88 lines
2.7 KiB
Python
88 lines
2.7 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"
|
||
API_KEY = "cma-bot-finance-2026"
|
||
|
||
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()
|