"""OpenAlex sources data source.

Downloads the **bulk snapshot** of ~280K journals + conferences from
``openalex.s3.amazonaws.com`` and writes a compact gzipped JSON file
used by the in-memory tier-2 lookups.

This is the public OpenAlex S3 snapshot — CC0 licensed, no auth, no
rate limits, ~350 MB compressed across ~120 partition files. We stream
each part, extract the few fields we need, and discard the rest. Total
wall-clock is ~30–60 s on a normal connection.

Why bulk snapshot instead of the REST API
-----------------------------------------

The previous implementation paginated ``/api/sources`` with
``per_page=200`` and a 100 ms polite-rate-limiting sleep between
requests. For ~280K records that meant ~1,400 HTTP requests + ~140 s
of pure sleep, plus actual transfer time — wall-clock 5–10 minutes,
and that's the time the user spends staring at the dashboard
"Download Data" button. The S3 dump is the recommended bulk path per
OpenAlex docs and finishes in well under a minute.
"""

from __future__ import annotations

import gzip
import json
import time
from pathlib import Path

from loguru import logger

from ...utilities.citation_normalizer import normalize_issn
from .base import DataSource

from ._openalex_common import (
    OPENALEX_S3_BASE,
    iter_partitions,
    validate_manifest_entries,
)

_OPENALEX_SOURCES_MANIFEST = (
    f"{OPENALEX_S3_BASE}/data/jsonl/sources/manifest.json"
)


class SchemaDriftError(RuntimeError):
    """OpenAlex renamed / removed a required field in the snapshot.

    A row-count floor catches the case where the whole fetch collapses,
    but not the case where every row loads but a key field (``h_index``,
    ``cited_by_count``) is silently None. We refuse to overwrite the
    existing snapshot in that case — better to keep the old data than
    rebuild an all-None DB that would quietly reclassify every journal
    into the "unknown" quality tier.
    """


class OpenAlexSource(DataSource):
    key = "openalex"  # gitleaks:allow
    name = "OpenAlex"
    url = "https://openalex.org"
    dataset_url = (
        "https://docs.openalex.org/download-all-data/openalex-snapshot"
    )
    license = "CC0 1.0"
    license_url = "https://creativecommons.org/publicdomain/zero/1.0/"
    description = (
        "~280K journals and conferences with h-index, impact factor, "
        "and publisher metadata"
    )
    filename = "openalex_sources.json.gz"
    count_label = "OpenAlex sources"
    auto_download = False  # large; user opts in via dashboard
    required = True  # bulk-download fatal-on-failure
    approx_size_mb = 13.0  # final compact output, NOT the raw snapshot

    def fetch(self, data_dir: Path, progress_cb=None) -> int:
        from ...security.safe_requests import (
            safe_get_with_retries as safe_get,
        )

        # 1. Fetch the manifest. Tells us which partition files exist
        #    and how many records to expect — so we can give the user
        #    an accurate progress log instead of just dots.
        logger.info(
            f"Fetching OpenAlex sources manifest: {_OPENALEX_SOURCES_MANIFEST}"
        )
        # consume_body=True: the manifest is small but a serial
        # bottleneck for the whole download. A body-read transient
        # here aborts everything.
        manifest_resp = safe_get(
            _OPENALEX_SOURCES_MANIFEST, timeout=30, consume_body=True
        )
        manifest_resp.raise_for_status()
        manifest = manifest_resp.json()

        # OpenAlex's 2026-06 "standard-format" snapshot moved the data to
        # ``data/jsonl/<entity>/`` and renamed the manifest's part list from
        # ``entries`` to ``files``. Each file entry still has ``url``
        # (s3://openalex/...) + ``meta.record_count`` / ``meta.content_length``.
        entries = manifest.get("files", [])
        total_records = sum(
            e.get("meta", {}).get("record_count", 0) for e in entries
        )
        total_bytes = sum(
            e.get("meta", {}).get("content_length", 0) for e in entries
        )
        logger.info(
            f"OpenAlex sources snapshot: {len(entries)} parts, "
            f"{total_records:,} records, "
            f"{total_bytes / 1024 / 1024:.0f} MB compressed"
        )

        # Validate manifest URLs before fetching anything, so a
        # compromised manifest cannot redirect fetches to an arbitrary
        # host. Legitimate OpenAlex manifest entries always use the
        # s3://openalex/ prefix.
        validate_manifest_entries(entries, "OpenAlex sources")

        # 2. Stream-process each part. The ``iter_partitions`` helper
        #    in ``_openalex_common`` owns the tmp-file lifecycle + the
        #    first-10 malformed-JSON suppression; we just consume
        #    records and track our own schema-drift counters.
        type_map = {"journal": "j", "conference": "c"}
        sources: dict = {}
        # Raw parse counters feed the ``id``-rename drift check below —
        # records without an ``id`` are silently skipped at the point of
        # extraction, so we need to know the ratio to distinguish a
        # collapsed fetch from a renamed identifier field.
        parsed_records = 0
        parsed_with_id = 0
        start = time.time()

        for idx, total_parts, records in iter_partitions(
            entries,
            data_dir,
            file_prefix="openalex_sources",
            label="OpenAlex sources",
            safe_get=safe_get,
        ):
            for rec in records:
                parsed_records += 1
                src_id = (rec.get("id") or "").split("/")[-1]
                if not src_id:
                    continue
                parsed_with_id += 1

                stats = rec.get("summary_stats") or {}
                compact = {
                    "n": rec.get("display_name", ""),
                    "t": type_map.get(rec.get("type", ""), rec.get("type", "")),
                    "h": stats.get("h_index"),
                    "if": stats.get("2yr_mean_citedness"),
                    "cb": rec.get("cited_by_count"),
                    "p": rec.get("host_organization_name") or "",
                    "i": normalize_issn(rec.get("issn_l")),
                }
                if rec.get("is_in_doaj"):
                    compact["d"] = 1
                if rec.get("is_core"):
                    compact["s"] = 1
                sources[src_id] = compact

            if (idx + 1) % 5 == 0 or idx == total_parts - 1:
                elapsed = time.time() - start
                logger.info(
                    f"OpenAlex sources: processed {idx + 1}/{total_parts} "
                    f"parts ({len(sources):,} records, {elapsed:.0f}s)"
                )
            # Report on EVERY partition, not just every 5th — the UI
            # needs smoother updates than the human-readable log.
            if progress_cb is not None:
                try:
                    progress_cb(
                        idx + 1,
                        total_parts,
                        f"{len(sources):,} records",
                    )
                except Exception:
                    logger.debug(
                        "OpenAlex progress_cb raised; continuing",
                        exc_info=True,
                    )

        # ``id``-rename drift runs *before* the row-count floor: a
        # renamed identifier makes every record drop out at parse time
        # (src_id empty → continue), so ``sources`` is empty and the
        # row-count floor would fire first with a generic RuntimeError
        # that hides the actual cause.
        if parsed_records >= 10_000 and parsed_with_id == 0:
            raise SchemaDriftError(
                f"OpenAlex snapshot parsed {parsed_records:,} records but "
                "none carried an 'id' field — the source identifier may "
                "have been renamed (e.g. to 'source_id'). Refusing to "
                "overwrite existing data — please check "
                "https://docs.openalex.org/download-all-data/"
                "snapshot-data-format for schema changes."
            )

        # 3. Write the compact snapshot in the same shape the existing
        #    build pipeline expects (`{"s": {src_id: compact}}`).
        #    Sanity check: OpenAlex normally has ~280K sources. If the
        #    fetch silently returned a tiny subset (e.g., every partition
        #    returned an empty shard), refuse to overwrite existing data.
        _MIN_OPENALEX_SOURCES = 10_000
        if len(sources) < _MIN_OPENALEX_SOURCES:
            raise RuntimeError(
                f"OpenAlex sources: suspiciously few records "
                f"({len(sources):,} < {_MIN_OPENALEX_SOURCES:,}); "
                "refusing to overwrite existing data"
            )

        # Field-level schema drift detection. The row-count floor above
        # catches a collapsed fetch, but not a silent upstream rename.
        # The journal-only sample avoids false-triggering on snapshots
        # that skew to non-journal types (conferences, repositories,
        # etc.) which legitimately lack ``h_index``.
        _SCHEMA_SAMPLE_SIZE = 100
        journal_sample = [r for r in sources.values() if r.get("t") == "j"][
            :_SCHEMA_SAMPLE_SIZE
        ]
        if len(journal_sample) >= _SCHEMA_SAMPLE_SIZE:
            has_hindex = any(r.get("h") is not None for r in journal_sample)
            has_cited = any(r.get("cb") is not None for r in journal_sample)
            if not has_hindex or not has_cited:
                raise SchemaDriftError(
                    "OpenAlex snapshot appears to have renamed a required "
                    "field: "
                    f"h_index present in journal sample={has_hindex}, "
                    f"cited_by_count present in journal sample={has_cited}. "
                    "Refusing to overwrite existing data — please check "
                    "https://docs.openalex.org/download-all-data/"
                    "snapshot-data-format for schema changes."
                )
        else:
            # The row-count floor above already refuses a collapsed
            # fetch. This branch only fires in unusual cases (truncated
            # test snapshot, snapshot with very few journal-typed
            # sources). Log at info so operators see the drift check
            # was bypassed — debug would be invisible at production
            # log levels.
            logger.info(
                "OpenAlex schema-drift check skipped: "
                f"only {len(journal_sample)} journal source(s) in sample "
                f"(< {_SCHEMA_SAMPLE_SIZE} required)"
            )

        output = data_dir / self.filename
        tmp = data_dir / f"{self.filename}.tmp"
        with gzip.open(tmp, "wt", encoding="utf-8") as f:
            json.dump({"s": sources}, f)
        tmp.rename(output)

        elapsed = time.time() - start
        logger.info(
            f"OpenAlex sources: saved {len(sources):,} sources in {elapsed:.0f}s"
        )
        return len(sources)
