fix: 移除ERP密码和API密钥硬编码，改为环境变量；erp-sync接口加鉴权

2026-06-05 07:51:20 +08:00
parent 2c92168e0e
commit 7b9ef8ac27
9 changed files with 844 additions and 8 deletions
@@ -0,0 +1,234 @@
+"""Step 4: ERP数据同步执行器
+直接连接ERP SQL Server，执行KPI-SQL并写入 kpi_values
+运行: python3 scripts/erp_data_sync.py [period] [--dry-run]
+"""
+
+import sys, os, json, logging, re, argparse
+from datetime import datetime
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+from app.database import get_session_local
+from sqlalchemy import text, create_engine
+
+logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
+logger = logging.getLogger("erp_data_sync")
+
+# ── ERP SQL Server 直连配置 ──
+ERP_DB_HOST = "211.149.143.215"
+ERP_DB_PORT = 1433
+ERP_DB_NAME = "SUBzxbtest"
+ERP_DB_USER = "zxbtest"
+ERP_DB_PASS = os.environ.get("ERP_DB_PASS")
+
+# ── KPI-SQL 映射（从 generate_kpi_sql.py 复制核心映射） ──
+KPI_SQL_MAP = {
+    "F_REVENUE_001": {
+        "sql": """SELECT COALESCE(SUM(SumMoney), 0) as value 
+FROM MasterBill 
+WHERE BillType=1 AND BillState>=3 AND Period=:period"""
+    },
+    "F_PROFIT_001": {
+        "sql": """SELECT 
+    CASE WHEN SUM(SumMoney) > 0 
+        THEN ROUND((SUM(SumMoney) - COALESCE(SUM(SumCostMoney),0)) / SUM(SumMoney) * 100, 2) 
+        ELSE 0 END as value
+FROM MasterBill 
+WHERE BillType=1 AND BillState>=3 AND Period=:period"""
+    },
+    "C_CUST_001": {
+        "sql": """SELECT COUNT(DISTINCT Unit_ID) as value
+FROM MasterBill 
+WHERE BillType=1 AND BillState>=3 AND Period=:period"""
+    },
+    "C_CUST_002": {
+        "sql": """SELECT 
+    CASE WHEN total_sales > 0 
+        THEN ROUND(top5_sales / total_sales * 100, 2) 
+        ELSE 0 END as value
+FROM (
+    SELECT 
+        SUM(CASE WHEN rn <= 5 THEN SumMoney ELSE 0 END) as top5_sales,
+        SUM(SumMoney) as total_sales
+    FROM (
+        SELECT SumMoney, 
+            ROW_NUMBER() OVER (ORDER BY SumMoney DESC) as rn
+        FROM (
+            SELECT SUM(SumMoney) as SumMoney
+            FROM MasterBill 
+            WHERE BillType=1 AND BillState>=3 AND Period=:period
+            GROUP BY Unit_ID
+        ) t
+    ) t2
+) t3"""
+    },
+    "F_AR_002": {
+        "sql": """SELECT 
+    CASE WHEN total_receivable > 0
+        THEN ROUND(overdue_receivable / total_receivable * 100, 2)
+        ELSE 0 END as value
+FROM (
+    SELECT 
+        SUM(CASE WHEN BillType=1 THEN SumMoney ELSE 0 END) as total_receivable,
+        SUM(CASE WHEN BillType=1 AND DATEDIFF(day, BillDate, GETDATE()) > 30 THEN SumMoney ELSE 0 END) as overdue_receivable
+    FROM MasterBill
+    WHERE Period<=:period AND BillState>=3
+) t""",
+    },
+    # ── 杜邦分析 ──
+    "F_ASSET_TOTAL": {
+        "sql": """SELECT 
+    COALESCE(
+        (SELECT SUM(CAST(Act_Tot AS FLOAT)) FROM BalanceInfo WHERE Act_ID=4 AND Period=:period)
+        + (SELECT SUM(CAST(Act_Tot AS FLOAT)) FROM BalanceInfo WHERE Act_ID=5 AND Period=:period)
+    , 0) as value"""
+    },
+    "F_EQUITY_TOTAL": {
+        "sql": """SELECT 
+    COALESCE(
+        (SELECT SUM(CAST(Act_Tot AS FLOAT)) FROM BalanceInfo WHERE Act_ID=3 AND Period=:period)
+        - (SELECT SUM(CAST(Act_Tot AS FLOAT)) FROM BalanceInfo WHERE Act_ID=2 AND Period=:period)
+    , 0) as value"""
+    },
+}
+
+
+def get_erp_engine():
+    """创建ERP直连引擎"""
+    conn_str = f"mssql+pymssql://{ERP_DB_USER}:{ERP_DB_PASS}@{ERP_DB_HOST}:{ERP_DB_PORT}/{ERP_DB_NAME}"
+    return create_engine(conn_str, pool_size=2, max_overflow=5, pool_pre_ping=True)
+
+
+def get_all_kpis(db) -> list:
+    """获取所有标记了erp的KPI定义"""
+    rows = db.execute(text("""
+        SELECT id, kpi_code, kpi_name, formula
+        FROM kpi_definitions 
+        WHERE data_source_type = 'erp'
+        ORDER BY id
+    """)).fetchall()
+    return [dict(r._mapping) for r in rows]
+
+
+def get_periods_to_sync(db) -> list:
+    """确定需要同步的期间（最近12个月）"""
+    rows = db.execute(text("""
+        SELECT DISTINCT period FROM kpi_values 
+        WHERE source_type = 'erp'
+        ORDER BY period DESC
+    """)).fetchall()
+    existing = set(r[0] for r in rows)
+    
+    # 生成最近12个月的期间
+    periods = []
+    now = datetime.now()
+    for i in range(12):
+        m = now.month - i
+        y = now.year
+        if m <= 0:
+            m += 12
+            y -= 1
+        period = f"{y}-{m:02d}"
+        periods.append(period)
+    
+    # 只同步已有期间中没有数据或需要更新的
+    # ERP中 Period 列: 1=1月, 2=2月 ... 12=12月（年度期间）
+    return periods
+
+
+def erp_period_to_int(period: str) -> str:
+    """将 2026-05 转为ERP的期间数字"""
+    return period.split("-")[1]  # "05"
+
+
+def execute_erp_sql(erp_engine, sql: str, period: str) -> float:
+    """在ERP SQL Server上执行SQL"""
+    period_month = period.split("-")[1]
+    period_year = period.split("-")[0]
+    
+    # 替换参数
+    exec_sql = sql.replace(":period", period_month)
+    exec_sql = re.sub(r"GETDATE\(\)", f"'{datetime.now().strftime('%Y-%m-%d')}'", exec_sql)
+    
+    with erp_engine.connect() as conn:
+        result = conn.execute(text(exec_sql))
+        row = result.fetchone()
+        return float(row[0]) if row and row[0] is not None else 0.0
+
+
+def main():
+    parser = argparse.ArgumentParser(description="ERP数据同步")
+    parser.add_argument("period", nargs="?", default=None, help="期间，如 2026-05")
+    parser.add_argument("--dry-run", action="store_true", help="试运行，不写入数据库")
+    args = parser.parse_args()
+    
+    db = get_session_local()()
+    
+    # 获取KPI
+    kpis = get_all_kpis(db)
+    logger.info(f"待同步KPI: {len(kpis)} 个")
+    for k in kpis:
+        has_sql = "✅" if k["kpi_code"] in KPI_SQL_MAP else "❌"
+        logger.info(f"  {has_sql} [{k['kpi_code']}] {k['kpi_name']}")
+    
+    # 获取期间
+    if args.period:
+        periods = [args.period]
+    else:
+        periods = get_periods_to_sync(db)
+        periods = periods[:3]  # 先只同步最近3个月
+    logger.info(f"期间: {periods}")
+    
+    # 连接ERP
+    logger.info("连接ERP SQL Server...")
+    try:
+        erp_engine = get_erp_engine()
+        with erp_engine.connect() as conn:
+            conn.execute(text("SELECT 1"))
+        logger.info("✅ ERP连接成功")
+    except Exception as e:
+        logger.error(f"❌ ERP连接失败: {e}")
+        db.close()
+        return
+    
+    # 逐KPI逐期间执行
+    total_written = 0
+    for kpi in kpis:
+        code = kpi["kpi_code"]
+        mapping = KPI_SQL_MAP.get(code)
+        if not mapping:
+            logger.info(f"  [{code}] 跳过（无SQL映射）")
+            continue
+        
+        for period in periods:
+            try:
+                value = execute_erp_sql(erp_engine, mapping["sql"], period)
+                logger.info(f"  [{code}] {period} = {value}")
+                
+                if not args.dry_run:
+                    # 写入 kpi_values
+                    db.execute(text("""
+                        INSERT INTO kpi_values (kpi_id, period, actual_value, source_type, source_batch, data_status, calculated_at)
+                        VALUES (:kpi_id, :period, :value, 'erp', :batch, 'verified', NOW())
+                        ON DUPLICATE KEY UPDATE actual_value=:value2, source_batch=:batch2, data_status='verified', calculated_at=NOW()
+                    """), {
+                        "kpi_id": kpi["id"],
+                        "period": period,
+                        "value": value,
+                        "batch": f"erp_sync_{datetime.now().strftime('%Y%m%d_%H%M')}",
+                        "value2": value,
+                        "batch2": f"erp_sync_{datetime.now().strftime('%Y%m%d_%H%M')}",
+                    })
+                    db.commit()
+                    total_written += 1
+            except Exception as e:
+                logger.error(f"  ❌ [{code}] {period} 失败: {e}")
+    
+    db.close()
+    logger.info(f"\n同步完成! 写入 {total_written} 条, 期间: {periods}")
+    
+    if args.dry_run:
+        logger.info("（试运行模式，未写入数据库）")
+
+
+if __name__ == "__main__":
+    main()