#!/usr/bin/env python3
"""Zero-dependency Hugging Bay SDK and CLI.

Download:
    curl -fsSL https://huggingbay.xyz/sdk/hugging_bay.py -o hugging_bay.py

Use as a module:
    from hugging_bay import HuggingBayClient
    client = HuggingBayClient()
    rows = client.search("rag embedding", limit=5)["rows"]

Use as a CLI:
    python hugging_bay.py search "rag embedding" --limit 5
"""

from __future__ import annotations

import argparse
import hashlib
import json
import os
import sys
import tempfile
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Dict, Iterable, Mapping, Optional
from urllib.error import HTTPError, URLError
from urllib.parse import parse_qsl, quote, urlencode, urlparse
from urllib.request import HTTPRedirectHandler, Request, build_opener, urlopen


Json = Dict[str, Any]


class HuggingBayError(RuntimeError):
    """Raised when the Hugging Bay API returns an error response."""

    def __init__(self, message: str, *, status: int = 0, payload: Optional[Json] = None) -> None:
        super().__init__(message)
        self.status = status
        self.payload = payload or {}


class DownloadError(RuntimeError):
    """Raised when a hosted download cannot be verified and attested."""


class _RejectRedirects(HTTPRedirectHandler):
    def redirect_request(self, req, fp, code, msg, headers, newurl):  # noqa: ANN001
        return None


def _open_without_redirect(request: Request, timeout: float):
    return build_opener(_RejectRedirects()).open(request, timeout=timeout)


def _response_status(response: Any) -> int:
    status = getattr(response, "status", None)
    return int(status if status is not None else response.getcode())


def _trusted_gcs_signed_url(value: str) -> bool:
    """Accept only short-lived, HTTPS GCS V4 signed URLs."""
    try:
        parsed = urlparse(str(value))
        host = (parsed.hostname or "").lower()
        if (
            parsed.scheme != "https"
            or not (host == "storage.googleapis.com" or host.endswith(".storage.googleapis.com"))
            or parsed.username
            or parsed.password
            or parsed.fragment
        ):
            return False
        params = {key.lower(): item for key, item in parse_qsl(parsed.query, keep_blank_values=True)}
        expires = float(params.get("x-goog-expires", "nan"))
        return (
            all(params.get(key) for key in (
                "x-goog-algorithm",
                "x-goog-credential",
                "x-goog-date",
                "x-goog-signature",
                "generation",
            ))
            and 0 < expires <= 300
        )
    except (TypeError, ValueError):
        return False


@dataclass
class HuggingBayClient:
    """Small public Hugging Bay API client for agents, scripts, and notebooks."""

    base_url: str = os.environ.get("HUGGING_BAY_API", "https://huggingbay.xyz")
    token: str = os.environ.get("HUGGING_BAY_TOKEN", "")
    timeout: float = 20.0
    user_agent: str = "hugging-bay-python-sdk/0.1"

    def _url(self, path: str, params: Optional[Mapping[str, Any]] = None) -> str:
        base = self.base_url.rstrip("/")
        query = ""
        if params:
            clean = {
                key: value
                for key, value in params.items()
                if value is not None and value != "" and value != []
            }
            if clean:
                query = "?" + urlencode(clean, doseq=True)
        return f"{base}{path if path.startswith('/') else '/' + path}{query}"

    def request(
        self,
        method: str,
        path: str,
        *,
        params: Optional[Mapping[str, Any]] = None,
        body: Optional[Mapping[str, Any]] = None,
        token: Optional[str] = None,
    ) -> Json:
        headers = {
            "Accept": "application/json",
            "User-Agent": self.user_agent,
        }
        active_token = self.token if token is None else token
        if active_token:
            headers["Authorization"] = f"Bearer {active_token}"
        data = None
        if body is not None:
            data = json.dumps(body).encode("utf-8")
            headers["Content-Type"] = "application/json"
        request = Request(self._url(path, params), data=data, headers=headers, method=method.upper())
        try:
            with urlopen(request, timeout=self.timeout) as response:
                raw = response.read().decode("utf-8")
                return json.loads(raw) if raw else {}
        except HTTPError as error:
            raw = error.read().decode("utf-8", errors="replace")
            try:
                payload = json.loads(raw) if raw else {}
            except json.JSONDecodeError:
                payload = {"text": raw}
            raise HuggingBayError(payload.get("error") or error.reason, status=error.code, payload=payload) from error
        except URLError as error:
            raise HuggingBayError(str(error.reason or error), status=0) from error

    def health(self) -> Json:
        return self.request("GET", "/api/health")

    def agent_discovery(self) -> Json:
        return self.request("GET", "/api/agent-discovery")

    def search(
        self,
        q: str = "",
        *,
        limit: int = 20,
        cursor: str = "0",
        sort: str = "best",
        type: str = "",
        license: str = "",
        hosting: str = "",
        source: str = "",
        works_with: str = "",
        review_status: str = "",
        semantic: bool = False,
        explain: bool = False,
    ) -> Json:
        return self.request("GET", "/api/artifacts", params={
            "summary": "1",
            "q": q,
            "limit": limit,
            "cursor": cursor,
            "sort": sort,
            "type": type,
            "license": license,
            "hosting": hosting,
            "source": source,
            "worksWith": works_with,
            "reviewStatus": review_status,
            "semantic": "1" if semantic else "",
            "explain": "1" if explain else "",
        })

    def resolve(self, repo: str, *, source: str = "") -> Json:
        return self.request("GET", "/api/resolve", params={"repo": repo, "source": source})

    def artifact(self, artifact_id: str) -> Json:
        return self.request("GET", f"/api/artifacts/{artifact_id}")

    def artifact_metadata(self, artifact_id: str) -> Json:
        return self.request("GET", f"/api/artifacts/{artifact_id}/metadata.json")

    def artifact_bundle(self, artifact_id: str) -> Json:
        return self.request("GET", f"/api/artifacts/{artifact_id}/agent-bundle")

    def artifact_versions(self, artifact_id: str) -> Json:
        return self.request("GET", f"/api/artifacts/{artifact_id}/versions")

    def artifact_files(self, artifact_id: str, *, limit: int = 100, cursor: str = "0", tool: str = "") -> Json:
        return self.request("GET", f"/api/artifacts/{artifact_id}/files", params={
            "limit": limit,
            "cursor": cursor,
            "tool": tool,
        })

    def download_plan(self, artifact_id: str, *, tool: str = "") -> Json:
        return self.request("GET", f"/api/artifacts/{artifact_id}/download-plan", params={"tool": tool})

    def download_file(
        self,
        artifact_id: str,
        path: str,
        destination: str,
        *,
        sha256: str,
        size_bytes: int,
    ) -> Json:
        """Stream, verify, attest, and atomically publish one hosted file."""
        expected_sha256 = str(sha256 or "").strip().lower()
        if expected_sha256.startswith("sha256:"):
            expected_sha256 = expected_sha256[7:]
        if len(expected_sha256) != 64 or any(character not in "0123456789abcdef" for character in expected_sha256):
            raise ValueError("sha256 must be a 64-character hexadecimal digest")
        try:
            expected_size = int(size_bytes)
        except (TypeError, ValueError) as error:
            raise ValueError("size_bytes must be a non-negative integer") from error
        if expected_size < 0 or expected_size != size_bytes:
            raise ValueError("size_bytes must be a non-negative integer")

        encoded_id = quote(str(artifact_id), safe="")
        encoded_path = "/".join(quote(part, safe="") for part in str(path).split("/"))
        initial_headers = {
            "Accept": "application/octet-stream",
            "User-Agent": self.user_agent,
        }
        if self.token:
            initial_headers["Authorization"] = f"Bearer {self.token}"
        initial_request = Request(
            self._url(f"/api/downloads/{encoded_id}/{encoded_path}"),
            headers=initial_headers,
            method="GET",
        )
        try:
            initial = _open_without_redirect(initial_request, self.timeout)
        except HTTPError as error:
            if error.code != 302:
                raise DownloadError(f"download redirect required, got HTTP {error.code}") from error
            location = error.headers.get("Location", "")
            completion_token = error.headers.get("X-Hugging-Bay-Completion-Token", "")
            error.close()
        except URLError as error:
            raise DownloadError(str(error.reason or error)) from error
        else:
            try:
                status = _response_status(initial)
                if status != 302:
                    raise DownloadError(f"download redirect required, got HTTP {status}")
                location = initial.headers.get("Location", "")
                completion_token = initial.headers.get("X-Hugging-Bay-Completion-Token", "")
            finally:
                initial.close()

        if not completion_token:
            raise DownloadError("download completion capability missing")
        if not _trusted_gcs_signed_url(location):
            raise DownloadError("untrusted GCS download redirect")

        target = Path(destination).expanduser()
        target.parent.mkdir(parents=True, exist_ok=True)
        temporary_path: Optional[Path] = None
        try:
            gcs_request = Request(
                location,
                headers={
                    "Accept": "application/octet-stream",
                    "User-Agent": self.user_agent,
                },
                method="GET",
            )
            try:
                gcs = _open_without_redirect(gcs_request, self.timeout)
            except HTTPError as error:
                raise DownloadError(f"GCS download failed with HTTP {error.code}") from error
            except URLError as error:
                raise DownloadError(str(error.reason or error)) from error
            try:
                status = _response_status(gcs)
                if status != 200:
                    raise DownloadError(f"GCS download failed with HTTP {status}")
                declared_size = gcs.headers.get("Content-Length")
                if declared_size and declared_size.isdigit() and int(declared_size) != expected_size:
                    raise DownloadError("GCS content length does not match the download plan")

                file_descriptor, temporary_name = tempfile.mkstemp(
                    dir=target.parent,
                    prefix=f".{target.name}.hbay-",
                    suffix=".part",
                )
                os.close(file_descriptor)
                temporary_path = Path(temporary_name)
                digest = hashlib.sha256()
                received = 0
                with temporary_path.open("wb") as output:
                    while True:
                        chunk = gcs.read(1024 * 1024)
                        if not chunk:
                            break
                        received += len(chunk)
                        if received > expected_size:
                            raise DownloadError("GCS response exceeded the download plan size")
                        digest.update(chunk)
                        output.write(chunk)
                    output.flush()
                    os.fsync(output.fileno())
            finally:
                gcs.close()

            actual_sha256 = digest.hexdigest()
            if received != expected_size or actual_sha256 != expected_sha256:
                raise DownloadError(
                    f"download verification failed: expected {expected_sha256}/{expected_size}, "
                    f"got {actual_sha256}/{received}"
                )
            completion = self.download_completion(
                artifact_id,
                path,
                sha256=actual_sha256,
                size_bytes=received,
                completion_token=completion_token,
            )
            completion_status = completion.get("status") if isinstance(completion, dict) else None
            if (
                completion_status not in {"client-attested", "already-recorded"}
                or completion.get("localHashAttested") is not True
            ):
                raise DownloadError("download completion was not recorded")
            if temporary_path is None:
                raise DownloadError("download temporary file missing")
            os.replace(temporary_path, target)
            temporary_path = None
            return {
                "status": completion_status,
                "destination": str(target),
                "sha256": actual_sha256,
                "sizeBytes": received,
                "completion": completion,
            }
        finally:
            if temporary_path is not None:
                try:
                    temporary_path.unlink()
                except FileNotFoundError:
                    pass

    def download_completion(
        self,
        artifact_id: str,
        path: str,
        *,
        sha256: str,
        size_bytes: int,
        completion_token: str,
    ) -> Json:
        """Close the honest download-completion loop.

        Call ONLY after downloading the file in full and locally recomputing its
        SHA-256 so it matches the download-plan's file.sha256. ``completion_token``
        is the single-use capability returned in the ``X-Hugging-Bay-Completion-Token``
        response header of the ``GET /api/downloads/{id}/{path}`` redirect; without
        it the pull is not counted. Redirect issuance is a download initiation, not a
        completed pull, and probes/HEAD/partial reads never count.
        """
        encoded_path = quote(path, safe="/")
        return self.request(
            "POST",
            f"/api/download-completions/{artifact_id}/{encoded_path}",
            body={
                "sha256": sha256,
                "sizeBytes": int(size_bytes),
                "completionToken": completion_token,
            },
        )

    def trust_bundle(self, artifact_id: str) -> Json:
        return self.request("GET", f"/api/trust-bundles/{artifact_id}")

    def reviews(self, artifact_id: str, *, limit: int = 50) -> Json:
        return self.request("GET", f"/api/artifacts/{artifact_id}/reviews", params={"limit": limit})

    def review_summary(self, artifact_id: str) -> Json:
        return self.request("GET", f"/api/artifacts/{artifact_id}/review-summary")

    def downloadable_files(self, *, artifact_id: str = "", limit: int = 100, tool: str = "", pack: str = "") -> Json:
        return self.request("GET", "/api/downloadable-files", params={
            "artifactId": artifact_id,
            "limit": limit,
            "tool": tool,
            "pack": pack,
        })

    def ranking(self, slug: str, *, limit: int = 50) -> Json:
        return self.request("GET", f"/api/rankings/{slug}", params={"limit": limit})

    def trending(self, *, window: str = "7d", limit: int = 25) -> Json:
        return self.request("GET", "/api/trending", params={"window": window, "limit": limit})

    def mirror_request(self, artifact_id: str, *, use_case: str = "", target_tool: str = "") -> Json:
        return self.request("POST", f"/api/artifacts/{artifact_id}/mirror-request", body={
            "useCase": use_case or "python-sdk",
            "targetTool": target_tool,
        })

    def source_request(self, repo: str, *, source: str = "", use_case: str = "", target_tool: str = "") -> Json:
        return self.request("POST", "/api/source-requests", body={
            "repo": repo,
            "source": source,
            "useCase": use_case or "python-sdk",
            "targetTool": target_tool,
        })


def _print_json(payload: Any) -> None:
    print(json.dumps(payload, indent=2, sort_keys=True))


def _rows(payload: Json) -> Iterable[Json]:
    rows = payload.get("rows")
    return rows if isinstance(rows, list) else []


def main(argv: Optional[list[str]] = None) -> int:
    parser = argparse.ArgumentParser(description="Hugging Bay Python SDK CLI")
    parser.add_argument("--api", default=os.environ.get("HUGGING_BAY_API", "https://huggingbay.xyz"))
    parser.add_argument("--token", default=os.environ.get("HUGGING_BAY_TOKEN", ""))
    sub = parser.add_subparsers(dest="command", required=True)

    health = sub.add_parser("health")
    health.set_defaults(func=lambda client, args: client.health())

    search = sub.add_parser("search")
    search.add_argument("query", nargs="?", default="")
    search.add_argument("--limit", type=int, default=20)
    search.add_argument("--sort", default="best")
    search.add_argument("--type", default="")
    search.add_argument("--hosting", default="")
    search.add_argument("--semantic", action="store_true")
    search.set_defaults(func=lambda client, args: client.search(
        args.query,
        limit=args.limit,
        sort=args.sort,
        type=args.type,
        hosting=args.hosting,
        semantic=args.semantic,
        explain=True,
    ))

    resolve = sub.add_parser("resolve")
    resolve.add_argument("repo")
    resolve.add_argument("--source", default="")
    resolve.set_defaults(func=lambda client, args: client.resolve(args.repo, source=args.source))

    artifact = sub.add_parser("artifact")
    artifact.add_argument("id")
    artifact.set_defaults(func=lambda client, args: client.artifact_bundle(args.id))

    versions = sub.add_parser("versions")
    versions.add_argument("id")
    versions.set_defaults(func=lambda client, args: client.artifact_versions(args.id))

    files = sub.add_parser("files")
    files.add_argument("id")
    files.add_argument("--limit", type=int, default=100)
    files.add_argument("--tool", default="")
    files.set_defaults(func=lambda client, args: client.artifact_files(args.id, limit=args.limit, tool=args.tool))

    downloads = sub.add_parser("downloads")
    downloads.add_argument("--artifact-id", default="")
    downloads.add_argument("--limit", type=int, default=100)
    downloads.add_argument("--tool", default="")
    downloads.add_argument("--pack", default="")
    downloads.set_defaults(func=lambda client, args: client.downloadable_files(
        artifact_id=args.artifact_id,
        limit=args.limit,
        tool=args.tool,
        pack=args.pack,
    ))

    ranking = sub.add_parser("ranking")
    ranking.add_argument("slug")
    ranking.add_argument("--limit", type=int, default=50)
    ranking.set_defaults(func=lambda client, args: client.ranking(args.slug, limit=args.limit))

    trending = sub.add_parser("trending")
    trending.add_argument("--window", default="7d")
    trending.add_argument("--limit", type=int, default=25)
    trending.set_defaults(func=lambda client, args: client.trending(window=args.window, limit=args.limit))

    args = parser.parse_args(argv)
    client = HuggingBayClient(base_url=args.api, token=args.token)
    try:
        payload = args.func(client, args)
    except HuggingBayError as error:
        _print_json({"error": str(error), "status": error.status, "payload": error.payload})
        return 1
    _print_json(payload)
    return 0


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