"""通用工具:年龄表达式、类型转换、日期解析、分页计算。""" from __future__ import annotations from datetime import date, datetime from decimal import Decimal from math import ceil from typing import Any from sqlalchemy import Integer, func, text from sqlalchemy.sql.elements import ColumnElement from app.core.database import dialect_name from app.core.exceptions import BusinessError # ---------------------------------------------------------------- 年龄 def age_expression(birth_date_column, dialect: str | None = None) -> ColumnElement: """跨方言的"按当前日期计算年龄"SQL 表达式。 模型里不落 age 字段:年龄是随时间漂移的派生值,落库第二天就会错。 统计接口需要按年龄筛选,所以这里给出能在 WHERE / GROUP BY 里用的表达式。 """ dialect = dialect or dialect_name() if dialect == "sqlite": return func.cast( (func.julianday(func.current_date()) - func.julianday(birth_date_column)) / 365.25, Integer, ) # MySQL: TIMESTAMPDIFF(YEAR, birth_date, CURDATE()) return func.timestampdiff(text("YEAR"), birth_date_column, func.curdate()) def date_diff_days_expr(end_column, start_column, dialect: str | None = None) -> ColumnElement: """两个日期列相减得到天数,跨方言。 MySQL 用 DATEDIFF(end, start);SQLite 用 julianday 差值取整。 就业时长(需求 2.6.3)用它直接在 SQL 里算,不用把整表拉到内存。 """ dialect = dialect or dialect_name() if dialect == "sqlite": return func.cast(func.julianday(end_column) - func.julianday(start_column), Integer) return func.datediff(end_column, start_column) def calc_age(birth: date | None, today: date | None = None) -> int | None: """Python 侧算年龄(列表出参用)。""" if not birth: return None today = today or date.today() return today.year - birth.year - ((today.month, today.day) < (birth.month, birth.day)) def guess_birth_date(age: int, today: date | None = None) -> date: """只给了年龄没有生日时的兜底推算(取当年某月 1 日,避免 2/29 问题)。""" today = today or date.today() return date(today.year - age, 1, 1) # ---------------------------------------------------------------- 类型转换 def to_float(value: Any) -> float | None: if value is None: return None if isinstance(value, Decimal): return float(value) try: return float(value) except (TypeError, ValueError): return None def parse_date(value: Any, field: str = "日期") -> date | None: """schema 层日期一律用 str,转 date 在这一层做。""" if value in (None, "", "null"): return None if isinstance(value, datetime): return value.date() if isinstance(value, date): return value if isinstance(value, str): for fmt in ("%Y-%m-%d", "%Y/%m/%d", "%Y.%m.%d", "%Y%m%d"): try: return datetime.strptime(value.strip(), fmt).date() except ValueError: continue raise BusinessError(f"{field}格式不正确,应为 YYYY-MM-DD") def parse_amount(value: Any, field: str = "金额") -> Decimal | None: if value in (None, ""): return None try: return Decimal(str(value)) except Exception as exc: # noqa: BLE001 raise BusinessError(f"{field}必须是数字") from exc # ---------------------------------------------------------------- 分页 def page_count(total: int, page_size: int) -> int: if page_size <= 0: return 0 return ceil(total / page_size) if total else 0 def paginate_params(page: int, page_size: int) -> tuple[int, int]: page = max(page or 1, 1) page_size = min(max(page_size or 10, 1), 200) return page, page_size