Crawl4ai Seo

config/defaults.json
config/config.json
cache/*
!cache/.gitkeep
reports/
.venv/
__pycache__/
*.pyc
.DS_Store

{
  "seed": {
    "limit": 200,
    "same_host_only": true,
    "include_patterns": [],
    "exclude_patterns": [
      "\\?.*utm_",
      "\\?.*fbclid=",
      "\\?.*gclid=",
      "/tag/",
      "/author/",
      "/feed/",
      "#"
    ],
    "request_timeout_seconds": 20
  },
  "crawl": {
    "cache_mode": "enabled",
    "max_urls": 100,
    "max_concurrency": 5,
    "word_count_threshold": 20,
    "page_timeout_ms": 45000,
    "respect_cache": true,
    "capture_markdown": true,
    "capture_fit_markdown": true
  },
  "browser": {
    "headless": true,
    "verbose": false
  },
  "output": {
    "preview_rows": 20,
    "store_html": false,
    "store_markdown": true
  },
  "scrapedo": {
    "enabled": false,
    "token_file": "../scrapedo-web-scraper/config/token.txt",
    "api_base": "http://api.scrape.do"
  }
}

#!/usr/bin/env python3
"""Build a navigation audit report from one seed job and one or more crawl jobs."""

from __future__ import annotations

import argparse
import csv
import json
import statistics
from collections import Counter, defaultdict
from pathlib import Path
from typing import Any
from urllib.parse import urlparse

from common import load_job, load_ndjson, load_seed_payload, normalize_domain, normalize_url, utc_now_iso, write_json


def path_depth(url: str) -> int:
    path = urlparse(url).path.strip("/")
    if not path:
        return 0
    return len([part for part in path.split("/") if part])


def prefix(url: str, parts_count: int = 1) -> str:
    parts = [part for part in urlparse(url).path.strip("/").split("/") if part]
    if not parts:
        return "/"
    return "/" + "/".join(parts[:parts_count]) + "/"


def url_parts(url: str) -> list[str]:
    return [part for part in urlparse(url).path.strip("/").split("/") if part]


def safe_list(value: Any) -> list[str]:
    if not isinstance(value, list):
        return []
    return [str(item).strip() for item in value if str(item).strip()]


def section_prefix(url: str) -> str:
    parts = url_parts(url)
    if len(parts) >= 2 and parts[0] in {"catalog", "articles", "news", "tags"}:
        return "/" + "/".join(parts[:2]) + "/"
    return prefix(url, 1)


def breadcrumb_signature(page: dict[str, Any]) -> str | None:
    breadcrumb_urls = [urlparse(normalize_url(item)).path or "/" for item in safe_list(page.get("breadcrumb_urls"))]
    if breadcrumb_urls:
        return " > ".join(breadcrumb_urls)

    breadcrumb_texts = safe_list(page.get("breadcrumb_texts"))
    if len(breadcrumb_texts) >= 2:
        return " > ".join(breadcrumb_texts[:-1] or breadcrumb_texts)
    return None


def nav_url_set(page: dict[str, Any]) -> set[str]:
    return {normalize_url(item) for item in safe_list(page.get("nav_urls_sample")) if normalize_url(item)}


def is_asset_like_url(url: str) -> bool:
    path = (urlparse(url).path or "").lower()
    suffixes = (
        ".jpg",
        ".jpeg",
        ".png",
        ".gif",
        ".svg",
        ".webp",
        ".pdf",
        ".doc",
        ".docx",
        ".xls",
        ".xlsx",
        ".zip",
        ".rar",
        ".mp4",
        ".webm",
    )
    if path.startswith("/upload/") or path.startswith("/downloads/"):
        return True
    return path.endswith(suffixes)


def classify_url_noise(url: str, page: dict[str, Any] | None = None) -> list[str]:
    parsed = urlparse(url)
    reasons: list[str] = []
    path = parsed.path or "/"
    query = parsed.query or ""

    if "login=yes" in query:
        reasons.append("login query URL")
    if "PAGEN_" in query:
        reasons.append("pagination URL")
    if path.endswith(".php"):
        reasons.append("php endpoint")
    if path.endswith("json.php") or path.endswith(".json"):
        reasons.append("json/service endpoint")
    if path in {"/personal/", "/personal/cart/"}:
        reasons.append("personal area URL")
    if path in {"/virtual.php", "/catalog/index1.php"}:
        reasons.append("alternate/technical endpoint")
    if path == "/policy/":
        reasons.append("document-like policy URL")
    if is_asset_like_url(url):
        reasons.append("asset/document URL")

    if page is not None:
        if page.get("success") and not page.get("title") and not page.get("has_breadcrumbs") and int(page.get("nav_link_count") or 0) == 0:
            reasons.append("document-like page without HTML nav signals")

    return reasons


def is_hub_candidate(url: str, page: dict[str, Any]) -> bool:
    parts = url_parts(url)
    depth = int(page.get("path_depth") or path_depth(url))
    if not parts:
        return True
    if depth <= 2:
        return True
    if len(parts) == 2 and parts[0] in {"catalog", "articles", "news", "tags"}:
        return True
    return False


def pick_best_page(current: dict[str, Any] | None, candidate: dict[str, Any]) -> dict[str, Any]:
    if current is None:
        return candidate

    current_score = (
        1 if current.get("success") else 0,
        1 if (current.get("status_code") or 0) < 400 else 0,
        1 if current.get("has_breadcrumbs") else 0,
        current.get("nav_link_count") or 0,
        current.get("internal_links_count") or 0,
        current.get("run_id") or "",
    )
    candidate_score = (
        1 if candidate.get("success") else 0,
        1 if (candidate.get("status_code") or 0) < 400 else 0,
        1 if candidate.get("has_breadcrumbs") else 0,
        candidate.get("nav_link_count") or 0,
        candidate.get("internal_links_count") or 0,
        candidate.get("run_id") or "",
    )
    return candidate if candidate_score >= current_score else current


def write_csv(path: Path, rows: list[dict[str, Any]], fieldnames: list[str]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    with path.open("w", encoding="utf-8", newline="") as handle:
        writer = csv.DictWriter(handle, fieldnames=fieldnames)
        writer.writeheader()
        for row in rows:
            writer.writerow(row)


def main() -> int:
    parser = argparse.ArgumentParser(description="Build a navigation audit report from cached crawl jobs.")
    parser.add_argument("--seed-job-id", help="Job id with seed.json that defines the full sitemap frontier")
    parser.add_argument("--seed-path", help="Direct path to seed.json")
    parser.add_argument("--crawl-job-id", action="append", default=[], help="Repeatable crawl job id")
    parser.add_argument("--report-dir", required=True, help="Directory for human-facing report artifacts")
    parser.add_argument("--label", default="navigation-audit", help="Free-form label for report metadata")
    args = parser.parse_args()

    if not args.seed_job_id and not args.seed_path:
        parser.error("Provide --seed-job-id or --seed-path.")
    if not args.crawl_job_id:
        parser.error("Provide at least one --crawl-job-id.")

    if args.seed_path:
        seed_path = Path(args.seed_path)
    else:
        seed_job = load_job(args.seed_job_id)
        seed_path = Path(seed_job["resolved"]["paths"]["seed_path"])

    seed_payload = load_seed_payload(seed_path)
    seed_urls = [normalize_url(url) for url in seed_payload.get("urls", []) if str(url).strip()]
    seed_set = set(seed_urls)
    domain = normalize_domain(seed_payload.get("target", {}).get("domain") or seed_payload.get("domain") or "")

    pages_by_url: dict[str, dict[str, Any]] = {}
    all_links: dict[tuple[str, str, str], dict[str, Any]] = {}
    crawl_jobs_meta: list[dict[str, Any]] = []

    for job_id in args.crawl_job_id:
        job = load_job(job_id)
        crawl_jobs_meta.append(
            {
                "job_id": job_id,
                "job_dir": str(job["job_dir"]),
                "label": job.get("resolved", {}).get("job", {}).get("label"),
            }
        )
        pages_path = Path(job["resolved"]["paths"]["pages_path"])
        links_path = Path(job["resolved"]["paths"]["links_path"])

        for page in load_ndjson(pages_path):
            url = normalize_url(page.get("url", ""))
            if not url:
                continue
            page["path_depth"] = page.get("path_depth") if page.get("path_depth") is not None else path_depth(url)
            pages_by_url[url] = pick_best_page(pages_by_url.get(url), page)

        for link in load_ndjson(links_path):
            if link.get("kind") != "internal":
                continue
            source_url = normalize_url(link.get("source_url", ""))
            target_url = normalize_url(link.get("target_url", ""))
            if not source_url or not target_url:
                continue
            if normalize_domain(source_url) != domain or normalize_domain(target_url) != domain:
                continue
            key = (source_url, target_url, link.get("anchor_text") or "")
            all_links[key] = {
                "source_url": source_url,
                "target_url": target_url,
                "anchor_text": link.get("anchor_text") or "",
            }

    crawled_urls = set(pages_by_url)
    links = list(all_links.values())

    incoming_sources: dict[str, set[str]] = defaultdict(set)
    outgoing_targets: dict[str, set[str]] = defaultdict(set)
    for link in links:
        source_url = link["source_url"]
        target_url = link["target_url"]
        if source_url == target_url:
            continue
        outgoing_targets[source_url].add(target_url)
        incoming_sources[target_url].add(source_url)

    successful_pages = [
        page
        for page in pages_by_url.values()
        if page.get("success") and (page.get("status_code") or 0) < 400
    ]
    hub_pages = [page for page in successful_pages if is_hub_candidate(page["url"], page)]

    nav_counts = [int(page.get("nav_link_count") or 0) for page in successful_pages if int(page.get("nav_link_count") or 0) > 0]
    nav_median = int(statistics.median(nav_counts)) if nav_counts else 0
    low_nav_threshold = max(3, nav_median // 2) if nav_median else 3

    nav_pages = [page for page in successful_pages if nav_url_set(page)]
    common_nav_counts: Counter[str] = Counter()
    for page in nav_pages:
        common_nav_counts.update(nav_url_set(page))
    common_nav_threshold = max(3, int(len(nav_pages) * 0.85)) if nav_pages else 999999
    common_nav_targets = {
        url
        for url, count in common_nav_counts.most_common(20)
        if count >= common_nav_threshold
    }

    page_inventory: list[dict[str, Any]] = []
    failed_pages: list[dict[str, Any]] = []
    orphan_like: list[dict[str, Any]] = []
    weakly_linked: list[dict[str, Any]] = []
    breadcrumb_issues: list[dict[str, Any]] = []
    breadcrumb_inconsistencies: list[dict[str, Any]] = []
    nav_issues: list[dict[str, Any]] = []
    menu_inconsistencies: list[dict[str, Any]] = []
    weak_hubs: list[dict[str, Any]] = []
    canonical_mismatches: list[dict[str, Any]] = []
    linked_not_seeded_rows: list[dict[str, Any]] = []
    technical_junk_rows: list[dict[str, Any]] = []

    title_map: dict[str, list[str]] = defaultdict(list)
    breadcrumb_signatures_by_section: dict[str, Counter[str]] = defaultdict(Counter)
    technical_junk_seen: set[tuple[str, str]] = set()

    for url, page in sorted(pages_by_url.items()):
        incoming = len(incoming_sources.get(url, set()))
        outgoing = len(outgoing_targets.get(url, set()))
        breadcrumb_sig = breadcrumb_signature(page)
        missing_common_nav = sorted(common_nav_targets - nav_url_set(page))
        record = {
            "url": url,
            "status_code": page.get("status_code"),
            "success": page.get("success"),
            "title": page.get("title"),
            "path_depth": page.get("path_depth"),
            "incoming_internal_links": incoming,
            "outgoing_internal_links": outgoing,
            "has_breadcrumbs": page.get("has_breadcrumbs"),
            "nav_link_count": page.get("nav_link_count"),
            "internal_links_count": page.get("internal_links_count"),
            "canonical": page.get("canonical"),
            "breadcrumb_signature": breadcrumb_sig,
            "menu_common_nav_missing": len(missing_common_nav),
        }
        page_inventory.append(record)

        title = (page.get("title") or "").strip().lower()
        if title:
            title_map[title].append(url)

        status_code = page.get("status_code") or 0
        if not page.get("success") or status_code >= 400:
            failed_pages.append(
                {
                    "type": "http_error",
                    "url": url,
                    "status_code": status_code,
                    "detail": page.get("error") or "",
                }
            )
            continue

        if page.get("canonical") and normalize_url(page["canonical"]) != url:
            canonical_mismatches.append(
                {
                    "type": "canonical_mismatch",
                    "url": url,
                    "status_code": status_code,
                    "detail": page.get("canonical"),
                }
            )

        if url != f"https://{domain}/" and incoming == 0:
            orphan_like.append(
                {
                    "type": "orphan_like",
                    "url": url,
                    "status_code": status_code,
                    "detail": f"path_depth={page.get('path_depth')}",
                }
            )
        elif incoming == 1:
            weakly_linked.append(
                {
                    "type": "weak_inlinks",
                    "url": url,
                    "status_code": status_code,
                    "detail": "only 1 internal inlink in crawled graph",
                }
            )

        if int(page.get("path_depth") or 0) >= 2 and not page.get("has_breadcrumbs"):
            breadcrumb_issues.append(
                {
                    "type": "missing_breadcrumbs",
                    "url": url,
                    "status_code": status_code,
                    "detail": "",
                }
            )

        if page.get("has_breadcrumbs") and breadcrumb_sig:
            breadcrumb_signatures_by_section[section_prefix(url)][breadcrumb_sig] += 1

        nav_link_count = int(page.get("nav_link_count") or 0)
        if is_hub_candidate(url, page) and nav_link_count < low_nav_threshold:
            nav_issues.append(
                {
                    "type": "weak_nav_template",
                    "url": url,
                    "status_code": status_code,
                    "detail": f"nav_link_count={nav_link_count}, median={nav_median}",
                }
            )

        if is_hub_candidate(url, page) and common_nav_targets and len(missing_common_nav) >= 2:
            menu_inconsistencies.append(
                {
                    "type": "menu_inconsistency",
                    "url": url,
                    "status_code": status_code,
                    "detail": f"missing_common_nav={len(missing_common_nav)}",
                }
            )

        if is_hub_candidate(url, page) and outgoing < 5:
            weak_hubs.append(
                {
                    "type": "weak_hub_outlinks",
                    "url": url,
                    "status_code": status_code,
                    "detail": f"outgoing_internal_links={outgoing}",
                }
            )

        for reason in classify_url_noise(url, page):
            key = (url, reason)
            if key in technical_junk_seen:
                continue
            technical_junk_seen.add(key)
            technical_junk_rows.append(
                {
                    "type": "technical_junk",
                    "url": url,
                    "status_code": status_code,
                    "detail": reason,
                }
            )

    all_linked_not_seeded_targets = sorted(
        {
            link["target_url"]
            for link in links
            if link["target_url"] not in seed_set
        }
    )
    linked_not_seeded_targets = []
    for url in all_linked_not_seeded_targets:
        if is_asset_like_url(url):
            for reason in classify_url_noise(url):
                key = (url, reason)
                if key in technical_junk_seen:
                    continue
                technical_junk_seen.add(key)
                technical_junk_rows.append(
                    {
                        "type": "technical_junk",
                        "url": url,
                        "status_code": "",
                        "detail": reason,
                    }
                )
            continue
        linked_not_seeded_targets.append(url)
        incoming = len(incoming_sources.get(url, set()))
        linked_not_seeded_rows.append(
            {
                "type": "linked_not_seeded",
                "url": url,
                "status_code": "",
                "detail": f"incoming_from_crawled={incoming}",
            }
        )
        for reason in classify_url_noise(url):
            key = (url, reason)
            if key in technical_junk_seen:
                continue
            technical_junk_seen.add(key)
            technical_junk_rows.append(
                {
                    "type": "technical_junk",
                    "url": url,
                    "status_code": "",
                    "detail": reason,
                }
            )

    for section, signatures in sorted(breadcrumb_signatures_by_section.items()):
        if sum(signatures.values()) < 2 or len(signatures) < 2:
            continue
        top_signature, top_count = signatures.most_common(1)[0]
        if top_count == sum(signatures.values()):
            continue
        breadcrumb_inconsistencies.append(
            {
                "type": "breadcrumb_inconsistency",
                "url": section,
                "status_code": "",
                "detail": f"signature_variants={len(signatures)}, dominant={top_count}/{sum(signatures.values())}, dominant_signature={top_signature}",
            }
        )

    duplicate_title_groups = [
        {"type": "duplicate_title", "title": title, "urls": urls}
        for title, urls in title_map.items()
        if len(urls) > 1
    ]
    duplicate_title_rows = [
        {
            "type": "duplicate_title",
            "url": ", ".join(item["urls"]),
            "status_code": "",
            "detail": item["title"],
        }
        for item in duplicate_title_groups
    ]

    crawled_prefix_counts = Counter(prefix(url, 1) for url in crawled_urls)
    seed_prefix_counts = Counter(prefix(url, 1) for url in seed_urls)
    backlog = sorted(seed_set - crawled_urls)
    backlog_prefix_counts = Counter(prefix(url, 1) for url in backlog)
    linked_not_seeded_prefix_counts = Counter(prefix(url, 1) for url in linked_not_seeded_targets)
    technical_junk_prefix_counts = Counter(prefix(row["url"], 1) for row in technical_junk_rows)

    structural_anti_patterns: list[str] = []
    catalog_seed = seed_prefix_counts.get("/catalog/", 0)
    if seed_urls and catalog_seed / len(seed_urls) >= 0.8 and (seed_prefix_counts.get("/tags/", 0) or seed_prefix_counts.get("/news/", 0)):
        structural_anti_patterns.append(
            f"sitemap frontier перегружен catalog/archive mix: /catalog/={catalog_seed}, /tags/={seed_prefix_counts.get('/tags/', 0)}, /news/={seed_prefix_counts.get('/news/', 0)}"
        )
    if linked_not_seeded_prefix_counts:
        top_prefix, top_count = linked_not_seeded_prefix_counts.most_common(1)[0]
        structural_anti_patterns.append(
            f"сайт линкует зоны вне seed frontier: linked-not-seeded={len(linked_not_seeded_targets)}, крупнейший префикс {top_prefix} ({top_count})"
        )
    if any("login query URL" in row["detail"] for row in technical_junk_rows):
        structural_anti_patterns.append("внутренний граф содержит login/query URL, что загрязняет навигационный слой и создаёт crawl waste")
    if any("php endpoint" in row["detail"] or "json/service endpoint" in row["detail"] for row in technical_junk_rows):
        structural_anti_patterns.append("в навигационный граф попадают php/json/service endpoints вместо чистых HTML landing pages")
    if menu_inconsistencies:
        structural_anti_patterns.append(f"есть menu/template drift: {len(menu_inconsistencies)} hub-like страниц отклоняются от common nav set")
    if breadcrumb_inconsistencies:
        structural_anti_patterns.append(f"есть breadcrumb drift по секциям: {len(breadcrumb_inconsistencies)} section groups с несколькими trail signature")

    summary = {
        "created_at": utc_now_iso(),
        "label": args.label,
        "domain": domain,
        "seed_url_count": len(seed_urls),
        "crawled_unique_urls": len(crawled_urls),
        "successful_pages": len(successful_pages),
        "internal_link_edges": len(links),
        "coverage_pct": round((len(crawled_urls) / len(seed_urls) * 100), 2) if seed_urls else 0,
        "nav_link_count_median": nav_median,
        "backlog_url_count": len(backlog),
        "failed_pages": len(failed_pages),
        "orphan_like_pages": len(orphan_like),
        "weakly_linked_pages": len(weakly_linked),
        "missing_breadcrumbs": len(breadcrumb_issues),
        "weak_nav_templates": len(nav_issues),
        "weak_hubs": len(weak_hubs),
        "canonical_mismatches": len(canonical_mismatches),
        "duplicate_title_groups": len(duplicate_title_groups),
        "linked_not_seeded": len(linked_not_seeded_targets),
        "technical_junk": len(technical_junk_rows),
        "menu_inconsistencies": len(menu_inconsistencies),
        "breadcrumb_inconsistencies": len(breadcrumb_inconsistencies),
        "structural_anti_patterns": len(structural_anti_patterns),
        "crawl_jobs": crawl_jobs_meta,
        "seed_path": str(seed_path),
    }

    issues_rows = (
        failed_pages
        + orphan_like[:200]
        + weakly_linked[:200]
        + breadcrumb_issues[:200]
        + breadcrumb_inconsistencies[:200]
        + nav_issues[:200]
        + menu_inconsistencies[:200]
        + weak_hubs[:200]
        + canonical_mismatches[:200]
        + linked_not_seeded_rows[:200]
        + technical_junk_rows[:200]
        + duplicate_title_rows[:200]
    )

    report_dir = Path(args.report_dir)
    report_dir.mkdir(parents=True, exist_ok=True)
    summary_path = report_dir / "summary.json"
    issues_csv_path = report_dir / "issues.csv"
    page_inventory_path = report_dir / "page-inventory.csv"
    report_path = report_dir / "navigation-audit-report.md"
    linked_not_seeded_path = report_dir / "linked-but-not-seeded.txt"

    write_json(summary_path, summary)
    linked_not_seeded_path.write_text("\n".join(linked_not_seeded_targets) + ("\n" if linked_not_seeded_targets else ""), encoding="utf-8")
    write_csv(
        page_inventory_path,
        page_inventory,
        [
            "url",
            "status_code",
            "success",
            "title",
            "path_depth",
            "incoming_internal_links",
            "outgoing_internal_links",
            "has_breadcrumbs",
            "nav_link_count",
            "internal_links_count",
            "canonical",
            "breadcrumb_signature",
            "menu_common_nav_missing",
        ],
    )
    write_csv(
        issues_csv_path,
        issues_rows,
        ["type", "url", "status_code", "detail"],
    )

    lines: list[str] = []
    lines.append(f"# Navigation Audit: {domain}")
    lines.append("")
    lines.append(f"Дата: `{summary['created_at']}`")
    lines.append("")
    lines.append("## Scope")
    lines.append("")
    lines.append(f"- Seed frontier: `{len(seed_urls)}` URL из sitemap/robots.")
    lines.append(f"- Crawled unique URLs: `{len(crawled_urls)}` (`{summary['coverage_pct']}%` seed frontier).")
    lines.append(f"- Successful pages: `{len(successful_pages)}`.")
    lines.append(f"- Internal link edges captured: `{len(links)}`.")
    lines.append(f"- Crawl jobs: `{', '.join(job['job_id'] for job in crawl_jobs_meta)}`.")
    lines.append("")
    lines.append("## Practical Plan")
    lines.append("")
    lines.append("- Seed source: sitemap index + robots.txt, как дешёвый и repeatable frontier.")
    lines.append("- Crawl order: pilot core pages -> hub/category coverage -> product sample expansion.")
    lines.append("- Resume model: каждый chunk = отдельный crawl job в skill cache; report rebuild не требует повторного fetch.")
    lines.append("- Reporting model: markdown/csv/json в отдельной `reports/...` директории; jobs/cache остаются внутри skill.")
    lines.append("")
    lines.append("## Coverage By Prefix")
    lines.append("")
    lines.append("| Prefix | Seed | Crawled | Backlog |")
    lines.append("|---|---:|---:|---:|")
    for pref, seed_count in seed_prefix_counts.most_common(15):
        lines.append(f"| `{pref}` | {seed_count} | {crawled_prefix_counts.get(pref, 0)} | {backlog_prefix_counts.get(pref, 0)} |")
    lines.append("")
    lines.append("## Findings")
    lines.append("")
    lines.append(f"1. `orphan-like / zero-inlink`: `{len(orphan_like)}` страниц среди crawled set не получили ни одной внутренней ссылки из других crawled pages.")
    lines.append(f"2. `weak inlinks`: `{len(weakly_linked)}` страниц имеют только одну внутреннюю ссылку в captured graph.")
    lines.append(f"3. `missing breadcrumbs`: `{len(breadcrumb_issues)}` страниц глубже первого уровня без заметного breadcrumb signal.")
    lines.append(f"4. `breadcrumb inconsistency`: `{len(breadcrumb_inconsistencies)}` section groups имеют несколько breadcrumb trail signatures.")
    lines.append(f"5. `weak nav template`: `{len(nav_issues)}` hub/category страниц имеют подозрительно слабый nav footprint относительно медианы `{nav_median}`.")
    lines.append(f"6. `menu inconsistency`: `{len(menu_inconsistencies)}` hub-like страниц теряют часть common nav targets.")
    lines.append(f"7. `weak hubs`: `{len(weak_hubs)}` hub/category страниц отдают меньше `5` уникальных внутренних ссылок.")
    lines.append(f"8. `linked-not-seeded`: `{len(linked_not_seeded_targets)}` внутренних URL линкуются сайтом, но отсутствуют в seed frontier.")
    lines.append(f"9. `technical junk`: `{len(technical_junk_rows)}` URL/страниц выглядят как query/php/service/document noise.")
    lines.append(f"10. `http errors / failed pages`: `{len(failed_pages)}`.")
    lines.append(f"11. `canonical mismatches`: `{len(canonical_mismatches)}`.")
    lines.append(f"12. `duplicate title groups`: `{len(duplicate_title_groups)}`.")
    lines.append("")

    def add_examples(title: str, rows: list[dict[str, Any]], limit: int = 10) -> None:
        lines.append(f"### {title}")
        lines.append("")
        if not rows:
            lines.append("- Не найдено в текущем покрытии.")
            lines.append("")
            return
        for row in rows[:limit]:
            detail = row.get("detail", "")
            suffix = f" — {detail}" if detail else ""
            lines.append(f"- `{row.get('url')}`{suffix}")
        lines.append("")

    add_examples("Zero-Inlink Examples", orphan_like)
    add_examples("Weakly Linked Examples", weakly_linked)
    add_examples("Missing Breadcrumbs Examples", breadcrumb_issues)
    add_examples("Breadcrumb Inconsistency Examples", breadcrumb_inconsistencies)
    add_examples("Weak Nav Template Examples", nav_issues)
    add_examples("Menu Inconsistency Examples", menu_inconsistencies)
    add_examples("Weak Hub Examples", weak_hubs)
    add_examples("Linked-Not-Seeded Examples", linked_not_seeded_rows)
    add_examples("Technical Junk Examples", technical_junk_rows)
    add_examples("Failed Pages", failed_pages)

    lines.append("## Structural Anti-Patterns")
    lines.append("")
    if structural_anti_patterns:
        for item in structural_anti_patterns:
            lines.append(f"- {item}")
    else:
        lines.append("- Явные structural anti-patterns не выделились на текущем покрытии.")
    lines.append("")

    lines.append("## Linked-Not-Seeded Zones")
    lines.append("")
    if linked_not_seeded_prefix_counts:
        for pref, count in linked_not_seeded_prefix_counts.most_common(10):
            lines.append(f"- `{pref}`: {count} URL")
    else:
        lines.append("- Не найдено в текущем покрытии.")
    lines.append("")

    lines.append("## Technical Junk By Prefix")
    lines.append("")
    if technical_junk_prefix_counts:
        for pref, count in technical_junk_prefix_counts.most_common(10):
            lines.append(f"- `{pref}`: {count} observations")
    else:
        lines.append("- Не найдено в текущем покрытии.")
    lines.append("")

    lines.append("## Remaining Frontier")
    lines.append("")
    lines.append(f"- Необойдённый backlog: `{len(backlog)}` URL.")
    if backlog_prefix_counts:
        for pref, count in backlog_prefix_counts.most_common(10):
            lines.append(f"- `{pref}`: {count} URL ещё не crawled.")
    lines.append("")
    lines.append("## Artifacts")
    lines.append("")
    lines.append(f"- Summary: `{summary_path}`")
    lines.append(f"- Issues CSV: `{issues_csv_path}`")
    lines.append(f"- Page inventory CSV: `{page_inventory_path}`")
    lines.append(f"- Linked but not seeded: `{linked_not_seeded_path}`")
    lines.append("")

    report_path.write_text("\n".join(lines) + "\n", encoding="utf-8")

    print(json.dumps(
        {
            "report_path": str(report_path),
            "summary_path": str(summary_path),
            "issues_csv_path": str(issues_csv_path),
            "page_inventory_path": str(page_inventory_path),
            "summary": summary,
        },
        ensure_ascii=False,
        indent=2,
    ))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

#!/usr/bin/env python3
"""Common helpers for the crawl4ai-seo skill."""

from __future__ import annotations

import hashlib
import json
import re
import uuid
from copy import deepcopy
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Iterable
from urllib.parse import urlparse


SCRIPT_DIR = Path(__file__).resolve().parent
SKILL_DIR = SCRIPT_DIR.parent
CONFIG_DIR = SKILL_DIR / "config"
CACHE_DIR = SKILL_DIR / "cache"
JOBS_DIR = CACHE_DIR / "jobs"
SITES_DIR = CACHE_DIR / "sites"
DEFAULTS_EXAMPLE_PATH = CONFIG_DIR / "defaults.example.json"
DEFAULTS_PATH = CONFIG_DIR / "defaults.json"
LEGACY_EXAMPLE_PATH = CONFIG_DIR / "config.example.json"
LEGACY_CONFIG_PATH = CONFIG_DIR / "config.json"


def utc_now_iso() -> str:
    return datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")


def deep_merge(base: dict, override: dict) -> dict:
    result = deepcopy(base)
    for key, value in override.items():
        if isinstance(value, dict) and isinstance(result.get(key), dict):
            result[key] = deep_merge(result[key], value)
        else:
            result[key] = value
    return result


def load_json(path: Path) -> dict:
    return json.loads(path.read_text(encoding="utf-8"))


def load_json_if_exists(path: Path) -> dict:
    if not path.exists():
        return {}
    return load_json(path)


def load_defaults() -> dict:
    defaults = {}
    if DEFAULTS_EXAMPLE_PATH.exists():
        defaults = load_json(DEFAULTS_EXAMPLE_PATH)
    elif LEGACY_EXAMPLE_PATH.exists():
        defaults = load_json(LEGACY_EXAMPLE_PATH)

    if DEFAULTS_PATH.exists():
        defaults = deep_merge(defaults, load_json(DEFAULTS_PATH))
    elif LEGACY_CONFIG_PATH.exists():
        defaults = deep_merge(defaults, load_json(LEGACY_CONFIG_PATH))

    return defaults


def resolve_skill_path(value: str | None) -> Path | None:
    if not value:
        return None
    path = Path(value)
    if path.is_absolute():
        return path
    return (SKILL_DIR / path).resolve()


def slugify(value: str) -> str:
    value = value.strip().lower()
    value = re.sub(r"^https?://", "", value)
    value = value.split("/", 1)[0]
    value = re.sub(r"[^a-z0-9]+", "-", value)
    value = value.strip("-")
    return value or "default"


def normalize_domain(value: str) -> str:
    if not value:
        return ""
    parsed = urlparse(value if "://" in value else f"https://{value}")
    return (parsed.netloc or parsed.path).lower().strip()


def normalize_url(value: str) -> str:
    value = value.strip()
    if not value:
        return value
    parsed = urlparse(value if "://" in value else f"https://{value}")
    scheme = parsed.scheme or "https"
    netloc = parsed.netloc or parsed.path
    path = parsed.path if parsed.netloc else ""
    normalized = f"{scheme}://{netloc}{path or '/'}"
    if parsed.query:
        normalized += f"?{parsed.query}"
    return normalized


def prune_none(value: Any) -> Any:
    if isinstance(value, dict):
        result = {}
        for key, item in value.items():
            cleaned = prune_none(item)
            if cleaned is not None:
                result[key] = cleaned
        return result
    if isinstance(value, list):
        return [prune_none(item) for item in value]
    return value


def load_launch_params(path: str | None) -> dict:
    if not path:
        return {}
    return prune_none(load_json(Path(path)))


def normalize_launch_params(payload: dict) -> dict:
    normalized = prune_none(deepcopy(payload))
    target = normalized.setdefault("target", {})
    domain = target.get("domain")
    if domain:
        target["domain"] = normalize_url(str(domain))
    site_slug = target.get("site_slug")
    if site_slug:
        target["site_slug"] = slugify(str(site_slug))
    elif domain:
        target["site_slug"] = slugify(normalize_domain(str(domain)))
    return normalized


def resolve_effective_config(launch_params: dict) -> dict:
    effective = deep_merge(load_defaults(), normalize_launch_params(launch_params))
    target = effective.setdefault("target", {})
    domain = target.get("domain")
    if domain:
        target["domain"] = normalize_url(str(domain))
    target["site_slug"] = slugify(str(target.get("site_slug") or normalize_domain(str(domain or "")) or "manual-list"))
    return effective


def make_job_id(site_slug: str) -> str:
    timestamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
    suffix = uuid.uuid4().hex[:8]
    site_part = slugify(site_slug) or "site"
    return f"{timestamp}-{site_part}-{suffix}"


def job_dir(job_id: str) -> Path:
    return JOBS_DIR / job_id


def site_ref_path(site_slug: str, job_id: str) -> Path:
    return SITES_DIR / slugify(site_slug) / "jobs" / f"{job_id}.json"


def build_job_paths(job_id: str, site_slug: str) -> dict[str, str]:
    root = job_dir(job_id)
    return {
        "job_dir": str(root),
        "launch_params_path": str(root / "launch_params.json"),
        "resolved_config_path": str(root / "resolved_config.json"),
        "manifest_path": str(root / "manifest.json"),
        "seed_path": str(root / "seed.json"),
        "pages_path": str(root / "pages.ndjson"),
        "links_path": str(root / "links.ndjson"),
        "summary_path": str(root / "summary.json"),
        "markdown_dir": str(root / "markdown"),
        "site_ref_path": str(site_ref_path(site_slug, job_id)),
    }


def ensure_job(launch_params: dict, job_id: str | None = None) -> dict:
    resolved = resolve_effective_config(launch_params)
    site_slug = resolved.get("target", {}).get("site_slug") or "manual-list"
    current_job_id = job_id or make_job_id(site_slug)
    root = job_dir(current_job_id)
    root.mkdir(parents=True, exist_ok=True)

    paths = build_job_paths(current_job_id, site_slug)
    created_at = utc_now_iso()
    normalized_launch = normalize_launch_params(launch_params)
    resolved_payload = deep_merge(
        resolved,
        {
            "job_meta": {
                "job_id": current_job_id,
                "created_at": created_at,
            },
            "paths": paths,
        },
    )
    manifest = {
        "job_id": current_job_id,
        "created_at": created_at,
        "updated_at": created_at,
        "status": "initialized",
        "target": resolved_payload.get("target", {}),
        "job": resolved_payload.get("job", {}),
        "paths": paths,
        "artifacts": {},
    }

    write_json(Path(paths["launch_params_path"]), normalized_launch)
    write_json(Path(paths["resolved_config_path"]), resolved_payload)
    write_json(Path(paths["manifest_path"]), manifest)
    write_json(
        Path(paths["site_ref_path"]),
        {
            "job_id": current_job_id,
            "site_slug": site_slug,
            "target": resolved_payload.get("target", {}),
            "job": resolved_payload.get("job", {}),
            "job_dir": paths["job_dir"],
            "manifest_path": paths["manifest_path"],
            "created_at": created_at,
        },
    )
    return load_job(current_job_id)


def load_job(job_id: str) -> dict:
    root = job_dir(job_id)
    resolved_path = root / "resolved_config.json"
    manifest_path = root / "manifest.json"
    launch_path = root / "launch_params.json"
    if not resolved_path.exists():
        raise FileNotFoundError(f"Job not found: {job_id}")
    return {
        "job_id": job_id,
        "job_dir": root,
        "launch_params": load_json_if_exists(launch_path),
        "resolved": load_json(resolved_path),
        "manifest": load_json_if_exists(manifest_path),
    }


def update_manifest(job_id: str, patch: dict) -> dict:
    current = load_job(job_id)
    manifest = deep_merge(current.get("manifest", {}), patch)
    manifest["job_id"] = job_id
    manifest["updated_at"] = utc_now_iso()
    write_json(Path(current["resolved"]["paths"]["manifest_path"]), manifest)
    return manifest


def url_fingerprint(value: str) -> str:
    return hashlib.sha1(normalize_url(value).encode("utf-8")).hexdigest()[:10]


def url_file_slug(value: str) -> str:
    normalized = normalize_url(value)
    return f"{slugify(normalized)[:80]}-{url_fingerprint(normalized)}"


def write_json(path: Path, payload: dict) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(
        json.dumps(payload, ensure_ascii=False, indent=2) + "\n",
        encoding="utf-8",
    )


def write_ndjson(path: Path, rows: Iterable[dict]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    with path.open("w", encoding="utf-8") as handle:
        for row in rows:
            handle.write(json.dumps(row, ensure_ascii=False) + "\n")


def load_seed_payload(path: Path) -> dict:
    return load_json(path)


def load_seed_urls(path: Path) -> list[str]:
    payload = load_seed_payload(path)
    urls = payload.get("urls") or []
    return [normalize_url(url) for url in urls if str(url).strip()]


def load_ndjson(path: Path) -> list[dict]:
    rows: list[dict] = []
    with path.open("r", encoding="utf-8") as handle:
        for line in handle:
            line = line.strip()
            if line:
                rows.append(json.loads(line))
    return rows


def print_preview(items: list[str], limit: int = 20) -> None:
    for idx, item in enumerate(items[:limit], start=1):
        print(f"{idx:>2}. {item}")
    if len(items) > limit:
        print(f"... and {len(items) - limit} more")

#!/usr/bin/env python3
"""Compare normalized page records from pages.ndjson."""

from __future__ import annotations

import argparse
import json
from pathlib import Path

from common import load_job, load_ndjson


def main() -> int:
    parser = argparse.ArgumentParser(description="Preview normalized page records for SEO comparison.")
    parser.add_argument("--pages", help="Path to pages.ndjson")
    parser.add_argument("--job-id", help="Read pages.ndjson from an existing job")
    parser.add_argument("--top", type=int, default=20, help="How many rows to preview")
    parser.add_argument("--json", action="store_true", help="Print JSON instead of a plain-text table")
    args = parser.parse_args()

    pages_path = args.pages
    if args.job_id:
        job = load_job(args.job_id)
        pages_path = job["resolved"]["paths"]["pages_path"]

    if not pages_path:
        parser.error("Provide --pages or --job-id.")

    rows = load_ndjson(Path(pages_path))
    preview = rows[: args.top]

    if args.json:
        print(json.dumps(preview, ensure_ascii=False, indent=2))
        return 0

    print("URL\tSuccess\tStatus\tTitle\tH1\tInternal\tExternal")
    for row in preview:
        print(
            "\t".join(
                [
                    str(row.get("url") or ""),
                    str(row.get("success")),
                    str(row.get("status_code") or ""),
                    str(row.get("title") or "")[:80],
                    str(row.get("h1") or "")[:80],
                    str(row.get("internal_links_count") or 0),
                    str(row.get("external_links_count") or 0),
                ]
            )
        )

    if len(rows) > args.top:
        print(f"... and {len(rows) - args.top} more rows in {pages_path}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.10"
# dependencies = ["crawl4ai>=0.8,<0.9", "beautifulsoup4>=4.12,<5"]
# ///
"""Batch crawl URLs via Crawl4AI and save normalized SEO research outputs."""

from __future__ import annotations

import argparse
import asyncio
from html.parser import HTMLParser
from pathlib import Path
from typing import Any
from urllib.parse import urljoin, urlparse

from bs4 import BeautifulSoup

from common import (
    deep_merge,
    ensure_job,
    load_job,
    load_launch_params,
    load_seed_payload,
    load_seed_urls,
    normalize_domain,
    normalize_launch_params,
    normalize_url,
    print_preview,
    utc_now_iso,
    update_manifest,
    url_file_slug,
    write_json,
    write_ndjson,
)


def first_non_empty(*values: Any) -> str | None:
    for value in values:
        if value is None:
            continue
        text = str(value).strip()
        if text:
            return text
    return None


class SEOHTMLParser(HTMLParser):
    def __init__(self, base_url: str) -> None:
        super().__init__(convert_charrefs=True)
        self.base_url = base_url
        self.title: str | None = None
        self.canonical: str | None = None
        self.meta: dict[str, str] = {}
        self.og: dict[str, str] = {}
        self.h1: list[str] = []
        self.headings: list[str] = []
        self._title_chunks: list[str] = []
        self._heading_chunks: list[str] = []
        self._active_heading: str | None = None
        self._inside_title = False

    def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None:
        attrs_map = {str(key).lower(): value for key, value in attrs if key}
        if tag == "title":
            self._inside_title = True
            self._title_chunks = []
            return
        if tag == "meta":
            content = first_non_empty(attrs_map.get("content"))
            if not content:
                return
            name = first_non_empty(attrs_map.get("name"))
            prop = first_non_empty(attrs_map.get("property"))
            if name and name.lower() not in self.meta:
                self.meta[name.lower()] = content
            if prop and prop.lower().startswith("og:") and prop.lower() not in self.og:
                self.og[prop.lower()] = content
            return
        if tag == "link":
            href = first_non_empty(attrs_map.get("href"))
            rel = first_non_empty(attrs_map.get("rel"))
            if href and rel:
                rel_values = {item.strip().lower() for item in rel.split() if item.strip()}
                if "canonical" in rel_values and not self.canonical:
                    self.canonical = normalize_url(urljoin(self.base_url, href))
            return
        if tag in {"h1", "h2", "h3", "h4", "h5", "h6"}:
            self._active_heading = tag
            self._heading_chunks = []

    def handle_endtag(self, tag: str) -> None:
        if tag == "title" and self._inside_title:
            title = " ".join(chunk.strip() for chunk in self._title_chunks if chunk.strip()).strip()
            if title and not self.title:
                self.title = title
            self._inside_title = False
            self._title_chunks = []
            return
        if tag == self._active_heading:
            heading = " ".join(chunk.strip() for chunk in self._heading_chunks if chunk.strip()).strip()
            if heading:
                self.headings.append(heading)
                if tag == "h1":
                    self.h1.append(heading)
            self._active_heading = None
            self._heading_chunks = []

    def handle_data(self, data: str) -> None:
        if self._inside_title:
            self._title_chunks.append(data)
        if self._active_heading:
            self._heading_chunks.append(data)


def extract_html_signals(page_url: str, html: Any) -> dict[str, Any]:
    if not isinstance(html, str) or not html.strip():
        return {
            "title": None,
            "description": None,
            "keywords": None,
            "robots": None,
            "canonical": None,
            "h1": [],
            "headings": [],
            "og": {},
        }

    parser = SEOHTMLParser(page_url)
    try:
        parser.feed(html)
        parser.close()
    except Exception:
        return {
            "title": None,
            "description": None,
            "keywords": None,
            "robots": None,
            "canonical": None,
            "h1": [],
            "headings": [],
            "og": {},
        }

    return {
        "title": parser.title,
        "description": parser.meta.get("description"),
        "keywords": parser.meta.get("keywords"),
        "robots": parser.meta.get("robots"),
        "canonical": parser.canonical,
        "h1": parser.h1,
        "headings": parser.headings,
        "og": parser.og,
    }


def extract_markdown_payload(markdown_obj: Any) -> tuple[str | None, str | None]:
    if markdown_obj is None:
        return None, None
    if isinstance(markdown_obj, str):
        return markdown_obj, None
    raw = getattr(markdown_obj, "raw_markdown", None)
    fit = getattr(markdown_obj, "fit_markdown", None)
    if raw is None and isinstance(markdown_obj, dict):
        raw = markdown_obj.get("raw_markdown") or markdown_obj.get("markdown")
        fit = markdown_obj.get("fit_markdown")
    return raw, fit


def extract_text_list(value: Any) -> list[str]:
    if value is None:
        return []
    if isinstance(value, list):
        return [str(item).strip() for item in value if str(item).strip()]
    if isinstance(value, dict):
        items: list[str] = []
        for item in value.values():
            if isinstance(item, list):
                items.extend(str(x).strip() for x in item if str(x).strip())
            elif item:
                items.append(str(item).strip())
        return items
    return [str(value).strip()]


def extract_link_items(links: Any, key: str) -> list[dict]:
    if not isinstance(links, dict):
        return []
    value = links.get(key)
    if not isinstance(value, list):
        return []
    return [item for item in value if isinstance(item, dict)]


def url_path_depth(value: str) -> int:
    path = urlparse(value).path.strip("/")
    if not path:
        return 0
    return len([part for part in path.split("/") if part])


def dedupe_texts(values: list[str], limit: int | None = None) -> list[str]:
    unique: list[str] = []
    seen: set[str] = set()
    for value in values:
        text = str(value).strip()
        if not text or text in seen:
            continue
        seen.add(text)
        unique.append(text)
        if limit is not None and len(unique) >= limit:
            break
    return unique


def extract_navigation_signals(page_url: str, html: Any) -> dict[str, Any]:
    payload = {
        "path_depth": url_path_depth(page_url),
        "has_breadcrumbs": False,
        "breadcrumb_texts": [],
        "breadcrumb_urls": [],
        "nav_block_count": 0,
        "nav_link_count": 0,
        "nav_urls_sample": [],
        "nav_texts_sample": [],
    }
    if not isinstance(html, str) or not html.strip():
        return payload

    try:
        soup = BeautifulSoup(html, "html.parser")
    except Exception:
        return payload

    page_domain = normalize_domain(page_url)

    def is_internal(url: str) -> bool:
        return normalize_domain(url) == page_domain

    def normalize_href(value: str | None) -> str | None:
        if not value:
            return None
        href = value.strip()
        if not href or href.startswith("#") or href.lower().startswith("javascript:"):
            return None
        return normalize_url(urljoin(page_url, href))

    def attr_blob(tag: Any) -> str:
        parts: list[str] = []
        for attr in ("class", "id", "role", "aria-label", "data-role"):
            value = tag.attrs.get(attr)
            if isinstance(value, list):
                parts.extend(str(item).lower() for item in value if str(item).strip())
            elif value:
                parts.append(str(value).lower())
        return " ".join(parts)

    breadcrumb_candidates: list[Any] = []
    nav_candidates: list[Any] = []
    seen_nodes: set[int] = set()

    for tag in soup.find_all(True):
        marker = attr_blob(tag)
        if "breadcrumb" in marker:
            identity = id(tag)
            if identity not in seen_nodes:
                breadcrumb_candidates.append(tag)
                seen_nodes.add(identity)

        is_nav_tag = tag.name in {"nav", "header"}
        has_nav_marker = (
            "navigation" in marker
            or " menu" in f" {marker}"
            or " nav" in f" {marker}"
            or "header" in marker
        )
        if is_nav_tag or has_nav_marker:
            nav_candidates.append(tag)

    breadcrumb_texts: list[str] = []
    breadcrumb_urls: list[str] = []
    for node in breadcrumb_candidates:
        texts = [text.strip() for text in node.stripped_strings if text.strip()]
        anchors = [normalize_href(anchor.get("href")) for anchor in node.find_all("a", href=True)]
        breadcrumb_texts.extend(texts[:10])
        breadcrumb_urls.extend(url for url in anchors if url and is_internal(url))

    breadcrumb_texts = dedupe_texts(breadcrumb_texts, limit=10)
    breadcrumb_urls = dedupe_texts(breadcrumb_urls, limit=10)
    payload["has_breadcrumbs"] = len(breadcrumb_texts) >= 2 or len(breadcrumb_urls) >= 1
    payload["breadcrumb_texts"] = breadcrumb_texts
    payload["breadcrumb_urls"] = breadcrumb_urls

    nav_urls: list[str] = []
    nav_texts: list[str] = []
    nav_block_count = 0
    for node in nav_candidates:
        local_urls: list[str] = []
        local_texts: list[str] = []
        for anchor in node.find_all("a", href=True):
            url = normalize_href(anchor.get("href"))
            if not url or not is_internal(url):
                continue
            local_urls.append(url)
            text = " ".join(anchor.stripped_strings).strip()
            if text:
                local_texts.append(text)
        local_urls = dedupe_texts(local_urls)
        local_texts = dedupe_texts(local_texts)
        if len(local_urls) >= 2:
            nav_block_count += 1
            nav_urls.extend(local_urls)
            nav_texts.extend(local_texts)

    payload["nav_block_count"] = nav_block_count
    payload["nav_urls_sample"] = dedupe_texts(nav_urls, limit=15)
    payload["nav_texts_sample"] = dedupe_texts(nav_texts, limit=15)
    payload["nav_link_count"] = len(dedupe_texts(nav_urls))
    return payload


def normalize_page(result: Any, run_id: str, markdown_root: Path, store_markdown: bool) -> tuple[dict, list[dict]]:
    metadata = getattr(result, "metadata", None) or {}
    links = getattr(result, "links", None) or {}
    html = getattr(result, "html", None) or getattr(result, "cleaned_html", None)
    markdown_obj = getattr(result, "markdown", None)
    raw_markdown, fit_markdown = extract_markdown_payload(markdown_obj)

    page_url = normalize_url(str(getattr(result, "url", "") or ""))
    page_domain = normalize_domain(page_url)
    html_signals = extract_html_signals(page_url, html)
    navigation_signals = extract_navigation_signals(page_url, html)

    markdown_path = None
    fit_markdown_path = None
    if store_markdown and raw_markdown:
        markdown_path = markdown_root / f"{url_file_slug(page_url)}.md"
        markdown_path.write_text(raw_markdown, encoding="utf-8")
    if store_markdown and fit_markdown:
        fit_markdown_path = markdown_root / f"{url_file_slug(page_url)}.fit.md"
        fit_markdown_path.write_text(fit_markdown, encoding="utf-8")

    internal_links = extract_link_items(links, "internal")
    external_links = extract_link_items(links, "external")
    headings = extract_text_list(metadata.get("headings"))
    h1_candidates = extract_text_list(metadata.get("h1") or metadata.get("h1_tags"))
    if not headings:
        headings = html_signals["headings"]
    if not h1_candidates:
        h1_candidates = html_signals["h1"]

    canonical = first_non_empty(metadata.get("canonical"), html_signals["canonical"])
    canonical = normalize_url(canonical) if canonical else None
    og_tags = html_signals["og"]

    page_row = {
        "run_id": run_id,
        "url": page_url,
        "domain": page_domain,
        "path_depth": navigation_signals["path_depth"],
        "success": bool(getattr(result, "success", False)),
        "status_code": getattr(result, "status_code", None),
        "title": first_non_empty(metadata.get("title"), html_signals["title"]),
        "description": first_non_empty(metadata.get("description"), html_signals["description"]),
        "canonical": canonical,
        "keywords": first_non_empty(metadata.get("keywords"), html_signals["keywords"]),
        "robots": first_non_empty(metadata.get("robots"), html_signals["robots"]),
        "og_title": first_non_empty(metadata.get("og:title"), og_tags.get("og:title")),
        "og_description": first_non_empty(metadata.get("og:description"), og_tags.get("og:description")),
        "og_image": first_non_empty(metadata.get("og:image"), og_tags.get("og:image")),
        "h1": h1_candidates[0] if h1_candidates else None,
        "headings": headings,
        "word_count": metadata.get("word_count"),
        "has_breadcrumbs": navigation_signals["has_breadcrumbs"],
        "breadcrumb_texts": navigation_signals["breadcrumb_texts"],
        "breadcrumb_urls": navigation_signals["breadcrumb_urls"],
        "nav_block_count": navigation_signals["nav_block_count"],
        "nav_link_count": navigation_signals["nav_link_count"],
        "nav_urls_sample": navigation_signals["nav_urls_sample"],
        "nav_texts_sample": navigation_signals["nav_texts_sample"],
        "internal_links_count": len(internal_links),
        "external_links_count": len(external_links),
        "markdown_path": str(markdown_path) if markdown_path else None,
        "fit_markdown_path": str(fit_markdown_path) if fit_markdown_path else None,
        "error": getattr(result, "error_message", None),
    }

    link_rows: list[dict] = []
    for kind, items in (("internal", internal_links), ("external", external_links)):
        for item in items or []:
            if not isinstance(item, dict):
                continue
            href = item.get("href")
            if not href:
                continue
            target_url = normalize_url(str(href))
            link_rows.append(
                {
                    "run_id": run_id,
                    "source_url": page_url,
                    "target_url": target_url,
                    "target_domain": normalize_domain(target_url),
                    "kind": kind,
                    "anchor_text": item.get("text"),
                    "nofollow": bool(item.get("nofollow")),
                    "same_domain": normalize_domain(target_url) == page_domain,
                }
            )

    return page_row, link_rows


async def crawl_urls(crawler: Any, urls: list[str], crawl_cfg: Any, max_concurrency: int) -> list[Any]:
    if hasattr(crawler, "arun"):
        semaphore = asyncio.Semaphore(max(1, max_concurrency))

        async def worker(url: str) -> Any:
            async with semaphore:
                return await crawler.arun(url=url, config=crawl_cfg)

        return list(await asyncio.gather(*(worker(url) for url in urls)))

    return list(await crawler.arun_many(urls=urls, config=crawl_cfg))


def build_cli_launch_params(args: argparse.Namespace) -> dict:
    return normalize_launch_params(
        {
            "target": {
                "domain": args.domain,
                "site_slug": args.site,
            },
            "job": {
                "project": args.project,
                "label": args.label,
            },
            "crawl": {
                "max_urls": args.limit,
                "cache_mode": "bypass" if args.fresh else None,
            },
        }
    )


async def run_crawl(args: argparse.Namespace) -> int:
    from crawl4ai import AsyncWebCrawler, BrowserConfig, CacheMode, CrawlerRunConfig

    existing_job = None
    if args.job_id:
        try:
            existing_job = load_job(args.job_id)
        except FileNotFoundError:
            existing_job = None

    seed_payload = None
    seed_path = Path(args.seed) if args.seed else None
    if existing_job and not seed_path:
        candidate = Path(existing_job["resolved"]["paths"]["seed_path"])
        if candidate.exists():
            seed_path = candidate

    urls = [normalize_url(url) for url in args.url]
    if seed_path:
        seed_payload = load_seed_payload(seed_path)
        urls.extend(load_seed_urls(seed_path))

    urls = list(dict.fromkeys(urls))
    if not urls:
        raise SystemExit("No URLs to crawl. Use --seed or one or more --url.")

    if existing_job:
        context = existing_job
    else:
        derived_domain = args.domain
        if not derived_domain and seed_payload:
            derived_domain = seed_payload.get("target", {}).get("domain") or seed_payload.get("domain")
        if not derived_domain:
            derived_domain = normalize_url(urls[0])
        launch_params = deep_merge(load_launch_params(args.params), build_cli_launch_params(args))
        launch_params = deep_merge(
            launch_params,
            {
                "target": {
                    "domain": derived_domain,
                    "site_slug": args.site or normalize_domain(derived_domain),
                }
            },
        )
        context = ensure_job(launch_params, job_id=args.job_id)

    config = context["resolved"]
    domain = config.get("target", {}).get("domain") or args.domain or normalize_domain(urls[0])
    max_urls = int(config.get("crawl", {}).get("max_urls", 100))
    urls = urls[:max_urls]

    run_id = context["job_id"]
    markdown_root = Path(config["paths"]["markdown_dir"])
    markdown_root.mkdir(parents=True, exist_ok=True)

    browser_cfg = BrowserConfig(
        headless=bool(config.get("browser", {}).get("headless", True)),
        verbose=bool(config.get("browser", {}).get("verbose", False)),
    )

    cache_mode_name = "BYPASS" if args.fresh else str(config.get("crawl", {}).get("cache_mode", "enabled")).upper()
    cache_mode = getattr(CacheMode, cache_mode_name, None)
    if cache_mode is None:
        cache_mode = getattr(CacheMode, "ENABLED")

    crawl_cfg = CrawlerRunConfig(
        cache_mode=cache_mode,
        word_count_threshold=int(config.get("crawl", {}).get("word_count_threshold", 20)),
        page_timeout=int(config.get("crawl", {}).get("page_timeout_ms", 45000)),
    )
    max_concurrency = int(config.get("crawl", {}).get("max_concurrency", 5))

    pages: list[dict] = []
    links: list[dict] = []

    async with AsyncWebCrawler(config=browser_cfg) as crawler:
        results = await crawl_urls(crawler, urls, crawl_cfg, max_concurrency=max_concurrency)

    for result in results:
        page_row, link_rows = normalize_page(
            result=result,
            run_id=run_id,
            markdown_root=markdown_root,
            store_markdown=bool(config.get("output", {}).get("store_markdown", True)),
        )
        pages.append(page_row)
        links.extend(link_rows)

    pages_path = Path(config["paths"]["pages_path"])
    links_path = Path(config["paths"]["links_path"])
    summary_path = Path(config["paths"]["summary_path"])
    manifest_path = Path(config["paths"]["manifest_path"])

    write_ndjson(pages_path, pages)
    write_ndjson(links_path, links)

    summary = {
        "job_id": run_id,
        "target": config.get("target", {}),
        "job": config.get("job", {}),
        "domain": normalize_domain(str(domain)),
        "run_id": run_id,
        "created_at": utc_now_iso(),
        "requested_urls": len(urls),
        "crawled_ok": sum(1 for page in pages if page["success"]),
        "crawled_failed": sum(1 for page in pages if not page["success"]),
        "internal_links": sum(1 for link in links if link["kind"] == "internal"),
        "external_links": sum(1 for link in links if link["kind"] == "external"),
        "pages_path": str(pages_path),
        "links_path": str(links_path),
    }

    write_json(summary_path, summary)
    update_manifest(
        context["job_id"],
        {
            "status": "crawl_completed",
            "artifacts": {
                "seed_path": str(seed_path) if seed_path else None,
                "pages_path": str(pages_path),
                "links_path": str(links_path),
                "summary_path": str(summary_path),
            },
            "crawl": {
                "requested_urls": len(urls),
                "seed_path": str(seed_path) if seed_path else None,
                "urls": urls,
            },
            "summary": summary,
        },
    )

    preview_rows = int(config.get("output", {}).get("preview_rows", 20))
    print(f"Job completed: {run_id}")
    print(f"Manifest:      {manifest_path}")
    print(f"Summary:       {summary_path}")
    print(f"Pages:         {pages_path}")
    print(f"Links:         {links_path}")
    print("")
    preview = [page["url"] for page in pages]
    print_preview(preview, limit=preview_rows)
    return 0


def main() -> int:
    parser = argparse.ArgumentParser(description="Batch crawl URLs with Crawl4AI.")
    parser.add_argument("--params", help="Path to JSON launch params")
    parser.add_argument("--job-id", help="Existing job_id to reuse, or new one to create")
    parser.add_argument("--site", help="Optional site slug override")
    parser.add_argument("--project", help="Optional project/client label for this job")
    parser.add_argument("--label", help="Optional free-form job label")
    parser.add_argument("--domain", help="Domain slug used in cache layout.")
    parser.add_argument("--seed", help="Path to seed.json produced by seed_urls.py.")
    parser.add_argument("--url", action="append", default=[], help="URL to crawl. Repeatable.")
    parser.add_argument("--limit", type=int, help="Override crawl.max_urls.")
    parser.add_argument("--fresh", action="store_true", help="Bypass Crawl4AI cache for this run.")
    args = parser.parse_args()
    return asyncio.run(run_crawl(args))


if __name__ == "__main__":
    raise SystemExit(main())