"""驾驶舱 API v2 — 支持时间区间""" from fastapi import APIRouter, Depends, Query, Request, HTTPException from sqlalchemy.orm import Session from sqlalchemy import case, func, or_ from datetime import datetime, timedelta from typing import Optional from app.database import get_db from app.auth_middleware import require_auth, require_role from app.deps import get_entity_id from app.models import KPIDefinition, KPIValue, KPIAlert, User, VoucherDetail from app.utils.cache import get as cache_get, set as cache_set from app.utils.voucher_guard import apply_voucher_guard from app.utils.threshold_parser import eval_threshold from app.utils.alert_entity_scope import month_period_clause # BEP 实体口径单一事实源（2026-10-03 扩面：酣客 entity1 / 博海 entity2 同表驱动派生器与卡面） from app.utils.bep_caliber import CLOSING_SUMMARY_LIKE, caliber_or_none # 「实测区 / 预估区」分区口径单一事实源（财务裁定 2026-10-02，任总「把 H2 预测行与实际分离」） # —— 预估行（data_status='estimated' 或 source_type='forecast'）禁进实测汇总 / 趋势序列 / # 预警依据；仅 include_forecast=true 时以独立字段（forecast_value / is_forecast）单列返回。 from app.utils.kpi_value_scope import actual_only_clause, forecast_only_clause, is_forecast_row # 指标方向（反向/正向）唯一权威源 = app.api.kpis.REVERSE_INDICATORS（2026-09-15 M2 准入2 收口）。 # 本文件原自带一份字面量副本，与 kpis.py 漂移后会与「能力评分」判定相反，故改为导入。 from app.api.kpis import REVERSE_INDICATORS import json import logging logger = logging.getLogger("cma.dashboard") router = APIRouter(prefix="/api/cma/dashboard", tags=["驾驶舱"], dependencies=[Depends(require_role("ceo", "finance", "business", "it"))], ) # ── 本量利·保本点 卡片只读数据（cma-bep-kpi Phase2 2026-09-08）── # 口径与 backend/scripts/kpi_bep_deriver.py 完全一致： # 月销 = voucher_details subject 5001(主营业务收入) 贷方 YTD 月均（万元） # 月管理费 = 5602% 借−贷 净额 YTD 月均（carry_forward=0 行域；账面口径含折旧） # 折旧 = 560213 借−贷 净额 YTD 月均（同上；口径4 见 _decisions/cma-expense-profit-net-caliber-20261002.md） # 渠补率 = **实测累计**（Σ571103借方 ÷ Σ5001贷方，YTD）——取自派生落库值 F_BEP_REBATE（单一事实源） # 保本额 = 月固定成本 ÷ (1-渠补率) # 2026-10-02 口径修订（任总 11:43 CST 放行「改，这样数据更加真实」）： # ① 渠补率不再用政策常量 84.61 参与计算，改用派生落库的实测累计值； # 政策常量仅作对照返回（inputs.rebate_policy_pct），**不参与计算**； # ② 账面/现金两条路径**统一读派生落库值**——此前 book_bep 实时复算 voucher、 # cash_bep 读派生值，9 月数据入库而派生器未跑时同卡两口径不同步； # ③ fail-closed：六条派生值（含渠补率）**同期齐全**才出卡；缺数 ⇒ available=False， # 不出保本值、不回落政策常量（不跨期回落，避免混期口径）。 # 账面口径(含折旧) / 现金口径(剔除折旧) 双口径只读返回，无写操作。 BEP_CODES = ("F_BEP_REVENUE", "F_BEP_COVER", "F_BEP_GAP", "F_BEP_REVENUE_CASH", "F_BEP_COVER_CASH", "F_BEP_REBATE") BEP_BOOK_CODES = ("F_BEP_REVENUE", "F_BEP_COVER", "F_BEP_GAP") # 账面3条必须齐全才 available BEP_CASH_CODES = ("F_BEP_REVENUE_CASH", "F_BEP_COVER_CASH") BEP_REBATE_CODE = "F_BEP_REBATE" # 变动/渠补率派生KPI编码（卡面单一事实源） # 展示用输入口径常量（**仅酣客(entity1)**；多实体一律走 app/utils/bep_caliber.py 的 ENTITY_CALIBERS， # 2026-10-03 扩面：博海 entity2 为 6001/6401/6601+6602+6603/1602） BEP_SALES_SUBJECT = "5001" BEP_ADMIN_PREFIX = "5602" BEP_DEP_SUBJECT = "560213" # 政策常量：仅作对照展示（2026-10-02 起不参与任何计算；博海无渠补科目 ⇒ None） BEP_REBATE_POLICY_PCT = 84.61 def _bep_sum_field(db, entity_id: int, period_from: str, period_to: str, field: str, subject_filter=None, *, exclude_closing: bool = False, subject_filters=None) -> float: """聚合 voucher_details 金额（元），subject_filter: ('eq', code) / ('like', prefix) field="net" = 借−贷 净额（口径4：费用/损益一律净额）。 ⚠️ 净额只在 carry_forward==0 行域成立（含结转腿时损益类净额恒 0） —— 见 /srv/hermes/wisdom/_decisions/cma-expense-profit-net-caliber-20261002.md。 exclude_closing（2026-10-03）：该账套结转腿未标 carry_forward 时（博海：全账套 cf 恒 0）， 须按 summary NOT LIKE '%结转%' 显式剔除，否则已结账期间净额恒 0。 subject_filters：多条 OR（固定成本 6601+6602+6603）。 作废/撤单标记过滤（2026-09-14 收口）: 已作废凭证不参与 BEP 均值。 """ expr = (VoucherDetail.debit_amount - VoucherDetail.credit_amount if field == "net" else getattr(VoucherDetail, field)) q = db.query(func.sum(expr)).filter( VoucherDetail.entity_id == entity_id, VoucherDetail.carry_forward == 0, VoucherDetail.period >= period_from, VoucherDetail.period <= period_to, ) if exclude_closing: q = q.filter(or_(VoucherDetail.summary.is_(None), ~VoucherDetail.summary.like(CLOSING_SUMMARY_LIKE))) filt = subject_filter if subject_filters: clauses = [] for kind, code in subject_filters: clauses.append(VoucherDetail.subject_code == code if kind == "eq" else VoucherDetail.subject_code.like(code + "%")) q = q.filter(or_(*clauses)) elif filt: kind, code = filt if kind == "eq": q = q.filter(VoucherDetail.subject_code == code) elif kind == "like": q = q.filter(VoucherDetail.subject_code.like(code + "%")) q = apply_voucher_guard(db, q, VoucherDetail) val = q.scalar() return float(val or 0.0) def _bep_inputs(db, entity_id: int, cal: dict, period_from: str, period: str, months: int) -> dict: """卡面展示用输入（万元/百分比）——按实体口径现算，与派生器同源同式。""" sales = _bep_sum_field(db, entity_id, period_from, period, cal["revenue_field"], cal["revenue_filter"], exclude_closing=cal["exclude_closing"]) / 10000.0 admin = _bep_sum_field(db, entity_id, period_from, period, "net", subject_filters=cal["fixed_filters"], exclude_closing=cal["exclude_closing"]) / 10000.0 dep = _bep_sum_field(db, entity_id, period_from, period, cal["dep_field"], cal["dep_filter"], exclude_closing=cal["exclude_closing"]) / 10000.0 return { "sales_avg": round(sales / months, 2), "admin_avg": round(admin / months, 2), "dep_avg": round(dep / months, 2), "cash_admin_avg": round((admin - dep) / months, 2), } def _bep_level(value, kpi) -> str: """按 KPI 静态阈值文本判定级别（绿→黄→红，命中即止），无阈值返回 none。 与 alert_generator / kpis._static_threshold_level 语义一致。""" import re as _re for level, field in (("green", "threshold_green"), ("yellow", "threshold_yellow"), ("red", "threshold_red")): expr = getattr(kpi, field, None) if not expr: continue m = _re.match(r"(>=|<=|>|<|=|!=)\s*([\d.]+)", str(expr).strip()) if not m: continue op, val = m.group(1), float(m.group(2)) hit = { ">=": value >= val, "<=": value <= val, ">": value > val, "<": value < val, "=": value == val, "!=": value != val, }[op] if hit: return level return "none" @router.get("/bep-card") def get_bep_card( db: Session = Depends(get_db), entity_id: int = Depends(get_entity_id), ): """本量利·保本点 卡片数据（只读，A区）——账面/现金双口径 2026-10-02 口径修订（任总 11:43 CST 放行）： · 渠补率 = 派生落库的**实测累计值**（F_BEP_REBATE，单一事实源）；政策常量仅对照返回； · 账面 / 现金两条路径**统一读派生落库值**（不再实时复算 voucher）； · fail-closed：六条派生值同期不齐（尤其缺渠补率）⇒ available=False，不出保本值、不回落常量。 """ cal = caliber_or_none(entity_id) if cal is None: return {"available": False, "entity_id": entity_id, "message": f"entity={entity_id} 未注册 BEP 口径（见 app/utils/bep_caliber.py）", "rebate_policy_pct": BEP_REBATE_POLICY_PCT, "readonly": True} kpis = {k.kpi_code: k for k in db.query(KPIDefinition).filter( KPIDefinition.entity_id == entity_id, KPIDefinition.status == "active", KPIDefinition.kpi_code.in_(BEP_CODES), ).all()} missing_book = [c for c in BEP_BOOK_CODES if c not in kpis] if missing_book: return {"available": False, "message": "BEP账面口径派生KPI未完整注册", "found": list(kpis.keys()), "missing": missing_book} # 派生落库值（同派生器：source_type='derived'）——按 code × period 建表，period desc 排序 # ⚠️ 不过滤 actual_value IS NULL：fail-closed 落库的 pending 期（值 NULL）必须参与「最新期」判定， # 否则卡面会悄悄跨期回落到上一期，掩盖「本期不出值」的事实。 rows = db.query(KPIValue, KPIDefinition.kpi_code).join( KPIDefinition, KPIDefinition.id == KPIValue.kpi_id, ).filter( KPIDefinition.entity_id == entity_id, KPIDefinition.kpi_code.in_(BEP_CODES), KPIValue.source_type == "derived", month_period_clause(KPIValue.period), ).order_by(KPIValue.period.desc(), KPIValue.id.desc()).all() by_code = {} # {code: {period: value|None}} remark_map = {} # {(code, period): remark} —— 用于区分「不适用」与「缺数」 for v, code in rows: by_code.setdefault(code, {}).setdefault(v.period, v.actual_value) remark_map.setdefault((code, v.period), v.remark or "") # 卡面期 = 账面三条派生行的**最新期间**（与派生器目标期一致） book_periods = sorted({p for c in BEP_BOOK_CODES for p in by_code.get(c, {})}, reverse=True) if not book_periods: return {"available": False, "entity_id": entity_id, "message": "尚无派生值，请先运行 kpi_bep_deriver.py", "rebate_policy_pct": BEP_REBATE_POLICY_PCT, "readonly": True} period = book_periods[0] year = period[:4] months = int(period[5:7]) period_from = f"{year}-01" # fail-closed：六条派生值（含变动成本率）本期必须齐全，缺一条即不出保本值 missing_at_period = [c for c in BEP_CODES if by_code.get(c, {}).get(period) is None] if missing_at_period: # ① 不适用（派生器 fail-closed 落库：remark 已注明「保本点不适用」） # —— 与「缺数」区分：卡面显示「不适用」而非「数据不齐」，且不得显示 0。 na_remarks = [remark_map.get((c, period), "") for c in missing_at_period] na_hit = [r for r in na_remarks if "不适用" in r] if na_hit and BEP_REBATE_CODE not in missing_at_period: reason = na_hit[0] label = ("边际贡献为负 · 保本点不适用" if "边际贡献非正" in reason else "固定成本退化 · 保本点不适用") return {"available": False, "not_applicable": True, "entity_id": entity_id, "period": period, "months": months, "label": label, "reason": reason, "rebate_pct": by_code.get(BEP_REBATE_CODE, {}).get(period), "rebate_policy_pct": (cal.get("policy_pct") if cal.get("policy_pct") is not None else BEP_REBATE_POLICY_PCT), "entity_name": cal["name"], "missing_codes": missing_at_period, "derived_periods": book_periods[:6], "data_source": "derived", "readonly": True} # ② 缺数（真不齐） if BEP_REBATE_CODE in missing_at_period: msg = (f"{cal['margin_label']}不可得：本期未落 F_BEP_REBATE 派生值 ⇒ 不出保本值" "（fail-closed，不回落政策常量、不跨期回落）") else: msg = ("本期派生值不齐 %s ⇒ 不出保本值（fail-closed，不跨期回落）" % ",".join(missing_at_period)) return {"available": False, "entity_id": entity_id, "period": period, "message": msg, "missing_codes": missing_at_period, "rebate_policy_pct": (cal.get("policy_pct") if cal.get("policy_pct") is not None else BEP_REBATE_POLICY_PCT), "derived_periods": book_periods[:6], "readonly": True} # 六条派生落库值（同期）—— 账面 / 现金两条路径同一来源，卡面不做实时复算 d = {c: by_code[c][period] for c in BEP_CODES} book_bep = float(d["F_BEP_REVENUE"]) book_cover = float(d["F_BEP_COVER"]) book_gap = float(d["F_BEP_GAP"]) cash_bep = float(d["F_BEP_REVENUE_CASH"]) cash_cover = float(d["F_BEP_COVER_CASH"]) cash_gap = round(100.0 - cash_cover, 1) # 缺口 = 100 − 覆盖率（与派生器同口径） rebate_pct = float(d[BEP_REBATE_CODE]) # 实测累计渠补率（单一事实源） cm_pct = round(100.0 - rebate_pct, 2) # 输入口径（万元，展示用；数值口径一律以派生落库值为准）——按实体口径现算 inp = _bep_inputs(db, entity_id, cal, period_from, period, months) sales_avg = inp["sales_avg"] admin_avg = inp["admin_avg"] dep_avg = inp["dep_avg"] cash_admin = inp["cash_admin_avg"] policy_pct = cal.get("policy_pct") # 博海无渠补政策常量 ⇒ None（前端不显示对照） # 回显六条派生落库值（同期） derived = {c: {"period": period, "value": d[c]} for c in BEP_CODES} cover_kpi = kpis["F_BEP_COVER"] book_level = _bep_level(book_cover, cover_kpi) cash_cover_kpi = kpis.get("F_BEP_COVER_CASH") or cover_kpi cash_level = _bep_level(cash_cover, cash_cover_kpi) gap_kpi = kpis["F_BEP_GAP"] book_gap_level = _bep_level(book_gap, gap_kpi) cash_gap_level = _bep_level(cash_gap, gap_kpi) return { "available": True, "entity_id": entity_id, "period": period, "months": months, "inputs": { "sales_avg": round(sales_avg, 2), "admin_avg": round(admin_avg, 2), "dep_avg": round(dep_avg, 2), "rebate_pct": rebate_pct, "rebate_policy_pct": BEP_REBATE_POLICY_PCT, # 政策常量：仅作对照，不参与计算 "rebate_source": "derived:F_BEP_REBATE(Σ571103借方÷Σ5001贷方, YTD累计)", "cm_pct": cm_pct, }, "book": { "bep_revenue": book_bep, "cover": book_cover, "gap": book_gap, "cover_level": book_level, "gap_level": book_gap_level, }, "cash": { "admin_cash": round(cash_admin, 2), "bep_revenue": cash_bep, "cover": cash_cover, "gap": cash_gap, "cover_level": cash_level, "gap_level": cash_gap_level, }, "derived": derived, "data_source": "derived", # 账面/现金同源：均读派生落库值 "formula": f"保本销售额 = 月固定成本 ÷ (1 − 实测渠补率{rebate_pct:.2f}%)", "readonly": True, } def parse_period(period_type: str, start_date: str = None, end_date: str = None): """解析时间区间""" today = datetime.now() if period_type == "month": start = today.replace(day=1, hour=0, minute=0, second=0, microsecond=0) end = today elif period_type == "quarter": q = (today.month - 1) // 3 start = today.replace(month=q*3+1, day=1, hour=0, minute=0, second=0, microsecond=0) end = today elif period_type == "year": start = today.replace(month=1, day=1, hour=0, minute=0, second=0, microsecond=0) end = today elif period_type == "custom" and start_date and end_date: start = datetime.strptime(start_date, "%Y-%m-%d") end = datetime.strptime(end_date, "%Y-%m-%d") + timedelta(days=1) else: start = today.replace(day=1, hour=0, minute=0, second=0, microsecond=0) end = today return start, end def period_prefix(period_type: str): """生成SQL期间前缀匹配""" if period_type == "month": return datetime.now().strftime("%Y-%m") elif period_type == "quarter": now = datetime.now() q = (now.month - 1) // 3 months = [f"{now.year}-{m:02d}" for m in range(q*3+1, q*3+4)] return months elif period_type == "year": return str(datetime.now().year) return None def kpi_target_by_frequency(k): """按考核频率返回对应周期的目标值（多粒度改造） monthly/weekly -> target_monthly; quarterly/half_year -> target_quarterly; yearly -> target_yearly 兼容: 对应列无值时回退 target_value """ freq = (k.frequency or "monthly").lower() if freq in ("monthly", "weekly"): val = getattr(k, "target_monthly", None) elif freq in ("quarterly", "half_year"): val = getattr(k, "target_quarterly", None) elif freq == "yearly": val = getattr(k, "target_yearly", None) else: val = None if val is None: val = k.target_value return val @router.get("/summary") def get_dashboard_summary(role: str = Query("ceo"), period: str = Query("month"), db: Session = Depends(get_db), entity_id: int = Depends(get_entity_id)): cache_key = f"summary:{role}:{period}:{entity_id}" cached = cache_get("dashboard", cache_key) if cached: return cached kpi_total = db.query(func.count(KPIDefinition.id)).filter( KPIDefinition.status == "active", KPIDefinition.entity_id == entity_id).scalar() alert_count = db.query(func.count(KPIAlert.id)).filter( KPIAlert.entity_id == entity_id, KPIAlert.status == "pending", ).scalar() dims = db.query(KPIDefinition.dimension, func.count(KPIDefinition.id)).filter( KPIDefinition.status == "active", KPIDefinition.entity_id == entity_id ).group_by(KPIDefinition.dimension).all() # 读取最近一次同步状态（从日志文件最后一行） sync_status = {"last_sync": None, "status": "unknown", "detail": ""} try: with open("/var/log/cma-daily-sync.log", "r") as f: lines = f.readlines() # 从最后往前找包含 "完成" 或 "失败" 的行 for line in reversed(lines[-50:]): if "全部完成" in line: sync_status["status"] = "success" sync_status["last_sync"] = line.strip() break elif "失败" in line or "ERROR" in line: sync_status["status"] = "failed" sync_status["last_sync"] = line.strip() break else: # 没找到完成/失败标记，取最后一行 sync_status["last_sync"] = lines[-1].strip() if lines else None except Exception as e: sync_status["detail"] = str(e) result = { "kpi_total": kpi_total or 0, "alert_count": alert_count or 0, "dimension_stats": [{"dimension": d[0], "count": d[1]} for d in dims], "sync_status": sync_status, } cache_set("dashboard", cache_key, result, ttl_seconds=30) return result def _sum_rows(values): """区间汇总辅助：保留原实现的「和为 0 视为无数据」历史语义（回归等值前提）""" if not values: return None total = sum(v.actual_value for v in values if v.actual_value) return total if total else None def _aggregate_actual(values, label): """实测区区间汇总 → 与原匿名对象同形（actual_value / period）""" total = _sum_rows(values) if total is None: return None return type('obj', (object,), {"actual_value": total, "period": label})() @router.get("/kpis") def get_dashboard_kpis(role: str = Query("ceo"), period: str = Query("month"), start_date: str = Query(None), end_date: str = Query(None), include_forecast: bool = Query( False, description="是否单列返回预估区（data_status=estimated 或 source_type=forecast）；" "默认 false=只取实测区，预估行禁与实测相加"), db: Session = Depends(get_db), entity_id: int = Depends(get_entity_id)): start, end = parse_period(period, start_date, end_date) period_str = start.strftime("%Y-%m") kpis = db.query(KPIDefinition).filter( KPIDefinition.status == "active", KPIDefinition.entity_id == entity_id ).all() result = [] forecast_section = [] # 预估区（单列，不计入上面的 actual_value） for k in kpis: base_query = db.query(KPIValue).filter(KPIValue.kpi_id == k.id) # 实测区恒过滤：预估行（estimated/forecast）不进实测取值与汇总 actual_query = base_query.filter(actual_only_clause()) forecast_query = base_query.filter(forecast_only_clause()) if include_forecast else None actual_value = None forecast_value = None is_forecast = False # 期间展示标签（各模式下与实测取值同口径） if period == "month": range_label = period_str elif period == "quarter": _months = period_prefix("quarter") range_label = f"{_months[0]}~{_months[-1]}" elif period == "year": range_label = period_str[:4] elif period == "custom" and start_date and end_date: range_label = f"{start_date}~{end_date}" else: range_label = None if period == "month": latest = actual_query.filter(KPIValue.period == period_str).order_by(KPIValue.id.desc()).first() if forecast_query is not None: fc_row = forecast_query.filter( KPIValue.period == period_str).order_by(KPIValue.id.desc()).first() if fc_row is not None: forecast_value = fc_row.actual_value if latest is None: # 该期只有预估行：值单列展示并打标，不并入实测 is_forecast = True elif period == "quarter": months = period_prefix("quarter") latest = _aggregate_actual(actual_query.filter(KPIValue.period.in_(months)).all(), range_label) if forecast_query is not None: forecast_value = _sum_rows(forecast_query.filter(KPIValue.period.in_(months)).all()) elif period == "year": latest = _aggregate_actual( actual_query.filter(KPIValue.period.like(f"{period_str[:4]}%")).all(), range_label) if forecast_query is not None: forecast_value = _sum_rows( forecast_query.filter(KPIValue.period.like(f"{period_str[:4]}%")).all()) elif period == "custom" and start_date and end_date: periods = [] d = start while d <= end: periods.append(d.strftime("%Y-%m")) d += timedelta(days=32) d = d.replace(day=1) latest = _aggregate_actual(actual_query.filter(KPIValue.period.in_(set(periods))).all(), range_label) if forecast_query is not None: forecast_value = _sum_rows( forecast_query.filter(KPIValue.period.in_(set(periods))).all()) else: latest = actual_query.filter( month_period_clause(KPIValue.period)).order_by(KPIValue.period.desc()).first() if forecast_query is not None: fc_row = forecast_query.filter( month_period_clause(KPIValue.period)).order_by(KPIValue.period.desc()).first() if fc_row is not None: forecast_value = fc_row.actual_value if latest is None: is_forecast = True range_label = fc_row.period alert = db.query(KPIAlert).filter( KPIAlert.kpi_id == k.id, KPIAlert.status == "pending", ).order_by(KPIAlert.id.desc()).first() if forecast_value is not None: forecast_section.append({ "kpi_id": k.id, "kpi_code": k.kpi_code, "kpi_name": k.kpi_name, "value": forecast_value, "unit": k.unit, "period": (latest.period if latest is not None else range_label), "data_status": "estimated", "is_forecast": True, "period_type": "forecast", }) result.append({ "id": k.id, "kpi_code": k.kpi_code, "kpi_name": k.kpi_name, "dimension": k.dimension, "unit": k.unit, "target_value": k.target_value, "target_monthly": k.target_monthly, "target_quarterly": k.target_quarterly, "target_yearly": k.target_yearly, "actual_value": latest.actual_value if latest else None, "period": latest.period if latest else None, "forecast_value": forecast_value, "is_forecast": is_forecast, "period_type": "forecast" if is_forecast else "actual", "alert_level": alert.alert_level if alert else "none", "alert_message": alert.alert_message if alert else None, "frequency": k.frequency, "responsible_dept": k.responsible_dept, }) resp = {"data": result, "period": period, "range": {"start": start.strftime("%Y-%m-%d"), "end": end.strftime("%Y-%m-%d")}} if include_forecast: resp["forecast"] = forecast_section return resp @router.get("/my-kpis") def get_my_kpis( current_user: User = Depends(require_auth), period: str = Query("month"), include_forecast: bool = Query(False, description="是否单列返回预估区；默认 false=只取实测区"), db: Session = Depends(get_db), entity_id: int = Depends(get_entity_id), ): """获取当前用户负责的KPI - business角色：只看自己负责的KPI - 其他角色：看所有有预警的KPI """ role = current_user.role username = current_user.username name = current_user.name period_str = datetime.now().strftime("%Y-%m") kpis = db.query(KPIDefinition).filter( KPIDefinition.status == "active", KPIDefinition.entity_id == entity_id, # 账套隔离 2026-09-21 ).all() result = [] forecast_section = [] # 预估区（单列，不进 actual_value / trend） for k in kpis: # business角色筛选 if role == "business": responsible = (k.responsible_user or "").strip() if responsible and responsible != username and responsible != name: continue # 实测区恒过滤（预估行禁作实测值/禁进趋势序列） latest = db.query(KPIValue).filter( KPIValue.kpi_id == k.id, KPIValue.period == period_str, actual_only_clause(), ).order_by(KPIValue.id.desc()).first() forecast_value = None is_forecast = False if include_forecast: fc_row = db.query(KPIValue).filter( KPIValue.kpi_id == k.id, KPIValue.period == period_str, forecast_only_clause(), ).order_by(KPIValue.id.desc()).first() # 无月期预估行时回退到最近一条预估行（半年度 H2 等伪期间），仅作单列展示 if fc_row is None: fc_row = db.query(KPIValue).filter( KPIValue.kpi_id == k.id, forecast_only_clause(), ).order_by(KPIValue.period.desc(), KPIValue.id.desc()).first() if fc_row is not None: forecast_value = fc_row.actual_value is_forecast = latest is None forecast_section.append({ "kpi_id": k.id, "kpi_code": k.kpi_code, "kpi_name": k.kpi_name, "value": forecast_value, "unit": k.unit, "period": fc_row.period, "data_status": "estimated", "is_forecast": True, "period_type": "forecast", }) alert = db.query(KPIAlert).filter( KPIAlert.kpi_id == k.id, KPIAlert.status == "pending", ).order_by(KPIAlert.id.desc()).first() trend_values = db.query(KPIValue).filter( KPIValue.kpi_id == k.id, month_period_clause(KPIValue.period), actual_only_clause(), ).order_by(KPIValue.period.desc()).limit(6).all() trend = [{"period": v.period, "value": v.actual_value} for v in reversed(trend_values)] result.append({ "id": k.id, "kpi_code": k.kpi_code, "kpi_name": k.kpi_name, "dimension": k.dimension, "unit": k.unit, "target_value": k.target_value, "target_monthly": k.target_monthly, "target_quarterly": k.target_quarterly, "target_yearly": k.target_yearly, "actual_value": latest.actual_value if latest else None, "period": latest.period if latest else period_str, "forecast_value": forecast_value, "is_forecast": is_forecast, "period_type": "forecast" if is_forecast else "actual", "alert_level": alert.alert_level if alert else "none", "alert_message": alert.alert_message if alert else None, "alert_id": alert.id if alert else None, "frequency": k.frequency, "responsible_dept": k.responsible_dept, "responsible_user": k.responsible_user, "trend": trend, "threshold_green": k.threshold_green, "threshold_yellow": k.threshold_yellow, "threshold_red": k.threshold_red, }) resp = {"data": result, "user_role": role, "user_name": name, "period": period_str} if include_forecast: resp["forecast"] = forecast_section return resp @router.get("/finance-analysis") def get_finance_analysis( current_user: User = Depends(require_auth), period: str = Query("month"), include_forecast: bool = Query(False, description="是否单列返回预估区；默认 false=只取实测区"), db: Session = Depends(get_db), entity_id: int = Depends(get_entity_id), ): """财务工作台分析数据""" period_str = datetime.now().strftime("%Y-%m") finance_kpis = db.query(KPIDefinition).filter( KPIDefinition.status == "active", KPIDefinition.dimension == "finance", KPIDefinition.entity_id == entity_id, ).all() kpi_data = [] forecast_section = [] # 预估区（单列，不进 trend / summary） for k in finance_kpis: # 实测区恒过滤（预估行禁作实测值/禁进趋势序列） latest = db.query(KPIValue).filter( KPIValue.kpi_id == k.id, KPIValue.period == period_str, actual_only_clause(), ).order_by(KPIValue.id.desc()).first() forecast_value = None if include_forecast: fc_row = db.query(KPIValue).filter( KPIValue.kpi_id == k.id, forecast_only_clause(), ).order_by(KPIValue.period.desc(), KPIValue.id.desc()).first() if fc_row is not None: forecast_value = fc_row.actual_value forecast_section.append({ "kpi_id": k.id, "kpi_code": k.kpi_code, "kpi_name": k.kpi_name, "value": forecast_value, "unit": k.unit, "period": fc_row.period, "data_status": "estimated", "is_forecast": True, "period_type": "forecast", }) trend_values = db.query(KPIValue).filter( KPIValue.kpi_id == k.id, month_period_clause(KPIValue.period), actual_only_clause(), ).order_by(KPIValue.period.desc()).limit(6).all() trend = [{"period": v.period, "value": v.actual_value} for v in reversed(trend_values)] alert = db.query(KPIAlert).filter( KPIAlert.kpi_id == k.id, KPIAlert.status == "pending", ).order_by(KPIAlert.id.desc()).first() kpi_data.append({ "id": k.id, "kpi_code": k.kpi_code, "kpi_name": k.kpi_name, "unit": k.unit, "target_value": k.target_value, "target_monthly": k.target_monthly, "target_quarterly": k.target_quarterly, "target_yearly": k.target_yearly, "actual_value": latest.actual_value if latest else None, "forecast_value": forecast_value, "is_forecast": False, "threshold_green": k.threshold_green, "threshold_yellow": k.threshold_yellow, "threshold_red": k.threshold_red, "trend": trend, "alert_level": alert.alert_level if alert else "none", "frequency": k.frequency, }) total_sales = next((k for k in kpi_data if k["kpi_code"] == "SALES_TOTAL"), None) gross_profit = next((k for k in kpi_data if k["kpi_code"] == "SALES_PROFIT_RATE"), None) cost_control = next((k for k in kpi_data if k["kpi_code"] == "COST_CONTROL_RATE"), None) receivable = next((k for k in kpi_data if k["kpi_code"] == "RECEIVABLE_TURNOVER"), None) resp = { "period": period_str, "kpis": kpi_data, "summary": { "total_sales": total_sales["actual_value"] if total_sales else None, "gross_profit_rate": gross_profit["actual_value"] if gross_profit else None, "cost_control_rate": cost_control["actual_value"] if cost_control else None, "receivable_turnover": receivable["actual_value"] if receivable else None, } } if include_forecast: resp["forecast"] = forecast_section return resp @router.get("/predict") def predict_kpis(include_forecast: bool = Query( False, description="是否单列返回预估区；默认 false=只取实测区（预估行禁进预测序列）"), db: Session = Depends(get_db), entity_id: int = Depends(get_entity_id)): """基于历史趋势预测下月KPI值（简单线性回归） 口径（财务裁定 2026-10-02）：预估行（estimated/forecast）**不进回归序列**—— 预估禁与实测相加、禁作预测/预警依据。`include_forecast=true` 仅额外单列返回 预估区（`forecast_rows`），不改变回归取数。 """ from datetime import datetime, timedelta period_str = datetime.now().strftime("%Y-%m") next_month = int(period_str[5:7]) + 1 next_year = int(period_str[:4]) if next_month > 12: next_month = 1 next_year += 1 next_period = f"{next_year}-{next_month:02d}" kpis = db.query(KPIDefinition).filter( KPIDefinition.status == "active", KPIDefinition.entity_id == entity_id ).all() predictions = [] forecast_rows = [] # 预估区（单列，不参与回归） for k in kpis: values = db.query(KPIValue).filter( KPIValue.kpi_id == k.id, actual_only_clause(), ).order_by(KPIValue.period.asc()).all() if include_forecast: for fc_row in db.query(KPIValue).filter( KPIValue.kpi_id == k.id, forecast_only_clause(), ).order_by(KPIValue.period.asc()).all(): forecast_rows.append({ "kpi_id": k.id, "kpi_code": k.kpi_code, "kpi_name": k.kpi_name, "period": fc_row.period, "value": fc_row.actual_value, "unit": k.unit, "data_status": "estimated", "is_forecast": True, "period_type": "forecast", }) # 需要至少3个数据点才能做预测 if len(values) < 3: continue # 简单线性回归: y = a + bx points = [(i, v.actual_value) for i, v in enumerate(values) if v.actual_value is not None] if len(points) < 3: continue n = len(points) sum_x = sum(p[0] for p in points) sum_y = sum(p[1] for p in points) sum_xy = sum(p[0] * p[1] for p in points) sum_xx = sum(p[0] ** 2 for p in points) # 斜率 b = (n*sum_xy - sum_x*sum_y) / (n*sum_xx - sum_x*sum_x) denom = n * sum_xx - sum_x * sum_x if denom == 0: continue b = (n * sum_xy - sum_x * sum_y) / denom a = (sum_y - b * sum_x) / n # 预测下个月（x = n，因为最后一个索引是 n-1） predicted_value = a + b * n # 检查预测值是否触发阈值 # 统一走 app.utils.threshold_parser（阈值语法单一事实源，2026-09-16 B1 修复）： # 原实现手工切操作符 + float()，只识别单条件标量表达式；生产库存在多条件 # 表达式（如 `<=36|>=124`）/区间/负数 → float() 抛 ValueError → 端点 500。 # eval_threshold 对非法/空表达式返回 False（不致命），并支持 A~B、C1|C2、负数。 alert_level = "none" if eval_threshold(predicted_value, k.threshold_red): alert_level = "red" elif eval_threshold(predicted_value, k.threshold_yellow): alert_level = "yellow" predictions.append({ "kpi_id": k.id, "kpi_code": k.kpi_code, "kpi_name": k.kpi_name, "target_value": k.target_value, "last_value": points[-1][1] if points else None, "predicted_value": round(predicted_value, 2), "predicted_period": next_period, "alert_level": alert_level, "trend": "up" if b > 0 else ("down" if b < 0 else "stable"), "confidence": "high" if len(points) >= 6 else ("medium" if len(points) >= 4 else "low"), "data_points": len(points), }) resp = { "current_period": period_str, "next_period": next_period, "predictions": predictions, "kpi_count": len(kpis), "predictable_count": len(predictions), } if include_forecast: resp["forecast_rows"] = forecast_rows return resp # ── 个人工作台 ────────────────────────────── # KPIValue 取数口径（2026-09-18 工作台「我的KPI」期间缺口修正） # 原实现（my_dashboard 内联）按 calculated_at 倒序取「最近写入」的一条：批量补录/复采会把 # **旧期间**的行写得更晚 → 卡片显示过期期间。实测 entity1：F_REVENUE/F_NET_PROFIT 的 # 2026-07 行写入时刻(2026-09-14 17:01:22, id=2128/2130)晚于 2026-08 行(17:00:30, id=2125/2127) # → 取到 7 月；C_REBATE_RATE 1–4 月补录行 calculated_at 并列(2026-08-24 17:17:38) → 取到 4 月。 # 新口径：优先取「月期间 YYYY-MM」中最新的一期（月期是采集/录入的常规粒度）；若该 KPI 只有 # H1/H2/FY 等汇总期行，则回退到原按 calculated_at 的取法（不改变无月期 KPI 的行为）。 # 注意：**不能**只按 period 字典序排 —— '2026-H1'/'2026-H2' 字典序大于 '2026-08'，会把 # 季度汇总行/半年度预估值(forecast)当成"最新"。 # 期间口径（R3 正则 / SQL 等价式）已于 2026-09-26 收口到 app/utils/alert_entity_scope.py # （is_month_period / month_period_clause），此处不再保留第二份副本。 # 主值优先（2026-09-18 同期间重复行判据收口，项目单 project-20260918-kpi-row-priority） # 同一期间可能并存两行：①主值（source_type='ledger' / data_status='verified'，结账/复核口径） # ②自动归集重复行（source_type='auto_collect'，备注自述「与主值为同一事实重复行」）。 # 原判据仅看写入时刻：重复行由复采批量写入、calculated_at 更晚 ⇒ 会被选中（实测 entity1 # 2026-08 的 F_REVENUE/F_NET_PROFIT：2136/2127 auto_collect 写于 09-14，晚于主值 1822/2102）。 # 今日两行数值一致（Δ=0）故用户不可见；一旦重复行与主值漂移，卡片会静默显示重复行的值。 # 故在**同一期间内**加「主值优先」：verified（已复核）> ledger（账面主值）> 其余。 # 注意：跨期间仍以 period 倒序为第一判据（不改变 d33b5e1 的期间口径——当期只有未复核行时， # 不会回退到上一期已复核行）。 _MAIN_VALUE_PRIORITY = case( (KPIValue.data_status == "verified", 0), (KPIValue.source_type == "ledger", 1), else_=2, ) def latest_kpi_value(db: Session, kpi_id: int) -> Optional[KPIValue]: """该 KPI 的最新可用值：月期间优先（同期内主值优先），无月期间行时回退 calculated_at 倒序。 口径（财务裁定 2026-10-02）：**只取实测区**——预估行（data_status='estimated' 或 source_type='forecast'）不得作为卡片「最新值」（工作台卡片=实测口径）。 """ monthly = ( db.query(KPIValue) .filter(KPIValue.kpi_id == kpi_id, month_period_clause(KPIValue.period), actual_only_clause()) .order_by( KPIValue.period.desc(), _MAIN_VALUE_PRIORITY, KPIValue.calculated_at.desc(), KPIValue.id.desc(), ) .first() ) if monthly is not None: return monthly return ( db.query(KPIValue) .filter(KPIValue.kpi_id == kpi_id, actual_only_clause()) .order_by(KPIValue.calculated_at.desc(), KPIValue.id.desc()) .first() ) @router.get("/my-dashboard") def my_dashboard( current_user: User = Depends(require_auth), include_forecast: bool = Query(False, description="是否单列返回预估区；默认 false=只取实测区"), db: Session = Depends(get_db), entity_id: int = Depends(get_entity_id), ): """个人工作台：返回我的KPI、改善行动、待办提醒（账套隔离: 按token企业过滤）""" username = current_user.username name = current_user.name role = current_user.role # 角色预设KPI编码 ROLE_PRESET_KPIS = { "ceo": ["F_REVENUE", "F_NET_PROFIT", "F_COST_RATIO", "C_REBATE_RATE", "P_DELIVERY", "F_FCF"], "finance": ["F_REVENUE", "F_NET_PROFIT", "F_COST_RATIO", "F_OP_CFLOW", "F_ROE"], "business": ["C_REBATE_RATE", "C_NEW_CLIENTS", "F_REVENUE"], "it": [], } preset_codes = ROLE_PRESET_KPIS.get(role, []) # 1. 我的KPI（responsible_user匹配用户名或姓名）+ 角色预设（均按企业隔离） assigned_kpis = db.query(KPIDefinition).filter( or_( KPIDefinition.responsible_user == username, KPIDefinition.responsible_user == name, ), KPIDefinition.status == "active", KPIDefinition.entity_id == entity_id, ).all() assigned_ids = {k.id for k in assigned_kpis} # 补充角色预设KPI（去重，按企业隔离） preset_kpis = [] if preset_codes: q = db.query(KPIDefinition).filter( KPIDefinition.kpi_code.in_(preset_codes), KPIDefinition.status == "active", KPIDefinition.entity_id == entity_id, ) if assigned_ids: q = q.filter(~KPIDefinition.id.in_(assigned_ids)) preset_kpis = q.all() all_kpis = assigned_kpis + preset_kpis kpi_list = [] forecast_section = [] # 预估区（单列，不作卡片实测值） for k in all_kpis: latest_v = latest_kpi_value(db, k.id) forecast_value = None if include_forecast: fc_row = db.query(KPIValue).filter( KPIValue.kpi_id == k.id, forecast_only_clause(), ).order_by(KPIValue.period.desc(), KPIValue.id.desc()).first() if fc_row is not None: forecast_value = fc_row.actual_value forecast_section.append({ "kpi_id": k.id, "kpi_code": k.kpi_code, "kpi_name": k.kpi_name, "value": forecast_value, "unit": k.unit, "period": fc_row.period, "data_status": "estimated", "is_forecast": True, "period_type": "forecast", }) actual = latest_v.actual_value if latest_v else None target = kpi_target_by_frequency(k) level = "gray" if actual is not None and target: # 反向指标（越低越好）集合 = 单一权威源 app.api.kpis.REVERSE_INDICATORS # （2026-09-15 M2 准入2：删本文件原字面量副本，避免与能力评分引擎方向漂移） if k.kpi_code in REVERSE_INDICATORS: # 反向：实际≤目标=绿；实际≤目标*1.1=黄；否则红 level = "green" if actual <= target else ( "yellow" if actual <= target * 1.1 else "red") else: ratio = actual / target level = "green" if ratio >= 0.9 else ("yellow" if ratio >= 0.7 else "red") kpi_list.append({ "id": k.id, "kpi_code": k.kpi_code, "kpi_name": k.kpi_name, "dimension": k.dimension, "category": k.category, "target_value": target, "target_monthly": k.target_monthly, "target_quarterly": k.target_quarterly, "target_yearly": k.target_yearly, "frequency": k.frequency, "actual_value": actual, "unit": k.unit, "level": level, "period": latest_v.period if latest_v else None, "forecast_value": forecast_value, "is_forecast": is_forecast_row(latest_v), "period_type": "actual", }) # 2. 我的改善行动（assignee匹配；CEO/管理员看全部） from app.models import ActionPlan if current_user.role == "ceo": # CEO/管理员查看全部行动方案（2026-08-26修复: 原逻辑只按assignee过滤导致工作台显示空） # 2026-08-31 隔离补漏: CEO 看全部也须按 entity_id 过滤，防跨账套待办串数据 my_plans = db.query(ActionPlan).filter( ActionPlan.entity_id == entity_id, ).order_by(ActionPlan.updated_at.desc()).all() else: my_plans = db.query(ActionPlan).filter( or_( ActionPlan.assignee == username, ActionPlan.assignee == name, ), ActionPlan.entity_id == entity_id, ).order_by(ActionPlan.updated_at.desc()).all() plan_list = [] for p in my_plans: overdue = False if p.due_date and p.status not in ("completed", "cancelled"): overdue = p.due_date < datetime.now() kpi_name = "" kpi = db.query(KPIDefinition).filter(KPIDefinition.id == p.kpi_id).first() if kpi: kpi_name = kpi.kpi_name plan_list.append({ "id": p.id, "kpi_id": p.kpi_id, "kpi_name": kpi_name, "title": p.title, "assignee": p.assignee, "priority": p.priority, "status": p.status, "progress": p.progress or 0, "due_date": p.due_date.isoformat() if p.due_date else None, "overdue": overdue, "created_at": p.created_at.isoformat() if p.created_at else None, }) # 3. 待办提醒 reminders = [] # 逾期行动 for p in plan_list: if p["overdue"]: reminders.append({ "type": "overdue_plan", "severity": "danger", "message": f"你负责的「{p['title']}」已逾期", "related_id": p["id"], "related_type": "action_plan", }) # 红色预警KPI for k in kpi_list: if k["level"] == "red": reminders.append({ "type": "red_kpi", "severity": "danger", "message": f"你负责的KPI「{k['kpi_name']}」处于红色预警", "related_id": k["id"], "related_type": "kpi", }) # 黄色预警KPI for k in kpi_list: if k["level"] == "yellow": reminders.append({ "type": "yellow_kpi", "severity": "warning", "message": f"你负责的KPI「{k['kpi_name']}」处于黄色预警", "related_id": k["id"], "related_type": "kpi", }) resp = { "kpis": kpi_list, "action_plans": plan_list, "reminders": reminders, } if include_forecast: resp["forecast"] = forecast_section return resp