From 71867c4c7a8183e5bbe1f60759be9a78f114705a Mon Sep 17 00:00:00 2001 From: fullstack-bot Date: Tue, 20 Oct 2026 23:41:31 +0800 Subject: [PATCH] =?UTF-8?q?feat(ar):=20=E5=BA=94=E6=94=B6=E8=B4=A6?= =?UTF-8?q?=E9=BE=84=E7=9C=8B=E6=9D=BF=20+=20=E9=80=BE=E6=9C=9F=E9=A2=84?= =?UTF-8?q?=E8=AD=A6=20overdue=20=E6=8E=A5=E5=85=A5=EF=BC=88=E9=98=B6?= =?UTF-8?q?=E6=AE=B5A=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit T1 逾期预警（§6 财务 2026-10-20 23:23 口径）： - alert_rules 支持 rule_type=ar_overdue；阈值/责任人/排除名单全量走 params（代码零业务常量） - 90 天占比红线 key 保留但默认 null 停用（不回落 0.25）；黄线 15%-25% 保留 - 新增逐户红线「>30 天且 >5 万」+ 聚合红线「150+ 占比环比上升 = 结构恶化」 - assignee 派生：manual → 近 3 个月主业务员（zxt_masterbill emp_id → employee_dict.u_Name， 缺表时降级同库摘要制单人姓名）→ 部门负责人（渠道 118 / 商务 114 / 直属酒窖 117）； 两路皆失败 ⇒ 该户不生成、进 skipped[]，禁落空 assignee - POST /run-ar-overdue（dry_run 默认 true）+ check-all 分支；只 INSERT 同一 message 幂等 T2 账龄看板： - GET /api/cma/ar-aging/{meta,trend,overview,customers,customers/export}；每条查询带 entity_id - 三口径分列、caliber=all → 400；缺期 has_data=false 且金额 null（不补 0） - ArAging.vue：口径后缀/截止期徽标/趋势断点/六档堆叠/逐户可排序导出 单测 21 项全通过（SQLite 夹具，不连生产库）；vite build 通过。 阶段A 零生产写：未 push、未部署、未 --apply、未 dry_run=false。 --- backend/app/api/alert_rules.py | 177 ++++- backend/app/api/ar_aging.py | 467 ++++++++++++ backend/app/main.py | 3 +- backend/app/services/ar_overdue_alert.py | 706 ++++++++++++++++++ backend/app/utils/ar_caliber.py | 71 ++ backend/scripts/seed_ar_overdue_rule.py | 157 ++++ .../tests/test_ar_aging_overdue_20261020.py | 480 ++++++++++++ frontend/src/api/index.ts | 15 + frontend/src/permission.ts | 1 + frontend/src/router/index.ts | 1 + frontend/src/views/ArAging.vue | 478 ++++++++++++ 11 files changed, 2552 insertions(+), 4 deletions(-) create mode 100644 backend/app/api/ar_aging.py create mode 100644 backend/app/services/ar_overdue_alert.py create mode 100644 backend/app/utils/ar_caliber.py create mode 100644 backend/scripts/seed_ar_overdue_rule.py create mode 100644 backend/tests/test_ar_aging_overdue_20261020.py create mode 100644 frontend/src/views/ArAging.vue diff --git a/backend/app/api/alert_rules.py b/backend/app/api/alert_rules.py index 419a792e..c56cc3bf 100644 --- a/backend/app/api/alert_rules.py +++ b/backend/app/api/alert_rules.py @@ -2,7 +2,7 @@ 预警规则智能化 — 任务6 后端组件: alert_rules 模型 + API + 预警引擎 """ -from fastapi import APIRouter, Depends, HTTPException, Query +from fastapi import APIRouter, Body, Depends, HTTPException, Query from sqlalchemy.orm import Session from sqlalchemy import text, Column, Integer, String, Text, Float, DateTime, JSON, Boolean, func from typing import Optional, List @@ -26,6 +26,8 @@ from app.utils.cash_thresholds import ( CASH_GAP_RULE_TYPE, validate_params as validate_cash_gap_params, ) +# 应收账龄逾期预警（T1 · 2026-10-20）：计算/落库单源在服务层，阈值全走 alert_rules.params +from app.services import ar_overdue_alert logger = logging.getLogger("alert_rules") @@ -38,6 +40,7 @@ VALID_RULE_TYPES = ( "forecast_deviation", # 预测偏差（升级2b 2026-08-25） "customer_concentration", # 客户集中度（P2 2026-09-10） "cash_gap_threshold", # 资金缺口阈值配置（G1 2026-10-17，配置行，不进预警生成链） + "ar_overdue", # 应收账龄逾期（T1 2026-10-20：90天以上占比 / 逐户逾期额 / 逾期户数环比） ) # ============================================================ @@ -142,6 +145,11 @@ def create_alert_rule(data: dict, db: Session = Depends(get_db), user=Depends(re err = validate_cash_gap_params(data.get("params")) if err: raise HTTPException(400, err) + # T1（2026-10-20）：ar_overdue 参数只校结构（阈值/责任人全量由 DB 行给，代码不设默认值） + if rule_type == ar_overdue_alert.RULE_TYPE: + err = ar_overdue_alert.validate_params(data.get("params")) + if err: + raise HTTPException(400, err) rule = AlertRule( entity_id=entity_id, @@ -180,6 +188,12 @@ def update_alert_rule(rule_id: int, data: dict, db: Session = Depends(get_db), e if err: db.rollback() raise HTTPException(400, err) + # T1（2026-10-20）：ar_overdue 改参同样只校结构（阈值可现场改，无需发版） + if rule.rule_type == ar_overdue_alert.RULE_TYPE: + err = ar_overdue_alert.validate_params(rule.params) + if err: + db.rollback() + raise HTTPException(400, err) db.commit() db.refresh(rule) return {"data": {c.name: getattr(rule, c.name) for c in AlertRule.__table__.columns}} @@ -307,6 +321,31 @@ def run_all_alert_checks(db: Session = Depends(get_db), entity_id: int = Depends # 不进 KPIValue 比价链（否则未来若绑定到有值的 KPI 会误生成预警）。 if rule.rule_type == CASH_GAP_RULE_TYPE: continue + # T1（2026-10-20）：应收账龄逾期 —— 与 POST /run-ar-overdue 同一计算函数， + # 不走 KPIValue 比价链（F_AR_002 无本期 kpi_values 也不得报错）。 + if rule.rule_type == ar_overdue_alert.RULE_TYPE: + rp = rule.params + if isinstance(rp, str): + rp = json.loads(rp) if rp else {} + rp = rp or {} + try: + ar_overdue_alert.ensure_assignee_configured(rp) + res = ar_overdue_alert.compute(db, entity_id, None, rp) + except (ar_overdue_alert.AssigneeMissing, ar_overdue_alert.ParamsInvalid) as exc: + logger.warning("ar_overdue 规则跳过（rule_id=%s）: %s", rule.id, exc) + continue + if res.get("skipped_reason"): + logger.info("ar_overdue 规则无数据跳过（rule_id=%s）: %s", + rule.id, res["skipped_reason"]) + continue + try: + w = ar_overdue_alert.write_alerts(db, entity_id, rule.kpi_id, res) + except Exception: # noqa: BLE001 — 单规则失败不拖垮整轮检查 + db.rollback() + logger.exception("ar_overdue 落库失败（rule_id=%s）", rule.id) + continue + alerts_generated += w["inserted"] + continue kpi = kpi_cache.get(rule.kpi_id) if kpi is None: kpi = db.query(KPIDefinition).filter(KPIDefinition.id == rule.kpi_id).first() @@ -316,7 +355,7 @@ def run_all_alert_checks(db: Session = Depends(get_db), entity_id: int = Depends if not kpi or kpi.entity_id != entity_id: continue - import json; params = json.loads(rule.params) if isinstance(rule.params, str) else (rule.params or {}) + params = json.loads(rule.params) if isinstance(rule.params, str) else (rule.params or {}) # 客户集中度规则（P2 2026-09-10）：不依赖同期KPIValue，直接看客户收入占比 if rule.rule_type == "customer_concentration": @@ -655,7 +694,7 @@ def _check_forecast_alerts(db: Session, entity_id: int = 1) -> int: continue # 检查预测值是否超限 - import json; params = json.loads(rule.params) if isinstance(rule.params, str) else (rule.params or {}) + params = json.loads(rule.params) if isinstance(rule.params, str) else (rule.params or {}) params["kpi"] = kpi for forecast in latest_forecasts: value = forecast.predicted_cash @@ -979,3 +1018,135 @@ def run_customer_concentration_check( continue db.commit() return {"message": f"客户集中度检查完成: {generated}条", "generated": generated} + + +# ============================================================ +# T1（2026-10-20）：应收账龄逾期预警 —— 运行入口 + 手工触发 +# ============================================================ +def _resolve_ar_kpi(db: Session, entity_id: int, params: dict, rule, override_kpi_id): + """解析 ar_overdue 绑定的 KPI：规则行 → 显式 override（仅 dry-run）→ params.kpi_code。""" + if rule is not None and rule.kpi_id: + kpi = db.query(KPIDefinition).filter( + KPIDefinition.id == rule.kpi_id, KPIDefinition.entity_id == entity_id).first() + return (kpi.id if kpi else rule.kpi_id), (kpi.kpi_code if kpi else None), "alert_rules.kpi_id" + if override_kpi_id: + kpi = db.query(KPIDefinition).filter( + KPIDefinition.id == override_kpi_id, KPIDefinition.entity_id == entity_id).first() + if not kpi: + raise HTTPException(400, f"KPI 不存在或不属于当前账套: kpi_id={override_kpi_id}") + return kpi.id, kpi.kpi_code, "query.kpi_id(dry-run override)" + code = (params or {}).get("kpi_code") + if code: + kpi = db.query(KPIDefinition).filter( + KPIDefinition.kpi_code == code, KPIDefinition.entity_id == entity_id).first() + if kpi: + return kpi.id, kpi.kpi_code, f"params.kpi_code={code}" + return None, code, f"params.kpi_code={code}(库内不存在)" + return None, None, "unresolved" + + +@router.post("/run-ar-overdue") +def run_ar_overdue_check( + period: Optional[str] = Query(None, description="账龄基准期 YYYY-MM；缺省=表内 MAX(period)"), + dry_run: bool = Query(True, description="默认 true：只算不写。false 走既有写通道插入 kpi_alerts"), + kpi_id: Optional[int] = Query(None, description="dry-run 专用：临时绑定 KPI（正式以规则行 kpi_id 为准）"), + data: Optional[dict] = Body(None, description="dry-run 专用：{'params': {...}} 覆盖规则参数（证明参数化，不碰 DB）"), + db: Session = Depends(get_db), + entity_id: int = Depends(get_entity_id), +): + """应收账龄逾期预警运行（T1）。 + + 参数优先级：dry-run body 覆盖 > `alert_rules` 行（rule_type='ar_overdue'）> 无规则 ⇒ 200 skipped。 + 阈值/责任人/排除名单全部来自参数；`assignee` 缺失 ⇒ 400（fail-closed）。 + `dry_run=false` 时禁止任何参数覆盖（防绕过 DB 配置），且只 INSERT 不 UPDATE/DELETE。 + """ + if is_excluded_entity(db, entity_id): + return {"message": f"账套 {entity_id} 为 demo 演示账套，按 R1 不参与预警生成", + "generated": 0, "skipped": True, "reason": "R1_demo_entity_excluded"} + + override_params = None + if isinstance(data, dict): + override_params = data.get("params") if isinstance(data.get("params"), dict) else data.get("params_override") + if (override_params is not None or kpi_id is not None) and not dry_run: + raise HTTPException( + 400, "dry_run=false 时禁止参数/KPI 覆盖：阈值与绑定必须来自 alert_rules 行（防绕过 DB 配置）") + + rule = db.query(AlertRule).filter( + AlertRule.entity_id == entity_id, + AlertRule.rule_type == ar_overdue_alert.RULE_TYPE, + AlertRule.enabled == 1, + ).order_by(AlertRule.id).first() + + params_source = "dry_run.body.params" + if override_params is not None: + params = override_params + elif rule is not None: + params = rule.params + if isinstance(params, str): + params = json.loads(params) if params else {} + params = params or {} + params_source = f"alert_rules#{rule.id}" + else: + return { + "dry_run": dry_run, "entity_id": entity_id, "period": period, + "skipped": True, "skipped_reason": "no_ar_overdue_rule", + "hint": "无 rule_type='ar_overdue' 规则行；阶段B 用 backend/scripts/seed_ar_overdue_rule.py 播种，" + "或 dry-run 时在 body 传 {'params': {...}} 演练", + } + + err = ar_overdue_alert.validate_params(params) + if err: + raise HTTPException(400, err) + try: + ar_overdue_alert.ensure_assignee_configured(params) + except ar_overdue_alert.AssigneeMissing as e: + raise HTTPException(400, str(e)) + + resolved_kpi_id, kpi_code, binding = _resolve_ar_kpi(db, entity_id, params, rule, kpi_id) + + try: + result = ar_overdue_alert.compute(db, entity_id, period, params) + except ar_overdue_alert.ParamsInvalid as e: + raise HTTPException(400, str(e)) + + hits = ar_overdue_alert.hits(result) + blockers = [] + if resolved_kpi_id is None: + blockers.append("kpi_id_unresolved:%s" % (kpi_code or "-")) + + payload = { + "dry_run": dry_run, + "entity_id": entity_id, + "period": result.get("period"), + "caliber": result.get("caliber"), + "params_source": params_source, + "kpi_id": resolved_kpi_id, + "kpi_code": kpi_code, + "kpi_binding": binding, + "skipped_reason": result.get("skipped_reason"), + "metrics": result.get("metrics"), + "aggregates": result.get("aggregates"), + "per_customer": result.get("per_customer"), + "skipped": result.get("skipped"), + "hit_count": len(hits), + "notes": result.get("notes"), + } + if result.get("skipped_reason"): + payload.update({"would_insert": 0, "write_blockers": blockers}) + return payload + + if dry_run: + payload.update({"would_insert": 0 if blockers else len(hits), "write_blockers": blockers}) + return payload + + if blockers: + raise HTTPException(400, "无法落库：" + "; ".join(blockers)) + written = ar_overdue_alert.write_alerts(db, entity_id, resolved_kpi_id, result) + log_op(db, "run_ar_overdue_alert", target_type="alert", target_id=resolved_kpi_id, + detail={"period": result.get("period"), "caliber": result.get("caliber"), + "hit": len(hits), **written}, + entity_id=entity_id) + db.commit() + payload.update({"would_insert": len(hits), "write_blockers": [], **written, + "generated": written["inserted"]}) + return payload diff --git a/backend/app/api/ar_aging.py b/backend/app/api/ar_aging.py new file mode 100644 index 00000000..d978b29f --- /dev/null +++ b/backend/app/api/ar_aging.py @@ -0,0 +1,467 @@ +"""应收账龄看板 API（T2 · 2026-10-20）— prefix `/api/cma/ar-aging` + +数据源：`hanke_ar_aging`（逐户·逐期·三口径分列，财务口径见 `app.utils.ar_caliber`）。 + +纪律（全部硬约束）： +1. **每条查询都带 `entity_id` 过滤**（防两企业混报）；本模块只读，无 ORM 写模型/无迁移。 +2. 三口径**禁合计**：`caliber=all` 或未知字面值 → 400；默认口径 = `借方应收户口径`。 +3. 缺期显「该期无数据」：`has_data=false` 且金额字段为 `null`（**禁补 0**）。 +4. `opening_amt`（期初）**只展示**，单独返回并标 `non_aging_base=true`。 +5. 表不存在/无权限 ⇒ `available=false` + 原因，**HTTP 200 不 500**。 +""" +from __future__ import annotations + +import csv +import io +import logging +from datetime import datetime +from decimal import Decimal, ROUND_HALF_UP +from typing import Any, Dict, List, Optional + +from fastapi import APIRouter, Depends, HTTPException, Query +from sqlalchemy import text +from sqlalchemy.orm import Session + +from app.database import get_db +from app.deps import get_entity_id +from app.auth_middleware import require_role +from app.utils import ar_caliber + +logger = logging.getLogger("ar_aging") + +router = APIRouter( + prefix="/api/cma/ar-aging", + tags=["应收账龄看板"], + dependencies=[Depends(require_role("ceo", "finance", "business"))], +) + +_CENT = Decimal("0.01") +_TABLE = ar_caliber.AR_AGING_TABLE +_BUCKET_COLS = ar_caliber.BUCKET_COLUMNS +_SORTABLE = { + "unit_code", "unit_name", "total_amt", "overdue_amt", "opening_amt", "oldest_days", + *_BUCKET_COLS, +} + + +# ============================================================ +# 公共helper +# ============================================================ +def _resolve_caliber(caliber: Optional[str]) -> str: + try: + return ar_caliber.validate_caliber(caliber, allow_none=True) + except ar_caliber.CaliberError as e: + raise HTTPException(400, str(e)) + + +def _dec(v: Any) -> Decimal: + return Decimal("0") if v is None else Decimal(str(v)) + + +def _money(v: Any) -> Optional[float]: + if v is None: + return None + return float(_dec(v).quantize(_CENT, rounding=ROUND_HALF_UP)) + + +def _money_str(v: Any) -> str: + return "{:.2f}".format(_dec(v).quantize(_CENT, rounding=ROUND_HALF_UP)) + + +def _available(db: Session) -> tuple[bool, Optional[str]]: + try: + db.execute(text("SELECT 1 FROM %s LIMIT 1" % _TABLE)).first() + return True, None + except Exception as e: # noqa: BLE001 — 表缺失/无权限一律降级 + db.rollback() + logger.warning("hanke_ar_aging 不可用: %s", e) + return False, "表 %s 不可用（不存在或无 SELECT 权限）: %s" % (_TABLE, str(e).split("\n")[0][:180]) + + +def _periods(db: Session, entity_id: int) -> List[str]: + rows = db.execute( + text("SELECT DISTINCT period FROM %s WHERE entity_id = :e ORDER BY period" % _TABLE), + {"e": entity_id}, + ).all() + return [r[0] for r in rows] + + +def _month_add(period: str, delta: int) -> str: + y, m = int(period[:4]), int(period[5:7]) + idx = y * 12 + (m - 1) + delta + return "%04d-%02d" % (idx // 12, idx % 12 + 1) + + +def _month_range(start: str, end: str) -> List[str]: + if start > end: + return [] + out, cur = [], start + while cur <= end: + out.append(cur) + cur = _month_add(cur, 1) + return out + + +def _rows(db: Session, entity_id: int, caliber: str, period: str) -> List[Dict[str, Any]]: + sql = ( + "SELECT id, period, caliber, unit_code, unit_name, opening_amt, total_amt, %s " + "FROM %s WHERE entity_id = :e AND caliber = :c AND period = :p " + "ORDER BY total_amt DESC, unit_code ASC" + % (", ".join(_BUCKET_COLS), _TABLE) + ) + return [dict(r) for r in db.execute( + text(sql), {"e": entity_id, "c": caliber, "p": period}).mappings().all()] + + +def _overdue_amt(row: Dict[str, Any]) -> Decimal: + """逾期额（>30 天）= total_amt − b_0_30；负档位不 ABS/不 COALESCE。""" + return _dec(row.get("total_amt")) - _dec(row.get("b_0_30")) + + +def _oldest_days(row: Dict[str, Any]) -> int: + days = 0 + for col, lower in ar_caliber.OVERDUE_BUCKETS: + if _dec(row.get(col)) != 0: + days = max(days, lower) + return days + + +def _thresholds(db: Session, entity_id: int) -> Dict[str, Any]: + """看板阈值标注来源：`alert_rules(rule_type='ar_overdue')` 行（无规则 ⇒ 全 null，前端显「阈值未配置」）。 + + 只读、只取展示用；判定仍以 `app.services.ar_overdue_alert` 单源为准。 + """ + from app.api.alert_rules import AlertRule + rule = db.query(AlertRule).filter( + AlertRule.entity_id == entity_id, + AlertRule.rule_type == "ar_overdue", + AlertRule.enabled == 1, + ).order_by(AlertRule.id).first() + if not rule: + return {"source": None, "share_90plus_red_gt": None, + "share_90plus_yellow_min": None, "share_90plus_yellow_max": None} + params = rule.params or {} + red = params.get("red") or {} + yellow = params.get("yellow") or {} + return { + "source": "alert_rules#%s" % rule.id, + "share_90plus_red_gt": red.get("share_90plus_gt"), + "share_90plus_yellow_min": yellow.get("share_90plus_min"), + "share_90plus_yellow_max": yellow.get("share_90plus_max"), + } + + +def _enrich(row: Dict[str, Any]) -> Dict[str, Any]: + total = _dec(row.get("total_amt")) + od = _overdue_amt(row) + o90 = sum((_dec(row.get(c)) for c in ar_caliber.NINETY_PLUS_COLUMNS), Decimal("0")) + out = { + "unit_code": row.get("unit_code"), + "unit_name": row.get("unit_name"), + "opening_amt": _money(row.get("opening_amt")), + "total_amt": _money(total), + "buckets": {col: _money(row.get(col)) for col in _BUCKET_COLS}, + "overdue_amt": _money(od), + "overdue_90plus": _money(o90), + "oldest_days": _oldest_days(row), + "pct_of_unit": (round(float(od / total), 6) if total > 0 else None), + "is_overdue": od > 0, + } + return out + + +# ============================================================ +# GET /meta +# ============================================================ +@router.get("/meta") +def get_meta(db: Session = Depends(get_db), entity_id: int = Depends(get_entity_id)): + """看板元信息：可用性、口径全集、可用期间、最新期、入表时刻、来源件。""" + ok, reason = _available(db) + base = { + "entity_id": entity_id, + "table": _TABLE, + "calibers": list(ar_caliber.CALIBERS), + "default_caliber": ar_caliber.DEFAULT_CALIBER, + "available": ok, + "reason": reason, + "periods_available": [], + "latest_period": None, + "case_insensitive_note": "口径字面值逐字匹配；贷方口径为 `贷方户单列(净额)`", + } + if not ok: + return base + + periods = _periods(db, entity_id) + base["periods_available"] = periods + base["latest_period"] = periods[-1] if periods else None + row = db.execute( + text("SELECT COUNT(*) n, MAX(loaded_at) loaded, MIN(period) pmin, MAX(period) pmax " + "FROM %s WHERE entity_id = :e" % _TABLE), + {"e": entity_id}, + ).mappings().first() + loaded = row["loaded"] + base.update({ + "row_count": int(row["n"] or 0), + "loaded_at": (loaded.isoformat(sep=" ") if hasattr(loaded, "isoformat") else (str(loaded) if loaded else None)), + "period_min": row["pmin"], + "period_max": row["pmax"], + }) + src = db.execute( + text("SELECT DISTINCT source_file FROM %s WHERE entity_id = :e ORDER BY source_file" % _TABLE), + {"e": entity_id}, + ).all() + base["source_files"] = [r[0] for r in src] + if not periods: + base["reason"] = "表内该账套无数据" + return base + + +# ============================================================ +# GET /trend +# ============================================================ +@router.get("/trend") +def get_trend( + caliber: Optional[str] = Query(None), + start: Optional[str] = Query(None, description="起始期 YYYY-MM"), + end: Optional[str] = Query(None, description="结束期 YYYY-MM（缺省=表内最新期与当前月取大）"), + db: Session = Depends(get_db), + entity_id: int = Depends(get_entity_id), +): + """逐期趋势（总额/户数/六档）；无数据期 has_data=false 且金额 null（禁补 0）。""" + cal = _resolve_caliber(caliber) + ok, reason = _available(db) + if not ok: + return {"available": False, "reason": reason, "entity_id": entity_id, + "caliber": cal, "points": []} + + periods = _periods(db, entity_id) + if not periods: + return {"available": True, "entity_id": entity_id, "caliber": cal, + "periods_available": [], "points": [], "reason": "表内该账套无数据"} + + now_period = datetime.now().strftime("%Y-%m") + start_p = start or periods[0] + end_p = end or max(periods[-1], now_period) + for p in (start_p, end_p): + if len(p) != 7 or p[4] != "-" or not (p[:4].isdigit() and p[5:].isdigit()): + raise HTTPException(400, f"期间格式非法（要求 YYYY-MM）: {p}") + + points = [] + for p in _month_range(start_p, end_p): + rows = _rows(db, entity_id, cal, p) + if not rows: + points.append({ + "period": p, "has_data": False, + "total_amt": None, "customers": None, "overdue_amt": None, + "buckets": {col: None for col in _BUCKET_COLS}, + "note": "该期无数据", + }) + continue + total = sum((_dec(r.get("total_amt")) for r in rows), Decimal("0")) + points.append({ + "period": p, + "has_data": True, + "total_amt": _money(total), + "customers": len(rows), + "overdue_amt": _money(sum((_overdue_amt(r) for r in rows), Decimal("0"))), + "buckets": {col: _money(sum((_dec(r.get(col)) for r in rows), Decimal("0"))) + for col in _BUCKET_COLS}, + "note": None, + }) + return { + "available": True, "entity_id": entity_id, "caliber": cal, + "start": start_p, "end": end_p, + "periods_available": periods, + "points": points, + } + + +# ============================================================ +# GET /overview +# ============================================================ +@router.get("/overview") +def get_overview( + period: Optional[str] = Query(None), + caliber: Optional[str] = Query(None), + db: Session = Depends(get_db), + entity_id: int = Depends(get_entity_id), +): + """期间概览：合计/户数/六档/逾期额/90天以上占比/环比；opening_amt 单列且标非账龄基数。""" + cal = _resolve_caliber(caliber) + th = _thresholds(db, entity_id) + ok, reason = _available(db) + if not ok: + return {"available": False, "reason": reason, "entity_id": entity_id, + "caliber": cal, "period": period, "has_data": False, "thresholds": th} + + periods = _periods(db, entity_id) + if not periods: + return {"available": True, "entity_id": entity_id, "caliber": cal, + "period": period, "has_data": False, "reason": "表内该账套无数据", + "thresholds": th} + p = period or periods[-1] + rows = _rows(db, entity_id, cal, p) + if not rows: + return { + "available": True, "entity_id": entity_id, "caliber": cal, "period": p, + "has_data": False, "reason": "该期无数据", + "total_amt": None, "customers": None, "overdue_amt": None, + "overdue_90plus": None, "share_90plus": None, + "buckets": {col: None for col in _BUCKET_COLS}, + "opening_amt": None, "opening_amt_non_aging_base": True, + "overdue_customers": None, "prev_period": None, "prev_overdue_amt": None, + "overdue_amt_mom_pct": None, "overdue_customers_mom_pct": None, + "thresholds": th, + } + + total = sum((_dec(r.get("total_amt")) for r in rows), Decimal("0")) + buckets = {col: sum((_dec(r.get(col)) for r in rows), Decimal("0")) for col in _BUCKET_COLS} + od = sum((_overdue_amt(r) for r in rows), Decimal("0")) + o90 = sum(buckets[c] for c in ar_caliber.NINETY_PLUS_COLUMNS) + overdue_customers = sum(1 for r in rows if _overdue_amt(r) > 0) + + idx = periods.index(p) if p in periods else -1 + prev_period = periods[idx - 1] if idx > 0 else None + prev_rows = _rows(db, entity_id, cal, prev_period) if prev_period else [] + prev_od = sum((_overdue_amt(r) for r in prev_rows), Decimal("0")) if prev_rows else None + prev_customers = (sum(1 for r in prev_rows if _overdue_amt(r) > 0) if prev_rows else None) + + def _mom(cur: Optional[Decimal], prev: Optional[Decimal]) -> Optional[float]: + if cur is None or prev is None or prev == 0: + return None + return round(float((cur - prev) / prev), 6) + + return { + "available": True, + "entity_id": entity_id, + "caliber": cal, + "period": p, + "has_data": True, + "total_amt": _money(total), + "customers": len(rows), + "buckets": {col: _money(v) for col, v in buckets.items()}, + "buckets_sum": _money(sum(buckets.values())), + "buckets_sum_matches_total": _money(sum(buckets.values())) == _money(total), + "overdue_amt": _money(od), + "overdue_90plus": _money(o90), + "share_90plus": (round(float(o90 / total), 6) if total > 0 else None), + "overdue_customers": overdue_customers, + # 期初只展示：禁作账龄基数、禁进分子分母（财务定死） + "opening_amt": _money(sum((_dec(r.get("opening_amt")) for r in rows), Decimal("0"))), + "opening_amt_non_aging_base": True, + "prev_period": prev_period, + "prev_has_data": bool(prev_rows), + "prev_overdue_amt": _money(prev_od) if prev_od is not None else None, + "prev_overdue_customers": prev_customers, + "overdue_amt_mom_pct": _mom(od, prev_od), + "overdue_customers_mom_pct": ( + _mom(Decimal(overdue_customers), Decimal(prev_customers)) + if prev_customers is not None else None), + "mom_note": None if prev_rows else "上期无数据，不计算环比", + "thresholds": th, + } + + +# ============================================================ +# GET /customers +# ============================================================ +@router.get("/customers") +def get_customers( + period: Optional[str] = Query(None), + caliber: Optional[str] = Query(None), + overdue_only: bool = Query(False), + sort: str = Query("overdue_amt"), + order: str = Query("desc"), + db: Session = Depends(get_db), + entity_id: int = Depends(get_entity_id), +): + """逐户明细（六档 + 逾期额 + 最老档），服务端排序。""" + cal = _resolve_caliber(caliber) + if sort not in _SORTABLE: + raise HTTPException(400, f"sort 不支持: {sort}（可选 {sorted(_SORTABLE)}）") + if order not in ("asc", "desc"): + raise HTTPException(400, "order 仅接受 asc/desc") + ok, reason = _available(db) + if not ok: + return {"available": False, "reason": reason, "entity_id": entity_id, + "caliber": cal, "period": period, "has_data": False, "items": [], "total": 0} + + periods = _periods(db, entity_id) + if not periods: + return {"available": True, "entity_id": entity_id, "caliber": cal, + "period": period, "has_data": False, "items": [], "total": 0} + p = period or periods[-1] + rows = _rows(db, entity_id, cal, p) + if not rows: + return {"available": True, "entity_id": entity_id, "caliber": cal, "period": p, + "has_data": False, "reason": "该期无数据", "items": [], "total": 0} + + items = [_enrich(r) for r in rows] + if overdue_only: + items = [i for i in items if i["is_overdue"]] + items.sort(key=lambda i: (i.get(sort) is None, i.get(sort)), reverse=(order == "desc")) + return { + "available": True, "entity_id": entity_id, "caliber": cal, "period": p, + "has_data": True, "total": len(items), "sort": sort, "order": order, + "overdue_only": overdue_only, "items": items, + } + + +# ============================================================ +# GET /customers/export +# ============================================================ +@router.get("/customers/export") +def export_customers( + period: Optional[str] = Query(None), + caliber: Optional[str] = Query(None), + overdue_only: bool = Query(False), + db: Session = Depends(get_db), + entity_id: int = Depends(get_entity_id), +): + """逐户明细 CSV（UTF-8 BOM，Excel 直开）；首段写「截止期/口径/导出时刻」。""" + from urllib.parse import quote + from starlette.responses import Response + + cal = _resolve_caliber(caliber) + ok, reason = _available(db) + if not ok: + raise HTTPException(503, reason or "数据表不可用") + periods = _periods(db, entity_id) + if not periods: + raise HTTPException(404, "表内该账套无数据") + p = period or periods[-1] + rows = _rows(db, entity_id, cal, p) + if not rows: + raise HTTPException(404, f"该期无数据: {p}") + + items = [_enrich(r) for r in rows] + if overdue_only: + items = [i for i in items if i["is_overdue"]] + items.sort(key=lambda i: (i["overdue_amt"] is None, -(i["overdue_amt"] or 0))) + + buf = io.StringIO() + w = csv.writer(buf) + w.writerow(["截止期", p, "口径", cal, + "导出时刻", datetime.now().strftime("%Y-%m-%d %H:%M:%S")]) + w.writerow(["单位编码", "单位名称", "期初（非账龄基数）", "期末应收合计", + "0-30", "31-60", "61-90", "90-120", "120-150", "150+", + "逾期额(>30天)", "90天以上", "最老逾期档(天起)", "逾期占该户%", "是否逾期"]) + for i in items: + w.writerow([ + i["unit_code"], i["unit_name"], _money_str(i["opening_amt"]), + _money_str(i["total_amt"]), + *[_money_str(i["buckets"][c]) for c in _BUCKET_COLS], + _money_str(i["overdue_amt"]), _money_str(i["overdue_90plus"]), + i["oldest_days"], + ("" if i["pct_of_unit"] is None else "%.2f" % (i["pct_of_unit"] * 100)), + "是" if i["is_overdue"] else "否", + ]) + + content = "\ufeff" + buf.getvalue() + filename = "应收账龄_%s_%s.csv" % (p, cal) + return Response( + content=content.encode("utf-8"), + media_type="text/csv; charset=utf-8", + headers={"Content-Disposition": "attachment; filename=\"ar_aging_%s.csv\"; filename*=UTF-8''%s" + % (p, quote(filename))}, + ) diff --git a/backend/app/main.py b/backend/app/main.py index e79135c3..05470e9a 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -7,7 +7,7 @@ from fastapi.responses import JSONResponse from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials from dotenv import load_dotenv from app.database import init_db, get_db -from app.api import auth, kpis, kpi_governance, templates, maps, dashboard, data, alerts, ai_analysis, ai_suggestions, alert_rules, users, thresholds, notifications, permissions, action_plans, alignment, org, objectives, versions, budget, cost, predict, reports, security, knowledge, bot_bridge, bot_bridge_v2, bot_suggestions, lead, tenant, customer_dashboard, deviation_push, budget_generate, knowledge_articles, kpi_causality, data_quality, bi_reports, entities, bsc_layers, okr, okr_templates, subjects, driver_budget, bot_kpis, ontology, bot_iron_law, analysis_results, expenses, cash, tax_compliance, verify, growth_quality, products, data_classification, value_sources, zero_based, derivation_rules, cash_classify, cashflow_structure, cash_vouchers_json, ext_input_ledgers, treasury, doc_center, map_completeness, map_summary, map_dimensions +from app.api import auth, kpis, kpi_governance, templates, maps, dashboard, data, alerts, ai_analysis, ai_suggestions, alert_rules, users, thresholds, notifications, permissions, action_plans, alignment, org, objectives, versions, budget, cost, predict, reports, security, knowledge, bot_bridge, bot_bridge_v2, bot_suggestions, lead, tenant, customer_dashboard, deviation_push, budget_generate, knowledge_articles, kpi_causality, data_quality, bi_reports, entities, bsc_layers, okr, okr_templates, subjects, driver_budget, bot_kpis, ontology, bot_iron_law, analysis_results, expenses, cash, tax_compliance, verify, growth_quality, products, data_classification, value_sources, zero_based, derivation_rules, cash_classify, cashflow_structure, cash_vouchers_json, ext_input_ledgers, treasury, doc_center, map_completeness, map_summary, map_dimensions, ar_aging from app.utils.cache import clear_all as clear_cache, delete as delete_cache from scripts.erp_sync import run_sync as run_erp_sync from app.auth_middleware import require_auth, require_auth_401 @@ -105,6 +105,7 @@ app.include_router(tax_compliance.router) app.include_router(verify.router) app.include_router(ext_input_ledgers.router) # 外部输入台账（M2 准入3 派生，2026-09-15） app.include_router(treasury.router) # 司库（资金）阶段一：账户池/头寸/内控闸门（2026-09-15） +app.include_router(ar_aging.router) # 应收账龄看板（T2 2026-10-20：逐户·六档·三口径禁合计） app.include_router(doc_center.router) # 文档中心：md 单一来源（运行期读盘，2026-09-24） @app.exception_handler(Exception) diff --git a/backend/app/services/ar_overdue_alert.py b/backend/app/services/ar_overdue_alert.py new file mode 100644 index 00000000..fd5ef0a2 --- /dev/null +++ b/backend/app/services/ar_overdue_alert.py @@ -0,0 +1,706 @@ +"""应收账龄逾期预警（`alert_rules.rule_type='ar_overdue'`）计算与落库 + +设计要点（任务 project-20261020-ar-overdue-alert-and-aging-board-fullstack）： + +* **阈值/责任人/排除名单全部来自 `alert_rules.params`（DB 行）**，本模块不出现任何 + 业务阈值数字；缺键即该判据关闭（显式失效，不回落隐式默认值）。 +* **只 INSERT**：命中写 `kpi_alerts(alert_type='overdue')`，同 + `(entity_id, kpi_id, alert_type, alert_message)` 已存在则跳过（`alert_message` + 含 period+对象 ⇒ 天然去重键）。**零 UPDATE / 零 DELETE**。 +* `assignee` 必填（`assignee_by_unit[unit_code]` → `assignee_default`），两者皆空 + ⇒ `AssigneeMissing`（调用方转 400，fail-closed）——禁落空 assignee 的行。 +* 口径：只读 `借方应收户口径`（`app.utils.ar_caliber`），负档位不抹平。 +""" +from __future__ import annotations + +import calendar +import logging +from decimal import Decimal, ROUND_HALF_UP +from typing import Any, Dict, List, Optional, Tuple + +from sqlalchemy import text +from sqlalchemy.orm import Session + +from app.models import KPIAlert +from app.utils import ar_caliber + +logger = logging.getLogger("ar_overdue_alert") + +RULE_TYPE = "ar_overdue" +ALERT_TYPE = "overdue" +MESSAGE_MAX = 500 + +_CENT = Decimal("0.01") + + +class AssigneeMissing(Exception): + """责任人未配置（fail-closed：拒绝运行，不落空 assignee 行）。""" + + +class ParamsInvalid(Exception): + """规则参数结构非法。""" + + +# ============================================================ +# 参数校验（只校结构，不设默认阈值） +# ============================================================ +def validate_params(params: Any) -> Optional[str]: + """返回错误文案；None = 合法。""" + if params is None: + return "params 不能为空（ar_overdue 规则参数全量走 DB，代码不设业务默认值）" + if not isinstance(params, dict): + return "params 必须是 JSON 对象" + try: + ar_caliber.validate_caliber(params.get("caliber") or ar_caliber.DEFAULT_CALIBER) + except ar_caliber.CaliberError as e: + return str(e) + for key in ("red", "yellow"): + if not isinstance(params.get(key), dict): + return "params.%s 必须是对象（阈值分组）" % key + for key in ("assignee_by_unit", "action_texts"): + if key in params and params[key] is not None and not isinstance(params[key], dict): + return "params.%s 必须是对象" % key + if "exclude_unit_codes" in params and params["exclude_unit_codes"] is not None: + v = params["exclude_unit_codes"] + if not isinstance(v, (list, tuple)): + return "params.exclude_unit_codes 必须是数组（排他名单）" + return None + + +def _params_caliber(params: Dict[str, Any]) -> str: + return params.get("caliber") or ar_caliber.DEFAULT_CALIBER + + +def _action_text(params: Dict[str, Any], level: str) -> str: + texts = params.get("action_texts") or {} + return str(texts.get(level) or texts.get("default") or "") + + +# ============================================================ +# assignee 解析（任务 §6.2 定值；全部只读 SELECT，禁止落空 assignee 行） +# 优先级：assignee_by_unit 手工点名 → 近 3 个月主业务员 → 部门负责人 +# 两路皆失败 ⇒ 该户不生成预警，进 `skipped[]`（含原因） +# ============================================================ +ASSIGNEE_MODES = ("salesman_then_dept", "manual_then_salesman_then_dept", "dept_only") + + +def _source_cfg(params: Dict[str, Any]) -> Dict[str, Any]: + src = params.get("assignee_source") + return src if isinstance(src, dict) else {} + + +def _assignee_by_unit(params: Dict[str, Any], unit_code: Optional[str]) -> Optional[str]: + by_unit = params.get("assignee_by_unit") or {} + if unit_code is None: + return None + v = by_unit.get(str(unit_code)) + return str(v).strip() if v else None + + +def ensure_assignee_configured(params: Dict[str, Any]) -> None: + """责任人来源整体缺失 ⇒ 400 fail-closed（§4.8 负测）。 + + §6.2 允许逐户派生（主业务员 → 部门负责人），因此「配了 assignee_source.mode」 + 也算已配置；派生失败的**单户**进 skipped[]，不落空 assignee 行。 + 完全没有任何来源（无 default / 无 by_unit / 无 mode）才拒绝整体运行。 + """ + if str(params.get("assignee_default") or "").strip(): + return + if params.get("assignee_by_unit"): + return + if str(_source_cfg(params).get("mode") or "") in ASSIGNEE_MODES: + return + raise AssigneeMissing( + "ar_overdue 规则未配置任何责任人来源：params.assignee_default / assignee_by_unit / " + "assignee_source.mode 均为空，按 fail-closed 拒绝运行（禁止落空 assignee 的预警行）" + ) + + +def _has_table(db: Session, name: str) -> bool: + """表是否可读（方言无关，不依赖 information_schema）。""" + try: + db.execute(text("SELECT 1 FROM %s LIMIT 1" % name)) + return True + except Exception: # noqa: BLE001 — 缺表/无权限一律降级 + return False + + +def _employee_name(db: Session, emp_id: Any) -> Optional[str]: + """employee_dict.u_Name（只读）。缺表/无该 id ⇒ None（调用方继续降级）。""" + try: + emp = int(emp_id) + except (TypeError, ValueError): + return None + if emp <= 0 or not _has_table(db, ar_caliber.EMPLOYEE_DICT_TABLE): + return None + try: + row = db.execute( + text("SELECT u_Name FROM %s WHERE id = :i" % ar_caliber.EMPLOYEE_DICT_TABLE), + {"i": emp}, + ).first() + return str(row[0]).strip() if row and row[0] else None + except Exception: # noqa: BLE001 + return None + + +def _operator_from_abst(abst: Any) -> Optional[str]: + """单据摘要尾段制单人姓名（`…：姓名` / `…元;姓名`）——employee_dict 缺表时的同库只读替代源。 + + 摘要里常混入金额/银行/备注（如 `…【建设银行】270000.00元;兀晨博`），因此逐段 + 取尾段后做**形状校验**：纯中文姓名、长度 ≤ 20、不含数字/括号/金额符号，否则判失败。 + """ + if not abst: + return None + s = str(abst).strip() + for sep in ("：", ";"): + if sep in s: + s = s.rsplit(sep, 1)[-1] + s = s.strip(" \u3000;；") + if not s or len(s) > 20: + return None + if any(ch.isdigit() for ch in s): + return None + if any(ch in s for ch in "【】[]()（）元,，.。:："): + return None + return s + + +def _period_window(period: str, months: int) -> Tuple[str, str]: + """账龄期 → 业务员回看窗口 [期末前 (months-1) 个月 1 日, 期末]。""" + y, m = int(period[:4]), int(period[5:7]) + ly, lm = y, m - (months - 1) + while lm <= 0: + lm += 12 + ly -= 1 + end_day = calendar.monthrange(y, m)[1] + return "%04d-%02d-01" % (ly, lm), "%04d-%02d-%02d" % (y, m, end_day) + + +def _resolve_unit_ref(db: Session, entity_id: int, unit_code: Any, unit_name: Any, + start: str, end: str) -> Tuple[Optional[int], str, str]: + """unit_code → zxt_masterbill.unit_id。返回 (unit_id, 来源, 失败原因)。""" + code = str(unit_code or "").strip() + if _has_table(db, ar_caliber.UNITS_DICT_TABLE): + for code_col in ("unit_code", "u_code", "code"): + try: + row = db.execute( + text("SELECT unit_id FROM %s WHERE %s = :c" + % (ar_caliber.UNITS_DICT_TABLE, code_col)), + {"c": code}, + ).first() + if row and row[0] is not None: + return int(row[0]), "units_dict.%s" % code_col, "" + except Exception: # noqa: BLE001 — 列名不匹配则试下一列 + continue + if code.isdigit(): + return int(code), "numeric_code", "" + if unit_name: + try: + row = db.execute( + text("SELECT unit_id, COUNT(*) AS n FROM %s WHERE entity_id = :e " + "AND abst LIKE :kw AND bill_date >= :s AND bill_date <= :d " + "GROUP BY unit_id ORDER BY n DESC LIMIT 1" % ar_caliber.ZXT_MASTERBILL_TABLE), + {"e": entity_id, "kw": "%%%s%%" % unit_name, "s": start, "d": end}, + ).first() + if row and row[0] is not None: + return int(row[0]), "abst_name_match", "" + except Exception: # noqa: BLE001 + pass + return None, "", "unit_link_unresolved(code=%s)" % (code or "-") + + +def _salesman_name(db: Session, entity_id: int, unit_id: int, start: str, end: str, + allow_abst_fallback: bool) -> Tuple[Optional[str], Optional[int], str]: + """该户近窗主业务员（单据数最多者；同数取最近单据）。""" + try: + row = db.execute( + text("SELECT emp_id, COUNT(*) AS n FROM %s WHERE entity_id = :e AND unit_id = :u " + "AND bill_date >= :s AND bill_date <= :d AND emp_id IS NOT NULL AND emp_id > 0 " + "GROUP BY emp_id ORDER BY n DESC, MAX(bill_date) DESC LIMIT 1" + % ar_caliber.ZXT_MASTERBILL_TABLE), + {"e": entity_id, "u": unit_id, "s": start, "d": end}, + ).first() + except Exception as exc: # noqa: BLE001 + return None, None, "salesman_query_failed:%s" % type(exc).__name__ + if not row or row[0] is None: + return None, None, "no_recent_bill(unit_id=%s)" % unit_id + emp_id = int(row[0]) + name = _employee_name(db, emp_id) + if name: + return name, emp_id, "" + if not allow_abst_fallback: + return None, emp_id, "salesman_name_unresolved(emp_id=%s, employee_dict 缺表)" % emp_id + try: + row2 = db.execute( + text("SELECT abst FROM %s WHERE entity_id = :e AND unit_id = :u AND emp_id = :m " + "AND bill_date >= :s AND bill_date <= :d ORDER BY bill_date DESC, id DESC LIMIT 1" + % ar_caliber.ZXT_MASTERBILL_TABLE), + {"e": entity_id, "u": unit_id, "m": emp_id, "s": start, "d": end}, + ).first() + name = _operator_from_abst(row2[0] if row2 else None) + except Exception: # noqa: BLE001 + name = None + if name: + return name, emp_id, "" + return None, emp_id, "salesman_name_unresolved(emp_id=%s, employee_dict 缺表且摘要无姓名)" % emp_id + + +def _dept_assignee(db: Session, unit_code: Any, + params: Dict[str, Any]) -> Tuple[Optional[str], str]: + """部门负责人兜底（§6.2：渠道 118 / 商务 114 / 直属酒窖 117）。 + + 部门归属无法自动判定时**不猜**：要求 params.assignee_source.dept_by_unit / + dept_by_prefix / dept_default 之一给出部门键，否则返回原因。 + """ + src = _source_cfg(params) + fb = src.get("dept_fallback") or {} + code = str(unit_code or "") + key = (src.get("dept_by_unit") or {}).get(code) or src.get("dept_default") + if key is None: + for prefix, k in (src.get("dept_by_prefix") or {}).items(): + if code.startswith(str(prefix)): + key = k + break + if key is None: + return None, "dept_unclassified" + if key not in fb: + return None, "dept_key_unmapped:%s" % key + name = _employee_name(db, fb[key]) + if name: + return name, "dept:%s(id=%s)" % (key, fb[key]) + return None, "dept_head_name_unresolved:%s(id=%s, employee_dict 缺表)" % (key, fb[key]) + + +def resolve_assignee(db: Session, entity_id: int, unit_code: Any, unit_name: Any, + params: Dict[str, Any], period: str) -> Dict[str, Any]: + """单户 assignee 解析结果：{name, source, emp_id, reason}。name=None ⇒ 不生成该户。""" + manual = _assignee_by_unit(params, str(unit_code) if unit_code is not None else None) + if manual: + return {"name": manual, "source": "manual:assignee_by_unit", "emp_id": None, "reason": ""} + + src = _source_cfg(params) + mode = str(src.get("mode") or "") + reasons: List[str] = [] + if mode in ASSIGNEE_MODES: + months = int(src.get("lookback_months") or 3) + start, end = _period_window(period, months) + if mode != "dept_only": + unit_id, link, link_reason = _resolve_unit_ref( + db, entity_id, unit_code, unit_name, start, end) + if unit_id is not None: + name, emp_id, why = _salesman_name( + db, entity_id, unit_id, start, end, + bool(src.get("abst_operator_fallback", True))) + if name: + return {"name": name, "source": "salesman:%s(emp_id=%s, %s)" + % (link, emp_id, "employee_dict" if _has_table(db, ar_caliber.EMPLOYEE_DICT_TABLE) + else "abst_operator"), "emp_id": emp_id, "reason": ""} + reasons.append(why) + else: + reasons.append(link_reason) + name, why = _dept_assignee(db, unit_code, params) + if name: + return {"name": name, "source": why, "emp_id": None, "reason": ""} + reasons.append(why) + return {"name": None, "source": "", "emp_id": None, "reason": "; ".join(r for r in reasons if r)} + + default = str(params.get("assignee_default") or "").strip() + if default: + return {"name": default, "source": "assignee_default", "emp_id": None, "reason": ""} + return {"name": None, "source": "", "emp_id": None, "reason": "no_assignee_source"} + + +def aggregate_assignee(params: Dict[str, Any]) -> Tuple[Optional[str], str]: + """聚合预警的责任人（整体口径）：assignee_default → assignee_source.aggregate_assignee。""" + d = str(params.get("assignee_default") or "").strip() + if d: + return d, "assignee_default" + a = str(_source_cfg(params).get("aggregate_assignee") or "").strip() + if a: + return a, "assignee_source.aggregate_assignee" + return None, "aggregate_assignee_unresolved" + + + +# ============================================================ +# 取数与计算 +# ============================================================ +def _dec(v: Any) -> Decimal: + if v is None: + return Decimal("0") + return Decimal(str(v)) + + +def _money(v: Decimal) -> float: + return float(v.quantize(_CENT, rounding=ROUND_HALF_UP)) + + +def _fmt_money(v: Decimal) -> str: + return "{:,.2f}".format(v.quantize(_CENT, rounding=ROUND_HALF_UP)) + + +def _fmt_pct(ratio: Optional[float]) -> str: + if ratio is None: + return "-" + return "{:.2f}".format(ratio * 100.0) + + +_SELECT_COLS = ( + "unit_code, unit_name, opening_amt, total_amt, " + + ", ".join(ar_caliber.BUCKET_COLUMNS) +) + + +def _periods(db: Session, entity_id: int, caliber: str) -> List[str]: + rows = db.execute( + text( + "SELECT DISTINCT period FROM %s WHERE entity_id = :e AND caliber = :c ORDER BY period" + % ar_caliber.AR_AGING_TABLE + ), + {"e": entity_id, "c": caliber}, + ).all() + return [r[0] for r in rows] + + +def _rows_for_period(db: Session, entity_id: int, caliber: str, period: str) -> List[Dict[str, Any]]: + sql = ( + "SELECT %s FROM %s WHERE entity_id = :e AND caliber = :c AND period = :p " + "ORDER BY total_amt DESC, unit_code ASC" % (_SELECT_COLS, ar_caliber.AR_AGING_TABLE) + ) + return [dict(r) for r in db.execute(text(sql), {"e": entity_id, "c": caliber, "p": period}).mappings().all()] + + +def _oldest_overdue_days(row: Dict[str, Any]) -> int: + """最老非零逾期档的下界天数（31/61/91/121/151）；无逾期 = 0。""" + days = 0 + for col, lower in ar_caliber.OVERDUE_BUCKETS: + if _dec(row.get(col)) != 0: + days = max(days, lower) + return days + + +def _level_share(share: Optional[float], params: Dict[str, Any]) -> Optional[str]: + red = params.get("red") or {} + yellow = params.get("yellow") or {} + gt = red.get("share_90plus_gt") + if gt is not None and share is not None and share > float(gt): + return "red" + ymin, ymax = yellow.get("share_90plus_min"), yellow.get("share_90plus_max") + if share is not None and ymin is not None and ymax is not None: + if float(ymin) <= share <= float(ymax): + return "yellow" + return None + + +def _level_mom(mom_pct: Optional[float], params: Dict[str, Any]) -> Optional[str]: + red = params.get("red") or {} + gte = red.get("count_mom_pct_gte") + if gte is not None and mom_pct is not None and mom_pct >= float(gte): + return "red" + return None + + +def _level_customer(overdue_amt: Decimal, oldest_days: int, params: Dict[str, Any]) -> Optional[str]: + red = params.get("red") or {} + yellow = params.get("yellow") or {} + red_amt = red.get("single_overdue_amt_gt") + red_min_days = red.get("single_overdue_min_days") + if red_amt is not None and overdue_amt > _dec(red_amt): + if red_min_days is None or oldest_days >= int(red_min_days): + return "red" + ymin = yellow.get("single_overdue_amt_min") + ymax = yellow.get("single_overdue_amt_max") + if ymin is not None and ymax is not None: + if _dec(ymin) <= overdue_amt <= _dec(ymax): + return "yellow" + return None + + +def _clip(msg: str) -> str: + return msg if len(msg) <= MESSAGE_MAX else msg[: MESSAGE_MAX - 1] + "…" + + +def compute( + db: Session, + entity_id: int, + period: Optional[str] = None, + params: Optional[Dict[str, Any]] = None, +) -> Dict[str, Any]: + """计算逾期预警命中集合（纯读，不写库）。 + + 返回: + {period, caliber, metrics, aggregates[], per_customer[], skipped[], skipped_reason, notes[]} + `skipped_reason` 非空表示本次不生成任何预警（缺期/无数据/分母非正…）。 + `skipped[]` 记录 assignee 解析失败而未生成的户/聚合项（§6.2 禁落空 assignee）。 + """ + params = params or {} + err = validate_params(params) + if err: + raise ParamsInvalid(err) + caliber = _params_caliber(params) + notes: List[str] = [] + out: Dict[str, Any] = { + "period": period, + "caliber": caliber, + "metrics": {}, + "aggregates": [], + "per_customer": [], + "skipped": [], + "skipped_reason": None, + "notes": notes, + } + + # ── 期间解析：默认表内 MAX(period)；缺期直接 skipped（不补 0）── + periods = _periods(db, entity_id, caliber) + if not periods: + out["skipped_reason"] = "no_data_in_table:%s/%s" % (entity_id, caliber) + return out + if not period: + period = periods[-1] + out["period"] = period + notes.append("period 未指定 → 取表内 MAX(period)=%s" % period) + if period not in periods: + out["skipped_reason"] = "no_data_for_period:%s" % period + return out + + rows = _rows_for_period(db, entity_id, caliber, period) + if not rows: + out["skipped_reason"] = "no_data_for_period:%s" % period + return out + + exclude = set(params.get("exclude_unit_codes") or []) + if exclude: + before = len(rows) + rows = [r for r in rows if str(r.get("unit_code")) not in exclude] + notes.append("排除名单命中 %d 行（exclude_unit_codes）" % (before - len(rows))) + if not rows: + out["skipped_reason"] = "all_rows_excluded" + return out + + total = sum((_dec(r.get("total_amt")) for r in rows), Decimal("0")) + overdue_90plus = sum((_dec(r.get(c)) for r in rows for c in ar_caliber.NINETY_PLUS_COLUMNS), Decimal("0")) + share = (float(overdue_90plus / total) if total > 0 else None) + if total <= 0: + notes.append("分母 SUM(total_amt)=%.2f ≤ 0 → 跳过聚合占比判据（禁除零）" % float(total)) + + # 逐户：逾期额 = total_amt - b_0_30（等价 31-60..150+ 之和，负档位不抹平） + customers: List[Dict[str, Any]] = [] + n_overdue = 0 + for r in rows: + o_i = _dec(r.get("total_amt")) - _dec(r.get("b_0_30")) + oldest = _oldest_overdue_days(r) + if o_i > 0: + n_overdue += 1 + level = _level_customer(o_i, oldest, params) if o_i > 0 else None + customers.append({"row": r, "overdue_amt": o_i, "oldest_days": oldest, "level": level}) + + # 环比：上一个有数据期（同口径同算法）；上期无数据 ⇒ 不算环比（文案写明，不得当 0） + idx = periods.index(period) + prev_period = periods[idx - 1] if idx > 0 else None + n_prev = None + prev_rows: List[Dict[str, Any]] = [] + if prev_period: + prev_rows = _rows_for_period(db, entity_id, caliber, prev_period) + if exclude: + prev_rows = [r for r in prev_rows if str(r.get("unit_code")) not in exclude] + n_prev = sum( + 1 for r in prev_rows + if (_dec(r.get("total_amt")) - _dec(r.get("b_0_30"))) > 0 + ) + mom_pct = None + if n_prev and n_prev > 0: + mom_pct = (n_overdue - n_prev) / n_prev + else: + notes.append("上期无数据（%s）→ 不计算环比" % (prev_period or "无更早期间")) + + metrics = { + "period": period, + "caliber": caliber, + "total_amt": _money(total), + "overdue_90plus": _money(overdue_90plus), + "share_90plus": (round(share, 6) if share is not None else None), + "overdue_amt": _money(sum((c["overdue_amt"] for c in customers), Decimal("0"))), + "customers": len(rows), + "overdue_customers": n_overdue, + "prev_period": prev_period, + "prev_overdue_customers": n_prev, + "count_mom_pct": (round(mom_pct, 6) if mom_pct is not None else None), + } + out["metrics"] = metrics + + # ── 聚合判据（§6.1）：90 天占比（red 键 null ⇒ 停用）/ 逾期户数环比 / 150+ 结构恶化 ── + agg_assignee, agg_src = aggregate_assignee(params) + + def _push_agg(key: str, level: str, metrics_dict: Dict[str, Any], msg: str, action: str) -> None: + blocked = None if agg_assignee else agg_src + out["aggregates"].append({ + "key": key, + "level": level, + "metrics": metrics_dict, + "alert_message": msg, + "suggestion": action, + "assignee": agg_assignee or "", + "assignee_source": agg_src, + "blocked": blocked, + }) + if blocked: + out["skipped"].append({"scope": "aggregate", "key": key, "level": level, + "reason": blocked}) + + red_cfg = params.get("red") or {} + yellow_cfg = params.get("yellow") or {} + red_gt = red_cfg.get("share_90plus_gt") + if share is not None and red_gt is None: + notes.append("red.share_90plus_gt 为 null ⇒ 90 天占比红线停用（不回落 0.25）") + + # 聚合判据 1：90 天以上占比（红线停用时只剩黄线 15%–25%） + if share is not None: + level = _level_share(share, params) + if level: + action = _action_text(params, level) + thr = ("红线 >%g%%" % (float(red_gt) * 100)) if red_gt is not None else "红线停用（DB 置 null）" + msg = _clip( + "【应收账款逾期】%s %s 90 天以上 ¥%s，占应收总额 %s%%（%s）｜建议：%s" + % (period, caliber, _fmt_money(overdue_90plus), _fmt_pct(share), thr, + action or "见规则配置") + ) + _push_agg("share_90plus", level, + {"overdue_90plus": _money(overdue_90plus), "total_amt": _money(total), + "share_90plus": round(share, 6)}, + msg, action) + + # 聚合判据 2：逾期户数环比 + level = _level_mom(mom_pct, params) + if level: + action = _action_text(params, level) + gte = red_cfg.get("count_mom_pct_gte") + msg = _clip( + "【应收账款逾期】%s 逾期户数 %d 户，环比 %s%%（红线 +%s%%）｜建议：%s" + % (period, n_overdue, + ("%+.2f" % (mom_pct * 100)) if mom_pct is not None else "-", + ("%g" % (float(gte) * 100)) if gte is not None else "-", + action or "见规则配置") + ) + _push_agg("count_mom", level, + {"overdue_customers": n_overdue, "prev_overdue_customers": n_prev, + "count_mom_pct": (round(mom_pct, 6) if mom_pct is not None else None)}, + msg, action) + + # 聚合判据 3：150+ 档占比环比上升 ⇒ 账龄结构恶化（§6.1 新增红线 2） + share_150 = None + prev_share_150 = None + if total > 0: + share_150 = float(sum((_dec(r.get("b_150p")) for r in rows), Decimal("0")) / total) + if prev_rows: + prev_total = sum((_dec(r.get("total_amt")) for r in prev_rows), Decimal("0")) + if prev_total > 0: + prev_share_150 = float( + sum((_dec(r.get("b_150p")) for r in prev_rows), Decimal("0")) / prev_total) + structure_on = bool(red_cfg.get("structure_worsen")) + if structure_on: + if share_150 is not None and prev_share_150 is not None and share_150 > prev_share_150: + action = _action_text(params, "red") + msg = _clip( + "【应收账款逾期】%s 150+ 档占比 %s%%（上期 %s%%）环比上升 %+.2fpp，账龄结构恶化" + "｜建议：%s" % (period, _fmt_pct(share_150), _fmt_pct(prev_share_150), + (share_150 - prev_share_150) * 100.0, action or "见规则配置") + ) + _push_agg("structure_150p", "red", + {"share_150p": round(share_150, 6), + "prev_period": prev_period, + "prev_share_150p": (round(prev_share_150, 6) + if prev_share_150 is not None else None), + "delta_pp": round((share_150 - prev_share_150) * 100.0, 4)}, + msg, action) + else: + notes.append( + "150+ 占比：本期 %s%% / 上期 %s%% ⇒ 结构恶化判据未触发" + % (_fmt_pct(share_150), _fmt_pct(prev_share_150))) + metrics["share_150p"] = (round(share_150, 6) if share_150 is not None else None) + metrics["prev_share_150p"] = (round(prev_share_150, 6) if prev_share_150 is not None else None) + + # ── 逐户：assignee 派生失败 ⇒ 不生成（进 skipped[]），禁落空 assignee ── + for c in customers: + if not c["level"]: + continue + r = c["row"] + who = resolve_assignee(db, entity_id, r.get("unit_code"), r.get("unit_name"), params, period) + if not who["name"]: + out["skipped"].append({ + "scope": "customer", + "unit_code": r.get("unit_code"), + "unit_name": r.get("unit_name"), + "level": c["level"], + "overdue_amt": _money(c["overdue_amt"]), + "oldest_days": c["oldest_days"], + "reason": who["reason"], + }) + continue + action = _action_text(params, c["level"]) + unit_total = _dec(r.get("total_amt")) + pct = (float(c["overdue_amt"] / unit_total) if unit_total > 0 else None) + msg = _clip( + "【应收账款逾期】%s（%s）%s 期末逾期 %d 天以上 ¥%s，占该户应收 %s%%｜建议：%s｜责任人：%s" + % ( + r.get("unit_name"), r.get("unit_code"), period, c["oldest_days"], + _fmt_money(c["overdue_amt"]), _fmt_pct(pct), action or "见规则配置", who["name"], + ) + ) + out["per_customer"].append({ + "unit_code": r.get("unit_code"), + "unit_name": r.get("unit_name"), + "overdue_amt": _money(c["overdue_amt"]), + "total_amt": _money(unit_total), + "oldest_days": c["oldest_days"], + "pct_of_unit": (round(pct, 6) if pct is not None else None), + "level": c["level"], + "alert_message": msg, + "suggestion": action, + "assignee": who["name"], + "assignee_source": who["source"], + "emp_id": who["emp_id"], + }) + return out + + +# ============================================================ +# 落库（只 INSERT；同 message 幂等跳过） +# ============================================================ +def hits(result: Dict[str, Any]) -> List[Dict[str, Any]]: + """可落库命中项：只收 assignee 非空的（§6.2 禁落空 assignee 行）。""" + all_hits = list(result.get("aggregates") or []) + list(result.get("per_customer") or []) + return [h for h in all_hits if str(h.get("assignee") or "").strip()] + + +def write_alerts(db: Session, entity_id: int, kpi_id: int, result: Dict[str, Any]) -> Dict[str, int]: + """把命中集合写成 `kpi_alerts(alert_type='overdue')`；历史行零触碰。""" + inserted = 0 + skipped = 0 + for h in hits(result): + msg = h.get("alert_message") or "" + if not msg: + continue + exists = db.query(KPIAlert.id).filter( + KPIAlert.entity_id == entity_id, + KPIAlert.kpi_id == kpi_id, + KPIAlert.alert_type == ALERT_TYPE, + KPIAlert.alert_message == msg, + ).first() + if exists: + skipped += 1 + continue + db.add(KPIAlert( + entity_id=entity_id, + kpi_id=kpi_id, + kpi_value_id=None, + alert_level=h.get("level") or "yellow", + alert_message=msg, + alert_type=ALERT_TYPE, + suggestion=h.get("suggestion") or None, + status="pending", + assignee=h.get("assignee") or None, + )) + inserted += 1 + if inserted: + db.commit() + return {"inserted": inserted, "skipped_duplicate": skipped} diff --git a/backend/app/utils/ar_caliber.py b/backend/app/utils/ar_caliber.py new file mode 100644 index 00000000..79a0ca7e --- /dev/null +++ b/backend/app/utils/ar_caliber.py @@ -0,0 +1,71 @@ +"""应收账龄口径单一事实源（`hanke_ar_aging`） + +财务已定死的口径纪律（ERRATA-02 §五 / DQ 台账 DQ-AR-03）： + +1. **三口径分列，禁跨口径合计**：`全户口径` / `借方应收户口径` / `贷方户单列(净额)`。 + 字面值逐字匹配——`贷方户单列(净额)` 写成 `贷方户单列` 会漏 42 行。 +2. **对外「超期金额」只取 `借方应收户口径`**；贷方户单列不进分母，任何 `caliber=all` 请求 400。 +3. **六档互斥**：0-30 / 31-60 / 61-90 / 90-120 / 120-150 / 150+，逐行六档和 ≡ `total_amt`。 +4. `opening_amt`（期初）**只展示**，禁作账龄基数、禁进任何分子分母。 +5. 负值保留：禁 `ABS()` / `COALESCE(...,0)` 抹平（DQ-AR-04 杨创 `b_0_30=-20`）。 +""" +from typing import Optional, Tuple + +# 表名（DDL：dama/_reports/prodwrite-cma-hanke-ar-aging-20261020/mirror/03_ddl.sql） +AR_AGING_TABLE = "hanke_ar_aging" + +# assignee 派生依赖的只读维度/事实表（§6.2）。三者均为**只读 SELECT**，不写不建。 +ZXT_MASTERBILL_TABLE = "zxt_masterbill" # 智享通单据流（unit_id / emp_id / bill_date / abst） +EMPLOYEE_DICT_TABLE = "employee_dict" # 员工字典（id → u_Name）；当前库内**未落库** +UNITS_DICT_TABLE = "units_dict" # 往来单位字典（unit_code → unit_id）；当前库内**未落库** + +# 三口径字面值（逐字匹配，勿改写） +CALIBER_FULL = "全户口径" +CALIBER_DEBIT = "借方应收户口径" +CALIBER_CREDIT_SEPARATE = "贷方户单列(净额)" +CALIBERS: Tuple[str, ...] = (CALIBER_FULL, CALIBER_DEBIT, CALIBER_CREDIT_SEPARATE) + +# 对外超期口径/看板默认口径 +DEFAULT_CALIBER = CALIBER_DEBIT + +# 跨口径合计请求字面值 —— 一律 400 +CALIBER_ALL = "all" + +# 六档列（顺序即展示顺序）：(列名, 档位标签, 档位下界天数) +BUCKETS: Tuple[Tuple[str, str, int], ...] = ( + ("b_0_30", "0-30", 0), + ("b_31_60", "31-60", 31), + ("b_61_90", "61-90", 61), + ("b_90_120", "90-120", 91), + ("b_120_150", "120-150", 121), + ("b_150p", "150+", 151), +) + +BUCKET_COLUMNS = tuple(b[0] for b in BUCKETS) +# 逾期档位（>30 天）列；最老档下界天数即「逾期 N 天以上」的 N +OVERDUE_BUCKETS: Tuple[Tuple[str, int], ...] = tuple((b[0], b[2]) for b in BUCKETS if b[2] > 0) +# 90 天以上（含 90-120 档） +NINETY_PLUS_COLUMNS = ("b_90_120", "b_120_150", "b_150p") + + +class CaliberError(ValueError): + """口径入参非法（未知值 / 跨口径合计）。""" + + +def validate_caliber(caliber: Optional[str], *, allow_none: bool = False) -> str: + """校验口径字面值；`all` 与未知值一律拒绝（禁跨口径合计）。""" + if caliber is None or caliber == "": + if allow_none: + return DEFAULT_CALIBER + raise CaliberError("caliber 不能为空") + if caliber == CALIBER_ALL: + raise CaliberError("禁止跨口径合计（caliber=all 不提供）：三口径必须分列查询") + if caliber not in CALIBERS: + raise CaliberError( + "未知口径 %r；仅接受字面值：%s" % (caliber, " / ".join(CALIBERS)) + ) + return caliber + + +def is_credit_separate(caliber: str) -> bool: + return caliber == CALIBER_CREDIT_SEPARATE diff --git a/backend/scripts/seed_ar_overdue_rule.py b/backend/scripts/seed_ar_overdue_rule.py new file mode 100644 index 00000000..f2176f78 --- /dev/null +++ b/backend/scripts/seed_ar_overdue_rule.py @@ -0,0 +1,157 @@ +#!/usr/bin/env python3 +"""播种应收账龄逾期预警规则（`alert_rules.rule_type='ar_overdue'`）— 阶段B 专用 + +用法： + # 只打印将写入的行（默认 dry-run，阶段A 唯一允许的形式） + python3 backend/scripts/seed_ar_overdue_rule.py --entity-id 1 + + # 真正写库（阶段B，须另铸门/执行件授权） + python3 backend/scripts/seed_ar_overdue_rule.py --entity-id 1 --apply + + # 把参数 JSON 打到 stdout/文件，供阶段A `POST /run-ar-overdue` 的 dry-run body 复演 + python3 backend/scripts/seed_ar_overdue_rule.py --entity-id 1 --emit-params /tmp/ar_params.json + +阈值与责任人**只存在于本脚本**（业务数字唯一定义处）；运行期一律读 DB 行， +代码侧不含任何阈值常量（任务硬约束 §3.4）。 +""" +from __future__ import annotations + +import argparse +import json +import os +import sys + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from dotenv import load_dotenv # noqa: E402 + +load_dotenv(os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), ".env")) + +from app.database import get_session_local # noqa: E402 +from app.models import KPIDefinition # noqa: E402 +from app.utils import ar_caliber # noqa: E402 + +RULE_TYPE = "ar_overdue" +DEFAULT_KPI_CODE = "F_AR_002" # 逾期应收账款率（财务口径）；库内暂缺时须由财务定绑定 + +# ── 业务数字唯一定义处（阶段B 放行后如需调阈值：改这里或直接改 DB 行）── +DEFAULT_PARAMS = { + "kpi_code": DEFAULT_KPI_CODE, + "caliber": ar_caliber.CALIBER_DEBIT, + "red": { + # 2026-10-20 财务裁决：90 天占比红线停用（10 期 10/10 恒红、无区分度）→ 键位保留、置 null + "share_90plus_gt": None, + "single_overdue_amt_gt": 50000, + "single_overdue_min_days": 30, + "count_mom_pct_gte": 0.20, + # 150+ 档占比环比上升 = 结构恶化（新增红线 2） + "structure_worsen": True, + }, + "yellow": { + "share_90plus_min": 0.15, + "share_90plus_max": 0.25, + "single_overdue_amt_min": 10000, + "single_overdue_amt_max": 50000, + }, + # 责任人：逐户派生「近 3 个月主业务员 → 部门负责人」，两路皆失败则该户不生成（skipped[]） + "assignee_source": { + "mode": "salesman_then_dept", + "lookback_months": 3, + "dept_fallback": {"渠道": 118, "商务": 114, "直属酒窖": 117}, + }, + "assignee_by_unit": {}, + "exclude_unit_codes": [], + "action_texts": {"red": "催收+停供评估", "yellow": "对账+催收"}, +} + + +def resolve_kpi(db, entity_id: int, kpi_code: str, kpi_id: int | None): + if kpi_id: + return db.query(KPIDefinition).filter( + KPIDefinition.id == kpi_id, KPIDefinition.entity_id == entity_id).first() + return db.query(KPIDefinition).filter( + KPIDefinition.kpi_code == kpi_code, KPIDefinition.entity_id == entity_id).first() + + +def main() -> int: + ap = argparse.ArgumentParser(description="播种 ar_overdue 预警规则") + ap.add_argument("--entity-id", type=int, required=True) + ap.add_argument("--kpi-code", default=DEFAULT_KPI_CODE) + ap.add_argument("--kpi-id", type=int, default=None, help="显式绑定 KPI（优先于 kpi-code）") + ap.add_argument("--assignee", default=None, help="覆盖 assignee_default（财务指定责任人）") + ap.add_argument("--apply", action="store_true", help="真正写库（阶段B 才允许）") + ap.add_argument("--emit-params", default=None, help="把最终 params JSON 写入指定文件") + args = ap.parse_args() + + params = json.loads(json.dumps(DEFAULT_PARAMS)) # deep copy + if args.assignee is not None: + params["assignee_default"] = args.assignee + if args.emit_params: + with open(args.emit_params, "w", encoding="utf-8") as f: + json.dump({"params": params}, f, ensure_ascii=False, indent=2) + print("params 已写出: %s" % args.emit_params) + + db = get_session_local()() + try: + kpi = resolve_kpi(db, args.entity_id, args.kpi_code, args.kpi_id) + if not kpi: + print("[BLOCK] entity=%s 内找不到 KPI code=%r（kpi_id=%s）——无法绑定，未写入任何行" + % (args.entity_id, args.kpi_code, args.kpi_id), file=sys.stderr) + print("[INFO] 该账套现存 AR 相关 KPI：", file=sys.stderr) + rows = db.query(KPIDefinition).filter( + KPIDefinition.entity_id == args.entity_id, + KPIDefinition.kpi_code.like("%AR%"), + ).order_by(KPIDefinition.id).all() + for r in rows: + print(" id=%s code=%s name=%s" % (r.id, r.kpi_code, r.kpi_name), file=sys.stderr) + print("[INFO] 处理：--kpi-id 显式指定，或财务先在 kpi_definitions 建 F_AR_002。", + file=sys.stderr) + return 2 + + from app.api.alert_rules import AlertRule + exist = db.query(AlertRule).filter( + AlertRule.entity_id == args.entity_id, + AlertRule.rule_type == RULE_TYPE, + ).all() + print("== 将写入的行（%s）==" % ("APPLY" if args.apply else "DRY-RUN")) + print(json.dumps({ + "entity_id": args.entity_id, + "kpi_id": kpi.id, + "kpi_code": kpi.kpi_code, + "kpi_name": kpi.kpi_name, + "rule_type": RULE_TYPE, + "trigger_on": "actual", + "enabled": 1, + "params": params, + }, ensure_ascii=False, indent=2)) + src = params.get("assignee_source") or {} + if not params.get("assignee_default") and not src.get("mode"): + print("[WARN] 责任人均未配置 → 运行期 fail-closed 400（阶段B 前须由财务给值）") + + if exist: + print("[SKIP] entity=%s 已有 %d 条 ar_overdue 规则（id=%s），不重复播种" + % (args.entity_id, len(exist), [r.id for r in exist])) + return 0 + if not args.apply: + print("[DRY-RUN] 未写库。阶段B 授权后加 --apply。") + return 0 + + rule = AlertRule( + entity_id=args.entity_id, + kpi_id=kpi.id, + rule_type=RULE_TYPE, + trigger_on="actual", + enabled=1, + params=params, + ) + db.add(rule) + db.commit() + db.refresh(rule) + print("[APPLIED] 已写入 alert_rules.id=%s" % rule.id) + return 0 + finally: + db.close() + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/backend/tests/test_ar_aging_overdue_20261020.py b/backend/tests/test_ar_aging_overdue_20261020.py new file mode 100644 index 00000000..0fa5cf9a --- /dev/null +++ b/backend/tests/test_ar_aging_overdue_20261020.py @@ -0,0 +1,480 @@ +# -*- coding: utf-8 -*- +"""应收账龄看板 + 逾期预警（T1/T2 · 2026-10-20 R2）单测 + +任务：`project-20261020-ar-overdue-alert-and-aging-board-fullstack`（阶段A） + +口径与边界（全部为硬断言）： + · 六档互斥：逐行 `b_0_30+…+b_150p == total_amt`； + · 逾期额 `o_i = total_amt − b_0_30`，负档位**不** ABS/COALESCE（DQ-AR-04 杨创 -20）； + · 三口径分列、禁跨口径合计（`caliber=all` → 400）； + · 缺期（2026-10）⇒ `has_data=false` 且金额 null（禁补 0）； + · §6 裁决：`red.share_90plus_gt=null` ⇒ 90 天占比红线停用（不回落 0.25）； + 新增「单户 >30 天且 >5 万」逐户红线 + 「150+ 占比环比上升」结构恶化红线； + · §6.2 assignee：手工点名 → 近 3 个月主业务员（employee_dict / 同库摘要降级） + → 部门负责人（渠道 118 / 商务 114 / 直属酒窖 117）；两路皆失败 ⇒ skipped[]，不落空行； + · 阈值/责任人全部走 `alert_rules.params`（本套件在本地 SQLite 夹具里改参数复跑， + 证明「改参数不改代码」）； + · 写通道只 INSERT，历史行零 UPDATE/DELETE，同 `alert_message` 幂等跳过。 + +本套件**不连任何生产库**：`tests/conftest.py` 已把引擎替换为内存 SQLite； +`hanke_ar_aging` 非 ORM 模型，故用原始 DDL 在夹具里建表。 +""" +from __future__ import annotations + +import json + +import pytest +from sqlalchemy import text +from sqlalchemy.orm import Session + +from app.api.alert_rules import AlertRule +from app.models import KPIAlert +from app.services import ar_overdue_alert +from app.utils import ar_caliber +from tests.conftest import create_test_kpi, create_test_user, get_token_for_user, auth_header + +ENT = 1 +DDL = """ +CREATE TABLE hanke_ar_aging ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + entity_id INTEGER NOT NULL, + period VARCHAR(7) NOT NULL, + caliber VARCHAR(24) NOT NULL, + unit_code VARCHAR(20) NOT NULL, + unit_name VARCHAR(100) NOT NULL, + opening_amt NUMERIC(18,2) NOT NULL DEFAULT 0, + total_amt NUMERIC(18,2) NOT NULL DEFAULT 0, + b_0_30 NUMERIC(18,2) NOT NULL DEFAULT 0, + b_31_60 NUMERIC(18,2) NOT NULL DEFAULT 0, + b_61_90 NUMERIC(18,2) NOT NULL DEFAULT 0, + b_90_120 NUMERIC(18,2) NOT NULL DEFAULT 0, + b_120_150 NUMERIC(18,2) NOT NULL DEFAULT 0, + b_150p NUMERIC(18,2) NOT NULL DEFAULT 0, + source_file VARCHAR(160) NOT NULL DEFAULT 'fixture.csv', + source_sha256 VARCHAR(64) NOT NULL DEFAULT 'fixture', + loaded_at DATETIME DEFAULT CURRENT_TIMESTAMP +) +""" + +# §6.1 参数最终形状（含 aggregate_assignee，仅测试用显式给值以验证聚合可落库） +BASE_PARAMS = { + "kpi_code": "F_AR_002", + "caliber": ar_caliber.CALIBER_DEBIT, + "red": {"share_90plus_gt": None, "single_overdue_amt_gt": 50000, + "single_overdue_min_days": 30, "count_mom_pct_gte": 0.20, + "structure_worsen": True}, + "yellow": {"share_90plus_min": 0.15, "share_90plus_max": 0.25, + "single_overdue_amt_min": 10000, "single_overdue_amt_max": 50000}, + "assignee_source": { + "mode": "salesman_then_dept", + "lookback_months": 3, + "dept_fallback": {"渠道": 118, "商务": 114, "直属酒窖": 117}, + "dept_by_unit": {"U3": "商务"}, + "aggregate_assignee": "财务应收会计", + }, + "assignee_by_unit": {}, + "exclude_unit_codes": [], + "action_texts": {"red": "催收+停供评估", "yellow": "对账+催收"}, +} + +ZXT_DDL = """ +CREATE TABLE zxt_masterbill ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + entity_id INTEGER, + unit_id INTEGER, + emp_id INTEGER, + bill_type INTEGER DEFAULT 1, + bill_date DATE, + sum_money NUMERIC(18,2) DEFAULT 0, + abst VARCHAR(500) +) +""" + + +def _reset_table(db: Session) -> None: + db.execute(text("DROP TABLE IF EXISTS hanke_ar_aging")) + db.execute(text(DDL)) + db.commit() + + +def _add(db: Session, period: str, code: str, name: str, total: float, buckets: list, + opening: float = 0.0, caliber: str = ar_caliber.CALIBER_DEBIT, entity_id: int = ENT) -> None: + assert abs(sum(buckets) - total) < 0.005, "夹具必须满足六档和 = 合计" + db.execute(text( + "INSERT INTO hanke_ar_aging (entity_id, period, caliber, unit_code, unit_name, opening_amt," + " total_amt, b_0_30, b_31_60, b_61_90, b_90_120, b_120_150, b_150p)" + " VALUES (:e,:p,:c,:code,:name,:op,:tot,:b0,:b1,:b2,:b3,:b4,:b5)" + ), {"e": entity_id, "p": period, "c": caliber, "code": code, "name": name, "op": opening, + "tot": total, "b0": buckets[0], "b1": buckets[1], "b2": buckets[2], + "b3": buckets[3], "b4": buckets[4], "b5": buckets[5]}) + db.commit() + + +@pytest.fixture() +def aging(db: Session): + """两期夹具（2026-08 / 2026-09），含负档位行 + 未逾期户 + 贷方口径干扰行。 + + 2026-09：合计 173,159；90+ = 63,179（36.48%）；逾期户 = U1/U3/U4 共 3 户。 + 2026-08：合计 163,159；90+ = 43,179（26.46%）；逾期户 3 户 ⇒ 环比 0%。 + 150+ 占比：2026-08 = 14.21% → 2026-09 = 16.27%（上升 ⇒ 结构恶化红）。 + """ + _reset_table(db) + _add(db, "2026-09", "U1", "甲公司", 100000, [20000, 10000, 10000, 20000, 15000, 25000], opening=5000) + _add(db, "2026-09", "U2", "乙公司", 40000, [40000, 0, 0, 0, 0, 0]) + _add(db, "2026-09", "U3", "丙公司", 30000, [0, 30000, 0, 0, 0, 0]) + # 负档位（DQ-AR-04 同形）：total=3159, b_0_30=-20, b_150p=3179 ⇒ o_i=3179（不得 ABS 抹成 3139/3159） + _add(db, "2026-09", "U4", "杨创", 3159, [-20, 0, 0, 0, 0, 3179]) + # 干扰行：若查询漏 caliber 过滤 → 会污染借方口径合计 + _add(db, "2026-09", "C1", "贷方户", -34010, [0, 0, 0, 0, 0, -34010], + caliber=ar_caliber.CALIBER_CREDIT_SEPARATE) + _add(db, "2026-08", "U1", "甲公司", 90000, [30000, 10000, 10000, 10000, 10000, 20000]) + _add(db, "2026-08", "U2", "乙公司", 50000, [50000, 0, 0, 0, 0, 0]) + _add(db, "2026-08", "U3", "丙公司", 20000, [0, 20000, 0, 0, 0, 0]) + _add(db, "2026-08", "U4", "杨创", 3159, [-20, 0, 0, 0, 0, 3179]) + return db + + +def _staff(db: Session, *, with_employee_dict: bool = True, with_units_dict: bool = True, + with_zxt: bool = True) -> None: + """维度夹具：units_dict(U1/U3/U4) + employee_dict(14/32/114) + zxt_masterbill。""" + for t in ("units_dict", "employee_dict", "zxt_masterbill"): + db.execute(text("DROP TABLE IF EXISTS %s" % t)) + if with_units_dict: + db.execute(text("CREATE TABLE units_dict (unit_code VARCHAR(20), unit_id INTEGER)")) + for code, uid in (("U1", 101), ("U3", 103), ("U4", 104)): + db.execute(text("INSERT INTO units_dict VALUES (:c,:u)"), {"c": code, "u": uid}) + if with_employee_dict: + db.execute(text("CREATE TABLE employee_dict (id INTEGER, u_Name VARCHAR(64))")) + for i, n in ((14, "李世昌"), (32, "兀晨博"), (114, "王经理"), (118, "渠道负责人"), + (117, "酒窖负责人")): + db.execute(text("INSERT INTO employee_dict VALUES (:i,:n)"), {"i": i, "n": n}) + if with_zxt: + db.execute(text(ZXT_DDL)) + rows = [ + (ENT, 101, 14, "2026-08-10", "向【甲公司】出售商品：李世昌"), + (ENT, 101, 14, "2026-08-11", "向【甲公司】出售商品：李世昌"), + (ENT, 101, 14, "2026-09-05", "向【甲公司】出售商品：李世昌"), + (ENT, 101, 32, "2026-09-20", "向【甲公司】出售商品：兀晨博"), + ] + for e, u, m, d, a in rows: + db.execute(text("INSERT INTO zxt_masterbill (entity_id, unit_id, emp_id, bill_date, abst)" + " VALUES (:e,:u,:m,:d,:a)"), + {"e": e, "u": u, "m": m, "d": d, "a": a}) + db.commit() + + +@pytest.fixture() +def staff(aging: Session): + _staff(aging) + return aging + + +# ─────────────────────── 数据口径 ─────────────────────── + +def test_six_buckets_equal_total_per_row(aging: Session): + rows = aging.execute(text( + "SELECT period, unit_code, total_amt, b_0_30+b_31_60+b_61_90+b_90_120+b_120_150+b_150p AS s" + " FROM hanke_ar_aging WHERE entity_id=1 AND caliber=:c" + ), {"c": ar_caliber.CALIBER_DEBIT}).mappings().all() + assert len(rows) == 8 + for r in rows: + assert abs(float(r["s"]) - float(r["total_amt"])) < 0.005, r + + +def test_caliber_literals_are_exact(): + assert ar_caliber.CALIBERS == ("全户口径", "借方应收户口径", "贷方户单列(净额)") + with pytest.raises(ar_caliber.CaliberError): + ar_caliber.validate_caliber("贷方户单列") # 少「(净额)」⇒ 会漏 42 行 + with pytest.raises(ar_caliber.CaliberError): + ar_caliber.validate_caliber("all") + assert ar_caliber.validate_caliber(None, allow_none=True) == ar_caliber.CALIBER_DEBIT + + +# ─────────────────────── 计算与分级（§6 阈值） ─────────────────────── + +def test_share_90plus_red_disabled_and_structure_worsen(staff: Session): + """§6.1：90 天占比红线置 null ⇒ 不生成该聚合项；150+ 环比上升 ⇒ 结构恶化红。""" + res = ar_overdue_alert.compute(staff, ENT, "2026-09", BASE_PARAMS) + assert res["skipped_reason"] is None + m = res["metrics"] + assert m["total_amt"] == pytest.approx(173159.0, abs=0.01) # 贷方口径行未被混入 + assert m["overdue_90plus"] == pytest.approx(63179.0, abs=0.01) + assert m["share_90plus"] == pytest.approx(63179 / 173159, rel=1e-6) + assert m["share_150p"] == pytest.approx(28179 / 173159, abs=1e-6) # metrics 保留 6 位 + assert m["prev_share_150p"] == pytest.approx(23179 / 163159, abs=1e-6) + + keys = [a["key"] for a in res["aggregates"]] + assert "share_90plus" not in keys, "红线停用后 36.48% 不应再出聚合项（无区分度）" + assert "structure_150p" in keys + agg = [a for a in res["aggregates"] if a["key"] == "structure_150p"][0] + assert agg["level"] == "red" + assert agg["assignee"] == "财务应收会计" + for token in ("2026-09", "150+ 档占比", "16.27%", "14.21%", "结构恶化", "催收+停供评估"): + assert token in agg["alert_message"], token + assert len(agg["alert_message"]) <= 500 + assert any("90 天占比红线停用" in n for n in res["notes"]) + + +def test_single_overdue_red_and_yellow_households(staff: Session): + res = ar_overdue_alert.compute(staff, ENT, "2026-09", BASE_PARAMS) + by_code = {c["unit_code"]: c for c in res["per_customer"]} + # 单户 >30 天且 >5 万 ⇒ 红 + assert by_code["U1"]["level"] == "red" + assert by_code["U1"]["overdue_amt"] == pytest.approx(80000.0) + assert by_code["U1"]["oldest_days"] == 151 + # 单户 1–5 万 ⇒ 黄 + assert by_code["U3"]["level"] == "yellow" and by_code["U3"]["oldest_days"] == 31 + # U4 逾期 3,179 < 1 万 ⇒ 不生成；U2 全额 0-30 ⇒ 不生成 + assert "U4" not in by_code and "U2" not in by_code + for c in res["per_customer"]: + assert c["assignee"], "禁落空 assignee" + assert c["alert_message"].count("｜建议：") == 1 + assert "｜责任人：" in c["alert_message"] + + +def test_assignee_chain_salesman_then_dept(staff: Session): + res = ar_overdue_alert.compute(staff, ENT, "2026-09", BASE_PARAMS) + by_code = {c["unit_code"]: c for c in res["per_customer"]} + # U1：units_dict → unit_id=101 → 主业务员 emp14 单据最多（3 vs 1）→ 李世昌 + assert by_code["U1"]["assignee"] == "李世昌" + assert "salesman" in by_code["U1"]["assignee_source"] + assert by_code["U1"]["emp_id"] == 14 + # U3：近 3 个月无单据 → 部门负责人 商务 114 → 王经理 + assert by_code["U3"]["assignee"] == "王经理" + assert by_code["U3"]["assignee_source"].startswith("dept:商务") + + +def test_assignee_manual_map_has_priority(staff: Session): + p = json.loads(json.dumps(BASE_PARAMS)) + p["assignee_by_unit"] = {"U1": "财务点名人甲"} + res = ar_overdue_alert.compute(staff, ENT, "2026-09", p) + by_code = {c["unit_code"]: c for c in res["per_customer"]} + assert by_code["U1"]["assignee"] == "财务点名人甲" + assert by_code["U1"]["assignee_source"] == "manual:assignee_by_unit" + + +def test_assignee_abst_fallback_when_employee_dict_absent(aging: Session): + """employee_dict 缺表 ⇒ 同库单据摘要尾段制单人姓名降级（仍非空、可审计）。""" + _staff(aging, with_employee_dict=False) + res = ar_overdue_alert.compute(aging, ENT, "2026-09", BASE_PARAMS) + by_code = {c["unit_code"]: c for c in res["per_customer"]} + assert by_code["U1"]["assignee"] == "李世昌" + assert "abst_operator" in by_code["U1"]["assignee_source"] + + +def test_assignee_unresolved_goes_to_skipped(aging: Session): + """维度表全缺 + 部门无法归类 ⇒ 该户不生成，skipped[] 写原因（禁落空 assignee）。""" + _staff(aging, with_employee_dict=False, with_units_dict=False, with_zxt=False) + p = json.loads(json.dumps(BASE_PARAMS)) + p["assignee_source"].pop("dept_by_unit") + res = ar_overdue_alert.compute(aging, ENT, "2026-09", p) + assert res["per_customer"] == [] + skipped = {s["unit_code"]: s for s in res["skipped"] if s["scope"] == "customer"} + assert set(skipped) == {"U1", "U3"} + assert "unit_link_unresolved" in skipped["U1"]["reason"] + assert "dept_unclassified" in skipped["U1"]["reason"] + assert [h for h in ar_overdue_alert.hits(res) if h.get("unit_code")] == [] + + +def test_assignee_missing_is_fail_closed(aging: Session): + p = {"red": {}, "yellow": {}, "assignee_default": "", "assignee_by_unit": {}} + with pytest.raises(ar_overdue_alert.AssigneeMissing): + ar_overdue_alert.ensure_assignee_configured(p) + # 配了逐户派生模式即视为已配置（派生失败按户 skipped，不整体 400） + p["assignee_source"] = {"mode": "salesman_then_dept"} + ar_overdue_alert.ensure_assignee_configured(p) + + +def test_level_boundaries_are_param_driven(aging: Session): + """改 DB 参数复跑 ⇒ 命中集合变化，代码零改动（§6.3 第 11 项：single_overdue_amt_gt）。""" + _staff(aging) + p = json.loads(json.dumps(BASE_PARAMS)) + assert {c["unit_code"] for c in ar_overdue_alert.compute(aging, ENT, "2026-09", p)["per_customer"]} == {"U1", "U3"} + p["red"]["single_overdue_amt_gt"] = 300000 # 5 万 → 30 万 ⇒ U1(8 万) 掉出红线且高于黄线上限 + res = ar_overdue_alert.compute(aging, ENT, "2026-09", p) + assert {c["unit_code"] for c in res["per_customer"]} == {"U3"} + p["red"]["single_overdue_amt_gt"] = 50000 + p["yellow"]["single_overdue_amt_min"] = 100000 # 黄线下限 10 万 ⇒ U3 消失 + assert {c["unit_code"] for c in ar_overdue_alert.compute(aging, ENT, "2026-09", p)["per_customer"]} == {"U1"} + + +def test_missing_period_is_skipped_not_zero(aging: Session): + res = ar_overdue_alert.compute(aging, ENT, "2026-10", BASE_PARAMS) + assert res["skipped_reason"] == "no_data_for_period:2026-10" + assert res["aggregates"] == [] and res["per_customer"] == [] + + +def test_default_period_is_max_and_mom_computed(staff: Session): + res = ar_overdue_alert.compute(staff, ENT, None, BASE_PARAMS) + assert res["period"] == "2026-09" + assert res["metrics"]["prev_period"] == "2026-08" + assert res["metrics"]["prev_overdue_customers"] == 3 + assert res["metrics"]["overdue_customers"] == 3 + assert res["metrics"]["count_mom_pct"] == pytest.approx(0.0) + assert "count_mom" not in {a["key"] for a in res["aggregates"]} + + +def test_no_previous_period_no_mom(aging: Session): + res = ar_overdue_alert.compute(aging, ENT, "2026-08", BASE_PARAMS) + assert res["metrics"]["prev_period"] is None + assert res["metrics"]["count_mom_pct"] is None + assert any("不计算环比" in n for n in res["notes"]) + + +def test_exclude_unit_codes(aging: Session): + _staff(aging) + p = json.loads(json.dumps(BASE_PARAMS)) + p["exclude_unit_codes"] = ["U1"] + res = ar_overdue_alert.compute(aging, ENT, "2026-09", p) + assert "U1" not in {c["unit_code"] for c in res["per_customer"]} + assert res["metrics"]["total_amt"] < 173159 + + +# ─────────────────────── 落库（只 INSERT） ─────────────────────── + +def test_write_alerts_idempotent_and_history_intact(staff: Session): + kpi = create_test_kpi(staff, kpi_code="F_AR_002", kpi_name="逾期应收账款率") + hist = KPIAlert(entity_id=ENT, kpi_id=kpi.id, alert_level="red", + alert_message="历史行-不得被改写", alert_type="actual", status="pending", + assignee="old-owner") + staff.add(hist) + staff.commit() + hist_id = hist.id + + res = ar_overdue_alert.compute(staff, ENT, "2026-09", BASE_PARAMS) + first = ar_overdue_alert.write_alerts(staff, ENT, kpi.id, res) + assert first["inserted"] == len(ar_overdue_alert.hits(res)) > 0 + second = ar_overdue_alert.write_alerts(staff, ENT, kpi.id, res) + assert second == {"inserted": 0, "skipped_duplicate": first["inserted"]} + + rows = staff.query(KPIAlert).filter(KPIAlert.alert_type == "overdue").all() + assert len(rows) == first["inserted"] + assert all(r.assignee and r.status == "pending" and r.kpi_value_id is None for r in rows) + assert all(len(r.alert_message) <= 500 for r in rows) + old = staff.get(KPIAlert, hist_id) + assert old.alert_message == "历史行-不得被改写" and old.assignee == "old-owner" + + +# ─────────────────────── HTTP 面（TestClient + 内存库） ─────────────────────── + +@pytest.fixture() +def fin_headers(client, db): + create_test_user(db, username="fin_probe", role="finance") + return auth_header(get_token_for_user(client, username="fin_probe")) + + +def test_api_meta_trend_overview(staff, client, fin_headers): + r = client.get("/api/cma/ar-aging/meta", headers=fin_headers) + assert r.status_code == 200, r.text + body = r.json() + assert body["available"] is True + assert body["calibers"] == list(ar_caliber.CALIBERS) + assert body["latest_period"] == "2026-09" + assert body["row_count"] == 9 # 8 借方 + 1 贷方（三口径不混算，仅计数） + + r = client.get("/api/cma/ar-aging/trend", headers=fin_headers, + params={"caliber": ar_caliber.CALIBER_DEBIT}) + assert r.status_code == 200 + pts = {p["period"]: p for p in r.json()["points"]} + assert pts["2026-09"]["has_data"] is True + assert pts["2026-10"]["has_data"] is False + assert pts["2026-10"]["total_amt"] is None and pts["2026-10"]["customers"] is None + + r = client.get("/api/cma/ar-aging/overview", headers=fin_headers, + params={"period": "2026-09", "caliber": ar_caliber.CALIBER_DEBIT}) + ov = r.json() + assert ov["has_data"] is True + assert ov["buckets_sum_matches_total"] is True + assert ov["opening_amt"] == pytest.approx(5000.0) + assert ov["opening_amt_non_aging_base"] is True + assert ov["total_amt"] == pytest.approx(173159.0, abs=0.01) + + +def test_api_caliber_all_rejected(aging, client, fin_headers): + for ep in ("overview", "trend", "customers"): + r = client.get(f"/api/cma/ar-aging/{ep}", headers=fin_headers, params={"caliber": "all"}) + assert r.status_code == 400, (ep, r.text) + assert "禁止跨口径合计" in r.json()["detail"] + + +def test_api_customers_sort_and_export(aging, client, fin_headers): + r = client.get("/api/cma/ar-aging/customers", headers=fin_headers, + params={"period": "2026-09", "caliber": ar_caliber.CALIBER_DEBIT, + "overdue_only": "true", "sort": "overdue_amt", "order": "desc"}) + items = r.json()["items"] + assert [i["unit_code"] for i in items][:3] == ["U1", "U3", "U4"] + assert items[0]["overdue_amt"] == pytest.approx(80000.0) + + r = client.get("/api/cma/ar-aging/customers/export", headers=fin_headers, + params={"period": "2026-09", "caliber": ar_caliber.CALIBER_DEBIT}) + assert r.status_code == 200 + assert r.content[:3] == b"\xef\xbb\xbf" # UTF-8 BOM（Excel 直开） + lines = r.content.decode("utf-8-sig").splitlines() + assert lines[0].startswith("截止期,2026-09,口径,借方应收户口径,导出时刻,") + assert "U1" in lines[2] or "U1" in lines[3] + + +def test_api_run_ar_overdue_rule_params_parameterized(staff, client, db, fin_headers): + """端到端：规则参数来自 DB 行 ⇒ 改 `single_overdue_amt_gt` 复跑命中变化，代码零改动。""" + kpi = create_test_kpi(db, kpi_code="F_AR_002", kpi_name="逾期应收账款率") + rule = AlertRule(entity_id=ENT, kpi_id=kpi.id, rule_type=ar_overdue_alert.RULE_TYPE, + trigger_on="actual", enabled=1, params=BASE_PARAMS) + db.add(rule) + db.commit() + + r = client.post("/api/cma/alert-rules/run-ar-overdue", + headers=fin_headers, params={"period": "2026-09", "dry_run": "true"}) + assert r.status_code == 200, r.text + j = r.json() + assert j["params_source"] == "alert_rules#%s" % rule.id + assert j["kpi_id"] == kpi.id + assert j["write_blockers"] == [] + assert j["hit_count"] == 3 # 结构恶化聚合红 + U1 红 + U3 黄 + assert {c["unit_code"] for c in j["per_customer"]} == {"U1", "U3"} + assert all(c["assignee"] for c in j["per_customer"]) + assert j["skipped"] == [] + assert "structure_150p" in {a["key"] for a in j["aggregates"]} + + params2 = json.loads(json.dumps(BASE_PARAMS)) + params2["red"]["single_overdue_amt_gt"] = 300000 # 5 万 → 30 万 + rule.params = params2 + db.commit() + r = client.post("/api/cma/alert-rules/run-ar-overdue", + headers=fin_headers, params={"period": "2026-09", "dry_run": "true"}) + assert r.status_code == 200 + j2 = r.json() + assert {c["unit_code"] for c in j2["per_customer"]} == {"U3"} + assert j2["hit_count"] == 2 and j2["hit_count"] < j["hit_count"] # 聚合红 + U3 黄 + + +def test_api_run_ar_overdue_assignee_empty_400(aging, client, db, fin_headers): + kpi = create_test_kpi(db, kpi_code="F_AR_002x", kpi_name="逾期应收账款率") + p = {"red": {}, "yellow": {}, "assignee_default": "", "assignee_by_unit": {}} + db.add(AlertRule(entity_id=ENT, kpi_id=kpi.id, rule_type=ar_overdue_alert.RULE_TYPE, + trigger_on="actual", enabled=1, params=p)) + db.commit() + r = client.post("/api/cma/alert-rules/run-ar-overdue", + headers=fin_headers, params={"period": "2026-09", "dry_run": "true"}) + assert r.status_code == 400 + assert "责任人" in r.json()["detail"] + + # dry_run=false + 参数覆盖 → 400（防绕过 DB 配置） + r = client.post("/api/cma/alert-rules/run-ar-overdue", headers=fin_headers, + params={"period": "2026-09", "dry_run": "false"}, + json={"params": BASE_PARAMS}) + assert r.status_code == 400 + assert "禁止参数/KPI 覆盖" in r.json()["detail"] + + +def test_api_meta_degrades_when_table_missing(client, db, fin_headers, monkeypatch): + _reset_table(db) + db.execute(text("DROP TABLE hanke_ar_aging")) + db.commit() + import app.api.ar_aging as aa + monkeypatch.setattr(aa, "_TABLE", "hanke_ar_aging_missing_zzz") + r = client.get("/api/cma/ar-aging/meta", headers=fin_headers) + assert r.status_code == 200 + body = r.json() + assert body["available"] is False and "不可用" in body["reason"] + r = client.get("/api/cma/ar-aging/trend", headers=fin_headers) + assert r.status_code == 200 and r.json()["available"] is False diff --git a/frontend/src/api/index.ts b/frontend/src/api/index.ts index 2fac2664..f93ea481 100644 --- a/frontend/src/api/index.ts +++ b/frontend/src/api/index.ts @@ -546,4 +546,19 @@ export const treasuryApi = { listHits: (params?: any) => api.get('/treasury/control-hits', { params }), } +// ── 应收账龄看板（T2 2026-10-20：逐户·六档·三口径禁合计，缺期不补 0）── +export const arAgingApi = { + meta: (params?: any) => api.get('/ar-aging/meta', { params }), + trend: (params?: any) => api.get('/ar-aging/trend', { params }), + overview: (params?: any) => api.get('/ar-aging/overview', { params }), + customers: (params?: any) => api.get('/ar-aging/customers', { params }), + exportUrl: (params: { period?: string; caliber?: string; overdue_only?: boolean }) => { + const q = new URLSearchParams() + if (params.period) q.set('period', params.period) + if (params.caliber) q.set('caliber', params.caliber) + if (params.overdue_only) q.set('overdue_only', 'true') + return `/api/cma/ar-aging/customers/export?${q.toString()}` + }, +} + export default api diff --git a/frontend/src/permission.ts b/frontend/src/permission.ts index f68ff91a..189aeeca 100644 --- a/frontend/src/permission.ts +++ b/frontend/src/permission.ts @@ -39,6 +39,7 @@ export const MENU_ITEMS: MenuItem[] = [ { path: '/receivables', label: '应收款催收', icon: 'Money', roles: ['ceo', 'finance', 'business'], group: '🟢 D 执行与控制' }, { path: '/treasury-dashboard', label: '资金总览', icon: 'Money', roles: ['ceo','finance'], group: '🟢 D 执行与控制' }, { path: '/treasury-payment', label: '资金内控', icon: 'Lock', roles: ['ceo','finance'], group: '🟢 D 执行与控制' }, + { path: '/ar-aging', label: '应收账龄', icon: 'TrendCharts', roles: ['ceo', 'finance', 'business'], group: '🟡 C 监控与评价' }, { path: '/growth-quality', label: '增长质量诊断', icon: 'TrendCharts', roles: ['ceo', 'finance', 'business', 'it'], group: '🟡 C 监控与评价' }, { path: '/product-matrix', label: '波士顿产品矩阵', icon: 'PieChart', roles: ['ceo', 'finance', 'business', 'it'], group: '🟡 C 监控与评价' }, { path: '/deviations', label: '差异分析', icon: 'DataAnalysis', roles: ['ceo', 'finance', 'business', 'it'], group: '🟢 D 执行与控制' }, diff --git a/frontend/src/router/index.ts b/frontend/src/router/index.ts index 284fcd66..843ebed0 100644 --- a/frontend/src/router/index.ts +++ b/frontend/src/router/index.ts @@ -60,6 +60,7 @@ const routes = [ { path: 'receivables', name: 'Receivables', component: () => import('@/views/Receivables.vue'), meta: { title: '应收款催收', roles: ['ceo', 'finance', 'business'], zone: 'B' } }, { path: 'treasury-dashboard', name: 'TreasuryDashboard', component: () => import('@/views/TreasuryDashboard.vue'), meta: { title: '资金总览', roles: ['ceo','finance'], zone: 'B', editable: true } }, { path: 'treasury-payment', name: 'TreasuryPayment', component: () => import('@/views/TreasuryPayment.vue'), meta: { title: '资金内控', roles: ['ceo','finance'], zone: 'B', editable: true } }, + { path: 'ar-aging', name: 'ArAging', component: () => import('@/views/ArAging.vue'), meta: { title: '应收账龄', roles: ['ceo', 'finance', 'business'], zone: 'A' } }, { path: 'growth-quality', name: 'GrowthQuality', component: () => import('@/views/GrowthQuality.vue'), meta: { title: '增长质量诊断', roles: ['ceo', 'finance', 'business', 'it'], zone: 'A' } }, { path: 'product-matrix', name: 'ProductMatrix', component: () => import('@/views/ProductMatrix.vue'), meta: { title: '波士顿产品矩阵', roles: ['ceo', 'finance', 'business', 'it'], zone: 'A' } }, // /tax-compliance: B区（页内职责=税务记录/发票/社保人工录入维护；见盘点表） diff --git a/frontend/src/views/ArAging.vue b/frontend/src/views/ArAging.vue new file mode 100644 index 00000000..80d0f918 --- /dev/null +++ b/frontend/src/views/ArAging.vue @@ -0,0 +1,478 @@ + + + + +