109 lines
3.8 KiB
Python
109 lines
3.8 KiB
Python
"""通用工具：年龄表达式、类型转换、日期解析、分页计算。"""
|
||
|
||
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
|