Initial commit

xuexiao/ui_sort.py

@@ -0,0 +1,140 @@
+"""表格列排序：解析参数、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),
+    }