feat: ChatBI自动报告生成—周报/月报/专项

2026-07-29 18:16:37 +08:00
parent 68e40ac9c5
commit bc02d09520
3 changed files with 895 additions and 1 deletions
@@ -0,0 +1,171 @@
+#!/usr/bin/env python3
+"""
+自动经营分析报告生成脚本 — ChatBI优化P1
+
+用途：
+  1. 定时触发：cron每周五17:00调用 python3 scripts/generate_report.py --type weekly
+  2. 定时触发：cron每月1日09:00调用 python3 scripts/generate_report.py --type monthly
+  3. 事件触发：python3 scripts/generate_report.py --type special --alert-ref 123
+  4. 手动触发：python3 scripts/generate_report.py --type weekly
+
+推送通道：
+  复用现有AI简报的企微推送通道（app/utils/notifier.py）
+"""
+import sys
+import os
+import json
+import argparse
+import logging
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+from dotenv import load_dotenv
+load_dotenv('/root/cma-management/backend/.env')
+
+from app.database import get_session_local
+
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
+)
+logger = logging.getLogger("cma.report_gen")
+
+
+def generate_and_push(report_type: str, trigger_type: str = "manual", alert_ref: str = None) -> dict:
+    """生成报告并通过企微推送"""
+    import httpx
+    from datetime import datetime
+
+    api_base = os.getenv("CMA_API_BASE", "http://127.0.0.1:8010")
+    api_url = f"{api_base}/api/cma/reports/generate"
+
+    payload = {
+        "report_type": report_type,
+        "trigger_type": trigger_type,
+    }
+    if alert_ref:
+        payload["alert_ref"] = alert_ref
+
+    try:
+        # 调用 CMA 系统 API 生成报告
+        resp = httpx.post(
+            api_url,
+            json=payload,
+            timeout=30,
+        )
+        resp.raise_for_status()
+        result = resp.json()
+        logger.info(f"报告生成成功: type={report_type}, id={result.get('id')}")
+    except Exception as e:
+        logger.error(f"报告生成API调用失败: {e}")
+        return {"success": False, "error": str(e)}
+
+    # 推送到企微（复用现有通知通道）
+    markdown = result.get("markdown", "")
+    title = result.get("title", f"经营分析报告")
+
+    if not markdown:
+        logger.warning("报告内容为空，跳过推送")
+        return {"success": True, "pushed": 0, "report_id": result.get("id")}
+
+    # 通过系统通知通道推送
+    db = get_session_local()()
+    try:
+        from app.models import NotificationChannel
+        from app.utils.notifier import send_wecom_app
+
+        channels = db.query(NotificationChannel).filter(
+            NotificationChannel.enabled == True,
+            NotificationChannel.channel_type == "wecom_app",
+        ).all()
+
+        # 也推送到机器人 webhook
+        robot_channels = db.query(NotificationChannel).filter(
+            NotificationChannel.enabled == True,
+            NotificationChannel.channel_type == "wecom",
+        ).all()
+
+        pushed = 0
+        for ch in channels + robot_channels:
+            config = ch.config or {}
+            if ch.channel_type == "wecom_app":
+                result_push = send_wecom_app(
+                    corp_id=config.get("corp_id", ""),
+                    corp_secret=config.get("corp_secret", ""),
+                    agent_id=config.get("agent_id", ""),
+                    touser=config.get("touser", "@all"),
+                    title=title,
+                    content=markdown,
+                    alert_level="green",
+                )
+            elif ch.channel_type == "wecom":
+                from app.utils.notifier import send_wecom_robot
+                webhook = config.get("webhook_url", "")
+                if webhook:
+                    result_push = send_wecom_robot(
+                        webhook_url=webhook,
+                        title=title,
+                        content=markdown,
+                        alert_level="green",
+                    )
+                else:
+                    continue
+            else:
+                continue
+
+            if result_push.get("success"):
+                pushed += 1
+                logger.info(f"报告推送成功: {ch.name}")
+
+        # 更新报告状态
+        from app.models import ReportHistory
+        record = db.query(ReportHistory).filter(
+            ReportHistory.id == result.get("id")
+        ).first()
+        if record:
+            record.status = "pushed"
+            db.commit()
+
+        logger.info(f"报告推送完成: {pushed} 个渠道")
+        return {
+            "success": True,
+            "pushed": pushed,
+            "report_id": result.get("id"),
+            "title": title,
+            "markdown_length": len(markdown),
+        }
+    except Exception as e:
+        logger.error(f"报告推送异常: {e}")
+        return {"success": False, "error": str(e), "report_id": result.get("id")}
+    finally:
+        db.close()
+
+
+def main():
+    parser = argparse.ArgumentParser(description="自动经营分析报告生成")
+    parser.add_argument("--type", choices=["weekly", "monthly", "special"],
+                        default="weekly", help="报告类型")
+    parser.add_argument("--trigger", choices=["manual", "scheduled", "event"],
+                        default="manual", help="触发方式")
+    parser.add_argument("--alert-ref", type=str, default=None,
+                        help="事件触发时的预警ID")
+    parser.add_argument("--no-push", action="store_true",
+                        help="仅生成不推送")
+    args = parser.parse_args()
+
+    logger.info(f"开始生成报告: type={args.type}, trigger={args.trigger}")
+
+    result = generate_and_push(
+        report_type=args.type,
+        trigger_type=args.trigger,
+        alert_ref=args.alert_ref,
+    )
+
+    print(json.dumps(result, ensure_ascii=False, indent=2))
+
+    if not result.get("success"):
+        sys.exit(1)
+
+
+if __name__ == "__main__":
+    main()