""" 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))