""" CMA 财务示范数据 — 供财务BOT分析使用 运行: (cd /root/cma-management/backend && venv/bin/python3 scripts/seed_finance_data.py) """ import sys, os sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from app.database import get_session_local from app.models import KPIDefinition, KPIValue, KPIAlert, ActionPlan from app.models.budget_plan import BudgetPlan from datetime import datetime import logging logging.basicConfig(level=logging.INFO, format='%(asctime)s [%(levelname)s] %(message)s') logger = logging.getLogger("seed_finance") def seed(): db = get_session_local()() try: # ── 1. KPI字典 ── if db.query(KPIDefinition).count() == 0: kpis_data = [ # code, name, dim, cat, unit, target, freq, src, # g_min, y_min, r_min, dept, user ("F_REVENUE", "营业收入", "finance", "revenue_growth", "万元", 5000, "monthly", "erp", ">=4500", ">=4000", "<4000", "销售部", "张经理"), ("F_GROSS_MARGIN","毛利率", "finance", "profitability", "%", 35, "monthly", "erp", ">=35", ">=30", "<30", "财务部", "李会计"), ("F_NET_PROFIT", "净利润", "finance", "profitability", "万元", 800, "monthly", "erp", ">=800", ">=600", "<600", "财务部", ""), ("F_OP_CFLOW", "经营性现金流", "finance", "cash_risk", "万元", 1000, "monthly", "erp", ">=1000", ">=500", "<500", "财务部", ""), ("F_COST_RATIO", "费用率", "finance", "cost_control", "%", 20, "monthly", "erp", "<=20", "<=25", ">25", "财务部", ""), ("F_AR_DAYS", "应收账款周转天数","finance","asset_efficiency", "天", 45, "monthly", "erp", "<=45", "<=60", ">60", "销售部", ""), ("C_SATISFACTION","客户满意度", "customer","customer_satisfaction","分",92, "monthly","manual",">=92",">=85","<85","",""), ("C_NEW_CLIENTS", "新客户数", "customer","customer_scale", "个", 10, "monthly","business",">=10",">=5","<5","销售部",""), ("P_DELIVERY", "交付及时率", "process", "delivery_quality", "%", 95, "monthly", "erp", ">=95", ">=90", "<90", "交付部", ""), ("P_BUG_RATE", "缺陷率", "process", "delivery_quality", "%", 2, "monthly", "erp", "<=2", "<=5", ">5", "", ""), ("L_TRAINING", "培训完成率", "learning","talent_pipeline", "%", 90, "monthly","manual",">=90",">=80","<80","",""), ("L_EMPLOYEE_SAT","员工满意度", "learning","employee_engagement","分", 85, "quarterly","manual",None,None,None,"",""), ] formula_map = { "F_REVENUE": "SUM(erp_sales.amount)", "F_GROSS_MARGIN": "(收入-成本)/收入*100", "F_NET_PROFIT": "收入-成本-费用-税金", "F_COST_RATIO": "期间费用/收入*100", "F_AR_DAYS": "360/应收账款周转率", "P_BUG_RATE": "缺陷数/总功能点*100", } kpis = [] for code, name, dim, cat, unit, target, freq, src, g, y, r, dept, user in kpis_data: k = KPIDefinition( kpi_code=code, kpi_name=name, dimension=dim, category=cat, unit=unit, target_value=target, frequency=freq, data_source_type=src, threshold_green=g, threshold_yellow=y, threshold_red=r, responsible_dept=dept, responsible_user=user, status="active", ) if code in formula_map: k.formula = formula_map[code] db.add(k) kpis.append(k) db.flush() logger.info(f"✅ KPI字典: {len(kpis)}条") else: logger.info("KPI字典已有数据，跳过") # ── 2. KPI实际值（1-6月趋势）── if db.query(KPIValue).count() == 0: kpi_map = {k.kpi_code: k.id for k in db.query(KPIDefinition).all()} values_data = { "F_REVENUE": [4200, 4500, 4700, 4900, 5100, 4800], "F_GROSS_MARGIN":[32, 33, 34, 35, 34, 31], "F_NET_PROFIT": [650, 700, 750, 800, 820, 700], "F_OP_CFLOW": [800, 900, 950, 1000, 1050, 850], "F_COST_RATIO": [22, 21, 20, 19, 20, 23], "F_AR_DAYS": [50, 48, 46, 45, 44, 52], "C_SATISFACTION":[88, 89, 90, 91, 92, 90], "C_NEW_CLIENTS": [7, 8, 9, 10, 11, 8], "P_DELIVERY": [92, 93, 94, 95, 96, 93], "P_BUG_RATE": [3.0, 2.5, 2.0, 1.8, 1.5, 2.2], "L_TRAINING": [85, 87, 88, 90, 91, 86], } months = ["2026-01","2026-02","2026-03","2026-04","2026-05","2026-06"] cnt = 0 for code, vals in values_data.items(): kid = kpi_map.get(code) if not kid: continue for m, v in zip(months, vals): db.add(KPIValue(kpi_id=kid, period=m, actual_value=v, source_type="manual", data_status="verified")) cnt += 1 db.flush() logger.info(f"✅ KPI实际值: {cnt}条") else: logger.info("KPI实际值已有数据，跳过") # ── 3. 预警 ── if db.query(KPIAlert).count() == 0: kpi_map = {k.kpi_code: k.id for k in db.query(KPIDefinition).all()} alerts_data = [ ("F_REVENUE", "yellow", "6月营收4800万，低于目标5000万（缺口4%）"), ("F_GROSS_MARGIN", "red", "6月毛利率31%，跌破30%预警线"), ("F_NET_PROFIT", "yellow", "6月净利润700万，低于目标800万（-12.5%）"), ("F_OP_CFLOW", "yellow", "经营性现金流850万，低于目标1000万"), ("F_COST_RATIO", "yellow", "费用率23%，超目标20%"), ("F_AR_DAYS", "yellow", "应收账款周转52天，超目标45天"), ("C_NEW_CLIENTS", "yellow", "6月新客户仅8家，低于目标10家"), ] for code, level, msg in alerts_data: db.add(KPIAlert( kpi_id=kpi_map.get(code), alert_level=level, alert_message=msg, status="pending", )) db.flush() logger.info(f"✅ 预警: {len(alerts_data)}条") else: logger.info("预警已有数据，跳过") # ── 4. 改善行动 ── if db.query(ActionPlan).count() == 0: kpi_map = {k.kpi_code: k.id for k in db.query(KPIDefinition).all()} actions_data = [ (kpi_map.get("F_GROSS_MARGIN"), "成本优化专项 - 降低采购成本5%", "重新谈判供应商合同，目标毛利率回升至35%", "采购部王经理", "high", "in_progress", 40, datetime(2026,8,31)), (kpi_map.get("F_REVENUE"), "Q3客户拓展计划", "新增3个大客户，目标月度营收5500万", "销售部张经理", "high", "pending", 0, datetime(2026,9,30)), (kpi_map.get("F_AR_DAYS"), "应收账款催收行动", "集中催收超60天应收款，目标周转天降至45天", "财务部李会计", "medium", "in_progress", 30, datetime(2026,7,31)), (kpi_map.get("P_DELIVERY"), "交付流程优化", "优化项目管理流程，交付及时率提升至95%+", "交付部赵主管", "medium", "pending", 10, datetime(2026,8,15)), ] for kid, title, desc, assignee, pri, status, prog, due in actions_data: db.add(ActionPlan( kpi_id=kid, title=title, description=desc, assignee=assignee, priority=pri, status=status, progress=prog, due_date=due, )) db.flush() logger.info(f"✅ 改善行动: {len(actions_data)}条") else: logger.info("改善行动已有数据，跳过") # ── 5. 预算数据 ── if db.query(BudgetPlan).count() == 0: kpi_map = {k.kpi_code: k.id for k in db.query(KPIDefinition).all()} # 每个财务KPI全年12个月预算 budget_map = { "F_REVENUE": [3800,4000,4200,4400,4600,4800,4800,4900,5000,5000,5100,5200], "F_GROSS_MARGIN":[33]*12, "F_NET_PROFIT": [600,620,650,680,700,720,720,750,780,800,820,850], "F_OP_CFLOW": [700,750,800,850,900,950,950,1000,1000,1050,1050,1100], "F_COST_RATIO": [22]*12, "F_AR_DAYS": [50,49,48,47,46,45,45,44,44,43,43,42], } cnt = 0 for code, vals in budget_map.items(): kid = kpi_map.get(code) if not kid: continue for month in range(1, 13): db.add(BudgetPlan( kpi_id=kid, period=f"2026-{month:02d}", budget_value=vals[month-1], budget_year=2026, budget_month=month, version="v1.0", status="active", )) cnt += 1 db.flush() logger.info(f"✅ 预算数据: {cnt}条") else: logger.info("预算数据已有数据，跳过") db.commit() logger.info("") logger.info("🎉 财务示范数据导入完成！") logger.info(f"📊 KPI指标: {db.query(KPIDefinition).count()}") logger.info(f"📊 KPI实际值: {db.query(KPIValue).count()}") logger.info(f"📊 预警: {db.query(KPIAlert).count()}") logger.info(f"📊 改善行动: {db.query(ActionPlan).count()}") logger.info(f"📊 预算: {db.query(BudgetPlan).count()}") except Exception as e: db.rollback() logger.error(f"❌ 导入失败: {e}") import traceback; traceback.print_exc() raise finally: db.close() if __name__ == "__main__": seed()