Enterprise AI in practice: how to build a custom SEO monitoring and optimization system (SEO Copilot)

Example Keyword
Average Position: 13
Impressions (last 90 days): 15000
Clicks (last 90 days): 200

COPY

Page: /example-product-guide/
Keyword: Example Keyword
Position: 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

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: trending up
query: commercially relevant

COPY

Keyword: Example Keyword
Current position: 13
Opportunity: page one
Recommended: expand the page content
add internal links
strengthen E-E-A-T signals

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

VPS server
├── Frontend Dashboard
├── Backend API
├── Database
├── SEO Analysis Worker
└── Scheduler

COPY

# SEO Copilot Single-Tenant Development and Deployment SOP (Agent Development SOP)

> **Document type**: A system-building and engineering-delivery standard operating procedure for AI coding agents such as Claude Code, Codex and Antigravity.
> **Engineering goal**: Build and deliver, from scratch, a single-user SEO intelligence loop for a small fixed set of websites (1–N sites hard-coded in configuration, not a self-service multi-tenant SaaS). The complete pipeline is: data ingestion → seven diagnosis engines → keyword selection and cannibalization prevention → prioritized to-do generation → 28-day attribution backtesting → LLM weekly report.
> **Execution constraint**: Avoid redundant theory. Strictly follow the architecture boundaries, data contracts, directory conventions and defensive rules in this SOP.
> **Source note**: This SOP was checked file by file against a real single-tenant system that has been stable in production for months. It covers all 10 Python modules, DDL and deployment configuration. Items marked “production-validated” describe real code behavior and can be copied directly; items marked “design suggestion” simplify areas that the production system may not implement. **This system is intentionally rough: no ORM, no migration tool, no login system and no task queue.** It serves one person managing a small number of their own websites, not a multi-user product. Do not over-engineer it.

---

## Contents
1. [System architecture and prerequisite resources](#1-system-architecture-and-prerequisite-resources)
2. [Project directory and configuration conventions](#2-project-directory-and-configuration-conventions)
3. [Database model and table contracts (PostgreSQL)](#3-database-model-and-table-contracts-postgresql)
4. [Core module implementation guide (M1–M8)](#4-core-module-implementation-guide-m1m8)
5. [Scheduling: in-process jobs, no independent worker](#5-scheduling-in-process-jobs-no-independent-worker)
6. [Docker orchestration and deployment](#6-docker-orchestration-and-deployment)
7. [Hidden dependencies and environment preparation](#7-hidden-dependencies-and-environment-preparation)
8. [Agent automation acceptance and smoke-test checklist](#8-agent-automation-acceptance-and-smoke-test-checklist)
9. [Definition of Done](#9-definition-of-done)

---

## 1. System architecture and prerequisite resources

### 1.1 Baseline technology stack (production-validated)

* **Runtime**: Python 3.12, based on the `python:3.12-slim` image.
* **Web framework**: FastAPI + Jinja2 server-side rendering. **The application layer has no login or session system**; authentication is completely delegated to HTTP Basic Auth at the reverse proxy (Traefik). Every FastAPI route allows requests that have already passed the proxy. This is intentional simplification: a personal system does not need an account system.
* **Data access**: **No ORM and no Alembic**. Use `psycopg[binary]==3.2.*` (psycopg3) with the `dict_row` row factory. Write SQL directly in each module and use `%(name)s` placeholders. Each module embeds its own `CREATE TABLE IF NOT EXISTS ...` DDL string; its entry function runs the DDL first, so schema creation is naturally idempotent and needs no migration tool.
* **Two storage layers**:
  - **PostgreSQL 16-alpine** is the working database queried directly by the application. Diagnosis engines, the dashboard, to-dos and weekly reports all read and write here.
  - **Google BigQuery is the upstream warehouse, not something this system builds**. GA4’s BigQuery linking and Search Console’s bulk data export are **native Google features** enabled in their respective admin screens. Google then writes data daily to the BigQuery dataset you specify; no code is required for that step. The system copies (does not query) the matching tables from BigQuery into local Postgres once a week. All analysis then runs as SQL against Postgres, avoiding extra BigQuery query charges.
* **Scheduling**: APScheduler **runs inside the FastAPI main process** and starts in `lifespan`. There is no separate worker container and no persistent task queue. Manual triggers, such as “run diagnosis now”, use a Python `threading.Thread` to execute in the background. There is no persistence or retry; a process restart loses the execution state. This is sufficient for a weekly batch that one person can rerun manually. Do not introduce Celery, Redis or a database queue just because this looks less elaborate.
* **Sites are hard-coded**: `db.py` contains a Python dict such as `SITES = {"site_a": {...}, "site_b": {...}}`. Adding a site means changing code and redeploying, not creating it through a web form. This is the fundamental difference between this system and self-service SaaS. Keep it simple.

### 1.2 Prerequisite resource checklist

Before initializing the project, the Agent must verify that the following credentials and resources are available:

| Credential / resource | Environment variable or location | Required | How to obtain it and what it is for |
|---|---|---:|---|
| **Database connection** | `DATABASE_URL` | Yes | `postgresql://seocopilot:<pwd>@db:5432/seocopilot` (native psycopg3 URL; do not add the SQLAlchemy `+psycopg` dialect prefix) |
| **Postgres password** | `POSTGRES_PASSWORD` | Yes | Shared by the `db` container and `DATABASE_URL` |
| **Rank-check credential** | `SERPER_API_KEY` | Yes | Serper.dev API key for live Google SERP checks |
| **LLM weekly report** | `DEEPSEEK_API_KEY` | Yes | Official DeepSeek API key, using the `deepseek-chat` model |
| **BigQuery service account** | A file, not an environment variable | Yes | Google Cloud service-account JSON mounted at `/srv/gcp-service-account.json`. `GOOGLE_APPLICATION_CREDENTIALS` points to it; `bigquery.py` already defaults to this path. See Section 7. |

> [!NOTE]
> This system does **not** use `SECRET_KEY`, `FERNET_KEY`, `GOOGLE_CLIENT_ID/SECRET` or `RESEND_API_KEY`. There are no sessions to sign, no OAuth token to encrypt and no outbound email. If you see those variables in another document, they belong to a more complex multi-tenant SaaS version, not this single-tenant SOP.

---

## 2. Project directory and configuration conventions

### 2.1 Directory structure (production-validated)

```text
seocopilot/
├── docker-compose.yml
├── .env.example                 # Create this yourself; never commit the real .env
├── gcp-service-account.json     # Google Cloud service-account key; read-only mount, never commit
├── raw-data/                    # One-time / periodic raw data import directory, read-only mounted
│   ├── <site-a-dir>/
│   │   ├── bigquery-data/       # GSC/GA4 CSV exports for historical cold start
│   │   │   ├── <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 export; Chinese headers are mapped in M1
│   │   ├── 竞争对手关键词/        # Semrush/Ahrefs Excel/CSV exports; filename format is required
│   │   └── 网站content/          # WordPress JSON exports: pages.json / posts.json
│   └── <site-b-dir>/ ...
└── web/
    ├── Dockerfile
    ├── requirements.txt
    └── app/
        ├── main.py               # FastAPI entry point: routes + APScheduler lifecycle
        ├── db.py                 # Connection factory + SITES list; the only configuration center
        ├── bigquery.py           # Weekly incremental GSC/GA4 sync from BigQuery to local Postgres
        ├── importer.py           # One-time/periodic Ads, competitor and WordPress imports
        ├── selection.py          # Keyword selection: ads/gap/wishlist + site ownership
        ├── serper.py             # Rank checks, including pagination offset handling
        ├── diagnose.py           # Seven SEO diagnosis rules
        ├── todos.py              # To-do generation + 28-day attribution backtest
        ├── report.py             # Combined multi-site weekly report generated by DeepSeek
        ├── queries.py            # Dashboard aggregation queries
        ├── templates/            # Jinja2: base / index / site / tracking / diagnosis / todos / reports / empty
        └── static/               # Site logo, favicon and other static assets
```

> [!IMPORTANT]
> There is no `models.py`, `alembic/`, `auth.py` or `security.py`. Those belong to another multi-tenant version. **Do not add them back by assumption.**

### 2.2 Route table (production-validated: 8 routes + 4 manual triggers)

| Method | Path | Purpose |
|---|---|---|
| GET | `/health` | Health check; probes `SELECT 1` |
| GET | `/` | Redirects to the overview page of the first site |
| GET | `/site/{site}` | Single-site overview dashboard; renders `empty.html` if the site tables are not ready |
| GET | `/site/{site}/tracking` | Keyword rank-tracking table |
| GET | `/site/{site}/diagnosis` | Latest diagnosis results |
| GET | `/site/{site}/todos` | To-do list |
| POST | `/todo/{todo_id}/update` | Updates to-do status and `feedback`, then 303-redirects to the referrer |
| GET | `/reports` | Historical weekly reports, rendered from Markdown; **not separated by site**, one report covers all sites |
| POST | `/admin/generate-todos` `/admin/generate-report` `/admin/rank-check` `/admin/diagnose` | Manually triggers a task in a background thread and 303-redirects to the referrer |

The `{site}` path parameter must be a key in `db.SITES`; unknown keys return 404. That is the entire site-ownership check. No additional authentication dependency is required.

---

## 3. Database model and table contracts (PostgreSQL)

**There is no central schema file.** Each table’s DDL lives in the module that owns it. Use `CREATE TABLE IF NOT EXISTS` and run it whenever the module entry function starts. When rebuilding the system, put each DDL block in the top of its corresponding module instead of creating a centralized `models.py`.

### 3.1 Ingestion layer (`importer.py` creates/imports; `bigquery.py` refreshes three tables)

```sql
-- gsc_url_daily: GSC detail by day and landing page
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 detail by day, query and landing page
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 detail by day and landing page
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 keyword performance, manually imported
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 competitor keyword export
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 content snapshot; this system has no page crawler
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 Analysis and execution layer (DDL embedded in each module)

```sql
-- tracked_keywords: selected terms, created by 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: rank-check batches and results
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);              -- NULL position = not in the top 30
CREATE INDEX IF NOT EXISTS idx_rc_site_kw ON rank_checks (site, keyword, check_date);

-- diagnostics: diagnosis results; delete and reinsert on same-site same-day rerun
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: closed-loop action items
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-day snapshot at creation
  created_at timestamptz DEFAULT now(),
  completed_at timestamptz,
  UNIQUE (site, rule, subject));

-- weekly_reports: one combined report for all sites
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]
> **Attribution backtesting is also calculated on demand and not persisted.** `todos.py` calculates the before/after 28-day comparison from `completed_at` whenever the to-do list is rendered. The `todos` table has no dedicated attribution-result field.

---

## 4. Core module implementation guide (M1–M8)

### M1 · Data ingestion: two independent paths (production-validated)

**Path A: automatic BigQuery incremental sync (continuous, no manual work)**

Prerequisite: GA4 BigQuery linking and Search Console bulk data export are enabled in Google’s admin screens. Both are native Google features and require no code. After activation, Google writes data daily to the selected BigQuery dataset; this system does not participate in that step.

`bigquery.py` does the following:

- `discover(client)`: walks all BigQuery datasets and tables in the GCP project and fuzzy-matches a site key in the dataset/table name plus a keyword from `TABLE_PATTERNS`. GA4 view names often contain partitions, dates or site names, so keyword matching is safer than exact table names:
  ```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)`: for every matched (site, local-table) pair, selects all BigQuery columns, runs `DELETE FROM <table> WHERE site = ?` locally and bulk loads with `cur.copy(...)` (`COPY ... FROM STDIN`). This is **idempotent**: it deletes and reloads rather than upserting, so reruns are safe.
- Pulls only the three GSC/GA4 tables. Google Ads, competitor data and WordPress content use Path B because they do not exist in BigQuery.
- `python -m app.bigquery test` runs a standalone connection test and lists every dataset/table and row count, making “why was nothing matched?” easy to diagnose.

**Path B: one-time/periodic raw-file imports (a person drops files into `raw-data/`)**

`importer.py` handles four data types. Directory and filename conventions must be followed exactly or the parser will not recognize the files:

1. **Historical GSC/GA4 CSV** (`raw-data/<site>/bigquery-data/*.csv`): export the historical range from BigQuery for the initial cold start and load it locally. After that, Path A handles incremental refreshes. Filenames only need to contain `gsc_url_daily`, `gsc_query_url_daily` or `ga4_landing_page_daily` for the glob to find them.
2. **Google Ads export** (`raw-data/<site>/google-ads-data/*.csv`): export from Google Ads. The Chinese-interface CSV has two title/summary rows, so the importer skips `lines[2:]`. The field mapping is:
   ```python
   ADS_COLMAP = {"关键字": "keyword", "匹配类型": "match_type", "广告系列": "campaign",
                 "广告组": "ad_group", "关键字状态": "kw_status", "状态": "status",
                 "最终到达网址": "final_url", "点击次数": "clicks", "展示次数": "impressions",
                 "点击率": "ctr", "转化次数": "conversions", "费用": "cost"}
   ```
   If the Google Ads account uses an English interface, change the source headers accordingly. **There is no API automation; this is a periodic manual export/import.**
3. **Competitor keyword export** (`raw-data/<site>/竞争对手关键词/*.xlsx|*.csv`, Semrush format): filenames must match `(.+?)-organic\\.Positions-(\\w{2})-(\\d{8})-`. The parser extracts competitor domain, region and export date from the default filename, so no extra form is needed.
4. **WordPress content export** (`raw-data/<site>/网站content/*.json`): expected shape is `{"items": [...], "collection"/"post_type": "..." }`. Each item needs `title`, `content` or `content_text`, `url` or `link`, and `slug`. HTML is stripped with a regex and tokenized to calculate `word_count`. **Generating this JSON is outside this system**; write a small script or use a WordPress plugin. This SOP does not include that exporter.

`importer.py` entry point `main(root)` runs all four imports. Each table is truncated and fully reloaded, not incrementally updated, so the process is idempotent and safe to repeat.

### M2 · Site-monitoring dashboard (`queries.py`, production-validated)

- `weekly_series(site)`: aggregates query-level and page-level Top 10/Page 2–3 keyword counts by `date_trunc('week', data_date)`, plus impressions, clicks, CTR, GA4 sessions and key events.
- `kpis(site)`: compares the most recent 28 days with the preceding 28 days.
- `query_table(site, bucket)`: `bucket='p1'` means `position≤10`; otherwise it returns the detail table for `10<position≤30`.
- `table_counts(site)`: row counts for six ingestion tables, so the dashboard can quickly show which data type has not been imported.
- `tracking_data(site)`: tracked keywords, latest/previous ranking and the most recent 12 checks, using a `LATERAL` subquery.
- `diagnosis_data(site)`: all findings from the latest diagnosis, sorted by rule and severity.

### M3 · Keyword rank tracking (Serper API, production-validated)

> [!CAUTION]
> **Serper pagination trap (hard constraint)**: when requesting `page=2/3`, `organic[i].position` starts counting from 1 again, and a page can contain fewer than ten results. Use the cumulative offset (the real number of results returned by earlier pages) plus the index within the current page. Add `len(organic)`, never a fixed `+10`.

**Additional optimization in this system**: if the same keyword is tracked by multiple sites, send only one SERP request and find each site’s domain in that response:

```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   # all sites found, or no more results, or past the top 30
            time.sleep(0.2)
        # Write position/url as None for sites not found in the top 30
```

- Use `threading.Lock()` (`_lock.acquire(blocking=False)`) to prevent a scheduled job and a manual trigger from running at the same time. This is an in-process lock, not a database lock; it is enough for the current single-container deployment. If the system later scales horizontally, it will no longer be sufficient.
- Insert one `rank_check_runs` row per batch and write one `rank_checks` row per keyword.

### M4 · Seven core SEO diagnosis rules (production-validated; do not change the thresholds)

Implement these in `diagnose.py`, using SQL aggregates over the latest **28-day** window:

1. **`low_ctr` high impressions, low CTR**: `impressions≥40`, weighted average `position≤10` and `CTR<1%`. `imp≥100` is `high`; otherwise `medium`.
2. **`page2` second-page opportunity**: `impressions≥30` and weighted average `position BETWEEN 8 AND 20`.
3. **`cannibalization` keyword cannibalization**: the same query reaches `≥2` different landing pages within 28 days and has at least 30 combined impressions; `≥4` pages is `high`.
4. **`thin_page` thin content** (depends on `wp_content`): `word_count<150` and `impressions≥20`; limit `object_type` to `page`/ `post`.
5. **`click_no_session` clicks without sessions** (depends on GA4): `clicks≥3` but `sessions==0`.
6. **`no_conversion` high traffic without conversion** (depends on GA4): `sessions≥30` but `sum(key_events)==0`.
7. **`meta` meta-tag audit** (depends on `wp_content`): `seo_title` (not `title`; page SEO titles are stored separately) longer than 60 is too long; a title that lacks the brand name is also an issue after normalizing case and spaces:
   ```python
   BRAND = {"site_a": "Brand A", "site_b": "Brand B"}
   if brand.lower().replace(" ", "") not in seo_title.lower().replace(" ", ""):
       emit("Missing brand", "low", subj, url, {"cur": seo_title}, 1, f"Append '| {brand}' to the title")
   ```
   `meta_description` shorter than 70 is too short, longer than 200 is probably truncated body copy, and missing is a missing description. If the same `seo_title` appears on more than one page, report a duplicate title.

> [!IMPORTANT]
> **The system does not gracefully skip missing GA4 or crawler data.** The site list is a fixed handful and ingestion is expected to be complete before diagnosis. If your sites are added gradually or sources arrive at different times, add a per-site data-existence check in `run()` and skip rules that lack their required data.
>
> **Noise suppression for `meta` only**: emit every issue on pages with impressions (`impressions≥20`). Aggregate no/low-impression pages by issue type, such as `[batch] title too long ×N`, with up to six example URLs.
>
> **Idempotent reruns**: first `DELETE FROM diagnostics WHERE site=? AND run_date=current_date`, then insert the new results.

### M5 · Keyword selection and cross-site cannibalization prevention (production-validated; implement fully)

`selection.py` selects candidates from three sources and writes them into `tracked_keywords`:

1. **`ads` verified Google Ads terms**: a paid term is valuable if it has conversions, clicks or at least 30 impressions. Score it as `conversions*100 + clicks*3 + impressions*0.02`.
2. **`gap` competitor-gap terms**: a competitor has ranked in the top 15 and either we do not track the term or our weighted average position is `>20`. Score:
   $$	ext{score} = rac{	ext{search_volume}}{	ext{KD}+5} 	imes (	ext{commercial/transactional intent ? } 1.5 : 1) 	imes min(1+0.1	imes(	ext{competitor count}-1), 1.5)$$
3. **`wishlist` target terms**: a manually maintained seed list (`raw-data/wishlist-keywords.csv`, one `term` column). Join competitor data to obtain volume/KD and score it. Do not track terms already in the top 10. A term containing digits, such as a model number or specification, signals purchase intent and receives `×1.3`.

**Cross-site ownership prevention (original highlight; implement it)**: run each candidate through ownership rules; the first regex match from top to bottom owns the term, so a term is never tracked by two owned sites:

```python
SITE_RULES = [
    ("site_a", r"pattern1|pattern2|..."),   # category-specific terms
    ("site_b", r"pattern3|pattern4|..."),
    ("site_c", r"pattern5|..."),
]

def assign_site(kw):
    for site, pat in SITE_RULES:
        if re.search(pat, kw):
            return site
    return None   # discard terms that match none of the owned sites
```

Limit each source/site to 100 terms, sorted by score. Write with `ON CONFLICT (site, keyword, source) DO UPDATE` to refresh score/reason. Before the current run, set all existing terms to `active=false`; keep their historical ranks rather than deleting them.

### M6 · AI search visibility (GEO) monitoring (design suggestion; not implemented in production)

Follow the multi-tenant SOP idea: use a non-branded commercial-question matrix, call an LLM API with web search, and compare citations against owned domains with regex. If implemented, use an independent `geo_checks` table (`site`, `query`, `platform`, `cited`, `citation_rank`, `checked_at`) and a separate module. Do not mix GEO checks into the existing diagnosis rules.

### M7 · Action loop and 28-day attribution backtest (production-validated)

1. **To-do generation (`todos.generate`)**: for each site and rule, take the Top N from the day’s latest diagnoses, sorted by `severity` then `score`. `RULE_CAPS`: `low_ctr=10, cannibalization=6, page2=10, thin_page=5, click_no_session=1, no_conversion=5, meta=8`. If `(site, rule, subject)` exists, use `ON CONFLICT DO NOTHING`; do not duplicate or overwrite an existing status.
2. **Creation-time baseline**: query a 28-day window by query first when `subject` is a keyword; fall back to landing-page dimensions if no query result exists.
3. **Status flow**: `open → in_progress → done → dismissed`. The form submits `status` and `feedback`; if `feedback` is empty, preserve the existing value with `coalesce(nullif(%s,''), feedback)`.
4. **Attribution `impact()`**: calculate on demand, never persist. Use `completed_at` as the split point and compare impressions, clicks and weighted position for the 28 days before completion with the period from completion until now. Return `days_after`; the UI should show “data accumulating” when `days_after<7`.

### M8 · Combined multi-site LLM weekly report (production-validated)

1. **Assemble context (`report._collect`)**: iterate over all sites. For each, collect eight weeks of trends, 28-day KPI changes, the latest two rank-check comparisons (Top 10 gains/losses), diagnosis counts by rule/severity, to-do counts by status and the five most recent completed feedback items. Combine everything into one JSON object such as `{site_a: {...}, site_b: {...}, ...}`.
2. **One LLM call for one report covering every site**, not one call per site:
   ```text
   System prompt:
   You are a senior SEO consultant for B2B industrial websites. Based on the data JSON
   for multiple sites, write this week's SEO report in Markdown, without an H1.
   Structure:
   ## Overall summary — 3–5 sentences on the most important changes across all sites
   ## SWOT by site — one H3 per site, with 1–3 concrete S/W/O/T items backed by data
   ## Action priorities — global Top 5 sorted by expected impact; identify site/action/expected result
   Requirements: use only the supplied data, do not invent facts, preserve numbers,
   keep the language concise and direct, and avoid filler.
   ```
   Use DeepSeek `deepseek-chat` with `temperature=0.4` and `max_tokens=3000`.
3. **Persist; do not email**: store `content_md` in `weekly_reports.content_md` and use the current date for `week_start`. The report is rendered from Markdown on `/reports`; this system sends no email. Add Resend/SMTP only as an out-of-scope enhancement.

---

## 5. Scheduling: in-process jobs, no independent worker

### 5.1 Scheduled pipeline (production-validated; registered in `main.py` `lifespan`)

```python
@asynccontextmanager
async def lifespan(app):
    sched = BackgroundScheduler(timezone="UTC")
    # Monday pipeline (UTC): 02:00 BigQuery → 03:00 rank check → 05:00 diagnosis
    # → 05:30 to-dos → 06:00 LLM report
    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)
```

Everything runs serially on Monday. There is no daily incremental refresh: Google already updates the BigQuery source daily, and a weekly pull is enough for this system. If your use case needs fresher data, split `bigquery.pull` into a daily cron while keeping diagnosis, selection and reports weekly.

### 5.2 Manual triggers: background thread, not a task queue

```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)
```

Clicking a button starts a daemon thread, immediately returns a 303, and does not wait for the job to finish. The user refreshes the page to see results. There is no execution-state tracking or retry; a process restart loses a running thread. This is an intentional trade-off for one person managing a few sites once a week. Check container logs and click again if needed.

### 5.3 Idempotency guarantees

- `gsc_url_daily` / `gsc_query_url_daily` / `ga4_landing_page_daily`: `bigquery.pull()` deletes by site, then copies the fresh rows.
- `diagnostics`: delete by site and current date, then insert.
- `todos`: `UNIQUE (site, rule, subject)` + `ON CONFLICT DO NOTHING`.
- `tracked_keywords`: `ON CONFLICT (site, keyword, source) DO UPDATE`; terms not selected this run become inactive.
- Four `importer.py` source types: truncate each target table, then fully reload.

---

## 6. Docker orchestration and deployment

### 6.1 `docker-compose.yml` (production-validated)

```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; escape $ as $$ for Compose interpolation
      - "traefik.http.middlewares.seocopilot-auth.basicauth.users=<user>:<htpasswd-hash>"

  # One-off tools container; no Traefik labels, so it is not load-balanced.
  # 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:
```

There is **no independent worker container**. Scheduling and manual triggers run inside `web`; `jobs` exists only for infrequent one-off imports and reselection, and does not start during normal `docker compose up` because it uses `profiles: ["tools"]`.

**Basic Auth generation**: `htpasswd -nb <username> <password>` produces a `user:$apr1$...` hash. Put it into the label and escape each `$` as `$$` in Compose. Alternatively use `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"]
```

This is simpler than a multi-tenant version: no `alembic upgrade head` step and no health-check probe beyond what the Traefik layer needs.

### 6.3 `requirements.txt` (production-validated)

```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.*
```

Not included: SQLAlchemy, Alembic, argon2-cffi, cryptography, google-auth/google-auth-oauthlib, or beautifulsoup4. There is no ORM, migration system, OAuth self-service flow or crawler. `pandas`/`openpyxl` read competitor Excel exports; `google-cloud-bigquery` is the official client used only by `bigquery.py`.

---

## 7. Hidden dependencies and environment preparation

The easiest place for a newcomer to get stuck is not code, but a few Google admin switches. Follow these steps:

1. **GA4 → BigQuery linking**: GA4 → Admin → Product links → BigQuery links → Create link. Select a GCP project and enable daily export. The free plan has limits but is sufficient for personal-scale use. Daily partitioned tables usually appear from the next day.
2. **Search Console → bulk data export**: Search Console → Settings → Bulk data export. Choose a GCP project and dataset. You need Owner/Editor access to that project and verified ownership of the Search Console property. Data begins writing daily from the next day.
3. **GCP service account**: in the project with BigQuery API enabled, create a service account with **BigQuery Data Viewer** and **BigQuery Job User**. Download its JSON key and mount it at `/srv/gcp-service-account.json`. **Never commit this file.**
4. **Historical cold start**: BigQuery linking only provides data from activation onward. For earlier history, export the desired range as CSV in BigQuery, put it under `raw-data/<site>/bigquery-data/`, run `importer.py` once, and then let `bigquery.pull()` handle ongoing refreshes.
5. **Google Ads is manual**: export the keyword report from Google Ads periodically, put it under `raw-data/<site>/google-ads-data/`, and run `importer.py`. This is the one recurring human step. Google Ads API automation is a reasonable future improvement, but outside this SOP.
6. **Competitor data**: export Organic Positions from Semrush/Ahrefs as Excel/CSV and keep the default filename format, such as `domain-organic.Positions-region-date-...`, under `raw-data/<site>/竞争对手关键词/`.
7. **WordPress content**: provide structured JSON such as `{"items": [...]}`. This SOP does not provide the exporter. A small WordPress REST API script can fetch `posts`, `pages` and `products` and output the fields required by the `wp_content` table.
8. **Traefik + Basic Auth**: if Traefik already runs, add a router and middleware labels. Without a reverse proxy, add `ports: ["8000:8000"]` to `web` and run locally; for personal use, do not force HTTPS prematurely.
9. **Serper.dev / DeepSeek accounts**: create accounts and obtain API keys.

---

## 8. Agent automation acceptance and smoke-test checklist

> The production project has no `tests/` directory and no pytest suite. Use this manual smoke-test sequence:

```bash
# 1. Health check and database connectivity
curl -s http://localhost:8000/health

# 2. BigQuery connection test and table discovery
docker compose run --rm jobs python -m app.bigquery test

# 3. Initial cold-start import
docker compose run --rm jobs python -m app.importer /srv/raw-data

# 4. Keyword selection
docker compose run --rm jobs python -m app.selection /srv/raw-data

# 5. Rank check
docker compose run --rm jobs python -c "from app.serper import run_check; print(run_check('manual'))"

# 6. Diagnosis
docker compose run --rm jobs python -c "from app.diagnose import run; run('manual')"

# 7. Generate to-dos
docker compose run --rm jobs python -c "from app.todos import generate; print(generate('manual'))"

# 8. Generate the report
docker compose run --rm jobs python -c "from app.report import generate; print(len(generate('manual')))"

# 9. Browser walkthrough
# Open http://localhost:8000/site/<your-site-key> and confirm that
# the overview, tracking, diagnosis and to-do pages all contain data.
```

Run steps 3 → 4 → 5 → 6 → 7 → 8 in order. If any step returns zero results, the previous import is wrong; check the `raw-data/` directory structure and filenames against Section 7.

---

## 9. Definition of Done

1. **BigQuery sync works**: GA4/GSC links are enabled in Google, and `bigquery.py test` lists and matches all three source tables for every site.
2. **Cold start is complete**: after `importer.py`, all six ingestion tables have non-zero row counts.
3. **Selection and ownership work**: every tracked keyword belongs to exactly one site because `SITE_RULES` matches exclusively.
4. **Serper pagination is correct**: cumulative offset is implemented and one request is shared when the same term belongs to multiple sites.
5. **All seven diagnosis rules run**, including the brand-suffix check in `meta`.
6. **The action loop works**: diagnosis → to-do → user marks complete → `impact()` returns the before/after 28-day comparison.
7. **The weekly report works**: all sites are combined into one Markdown report and can be viewed at `/reports`; email is not required.
8. **Scheduling is confirmed**: APScheduler runs in the `web` process on the Monday sequence, and manual endpoints start a background thread without blocking the request.
9. **Deployment is isolated**: `web` has Traefik + Basic Auth labels; `jobs` has no proxy labels and is used only for one-off commands.

---

## Appendix: Disclaimer

This SOP was organized by checking a real single-tenant production system file by file. Its choices—no ORM, no migration tool, no login system and no persistent task queue—are deliberate simplifications for one person managing a small number of their own websites, not a lack of capability. If the project grows into self-service multi-user registration, arbitrary site onboarding, auditing and permissions, it has crossed this SOP’s design boundary. Use a different architecture with ORM/migrations, OAuth self-service authorization, persistent queues and multi-tenant data isolation. When the Agent encounters a case not covered here, preserve the principles of simplicity, idempotency and single-person maintainability instead of adding abstractions by default.

COPY