fetch_url_inspection.py

"""URL Inspection API でインデックス状況を取得 → inspect_metrics に upsert。

- searchconsole v1 の urlInspection().index().inspect(body=...) を使う。
- 1リクエスト1URL（まとめて検査はできない）。sitemap.xml のURLを1本ずつ検査する。
- 既存の `webmasters.readonly` スコープで動く（追加の認証は不要）。
- 読み取り専用（検査するだけで、インデックス登録のリクエストは送らない）。
- 部分失敗を許容: 1URLの検査失敗は飲み込んで継続。全滅時のみ例外送出。

fetch のシグネチャは fetch_gsc.fetch と同一（main.py の run_source から呼ばれる）。
service は fetch_gsc.build_service で作った searchconsole サービスをそのまま流用する。
"""

from __future__ import annotations

import logging
import sqlite3
import time
import urllib.request
import xml.etree.ElementTree as ET
from datetime import date
from typing import Any

import common

# sitemap.xml の名前空間
SITEMAP_NS = {"sm": "http://www.sitemaps.org/schemas/sitemap/0.9"}
DEFAULT_DELAY = 0.15  # 各検査後の小スリープ（API のレート上限を超えないため）


def load_target_urls(site_cfg: dict, cfg: dict, logger: logging.Logger) -> list[str]:
    """sitemap.xml から検査対象URLを取得。sitemapindex / urlset 両対応・重複除去。"""
    sitemap_url = site_cfg.get("sitemap_url")
    if not sitemap_url:
        return []
    req = urllib.request.Request(
        sitemap_url, headers={"User-Agent": "analytics-pipeline/1.0 (+url-inspection)"}
    )
    with urllib.request.urlopen(req, timeout=30) as resp:  # noqa: S310 - 設定した自サイトのsitemap
        xml = resp.read()
    root = ET.fromstring(xml)

    urls: list[str] = []
    if root.tag.lower().endswith("sitemapindex"):
        # サイトマップインデックス → 子サイトマップを再帰取得
        for loc in root.findall(".//sm:sitemap/sm:loc", SITEMAP_NS):
            if loc.text:
                child = dict(site_cfg, sitemap_url=loc.text.strip())
                urls.extend(load_target_urls(child, cfg, logger))
    else:
        for loc in root.findall(".//sm:url/sm:loc", SITEMAP_NS):
            if loc.text:
                urls.append(loc.text.strip())

    # 重複除去（順序保持）
    seen: set[str] = set()
    uniq = [u for u in urls if not (u in seen or seen.add(u))]
    return uniq


def _inspect(service: Any, site_url: str, url: str) -> dict:
    # Python の discovery クライアントでのメソッド名は inspect()
    # （REST: POST urlInspection/index:inspect）。
    return (
        service.urlInspection()
        .index()
        .inspect(body={"inspectionUrl": url, "siteUrl": site_url})
        .execute()
    )


def fetch(
    conn: sqlite3.Connection,
    service: Any,
    site_cfg: dict,
    cfg: dict,
    today: date,
    logger: logging.Logger,
) -> int:
    """1サイト分のURL Inspectionを実行 → inspect_metrics に upsert。書込行数を返す。"""
    site = site_cfg["site"]
    site_url = site_cfg["gsc_site_url"]
    domain = site_cfg["domain"]
    inspected = common.ymd(today)
    max_retries = cfg.get("retry_max", 3)
    base_delay = cfg.get("retry_base_delay", 2.0)
    delay = cfg.get("inspect_request_delay", DEFAULT_DELAY)

    urls = load_target_urls(site_cfg, cfg, logger)
    max_urls = cfg.get("inspect_max_urls")  # テスト用上限（None=全件）
    if max_urls:
        urls = urls[: int(max_urls)]
    if not urls:
        logger.info("[%s][url_inspection] 検査対象URLが0件（sitemap_url未設定？）。スキップ。", site)
        return 0
    logger.info("[%s][url_inspection] %d URL を検査します", site, len(urls))

    out = []
    errors = 0
    for i, u in enumerate(urls, 1):
        try:
            resp = common.with_retry(
                _inspect, service, site_url, u,
                max_retries=max_retries, base_delay=base_delay, logger=logger,
            )
        except Exception as exc:  # noqa: BLE001 - 1URL失敗は飲み込んで継続
            errors += 1
            logger.warning("[%s][url_inspection] 検査失敗 %s: %s", site, u, exc)
            time.sleep(delay)
            continue

        idx = (resp.get("inspectionResult") or {}).get("indexStatusResult") or {}
        out.append([
            site, inspected, common.normalize_path(u, domain),
            idx.get("coverageState"), idx.get("verdict"),
            idx.get("robotsTxtState"), idx.get("indexingState"),
            idx.get("lastCrawlTime"), idx.get("googleCanonical"),
            idx.get("userCanonical"),
        ])
        time.sleep(delay)
        if i % 50 == 0:
            logger.info("[%s][url_inspection] %d/%d 検査済", site, i, len(urls))

    if not out and urls:
        raise RuntimeError(f"全URL({len(urls)})の検査に失敗しました")

    n = common.upsert(
        conn, "inspect_metrics",
        ["site", "inspected_date", "page_path", "coverage_state", "verdict",
         "robots_state", "indexing_state", "last_crawl_time",
         "google_canonical", "user_canonical"],
        out,
    )
    conn.commit()
    if errors:
        logger.warning("[%s][url_inspection] %d/%d URLが検査失敗（スキップ）", site, errors, len(urls))
    logger.info("[%s][url_inspection] inspect_metrics: %d 行", site, n)
    return n