Agent/examples/auto_put_ad_mini/roi_control/feishu.py

"""Publish editable ROI workbooks and row-level approval instructions."""

from __future__ import annotations

import json
import os
import time
from pathlib import Path
from typing import Any

import httpx


BASE_URL = "https://open.feishu.cn/open-apis"


class RoiFeishuPublisher:
    def __init__(
        self,
        timeout: float = 30.0,
        *,
        require_chat_ids: bool = True,
    ) -> None:
        self.app_id = os.getenv("FEISHU_APP_ID", "").strip()
        self.app_secret = os.getenv("FEISHU_APP_SECRET", "").strip()
        primary_chat_id = (
            os.getenv("ROI_FEISHU_CHAT_ID", "").strip()
            or os.getenv("FEISHU_AD_PROJECT_CHAT_ID", "").strip()
            or os.getenv("RTC_COMMAND_CHAT_ID", "").strip()
        )
        operator_chat_id = os.getenv("FEISHU_OPERATOR_CHAT_ID", "").strip()
        self.chat_ids = list(
            dict.fromkeys(
                chat_id
                for chat_id in (primary_chat_id, operator_chat_id)
                if chat_id
            )
        )
        required = {
            "FEISHU_APP_ID": self.app_id,
            "FEISHU_APP_SECRET": self.app_secret,
        }
        if require_chat_ids:
            required[
                "ROI_FEISHU_CHAT_ID/FEISHU_AD_PROJECT_CHAT_ID/"
                "RTC_COMMAND_CHAT_ID/FEISHU_OPERATOR_CHAT_ID"
            ] = self.chat_ids
        missing = [name for name, value in required.items() if not value]
        if missing:
            raise RuntimeError(f"Missing Feishu configuration: {', '.join(missing)}")
        self.client = httpx.Client(timeout=timeout)

    def close(self) -> None:
        self.client.close()

    @staticmethod
    def _json(response: httpx.Response, action: str) -> dict[str, Any]:
        response.raise_for_status()
        payload = response.json()
        if payload.get("code") != 0:
            raise RuntimeError(f"{action} failed: {payload.get('msg', payload)}")
        return payload

    def _token(self) -> str:
        response = self.client.post(
            f"{BASE_URL}/auth/v3/tenant_access_token/internal",
            json={"app_id": self.app_id, "app_secret": self.app_secret},
        )
        return self._json(response, "get tenant token")["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 handle:
            response = self.client.post(
                f"{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, handle, "application/octet-stream")},
                timeout=60,
            )
        return self._json(response, "upload ROI workbook")["data"]["file_token"]

    def _import_sheet(self, token: str, file_token: str, path: Path) -> str:
        response = self.client.post(
            f"{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": ""},
            },
        )
        return self._json(response, "create ROI sheet import")["data"]["ticket"]

    def _wait_import(self, token: str, ticket: str) -> dict[str, Any]:
        deadline = time.monotonic() + 60
        while time.monotonic() < deadline:
            response = self.client.get(
                f"{BASE_URL}/drive/v1/import_tasks/{ticket}",
                headers=self._headers(token),
            )
            result = (
                self._json(response, "wait ROI sheet import")
                .get("data", {})
                .get("result", {})
            )
            if result.get("job_status") == 0:
                return result
            if result.get("job_status") == 3:
                raise RuntimeError(
                    f"ROI sheet import failed: {result.get('job_error_msg', 'unknown')}"
                )
            time.sleep(2)
        raise RuntimeError("ROI sheet import timed out after 60 seconds")

    def _set_editable_link(self, token: str, sheet_token: str) -> None:
        response = self.client.patch(
            f"{BASE_URL}/drive/v1/permissions/{sheet_token}/public",
            headers={**self._headers(token), "Content-Type": "application/json"},
            params={"type": "sheet"},
            json={
                "external_access_entity": "open",
                "link_share_entity": "anyone_editable",
            },
        )
        self._json(response, "set ROI sheet editable permission")

    def _send_card(
        self,
        token: str,
        *,
        run_id: str,
        batch_name: str,
        chat_id: str,
        url: str,
        summary: str,
        requires_approval: bool,
    ) -> str:
        if not requires_approval:
            instructions = "\n\n本批次没有可执行动作，无需审批。"
        else:
            instructions = (
                "\n\n请在有效期内打开表格，取消隐藏小程序三日汇总表的黄色"
                "【审批选择】列后逐行选择批准或拒绝。批准即为最终确认，系统自动执行。"
            )
        card = {
            "config": {"wide_screen_mode": True},
            "header": {
                "template": "orange" if requires_approval else "blue",
                "title": {"tag": "plain_text", "content": "日级 ROI 调控"},
            },
            "elements": [
                {
                    "tag": "div",
                    "text": {
                        "tag": "lark_md",
                        "content": (
                            f"批次：**{batch_name}**\n"
                            f"审批ID：`{run_id}`\n{summary}{instructions}"
                        ),
                    },
                },
                {"tag": "hr"},
                {
                    "tag": "action",
                    "actions": [
                        {
                            "tag": "button",
                            "type": "primary",
                            "text": {"tag": "plain_text", "content": "查看 ROI 明细"},
                            "url": url,
                        }
                    ],
                },
            ],
        }
        response = self.client.post(
            f"{BASE_URL}/im/v1/messages",
            headers={**self._headers(token), "Content-Type": "application/json"},
            params={"receive_id_type": "chat_id"},
            json={
                "receive_id": chat_id,
                "msg_type": "interactive",
                "content": json.dumps(card, ensure_ascii=False),
            },
        )
        return self._json(response, "send ROI batch card")["data"]["message_id"]

    def send_execution_results(self, rows: list[dict[str, Any]]) -> None:
        if not rows:
            return
        token = self._token()
        for offset in range(0, len(rows), 15):
            chunk = rows[offset : offset + 15]
            details = []
            for row in chunk:
                cost = float(row.get("cost") or 0)
                roi = float(row.get("roi") or 0)
                result = str(row.get("execution_status") or "")
                error = str(row.get("error_message") or row.get("skip_reason") or "")
                details.append(
                    f"- 账户 `{row['account_id']}` / 广告 `{row['adgroup_id']}` / "
                    f"创意 `{row.get('dynamic_creative_id') or '-'}`\n"
                    f"  {row.get('adgroup_name') or ''} | 窗口成本 {cost:.2f} 元 | "
                    f"预测ROI {roi:.3f} | **{result}**"
                    + (f" | {error[:160]}" if error else "")
                )
            success_count = sum(
                row.get("execution_status") == "SUCCESS" for row in chunk
            )
            content = (
                f"本次处理 {len(chunk)} 条，成功 {success_count} 条，"
                f"其他 {len(chunk) - success_count} 条。\n\n"
                + "\n".join(details)
            )
            card = {
                "config": {"wide_screen_mode": True},
                "header": {
                    "template": "blue" if success_count == len(chunk) else "orange",
                    "title": {"tag": "plain_text", "content": "日级 ROI 审批执行结果"},
                },
                "elements": [
                    {"tag": "div", "text": {"tag": "lark_md", "content": content}},
                    {
                        "tag": "action",
                        "actions": [
                            {
                                "tag": "button",
                                "type": "primary",
                                "text": {"tag": "plain_text", "content": "查看审批表"},
                                "url": str(chunk[0].get("sheet_url") or ""),
                            }
                        ],
                    },
                ],
            }
            for chat_id in self.chat_ids:
                response = self.client.post(
                    f"{BASE_URL}/im/v1/messages",
                    headers={**self._headers(token), "Content-Type": "application/json"},
                    params={"receive_id_type": "chat_id"},
                    json={
                        "receive_id": chat_id,
                        "msg_type": "interactive",
                        "content": json.dumps(card, ensure_ascii=False),
                    },
                )
                self._json(response, "send ROI execution result")

    def send_service_alert(
        self,
        *,
        title: str,
        content: str,
        chat_id: str | None = None,
    ) -> str:
        target_chat_id = (
            (chat_id or "").strip()
            or os.getenv("ROI_FAILURE_FEISHU_CHAT_ID", "").strip()
            or os.getenv("FEISHU_OPERATOR_CHAT_ID", "").strip()
        )
        if not target_chat_id:
            raise RuntimeError("Missing ROI failure alert chat_id")
        card = {
            "config": {"wide_screen_mode": True},
            "header": {
                "template": "red",
                "title": {"tag": "plain_text", "content": title},
            },
            "elements": [
                {
                    "tag": "div",
                    "text": {"tag": "lark_md", "content": content},
                }
            ],
        }
        token = self._token()
        response = self.client.post(
            f"{BASE_URL}/im/v1/messages",
            headers={**self._headers(token), "Content-Type": "application/json"},
            params={"receive_id_type": "chat_id"},
            json={
                "receive_id": target_chat_id,
                "msg_type": "interactive",
                "content": json.dumps(card, ensure_ascii=False),
            },
        )
        return self._json(response, "send ROI service failure alert")["data"][
            "message_id"
        ]

    def publish(
        self,
        path: Path,
        *,
        run_id: str,
        batch_name: str,
        summary: str,
        requires_approval: bool,
    ) -> dict[str, str]:
        imported = self.upload_workbook(path)
        token = self._token()
        message_ids = [
            self._send_card(
                token,
                run_id=run_id,
                batch_name=batch_name,
                chat_id=chat_id,
                url=imported["url"],
                summary=summary,
                requires_approval=requires_approval,
            )
            for chat_id in self.chat_ids
        ]
        return {
            **imported,
            "message_id": message_ids[0],
        }

    def upload_workbook(self, path: Path) -> dict[str, str]:
        """Upload one workbook as an editable online sheet without notifying chats."""

        if not path.is_file():
            raise FileNotFoundError(path)
        token = self._token()
        file_token = self._upload(token, path)
        ticket = self._import_sheet(token, file_token, path)
        result = self._wait_import(token, ticket)
        url = str(result.get("url") or "")
        sheet_token = str(result.get("token") or "")
        if not url or not sheet_token:
            raise RuntimeError("ROI sheet import returned no URL/token")
        self._set_editable_link(token, sheet_token)
        return {
            "url": url,
            "sheet_token": sheet_token,
        }