"""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}" )