187 lines
6.9 KiB
Python
187 lines
6.9 KiB
Python
"""
|
|
CMA管理会计OS - 研学BOT桥接脚本
|
|
供研学BOT调用，获取CMA系统数据进行AI分析
|
|
|
|
用法:
|
|
python3 cma_bot_bridge.py --action dashboard 获取驾驶舱摘要数据
|
|
python3 cma_bot_bridge.py --action kpis 获取全部KPI列表
|
|
python3 cma_bot_bridge.py --action alerts 获取预警列表
|
|
python3 cma_bot_bridge.py --action brief 获取CEO简报数据
|
|
"""
|
|
import os, sys, json, hashlib, argparse
|
|
from datetime import datetime
|
|
from dotenv import load_dotenv
|
|
|
|
load_dotenv()
|
|
|
|
CMA_API = "http://127.0.0.1:8010/api/cma"
|
|
ADMIN_USER = "admin"
|
|
ADMIN_PASS = "cma2026"
|
|
|
|
# ── 缓存token ──
|
|
_token_cache = {"token": None, "expires": 0}
|
|
|
|
|
|
def _get_token() -> str:
|
|
"""登录获取token"""
|
|
import httpx
|
|
now = datetime.now().timestamp()
|
|
if _token_cache["token"] and now < _token_cache["expires"]:
|
|
return _token_cache["token"]
|
|
|
|
resp = httpx.post(f"{CMA_API}/auth/login", json={
|
|
"username": ADMIN_USER,
|
|
"password": ADMIN_PASS,
|
|
}, timeout=10)
|
|
data = resp.json()
|
|
_token_cache["token"] = data["token"]
|
|
_token_cache["expires"] = now + 3500 # token有效期1小时，提前100秒刷新
|
|
return data["token"]
|
|
|
|
|
|
def get_headers() -> dict:
|
|
return {"Authorization": f"Bearer {_get_token()}"}
|
|
|
|
|
|
def get_dashboard_summary() -> dict:
|
|
"""获取驾驶舱摘要数据"""
|
|
import httpx
|
|
headers = get_headers()
|
|
|
|
# 获取KPI列表
|
|
resp = httpx.get(f"{CMA_API}/kpis", headers=headers, params={"page_size": 100}, timeout=10)
|
|
kpis = resp.json()
|
|
|
|
# 获取预警
|
|
resp2 = httpx.get(f"{CMA_API}/alerts", headers=headers, params={"page_size": 50}, timeout=10)
|
|
alerts = resp2.json()
|
|
|
|
return {
|
|
"kpi_count": kpis.get("total", 0),
|
|
"kpis": kpis.get("data", []),
|
|
"alert_count": alerts.get("total", 0),
|
|
"alerts": alerts.get("data", []),
|
|
"fetched_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
|
|
}
|
|
|
|
|
|
def get_kpis() -> list:
|
|
"""获取所有KPI"""
|
|
import httpx
|
|
headers = get_headers()
|
|
resp = httpx.get(f"{CMA_API}/kpis", headers=headers, params={"page_size": 200}, timeout=10)
|
|
return resp.json().get("data", [])
|
|
|
|
|
|
def get_alerts() -> list:
|
|
"""获取预警列表"""
|
|
import httpx
|
|
headers = get_headers()
|
|
resp = httpx.get(f"{CMA_API}/alerts", headers=headers, params={"page_size": 50}, timeout=10)
|
|
return resp.json().get("data", [])
|
|
|
|
|
|
def get_strategic_maps() -> list:
|
|
"""获取战略地图"""
|
|
import httpx
|
|
headers = get_headers()
|
|
resp = httpx.get(f"{CMA_API}/maps", headers=headers, params={"page_size": 20}, timeout=10)
|
|
return resp.json().get("data", [])
|
|
|
|
|
|
def format_for_ai(action: str) -> str:
|
|
"""格式化数据供AI分析"""
|
|
if action == "dashboard":
|
|
data = get_dashboard_summary()
|
|
lines = [f"📊 CMA管理会计OS - 系统摘要 ({data['fetched_at']})"]
|
|
lines.append(f"")
|
|
lines.append(f"KPI指标总数: {data['kpi_count']}")
|
|
lines.append(f"待处理预警: {data['alert_count']}")
|
|
if data['kpis']:
|
|
lines.append(f"\n--- KPI列表 ---")
|
|
for k in data['kpis']:
|
|
dim_icon = {"finance": "💰", "customer": "👥", "process": "⚙️", "learning": "📚"}
|
|
icon = dim_icon.get(k.get("dimension", ""), "📌")
|
|
lines.append(f"{icon} {k.get('kpi_name','')} ({k.get('kpi_code','')}) - {k.get('dimension','')}")
|
|
lines.append(f" 目标: {k.get('target_value','未设置')}{k.get('unit','')}")
|
|
lines.append(f" 公式: {k.get('formula','') or '无'}")
|
|
if data['alerts']:
|
|
lines.append(f"\n--- 预警列表 ---")
|
|
for a in data['alerts']:
|
|
lines.append(f"⚠️ [{a.get('alert_level','')}] {a.get('message','')}")
|
|
return "\n".join(lines)
|
|
|
|
elif action == "kpis":
|
|
kpis = get_kpis()
|
|
lines = [f"📋 CMA KPI字典 ({len(kpis)}个)"]
|
|
for k in kpis:
|
|
lines.append(f"\n- {k.get('kpi_name','')} ({k.get('kpi_code','')})")
|
|
lines.append(f" 维度: {k.get('dimension','')} | 目标: {k.get('target_value','')}{k.get('unit','')}")
|
|
return "\n".join(lines)
|
|
|
|
elif action == "alerts":
|
|
alerts = get_alerts()
|
|
lines = [f"🚨 CMA预警列表 ({len(alerts)}条待处理)"]
|
|
for a in alerts:
|
|
lines.append(f"\n[{a.get('alert_level','')}] {a.get('message','')}")
|
|
lines.append(f" 时间: {a.get('created_at','')}")
|
|
return "\n".join(lines)
|
|
|
|
elif action == "brief":
|
|
data = get_dashboard_summary()
|
|
lines = [
|
|
f"CMA管理会计OS - 快速简报",
|
|
f"采集时间: {data['fetched_at']}",
|
|
f"",
|
|
f"📊 概况: {data['kpi_count']}个KPI, {data['alert_count']}条预警",
|
|
]
|
|
if data['kpis']:
|
|
lines.append(f"\n├─ 财务维度:")
|
|
for k in data['kpis']:
|
|
if k.get("dimension") == "finance":
|
|
lines.append(f"│ {k['kpi_name']}: 目标={k.get('target_value','')}")
|
|
lines.append(f"\n├─ 客户维度:")
|
|
for k in data['kpis']:
|
|
if k.get("dimension") == "customer":
|
|
lines.append(f"│ {k['kpi_name']}: 目标={k.get('target_value','')}")
|
|
lines.append(f"\n├─ 流程维度:")
|
|
for k in data['kpis']:
|
|
if k.get("dimension") == "process":
|
|
lines.append(f"│ {k['kpi_name']}: 目标={k.get('target_value','')}")
|
|
lines.append(f"\n└─ 学习成长维度:")
|
|
for k in data['kpis']:
|
|
if k.get("dimension") == "learning":
|
|
lines.append(f" {k['kpi_name']}: 目标={k.get('target_value','')}")
|
|
return "\n".join(lines)
|
|
|
|
else:
|
|
return f"未知action: {action}"
|
|
|
|
|
|
if __name__ == "__main__":
|
|
parser = argparse.ArgumentParser(description="CMA Bot Bridge")
|
|
parser.add_argument("--action", choices=["dashboard", "kpis", "alerts", "brief", "test"],
|
|
default="dashboard", help="数据action")
|
|
parser.add_argument("--format", choices=["json", "text"], default="text", help="输出格式")
|
|
args = parser.parse_args()
|
|
|
|
if args.action == "test":
|
|
# 连通性测试
|
|
try:
|
|
token = _get_token()
|
|
print(f"✅ CMA API连通成功, token前缀: {token[:10]}...")
|
|
sys.exit(0)
|
|
except Exception as e:
|
|
print(f"❌ CMA API连接失败: {e}")
|
|
sys.exit(1)
|
|
|
|
if args.format == "json":
|
|
if args.action == "dashboard":
|
|
print(json.dumps(get_dashboard_summary(), ensure_ascii=False, indent=2))
|
|
elif args.action == "kpis":
|
|
print(json.dumps(get_kpis(), ensure_ascii=False, indent=2))
|
|
elif args.action == "alerts":
|
|
print(json.dumps(get_alerts(), ensure_ascii=False, indent=2))
|
|
else:
|
|
print(format_for_ai(args.action))
|