feat(ad-platform): unify daily and realtime control services

AGENTS.md

@@ -11,6 +11,22 @@
 - 每日自动流程应根据腾讯侧当前状态自行判断是否需要创建广告或补创意。
 - 飞书配置行关闭时,只代表不再新建广告或补创意,不能因此暂停、删除或改动既有广告。
 - 尽量保持幂等。重复运行时应跳过已完成的工作,只补缺口。
+- 当前新建广告统一使用每天 `06:00-20:00` 的投放时段,由 `ad_delivery_template.time_series_json` 管理;修改全局时段时要同时保持代码默认值与数据库启用模板一致。
+- 飞书「自动化账户」表按 `日期` 作为配置生效批次;主流程只处理当天日期的行,其他日期不覆盖或禁用已有账户配置。`年龄` 支持如 `30-66+`; `地域` 支持中文省市或 `不限`。
+
+## 生产服务架构
+
+- 生产使用同一个完整 `ad-put-agent` 镜像启动两个容器:`ad-control-service` 和 `ad-daily-service`。
+- `ad-control-service` 负责唯一飞书 WebSocket、运营暂停/停止/恢复和每 10 分钟实时 CPM 调控。
+- 实时调控账户范围是历史 `ad_creation_account_config` 与启用白名单的交集;飞书关闭创建配置只停止新建/补创意,不能让既有广告退出实时管理。
+- `ad-daily-service` 每天 10:30 调用现有广告/创意创建流程,每 2 小时扫描腾讯创意正式审核结果。
+- 旧 `server.py`、`execute_once.py` 和 `run.py` 的模型调控链路不再作为生产入口,但暂不删除历史代码。
+- 飞书生产控制命令使用确定性解析,不能让模型直接决定账户范围或执行腾讯写操作。
+- `暂停` 表示仅暂停到下一投放日 06:00;`停止` 表示持续停止;`恢复` 只解除运营暂停,不能解除仍生效的 CPM 暂停。
+- 所有飞书触发的腾讯写操作都必须二次确认。群聊命令必须来自配置群并 @机器人,发送人必须在允许列表。
+- 运营暂停状态和 CPM 暂停状态必须分开持久化。原本人工暂停的广告不能被系统认领或自动开启;腾讯后台人工重新开启时以人工操作为准。
+- 实时控制和飞书写操作必须共用 MySQL advisory lock,并执行腾讯写后回读校验。
+- `RTC_APPLY_ENABLED` 默认关闭;生产切换前必须先完成 dry-run。日级 ROI 和预算节奏当前只保留职责边界,不能自动执行。
 
 ## 模块 B 创意创建规则
 

Dockerfile.auto_put_ad_mini

@@ -5,7 +5,7 @@ WORKDIR /app
 # 安装系统依赖（使用阿里云镜像源）
 RUN sed -i 's/deb.debian.org/mirrors.aliyun.com/g' /etc/apt/sources.list.d/debian.sources \
     && apt-get update \
-    && apt-get install -y --no-install-recommends curl \
+    && apt-get install -y --no-install-recommends curl tzdata \
     && rm -rf /var/lib/apt/lists/*
 
 # 复制依赖文件：Agent 框架依赖 + 业务特有依赖
@@ -23,6 +23,7 @@ COPY agent/ ./agent/
 COPY im-client/ ./im-client/
 COPY examples/auto_put_ad/ ./examples/auto_put_ad/
 COPY examples/auto_put_ad_mini/ ./examples/auto_put_ad_mini/
+COPY examples/tencent_realtime_control/ ./examples/tencent_realtime_control/
 
 # 设置工作目录到业务目录（参考成功案例）
 WORKDIR /app/examples/auto_put_ad_mini
@@ -32,8 +33,8 @@ RUN mkdir -p outputs/{raw,ad_status,reports,execution_log,data}
 
 # 设置环境变量
 ENV PYTHONUNBUFFERED=1 \
-    TZ=UTC \
-    PYTHONPATH=/app
+    TZ=Asia/Shanghai \
+    PYTHONPATH=/app:/app/examples/auto_put_ad_mini:/app/examples/tencent_realtime_control
 
 # 暴露端口
 EXPOSE 8080
@@ -42,5 +43,5 @@ EXPOSE 8080
 HEALTHCHECK --interval=30s --timeout=10s --start-period=40s --retries=3 \
     CMD curl -f http://localhost:8080/health || python -c "import pathlib; pathlib.Path('/app/examples/auto_put_ad_mini/outputs').exists()" || exit 1
 
-# 启动命令
-CMD ["python", "server.py"]
+# Compose 在同一镜像上分别覆盖为日级服务和实时控制服务。
+CMD ["python", "run_daily_service.py"]

docker-compose.yml

@@ -1,67 +1,55 @@
+x-ad-service: &ad-service
+  image: registry.cn-hangzhou.aliyuncs.com/stuuudy/ad-put-agent:${VERSION:-latest}
+  env_file:
+    - path: ${AD_RUNTIME_ENV_FILE:-.env}
+      format: raw
+  environment:
+    TZ: ${TZ:-Asia/Shanghai}
+    PYTHONPATH: /app:/app/examples/auto_put_ad_mini:/app/examples/tencent_realtime_control
+  volumes:
+    - ./examples/auto_put_ad_mini/outputs:/app/examples/auto_put_ad_mini/outputs
+    - ./examples/tencent_realtime_control/outputs:/app/examples/tencent_realtime_control/outputs
+  networks:
+    - ad_network
+  restart: unless-stopped
+
 services:
-  auto_put_ad_mini:
-    image: registry.cn-hangzhou.aliyuncs.com/stuuudy/ad-put-agent:${VERSION:-latest}
+  ad_control_service:
+    <<: *ad-service
     build:
-      context: .  # 项目根目录
-      dockerfile: Dockerfile.auto_put_ad_mini  # 根目录的 Dockerfile（参考成功案例）
-    container_name: auto_put_ad_mini
-    environment:
-      # 基础配置
-      - TZ=${TZ:-Asia/Shanghai}
-      - EXECUTION_ENABLED=${EXECUTION_ENABLED:-false}
-      - WHITELIST_ENABLED=${WHITELIST_ENABLED:-true}
-      - WHITELIST_ACCOUNTS=${WHITELIST_ACCOUNTS:-80769799}
-
-      # 腾讯广告 API
-      - TENCENT_AD_ACCOUNT_ID=${TENCENT_AD_ACCOUNT_ID:-80769799}
-      - TENCENT_AD_USER_TOKEN=${TENCENT_AD_USER_TOKEN}
-      - TENCENT_AD_ACCESS_TOKEN=${TENCENT_AD_ACCESS_TOKEN:-}
-      - TENCENT_AD_BASE_URL=${TENCENT_AD_BASE_URL:-https://api.e.qq.com/v3.0}
-
-      # 飞书配置
-      - FEISHU_APP_ID=${FEISHU_APP_ID}
-      - FEISHU_APP_SECRET=${FEISHU_APP_SECRET}
-      - FEISHU_OPERATOR_OPEN_ID=${FEISHU_OPERATOR_OPEN_ID}
-      - FEISHU_OPERATOR_CHAT_ID=${FEISHU_OPERATOR_CHAT_ID}
-      - FEISHU_AD_PROJECT_CHAT_ID=${FEISHU_AD_PROJECT_CHAT_ID:-}
-
-      # 数据库配置
-      - DB_HOST=${DB_HOST:-localhost}
-      - DB_PORT=${DB_PORT:-3306}
-      - DB_USER=${DB_USER:-ad_rw}
-      - DB_PASSWORD=${DB_PASSWORD}
-      - DB_NAME=${DB_NAME:-auto_put_ad_mini}
-
-      # ODPS 数据平台（可选）
-      - ODPS_ACCESS_ID=${ODPS_ACCESS_ID:-}
-      - ODPS_ACCESS_SECRET=${ODPS_ACCESS_SECRET:-}
-      - ODPS_PROJECT=${ODPS_PROJECT:-loghubods}
-
-      # LLM API Keys
-      - QWEN_API_KEY=${QWEN_API_KEY:-}
-      - OPEN_ROUTER_API_KEY=${OPEN_ROUTER_API_KEY:-}
-
-      # 代理配置（可选）
-      - HTTP_PROXY=${HTTP_PROXY:-}
-      - HTTPS_PROXY=${HTTPS_PROXY:-}
-
-      # APScheduler 定时任务
-      - CRON_SCHEDULE=${CRON_SCHEDULE:-0 2 * * *}
-      - RUN_ON_STARTUP=${RUN_ON_STARTUP:-false}
-      - PORT=${PORT:-8080}
-    ports:
-      - "8080:8080"
-    volumes:
-      - ./examples/auto_put_ad_mini/outputs:/app/outputs
-    networks:
-      - ad_network
-    restart: unless-stopped
+      context: .
+      dockerfile: Dockerfile.auto_put_ad_mini
+    container_name: ad-control-service
+    command:
+      - python
+      - /app/examples/tencent_realtime_control/run_control_service.py
+    healthcheck:
+      test:
+        - CMD
+        - python
+        - -c
+        - "from pathlib import Path; raise SystemExit(0 if b'run_control_service.py' in Path('/proc/1/cmdline').read_bytes() else 1)"
+      interval: 30s
+      timeout: 5s
+      retries: 3
+      start_period: 30s
+
+  ad_daily_service:
+    <<: *ad-service
+    container_name: ad-daily-service
+    command:
+      - python
+      - /app/examples/auto_put_ad_mini/run_daily_service.py
     healthcheck:
-      test: ["CMD", "curl", "-f", "http://localhost:8080/health"]
+      test:
+        - CMD
+        - python
+        - -c
+        - "from pathlib import Path; raise SystemExit(0 if b'run_daily_service.py' in Path('/proc/1/cmdline').read_bytes() else 1)"
       interval: 30s
-      timeout: 10s
+      timeout: 5s
       retries: 3
-      start_period: 40s
+      start_period: 30s
 
 networks:
   ad_network:

examples/auto_put_ad_mini/.env.example

@@ -13,14 +13,40 @@ TENCENT_AD_ACCOUNT_ID=80769799
 # ⚠️ 重要：写操作（暂停广告/修改出价）必须配置 user_token
 # 获取方式：https://docs.qq.com/doc/DSVdkdk1LQ1hOam5n
 TENCENT_AD_USER_TOKEN=xxx
+# 可选,默认使用内部接口按 accountId 动态获取最新 user_token
+# TENCENT_AD_USER_TOKEN_API=https://api.piaoquantv.com/ad/put/tencent/getUserToken
 
 # TENCENT_AD_BASE_URL=https://api.e.qq.com/v3.0
 
+# 统一生产服务
+DAILY_CREATION_ENABLED=0
+DAILY_CREATION_HOUR=10
+DAILY_CREATION_MINUTE=30
+DAILY_REVIEW_ENABLED=0
+DAILY_REVIEW_INTERVAL_HOURS=2
+DAILY_RUN_ON_STARTUP=0
+DAILY_REVIEW_RUN_ON_STARTUP=0
+DAILY_SYNC_DELIVERY_TEMPLATE=0
+
+RTC_START_HOUR=6
+RTC_STOP_HOUR=20
+RTC_POLL_SECONDS=600
+RTC_HIGH_CPM=250
+RTC_LOW_CPM=190
+RTC_BID_UP_RATIO=1.25
+RTC_MAX_PARTITION_LAG_HOURS=2
+RTC_APPLY_ENABLED=0
+RTC_COMMAND_ENABLED=1
+RTC_COMMAND_CHAT_ID=
+RTC_COMMAND_ALLOWED_OPEN_IDS=
+RTC_COMMAND_CONFIRM_TTL_MINUTES=10
+RTC_COMMAND_WORKERS=2
+
 # ========================================
 # 飞书 IM 审批配置
 # ========================================
-FEISHU_APP_ID=cli_a955e97067f85cb3
-FEISHU_APP_SECRET=NQaG4ci1plXRDTgwCqrLJgMLLoA2tdF8
+FEISHU_APP_ID=cli_xxx
+FEISHU_APP_SECRET=xxx
 
 # 运营审批群聊（需要审批回复的群）
 FEISHU_OPERATOR_OPEN_ID=ou_498988d823b61ab89c9afe4310f85bb4

examples/auto_put_ad_mini/config.py

@@ -9,7 +9,7 @@
 """
 import os
 import logging
-from datetime import datetime
+from datetime import date, datetime
 from pathlib import Path
 from typing import Optional
 from zoneinfo import ZoneInfo
@@ -325,7 +325,10 @@ def get_account_creation_config(account_id: int) -> dict:
                        c.audience_tier_label, c.bid_min_fen, c.bid_max_fen,
                        c.bid_amount_fen, c.bid_scene, c.custom_cost_cap_fen,
                        c.automatic_site_enabled, c.site_set_json AS account_site_set_json,
+                       c.location_types_json AS account_location_types_json,
+                       c.region_ids_json AS account_region_ids_json,
                        c.age_min, c.age_max,
+                       c.config_date,
                        c.daily_budget_fen AS account_daily_budget_fen,
                        t.site_set_json, t.location_types_json, t.region_ids_json,
                        t.daily_budget_fen, t.time_series_json,
@@ -398,8 +401,16 @@ def get_account_creation_config(account_id: int) -> dict:
     import json as _json
     from tools.delivery_config import parse_bid_scene
     site_set = _json.loads(row.get("account_site_set_json") or row["site_set_json"])
-    location_types = _json.loads(row["location_types_json"])
-    region_ids = _json.loads(row["region_ids_json"])
+    location_types = _json.loads(
+        row["account_location_types_json"]
+        if row.get("account_location_types_json") is not None
+        else row["location_types_json"]
+    )
+    region_ids = _json.loads(
+        row["account_region_ids_json"]
+        if row.get("account_region_ids_json") is not None
+        else row["region_ids_json"]
+    )
     time_series = _json.loads(row["time_series_json"])
     automatic_site_enabled = row.get("automatic_site_enabled")
 
@@ -422,6 +433,10 @@ def get_account_creation_config(account_id: int) -> dict:
         "automatic_site_enabled": bool(automatic_site_enabled) if automatic_site_enabled is not None else False,
         "location_types": location_types,
         "region_ids": region_ids,
+        "config_date": (
+            row["config_date"].isoformat()
+            if row.get("config_date") is not None else None
+        ),
         "daily_budget_fen": int(
             row["account_daily_budget_fen"]
             if row.get("account_daily_budget_fen") is not None
@@ -437,8 +452,8 @@ def get_account_audience_pack(account_id: int) -> tuple[Optional[int], str]:
     return cfg["audience_pack_id"], cfg["audience_tier_label"]
 
 
-def get_creation_account_ids() -> list[int]:
-    """读取当前启用的待投放账户,作为 Phase 0 创建广告的账户范围。"""
+def get_creation_account_ids(config_date: Optional[date] = None) -> list[int]:
+    """读取指定飞书配置日期启用的账户,作为 Phase 0/1 自动化范围。"""
     try:
         from db.connection import get_connection
     except Exception as e:
@@ -447,6 +462,7 @@ def get_creation_account_ids() -> list[int]:
 
     conn = get_connection()
     try:
+        target_config_date = config_date or now_in_timezone().date()
         with conn.cursor() as cur:
             cur.execute(
                 """
@@ -458,8 +474,10 @@ def get_creation_account_ids() -> list[int]:
                  AND t.enabled = TRUE
                 WHERE c.enabled = TRUE
                   AND w.enabled = TRUE
+                  AND c.config_date = %s
                 ORDER BY c.id ASC
-                """
+                """,
+                (target_config_date,),
             )
             rows = cur.fetchall()
             return [int(r["account_id"]) for r in rows]
@@ -846,9 +864,9 @@ WECHAT_POSITION_TARGETED_PRESET = [
 # 1 账户广告数(2026-06-09 用户确认:2 条,一条有 wechat_position 定投,一条无定投)
 ADS_PER_ACCOUNT = 2
 
-# --- 时段 / 日期(真实样本 95205841163 反推:6:00-22:30 投放)---
+# --- 时段 / 日期(当前统一为每天 06:00-20:00)---
 # 一天 48 段 × 7 天 = 336 位字符串
-TIME_SERIES_ONE_DAY = "000000000000" + "1" * 34 + "00"   # 0:00-6:00 关 / 6:00-23:00 投 / 23:00-24:00 关
+TIME_SERIES_ONE_DAY = "0" * 12 + "1" * 28 + "0" * 8
 TIME_SERIES_DEFAULT = TIME_SERIES_ONE_DAY * 7
 # 长度自验
 assert len(TIME_SERIES_DEFAULT) == 336, f"time_series 长度错误:{len(TIME_SERIES_DEFAULT)}"

examples/auto_put_ad_mini/configure_creation_accounts.py

@@ -15,6 +15,7 @@ import csv
 import json
 import sys
 from dataclasses import dataclass
+from datetime import date
 from decimal import Decimal, ROUND_HALF_UP
 from pathlib import Path
 
@@ -49,6 +50,11 @@ class AccountConfigInput:
     custom_cost_cap_fen: int | None = None
     automatic_site_enabled: bool | None = None
     site_set: list[str] | None = None
+    age_min: int = DEFAULT_AGE_MIN
+    age_max: int = DEFAULT_AGE_MAX
+    location_types: list[str] | None = None
+    region_ids: list[int] | None = None
+    config_date: date | None = None
 
 
 def _bid_to_fen(raw: str) -> int:
@@ -97,7 +103,8 @@ def ensure_account_delivery_config_columns() -> None:
                   AND TABLE_NAME = 'ad_creation_account_config'
                   AND COLUMN_NAME IN (
                     'bid_scene', 'custom_cost_cap_fen',
-                    'automatic_site_enabled', 'site_set_json'
+                    'automatic_site_enabled', 'site_set_json',
+                    'location_types_json', 'region_ids_json', 'config_date'
                   )
                 """
             )
@@ -138,6 +145,33 @@ def ensure_account_delivery_config_columns() -> None:
                     AFTER automatic_site_enabled
                     """
                 )
+            if "location_types_json" not in existing:
+                cur.execute(
+                    """
+                    ALTER TABLE ad_creation_account_config
+                    ADD COLUMN location_types_json TEXT DEFAULT NULL
+                    COMMENT '账户级地域类型覆盖JSON;NULL使用投放模板'
+                    AFTER site_set_json
+                    """
+                )
+            if "region_ids_json" not in existing:
+                cur.execute(
+                    """
+                    ALTER TABLE ad_creation_account_config
+                    ADD COLUMN region_ids_json MEDIUMTEXT DEFAULT NULL
+                    COMMENT '账户级地域ID覆盖JSON;NULL使用投放模板,空数组表示不限地域'
+                    AFTER location_types_json
+                    """
+                )
+            if "config_date" not in existing:
+                cur.execute(
+                    """
+                    ALTER TABLE ad_creation_account_config
+                    ADD COLUMN config_date DATE DEFAULT NULL
+                    COMMENT '飞书配置生效日期;仅当天配置参与自动创建/补创意'
+                    AFTER region_ids_json
+                    """
+                )
         conn.commit()
     finally:
         conn.close()
@@ -194,7 +228,9 @@ def upsert_account_config(row: AccountConfigInput, dry_run: bool = False) -> Non
             f"status={grant_status} material_source={row.material_source} "
             f"ai_fallback={row.ai_fallback_to_history} bid_scene={row.bid_scene} "
             f"custom_cost_cap={row.custom_cost_cap_fen} "
-            f"automatic_site={row.automatic_site_enabled} site_set={row.site_set}"
+            f"automatic_site={row.automatic_site_enabled} site_set={row.site_set} "
+            f"age={row.age_min}-{row.age_max} regions={row.region_ids} "
+            f"config_date={row.config_date}"
         )
         return
 
@@ -257,7 +293,8 @@ def upsert_account_config(row: AccountConfigInput, dry_run: bool = False) -> Non
                    bid_min_fen, bid_max_fen, bid_amount_fen, bid_scene,
                    custom_cost_cap_fen,
                    daily_budget_fen, automatic_site_enabled, site_set_json,
-                   age_min, age_max, audience_source_account_id, audience_grant_status,
+                   location_types_json, region_ids_json,
+                   age_min, age_max, config_date, audience_source_account_id, audience_grant_status,
                    remark, created_by, updated_by)
                 VALUES
                   (%s, TRUE, %s, %s, %s,
@@ -266,7 +303,8 @@ def upsert_account_config(row: AccountConfigInput, dry_run: bool = False) -> Non
                    %s, %s, %s, %s,
                    %s,
                    %s, %s, %s,
-                   %s, %s, %s, %s,
+                   %s, %s,
+                   %s, %s, %s, %s, %s,
                    %s, %s, %s)
                 ON DUPLICATE KEY UPDATE
                   enabled=TRUE,
@@ -289,8 +327,11 @@ def upsert_account_config(row: AccountConfigInput, dry_run: bool = False) -> Non
                   daily_budget_fen=VALUES(daily_budget_fen),
                   automatic_site_enabled=VALUES(automatic_site_enabled),
                   site_set_json=VALUES(site_set_json),
+                  location_types_json=VALUES(location_types_json),
+                  region_ids_json=VALUES(region_ids_json),
                   age_min=VALUES(age_min),
                   age_max=VALUES(age_max),
+                  config_date=VALUES(config_date),
                   audience_source_account_id=IF(
                     ad_creation_account_config.audience_name = VALUES(audience_name),
                     ad_creation_account_config.audience_source_account_id,
@@ -328,8 +369,11 @@ def upsert_account_config(row: AccountConfigInput, dry_run: bool = False) -> Non
                     row.daily_budget_fen,
                     None if row.automatic_site_enabled is None else (1 if row.automatic_site_enabled else 0),
                     json.dumps(row.site_set, ensure_ascii=False) if row.site_set is not None else None,
-                    DEFAULT_AGE_MIN,
-                    DEFAULT_AGE_MAX,
+                    json.dumps(row.location_types, ensure_ascii=False) if row.location_types is not None else None,
+                    json.dumps(row.region_ids, ensure_ascii=False) if row.region_ids is not None else None,
+                    row.age_min,
+                    row.age_max,
+                    row.config_date,
                     source_account_id,
                     grant_status,
                     "configured from account_id/audience_name/bid triple",
@@ -345,13 +389,19 @@ def upsert_account_config(row: AccountConfigInput, dry_run: bool = False) -> Non
         f"configured account={row.account_id} audience={row.audience_name} "
         f"bid={row.bid_min_fen / 100:.2f}-{row.bid_max_fen / 100:.2f} "
         f"bid_scene={row.bid_scene} custom_cost_cap={row.custom_cost_cap_fen} "
-        f"status={grant_status}"
+        f"age={row.age_min}-{row.age_max} regions={row.region_ids} "
+        f"config_date={row.config_date} status={grant_status}"
     )
 
 
-def set_account_automation_enabled(account_id: int, enabled: bool, dry_run: bool = False) -> None:
+def set_account_automation_enabled(
+    account_id: int,
+    enabled: bool,
+    config_date: date | None = None,
+    dry_run: bool = False,
+) -> None:
     if dry_run:
-        print(f"[dry-run] set account={account_id} enabled={enabled}")
+        print(f"[dry-run] set account={account_id} enabled={enabled} config_date={config_date}")
         return
 
     from db.connection import get_connection
@@ -362,10 +412,10 @@ def set_account_automation_enabled(account_id: int, enabled: bool, dry_run: bool
             cur.execute(
                 """
                 UPDATE ad_creation_account_config
-                SET enabled=%s, updated_by='configure_creation_accounts'
+                SET enabled=%s, config_date=%s, updated_by='configure_creation_accounts'
                 WHERE account_id=%s
                 """,
-                (1 if enabled else 0, account_id),
+                (1 if enabled else 0, config_date, account_id),
             )
         conn.commit()
     finally:

examples/auto_put_ad_mini/db/schema.sql

@@ -67,8 +67,11 @@ CREATE TABLE IF NOT EXISTS ad_creation_account_config (
     daily_budget_fen INT DEFAULT NULL COMMENT '账户级单广告日预算(分);NULL则使用投放模板默认预算',
     automatic_site_enabled BOOLEAN DEFAULT NULL COMMENT '账户级是否开启智能版位;NULL使用模板默认手动版位',
     site_set_json TEXT DEFAULT NULL COMMENT '账户级投放版位覆盖JSON;NULL使用投放模板',
+    location_types_json TEXT DEFAULT NULL COMMENT '账户级地域类型覆盖JSON;NULL使用投放模板',
+    region_ids_json MEDIUMTEXT DEFAULT NULL COMMENT '账户级地域ID覆盖JSON;NULL使用投放模板,空数组表示不限地域',
     age_min INT NOT NULL COMMENT '最小年龄',
     age_max INT NOT NULL COMMENT '最大年龄',
+    config_date DATE DEFAULT NULL COMMENT '飞书配置生效日期;仅当天配置参与自动创建/补创意',
     audience_source_account_id BIGINT DEFAULT NULL COMMENT '人群包来源账户;默认可由 TENCENT_AUDIENCE_SOURCE_ACCOUNT_ID 提供',
     audience_grant_status VARCHAR(50) DEFAULT NULL COMMENT '授权/验证状态:not_required_no_audience_pack/resolved/grant_submitted/verified',
     audience_verified_at TIMESTAMP NULL DEFAULT NULL COMMENT '目标账户可见验证时间',

examples/auto_put_ad_mini/docs/ad_automation_platform_architecture_plan_2026-07-25.md

@@ -0,0 +1,134 @@
+# 自动投放平台统一架构
+
+> 状态：已实现
+> 日期：2026-07-25
+
+## 1. 服务边界
+
+生产只启动两个服务，共用 `ad-put-agent` 完整镜像。
+
+| 服务 | 运行方式 | 职责 |
+|---|---|---|
+| `ad-control-service` | 常驻 | 飞书控制命令、运营暂停状态、每 10 分钟 CPM 调控、次日恢复 |
+| `ad-daily-service` | 常驻调度 | 每日 10:30 广告/创意创建、每 2 小时创意审核扫描 |
+
+旧 `server.py`、`execute_once.py` 和 `run.py` 的模型调控链路不再作为生产入口。
+现有 ROI 纯计算函数保留，未来日级 ROI 服务可以复用，但本期不自动调价。
+
+## 2. 日级流程
+
+`ad-daily-service` 不重写业务流程，只调度已有脚本：
+
+```text
+每天 10:30
+  execute_creation_once.py
+    飞书配置同步
+    -> 广告缺口检查与创建
+    -> 创意缺口检查
+    -> 历史素材召回 / AI 生成
+    -> 现有飞书在线表格审批
+    -> 审批通过项提交腾讯
+
+每 2 小时
+  scan_creative_reviews.py
+    -> 查询腾讯正式审核结果
+    -> 写入审核结果和拒审事实
+```
+
+每类任务使用独立 MySQL advisory lock，防止多 Pod 重复运行。创建流程仍保持
+原有幂等、审批和副作用边界。创建、审核扫描和模板同步默认均关闭，生产必须
+通过 `DAILY_*_ENABLED` 和 `DAILY_SYNC_DELIVERY_TEMPLATE` 显式启用。
+
+## 3. 实时流程
+
+`ad-control-service` 内有两个并发入口：
+
+```text
+飞书 WebSocket
+  -> 确定性解析
+  -> 权限校验
+  -> 生成操作预览
+  -> 二次确认
+  -> 腾讯写入和回读
+
+每 10 分钟实时循环
+  -> 运营暂停状态对账
+  -> 当天暂停到期处理
+  -> 获取最新小时 CPM
+  -> CPM 决策
+  -> 合并最终广告状态和出价
+  -> 腾讯写入和回读
+  -> 动作审计与飞书通知
+```
+
+两个入口共用 `RTC_DB_LOCK_NAME`，同一时刻只允许一个流程修改腾讯广告。
+
+实时 CPM 规则保持：
+
+- `CPM > 250`：基础出价上调 25%，每天最多一次。
+- `190 <= CPM <= 250`：恢复基础出价和 CPM 策略暂停。
+- `CPM < 190`：恢复基础出价后暂停。
+- 每天 20:00：恢复基础出价并执行当日收尾暂停。
+- 次日 06:00：恢复实时策略暂停，再进入当日 CPM 判断。
+
+## 4. 飞书命令
+
+首期只接受确定性生产命令，不接入开放式模型问答：
+
+```text
+暂停全部
+暂停账户 86197363,86748325
+停止全部
+停止账户 86197363 86748325
+恢复全部
+恢复账户 86197363、86748325
+查看暂停状态
+确认 cmd_xxx
+取消 cmd_xxx
+```
+
+- `暂停`：暂停到下一投放日 06:00。
+- `停止`：持续停止，只有明确恢复或腾讯后台人工开启才解除。
+- `恢复`：解除运营暂停；如果仍有 CPM 暂停，不会误开广告。
+- 所有腾讯写命令都必须在 10 分钟内二次确认。
+- 群聊必须来自配置群并 @机器人；发送人必须在允许列表。
+- 飞书消息 ID 是幂等键，重复事件不会重复创建命令。
+- “全部”表示 `ad_creation_account_config` 历史账户与启用白名单的交集，
+  不受当前是否继续新建广告影响。
+
+## 5. 状态与优先级
+
+`realtime_control_ad_state` 分开保存：
+
+- `paused_by_strategy`：CPM 实时策略暂停。
+- `operator_pause_mode=UNTIL_NEXT_DELIVERY`：运营当天暂停。
+- `operator_pause_mode=UNTIL_MANUAL`：运营持续停止。
+
+优先级为：
+
+```text
+未到期运营暂停 > CPM 状态
+```
+
+只认领命令执行时腾讯状态为 `NORMAL` 的广告。原本人工暂停的广告只记录跳过，
+不会被后续自动恢复。腾讯后台人工重新开启运营停止广告时，系统尊重人工操作，
+清除运营暂停状态并发送通知。
+
+命令、广告明细、写前状态、目标状态、回读状态和错误分别保存到
+`operator_command`、`operator_command_item`。腾讯实时动作继续写入
+`realtime_control_action_log`。
+
+## 6. 配置所有权
+
+- 飞书/数据库：账户、人群包、出价、预算、素材来源、年龄、地域、版位。
+- 日级服务：下一投放日的广告和创意供给；未来拥有 `base_bid_fen`。
+- 实时服务：盘中状态和临时倍率，不改账户配置和长期投放日期。
+- 新建广告统一使用 `06:00-20:00`，代码默认值和启用投放模板在日级服务启动时校准。
+- 历史广告不会因本次代码发布自动批量修改投放时段。
+
+## 7. 暂不实现
+
+- 不实现自动日级 ROI 调价，只保留基础出价所有权边界。
+- 不实现预算节奏调控。
+- 不引入队列、Redis、Celery 或额外 worker。
+- 不让模型解析账户 ID 或直接执行腾讯写操作。

examples/auto_put_ad_mini/docs/unified_services_deployment.md

@@ -0,0 +1,163 @@
+# 自动投放双服务部署
+
+## 1. Jenkins 构建
+
+构建统一完整镜像，不再构建 `tencent-realtime-control` 独立镜像：
+
+```bash
+set -e
+
+REGISTRY="registry.cn-hangzhou.aliyuncs.com/stuuudy"
+IMAGE="${REGISTRY}/ad-put-agent"
+
+cd "${WORKSPACE}"
+docker build \
+  -f Dockerfile.auto_put_ad_mini \
+  -t "${IMAGE}:latest" \
+  -t "${IMAGE}:${BUILD_TIMESTAMP}" \
+  .
+docker push "${IMAGE}:latest"
+docker push "${IMAGE}:${BUILD_TIMESTAMP}"
+
+ssh -i /root/.ssh/piaoquan.pem root@8.222.142.37 \
+  "sh /home/sh/deploy.sh ${BUILD_TIMESTAMP}"
+```
+
+## 2. ECS 文件
+
+将仓库中的 `docker-compose.yml` 放到 `/home/server/docker-compose.yml`。
+要求 ECS 的 Docker Compose 版本不低于 `2.30`，以支持 `env_file.format=raw`。
+
+两个服务共用现有的 `/home/server/.env`：
+
+```bash
+chmod 600 /home/server/.env
+docker compose version
+```
+
+原有数据库、ODPS、腾讯、飞书、OSS 和 AI 环境变量全部保留。
+
+首次部署增加：
+
+```dotenv
+RTC_START_HOUR=6
+RTC_STOP_HOUR=20
+RTC_POLL_SECONDS=600
+RTC_HIGH_CPM=250
+RTC_LOW_CPM=190
+RTC_BID_UP_RATIO=1.25
+RTC_MAX_PARTITION_LAG_HOURS=2
+RTC_APPLY_ENABLED=0
+RTC_COMMAND_ENABLED=1
+RTC_COMMAND_CHAT_ID=oc_xxx
+RTC_COMMAND_ALLOWED_OPEN_IDS=ou_xxx
+RTC_COMMAND_CONFIRM_TTL_MINUTES=10
+
+DAILY_CREATION_ENABLED=0
+DAILY_CREATION_HOUR=10
+DAILY_CREATION_MINUTE=30
+DAILY_REVIEW_ENABLED=0
+DAILY_REVIEW_INTERVAL_HOURS=2
+DAILY_RUN_ON_STARTUP=0
+DAILY_REVIEW_RUN_ON_STARTUP=0
+DAILY_SYNC_DELIVERY_TEMPLATE=0
+```
+
+多个操作人使用英文逗号分隔 `RTC_COMMAND_ALLOWED_OPEN_IDS`。
+代码默认 `RTC_APPLY_ENABLED=0`，日级创建、审核扫描和模板同步也默认关闭；
+缺少显式生产配置时不会创建广告或修改腾讯。
+
+## 3. ECS 部署脚本
+
+`/home/sh/deploy.sh` 使用：
+
+```bash
+#!/bin/bash
+set -e
+
+echo "Qingqu@2019" |
+  docker login \
+    --username=stuuudys \
+    registry.cn-hangzhou.aliyuncs.com \
+    --password-stdin
+
+VERSION=${1:-latest}
+cd /home/server
+
+VERSION=${VERSION} docker compose --env-file /dev/null pull
+VERSION=${VERSION} docker compose --env-file /dev/null down --remove-orphans
+VERSION=${VERSION} docker compose --env-file /dev/null up -d
+VERSION=${VERSION} docker compose --env-file /dev/null ps
+```
+
+Compose 以 raw 模式读取 `/home/server/.env`，不展开密码或密钥中的 `$`，
+也不再依赖 `/home/sh/.env` 向容器透传变量。
+
+## 4. 首次 dry-run
+
+保持 `RTC_APPLY_ENABLED=0` 和两个 `DAILY_*_ENABLED=0`：
+
+```bash
+cd /home/server
+docker compose --env-file /dev/null config --services
+docker compose --env-file /dev/null up -d
+docker compose --env-file /dev/null ps
+docker compose --env-file /dev/null logs --tail=200 ad_control_service
+docker compose --env-file /dev/null logs --tail=200 ad_daily_service
+```
+
+控制服务启动时自动执行实时状态表增量迁移。也可以显式验证：
+
+```bash
+docker compose --env-file /dev/null run --rm \
+  ad_control_service \
+  python /app/examples/tencent_realtime_control/init_db.py
+```
+
+执行无腾讯写入的 CPM dry-run：
+
+```bash
+docker compose --env-file /dev/null run --rm \
+  ad_control_service \
+  python /app/examples/tencent_realtime_control/run_once.py \
+  --at 2026-07-25T12:00:00 \
+  --cpm-override 220 \
+  --force-refresh
+```
+
+飞书验证：
+
+1. 在配置群 @机器人发送 `查看暂停状态`。
+2. 发送一个单账户暂停命令，确认机器人返回预览和 `cmd_xxx`。
+3. 回复确认，服务必须提示 dry-run 禁止写腾讯。
+
+## 5. 生产切换
+
+1. 确认旧 `auto_put_ad_mini/server.py` 和旧实时容器不再运行。
+2. 设置 `RTC_APPLY_ENABLED=1`，保持日级服务关闭。
+3. 重新 `docker compose up -d`，观察一个完整实时周期。
+4. 用单账户命令验证暂停、恢复、腾讯回读和飞书结果。
+5. 设置 `DAILY_REVIEW_ENABLED=1`、`DAILY_SYNC_DELIVERY_TEMPLATE=1`。
+6. 审核扫描稳定后设置 `DAILY_CREATION_ENABLED=1`。
+7. 保持 `DAILY_RUN_ON_STARTUP=0`，防止部署时额外触发完整创建。
+
+## 6. 验收与回滚
+
+验收：
+
+```bash
+docker compose --env-file /dev/null ps
+docker compose --env-file /dev/null logs --since=30m ad_control_service
+docker compose --env-file /dev/null logs --since=30m ad_daily_service
+```
+
+应确认：
+
+- 两个容器均为 healthy。
+- 实时循环每 10 分钟运行且没有双实例锁冲突。
+- 飞书只有一个 WebSocket 消费者。
+- 10:30 只出现一次创建任务。
+- 审核扫描每 2 小时最多一个实例。
+
+回滚时将 `VERSION` 改为上一个镜像版本并重新部署。数据库新增表和列可以保留，
+旧代码不会读取这些字段，不需要执行破坏性回滚。

examples/auto_put_ad_mini/execute_creation_once.py

@@ -21,12 +21,14 @@
 开关 CREATION_APPROVAL_REQUIRED=False 时跳过 Phase 2,全 records 直接 approve。
 """
 
+import argparse
 import json
 import logging
 import os
 import sys
 import time
 from concurrent.futures import FIRST_COMPLETED, ThreadPoolExecutor, wait
+from datetime import date
 from pathlib import Path
 
 _HERE = Path(__file__).parent
@@ -231,7 +233,10 @@ def _fetch_existing_fingerprints_for_account(account_id: int) -> set[str]:
     return fingerprints
 
 
-def phase0_create_ads(target_ads: int = ADS_PER_ACCOUNT) -> list[dict]:
+def phase0_create_ads(
+    target_ads: int = ADS_PER_ACCOUNT,
+    config_date: date | None = None,
+) -> list[dict]:
     """Phase 0:对每账户检查广告数,不足则建到 target_ads 条(模块 A,2026-06-09 P1-G)。
 
     流程:
@@ -244,7 +249,7 @@ def phase0_create_ads(target_ads: int = ADS_PER_ACCOUNT) -> list[dict]:
     from tools.ad_api import _get
     from tools.im_approval_ad_creation import run_ad_approval_workflow
 
-    creation_accounts = get_creation_account_ids()
+    creation_accounts = get_creation_account_ids(config_date=config_date)
     if not creation_accounts:
         logger.error("[phase0] 待投放账户配置为空,退出")
         return []
@@ -438,13 +443,16 @@ def _wait_created_ads_visible(created_ads: list[dict], timeout_seconds: int = 60
         logger.info("[phase0] 新建广告均已在 list 接口可见")
 
 
-def phase1_prepare(target_creatives: int = TARGET_CREATIVES_PER_AD) -> list[dict]:
+def phase1_prepare(
+    target_creatives: int = TARGET_CREATIVES_PER_AD,
+    config_date: date | None = None,
+) -> list[dict]:
     """Phase 1:扫账户 → 关联点过滤 → 准备 pending records(不 POST 腾讯)。
 
     Returns:
         pending_records — 每个元素是 prepare_one_creative_for_ad 返回的 dict
     """
-    creation_accounts = get_creation_account_ids()
+    creation_accounts = get_creation_account_ids(config_date=config_date)
     if not creation_accounts:
         logger.error("[phase1] 待投放账户配置为空,Phase 1 退出")
         return []
@@ -876,7 +884,7 @@ def phase2_approval(records: list[dict], xlsx_output_dir: Path) -> dict[int, str
     return actions
 
 
-def run_once() -> dict:
+def run_once(config_date: str | None = None) -> dict:
     """完整主循环:Phase 0 → Phase 1 → Phase 2(若开关 True)→ Phase 3。"""
     run_started = now_in_timezone().isoformat()
     sync_stats: dict = {}
@@ -887,7 +895,7 @@ def run_once() -> dict:
     try:
         from sync_feishu_account_config import sync_from_feishu
 
-        sync_stats = sync_from_feishu()
+        sync_stats = sync_from_feishu(config_date=config_date)
         logger.info("[main] Phase -1 完成: %s", sync_stats)
     except Exception as e:
         logger.exception("[main] Phase -1 同步飞书配置失败,本轮停止:%s", e)
@@ -900,6 +908,8 @@ def run_once() -> dict:
             "total": {"phase1_prepared": 0},
         }
 
+    effective_config_date = date.fromisoformat(sync_stats["config_date"])
+
     # Phase 0:模块 A 建广告(满足每账户 ADS_PER_ACCOUNT 条)
     logger.info("=" * 60)
     if _env_flag("CREATION_SKIP_PHASE0"):
@@ -907,14 +917,14 @@ def run_once() -> dict:
         created_ads = []
     else:
         logger.info("[main] Phase 0 启动 — 模块 A 检查 + 建广告")
-        created_ads = phase0_create_ads()
+        created_ads = phase0_create_ads(config_date=effective_config_date)
         logger.info("[main] Phase 0 完成:本轮新建广告 %d 条", len(created_ads))
         _wait_created_ads_visible(created_ads)
 
     # Phase 1:模块 B 给所有广告(新+旧)补创意
     logger.info("=" * 60)
     logger.info("[main] Phase 1 启动 — 模块 B 给广告补创意")
-    pending_records = phase1_prepare()
+    pending_records = phase1_prepare(config_date=effective_config_date)
     if not pending_records:
         logger.info("[main] Phase 1 无 pending records,主循环退出")
         return {
@@ -962,6 +972,13 @@ def run_once() -> dict:
 
 
 def main() -> int:
+    parser = argparse.ArgumentParser(description="执行广告与创意创建主流程")
+    parser.add_argument(
+        "--config-date",
+        help="执行指定飞书配置日期,格式 YYYYMMDD 或 YYYY-MM-DD;默认当天",
+    )
+    args = parser.parse_args()
+
     _setup_logging()
     logger.info("=" * 60)
     logger.info("模块 B 创意搭建子系统 — 主循环启动")
@@ -970,7 +987,7 @@ def main() -> int:
     logger.info("WHITELIST_ACCOUNTS            = %s", WHITELIST_ACCOUNTS)
     logger.info("=" * 60)
 
-    summary = run_once()
+    summary = run_once(config_date=args.config_date)
     t = summary.get("total") or {}
     logger.info("=" * 60)
     logger.info("[main] 主循环结束")

examples/auto_put_ad_mini/run_daily_service.py

@@ -0,0 +1,157 @@
+#!/usr/bin/env python
+"""Schedule daily creation and two-hour creative review scans."""
+
+from __future__ import annotations
+
+import json
+import logging
+import os
+import subprocess
+import sys
+from pathlib import Path
+
+from apscheduler.schedulers.blocking import BlockingScheduler
+from apscheduler.triggers.cron import CronTrigger
+from apscheduler.triggers.interval import IntervalTrigger
+from dotenv import load_dotenv
+
+
+HERE = Path(__file__).resolve().parent
+ROOT = HERE.parents[1]
+RTC_DIR = ROOT / "examples" / "tencent_realtime_control"
+for path in (ROOT, HERE, RTC_DIR):
+    if str(path) not in sys.path:
+        sys.path.insert(0, str(path))
+
+load_dotenv(HERE / ".env", override=False)
+load_dotenv(Path.cwd() / ".env", override=False)
+
+from config import TIME_SERIES_DEFAULT  # noqa: E402
+from db.connection import get_connection  # noqa: E402
+from storage import advisory_lock  # noqa: E402
+
+
+logger = logging.getLogger("auto_put_ad_mini.daily_service")
+
+
+def _env_flag(name: str, default: bool) -> bool:
+    raw = os.getenv(name)
+    if raw is None:
+        return default
+    return raw.strip().lower() in {"1", "true", "yes", "on"}
+
+
+def sync_enabled_delivery_templates() -> int:
+    """Keep enabled DB templates aligned with the code's global delivery window."""
+    serialized = json.dumps(TIME_SERIES_DEFAULT)
+    connection = get_connection()
+    try:
+        with connection.cursor() as cursor:
+            return cursor.execute(
+                """
+                UPDATE ad_delivery_template
+                SET time_series_json=%s,
+                    updated_by='ad_daily_service'
+                WHERE enabled=TRUE
+                  AND time_series_json<>%s
+                """,
+                (serialized, serialized),
+            )
+    finally:
+        connection.close()
+
+
+def _run_script(script_name: str, lock_name: str) -> None:
+    with advisory_lock(lock_name) as acquired:
+        if not acquired:
+            logger.warning("Skip %s: another instance holds %s", script_name, lock_name)
+            return
+        logger.info("Starting %s", script_name)
+        completed = subprocess.run(
+            [sys.executable, str(HERE / script_name)],
+            cwd=HERE,
+            check=False,
+        )
+        if completed.returncode:
+            raise RuntimeError(
+                f"{script_name} exited with code {completed.returncode}"
+            )
+        logger.info("Finished %s", script_name)
+
+
+def run_creation() -> None:
+    _run_script(
+        "execute_creation_once.py",
+        os.getenv("DAILY_CREATION_LOCK_NAME", "ad_daily_creation"),
+    )
+
+
+def run_creative_review() -> None:
+    _run_script(
+        "scan_creative_reviews.py",
+        os.getenv("DAILY_REVIEW_LOCK_NAME", "ad_creative_review_scan"),
+    )
+
+
+def main() -> None:
+    if _env_flag("DAILY_SYNC_DELIVERY_TEMPLATE", False):
+        changed = sync_enabled_delivery_templates()
+        logger.info("Synchronized %d enabled delivery template(s)", changed)
+
+    scheduler = BlockingScheduler(timezone="Asia/Shanghai")
+    if _env_flag("DAILY_CREATION_ENABLED", False):
+        creation_hour = int(os.getenv("DAILY_CREATION_HOUR", "10"))
+        creation_minute = int(os.getenv("DAILY_CREATION_MINUTE", "30"))
+        scheduler.add_job(
+            run_creation,
+            CronTrigger(
+                hour=creation_hour,
+                minute=creation_minute,
+                timezone="Asia/Shanghai",
+            ),
+            id="daily_creation",
+            name="广告与创意创建",
+            max_instances=1,
+            coalesce=True,
+            misfire_grace_time=int(
+                os.getenv("DAILY_CREATION_MISFIRE_GRACE_SECONDS", "3600")
+            ),
+        )
+    if _env_flag("DAILY_REVIEW_ENABLED", False):
+        scheduler.add_job(
+            run_creative_review,
+            IntervalTrigger(
+                hours=int(os.getenv("DAILY_REVIEW_INTERVAL_HOURS", "2")),
+                timezone="Asia/Shanghai",
+            ),
+            id="creative_review_scan",
+            name="腾讯创意审核扫描",
+            max_instances=1,
+            coalesce=True,
+            misfire_grace_time=1800,
+        )
+    if _env_flag("DAILY_RUN_ON_STARTUP", False):
+        scheduler.add_job(
+            run_creation,
+            id="daily_creation_startup",
+            name="启动时创建检查",
+        )
+    if _env_flag("DAILY_REVIEW_RUN_ON_STARTUP", False):
+        scheduler.add_job(
+            run_creative_review,
+            id="creative_review_startup",
+            name="启动时审核扫描",
+        )
+    logger.info(
+        "Daily service started jobs=%s",
+        [job.id for job in scheduler.get_jobs()],
+    )
+    scheduler.start()
+
+
+if __name__ == "__main__":
+    logging.basicConfig(
+        level=os.getenv("LOG_LEVEL", "INFO"),
+        format="%(asctime)s %(levelname)s %(name)s %(message)s",
+    )
+    main()

examples/auto_put_ad_mini/sync_feishu_account_config.py

@@ -16,15 +16,21 @@
   - 「生成失败是否回退历史素材」为空/否默认不回退。
   - 「出价方式」为空默认稳定成本;填「最大转化量」时使用「最大转化量出价」作为控制成本。
   - 「投放版位」为空使用投放模板默认;填 AIM+ 时使用智能版位。
+  - 「年龄」支持 30-66+ 或 45-66;为空使用 45-66。
+  - 「地域」支持省市中文名逗号分隔;填 不限 时不传地域定向;为空使用投放模板。
+  - 「日期」是飞书配置生效日期,每日只同步当天日期的行。
 """
 
 from __future__ import annotations
 
 import argparse
 import json
+import re
 import sys
+from datetime import date, datetime
 from pathlib import Path
 from typing import Any
+from zoneinfo import ZoneInfo
 
 import httpx
 from dotenv import load_dotenv
@@ -36,6 +42,8 @@ sys.path.insert(0, str(_HERE))
 
 from configure_creation_accounts import (  # noqa: E402
     AccountConfigInput,
+    DEFAULT_AGE_MAX,
+    DEFAULT_AGE_MIN,
     parse_bid,
     parse_budget_fen,
     set_account_automation_enabled,
@@ -48,6 +56,7 @@ from tools.account_material_strategy import (  # noqa: E402
 from tools.delivery_config import (  # noqa: E402
     BID_MODE_MAX_CONVERSION,
     parse_bid_scene,
+    parse_geo_location_config,
     parse_placement_config,
 )
 from tools.feishu_doc import (  # noqa: E402
@@ -59,12 +68,13 @@ from tools.feishu_doc import (  # noqa: E402
 
 DEFAULT_SPREADSHEET_TOKEN = "D8f4sKEb2hfBnNttX1ucZTklnHf"
 DEFAULT_SHEET_ID = "y3uCcz"
-DEFAULT_RANGE = "A1:L300"
+DEFAULT_RANGE = "A1:N300"
 
 HEADER_ROW_INDEX = 1
 DATA_START_INDEX = 2
 
 REQUIRED_COLUMNS = {
+    "日期": "config_date",
     "账户id": "account_id",
     "人群包": "audience_name",
     "初始出价": "bid",
@@ -79,6 +89,9 @@ OPTIONAL_COLUMNS = {
     "最大转化量出价": "max_conversion_bid",
     "投放版位": "placement",
     "版位": "placement",
+    "年龄": "age",
+    "地域": "geo_location",
+    "投放地域": "geo_location",
 }
 
 
@@ -136,17 +149,52 @@ def _parse_account_id(raw: str) -> int | None:
         raise ValueError(f"账户id非法:{raw}") from e
 
 
+def _parse_config_date(raw: str) -> date:
+    text = raw.strip().replace("/", "-")
+    for fmt in ("%Y%m%d", "%Y-%m-%d"):
+        try:
+            return datetime.strptime(text, fmt).date()
+        except ValueError:
+            continue
+    raise ValueError(f"日期非法:{raw}; 请输入 YYYYMMDD 或 YYYY-MM-DD")
+
+
+def _parse_age_range(raw: str) -> tuple[int, int]:
+    text = raw.strip()
+    if not text:
+        return DEFAULT_AGE_MIN, DEFAULT_AGE_MAX
+    matched = re.fullmatch(r"(\d{1,3})\s*[-~至]\s*(\d{1,3})\+?", text)
+    if not matched:
+        raise ValueError(f"年龄格式非法:{raw}; 示例 30-66+")
+    age_min, age_max = (int(value) for value in matched.groups())
+    if age_min > age_max:
+        raise ValueError(f"年龄范围非法:{raw}")
+    return age_min, age_max
+
+
 def sync_from_feishu(
     spreadsheet_token: str = DEFAULT_SPREADSHEET_TOKEN,
     sheet_id: str = DEFAULT_SHEET_ID,
     cell_range: str = DEFAULT_RANGE,
+    config_date: str | None = None,
     dry_run: bool = False,
     strict: bool = True,
-) -> dict[str, int]:
+) -> dict[str, Any]:
     rows = _fetch_rows(spreadsheet_token, sheet_id, cell_range)
     mapping = _header_map(rows)
+    target_config_date = (
+        _parse_config_date(config_date)
+        if config_date else datetime.now(ZoneInfo("Asia/Shanghai")).date()
+    )
 
-    stats = {"enabled": 0, "disabled": 0, "skipped": 0, "errors": 0}
+    stats = {
+        "config_date": target_config_date.isoformat(),
+        "enabled": 0,
+        "disabled": 0,
+        "skipped": 0,
+        "date_skipped": 0,
+        "errors": 0,
+    }
     enabled_records: list[AccountConfigInput] = []
     disabled_accounts: list[int] = []
     for row_num, row in enumerate(rows[DATA_START_INDEX:], start=DATA_START_INDEX + 1):
@@ -156,6 +204,11 @@ def sync_from_feishu(
                 stats["skipped"] += 1
                 continue
 
+            row_config_date = _parse_config_date(_cell(row, mapping["日期"]))
+            if row_config_date != target_config_date:
+                stats["date_skipped"] += 1
+                continue
+
             enabled_text = _cell(row, mapping["是否自动化执行"])
             enabled = enabled_text == "是"
             if not enabled:
@@ -187,6 +240,9 @@ def sync_from_feishu(
                 _cell(row, mapping[placement_col])
                 if placement_col in mapping else ""
             )
+            age_text = _cell(row, mapping["年龄"]) if "年龄" in mapping else ""
+            geo_col = "地域" if "地域" in mapping else "投放地域"
+            geo_text = _cell(row, mapping[geo_col]) if geo_col in mapping else ""
             bid_min, bid_max, bid_amount = parse_bid(bid_text)
             bid_scene = parse_bid_scene(bid_scene_text)
             custom_cost_cap_fen = None
@@ -198,6 +254,8 @@ def sync_from_feishu(
                     raise ValueError("最大转化量出价必须是固定值")
                 custom_cost_cap_fen = max_bid_amount
             placement = parse_placement_config(placement_text)
+            age_min, age_max = _parse_age_range(age_text)
+            geo_location = parse_geo_location_config(geo_text)
             enabled_records.append(AccountConfigInput(
                 account_id=account_id,
                 audience_name=audience_name,
@@ -214,6 +272,13 @@ def sync_from_feishu(
                     placement.automatic_site_enabled if placement is not None else None
                 ),
                 site_set=(placement.site_set if placement is not None else None),
+                age_min=age_min,
+                age_max=age_max,
+                location_types=(
+                    geo_location.location_types if geo_location is not None else None
+                ),
+                region_ids=(geo_location.region_ids if geo_location is not None else None),
+                config_date=row_config_date,
             ))
             stats["enabled"] += 1
         except Exception as e:
@@ -225,7 +290,12 @@ def sync_from_feishu(
         raise RuntimeError(f"飞书配置表存在 {stats['errors']} 行错误,停止本轮自动化")
 
     for account_id in disabled_accounts:
-        set_account_automation_enabled(account_id, False, dry_run=dry_run)
+        set_account_automation_enabled(
+            account_id,
+            False,
+            config_date=target_config_date,
+            dry_run=dry_run,
+        )
     for record in enabled_records:
         upsert_account_config(record, dry_run=dry_run)
 
@@ -238,6 +308,7 @@ def main() -> int:
     parser.add_argument("--spreadsheet-token", default=DEFAULT_SPREADSHEET_TOKEN)
     parser.add_argument("--sheet-id", default=DEFAULT_SHEET_ID)
     parser.add_argument("--range", default=DEFAULT_RANGE)
+    parser.add_argument("--config-date", help="只同步该配置日期,格式 YYYYMMDD 或 YYYY-MM-DD")
     parser.add_argument("--dry-run", action="store_true")
     parser.add_argument("--allow-row-errors", action="store_true")
     args = parser.parse_args()
@@ -246,6 +317,7 @@ def main() -> int:
         spreadsheet_token=args.spreadsheet_token,
         sheet_id=args.sheet_id,
         cell_range=args.range,
+        config_date=args.config_date,
         dry_run=args.dry_run,
         strict=not args.allow_row_errors,
     )

examples/auto_put_ad_mini/tools/ad_api.py

@@ -42,9 +42,14 @@ TOKEN_API_URL = os.getenv(
     "TENCENT_AD_TOKEN_API",
     "https://api.piaoquantv.com/ad/put/tencent/getAccessToken",
 )
+USER_TOKEN_API_URL = os.getenv(
+    "TENCENT_AD_USER_TOKEN_API",
+    "https://api.piaoquantv.com/ad/put/tencent/getUserToken",
+)
 
 # Token 缓存：避免每次 API 调用都重新获取
 _token_cache: Dict[int, Dict[str, Any]] = {}
+_user_token_cache: Dict[int, Dict[str, Any]] = {}
 _TOKEN_CACHE_TTL = 1800  # 缓存有效期 30 分钟
 
 
@@ -123,11 +128,34 @@ def _get(path: str, params: Dict[str, Any]) -> Dict[str, Any]:
 
 
 def _get_user_token_for_account(account_id: int) -> str:
-    """获取 user_token。优先 DB(account_whitelist.user_token),fallback 到环境变量。
+    """动态获取 user_token,接口不可用时 fallback 到 DB/环境变量。
 
     DB 设计原因(2026-06-05):每个账户可能有不同的经办人授权 token,
     全局环境变量无法覆盖多账户场景。
     """
+    cached = _user_token_cache.get(account_id)
+    if cached and time.time() - cached["ts"] < _TOKEN_CACHE_TTL:
+        return cached["token"]
+
+    try:
+        resp = httpx.get(
+            USER_TOKEN_API_URL,
+            params={"accountId": account_id},
+            timeout=10,
+        )
+        resp.raise_for_status()
+        token = resp.text.strip()
+        if token and len(token) > 10:
+            _user_token_cache[account_id] = {"token": token, "ts": time.time()}
+            logger.info(
+                "[UserTokenAPI] 已获取 account=%s 的 user_token (缓存30分钟)",
+                account_id,
+            )
+            return token
+        logger.warning("[UserTokenAPI] 返回内容异常,降级使用 DB/env")
+    except Exception as e:
+        logger.warning("[UserTokenAPI] 请求失败,降级使用 DB/env: %s", e)
+
     if account_id:
         try:
             from db.connection import get_connection
@@ -157,7 +185,7 @@ def _post(path: str, body: Dict[str, Any]) -> Dict[str, Any]:
     公共参数在 URL query，业务参数在 JSON body。
 
     ⚠️ 重要：腾讯广告写操作需要 user_token（实名认证令牌）
-    优先级:DB account_whitelist.user_token > 环境变量 TENCENT_AD_USER_TOKEN
+    优先级:动态 user token API > DB account_whitelist.user_token > 环境变量
     """
     account_id = body.get("account_id", 0)
     params = _common_params(account_id)

examples/auto_put_ad_mini/tools/ad_creation.py

@@ -289,14 +289,13 @@ def build_ad_request_body(
     """
     begin_date = begin_date or datetime.now().strftime("%Y-%m-%d")
 
-    # 定向 — 固定地域 + 固定年龄 + 可选人群包
-    targeting: dict = {
-        "geo_location": {
+    # 定向 — 账户级年龄/地域 + 可选人群包。
+    targeting: dict = {"age": candidate.age}
+    if candidate.location_types or candidate.region_ids:
+        targeting["geo_location"] = {
             "location_types": candidate.location_types,
             "regions": candidate.region_ids,
-        },
-        "age": candidate.age,
-    }
+        }
     if candidate.custom_audience:
         targeting["custom_audience"] = candidate.custom_audience
 

examples/auto_put_ad_mini/tools/delivery_config.py

@@ -3,6 +3,9 @@
 from __future__ import annotations
 
 from dataclasses import dataclass
+from functools import lru_cache
+import json
+from pathlib import Path
 
 BID_MODE_AVERAGE_COST = "average_cost"
 BID_MODE_MAX_CONVERSION = "max_conversion"
@@ -14,6 +17,12 @@ class PlacementConfig:
     site_set: list[str]
 
 
+@dataclass(frozen=True)
+class GeoLocationConfig:
+    location_types: list[str]
+    region_ids: list[int]
+
+
 _BID_SCENE_VALUES = {
     "": BID_MODE_AVERAGE_COST,
     "平均成本": BID_MODE_AVERAGE_COST,
@@ -72,3 +81,46 @@ def parse_placement_config(raw: object) -> PlacementConfig | None:
     if not site_set:
         raise ValueError(f"投放版位为空:{raw}")
     return PlacementConfig(automatic_site_enabled=False, site_set=site_set)
+
+
+@lru_cache(maxsize=1)
+def _region_name_mapping() -> dict[str, int]:
+    path = Path(__file__).resolve().parents[1] / "data" / "tencent_constants" / "regions_all.json"
+    regions = json.loads(path.read_text(encoding="utf-8"))
+    mapping: dict[str, int] = {}
+    for region in regions:
+        name = str(region.get("name") or "").strip()
+        if not name:
+            continue
+        region_id = int(region["id"])
+        mapping.setdefault(name, region_id)
+        for suffix in ("特别行政区", "壮族自治区", "回族自治区", "维吾尔自治区", "自治区", "省", "市"):
+            if name.endswith(suffix):
+                mapping.setdefault(name[:-len(suffix)], region_id)
+                break
+    return mapping
+
+
+def parse_geo_location_config(raw: object) -> GeoLocationConfig | None:
+    """Parse Feishu geography: blank keeps the delivery template; 不限 omits geo targeting."""
+    text = str(raw or "").strip()
+    if not text:
+        return None
+    if text.lower() in {"不限", "不限地域", "全国", "all"}:
+        return GeoLocationConfig(location_types=[], region_ids=[])
+
+    names = [part.strip() for part in text.replace("，", ",").split(",") if part.strip()]
+    if not names:
+        raise ValueError(f"地域为空:{raw}")
+    name_mapping = _region_name_mapping()
+    region_ids: list[int] = []
+    for name in names:
+        try:
+            region_id = int(name)
+        except ValueError:
+            region_id = name_mapping.get(name, 0)
+        if not region_id:
+            raise ValueError(f"未知地域:{name}")
+        if region_id not in region_ids:
+            region_ids.append(region_id)
+    return GeoLocationConfig(location_types=["LIVE_IN"], region_ids=region_ids)

examples/tencent_realtime_control/.env.example

@@ -1,5 +1,5 @@
-RTC_START_HOUR=7
-RTC_STOP_HOUR=21
+RTC_START_HOUR=6
+RTC_STOP_HOUR=20
 RTC_POLL_SECONDS=600
 RTC_HIGH_CPM=250
 RTC_LOW_CPM=190
@@ -12,6 +12,12 @@ RTC_FEISHU_CHAT_ID=
 RTC_FEISHU_TIMEOUT_SECONDS=30
 RTC_VERIFY_ATTEMPTS=3
 RTC_VERIFY_DELAY_SECONDS=1
+RTC_APPLY_ENABLED=0
+RTC_COMMAND_ENABLED=1
+RTC_COMMAND_CHAT_ID=
+RTC_COMMAND_ALLOWED_OPEN_IDS=
+RTC_COMMAND_CONFIRM_TTL_MINUTES=10
+RTC_COMMAND_WORKERS=2
 
 ODPS_ACCESS_ID=
 ODPS_ACCESS_SECRET=

examples/tencent_realtime_control/README.md

@@ -7,21 +7,22 @@
 - 从 `loghubods.advertiser_data_da_hour` 读取当天整体小时数据。
 - CPM 直接使用 `真实cpm_总`。
 - 固定过滤名称、广告主、公司、行业、客户和落地页类型均为 `SUM`。
-- 账户只取 `ad_creation_account_config.enabled=1` 与
-  `account_whitelist.enabled=1` 的交集。
+- 实时管理账户取历史 `ad_creation_account_config` 与
+  `account_whitelist.enabled=1` 的交集；关闭创建配置只停止新建和补创意，
+  不会让既有广告退出实时管理。
 - 普通出价使用 `bid_amount`，最大转化量使用 `custom_cost_cap`。
 - 每条广告首次纳管时将腾讯当前实际出价持久化为基础出价。
 
 ## 调控规则
 
-北京时间每天 `07:00-21:00` 每 10 分钟检查一次：
+北京时间每天 `06:00-20:00` 每 10 分钟检查一次：
 
 - `CPM > 250`：基础出价上调 25%。
 - `190 <= CPM <= 250`：恢复基础出价并恢复策略暂停的广告。
 - `CPM < 190`：先恢复基础出价，再暂停广告。
 - 同一广告每天最多上调一次；恢复基础价后，当天也不会再次上调。
-- `21:00`：先恢复基础出价，再暂停广告。
-- 次日 `07:00`：只恢复由本策略暂停的广告，不打开人工暂停广告。
+- `20:00`：先恢复基础出价，再暂停广告。
+- 次日 `06:00`：只恢复由本策略暂停的广告，不打开人工暂停广告。
 
 每个真实发生调价或关停动作的窗口会生成一张飞书在线表格并发送到调控群。
 群消息包含触发时间、数据分区、当前 CPM、决策、影响账户/广告数、动作分布和
@@ -39,7 +40,7 @@
 
 ODPS 小时分区可能延迟。当天尚未出现 `06` 点及之后的分区时只等待，不使用
 夜间低流量分区做关停判断。最新分区相对当前时间最多允许延迟 2 小时，超过后
-本轮不执行基于 CPM 的调价、恢复或关停；每天 07:00 的定时恢复不依赖 CPM。
+本轮不执行基于 CPM 的调价、恢复或关停；每天 06:00 的定时恢复不依赖 CPM。
 
 状态保存在 MySQL，使用数据库锁避免多 Pod 同时执行。相同分区和相同决策默认
 不重复扫描腾讯；每 60 分钟至少刷新一次广告清单以纳管新广告。
@@ -99,27 +100,94 @@ ODPS 小时分区可能延迟。当天尚未出现 `06` 点及之后的分区时
 .venv/bin/python examples/tencent_realtime_control/run_once.py --apply
 ```
 
+## 一次性延后生效日期
+
+需要让全部自动化广告在指定日期才开始投放时，先 dry-run：
+
+```bash
+.venv/bin/python \
+  examples/tencent_realtime_control/schedule_automated_ads_from_date.py \
+  --begin-date 2026-07-26
+```
+
+确认账户数、广告数、待更新数和待恢复数后真实执行：
+
+```bash
+.venv/bin/python \
+  examples/tencent_realtime_control/schedule_automated_ads_from_date.py \
+  --begin-date 2026-07-26 \
+  --apply
+```
+
+脚本只处理当前启用的自动化账户。它保留广告原有 `end_date` 和
+`time_series`，逐账户更新 `begin_date` 并回读校验；日期更新成功后只恢复
+由实时策略暂停的广告，不会打开人工暂停广告。默认 dry-run，必须显式传
+`--apply` 才会修改腾讯广告。
+
 ## Docker 调度
 
-长驻进程会在投放时段按 10 分钟边界执行，21 点收尾后休眠到次日 7 点：
+统一控制服务在投放时段按 10 分钟边界执行，20 点收尾后休眠到次日 6 点：
 
 ```bash
-.venv/bin/python examples/tencent_realtime_control/run_scheduler.py --apply
+RTC_APPLY_ENABLED=0 \
+  .venv/bin/python \
+  examples/tencent_realtime_control/run_control_service.py
 ```
 
-默认所有执行入口均为 dry-run，必须显式传 `--apply` 才会调用腾讯写接口。
+默认所有执行入口均为 dry-run，必须设置 `RTC_APPLY_ENABLED=1` 或显式传
+`--apply` 才会调用腾讯写接口。
 `--at` 和 `--cpm-override` 只允许 dry-run。
 
-构建独立生产镜像：
+生产使用统一完整镜像：
 
 ```bash
 docker build \
-  -f Dockerfile.tencent_realtime_control \
-  -t tencent-realtime-control:local .
+  -f Dockerfile.auto_put_ad_mini \
+  -t ad-put-agent:local .
+```
+
+Compose 从同一镜像启动 `ad-control-service` 和 `ad-daily-service`。完整部署和
+dry-run 切换步骤见
+`examples/auto_put_ad_mini/docs/unified_services_deployment.md`。
+
+## ECS 首次迁移 ODPS 配置
+
+仅用于旧 `auto_put_ad_mini` 容器仍在运行、但 ECS 的
+`/home/server/.env` 缺少 ODPS 环境变量时。命令不会在终端打印密钥值。
+
+先从旧容器迁移配置：
+
+```bash
+cd /home/server
+
+sed -i \
+  '/^ODPS_ACCESS_ID=/d;/^ODPS_ACCESS_SECRET=/d;/^ODPS_PROJECT=/d' \
+  /home/server/.env
+
+docker exec auto_put_ad_mini python -c 'import ast,shlex; tree=ast.parse(open("/app/examples/auto_put_ad_mini/tools/odps_module.py",encoding="utf-8").read()); values={target.attr:node.value.value for node in ast.walk(tree) if isinstance(node,ast.Assign) and isinstance(node.value,ast.Constant) for target in node.targets if isinstance(target,ast.Attribute)}; print("ODPS_ACCESS_ID="+shlex.quote(values["accessId"])); print("ODPS_ACCESS_SECRET="+shlex.quote(values["accessSecret"])); print("ODPS_PROJECT=loghubods")' >> /home/server/.env
+
+chmod 600 /home/server/.env
+```
+
+确认变量名已经写入，不显示变量值：
+
+```bash
+grep -E '^ODPS_(ACCESS_ID|ACCESS_SECRET|PROJECT)=' /home/server/.env \
+  | cut -d= -f1
+```
+
+验证 ODPS 查询：
+
+```bash
+cd /home/server
+VERSION=latest docker compose --env-file /dev/null run --rm \
+  ad_control_service \
+  python /app/examples/tencent_realtime_control/fetch_daily_hourly_cpm.py \
+  --date 20260724
 ```
 
-镜像默认命令是 `python run_scheduler.py --apply`。首次上线应先覆盖默认命令执行
-`python run_once.py` 完成 dry-run，确认无误后再启动默认调度命令。
+这只是旧配置迁移手段。迁移完成后应轮换已经硬编码在旧代码和 Git 历史中的
+AccessKey，并将新值只保存在受权限保护的生产环境配置中。
 
 ## 环境变量
 

examples/tencent_realtime_control/feishu_command_service.py

@@ -0,0 +1,201 @@
+"""Feishu WebSocket entry for deterministic advertising control commands."""
+
+from __future__ import annotations
+
+import logging
+import os
+from concurrent.futures import ThreadPoolExecutor
+from datetime import datetime
+from typing import Any
+from zoneinfo import ZoneInfo
+
+from agent.tools.builtin.feishu.feishu_client import (
+    ChatType,
+    FeishuClient,
+    FeishuMessageEvent,
+)
+
+from operator_commands import (
+    ACTION_CANCEL,
+    ACTION_CONFIRM,
+    ACTION_DAY_PAUSE,
+    ACTION_RESUME,
+    ACTION_STATUS,
+    ACTION_STOP,
+    parse_command,
+)
+from operator_control import (
+    cancel_command,
+    execute_confirmed_command,
+    pause_status_summary,
+    preview_write_command,
+)
+from realtime_config import RealtimeControlConfig
+
+
+SHANGHAI = ZoneInfo("Asia/Shanghai")
+logger = logging.getLogger("tencent_realtime_control.feishu_commands")
+
+_ACTION_LABELS = {
+    ACTION_DAY_PAUSE: "仅暂停今天",
+    ACTION_STOP: "持续停止",
+    ACTION_RESUME: "恢复投放",
+}
+
+
+def _split_ids(raw: str) -> set[str]:
+    normalized = str(raw or "").replace("，", ",").replace(" ", ",")
+    return {value.strip() for value in normalized.split(",") if value.strip()}
+
+
+class FeishuCommandService:
+    def __init__(self, *, apply: bool) -> None:
+        app_id = os.getenv("FEISHU_APP_ID", "").strip()
+        app_secret = os.getenv("FEISHU_APP_SECRET", "").strip()
+        if not app_id or not app_secret:
+            raise RuntimeError("FEISHU_APP_ID and FEISHU_APP_SECRET are required")
+        self.allowed_chat_id = (
+            os.getenv("RTC_COMMAND_CHAT_ID", "").strip()
+            or os.getenv("FEISHU_AD_PROJECT_CHAT_ID", "").strip()
+        )
+        self.allowed_open_ids = _split_ids(
+            os.getenv("RTC_COMMAND_ALLOWED_OPEN_IDS", "")
+            or os.getenv("FEISHU_OPERATOR_OPEN_ID", "")
+        )
+        if not self.allowed_chat_id:
+            raise RuntimeError(
+                "RTC_COMMAND_CHAT_ID/FEISHU_AD_PROJECT_CHAT_ID is required"
+            )
+        if not self.allowed_open_ids:
+            raise RuntimeError(
+                "RTC_COMMAND_ALLOWED_OPEN_IDS/FEISHU_OPERATOR_OPEN_ID is required"
+            )
+        self.apply = apply
+        self.config = RealtimeControlConfig.from_env()
+        self.confirmation_ttl_minutes = int(
+            os.getenv("RTC_COMMAND_CONFIRM_TTL_MINUTES", "10")
+        )
+        if self.confirmation_ttl_minutes < 1:
+            raise ValueError("RTC_COMMAND_CONFIRM_TTL_MINUTES must be positive")
+        self.client = FeishuClient(app_id=app_id, app_secret=app_secret)
+        self.executor = ThreadPoolExecutor(
+            max_workers=int(os.getenv("RTC_COMMAND_WORKERS", "2")),
+            thread_name_prefix="feishu-ad-command",
+        )
+
+    def start(self) -> Any:
+        return self.client.start_websocket(
+            on_message=self._enqueue_message,
+            blocking=False,
+        )
+
+    def _enqueue_message(self, event: FeishuMessageEvent) -> None:
+        self.executor.submit(self.handle_message, event)
+
+    def _reply(self, event: FeishuMessageEvent, text: str) -> None:
+        self.client.send_message(
+            to=event.chat_id,
+            text=text,
+            reply_to_message_id=event.message_id,
+        )
+
+    def _authorized(self, event: FeishuMessageEvent) -> bool:
+        if event.sender_open_id not in self.allowed_open_ids:
+            return False
+        if event.chat_type == ChatType.GROUP:
+            return (
+                event.chat_id == self.allowed_chat_id
+                and event.mentioned_bot
+            )
+        return event.chat_type == ChatType.P2P
+
+    def handle_message(self, event: FeishuMessageEvent) -> None:
+        if event.content_type not in {"text", "post"}:
+            return
+        if not self._authorized(event):
+            return
+        try:
+            parsed = parse_command(event.content)
+            if parsed is None:
+                return
+            now = datetime.now(SHANGHAI)
+            if parsed.action == ACTION_STATUS:
+                summary = pause_status_summary()
+                accounts = ",".join(str(value) for value in summary["accounts"]) or "无"
+                self._reply(
+                    event,
+                    "当前运营暂停状态\n"
+                    f"- 暂停广告：{summary['total']} 条\n"
+                    f"- 仅暂停今天：{summary['until_next_delivery']} 条\n"
+                    f"- 持续停止：{summary['until_manual']} 条\n"
+                    f"- 涉及账户：{accounts}",
+                )
+                return
+            if parsed.action == ACTION_CANCEL:
+                command = cancel_command(
+                    parsed.command_id or "",
+                    event.sender_open_id,
+                    now,
+                )
+                message = (
+                    f"命令 {command['command_id']} 已取消"
+                    if command["status"] == "CANCELLED"
+                    else (
+                        f"命令 {command['command_id']} 未取消，"
+                        f"当前状态：{command['status']}"
+                    )
+                )
+                self._reply(
+                    event,
+                    message,
+                )
+                return
+            if parsed.action == ACTION_CONFIRM:
+                if not self.apply:
+                    self._reply(event, "当前服务为 dry-run，禁止执行腾讯写操作。")
+                    return
+                command = execute_confirmed_command(
+                    parsed.command_id or "",
+                    sender_open_id=event.sender_open_id,
+                    now=now,
+                    start_hour=self.config.start_hour,
+                    lock_name=self.config.lock_name,
+                )
+                self._reply(
+                    event,
+                    f"命令 {command['command_id']} 执行完成\n"
+                    f"- 状态：{command['status']}\n"
+                    f"- 成功：{command.get('successes', 0)} 条\n"
+                    f"- 失败：{command.get('failures', 0)} 条",
+                )
+                return
+
+            command = preview_write_command(
+                parsed,
+                now=now,
+                source_message_id=event.message_id,
+                chat_id=event.chat_id,
+                sender_open_id=event.sender_open_id,
+                sender_name=event.sender_name,
+                confirmation_ttl_minutes=self.confirmation_ttl_minutes,
+                start_hour=self.config.start_hour,
+            )
+            resume_text = (
+                command["resume_at"].strftime("%Y-%m-%d %H:%M")
+                if command.get("resume_at")
+                else "仅人工明确恢复"
+            )
+            self._reply(
+                event,
+                f"操作预览：{_ACTION_LABELS[command['action']]}\n"
+                f"- 账户：{command['preview_account_count']} 个\n"
+                f"- 预计涉及广告：{command['preview_ad_count']} 条\n"
+                f"- 恢复时间：{resume_text}\n"
+                f"- 命令ID：{command['command_id']}\n"
+                f"请在 {self.confirmation_ttl_minutes} 分钟内回复："
+                f"确认 {command['command_id']}\n"
+                f"取消命令请回复：取消 {command['command_id']}",
+            )
+        except Exception as exc:
+            logger.exception("Feishu command failed")
+            self._reply(event, f"命令处理失败：{exc}")

examples/tencent_realtime_control/feishu_notifier.py

@@ -395,6 +395,25 @@ class FeishuNotifier:
         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,
         *,
@@ -550,3 +569,23 @@ def send_action_notification(
             "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}"
+    )

examples/tencent_realtime_control/operator_commands.py

@@ -0,0 +1,94 @@
+"""Deterministic parsing for Feishu advertising control commands."""
+
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass
+
+
+ACTION_DAY_PAUSE = "DAY_PAUSE"
+ACTION_STOP = "STOP"
+ACTION_RESUME = "RESUME"
+ACTION_STATUS = "STATUS"
+ACTION_CONFIRM = "CONFIRM"
+ACTION_CANCEL = "CANCEL"
+
+WRITE_ACTIONS = {ACTION_DAY_PAUSE, ACTION_STOP, ACTION_RESUME}
+
+_COMMAND_ID_RE = re.compile(r"\bcmd_[0-9A-Za-z_-]+\b", re.IGNORECASE)
+_ACCOUNT_ID_RE = re.compile(r"(?<!\d)(\d{7,12})(?!\d)")
+
+
+@dataclass(frozen=True)
+class ParsedCommand:
+    action: str
+    scope_type: str = "ACCOUNTS"
+    account_ids: tuple[int, ...] = ()
+    command_id: str | None = None
+
+
+def normalize_text(raw: str) -> str:
+    text = str(raw or "").strip()
+    translations = str.maketrans(
+        {
+            "，": ",",
+            "、": ",",
+            "；": ",",
+            ";": ",",
+            "：": " ",
+            ":": " ",
+            "\t": " ",
+            "\r": " ",
+            "\n": " ",
+        }
+    )
+    return re.sub(r"\s+", " ", text.translate(translations)).strip()
+
+
+def parse_command(raw: str) -> ParsedCommand | None:
+    text = normalize_text(raw)
+    if not text:
+        return None
+
+    lowered = text.lower()
+    command_match = _COMMAND_ID_RE.search(text)
+    if lowered.startswith("确认"):
+        if not command_match:
+            raise ValueError("确认命令缺少 command_id")
+        return ParsedCommand(
+            action=ACTION_CONFIRM,
+            command_id=command_match.group(0).lower(),
+        )
+    if lowered.startswith("取消"):
+        if not command_match:
+            raise ValueError("取消命令缺少 command_id")
+        return ParsedCommand(
+            action=ACTION_CANCEL,
+            command_id=command_match.group(0).lower(),
+        )
+
+    if "查看暂停状态" in text or "查询暂停状态" in text:
+        return ParsedCommand(action=ACTION_STATUS, scope_type="ALL")
+
+    if text.startswith(("停止", "持续停止", "永久停止")):
+        action = ACTION_STOP
+    elif text.startswith(("暂停", "今天暂停", "暂停到明天", "今天不投")):
+        action = ACTION_DAY_PAUSE
+    elif text.startswith(("恢复", "开启", "继续投放")):
+        action = ACTION_RESUME
+    else:
+        return None
+
+    if "全部" in text or "所有" in text:
+        return ParsedCommand(action=action, scope_type="ALL")
+
+    account_ids = tuple(
+        dict.fromkeys(int(value) for value in _ACCOUNT_ID_RE.findall(text))
+    )
+    if not account_ids:
+        raise ValueError("命令中没有有效账户 ID，也没有指定“全部”")
+    return ParsedCommand(
+        action=action,
+        scope_type="ACCOUNTS",
+        account_ids=account_ids,
+    )

examples/tencent_realtime_control/operator_control.py

@@ -0,0 +1,531 @@
+"""Preview and execute confirmed Feishu operator commands."""
+
+from __future__ import annotations
+
+import logging
+import uuid
+from datetime import datetime, timedelta
+from typing import Any
+
+from operator_commands import (
+    ACTION_DAY_PAUSE,
+    ACTION_RESUME,
+    ACTION_STOP,
+    ParsedCommand,
+)
+from storage import (
+    advisory_lock,
+    clear_operator_pause,
+    create_operator_command,
+    insert_operator_command_item,
+    load_ad_states,
+    load_managed_accounts,
+    load_operator_command,
+    load_operator_pauses,
+    set_operator_pause,
+    transition_operator_command,
+    update_operator_command,
+    upsert_ad_state,
+)
+from tencent_client import (
+    ACTIVE_STATUS,
+    SUSPEND_STATUS,
+    PostWriteVerificationError,
+    TencentWriteNotSentError,
+    TencentWriteOutcomeUnknownError,
+    TencentWriteRejectedError,
+    TencentClient,
+    current_bid_fen,
+    resolve_bid_field,
+)
+
+
+COMMAND_PENDING = "PENDING_CONFIRMATION"
+COMMAND_EXECUTING = "EXECUTING"
+COMMAND_SUCCEEDED = "SUCCEEDED"
+COMMAND_PARTIAL = "PARTIAL"
+COMMAND_FAILED = "FAILED"
+COMMAND_CANCELLED = "CANCELLED"
+COMMAND_EXPIRED = "EXPIRED"
+FINAL_COMMAND_STATUSES = {
+    COMMAND_SUCCEEDED,
+    COMMAND_PARTIAL,
+    COMMAND_FAILED,
+    COMMAND_CANCELLED,
+    COMMAND_EXPIRED,
+}
+
+PAUSE_UNTIL_NEXT_DELIVERY = "UNTIL_NEXT_DELIVERY"
+PAUSE_UNTIL_MANUAL = "UNTIL_MANUAL"
+logger = logging.getLogger("tencent_realtime_control.operator_control")
+
+
+def _managed_account_map() -> dict[int, dict[str, Any]]:
+    return {
+        int(row["account_id"]): row
+        for row in load_managed_accounts()
+    }
+
+
+def resolve_target_accounts(parsed: ParsedCommand) -> list[dict[str, Any]]:
+    managed = _managed_account_map()
+    if parsed.scope_type == "ALL":
+        accounts = list(managed.values())
+        if not accounts:
+            raise ValueError("当前没有启用白名单的自动化管理账户")
+        return accounts
+    missing = [account_id for account_id in parsed.account_ids if account_id not in managed]
+    if missing:
+        raise ValueError(f"账户不在自动化管理范围: {missing}")
+    return [managed[account_id] for account_id in parsed.account_ids]
+
+
+def _next_delivery_start(now: datetime, start_hour: int) -> datetime:
+    return datetime.combine(
+        now.date() + timedelta(days=1),
+        datetime.min.time().replace(hour=start_hour),
+        now.tzinfo,
+    )
+
+
+def preview_write_command(
+    parsed: ParsedCommand,
+    *,
+    now: datetime,
+    source_message_id: str,
+    chat_id: str,
+    sender_open_id: str,
+    sender_name: str | None,
+    confirmation_ttl_minutes: int,
+    start_hour: int,
+    tencent: TencentClient | None = None,
+) -> dict[str, Any]:
+    accounts = resolve_target_accounts(parsed)
+    client = tencent or TencentClient()
+    preview_ad_count = 0
+    if parsed.action in {ACTION_DAY_PAUSE, ACTION_STOP}:
+        for account in accounts:
+            ads = client.get_ads(int(account["account_id"]))
+            preview_ad_count += sum(
+                str(ad.get("configured_status") or "") == ACTIVE_STATUS
+                for ad in ads
+            )
+    elif parsed.action == ACTION_RESUME:
+        preview_ad_count = len(
+            load_operator_pauses([int(row["account_id"]) for row in accounts])
+        )
+    else:
+        raise ValueError(f"Unsupported write action: {parsed.action}")
+
+    command_id = f"cmd_{now.strftime('%Y%m%d%H%M%S')}_{uuid.uuid4().hex[:8]}"
+    expires_at = now + timedelta(minutes=confirmation_ttl_minutes)
+    command = create_operator_command(
+        {
+            "command_id": command_id,
+            "source_message_id": source_message_id,
+            "chat_id": chat_id,
+            "sender_open_id": sender_open_id,
+            "sender_name": sender_name,
+            "action": parsed.action,
+            "scope_type": parsed.scope_type,
+            "target_account_ids": [int(row["account_id"]) for row in accounts],
+            "status": COMMAND_PENDING,
+            "preview_account_count": len(accounts),
+            "preview_ad_count": preview_ad_count,
+            "expires_at": expires_at,
+        }
+    )
+    command["resume_at"] = (
+        _next_delivery_start(now, start_hour)
+        if parsed.action == ACTION_DAY_PAUSE
+        else None
+    )
+    return command
+
+
+def cancel_command(command_id: str, sender_open_id: str, now: datetime) -> dict[str, Any]:
+    command = load_operator_command(command_id)
+    if not command:
+        raise ValueError(f"命令不存在: {command_id}")
+    if command["sender_open_id"] != sender_open_id:
+        raise PermissionError("只能取消自己发起的命令")
+    if command["status"] != COMMAND_PENDING:
+        return command
+    transition_operator_command(
+        command_id,
+        expected_statuses={COMMAND_PENDING},
+        target_status=COMMAND_CANCELLED,
+        executed_at=now,
+    )
+    return load_operator_command(command_id) or command
+
+
+def _ensure_state(
+    *,
+    account: dict[str, Any],
+    ad: dict[str, Any],
+    states: dict[int, dict[str, Any]],
+    now: datetime,
+) -> dict[str, Any]:
+    adgroup_id = int(ad["adgroup_id"])
+    state = states.get(adgroup_id)
+    if state:
+        return state
+    bid_field = resolve_bid_field(ad, account.get("bid_scene"))
+    base_bid = current_bid_fen(ad, bid_field)
+    if base_bid is None or base_bid <= 0:
+        raise ValueError(
+            f"广告缺少有效基础出价: account={account['account_id']} ad={adgroup_id}"
+        )
+    upsert_ad_state(
+        account_id=int(account["account_id"]),
+        adgroup_id=adgroup_id,
+        adgroup_name=str(ad.get("adgroup_name") or ""),
+        bid_field=bid_field,
+        base_bid_fen=base_bid,
+        boosted_date=None,
+        paused_by_strategy=False,
+        pause_reason=None,
+        last_action="REGISTER_BASE",
+        action_at=now,
+    )
+    state = {
+        "account_id": int(account["account_id"]),
+        "adgroup_id": adgroup_id,
+        "bid_field": bid_field,
+        "base_bid_fen": base_bid,
+        "paused_by_strategy": False,
+    }
+    states[adgroup_id] = state
+    return state
+
+
+def _record_item(command: dict[str, Any], account: dict[str, Any], **values: Any) -> None:
+    insert_operator_command_item(
+        {
+            "command_id": command["command_id"],
+            "account_id": int(account["account_id"]),
+            "audience_name": account.get("audience_name"),
+            **values,
+        }
+    )
+
+
+def _execute_pause(
+    command: dict[str, Any],
+    account: dict[str, Any],
+    *,
+    now: datetime,
+    start_hour: int,
+    client: TencentClient,
+) -> tuple[int, int]:
+    account_id = int(account["account_id"])
+    ads = client.get_ads(account_id)
+    states = load_ad_states(account_id)
+    successes = failures = 0
+    mode = (
+        PAUSE_UNTIL_NEXT_DELIVERY
+        if command["action"] == ACTION_DAY_PAUSE
+        else PAUSE_UNTIL_MANUAL
+    )
+    resume_at = _next_delivery_start(now, start_hour) if mode == PAUSE_UNTIL_NEXT_DELIVERY else None
+    for ad in ads:
+        adgroup_id = int(ad.get("adgroup_id") or 0)
+        before_status = str(ad.get("configured_status") or "")
+        if before_status != ACTIVE_STATUS:
+            _record_item(
+                command,
+                account,
+                adgroup_id=adgroup_id,
+                adgroup_name=ad.get("adgroup_name"),
+                before_status=before_status,
+                target_status=SUSPEND_STATUS,
+                readback_status=before_status,
+                execution_status="SKIPPED_NOT_ACTIVE",
+            )
+            continue
+        tencent_updated = False
+        try:
+            _ensure_state(account=account, ad=ad, states=states, now=now)
+            set_operator_pause(
+                account_id=account_id,
+                adgroup_id=adgroup_id,
+                mode=mode,
+                resume_at=resume_at,
+                command_id=command["command_id"],
+                paused_from_status=before_status,
+                paused_at=now,
+            )
+            readback = client.update_ad(
+                account_id,
+                adgroup_id,
+                target_status=SUSPEND_STATUS,
+            )
+            tencent_updated = True
+            _record_item(
+                command,
+                account,
+                adgroup_id=adgroup_id,
+                adgroup_name=ad.get("adgroup_name"),
+                before_status=before_status,
+                target_status=SUSPEND_STATUS,
+                readback_status=readback.get("configured_status"),
+                execution_status="SUCCESS",
+            )
+            successes += 1
+        except Exception as exc:
+            if isinstance(
+                exc,
+                (PostWriteVerificationError, TencentWriteOutcomeUnknownError),
+            ):
+                tencent_updated = True
+            safe_to_clear = isinstance(
+                exc,
+                (TencentWriteNotSentError, TencentWriteRejectedError),
+            )
+            if not tencent_updated and safe_to_clear:
+                clear_operator_pause(
+                    account_id,
+                    adgroup_id,
+                    action="OPERATOR_PAUSE_FAILED",
+                    action_at=now,
+                )
+            try:
+                _record_item(
+                    command,
+                    account,
+                    adgroup_id=adgroup_id,
+                    adgroup_name=ad.get("adgroup_name"),
+                    before_status=before_status,
+                    target_status=SUSPEND_STATUS,
+                    readback_status=(
+                        exc.actual.get("configured_status")
+                        if isinstance(exc, PostWriteVerificationError)
+                        else SUSPEND_STATUS
+                        if tencent_updated
+                        else None
+                    ),
+                    execution_status=(
+                        "VERIFY_FAILED"
+                        if isinstance(exc, PostWriteVerificationError)
+                        else "OUTCOME_UNKNOWN"
+                        if isinstance(exc, TencentWriteOutcomeUnknownError)
+                        else "AUDIT_FAILED"
+                        if tencent_updated
+                        else "FAILED"
+                    ),
+                    error_message=str(exc),
+                )
+            except Exception:
+                logger.exception(
+                    "Failed to persist operator command failure item "
+                    "command=%s account=%s ad=%s",
+                    command["command_id"],
+                    account_id,
+                    adgroup_id,
+                )
+            failures += 1
+    return successes, failures
+
+
+def _execute_resume(
+    command: dict[str, Any],
+    account: dict[str, Any],
+    *,
+    now: datetime,
+    client: TencentClient,
+) -> tuple[int, int]:
+    account_id = int(account["account_id"])
+    pauses = load_operator_pauses([account_id])
+    if not pauses:
+        return 0, 0
+    ads = {
+        int(ad.get("adgroup_id") or 0): ad
+        for ad in client.get_ads(account_id)
+    }
+    successes = failures = 0
+    for state in pauses:
+        adgroup_id = int(state["adgroup_id"])
+        ad = ads.get(adgroup_id)
+        if not ad:
+            _record_item(
+                command,
+                account,
+                adgroup_id=adgroup_id,
+                adgroup_name=state.get("adgroup_name"),
+                execution_status="FAILED",
+                error_message="Tencent ad not found",
+            )
+            failures += 1
+            continue
+        before_status = str(ad.get("configured_status") or "")
+        try:
+            readback_status = before_status
+            target_status = None
+            if before_status == SUSPEND_STATUS and not bool(state.get("paused_by_strategy")):
+                target_status = ACTIVE_STATUS
+                readback = client.update_ad(
+                    account_id,
+                    adgroup_id,
+                    target_status=ACTIVE_STATUS,
+                )
+                readback_status = str(readback.get("configured_status") or "")
+            clear_operator_pause(
+                account_id,
+                adgroup_id,
+                action="OPERATOR_RESUME",
+                action_at=now,
+            )
+            _record_item(
+                command,
+                account,
+                adgroup_id=adgroup_id,
+                adgroup_name=ad.get("adgroup_name"),
+                before_status=before_status,
+                target_status=target_status,
+                readback_status=readback_status,
+                execution_status="SUCCESS",
+            )
+            successes += 1
+        except Exception as exc:
+            _record_item(
+                command,
+                account,
+                adgroup_id=adgroup_id,
+                adgroup_name=ad.get("adgroup_name"),
+                before_status=before_status,
+                target_status=ACTIVE_STATUS,
+                execution_status="FAILED",
+                error_message=str(exc),
+            )
+            failures += 1
+    return successes, failures
+
+
+def execute_confirmed_command(
+    command_id: str,
+    *,
+    sender_open_id: str,
+    now: datetime,
+    start_hour: int,
+    lock_name: str,
+    tencent: TencentClient | None = None,
+) -> dict[str, Any]:
+    command = load_operator_command(command_id)
+    if not command:
+        raise ValueError(f"命令不存在: {command_id}")
+    if command["sender_open_id"] != sender_open_id:
+        raise PermissionError("只能确认自己发起的命令")
+    if command["status"] in FINAL_COMMAND_STATUSES:
+        return command
+    if command["status"] not in {COMMAND_PENDING, COMMAND_EXECUTING}:
+        raise RuntimeError(f"命令当前不可确认: {command['status']}")
+    with advisory_lock(lock_name) as acquired:
+        if not acquired:
+            raise RuntimeError("实时控制正在执行，请稍后再次确认")
+        command = load_operator_command(command_id) or command
+        if command["status"] in FINAL_COMMAND_STATUSES:
+            return command
+        if command["status"] not in {COMMAND_PENDING, COMMAND_EXECUTING}:
+            raise RuntimeError(f"命令当前不可确认: {command['status']}")
+        if (
+            command["status"] == COMMAND_PENDING
+            and command.get("expires_at")
+            and now.replace(tzinfo=None) > command["expires_at"]
+        ):
+            transition_operator_command(
+                command_id,
+                expected_statuses={COMMAND_PENDING},
+                target_status=COMMAND_EXPIRED,
+                executed_at=now,
+            )
+            return load_operator_command(command_id) or command
+        if command["status"] == COMMAND_PENDING:
+            transitioned = transition_operator_command(
+                command_id,
+                expected_statuses={COMMAND_PENDING},
+                target_status=COMMAND_EXECUTING,
+                confirmed_at=now,
+            )
+            if not transitioned:
+                latest = load_operator_command(command_id) or command
+                if latest["status"] in FINAL_COMMAND_STATUSES:
+                    return latest
+                raise RuntimeError(
+                    f"命令状态并发变化: {latest['status']}"
+                )
+        try:
+            managed = _managed_account_map()
+            missing = [
+                account_id
+                for account_id in command["target_account_ids"]
+                if account_id not in managed
+            ]
+            if missing:
+                raise RuntimeError(
+                    f"确认时账户已不在自动化管理范围: {missing}"
+                )
+            accounts = [
+                managed[account_id]
+                for account_id in command["target_account_ids"]
+            ]
+            client = tencent or TencentClient()
+            successes = failures = 0
+            for account in accounts:
+                if command["action"] in {ACTION_DAY_PAUSE, ACTION_STOP}:
+                    ok, failed = _execute_pause(
+                        command,
+                        account,
+                        now=now,
+                        start_hour=start_hour,
+                        client=client,
+                    )
+                elif command["action"] == ACTION_RESUME:
+                    ok, failed = _execute_resume(
+                        command,
+                        account,
+                        now=now,
+                        client=client,
+                    )
+                else:
+                    raise ValueError(f"Unsupported command action: {command['action']}")
+                successes += ok
+                failures += failed
+        except Exception as exc:
+            update_operator_command(
+                command_id,
+                COMMAND_FAILED,
+                executed_at=now,
+                error_message=str(exc),
+            )
+            raise
+
+        status = (
+            COMMAND_FAILED
+            if failures and not successes
+            else COMMAND_PARTIAL
+            if failures
+            else COMMAND_SUCCEEDED
+        )
+        update_operator_command(command_id, status, executed_at=now)
+        result = load_operator_command(command_id) or command
+        result["successes"] = successes
+        result["failures"] = failures
+        return result
+
+
+def pause_status_summary() -> dict[str, Any]:
+    account_ids = [int(row["account_id"]) for row in load_managed_accounts()]
+    pauses = load_operator_pauses(account_ids)
+    return {
+        "total": len(pauses),
+        "until_next_delivery": sum(
+            row["operator_pause_mode"] == PAUSE_UNTIL_NEXT_DELIVERY
+            for row in pauses
+        ),
+        "until_manual": sum(
+            row["operator_pause_mode"] == PAUSE_UNTIL_MANUAL
+            for row in pauses
+        ),
+        "accounts": sorted({int(row["account_id"]) for row in pauses}),
+    }

examples/tencent_realtime_control/realtime_config.py

@@ -9,8 +9,8 @@ from decimal import Decimal
 
 @dataclass(frozen=True)
 class RealtimeControlConfig:
-    start_hour: int = 7
-    stop_hour: int = 21
+    start_hour: int = 6
+    stop_hour: int = 20
     poll_seconds: int = 600
     high_cpm: Decimal = Decimal("250")
     low_cpm: Decimal = Decimal("190")
@@ -22,8 +22,8 @@ class RealtimeControlConfig:
     @classmethod
     def from_env(cls) -> "RealtimeControlConfig":
         config = cls(
-            start_hour=int(os.getenv("RTC_START_HOUR", "7")),
-            stop_hour=int(os.getenv("RTC_STOP_HOUR", "21")),
+            start_hour=int(os.getenv("RTC_START_HOUR", "6")),
+            stop_hour=int(os.getenv("RTC_STOP_HOUR", "20")),
             poll_seconds=int(os.getenv("RTC_POLL_SECONDS", "600")),
             high_cpm=Decimal(os.getenv("RTC_HIGH_CPM", "250")),
             low_cpm=Decimal(os.getenv("RTC_LOW_CPM", "190")),

examples/tencent_realtime_control/run_control_service.py

@@ -0,0 +1,66 @@
+#!/usr/bin/env python
+"""Run Feishu controls and the ten-minute real-time control loop."""
+
+from __future__ import annotations
+
+import argparse
+import logging
+import os
+import sys
+from pathlib import Path
+
+
+ROOT = Path(__file__).resolve().parents[2]
+if str(ROOT) not in sys.path:
+    sys.path.insert(0, str(ROOT))
+
+from run_once import load_environment  # noqa: E402
+from run_scheduler import run_forever  # noqa: E402
+from storage import initialize_schema  # noqa: E402
+
+
+logger = logging.getLogger("tencent_realtime_control.service")
+
+
+def _env_flag(name: str, default: bool = False) -> bool:
+    raw = os.getenv(name)
+    if raw is None:
+        return default
+    return raw.strip().lower() in {"1", "true", "yes", "on"}
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument(
+        "--apply",
+        action="store_true",
+        help="Enable confirmed Feishu commands and real Tencent updates.",
+    )
+    return parser.parse_args()
+
+
+def main() -> None:
+    load_environment()
+    args = parse_args()
+    apply = args.apply or _env_flag("RTC_APPLY_ENABLED")
+    initialize_schema()
+    if os.getenv("RTC_COMMAND_ENABLED", "1").strip().lower() in {
+        "1",
+        "true",
+        "yes",
+    }:
+        from feishu_command_service import FeishuCommandService
+
+        FeishuCommandService(apply=apply).start()
+        logger.info("Feishu command WebSocket started")
+    else:
+        logger.warning("Feishu command WebSocket is disabled")
+    run_forever(apply=apply)
+
+
+if __name__ == "__main__":
+    logging.basicConfig(
+        level=os.getenv("LOG_LEVEL", "INFO"),
+        format="%(asctime)s %(levelname)s %(name)s %(message)s",
+    )
+    main()

examples/tencent_realtime_control/run_once.py

@@ -93,9 +93,11 @@ def target_for_ad(
     base_bid: int,
     boosted_date: Any,
     paused_by_strategy: bool,
+    operator_pause_expired: bool,
     bid_up_ratio: Decimal,
 ) -> dict[str, Any] | None:
-    manual_suspend = current_status != ACTIVE_STATUS and not paused_by_strategy
+    managed_suspend = paused_by_strategy or operator_pause_expired
+    manual_suspend = current_status != ACTIVE_STATUS and not managed_suspend
 
     target_bid = base_bid
     target_status: str | None = None
@@ -127,10 +129,10 @@ def target_for_ad(
         else:
             target_bid = boosted_bid(base_bid, bid_up_ratio)
             next_boosted_date = control_date
-        if paused_by_strategy:
+        if managed_suspend:
             target_status = ACTIVE_STATUS
     elif decision == DECISION_RESTORE:
-        if paused_by_strategy:
+        if managed_suspend:
             target_status = ACTIVE_STATUS
     elif decision in (DECISION_PAUSE, DECISION_CUTOFF):
         target_status = SUSPEND_STATUS if current_status != SUSPEND_STATUS else None
@@ -175,7 +177,12 @@ def execute_inventory_action(
     tencent: Any,
     test_notification: bool = False,
 ) -> dict[str, Any]:
-    from storage import insert_action_log, load_ad_states, upsert_ad_state
+    from storage import (
+        clear_operator_pause,
+        insert_action_log,
+        load_ad_states,
+        upsert_ad_state,
+    )
     from tencent_client import current_bid_fen, resolve_bid_field
 
     results: list[dict[str, Any]] = []
@@ -205,9 +212,56 @@ def execute_inventory_action(
             paused_by_strategy = bool(
                 (state or {}).get("paused_by_strategy")
             )
+            operator_pause_mode = str(
+                (state or {}).get("operator_pause_mode") or ""
+            )
+            operator_resume_at = (state or {}).get("operator_resume_at")
+            if operator_resume_at and operator_resume_at.tzinfo is None:
+                operator_resume_at = operator_resume_at.replace(tzinfo=SHANGHAI)
+            operator_manually_opened = bool(
+                operator_pause_mode and current_status == ACTIVE_STATUS
+            )
+            operator_pause_expired = bool(
+                operator_pause_mode == "UNTIL_NEXT_DELIVERY"
+                and operator_resume_at
+                and operator_resume_at <= now
+            )
+            operator_pause_active = bool(
+                operator_pause_mode
+                and not operator_manually_opened
+                and not operator_pause_expired
+            )
+
+            if operator_manually_opened and apply:
+                clear_operator_pause(
+                    account_id,
+                    adgroup_id,
+                    action="OPERATOR_MANUAL_OVERRIDE",
+                    action_at=now,
+                )
+                results.append(
+                    {
+                        "account_id": account_id,
+                        "audience_name": account.get("audience_name"),
+                        "adgroup_id": adgroup_id,
+                        "adgroup_name": ad.get("adgroup_name"),
+                        "decision": "OPERATOR_MANUAL_OVERRIDE",
+                        "before_status": current_status,
+                        "target_status": current_status,
+                        "status": "success",
+                        "bid_change_needed": False,
+                        "status_change_needed": False,
+                    }
+                )
+                continue
+
+            if operator_pause_active:
+                continue
+
             if (
                 current_status != ACTIVE_STATUS
                 and not paused_by_strategy
+                and not operator_pause_expired
                 and (
                     state is None
                     or decision not in (DECISION_PAUSE, DECISION_CUTOFF)
@@ -242,6 +296,7 @@ def execute_inventory_action(
                 base_bid=base_bid,
                 boosted_date=(state or {}).get("boosted_date"),
                 paused_by_strategy=paused_by_strategy,
+                operator_pause_expired=operator_pause_expired,
                 bid_up_ratio=config.bid_up_ratio,
             )
             if plan is None:
@@ -337,6 +392,13 @@ def execute_inventory_action(
                     last_action=decision,
                     action_at=now,
                 )
+                if operator_pause_expired:
+                    clear_operator_pause(
+                        account_id,
+                        adgroup_id,
+                        action="OPERATOR_DAY_PAUSE_EXPIRED",
+                        action_at=now,
+                    )
             if apply and (needs_api or state is None or execution_status == "failed"):
                 insert_action_log(
                     {
@@ -416,6 +478,17 @@ def execute_inventory_action(
             observed_cpm=observed_cpm,
             test_mode=test_notification,
         )
+        if apply:
+            try:
+                from feishu_notifier import (
+                    send_operator_reconciliation_notification,
+                )
+
+                send_operator_reconciliation_notification(results)
+            except Exception:
+                logger.exception(
+                    "Failed to notify operator manual override reconciliation"
+                )
 
     return {
         "failures": failures,
@@ -438,7 +511,7 @@ def run_cycle(
     from storage import (
         advisory_lock,
         load_daily_state,
-        load_enabled_accounts,
+        load_realtime_accounts,
         save_daily_state,
     )
     from tencent_client import TencentClient
@@ -461,7 +534,7 @@ def run_cycle(
             return payload
 
         daily_state = load_daily_state(now.date())
-        accounts = load_enabled_accounts()
+        accounts = load_realtime_accounts()
         if now.hour < config.start_hour:
             payload["decision"] = DECISION_OFF_HOURS
             return payload

examples/tencent_realtime_control/run_scheduler.py

@@ -56,16 +56,15 @@ def parse_args() -> argparse.Namespace:
     return parser.parse_args()
 
 
-def main() -> None:
-    load_environment()
-    args = parse_args()
+def run_forever(*, apply: bool, install_signal_handlers: bool = True) -> None:
     config = RealtimeControlConfig.from_env()
-    signal.signal(signal.SIGINT, request_stop)
-    signal.signal(signal.SIGTERM, request_stop)
+    if install_signal_handlers:
+        signal.signal(signal.SIGINT, request_stop)
+        signal.signal(signal.SIGTERM, request_stop)
 
     logger.info(
         "Scheduler started apply=%s hours=%02d:00-%02d:00 interval=%ss",
-        args.apply,
+        apply,
         config.start_hour,
         config.stop_hour,
         config.poll_seconds,
@@ -74,7 +73,7 @@ def main() -> None:
         try:
             payload = run_cycle(
                 now=datetime.now(SHANGHAI),
-                apply=args.apply,
+                apply=apply,
             )
             report = write_run_report(payload)
             logger.info(
@@ -99,6 +98,12 @@ def main() -> None:
             time.sleep(min(remaining, 5))
 
 
+def main() -> None:
+    load_environment()
+    args = parse_args()
+    run_forever(apply=args.apply)
+
+
 if __name__ == "__main__":
     logging.basicConfig(
         level=logging.INFO,

examples/tencent_realtime_control/schema.sql

@@ -20,6 +20,11 @@ CREATE TABLE IF NOT EXISTS realtime_control_ad_state (
     boosted_date DATE DEFAULT NULL,
     paused_by_strategy BOOLEAN NOT NULL DEFAULT FALSE,
     pause_reason VARCHAR(32) DEFAULT NULL,
+    operator_pause_mode VARCHAR(32) DEFAULT NULL,
+    operator_resume_at DATETIME DEFAULT NULL,
+    operator_command_id VARCHAR(64) DEFAULT NULL,
+    operator_paused_from_status VARCHAR(50) DEFAULT NULL,
+    operator_paused_at DATETIME DEFAULT NULL,
     last_action VARCHAR(32) DEFAULT NULL,
     last_action_at DATETIME DEFAULT NULL,
     last_seen_at DATETIME DEFAULT NULL,
@@ -27,6 +32,7 @@ CREATE TABLE IF NOT EXISTS realtime_control_ad_state (
     updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
     PRIMARY KEY (account_id, adgroup_id),
     KEY idx_strategy_pause (paused_by_strategy),
+    KEY idx_operator_pause (operator_pause_mode, operator_resume_at),
     KEY idx_boosted_date (boosted_date),
     KEY idx_last_seen (last_seen_at)
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='实时CPM调控广告基准与状态';
@@ -56,3 +62,44 @@ CREATE TABLE IF NOT EXISTS realtime_control_action_log (
     KEY idx_adgroup (account_id, adgroup_id),
     KEY idx_execution_status (execution_status)
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='实时CPM调控动作审计';
+
+CREATE TABLE IF NOT EXISTS operator_command (
+    command_id VARCHAR(64) NOT NULL PRIMARY KEY,
+    source_message_id VARCHAR(128) NOT NULL,
+    chat_id VARCHAR(128) NOT NULL,
+    sender_open_id VARCHAR(128) NOT NULL,
+    sender_name VARCHAR(255) DEFAULT NULL,
+    action VARCHAR(32) NOT NULL,
+    scope_type VARCHAR(32) NOT NULL,
+    target_account_ids TEXT NOT NULL,
+    status VARCHAR(32) NOT NULL,
+    preview_account_count INT NOT NULL DEFAULT 0,
+    preview_ad_count INT NOT NULL DEFAULT 0,
+    expires_at DATETIME DEFAULT NULL,
+    confirmed_at DATETIME DEFAULT NULL,
+    executed_at DATETIME DEFAULT NULL,
+    error_message TEXT DEFAULT NULL,
+    created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
+    UNIQUE KEY uk_operator_source_message (source_message_id),
+    KEY idx_operator_status_expiry (status, expires_at),
+    KEY idx_operator_sender (sender_open_id, created_at)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='飞书运营控制命令';
+
+CREATE TABLE IF NOT EXISTS operator_command_item (
+    id BIGINT NOT NULL AUTO_INCREMENT PRIMARY KEY,
+    command_id VARCHAR(64) NOT NULL,
+    account_id BIGINT NOT NULL,
+    audience_name VARCHAR(200) DEFAULT NULL,
+    adgroup_id BIGINT DEFAULT NULL,
+    adgroup_name VARCHAR(255) DEFAULT NULL,
+    before_status VARCHAR(50) DEFAULT NULL,
+    target_status VARCHAR(50) DEFAULT NULL,
+    readback_status VARCHAR(50) DEFAULT NULL,
+    execution_status VARCHAR(32) NOT NULL,
+    error_message TEXT DEFAULT NULL,
+    created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    KEY idx_operator_item_command (command_id),
+    KEY idx_operator_item_ad (account_id, adgroup_id),
+    KEY idx_operator_item_status (execution_status)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='飞书运营控制命令广告明细';

examples/tencent_realtime_control/storage.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import os
+import json
 from contextlib import contextmanager
 from datetime import date, datetime
 from pathlib import Path
@@ -46,6 +47,47 @@ def initialize_schema() -> None:
         with connection.cursor() as cursor:
             for statement in statements:
                 cursor.execute(statement)
+            cursor.execute(
+                """
+                SELECT COLUMN_NAME
+                FROM information_schema.COLUMNS
+                WHERE TABLE_SCHEMA=%s
+                  AND TABLE_NAME='realtime_control_ad_state'
+                """,
+                (os.environ["DB_NAME"],),
+            )
+            existing_columns = {row["COLUMN_NAME"] for row in cursor.fetchall()}
+            migrations = {
+                "operator_pause_mode": "VARCHAR(32) DEFAULT NULL",
+                "operator_resume_at": "DATETIME DEFAULT NULL",
+                "operator_command_id": "VARCHAR(64) DEFAULT NULL",
+                "operator_paused_from_status": "VARCHAR(50) DEFAULT NULL",
+                "operator_paused_at": "DATETIME DEFAULT NULL",
+            }
+            for column, definition in migrations.items():
+                if column not in existing_columns:
+                    cursor.execute(
+                        f"ALTER TABLE realtime_control_ad_state "
+                        f"ADD COLUMN {column} {definition}"
+                    )
+            cursor.execute(
+                """
+                SELECT INDEX_NAME
+                FROM information_schema.STATISTICS
+                WHERE TABLE_SCHEMA=%s
+                  AND TABLE_NAME='realtime_control_ad_state'
+                  AND INDEX_NAME='idx_operator_pause'
+                """,
+                (os.environ["DB_NAME"],),
+            )
+            if not cursor.fetchone():
+                cursor.execute(
+                    """
+                    CREATE INDEX idx_operator_pause
+                    ON realtime_control_ad_state
+                        (operator_pause_mode, operator_resume_at)
+                    """
+                )
     finally:
         connection.close()
 
@@ -88,6 +130,44 @@ def load_enabled_accounts() -> list[dict[str, Any]]:
         connection.close()
 
 
+def load_realtime_accounts() -> list[dict[str, Any]]:
+    """All historical automation accounts still enabled by the whitelist."""
+    connection = connect()
+    try:
+        with connection.cursor() as cursor:
+            cursor.execute(
+                """
+                SELECT c.account_id, c.audience_name, c.bid_scene
+                FROM ad_creation_account_config c
+                JOIN account_whitelist w ON w.account_id = c.account_id
+                WHERE w.enabled = TRUE
+                ORDER BY c.account_id
+                """
+            )
+            return list(cursor.fetchall())
+    finally:
+        connection.close()
+
+
+def load_managed_accounts() -> list[dict[str, Any]]:
+    """Historical automation accounts still allowed by the account whitelist."""
+    connection = connect()
+    try:
+        with connection.cursor() as cursor:
+            cursor.execute(
+                """
+                SELECT c.account_id, c.audience_name, c.bid_scene, c.enabled
+                FROM ad_creation_account_config c
+                JOIN account_whitelist w ON w.account_id = c.account_id
+                WHERE w.enabled = TRUE
+                ORDER BY c.account_id
+                """
+            )
+            return list(cursor.fetchall())
+    finally:
+        connection.close()
+
+
 def load_daily_state(control_date: date) -> dict[str, Any]:
     connection = connect()
     try:
@@ -223,3 +303,249 @@ def insert_action_log(record: dict[str, Any]) -> None:
             )
     finally:
         connection.close()
+
+
+def create_operator_command(record: dict[str, Any]) -> dict[str, Any]:
+    connection = connect()
+    try:
+        with connection.cursor() as cursor:
+            try:
+                cursor.execute(
+                    """
+                    INSERT INTO operator_command
+                        (command_id, source_message_id, chat_id, sender_open_id,
+                         sender_name, action, scope_type, target_account_ids,
+                         status, preview_account_count, preview_ad_count, expires_at)
+                    VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)
+                    """,
+                    (
+                        record["command_id"],
+                        record["source_message_id"],
+                        record["chat_id"],
+                        record["sender_open_id"],
+                        record.get("sender_name"),
+                        record["action"],
+                        record["scope_type"],
+                        json.dumps(record["target_account_ids"]),
+                        record["status"],
+                        record.get("preview_account_count", 0),
+                        record.get("preview_ad_count", 0),
+                        record.get("expires_at"),
+                    ),
+                )
+            except pymysql.err.IntegrityError:
+                cursor.execute(
+                    "SELECT * FROM operator_command WHERE source_message_id=%s",
+                    (record["source_message_id"],),
+                )
+                existing = cursor.fetchone()
+                if existing:
+                    existing["target_account_ids"] = json.loads(
+                        existing.get("target_account_ids") or "[]"
+                    )
+                    return existing
+                raise
+        return load_operator_command(record["command_id"]) or {}
+    finally:
+        connection.close()
+
+
+def load_operator_command(command_id: str) -> dict[str, Any] | None:
+    connection = connect()
+    try:
+        with connection.cursor() as cursor:
+            cursor.execute(
+                "SELECT * FROM operator_command WHERE command_id=%s",
+                (command_id,),
+            )
+            row = cursor.fetchone()
+            if row:
+                row["target_account_ids"] = json.loads(
+                    row.get("target_account_ids") or "[]"
+                )
+            return row
+    finally:
+        connection.close()
+
+
+def update_operator_command(command_id: str, status: str, **values: Any) -> None:
+    allowed = {"confirmed_at", "executed_at", "error_message"}
+    unknown = set(values) - allowed
+    if unknown:
+        raise ValueError(f"Unsupported operator-command fields: {sorted(unknown)}")
+    assignments = ["status=%s"]
+    params: list[Any] = [status]
+    for column, value in values.items():
+        assignments.append(f"{column}=%s")
+        params.append(value)
+    params.append(command_id)
+    connection = connect()
+    try:
+        with connection.cursor() as cursor:
+            cursor.execute(
+                f"UPDATE operator_command SET {', '.join(assignments)} "
+                "WHERE command_id=%s",
+                params,
+            )
+    finally:
+        connection.close()
+
+
+def transition_operator_command(
+    command_id: str,
+    *,
+    expected_statuses: set[str],
+    target_status: str,
+    **values: Any,
+) -> bool:
+    allowed = {"confirmed_at", "executed_at", "error_message"}
+    unknown = set(values) - allowed
+    if unknown:
+        raise ValueError(f"Unsupported operator-command fields: {sorted(unknown)}")
+    if not expected_statuses:
+        raise ValueError("expected_statuses must not be empty")
+    assignments = ["status=%s"]
+    params: list[Any] = [target_status]
+    for column, value in values.items():
+        assignments.append(f"{column}=%s")
+        params.append(value)
+    placeholders = ", ".join(["%s"] * len(expected_statuses))
+    params.extend([command_id, *sorted(expected_statuses)])
+    connection = connect()
+    try:
+        with connection.cursor() as cursor:
+            affected = cursor.execute(
+                f"""
+                UPDATE operator_command
+                SET {', '.join(assignments)}
+                WHERE command_id=%s
+                  AND status IN ({placeholders})
+                """,
+                params,
+            )
+            return affected == 1
+    finally:
+        connection.close()
+
+
+def insert_operator_command_item(record: dict[str, Any]) -> None:
+    columns = [
+        "command_id",
+        "account_id",
+        "audience_name",
+        "adgroup_id",
+        "adgroup_name",
+        "before_status",
+        "target_status",
+        "readback_status",
+        "execution_status",
+        "error_message",
+    ]
+    connection = connect()
+    try:
+        with connection.cursor() as cursor:
+            cursor.execute(
+                f"INSERT INTO operator_command_item ({', '.join(columns)}) "
+                f"VALUES ({', '.join(['%s'] * len(columns))})",
+                [record.get(column) for column in columns],
+            )
+    finally:
+        connection.close()
+
+
+def set_operator_pause(
+    *,
+    account_id: int,
+    adgroup_id: int,
+    mode: str,
+    resume_at: datetime | None,
+    command_id: str,
+    paused_from_status: str,
+    paused_at: datetime,
+) -> None:
+    connection = connect()
+    try:
+        with connection.cursor() as cursor:
+            cursor.execute(
+                """
+                UPDATE realtime_control_ad_state
+                SET operator_pause_mode=%s,
+                    operator_resume_at=%s,
+                    operator_command_id=%s,
+                    operator_paused_from_status=%s,
+                    operator_paused_at=%s,
+                    last_action='OPERATOR_PAUSE',
+                    last_action_at=%s,
+                    last_seen_at=%s
+                WHERE account_id=%s AND adgroup_id=%s
+                """,
+                (
+                    mode,
+                    resume_at,
+                    command_id,
+                    paused_from_status,
+                    paused_at,
+                    paused_at,
+                    paused_at,
+                    account_id,
+                    adgroup_id,
+                ),
+            )
+    finally:
+        connection.close()
+
+
+def clear_operator_pause(
+    account_id: int,
+    adgroup_id: int,
+    *,
+    action: str,
+    action_at: datetime,
+) -> None:
+    connection = connect()
+    try:
+        with connection.cursor() as cursor:
+            cursor.execute(
+                """
+                UPDATE realtime_control_ad_state
+                SET operator_pause_mode=NULL,
+                    operator_resume_at=NULL,
+                    operator_command_id=NULL,
+                    operator_paused_from_status=NULL,
+                    operator_paused_at=NULL,
+                    last_action=%s,
+                    last_action_at=%s,
+                    last_seen_at=%s
+                WHERE account_id=%s AND adgroup_id=%s
+                """,
+                (action, action_at, action_at, account_id, adgroup_id),
+            )
+    finally:
+        connection.close()
+
+
+def load_operator_pauses(
+    account_ids: list[int] | None = None,
+) -> list[dict[str, Any]]:
+    params: list[Any] = []
+    where = "WHERE operator_pause_mode IS NOT NULL"
+    if account_ids is not None:
+        if not account_ids:
+            return []
+        where += f" AND account_id IN ({', '.join(['%s'] * len(account_ids))})"
+        params.extend(account_ids)
+    connection = connect()
+    try:
+        with connection.cursor() as cursor:
+            cursor.execute(
+                f"""
+                SELECT *
+                FROM realtime_control_ad_state
+                {where}
+                ORDER BY account_id, adgroup_id
+                """,
+                params,
+            )
+            return list(cursor.fetchall())
+    finally:
+        connection.close()

examples/tencent_realtime_control/tencent_client.py

@@ -25,9 +25,46 @@ AD_FIELDS = [
     "custom_cost_cap",
     "smart_bid_type",
     "cost_constraint_scene",
+    "begin_date",
+    "end_date",
+    "time_series",
 ]
 
 
+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(
@@ -194,18 +231,40 @@ class TencentClient:
             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")
+        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:
@@ -215,29 +274,99 @@ class TencentClient:
 
         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):
+            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 actual != target:
-                    matches = False
-                    break
-            if matches:
-                return ad
+                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 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(
-            "Post-write verification failed: "
-            f"account={account_id} adgroup={adgroup_id} "
-            f"expected={expected} actual={last_actual}"
+            "Tencent begin_date verification failed: "
+            f"account={account_id} expected={begin_date} actual={last_actual}"
         )
 
     def get_today_ad_metrics(