日志输出到阿里云sls、审批开关配置、合作平台token刷新

examples/auto_put_ad_mini/.env.example

@@ -1,6 +1,25 @@
 # 兼容本地开发的组件级样例。生产部署统一以仓库根目录 runtime.env.example 为准。
 # 不要在两个组件样例中分别维护生产配置。
 
+# ========================================
+# 本地日志与阿里云 SLS
+# ========================================
+LOG_LEVEL=INFO
+SLS_ENDPOINT=cn-hangzhou.log.aliyuncs.com
+SLS_ACCESS_KEY_ID=xxx
+SLS_ACCESS_KEY_SECRET=xxx
+SLS_PROJECT=auto-put-tecent
+D=info-log
+SLS_ERROR_LOGSTORE=error-log
+# SLS_INFO_LOGSTORE 未设置时的回退 logstore
+# SLS_LOGSTORE=info-log
+# SLS 上报最低日志级别，默认 INFO
+# SLS_LOG_LEVEL=INFO
+# SLS 批量上报最大条数，默认 1024
+# SLS_BATCH_SIZE_MAX=1024
+# SLS 批量上报等待间隔(毫秒)，默认 2000
+# SLS_PUT_WAIT_MS=2000
+
 # ========================================
 # 腾讯广告 API 配置
 # ========================================
@@ -18,6 +37,38 @@ TENCENT_AD_USER_TOKEN=xxx
 
 # TENCENT_AD_BASE_URL=https://api.e.qq.com/v3.0
 
+# ========================================
+# 创意搭建子系统（模块 B）
+# ========================================
+# 每广告目标创意数，默认 8
+TARGET_CREATIVES_PER_AD=8
+# 创意准备阶段并发数，默认 1（串行）
+# CREATIVE_PREPARE_MAX_WORKERS=1
+# 创意准备任务缓冲数，默认 16
+# CREATIVE_PREPARE_TASK_BUFFER=16
+# 是否走飞书审批流程，默认 true
+# CREATION_APPROVAL_REQUIRED=true
+# 飞书审批消息中是否嵌入素材预览图，默认 1
+# CREATION_APPROVAL_EMBED_IMAGES=1
+# 审批超时分钟数，默认 120
+CREATION_APPROVAL_TIMEOUT_MINUTES=120
+# 同一广告单次运行中最多使用同一落地页数，默认 1
+# MAX_SAME_LANDING_PER_AD_IN_RUN=1
+# 落地页品类排除列表（逗号分隔），默认排除常见品类
+# LANDING_EXCLUDED_CATEGORIES=早中晚好,祝福音乐,历史名人
+# 仅处理指定账户（逗号分隔），已废弃，请使用 --account-id CLI 参数
+# CREATION_ONLY_ACCOUNT_IDS=
+# 仅处理指定人群包（逗号分隔），为空=全部
+# CREATION_ONLY_CROWD_PACKAGES=
+# 跳过 Phase0（广告创建），仅执行后续步骤。调试用，默认 0
+# CREATION_SKIP_PHASE0=0
+# 素材去重回溯天数，默认 7
+# CREATIVE_MATERIAL_DEDUPE_LOOKBACK_DAYS=7
+# 落地页去重回溯天数，默认 7
+# CREATIVE_LANDING_DEDUPE_LOOKBACK_DAYS=7
+# 创建恢复回溯小时数，默认 24
+# CREATION_RECOVERY_LOOKBACK_HOURS=24
+
 # 统一生产服务
 DAILY_CREATION_ENABLED=0
 DAILY_CREATION_HOUR=10
@@ -111,22 +162,85 @@ FEISHU_CREATION_CHAT_ID=
 # 如何获取群聊ID：运行 python3 get_chat_id.py
 # FEISHU_AD_PROJECT_CHAT_ID=oc_xxxxxxxxxxxxxxxxxxxxxx
 
+# 飞书 IM 总开关（1=启用, 0=关闭所有飞书消息），默认 1
+# IM_ENABLED=1
+
 # ========================================
 # ODPS 数据平台配置
 # ========================================
 # ODPS_ACCESS_ID=xxx
 # ODPS_ACCESS_SECRET=xxx
 # ODPS_PROJECT=loghubods
+# ODPS 服务端点，默认 http://service.odps.aliyun.com/api
+# ODPS_ENDPOINT=http://service.odps.aliyun.com/api
+
+# Datanexus 数据平台（用于 TencentAD 数据查询）
+# DATANEXUS_ACCESS_TOKEN=
+# DATANEXUS_TOKEN_API=
 
 # ========================================
-# LLM API Key
+# LLM / AI 模型配置
 # ========================================
 # QWEN_API_KEY=xxx
 # OPEN_ROUTER_API_KEY=xxx
-# AI 图片生成模型。当前默认使用 OpenRouter Images API 可用的 pro 图片模型。
-# OPENROUTER_IMAGE_MODEL=google/gemini-3-pro-image
+# OpenRouter 对话补全端点，默认 https://openrouter.ai/api/v1/chat/completions
+# OPENROUTER_CHAT_COMPLETIONS_URL=https://openrouter.ai/api/v1/chat/completions
+# OpenRouter 图片生成端点，默认 https://openrouter.ai/api/v1/images
+# OPENROUTER_IMAGES_URL=https://openrouter.ai/api/v1/images
+# AI 图片生成模型，默认 google/gemini-3.1-flash-image
+# OPENROUTER_IMAGE_MODEL=google/gemini-3.1-flash-image
+# AI 文本生成模型，默认 google/gemini-3-flash-preview
+# OPENROUTER_TEXT_MODEL=google/gemini-3-flash-preview
+# AI 素材审核模型，默认 google/gemini-3-flash-preview
+# AI_MATERIAL_REVIEW_MODEL=google/gemini-3-flash-preview
+
+# ========================================
+# AI 素材生成（图片 + 封面文案）
+# ========================================
+# 是否使用图案选择器（1=启用, 0=关闭），默认 1
+# AI_IMAGE_USE_PATTERN_SELECTOR=1
+# 图案选择器返回 Top K 结果，默认 1
+# AI_IMAGE_PATTERN_TOP_K=1
+# 图片生成版位，默认 WECHAT_OFFICIAL_ACCOUNTS
+# AI_IMAGE_PATTERN_PLACEMENT=WECHAT_OFFICIAL_ACCOUNTS
+# 是否生成封面文案（1=启用, 0=关闭），默认 1
+# AI_COVER_COPY_REQUIRED=1
+# 封面文案最大 token 数，默认 900
+# AI_COVER_COPY_MAX_TOKENS=900
+# 封面文案标题最大长度（字符），默认 22
+# AI_COVER_COPY_TITLE_MAX_LEN=22
+# OSS 上传路径前缀，默认 auto_put_tencent/image
+# AI_IMAGE_OSS_PREFIX=auto_put_tencent/image
+# 图片宽高比，默认 16:9
+# AI_IMAGE_ASPECT_RATIO=16:9
+# 图片分辨率，默认 1K
+# AI_IMAGE_RESOLUTION=1K
+# 图片输出格式，默认 jpeg
+# AI_IMAGE_OUTPUT_FORMAT=jpeg
+# 图片目标宽度(px)，默认 1280
+# AI_IMAGE_TARGET_WIDTH=1280
+# 图片目标高度(px)，默认 720
+# AI_IMAGE_TARGET_HEIGHT=720
+# 自定义图片生成 Prompt 模板路径（留空=使用内置模板）
+# AI_IMAGE_PROMPT_TEMPLATE_PATH=
+# 自定义内容安全 Prompt 模板路径（留空=使用内置模板）
+# AI_SANITIZE_PROMPT_TEMPLATE_PATH=
+# 自定义封面文案 Prompt 模板路径（留空=使用内置模板）
+# AI_COVER_COPY_PROMPT_TEMPLATE_PATH=
+
+# ========================================
+# 阿里云 OSS（AI 图片上传存储）
+# ========================================
+# ALIYUN_OSS_ENDPOINT=oss-cn-hangzhou.aliyuncs.com
+# ALIYUN_OSS_BUCKET=your-bucket
+# 也可用 AI_IMAGE_OSS_BUCKET 替代 ALIYUN_OSS_BUCKET
+# AI_IMAGE_OSS_BUCKET=your-bucket
+# ALIYUN_OSS_ACCESS_KEY_ID=xxx
+# ALIYUN_OSS_ACCESS_KEY_SECRET=xxx
+# OSS 公开访问基础 URL（用于生成图片外链）
+# AI_IMAGE_PUBLIC_BASE_URL=https://your-bucket.oss-cn-hangzhou.aliyuncs.com
 
-# 外部素材召回账户专用。飞书「素材来源」填“外部素材”或“外部素材召回”时生效。
+# 外部素材召回账户专用。飞书「素材来源」填”外部素材”或”外部素材召回”时生效。
 EXTERNAL_RECALL_SOURCE_LABEL=外部合作
 EXTERNAL_RECALL_CANDIDATE_LIMIT=300
 EXTERNAL_RECALL_EDIT_LIMIT_PER_LANDING=3
@@ -134,6 +248,54 @@ EXTERNAL_RECALL_UV_WINDOW_DAYS=90
 # 默认复用 OPENROUTER_IMAGE_MODEL；可单独指定支持 input_references 的图片模型。
 # EXTERNAL_IMAGE_MODEL=google/gemini-3.1-flash-image
 
+# ========================================
+# 视频召回与风控
+# ========================================
+# 视频风控 API 地址
+# VIDEO_RISK_API_URL=https://longvideoapi.piaoquantv.com/longvideoapi/openapi/video/getVideoTagIds
+# 视频风控最大允许级别，默认 5（超过则拒绝）
+# VIDEO_RISK_MAX_ALLOWED_LEVEL=5
+# 视频风控 API 超时秒数，默认 10
+# VIDEO_RISK_API_TIMEOUT_SECONDS=10
+# 视频召回并发数，默认 4
+# RECALL_PARALLEL_MAX_WORKERS=4
+# 单视频召回查询上限，默认 12
+# RECALL_QUERY_LIMIT_PER_VIDEO=12
+# 飘泉 TV 登录手机号（素材召回用）
+# CONTENTPLATFORM_TEL_NUM=
+# 飘泉 TV 登录密码（素材召回用）
+# CONTENTPLATFORM_PASSWORD=
+# 飘泉 TV 静态 Token（可选，优先使用动态登录）
+# PIAOQUANTV_TOKEN=
+# 飘泉 TV 视频来源过滤
+# PIAOQUANTV_VIDEO_SOURCE=
+# 飘泉 TV 视频类型过滤
+# PIAOQUANTV_VIDEO_TYPE=
+# 热门素材兜底开关（1=启用，召回量不足时自动回退热门源），默认 1
+# PIAOQUANTV_HOT_FALLBACK_ENABLED=1
+# 热门素材兜底来源标签，默认 hot
+# PIAOQUANTV_HOT_FALLBACK_SOURCE=hot
+# 飘泉 TV 视频最大页数，默认 3
+# PIAOQUANTV_VIDEO_MAX_PAGES=3
+# 落地计划保存接口地址
+# PIAOQUANTV_XCX_SAVE_URL=https://tp-open.piaoquantv.com/contentPlatform/plan/xcx/save
+# 视频向量检索服务地址
+# VECTOR_BASE_URL=https://api-internal.piaoquantv.com/videoVector
+# 人群包→素材来源映射（JSON），为空时使用内置默认值
+# VIDEO_RECALL_CROWD_PACKAGE_MAP=
+# 素材来源标签映射（JSON），为空时使用内置默认值
+# VIDEO_RECALL_SOURCE_MAP=
+
+# ========================================
+# 创意审核扫描
+# ========================================
+# 审核扫描回溯小时数，默认 72
+# CREATIVE_REVIEW_LOOKBACK_HOURS=72
+# 审核扫描单次上限，默认 1000
+# CREATIVE_REVIEW_SCAN_LIMIT=1000
+# 审核扫描批次大小，默认 100
+# CREATIVE_REVIEW_BATCH_SIZE=100
+
 # ========================================
 # 数据库配置（MySQL）
 # ========================================
@@ -142,6 +304,12 @@ DB_PORT=3306
 DB_USER=ad_rw
 DB_PASSWORD=your_password
 DB_NAME=auto_put_ad_mini
+# 数据库连接超时秒数，默认 10
+# DB_CONNECT_TIMEOUT=10
+# 数据库读超时秒数，默认 60
+# DB_READ_TIMEOUT=60
+# 数据库写超时秒数，默认 60
+# DB_WRITE_TIMEOUT=60
 
 # ========================================
 # 生产环境配置（海外部署）
@@ -170,3 +338,51 @@ EXECUTION_ENABLED=false
 # CRON_SCHEDULE=0 2 * * *  # 每天凌晨2点UTC
 # RUN_ON_STARTUP=false     # 启动时是否立即执行
 # PORT=8080                # FastAPI 服务端口
+
+# ========================================
+# 高级配置（通常无需修改）
+# ========================================
+
+# -- 腾讯广告 AccessToken 动态获取 API --
+# 默认 https://api.piaoquantv.com/ad/put/tencent/getAccessToken
+# TENCENT_AD_TOKEN_API=https://api.piaoquantv.com/ad/put/tencent/getAccessToken
+
+# -- ROI 策略补充参数 --
+# 自投提价最小广告天数，默认 3
+# ROI_SELF_UP_MIN_AGE=3
+# 自投最小日均 UV，默认 200（也支持旧名 ROI_SELF_MIN_AVG_UV）
+# ROI_SELF_MIN_DAILY_UV=200
+# 代投最小日均 UV，默认 200（也支持旧名 ROI_PARTNER_MIN_AVG_UV）
+# ROI_PARTNER_MIN_DAILY_UV=200
+# 关停分位数（ROI_PARTNER_STOP_QUANTILE 的回退值），默认 0.20
+# ROI_STOP_QUANTILE=0.20
+# 关停权重上限分位数，默认 0.95
+# ROI_STOP_WEIGHT_CAP_QUANTILE=0.95
+# 素材策略学习 - 模型最低评分阈值，默认 75
+# PATTERN_SELECTOR_MIN_MODEL_SCORE=75
+# ROI 参数发布者标识，默认取系统 USER 环境变量
+# ROI_PARAMETER_PUBLISHED_BY=
+
+# -- 定时任务调度锁（多实例互斥） --
+# 每日创建任务锁名，默认 ad_daily_creation
+# DAILY_CREATION_LOCK_NAME=ad_daily_creation
+# 每日创建 misfire 容忍秒数，默认 3600
+# DAILY_CREATION_MISFIRE_GRACE_SECONDS=3600
+# 每日审核扫描锁名，默认 ad_creative_review_scan
+# DAILY_REVIEW_LOCK_NAME=ad_creative_review_scan
+# 每日 ROI 任务锁名，默认 ad_daily_roi
+# DAILY_ROI_LOCK_NAME=ad_daily_roi
+# 每日 ROI misfire 容忍秒数，默认 3600
+# DAILY_ROI_MISFIRE_GRACE_SECONDS=3600
+# 每日 ROI 内部测试锁名，默认 ad_daily_roi
+# DAILY_ROI_INTERNAL_TEST_LOCK_NAME=ad_daily_roi
+# ROI 内部测试 misfire 容忍秒数，默认 3600
+# ROI_INTERNAL_TEST_MISFIRE_GRACE_SECONDS=3600
+# 实时调控 DB 锁名，默认 tencent_realtime_control
+# RTC_DB_LOCK_NAME=tencent_realtime_control
+
+# -- 人群包授权 --
+# 人群包来源账户 ID，默认 55615440
+# TENCENT_AUDIENCE_SOURCE_ACCOUNT_ID=55615440
+# 人群包授权业务 ID，默认 0
+# TENCENT_AUDIENCE_GRANT_BUSINESS_ID=0

examples/auto_put_ad_mini/DEPLOYMENT.md

@@ -199,6 +199,23 @@ curl -X POST http://localhost:8080/trigger | jq .
 
 ### 环境变量
 
+日志由本工程自己的 `logging_setup.py` 初始化，不依赖 `agent/utils/logging.py`。
+所有 `INFO` 及以上 Python 日志、`print` 输出和未捕获异常直接输出到控制台，
+不再写入本地文件。配置 SLS 凭证后，`INFO/WARNING` 发送到
+`auto-put-tecent/info-log`，`ERROR/CRITICAL` 发送到
+`auto-put-tecent/error-log`。可使用 `event`、`task`、`stage`、`result`、
+`account_id` 和 `duration_ms` 检索执行路径。
+
+```bash
+LOG_LEVEL=INFO
+SLS_ENDPOINT=cn-hangzhou.log.aliyuncs.com
+SLS_ACCESS_KEY_ID=<RAM AccessKey ID>
+SLS_ACCESS_KEY_SECRET=<RAM AccessKey Secret>
+SLS_PROJECT=auto-put-tecent
+SLS_INFO_LOGSTORE=info-log
+SLS_ERROR_LOGSTORE=error-log
+```
+
 | 变量名 | 说明 | 默认值 | 必需 |
 |--------|------|--------|------|
 | `WHITELIST_ENABLED` | 启用账户白名单 | true | 否 |

examples/auto_put_ad_mini/PROJECT_STRUCTURE.md

@@ -1,7 +1,7 @@
 # auto_put_ad_mini — 项目结构文档
 
 > **定位**: 微信小程序投流 — 面向 ROI 的广告粒度自动调控 + 自动创建投放系统
-> **最后更新**: 2026-08-05
+> **最后更新**: 2026-08-06
 
 ---
 
@@ -37,7 +37,8 @@ examples/auto_put_ad_mini/
 │   ├── execute_creation_apply.py       # Flow B Phase 3：独立提交创意到腾讯
 │   ├── run_full_analysis.py            # 仅数据准备步骤（无 LLM，用于调试）
 │   ├── fetch_data.py                   # 独立数据拉取脚本
-│   └── merge_data.py                   # 独立数据合并脚本
+│   ├── merge_data.py                   # 独立数据合并脚本
+│   └── logging_setup.py                # ★统一日志基础设施（SLS + 本地文件）
 │
 ├── 🟢 生产服务
 │   ├── run_daily_service.py            # ★ 生产日级服务入口（APScheduler，Flow B + C + D）
@@ -49,7 +50,7 @@ examples/auto_put_ad_mini/
 │   └── requirements.txt                # 项目专属 Python 依赖
 │
 ├── 🟡 配置与技能
-│   ├── config.py                       # 核心业务配置（~1073 行）
+│   ├── config.py                       # 核心业务配置
 │   ├── presets.json                    # Agent 预设参数
 │   ├── strategy_params.json            # 策略参数
 │   ├── whitelist.json                  # 账户白名单（可被 DB 覆盖）
@@ -170,7 +171,7 @@ examples/auto_put_ad_mini/
 │   ├── debug_generate_ai_material.py   # AI 素材生成调试
 │   └── debug_select_creative_patterns.py # 创意范式选择调试
 │
-├── 🧪 测试文件（26 个）
+├── 🧪 测试文件（27 个）
 │   ├── test_e2e_full_flow.py           # 端到端全流程测试
 │   ├── test_approval_flow_e2e.py       # 审批流程端到端测试
 │   ├── test_real_approval_flow.py      # 真实审批流程测试
@@ -183,6 +184,7 @@ examples/auto_put_ad_mini/
 │   ├── test_video_recall_pagination.py # 视频召回分页测试
 │   ├── test_compute_signal_scores.py   # 信号分数计算测试
 │   ├── test_ad_creation_status.py      # 广告创建状态测试
+│   ├── test_production_creation_entry.py # 生产创建入口集成测试
 │   ├── test_feishu_approval.py         # 飞书审批测试
 │   ├── test_feishu_import.py           # 飞书数据导入测试
 │   ├── test_im_approval_creation.py    # IM 审批创建测试
@@ -211,7 +213,8 @@ examples/auto_put_ad_mini/
 │
 ├── 📁 数据与 SQL
 │   ├── data/tencent_constants/
-│   │   └── regions_sop_current.json    # SOP 地域参考数据
+│   │   ├── regions_sop_current.json    # SOP 地域参考数据
+│   │   └── regions_all.json            # 全量地域数据
 │   ├── sql/
 │   │   └── high_consumption_materials_30d.sql  # 30 日高消耗素材查询
 │   ├── shared/reports/                 # 共享分析报告

examples/auto_put_ad_mini/config.py

@@ -76,7 +76,6 @@ MAIN_CONFIG = RunConfig(
 SKILLS_DIR = str(Path(__file__).parent / "skills")
 TRACE_STORE_PATH = ".trace"
 LOG_LEVEL = "INFO"
-LOG_FILE = None
 
 # ═══════════════════════════════════════════
 # 时区配置
@@ -188,7 +187,8 @@ FEEDBACK_CHECK_HOURS = 6
 # ═══════════════════════════════════════════
 # IM 审批配置（飞书直连）
 # ═══════════════════════════════════════════
-IM_ENABLED = True                  # IM 主开关（True 时审批消息发飞书）
+# 环境变量 IM_ENABLED 优先（"1"/"true"/"yes"=启用，"0"/"false"/"no"=禁用）
+IM_ENABLED = os.getenv("IM_ENABLED", "true").strip().lower() not in {"0", "false", "no"}
 IM_APPROVAL_TIMEOUT_MINUTES = 120  # 审批超时（分钟）— 2小时
 IM_APPROVAL_POLL_INTERVAL_SECONDS = 30  # 审批轮询间隔（秒）
 
@@ -1047,7 +1047,10 @@ CREATIVE_DESCRIPTION_COUNT_PER_AD = 1
 # --- 审批开关 ---
 # True(默认):Phase 1 准备后写待审批 CSV + 发飞书 sheet → 等运营审批 → Phase 3 POST
 # False:Phase 1 跑完直接 Phase 3 POST(skip 飞书)— 用于自动化 cron + 信任规则的场景
-CREATION_APPROVAL_REQUIRED = True
+# 环境变量 CREATION_APPROVAL_REQUIRED 优先（"1"/"true"/"yes"=需要审批，"0"/"false"/"no"=跳过）
+CREATION_APPROVAL_REQUIRED = os.getenv(
+    "CREATION_APPROVAL_REQUIRED", "true"
+).strip().lower() not in {"0", "false", "no"}
 
 # 审批超时(分钟,与现有调控审批配置对齐)
 CREATION_APPROVAL_TIMEOUT_MINUTES = 120
@@ -1062,16 +1065,20 @@ SLS_ENDPOINT          = os.environ.get("SLS_ENDPOINT", "")          # 例 cn-han
 SLS_ACCESS_KEY_ID     = os.environ.get("SLS_ACCESS_KEY_ID", "")     # RAM 子账号 AK(只授 Log:PutLogs 权限)
 SLS_ACCESS_KEY_SECRET = os.environ.get("SLS_ACCESS_KEY_SECRET", "") # 同上的 SK
 SLS_PROJECT           = os.environ.get("SLS_PROJECT", "auto-put-tecent")
-SLS_LOGSTORE          = os.environ.get("SLS_LOGSTORE", "info-log")
+SLS_INFO_LOGSTORE     = os.environ.get("SLS_INFO_LOGSTORE", os.environ.get("SLS_LOGSTORE", "info-log"))
+SLS_ERROR_LOGSTORE    = os.environ.get("SLS_ERROR_LOGSTORE", "error-log")
+SLS_LOGSTORE          = SLS_INFO_LOGSTORE  # 兼容旧变量名,默认对应 info-log
 
 # 全开:任一缺失则降级为 False,主链路不受影响
-SLS_ENABLED = bool(SLS_ENDPOINT and SLS_ACCESS_KEY_ID and SLS_ACCESS_KEY_SECRET
-                   and SLS_PROJECT and SLS_LOGSTORE)
+SLS_ENABLED = bool(
+    SLS_ENDPOINT and SLS_ACCESS_KEY_ID and SLS_ACCESS_KEY_SECRET
+    and SLS_PROJECT and SLS_INFO_LOGSTORE and SLS_ERROR_LOGSTORE
+)
 
 # 上报等级 — 用户决策(2026-06-11):所有 INFO+ 上报
 # 注:material_recall 每个 landing 打 4 条 INFO,日 cron 量级约 5k-20k 条,SLS 流量成本几 RMB/月
-SLS_LOG_LEVEL = "INFO"
+SLS_LOG_LEVEL = os.environ.get("SLS_LOG_LEVEL", "INFO").upper()
 
 # QueuedLogHandler 内部异步队列参数(SDK 默认 + 微调,避免长连接 idle 断)
-SLS_BATCH_SIZE_MAX = 1024      # 单次 PutLogs 最多条数
-SLS_PUT_WAIT_MS    = 2000      # 队列攒到 batch_size 或等 2s flush 一次
+SLS_BATCH_SIZE_MAX = int(os.environ.get("SLS_BATCH_SIZE_MAX", "1024"))
+SLS_PUT_WAIT_MS = int(os.environ.get("SLS_PUT_WAIT_MS", "2000"))

examples/auto_put_ad_mini/execute_creation_apply.py

@@ -16,7 +16,9 @@ import asyncio
 import argparse
 import json
 import logging
+import os
 import sys
+import time
 from pathlib import Path
 
 _HERE = Path(__file__).parent
@@ -324,20 +326,8 @@ def write_summary(summary: dict, output_dir: Path) -> Path:
 
 def main() -> int:
     """独立模式:从 JSON 文件读 pending records → apply → 写 summary + 发飞书。"""
-    logging.basicConfig(
-        level=logging.INFO,
-        format="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
-        datefmt="%H:%M:%S",
-    )
-    for noisy in ("httpx", "httpcore", "config", "db.config"):
-        logging.getLogger(noisy).setLevel(logging.WARNING)
-
-    # SLS 上报(2026-06-11 接入)— 配置缺失自动降级
-    try:
-        from tools.sls_setup import attach_sls_handler
-        attach_sls_handler()
-    except Exception as e:
-        logger.warning("[sls] 挂载异常(降级为本地 only):%s", e)
+    from logging_setup import setup_logging
+    setup_logging(level=os.getenv("LOG_LEVEL", "INFO"))
 
     parser = argparse.ArgumentParser(
         description="从 pending records 执行创意创建;可选择读取已有飞书审批表决策。",
@@ -352,14 +342,30 @@ def main() -> int:
         help="读取已有审批表时等待决策的分钟数",
     )
     args = parser.parse_args()
+    started = time.monotonic()
+    logger.info(
+        "event=creation_apply stage=load_pending result=started file=%s",
+        args.pending_records_json,
+    )
 
     pending_path = Path(args.pending_records_json)
     if not pending_path.exists():
         logger.error(f"文件不存在: {pending_path}")
         return 1
 
-    with open(pending_path, encoding="utf-8") as f:
-        records = json.load(f)
+    try:
+        with open(pending_path, encoding="utf-8") as f:
+            records = json.load(f)
+    except Exception:
+        logger.exception(
+            "event=creation_apply stage=load_pending result=failed file=%s",
+            pending_path,
+        )
+        raise
+    logger.info(
+        "event=creation_apply stage=load_pending result=succeeded records=%d",
+        len(records),
+    )
 
     if args.sheet_token and args.sheet_id:
         from tools.im_approval_creation import poll_approval_actions
@@ -379,7 +385,14 @@ def main() -> int:
         logger.info("审批决策读取完成: %d/%d", len(actions), len(records))
 
     logger.info(f"读到 {len(records)} 条 pending records,开始 Phase 3 执行")
-    summary = apply_pending_records(records)
+    try:
+        summary = apply_pending_records(records)
+    except Exception:
+        logger.exception(
+            "event=creation_apply stage=apply result=failed records=%d duration_ms=%d",
+            len(records), int((time.monotonic() - started) * 1000),
+        )
+        raise
     out_path = write_summary(summary, _HERE / "outputs" / "data")
     _send_apply_summary_to_feishu(summary)
 
@@ -388,6 +401,13 @@ def main() -> int:
         f"Phase 3 完成: approve={t['approved']} ok={t['posted_ok']} fail={t['posted_failed']}"
     )
     logger.info(f"summary: {out_path}")
+    logger.info(
+        "event=creation_apply stage=apply result=%s approved=%d posted_ok=%d "
+        "posted_failed=%d duration_ms=%d summary=%s",
+        "succeeded" if t["posted_failed"] == 0 else "failed",
+        t["approved"], t["posted_ok"], t["posted_failed"],
+        int((time.monotonic() - started) * 1000), out_path,
+    )
     return 0 if t["posted_failed"] == 0 else 1
 
 

examples/auto_put_ad_mini/execute_creation_once.py

@@ -191,20 +191,8 @@ def _filter_creation_accounts(
 
 def _setup_logging() -> None:
     """配置日志格式与级别，挂载 SLS 上报（配置缺失时自动降级）。"""
-    logging.basicConfig(
-        level=logging.INFO,
-        format="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
-        datefmt="%H:%M:%S",
-    )
-    for noisy in ("httpx", "httpcore", "config", "db.config"):
-        logging.getLogger(noisy).setLevel(logging.WARNING)
-
-    # SLS 上报(2026-06-11 接入,K8s pod 直发)— 配置缺失自动降级,不阻塞主链路
-    try:
-        from tools.sls_setup import attach_sls_handler
-        attach_sls_handler()
-    except Exception as e:
-        logger.warning("[sls] 挂载异常(降级为本地 only):%s", e)
+    from logging_setup import setup_logging
+    setup_logging(level=os.getenv("LOG_LEVEL", "INFO"))
 
 
 def _fetch_existing_fingerprints_for_account(account_id: int) -> set[str]:
@@ -1067,6 +1055,11 @@ def main() -> int:
     logger.info("WHITELIST_ACCOUNTS            = %s", WHITELIST_ACCOUNTS)
     logger.info("=" * 60)
 
+    started = time.monotonic()
+    logger.info(
+        "event=creation_run stage=run_once result=started config_date=%s",
+        args.config_date or "today",
+    )
     try:
         summary = run_once(
             config_date=args.config_date,
@@ -1074,6 +1067,14 @@ def main() -> int:
         )
     except ValueError as exc:
         parser.error(str(exc))
+    except Exception:
+        logger.exception(
+            "event=creation_run stage=run_once result=failed config_date=%s "
+            "duration_ms=%d",
+            args.config_date or "today",
+            int((time.monotonic() - started) * 1000),
+        )
+        raise
     if (summary.get("sync_stats") or {}).get("error"):
         return 1
     t = summary.get("total") or {}
@@ -1087,6 +1088,15 @@ def main() -> int:
         logger.info("  posted_ok       = %d", t.get("posted_ok", 0))
         logger.info("  posted_failed   = %d", t.get("posted_failed", 0))
     logger.info("=" * 60)
+    logger.info(
+        "event=creation_run stage=run_once result=%s duration_ms=%d "
+        "prepared=%d posted_ok=%d posted_failed=%d",
+        "succeeded" if t.get("posted_failed", 0) == 0 else "failed",
+        int((time.monotonic() - started) * 1000),
+        t.get("phase1_prepared", t.get("records", 0)),
+        t.get("posted_ok", 0),
+        t.get("posted_failed", 0),
+    )
     return 0 if t.get("posted_failed", 0) == 0 else 1
 
 

examples/auto_put_ad_mini/execute_once.py

@@ -4,6 +4,7 @@
 import asyncio
 import os
 import sys
+import time
 from pathlib import Path
 from datetime import datetime, timedelta
 import logging
@@ -24,10 +25,10 @@ load_dotenv()
 from agent.core.runner import AgentRunner
 from agent.trace import FileSystemTraceStore, Trace, Message
 from agent.llm import create_openrouter_llm_call
-from agent.utils import setup_logging
+from examples.auto_put_ad_mini.logging_setup import setup_logging, get_trace_id
 
 from examples.auto_put_ad_mini.config import (
-    MAIN_CONFIG, SKILLS_DIR, TRACE_STORE_PATH, LOG_LEVEL, LOG_FILE,
+    MAIN_CONFIG, SKILLS_DIR, TRACE_STORE_PATH, LOG_LEVEL,
 )
 
 # 导入自定义工具
@@ -52,7 +53,15 @@ async def main():
     """广告智能调控主流程：数据拉取 → ROI计算 → 候选筛选 → AI推理 → 护栏验证 → 报告生成。"""
     base_dir = Path(__file__).parent
 
-    setup_logging(level=LOG_LEVEL, file=LOG_FILE)
+    setup_logging(level=LOG_LEVEL)
+    pipeline_logger = logging.getLogger("auto_put_ad_mini.pipeline")
+    trace_id = get_trace_id() or "-"
+    started_ms = int(time.monotonic() * 1000)
+
+    pipeline_logger.info(
+        "event=pipeline stage=run result=started trace_id=%s date=%s",
+        trace_id, target_date if 'target_date' in dir() else "auto",
+    )
 
     # 加载 system prompt
     prompt_path = base_dir / "prompts" / "system.prompt"
@@ -119,64 +128,46 @@ async def main():
                         result = content.get("result", content.get("text", str(content)))
 
                         # 识别关键步骤
+                        _step_label = None
                         if tool_name == "fetch_creative_data":
                             step_count += 1
-                            print(f"\n{'='*70}")
-                            print(f"📌 步骤 {step_count}: 数据拉取")
-                            print(f"{'='*70}")
-
+                            _step_label = "数据拉取"
                         elif tool_name == "calculate_roi_metrics":
                             step_count += 1
-                            print(f"\n{'='*70}")
-                            print(f"📌 步骤 {step_count}: ROI 计算")
-                            print(f"{'='*70}")
-
+                            _step_label = "ROI 计算"
                         elif tool_name == "calculate_portfolio_summary":
                             step_count += 1
-                            print(f"\n{'='*70}")
-                            print(f"📌 步骤 {step_count}: 人群包基线计算")
-                            print(f"{'='*70}")
-
+                            _step_label = "人群包基线计算"
                         elif tool_name == "get_ads_for_review":
                             step_count += 1
-                            print(f"\n{'='*70}")
-                            print(f"📌 步骤 {step_count}: 候选筛选（零消耗/待评估/正常运行）")
-                            print(f"{'='*70}")
-
+                            _step_label = "候选筛选"
                         elif tool_name == "apply_decisions":
                             step_count += 1
-                            print(f"\n{'='*70}")
-                            print(f"📌 步骤 {step_count}: 保存智能引擎决策")
-                            print(f"{'='*70}")
-
+                            _step_label = "保存智能引擎决策"
                         elif tool_name == "validate_decisions":
                             step_count += 1
-                            print(f"\n{'='*70}")
-                            print(f"📌 步骤 {step_count}: 安全护栏验证")
-                            print(f"{'='*70}")
-
+                            _step_label = "安全护栏验证"
                         elif tool_name == "execute_decisions":
                             step_count += 1
-                            print(f"\n{'='*70}")
-                            print(f"📌 步骤 {step_count}: 分级执行")
-                            print(f"{'='*70}")
-
+                            _step_label = "分级执行"
                         elif tool_name == "send_approval_request":
                             step_count += 1
-                            print(f"\n{'='*70}")
-                            print(f"📌 步骤 {step_count}: IM 审批请求")
-                            print(f"{'='*70}")
-
+                            _step_label = "IM 审批请求"
                         elif tool_name == "generate_report":
                             step_count += 1
-                            print(f"\n{'='*70}")
-                            print(f"📌 步骤 {step_count}: 生成最终报告")
-                            print(f"{'='*70}")
-
+                            _step_label = "生成最终报告"
                         elif tool_name == "check_execution_feedback":
                             step_count += 1
+                            _step_label = "执行效果检查"
+
+                        if _step_label:
+                            step_slug = _step_label.replace(" ", "_")
+                            pipeline_logger.info(
+                                "event=step stage=%s step=%d result=started trace_id=%s",
+                                step_slug, step_count, trace_id,
+                            )
                             print(f"\n{'='*70}")
-                            print(f"📌 步骤 {step_count}: 执行效果检查")
+                            print(f"📌 步骤 {step_count}: {_step_label}")
                             print(f"{'='*70}")
 
                         # 打印简化结果
@@ -197,10 +188,19 @@ async def main():
         print(f"  - 最终报告（带格式）：examples/auto_put_ad_mini/outputs/reports/decision_*.xlsx")
         print()
 
+        pipeline_logger.info(
+            "event=pipeline stage=run result=completed duration_ms=%d steps=%d trace_id=%s",
+            int(time.monotonic() * 1000) - started_ms, step_count, trace_id,
+        )
+
     except Exception as e:
         print(f"\n❌ 执行失败: {e}")
         import traceback
         traceback.print_exc()
+        pipeline_logger.error(
+            "event=pipeline stage=run result=failed duration_ms=%d error=%s trace_id=%s",
+            int(time.monotonic() * 1000) - started_ms, str(e)[:200], trace_id,
+        )
 
 
 if __name__ == "__main__":

examples/auto_put_ad_mini/execute_once_test.py

@@ -25,10 +25,10 @@ load_dotenv()
 from agent.core.runner import AgentRunner
 from agent.trace import FileSystemTraceStore, Trace, Message
 from agent.llm import create_openrouter_llm_call
-from agent.utils import setup_logging
+from examples.auto_put_ad_mini.logging_setup import setup_logging
 
 from examples.auto_put_ad_mini.config import (
-    MAIN_CONFIG, SKILLS_DIR, TRACE_STORE_PATH, LOG_LEVEL, LOG_FILE,
+    MAIN_CONFIG, SKILLS_DIR, TRACE_STORE_PATH, LOG_LEVEL,
 )
 
 # 导入自定义工具
@@ -49,7 +49,7 @@ except ImportError:
 
 async def main():
     base_dir = Path(__file__).parent
-    setup_logging(level=LOG_LEVEL, file=LOG_FILE)
+    setup_logging(level=LOG_LEVEL)
 
     prompt_path = base_dir / "prompts" / "system.prompt"
     system_prompt = prompt_path.read_text(encoding="utf-8") if prompt_path.exists() else ""

examples/auto_put_ad_mini/logging_setup.py

@@ -0,0 +1,361 @@
+"""Logging setup owned by the auto_put_ad_mini example.
+
+The example is also run as standalone scripts, so this module deliberately
+does not depend on ``agent.utils.logging``.  Logs go to the console and,
+when configured, to separate Alibaba Cloud SLS Logstores (INFO/WARNING and
+ERROR/CRITICAL).  No local file logging is performed.
+"""
+
+from __future__ import annotations
+
+import atexit
+import logging
+import os
+import random
+import string
+import sys
+import threading
+from datetime import datetime
+from typing import Iterable, Optional
+
+
+_FORMAT = "%Y-%m-%d %H:%M:%S"
+_LOCAL_LOG_FORMAT = "%(asctime)s | %(levelname)s | %(name)s | %(message)s"
+_SLS_LOG_FORMAT = "%(message)s"
+_SLS_HANDLERS: list[logging.Handler] = []
+_CAPTURE_INSTALLED = False
+_CONFIGURED = False
+_EXCEPTION_HOOKS_INSTALLED = False
+_ATEXIT_REGISTERED = False
+_TRACE_ID: str | None = None
+
+
+def get_trace_id() -> str | None:
+    """Return the trace_id of the current session, or None if not yet initialised."""
+    return _TRACE_ID
+
+
+class _LevelFilter(logging.Filter):
+    def __init__(self, minimum: int, maximum: Optional[int] = None):
+        super().__init__()
+        self.minimum = minimum
+        self.maximum = maximum
+
+    def filter(self, record: logging.LogRecord) -> bool:
+        return record.levelno >= self.minimum and (
+            self.maximum is None or record.levelno <= self.maximum
+        )
+
+
+class _ExcludeCapturedOutput(logging.Filter):
+    def filter(self, record: logging.LogRecord) -> bool:
+        return not getattr(record, "captured_output", False)
+
+
+class _TraceIdFilter(logging.Filter):
+    """Inject ``trace_id`` into every LogRecord for structured SLS queries.
+
+    The filter reads from the module-level ``_TRACE_ID`` so that it works even
+    when ``setup_logging()`` is called after the filter is added.
+    """
+
+    def filter(self, record: logging.LogRecord) -> bool:
+        record.trace_id = _TRACE_ID or "-"
+        return True
+
+
+class _SafeQueuedLogHandler:
+    """Create an SDK handler whose emit failures cannot break the app logger,
+    and whose timestamps are forced to second-level precision.
+
+    The SLS protobuf schema includes ``optional fixed32 Time_ns`` — when set
+    (even to 0) the console renders ``.000000000``.  We strip it so only
+    integer-second timestamps appear.
+    """
+
+    @staticmethod
+    def create(handler_cls, **kwargs) -> logging.Handler:
+        handler = handler_cls(**kwargs)
+
+        # -- safe emit ----------------------------------------------------
+        original_emit = handler.emit
+
+        def safe_emit(record):
+            try:
+                original_emit(record)
+            except Exception:
+                logging.Handler.handleError(handler, record)
+
+        handler.emit = safe_emit
+
+        # -- strip Time_ns from protobuf before serialization -------------
+        # The SDK populates LogItem.time_nano_part from ``time.time()``
+        # rather than ``record.created``, injecting a random 9-digit value.
+        # Even when forced to 0 the optional protobuf field *is* serialised
+        # (fixed32 0 ≠ absent) and the console renders ``.000000000``.
+        # Monkey-patching ``LogGroup.SerializeToString`` (called once per
+        # batch inside ``LogClient.put_logs``) clears the field right before
+        # encoding without duplicating the ~40-line SDK method.
+        import aliyun.log.log_logs_pb2 as _sls_pb
+
+        _orig_serialize = _sls_pb.LogGroup.SerializeToString
+        original_send = handler.send
+
+        def patched_send(req):
+            def _no_nano_serialize(self):
+                for log in self.Logs:
+                    if log.HasField("Time_ns"):
+                        log.ClearField("Time_ns")
+                return _orig_serialize(self)
+
+            _sls_pb.LogGroup.SerializeToString = _no_nano_serialize
+            try:
+                return original_send(req)
+            finally:
+                _sls_pb.LogGroup.SerializeToString = _orig_serialize
+
+        handler.send = patched_send
+        return handler
+
+
+class _CapturedStream:
+    """Mirror a process stream into logging while preserving interactive output."""
+
+    def __init__(self, original, level: int, logger_name: str):
+        self._original = original
+        self._level = level
+        self._logger = logging.getLogger(logger_name)
+        self._buffer = ""
+
+    def write(self, value: str) -> int:
+        if not value:
+            return 0
+        self._original.write(value)
+        self._original.flush()
+        self._buffer += value
+        while "\n" in self._buffer:
+            line, self._buffer = self._buffer.split("\n", 1)
+            line = line.rstrip("\r")
+            if line.strip():
+                self._logger.log(
+                    self._level,
+                    "%s",
+                    line,
+                    extra={"captured_output": True},
+                )
+        return len(value)
+
+    def flush(self) -> None:
+        self._original.flush()
+        if self._buffer.strip():
+            self._logger.log(
+                self._level,
+                "%s",
+                self._buffer.strip(),
+                extra={"captured_output": True},
+            )
+            self._buffer = ""
+
+    def isatty(self) -> bool:
+        return self._original.isatty()
+
+    def fileno(self) -> int:
+        return self._original.fileno()
+
+    @property
+    def encoding(self):
+        return getattr(self._original, "encoding", "utf-8")
+
+
+def _configured_sls_handlers(log_level: int) -> Iterable[logging.Handler]:
+    try:
+        from aliyun.log import QueuedLogHandler
+    except ImportError:
+        logging.getLogger(__name__).warning(
+            "[sls] aliyun-log-python-sdk 未安装，跳过 SLS 上报"
+        )
+        return []
+
+    sls_endpoint = os.getenv("SLS_ENDPOINT", "").strip()
+    access_key_id = os.getenv("SLS_ACCESS_KEY_ID", "").strip()
+    access_key_secret = os.getenv("SLS_ACCESS_KEY_SECRET", "").strip()
+    project = os.getenv("SLS_PROJECT", "auto-put-tecent").strip()
+    info_logstore = os.getenv(
+        "SLS_INFO_LOGSTORE", os.getenv("SLS_LOGSTORE", "info-log")
+    ).strip()
+    error_logstore = os.getenv("SLS_ERROR_LOGSTORE", "error-log").strip()
+    batch_size = int(os.getenv("SLS_BATCH_SIZE_MAX", "1024"))
+    put_wait_ms = int(os.getenv("SLS_PUT_WAIT_MS", "2000"))
+    sls_level = getattr(
+        logging, os.getenv("SLS_LOG_LEVEL", "INFO").upper(), logging.INFO
+    )
+
+    if not all(
+        (sls_endpoint, access_key_id, access_key_secret,
+         project, info_logstore, error_logstore)
+    ):
+        logging.getLogger(__name__).warning(
+            "[sls] SLS 凭证或 Logstore 未完整配置，跳过上报"
+        )
+        return []
+
+    # SLS uses its own __time__ field, and level/name/func/file/line/thread are
+    # already sent as separate fields via the ``fields`` list. Keep only the
+    # message body to avoid redundant data cluttering the SLS console.
+    sls_formatter = logging.Formatter(_SLS_LOG_FORMAT)
+    common = dict(
+        end_point=sls_endpoint,
+        access_key_id=access_key_id,
+        access_key=access_key_secret,
+        project=project,
+        fields=[
+            "record_name", "level", "func_name", "module", "file_path", "line_no",
+            "process_id", "process_name", "thread_id", "thread_name",
+        ],
+        extract_kv=True,
+        batch_size=batch_size,
+        put_wait=max(float(put_wait_ms) / 1000, 0.1),
+    )
+    handlers = []
+    for store, level_filter in (
+        (info_logstore, _LevelFilter(sls_level, logging.WARNING)),
+        (error_logstore, _LevelFilter(logging.ERROR)),
+    ):
+        handler = _SafeQueuedLogHandler.create(
+            QueuedLogHandler, log_store=store, **common
+        )
+        handler.setLevel(sls_level if store == info_logstore else logging.ERROR)
+        handler.addFilter(level_filter)
+        handler.setFormatter(sls_formatter)
+        handlers.append(handler)
+    return handlers
+
+
+def attach_sls_handler(root_logger: Optional[logging.Logger] = None) -> bool:
+    """Attach the example-owned SLS handlers once, returning whether enabled."""
+    global _SLS_HANDLERS
+    root_logger = root_logger or logging.getLogger()
+    if _SLS_HANDLERS and all(handler in root_logger.handlers for handler in _SLS_HANDLERS):
+        return True
+    # 清理残留的旧 SLS handler（如 setup_logging 被多次调用导致部分 handler 残留）
+    for old_handler in list(_SLS_HANDLERS):
+        if old_handler in root_logger.handlers:
+            root_logger.removeHandler(old_handler)
+            try:
+                old_handler.close()
+            except Exception:
+                pass
+    _SLS_HANDLERS = []
+    try:
+        handlers = list(_configured_sls_handlers(root_logger.level or logging.INFO))
+        for handler in handlers:
+            root_logger.addHandler(handler)
+        _SLS_HANDLERS = handlers
+        if handlers:
+            logging.getLogger(__name__).info(
+                "[sls] 已挂载双 Logstore: project=%s info-log/error-log",
+                os.getenv("SLS_PROJECT", "auto-put-tecent"),
+            )
+        return bool(handlers)
+    except Exception:
+        logging.getLogger(__name__).exception("[sls] handler 挂载失败，降级为本地日志")
+        return False
+
+
+def _close_sls_handlers() -> None:
+    for handler in _SLS_HANDLERS:
+        try:
+            handler.close()
+        except Exception:
+            pass
+
+
+def _install_exception_hooks() -> None:
+    global _EXCEPTION_HOOKS_INSTALLED
+    if _EXCEPTION_HOOKS_INSTALLED:
+        return
+
+    original_sys_hook = sys.excepthook
+
+    def log_uncaught_exception(exc_type, exc_value, exc_traceback) -> None:
+        if issubclass(exc_type, KeyboardInterrupt):
+            original_sys_hook(exc_type, exc_value, exc_traceback)
+            return
+        logging.getLogger("auto_put_ad_mini.uncaught").critical(
+            "event=uncaught_exception thread=main result=failed",
+            exc_info=(exc_type, exc_value, exc_traceback),
+        )
+
+    def log_thread_exception(args: threading.ExceptHookArgs) -> None:
+        logging.getLogger("auto_put_ad_mini.uncaught").critical(
+            "event=uncaught_exception thread=%s result=failed",
+            args.thread.name if args.thread else "unknown",
+            exc_info=(args.exc_type, args.exc_value, args.exc_traceback),
+        )
+
+    sys.excepthook = log_uncaught_exception
+    threading.excepthook = log_thread_exception
+    _EXCEPTION_HOOKS_INSTALLED = True
+
+
+def setup_logging(
+    level: str = "INFO",
+    capture_output: bool = True,
+) -> None:
+    """Configure root logging: console + optional SLS, no local file."""
+    global _CAPTURE_INSTALLED, _CONFIGURED, _TRACE_ID
+
+    # Generate a per-session trace_id once per process lifetime.
+    if _TRACE_ID is None:
+        _TRACE_ID = _generate_trace_id()
+
+    log_level = getattr(logging, str(level).upper(), logging.INFO)
+    stdout = sys.__stdout__
+    stderr = sys.__stderr__
+    trace_filter = _TraceIdFilter()
+    console = logging.StreamHandler(stdout)
+    console.setLevel(log_level)
+    console.addFilter(_ExcludeCapturedOutput())
+    console.addFilter(trace_filter)
+    formatter = logging.Formatter(_LOCAL_LOG_FORMAT, datefmt=_FORMAT)
+    console.setFormatter(formatter)
+
+    root_logger = logging.getLogger()
+    # Handler failures (for example an SLS network outage) must never write
+    # recursive logging tracebacks to stderr or interrupt the business flow.
+    logging.raiseExceptions = False
+    if not _CONFIGURED:
+        logging.basicConfig(
+            level=log_level,
+            handlers=[console],
+            force=True,
+        )
+        _CONFIGURED = True
+    else:
+        root_logger.setLevel(log_level)
+    for noisy in ("httpx", "httpcore", "urllib3", "apscheduler", "aliyun.log"):
+        logging.getLogger(noisy).setLevel(logging.WARNING)
+
+    attach_sls_handler(root_logger)
+    _install_exception_hooks()
+    if capture_output and not _CAPTURE_INSTALLED:
+        sys.stdout = _CapturedStream(stdout, logging.INFO, "auto_put_ad_mini.stdout")
+        sys.stderr = _CapturedStream(stderr, logging.ERROR, "auto_put_ad_mini.stderr")
+        _CAPTURE_INSTALLED = True
+    global _ATEXIT_REGISTERED
+    if not _ATEXIT_REGISTERED:
+        atexit.register(_close_sls_handlers)
+        _ATEXIT_REGISTERED = True
+    logging.getLogger(__name__).info("[logging] 日志初始化完成 trace_id=%s", _TRACE_ID)
+
+
+def _generate_trace_id() -> str:
+    """Generate a compact, sortable, unique trace_id for one process invocation.
+
+    Format: ``YYYYMMDD-HHMMSS-{8 random hex chars}``.
+    ~16M combinations per second; sufficient for cron-triggered processes.
+    """
+    now = datetime.now()
+    ts = now.strftime("%Y%m%d-%H%M%S")
+    suffix = "".join(random.choices(string.ascii_lowercase + string.digits, k=8))
+    return f"{ts}-{suffix}"

examples/auto_put_ad_mini/quick_analysis.py

@@ -12,7 +12,7 @@ sys.path.insert(0, str(Path(__file__).parent.parent.parent))
 from dotenv import load_dotenv
 load_dotenv()
 
-from agent.utils import setup_logging
+from examples.auto_put_ad_mini.logging_setup import setup_logging
 setup_logging(level="INFO")
 
 # 设置环境

examples/auto_put_ad_mini/refresh_roi_fission_multiplier.py

@@ -20,6 +20,8 @@ for path in (ROOT, HERE, RTC_DIR):
 load_dotenv(HERE / ".env", override=False)
 load_dotenv(Path.cwd() / ".env", override=False)
 
+from logging_setup import setup_logging
+
 from roi_control.fission_multiplier import (
     DEFAULT_FISSION_PARAMETER_VERSION,
     FISSION_HORIZON_DAYS,
@@ -108,6 +110,7 @@ def publish_version(version: str, published_by: str) -> int:
 
 
 def main() -> int:
+    setup_logging(level=os.getenv("LOG_LEVEL", "INFO"))
     args = build_parser().parse_args()
     if args.publish_version:
         if args.sql_only:

examples/auto_put_ad_mini/requirements.txt

@@ -40,5 +40,8 @@ pymysql>=1.1.0
 # Feishu SDK
 lark-oapi>=1.5.0
 
+# Alibaba Cloud Simple Log Service (SLS)
+aliyun-log-python-sdk>=0.7.16
+
 # Optional: Prometheus metrics
-# prometheus-client>=0.19.0
+prometheus-client>=0.19.0

examples/auto_put_ad_mini/run.py

@@ -28,11 +28,11 @@ if http_proxy or https_proxy:
 from agent.core.runner import AgentRunner
 from agent.trace import FileSystemTraceStore, Trace, Message
 from agent.llm import create_openrouter_llm_call
-from agent.utils import setup_logging
+from examples.auto_put_ad_mini.logging_setup import setup_logging
 
 # 导入配置
 from examples.auto_put_ad_mini.config import (
-    MAIN_CONFIG, SKILLS_DIR, TRACE_STORE_PATH, LOG_LEVEL, LOG_FILE,
+    MAIN_CONFIG, SKILLS_DIR, TRACE_STORE_PATH, LOG_LEVEL,
 )
 
 # 导入自定义工具（触发 @tool 注册）
@@ -99,7 +99,7 @@ async def main():
     """交互式主入口：初始化 Agent 后循环读取用户指令并流式输出执行过程"""
     base_dir = Path(__file__).parent
 
-    setup_logging(level=LOG_LEVEL, file=LOG_FILE)
+    setup_logging(level=LOG_LEVEL)
 
     system_prompt = _load_system_prompt(base_dir)
     _load_presets(base_dir)

examples/auto_put_ad_mini/run_daily_roi.py

@@ -8,6 +8,7 @@ import json
 import logging
 import os
 import sys
+import time
 from pathlib import Path
 
 from dotenv import load_dotenv
@@ -48,20 +49,37 @@ def parse_args() -> argparse.Namespace:
 
 
 def main() -> None:
+    from logging_setup import setup_logging
+    setup_logging(level=os.getenv("LOG_LEVEL", "INFO"))
     args = parse_args()
-    result = run_daily_roi(
-        requested_end_date=args.end_date,
-        output_dir=args.output_dir,
-        send_feishu=args.send_feishu,
-        source_revision=args.source_revision,
-        internal_test=args.internal_test,
+    logger = logging.getLogger("auto_put_ad_mini.run_daily_roi")
+    started = time.monotonic()
+    logger.info(
+        "event=daily_roi stage=run result=started end_date=%s send_feishu=%s "
+        "internal_test=%s output_dir=%s",
+        args.end_date or "auto", args.send_feishu, args.internal_test, args.output_dir,
+    )
+    try:
+        result = run_daily_roi(
+            requested_end_date=args.end_date,
+            output_dir=args.output_dir,
+            send_feishu=args.send_feishu,
+            source_revision=args.source_revision,
+            internal_test=args.internal_test,
+        )
+    except Exception:
+        logger.exception(
+            "event=daily_roi stage=run result=failed duration_ms=%d",
+            int((time.monotonic() - started) * 1000),
+        )
+        raise
+    logger.info(
+        "event=daily_roi stage=run result=succeeded duration_ms=%d run_id=%s",
+        int((time.monotonic() - started) * 1000),
+        result.get("run_id", "") if isinstance(result, dict) else "",
     )
     print(json.dumps(result, ensure_ascii=False, default=str))
 
 
 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/run_daily_service.py

@@ -9,6 +9,7 @@ import logging
 import os
 import subprocess
 import sys
+import time
 from pathlib import Path
 
 from apscheduler.schedulers.blocking import BlockingScheduler
@@ -101,22 +102,46 @@ def _run_script(
     lock_name: str,
     extra_args: list[str] | None = None,
 ) -> None:
+    task_args = extra_args or []
+    started = time.monotonic()
+    logger.info(
+        "event=scheduled_task task=%s stage=lock_wait result=started lock=%s args=%s",
+        script_name, lock_name, task_args,
+    )
     with advisory_lock(lock_name) as acquired:
         if not acquired:
-            logger.warning("Skip %s: another instance holds %s", script_name, lock_name)
+            logger.warning(
+                "event=scheduled_task task=%s stage=lock_wait result=skipped "
+                "reason=lock_held lock=%s",
+                script_name, lock_name,
+            )
             return
-        logger.info("Starting %s", script_name)
+        logger.info(
+            "event=scheduled_task task=%s stage=subprocess result=started lock=%s",
+            script_name, lock_name,
+        )
         try:
             completed = subprocess.run(
-                [sys.executable, str(HERE / script_name), *(extra_args or [])],
+                [sys.executable, str(HERE / script_name), *task_args],
                 cwd=HERE,
                 check=False,
             )
         except Exception as exc:
+            logger.exception(
+                "event=scheduled_task task=%s stage=subprocess result=failed "
+                "duration_ms=%d",
+                script_name, int((time.monotonic() - started) * 1000),
+            )
             if script_name == "run_daily_roi.py":
                 _notify_roi_failure(extra_args=extra_args, error=str(exc))
             raise
         if completed.returncode:
+            logger.error(
+                "event=scheduled_task task=%s stage=subprocess result=failed "
+                "returncode=%d duration_ms=%d",
+                script_name, completed.returncode,
+                int((time.monotonic() - started) * 1000),
+            )
             if script_name == "run_daily_roi.py":
                 _notify_roi_failure(
                     extra_args=extra_args,
@@ -125,7 +150,11 @@ def _run_script(
             raise RuntimeError(
                 f"{script_name} exited with code {completed.returncode}"
             )
-        logger.info("Finished %s", script_name)
+        logger.info(
+            "event=scheduled_task task=%s stage=subprocess result=succeeded "
+            "returncode=0 duration_ms=%d",
+            script_name, int((time.monotonic() - started) * 1000),
+        )
 
 
 def run_creation() -> None:
@@ -191,6 +220,9 @@ def _positive_account_id(raw: str) -> int:
 
 
 def main(argv: list[str] | None = None) -> None:
+    from logging_setup import setup_logging
+    setup_logging(level=os.getenv("LOG_LEVEL", "INFO"))
+    logger.info("event=daily_service stage=startup result=started")
     parser = argparse.ArgumentParser(description="广告日级生产调度服务")
     parser.add_argument(
         "--run-creation-once",
@@ -315,15 +347,11 @@ def main(argv: list[str] | None = None) -> None:
             name="启动时ROI计算",
         )
     logger.info(
-        "Daily service started jobs=%s",
+        "event=daily_service stage=scheduler result=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/run_decision_test.py

@@ -172,7 +172,7 @@ async def check_im_config():
     print(f"\n💡 当前状态分析:")
     if not IM_ENABLED:
         print(f"   → 飞书审批已关闭，决策不会发送到飞书")
-        print(f"   → 建议：config.py 中设置 IM_ENABLED=True")
+        print(f"   → 设置环境变量 IM_ENABLED=true 或在 config.py 中修改默认值")
     elif not EXECUTION_ENABLED:
         print(f"   → 执行开关关闭，系统处于【只验证不执行】模式")
         print(f"   → 在此模式下，决策会被验证但不会真正调用API")

examples/auto_put_ad_mini/scan_creative_reviews.py

@@ -8,6 +8,7 @@ import json
 import logging
 import os
 import sys
+import time
 from pathlib import Path
 
 _HERE = Path(__file__).parent
@@ -28,19 +29,8 @@ logger = logging.getLogger("scan_creative_reviews")
 
 def _setup_logging() -> None:
     """初始化日志配置：设置格式、降噪第三方 logger，并尝试挂载 SLS 上报。"""
-    logging.basicConfig(
-        level=os.getenv("LOG_LEVEL", "INFO"),
-        format="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
-        datefmt="%H:%M:%S",
-    )
-    for noisy in ("httpx", "httpcore", "config", "db.config"):
-        logging.getLogger(noisy).setLevel(logging.WARNING)
-
-    try:
-        from tools.sls_setup import attach_sls_handler
-        attach_sls_handler()
-    except Exception as e:
-        logger.warning("[sls] 挂载异常(降级为本地 only):%s", e)
+    from logging_setup import setup_logging
+    setup_logging(level=os.getenv("LOG_LEVEL", "INFO"))
 
 
 def main() -> int:
@@ -51,16 +41,27 @@ def main() -> int:
     batch_size = int(os.getenv("CREATIVE_REVIEW_BATCH_SIZE", "100"))
 
     logger.info(
-        "[creative_review] 扫描启动 lookback_hours=%d limit=%d batch_size=%d",
+        "event=creative_review stage=scan result=started lookback_hours=%d "
+        "limit=%d batch_size=%d",
         lookback_hours, limit, batch_size,
     )
-    summary = scan_pending_reviews(
-        lookback_hours=lookback_hours,
-        limit=limit,
-        batch_size=batch_size,
-    )
+    started = time.monotonic()
+    try:
+        summary = scan_pending_reviews(
+            lookback_hours=lookback_hours,
+            limit=limit,
+            batch_size=batch_size,
+        )
+    except Exception:
+        logger.exception(
+            "event=creative_review stage=scan result=failed duration_ms=%d",
+            int((time.monotonic() - started) * 1000),
+        )
+        raise
     logger.info(
-        "[creative_review] 扫描完成 %s",
+        "event=creative_review stage=scan result=%s duration_ms=%d summary=%s",
+        "failed" if summary.get("errors") else "succeeded",
+        int((time.monotonic() - started) * 1000),
         json.dumps(summary, ensure_ascii=False),
     )
     return 0 if not summary.get("errors") else 1

examples/auto_put_ad_mini/server.py

@@ -14,6 +14,8 @@ from apscheduler.triggers.cron import CronTrigger
 from fastapi import FastAPI, HTTPException
 from fastapi.responses import JSONResponse
 
+from logging_setup import setup_logging
+
 # 添加项目根目录到 Python 路径
 sys.path.insert(0, str(Path(__file__).parent.parent.parent))
 
@@ -32,12 +34,22 @@ scheduler: AsyncIOScheduler = None
 # ═══════════════════════════════════════════
 async def scheduled_decision_job():
     """定时执行决策流程"""
-    logger.info(f"[定时任务] 开始执行决策流程 - {datetime.now(timezone.utc)}")
+    import time as _time
+    _started_ms = int(_time.monotonic() * 1000)
+    logger.info("event=server stage=scheduled_job result=started")
     try:
         await execute_decision_pipeline()
-        logger.info(f"[定时任务] 决策流程执行完成")
+        logger.info(
+            "event=server stage=scheduled_job result=completed duration_ms=%d",
+            int(_time.monotonic() * 1000) - _started_ms,
+        )
     except Exception as e:
-        logger.error(f"[定时任务] 执行失败: {e}", exc_info=True)
+        logger.error(
+            "event=server stage=scheduled_job result=failed error=%s duration_ms=%d",
+            str(e)[:200],
+            int(_time.monotonic() * 1000) - _started_ms,
+            exc_info=True,
+        )
 
 # ═══════════════════════════════════════════
 # FastAPI 应用
@@ -53,6 +65,9 @@ async def startup():
     """服务启动时初始化"""
     global scheduler
 
+    setup_logging(level=os.getenv("LOG_LEVEL", "INFO"))
+    logger.info("event=server stage=startup result=started")
+
     logger.info("=" * 60)
     logger.info("广告智能调控服务启动中...")
     logger.info("启动 APScheduler")
@@ -206,12 +221,7 @@ async def manual_trigger():
 if __name__ == "__main__":
     import uvicorn
 
-    # 配置日志输出到 stdout（Kubernetes 自动收集）
-    logging.basicConfig(
-        level=logging.INFO,
-        format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
-        handlers=[logging.StreamHandler(sys.stdout)]
-    )
+    setup_logging(level=os.getenv("LOG_LEVEL", "INFO"))
 
     port = int(os.getenv("PORT", 8080))
     uvicorn.run(

examples/auto_put_ad_mini/sync_ad_status.py

@@ -44,11 +44,8 @@ logger = logging.getLogger("sync_ad_status")
 
 def _setup_logging() -> None:
     """初始化日志配置：INFO 级别输出到 stdout。"""
-    logging.basicConfig(
-        level=logging.INFO,
-        format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
-        stream=sys.stdout,
-    )
+    from logging_setup import setup_logging
+    setup_logging(level="INFO")
 
 
 def _fetch_api_ad_ids(account_id: int, page_size: int) -> Set[int]:
@@ -140,8 +137,12 @@ def main() -> int:
         )
         try:
             api_ids = _fetch_api_ad_ids(acct, args.page_size)
-        except Exception as e:
-            logger.error("[account=%s] 拉取 API 失败：%s（跳过该账号）", acct, e)
+        except Exception:
+            logger.exception(
+                "event=sync_ad_status stage=fetch_account result=failed "
+                "account_id=%s（跳过该账号）",
+                acct,
+            )
             failed_accounts.append(acct)
             continue
 

examples/auto_put_ad_mini/sync_feishu_account_config.py

@@ -26,6 +26,7 @@ from __future__ import annotations
 
 import argparse
 import json
+import os
 import re
 import sys
 from datetime import date, datetime
@@ -40,6 +41,8 @@ load_dotenv(_HERE / ".env")
 sys.path.insert(0, str(_HERE.parent.parent))
 sys.path.insert(0, str(_HERE))
 
+from logging_setup import setup_logging  # noqa: E402
+
 from configure_creation_accounts import (  # noqa: E402
     AccountConfigInput,
     DEFAULT_AGE_MAX,
@@ -351,6 +354,7 @@ def main() -> int:
     parser.add_argument("--allow-row-errors", action="store_true")
     args = parser.parse_args()
 
+    setup_logging(level=os.getenv("LOG_LEVEL", "INFO"))
     stats = sync_from_feishu(
         spreadsheet_token=args.spreadsheet_token,
         sheet_id=args.sheet_id,

examples/auto_put_ad_mini/test_analysis_0415.py

@@ -19,10 +19,10 @@ load_dotenv()
 from agent.core.runner import AgentRunner
 from agent.trace import FileSystemTraceStore
 from agent.llm import create_openrouter_llm_call
-from agent.utils import setup_logging
+from examples.auto_put_ad_mini.logging_setup import setup_logging
 
 from examples.auto_put_ad_mini.config import (
-    MAIN_CONFIG, SKILLS_DIR, TRACE_STORE_PATH, LOG_LEVEL, LOG_FILE,
+    MAIN_CONFIG, SKILLS_DIR, TRACE_STORE_PATH, LOG_LEVEL,
 )
 
 # 导入自定义工具
@@ -41,7 +41,7 @@ from examples.auto_put_ad_mini.tools.im_approval import send_approval_request, c
 async def main():
     base_dir = Path(__file__).parent
 
-    setup_logging(level=LOG_LEVEL, file=LOG_FILE)
+    setup_logging(level=LOG_LEVEL)
 
     # 加载 system prompt
     prompt_path = base_dir / "prompts" / "system.prompt"

examples/auto_put_ad_mini/test_approval_replay.py

@@ -22,10 +22,10 @@ load_dotenv()
 from agent.core.runner import AgentRunner
 from agent.trace import FileSystemTraceStore
 from agent.llm import create_openrouter_llm_call
-from agent.utils import setup_logging
+from examples.auto_put_ad_mini.logging_setup import setup_logging
 
 from examples.auto_put_ad_mini.config import (
-    MAIN_CONFIG, SKILLS_DIR, TRACE_STORE_PATH, LOG_LEVEL, LOG_FILE,
+    MAIN_CONFIG, SKILLS_DIR, TRACE_STORE_PATH, LOG_LEVEL,
 )
 
 # 触发 @tool 注册
@@ -49,7 +49,7 @@ def _load_system_prompt(base_dir: Path) -> str:
 
 
 async def main():
-    setup_logging(level=LOG_LEVEL, file=LOG_FILE)
+    setup_logging(level=LOG_LEVEL)
     base_dir = Path(__file__).parent
 
     validated_csv = sys.argv[1] if len(sys.argv) > 1 else str(

examples/auto_put_ad_mini/test_feishu_approval.py

@@ -33,7 +33,7 @@ async def test_feishu_message():
 
     if not IM_ENABLED:
         print("\n❌ IM_ENABLED=False，飞书功能未启用")
-        print("   请在 config.py 中设置 IM_ENABLED=True")
+        print("   请设置环境变量 IM_ENABLED=true 或在 config.py 中修改默认值")
         return False
 
     print(f"\n✅ IM_ENABLED=True，飞书功能已启用")

examples/auto_put_ad_mini/tools/execution_engine.py

@@ -456,6 +456,15 @@ async def execute_decisions(
         filter_ad_ids: 仅执行该 ad_id 列表中的决策（部分批准型协议主路径）；为 None 时执行全部
     """
     try:
+        import time as _time
+        _started_ms = int(_time.monotonic() * 1000)
+        logger.info(
+            "event=execute_decisions stage=run result=started "
+            "validated_csv=%s approval_mode=%s filter_ids=%s",
+            validated_csv or "auto", approval_mode,
+            len(filter_ad_ids) if filter_ad_ids else "all",
+        )
+
         if not EXECUTION_ENABLED:
             return ToolResult(
                 title="执行引擎未启用",
@@ -624,6 +633,12 @@ async def execute_decisions(
             else:
                 failed += 1
 
+            logger.info(
+                "event=execute_decisions stage=auto_exec ad_id=%d account_id=%d "
+                "action=%s tier=1 result=%s api_code=%d",
+                ad_id, account_id, action, exec_status, api_code,
+            )
+
             post_state = await executor.get_ad_state(ad_id, account_id) if exec_status in ("success", "partial") else None
 
             audit_entry = {
@@ -812,6 +827,12 @@ async def execute_decisions(
                         audit_entry.update(pause_extra)
                     audit.log(audit_entry)
 
+                    logger.info(
+                        "event=execute_decisions stage=approved_exec ad_id=%d account_id=%d "
+                        "action=%s tier=%d result=%s api_code=%d",
+                        ad_id, account_id, action, tier, exec_status, api_code,
+                    )
+
         total_executed = executed + approved_executed
 
         output_lines = [
@@ -838,6 +859,13 @@ async def execute_decisions(
             f"  Tier 3 (高价值): {tier_summary.get(3, 0)} 个",
         ])
 
+        logger.info(
+            "event=execute_decisions stage=run result=completed "
+            "tier1_executed=%d tier1_failed=%d approved=%d rejected=%d pending=%d "
+            "duration_ms=%d",
+            executed, failed, approved_executed, rejected_count, pending_approval,
+            int(_time.monotonic() * 1000) - _started_ms,
+        )
         return ToolResult(
             title=f"执行完成（自动{executed}/审批通过{approved_executed}/拒绝{rejected_count}/超时{timeout_count}）",
             output="\n".join(output_lines),
@@ -854,7 +882,12 @@ async def execute_decisions(
         )
 
     except Exception as e:
-        logger.error("execute_decisions 失败: %s", e, exc_info=True)
+        logger.error(
+            "event=execute_decisions stage=run result=failed duration_ms=%d error=%s",
+            int(_time.monotonic() * 1000) - _started_ms if '_started_ms' in dir() else 0,
+            str(e)[:200],
+            exc_info=True,
+        )
         return ToolResult(title="execute_decisions 失败", output=str(e))
 
 
@@ -877,6 +910,14 @@ async def check_execution_feedback(
         hours_after: 操作后等待时间（小时），仅检查超过此时间的操作
     """
     try:
+        import time as _time
+        _started_ms = int(_time.monotonic() * 1000)
+        logger.info(
+            "event=check_execution_feedback stage=run result=started "
+            "log_path=%s hours_after=%d",
+            execution_log_path or "auto", hours_after,
+        )
+
         # 查找最新执行日志
         if not execution_log_path:
             log_dir = EXECUTION_LOG_DIR
@@ -973,6 +1014,12 @@ async def check_execution_feedback(
                         f"(执行于 {r['executed_at']})"
                     )
 
+        logger.info(
+            "event=check_execution_feedback stage=run result=completed "
+            "total=%d effective=%d ineffective=%d unknown=%d duration_ms=%d",
+            len(results), effective, ineffective, unknown,
+            int(_time.monotonic() * 1000) - _started_ms,
+        )
         return ToolResult(
             title=f"效果检查（有效{effective}/无效{ineffective}）",
             output="\n".join(output_lines),
@@ -986,5 +1033,8 @@ async def check_execution_feedback(
         )
 
     except Exception as e:
-        logger.error("check_execution_feedback 失败: %s", e, exc_info=True)
+        logger.error(
+            "event=check_execution_feedback stage=run result=failed error=%s",
+            str(e)[:200], exc_info=True,
+        )
         return ToolResult(title="check_execution_feedback 失败", output=str(e))

examples/auto_put_ad_mini/tools/guardrails.py

@@ -827,6 +827,14 @@ async def validate_decisions(
         dry_run: 是否强制干运行模式
     """
     try:
+        import time as _time
+        _started_ms = int(_time.monotonic() * 1000)
+        logger.info(
+            "event=validate_decisions stage=run result=started "
+            "decisions_csv=%s end_date=%s",
+            decisions_csv or "auto", end_date,
+        )
+
         if not GUARDRAILS_ENABLED:
             return ToolResult(
                 title="护栏已禁用",
@@ -940,6 +948,12 @@ async def validate_decisions(
             output_lines.append("")
             output_lines.append("⚠️ 当前为干运行模式（DRY_RUN），操作不会实际执行")
 
+        logger.info(
+            "event=validate_decisions stage=run result=completed "
+            "total=%d approved=%d blocked=%d modified=%d duration_ms=%d",
+            total, approved, blocked, modified,
+            int(_time.monotonic() * 1000) - _started_ms,
+        )
         return ToolResult(
             title=f"护栏验证（{total}条，拦截{blocked}）",
             output="\n".join(output_lines),

examples/auto_put_ad_mini/tools/im_approval.py

@@ -529,6 +529,14 @@ async def send_approval_request(
         poll_interval_seconds: 轮询间隔（秒）
     """
     try:
+        import time as _time
+        _started_ms = int(_time.monotonic() * 1000)
+        logger.info(
+            "event=send_approval_request stage=run result=started "
+            "validated_csv=%s wait_for_reply=%s timeout_min=%d",
+            validated_csv or "auto", wait_for_reply, timeout_minutes,
+        )
+
         if not IM_ENABLED:
             return ToolResult(
                 title="IM 审批未启用",
@@ -991,7 +999,10 @@ async def send_approval_request(
         )
 
     except Exception as e:
-        logger.error("send_approval_request 失败: %s", e, exc_info=True)
+        logger.error(
+            "event=send_approval_request stage=run result=failed error=%s",
+            str(e)[:200], exc_info=True,
+        )
         return ToolResult(title="send_approval_request 失败", output=str(e))
 
 

examples/auto_put_ad_mini/tools/landing_plan.py

@@ -8,7 +8,7 @@
 
 接口契约(严格按用户给的 curl,不增不减字段):
   POST https://tp-open.piaoquantv.com/contentPlatform/plan/xcx/save
-  header: token: <fixed>
+  认证:动态登录获取 token(POST /contentPlatform/account/login),请求头带 x-nonce + x-timestamp
   body: {
     "audiencePackage": str,              # 账户级 crowd_package
     "videoList": [                       # 单条逐次调,不批量
@@ -28,7 +28,7 @@ from urllib.parse import parse_qs, unquote
 
 import httpx
 
-from tools.video_recall import LandingVideo, PIAOQUANTV_TOKEN
+from tools.video_recall import LandingVideo, get_piaoquantv_token, refresh_piaoquantv_token, _build_auth_headers, _is_not_login_error
 
 logger = logging.getLogger(__name__)
 
@@ -110,20 +110,36 @@ def create_landing_plan(
             "experimentId": video.experiment_id,
         }],
     }
-    headers = {
-        "Content-Type": "application/json",
-        "token": PIAOQUANTV_TOKEN,
-        "Accept": "application/json, text/plain, */*",
-    }
+    token = get_piaoquantv_token()
+    headers = _build_auth_headers(token)
+
+    def _do_request():
+        return httpx.post(XCX_SAVE_URL, json=body, headers=headers, timeout=timeout)
 
     logger.info(
         "[landing_plan] POST xcx/save crowd=%r videoId=%d experimentId=%s",
         crowd_package, video.video_id, video.experiment_id,
     )
-    resp = httpx.post(XCX_SAVE_URL, json=body, headers=headers, timeout=timeout)
+    resp = _do_request()
+
+    # 401 或 Not_LOGIN(code=1000) 时刷新 token 并重试一次
+    if resp.status_code == 401:
+        logger.warning("[landing_plan] 收到 401,刷新 token 并重试")
+        token = refresh_piaoquantv_token()
+        headers = _build_auth_headers(token)
+        resp = _do_request()
+
     resp.raise_for_status()
     data = resp.json()
 
+    if _is_not_login_error(data):
+        logger.warning("[landing_plan] 收到 Not_LOGIN(1000),刷新 token 并重试")
+        token = refresh_piaoquantv_token()
+        headers = _build_auth_headers(token)
+        resp = _do_request()
+        resp.raise_for_status()
+        data = resp.json()
+
     if data.get("code") != 0 or not data.get("success"):
         raise RuntimeError(
             f"xcx/save 失败 videoId={video.video_id} "

examples/auto_put_ad_mini/tools/report_generator.py

@@ -203,6 +203,14 @@ async def generate_report(
         ToolResult，包含报告路径
     """
     try:
+        import time as _time
+        _started_ms = int(_time.monotonic() * 1000)
+        logger.info(
+            "event=generate_report stage=run result=started "
+            "decision_csv=%s end_date=%s",
+            decision_csv or "auto", end_date,
+        )
+
         if not decision_csv:
             decision_csv = str(_MINI_DIR / "outputs" / "decision_temp.csv")
 
@@ -252,6 +260,12 @@ async def generate_report(
         pause_count = (df_out["action"] == "pause").sum() if "action" in df_out.columns else 0
         total_count = len(df_out)
 
+        logger.info(
+            "event=generate_report stage=run result=completed "
+            "total_ads=%d pause_ads=%d duration_ms=%d",
+            total_count, pause_count,
+            int(_time.monotonic() * 1000) - _started_ms,
+        )
         return ToolResult(
             title=f"报告已生成（{total_count}个广告）",
             output=f"报告已生成:\n  CSV: {csv_path}\n  XLSX: {xlsx_path}\n\n共 {total_count} 个广告，其中 {pause_count} 个建议关停",
@@ -264,7 +278,9 @@ async def generate_report(
         )
 
     except Exception as e:
-        logger.error("报告生成失败: %s", e, exc_info=True)
-        return ToolResult(title="报告生成失败", output=str(e))
+        logger.error(
+            "event=generate_report stage=run result=failed error=%s",
+            str(e)[:200], exc_info=True,
+        )
 
 

examples/auto_put_ad_mini/tools/sls_setup.py

@@ -1,108 +1,60 @@
-"""阿里云 SLS 日志上报接入(2026-06-11)。
+"""阿里云 SLS 日志上报 — 工具集。
 
 数据流:
-  Python logger → 本地 FileHandler(全量保留)
-                → QueuedLogHandler(aliyun-log-python-sdk 自带,异步队列 + 批量 flush)
-                                                 → SLS Logstore
+  Python logger → logging_setup.setup_logging()
+                 → 本地 FileHandler（全量保留）
+                 → QueuedLogHandler（异步队列 + 批量 flush）
+                                                → SLS info-log / error-log
 
 设计原则:
-  · 配置缺失 → SLS_ENABLED=False → 不上报,主链路不受影响(降级)
-  · 上报失败 → SDK 内部 retry + 异常吞掉(不抛出),主链路不受影响
-  · INFO+ 全量上报(用户 2026-06-11 决策,日量级约 5k-20k 条)
+  · 配置缺失 → logging_setup 内部自动降级 → 不上报，主链路不受影响
+  · 上报失败 → SDK 内部 retry + SafeQueuedLogHandler 吞异常，主链路不受影响
+  · INFO/WARNING 进 info-log，ERROR/CRITICAL 进 error-log
+
+注意:
+  · 生产环境日志初始化入口是 ``logging_setup.setup_logging()``，已内置 SLS 挂载
+  · 本模块仅提供 smoke_test 等辅助工具，不重复实现日志初始化
 """
 
 from __future__ import annotations
 
 import logging
-from typing import Optional
+import os
+import sys
 
 logger = logging.getLogger(__name__)
 
 
-def attach_sls_handler(root_logger: Optional[logging.Logger] = None) -> bool:
-    """给 root_logger 挂上 SLS QueuedLogHandler(若 SLS_ENABLED)。
-
-    Returns:
-        True 已挂上 / False 跳过(配置缺失 / SDK 缺失 / 异常)
-    """
-    if root_logger is None:
-        root_logger = logging.getLogger()
-
-    try:
-        from config import (
-            SLS_ENABLED,
-            SLS_ENDPOINT,
-            SLS_ACCESS_KEY_ID,
-            SLS_ACCESS_KEY_SECRET,
-            SLS_PROJECT,
-            SLS_LOGSTORE,
-            SLS_LOG_LEVEL,
-        )
-    except ImportError as e:
-        logger.warning("[sls] config 缺 SLS_* 变量,跳过:%s", e)
-        return False
-
-    if not SLS_ENABLED:
-        logger.info("[sls] SLS_ENABLED=False(凭证未完整配置),跳过上报")
-        return False
+def smoke_test() -> bool:
+    """烟囱测试：发一条 INFO + 一条 WARNING + 一条 ERROR，等 3 秒让队列 flush。
 
-    try:
-        from aliyun.log import QueuedLogHandler
-    except ImportError:
-        logger.warning("[sls] aliyun-log-python-sdk 未安装,跳过上报(.venv/bin/pip3 install aliyun-log-python-sdk)")
-        return False
+    用法（从项目根目录）:
+        cd examples/auto_put_ad_mini
+        python -c "from tools.sls_setup import smoke_test; smoke_test()"
 
-    try:
-        sls_handler = QueuedLogHandler(
-            end_point=SLS_ENDPOINT,
-            access_key_id=SLS_ACCESS_KEY_ID,
-            access_key=SLS_ACCESS_KEY_SECRET,
-            project=SLS_PROJECT,
-            log_store=SLS_LOGSTORE,
-            # 自动提取 record 标准字段
-            fields=["levelname", "name", "filename", "lineno", "thread", "process"],
-            # message 中 k=v 形式自动解析为独立字段(便于 SLS 索引/查询)
-            extract_kv=True,
-        )
-        sls_handler.setLevel(getattr(logging, SLS_LOG_LEVEL, logging.INFO))
-        sls_handler.setFormatter(logging.Formatter(
-            "%(asctime)s | %(levelname)s | %(name)s | %(message)s",
-            datefmt="%Y-%m-%d %H:%M:%S",
-        ))
-        root_logger.addHandler(sls_handler)
-        logger.info(
-            "[sls] 已挂载 QueuedLogHandler → %s/%s/%s (level=%s)",
-            SLS_ENDPOINT, SLS_PROJECT, SLS_LOGSTORE, SLS_LOG_LEVEL,
-        )
-        return True
-    except Exception as e:
-        logger.warning("[sls] 挂载失败(降级为不上报,主链路不受影响):%s", e)
-        return False
+    然后去 SLS 控制台 logstore=info-log / error-log 查看是否收到。
+    """
 
+    # 确保 examples/auto_put_ad_mini 在 sys.path 中，兼容不同工作目录
+    _project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+    if _project_root not in sys.path:
+        sys.path.insert(0, _project_root)
 
-def smoke_test() -> bool:
-    """烟囱测试:发一条 INFO + 一条 WARNING + 一条 ERROR,等 3 秒让队列 flush。
+    # 加载 .env 中的 SLS 凭证（logging_setup 不自行 load_dotenv，由调用方负责）
+    from dotenv import load_dotenv
+    load_dotenv(os.path.join(_project_root, ".env"))
 
-    用法:python -c "from tools.sls_setup import smoke_test; smoke_test()"
-    然后去 SLS 控制台 logstore=info-log 查看是否收到。
-    """
-    import time
+    from logging_setup import setup_logging
 
-    logging.basicConfig(
-        level=logging.INFO,
-        format="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
-        datefmt="%H:%M:%S",
-    )
-    ok = attach_sls_handler()
-    if not ok:
-        logger.error("[smoke_test] SLS 未挂上,无法测试")
-        return False
+    # setup_logging() 内部已调用 attach_sls_handler()，无需重复挂载
+    setup_logging(level="INFO", capture_output=False)
 
     test_logger = logging.getLogger("sls_smoke_test")
     test_logger.info("[smoke_test] hello SLS - INFO 消息 event=smoke_info")
     test_logger.warning("[smoke_test] hello SLS - WARNING 消息 event=smoke_warn")
     test_logger.error("[smoke_test] hello SLS - ERROR 消息 event=smoke_error code=999")
-    logger.info("[smoke_test] 已发 3 条,等 3s 让 QueuedLogHandler flush")
+    logger.info("[smoke_test] 已发 3 条，等 3s 让 QueuedLogHandler flush")
+    import time
     time.sleep(3)
-    logger.info("[smoke_test] 测试完成,请去 SLS 控制台 logstore 查看")
+    logger.info("[smoke_test] 测试完成，请去 SLS 控制台 logstore 查看")
     return True

examples/auto_put_ad_mini/tools/video_recall.py

@@ -7,7 +7,7 @@
     (title / standardElement / categoryName / demandContentTitle 等)做 query 召回
 
 接口:POST https://tp-open.piaoquantv.com/contentPlatform/plan/videoContentList
-认证:token header(沿用用户给的值,可通过 env PIAOQUANTV_TOKEN 覆盖)
+认证:动态登录获取 token(POST /contentPlatform/account/login),请求头带 x-nonce + x-timestamp
 
 字段参考(2026-06-08 反推自 sample 调用):
   返回 data.objs[*] 39 个字段,关键:
@@ -20,6 +20,8 @@
 import logging
 import os
 import json
+import time
+import uuid
 from dataclasses import dataclass, field
 from pathlib import Path
 from typing import List, Optional
@@ -35,11 +37,106 @@ except Exception:
 
 logger = logging.getLogger(__name__)
 
+# ── piaoquantv contentPlatform 认证 & 请求头 ──────────────────────────
+PIAOQUANTV_LOGIN_URL = "https://tp-open.piaoquantv.com/contentPlatform/account/login"
 PIAOQUANTV_VIDEO_API = "https://tp-open.piaoquantv.com/contentPlatform/plan/videoContentList"
+
+# 登录凭据(环境变量注入)
+_CONTENTPLATFORM_TEL_NUM = os.getenv("CONTENTPLATFORM_TEL_NUM", "")
+_CONTENTPLATFORM_PASSWORD = os.getenv("CONTENTPLATFORM_PASSWORD", "")
+
+# 模块级 token 缓存(登录一次,后续复用;401 / Not_LOGIN 时刷新)
+_cached_token: Optional[str] = None
+_cached_sign_key: Optional[str] = None
+
+# contentPlatform 登录失效错误码
+_NOT_LOGIN_CODE = 1000
+
+
+def _is_not_login_error(data: dict) -> bool:
+    """判断 contentPlatform 返回是否因登录失效被拒绝。"""
+    return data.get("code") == _NOT_LOGIN_CODE
+
+
+def _build_auth_headers(token: str) -> dict:
+    """构造 contentPlatform 请求必需的鉴权头。
+
+    每次请求生成新的 x-nonce (UUID) 和 x-timestamp (epoch millis),
+    防止重放攻击。
+    """
+    return {
+        "content-type": "application/json",
+        "accept": "application/json, text/plain, */*",
+        "token": token,
+        "x-nonce": str(uuid.uuid4()),
+        "x-timestamp": str(int(time.time() * 1000)),
+    }
+
+
+def _login_piaoquantv() -> tuple[str, str]:
+    """调 contentPlatform 登录接口,返回 (token, signKey)。
+
+    Raises:
+        RuntimeError: 登录失败时抛出。
+    """
+    if not _CONTENTPLATFORM_TEL_NUM or not _CONTENTPLATFORM_PASSWORD:
+        raise RuntimeError(
+            "CONTENTPLATFORM_TEL_NUM / CONTENTPLATFORM_PASSWORD 环境变量未设置,无法登录 contentPlatform"
+        )
+    body = {
+        "telNum": _CONTENTPLATFORM_TEL_NUM,
+        "password": _CONTENTPLATFORM_PASSWORD,
+    }
+    logger.info("[piaoquantv] 登录 contentPlatform telNum=%s", _CONTENTPLATFORM_TEL_NUM)
+    resp = httpx.post(
+        PIAOQUANTV_LOGIN_URL,
+        json=body,
+        headers={"content-type": "application/json"},
+        timeout=15,
+    )
+    resp.raise_for_status()
+    data = resp.json()
+    if data.get("code") != 0 or not data.get("success"):
+        raise RuntimeError(
+            f"contentPlatform 登录失败: code={data.get('code')} msg={data.get('msg')}"
+        )
+    token = (data.get("data") or {}).get("token") or ""
+    sign_key = (data.get("data") or {}).get("signKey") or ""
+    if not token:
+        raise RuntimeError("contentPlatform 登录返回 token 为空")
+    logger.info("[piaoquantv] 登录成功, token=%s...", token[:8])
+    return token, sign_key
+
+
+def get_piaoquantv_token() -> str:
+    """获取 contentPlatform 的认证 token(带内存缓存)。
+
+    首次调用时登录获取 token;后续直接返回缓存值。
+    外部调用者无需关心 token 生命周期。
+    """
+    global _cached_token, _cached_sign_key
+    if _cached_token:
+        return _cached_token
+    token, sign_key = _login_piaoquantv()
+    _cached_token = token
+    _cached_sign_key = sign_key
+    return token
+
+
+def refresh_piaoquantv_token() -> str:
+    """强制刷新 token(用于 401 响应后的重试)。"""
+    global _cached_token, _cached_sign_key
+    _cached_token = None
+    _cached_sign_key = None
+    return get_piaoquantv_token()
+
+
+# 兼容旧代码的模块级常量(推荐改用 get_piaoquantv_token())
 PIAOQUANTV_TOKEN = os.getenv(
     "PIAOQUANTV_TOKEN",
-    "f6e07dba7fe3476cb31fd3733d607c5c",  # 用户 2026-06-08 提供
+    "f6e07dba7fe3476cb31fd3733d607c5c",  # 兜底:仅当未配置登录凭据时使用
 )
+
 PIAOQUANTV_VIDEO_SOURCE = os.getenv("PIAOQUANTV_VIDEO_SOURCE", "")
 _PIAOQUANTV_VIDEO_TYPE_RAW = os.getenv("PIAOQUANTV_VIDEO_TYPE", "").strip()
 PIAOQUANTV_VIDEO_TYPE = int(_PIAOQUANTV_VIDEO_TYPE_RAW) if _PIAOQUANTV_VIDEO_TYPE_RAW else None
@@ -222,20 +319,36 @@ def fetch_landing_videos(
         body["type"] = video_business_type
     if source is not None:
         body["source"] = source
-    headers = {
-        "content-type": "application/json",
-        "token": PIAOQUANTV_TOKEN,
-        "accept": "application/json",
-    }
+    token = get_piaoquantv_token()
+    headers = _build_auth_headers(token)
+
+    def _do_request():
+        return httpx.post(PIAOQUANTV_VIDEO_API, json=body, headers=headers, timeout=timeout)
 
     logger.info(
         "[video_recall] fetch crowd=%r source=%r page=%d size=%d",
         crowd_package, source, page_num, page_size,
     )
-    resp = httpx.post(PIAOQUANTV_VIDEO_API, json=body, headers=headers, timeout=timeout)
+    resp = _do_request()
+
+    # 401 或 Not_LOGIN(code=1000) 时刷新 token 并重试一次
+    if resp.status_code == 401:
+        logger.warning("[video_recall] 收到 401,刷新 token 并重试")
+        token = refresh_piaoquantv_token()
+        headers = _build_auth_headers(token)
+        resp = _do_request()
+
     resp.raise_for_status()
     data = resp.json()
 
+    if _is_not_login_error(data):
+        logger.warning("[video_recall] 收到 Not_LOGIN(1000),刷新 token 并重试")
+        token = refresh_piaoquantv_token()
+        headers = _build_auth_headers(token)
+        resp = _do_request()
+        resp.raise_for_status()
+        data = resp.json()
+
     if data.get("code") != 0 or not data.get("success"):
         raise RuntimeError(
             f"piaoquantv 接口失败:code={data.get('code')} msg={data.get('msg')}"