"""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_NAMES = ( "小程序创意级三日汇总", "小程序广告级三日汇总", ) LEGACY_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_ids(self, token: str, sheet_token: str) -> list[tuple[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", []) ) by_title = { str(sheet.get("title") or ""): str(sheet["sheet_id"]) for sheet in sheets } targets = [ (title, by_title[title]) for title in APPROVAL_SHEET_NAMES if title in by_title ] if not targets and LEGACY_APPROVAL_SHEET_NAME in by_title: targets.append( (LEGACY_APPROVAL_SHEET_NAME, by_title[LEGACY_APPROVAL_SHEET_NAME]) ) if not targets: expected = "、".join((*APPROVAL_SHEET_NAMES, LEGACY_APPROVAL_SHEET_NAME)) raise RuntimeError(f"ROI审批表缺少工作表: {expected}") return targets 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) -> list[dict[str, Any]]: token = self._token() decisions: list[dict[str, Any]] = [] for sheet_name, sheet_id in self._sheet_ids(token, sheet_token): decisions.extend( self._read_sheet_approvals( token, sheet_token, sheet_name=sheet_name, sheet_id=sheet_id, ) ) return decisions def _read_sheet_approvals( self, token: str, sheet_token: str, *, sheet_name: str, sheet_id: str, ) -> list[dict[str, Any]]: header_rows = self._read_values( token, sheet_token, f"{sheet_id}!A1:ZZ1", ) if not header_rows: return [] 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, "sheet_id": sheet_id, "sheet_name": sheet_name, } ) return decisions def write_results( self, sheet_token: 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"{decision['sheet_id']}!{start}{decision['row_number']}:" f"{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_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"], 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