AI SEOワークフローにおける検索意図ドリフトモニタリング

from dataclasses import dataclass

@dataclass(frozen=True)
class QueryCohort:
    cohort_id: str
    market: str
    language: str
    device: str
    canonical_topic: str
    queries: tuple[str, ...]
    owned_page_group: tuple[str, ...]
    business_purpose: str

cohort = QueryCohort(
    cohort_id="captcha-api-comparison-us-en-desktop",
    market="US",
    language="en",
    device="DESKTOP",
    canonical_topic="captcha api comparison",
    queries=(
        "captcha api comparison",
        "compare captcha solving api",
        "captcha service comparison",
    ),
    owned_page_group=("/products/", "/blog/ai/"),
    business_purpose="content planning",
)

from googleapiclient.discovery import build
from google.oauth2 import service_account

SCOPES = ["https://www.googleapis.com/auth/webmasters.readonly"]

credentials = service_account.Credentials.from_service_account_file(
    "search-console-reader.json",
    scopes=SCOPES,
)
service = build("searchconsole", "v1", credentials=credentials)


def search_console_rows(
    site_url: str,
    start_date: str,
    end_date: str,
    row_limit: int = 25000,
) -> list[dict]:
    body = {
        "startDate": start_date,
        "endDate": end_date,
        "dimensions": ["query", "page", "country", "device"],
        "rowLimit": row_limit,
        "dataState": "final",
    }
    response = service.searchanalytics().query(
        siteUrl=site_url,
        body=body,
    ).execute()
    return response.get("rows", [])

from dataclasses import dataclass

@dataclass
class SearchPerformance:
    query: str
    page: str
    country: str
    device: str
    clicks: float
    impressions: float
    ctr: float
    position: float
    start_date: str
    end_date: str
    extraction_id: str
    coverage_note: str


def normalize_sc_row(row: dict, start_date: str, end_date: str, run_id: str):
    keys = row.get("keys") or []
    if len(keys) != 4:
        raise ValueError("Unexpected Search Console dimensions")

    return SearchPerformance(
        query=keys[0],
        page=keys[1],
        country=keys[2],
        device=keys[3],
        clicks=float(row.get("clicks", 0)),
        impressions=float(row.get("impressions", 0)),
        ctr=float(row.get("ctr", 0)),
        position=float(row.get("position", 0)),
        start_date=start_date,
        end_date=end_date,
        extraction_id=run_id,
        coverage_note="API may prioritize top rows; not guaranteed exhaustive",
    )

from dataclasses import dataclass, field
from datetime import datetime, timezone

@dataclass
class SerpSnapshot:
    cohort_id: str
    query: str
    market: str
    language: str
    device: str
    source: str
    collection_id: str
    organic_results: list[dict]
    features: list[str]
    evidence_hash: str
    collected_at: str = field(
        default_factory=lambda: datetime.now(timezone.utc).isoformat()
    )

INTENT_LABELS = {
    "INFORMATIONAL_GUIDE",
    "TROUBLESHOOTING",
    "COMMERCIAL_COMPARISON",
    "TRANSACTIONAL_CATEGORY",
    "TRANSACTIONAL_PRODUCT",
    "NAVIGATIONAL_BRAND",
    "LOCAL_VISIT",
    "NEWS_FRESHNESS",
    "VIDEO_LEARNING",
    "COMMUNITY_DISCUSSION",
    "MIXED",
}

from urllib.parse import urlparse


def classify_archetype(result: dict) -> str:
    url = result.get("url", "")
    title = result.get("title", "").lower()
    path = urlparse(url).path.lower()

    if any(part in path for part in ["/product/", "/products/"]):
        return "PRODUCT_OR_CATEGORY"
    if any(part in path for part in ["/compare", "/best-"]):
        return "COMPARISON"
    if any(word in title for word in ["how to", "guide", "tutorial"]):
        return "GUIDE"
    if any(part in path for part in ["/news/", "/press/"]):
        return "NEWS"
    if any(part in path for part in ["/forum/", "/community/"]):
        return "COMMUNITY"
    return "OTHER"

from collections import Counter


def snapshot_features(snapshot: SerpSnapshot) -> dict:
    top_results = snapshot.organic_results[:10]
    archetypes = Counter(
        item.get("archetype", "OTHER") for item in top_results
    )
    hosts = {urlparse(item["url"]).hostname for item in top_results}

    return {
        "archetype_share": {
            key: value / max(len(top_results), 1)
            for key, value in archetypes.items()
        },
        "hosts": hosts,
        "features": set(snapshot.features),
        "top_urls": {item["url"] for item in top_results},
    }

def jaccard(left: set, right: set) -> float:
    union = left | right
    if not union:
        return 1.0
    return len(left & right) / len(union)


def serp_change_metrics(previous: SerpSnapshot, current: SerpSnapshot) -> dict:
    old = snapshot_features(previous)
    new = snapshot_features(current)

    return {
        "url_similarity": jaccard(old["top_urls"], new["top_urls"]),
        "host_similarity": jaccard(old["hosts"], new["hosts"]),
        "feature_similarity": jaccard(old["features"], new["features"]),
        "previous_archetypes": old["archetype_share"],
        "current_archetypes": new["archetype_share"],
    }

def archetype_delta(previous: dict, current: dict) -> dict:
    labels = set(previous) | set(current)
    return {
        label: round(current.get(label, 0) - previous.get(label, 0), 3)
        for label in labels
    }


def dominant_archetype(shares: dict) -> str:
    if not shares:
        return "UNKNOWN"
    return max(shares.items(), key=lambda item: item[1])[0]

def performance_delta(previous: SearchPerformance, current: SearchPerformance):
    return {
        "impression_delta": current.impressions - previous.impressions,
        "click_delta": current.clicks - previous.clicks,
        "ctr_delta": current.ctr - previous.ctr,
        "position_delta": current.position - previous.position,
    }

def intent_drift_score(metrics: dict, behavior: dict) -> dict:
    url_change = 1 - metrics["url_similarity"]
    host_change = 1 - metrics["host_similarity"]
    feature_change = 1 - metrics["feature_similarity"]

    old_dom = dominant_archetype(metrics["previous_archetypes"])
    new_dom = dominant_archetype(metrics["current_archetypes"])
    archetype_changed = float(old_dom != new_dom)

    score = (
        0.20 * url_change
        + 0.15 * host_change
        + 0.20 * feature_change
        + 0.35 * archetype_changed
        + 0.10 * min(abs(behavior["ctr_delta"]) * 5, 1)
    )

    return {
        "score": round(score, 3),
        "previous_dominant": old_dom,
        "current_dominant": new_dom,
        "archetype_changed": bool(archetype_changed),
    }

def confirmed_drift(windows: list[dict], minimum_score: float = 0.55):
    if len(windows) < 3:
        return False

    recent = windows[-3:]
    high = [item for item in recent if item["score"] >= minimum_score]
    labels = {
        item["current_dominant"]
        for item in recent
        if item["current_dominant"] != "UNKNOWN"
    }

    return len(high) >= 2 and len(labels) == 1

def build_drift_alert(cohort: QueryCohort, analysis: dict) -> dict:
    return {
        "event": "SEARCH_INTENT_DRIFT_CONFIRMED",
        "cohort_id": cohort.cohort_id,
        "topic": cohort.canonical_topic,
        "market": cohort.market,
        "device": cohort.device,
        "previous_intent": analysis["previous_dominant"],
        "current_intent": analysis["current_dominant"],
        "score": analysis["score"],
        "snapshot_ids": analysis["snapshot_ids"],
        "search_console_run_ids": analysis["search_console_run_ids"],
        "recommended_action": "human_content_review",
    }

from dataclasses import dataclass

@dataclass(frozen=True)
class AnalysisCase:
    case_id: str
    cohort_id: str
    approved_purpose: str
    snapshot_ids: tuple[str, ...]
    search_console_run_ids: tuple[str, ...]

CASES: dict[str, AnalysisCase] = {}

async def analyze_search_intent_case(case_id: str) -> dict:
    """保存された証拠から事前に承認された検索意図ケースを分析します。"""
    case = CASES.get(case_id)
    if case is None:
        raise PermissionError("不明な分析ケース")

    result = analysis_service.compute(case)
    return {
        "case_id": case.case_id,
        "status": result.status,
        "previous_intent": result.previous_intent,
        "current_intent": result.current_intent,
        "confidence": result.confidence,
        "evidence_references": result.evidence_references,
        "review_required": True,
    }

CONTENT_ACTIONS = {
    ("GUIDE", "COMPARISON"): "決定基準と比較表を追加する",
    ("GUIDE", "PRODUCT_OR_CATEGORY"): "カテゴリまたは製品ランディングページの適合性を確認する",
    ("EVERGREEN", "NEWS"): "新鮮さワークフローまたはニュースコンパニオンページを追加する",
    ("NATIONAL", "LOCAL"): "ローカルページとローカル証拠を確認する",
    ("TEXT", "VIDEO"): "検証済みビデオまたは視覚的説明を追加する",
}