#!/usr/bin/env python3
"""Fetch and derive a reproducible Google Trends evidence bundle.

This uses the same official web endpoints as Google Trends Explore. These
endpoints are not the restricted Google Trends API alpha and may rate-limit
automated clients. Every HTTP attempt and failure is retained in the run
manifest/error log; no synthetic values are generated.
"""

from __future__ import annotations

import csv
import hashlib
import http.cookiejar
import json
import math
import re
import statistics
import time
import urllib.error
import urllib.parse
import urllib.request
from dataclasses import dataclass
from datetime import UTC, datetime
from pathlib import Path
from typing import Any


ROOT = Path(__file__).resolve().parent
SPEC_PATH = ROOT / "query_spec.json"
BASE_URL = "https://trends.google.com"
USER_AGENT = (
    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
    "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/140.0.0.0 Safari/537.36"
)
ANTI_XSSI = re.compile(r"^\)\]\}',?\s*")


class TrendsFetchError(RuntimeError):
    """Raised when an official Google Trends endpoint cannot be fetched."""


@dataclass(frozen=True)
class FetchResult:
    url: str
    status: int
    body: bytes
    fetched_at_utc: str


class TrendsClient:
    """Small stdlib-only client for the Google Trends Explore web endpoints."""

    def __init__(self, error_log_path: Path, min_interval_seconds: float = 1.25) -> None:
        self.cookie_jar = http.cookiejar.CookieJar()
        self.opener = urllib.request.build_opener(
            urllib.request.HTTPCookieProcessor(self.cookie_jar)
        )
        self.error_log_path = error_log_path
        self.min_interval_seconds = min_interval_seconds
        self.last_request_monotonic = 0.0
        self.request_log: list[dict[str, Any]] = []

    def _throttle(self) -> None:
        elapsed = time.monotonic() - self.last_request_monotonic
        if elapsed < self.min_interval_seconds:
            time.sleep(self.min_interval_seconds - elapsed)

    def _record(
        self,
        *,
        url: str,
        status: int,
        fetched_at_utc: str,
        byte_count: int,
        attempt: int,
        error: str | None = None,
    ) -> None:
        entry = {
            "url": url,
            "status": status,
            "fetched_at_utc": fetched_at_utc,
            "bytes": byte_count,
            "attempt": attempt,
            "error": error,
        }
        self.request_log.append(entry)
        if error is not None:
            with self.error_log_path.open("a", encoding="utf-8") as handle:
                handle.write(json.dumps(entry, ensure_ascii=False) + "\n")

    def fetch(
        self,
        url: str,
        *,
        retries: int = 4,
        accept_statuses: frozenset[int] = frozenset({200}),
    ) -> FetchResult:
        backoffs = (1.0, 5.0, 15.0, 30.0)
        last_error: Exception | None = None
        for attempt in range(1, retries + 1):
            self._throttle()
            fetched_at_utc = datetime.now(UTC).isoformat()
            request = urllib.request.Request(
                url,
                headers={
                    "User-Agent": USER_AGENT,
                    "Accept": "application/json,text/plain,*/*",
                    "Accept-Language": "en-US,en;q=0.9,ru;q=0.8",
                    "Referer": f"{BASE_URL}/trends/explore",
                },
            )
            try:
                with self.opener.open(request, timeout=45) as response:
                    body = response.read()
                    status = response.status
                self.last_request_monotonic = time.monotonic()
                self._record(
                    url=url,
                    status=status,
                    fetched_at_utc=fetched_at_utc,
                    byte_count=len(body),
                    attempt=attempt,
                )
                if status in accept_statuses:
                    return FetchResult(url, status, body, fetched_at_utc)
                last_error = TrendsFetchError(f"unexpected HTTP {status} for {url}")
            except urllib.error.HTTPError as exc:
                body = exc.read()
                self.last_request_monotonic = time.monotonic()
                error = f"HTTPError {exc.code}: {body[:300].decode('utf-8', 'replace')}"
                self._record(
                    url=url,
                    status=exc.code,
                    fetched_at_utc=fetched_at_utc,
                    byte_count=len(body),
                    attempt=attempt,
                    error=error,
                )
                if exc.code in accept_statuses:
                    return FetchResult(url, exc.code, body, fetched_at_utc)
                last_error = exc
            except (TimeoutError, urllib.error.URLError) as exc:
                self.last_request_monotonic = time.monotonic()
                self._record(
                    url=url,
                    status=0,
                    fetched_at_utc=fetched_at_utc,
                    byte_count=0,
                    attempt=attempt,
                    error=f"{type(exc).__name__}: {exc}",
                )
                last_error = exc
            if attempt < retries:
                time.sleep(backoffs[min(attempt - 1, len(backoffs) - 1)])
        raise TrendsFetchError(f"failed after {retries} attempts: {url}: {last_error}")

    def bootstrap(self) -> None:
        url = (
            f"{BASE_URL}/trends/explore?"
            + urllib.parse.urlencode(
                {
                    "hl": "en-US",
                    "geo": "RU",
                    "date": "today 12-m",
                    "q": "Dungeons & Dragons",
                }
            )
        )
        # Google commonly answers a fresh cookie jar with one 429 while setting
        # NID, then accepts the same browser-like session on the next request.
        first = self.fetch(url, retries=1, accept_statuses=frozenset({200, 429}))
        if first.status == 429:
            self.fetch(url, retries=4)

    def fetch_json(self, url: str) -> tuple[dict[str, Any], FetchResult]:
        result = self.fetch(url)
        text = result.body.decode("utf-8")
        payload = json.loads(ANTI_XSSI.sub("", text, count=1))
        return payload, result


def sha256_bytes(data: bytes) -> str:
    return hashlib.sha256(data).hexdigest()


def write_json(path: Path, payload: Any) -> 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_csv(path: Path, fieldnames: list[str], rows: list[dict[str, Any]]) -> 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()
        writer.writerows(rows)


def api_url(path: str, params: dict[str, Any]) -> str:
    serialized = {
        key: json.dumps(value, ensure_ascii=False, separators=(",", ":"))
        if isinstance(value, (dict, list))
        else str(value)
        for key, value in params.items()
    }
    return f"{BASE_URL}{path}?{urllib.parse.urlencode(serialized)}"


def explore_url(group: dict[str, Any], period: str) -> str:
    return f"{BASE_URL}/trends/explore?" + urllib.parse.urlencode(
        {
            "hl": group["hl"],
            "date": period,
            "geo": group["geo"],
            "q": ",".join(item["keyword"] for item in group["items"]),
        }
    )


def slug(value: str) -> str:
    normalized = re.sub(r"[^a-zA-Z0-9а-яА-ЯёЁ]+", "-", value).strip("-").lower()
    return normalized or "item"


def widget_data_url(widget: dict[str, Any], *, hl: str, tz: int) -> str:
    endpoint_by_id = {
        "TIMESERIES": "/trends/api/widgetdata/multiline",
        "GEO_MAP": "/trends/api/widgetdata/comparedgeo",
    }
    widget_id = widget["id"]
    if widget_id.startswith("RELATED_QUERIES_"):
        endpoint = "/trends/api/widgetdata/relatedsearches"
    elif widget_id.startswith("GEO_MAP_"):
        endpoint = "/trends/api/widgetdata/comparedgeo"
    else:
        endpoint = endpoint_by_id[widget_id]
    return api_url(
        endpoint,
        {
            "hl": hl,
            "tz": tz,
            "req": widget["request"],
            "token": widget["token"],
        },
    )


def derive_timeseries(
    payload: dict[str, Any], group: dict[str, Any]
) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
    default = payload.get("default", {})
    timeline = default.get("timelineData", [])
    items = group["items"]
    rows: list[dict[str, Any]] = []
    values_by_item: list[list[tuple[str, int]]] = [[] for _ in items]
    for point in timeline:
        values = point.get("value", [])
        has_data = point.get("hasData", [True] * len(items))
        for index, item in enumerate(items):
            value = int(values[index]) if index < len(values) else 0
            available = bool(has_data[index]) if index < len(has_data) else False
            row = {
                "timestamp_utc": datetime.fromtimestamp(int(point["time"]), UTC).isoformat(),
                "period_label": point.get("formattedTime", ""),
                "item": item["label"],
                "kind": item["kind"],
                "value": value,
                "has_data": str(available).lower(),
                "is_partial": str(bool(point.get("isPartial", False))).lower(),
            }
            rows.append(row)
            if available and not point.get("isPartial", False):
                values_by_item[index].append((row["timestamp_utc"], value))

    summaries: list[dict[str, Any]] = []
    google_averages = default.get("averages", [])
    for index, item in enumerate(items):
        observations = values_by_item[index]
        numeric = [value for _, value in observations]
        peak_time, peak_value = max(observations, key=lambda pair: pair[1]) if observations else ("", 0)
        recent = numeric[-13:]
        previous = numeric[-26:-13]
        recent_mean = statistics.fmean(recent) if recent else math.nan
        previous_mean = statistics.fmean(previous) if previous else math.nan
        summaries.append(
            {
                "item": item["label"],
                "kind": item["kind"],
                "google_average_index": google_averages[index]
                if index < len(google_averages)
                else "",
                "computed_mean_index": round(statistics.fmean(numeric), 3)
                if numeric
                else "",
                "median_index": round(statistics.median(numeric), 3) if numeric else "",
                "latest_complete_index": numeric[-1] if numeric else "",
                "peak_index": peak_value,
                "peak_timestamp_utc": peak_time,
                "nonzero_share": round(sum(value > 0 for value in numeric) / len(numeric), 4)
                if numeric
                else "",
                "last_13_points_mean": round(recent_mean, 3) if recent else "",
                "previous_13_points_mean": round(previous_mean, 3) if previous else "",
                "last_vs_previous_13_pct": round(
                    (recent_mean / previous_mean - 1) * 100, 2
                )
                if previous and previous_mean > 0
                else "",
                "complete_points": len(numeric),
            }
        )
    return rows, summaries


def derive_geography(
    payload: dict[str, Any], group: dict[str, Any]
) -> list[dict[str, Any]]:
    rows: list[dict[str, Any]] = []
    items = group["items"]
    for entry in payload.get("default", {}).get("geoMapData", []):
        values = entry.get("value", [])
        has_data = entry.get("hasData", [True] * len(items))
        for index, item in enumerate(items):
            rows.append(
                {
                    "geo_code": entry.get("geoCode", ""),
                    "geo_name": entry.get("geoName", ""),
                    "item": item["label"],
                    "value": values[index] if index < len(values) else 0,
                    "has_data": str(
                        bool(has_data[index]) if index < len(has_data) else False
                    ).lower(),
                }
            )
    return rows


def derive_related_queries(
    payload: dict[str, Any], item_label: str
) -> list[dict[str, Any]]:
    rows: list[dict[str, Any]] = []
    ranked_lists = payload.get("default", {}).get("rankedList", [])
    for metric_index, ranked_list in enumerate(ranked_lists):
        metric = "top" if metric_index == 0 else "rising"
        for rank, entry in enumerate(ranked_list.get("rankedKeyword", []), start=1):
            rows.append(
                {
                    "source_item": item_label,
                    "metric": metric,
                    "rank": rank,
                    "query": entry.get("query", ""),
                    "value": entry.get("value", ""),
                    "formatted_value": entry.get("formattedValue", ""),
                    "link": entry.get("link", ""),
                }
            )
    return rows


def main() -> int:
    spec = json.loads(SPEC_PATH.read_text(encoding="utf-8"))
    run_started = datetime.now(UTC)
    run_id = run_started.strftime("%Y%m%dT%H%M%SZ")
    run_dir = ROOT / "runs" / run_id
    raw_dir = run_dir / "raw"
    derived_dir = run_dir / "derived"
    run_dir.mkdir(parents=True, exist_ok=False)
    error_log_path = run_dir / "errors.jsonl"
    error_log_path.touch()

    client = TrendsClient(error_log_path)
    failures: list[dict[str, str]] = []
    artifacts: list[dict[str, Any]] = []
    client.bootstrap()

    for query in spec["autocomplete_checks"]:
        url = f"{BASE_URL}/trends/api/autocomplete/{urllib.parse.quote(query, safe='')}?hl=en-US"
        try:
            payload, result = client.fetch_json(url)
            path = raw_dir / "autocomplete" / f"{slug(query)}.json"
            write_json(path, {"query": query, "endpoint": url, "response": payload})
            artifacts.append(
                {
                    "path": str(path.relative_to(ROOT)),
                    "source_url": result.url,
                    "fetched_at_utc": result.fetched_at_utc,
                    "sha256": sha256_bytes(path.read_bytes()),
                }
            )
        except TrendsFetchError as exc:
            failures.append({"stage": "autocomplete", "query": query, "error": str(exc)})

    for group in spec["groups"]:
        if len(group["items"]) > 5:
            raise ValueError(f"group {group['id']} exceeds five comparison items")
        for period_id, period in spec["periods"].items():
            group_dir = raw_dir / group["id"] / period_id
            derived_group_dir = derived_dir / group["id"] / period_id
            request_payload = {
                "comparisonItem": [
                    {
                        "keyword": item["keyword"],
                        "geo": group["geo"],
                        "time": period,
                    }
                    for item in group["items"]
                ],
                "category": spec["category"],
                "property": spec["search_property"],
            }
            explore_api = api_url(
                "/trends/api/explore",
                {
                    "hl": group["hl"],
                    "tz": spec["timezone_offset_minutes"],
                    "req": request_payload,
                },
            )
            try:
                explore_payload, explore_result = client.fetch_json(explore_api)
                explore_path = group_dir / "explore_response.json"
                write_json(
                    explore_path,
                    {
                        "request": request_payload,
                        "human_explore_url": explore_url(group, period),
                        "api_endpoint": explore_api,
                        "fetched_at_utc": explore_result.fetched_at_utc,
                        "response": explore_payload,
                    },
                )
                artifacts.append(
                    {
                        "path": str(explore_path.relative_to(ROOT)),
                        "source_url": explore_result.url,
                        "fetched_at_utc": explore_result.fetched_at_utc,
                        "sha256": sha256_bytes(explore_path.read_bytes()),
                    }
                )
            except TrendsFetchError as exc:
                failures.append(
                    {
                        "stage": "explore",
                        "group": group["id"],
                        "period": period_id,
                        "error": str(exc),
                    }
                )
                continue

            widgets = {widget["id"]: widget for widget in explore_payload.get("widgets", [])}
            timeseries_widget = widgets.get("TIMESERIES")
            if timeseries_widget is not None:
                try:
                    url = widget_data_url(
                        timeseries_widget,
                        hl=group["hl"],
                        tz=spec["timezone_offset_minutes"],
                    )
                    payload, result = client.fetch_json(url)
                    path = group_dir / "timeseries.json"
                    write_json(path, {"endpoint": url, "response": payload})
                    rows, summaries = derive_timeseries(payload, group)
                    write_csv(
                        derived_group_dir / "interest_over_time.csv",
                        [
                            "timestamp_utc",
                            "period_label",
                            "item",
                            "kind",
                            "value",
                            "has_data",
                            "is_partial",
                        ],
                        rows,
                    )
                    summary_fields = [
                        "item",
                        "kind",
                        "google_average_index",
                        "computed_mean_index",
                        "median_index",
                        "latest_complete_index",
                        "peak_index",
                        "peak_timestamp_utc",
                        "nonzero_share",
                        "last_13_points_mean",
                        "previous_13_points_mean",
                        "last_vs_previous_13_pct",
                        "complete_points",
                    ]
                    write_csv(derived_group_dir / "summary.csv", summary_fields, summaries)
                    artifacts.append(
                        {
                            "path": str(path.relative_to(ROOT)),
                            "source_url": result.url,
                            "fetched_at_utc": result.fetched_at_utc,
                            "sha256": sha256_bytes(path.read_bytes()),
                        }
                    )
                except TrendsFetchError as exc:
                    failures.append(
                        {
                            "stage": "timeseries",
                            "group": group["id"],
                            "period": period_id,
                            "error": str(exc),
                        }
                    )

            if period_id != "five_years":
                continue

            if group.get("fetch_geography") and widgets.get("GEO_MAP") is not None:
                try:
                    widget = widgets["GEO_MAP"]
                    url = widget_data_url(
                        widget,
                        hl=group["hl"],
                        tz=spec["timezone_offset_minutes"],
                    )
                    payload, result = client.fetch_json(url)
                    path = group_dir / "geography_compared.json"
                    write_json(path, {"endpoint": url, "response": payload})
                    rows = derive_geography(payload, group)
                    write_csv(
                        derived_group_dir / "geography_compared.csv",
                        ["geo_code", "geo_name", "item", "value", "has_data"],
                        rows,
                    )
                    artifacts.append(
                        {
                            "path": str(path.relative_to(ROOT)),
                            "source_url": result.url,
                            "fetched_at_utc": result.fetched_at_utc,
                            "sha256": sha256_bytes(path.read_bytes()),
                        }
                    )
                except TrendsFetchError as exc:
                    failures.append(
                        {
                            "stage": "geography",
                            "group": group["id"],
                            "period": period_id,
                            "error": str(exc),
                        }
                    )

            if group.get("fetch_related_queries"):
                related_rows: list[dict[str, Any]] = []
                for index, item in enumerate(group["items"]):
                    widget = widgets.get(f"RELATED_QUERIES_{index}")
                    if widget is None:
                        failures.append(
                            {
                                "stage": "related_queries",
                                "group": group["id"],
                                "period": period_id,
                                "item": item["label"],
                                "error": "widget absent in Explore response",
                            }
                        )
                        continue
                    try:
                        url = widget_data_url(
                            widget,
                            hl=group["hl"],
                            tz=spec["timezone_offset_minutes"],
                        )
                        payload, result = client.fetch_json(url)
                        path = group_dir / "related_queries" / f"{index}-{slug(item['label'])}.json"
                        write_json(path, {"endpoint": url, "response": payload})
                        related_rows.extend(derive_related_queries(payload, item["label"]))
                        artifacts.append(
                            {
                                "path": str(path.relative_to(ROOT)),
                                "source_url": result.url,
                                "fetched_at_utc": result.fetched_at_utc,
                                "sha256": sha256_bytes(path.read_bytes()),
                            }
                        )
                    except TrendsFetchError as exc:
                        failures.append(
                            {
                                "stage": "related_queries",
                                "group": group["id"],
                                "period": period_id,
                                "item": item["label"],
                                "error": str(exc),
                            }
                        )
                write_csv(
                    derived_group_dir / "related_queries.csv",
                    [
                        "source_item",
                        "metric",
                        "rank",
                        "query",
                        "value",
                        "formatted_value",
                        "link",
                    ],
                    related_rows,
                )

    run_finished = datetime.now(UTC)
    manifest = {
        "schema_version": 1,
        "run_id": run_id,
        "started_at_utc": run_started.isoformat(),
        "finished_at_utc": run_finished.isoformat(),
        "local_timezone_context": "Asia/Ho_Chi_Minh",
        "trends_timezone_offset_minutes": spec["timezone_offset_minutes"],
        "trends_time_basis_note": (
            "Google says Explore charts for ranges of 30 days or longer use UTC. "
            "Both requested ranges exceed 30 days, and tz=0 was sent."
        ),
        "source": "Official Google Trends Explore web endpoints at trends.google.com",
        "official_api_alpha_note": (
            "The public Google Trends API remains an application-only alpha. "
            "This bundle therefore uses the official Explore web endpoints."
        ),
        "normalization_warning": (
            "Values are sampled and normalized 0-100 within each request. "
            "They are not absolute search volumes and cannot be compared across "
            "independently scaled groups or time windows without a shared anchor."
        ),
        "query_spec_sha256": sha256_bytes(SPEC_PATH.read_bytes()),
        "request_count": len(client.request_log),
        "failure_count": len(failures),
        "failures": failures,
        "requests": client.request_log,
        "artifacts": artifacts,
    }
    write_json(run_dir / "manifest.json", manifest)
    (ROOT / "latest_run.txt").write_text(run_id + "\n", encoding="utf-8")
    print(run_dir)
    print(f"requests={len(client.request_log)} failures={len(failures)}")
    return 0 if not failures else 2


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