Agent/examples/auto_put_ad_mini/roi_control/sheet_approval.py

"""Poll editable ROI sheets and execute approved database action items."""

from __future__ import annotations

import logging
import os
import threading
from datetime import datetime
from typing import Any
from zoneinfo import ZoneInfo

import httpx
from openpyxl.utils import get_column_letter

from .config import RoiConfig
from .execution import execute_approved_roi_actions
from .feishu import BASE_URL, RoiFeishuPublisher
from .repository import (
    expire_pending_sheet_runs,
    finalize_sheet_run_if_resolved,
    load_actions_by_ids,
    load_pending_sheet_runs,
    load_unnotified_action_details,
    mark_action_notifications,
    record_sheet_decision,
)


SHANGHAI = ZoneInfo("Asia/Shanghai")
APPROVAL_SHEET_NAME = "小程序投流"
APPROVED_VALUES = {"批准", "approve", "approved"}
REJECTED_VALUES = {"拒绝", "reject", "rejected"}
logger = logging.getLogger("auto_put_ad_mini.roi_sheet_approval")


def parse_approval_rows(values: list[list[Any]]) -> list[dict[str, Any]]:
    if not values:
        return []
    headers = {
        str(value or "").strip(): index
        for index, value in enumerate(values[0])
        if str(value or "").strip()
    }
    required = {"审批选择", "动作幂等键"}
    missing = required - headers.keys()
    if missing:
        raise RuntimeError(f"ROI审批表缺少列: {', '.join(sorted(missing))}")
    decisions: list[dict[str, Any]] = []
    for row_number, row in enumerate(values[1:], start=2):
        approval_index = headers["审批选择"]
        key_index = headers["动作幂等键"]
        approval = str(row[approval_index] if approval_index < len(row) else "").strip()
        key = str(row[key_index] if key_index < len(row) else "").strip()
        normalized = approval.lower()
        if not key:
            continue
        if normalized in APPROVED_VALUES:
            decision = "APPROVED"
        elif normalized in REJECTED_VALUES:
            decision = "REJECTED"
        else:
            continue
        decisions.append(
            {
                "row_number": row_number,
                "idempotency_key": key,
                "decision": decision,
                "headers": headers,
            }
        )
    return decisions


class RoiSheetClient:
    def __init__(self, timeout: float = 30.0) -> None:
        self.app_id = os.getenv("FEISHU_APP_ID", "").strip()
        self.app_secret = os.getenv("FEISHU_APP_SECRET", "").strip()
        if not self.app_id or not self.app_secret:
            raise RuntimeError("FEISHU_APP_ID and FEISHU_APP_SECRET are required")
        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 _sheet_id(self, token: str, sheet_token: str) -> str:
        response = self.client.get(
            f"{BASE_URL}/sheets/v3/spreadsheets/{sheet_token}/sheets/query",
            headers=self._headers(token),
        )
        sheets = (
            self._json(response, "query ROI sheets")
            .get("data", {})
            .get("sheets", [])
        )
        target = next(
            (sheet for sheet in sheets if sheet.get("title") == APPROVAL_SHEET_NAME),
            None,
        )
        if not target:
            raise RuntimeError(f"ROI审批表缺少工作表: {APPROVAL_SHEET_NAME}")
        return str(target["sheet_id"])

    def _read_values(
        self,
        token: str,
        sheet_token: str,
        cell_range: str,
    ) -> list[list[Any]]:
        response = self.client.get(
            f"{BASE_URL}/sheets/v2/spreadsheets/{sheet_token}/values/{cell_range}",
            headers=self._headers(token),
            params={"valueRenderOption": "ToString"},
        )
        return (
            self._json(response, "read ROI approvals")
            .get("data", {})
            .get("valueRange", {})
            .get("values", [])
        ) or []

    def read_approvals(self, sheet_token: str) -> tuple[str, list[dict[str, Any]]]:
        token = self._token()
        sheet_id = self._sheet_id(token, sheet_token)
        header_rows = self._read_values(
            token,
            sheet_token,
            f"{sheet_id}!A1:ZZ1",
        )
        if not header_rows:
            return sheet_id, []
        headers = {
            str(value or "").strip(): index
            for index, value in enumerate(header_rows[0])
            if str(value or "").strip()
        }
        required = {"审批选择", "动作幂等键"}
        missing = required - headers.keys()
        if missing:
            raise RuntimeError(f"ROI审批表缺少列: {', '.join(sorted(missing))}")
        approval_letter = get_column_letter(headers["审批选择"] + 1)
        key_letter = get_column_letter(headers["动作幂等键"] + 1)
        approval_rows = self._read_values(
            token,
            sheet_token,
            f"{sheet_id}!{approval_letter}2:{approval_letter}5000",
        )
        key_rows = self._read_values(
            token,
            sheet_token,
            f"{sheet_id}!{key_letter}2:{key_letter}5000",
        )
        decisions: list[dict[str, Any]] = []
        for offset in range(max(len(approval_rows), len(key_rows))):
            approval_row = approval_rows[offset] if offset < len(approval_rows) else []
            key_row = key_rows[offset] if offset < len(key_rows) else []
            approval = str(approval_row[0] if approval_row else "").strip().lower()
            key = str(key_row[0] if key_row else "").strip()
            if not key:
                continue
            if approval in APPROVED_VALUES:
                decision = "APPROVED"
            elif approval in REJECTED_VALUES:
                decision = "REJECTED"
            else:
                continue
            decisions.append(
                {
                    "row_number": offset + 2,
                    "idempotency_key": key,
                    "decision": decision,
                    "headers": headers,
                }
            )
        return sheet_id, decisions

    def write_results(
        self,
        sheet_token: str,
        sheet_id: str,
        decisions: list[dict[str, Any]],
        actions: list[dict[str, Any]],
    ) -> None:
        by_id = {int(item["id"]): item for item in actions}
        token = self._token()
        value_ranges = []
        for decision in decisions:
            item = by_id.get(int(decision["item_id"]))
            if not item:
                continue
            headers = decision["headers"]
            status_column = headers.get("执行状态")
            result_column = headers.get("执行结果")
            if status_column is None or result_column is None:
                continue
            execution_status = str(item.get("execution_status") or "")
            status = {
                "SUCCESS": "执行成功",
                "REJECTED": "已拒绝",
                "PENDING": "待执行",
                "PREPARED": "执行中",
            }.get(execution_status, "执行失败" if execution_status else "")
            result = str(item.get("error_message") or item.get("skip_reason") or "")
            start = get_column_letter(status_column + 1)
            end = get_column_letter(result_column + 1)
            value_ranges.append(
                {
                    "range": f"{sheet_id}!{start}{decision['row_number']}:{end}{decision['row_number']}",
                    "values": [[status, result]],
                }
            )
        if not value_ranges:
            return
        response = self.client.post(
            f"{BASE_URL}/sheets/v2/spreadsheets/"
            f"{sheet_token}/values_batch_update",
            headers={**self._headers(token), "Content-Type": "application/json"},
            json={"valueRanges": value_ranges},
        )
        self._json(response, "write ROI execution results")


class RoiSheetApprovalService:
    def __init__(self) -> None:
        self.config = RoiConfig.from_env()
        self.lock_name = os.getenv("RTC_DB_LOCK_NAME", "tencent_realtime_control")
        self.client = RoiSheetClient()
        self.stop_event = threading.Event()

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

    def process_once(self, now: datetime | None = None) -> dict[str, int]:
        current = now or datetime.now(SHANGHAI)
        expired = expire_pending_sheet_runs(current)
        run_count = 0
        decision_count = 0
        failed_runs = 0
        for run in load_pending_sheet_runs(current):
            run_count += 1
            try:
                decision_count += self._process_run(run, current)
            except Exception:
                failed_runs += 1
                logger.exception(
                    "Skip incompatible or unavailable ROI sheet run=%s",
                    run["run_id"],
                )
        self._notify_results(current)
        return {
            "runs": run_count,
            "decisions": decision_count,
            "expired": expired,
            "failed_runs": failed_runs,
        }

    def _process_run(self, run: dict[str, Any], now: datetime) -> int:
        sheet_id, sheet_decisions = self.client.read_approvals(run["sheet_token"])
        approved_ids: list[int] = []
        processed_decisions: list[dict[str, Any]] = []
        changed_count = 0
        for decision in sheet_decisions:
            item = record_sheet_decision(
                run_id=run["run_id"],
                idempotency_key=decision["idempotency_key"],
                decision=decision["decision"],
                sheet_row_number=decision["row_number"],
                now=now,
            )
            if not item:
                continue
            changed_count += int(bool(item.get("decision_changed")))
            decision["item_id"] = int(item["id"])
            processed_decisions.append(decision)
            if item.get("approval_status") == "APPROVED":
                approved_ids.append(int(item["id"]))
        if approved_ids:
            execute_approved_roi_actions(
                approved_ids,
                now=now,
                lock_name=self.lock_name,
            )
        finalize_sheet_run_if_resolved(run["run_id"], now=now)
        if processed_decisions:
            actions = load_actions_by_ids(
                [decision["item_id"] for decision in processed_decisions]
            )
            try:
                self.client.write_results(
                    run["sheet_token"],
                    sheet_id,
                    processed_decisions,
                    actions,
                )
            except Exception:
                logger.exception("Failed to write ROI execution result back to sheet")
        return changed_count

    def _notify_results(self, now: datetime) -> None:
        rows = load_unnotified_action_details()
        if not rows:
            return
        item_ids = [int(row["id"]) for row in rows]
        publisher = RoiFeishuPublisher()
        try:
            publisher.send_execution_results(rows)
        except Exception as exc:
            mark_action_notifications(item_ids, notified_at=None, error=str(exc))
            raise
        finally:
            publisher.close()
        mark_action_notifications(item_ids, notified_at=now, error=None)

    def run_forever(self) -> None:
        logger.info(
            "ROI sheet approval poller started interval=%ss",
            self.config.sheet_approval_poll_seconds,
        )
        while not self.stop_event.is_set():
            try:
                result = self.process_once()
                if result["runs"] or result["expired"]:
                    logger.info("ROI sheet approval cycle result=%s", result)
            except Exception:
                logger.exception("ROI sheet approval cycle failed")
            self.stop_event.wait(self.config.sheet_approval_poll_seconds)

    def start(self) -> threading.Thread:
        thread = threading.Thread(
            target=self.run_forever,
            name="roi-sheet-approval",
            daemon=True,
        )
        thread.start()
        return thread