"""Admin dashboard aggregates over existing B-43 eligible events and review queues. No fabricated demo amounts. Opening balances are unavailable until B-45; peer group ``opening`` is always null and ``ending`` equals period net change. Period closing status is unknown without the monthly-close module — returned as null so the UI shows an em dash rather than a guessed label. """ from __future__ import annotations from calendar import monthrange from datetime import datetime, timedelta, timezone from decimal import Decimal, ROUND_HALF_UP import sqlite3 WAN = Decimal("10000") ZERO = Decimal("0") TWOPLACES = Decimal("0.01") def today_shanghai() -> str: return datetime.now(timezone(timedelta(hours=8))).date().isoformat() def _month_bounds(year_month: str) -> tuple[str, str]: year, month = (int(part) for part in year_month.split("-")) last = monthrange(year, month)[1] return f"{year_month}-01", f"{year_month}-{last:02d}" def _q2(value: Decimal) -> str: return str(value.quantize(TWOPLACES, rounding=ROUND_HALF_UP)) def _to_wan(amount: Decimal) -> str: return _q2(amount / WAN) def _parse_day(iso_ts: str | None) -> str | None: if not iso_ts: return None text = str(iso_ts) return text[:10] if len(text) >= 10 else None def audit_counts(connection: sqlite3.Connection) -> dict[str, int]: """Derive pending review counts from real queues only. Priority mapping (aligned with current admin audit UI semantics): - high: unresolved / needs_review match exceptions - medium: pending bank-account registrations + pending manual records - low: reserved for future low-risk queues (currently always 0) Homepage card, sidebar badge and audit-center heading must all read this same payload (via ``/api/admin/dashboard`` → ``audit``). """ high = connection.execute( """ SELECT COUNT(*) AS n FROM current_transfer_decisions c JOIN transfer_match_decisions d ON d.id = c.decision_id JOIN canonical_transfer_events e ON e.id = c.event_id WHERE e.lifecycle = 'active' AND d.classification IN ('unresolved', 'needs_review') """ ).fetchone()["n"] medium_accounts = connection.execute( "SELECT COUNT(*) AS n FROM bank_accounts WHERE status = 'pending'" ).fetchone()["n"] medium_manuals = connection.execute( """ SELECT COUNT(*) AS n FROM manual_records m JOIN current_manual_record_decisions c ON c.record_id = m.id JOIN manual_record_decisions d ON d.id = c.decision_id WHERE d.state = 'pending' """ ).fetchone()["n"] medium = int(medium_accounts) + int(medium_manuals) low = 0 return { "total": int(high) + int(medium) + int(low), "high": int(high), "medium": int(medium), "low": int(low), } def _load_eligible( connection: sqlite3.Connection, *, from_date: str, cutoff: str ) -> list[sqlite3.Row]: return connection.execute( """ SELECT e.event_id, e.decision_id, e.effective_at, e.amount, e.currency, e.payer_company_id, e.payee_company_id, e.pairing, cpayer.name AS payer_name, cpayee.name AS payee_name FROM eligible_intercompany_events e JOIN companies cpayer ON cpayer.id = e.payer_company_id JOIN companies cpayee ON cpayee.id = e.payee_company_id WHERE date(e.effective_at) >= date(?) AND date(e.effective_at) <= date(?) ORDER BY e.effective_at, e.event_id """, (from_date, cutoff), ).fetchall() def _company_rows(connection: sqlite3.Connection) -> list[sqlite3.Row]: return connection.execute( """ SELECT id, name FROM companies WHERE status != 'disabled' ORDER BY id """ ).fetchall() def _period_status_for_cutoff( connection: sqlite3.Connection, cutoff: str ) -> tuple[str | None, str]: ym = str(cutoff or "")[:7] exists = connection.execute( "SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'period_close_runs'" ).fetchone() if exists is None: return None, "—" row = connection.execute( """ SELECT status FROM period_close_runs WHERE year_month = ? ORDER BY version DESC LIMIT 1 """, (ym,), ).fetchone() mapping = { "closed": ("closed", "已锁定"), "reopened": ("reopened", "已重开"), "failed": ("failed", "结账失败"), "pending": ("pending", "待结账"), "closing": ("closing", "处理中"), } if row is not None: return mapping.get(row["status"], (row["status"], str(row["status"]))) locked = connection.execute( "SELECT 1 FROM closed_periods WHERE year_month = ?", (ym,) ).fetchone() if locked is not None: return "closed", "已锁定" return None, "—" def company_summaries( connection: sqlite3.Connection, *, from_date: str, cutoff: str ) -> tuple[list[dict[str, object]], dict[str, object]]: companies = _company_rows(connection) events = _load_eligible(connection, from_date=from_date, cutoff=cutoff) period_status, period_label = _period_status_for_cutoff(connection, cutoff) debit_total = ZERO credit_total = ZERO by_id: dict[int, dict[str, object]] = {} for company in companies: by_id[int(company["id"])] = { "id": int(company["id"]), "name": company["name"], "detail_count": 0, "debit": ZERO, "credit": ZERO, "period_status": period_status, "period_status_label": period_label, } for event in events: amount = Decimal(str(event["amount"])) debit_total += amount credit_total += amount payer_id = int(event["payer_company_id"]) payee_id = int(event["payee_company_id"]) if payer_id in by_id: row = by_id[payer_id] row["detail_count"] = int(row["detail_count"]) + 1 row["debit"] = Decimal(row["debit"]) + amount if payee_id in by_id: row = by_id[payee_id] row["detail_count"] = int(row["detail_count"]) + 1 row["credit"] = Decimal(row["credit"]) + amount items: list[dict[str, object]] = [] for company in companies: row = by_id[int(company["id"])] debit = Decimal(row["debit"]) credit = Decimal(row["credit"]) # Payee inflow − payer outflow = signed net from this company's view. net = credit - debit items.append( { "id": row["id"], "name": row["name"], "detail_count": row["detail_count"], "debit_wan": _to_wan(debit), "credit_wan": _to_wan(credit), "net_wan": _to_wan(net), "period_status": row["period_status"], "period_status_label": row["period_status_label"], } ) totals = { "company_count": len(items), "detail_count": len(events), "debit_wan": _to_wan(debit_total), "credit_wan": _to_wan(credit_total), "net_wan": _to_wan(debit_total - credit_total), } return items, totals def period_progress( connection: sqlite3.Connection, *, year_month: str ) -> dict[str, object]: """Per-company coverage of the cutoff month: done / in progress / unsubmitted. Only companies with at least one active bank account are counted — the same rule as monthly-close submission checks. Covered = confirmed source rows in the month, or an approved no-business attestation that spans the month. """ start, end = _month_bounds(year_month) enabled_rows = connection.execute( "SELECT DISTINCT company_id FROM bank_accounts WHERE status = 'active'" ).fetchall() enabled_ids = {int(row["company_id"]) for row in enabled_rows} empty = { "year_month": year_month, "enabled_count": 0, "done": 0, "in_progress": 0, "unsubmitted": 0, "percent": 0, } if not enabled_ids: return empty submitted = { int(row["company_id"]) for row in connection.execute( """ SELECT DISTINCT b.company_id AS company_id FROM source_rows r JOIN sheet_batches s ON s.id = r.sheet_batch_id JOIN import_batches b ON b.id = s.import_batch_id JOIN sheet_reviews rv ON rv.sheet_batch_id = s.id AND rv.review_status = 'confirmed' WHERE date(r.transaction_at) >= date(?) AND date(r.transaction_at) <= date(?) """, (start, end), ).fetchall() } attested = { int(row["company_id"]) for row in connection.execute( """ SELECT DISTINCT a.company_id AS company_id FROM no_business_attestations a WHERE a.status = 'approved' AND date(a.gap_start) <= date(?) AND date(a.gap_end) >= date(?) """, (end, start), ).fetchall() } covered = (submitted | attested) & enabled_ids pending: set[int] = set() for row in connection.execute( """ SELECT DISTINCT p.company_id AS company_id FROM current_transfer_decisions c JOIN transfer_match_decisions d ON d.id = c.decision_id JOIN canonical_transfer_events e ON e.id = c.event_id JOIN transfer_decision_participants p ON p.decision_id = d.id WHERE e.lifecycle = 'active' AND d.classification IN ('unresolved', 'needs_review') AND date(d.effective_at) >= date(?) AND date(d.effective_at) <= date(?) """, (start, end), ): pending.add(int(row["company_id"])) for row in connection.execute( """ SELECT DISTINCT m.company_id AS company_id FROM manual_records m JOIN current_manual_record_decisions c ON c.record_id = m.id JOIN manual_record_decisions d ON d.id = c.decision_id WHERE d.state = 'pending' AND date(m.occurred_at) >= date(?) AND date(m.occurred_at) <= date(?) """, (start, end), ): pending.add(int(row["company_id"])) for row in connection.execute( "SELECT DISTINCT company_id FROM bank_accounts WHERE status = 'pending'" ): pending.add(int(row["company_id"])) pending_enabled = pending & enabled_ids done = covered - pending_enabled in_progress = pending_enabled unsubmitted = enabled_ids - covered - pending_enabled enabled_count = len(enabled_ids) percent = round(100 * len(done) / enabled_count) if enabled_count else 0 return { "year_month": year_month, "enabled_count": enabled_count, "done": len(done), "in_progress": len(in_progress), "unsubmitted": len(unsubmitted), "percent": percent, } def company_peer_groups( connection: sqlite3.Connection, company_id: int, *, from_date: str, cutoff: str, ) -> dict[str, object]: company = connection.execute( "SELECT id, name FROM companies WHERE id = ?", (company_id,) ).fetchone() if company is None: raise KeyError(company_id) events = connection.execute( """ SELECT e.event_id, e.effective_at, e.amount, e.currency, e.payer_company_id, e.payee_company_id, cpayer.name AS payer_name, cpayee.name AS payee_name FROM eligible_intercompany_events e JOIN companies cpayer ON cpayer.id = e.payer_company_id JOIN companies cpayee ON cpayee.id = e.payee_company_id WHERE (e.payer_company_id = ? OR e.payee_company_id = ?) AND date(e.effective_at) >= date(?) AND date(e.effective_at) <= date(?) ORDER BY e.effective_at, e.event_id """, (company_id, company_id, from_date, cutoff), ).fetchall() groups: dict[int, dict[str, object]] = {} for event in events: amount = Decimal(str(event["amount"])) payer_id = int(event["payer_company_id"]) payee_id = int(event["payee_company_id"]) if payer_id == company_id: peer_id = payee_id peer_name = event["payee_name"] direction = "debit" summary = f"付往 {peer_name}" else: peer_id = payer_id peer_name = event["payer_name"] direction = "credit" summary = f"收自 {peer_name}" bucket = groups.get(peer_id) if bucket is None: bucket = { "peer_id": peer_id, "peer_name": peer_name, "count": 0, "opening": None, "opening_status": "unavailable", "debit": ZERO, "credit": ZERO, "lines": [], } groups[peer_id] = bucket bucket["count"] = int(bucket["count"]) + 1 if direction == "debit": bucket["debit"] = Decimal(bucket["debit"]) + amount else: bucket["credit"] = Decimal(bucket["credit"]) + amount day = _parse_day(event["effective_at"]) or "" bucket["lines"].append( { "event_id": int(event["event_id"]), "date": day, "direction": direction, "summary": summary, "amount_wan": _to_wan(amount), "currency": event["currency"] or "CNY", } ) result_groups: list[dict[str, object]] = [] for peer_id in sorted(groups.keys(), key=lambda i: groups[i]["peer_name"]): bucket = groups[peer_id] debit = Decimal(bucket["debit"]) credit = Decimal(bucket["credit"]) ending = credit - debit result_groups.append( { "peer_id": bucket["peer_id"], "peer_name": bucket["peer_name"], "count": bucket["count"], "opening": None, "opening_status": "unavailable", "debit_wan": _to_wan(debit), "credit_wan": _to_wan(credit), "ending_wan": _to_wan(ending), "result_kind": "period_net_change", "lines": bucket["lines"], } ) return { "company_id": int(company["id"]), "company_name": company["name"], "from_date": from_date, "cutoff": cutoff, "groups": result_groups, } def weekly_flow( connection: sqlite3.Connection, *, cutoff: str, days: int = 7 ) -> dict[str, object]: end = datetime.strptime(cutoff, "%Y-%m-%d").date() start = end - timedelta(days=days - 1) labels: list[str] = [] inflow = [ZERO] * days outflow = [ZERO] * days index: dict[str, int] = {} for offset in range(days): day = start + timedelta(days=offset) key = day.isoformat() index[key] = offset labels.append(f"{day.month:02d}-{day.day:02d}") rows = connection.execute( """ SELECT date(e.effective_at) AS day, e.amount FROM eligible_intercompany_events e WHERE date(e.effective_at) >= date(?) AND date(e.effective_at) <= date(?) """, (start.isoformat(), cutoff), ).fetchall() for row in rows: day = row["day"] if day not in index: continue amount = Decimal(str(row["amount"])) # Group-level flow: every eligible transfer is both an outflow (payer) # and an inflow (payee); plot both series with the same absolute amount. inflow[index[day]] += amount outflow[index[day]] += amount return { "labels": labels, "inflow_wan": [_to_wan(v) for v in inflow], "outflow_wan": [_to_wan(v) for v in outflow], } def build_dashboard( connection: sqlite3.Connection, *, from_date: str = "2026-01-01", cutoff: str | None = None, ) -> dict[str, object]: cutoff_date = cutoff or today_shanghai() try: datetime.strptime(from_date, "%Y-%m-%d") datetime.strptime(cutoff_date, "%Y-%m-%d") except ValueError as exc: raise ValueError("日期必须是 YYYY-MM-DD") from exc if from_date > cutoff_date: raise ValueError("from 不能晚于 cutoff") period = datetime.strptime(cutoff_date, "%Y-%m-%d") # Display month for the status column header: use the calendar month of cutoff. period_month = period.month companies, totals = company_summaries( connection, from_date=from_date, cutoff=cutoff_date ) period_label = f"{period.year}-{period.month:02d}" return { "from_date": from_date, "cutoff": cutoff_date, "period_month": period_month, "period_label": period_label, "audit": audit_counts(connection), "totals": totals, "period_progress": period_progress(connection, year_month=period_label), "companies": companies, "weekly_flow": weekly_flow(connection, cutoff=cutoff_date), "opening_status": "unavailable", }