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",
]


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
        params = {
            **self._common_params(account_id),
            "user_token": self._user_token(account_id),
        }
        response = self.session.post(
            f"{self.base_url}/adgroups/update",
            params=params,
            json=body,
            timeout=self.timeout,
        )
        response.raise_for_status()
        self._check(response.json(), "update_ad")

        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):
            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
            if attempt < self.verify_attempts:
                time.sleep(self.verify_delay_seconds)

        raise RuntimeError(
            "Post-write verification failed: "
            f"account={account_id} adgroup={adgroup_id} "
            f"expected={expected} 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