From ac6ea93ed56bc8162689ddefb1388e7d42826b5b Mon Sep 17 00:00:00 2001 From: Hermes CI Fix Date: Fri, 7 Aug 2026 17:42:26 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=A4=9A=E7=A7=9F=E6=88=B7=E8=81=94?= =?UTF-8?q?=E5=8A=A8=20=E2=80=94=20CMA=E5=88=87=E6=8D=A2=E4=BC=81=E4=B8=9A?= =?UTF-8?q?=E6=B3=A8=E5=85=A5tenant=5Fid=20+=20a2a=5Fdispatch=E5=88=86?= =?UTF-8?q?=E5=8F=91?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/app/api/action_plans.py | 2 +- backend/app/api/tenant.py | 63 ++++++++ backend/app/main.py | 3 +- backend/scripts/erp_sync.py | 217 +++++++++++++++++++--------- frontend/src/layouts/MainLayout.vue | 4 + 5 files changed, 216 insertions(+), 73 deletions(-) create mode 100644 backend/app/api/tenant.py diff --git a/backend/app/api/action_plans.py b/backend/app/api/action_plans.py index dfd4379c..8f910463 100644 --- a/backend/app/api/action_plans.py +++ b/backend/app/api/action_plans.py @@ -216,7 +216,7 @@ def plan_stats(db: Session = Depends(get_db), current_user: User = Depends(requi in_progress = query.filter(ActionPlan.status == "in_progress").count() completed = query.filter(ActionPlan.status == "completed").count() from datetime import datetime - overdue = query.filter(ActionPlan.status.in_(["pending", "in_progress"]), ActionPlan.deadline < datetime.now()).count() + overdue = query.filter(ActionPlan.status.in_(["pending", "in_progress"]), ActionPlan.due_date < datetime.now()).count() return { "total": total, "pending": pending, diff --git a/backend/app/api/tenant.py b/backend/app/api/tenant.py new file mode 100644 index 00000000..daba01ef --- /dev/null +++ b/backend/app/api/tenant.py @@ -0,0 +1,63 @@ +"""租户状态API — CMA系统切换公司时记录当前tenant，供项目Bot分发A2A任务""" +from fastapi import APIRouter, Depends, HTTPException +from pydantic import BaseModel +from sqlalchemy.orm import Session +from app.database import get_db +from app.models import SystemConfig, Entity + +router = APIRouter(prefix="/api/cma/tenant", tags=["多租户"]) + +TENANT_KEY = "current_tenant" + +class TenantSwitchRequest(BaseModel): + entity_id: int + source: str = "cma-system" + + +def _tenant_name(entity_id: int) -> str: + """entity_id → tenant_id""" + return "company_b" if entity_id == 2 else "company_a" + + +@router.get("/current") +def get_current_tenant(db: Session = Depends(get_db)): + """查询当前租户""" + cfg = db.query(SystemConfig).filter(SystemConfig.config_key == TENANT_KEY).first() + if cfg and cfg.config_value: + import json + try: + return json.loads(cfg.config_value) + except: + pass + return {"tenant_id": "company_a", "entity_id": 1, "name": "陕西酣客文化传媒"} + + +@router.post("/switch") +def switch_tenant(data: TenantSwitchRequest, db: Session = Depends(get_db)): + """切换当前租户（CMA前端企业切换器调用）""" + ent = db.query(Entity).filter(Entity.id == data.entity_id).first() + if not ent: + raise HTTPException(404, "企业不存在") + + import json + state = { + "tenant_id": _tenant_name(data.entity_id), + "entity_id": data.entity_id, + "name": ent.name, + "short_name": ent.short_name, + "source": data.source, + "switched_at": __import__("datetime").datetime.now().isoformat(), + } + + cfg = db.query(SystemConfig).filter(SystemConfig.config_key == TENANT_KEY).first() + if cfg: + cfg.config_value = json.dumps(state, ensure_ascii=False) + else: + db.add(SystemConfig( + config_key=TENANT_KEY, + config_value=json.dumps(state, ensure_ascii=False), + description="当前租户状态（CMA企业切换联动）", + )) + db.commit() + + return {"success": True, "current": state} diff --git a/backend/app/main.py b/backend/app/main.py index 05999c2a..d8686f34 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -5,7 +5,7 @@ from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import JSONResponse from dotenv import load_dotenv from app.database import init_db -from app.api import auth, kpis, templates, maps, dashboard, data, alerts, ai_analysis, alert_rules, users, thresholds, notifications, permissions, action_plans, alignment, org, objectives, versions, budget, cost, predict, reports, security, knowledge, bot_bridge, bot_bridge_v2, lead, customer_dashboard, deviation_push, budget_generate, knowledge_articles, kpi_causality, data_quality, bi_reports, entities, bsc_layers, okr, okr_templates, subjects, driver_budget, bot_kpis, bot_iron_law, analysis_results, expenses, cash, tax_compliance +from app.api import auth, kpis, templates, maps, dashboard, data, alerts, ai_analysis, alert_rules, users, thresholds, notifications, permissions, action_plans, alignment, org, objectives, versions, budget, cost, predict, reports, security, knowledge, bot_bridge, bot_bridge_v2, lead, tenant, customer_dashboard, deviation_push, budget_generate, knowledge_articles, kpi_causality, data_quality, bi_reports, entities, bsc_layers, okr, okr_templates, subjects, driver_budget, bot_kpis, bot_iron_law, analysis_results, expenses, cash, tax_compliance from app.utils.cache import clear_all as clear_cache, delete as delete_cache from scripts.erp_sync import run_sync as run_erp_sync from app.auth_middleware import require_auth @@ -57,6 +57,7 @@ app.include_router(knowledge.router) app.include_router(bot_bridge.router) app.include_router(bot_bridge_v2.router) app.include_router(lead.router) +app.include_router(tenant.router) app.include_router(customer_dashboard.router) app.include_router(deviation_push.router) app.include_router(budget_generate.router) diff --git a/backend/scripts/erp_sync.py b/backend/scripts/erp_sync.py index 7381c85a..22e3e576 100644 --- a/backend/scripts/erp_sync.py +++ b/backend/scripts/erp_sync.py @@ -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}") diff --git a/frontend/src/layouts/MainLayout.vue b/frontend/src/layouts/MainLayout.vue index e0c2494d..13dd33ca 100644 --- a/frontend/src/layouts/MainLayout.vue +++ b/frontend/src/layouts/MainLayout.vue @@ -71,6 +71,10 @@ function onEntityChange(id: number) { currentEntityName.value = e?.short_name || e?.name || '' localStorage.setItem('cma_entity_id', String(id)) localStorage.setItem('cma_entity_name', currentEntityName.value) + // 多租户联动：通知后端记录当前tenant（供项目Bot A2A分发） + try { + api.post('/tenant/switch', { entity_id: id, source: 'cma-frontend' }) + } catch (_) {} location.reload() } onMounted(async () => {