cma-management/backend/scripts/generate_report.py

#!/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()