Agent/examples/tencent_realtime_control/tencent_client.py

"""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
        ]
        params = {
            **self._common_params(account_id),
            "user_token": self._user_token(account_id),
        }
        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,
        )
        response.raise_for_status()
        data = self._check(response.json(), "update_ad_begin_dates")
        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 RuntimeError(
                "update_ad_begin_dates partially failed: "
                f"items={item_failures} fail_id_list={failed_ids}"
            )

        target_ids = set(adgroup_ids)
        last_actual: dict[int, str] = {}
        for attempt in range(1, self.verify_attempts + 1):
            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
            }
            last_actual = {
                adgroup_id: str(
                    (ads.get(adgroup_id) or {}).get("begin_date") or ""
                )
                for adgroup_id in adgroup_ids
            }
            if all(value == begin_date for value in last_actual.values()):
                return [ads[adgroup_id] for adgroup_id in adgroup_ids]
            if attempt < self.verify_attempts:
                time.sleep(self.verify_delay_seconds)

        raise RuntimeError(
            "Tencent begin_date verification failed: "
            f"account={account_id} expected={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 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