from __future__ import annotations

import hashlib
import json
import os
import tempfile
import threading
import time
from contextlib import contextmanager
from datetime import date, datetime, timedelta
from decimal import Decimal
from pathlib import Path
from typing import Any, Iterator


def canonical_bytes(value: Any) -> bytes:
    def encode(item: Any) -> Any:
        if isinstance(item, Decimal):
            return format(item, "f")
        if isinstance(item, Path):
            return str(item)
        raise TypeError(f"Object of type {type(item).__name__} is not JSON serializable")

    return json.dumps(
        value,
        ensure_ascii=False,
        sort_keys=True,
        separators=(",", ":"),
        default=encode,
    ).encode("utf-8")


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


def atomic_write(path: Path, data: bytes) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    fd, raw_temp = tempfile.mkstemp(prefix=f".{path.name}.", dir=path.parent)
    temp = Path(raw_temp)
    try:
        with os.fdopen(fd, "wb") as handle:
            handle.write(data)
            handle.flush()
            os.fsync(handle.fileno())
        os.replace(temp, path)
    finally:
        if temp.exists():
            temp.unlink()


def request_fingerprint(preimage: dict[str, Any]) -> str:
    return sha256_bytes(canonical_bytes(preimage))


DATA_KIND_TTLS = {
    "stock_identity": 7 * 86400,
    "announcement_index_recent": 6 * 3600,
    "announcement_index_historical": 30 * 86400,
    "market_close": 30 * 86400,
    "shares_market_cap": 15 * 60,
    "finance_main_recent": 6 * 3600,
    "finance_main_historical": 30 * 86400,
    "finance_income_recent": 6 * 3600,
    "finance_income_historical": 30 * 86400,
    "finance_balance_recent": 6 * 3600,
    "finance_balance_historical": 30 * 86400,
    "finance_cashflow_recent": 6 * 3600,
    "finance_cashflow_historical": 30 * 86400,
    "forecast_summary": 6 * 3600,
    "forecast_detail": 6 * 3600,
}

PROVIDER_KINDS = {
    "announcements": ("stock_identity", "announcement_index"),
    "market": ("market_close", "shares_market_cap"),
    "finance": ("finance_main", "finance_income", "finance_balance", "finance_cashflow"),
    "forecast": ("forecast_summary", "forecast_detail"),
}


class BlobIntegrityError(OSError):
    """A content-addressed object could not be written and verified safely."""

_PROCESS_LOCKS: dict[str, threading.Lock] = {}
_PROCESS_LOCKS_GUARD = threading.Lock()


@contextmanager
def _ticker_lock(path: Path) -> Iterator[None]:
    key = os.path.normcase(os.path.abspath(path))
    with _PROCESS_LOCKS_GUARD:
        local = _PROCESS_LOCKS.setdefault(key, threading.Lock())
    with local:
        path.parent.mkdir(parents=True, exist_ok=True)
        handle = open(path, "a+b")
        try:
            handle.seek(0)
            if handle.tell() == 0 and path.stat().st_size == 0:
                handle.write(b"0")
                handle.flush()
            if os.name == "nt":
                import msvcrt

                handle.seek(0)
                msvcrt.locking(handle.fileno(), msvcrt.LK_LOCK, 1)
            else:
                import fcntl

                fcntl.flock(handle.fileno(), fcntl.LOCK_EX)
            yield
        finally:
            try:
                handle.seek(0)
                if os.name == "nt":
                    import msvcrt

                    msvcrt.locking(handle.fileno(), msvcrt.LK_UNLCK, 1)
                else:
                    import fcntl

                    fcntl.flock(handle.fileno(), fcntl.LOCK_UN)
            finally:
                handle.close()


class ContentCache:
    def __init__(self, root: Path):
        self.root = root

    def load_reusable(
        self, provider: str, fingerprint: str, process_start: datetime
    ) -> dict[str, Any] | None:
        index = self.root / "reusable" / provider / f"{fingerprint}.json"
        if not index.is_file():
            return None
        try:
            meta = json.loads(index.read_text(encoding="utf-8"))
            expires = datetime.fromisoformat(meta["expires_at"])
            blob = self.root / meta["blob_path"]
            data = blob.read_bytes()
        except (OSError, ValueError, KeyError, json.JSONDecodeError):
            return None
        if process_start > expires or sha256_bytes(data) != meta.get("blob_hash"):
            return None
        if not (200 <= int(meta.get("http_status", 0)) < 300):
            return None
        if not all(
            meta.get(key) is True
            for key in ("transport_complete", "parse_ok", "schema_ok", "as_of_ok")
        ) or meta.get("semantic_status") != "OK":
            return None
        return {"body": data, "meta": meta}

    def store(
        self,
        provider: str,
        fingerprint: str,
        body: bytes,
        fetched_at: datetime,
        ttl_seconds: int,
        reusable: bool,
        raw_meta: dict[str, Any],
    ) -> dict[str, Any]:
        digest = sha256_bytes(body)
        blob_rel = Path("blobs") / "sha256" / digest[:2] / f"{digest}.bin"
        blob = self.root / blob_rel
        blob_valid = False
        if blob.is_file():
            try:
                blob_valid = sha256_bytes(blob.read_bytes()) == digest
            except OSError:
                blob_valid = False
        if not blob_valid:
            try:
                atomic_write(blob, body)
            except OSError as exc:
                raise BlobIntegrityError(
                    f"内容寻址 blob 无法原子写入：{digest}"
                ) from exc
        try:
            if sha256_bytes(blob.read_bytes()) != digest:
                raise BlobIntegrityError(f"内容寻址 blob 写后校验失败：{digest}")
        except BlobIntegrityError:
            raise
        except OSError as exc:
            raise BlobIntegrityError(f"内容寻址 blob 无法形成可信实物：{digest}") from exc
        meta = {
            **raw_meta,
            "schema_version": 1,
            "complete": True,
            "provider_id": provider,
            "fingerprint": fingerprint,
            "fetched_at": fetched_at.isoformat(),
            "expires_at": (fetched_at + timedelta(seconds=ttl_seconds)).isoformat(),
            "blob_path": blob_rel.as_posix(),
            "blob_hash": digest,
            "bytes": len(body),
        }
        run_id = raw_meta.get("run_id", "unknown")
        atomic_write(
            self.root / "raw-index" / provider / fingerprint / f"{run_id}.json",
            canonical_bytes(meta),
        )
        if reusable:
            atomic_write(
                self.root / "reusable" / provider / f"{fingerprint}.json",
                canonical_bytes(meta),
            )
        return meta

    def promote_reusable(
        self, provider: str, fingerprint: str, meta: dict[str, Any]
    ) -> dict[str, Any]:
        promoted = {
            **meta,
            "parse_ok": True,
            "schema_ok": True,
            "as_of_ok": True,
            "semantic_status": "OK",
        }
        atomic_write(
            self.root / "reusable" / provider / f"{fingerprint}.json",
            canonical_bytes(promoted),
        )
        return promoted

    def _baseline_material(self, baseline: dict[str, Any]) -> dict[str, Any]:
        return {k: v for k, v in baseline.items() if k not in {"created_at", "data_hash"}}

    def _baseline_valid(self, baseline: dict[str, Any]) -> bool:
        """Validate only the immutable baseline envelope, not every kind blob."""
        expected = baseline.get("data_hash")
        if not isinstance(expected, str) or len(expected) != 64:
            return False
        if sha256_bytes(canonical_bytes(self._baseline_material(baseline))) != expected:
            return False
        return True

    def _data_kind_blob_valid(self, entry: dict[str, Any]) -> bool:
        digest = str(entry.get("raw_hash", "")).lower()
        if len(digest) != 64:
            return False
        blob = self.root / "blobs" / "sha256" / digest[:2] / f"{digest}.bin"
        try:
            return sha256_bytes(blob.read_bytes()) == digest
        except OSError:
            return False

    @staticmethod
    def _selection_key(item: dict[str, Any]) -> tuple[str, bytes, str]:
        return (
            item.get("baseline_as_of", ""),
            canonical_bytes(item.get("watermark", {})),
            str(item.get("data_hash", "")).lower(),
        )

    def select_baseline(self, ticker: str, requested_as_of: str) -> dict[str, Any] | None:
        root = self.root / "companies" / ticker / "baselines"
        candidates: list[dict[str, Any]] = []
        if not root.is_dir():
            return None
        for path in root.glob("*.json"):
            try:
                item = json.loads(path.read_text(encoding="utf-8"))
            except (OSError, json.JSONDecodeError):
                continue
            if item.get("ticker") != ticker or item.get("baseline_as_of", "9999-12-31") > requested_as_of:
                continue
            if self._baseline_valid(item):
                candidates.append(item)
        return max(candidates, key=self._selection_key) if candidates else None

    @staticmethod
    def data_kind_state(
        entry: dict[str, Any] | None,
        requested_as_of: str,
        process_start: datetime,
        adapter_version: str = "1.0.0",
    ) -> str:
        if entry is None:
            return "MISSING"
        required = (
            "data_kind",
            "provider_id",
            "request_fingerprint",
            "ticker",
            "requested_as_of",
            "baseline_as_of",
            "data_date",
            "fetched_at",
            "expires_at",
            "adapter_version",
            "raw_hash",
            "schema_status",
            "as_of_status",
            "semantic_status",
            "watermark",
        )
        if any(key not in entry for key in required):
            return "STALE"
        try:
            expires = datetime.fromisoformat(entry["expires_at"])
        except (TypeError, ValueError):
            return "STALE"
        if entry["baseline_as_of"] > requested_as_of or entry["data_date"] > requested_as_of:
            return "STALE"
        if entry["requested_as_of"] != requested_as_of:
            return "STALE"
        publish = entry.get("publish_date")
        if entry.get("requires_publish_date", False) and (
            not publish or publish > requested_as_of
        ):
            return "STALE"
        if process_start > expires or entry["adapter_version"] != adapter_version:
            return "STALE"
        if entry["schema_status"] != "PASS" or entry["as_of_status"] != "PASS":
            return "STALE"
        if entry["semantic_status"] != "OK" or len(str(entry["raw_hash"])) != 64:
            return "STALE"
        kind = entry["data_kind"]
        if kind == "market_close" and not entry.get("is_latest_eligible_trade_date"):
            return "STALE"
        if kind == "shares_market_cap" and not entry.get("historical_capture_valid"):
            return "STALE"
        if kind.startswith("finance_") and not entry.get("required_periods_complete"):
            return "STALE"
        if kind == "forecast_summary" and not entry.get("summary_complete"):
            return "STALE"
        if kind == "forecast_detail" and not entry.get("detail_complete"):
            return "STALE"
        return "FRESH"

    def load_baseline_blob(
        self,
        entry: dict[str, Any],
        provider_id: str,
        request_fingerprint_value: str,
    ) -> dict[str, Any] | None:
        """Load a semantically validated per-kind baseline without a request index."""
        if (
            entry.get("provider_id") != provider_id
            or entry.get("request_fingerprint") != request_fingerprint_value
        ):
            return None
        digest = str(entry.get("raw_hash", "")).lower()
        if len(digest) != 64:
            return None
        blob = self.root / "blobs" / "sha256" / digest[:2] / f"{digest}.bin"
        try:
            body = blob.read_bytes()
        except OSError:
            return None
        if sha256_bytes(body) != digest:
            return None
        meta = {
            "schema_version": 1,
            "provider_id": provider_id,
            "fingerprint": request_fingerprint_value,
            "blob_path": blob.relative_to(self.root).as_posix(),
            "blob_hash": digest,
            "bytes": len(body),
            "http_status": 200,
            "transport_complete": True,
            "parse_ok": True,
            "schema_ok": True,
            "as_of_ok": True,
            "semantic_status": "OK",
            "fetched_at": entry["fetched_at"],
            "expires_at": entry["expires_at"],
            "data_kind": entry["data_kind"],
            "started_at": None,
            "finished_at": None,
            "remaining_before": None,
            "remaining_after": None,
            "attempts": [],
            "baseline_reused": True,
        }
        return {"body": body, "meta": meta}

    def baseline_states(
        self, ticker: str, requested_as_of: str, process_start: datetime
    ) -> tuple[dict[str, Any] | None, dict[str, str]]:
        baseline = self.select_baseline(ticker, requested_as_of)
        entries = (baseline or {}).get("data_kinds", {})
        all_kinds = {kind for kinds in PROVIDER_KINDS.values() for kind in kinds}
        states: dict[str, str] = {}
        for kind in sorted(all_kinds):
            entry = entries.get(kind)
            state = self.data_kind_state(entry, requested_as_of, process_start)
            if state == "FRESH" and not self._data_kind_blob_valid(entry):
                state = "STALE"
            states[kind] = state
        return baseline, states

    def baseline_entries_complete_and_valid(
        self, entries: dict[str, dict[str, Any]]
    ) -> bool:
        """A company baseline advances only when every kind has a trusted blob."""
        expected = {kind for kinds in PROVIDER_KINDS.values() for kind in kinds}
        return set(entries) == expected and all(
            self._data_kind_blob_valid(entries[kind]) for kind in expected
        )

    def fresh_provider_results(
        self, ticker: str, requested_as_of: str, process_start: datetime
    ) -> tuple[dict[str, dict[str, Any]], dict[str, str]]:
        baseline, states = self.baseline_states(ticker, requested_as_of, process_start)
        fresh: dict[str, dict[str, Any]] = {}
        if baseline:
            for name, kinds in PROVIDER_KINDS.items():
                if all(states.get(kind) == "FRESH" for kind in kinds):
                    result = json.loads(json.dumps(baseline["provider_results"][name]))
                    result["baseline_reused"] = True
                    result["request_telemetry"] = []
                    fresh[name] = result
        return fresh, states

    def write_baseline(
        self,
        ticker: str,
        baseline: dict[str, Any],
        supersede_data_hash: str | None = None,
    ) -> Path:
        baseline = dict(baseline)
        baseline["data_hash"] = sha256_bytes(canonical_bytes(self._baseline_material(baseline)))
        target = (
            self.root
            / "companies"
            / ticker
            / "baselines"
            / f"{baseline['baseline_as_of']}-{baseline['data_hash']}.json"
        )
        lock_path = self.root / "companies" / ticker / ".current.lock"
        with _ticker_lock(lock_path):
            atomic_write(target, canonical_bytes(baseline))
            if supersede_data_hash and supersede_data_hash != baseline["data_hash"]:
                old = (
                    self.root
                    / "companies"
                    / ticker
                    / "baselines"
                    / f"{baseline['baseline_as_of']}-{supersede_data_hash}.json"
                )
                if old.is_file():
                    history = self.root / "companies" / ticker / "history" / old.name
                    history.parent.mkdir(parents=True, exist_ok=True)
                    os.replace(old, history)
            current = self.root / "companies" / ticker / "current.json"
            candidate = {
                "baseline_as_of": baseline["baseline_as_of"],
                "watermark": baseline.get("watermark", {}),
                "data_hash": baseline["data_hash"].lower(),
                "path": target.relative_to(self.root).as_posix(),
            }
            existing = None
            if current.is_file():
                try:
                    existing = json.loads(current.read_text(encoding="utf-8"))
                except (OSError, json.JSONDecodeError):
                    pass
            existing_is_superseded = (
                existing is not None
                and supersede_data_hash is not None
                and existing.get("data_hash") == supersede_data_hash.lower()
            )
            if (
                existing is None
                or existing_is_superseded
                or self._selection_key(candidate) > self._selection_key(existing)
            ):
                atomic_write(current, canonical_bytes(candidate))
        return target