141 lines
4.9 KiB
Python
141 lines
4.9 KiB
Python
"""表格列排序：解析参数、SQL 排序、内存排序"""
|
||
from __future__ import annotations
|
||
|
||
from typing import Any, Callable, Optional
|
||
|
||
from sqlalchemy import func
|
||
from sqlalchemy.orm import Query
|
||
|
||
from models import MonthlySnapshot, Student, Course
|
||
|
||
|
||
def parse_sort(
|
||
sort: Optional[str],
|
||
order: Optional[str],
|
||
allowed: set[str],
|
||
default: str = '',
|
||
) -> tuple[str, str]:
|
||
s = (sort or '').strip()
|
||
o = (order or 'asc').strip().lower()
|
||
if o not in ('asc', 'desc'):
|
||
o = 'asc'
|
||
if s not in allowed:
|
||
s = default
|
||
return s, o
|
||
|
||
|
||
def toggle_order(current_sort: str, field: str, current_order: str) -> str:
|
||
return 'desc' if current_sort == field and current_order == 'asc' else 'asc'
|
||
|
||
|
||
def _course_label(acc) -> str:
|
||
if not acc or not getattr(acc, 'course', None):
|
||
return ''
|
||
c = acc.course
|
||
return (c.course_code or c.name or '').lower()
|
||
|
||
|
||
def _num(val) -> float:
|
||
if val is None:
|
||
return float('-inf')
|
||
try:
|
||
return float(val)
|
||
except (TypeError, ValueError):
|
||
return float('-inf')
|
||
|
||
|
||
def sort_rows_in_memory(
|
||
rows: list,
|
||
sort: str,
|
||
order: str,
|
||
key_map: dict[str, Callable[[Any], Any]],
|
||
) -> list:
|
||
if not sort or sort not in key_map:
|
||
return rows
|
||
reverse = order == 'desc'
|
||
return sorted(rows, key=key_map[sort], reverse=reverse)
|
||
|
||
|
||
def apply_snapshot_query_sort(
|
||
query: Query,
|
||
sort: str,
|
||
order: str,
|
||
) -> Query:
|
||
"""课时余额列表：数据库分页前排序"""
|
||
desc = order == 'desc'
|
||
col = func.coalesce
|
||
|
||
mapping = {
|
||
'student': Student.name,
|
||
'course': Course.course_code,
|
||
'end_lessons': col(MonthlySnapshot.end_lessons, MonthlySnapshot.prev_lessons),
|
||
'end_gift': col(MonthlySnapshot.end_gift_lessons, 0),
|
||
'end_total': col(MonthlySnapshot.end_total_lessons, 0),
|
||
'end_balance': col(MonthlySnapshot.end_balance, MonthlySnapshot.prev_balance),
|
||
'unit_price': col(MonthlySnapshot.unit_price, 0),
|
||
'consumed': col(MonthlySnapshot.consumed_lessons, 0),
|
||
}
|
||
expr = mapping.get(sort)
|
||
if expr is None:
|
||
return query.order_by(
|
||
MonthlySnapshot.seq_no.asc(),
|
||
Student.name.asc(),
|
||
MonthlySnapshot.id.asc(),
|
||
)
|
||
return query.order_by(expr.desc() if desc else expr.asc(), MonthlySnapshot.id.asc())
|
||
|
||
|
||
def snapshot_row_sort_keys(disp: dict, snap: MonthlySnapshot) -> dict[str, Callable]:
|
||
"""内存排序用（预览表、账户流水等）"""
|
||
return {
|
||
'end_lessons': lambda r: _num(
|
||
(r.get('disp') or {}).get('end_lessons')
|
||
if isinstance(r.get('disp'), dict)
|
||
else getattr(r.get('snap'), 'end_lessons', None),
|
||
),
|
||
'end_gift': lambda r: _num((r.get('disp') or {}).get('end_gift_lessons')),
|
||
'end_total': lambda r: _num((r.get('disp') or {}).get('end_total_lessons')),
|
||
'end_balance': lambda r: _num((r.get('disp') or {}).get('end_balance')),
|
||
'unit_price': lambda r: _num((r.get('disp') or {}).get('unit_price')),
|
||
'consumed': lambda r: _num(getattr(r.get('snap'), 'consumed_lessons', None)),
|
||
'student': lambda r: (getattr(getattr(r.get('acc'), 'student', None), 'name', '') or ''),
|
||
'course': lambda r: _course_label(r.get('acc')),
|
||
}
|
||
|
||
|
||
def account_ledger_sort_keys() -> dict[str, Callable]:
|
||
return {
|
||
'course': lambda x: _course_label(x[0]),
|
||
'end_lessons': lambda x: _num(x[1].get('end_lessons')),
|
||
'end_gift': lambda x: _num(x[1].get('end_gift_lessons')),
|
||
'end_total': lambda x: _num(x[1].get('end_total_lessons')),
|
||
'end_balance': lambda x: _num(x[1].get('end_balance')),
|
||
'unit_price': lambda x: _num(x[1].get('unit_price')),
|
||
'refund_due': lambda x: _num(x[2].get('refund_due') if x[2].get('has_original') else -1),
|
||
}
|
||
|
||
|
||
def preview_row_sort_keys() -> dict[str, Callable]:
|
||
return {
|
||
'student': lambda r: (r.get('student_name') or '').lower(),
|
||
'course': lambda r: (r.get('course') or '').lower(),
|
||
'end_lessons': lambda r: _num(r.get('end_lessons')),
|
||
'end_gift': lambda r: _num(r.get('end_gift_lessons')),
|
||
'end_total': lambda r: _num(r.get('end_total_lessons')),
|
||
'end_balance': lambda r: _num(r.get('end_balance')),
|
||
'unit_price': lambda r: _num(r.get('unit_price')),
|
||
}
|
||
|
||
|
||
def monthly_snapshot_sort_keys() -> dict[str, Callable]:
|
||
return {
|
||
'period': lambda s: (s.year, s.month),
|
||
'unit_price': lambda s: _num(s.unit_price),
|
||
'end_lessons': lambda s: _num(s.end_lessons),
|
||
'end_gift': lambda s: _num(s.end_gift_lessons),
|
||
'end_total': lambda s: _num(s.end_total_lessons),
|
||
'end_balance': lambda s: _num(s.end_balance),
|
||
'consumed': lambda s: _num(s.consumed_lessons),
|
||
'consumed_amount': lambda s: _num(s.consumed_amount),
|
||
}
|