wecom-fullstackandHermes CI Fix
a9dcbf22bc
- knowledge_service: 移除 os.getenv 默认值占位，改 _require_deepseek_key() 必填，缺 env 抛 DeepSeekKeyMissing
- knowledge API: generate/{daily,weekly,monthly} 缺 Key 返回 503（可观测，不再静默走模板兜底）
- finance_bot / verify_zxt_vouchers_json / bot_bridge_push: 凭据改 env 优先（CMA_BOT_KEY_* / CMA_BOT_API_KEYS / CMA_BRIDGE_TOKEN），历史共享键仅作兼容期兜底并告警
- tests: 新增 5 例回归（缺 Key 服务层抛错、API 503；有 Key 正常 200）
(cherry picked from commit 859bd8276d7c5aad4bf10ee3399b6cccc188f064)
110 lines
3.5 KiB
Python
110 lines
3.5 KiB
Python
#!/usr/bin/env python3
|
||
"""财务BOT — CMA财务日报脚本
|
||
每天9点自动生成财务摘要，直接返回文本给Hermes cron递送
|
||
"""
|
||
import sys, os, json
|
||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||
|
||
# 直接调用BOT API
|
||
CMA_BOT_API = "http://127.0.0.1:8010/api/cma/bot"
|
||
|
||
# 凭据来源优先级：CMA_BOT_KEY_FINANCE（单键）→ CMA_BOT_API_KEYS（JSON 注册表，取 role=finance）
|
||
# 兼容期兜底：历史共享键（P3 待轮换，轮换后删除本兜底）
|
||
_LEGACY_SHARED_KEY = "cma-bot-finance-2026"
|
||
|
||
|
||
def _resolve_api_key() -> str:
|
||
key = (os.getenv("CMA_BOT_KEY_FINANCE") or "").strip()
|
||
if key:
|
||
return key
|
||
raw = (os.getenv("CMA_BOT_API_KEYS") or "").strip()
|
||
if raw:
|
||
try:
|
||
for k, v in json.loads(raw).items():
|
||
if isinstance(v, dict) and v.get("role") == "finance":
|
||
return k
|
||
except Exception:
|
||
pass
|
||
print("[WARN] 未配置 CMA_BOT_KEY_FINANCE / CMA_BOT_API_KEYS，回退历史共享键（待轮换）", file=sys.stderr)
|
||
return _LEGACY_SHARED_KEY
|
||
|
||
|
||
API_KEY = _resolve_api_key()
|
||
|
||
import httpx
|
||
from datetime import datetime
|
||
|
||
def fetch(path, params=None):
|
||
url = f"{CMA_BOT_API}{path}"
|
||
r = httpx.get(url, headers={"X-BOT-KEY": API_KEY}, params=params, timeout=10)
|
||
r.raise_for_status()
|
||
return r.json()
|
||
|
||
def format_finance_brief(data: dict) -> str:
|
||
"""格式化财务简报"""
|
||
lines = []
|
||
lines.append(f"📊 财务BOT日报 | {datetime.now().strftime('%Y-%m-%d %H:%M')}")
|
||
lines.append("")
|
||
|
||
stats = data.get("overview", {})
|
||
lines.append(f"📌 系统概览：{stats.get('kpis', 0)}个KPI | "
|
||
f"{stats.get('alerts_open', 0)}条预警 | "
|
||
f"{stats.get('budget_plans', 0)}个预算计划")
|
||
|
||
# 财务维度KPI
|
||
kpis = data.get("kpis", [])
|
||
fin_kpis = [k for k in kpis if k.get("dimension") == "finance"]
|
||
if fin_kpis:
|
||
lines.append("")
|
||
lines.append("💰 财务维度KPI：")
|
||
for k in fin_kpis:
|
||
line = f" • {k['name']}（{k['code']}）"
|
||
target = k.get("target")
|
||
unit = k.get("unit", "")
|
||
if target is not None:
|
||
line += f" 目标{target}{unit}"
|
||
lines.append(line)
|
||
|
||
# 预警
|
||
alerts = data.get("alerts", [])
|
||
if alerts:
|
||
lines.append("")
|
||
lines.append("🚨 待处理预警：")
|
||
for a in alerts[:5]:
|
||
level_icon = {"red": "🔴", "yellow": "🟡", "green": "🟢"}
|
||
icon = level_icon.get(a.get("level", ""), "⚠️")
|
||
lines.append(f" {icon} [{a['level']}] {a['message']}")
|
||
|
||
# 预算异常
|
||
budget = data.get("budget", [])
|
||
if budget:
|
||
lines.append("")
|
||
lines.append("📋 预算计划：共{}条".format(len(budget)))
|
||
|
||
# 行动方案
|
||
actions = data.get("actions", [])
|
||
if actions:
|
||
lines.append("")
|
||
lines.append("📋 进行中改善行动：")
|
||
for a in actions[:3]:
|
||
prog = a.get("progress", 0)
|
||
bar = "▓" * (prog // 10) + "░" * (10 - prog // 10)
|
||
lines.append(f" {bar} {a['title']}（{prog}%）")
|
||
|
||
lines.append("")
|
||
lines.append("💡 输入「财务分析」获取详细解读")
|
||
return "\n".join(lines)
|
||
|
||
|
||
def main():
|
||
try:
|
||
data = fetch("/query", {"q": "all"})
|
||
report = format_finance_brief(data)
|
||
print(report)
|
||
except Exception as e:
|
||
print(f"❌ 财务BOT获取数据失败: {e}")
|
||
sys.exit(1)
|
||
|
||
if __name__ == "__main__":
|
||
main()
|