"""Bounded authenticated object reads and expiring CDN grants for private assets."""
from pathlib import Path
import os
import re
import hashlib
from http.client import HTTPException
import threading
import time
from urllib.parse import parse_qs, quote, urlsplit

import requests
from urllib3.exceptions import HTTPError as TransportError

BUCKET = os.environ.get('SUPRATEX_DATA_BUCKET', 'UniLat3D/supra-tex-models')
MOUNT = Path('/data')
CONNECT_TIMEOUT = 8
READ_TIMEOUT = 20
ATTEMPT_SECONDS = 180
ATTEMPTS = 3
_guard = threading.Lock()
_grants = {}


def bucket_url(path):
    # Lexical paths come from the Store's verified, account-owned manifest.
    # Do not resolve/stat the remote mount merely to create a download URL.
    absolute = Path(os.path.abspath(os.fspath(path)))
    if not absolute.is_relative_to(MOUNT) or '..' in absolute.parts:
        return None
    key = absolute.relative_to(MOUNT).as_posix()
    return f'https://huggingface.co/buckets/{BUCKET}/resolve/' + quote(key, safe='')


def _headers(offset=0):
    from huggingface_hub.utils import build_hf_headers
    headers = dict(build_hf_headers())
    headers['Accept-Encoding'] = 'identity'
    if offset:
        headers['Range'] = f'bytes={offset}-'
    return headers


def describe(path):
    url = bucket_url(path)
    if not url:
        return None
    headers = _headers(); headers['Range'] = 'bytes=0-0'
    with requests.get(url,headers=headers,stream=True,timeout=(CONNECT_TIMEOUT,READ_TIMEOUT)) as response:
        if response.status_code == 404:
            raise FileNotFoundError('Artifact not found')
        response.raise_for_status()
        match = re.fullmatch(r'bytes 0-0/(\d+)',response.headers.get('Content-Range',''))
        if not match:
            raise ValueError('Artifact storage did not provide its size')
        identity = str(path)+'\0'+response.headers.get('ETag','')+'\0'+match[1]
        return {'path':str(path),'size':int(match[1]),'version':hashlib.sha256(identity.encode()).hexdigest()}


def _signed_target(value):
    parsed = urlsplit(value)
    if (parsed.scheme != 'https' or parsed.username or parsed.password or parsed.port not in (None, 443)
            or not parsed.hostname or not (parsed.hostname.endswith('.cdn.hf.co')
                or parsed.hostname.endswith('.xethub.hf.co'))):
        return None
    fields = parse_qs(parsed.query)
    expires = fields.get('Expires', [''])[0]
    if not expires.isdigit() or not fields.get('Signature'):
        return None
    remaining = int(expires) - time.time()
    # Never pass an unbounded or long-lived bearer capability to a browser.
    if not 20 <= remaining <= 3660:
        return None
    return int(expires)


def cdn_grant(owner, descriptor):
    """Return an object-scoped grant; caller must authorize ownership first."""
    origin = bucket_url(descriptor['path'])
    if not origin or os.environ.get('SUPRATEX_DIRECT_ASSETS', '1') == '0':
        return None
    key = (owner, descriptor['path'], descriptor['version'])
    now = time.time()
    with _guard:
        cached = _grants.get(key)
        if cached and cached['expires_at'] > now + 20:
            return dict(cached)
    try:
        # Fetch headers only, with no credential forwarding to the CDN.
        headers = _headers()
        headers['Range'] = 'bytes=0-0'
        with requests.get(origin, headers=headers, stream=True, allow_redirects=False,
                          timeout=(3, 6)) as response:
            target = response.headers.get('Location', '')
            expires = _signed_target(target) if response.status_code in (302, 303, 307) else None
            if not expires:
                return None
        result = {'url': target, 'expires_at': expires}
        with _guard:
            for old in list(_grants):
                if _grants[old]['expires_at'] < now + 20:
                    _grants.pop(old, None)
            while len(_grants) >= 256:
                _grants.pop(next(iter(_grants)))
            _grants[key] = result
        return dict(result)
    except (requests.RequestException, ValueError, OSError):
        return None


def download(path, destination, size, progress=None):
    """Read the mount's object over finite HTTP attempts with byte-range resume."""
    url = bucket_url(path)
    if not url:
        return False
    return download_url(url, destination, size, progress=progress)


def download_url(url, destination, size, progress=None, *, authenticated=True):
    """The same finite transport for pinned public fallbacks and private data."""
    destination = Path(destination)
    for attempt in range(ATTEMPTS):
        offset = destination.stat().st_size if destination.exists() else 0
        try:
            began = time.monotonic()
            headers = _headers(offset) if authenticated else {'Accept-Encoding':'identity', **({'Range':f'bytes={offset}-'} if offset else {})}
            with requests.get(url, headers=headers, stream=True,
                              timeout=(CONNECT_TIMEOUT, READ_TIMEOUT)) as response:
                if response.status_code == 404:
                    raise FileNotFoundError('Artifact not found')
                if response.status_code == 200 and offset:
                    offset = 0
                elif response.status_code == 206:
                    match = re.fullmatch(r'bytes (\d+)-(\d+)/(\d+)', response.headers.get('Content-Range', ''))
                    if not match or int(match[1]) != offset or int(match[2]) != size-1 or int(match[3]) != size:
                        raise ValueError('Invalid artifact byte range')
                elif response.status_code != 200:
                    raise ValueError('Artifact storage rejected the request')
                if response.headers.get('Content-Encoding', 'identity') != 'identity':
                    raise ValueError('Unexpected artifact encoding')
                completed = offset
                with destination.open('ab' if offset else 'wb') as output:
                    while True:
                        if time.monotonic() - began > ATTEMPT_SECONDS:
                            raise TimeoutError('Artifact storage transfer timed out')
                        read = getattr(response.raw, 'read1', None)
                        chunk = read(1024**2, decode_content=False) if read else response.raw._fp.read1(1024**2)
                        if not chunk:
                            break
                        completed += len(chunk)
                        if completed > size:
                            raise ValueError('Artifact exceeds its recorded size')
                        output.write(chunk)
                        if progress:
                            progress(completed, size)
                if completed != size:
                    raise ValueError('Artifact transfer was interrupted')
                return True
        except FileNotFoundError:
            raise
        except (requests.RequestException, TransportError, HTTPException, OSError, ValueError, TimeoutError):
            if attempt + 1 == ATTEMPTS:
                raise RuntimeError('Artifact storage is temporarily unavailable. Please retry.') from None
            time.sleep(min(2**attempt, 4))
    return False