477 lines
18 KiB
Python
477 lines
18 KiB
Python
"""
|
||
ERP数据同步脚本 — 管理会计OS
|
||
根据 kpi_definitions.formula 中的规则从ERP系统拉取数据并写入 kpi_values
|
||
支持:
|
||
- HTTP API 模式: 通过 erp-api-gateway 查询实时数据
|
||
- Fallback 模式: API不可达时使用本地已有数据或标记待同步
|
||
- 定时执行(crontab) + 手动触发
|
||
"""
|
||
|
||
import sys
|
||
import os
|
||
import json
|
||
import logging
|
||
import re
|
||
import urllib.request
|
||
import urllib.error
|
||
from datetime import datetime, timedelta
|
||
|
||
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, get_session_local
|
||
from app.models import KPIDefinition, KPIValue, OperationLog, DataSourceConfig
|
||
|
||
logger = logging.getLogger("erp_sync")
|
||
|
||
# ERP API 配置
|
||
ERP_API_BASE = os.getenv("ERP_API_BASE", "http://127.0.0.1:8300/api/v1")
|
||
ERP_API_KEY = os.getenv("ERP_API_KEY", "erp-gateway-key-bhwl-2026")
|
||
|
||
# 无DB会话时使用的静态映射（保底，使用真实KPI编码）
|
||
# 对应 data_source_config 表的 active 端点（id=1,2,3,4）
|
||
# {period_year} 和 {period_month} 在 _resolve_endpoint 中替换
|
||
STATIC_API_MAP = {
|
||
"F_REVENUE": f"{ERP_API_BASE}/stats/monthly?year={{{'period_year'}}}",
|
||
"F_GROSS_MARGIN": f"{ERP_API_BASE}/stats/gross-profit?year={{{'period_year'}}}&month={{{'period_month'}}}",
|
||
"F_NET_PROFIT": f"{ERP_API_BASE}/stats/monthly?year={{{'period_year'}}}",
|
||
"F_COST_RATIO": f"{ERP_API_BASE}/stats/monthly-cost?year={{{'period_year'}}}",
|
||
}
|
||
|
||
# ============================================================
|
||
# 公式解析
|
||
# ============================================================
|
||
|
||
def parse_formula(formula: str) -> dict:
|
||
"""解析KPI公式，提取ERP表和字段映射"""
|
||
result = {"table": "MasterBill", "field": None, "agg": "SUM",
|
||
"where": None, "raw": formula, "erp_direct": True}
|
||
|
||
# 特殊处理中文描述的公式
|
||
ZH_PATTERNS = {
|
||
"前5客户销售额/总销售额*100": ("TOP5_CUSTOMER", "MasterBill"),
|
||
"前5客户集中度": ("TOP5_CUSTOMER", "MasterBill"),
|
||
"满意客户数/总客户数*100": ("CUSTOMER_SAT_RATIO", "MasterBill"),
|
||
"准时交付订单/总订单*100": ("DELIVERY_RATE", "MasterBill"),
|
||
"完成培训人数/应培训人数*100": ("TRAINING_RATE", "MasterBill"),
|
||
}
|
||
for zh_pattern, (agg_type, table) in ZH_PATTERNS.items():
|
||
if zh_pattern in formula:
|
||
result.update({"agg": agg_type, "table": table, "field_expr": formula,
|
||
"erp_direct": False}) # 不能直接跑SQL
|
||
return result
|
||
|
||
# 优先检测比率型公式: SUM(A)/SUM(B)*100
|
||
ratio_m = re.match(r"(SUM|COUNT|AVG)\s*\((.+?)\)\s*/\s*(SUM|COUNT|AVG)\s*\((.+?)\)", formula, re.I)
|
||
if ratio_m:
|
||
result["agg"] = f"RATIO_{ratio_m.group(1)}"
|
||
result["field_expr"] = f"({ratio_m.group(2)})/({ratio_m.group(4)})"
|
||
return result
|
||
|
||
# 匹配完整聚合: SUM(...), COUNT(DISTINCT ...), COUNT(...), AVG(...)
|
||
m = re.match(r"(SUM|COUNT(?:\s+DISTINCT)?|AVG|MAX|MIN)\s*\((.+?)\)", formula, re.I)
|
||
if not m:
|
||
result["field_expr"] = "1"
|
||
result["table"] = "MasterBill"
|
||
result["erp_direct"] = False
|
||
return result
|
||
|
||
agg_func = m.group(1).strip().upper()
|
||
field_expr = m.group(2).strip()
|
||
|
||
if agg_func.startswith("COUNT") and field_expr.startswith("DISTINCT "):
|
||
result["agg"] = "COUNT_DISTINCT"
|
||
cleaned = field_expr.replace("DISTINCT ", "").strip()
|
||
result["field_expr"] = cleaned
|
||
parts = cleaned.split(".")
|
||
if parts:
|
||
result["table"] = parts[0]
|
||
else:
|
||
result["agg"] = agg_func
|
||
result["field_expr"] = field_expr
|
||
parts = field_expr.split(".")
|
||
if len(parts) >= 2:
|
||
candidate = parts[0].strip()
|
||
if candidate and candidate[0].isupper():
|
||
result["table"] = candidate
|
||
|
||
wm = re.search(r"WHERE\s+(.+)$", formula, re.I)
|
||
if wm:
|
||
result["where"] = wm.group(1).strip()
|
||
|
||
return result
|
||
|
||
|
||
# ============================================================
|
||
# API 模式: 通过 ERP 接口查询（表驱动）
|
||
# ============================================================
|
||
|
||
# 端点类型 → 解析方式映射
|
||
# key: 端点路径片段, value: 解析类型
|
||
# 注意: 更具体的路径要放在前面（如 monthly-cost 在 monthly 之前）
|
||
ENDPOINT_PARSER = {
|
||
"/stats/monthly-cost": "monthly_cost", # data:[{period, sales, cost, orders}]
|
||
"/stats/monthly": "monthly", # data:[{period, orders, customers, amount}]
|
||
"/stats/gross-profit": "gross_profit", # {gross_profit_rate}
|
||
"/stats/customer-top": "customer_top", # data:[{name, orders, amount}]
|
||
"/stats/product-top": "product_top", # data:[{name, qty, amount, orders}]
|
||
"/sales/summary": "sales_summary", # {total_amount, customer_count, ...}
|
||
"/sales/trend": "trend", # data:[{period, amount}]
|
||
"/data/query": "raw_query", # {data:[...]}
|
||
"/crm/retention-rate": "crm_retention", # {retention_rate|value}
|
||
"/crm/new-customers": "crm_new", # {new_customers|value}
|
||
"/production/quality-rate": "quality", # {quality_rate|value}
|
||
"/production/rework-rate": "rework", # {rework_rate|value}
|
||
"/cashflow": "cashflow", # {value}
|
||
"/ar/aging": "ar_aging", # {value}
|
||
"/delivery/rate": "delivery_rate", # {value}
|
||
"/quality/defect": "quality_defect", # {value}
|
||
}
|
||
|
||
|
||
def _resolve_endpoint(kpi: KPIDefinition, db_session, period: str) -> tuple:
|
||
"""从 data_source_config 表解析KPI的API端点。
|
||
返回 (url, parser_type) 或 (None, None)
|
||
"""
|
||
period_month = int(period[5:7])
|
||
period_year = int(period[:4])
|
||
|
||
# 1. 优先从 kpi.data_source_config JSON (source_ids) 查 data_source_config 表
|
||
if db_session is not None:
|
||
cfg = getattr(kpi, "data_source_config", None)
|
||
if isinstance(cfg, dict) and cfg.get("source_ids"):
|
||
for sid in cfg["source_ids"]:
|
||
src = db_session.query(DataSourceConfig).filter(
|
||
DataSourceConfig.id == sid,
|
||
DataSourceConfig.status == "active",
|
||
).first()
|
||
if src and src.api_endpoint:
|
||
url = src.api_endpoint
|
||
# 兼容 {period_year}/{period_month} 模板
|
||
url = url.replace("{period_year}", str(period_year)) \
|
||
.replace("{period_month}", str(period_month))
|
||
parser = _detect_parser(url)
|
||
logger.info(f" [{kpi.kpi_code}] 表驱动: source_id={sid} → {url}")
|
||
return url, parser
|
||
|
||
# 2. 静态映射保底（旧版硬编码，含真实KPI编码）
|
||
url = STATIC_API_MAP.get(kpi.kpi_code)
|
||
if url:
|
||
url = url.replace("{period_year}", str(period_year)) \
|
||
.replace("{period_month}", str(period_month))
|
||
return url, _detect_parser(url)
|
||
|
||
return None, None
|
||
|
||
|
||
def _detect_parser(url: str) -> str:
|
||
"""根据URL路径推断解析类型"""
|
||
for fragment, parser in ENDPOINT_PARSER.items():
|
||
if fragment in url:
|
||
return parser
|
||
return "generic"
|
||
|
||
|
||
def _parse_response(kpi_code: str, parser: str, data: dict, period: str) -> float:
|
||
"""按端点类型解析响应，返回数值"""
|
||
# 通用兜底
|
||
if isinstance(data, (int, float)):
|
||
return float(data)
|
||
|
||
# data[] 数组类型
|
||
rows = data.get("data", []) if isinstance(data, dict) else []
|
||
|
||
if parser == "monthly":
|
||
for m in rows:
|
||
if m.get("period") == period:
|
||
return float(m.get("amount", 0))
|
||
return float(data.get("summary", {}).get("total_amount", 0)) if isinstance(data, dict) else 0
|
||
|
||
elif parser == "monthly_cost":
|
||
for m in rows:
|
||
if m.get("period") == period:
|
||
return float(m.get("cost", 0))
|
||
return 0
|
||
|
||
elif parser == "customer_top":
|
||
return float(sum(c.get("amount", 0) for c in rows))
|
||
|
||
elif parser == "product_top":
|
||
return float(sum(c.get("amount", 0) for c in rows))
|
||
|
||
elif parser == "trend":
|
||
for m in rows:
|
||
if m.get("period") == period:
|
||
return float(m.get("amount", 0))
|
||
return 0
|
||
|
||
elif parser == "raw_query":
|
||
# /data/query 返回 {data:[{...}]} 取第一行第一个数值
|
||
if rows:
|
||
first = rows[0]
|
||
for v in first.values():
|
||
if isinstance(v, (int, float)):
|
||
return float(v)
|
||
return 0
|
||
|
||
elif parser == "gross_profit":
|
||
return float(data.get("gross_profit_rate", data.get("value", 0)))
|
||
|
||
elif parser == "sales_summary":
|
||
return float(data.get("total_amount", data.get("value", 0)))
|
||
|
||
elif parser in ("crm_retention", "crm_new", "quality", "rework",
|
||
"cashflow", "ar_aging", "delivery_rate", "quality_defect"):
|
||
# 所有财务/运营扩展端点统一返回 {value: xxx}
|
||
return float(data.get("value", data.get("retention_rate", data.get("new_customers",
|
||
data.get("quality_rate", data.get("rework_rate", 0))))))
|
||
|
||
# generic: 取常见字段
|
||
if isinstance(data, dict):
|
||
for key in ("value", "amount", "total", "actual_value", "result"):
|
||
if key in data and data[key] is not None:
|
||
return float(data[key])
|
||
return 0.0
|
||
|
||
|
||
def fetch_via_api(kpi: KPIDefinition, parsed: dict, period: str, db_session=None) -> float:
|
||
"""通过 erp-api-gateway 查询ERP数据（表驱动）"""
|
||
kpi_code = kpi.kpi_code
|
||
|
||
url, parser = _resolve_endpoint(kpi, db_session, period)
|
||
if not url:
|
||
raise ValueError(f"未配置API映射: {kpi_code}")
|
||
|
||
headers = {"X-API-Key": ERP_API_KEY, "User-Agent": "CMA-ERP-SYNC/1.0"}
|
||
logger.info(f" [{kpi_code}] API请求: {url} (parser={parser})")
|
||
|
||
req = urllib.request.Request(url, headers=headers)
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=15) as resp:
|
||
data = json.loads(resp.read().decode())
|
||
except urllib.error.HTTPError as e:
|
||
raise ConnectionError(f"API返回 {e.code}: {e.read().decode()[:200]}")
|
||
except Exception as e:
|
||
raise ConnectionError(f"API请求失败: {e}")
|
||
|
||
return _parse_response(kpi_code, parser, data, period)
|
||
|
||
|
||
# ============================================================
|
||
# Fallback 模式: 本地已有数据推算
|
||
# ============================================================
|
||
|
||
def fetch_fallback(kpi: KPIDefinition, parsed: dict, db_session, period: str) -> float:
|
||
"""Fallback: 从本地已有 kpi_values 推算或返回 None"""
|
||
kpi_code = kpi.kpi_code
|
||
|
||
# 对于已有数据的KPI，沿用最近月份的值（标注为estimated）
|
||
existing = db_session.query(KPIValue).filter(
|
||
KPIValue.kpi_id == kpi.id,
|
||
KPIValue.source_type.in_(["erp", "manual"]),
|
||
).order_by(KPIValue.period.desc()).first()
|
||
|
||
if existing and existing.actual_value is not None and existing.data_status == "verified":
|
||
logger.info(f" [{kpi_code}] Fallback: 沿用最近期 verified数据 {existing.period}={existing.actual_value}")
|
||
return existing.actual_value
|
||
|
||
# Bug3修复: 不沿用estimated旧值（可能失真，如4800万收入），避免污染当月数据
|
||
if existing and existing.actual_value is not None and existing.data_status != "verified":
|
||
logger.info(f" [{kpi_code}] Fallback: 最近期 {existing.period} 为estimated({existing.actual_value})，不沿用，返回None")
|
||
return None
|
||
|
||
# 特殊KPI的默认值
|
||
DEFAULTS = {
|
||
"SALES_TOTAL": 800000,
|
||
"CUSTOMER_COUNT": 25,
|
||
"SALES_PROFIT_RATE": 25.0,
|
||
"TOP5_CUSTOMER_RATIO": 50.0,
|
||
}
|
||
if kpi_code in DEFAULTS:
|
||
logger.info(f" [{kpi_code}] Fallback: 使用默认值 {DEFAULTS[kpi_code]}")
|
||
return DEFAULTS[kpi_code]
|
||
|
||
return None
|
||
|
||
|
||
# ============================================================
|
||
# TOP5_CUSTOMER_RATIO 的公式补充处理
|
||
# ============================================================
|
||
|
||
def compute_top5_ratio(db_session, period: str) -> float:
|
||
"""从 ERP schema 采集数据计算：前5客户销售额/总销售额*100"""
|
||
# 先检查 erp_schema 是否有 MasterBill 的完整数据
|
||
# 如果有物化数据，可以在这里做本地计算
|
||
# 目前 erp_schema 只有元数据没有数据，返回 None 表示需要 API
|
||
return None
|
||
|
||
|
||
# ============================================================
|
||
# 核心同步函数
|
||
# ============================================================
|
||
|
||
def sync_kpi(kpi: KPIDefinition, db_session, dry_run: bool = False,
|
||
use_api: bool = True, target_period: str = None) -> bool:
|
||
"""同步单个KPI的ERP数据"""
|
||
if kpi.data_source_type not in ("erp",):
|
||
return False
|
||
|
||
formula = kpi.formula
|
||
if not formula:
|
||
logger.warning(f" [{kpi.kpi_code}] 无公式定义")
|
||
return False
|
||
|
||
parsed = parse_formula(formula)
|
||
logger.info(f" [{kpi.kpi_code}] 解析: table={parsed['table']}, agg={parsed['agg']}, "
|
||
f"erp_direct={parsed.get('erp_direct',True)}")
|
||
|
||
current_period = target_period if target_period else datetime.now().strftime("%Y-%m")
|
||
|
||
# 尝试通过 API 获取
|
||
value = None
|
||
api_ok = False
|
||
if use_api:
|
||
try:
|
||
value = fetch_via_api(kpi, parsed, current_period, db_session=db_session)
|
||
if value is not None:
|
||
api_ok = True
|
||
logger.info(f" [{kpi.kpi_code}] API结果: {current_period}={value}")
|
||
except Exception as e:
|
||
logger.warning(f" [{kpi.kpi_code}] API失败: {e}")
|
||
|
||
# API 失败则 fallback
|
||
if not api_ok:
|
||
try:
|
||
value = fetch_fallback(kpi, parsed, db_session, current_period)
|
||
if value is not None:
|
||
source_note = "estimated"
|
||
logger.info(f" [{kpi.kpi_code}] Fallback结果: {current_period}={value}")
|
||
else:
|
||
logger.warning(f" [{kpi.kpi_code}] 无可用数据, 跳过")
|
||
return False
|
||
except Exception as e:
|
||
logger.error(f" [{kpi.kpi_code}] Fallback失败: {e}")
|
||
return False
|
||
|
||
if dry_run:
|
||
logger.info(f" [{kpi.kpi_code}] DRY RUN: 跳过写入 value={value}")
|
||
return True
|
||
|
||
# 写入 kpi_values
|
||
try:
|
||
existing = db_session.query(KPIValue).filter(
|
||
KPIValue.kpi_id == kpi.id,
|
||
KPIValue.period == current_period,
|
||
KPIValue.source_type == "erp",
|
||
).first()
|
||
|
||
remark = f"ERP自动同步{' (API)' if api_ok else ' (估算)'} {datetime.now().strftime('%Y-%m-%d %H:%M')}"
|
||
|
||
if existing:
|
||
existing.actual_value = value
|
||
existing.data_status = "verified" if api_ok else "estimated"
|
||
existing.remark = remark
|
||
existing.source_type = "erp"
|
||
logger.info(f" [{kpi.kpi_code}] 更新 {current_period}: {value}")
|
||
else:
|
||
kv = KPIValue(
|
||
kpi_id=kpi.id,
|
||
period=current_period,
|
||
actual_value=value,
|
||
source_type="erp",
|
||
source_batch=f"sync_{current_period}",
|
||
data_status="verified" if api_ok else "estimated",
|
||
remark=remark,
|
||
)
|
||
db_session.add(kv)
|
||
logger.info(f" [{kpi.kpi_code}] 新增 {current_period}: {value}")
|
||
|
||
db_session.commit()
|
||
return True
|
||
|
||
except Exception as e:
|
||
db_session.rollback()
|
||
logger.error(f" [{kpi.kpi_code}] 写入失败: {e}")
|
||
return False
|
||
|
||
|
||
def run_sync(dry_run: bool = False, kpi_codes: list = None, use_api: bool = True, period: str = None):
|
||
"""执行全部ERP KPI同步"""
|
||
db = get_session_local()()
|
||
try:
|
||
query = db.query(KPIDefinition).filter(
|
||
KPIDefinition.status == "active",
|
||
KPIDefinition.data_source_type == "erp",
|
||
)
|
||
if kpi_codes:
|
||
query = query.filter(KPIDefinition.kpi_code.in_(kpi_codes))
|
||
|
||
kpis = query.all()
|
||
target_period = period if period else datetime.now().strftime("%Y-%m")
|
||
logger.info(f"开始同步ERP数据: {len(kpis)} 个KPI (API模式={use_api}, 期间={target_period})")
|
||
|
||
success = 0
|
||
fail = 0
|
||
for kpi in kpis:
|
||
if sync_kpi(kpi, db, dry_run, use_api, target_period):
|
||
success += 1
|
||
else:
|
||
fail += 1
|
||
|
||
if not dry_run:
|
||
log = OperationLog(
|
||
action="erp_sync",
|
||
target_type="kpi",
|
||
detail=json.dumps({
|
||
"total": len(kpis), "success": success,
|
||
"failed": fail, "api_mode": use_api,
|
||
"period": datetime.now().strftime("%Y-%m"),
|
||
}, ensure_ascii=False),
|
||
)
|
||
db.add(log)
|
||
db.commit()
|
||
|
||
logger.info(f"同步完成: {success}成功 / {fail}失败 / {len(kpis)}总计")
|
||
|
||
finally:
|
||
db.close()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
import argparse
|
||
parser = argparse.ArgumentParser(description="ERP数据同步")
|
||
parser.add_argument("--dry-run", action="store_true", help="仅预览，不写入数据库")
|
||
parser.add_argument("--kpi", nargs="+", help="指定KPI编码")
|
||
parser.add_argument("--no-api", action="store_true", help="禁用API模式，仅用本地fallback")
|
||
parser.add_argument("--backfill", type=int, default=0,
|
||
help="回填历史月份数（如 --backfill 6 回填最近6个月）")
|
||
args = parser.parse_args()
|
||
|
||
if args.backfill:
|
||
from datetime import datetime, timedelta
|
||
from app.database import get_session_local
|
||
|
||
today = datetime.now()
|
||
months_backfilled = 0
|
||
for i in range(1, args.backfill + 1):
|
||
# 计算目标月份
|
||
m = today.month - i
|
||
y = today.year
|
||
while m <= 0:
|
||
m += 12
|
||
y -= 1
|
||
period = f"{y}-{m:02d}"
|
||
|
||
print(f"回填 {period}...")
|
||
try:
|
||
run_sync(dry_run=False, kpi_codes=args.kpi, use_api=not args.no_api, period=period)
|
||
months_backfilled += 1
|
||
except Exception as e:
|
||
print(f" {period} 失败: {e}")
|
||
|
||
print(f"回填完成: {months_backfilled} 个月")
|
||
else:
|
||
run_sync(dry_run=args.dry_run, kpi_codes=args.kpi, use_api=not args.no_api)
|