feat: bot-bridge数据通道 — POST /kpi-result回填KPI+触发预警+MPM自动回填脚本
- 新增 POST /api/cma/bot-bridge/kpi-result: (entity_id,kpi_code)查KPI, 写入KPIValue(source_type=bot, data_status=verified), 复用run_alert_check 触发预警(自动联动行动计划), 返回new_alerts; KPI不存在/缺字段返回错误 - 修复verify引擎_get_kpi_current_value排序: 按calculated_at取最新值, bot回填值可被auto-verify读到 (修复2026H1字符串排序遮蔽月值问题) - 新增scripts/bot_bridge_push.py: 财务Bot MPM结果自动回填(MPM字段→KPI 映射), 支持--mpm/--mpm-file/单KPI直推/--ping
This commit is contained in:
@@ -60,7 +60,7 @@ def _get_kpi_current_value(db: Session, kpi_id: int, entity_id: int = None) -> O
|
|||||||
latest = db.query(KPIValue).filter(
|
latest = db.query(KPIValue).filter(
|
||||||
KPIValue.kpi_id == kpi_id,
|
KPIValue.kpi_id == kpi_id,
|
||||||
KPIValue.data_status == "verified",
|
KPIValue.data_status == "verified",
|
||||||
).order_by(KPIValue.period.desc()).first()
|
).order_by(KPIValue.calculated_at.desc(), KPIValue.period.desc()).first()
|
||||||
return {
|
return {
|
||||||
"kpi_id": kpi.id,
|
"kpi_id": kpi.id,
|
||||||
"kpi_code": kpi.kpi_code,
|
"kpi_code": kpi.kpi_code,
|
||||||
@@ -403,6 +403,123 @@ def receive_mpm_result(
|
|||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
|
@router.post("/kpi-result")
|
||||||
|
def push_kpi_result(
|
||||||
|
data: dict,
|
||||||
|
bridge_bot: str = Depends(verify_bridge_token),
|
||||||
|
db: Session = Depends(get_db),
|
||||||
|
):
|
||||||
|
"""
|
||||||
|
Bot分析结果回填KPI值 + 触发预警(PRD第七部分 bot-bridge数据通道)
|
||||||
|
|
||||||
|
入参:
|
||||||
|
entity_id (int, 默认1): 企业实体ID
|
||||||
|
kpi_code (str, 必填): KPI编码
|
||||||
|
period (str, 必填): 期间 YYYY-MM
|
||||||
|
value (num, 必填): 实际值
|
||||||
|
source (str, 默认finance-bot): 来源Bot标识
|
||||||
|
remark (str, 可选): 备注
|
||||||
|
|
||||||
|
行为:
|
||||||
|
1. 按(entity_id, kpi_code)查KPI → 不存在返回错误
|
||||||
|
2. 写入KPIValue (source_type=bot, data_status=verified) — 与auto-verify引擎兼容
|
||||||
|
3. 调用 run_alert_check 触发阈值预警 → 返回 new_alerts
|
||||||
|
"""
|
||||||
|
entity_id = data.get("entity_id", 1)
|
||||||
|
kpi_code = data.get("kpi_code")
|
||||||
|
period = data.get("period")
|
||||||
|
value = data.get("value")
|
||||||
|
source = data.get("source", "finance-bot")
|
||||||
|
remark = data.get("remark")
|
||||||
|
|
||||||
|
if not kpi_code:
|
||||||
|
raise HTTPException(400, "缺少必填字段: kpi_code")
|
||||||
|
if not period:
|
||||||
|
raise HTTPException(400, "缺少必填字段: period")
|
||||||
|
if value is None or value == "":
|
||||||
|
raise HTTPException(400, "缺少必填字段: value")
|
||||||
|
try:
|
||||||
|
actual_value = float(value)
|
||||||
|
except (TypeError, ValueError):
|
||||||
|
raise HTTPException(400, f"value不是有效数值: {value!r}")
|
||||||
|
|
||||||
|
# 1. 按(entity_id, kpi_code)查KPI
|
||||||
|
kpi = db.query(KPIDefinition).filter(
|
||||||
|
KPIDefinition.entity_id == entity_id,
|
||||||
|
KPIDefinition.kpi_code == kpi_code,
|
||||||
|
).first()
|
||||||
|
if not kpi:
|
||||||
|
raise HTTPException(404, f"KPI {kpi_code} 不存在 (entity_id={entity_id})")
|
||||||
|
|
||||||
|
# 2. 写入KPIValue(同period已存在则更新,幂等upsert)
|
||||||
|
existing = db.query(KPIValue).filter(
|
||||||
|
KPIValue.kpi_id == kpi.id,
|
||||||
|
KPIValue.period == period,
|
||||||
|
).first()
|
||||||
|
if existing:
|
||||||
|
existing.actual_value = actual_value
|
||||||
|
existing.source_type = "bot"
|
||||||
|
existing.source_batch = source
|
||||||
|
existing.data_status = "verified"
|
||||||
|
existing.calculated_at = datetime.now()
|
||||||
|
if remark:
|
||||||
|
existing.remark = remark
|
||||||
|
val = existing
|
||||||
|
else:
|
||||||
|
val = KPIValue(
|
||||||
|
kpi_id=kpi.id,
|
||||||
|
period=period,
|
||||||
|
actual_value=actual_value,
|
||||||
|
source_type="bot",
|
||||||
|
source_batch=source,
|
||||||
|
data_status="verified",
|
||||||
|
calculated_at=datetime.now(),
|
||||||
|
remark=remark,
|
||||||
|
)
|
||||||
|
db.add(val)
|
||||||
|
db.commit()
|
||||||
|
db.refresh(val)
|
||||||
|
|
||||||
|
# 3. 触发预警检查(复用alert_generator引擎,yellow/red自动联动行动计划)
|
||||||
|
from scripts.alert_generator import run_alert_check
|
||||||
|
new_count = run_alert_check(db, period)
|
||||||
|
|
||||||
|
# 收集本次写入值直接触发的预警(kpi_value_id关联)
|
||||||
|
new_alerts = []
|
||||||
|
if new_count:
|
||||||
|
triggered = db.query(KPIAlert).filter(
|
||||||
|
KPIAlert.kpi_value_id == val.id,
|
||||||
|
).order_by(KPIAlert.created_at.desc()).all()
|
||||||
|
new_alerts = [
|
||||||
|
{
|
||||||
|
"id": a.id,
|
||||||
|
"kpi_id": a.kpi_id,
|
||||||
|
"kpi_code": kpi.kpi_code,
|
||||||
|
"level": a.alert_level,
|
||||||
|
"message": a.alert_message,
|
||||||
|
"action_plan_id": a.action_plan_linked_id,
|
||||||
|
"created_at": a.created_at.isoformat() if a.created_at else None,
|
||||||
|
} for a in triggered
|
||||||
|
]
|
||||||
|
|
||||||
|
logger.info(
|
||||||
|
f"[bot-bridge] KPI回填: {kpi.kpi_code}@{period}={actual_value} "
|
||||||
|
f"source={source} entity={entity_id} alerts={new_count}"
|
||||||
|
)
|
||||||
|
|
||||||
|
return {
|
||||||
|
"status": "ok",
|
||||||
|
"kpi_value_id": val.id,
|
||||||
|
"kpi_code": kpi.kpi_code,
|
||||||
|
"kpi_name": kpi.kpi_name,
|
||||||
|
"period": period,
|
||||||
|
"value": actual_value,
|
||||||
|
"source": source,
|
||||||
|
"new_alerts_count": len(new_alerts),
|
||||||
|
"new_alerts": new_alerts,
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
@router.post("/verify/{action_plan_id}")
|
@router.post("/verify/{action_plan_id}")
|
||||||
def verify_action_plan(
|
def verify_action_plan(
|
||||||
action_plan_id: int,
|
action_plan_id: int,
|
||||||
|
|||||||
@@ -0,0 +1,138 @@
|
|||||||
|
"""
|
||||||
|
bot-bridge 推送脚本 — 财务Bot分析结果自动回填CMA KPI值 + 触发预警
|
||||||
|
|
||||||
|
数据通道: 财务Bot MPM结果 → POST /api/cma/bot-bridge/kpi-result → KPIValue写入
|
||||||
|
→ run_alert_check 触发预警 → 联动行动计划
|
||||||
|
|
||||||
|
用法:
|
||||||
|
# 1) 推送MPM分析结果(自动映射MPM字段→KPI编码,逐项调kpi-result)
|
||||||
|
python3 bot_bridge_push.py --entity-id 1 --period 2026-08 \
|
||||||
|
--mpm '{"gross_margin_adjusted": 22.5, "channel_rebate_rate": 78.3}'
|
||||||
|
|
||||||
|
# 2) 从JSON文件读取MPM结果
|
||||||
|
python3 bot_bridge_push.py --entity-id 1 --period 2026-08 --mpm-file mpm_result.json
|
||||||
|
|
||||||
|
# 3) 单KPI直接回填
|
||||||
|
python3 bot_bridge_push.py --entity-id 1 --period 2026-08 \
|
||||||
|
--kpi-code C_REBATE_RATE --value 78.3 --source finance-bot
|
||||||
|
|
||||||
|
# 4) 测试连通性
|
||||||
|
python3 bot_bridge_push.py --ping
|
||||||
|
|
||||||
|
环境变量:
|
||||||
|
CMA_API=http://127.0.0.1:8010/api/cma/bot-bridge (默认)
|
||||||
|
CMA_BRIDGE_TOKEN=cma-bot-bridge-2026 (默认, 与migrate_bot_bridge_v2.sql一致)
|
||||||
|
"""
|
||||||
|
import argparse
|
||||||
|
import json
|
||||||
|
import os
|
||||||
|
import sys
|
||||||
|
from datetime import datetime
|
||||||
|
from typing import Optional
|
||||||
|
|
||||||
|
import httpx
|
||||||
|
|
||||||
|
CMA_API = os.getenv("CMA_API", "http://127.0.0.1:8010/api/cma/bot-bridge")
|
||||||
|
BRIDGE_TOKEN = os.getenv("CMA_BRIDGE_TOKEN", "cma-bot-bridge-2026")
|
||||||
|
|
||||||
|
# MPM字段 → KPI编码映射(与 bot_bridge_v2.MPM_TO_KPI_MAP 保持一致)
|
||||||
|
MPM_TO_KPI_MAP = {
|
||||||
|
"gross_margin_adjusted": "F_GROSS_MARGIN",
|
||||||
|
"net_profit_adjusted": "F_NET_PROFIT",
|
||||||
|
"channel_rebate_rate": "C_REBATE_RATE",
|
||||||
|
"mgmt_expense_ratio": "F_COST_RATIO",
|
||||||
|
}
|
||||||
|
|
||||||
|
_HEADERS = {"X-BRIDGE-TOKEN": BRIDGE_TOKEN, "Content-Type": "application/json"}
|
||||||
|
|
||||||
|
|
||||||
|
def push_kpi_result(entity_id: int, period: str, kpi_code: str, value,
|
||||||
|
source: str = "finance-bot", remark: Optional[str] = None) -> dict:
|
||||||
|
"""调用 POST /api/cma/bot-bridge/kpi-result"""
|
||||||
|
payload = {
|
||||||
|
"entity_id": entity_id,
|
||||||
|
"kpi_code": kpi_code,
|
||||||
|
"period": period,
|
||||||
|
"value": value,
|
||||||
|
"source": source,
|
||||||
|
}
|
||||||
|
if remark:
|
||||||
|
payload["remark"] = remark
|
||||||
|
resp = httpx.post(f"{CMA_API}/kpi-result", json=payload,
|
||||||
|
headers=_HEADERS, timeout=30)
|
||||||
|
if resp.status_code >= 400:
|
||||||
|
try:
|
||||||
|
detail = resp.json().get("detail", resp.text)
|
||||||
|
except Exception:
|
||||||
|
detail = resp.text
|
||||||
|
return {"kpi_code": kpi_code, "success": False, "error": detail}
|
||||||
|
return {"kpi_code": kpi_code, "success": True, "data": resp.json()}
|
||||||
|
|
||||||
|
|
||||||
|
def push_mpm_results(entity_id: int, period: str, results: dict,
|
||||||
|
source: str = "finance-bot") -> dict:
|
||||||
|
"""MPM结果自动回填:映射字段→KPI编码,逐项推送"""
|
||||||
|
summary = {"kpi_pushed": 0, "kpi_failed": 0, "alerts": [], "items": []}
|
||||||
|
for field, kpi_code in MPM_TO_KPI_MAP.items():
|
||||||
|
if not kpi_code or results.get(field) is None:
|
||||||
|
continue
|
||||||
|
result = push_kpi_result(
|
||||||
|
entity_id, period, kpi_code, results[field], source,
|
||||||
|
remark=f"MPM自动回填[{field}] {datetime.now().strftime('%Y-%m-%d %H:%M')}",
|
||||||
|
)
|
||||||
|
summary["items"].append(result)
|
||||||
|
if result["success"]:
|
||||||
|
summary["kpi_pushed"] += 1
|
||||||
|
summary["alerts"].extend(result["data"].get("new_alerts", []))
|
||||||
|
else:
|
||||||
|
summary["kpi_failed"] += 1
|
||||||
|
return summary
|
||||||
|
|
||||||
|
|
||||||
|
def main():
|
||||||
|
parser = argparse.ArgumentParser(description="财务Bot分析结果回填CMA")
|
||||||
|
parser.add_argument("--entity-id", type=int, default=1, help="企业实体ID")
|
||||||
|
parser.add_argument("--period", default=datetime.now().strftime("%Y-%m"), help="期间 YYYY-MM")
|
||||||
|
parser.add_argument("--mpm", type=json.loads, help="MPM结果JSON (内联)")
|
||||||
|
parser.add_argument("--mpm-file", help="MPM结果JSON文件路径")
|
||||||
|
parser.add_argument("--kpi-code", help="单KPI回填: KPI编码")
|
||||||
|
parser.add_argument("--value", type=float, help="单KPI回填: 数值")
|
||||||
|
parser.add_argument("--source", default="finance-bot", help="来源Bot标识")
|
||||||
|
parser.add_argument("--ping", action="store_true", help="连通性测试")
|
||||||
|
args = parser.parse_args()
|
||||||
|
|
||||||
|
if args.ping:
|
||||||
|
try:
|
||||||
|
resp = httpx.get(f"{CMA_API}/../bot/ping", headers=_HEADERS, timeout=10)
|
||||||
|
print(f"✅ bot-bridge API连通 (HTTP {resp.status_code})")
|
||||||
|
return 0
|
||||||
|
except Exception as e:
|
||||||
|
print(f"❌ 连接失败: {e}")
|
||||||
|
return 1
|
||||||
|
|
||||||
|
if args.mpm_file:
|
||||||
|
with open(args.mpm_file, encoding="utf-8") as f:
|
||||||
|
mpm = json.load(f)
|
||||||
|
else:
|
||||||
|
mpm = args.mpm
|
||||||
|
|
||||||
|
if mpm:
|
||||||
|
summary = push_mpm_results(args.entity_id, args.period, mpm, args.source)
|
||||||
|
print(json.dumps(summary, ensure_ascii=False, indent=2))
|
||||||
|
return 0 if summary["kpi_failed"] == 0 else 1
|
||||||
|
|
||||||
|
if args.kpi_code:
|
||||||
|
if args.value is None:
|
||||||
|
print("❌ 单KPI回填需要 --value")
|
||||||
|
return 1
|
||||||
|
result = push_kpi_result(args.entity_id, args.period, args.kpi_code,
|
||||||
|
args.value, args.source)
|
||||||
|
print(json.dumps(result, ensure_ascii=False, indent=2))
|
||||||
|
return 0 if result["success"] else 1
|
||||||
|
|
||||||
|
parser.print_help()
|
||||||
|
return 1
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
sys.exit(main())
|
||||||
Reference in New Issue
Block a user