"""Python 3.10+ client. Uses only the standard library; keep API keys server-side."""
import json
import os
import re
import shutil
import time
import uuid
from concurrent.futures import ThreadPoolExecutor
from email.utils import parsedate_to_datetime
from http.client import HTTPException
from pathlib import Path
from urllib.error import HTTPError, URLError
from urllib.parse import urljoin, urlsplit
from urllib.request import HTTPRedirectHandler, Request, build_opener

MAX_BYTES = 15 * 1024 * 1024
TYPES = {".png": "image/png", ".jpg": "image/jpeg", ".jpeg": "image/jpeg", ".webp": "image/webp"}


class ApiError(Exception):
    def __init__(self, code: str, status: int = 0):
        super().__init__(code)
        self.code, self.status = code, status
        self.items: list[dict[str, int | str | None]] = []
        self.idempotency_key: str | None = None
        self.batch_id: str | None = None
        self.status_url: str | None = None


class _NoRedirect(HTTPRedirectHandler):
    def redirect_request(self, req, fp, code, msg, headers, newurl):
        return None


def _safe_origin(origin):
    value = urlsplit(origin)
    return bool(value.hostname) and (value.scheme == "https" or value.scheme == "http" and value.hostname in ("localhost", "127.0.0.1", "::1")) and not value.username and not value.password


def _client(origin, api_key):
    parts = urlsplit(origin)
    if not _safe_origin(origin) or parts.path not in ("", "/") or parts.query or parts.fragment or not api_key:
        raise ApiError("invalid_client_options")

    def url(path):
        if not isinstance(path, str) or not path.startswith("/api/v1/") or "\\" in path:
            raise ApiError("invalid_api_url")
        return urljoin(origin.rstrip("/") + "/", path)

    return url, {"Authorization": "Bearer " + api_key}


def _deadline(value):
    from datetime import datetime
    try:
        return datetime.fromisoformat(value.replace("Z", "+00:00")).timestamp()
    except (AttributeError, TypeError, ValueError):
        raise ApiError("invalid_deadline") from None


def _delay(value, attempt):
    if not value:
        return 2 ** attempt
    try:
        return max(0, float(value))
    except ValueError:
        try:
            return max(0, parsedate_to_datetime(value).timestamp() - time.time())
        except (TypeError, ValueError):
            return 2 ** attempt


def _request(url, make_request, deadline=None, binary=False):
    deadline = deadline if deadline is not None else time.time() + 240
    for attempt in range(4):
        remaining = deadline - time.time()
        if remaining <= 0:
            raise ApiError("request_deadline")
        handle = None
        delay, status, code = 2 ** attempt, 0, "network_error"
        try:
            request, handle = make_request(url)
            response = build_opener(_NoRedirect()).open(request, timeout=min(90, remaining))
            if binary:
                return response
            with response:
                try:
                    return json.load(response)
                except (ValueError, UnicodeError):
                    code = "invalid_response"
        except HTTPError as error:
            status, code = error.code, "http_" + str(error.code)
            delay = _delay(error.headers.get("Retry-After"), attempt)
            with error:
                if "application/json" in error.headers.get("Content-Type", ""):
                    try:
                        value = json.loads(error.read(65536)).get("code", "")
                        if re.fullmatch(r"[a-z0-9_]+", value):
                            code = value
                    except (ValueError, TypeError, AttributeError, HTTPException):
                        pass
            if status not in (408, 429) and status < 500:
                raise ApiError(code, status) from None
        except (URLError, OSError, TimeoutError, HTTPException):
            pass
        finally:
            if handle is not None:
                handle.close()
        if attempt == 3 or time.time() + delay >= deadline:
            raise ApiError(code, status) from None
        time.sleep(delay)


def upload_images(paths, *, origin, api_key, idempotency_key=None):
    """Upload one path or 1-20 paths. Reuse idempotency_key to resume the same files."""
    key = idempotency_key if idempotency_key is not None else str(uuid.uuid4())
    if not isinstance(key, str) or not re.fullmatch(r"[A-Za-z0-9._:-]{8,128}", key):
        raise ApiError("invalid_idempotency_key")
    url, headers = _client(origin, api_key)
    names = [paths] if isinstance(paths, (str, os.PathLike)) else list(paths)
    if not 1 <= len(names) <= 20:
        raise ApiError("invalid_image_count")
    files = []
    for name in names:
        path = Path(name)
        size, content_type = path.stat().st_size, TYPES.get(path.suffix.lower())
        if not path.is_file() or not content_type or not 0 < size <= MAX_BYTES:
            raise ApiError("invalid_file")
        files.append({"filename": path.name, "contentType": content_type, "sizeBytes": size})
    batch = None
    try:
        body = json.dumps({"files": files}).encode()
        batch = _request(url("/api/v1/images"), lambda target: (Request(target, data=body, method="POST",
            headers={**headers, "Content-Type": "application/json", "Idempotency-Key": key}), None))
        deadline = _deadline(batch["uploadExpiresAt"])

        def transfer(upload):
            try:
                ordinal = upload["ordinal"]
                if not isinstance(ordinal, int) or not 0 <= ordinal < len(files):
                    raise ApiError("invalid_upload")
                file = files[ordinal]
                if not _safe_origin(upload["url"]) or upload["contentType"] != file["contentType"]:
                    raise ApiError("invalid_upload")

                def put(target):
                    handle = open(names[ordinal], "rb")
                    if os.fstat(handle.fileno()).st_size != file["sizeBytes"]:
                        handle.close()
                        raise ApiError("file_changed")
                    return Request(target, data=handle, method="PUT", headers={
                        "Content-Type": file["contentType"], "Content-Length": str(file["sizeBytes"])}), handle

                with _request(upload["url"], put, deadline, binary=True):
                    pass
                _request(url(upload["confirmUrl"]), lambda target: (Request(target, data=b"", method="POST", headers=headers), None), deadline)
                return None
            except (ApiError, OSError, HTTPException, KeyError, TypeError, ValueError) as error:
                return {"ordinal": upload.get("ordinal"), "code": error.code if isinstance(error, ApiError) else "upload_error"}

        with ThreadPoolExecutor(max_workers=2) as pool:
            errors = [error for error in pool.map(transfer, batch.get("uploads", [])) if error]
        if errors:
            error = ApiError("upload_incomplete")
            error.items = errors
            raise error
        result = _request(url(batch["statusUrl"]), lambda target: (Request(target, headers=headers), None))
        return {**result, "idempotencyKey": key}
    except (ApiError, OSError, HTTPException, KeyError, TypeError, ValueError) as error:
        failure = error if isinstance(error, ApiError) else ApiError("upload_error")
        failure.idempotency_key = key
        if batch:
            failure.batch_id, failure.status_url = batch["id"], batch["statusUrl"]
        raise failure from None


def wait_for_batch(batch, *, origin, api_key):
    """Wait for terminal per-image outcomes, including partial success."""
    url, headers = _client(origin, api_key)
    key = batch.get("idempotencyKey")
    deadline = _deadline(batch["processingDeadlineAt"])
    while batch.get("pollAfterSeconds"):
        delay = max(1, float(batch["pollAfterSeconds"]))
        if time.time() + delay >= deadline:
            raise ApiError("processing_deadline")
        time.sleep(delay)
        batch = _request(url(batch["statusUrl"]), lambda target: (Request(target, headers=headers), None), deadline)
    return {**batch, **({"idempotencyKey": key} if key else {})}


def download_result(path, destination, *, origin, api_key):
    """Save an image downloadUrl or batch zipUrl without overwriting an existing file."""
    url, headers = _client(origin, api_key)
    with _request(url(path), lambda target: (Request(target, headers=headers), None), binary=True) as response:
        try:
            with open(destination, "xb") as output:
                shutil.copyfileobj(response, output, length=65536)
        except OSError:
            raise ApiError("download_failed") from None