"""表格列排序:解析参数、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), }