cma-management/backend/scripts/daily_push.py

"""每日数据找人推送 — 路线图R2 (2026-08-30) / 分企业推送 (2026-09-03)

北极星③：主动推送扩大 —— 异常 + 机会两类。
- 异常类：待处理预警（kpi_alerts pending，按 entity_id 隔离）
- 机会类：KPI向好 / 预算余量 / 预测上行（opportunity_detector，按 entity_id 隔离）
复用企微通道 8800/send（公司群中继服务）。

v2 分企业版（2026-09-03，任总指示：陕西酣客与陕西博海分开汇报避免混淆）：
- 默认遍历 entities 表 status='active' 的全部企业，每个企业独立组装+推送
- 消息标题带企业标识：【陕西酣客】/【陕西博海】…，一眼分清
- --entity-id 传入时只推该企业（向后兼容）

用法: /root/cma-management/backend/venv/bin/python3 scripts/daily_push.py [--dry-run] [--entity-id N]
cron: 15 9 * * *  (alert_generator 9:00 之后)
"""
import sys
import os
import json
import logging
import argparse
from datetime import datetime

sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from app.database import get_session_local
from app.models import Entity, KPIDefinition, KPIAlert
from scripts.opportunity_detector import detect_all, flatten

logger = logging.getLogger("cma.daily_push")

RELAY_URL = "http://127.0.0.1:8800/send"
SOURCE = "管理会计OS"


def entity_display(e) -> str:
    """企业展示名：优先 short_name（酣客/博海），无则用 name"""
    if getattr(e, "short_name", None):
        return e.short_name
    return e.name


def collect_exceptions(db, entity_id: int, limit: int = 10) -> list:
    """异常类：待处理预警（red/yellow），限定企业"""
    out = []
    alerts = db.query(KPIAlert).filter(
        KPIAlert.status == "pending",
        KPIAlert.alert_level.in_(["red", "yellow"]),
        KPIAlert.entity_id == entity_id,
    ).order_by(KPIAlert.created_at.desc()).limit(limit).all()
    for a in alerts:
        k = db.query(KPIDefinition).filter(KPIDefinition.id == a.kpi_id).first()
        kpi_name = k.kpi_name if k else f"KPI#{a.kpi_id}"
        icon = "🔴" if a.alert_level == "red" else "🟡"
        out.append({
            "type": "exception",
            "title": f"{icon} {kpi_name} 预警",
            "detail": f"({a.alert_level}) {a.alert_message}",
            "kpi_id": a.kpi_id,
            "kpi_name": kpi_name,
            "period": "",
        })
    return out


def build_message(exceptions: list, opportunities: list, entity_name: str = "") -> str:
    """组装 markdown 推送内容（标题带企业标识）"""
    now = datetime.now().strftime("%Y-%m-%d %H:%M")
    tag = f"【{entity_name}】" if entity_name else ""
    lines = [f"## 📊 {tag}管理会计OS · 每日经营播报", f"**{now}**", ""]

    lines.append("### ⚠️ 异常关注")
    if exceptions:
        for e in exceptions:
            lines.append(f"- {e['title']}")
            lines.append(f"  {e['detail']}")
    else:
        lines.append("- 今日无待处理预警 ✅")

    lines.append("")
    lines.append("### 🎯 机会发现")
    if opportunities:
        for o in opportunities:
            lines.append(f"- {o['title']}")
            lines.append(f"  {o['detail']}")
    else:
        lines.append("- 今日暂无显著机会")

    lines.append("")
    lines.append("---")
    lines.append("💡 数据找人：异常要处理，机会要把握。详情见 CMA 系统。")
    return "\n".join(lines)


def push_wecom(msg: str) -> dict:
    """通过8800中继推送企微"""
    import urllib.request
    import urllib.parse
    data = urllib.parse.urlencode({
        "msg": msg,
        "source": SOURCE,
        "msgtype": "markdown",
    }).encode("utf-8")
    req = urllib.request.Request(RELAY_URL, data=data,
                                 headers={"Content-Type": "application/x-www-form-urlencoded"})
    try:
        with urllib.request.urlopen(req, timeout=15) as resp:
            result = json.loads(resp.read().decode("utf-8"))
            return result
    except Exception as e:
        return {"ok": False, "error": f"推送异常: {e}"}


def resolve_targets(db, entity_id) -> list:
    """--entity-id 传入 → 单企业；否则全部 active 企业（自动适配未来新增企业）"""
    if entity_id is not None:
        e = db.query(Entity).filter(Entity.id == entity_id).first()
        if not e:
            raise SystemExit(f"entity_id={entity_id} 不存在")
        return [e]
    entities = db.query(Entity).filter(Entity.status == "active").order_by(Entity.id).all()
    if not entities:
        # 兜底：无 active 企业时退回 entity 1（兼容旧默认行为），不静默空跑
        e = db.query(Entity).filter(Entity.id == 1).first()
        if not e:
            raise SystemExit("entities 表无 active 企业且无 entity 1，无法推送")
        logger.warning("无 active 企业，退回 entity 1")
        return [e]
    return entities


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--dry-run", action="store_true", help="只打印不推送")
    parser.add_argument("--entity-id", type=int, default=None,
                        help="只推指定企业；不传则遍历全部 active 企业")
    args = parser.parse_args()

    db = get_session_local()()
    try:
        targets = resolve_targets(db, args.entity_id)
        logger.info(f"推送目标企业: {[(e.id, entity_display(e)) for e in targets]}")

        failures = 0
        for e in targets:
            # 异常隔离：单个企业失败不中断其他企业
            try:
                entity_name = entity_display(e)
                exceptions = collect_exceptions(db, e.id)
                opportunities = flatten(detect_all(db, e.id))
                msg = build_message(exceptions, opportunities, entity_name)

                if args.dry_run:
                    print("=" * 60)
                    print(f"企业: {e.name} (entity_id={e.id})")
                    print("=" * 60)
                    print(msg)
                    print(f"\n[DRY-RUN][{entity_name}] 异常{len(exceptions)}条 / 机会{len(opportunities)}条")
                    continue

                result = push_wecom(msg)
                ok = result.get("ok", False)
                print(f"[{entity_name}] 推送结果: {json.dumps(result, ensure_ascii=False)}")
                print(f"[{entity_name}] 统计: 异常{len(exceptions)}条 / 机会{len(opportunities)}条")
                if not ok:
                    failures += 1
            except Exception as exc:
                failures += 1
                logger.exception(f"企业 {getattr(e, 'id', '?')} 推送失败: {exc}")

        if failures and not args.dry_run:
            raise SystemExit(f"{failures} 个企业推送失败")
    finally:
        db.close()


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    main()