sxbh_admin/cma-management
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 多租户联动 — CMA切换企业注入tenant_id + a2a_dispatch分发

Browse Source
This commit is contained in:
Hermes CI Fix
2026-08-07 17:42:26 +08:00
parent f33162b6c9
commit ac6ea93ed5
5 changed files with 216 additions and 73 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/scripts/erp_sync.py
+146 -71
View File
@@ -22,7 +22,7 @@ from dotenv import load_dotenv
load_dotenv()
from app.database import get_engine, get_session_local
from app.models import KPIDefinition, KPIValue, OperationLog
from app.models import KPIDefinition, KPIValue, OperationLog, DataSourceConfig
logger = logging.getLogger("erp_sync")
@@ -30,6 +30,16 @@ logger = logging.getLogger("erp_sync")
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'}}}",
}
# ============================================================
# 公式解析
# ============================================================
@@ -95,36 +105,147 @@ def parse_formula(formula: str) -> dict:
# ============================================================
# API 模式: 通过 ERP 接口查询
# API 模式: 通过 ERP 接口查询（表驱动）
# ============================================================
def fetch_via_api(kpi: KPIDefinition, parsed: dict, period: str) -> float:
"""通过 erp-api-gateway 查询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
# 各KPI对应的API路径
API_MAP = {
"SALES_TOTAL": f"{ERP_API_BASE}/stats/monthly?year={period_year}",
"CUSTOMER_COUNT": f"{ERP_API_BASE}/stats/monthly?year={period_year}",
"SALES_PROFIT_RATE": f"{ERP_API_BASE}/stats/gross-profit?year={period_year}&month={period_month}",
"TOP5_CUSTOMER_RATIO": f"{ERP_API_BASE}/stats/customer-top?year={period_year}&limit=5",
# P1: CRM模块
"C_RETENTION_RATE": f"{ERP_API_BASE}/crm/retention-rate?year={period_year}&month={period_month}",
"C_NEW_CLIENTS": f"{ERP_API_BASE}/crm/new-customers?year={period_year}&month={period_month}",
# P1: 生产模块
"F_QUALITY_RATE": f"{ERP_API_BASE}/production/quality-rate?year={period_year}&month={period_month}",
"F_REWORK_RATE": f"{ERP_API_BASE}/production/rework-rate?year={period_year}&month={period_month}",
}
headers = {"X-API-Key": ERP_API_KEY, "User-Agent": "CMA-ERP-SYNC/1.0"}
if kpi_code not in API_MAP:
url, parser = _resolve_endpoint(kpi, db_session, period)
if not url:
raise ValueError(f"未配置API映射: {kpi_code}")
url = API_MAP[kpi_code]
logger.info(f" [{kpi_code}] API请求: {url}")
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:
@@ -135,53 +256,7 @@ def fetch_via_api(kpi: KPIDefinition, parsed: dict, period: str) -> float:
except Exception as e:
raise ConnectionError(f"API请求失败: {e}")
if kpi_code == "SALES_TOTAL":
# 从 monthly trend 中取对应月份
for m in data.get("data", []):
if m["period"] == period:
return float(m["amount"])
# fallback: 取汇总
return float(data.get("summary", {}).get("total_amount", 0))
elif kpi_code == "CUSTOMER_COUNT":
for m in data.get("data", []):
if m["period"] == period:
return float(m["customers"])
return 0
elif kpi_code == "SALES_PROFIT_RATE":
return float(data.get("gross_profit_rate", 0))
elif kpi_code == "TOP5_CUSTOMER_RATIO":
top5 = data.get("data", [])
top5_total = sum(c["amount"] for c in top5)
# 同时获取全年总额
total_url = f"{ERP_API_BASE}/stats/monthly?year={period_year}"
req2 = urllib.request.Request(total_url, headers=headers)
with urllib.request.urlopen(req2, timeout=15) as resp2:
total_data = json.loads(resp2.read().decode())
total_amount = sum(m["amount"] for m in total_data.get("data", []))
if total_amount > 0:
return round(top5_total / total_amount * 100, 2)
return 0
# P1: CRM模块 — 客户保留率
elif kpi_code == "C_RETENTION_RATE":
return float(data.get("retention_rate", data.get("value", 0)))
# P1: CRM模块 — 新客户数
elif kpi_code == "C_NEW_CLIENTS":
return float(data.get("new_customers", data.get("value", 0)))
# P1: 生产模块 — 产品合格率
elif kpi_code == "F_QUALITY_RATE":
return float(data.get("quality_rate", data.get("value", 0)))
# P1: 生产模块 — 返工率
elif kpi_code == "F_REWORK_RATE":
return float(data.get("rework_rate", data.get("value", 0)))
raise ValueError(f"未实现的API映射: {kpi_code}")
return _parse_response(kpi_code, parser, data, period)
# ============================================================
@@ -254,7 +329,7 @@ def sync_kpi(kpi: KPIDefinition, db_session, dry_run: bool = False,
api_ok = False
if use_api:
try:
value = fetch_via_api(kpi, parsed, current_period)
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}")