Agent/examples/tencent_realtime_control/feishu_notifier.py

"""Send successful real-time control actions as a Feishu spreadsheet."""

from __future__ import annotations

import json
import logging
import os
import time
from datetime import datetime
from decimal import Decimal
from pathlib import Path
from typing import Any

import requests
from openpyxl import Workbook
from openpyxl.styles import Alignment, Font, PatternFill


ROOT = Path(__file__).resolve().parent
FEISHU_BASE_URL = "https://open.feishu.cn/open-apis"
HEADERS = [
    "日期",
    "数据分区",
    "当前CPM",
    "账户ID",
    "广告ID",
    "广告名称",
    "人群包",
    "今日消耗(元)",
    "今日曝光",
    "今日点击",
    "基础出价(元)",
    "调整前出价(元)",
    "当前出价(元)",
    "调整前开始日期",
    "调整后开始日期",
    "动作",
    "执行结果",
    "错误信息",
]

logger = logging.getLogger("tencent_realtime_control.feishu")


def _yuan(fen: int) -> float:
    return float((Decimal(fen) / Decimal("100")).quantize(Decimal("0.00")))


def _action_text(result: dict[str, Any], bid_up_ratio: Decimal) -> str:
    decision = result["decision"]
    bid_changed = bool(result.get("bid_change_needed"))
    status_changed = bool(result.get("status_change_needed"))

    if decision == "BOOST":
        percent = int((bid_up_ratio - Decimal("1")) * Decimal("100"))
        return f"出价上调{percent}%"
    if decision == "RESTORE":
        if bid_changed and status_changed:
            return "恢复基础出价并开启投放"
        if bid_changed:
            return "恢复基础出价"
        return "恢复投放"
    if decision == "PAUSE":
        target_date = str(result.get("target_begin_date") or "次日")
        return (
            f"恢复基础出价并延后至{target_date}（CPM过低）"
            if bid_changed
            else f"延后至{target_date}（CPM过低）"
        )
    if decision == "CUTOFF":
        return "恢复基础出价并关停（投放截止）" if bid_changed else "关停（投放截止）"
    return decision


def notification_rows(
    *,
    now: datetime,
    results: list[dict[str, Any]],
    bid_up_ratio: Decimal,
    observed_partition: str | None,
    observed_cpm: Decimal | None,
    test_mode: bool = False,
) -> list[list[Any]]:
    rows: list[list[Any]] = []
    for result in results:
        expected_statuses = {"planned"} if test_mode else {"success", "failed"}
        if result.get("status") not in expected_statuses:
            continue

        is_bid_change = bool(result.get("bid_change_needed"))
        is_begin_date_change = bool(result.get("begin_date_change_needed"))
        is_pause = (
            result.get("decision") in {"PAUSE", "CUTOFF"}
            and bool(result.get("status_change_needed"))
        )
        if not is_bid_change and not is_begin_date_change and not is_pause:
            continue

        rows.append(
            [
                now.strftime("%Y-%m-%d %H:%M:%S"),
                observed_partition or "",
                float(observed_cpm) if observed_cpm is not None else "",
                str(result["account_id"]),
                str(result["adgroup_id"]),
                str(result.get("adgroup_name") or ""),
                str(result.get("audience_name") or ""),
                _yuan(int(result.get("today_cost_fen") or 0)),
                int(result.get("today_impressions") or 0),
                int(result.get("today_clicks") or 0),
                _yuan(int(result["base_bid_fen"])),
                _yuan(int(result["before_bid_fen"])),
                (
                    _yuan(int(result["target_bid_fen"]))
                    if result.get("status") != "failed"
                    else ""
                ),
                str(result.get("before_begin_date") or ""),
                str(result.get("target_begin_date") or ""),
                (
                    f"【模拟】{_action_text(result, bid_up_ratio)}"
                    if test_mode
                    else _action_text(result, bid_up_ratio)
                ),
                (
                    "模拟"
                    if test_mode
                    else "成功"
                    if result.get("status") == "success"
                    else "失败"
                ),
                str(result.get("error") or ""),
            ]
        )
    return rows


def create_notification_xlsx(
    *,
    run_id: str,
    now: datetime,
    rows: list[list[Any]],
    test_mode: bool = False,
) -> Path:
    output_dir = ROOT / "outputs"
    output_dir.mkdir(parents=True, exist_ok=True)
    mode = "test" if test_mode else "apply"
    path = output_dir / (
        f"realtime_actions_{mode}_{now.strftime('%Y%m%d_%H%M%S')}_"
        f"{run_id[:8]}.xlsx"
    )

    workbook = Workbook()
    sheet = workbook.active
    sheet.title = "实时调控动作_模拟" if test_mode else "实时调控动作"
    sheet.append(HEADERS)
    for row in rows:
        sheet.append(row)

    header_fill = PatternFill("solid", fgColor="1F4E78")
    for cell in sheet[1]:
        cell.fill = header_fill
        cell.font = Font(color="FFFFFF", bold=True)
        cell.alignment = Alignment(horizontal="center", vertical="center")

    widths = [
        21, 18, 12, 14, 18, 30, 22, 16, 14, 14, 16, 16, 16, 18, 18, 38, 12, 55
    ]
    for index, width in enumerate(widths, start=1):
        sheet.column_dimensions[chr(64 + index)].width = width
    for row in sheet.iter_rows(min_row=2):
        for cell in row:
            cell.alignment = Alignment(horizontal="center", vertical="center")
    for column in ("C", "H", "K", "L", "M"):
        for cell in sheet[column][1:]:
            cell.number_format = "0.00"
    sheet.freeze_panes = "A2"
    sheet.auto_filter.ref = sheet.dimensions
    workbook.save(path)
    return path


class FeishuNotifier:
    def __init__(self) -> None:
        self.app_id = os.getenv("FEISHU_APP_ID", "").strip()
        self.app_secret = os.getenv("FEISHU_APP_SECRET", "").strip()
        self.chat_id = (
            os.getenv("RTC_FEISHU_CHAT_ID", "").strip()
            or os.getenv("FEISHU_AD_PROJECT_CHAT_ID", "").strip()
        )
        self.timeout = int(os.getenv("RTC_FEISHU_TIMEOUT_SECONDS", "30"))
        self.session = requests.Session()

    def _require_config(self) -> None:
        missing = [
            name
            for name, value in (
                ("FEISHU_APP_ID", self.app_id),
                ("FEISHU_APP_SECRET", self.app_secret),
                ("RTC_FEISHU_CHAT_ID/FEISHU_AD_PROJECT_CHAT_ID", self.chat_id),
            )
            if not value
        ]
        if missing:
            raise RuntimeError(
                f"Missing Feishu notification configuration: {', '.join(missing)}"
            )

    def _tenant_token(self) -> str:
        response = self.session.post(
            f"{FEISHU_BASE_URL}/auth/v3/tenant_access_token/internal",
            json={"app_id": self.app_id, "app_secret": self.app_secret},
            timeout=self.timeout,
        )
        response.raise_for_status()
        payload = response.json()
        if payload.get("code") != 0:
            raise RuntimeError(f"Feishu token failed: {payload}")
        return str(payload["tenant_access_token"])

    @staticmethod
    def _headers(token: str) -> dict[str, str]:
        return {"Authorization": f"Bearer {token}"}

    def _upload(self, token: str, path: Path) -> str:
        with path.open("rb") as file:
            response = self.session.post(
                f"{FEISHU_BASE_URL}/drive/v1/medias/upload_all",
                headers=self._headers(token),
                data={
                    "file_name": path.name,
                    "parent_type": "explorer",
                    "parent_node": "",
                    "size": str(path.stat().st_size),
                },
                files={
                    "file": (
                        path.name,
                        file,
                        "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
                    )
                },
                timeout=max(self.timeout, 60),
            )
        response.raise_for_status()
        payload = response.json()
        if payload.get("code") != 0:
            raise RuntimeError(f"Feishu upload failed: {payload}")
        return str(payload["data"]["file_token"])

    def _import_sheet(self, token: str, file_token: str, path: Path) -> dict[str, Any]:
        response = self.session.post(
            f"{FEISHU_BASE_URL}/drive/v1/import_tasks",
            headers={**self._headers(token), "Content-Type": "application/json"},
            json={
                "file_extension": "xlsx",
                "file_token": file_token,
                "type": "sheet",
                "file_name": path.stem,
                "point": {"mount_type": 1, "mount_key": ""},
            },
            timeout=self.timeout,
        )
        response.raise_for_status()
        payload = response.json()
        if payload.get("code") != 0:
            raise RuntimeError(f"Feishu import task failed: {payload}")
        ticket = str(payload["data"]["ticket"])

        deadline = time.monotonic() + 60
        while time.monotonic() < deadline:
            response = self.session.get(
                f"{FEISHU_BASE_URL}/drive/v1/import_tasks/{ticket}",
                headers=self._headers(token),
                timeout=self.timeout,
            )
            response.raise_for_status()
            payload = response.json()
            if payload.get("code") != 0:
                raise RuntimeError(f"Feishu import query failed: {payload}")
            result = (payload.get("data") or {}).get("result") or {}
            if result.get("job_status") == 0:
                return result
            if result.get("job_status") == 3:
                raise RuntimeError(
                    f"Feishu import failed: {result.get('job_error_msg')}"
                )
            time.sleep(2)
        raise RuntimeError("Feishu spreadsheet import timed out after 60 seconds")

    def _set_read_permission(
        self, token: str, sheet_token: str, file_type: str
    ) -> None:
        response = self.session.patch(
            f"{FEISHU_BASE_URL}/drive/v1/permissions/{sheet_token}/public",
            headers={**self._headers(token), "Content-Type": "application/json"},
            params={"type": file_type},
            json={
                "external_access_entity": "open",
                "link_share_entity": "anyone_readable",
            },
            timeout=self.timeout,
        )
        response.raise_for_status()
        payload = response.json()
        if payload.get("code") != 0:
            logger.warning("Feishu permission update failed: %s", payload)

    def _send_card(
        self,
        token: str,
        *,
        url: str,
        now: datetime,
        summary: dict[str, Any],
        test_mode: bool,
    ) -> None:
        action_counts = summary["action_counts"]
        action_text = "，".join(
            f"{name} {count} 条"
            for name, count in action_counts.items()
            if count
        )
        cpm_text = (
            f"{summary['observed_cpm']:.2f}"
            if summary.get("observed_cpm") is not None
            else "无（定时收尾）"
        )
        summary_lines = [
            f"**触发时间：** {now.strftime('%Y-%m-%d %H:%M:%S')}",
        ]
        if test_mode:
            summary_lines.append("**执行模式：** 仅模拟，未修改腾讯广告")
        summary_lines.extend(
            [
                f"**数据分区：** {summary.get('observed_partition') or '无'}",
                f"**当前 CPM：** {cpm_text}",
                f"**本轮决策：** {summary['decision']}",
                (
                    f"**影响范围：** {summary['account_count']} 个账户，"
                    f"{summary['row_count']} 条广告"
                ),
                f"**动作概要：** {action_text}",
                (
                    "**受影响广告今日消耗：** "
                    f"{summary['today_cost_yuan']:.2f} 元"
                ),
                f"**执行结果：** 成功 {summary['success_count']} 条，失败 "
                f"{summary['failure_count']} 条",
            ]
        )
        card = {
            "config": {"wide_screen_mode": True},
            "header": {
                "title": {
                    "tag": "plain_text",
                    "content": (
                        "腾讯广告实时调控通知（模拟）"
                        if test_mode
                        else "腾讯广告实时调控通知"
                    ),
                },
                "template": (
                    "orange"
                    if test_mode
                    else "red"
                    if summary["failure_count"]
                    else "blue"
                ),
            },
            "elements": [
                {
                    "tag": "div",
                    "text": {
                        "tag": "lark_md",
                        "content": "\n".join(summary_lines),
                    },
                },
                {
                    "tag": "action",
                    "actions": [
                        {
                            "tag": "button",
                            "text": {"tag": "plain_text", "content": "打开动作明细"},
                            "type": "primary",
                            "url": url,
                        }
                    ],
                },
            ],
        }
        response = self.session.post(
            f"{FEISHU_BASE_URL}/im/v1/messages",
            headers={**self._headers(token), "Content-Type": "application/json"},
            params={"receive_id_type": "chat_id"},
            json={
                "receive_id": self.chat_id,
                "msg_type": "interactive",
                "content": json.dumps(card, ensure_ascii=False),
            },
            timeout=self.timeout,
        )
        response.raise_for_status()
        payload = response.json()
        if payload.get("code") != 0:
            raise RuntimeError(f"Feishu message failed: {payload}")

    def send_text(self, text: str) -> None:
        self._require_config()
        token = self._tenant_token()
        response = self.session.post(
            f"{FEISHU_BASE_URL}/im/v1/messages",
            headers={**self._headers(token), "Content-Type": "application/json"},
            params={"receive_id_type": "chat_id"},
            json={
                "receive_id": self.chat_id,
                "msg_type": "text",
                "content": json.dumps({"text": text}, ensure_ascii=False),
            },
            timeout=self.timeout,
        )
        response.raise_for_status()
        payload = response.json()
        if payload.get("code") != 0:
            raise RuntimeError(f"Feishu text message failed: {payload}")

    def send(
        self,
        *,
        path: Path,
        now: datetime,
        summary: dict[str, Any],
        test_mode: bool = False,
    ) -> str:
        self._require_config()
        token = self._tenant_token()
        file_token = self._upload(token, path)
        result = self._import_sheet(token, file_token, path)
        url = str(result.get("url") or "")
        if not url:
            raise RuntimeError("Feishu import succeeded without spreadsheet URL")
        sheet_token = str(result.get("token") or "")
        if sheet_token:
            self._set_read_permission(
                token,
                sheet_token,
                str(result.get("type") or "sheet"),
            )
        self._send_card(
            token,
            url=url,
            now=now,
            summary=summary,
            test_mode=test_mode,
        )
        return url


def build_notification_summary(
    *,
    rows: list[list[Any]],
    results: list[dict[str, Any]],
    decision: str,
    observed_partition: str | None,
    observed_cpm: Decimal | None,
    test_mode: bool = False,
) -> dict[str, Any]:
    expected_statuses = {"planned"} if test_mode else {"success", "failed"}
    notified = [
        result
        for result in results
        if result.get("status") in expected_statuses
        and (
            result.get("bid_change_needed")
            or result.get("begin_date_change_needed")
            or (
                result.get("decision") in {"PAUSE", "CUTOFF"}
                and result.get("status_change_needed")
            )
        )
    ]
    return {
        "decision": decision,
        "observed_partition": observed_partition,
        "observed_cpm": float(observed_cpm) if observed_cpm is not None else None,
        "row_count": len(rows),
        "account_count": len({int(row["account_id"]) for row in notified}),
        "today_cost_yuan": sum(
            int(row.get("today_cost_fen") or 0) for row in notified
        )
        / 100,
        "success_count": sum(
            row.get("status") in {"success", "planned"} for row in notified
        ),
        "failure_count": sum(row.get("status") == "failed" for row in notified),
        "action_counts": {
            "上调": sum(row.get("decision") == "BOOST" for row in notified),
            "恢复基础价": sum(
                row.get("decision") == "RESTORE"
                and row.get("bid_change_needed")
                for row in notified
            ),
            "关停": sum(
                row.get("decision") == "CUTOFF"
                and row.get("status_change_needed")
                for row in notified
            ),
            "延后至次日": sum(
                row.get("decision") == "PAUSE"
                and row.get("begin_date_change_needed")
                for row in notified
            ),
        },
    }


def send_action_notification(
    *,
    run_id: str,
    now: datetime,
    results: list[dict[str, Any]],
    bid_up_ratio: Decimal,
    decision: str,
    observed_partition: str | None,
    observed_cpm: Decimal | None,
    test_mode: bool = False,
) -> dict[str, Any]:
    rows = notification_rows(
        now=now,
        results=results,
        bid_up_ratio=bid_up_ratio,
        observed_partition=observed_partition,
        observed_cpm=observed_cpm,
        test_mode=test_mode,
    )
    if not rows:
        return {"status": "no_actions", "row_count": 0}

    path = create_notification_xlsx(
        run_id=run_id,
        now=now,
        rows=rows,
        test_mode=test_mode,
    )
    summary = build_notification_summary(
        rows=rows,
        results=results,
        decision=decision,
        observed_partition=observed_partition,
        observed_cpm=observed_cpm,
        test_mode=test_mode,
    )
    enabled = os.getenv("RTC_FEISHU_NOTIFY_ENABLED", "1").strip().lower() in {
        "1",
        "true",
        "yes",
        "y",
        "on",
    }
    if not enabled:
        return {
            "status": "disabled",
            "row_count": len(rows),
            "xlsx_path": str(path),
        }

    try:
        url = FeishuNotifier().send(
            path=path,
            now=now,
            summary=summary,
            test_mode=test_mode,
        )
        return {
            "status": "sent",
            "row_count": len(rows),
            "xlsx_path": str(path),
            "url": url,
        }
    except Exception as exc:
        logger.exception("Feishu action notification failed")
        return {
            "status": "failed",
            "row_count": len(rows),
            "xlsx_path": str(path),
            "error": str(exc),
        }


def send_operator_reconciliation_notification(
    results: list[dict[str, Any]],
) -> None:
    rows = [
        result
        for result in results
        if result.get("decision") == "OPERATOR_MANUAL_OVERRIDE"
    ]
    if not rows:
        return
    details = "\n".join(
        f"- 账户 {row['account_id']} / 广告 {row['adgroup_id']}"
        for row in rows
    )
    FeishuNotifier().send_text(
        "检测到腾讯后台人工开启广告，已清除系统运营停止状态并尊重人工操作：\n"
        f"{details}"
    )