跨境出海企业 AI 应用案例：如何搭建一套定制化 SEO 监控及优化系统（SEO Copilot）

Example Keyword
Average Position: 13
过去 90 天 Impression: 15000
过去 90 天 Click: 200

COPY

页面：/example-product-guide/
关键词：Example Keyword
排名：12
CTR: 0.8%

COPY

project/
├── README.md
├── BUSINESS.md
├── DATA_SCHEMA.md
├── PLAN.md
├── TODO.md
└── CLAUDE.md

COPY

Search Console
keyword
click
impression
position
CTR

COPY

keyword: example keyword
position: 14
impression: 20000
click: 100

COPY

机会类型：首页提升机会

原因：排名已经进入第二页
曝光持续增长
CTR 偏低

建议：优化 Title
补充内容
增加内部链接

COPY

PROJECT_CONTEXT.md
DATA_SCHEMA.md
FEATURE_LIST.md
TECH_STACK.md

COPY

page
date
traffic
ranking
conversion
content_quality

COPY

position: 8-20
impression: 持续增长
query: 商业相关

COPY

关键词：Example Keyword
当前排名：13
机会：进入首页
建议：优化页面内容
增加内部链接
提升 E-E-A-T 信号

COPY

impression: 50000
click: 200
CTR: 0.4%

COPY

keyword: example keyword
URL A: position 8
URL B: position 15
URL C: position 22

COPY

关键词：Example Keyword
当前排名：14
曝光：持续增长
CTR：较低

COPY

排名：11-20
曝光：持续增长

COPY

Impression: 50000
Click: 200
CTR: 0.4%

COPY

VPS 服务器
├── Frontend Dashboard
├── Backend API
├── Database
├── SEO Analysis Worker
└── Scheduler

COPY

# SEO-Copilot 单租户版开发与部署标准作业程序 (Agent Development SOP)

> **文档性质**：面向 AI Coding Agent（如 Claude Code / Codex / Antigravity）的系统构建与工程交付标准作业规范。
> **工程目标**：从零搭建并交付一套**单人自用**的 SEO 智能闭环系统（SEO Copilot），管理自己名下固定的几个网站（1~N 个，写死在配置里，不是自助注册的多租户 SaaS）。包含"数据接入 → 7 类诊断引擎 → 关键词选品与防内耗 → 优先级待办生成 → 28 天归因回测 → LLM 周报"完整业务流水线。
> **执行约束**：严禁冗余理论阐述；严格遵循本 SOP 规定的架构边界、数据契约、目录规范与避坑防御准则。
> **溯源说明**：本 SOP 对照一套已在生产环境稳定运行数月的真实单租户系统源码逐文件校对，覆盖全部 10 个 Python 模块、DDL、部署配置。凡标注"生产验证"的内容均为真实代码行为，可直接照抄；凡标注"设计建议"的内容为简化方案，生产系统未必有对应实现。**这套系统刻意做得很"糙"——没有 ORM、没有迁移工具、没有登录系统、没有任务队列——因为它服务的读者是"一个人管少数几个自己的网站"，不是要撑起一个多用户产品。不要把它过度工程化。**

---

## 目录
1. [系统架构与前置资源清单](#一系统架构与前置资源清单)
2. [工程目录与配置文件规范](#二工程目录与配置文件规范)
3. [数据库建模与数据表契约 (PostgreSQL)](#三数据库建模与数据表契约-postgresql)
4. [核心功能模块开发指引 (M1 ~ M8)](#四核心功能模块开发指引-m1--m8)
5. [调度规范：进程内定时任务，无独立 Worker](#五调度规范进程内定时任务无独立-worker)
6. [Docker 编排与部署规范](#六docker-编排与部署规范)
7. [隐性依赖与环境准备清单（从零开始必读）](#七隐性依赖与环境准备清单从零开始必读)
8. [Agent 自动化验收与冒烟测试检查表](#八agent-自动化验收与冒烟测试检查表)
9. [交付终态标准 (Definition of Done)](#九交付终态标准-definition-of-done)

---

## 一、系统架构与前置资源清单

### 1.1 技术栈基准（生产验证）
* **运行环境**：Python 3.12（`python:3.12-slim` 基础镜像）
* **Web 框架**：FastAPI + Jinja2 服务端渲染。**应用层没有任何登录/会话系统**——鉴权完全交给反向代理（Traefik）的 HTTP Basic Auth，FastAPI 里所有路由对内网/已过反代认证的请求一律放行。这是刻意的简化：单人自用不需要一套账户体系。
* **数据访问**：**没有 ORM，没有 Alembic**。直接用 `psycopg[binary]==3.2.*`（psycopg3）+ `dict_row` 行工厂，SQL 语句手写在各模块里，参数用 `%(name)s` 占位符。每个模块顶部内嵌一段 `CREATE TABLE IF NOT EXISTS ...` 的 DDL 字符串，模块的入口函数一执行就先跑一遍 DDL，天然幂等，不需要迁移工具管理 schema 版本。
* **两层存储**：
  - **PostgreSQL 16-alpine** 是应用直接查询的"工作库"——诊断引擎、看板、待办、周报全部读写这里。
  - **Google BigQuery 是上游数据仓库，不是本系统自己搭的**：GA4 的「BigQuery 关联」和 Search Console 的「批量数据导出到 BigQuery」都是 **Google 官方原生功能**，在各自后台点几下开启，数据会自动、按日、持续写入你指定的 BigQuery 数据集，全程不需要写一行代码。本系统只负责每周从 BigQuery 里把匹配的表 **复制（不是查询）** 一份到本地 Postgres，之后所有分析都在 Postgres 里跑 SQL，不产生额外 BigQuery 查询费用。
* **调度**：APScheduler **直接跑在 FastAPI 主进程内**（`lifespan` 里启动），不是独立的 worker 容器，也没有持久化任务队列表。手动触发的操作（如"立即跑一次诊断"）用 Python 原生 `threading.Thread` 起一个后台线程执行，不落库、没有重试机制、进程重启就丢失执行状态。这在"每周批一次、失败了人工重跑"的场景下完全够用，不要因为"看起来不够可靠"就引入 Celery/Redis 或数据库任务队列——那是在为不存在的规模问题预先设计。
* **站点清单是写死在代码里的**：`db.py` 里一个 Python dict（`SITES = {"site_a": {...}, "site_b": {...}}`），新增一个站点 = 改代码 + 重新部署，不是通过网页表单创建。这是本系统与"自助 SaaS"最根本的区别，务必保持这个简单性。

### 1.2 前置依赖资源清单
Agent 在初始化项目前，必须核验以下凭据已在环境变量中声明：

| 凭证 / 资源名称 | 环境变量标识 | 必填度 | 获取途径与用途说明 |
|---|---|---|---|
| **数据库连接** | `DATABASE_URL` | 必须 | `postgresql://seocopilot:<pwd>@db:5432/seocopilot`（psycopg3 原生 URL 格式，不需要 `+psycopg` 这种 SQLAlchemy 方言前缀） |
| **Postgres 密码** | `POSTGRES_PASSWORD` | 必须 | 供 `db` 容器和 `DATABASE_URL` 共用 |
| **排名检测凭据** | `SERPER_API_KEY` | 必须 | serper.dev API Key，用于 Google 实时 SERP 抓取 |
| **LLM 周报生成** | `DEEPSEEK_API_KEY` | 必须 | DeepSeek 官方 API Key（`deepseek-chat` 模型） |
| **BigQuery 服务账号** | 不是环境变量，是文件 | 必须 | Google Cloud 服务账号 JSON 密钥，挂载到容器内固定路径 `/srv/gcp-service-account.json`，代码里通过 `GOOGLE_APPLICATION_CREDENTIALS` 环境变量指向它（`bigquery.py` 里 `os.environ.setdefault(...)` 已写死默认路径）。获取方式见第七节。 |

> [!NOTE]
> 本系统**没有** `SECRET_KEY`、`FERNET_KEY`、`GOOGLE_CLIENT_ID/SECRET`、`RESEND_API_KEY` 这些字段——没有会话签名（无登录系统）、没有 OAuth token 需要加密存储（不走自助授权，走服务账号）、没有邮件发送（周报只在网页里看，不外发）。如果你在别的资料里看到这些变量名，那是另一套更复杂的多租户 SaaS 版本的东西，不属于这份单租户 SOP。

---

## 二、工程目录与配置文件规范

### 2.1 目录结构（生产验证）
```text
seocopilot/
├── docker-compose.yml
├── .env.example                 # 需自行创建（真实 .env 不进版本库）
├── gcp-service-account.json     # Google Cloud 服务账号密钥，只读挂载进容器（不进版本库）
├── raw-data/                    # 一次性/周期性原始数据导入目录（只读挂载）
│   ├── <site-a-dir>/
│   │   ├── bigquery-data/       # GSC/GA4 从 BigQuery 导出的 CSV（历史冷启动用；日常增量走 bigquery.py 自动拉）
│   │   │   ├── <site>_gsc_url_daily_web_000000000000.csv
│   │   │   ├── <site>_gsc_query_url_daily_web_000000000000.csv
│   │   │   └── <site>_ga4_landing_page_daily_000000000000.csv
│   │   ├── google-ads-data/     # Google Ads 后台手动导出的 CSV（中文界面表头，见 M1）
│   │   ├── 竞争对手关键词/        # Semrush/Ahrefs 导出的 Excel/CSV，文件名需符合特定格式（见 M1）
│   │   └── 网站content/          # WordPress 内容导出 JSON（pages.json / posts.json，见 M1）
│   └── <site-b-dir>/ ...
└── web/
    ├── Dockerfile
    ├── requirements.txt
    └── app/
        ├── main.py               # FastAPI 入口：路由 + APScheduler 生命周期注册
        ├── db.py                 # 连接工厂 + SITES 站点清单（唯一的"配置中心"）
        ├── bigquery.py           # 每周从 BigQuery 增量同步 GSC/GA4 到本地 Postgres
        ├── importer.py           # 一次性/周期性从 raw-data/ 导入 Ads/竞对/WP内容
        ├── selection.py          # 关键词选品：Ads词/竞对缺口词/期望词三路 + 站点归属防内耗
        ├── serper.py             # 排名检测（含分页偏移量避坑逻辑）
        ├── diagnose.py           # 7 类诊断规则引擎
        ├── todos.py              # 待办生成 + 28 天归因回测
        ├── report.py             # 多站合并周报（DeepSeek 生成）
        ├── queries.py            # 看板聚合查询（周趋势/KPI/排名追踪/诊断明细）
        ├── templates/            # Jinja2 模板：base / index / site / tracking / diagnosis / todos / reports / empty
        └── static/               # 站点 logo、favicon 等静态资源
```

> [!IMPORTANT]
> 没有 `models.py`，没有 `alembic/`，没有 `auth.py`，没有 `security.py`——这些在别的（多租户）版本里存在，这套单租户系统统统不需要。**不要凭空加回来。**

### 2.2 路由表（生产验证，全部 8 个路由 + 4 个手动触发端点）

| 方法 | 路径 | 作用 |
|---|---|---|
| GET | `/health` | 健康检查，探测 `SELECT 1` |
| GET | `/` | 重定向到第一个站点的总览页 |
| GET | `/site/{site}` | 单站总览看板 (M2)，站点表还没建好时渲染 `empty.html` |
| GET | `/site/{site}/tracking` | 关键词排名追踪表 (M3) |
| GET | `/site/{site}/diagnosis` | 最近一次诊断结果 (M4) |
| GET | `/site/{site}/todos` | 待办列表 (M7) |
| POST | `/todo/{todo_id}/update` | 更新待办状态 + 填写 `feedback`，成功后 303 跳回来源页（`Referer` 头） |
| GET | `/reports` | 历史周报列表（Markdown 渲染成 HTML），**不区分站点**，一份报告覆盖所有站点 |
| POST | `/admin/generate-todos` `/admin/generate-report` `/admin/rank-check` `/admin/diagnose` | 手动立即触发对应任务，起一个后台线程执行，303 跳回来源页 |

`{site}` 路径参数必须是 `db.SITES` 字典里的 key，不存在就 404——这就是站点归属校验的全部逻辑，不需要额外的鉴权依赖。

---

## 三、数据库建模与数据表契约 (PostgreSQL)

**没有集中的 schema 文件**——每张表的 DDL 分散内嵌在"谁用它谁负责建"的模块里，`CREATE TABLE IF NOT EXISTS` + 模块入口函数执行时自动跑一遍。Agent 复刻时把下面每段 DDL 原样放进对应模块文件顶部即可，不要另建一个 `models.py` 集中管理。

### 3.1 数据接入层（`importer.py` 负责建表 + 首次导入；`bigquery.py` 之后负责增量刷新其中 3 张）
```sql
-- gsc_url_daily：GSC 明细，按日 × 落地页
CREATE TABLE IF NOT EXISTS gsc_url_daily (
  site text NOT NULL, data_date date NOT NULL, landing_page text NOT NULL,
  country text, device text,
  impressions int, clicks int, ctr double precision, avg_position double precision);
CREATE INDEX IF NOT EXISTS idx_gud_site_date ON gsc_url_daily (site, data_date);

-- gsc_query_url_daily：GSC 明细，按日 × Query × 落地页
CREATE TABLE IF NOT EXISTS gsc_query_url_daily (
  site text NOT NULL, data_date date NOT NULL, query text NOT NULL, landing_page text NOT NULL,
  country text, device text,
  impressions int, clicks int, ctr double precision, avg_position double precision);
CREATE INDEX IF NOT EXISTS idx_gqud_site_date ON gsc_query_url_daily (site, data_date);
CREATE INDEX IF NOT EXISTS idx_gqud_site_query ON gsc_query_url_daily (site, query);

-- ga4_landing_page_daily：GA4 明细，按日 × 落地页
CREATE TABLE IF NOT EXISTS ga4_landing_page_daily (
  site text NOT NULL, data_date date NOT NULL, landing_page text NOT NULL,
  country text, device text,
  sessions int, engaged_sessions int, engagement_rate double precision,
  engagement_time_seconds double precision, key_events double precision,
  conversion_rate double precision);
CREATE INDEX IF NOT EXISTS idx_ga4_site_date ON ga4_landing_page_daily (site, data_date);

-- ads_keywords：Google Ads 投放关键词表现（手动周期性导入，无 API 自动化）
CREATE TABLE IF NOT EXISTS ads_keywords (
  site text NOT NULL, keyword text NOT NULL, keyword_clean text,
  match_type text, campaign text, ad_group text, kw_status text, status text,
  final_url text, clicks int, impressions int, ctr double precision,
  conversions double precision, cost double precision);

-- competitor_keywords：竞对关键词导出（Semrush/Ahrefs）
CREATE TABLE IF NOT EXISTS competitor_keywords (
  site text NOT NULL, competitor text NOT NULL, geo text,
  keyword text NOT NULL, position int, prev_position int,
  search_volume int, kd double precision, cpc double precision,
  url text, traffic double precision, intents text, position_type text, export_date date);
CREATE INDEX IF NOT EXISTS idx_ck_site_kw ON competitor_keywords (site, keyword);

-- wp_content：WordPress 内容导出快照（title/meta/正文字数——本系统没有网页爬虫，内容信号全部来自这里）
CREATE TABLE IF NOT EXISTS wp_content (
  site text NOT NULL, wp_id bigint, object_type text, slug text, url text,
  title text, seo_title text, meta_description text,
  content_text text, word_count int, published text, modified text);
CREATE INDEX IF NOT EXISTS idx_wp_site_type ON wp_content (site, object_type);
```

### 3.2 分析与执行层（各自模块内嵌 DDL）
```sql
-- tracked_keywords：选品结果（selection.py 建表）
CREATE TABLE IF NOT EXISTS tracked_keywords (
  id serial PRIMARY KEY,
  site text NOT NULL, keyword text NOT NULL, source text NOT NULL,  -- ads | gap | wishlist
  score double precision, reason text,
  search_volume int, kd double precision,
  active boolean DEFAULT true,
  created_at timestamptz DEFAULT now(),
  UNIQUE (site, keyword, source));

-- rank_check_runs / rank_checks：排名检测批次与结果（selection.py 建表，serper.py 写入）
CREATE TABLE IF NOT EXISTS rank_check_runs (
  id serial PRIMARY KEY,
  started_at timestamptz DEFAULT now(), finished_at timestamptz,
  trigger text, keywords int DEFAULT 0, errors int DEFAULT 0);

CREATE TABLE IF NOT EXISTS rank_checks (
  id serial PRIMARY KEY,
  run_id int REFERENCES rank_check_runs (id),
  site text NOT NULL, keyword text NOT NULL,
  check_date date DEFAULT current_date,
  position int, url text);              -- position NULL = 未进前 30
CREATE INDEX IF NOT EXISTS idx_rc_site_kw ON rank_checks (site, keyword, check_date);

-- diagnostics：诊断结果（diagnose.py 建表，同站同日重跑先删后插）
CREATE TABLE IF NOT EXISTS diagnostics (
  id serial PRIMARY KEY,
  site text NOT NULL, run_date date DEFAULT current_date,
  rule text NOT NULL, severity text NOT NULL,
  subject text NOT NULL, url text, details jsonb, score double precision);
CREATE INDEX IF NOT EXISTS idx_diag_site_date ON diagnostics (site, run_date);

-- todos：闭环待办（todos.py 建表）
CREATE TABLE IF NOT EXISTS todos (
  id serial PRIMARY KEY,
  site text NOT NULL,
  rule text NOT NULL, severity text,
  title text NOT NULL, description text,
  subject text, url text,
  status text NOT NULL DEFAULT 'open',   -- open | in_progress | done | dismissed
  feedback text,
  baseline jsonb,                         -- 创建时的 28 天窗口快照
  created_at timestamptz DEFAULT now(),
  completed_at timestamptz,
  UNIQUE (site, rule, subject));

-- weekly_reports：周报持久化（report.py 建表）——注意没有 site_id，一份报告覆盖全部站点
CREATE TABLE IF NOT EXISTS weekly_reports (
  id serial PRIMARY KEY,
  week_start date NOT NULL,
  content_md text NOT NULL,
  model text, created_at timestamptz DEFAULT now());
```

> [!CAUTION]
> **归因回测同样是按需计算，不落库。** `todos.py` 的 `impact()` 函数在每次渲染待办列表时用 `completed_at` 现算前后 28 天对比，`todos` 表里没有专门的归因结果字段。

---

## 四、核心功能模块开发指引 (M1 ~ M8)

### M1 · 数据接入：两条独立的线（生产验证）

**线 A：BigQuery 自动增量同步（日常持续运行，无需人工干预）**

前提：GA4 已在后台开启「BigQuery 关联」，Search Console 已开启「批量数据导出」——这两项都是 Google 原生功能，在对应产品的管理后台点开启即可，不需要写代码（详见第七节）。开启后 Google 会自动、按日、持续把数据写进指定的 BigQuery 数据集，本系统完全不参与这个环节。

`bigquery.py` 做的事：
- `discover(client)`：遍历 GCP 项目下所有 BigQuery 数据集和表，按"数据集名或表名包含站点 key" + "表名命中 `TABLE_PATTERNS` 关键字"做模糊匹配（GA4 视图名里通常带日期分区/站点名，命名不完全规整，所以用关键字匹配而不是精确表名）：
  ```python
  TABLE_PATTERNS = {
      "gsc_url_daily": ["gsc_url_daily_web"],
      "gsc_query_url_daily": ["gsc_query_url_daily_web"],
      "ga4_landing_page_daily": ["ga4", "landing_page_daily"],
  }
  ```
- `pull(trigger)`：对每个匹配到的 (站点, 本地表) 组合，`SELECT` 出 BigQuery 里的全部列，本地先 `DELETE FROM <table> WHERE site = ?` 再用 `cur.copy(...)` 批量写入（`COPY ... FROM STDIN`，比逐行 `INSERT` 快得多）。**幂等**：每次都是"删了重灌"，不是增量 upsert，所以可以放心重跑。
- 只拉 **GSC + GA4** 三张表，Google Ads、竞对数据、WordPress 内容不走这条线（BigQuery 里没有这些，见线 B）。
- `python -m app.bigquery test` 可以单独跑连接测试，列出项目里所有数据集/表及行数，方便排查"为什么没匹配到"。

**线 B：一次性/周期性原始文件导入（人工往 `raw-data/` 里丢文件，跑一次导入脚本）**

`importer.py` 处理四类数据，各自的目录/文件名约定必须严格遵守（否则解析器识别不到）：

1. **GSC/GA4 历史 CSV**（`raw-data/<site>/bigquery-data/*.csv`）——首次冷启动时，从 BigQuery 手动导出历史数据（多个月/年）批量灌入本地库；此后交给线 A 自动增量，这里就不用再动了。文件名里带 `gsc_url_daily` / `gsc_query_url_daily` / `ga4_landing_page_daily` 关键字即可被通配符 `glob` 匹配到。
2. **Google Ads 导出**（`raw-data/<site>/google-ads-data/*.csv`）——Google Ads 后台手动导出，注意**中文界面导出的 CSV 前两行是标题/汇总行，代码里 `lines[2:]` 跳过**，表头字段用中文名映射：
   ```python
   ADS_COLMAP = {"关键字": "keyword", "匹配类型": "match_type", "广告系列": "campaign",
                 "广告组": "ad_group", "关键字状态": "kw_status", "状态": "status",
                 "最终到达网址": "final_url", "点击次数": "clicks", "展示次数": "impressions",
                 "点击率": "ctr", "转化次数": "conversions", "费用": "cost"}
   ```
   如果你的 Google Ads 账号界面语言是英文，字段名要相应改成英文表头。**没有 API 自动化，纯手动周期性导出+导入**。
3. **竞对关键词导出**（`raw-data/<site>/竞争对手关键词/*.xlsx|*.csv`，Semrush 格式）——文件名必须匹配正则 `(.+?)-organic\.Positions-(\w{2})-(\d{8})-`（Semrush 默认导出文件名就是这个格式：`<竞对域名>-organic.Positions-<地区>-<日期>-...`），从文件名里反解出竞对域名/地区/导出日期，不需要额外填表单。
4. **WordPress 内容导出**（`raw-data/<site>/网站content/*.json`）——JSON 结构预期为 `{"items": [...], "collection"/"post_type": "..."}`，每个 item 需要有 `title`/`content`(或 `content_text`)/`url`(或 `link`)/`slug` 等字段；HTML 标签会被正则剥离、按空格分词统计 `word_count`。**这份 JSON 从哪来不属于本系统范围**——你需要自己写一个小脚本或用现成 WP 插件把内容导出成这个结构，本 SOP 不含该脚本。

`importer.py` 的入口 `main(root)` 对四类数据挨个跑一遍，每张表先 `TRUNCATE` 再全量重载（不是增量），所以整个导入过程是幂等的，可以放心重复执行。

### M2 · 站点监控看板 (`queries.py`，生产验证)
- `weekly_series(site)`：按 `date_trunc('week', data_date)` 聚合 query 维度和 page 维度两套 Top10/Page2-3 词数，加曝光/点击/CTR/GA4 sessions/key_events。
- `kpis(site)`：近 28 天 vs 前 28 天环比。
- `query_table(site, bucket)`：`bucket='p1'` → position≤10，否则 10<position≤30 的明细表。
- `table_counts(site)`：六张接入表各自的行数，用于在页面上快速判断"这个站点哪类数据还没导入"。
- `tracking_data(site)`：追踪关键词 + 最新/上一次排名 + 最近 12 次历史（`LATERAL` 子查询取每个关键词最近的 `rank_checks` 记录）。
- `diagnosis_data(site)`：最近一次诊断的全部发现，按规则/严重度排序。

### M3 · 关键词排名追踪 (Serper API，生产验证，可直接照抄)
> [!CAUTION]
> **Serper 分页陷阱（硬性约束）**：请求 `page=2/3` 时 `organic[i].position` 从 1 重新计数；单页结果可能不足 10 条。必须用**累计偏移量**（历史各页真实返回条数之和）+ 页内下标算全局排名，累加值用 `len(organic)` 而不是固定 `+10`。

**本系统的额外优化**：如果同一个关键词被多个站点同时追踪，**只发一次 SERP 请求，结果里各自找各自的域名**，不重复调用 API：
```python
def _run(trigger):
    key = os.environ["SERPER_API_KEY"]
    domains = {site: cfg["domain"].replace("https://", "").replace("http://", "")
               for site, cfg in SITES.items()}
    rows = ...  # SELECT DISTINCT site, keyword FROM tracked_keywords WHERE active
    by_kw = {}
    for r in rows:
        by_kw.setdefault(r["keyword"], []).append(r["site"])

    for kw, sites in by_kw.items():
        found = {}   # site -> (position, url)
        offset = 0
        for page in (1, 2, 3):
            resp = requests.post(SERPER_URL, json={"q": kw, "gl": "us", "hl": "en", "page": page},
                                 headers={"X-API-KEY": key, "Content-Type": "application/json"},
                                 timeout=30)
            organic = resp.json().get("organic", [])
            for i, item in enumerate(organic):
                link = item.get("link", "")
                for site in sites:
                    if site not in found and domains[site] in link:
                        found[site] = (offset + i + 1, link)
            offset += len(organic)
            if len(found) == len(sites) or len(organic) == 0 or offset >= 30:
                break   # 该词命中的站点已全部找到位置，或翻到头了，提前结束
            time.sleep(0.2)
        # 未命中的站点 position/url 写 None（未进前 30）
```
- 用 `threading.Lock()`（`_lock.acquire(blocking=False)`）防止定时任务和手动触发同时跑——**不是数据库锁，是进程内锁**，单容器场景下够用；如果以后要横向扩容多个 web 副本，这个锁就失效了（当前系统就是单容器部署，不存在这个问题）。
- 每次批量检测先插一条 `rank_check_runs`，逐关键词写 `rank_checks`。

### M4 · 七类核心 SEO 诊断规则引擎（生产验证，阈值来自真实生产数据，不要改数字）
在 `diagnose.py` 实现，均以最近 **28 天**窗口 SQL 聚合为输入：

1. **`low_ctr` 高曝光低 CTR**：`impressions≥40` 且加权平均 `position≤10` 且 `CTR<1%`。`imp≥100` 记 `high`，否则 `medium`。
2. **`page2` 第二页机会**：`impressions≥30` 且加权平均 `position BETWEEN 8 AND 20`。
3. **`cannibalization` 关键词内耗**：同一 query 28 天内命中 `≥2` 个不同落地页且合计 `impressions≥30`；命中页数 `≥4` 记 `high`。
4. **`thin_page` 薄内容**（依赖 `wp_content`）：`word_count<150` 且 `impressions≥20`，`object_type` 限 `page`/`post`。
5. **`click_no_session` 有点击无会话**（依赖 GA4）：`clicks≥3` 但 `sessions==0`。
6. **`no_conversion` 高访问无转化**（依赖 GA4）：`sessions≥30` 但 `sum(key_events)==0`。
7. **`meta` Meta 标签体检**（依赖 `wp_content`）：`seo_title`（不是 `title`！本系统里页面的 SEO 标题单独存一列）长度 `>60` 记标题过长；**标题不含品牌名**（大小写、空格归一化后比对）也算一条问题，附带具体品牌名传进来：
   ```python
   BRAND = {"site_a": "Brand A", "site_b": "Brand B"}
   if brand.lower().replace(" ", "") not in seo_title.lower().replace(" ", ""):
       emit("标题缺品牌", "low", subj, url, {"cur": seo_title}, 1, f"标题追加 '| {brand}'")
   ```
   `meta_description` 长度 `<70` 记太短，`>200` 记疑似正文截断，缺失记缺描述；同一 `seo_title` 出现在 `>1` 个页面记标题重复。

> [!IMPORTANT]
> **本系统没有"GA4/爬虫数据缺失时优雅跳过"的探测逻辑**——因为站点清单是固定的三五个，接入数据是一次性导入齐全的，不存在"半路加个新站点还没导数据"的中间态。如果你的场景里站点会陆续增加、数据接入有先后顺序，自己在 `run()` 里加一个"该站点是否已有 GA4/wp_content 数据"的判断，跳过对应规则，不要空跑报错。
>
> **噪声抑制（仅 `meta` 规则）**：有曝光（`impressions≥20`）的问题逐条输出，无曝光/低曝光页面同一问题类型聚合成一条 `"[批量] 标题过长 ×N"`，附最多 6 个示例 URL。
>
> **幂等重跑**：每次先 `DELETE FROM diagnostics WHERE site=? AND run_date=current_date` 再插入。

### M5 · 关键词选品与站点防内耗（生产验证，这是本系统真正的核心亮点，务必完整实现）
`selection.py` 从**三路来源**筛出候选词，写入 `tracked_keywords`：

1. **`ads` Google Ads 已验证词**：有转化、或有点击、或曝光 `≥30` 的投放词才算"有价值"，打分 `conversions*100 + clicks*3 + impressions*0.02`。
2. **`gap` 竞对缺口词**：竞对进入过前 15 名、且（我方未追踪 或 加权平均排名 `>20`）的词。机会分：
   $$\text{score} = \frac{\text{search\_volume}}{\text{KD}+5} \times (\text{commercial/transactional 意图 ? } 1.5 : 1) \times \min(1+0.1\times(\text{命中竞对数}-1),\ 1.5)$$
3. **`wishlist` 期望词**：一份手工维护的种子词表（`raw-data/wishlist-keywords.csv`，一列 `term`），join 竞对数据拿 volume/KD 打分；已进前 10 的不再追踪；词里带数字（型号/规格）视为购买意图强，分数 `×1.3`。

**跨站防内耗（原创亮点，务必实现）**：每个候选词先跑一遍归属规则，**从上到下第一个正则命中就归给对应站点**，同一个词绝不会被两个自营站点同时追踪：
```python
SITE_RULES = [
    ("site_a", r"关键词模式1|关键词模式2|..."),   # 例如某类目专属术语
    ("site_b", r"关键词模式3|关键词模式4|..."),
    ("site_c", r"关键词模式5|..."),
]

def assign_site(kw):
    for site, pat in SITE_RULES:
        if re.search(pat, kw):
            return site
    return None   # 三个站都不命中的词直接丢弃，不追踪
```
每个来源每站上限 100 个词（按 score 降序截断）。写库用 `ON CONFLICT (site, keyword, source) DO UPDATE`（存在则刷新分数/原因，不存在则插入），本轮未再入选的词先整体 `active=false`（保留历史排名数据，不删除）。

### M6 · AI 搜索可见度 (GEO) 监控管道（设计建议，生产系统未实现）
同多租户版本 SOP 的描述：非品牌通用商业提问矩阵 + 调用带联网搜索的 LLM API + 正则比对 citations 是否命中自家域名。如果要做，建议独立一张 `geo_checks` 表（`site`, `query`, `platform`, `cited`, `citation_rank`, `checked_at`），单独一个模块，不要和现有诊断规则混在一起。

### M7 · 行动闭环与 28 天归因回测（生产验证）
1. **待办生成 (`todos.generate`)**：每站每规则从当天最新诊断里按 `severity` 再按 `score` 取 Top N（`RULE_CAPS`：`low_ctr=10, cannibalization=6, page2=10, thin_page=5, click_no_session=1, no_conversion=5, meta=8`）。`(site, rule, subject)` 已存在则 `ON CONFLICT DO NOTHING`，不重复插入、不覆盖已有状态。
2. **创建时基线快照**：优先按 query 维度查 28 天窗口（`subject` 是关键词场景），查不到退化到按落地页维度查。
3. **状态流转**：`open → in_progress → done → dismissed`，前端表单提交 `status` + `feedback`（做了什么改动），`feedback` 为空时保留原值不覆盖（`coalesce(nullif(%s,''), feedback)`）。
4. **归因回测 `impact()`**：按需现算，不落库。用 `completed_at` 切一刀，对比"完成前 28 天"与"完成后至今"两段窗口的 impressions/clicks/加权 position，返回 `days_after`。UI 层 `days_after<7` 应提示"数据积累中"。

### M8 · LLM 多站合并周报（生产验证，注意这里和"每站一份"的设计不同）
1. **上下文组装 (`report._collect`)**：**遍历全部站点**，为每个站点收集最近 8 周趋势、28 天 KPI 环比、最新两次排名检测对比（涨跌 Top 10）、最新诊断规则×严重度计数、待办状态计数+最近 5 条已完成反馈，汇总成 `{site_a: {...}, site_b: {...}, ...}` 一个大 JSON。
2. **一次 LLM 调用生成一份覆盖所有站点的周报**（不是每站单独调一次）：
   ```text
   系统 Prompt（中文站点用中文写）：
   你是一名资深 B2B 工业品独立站 SEO 顾问。基于给定的多个站点的数据 JSON，
   写一份本周 SEO 周报（markdown，无需一级标题）。结构：
   ## 总体概述 —— 所有站点本周核心变化，3-5 句，指出最重要的信号
   ## 各站 SWOT —— 每站一个三级标题，S/W/O/T 各 1-3 条，必须引用具体数据
   ## 本周行动优先级 —— 全局 Top 5，按预期影响排序，每条注明站点/动作/预期效果
   要求：只依据给定数据，不编造；数字保留原值；语言精炼直接，避免空话。
   ```
   DeepSeek `deepseek-chat`，`temperature=0.4, max_tokens=3000`。
3. **持久化，无邮件发送**：`content_md` 存 `weekly_reports.content_md`，`week_start` 记当天日期。**这个系统不发邮件**——周报只在 `/reports` 页面用 `markdown` 库渲染成 HTML 查看。如果你想要邮件推送，自己接一个 Resend/SMTP，属于本 SOP 范围外的增强。

---

## 五、调度规范：进程内定时任务，无独立 Worker

### 5.1 定时流水线（生产验证，`main.py` 的 `lifespan` 里注册）
```python
@asynccontextmanager
async def lifespan(app):
    sched = BackgroundScheduler(timezone="UTC")
    # 周一流水线（UTC）：02:00 拉 BigQuery → 03:00 查排名 → 05:00 诊断 → 05:30 生成 to-do → 06:00 LLM 周报
    sched.add_job(lambda: bigquery.pull("weekly"),  CronTrigger(day_of_week="mon", hour=2), id="weekly_bq_pull")
    sched.add_job(lambda: serper.run_check("weekly"), CronTrigger(day_of_week="mon", hour=3), id="weekly_rank_check")
    sched.add_job(lambda: diagnose.run("weekly"),   CronTrigger(day_of_week="mon", hour=5), id="weekly_diagnose")
    sched.add_job(lambda: todos.generate("weekly"), CronTrigger(day_of_week="mon", hour=5, minute=30), id="weekly_todos")
    sched.add_job(lambda: report.generate("weekly"), CronTrigger(day_of_week="mon", hour=6), id="weekly_report")
    sched.start()
    yield
    sched.shutdown(wait=False)
```
**全部串行、全部周一跑**，没有每日增量刷新——因为 BigQuery 那边 Google 已经每天在自动更新底层数据，本系统每周批一次拉过来就够用（周报也是周维度产出，没必要日更）。如果你的使用场景需要更高频率的数据新鲜度，把 `bigquery.pull` 单独拆一条每日 cron 出来即可，其余诊断/选品/周报维持周频没问题。

### 5.2 手动触发：后台线程，不是任务队列
```python
@app.post("/admin/diagnose")
def trigger_diagnose(request: Request):
    threading.Thread(target=diagnose.run, args=("manual",), daemon=True).start()
    return RedirectResponse(request.headers.get("referer") or "/", status_code=303)
```
点击按钮 → 起一个 daemon 线程跑完就完事 → 页面立刻 303 跳转（**不等任务跑完**，用户需要手动刷新页面看结果）。**没有执行状态追踪、没有失败重试、进程重启会直接丢失正在跑的线程**。这是刻意的权衡：一个人管几个网站，一周批一次，手动点一下立即触发，出了问题看容器日志、重新点一下按钮就行——为这个规模引入持久化任务队列纯属过度设计。

### 5.3 幂等性保障
- `gsc_url_daily`/`gsc_query_url_daily`/`ga4_landing_page_daily`：`bigquery.pull()` 每次先 `DELETE WHERE site=?` 再 `COPY` 写入。
- `diagnostics`：先 `DELETE WHERE site=? AND run_date=current_date` 再插入。
- `todos`：靠 `UNIQUE (site, rule, subject)` + `ON CONFLICT DO NOTHING` 天然防重复。
- `tracked_keywords`：`ON CONFLICT (site, keyword, source) DO UPDATE`，本轮未再入选的整体 `active=false`。
- `importer.py` 四类原始数据：每次先 `TRUNCATE` 对应表再全量重载。

---

## 六、Docker 编排与部署规范

### 6.1 `docker-compose.yml`（生产验证）
```yaml
services:
  web:
    build: ./web
    container_name: seocopilot-web
    restart: unless-stopped
    env_file: .env
    volumes:
      - ./raw-data:/srv/raw-data:ro
      - ./gcp-service-account.json:/srv/gcp-service-account.json:ro
    depends_on:
      db:
        condition: service_healthy
    labels:
      - "traefik.enable=true"
      - "traefik.http.routers.seocopilot.entrypoints=websecure"
      - "traefik.http.routers.seocopilot.rule=Host(`seocopilot.yourdomain.com`)"
      - "traefik.http.routers.seocopilot.tls.certresolver=letsencrypt"
      - "traefik.http.routers.seocopilot.middlewares=seocopilot-auth"
      - "traefik.http.services.seocopilot.loadbalancer.server.port=8000"
      # htpasswd 生成的 Basic Auth 凭证，$ 要转义成 $$（compose 变量插值规则）：
      - "traefik.http.middlewares.seocopilot-auth.basicauth.users=<user>:<htpasswd-hash>"

  # 一次性任务专用容器：不挂 Traefik 标签，避免被反代误纳入负载均衡
  # 用法：docker compose run --rm jobs python -m app.importer
  #       docker compose run --rm jobs python -m app.selection
  jobs:
    build: ./web
    profiles: ["tools"]
    env_file: .env
    volumes:
      - ./raw-data:/srv/raw-data:ro
      - ./gcp-service-account.json:/srv/gcp-service-account.json:ro
    depends_on:
      db:
        condition: service_healthy

  db:
    image: postgres:16-alpine
    container_name: seocopilot-db
    restart: unless-stopped
    environment:
      POSTGRES_DB: seocopilot
      POSTGRES_USER: seocopilot
      POSTGRES_PASSWORD: ${POSTGRES_PASSWORD}
    volumes:
      - pgdata:/var/lib/postgresql/data
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U seocopilot -d seocopilot"]
      interval: 5s
      timeout: 3s
      retries: 10

volumes:
  pgdata:
```
**没有独立 worker 容器**——第五节讲过，调度和手动触发都在 `web` 进程里完成，不需要第二个跑业务逻辑的容器。`jobs` 这个 profile 容器只是为了不常跑的一次性命令（导入原始文件、重新选品）单独开一个 shell，平时不启动（`profiles: ["tools"]` 意味着 `docker compose up` 默认不会拉起它）。

**Basic Auth 生成方式**：`htpasswd -nb <用户名> <密码>` 生成一行 `user:$apr1$...` 格式的哈希，填进上面的 label，注意 compose 文件里 `$` 要写成 `$$` 转义。不想装 htpasswd 也可以用在线工具或 `openssl passwd -apr1`。

### 6.2 `Dockerfile`
```dockerfile
FROM python:3.12-slim
WORKDIR /srv
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY app ./app
EXPOSE 8000
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]
```
比多租户版本简单得多：没有 `alembic upgrade head` 前置步骤（没有迁移工具），没有健康检查探针（Traefik 层面探测就够）。

### 6.3 `requirements.txt`（生产验证）
```text
fastapi==0.116.*
uvicorn[standard]==0.35.*
jinja2==3.1.*
psycopg[binary]==3.2.*
pandas==2.2.*
openpyxl==3.1.*
requests==2.32.*
apscheduler==3.10.*
markdown==3.7
python-multipart==0.0.*
google-cloud-bigquery==3.*
```
**没有**：SQLAlchemy、Alembic、argon2-cffi、cryptography、google-auth/google-auth-oauthlib（没有 OAuth 自助连接流程）、beautifulsoup4（没有网页爬虫）。`pandas`/`openpyxl` 是为了 `importer.py` 读竞对 Excel 导出文件；`google-cloud-bigquery` 是官方 BigQuery Python 客户端，只在 `bigquery.py` 里用到。

---

## 七、隐性依赖与环境准备清单（从零开始必读）

这套系统最容易卡住陌生人的地方，不是写代码，是 **Google 那边几个后台开关**，本 SOP 之前的版本完全没提到过，务必照做：

1. **GA4 → BigQuery 关联（Google 原生功能，免费额度内足够个人用量）**：GA4 后台 → 管理 → 产品链接 → BigQuery 链接 → 新建链接，选一个 GCP 项目，勾选每日导出（免费版有次数限制，够用）。开启后通常次日开始在指定数据集里出现每日分区表。
2. **Search Console → 批量数据导出（同样是 Google 原生功能）**：Search Console 设置 → 批量数据导出 → 选同一个（或另一个）GCP 项目和数据集。**前提是你对该 GCP 项目有 Owner/Editor 权限**，且该 Search Console 资源已验证所有权。开启后同样是次日开始按天写入。
3. **GCP 服务账号**：在启用了 BigQuery API 的 GCP 项目里创建一个服务账号，赋予 **BigQuery Data Viewer**（读表）+ **BigQuery Job User**（跑查询）角色即可，不需要更高权限；下载 JSON 密钥文件，挂载进容器路径 `/srv/gcp-service-account.json`（对应 compose 里的 volume 映射），**这个文件不能进版本库**。
4. **首次冷启动数据从哪来**：BigQuery 关联开启后只会有"从开启那天起"的数据，没有历史。如果需要更早的历史数据，去 BigQuery 控制台手动跑 SQL 导出历史范围的 CSV，放进 `raw-data/<site>/bigquery-data/`，跑一次 `importer.py` 做冷启动灌入；之后交给 `bigquery.pull()` 自动增量。
5. **Google Ads 数据是纯手动的**：没有 API 自动拉取，需要定期去 Google Ads 后台手动导出关键词报表 CSV，放进 `raw-data/<site>/google-ads-data/`，跑一次 `importer.py`（这一步没有自动化，是本系统唯一"需要人隔三差五手动操作"的环节，如果你想自动化，接 Google Ads API 是合理的后续改进方向，不在本 SOP 范围内）。
6. **竞对数据**：Semrush/Ahrefs 之类工具手动导出 Organic Positions 报告（Excel/CSV），文件名保持工具默认导出格式（形如 `域名-organic.Positions-地区-日期-...`），扔进 `raw-data/<site>/竞争对手关键词/`。
7. **WordPress 内容导出**：需要一份结构化 JSON（`{"items": [...]}`），本 SOP 不含生成这份 JSON 的具体脚本/插件——如果你用 WordPress REST API，写一个几十行的脚本批量拉 `posts`/`pages`/`products` 并按上文 3.1 节 `wp_content` 表要求的字段整理输出即可。
8. **Traefik + Basic Auth**：如果你已经有 Traefik 在跑，加一个 router + middleware 标签就行；如果没有反代，本地部署直接用 `docker compose up`，`web` 容器加一行 `ports: ["8000:8000"]` 本机访问即可，个人使用没必要强求 HTTPS。
9. **Serper.dev / DeepSeek 账号**：注册拿 API Key，跟前面版本 SOP 一致。

---

## 八、Agent 自动化验收与冒烟测试检查表

> 生产项目没有 `tests/` 目录，没有 pytest。以下是手工冒烟验证序列：

```bash
# 1. 健康检查 + 数据库连通性
curl -s http://localhost:8000/health

# 2. BigQuery 连接测试（列出项目下所有数据集/表，确认能匹配到你的站点）
docker compose run --rm jobs python -m app.bigquery test

# 3. 首次冷启动导入（需要 raw-data/ 已按第七节目录约定准备好文件）
docker compose run --rm jobs python -m app.importer /srv/raw-data

# 4. 关键词选品（依赖第 3 步的 ads_keywords / competitor_keywords 已导入）
docker compose run --rm jobs python -m app.selection /srv/raw-data

# 5. 排名检测（依赖第 4 步已生成 tracked_keywords）
docker compose run --rm jobs python -c "from app.serper import run_check; print(run_check('manual'))"

# 6. 诊断引擎（依赖第 3 步 GSC 数据 + wp_content 已导入）
docker compose run --rm jobs python -c "from app.diagnose import run; run('manual')"

# 7. 待办生成
docker compose run --rm jobs python -c "from app.todos import generate; print(generate('manual'))"

# 8. 周报生成（依赖前面全部步骤都有数据）
docker compose run --rm jobs python -c "from app.report import generate; print(len(generate('manual')))"

# 9. 页面走查：浏览器打开 http://localhost:8000/site/<你的站点key> 确认看板/排名/诊断/待办四个页面都有数据
```
按 3→4→5→6→7→8 的顺序跑一遍，任何一步空跑（0 条结果）说明上一步的数据没导对，回头检查 `raw-data/` 目录结构和文件名是否符合第七节的约定。

---

## 九、交付终态标准 (Definition of Done)

1. **BigQuery 自动同步跑通**：GA4/GSC 的 BigQuery 关联已在 Google 后台开启，`bigquery.py test` 能列出并匹配到全部站点的三张表。
2. **原始数据冷启动完成**：`importer.py` 跑完后 `table_counts()` 六张表都有非零行数。
3. **选品与防内耗生效**：`tracked_keywords` 里每个关键词只归属一个站点（`SITE_RULES` 正则命中且互斥）。
4. **Serper 翻页避坑已落地**：累计偏移量算法，且同词多站去重成一次请求。
5. **7 类诊断规则全部跑通**，`meta` 规则含品牌后缀检查。
6. **待办闭环生效**：诊断 → 待办生成 → 用户标记完成 → `impact()` 能算出前后 28 天对比。
7. **周报可生成**：多站合并成一份 Markdown，存库可在 `/reports` 查看（不要求邮件发送）。
8. **调度确认**：APScheduler 在 `web` 进程内按周一流水线顺序触发，手动触发端点能正常起后台线程执行且不阻塞请求。
9. **部署隔离**：`web` 挂 Traefik + Basic Auth 标签，`jobs` 容器不挂任何反代标签，仅用于一次性命令。

---

## 附：免责声明

本 SOP 基于一套真实运行的单租户生产系统源码逐文件校对整理，技术选型（无 ORM、无迁移工具、无登录系统、无持久化任务队列）都是服务于"一个人管理少数几个自己的网站"这个明确场景的刻意简化，不是能力不足。如果你的场景膨胀到"需要自助多用户注册、按需接入任意数量站点、需要审计与权限体系"，那已经超出本 SOP 的设计边界，需要另一套架构（引入 ORM/迁移工具/OAuth 自助授权/持久化任务队列/多租户数据隔离），不要在这套单租户系统上硬堆这些能力。使用本 SOP 的 Agent 在实现细节上遇到本文未覆盖的情况，应优先保持"简单、幂等、单人可维护"的原则，而不是引入不必要的抽象。

COPY