competitive-analysis

competitive-analysis — discovery skill (Phase 7 Wave 2 / W-B3)

10-step protocol. Steps map 1:1 to workflow_runner invocations + the spec §16.5 8-step MCP discipline + §16.8 budget pre-flight + §14.5 canonical tier ladder. Raw JSON drift recovery is mandatory: every DFS / Scrapling response is dropped into inbox/dfs/ or inbox/scrapling/ before any transform runs, so a transform bug never costs us the upstream payload (DFS competitors_domain is paid; re-fetch costs credits).

This skill follows the convention authority of:

• skills/ingestion/scrapling-ops/SKILL.md — Phase 6 generic helper; authoritative for the §14.5 tier_escalation invariant + bulk URL routing + inbox/scrapling/ raw-recovery discipline.
• skills/ingestion/dfs-pull/SKILL.md — paid-MCP authority; budget pre-flight at Step 1, cost.credits provenance fields, _normalize_dfs_response for REST envelope vs flat wrapper.
• skills/discovery/quick-wins/SKILL.md (Phase 5) and skills/discovery/cannibalization/SKILL.md (Phase 7 Wave 1) — 10-step shape, raw inbox, D-03 URL invariant, DURUR + flag rule, provenance event format.

ADR-025 Phase 7+ activation (NEW): this skill is the first concrete consumer of templates/scrapling/S1_competitor_snapshot.schema.json. Every projected staging row passes Draft7 validation against that sub-schema before it lands on disk; failure raises StagingSchemaError (DURUR #3). The S1 schema is locked alongside this skill — do not modify either side without an ADR.

Inputs (frontmatter contract)

Name	Type	Default	Notes
project_slug	string	—	Required. Resolves projects/{slug}/master.xlsx + staging dir.
target_domain	string	—	Required. Used for DFS competitors_domain lookup.
competitor_urls	array	—	Optional. Direct override; otherwise derived from competitors_domain.
max_urls	integer	50	Hard cap; scrapling-ops UrlBudgetExceededError upstream.
bulk_threshold	number	0.5	Above-threshold all-tiers-fail rate → BulkUnhealthyError.

workspace_root is resolved via PSEO_WORKSPACE_ROOT env or explicit test override (mirrors workflow_runner / events_writer).

Locale resolution (config-first — NO engine-level country default)

The DFS location_code / language_code used by Step 2 (competitors_domain) are resolved config-first from the project's own project.config.dataforseo block — schema-required fields (schemas/project-config.schema.json). There is no engine-level country default and this skill does not inherit a hardcoded locale from dfs-pull: the engine is project-agnostic and must never assume Turkey (or any country). Resolution order:

1. Explicit location_code / language_code (operator override).
2. project.config.dataforseo.location_code + project.config.dataforseo.language_code.
3. Neither available → DURUR (config locale unresolvable): STOP before any paid MCP call; never fall back to a country.

Per-market reference (codes VERIFIED from live project configs)

market	content_locale	location_code	language_code	Note
TR	tr-TR	2792	tr	Turkey (country).
CA	en-CA	20120	en	"Ontario,Canada" — a sub-country location; a coarse country code would be wrong.
NG	en-NG	2566	en	Nigeria (country).

Every code above is read straight from that project's project.config.dataforseo (the authoritative source — not hardcoded here). For a NEW market, resolve the DFS location_code at init via a serp_locations lookup and persist it into project.config.dataforseo — never guess a country code in skill logic. (Same contract as skills/ingestion/dfs-pull/SKILL.md, the paid-MCP convention authority.)

Outputs (artifacts produced)

• projects/{slug}/_state/staging/competitive_analysis_{date}_{slug}.json — JSON array of S1-validated competitor snapshot rows. NO Excel write — Phase 7+ S1 path is staging-only per ADR-025.
• projects/{slug}/inbox/scrapling/{date}-bulk_stealthy_fetch-competitor-{slug}.json — raw Scrapling response envelope (drift recovery).
• projects/{slug}/inbox/dfs/{date}-competitors_domain-{slug}.json — raw DataForSEO competitors_domain response (drift recovery; paid).
• projects/{slug}/outputs/reports/{date}-competitive-analysis.md — human-readable summary (top competitors, content_hash deltas vs prior snapshot, tier distribution, durur list).
• projects/{slug}/_state/events.jsonl — event_kind=provenance entries with source.kind ∈ {scrapling_mcp, dataforseo_mcp} and target_excel_sheet=null (staging-only).

Tier Escalation (canonical, schema-locked §14.5)

The state machine is fixed by schemas/scrapling-output-mapping.schema .json (tierEscalation definition: ["get", "fetch", "stealthy_fetch"], minItems=3, maxItems=3, const-equal). This skill enforces the same invariant on the transform side via competitive_analysis_transform.assert_canonical_tier_order — any caller who supplies a custom tier_escalation parameter that diverges from the canonical sequence raises TierEscalationOrderError (DURUR #7).

Tier 0  get             — basic HTTP GET, no JS, no anti-bot.
                          Cheapest tier; tries first per-URL.
Tier 1  fetch           — browser-like headers, optional JS render.
Tier 2  stealthy_fetch  — anti-bot bypass (Cloudflare-aware,
                          fingerprint randomization).
                          Failure modes: still blocked → DURUR
                          (do not invent tier 3).

Fixed sequence: ['get', 'fetch', 'stealthy_fetch']. Order is canonical (cheap → expensive). Each tier failure escalates to the next. All 3 fail → DURUR (the URL is recorded in events.jsonl with source.kind=scrapling_mcp and excluded from staging; if the aggregate failure rate exceeds bulk_threshold the whole run halts via BulkUnhealthyError).

10-Step Body Protocol

Each step name must match the steps[*].name passed to workflow_runner.create_run. Names are stable identifiers across runs.

Step 1 — create_run + preflight_budget

from scripts.state import workflow_runner
from scripts.discovery import competitive_analysis_transform as ca
handle = workflow_runner.create_run(
    skill="competitive-analysis",
    project_slug=project_slug,
    steps=[
        {"name": "preflight_budget"},
        {"name": "fetch_competitors_domain"},
        {"name": "fetch_scrapling_bulk"},
        {"name": "persist_inbox"},
        {"name": "transform_and_validate"},
        {"name": "request_approval"},
        {"name": "write_staging"},
        {"name": "render_report"},
    ],
)

# §16.8 budget pre-flight — DURUR #2 on FAIL.
ca.preflight_budget(
    estimated_credits=ca.estimate_credits(1),     # competitors_domain ~5 credits
    project_config_path=str(workspace_root / "projects" / project_slug / "project.config.json"),
    events_path=str(workspace_root / "projects" / project_slug / "_state" / "events.jsonl"),
)

Step 2 — fetch_competitors_domain (DFS, paid)

# location_code / language_code resolved config-first from
# project.config.dataforseo (see "Locale resolution") — no engine default.
raw_competitors = mcp__dataforseo__dataforseo_labs_google_competitors_domain(
    target=target_domain,
    location_code=location_code,   # from project.config.dataforseo
    language_code=language_code,   # from project.config.dataforseo
    limit=20,
)

Parse via competitive_analysis_transform.extract_competitor_domains (tolerant of both REST envelope AND flat wrapper, identical to dfs_pull's _normalize_dfs_response).

Step 3 — fetch_scrapling_bulk (Scrapling, free, tier 2)

For weekly competitor monitoring (S1) we go directly to bulk_stealthy_fetch — competitor sites typically gate at tier 1. The §14.5 invariant still applies: a per-URL run that ascends the ladder records tier_attempts honestly (get success → 1, fetch → 2, stealthy_fetch → 3). Callers wanting full ladder traversal use scripts.ingestion.scrapling_ops.bulk_tier_escalate upstream and hand FetchEntry instances to this transform.

raw_scrapling = mcp__ScraplingServer__bulk_stealthy_fetch(
    urls=competitor_urls,
)

Step 4 — persist_inbox (raw drift recovery, §16.5 step 3)

ca.write_inbox_raw(
    raw_competitors,
    workspace_root / "projects" / project_slug / "inbox" / "dfs"
    / f"{date}-competitors_domain-{project_slug}.json",
)
ca.write_inbox_raw(
    raw_scrapling,
    workspace_root / "projects" / project_slug / "inbox" / "scrapling"
    / f"{date}-bulk_stealthy_fetch-competitor-{project_slug}.json",
)

Inbox path failure → InboxPathError (DURUR #6).

Step 5 — transform_and_validate

Pure compute via scripts/discovery/competitive_analysis_transform.py:

entries = ca.fetch_entries_from_scrapling_envelope(raw_scrapling)
result = ca.transform(
    entries,
    project_slug=project_slug,
    snapshot_date=snapshot_iso_utc,
    raw_competitors_domain=raw_competitors,
    bulk_threshold=bulk_threshold,
)

Each row goes through validate_s1_row(row) — Draft7 validation against templates/scrapling/S1_competitor_snapshot.schema.json. Failure raises StagingSchemaError (DURUR #3). All-tiers-fail entries are excluded from result["rows"] and surface in result["durur_urls"] for the events.jsonl + report layers.

If result["rows"] is empty AND result["durur_urls"] is empty (no input), the skill exits as a clean no-op (recorded in events.jsonl, no staging file written).

Step 6 — request_approval (skill EXIT awaiting_approval)

workflow_runner.request_approval(
    handle.run_id, project_slug=project_slug,
    approver="user",
    subject=f"{result['meta']['row_count']} rakip URL snapshot alındı, "
            f"_state/staging'e yazalım mı?",
    step_index=4,
)
# Skill exits here. The user replies in a fresh session; resume below.

Step 7 — write_staging (NO Excel — staging-only per ADR-025)

output_path = (
    workspace_root / "projects" / project_slug
    / "_state" / "staging"
    / ca.staging_filename(date=date, project_slug=project_slug)
)
ca.write_staging_json(result["rows"], output_path)

Staging path failure → StagingPathError (DURUR #5). The writer re-validates every row against S1 immediately before flushing the tmp file to its final name — never persists a drifted row.

Note: transaction.append is NOT called — there is no Excel target for the S1 path. Phase 8 may project staging → master.xlsx in a separate skill (parallel to dfs-pull → cluster-map). The transform module imports nothing from scripts.excel — enforced by test_transform_does_not_ import_transaction.

Step 8 — render_report

render_template.py templates/reports/competitive-analysis.template.md data.json → outputs/reports/{date}-competitive-analysis.md. Variables: $project_slug, $date, $target_domain, $row_count, $durur_count, $competitor_domain_count, $top_competitors, $tier_distribution, $staging_path, $run_id.

Step 9 — Provenance events (TWO emits — one per source)

from scripts.state import events_writer

# Scrapling — free, target_excel_sheet=null per staging-only.
events_writer.append_provenance(
    project_id=project_slug,
    source={
        "kind": "scrapling_mcp",
        "mcp_server": "ScraplingServer",
        "mcp_tool": "ScraplingServer__bulk_stealthy_fetch",
        "response_bytes": len(json.dumps(raw_scrapling)),
    },
    operation="ingest",
    target_excel_sheet=None,
    target_table="scrapling_competitor_snapshot",
    rows_written=result["meta"]["row_count"],
)

# DataForSEO — paid, cost.credits populated.
events_writer.append_provenance(
    project_id=project_slug,
    source={
        "kind": "dataforseo_mcp",
        "mcp_server": "dataforseo",
        "mcp_tool": "dataforseo__dataforseo_labs_google_competitors_domain",
        "response_bytes": len(json.dumps(raw_competitors)),
    },
    operation="ingest",
    target_excel_sheet=None,
    target_table="dfs_competitors_domain",
    rows_written=result["meta"]["competitor_domain_count"],
    cost={
        "provider": "dataforseo",
        "credits": ca.estimate_credits(1),
        "budget_key": "project.config.dataforseo.budget_credits_per_day",
    },
)

Per-URL all-tiers-fail entries in result["durur_urls"] ALSO get one source.kind=scrapling_mcp provenance event each (operation=ingest, rows_written=0, validation_status=fail) so drift-check can audit them without scanning the staging file.

target_excel_sheet=null is intentional and supported by schemas/events.schema.json (oneOf: logicalSheet | null).

Step 10 — complete

workflow_runner.complete(handle.run_id, project_slug=project_slug, outputs={
    # F5: outputs.* must be STRING-TYPED.
    "row_count": str(result["meta"]["row_count"]),
    "durur_count": str(result["meta"]["durur_count"]),
    "competitor_domain_count": str(result["meta"]["competitor_domain_count"]),
    "staging_path": str(output_path),
    "report_path": str(report_path),
})

Workflow status flips running → done (workflow-run schema) and a workflow_action=done event lands in events.jsonl (ADR-020).

S1 Staging Row Shape (templates/scrapling/S1_competitor_snapshot.schema.json)

Field	Type	Notes
snapshot_date	string (date-time)	Shared across all rows of a batch (snapshot identity).
domain	string	Competitor root domain (lowercase, no scheme).
url_normalized	string (uri)	D-03 canonical form; cross-skill join key.
url_original	string (uri)	URL as fetched, pre-normalization.
fetch_method	enum	get / fetch / stealthy_fetch (§14.5).
tier_attempts	int 1..3	Tiers attempted before success.
content_hash	string [a-f0-9]{64}	SHA-256 hex (no sha256: prefix on this sub-schema).
content_excerpt	string ≤500	First 500 chars; for diff visibility.
meta_title	string | null	Page <title>.
meta_description	string | null	<meta name="description">.
h1	string[]	All H1 headings (document order).
schema_org_count	int ≥0	JSON-LD schema.org block count.
page_size_bytes	int ≥0	Body byte count.
status_code	int 100..599	HTTP status code.
fetch_duration_ms	int ≥0	End-to-end fetch wall time.

Required: snapshot_date, domain, url_normalized, fetch_method, status_code. The remaining columns are optional in the schema but written deterministically by the transform (defaults used when the fetched envelope omits them).

URL normalization (D-03 invariant)

Every URL passing through this skill is normalized via competitive_analysis_transform.normalize_url. The function is idempotent: normalize_url(normalize_url(u)) == normalize_url(u). Rules: lowercase scheme+host, IDN→punycode, strip default ports, strip trailing slash on non-root, drop fragment, drop tracking params, sort remaining query keys. Same shape as cannibalization / quick-wins — cross-skill drift would break the join key contract.

DURUR conditions (9)

Stop and flag the manager — do not patch, do not fall back.

1. All 3 Scrapling tiers fail for a URL (recorded in events.jsonl with source.kind=scrapling_mcp; excluded from staging file). The batch as a whole halts only when durur_count / total > bulk_threshold — see DURUR #4.
2. Budget pre-flight FAIL — preflight_budget raises BudgetExceededError. STOP, awaiting_approval; no DFS or Scrapling call is dispatched.
3. S1 schema validate fail on a transform staging row — raises StagingSchemaError. Caused by sub-schema drift; STOP, manager confirms before re-run.
4. Bulk fetch >50% URL fail rate (across all 3 tiers) — raises BulkUnhealthyError. Competitor batch unhealthy; STOP, no partial staging accepted. Override via bulk_threshold input only with manager sign-off.
5. _state/staging/ path cannot be created / not writable — StagingPathError. STOP, surface to manager.
6. inbox/scrapling/ or inbox/dfs/ path cannot be created — InboxPathError. STOP; without inbox the raw payload cannot be recovered on transform-bug re-runs.
7. tier_escalation non-canonical order — caller-supplied sequence diverges from ['get','fetch','stealthy_fetch']; raises TierEscalationOrderError. §14.5 invariant locked.
8. workflow_runner.create_run fails schema validation (schemas/workflow-run.schema.json). STOP.
9. PSEO_WORKSPACE_ROOT env var unset and no explicit workspace_root arg passed to workflow_runner / events_writer. STOP, surface to manager.

Cross-references

• Schemas: templates/scrapling/S1_competitor_snapshot.schema.json (this skill's S1 sub-schema, ADR-025 Phase 7+ activation), schemas/scrapling-output-mapping.schema.json (§14.5 const-locked tier ladder; §14.2 S1 scenario contract), schemas/events.schema.json (source.kind ∈ {scrapling_mcp, dataforseo_mcp}, target_excel_sheet=null), schemas/skill-frontmatter.schema.json (this frontmatter, Q-W-A4-01 budget block lock).
• Cross-modules (IMPORT-only): scripts/state/workflow_runner.py, scripts/state/events_writer.py, scripts/reporting/render_template.py, scripts/budget/check_budget.py (via preflight_budget helper), scripts/ingestion/dfs_pull.py (_normalize_dfs_response re-use for competitors_domain envelope tolerance), scripts/ingestion/scrapling_ops.py (TIER_LADDER, bulk_tier_escalate for full-ladder runs).
• Transform: scripts/discovery/competitive_analysis_transform.py.
• Tests: tests/skills/test_competitive_analysis.py (≥7 cases incl. S1 schema self-validate, tier_escalation §14.5 order rejection, bulk-unhealthy threshold, budget pre-flight PASS/FAIL).
• Template: templates/reports/competitive-analysis.template.md.
• ADR: ADR-025 (Phase 7+ sub-schema activation; this skill is the first consumer), ADR-026 (Q-015 hard cap on URL count).

Discipline checklist

• TODO/fallback YASAK — every DURUR raises, none silently downgrade.
• Schema-first — frontmatter validates against schemas/skill-frontmatter.schema.json Draft 7; every row validates against templates/scrapling/S1_competitor_snapshot.schema.json Draft 7.
• Plugin-agnostik — no slug literals; project_slug flows through every path; transform has 0 hardcoded slug words.
• ADR-013: Use when / Also use when / Do not use when are STRING content inside description, not separate fields.
• Cross-module IMPORT discipline — workflow_runner / events_writer / dfs_pull._normalize_dfs_response / scrapling_ops are imported where helpful, never modified. scripts.excel.transaction is NOT imported (staging-only).
• Q-W-A4-01: budget block carries only uses_paid_mcp and estimated_credits — schema rejects _per_call/_per_url keys.
• §14.5 invariant: assert_canonical_tier_order enforces ['get','fetch','stealthy_fetch'] const-locked sequence.
• Append-only state — events.jsonl only grows; no in-place rewrite.
• STAGING-ONLY: no transaction.append; output path is _state/staging/competitive_analysis_{date}_{slug}.json.