""" CMA管理报表中心 — 管理会计OS 非传统财务报表，聚焦管理决策分析 报表: 1. 管理利润表 — 收入→变动成本→边际贡献→固定成本→息税前利润 2. 预算执行报告 — 各KPI预算vs实际vs差异率 3. KPI趋势报告 — 选定KPI的历史趋势 4. 四维度绩效评分卡 — BSC健康度雷达图 """ from fastapi import APIRouter, Depends, Query, HTTPException import json from pydantic import BaseModel from sqlalchemy.orm import Session from sqlalchemy import func from typing import Optional from datetime import datetime, date from app.database import get_db from app.auth_middleware import require_role, require_auth from app.models import KPIDefinition, KPIValue, BudgetPlan, StrategicMap, KPIAlert, User, Subject, ActionPlan, OperationLog, ReportHistory from app.utils.deviation_engine import calc_period_deviation, calc_period_diff import logging logger = logging.getLogger("cma.reports") router = APIRouter(prefix="/api/cma/reports", tags=["管理报表"], dependencies=[Depends(require_role("ceo", "finance", "business"))], ) # ============================================================ # 报表1: 管理利润表 # ============================================================ @router.get("/profit-summary") def get_profit_summary( period: str = Query(None, description="格式 YYYY-MM"), db: Session = Depends(get_db), ): """管理利润表 — 收入→变动成本→边际贡献→固定成本→息税前利润""" if period is None: period = datetime.now().strftime("%Y-%m") # 从KPI数据中获取各利润要素 def get_val(code: str): kpi = db.query(KPIDefinition).filter(KPIDefinition.kpi_code == code).first() if not kpi: return None v = db.query(KPIValue).filter( KPIValue.kpi_id == kpi.id, KPIValue.period == period ).order_by(KPIValue.id.desc()).first() return v.actual_value if v else None revenue = get_val("F_REVENUE") gross_profit_rate = get_val("F_PROFIT_RATE") net_profit_rate = get_val("F_NET_PROFIT_RATE") cost_ratio = get_val("F_COST_RATIO") # 计算利润要素 # 营收已知，用毛利率算毛利，用成本率算成本 gross_profit = round(revenue * (gross_profit_rate / 100), 2) if revenue and gross_profit_rate else None total_cost = round(revenue * (cost_ratio / 100), 2) if revenue and cost_ratio else None net_profit = round(revenue * (net_profit_rate / 100), 2) if revenue and net_profit_rate else None # 边际贡献 ≈ 毛利（简化模型） contribution_margin = gross_profit # 固定成本 ≈ 总成本 - 变动成本（假设变动成本=营收*50%） variable_cost = round(revenue * 0.50, 2) if revenue else None fixed_cost = round(total_cost - variable_cost, 2) if total_cost and variable_cost else None # 找上期做环比 prev_year, prev_month = period.split("-") py, pm = int(prev_year), int(prev_month) pm -= 1 if pm <= 0: pm += 12 py -= 1 prev_period = f"{py}-{pm:02d}" def get_prev_val(code: str): kpi = db.query(KPIDefinition).filter(KPIDefinition.kpi_code == code).first() if not kpi: return None v = db.query(KPIValue).filter( KPIValue.kpi_id == kpi.id, KPIValue.period == prev_period ).order_by(KPIValue.id.desc()).first() return v.actual_value if v else None prev_revenue = get_prev_val("F_REVENUE") prev_gross_profit_rate = get_prev_val("F_PROFIT_RATE") prev_net_profit_rate = get_prev_val("F_NET_PROFIT_RATE") prev_cost_ratio = get_prev_val("F_COST_RATIO") prev_gross_profit = round(prev_revenue * (prev_gross_profit_rate / 100), 2) if prev_revenue and prev_gross_profit_rate else None prev_total_cost = round(prev_revenue * (prev_cost_ratio / 100), 2) if prev_revenue and prev_cost_ratio else None prev_net_profit = round(prev_revenue * (prev_net_profit_rate / 100), 2) if prev_revenue and prev_net_profit_rate else None prev_contribution_margin = prev_gross_profit prev_variable_cost = round(prev_revenue * 0.50, 2) if prev_revenue else None prev_fixed_cost = round(prev_total_cost - prev_variable_cost, 2) if prev_total_cost and prev_variable_cost else None def calc_chg(cur, prev): if cur is not None and prev is not None and prev != 0: return round((cur - prev) / prev * 100, 2) return None items = [ { "name": "营业收入", "value": revenue, "prev_value": prev_revenue, "change_rate": calc_chg(revenue, prev_revenue), "ratio": 100.0, }, { "name": "减：变动成本", "value": variable_cost, "prev_value": prev_variable_cost, "change_rate": calc_chg(variable_cost, prev_variable_cost), "ratio": round(variable_cost / revenue * 100, 2) if variable_cost and revenue else None, }, { "name": "＝ 边际贡献", "value": contribution_margin, "prev_value": prev_contribution_margin, "change_rate": calc_chg(contribution_margin, prev_contribution_margin), "ratio": round(contribution_margin / revenue * 100, 2) if contribution_margin and revenue else None, "is_subtotal": True, }, { "name": "减：固定成本", "value": fixed_cost, "prev_value": prev_fixed_cost, "change_rate": calc_chg(fixed_cost, prev_fixed_cost), "ratio": round(fixed_cost / revenue * 100, 2) if fixed_cost and revenue else None, }, { "name": "＝ 息税前利润", "value": net_profit, "prev_value": prev_net_profit, "change_rate": calc_chg(net_profit, prev_net_profit), "ratio": round(net_profit / revenue * 100, 2) if net_profit and revenue else None, "is_total": True, }, ] return { "period": period, "prev_period": prev_period, "items": items, } # ============================================================ # 报表2: 预算执行报告 # ============================================================ @router.get("/budget-execution") def get_budget_execution( period: str = Query(None, description="格式 YYYY-MM"), dimension: Optional[str] = Query(None), alert_level: Optional[str] = Query(None), db: Session = Depends(get_db), ): """预算执行报告 — 各KPI预算vs实际vs差异率""" if period is None: period = datetime.now().strftime("%Y-%m") query = db.query(KPIDefinition).filter(KPIDefinition.status == "active") if dimension: query = query.filter(KPIDefinition.dimension == dimension) kpis = query.order_by(KPIDefinition.dimension, KPIDefinition.kpi_code).all() items = [] summary = {"total": 0, "with_budget": 0, "over_budget": 0, "normal": 0, "under_budget": 0} for kpi in kpis: dev = calc_period_deviation(db, kpi.id, period) if dev.get("actual_value") is None and dev.get("budget_value") is None: continue # 跳过完全无数据的KPI summary["total"] += 1 if dev.get("deviation_rate") is not None: rate = dev["deviation_rate"] level = "red" if abs(rate) > 20 else "yellow" if abs(rate) > 10 else "normal" if level == "red": summary["over_budget"] += 1 if rate > 0 else 0 summary["under_budget"] += 1 if rate < 0 else 0 else: summary["normal"] += 1 else: level = "gray" summary["normal"] += 1 if dev.get("budget_value") is not None: summary["with_budget"] += 1 items.append({ "kpi_id": kpi.id, "kpi_code": kpi.kpi_code, "kpi_name": kpi.kpi_name, "dimension": kpi.dimension, "unit": kpi.unit, "actual_value": dev.get("actual_value"), "budget_value": dev.get("budget_value"), "deviation_amount": dev.get("deviation_amount"), "deviation_rate": dev.get("deviation_rate"), "is_over_budget": dev.get("is_over_budget"), "alert_level": level, }) # alert_level 过滤 if alert_level: items = [i for i in items if i["alert_level"] == alert_level] return {"period": period, "summary": summary, "items": items} # ============================================================ # 报表3: KPI趋势报告 # ============================================================ @router.get("/kpi-trends") def get_kpi_trends( kpi_id: Optional[int] = Query(None), dimension: Optional[str] = Query(None), months: int = Query(12, ge=3, le=36), db: Session = Depends(get_db), ): """KPI趋势报告 — 选定KPI的历史趋势线""" query = db.query(KPIDefinition).filter(KPIDefinition.status == "active") if kpi_id: query = query.filter(KPIDefinition.id == kpi_id) if dimension: query = query.filter(KPIDefinition.dimension == dimension) kpis = query.order_by(KPIDefinition.dimension, KPIDefinition.kpi_code).all() results = [] for kpi in kpis: values = db.query(KPIValue).filter( KPIValue.kpi_id == kpi.id ).order_by(KPIValue.period.desc()).limit(months).all() values.reverse() trend = [{"period": v.period, "value": v.actual_value} for v in values] vals = [v.actual_value for v in values if v.actual_value is not None] target = kpi.target_value avg_val = round(sum(vals) / len(vals), 2) if vals else None max_val = max(vals) if vals else None min_val = min(vals) if vals else None # 趋势方向 if len(vals) >= 2: first_half = sum(vals[:len(vals)//2]) / (len(vals)//2) second_half = sum(vals[len(vals)//2:]) / (len(vals) - len(vals)//2) trend_dir = "up" if second_half > first_half * 1.05 else "down" if second_half < first_half * 0.95 else "stable" else: trend_dir = "stable" results.append({ "kpi_id": kpi.id, "kpi_code": kpi.kpi_code, "kpi_name": kpi.kpi_name, "dimension": kpi.dimension, "unit": kpi.unit, "target_value": target, "trend": trend, "trend_dir": trend_dir, "avg": avg_val, "max": max_val, "min": min_val, }) return {"data": results} # ============================================================ # 报表4: 四维度绩效评分卡 # ============================================================ DIM_CONFIG = { "finance": {"name": "财务维度", "icon": "💰", "color": "#409eff"}, "customer": {"name": "客户维度", "icon": "🤝", "color": "#67c23a"}, "process": {"name": "内部流程", "icon": "⚙️", "color": "#e6a23c"}, "learning": {"name": "学习成长", "icon": "📚", "color": "#f56c6c"}, } @router.get("/bsc-scorecard") def get_bsc_scorecard( map_id: Optional[int] = Query(None), period: Optional[str] = Query(None), db: Session = Depends(get_db), ): """四维度绩效评分卡 — BSC健康度""" if period is None: period = datetime.now().strftime("%Y-%m") # 取最新的已发布地图 map_query = db.query(StrategicMap).filter(StrategicMap.status == "published") if map_id: map_query = map_query.filter(StrategicMap.id == map_id) sm = map_query.order_by(StrategicMap.updated_at.desc()).first() if not sm: # 没有已发布地图，按维度聚合KPI return _build_scorecard_from_kpis(db, period) # 从战略地图维度数据构建评分卡 dims = sm.dimensions if isinstance(dims, str): import json dims = json.loads(dims) dimensions = [] total_score = 0 dim_count = 0 for dim in dims: dim_key = dim.get("key", "") config = DIM_CONFIG.get(dim_key, {"name": dim.get("name", dim_key), "icon": "📊", "color": "#999"}) objectives = dim.get("objectives", []) obj_results = [] dim_total = 0 dim_valid = 0 for obj in objectives: kpi_codes = obj.get("kpis", []) kpi_scores = [] for code in kpi_codes: kpi = db.query(KPIDefinition).filter(KPIDefinition.kpi_code == code).first() if not kpi: continue v = db.query(KPIValue).filter( KPIValue.kpi_id == kpi.id, KPIValue.period == period ).order_by(KPIValue.id.desc()).first() if v and v.actual_value and kpi.target_value: ratio = v.actual_value / kpi.target_value score = min(round(ratio * 100, 1), 100) level = "green" if ratio >= 0.9 else "yellow" if ratio >= 0.7 else "red" kpi_scores.append({"code": code, "name": kpi.kpi_name, "actual": v.actual_value, "target": kpi.target_value, "score": score, "level": level}) dim_total += score dim_valid += 1 obj_results.append({ "name": obj.get("name", ""), "kpi_count": len(kpi_codes), "kpi_with_data": dim_valid, "kpis": kpi_scores, }) dim_score = round(dim_total / dim_valid, 1) if dim_valid > 0 else 0 dimensions.append({ "key": dim_key, "name": config["name"], "icon": config["icon"], "color": config["color"], "score": dim_score, "objectives": obj_results, }) total_score += dim_score dim_count += 1 overall = round(total_score / dim_count, 1) if dim_count > 0 else 0 return { "period": period, "map_id": sm.id, "map_title": sm.title, "overall_score": overall, "dimensions": dimensions, } def _build_scorecard_from_kpis(db: Session, period: str) -> dict: """没有战略地图时，直接按维度聚合KPI算分""" kpis = db.query(KPIDefinition).filter(KPIDefinition.status == "active").all() dims: dict = {} for kpi in kpis: dim = kpi.dimension or "other" if dim not in dims: dims[dim] = {"kpis": [], "total_score": 0, "valid": 0} v = db.query(KPIValue).filter( KPIValue.kpi_id == kpi.id, KPIValue.period == period ).order_by(KPIValue.id.desc()).first() score = None level = "gray" if v and v.actual_value and kpi.target_value: ratio = v.actual_value / kpi.target_value score = min(round(ratio * 100, 1), 100) level = "green" if ratio >= 0.9 else "yellow" if ratio >= 0.7 else "red" dims[dim]["total_score"] += score dims[dim]["valid"] += 1 dims[dim]["kpis"].append({ "code": kpi.kpi_code, "name": kpi.kpi_name, "actual": v.actual_value if v else None, "target": kpi.target_value, "score": score, "level": level, }) dimensions = [] total_score = 0 dim_count = 0 for key, data in dims.items(): config = DIM_CONFIG.get(key, {"name": key, "icon": "📊", "color": "#999"}) dim_score = round(data["total_score"] / data["valid"], 1) if data["valid"] > 0 else 0 dimensions.append({ "key": key, "name": config["name"], "icon": config["icon"], "color": config["color"], "score": dim_score, "objectives": [{"name": "全部KPI", "kpis": data["kpis"], "kpi_count": len(data["kpis"]), "kpi_with_data": data["valid"]}], }) total_score += dim_score dim_count += 1 return { "period": period, "map_id": None, "map_title": None, "overall_score": round(total_score / dim_count, 1) if dim_count > 0 else 0, "dimensions": dimensions, } # ============================================================ # 利润表: 新30号准则五板块结构 (2027) # ============================================================ # 科目编码 → 新30号准则板块映射（PRD第118-145行） NEW_STANDARD_MAP = { # 经营类 "6001": "operating", # 主营业务收入 "6051": "operating", # 其他业务收入 "6401": "operating", # 主营业务成本 "6402": "operating", # 其他业务成本 "6601": "operating", # 销售费用 "6602": "operating", # 管理费用 "660204": "operating_rd", # 研发费用（从管理费剥离） "6603": "operating_fx", # 经营汇兑损益 "6701": "operating", # 经营资产减值损失 # 投资类 "6011": "investing", # 利息收入（银行存款） "6111": "investing", # 投资收益 "611101": "investing", # 股权投资 "670101": "investing", # 投资类资产减值 # 筹资类 "660301": "financing", # 利息支出（借款） "660302": "financing_fx", # 筹资汇兑损益 # 所得税 "6801": "tax", # 所得税费用 # 终止经营 "6901": "discontinued", # 终止经营损益 } # 板块 → 展示信息 BLOCK_INFO = { "operating": { "name": "一、经营类损益", "short_name": "经营类", "items": [ {"code": "6001", "name": "营业收入", "sign": 1}, {"code": "6051", "name": "其他业务收入", "sign": 1}, {"code": "6401", "name": "减：营业成本", "sign": -1}, {"code": "6402", "name": "减：其他业务成本", "sign": -1}, {"code": "6601", "name": "减：销售费用", "sign": -1}, {"code": "6602", "name": "减：管理费用", "sign": -1}, {"code": "660204", "name": "减：研发费用", "sign": -1}, {"code": "6603", "name": "经营汇兑损益", "sign": 1}, {"code": "6701", "name": "减：经营资产减值损失", "sign": -1}, ], "result_key": "operating_profit", "result_name": "经营利润", }, "investing": { "name": "二、投资类损益", "short_name": "投资类", "items": [ {"code": "6011", "name": "利息收入", "sign": 1}, {"code": "6111", "name": "投资收益", "sign": 1}, {"code": "670101", "name": "减：投资类资产减值", "sign": -1}, ], "result_key": "investing_profit", "result_name": "投资净收益", }, "financing": { "name": "三、筹资类损益", "short_name": "筹资类", "items": [ {"code": "660301", "name": "减：利息支出", "sign": -1}, {"code": "660302", "name": "筹资汇兑损益", "sign": 1}, ], "result_key": "financing_profit", "result_name": "筹资费用净额", }, "tax": { "name": "四、所得税费用", "short_name": "所得税", "items": [ {"code": "6801", "name": "减：所得税费用", "sign": -1}, ], "result_key": "tax_profit", "result_name": "所得税费用", }, "discontinued": { "name": "五、终止经营损益", "short_name": "终止经营", "items": [ {"code": "6901", "name": "终止经营损益", "sign": 1}, ], "result_key": "discontinued_profit", "result_name": "终止经营损益", }, } def _get_subject_amount(db: Session, code: str, period: str) -> Optional[float]: """从 subjects + kpi_values 获取科目金额数据""" # 尝试从KPI数据获取（KPI编码与科目编码映射） kpi_code_map = { "6001": "F_REVENUE", "6051": "F_REVENUE_OTHER", "6401": "F_COST", "6402": "F_COST_OTHER", "6601": "F_SELLING_EXP", "6602": "F_ADMIN_EXP", "660204": "F_RD_EXP", "6603": "F_FINANCE_EXP", "6701": "F_IMPAIRMENT_LOSS", "6011": "F_INTEREST_INCOME", "6111": "F_INVEST_INCOME", "611101": "F_INVEST_INCOME", "660301": "F_INTEREST_EXP", "660302": "F_FX_LOSS", "6801": "F_TAX_EXP", "6901": "F_DISCONTINUED", } # 1. 优先从 kpi_values 取 if code in kpi_code_map: kpi_code = kpi_code_map[code] kpi = db.query(KPIDefinition).filter(KPIDefinition.kpi_code == kpi_code).first() if kpi: v = db.query(KPIValue).filter( KPIValue.kpi_id == kpi.id, KPIValue.period == period ).order_by(KPIValue.id.desc()).first() if v and v.actual_value is not None: return float(v.actual_value) # 2. 从 subjects + voucher_details 取（如果存在） try: from app.models import VoucherDetail result = db.query( func.sum(VoucherDetail.debit_amount - VoucherDetail.credit_amount) ).filter( VoucherDetail.subject_code == code, VoucherDetail.period == period ).scalar() if result is not None: return float(result) except Exception: pass return None @router.get("/profit-statement") def get_profit_statement( period: str = Query(None, description="格式 YYYY-MM"), format: str = Query("old", description="old/new/dual"), db: Session = Depends(get_db), ): """利润表 — 支持旧格式、新30号准则五板块格式、双列对比""" if period is None: period = datetime.now().strftime("%Y-%m") if format == "old": # 旧30号准则格式（保留兼容） return get_profit_summary(period=period, db=db) if format == "dual": # 双列对比：旧准则 vs 新准则 old_data = get_profit_summary(period=period, db=db) new_data = _build_new_format_profit(db, period) return { "period": period, "format": "dual", "title": f"利润表双列对比（{period}）", "old_format": old_data, "new_format": new_data, } # === 新30号准则：五板块结构 === return _build_new_format_profit(db, period) def _build_new_format_profit(db: Session, period: str) -> dict: """构建新30号准则五板块利润表""" blocks = [] total_net_profit = 0 all_items_have_data = True for block_key in ["operating", "investing", "financing", "tax", "discontinued"]: block_cfg = BLOCK_INFO[block_key] items = [] block_subtotal = 0 block_has_data = False for item_cfg in block_cfg["items"]: amount = _get_subject_amount(db, item_cfg["code"], period) if amount is not None: effective = amount * item_cfg["sign"] block_subtotal += effective block_has_data = True items.append({ "code": item_cfg["code"], "name": item_cfg["name"], "amount": round(amount, 2) if amount is not None else None, "sign": item_cfg["sign"], "effective": round(amount * item_cfg["sign"], 2) if amount is not None else None, }) # Fallback: 使用PRD示例数据 if not block_has_data: all_items_have_data = False block_subtotal = _get_demo_block_total(block_key) block_result = { "key": block_key, "name": block_cfg["name"], "short_name": block_cfg["short_name"], "subtotal": round(block_subtotal, 2), "subtotal_name": block_cfg["result_name"], "items": items, "expanded": True, "has_real_data": block_has_data, } blocks.append(block_result) total_net_profit += block_subtotal # 合计行：净利润 = 一二三+四+五 return { "period": period, "format": "new", "title": f"利润表 — 新30号准则（{period}）", "blocks": blocks, "net_profit": round(total_net_profit, 2), "net_profit_name": "净利润", "all_items_have_data": all_items_have_data, "prev_period": None, # TODO: P1追溯调整 } # ============================================================ # MPM管理层指标计算器 (P2) # ============================================================ # MPM指标类型定义 MPM_INDICATOR_TYPES = { "ebitda": { "name": "EBITDA", "description": "息税折旧摊销前利润", "base_label": "净利润", "default_adjustments": [ {"code": "tax", "name": "加：所得税费用", "sign": 1, "checked": True}, {"code": "interest", "name": "加：利息支出", "sign": 1, "checked": True}, {"code": "depreciation", "name": "加：折旧与摊销", "sign": 1, "checked": True}, {"code": "impairment", "name": "加：资产减值损失", "sign": 1, "checked": False}, ], }, "adjusted_net_profit": { "name": "调整后净利润", "description": "剔除非经常性项目后的可持续净利润", "base_label": "净利润（准则）", "default_adjustments": [ {"code": "impairment", "name": "加：资产减值损失", "sign": 1, "checked": True}, {"code": "equity_incentive", "name": "加：股权激励费用", "sign": 1, "checked": True}, {"code": "ma_cost", "name": "加：并购相关费用", "sign": 1, "checked": False}, {"code": "nonrecurring_income", "name": "减：非经常性投资收益", "sign": -1, "checked": True}, {"code": "asset_disposal", "name": "减：资产处置收益", "sign": -1, "checked": False}, {"code": "government_grant", "name": "减：政府补助", "sign": -1, "checked": False}, ], }, "free_cash_flow": { "name": "自由现金流", "description": "经营现金流扣除资本支出后的可自由支配现金流", "base_label": "经营现金流", "default_adjustments": [ {"code": "capex", "name": "减：资本支出", "sign": -1, "checked": True}, {"code": "working_capital", "name": "减：营运资本增加", "sign": -1, "checked": True}, {"code": "maintenance_capex", "name": "减：维护性资本支出", "sign": -1, "checked": False}, {"code": "dividend", "name": "加：股息收入", "sign": 1, "checked": False}, ], }, "custom": { "name": "自定义指标", "description": "自定义管理层指标", "base_label": "净利润", "default_adjustments": [ {"code": "adjustment_1", "name": "调整项目1", "sign": 1, "checked": False, "amount": None}, {"code": "adjustment_2", "name": "调整项目2", "sign": -1, "checked": False, "amount": None}, {"code": "adjustment_3", "name": "调整项目3", "sign": 1, "checked": False, "amount": None}, ], }, } # 调整项默认金额（从利润表自动取值映射） ADJUSTMENT_VALUE_MAP = { "tax": {"code": "6801", "sign_inverse": True}, # 所得税费用，加回需取绝对值 "interest": {"code": "660301", "sign_inverse": True}, "impairment": {"code": "6701", "sign_inverse": False}, # 资产减值损失本身是费用 "equity_incentive": None, # 无映射，需用户输入 "ma_cost": None, "nonrecurring_income": {"code": "6111", "sign_inverse": False}, "asset_disposal": None, "government_grant": None, "capex": None, "working_capital": None, "maintenance_capex": None, "dividend": None, } class MpmCalculateRequest(BaseModel): indicator_type: str = "adjusted_net_profit" period: str = None adjustments: list[dict] = None # [{code, name, sign, checked, amount}] @router.post("/mpm-calculate") def mpm_calculate( req: MpmCalculateRequest, db: Session = Depends(get_db), ): """MPM管理层指标计算器 — 生成合规调节表""" if req.period is None: req.period = datetime.now().strftime("%Y-%m") indicator_cfg = MPM_INDICATOR_TYPES.get(req.indicator_type) if not indicator_cfg: raise HTTPException(status_code=400, detail=f"不支持的指标类型: {req.indicator_type}") # 获取基准值：净利润 net_profit = _calc_new_net_profit(db, req.period) if net_profit is None: net_profit = 0 # 经营现金流（自由现金流的基准） operating_cf = _get_kpi_val(db, "F_OPERATING_CF", req.period) # 确定基准值 if req.indicator_type == "free_cash_flow": base_value = operating_cf or net_profit # fallback base_label = "经营现金流" else: base_value = net_profit base_label = indicator_cfg["base_label"] # 获取调整项（来自请求或默认） adjustments = req.adjustments if req.adjustments else indicator_cfg["default_adjustments"] # 自动填充调整项金额 reconciliation_items = [] running_total = base_value # 第一步：基准值 reconciliation_items.append({ "step": 0, "code": "_base", "name": base_label, "sign": 1, "amount": round(base_value, 2), "effective": round(base_value, 2), "is_base": True, "running_total": round(base_value, 2), }) for adj in adjustments: code = adj.get("code", "") checked = adj.get("checked", False) sign = adj.get("sign", 1) name = adj.get("name", "") amount = adj.get("amount") # 尝试自动取值 if amount is None and checked: amount = _get_adjustment_value(db, code, req.period) effective = round(amount * sign, 2) if amount is not None else None item = { "step": len(reconciliation_items), "code": code, "name": name, "sign": sign, "amount": round(amount, 2) if amount is not None else None, "effective": effective, "checked": checked, "is_base": False, "running_total": None, } if checked and effective is not None: running_total += effective item["running_total"] = round(running_total, 2) reconciliation_items.append(item) # 最终结果 final_value = round(running_total, 2) return { "indicator_type": req.indicator_type, "indicator_name": indicator_cfg["name"], "indicator_desc": indicator_cfg["description"], "period": req.period, "base_value": round(base_value, 2), "base_label": base_label, "final_value": final_value, "final_label": indicator_cfg["name"], "adjustment_count": sum(1 for a in adjustments if a.get("checked", False)), "total_adjustments": len(adjustments), "reconciliation_items": reconciliation_items, "has_real_data": net_profit != 0, } def _get_kpi_val(db: Session, code: str, period: str) -> Optional[float]: """从KPI定义+值获取数值""" kpi = db.query(KPIDefinition).filter(KPIDefinition.kpi_code == code).first() if not kpi: return None v = db.query(KPIValue).filter( KPIValue.kpi_id == kpi.id, KPIValue.period == period ).order_by(KPIValue.id.desc()).first() return float(v.actual_value) if v and v.actual_value is not None else None def _calc_new_net_profit(db: Session, period: str) -> Optional[float]: """计算新30号准则下的净利润""" total = 0 has_data = False for block_key in ["operating", "investing", "financing", "tax", "discontinued"]: block_cfg = BLOCK_INFO[block_key] for item_cfg in block_cfg["items"]: amount = _get_subject_amount(db, item_cfg["code"], period) if amount is not None: total += amount * item_cfg["sign"] has_data = True if not has_data: return None return round(total, 2) def _get_adjustment_value(db: Session, adj_code: str, period: str) -> Optional[float]: """获取调整项的自动取值""" mapping = ADJUSTMENT_VALUE_MAP.get(adj_code) if mapping is None: return None # 需要用户输入 code = mapping["code"] amount = _get_subject_amount(db, code, period) if amount is None: return None # sign_inverse: 如果调整项是"加回"费用，费用本身的sign是负的（在利润表中是减项） # 但在MPM调节中，加回费用取绝对值 return abs(amount) if mapping.get("sign_inverse", False) else amount def _get_demo_block_total(block_key: str) -> float: """PRD示例数据 fallback""" demo = { "operating": -567883, "investing": 123456, "financing": -98765, "tax": -43210, "discontinued": 0, } return demo.get(block_key, 0) # ============================================================ # 追溯调整: 2026年数据按新30号准则重述 (P1) # ============================================================ @router.get("/restatement") def get_restatement( period: str = Query(None, description="格式 YYYY-MM"), db: Session = Depends(get_db), ): """2026年数据按新准则重述 — 旧口径vs新口径双列对比，自动标记调整项""" if period is None: period = datetime.now().strftime("%Y-%m") # 旧口径数据 (传统利润表项目) def _old_kpi_val(code: str): kpi = db.query(KPIDefinition).filter(KPIDefinition.kpi_code == code).first() if not kpi: return None v = db.query(KPIValue).filter( KPIValue.kpi_id == kpi.id, KPIValue.period == period ).order_by(KPIValue.id.desc()).first() return float(v.actual_value) if v and v.actual_value is not None else None old_revenue = _old_kpi_val("F_REVENUE") old_revenue_other = _old_kpi_val("F_REVENUE_OTHER") old_cost = _old_kpi_val("F_COST") old_cost_other = _old_kpi_val("F_COST_OTHER") old_selling_exp = _old_kpi_val("F_SELLING_EXP") old_admin_exp = _old_kpi_val("F_ADMIN_EXP") old_finance_exp = _old_kpi_val("F_FINANCE_EXP") old_rd_exp = _old_kpi_val("F_RD_EXP") # 新口径数据 (从科目映射或kpi_values获取) new_revenue = _get_subject_amount(db, "6001", period) new_revenue_other = _get_subject_amount(db, "6051", period) new_cost = _get_subject_amount(db, "6401", period) new_cost_other = _get_subject_amount(db, "6402", period) new_selling = _get_subject_amount(db, "6601", period) new_admin = _get_subject_amount(db, "6602", period) new_rd = _get_subject_amount(db, "660204", period) new_interest_income = _get_subject_amount(db, "6011", period) new_interest_exp = _get_subject_amount(db, "660301", period) new_fx = _get_subject_amount(db, "6603", period) new_fx_financing = _get_subject_amount(db, "660302", period) new_invest_income = _get_subject_amount(db, "6111", period) new_impairment = _get_subject_amount(db, "6701", period) new_invest_impairment = _get_subject_amount(db, "670101", period) new_tax = _get_subject_amount(db, "6801", period) new_discontinued = _get_subject_amount(db, "6901", period) # 旧口径汇总计算 old_operating_items = [ {"name": "营业收入", "old_value": old_revenue, "category": "operating"}, {"name": "其他业务收入", "old_value": old_revenue_other, "category": "operating"}, {"name": "减：营业成本", "old_value": old_cost, "category": "operating"}, {"name": "减：其他业务成本", "old_value": old_cost_other, "category": "operating"}, {"name": "减：销售费用", "old_value": old_selling_exp, "category": "operating"}, {"name": "减：管理费用", "old_value": old_admin_exp, "category": "operating"}, {"name": "减：财务费用", "old_value": old_finance_exp, "category": "financing"}, ] # 新口径项目明细 new_items_map = [ # 经营类 {"name": "营业收入", "new_value": new_revenue, "category": "operating"}, {"name": "其他业务收入", "new_value": new_revenue_other, "category": "operating"}, {"name": "减：营业成本", "new_value": new_cost, "category": "operating"}, {"name": "减：其他业务成本", "new_value": new_cost_other, "category": "operating"}, {"name": "减：销售费用", "new_value": new_selling, "category": "operating"}, {"name": "减：管理费用（不含研发）", "new_value": new_admin, "category": "operating"}, {"name": "减：研发费用", "new_value": new_rd, "category": "operating"}, {"name": "减：经营资产减值损失", "new_value": new_impairment, "category": "operating"}, {"name": "经营汇兑损益", "new_value": new_fx, "category": "operating"}, # 投资类 {"name": "利息收入", "new_value": new_interest_income, "category": "investing"}, {"name": "投资收益", "new_value": new_invest_income, "category": "investing"}, {"name": "减：投资类资产减值", "new_value": new_invest_impairment, "category": "investing"}, # 筹资类 {"name": "减：利息支出", "new_value": new_interest_exp, "category": "financing"}, {"name": "筹资汇兑损益", "new_value": new_fx_financing, "category": "financing"}, # 所得税 {"name": "减：所得税费用", "new_value": new_tax, "category": "tax"}, # 终止经营 {"name": "终止经营损益", "new_value": new_discontinued, "category": "discontinued"}, ] # 构建对比行 items = [] adjusted_count = 0 new_count = 0 unchanged_count = 0 # 先处理旧口径中存在的项目，匹配新口径 old_new_mapping = { "营业收入": "营业收入", "其他业务收入": "其他业务收入", "减：营业成本": "减：营业成本", "减：其他业务成本": "减：其他业务成本", "减：销售费用": "减：销售费用", "减：管理费用": "减：管理费用（不含研发）", } for old_item in old_operating_items: name = old_item["name"] old_val = old_item["old_value"] new_name = old_new_mapping.get(name, name) new_item = next((n for n in new_items_map if n["name"] == new_name), None) new_val = new_item["new_value"] if new_item else None # 管理费用：旧口径含研发，新口径不含 → 自动调整 if name == "减：管理费用": # 旧管理费 - 旧研发费 = 新管理费（不含研发） computed_new = old_val if old_admin_exp is not None and old_rd_exp is not None: # 如果新口径取不到值，用旧口径推算 if new_val is None: new_val = old_admin_exp - old_rd_exp if old_rd_exp else old_admin_exp diff = round(new_val - old_val, 2) if old_val is not None and new_val is not None else None needs_adj = diff is not None and abs(diff) > 0.01 if needs_adj: adjusted_count += 1 items.append({ "item_name": name, "old_value": old_val, "new_value": new_val, "difference": diff, "needs_adjustment": needs_adj, "adjustment_reason": "研发费用剥离" if needs_adj else None, "category": old_item["category"], }) # 自动带上研发费用行 rd_old = None # 旧口径无单独研发费用 rd_new = new_rd or old_rd_exp if rd_new is not None: adjusted_count += 1 items.append({ "item_name": "减：研发费用（单独列示）", "old_value": rd_old, "new_value": rd_new, "difference": rd_new if rd_new is not None else None, "needs_adjustment": rd_new is not None, "adjustment_reason": "新30号准则单独列示", "category": "operating", }) continue # 财务费用：旧口径一行汇总，新口径拆解为利息收入(投资类)+利息支出(筹资类) if name == "减：财务费用": fin_old = old_val fin_new_total = 0 fin_new_breakdown = [] # 利息支出（筹资类） int_exp_val = new_interest_exp if int_exp_val is not None: fin_new_total += -int_exp_val fin_new_breakdown.append({ "item_name": "减：利息支出（筹资类）", "old_value": None, "new_value": int_exp_val, "difference": None, "needs_adjustment": True, "adjustment_reason": "财务费用拆解", "category": "financing", }) adjusted_count += 1 # 筹资汇兑损益 fx_fin_val = new_fx_financing if fx_fin_val is not None: fin_new_total += fx_fin_val fin_new_breakdown.append({ "item_name": "筹资汇兑损益", "old_value": None, "new_value": fx_fin_val, "difference": None, "needs_adjustment": True, "adjustment_reason": "财务费用拆解", "category": "financing", }) adjusted_count += 1 # 利息收入（投资类） int_inc_val = new_interest_income if int_inc_val is not None: fin_new_breakdown.append({ "item_name": "利息收入（投资类）", "old_value": None, "new_value": int_inc_val, "difference": None, "needs_adjustment": True, "adjustment_reason": "财务费用拆解", "category": "investing", }) adjusted_count += 1 # 经营汇兑损益（经营类） fx_op_val = new_fx if fx_op_val is not None: fin_new_breakdown.append({ "item_name": "经营汇兑损益（经营类）", "old_value": None, "new_value": fx_op_val, "difference": None, "needs_adjustment": True, "adjustment_reason": "财务费用拆解", "category": "operating", }) adjusted_count += 1 items.append({ "item_name": "减：财务费用", "old_value": fin_old, "new_value": None, "difference": None, "needs_adjustment": True, "adjustment_reason": "财务费用拆解为投资类利息收入和筹资类利息支出", "category": "financing", "breakdown": fin_new_breakdown, }) # 把拆解项加到主列表 for brk in fin_new_breakdown: items.append(brk) continue # 普通项目直接对比 diff = round(new_val - old_val, 2) if old_val is not None and new_val is not None else None needs_adj = diff is not None and abs(diff) > 0.01 if needs_adj: adjusted_count += 1 else: unchanged_count += 1 items.append({ "item_name": name, "old_value": old_val, "new_value": new_val, "difference": diff, "needs_adjustment": needs_adj, "adjustment_reason": None, "category": old_item["category"], }) # 新增项目（只有新口径有） existing_names = [i["item_name"] for i in items] for new_item in new_items_map: if new_item["name"] not in existing_names and new_item["new_value"] is not None: items.append({ "item_name": new_item["name"], "old_value": None, "new_value": new_item["new_value"], "difference": None, "needs_adjustment": True, "adjustment_reason": "新30号准则新增项目", "category": new_item["category"], }) new_count += 1 adjusted_count += 1 # 计算旧口径净利润和新口径净利润 def _calc_net_profit_old(): """旧口径净利润 (简化)""" rev = old_revenue or 0 rev_other = old_revenue_other or 0 c = old_cost or 0 c_other = old_cost_other or 0 sell = old_selling_exp or 0 admin = old_admin_exp or 0 fin = old_finance_exp or 0 return rev + rev_other - c - c_other - sell - admin - fin def _calc_net_profit_new(): """新口径净利润 = 经营利润 + 投资净收益 + 筹资净费用 + 所得税 + 终止经营""" op_items = ["6001", "6051", "6401", "6402", "6601", "6602", "660204", "6603", "6701"] inv_items = ["6011", "6111", "670101"] fin_items = ["660301", "660302"] tax_items = ["6801"] dis_items = ["6901"] total = 0 for codes in [op_items, inv_items, fin_items, tax_items, dis_items]: for code in codes: amt = _get_subject_amount(db, code, period) if amt is not None: # 根据BLOCK_INFO中的sign处理 for bk in BLOCK_INFO.values(): for ic in bk["items"]: if ic["code"] == code: total += amt * ic["sign"] return round(total, 2) old_net = _calc_net_profit_old() new_net = _calc_net_profit_new() return { "period": period, "items": items, "summary": { "total_items": len(items), "adjusted_items": adjusted_count, "new_items": new_count, "unchanged_items": unchanged_count, }, "net_profit_comparison": { "old_net_profit": round(old_net, 2), "new_net_profit": new_net, "difference": round(new_net - old_net, 2), }, } @router.get("/category-map") def get_category_map( db: Session = Depends(get_db), ): """返回科目→新30号准则板块映射""" subjects_data = db.query(Subject).filter(Subject.is_active == 1).order_by(Subject.subject_code).all() map_list = [] for s in subjects_data: if s.new_standard_category: map_list.append({ "subject_code": s.subject_code, "subject_name": s.subject_name, "category": s.new_standard_category, }) # 如果没有数据库数据，返回硬编码映射 if not map_list: # 从 NEW_STANDARD_MAP 反向构造 all_subjects = db.query(Subject).filter(Subject.is_active == 1).all() subj_map = {s.subject_code: s.subject_name for s in all_subjects} for code, cat in NEW_STANDARD_MAP.items(): map_list.append({ "subject_code": code, "subject_name": subj_map.get(code, code), "category": cat, }) # 按板块分组 grouped = {"operating": [], "operating_rd": [], "operating_fx": [], "investing": [], "financing": [], "financing_fx": [], "tax": [], "discontinued": []} for m in map_list: cat = m["category"] if cat in grouped: grouped[cat].append(m) else: grouped.setdefault(cat, []).append(m) return { "mapping": NEW_STANDARD_MAP, "subjects": map_list, "grouped": grouped, "total": len(map_list), } @router.get("/dupont") def get_dupont_analysis( entity: str = Query("bohai"), db: Session = Depends(get_db), ): """杜邦分析 — ROE三级拆解 (CMA P2)""" if entity == "bohai": net_profit = 14.1 # 万 revenue = 383 # 万 total_assets = 533 # 万 equity = 114 # 万 net_profit_margin = round(net_profit / revenue * 100, 2) asset_turnover = round(revenue / total_assets, 4) financial_leverage = round(total_assets / equity, 2) roe = round(net_profit_margin / 100 * asset_turnover * financial_leverage * 100, 2) # 上期对比（模拟上一期数据） prev_roe = round(11.2, 2) roe_change = round(roe - prev_roe, 2) return { "entity": "bohai", "entity_name": "陕西博海科技（IT服务）", "period": "2026年H1", "roe": roe, "roe_change": roe_change, "roe_trend": "up" if roe_change > 0 else "down", "prev_roe": prev_roe, "factors": { "net_profit_margin": { "value": net_profit_margin, "label": "净利润率", "desc": "净利润/收入", "status": "🟡" if net_profit_margin < 5 else "✅", "assessment": "IT经销行业正常偏低", "raw": {"net_profit": net_profit, "revenue": revenue}, }, "asset_turnover": { "value": asset_turnover, "label": "资产周转率", "desc": "收入/总资产", "status": "🟡" if asset_turnover < 1 else "✅", "assessment": "资金效率中等", "raw": {"revenue": revenue, "total_assets": total_assets}, }, "financial_leverage": { "value": financial_leverage, "label": "财务杠杆", "desc": "总资产/净资产", "status": "🟡" if financial_leverage > 3 else "✅", "assessment": "负债率78.6%，偏高但可控", "raw": {"total_assets": total_assets, "equity": equity}, }, }, "raw_data": { "net_profit": net_profit, "revenue": revenue, "total_assets": total_assets, "equity": equity, }, "insight": { "improvement": "提高周转率或利润率，而非加杠杆", "detail": f"净利润率{net_profit_margin}%偏低，资产周转率{asset_turnover}x中等，财务杠杆{financial_leverage}x偏高。改善方向：提升毛利率或加快库存周转。", }, } return {"error": "不支持的实体"} # ============================================================ # 自动报告生成 — ChatBI优化P1 # 支持周报/月报/专项报告，定时/事件/手动触发 # ============================================================ WEEKDAY_CN = ["周一", "周二", "周三", "周四", "周五", "周六", "周日"] DIM_CN = {"finance": "财务维度", "customer": "客户维度", "process": "内部流程", "learning": "学习成长"} ALERT_LEVEL_CN = {"red": "🔴 紧急", "yellow": "🟡 预警", "green": "🟢 正常"} def _get_current_period(report_type: str) -> str: """根据报告类型自动计算当前期间""" now = datetime.now() if report_type == "weekly": iso = now.isocalendar() return f"{iso[0]}-W{iso[1]:02d}" elif report_type == "monthly": return now.strftime("%Y-%m") elif report_type == "special": return now.strftime("%Y-%m") return now.strftime("%Y-%m") def _calc_week_range(period: str) -> tuple: """周期间 → 起止日期""" import datetime as dt year, week = period.split("-W") year, week = int(year), int(week) # ISO week: week 1 is the week containing Jan 4 jan4 = dt.date(year, 1, 4) start_of_week1 = jan4 - dt.timedelta(days=jan4.isoweekday() - 1) monday = start_of_week1 + dt.timedelta(weeks=week - 1) sunday = monday + dt.timedelta(days=6) return monday.strftime("%Y-%m-%d"), sunday.strftime("%Y-%m-%d") def _get_month_period_prefix(period: str) -> str: """YYYY-MM 的前期""" y, m = period.split("-") y, m = int(y), int(m) m -= 1 if m <= 0: m += 12 y -= 1 return f"{y}-{m:02d}" def _fetch_kpi_data(db: Session) -> list: """获取所有活跃KPI的当前值、目标值、维度、预警""" kpis = db.query(KPIDefinition).filter(KPIDefinition.status == "active").all() result = [] for k in kpis: latest = db.query(KPIValue).filter( KPIValue.kpi_id == k.id, KPIValue.actual_value.isnot(None), ).order_by(KPIValue.period.desc()).first() alerts = db.query(KPIAlert).filter( KPIAlert.kpi_id == k.id, KPIAlert.status == "pending", ).order_by(KPIAlert.created_at.desc()).all() result.append({ "kpi_id": k.id, "kpi_code": k.kpi_code, "kpi_name": k.kpi_name, "dimension": k.dimension, "category": k.category, "unit": k.unit, "target_value": k.target_value, "current_value": latest.actual_value if latest else None, "current_period": latest.period if latest else None, "frequency": k.frequency, "alerts": [ {"level": a.alert_level, "message": a.alert_message} for a in alerts[:3] ], }) return result def _build_weekly_report(db: Session, period: str) -> dict: """生成周报""" kpis = _fetch_kpi_data(db) monday, sunday = _calc_week_range(period) now_str = datetime.now().strftime("%Y-%m-%d %H:%M") # 最近7天新增的预警 from datetime import timedelta seven_days_ago = datetime.now() - timedelta(days=7) recent_alerts = db.query(KPIAlert).filter( KPIAlert.created_at >= seven_days_ago, KPIAlert.status == "pending", ).order_by(KPIAlert.created_at.desc()).all() # 按维度分组统计 dim_stats = {} for k in kpis: d = k.get("dimension") or "other" if d not in dim_stats: dim_stats[d] = {"total": 0, "with_data": 0, "alert_count": 0} dim_stats[d]["total"] += 1 if k["current_value"] is not None: dim_stats[d]["with_data"] += 1 if k["alerts"]: dim_stats[d]["alert_count"] += len(k["alerts"]) # KPI变动（取有环比数据的） changes = [] for k in kpis: if k["current_value"] is None: continue prev_period = _get_month_period_prefix(k["current_period"]) if k["current_period"] else None if prev_period: prev_val = db.query(KPIValue).filter( KPIValue.kpi_id == k["kpi_id"], KPIValue.period == prev_period, ).first() if prev_val and prev_val.actual_value: diff = round(k["current_value"] - prev_val.actual_value, 2) rate = round(diff / prev_val.actual_value * 100, 2) if prev_val.actual_value != 0 else None changes.append({ **k, "prev_value": prev_val.actual_value, "change": diff, "change_rate": rate, }) changes.sort(key=lambda x: abs(x.get("change_rate") or 0), reverse=True) top_changes = changes[:8] # ── 构建 Markdown ── md_lines = [ f"📊 **经营分析周报**", f"📅 {monday} ~ {sunday} | 生成时间：{now_str}", "", "---", "", "## 一、本周概览", f"• 监控KPI：{len(kpis)} 个 | 有数据：{sum(1 for k in kpis if k['current_value'] is not None)} 个", f"• 待处理预警：{len(recent_alerts)} 条", ] # 按维度展示 for dim_key, dim_label in [("finance", "💰 财务"), ("customer", "🤝 客户"), ("process", "⚙️ 流程"), ("learning", "📚 学习成长")]: s = dim_stats.get(dim_key) if s: md_lines.append(f" - {dim_label}：{s['total']}个KPI | {s['with_data']}个有数据 | {s['alert_count']}条预警") md_lines.extend([ "", "## 二、关键KPI变动 TOP8", ]) for c in top_changes: direction = "📈" if (c.get("change_rate") or 0) > 0 else "📉" rate_str = f"{c['change_rate']:+.1f}%" if c.get("change_rate") is not None else "-" md_lines.append( f" {direction} **{c['kpi_name']}**：{c['current_value']}{c['unit']} " f"（上期{c.get('prev_value', '-')}，变动{rate_str}）" ) if recent_alerts: md_lines.extend([ "", "## 三、本周预警", ]) for a in recent_alerts[:10]: kpi = next((k for k in kpis if k["kpi_id"] == a.kpi_id), None) kpi_name = kpi["kpi_name"] if kpi else f"KPI#{a.kpi_id}" md_lines.append(f" {ALERT_LEVEL_CN.get(a.alert_level, '⚠️')} {kpi_name}：{a.alert_message}") md_lines.extend([ "", "## 四、改进行动", ]) actions = db.query(ActionPlan).filter( ActionPlan.status.in_(["pending", "in_progress"]), ).order_by(ActionPlan.created_at.desc()).limit(5).all() if actions: for a in actions: bar = "▓" * (a.progress // 10) + "░" * (10 - a.progress // 10) md_lines.append(f" • {bar} {a.title}（{a.progress}%）- {a.assignee or '未分配'}") else: md_lines.append(" （暂无进行中的改善行动）") md_lines.extend([ "", "---", f"💡 发送「分析报告」可重新生成", ]) markdown = "\n".join(md_lines) # ── 构建 JSON ── json_data = { "report_type": "weekly", "period": period, "date_range": {"start": monday, "end": sunday}, "generated_at": now_str, "overview": { "total_kpis": len(kpis), "kpis_with_data": sum(1 for k in kpis if k["current_value"] is not None), "pending_alerts": len(recent_alerts), }, "dimensions": {dk: { "label": DIM_CN.get(dk, dk), "kpi_count": ds["total"], "with_data": ds["with_data"], "alert_count": ds["alert_count"], } for dk, ds in dim_stats.items()}, "top_changes": [ { "kpi_code": c["kpi_code"], "kpi_name": c["kpi_name"], "current_value": c["current_value"], "prev_value": c.get("prev_value"), "change": c.get("change"), "change_rate": c.get("change_rate"), "unit": c["unit"], } for c in top_changes ], "alerts": [ { "kpi_id": a.kpi_id, "alert_level": a.alert_level, "alert_message": a.alert_message, } for a in recent_alerts[:10] ], } return {"markdown": markdown, "json": json_data, "title": f"经营分析周报 {monday}~{sunday}"} def _build_monthly_report(db: Session, period: str) -> dict: """生成月报""" kpis = _fetch_kpi_data(db) now_str = datetime.now().strftime("%Y-%m-%d %H:%M") prev_period = _get_month_period_prefix(period) # 预算执行数据 budget_items = [] for k in kpis: dev = calc_period_deviation(db, k["kpi_id"], period) if dev.get("actual_value") is not None or dev.get("budget_value") is not None: budget_items.append({ "kpi_name": k["kpi_name"], "kpi_code": k["kpi_code"], "dimension": k["dimension"], "actual": dev.get("actual_value"), "budget": dev.get("budget_value"), "deviation_rate": dev.get("deviation_rate"), "unit": k["unit"], }) # 同比/环比 comparisons = [] for k in kpis[:20]: if k["current_value"] is None: continue mom = calc_period_diff(db, k["kpi_id"], period, "mom") yoy = calc_period_diff(db, k["kpi_id"], period, "yoy") comparisons.append({ "kpi_name": k["kpi_name"], "kpi_code": k["kpi_code"], "current": k["current_value"], "unit": k["unit"], "mom_rate": mom.get("diff_rate"), "yoy_rate": yoy.get("diff_rate"), }) # 预警汇总 pending_alerts = db.query(KPIAlert).filter( KPIAlert.status == "pending", ).all() red_count = sum(1 for a in pending_alerts if a.alert_level == "red") yellow_count = sum(1 for a in pending_alerts if a.alert_level == "yellow") # 各维度达成情况 dim_summary = {} for k in kpis: d = k.get("dimension") or "other" if d not in dim_summary: dim_summary[d] = {"total": 0, "achieved": 0, "warning": 0, "failed": 0} dim_summary[d]["total"] += 1 if k["current_value"] is not None and k["target_value"]: ratio = k["current_value"] / k["target_value"] if ratio >= 0.9: dim_summary[d]["achieved"] += 1 elif ratio >= 0.7: dim_summary[d]["warning"] += 1 else: dim_summary[d]["failed"] += 1 # 改善行动 actions = db.query(ActionPlan).order_by(ActionPlan.created_at.desc()).limit(5).all() # ── 生成 Markdown ── md_lines = [ f"📊 **经营分析月报**", f"📅 {period} | 生成时间：{now_str}", "", "---", "", "## 一、月度总览", f"• 监控KPI：{len(kpis)} 个", f"• 预警状态：🔴 {red_count}条紧急 | 🟡 {yellow_count}条预警", "", "## 二、四维度达成情况", ] for dk in ["finance", "customer", "process", "learning"]: ds = dim_summary.get(dk) if ds: label = DIM_CN.get(dk, dk) total = ds["total"] achieved = ds["achieved"] pct = round(achieved / total * 100, 1) if total > 0 else 0 bar_len = 10 filled = int(pct / 10) bar = "▓" * filled + "░" * (bar_len - filled) md_lines.append(f"• {label}：{bar} {pct}%（{achieved}/{total}达标）") md_lines.extend([ "", "## 三、预算执行 TOP异常", ]) budget_with_dev = [b for b in budget_items if b.get("deviation_rate") is not None] budget_with_dev.sort(key=lambda x: abs(x["deviation_rate"]), reverse=True) for b in budget_with_dev[:8]: direction = "🔴" if (b["deviation_rate"] or 0) > 0 else "🟢" md_lines.append( f" {direction} **{b['kpi_name']}**：实际{b['actual']}{b['unit']} " f"vs 预算{b['budget']}{b['unit']}（差异率{b['deviation_rate']:+.1f}%）" ) md_lines.extend([ "", "## 四、同比/环比分析", ]) for c in comparisons[:8]: mom_str = f"环比{c.get('mom_rate'):+.1f}%" if c.get("mom_rate") is not None else "环比N/A" yoy_str = f"同比{c.get('yoy_rate'):+.1f}%" if c.get("yoy_rate") is not None else "同比N/A" md_lines.append(f" • **{c['kpi_name']}**：{c['current']}{c['unit']} | {mom_str} | {yoy_str}") md_lines.extend([ "", "## 五、改善行动进展", ]) if actions: for a in actions: bar = "▓" * (a.progress // 10) + "░" * (10 - a.progress // 10) status_cn = {"pending": "待开始", "in_progress": "进行中", "completed": "已完成"}.get(a.status, a.status) md_lines.append(f" • {bar} {a.title}（{a.progress}%）- {status_cn}") else: md_lines.append(" （暂无改善行动）") md_lines.extend([ "", "---", f"💡 发送「生成{period}经营报告」可重新生成", ]) markdown = "\n".join(md_lines) # ── 生成 JSON ── json_data = { "report_type": "monthly", "period": period, "generated_at": now_str, "overview": { "total_kpis": len(kpis), "red_alerts": red_count, "yellow_alerts": yellow_count, }, "dimensions": {dk: { "label": DIM_CN.get(dk, dk), "total": ds["total"], "achieved": ds["achieved"], "achievement_rate": round(ds["achieved"] / ds["total"] * 100, 1) if ds["total"] > 0 else 0, } for dk, ds in dim_summary.items()}, "budget_execution": [ { "kpi_code": b["kpi_code"], "kpi_name": b["kpi_name"], "actual": b.get("actual"), "budget": b.get("budget"), "deviation_rate": b.get("deviation_rate"), "unit": b["unit"], } for b in budget_with_dev[:15] ], "comparisons": [ { "kpi_code": c["kpi_code"], "kpi_name": c["kpi_name"], "current": c["current"], "mom_rate": c.get("mom_rate"), "yoy_rate": c.get("yoy_rate"), } for c in comparisons[:15] ], } return {"markdown": markdown, "json": json_data, "title": f"经营分析月报 {period}"} def _build_special_report(db: Session, period: str, alert_ref: str = None) -> dict: """生成专项分析报告 — 聚焦KPI异常""" kpis = _fetch_kpi_data(db) now_str = datetime.now().strftime("%Y-%m-%d %H:%M") # 按偏差率排序（当前值/目标值） scored = [] for k in kpis: if k["current_value"] is not None and k["target_value"] and k["target_value"] > 0: ratio = k["current_value"] / k["target_value"] deviation = round((ratio - 1) * 100, 2) scored.append({**k, "achievement_ratio": ratio, "deviation_pct": deviation}) scored.sort(key=lambda x: abs(x["deviation_pct"]), reverse=True) top_issues = scored[:10] worst_issues = [s for s in scored if s["deviation_pct"] < 0][:5] best_issues = [s for s in scored if s["deviation_pct"] > 0][:3] # 如果有预警引用，聚焦该预警关联的KPI focus_kpi_name = None if alert_ref: alert = db.query(KPIAlert).filter(KPIAlert.id == int(alert_ref)).first() if alert_ref.isdigit() else None if alert: target_kpi = next((k for k in kpis if k["kpi_id"] == alert.kpi_id), None) if target_kpi: focus_kpi_name = target_kpi["kpi_name"] # 维度分布 dim_issues = {} for s in scored: d = s.get("dimension") or "other" if d not in dim_issues: dim_issues[d] = {"on_track": 0, "at_risk": 0, "critical": 0} if s["achievement_ratio"] >= 0.9: dim_issues[d]["on_track"] += 1 elif s["achievement_ratio"] >= 0.7: dim_issues[d]["at_risk"] += 1 else: dim_issues[d]["critical"] += 1 # ── Markdown ── md_lines = [ f"📊 **经营分析专项报告**", f"📅 {period} | 生成时间：{now_str}", ] if focus_kpi_name: md_lines.append(f"🎯 触发事件：{focus_kpi_name} 异常预警") md_lines.extend([ "", "---", "", "## 一、风险总览", ]) for dk in ["finance", "customer", "process", "learning"]: d = dim_issues.get(dk) if d: label = DIM_CN.get(dk, dk) md_lines.append( f"• {label}：{d['on_track']}正常 / {d['at_risk']}预警 / {d['critical']}危险" ) md_lines.extend([ "", "## 二、风险KPI TOP 5（严重未达标）", ]) for w in worst_issues: md_lines.append( f" 🔴 **{w['kpi_name']}**：实际{w['current_value']}{w['unit']} " f"vs 目标{w['target_value']}{w['unit']}（达成率{w['achievement_ratio']*100:.1f}%）" ) for a in w.get("alerts", []): md_lines.append(f" ⚠️ {a['message']}") md_lines.extend([ "", "## 三、待处理预警详情", ]) pending_alerts = db.query(KPIAlert).filter( KPIAlert.status == "pending", ).order_by(KPIAlert.created_at.desc()).limit(10).all() if pending_alerts: for a in pending_alerts: target_kpi = next((k for k in kpis if k["kpi_id"] == a.kpi_id), None) name = target_kpi["kpi_name"] if target_kpi else f"KPI#{a.kpi_id}" md_lines.append(f" {ALERT_LEVEL_CN.get(a.alert_level, '⚠️')} {name}：{a.alert_message}") else: md_lines.append(" ✅ 无待处理预警") md_lines.extend([ "", "## 四、改善建议", ]) for w in worst_issues: if w["target_value"] and w["current_value"]: gap = round(w["target_value"] - w["current_value"], 2) md_lines.append(f" • **{w['kpi_name']}**：缺口{gap}{w['unit']}，需提升至{w['target_value']}{w['unit']}才能达标") md_lines.extend([ "", "## 五、表现优秀KPI", ]) for b in best_issues: md_lines.append(f" 🟢 **{b['kpi_name']}**：{b['current_value']}{b['unit']}，超目标{b['deviation_pct']:+.1f}%") md_lines.extend([ "", "---", "💡 如有疑问，请回复「分析详情」获取更细颗粒度的数据", ]) markdown = "\n".join(md_lines) # ── JSON ── json_data = { "report_type": "special", "period": period, "generated_at": now_str, "focus_kpi": focus_kpi_name, "alert_ref": alert_ref, "risk_summary": {dk: { "label": DIM_CN.get(dk, dk), "on_track": dim_issues.get(dk, {}).get("on_track", 0), "at_risk": dim_issues.get(dk, {}).get("at_risk", 0), "critical": dim_issues.get(dk, {}).get("critical", 0), } for dk in ["finance", "customer", "process", "learning"]}, "worst_kpis": [ { "kpi_code": w["kpi_code"], "kpi_name": w["kpi_name"], "current_value": w["current_value"], "target_value": w["target_value"], "achievement_ratio": round(w["achievement_ratio"], 4), "gap": round(w["target_value"] - w["current_value"], 2) if w["target_value"] and w["current_value"] else None, "unit": w["unit"], } for w in worst_issues ], "best_kpis": [ { "kpi_code": b["kpi_code"], "kpi_name": b["kpi_name"], "current_value": b["current_value"], "target_value": b["target_value"], "achievement_ratio": round(b["achievement_ratio"], 4), "unit": b["unit"], } for b in best_issues ], "alerts": [ { "kpi_id": a.kpi_id, "alert_level": a.alert_level, "alert_message": a.alert_message, } for a in pending_alerts[:10] ], } return {"markdown": markdown, "json": json_data, "title": f"经营分析专项报告 {period}"} # ============================================================ # POST /api/cma/reports/generate — 自动报告生成主入口 # ============================================================ class GenerateReportRequest(BaseModel): report_type: str = "monthly" # weekly / monthly / special period: Optional[str] = None # 自动计算 if None trigger_type: str = "manual" # manual / scheduled / event alert_ref: Optional[str] = None # 事件触发时的预警ID @router.post("/generate", response_model=None) def generate_report( req: GenerateReportRequest, db: Session = Depends(get_db), current_user=Depends(require_auth), ): """生成经营分析报告（周报/月报/专项），返回markdown+JSON 触发方式： - POST ?trigger_type=manual （用户主动触发） - POST ?trigger_type=scheduled （定时任务触发） - POST ?trigger_type=event&alert_ref=123 （KPI异常事件触发） """ # 校验报告类型 if req.report_type not in ("weekly", "monthly", "special"): raise HTTPException(400, f"不支持的报告类型: {req.report_type}，可选: weekly/monthly/special") # 确定期间 period = req.period or _get_current_period(req.report_type) # 生成报告 builders = { "weekly": lambda db, period: _build_weekly_report(db, period), "monthly": lambda db, period: _build_monthly_report(db, period), "special": lambda db, period: _build_special_report(db, period, alert_ref=req.alert_ref), } builder = builders[req.report_type] try: report_data = builder(db, period) except Exception as e: logger.error(f"报告生成异常: {e}", exc_info=True) raise HTTPException(500, f"报告生成失败: {str(e)}") # 保存到数据库 record = ReportHistory( report_type=req.report_type, period=period, title=report_data["title"], markdown_content=report_data["markdown"], json_content=report_data["json"], status="generated", trigger_type=req.trigger_type, alert_ref=req.alert_ref, ) db.add(record) db.flush() # 记录操作日志 log = OperationLog( action="generate_report", target_type="report", target_id=record.id, detail=json.dumps({ "report_type": req.report_type, "period": period, "trigger_type": req.trigger_type, }, ensure_ascii=False), ) db.add(log) db.commit() db.refresh(record) return { "id": record.id, "report_type": req.report_type, "period": period, "title": report_data["title"], "generated_at": record.created_at.strftime("%Y-%m-%d %H:%M:%S") if record.created_at else datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "markdown": report_data["markdown"], "json": report_data["json"], } @router.get("/history") def list_report_history( report_type: Optional[str] = Query(None), limit: int = Query(20, ge=1, le=100), db: Session = Depends(get_db), current_user=Depends(require_auth), ): """查看报告生成历史""" query = db.query(ReportHistory).order_by(ReportHistory.created_at.desc()) if report_type: query = query.filter(ReportHistory.report_type == report_type) records = query.limit(limit).all() return { "total": len(records), "data": [ { "id": r.id, "report_type": r.report_type, "period": r.period, "title": r.title, "status": r.status, "trigger_type": r.trigger_type, "created_at": r.created_at.strftime("%Y-%m-%d %H:%M:%S") if r.created_at else None, } for r in records ], } @router.get("/history/{report_id}") def get_report_detail( report_id: int, db: Session = Depends(get_db), current_user=Depends(require_auth), ): """获取单条报告详情（含完整markdown内容）""" r = db.query(ReportHistory).filter(ReportHistory.id == report_id).first() if not r: raise HTTPException(404, "报告不存在") return { "id": r.id, "report_type": r.report_type, "period": r.period, "title": r.title, "status": r.status, "trigger_type": r.trigger_type, "alert_ref": r.alert_ref, "markdown": r.markdown_content, "json": r.json_content, "created_at": r.created_at.strftime("%Y-%m-%d %H:%M:%S") if r.created_at else None, }