"""Minimal Tencent Ads read/write client for real-time control.""" from __future__ import annotations import json import os import time import uuid from datetime import date from typing import Any import requests from storage import connect ACTIVE_STATUS = "AD_STATUS_NORMAL" SUSPEND_STATUS = "AD_STATUS_SUSPEND" AD_FIELDS = [ "adgroup_id", "adgroup_name", "configured_status", "system_status", "bid_amount", "custom_cost_cap", "smart_bid_type", "cost_constraint_scene", "begin_date", "end_date", "time_series", ] DYNAMIC_CREATIVE_FIELDS = [ "dynamic_creative_id", "adgroup_id", "configured_status", "system_status", ] class PostWriteVerificationError(RuntimeError): """Tencent accepted a write, but the read-back did not converge in time.""" def __init__( self, *, account_id: int, adgroup_id: int, expected: dict[str, Any], actual: dict[str, Any], ) -> None: self.account_id = account_id self.adgroup_id = adgroup_id self.expected = expected self.actual = actual super().__init__( "Post-write verification failed: " f"account={account_id} adgroup={adgroup_id} " f"expected={expected} actual={actual}" ) class TencentWriteNotSentError(RuntimeError): """The request failed before Tencent received the write request.""" class TencentWriteRejectedError(RuntimeError): """Tencent explicitly rejected the write request.""" class TencentWriteOutcomeUnknownError(RuntimeError): """The request may have reached Tencent, but no definitive result exists.""" class TencentClient: def __init__(self) -> None: self.base_url = os.getenv( "TENCENT_AD_BASE_URL", "https://api.e.qq.com/v3.0" ).rstrip("/") self.access_token_api = os.getenv( "TENCENT_AD_TOKEN_API", "https://api.piaoquantv.com/ad/put/tencent/getAccessToken", ) self.user_token_api = os.getenv( "TENCENT_AD_USER_TOKEN_API", "https://api.piaoquantv.com/ad/put/tencent/getUserToken", ) self.timeout = int(os.getenv("TENCENT_AD_TIMEOUT_SECONDS", "30")) self.verify_attempts = int(os.getenv("RTC_VERIFY_ATTEMPTS", "3")) self.verify_delay_seconds = float( os.getenv("RTC_VERIFY_DELAY_SECONDS", "1") ) if self.verify_attempts < 1: raise ValueError("RTC_VERIFY_ATTEMPTS must be at least 1") if self.verify_delay_seconds < 0: raise ValueError("RTC_VERIFY_DELAY_SECONDS must not be negative") self._access_tokens: dict[int, str] = {} self._user_tokens: dict[int, str] = {} self.session = requests.Session() def _access_token(self, account_id: int) -> str: if account_id not in self._access_tokens: response = self.session.get( self.access_token_api, params={"accountId": account_id}, timeout=15, ) response.raise_for_status() token = response.text.strip() if len(token) <= 10: raise RuntimeError(f"Invalid access token for account={account_id}") self._access_tokens[account_id] = token return self._access_tokens[account_id] def _user_token(self, account_id: int) -> str: if account_id in self._user_tokens: return self._user_tokens[account_id] token = "" try: response = self.session.get( self.user_token_api, params={"accountId": account_id}, timeout=15, ) response.raise_for_status() candidate = response.text.strip() if len(candidate) > 10: token = candidate except Exception: pass if not token: connection = connect() try: with connection.cursor() as cursor: cursor.execute( "SELECT user_token FROM account_whitelist WHERE account_id=%s", (account_id,), ) row = cursor.fetchone() token = str((row or {}).get("user_token") or "") finally: connection.close() if not token: token = os.getenv("TENCENT_AD_USER_TOKEN", "").strip() if not token: raise RuntimeError(f"No user_token available for account={account_id}") self._user_tokens[account_id] = token return token def _common_params(self, account_id: int) -> dict[str, Any]: return { "access_token": self._access_token(account_id), "timestamp": int(time.time()), "nonce": uuid.uuid4().hex, } @staticmethod def _check(payload: dict[str, Any], operation: str) -> dict[str, Any]: if payload.get("code") != 0: message = payload.get("message_cn") or payload.get("message") or "unknown" raise RuntimeError( f"{operation} failed: code={payload.get('code')} message={message}" ) return payload.get("data") or {} def get_ads(self, account_id: int) -> list[dict[str, Any]]: ads: list[dict[str, Any]] = [] page = 1 while True: params = { **self._common_params(account_id), "account_id": account_id, "fields": json.dumps(AD_FIELDS, ensure_ascii=False), "page": page, "page_size": 100, } response = self.session.get( f"{self.base_url}/adgroups/get", params=params, timeout=self.timeout, ) response.raise_for_status() data = self._check(response.json(), "get_ads") rows = data.get("list") or [] ads.extend(rows) page_info = data.get("page_info") or {} if page >= int(page_info.get("total_page") or 1): break page += 1 return ads def get_ad(self, account_id: int, adgroup_id: int) -> dict[str, Any]: params = { **self._common_params(account_id), "account_id": account_id, "fields": json.dumps(AD_FIELDS, ensure_ascii=False), "filtering": json.dumps( [ { "field": "adgroup_id", "operator": "IN", "values": [str(adgroup_id)], } ] ), "page": 1, "page_size": 10, } response = self.session.get( f"{self.base_url}/adgroups/get", params=params, timeout=self.timeout, ) response.raise_for_status() data = self._check(response.json(), "get_ad") rows = data.get("list") or [] for row in rows: if int(row.get("adgroup_id") or 0) == adgroup_id: return row raise RuntimeError( f"Ad not found after update: account={account_id} adgroup={adgroup_id}" ) def update_ad( self, account_id: int, adgroup_id: int, *, bid_field: str | None = None, target_bid_fen: int | None = None, target_status: str | None = None, ) -> dict[str, Any]: body: dict[str, Any] = { "account_id": account_id, "adgroup_id": adgroup_id, } if bid_field and target_bid_fen is not None: body[bid_field] = target_bid_fen if target_status: body["configured_status"] = target_status try: params = { **self._common_params(account_id), "user_token": self._user_token(account_id), } except Exception as exc: raise TencentWriteNotSentError(str(exc)) from exc try: response = self.session.post( f"{self.base_url}/adgroups/update", params=params, json=body, timeout=self.timeout, ) except requests.RequestException as exc: raise TencentWriteOutcomeUnknownError(str(exc)) from exc if response.status_code == 408 or response.status_code >= 500: raise TencentWriteOutcomeUnknownError( f"Tencent HTTP {response.status_code}: {response.text[:500]}" ) try: response.raise_for_status() except requests.HTTPError as exc: raise TencentWriteRejectedError(str(exc)) from exc try: payload = response.json() except Exception as exc: raise TencentWriteOutcomeUnknownError( f"Tencent returned non-JSON success response: {response.text[:500]}" ) from exc try: self._check(payload, "update_ad") except Exception as exc: raise TencentWriteRejectedError(str(exc)) from exc expected: dict[str, Any] = {} if bid_field and target_bid_fen is not None: expected[bid_field] = target_bid_fen if target_status: expected["configured_status"] = target_status last_actual: dict[str, Any] = {} for attempt in range(1, self.verify_attempts + 1): try: ad = self.get_ad(account_id, adgroup_id) last_actual = {field: ad.get(field) for field in expected} matches = True for field, target in expected.items(): actual = ad.get(field) if field in {"bid_amount", "custom_cost_cap"}: try: actual = int(actual) except (TypeError, ValueError): matches = False break if actual != target: matches = False break if matches: return ad except Exception as exc: last_actual = {"verification_error": str(exc)} if attempt < self.verify_attempts: time.sleep(self.verify_delay_seconds) raise PostWriteVerificationError( account_id=account_id, adgroup_id=adgroup_id, expected=expected, actual=last_actual, ) def get_dynamic_creative( self, account_id: int, dynamic_creative_id: int, ) -> dict[str, Any]: params = { **self._common_params(account_id), "account_id": account_id, "fields": json.dumps(DYNAMIC_CREATIVE_FIELDS, ensure_ascii=False), "filtering": json.dumps( [ { "field": "dynamic_creative_id", "operator": "IN", "values": [str(dynamic_creative_id)], } ] ), "page": 1, "page_size": 10, } response = self.session.get( f"{self.base_url}/dynamic_creatives/get", params=params, timeout=self.timeout, ) response.raise_for_status() data = self._check(response.json(), "get_dynamic_creative") for row in data.get("list") or []: if int(row.get("dynamic_creative_id") or 0) == dynamic_creative_id: return row raise RuntimeError( "Dynamic creative not found: " f"account={account_id} creative={dynamic_creative_id}" ) def update_dynamic_creative_status( self, account_id: int, dynamic_creative_id: int, target_status: str, ) -> dict[str, Any]: try: params = { **self._common_params(account_id), "user_token": self._user_token(account_id), } except Exception as exc: raise TencentWriteNotSentError(str(exc)) from exc try: response = self.session.post( f"{self.base_url}/dynamic_creatives/update", params=params, json={ "account_id": account_id, "dynamic_creative_id": dynamic_creative_id, "configured_status": target_status, }, timeout=self.timeout, ) except requests.RequestException as exc: raise TencentWriteOutcomeUnknownError(str(exc)) from exc if response.status_code == 408 or response.status_code >= 500: raise TencentWriteOutcomeUnknownError( f"Tencent HTTP {response.status_code}: {response.text[:500]}" ) try: response.raise_for_status() payload = response.json() self._check(payload, "update_dynamic_creative") except requests.HTTPError as exc: raise TencentWriteRejectedError(str(exc)) from exc except (ValueError, TypeError) as exc: raise TencentWriteOutcomeUnknownError( f"Tencent returned invalid JSON: {response.text[:500]}" ) from exc except RuntimeError as exc: raise TencentWriteRejectedError(str(exc)) from exc last_actual: dict[str, Any] = {} for attempt in range(1, self.verify_attempts + 1): try: creative = self.get_dynamic_creative( account_id, dynamic_creative_id, ) last_actual = { "configured_status": creative.get("configured_status") } if creative.get("configured_status") == target_status: return creative except Exception as exc: last_actual = {"verification_error": str(exc)} if attempt < self.verify_attempts: time.sleep(self.verify_delay_seconds) raise PostWriteVerificationError( account_id=account_id, adgroup_id=dynamic_creative_id, expected={"configured_status": target_status}, actual=last_actual, ) def update_ad_begin_dates( self, account_id: int, adgroup_ids: list[int], begin_date: str, ) -> list[dict[str, Any]]: if not adgroup_ids: return [] if len(adgroup_ids) > 100: raise ValueError("Tencent update_datetime supports at most 100 ads") specs = [ {"adgroup_id": adgroup_id, "begin_date": begin_date} for adgroup_id in adgroup_ids ] try: params = { **self._common_params(account_id), "user_token": self._user_token(account_id), } except Exception as exc: raise TencentWriteNotSentError(str(exc)) from exc try: response = self.session.post( f"{self.base_url}/adgroups/update_datetime", params=params, json={ "account_id": account_id, "update_datetime_spec": specs, }, timeout=self.timeout, ) except requests.RequestException as exc: raise TencentWriteOutcomeUnknownError(str(exc)) from exc if response.status_code == 408 or response.status_code >= 500: raise TencentWriteOutcomeUnknownError( f"Tencent HTTP {response.status_code}: {response.text[:500]}" ) try: response.raise_for_status() except requests.HTTPError as exc: raise TencentWriteRejectedError(str(exc)) from exc try: payload = response.json() except Exception as exc: raise TencentWriteOutcomeUnknownError( f"Tencent returned non-JSON success response: {response.text[:500]}" ) from exc try: data = self._check(payload, "update_ad_begin_dates") except Exception as exc: raise TencentWriteRejectedError(str(exc)) from exc item_failures = [ item for item in data.get("list") or [] if int(item.get("code") or 0) != 0 ] failed_ids = [int(value) for value in data.get("fail_id_list") or []] if item_failures or failed_ids: raise TencentWriteRejectedError( "update_ad_begin_dates partially failed: " f"items={item_failures} fail_id_list={failed_ids}" ) target_ids = set(adgroup_ids) last_actual: dict[str, Any] = {} for attempt in range(1, self.verify_attempts + 1): try: ads = { int(ad.get("adgroup_id") or 0): ad for ad in self.get_ads(account_id) if int(ad.get("adgroup_id") or 0) in target_ids } begin_dates = { adgroup_id: str( (ads.get(adgroup_id) or {}).get("begin_date") or "" ) for adgroup_id in adgroup_ids } last_actual = {"begin_dates": begin_dates} if all(value == begin_date for value in begin_dates.values()): return [ads[adgroup_id] for adgroup_id in adgroup_ids] except Exception as exc: last_actual = {"verification_error": str(exc)} if attempt < self.verify_attempts: time.sleep(self.verify_delay_seconds) raise PostWriteVerificationError( account_id=account_id, adgroup_id=adgroup_ids[0], expected={"begin_date": begin_date}, actual=last_actual, ) def get_today_ad_metrics( self, account_id: int, adgroup_ids: list[int], data_date: date, ) -> dict[int, dict[str, int]]: if not adgroup_ids: return {} metrics: dict[int, dict[str, int]] = {} page = 1 while True: params = { **self._common_params(account_id), "account_id": account_id, "level": "REPORT_LEVEL_ADGROUP", "date_range": json.dumps( { "start_date": data_date.isoformat(), "end_date": data_date.isoformat(), } ), "group_by": json.dumps(["adgroup_id", "date"]), "fields": json.dumps( [ "account_id", "adgroup_id", "date", "view_count", "valid_click_count", "cost", "conversions_count", ] ), "filtering": json.dumps( [ { "field": "adgroup_id", "operator": "IN", "values": [str(value) for value in adgroup_ids], } ] ), "page": page, "page_size": 100, "time_line": "REQUEST_TIME", } response = self.session.get( f"{self.base_url}/daily_reports/get", params=params, timeout=self.timeout, ) response.raise_for_status() data = self._check(response.json(), "get_today_ad_metrics") for row in data.get("list") or []: adgroup_id = int(row.get("adgroup_id") or 0) if adgroup_id <= 0: continue metrics[adgroup_id] = { "cost_fen": int(row.get("cost") or 0), "impressions": int(row.get("view_count") or 0), "clicks": int(row.get("valid_click_count") or 0), "conversions": int(row.get("conversions_count") or 0), } page_info = data.get("page_info") or {} if page >= int(page_info.get("total_page") or 1): break page += 1 return metrics def get_today_account_metrics( self, account_id: int, data_date: date, ) -> dict[str, int]: ads = self.get_ads(account_id) adgroup_ids = [ int(ad.get("adgroup_id") or 0) for ad in ads if int(ad.get("adgroup_id") or 0) > 0 ] metrics: dict[int, dict[str, int]] = {} for start in range(0, len(adgroup_ids), 100): metrics.update( self.get_today_ad_metrics( account_id, adgroup_ids[start:start + 100], data_date, ) ) return { "cost_fen": sum(int(row.get("cost_fen") or 0) for row in metrics.values()), "impressions": sum( int(row.get("impressions") or 0) for row in metrics.values() ), "clicks": sum(int(row.get("clicks") or 0) for row in metrics.values()), "conversions": sum( int(row.get("conversions") or 0) for row in metrics.values() ), } def resolve_bid_field(ad: dict[str, Any], configured_bid_scene: str | None) -> str: smart_bid_type = str(ad.get("smart_bid_type") or "").upper() try: has_custom_cost_cap = int(ad.get("custom_cost_cap") or 0) > 0 except (TypeError, ValueError): has_custom_cost_cap = False if ( has_custom_cost_cap or smart_bid_type == "SMART_BID_TYPE_SYSTEMATIC" or configured_bid_scene == "max_conversion" ): return "custom_cost_cap" return "bid_amount" def current_bid_fen(ad: dict[str, Any], bid_field: str) -> int | None: try: return int(ad.get(bid_field)) except (TypeError, ValueError): return None