cma-management/backend/scripts/erp_p1_crm_prod.py

""""
ERP数据源扩展 — 任务5: CRM+生产模块对接
执行: cd /root/cma-management/backend && python3 scripts/erp_p1_crm_prod.py
"""
import sys, os
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from dotenv import load_dotenv
load_dotenv()

from app.database import get_engine
from sqlalchemy import text
from datetime import datetime

engine = get_engine()

def log(msg):
    print(f"[{datetime.now():%H:%M:%S}] {msg}")

# CRM + 生产模块数据源配置
NEW_SOURCES = [
    # ─── CRM模块 ───
    {
        "name": "ERP-CRM-客户保留率",
        "kpi_codes": ["C_RETENTION_RATE"],
        "source_type": "erp",
        "api_endpoint": "http://erp-api.sxbh.ltd/api/v1/crm/retention-rate",
        "query_sql": """SELECT CASE WHEN total_customers > 0 
THEN ROUND(renew_customers/total_customers*100, 2) ELSE 0 END as value 
FROM (SELECT COUNT(*) as total_customers, 
SUM(CASE WHEN DATEDIFF(day, last_order_date, GETDATE()) <= 365 THEN 1 ELSE 0 END) as renew_customers 
FROM Units WHERE UnitType='Customer') t""",
        "sync_type": "daily",
    },
    {
        "name": "ERP-CRM-新客户数",
        "kpi_codes": ["C_NEW_CLIENTS"],
        "source_type": "erp",
        "api_endpoint": "http://erp-api.sxbh.ltd/api/v1/crm/new-customers",
        "query_sql": "SELECT COUNT(*) as value FROM Units WHERE UnitType='Customer' AND DATEDIFF(day, CreateDate, GETDATE()) <= 30",
        "sync_type": "daily",
    },
    # ─── 生产模块 ───
    {
        "name": "ERP-生产-产品合格率",
        "kpi_codes": ["F_QUALITY_RATE"],
        "source_type": "erp",
        "api_endpoint": "http://erp-api.sxbh.ltd/api/v1/production/quality-rate",
        "query_sql": """SELECT CASE WHEN total_qty > 0 
THEN ROUND(qualified_qty/total_qty*100, 2) ELSE 0 END as value 
FROM (SELECT COUNT(*) as total_qty, 
SUM(CASE WHEN QualityStatus='OK' THEN 1 ELSE 0 END) as qualified_qty 
FROM QualityInspection WHERE Period=:period) t""",
        "sync_type": "daily",
    },
    {
        "name": "ERP-生产-返工率",
        "kpi_codes": ["F_REWORK_RATE"],
        "source_type": "erp",
        "api_endpoint": "http://erp-api.sxbh.ltd/api/v1/production/rework-rate",
        "query_sql": """SELECT CASE WHEN total_qty > 0 
THEN ROUND(rework_qty/total_qty*100, 2) ELSE 0 END as value 
FROM (SELECT COUNT(*) as total_qty, 
SUM(CASE WHEN ReworkStatus='Rework' THEN 1 ELSE 0 END) as rework_qty 
FROM ProductionOrder WHERE Period=:period) t""",
        "sync_type": "daily",
    },
]

with engine.connect() as conn:
    print("=" * 70)
    log("开始ERP数据源扩展 — 任务5: CRM+生产模块")
    print("=" * 70)

    created = 0
    for src in NEW_SOURCES:
        exists = conn.execute(
            text("SELECT id FROM data_source_config WHERE name=:name"),
            {"name": src["name"]}
        ).fetchone()
        if exists:
            log(f"  ⏭️  已存在: {src['name']} (id={exists[0]})")
            continue

        conn.execute(text("""
            INSERT INTO data_source_config 
            (name, source_type, api_endpoint, query_sql, sync_type, status, created_at)
            VALUES (:name, :source_type, :api_endpoint, :query_sql, :sync_type, 'active', NOW())
        """), {
            "name": src["name"],
            "source_type": src["source_type"],
            "api_endpoint": src["api_endpoint"],
            "query_sql": src["query_sql"],
            "sync_type": src["sync_type"],
        })
        created += 1
        log(f"  ✅ 新增: {src['name']} — 关联KPI: {', '.join(src['kpi_codes'])}")

    conn.commit()

    # 打印所有数据源
    rows = conn.execute(text(
        "SELECT id, name, source_type, sync_type, status FROM data_source_config ORDER BY id"
    )).fetchall()
    print("\n数据源清单:")
    for r in rows:
        print(f"  [{r[0]}] {r[1]:35s} type={r[2]:10s} sync={r[3]:10s} status={r[4]}")
    log(f"\n✅ CRM+生产模块数据源配置完成: 新增 {created} 条, 共 {len(rows)} 条")

    # 更新KPI的data_source_config字段（存储关联的数据源ID）
    log("\n更新KPI的data_source_config字段...")
    for src in NEW_SOURCES:
        src_row = conn.execute(
            text("SELECT id FROM data_source_config WHERE name=:name"),
            {"name": src["name"]}
        ).fetchone()
        if not src_row:
            continue
        for kpi_code in src["kpi_codes"]:
            kpi = conn.execute(
                text("SELECT id, data_source_config FROM kpi_definitions WHERE kpi_code=:code AND status='active'"),
                {"code": kpi_code}
            ).fetchone()
            if kpi:
                existing_config = kpi[1] or {}
                if isinstance(existing_config, str):
                    import json
                    try:
                        existing_config = json.loads(existing_config)
                    except:
                        existing_config = {}
                # 确保是dict
                if not isinstance(existing_config, dict):
                    existing_config = {"source_ids": []}
                if "source_ids" not in existing_config:
                    existing_config["source_ids"] = []
                if src_row[0] not in existing_config["source_ids"]:
                    existing_config["source_ids"].append(src_row[0])
                import json
                conn.execute(
                    text("UPDATE kpi_definitions SET data_source_config=:config WHERE id=:id"),
                    {"config": json.dumps(existing_config, ensure_ascii=False), "id": kpi[0]}
                )
                log(f"  ✅ KPI {kpi_code}: data_source_config 已更新")

    conn.commit()
    log("\n✅ 任务5: CRM+生产模块对接完成")
    print("=" * 70)