from __future__ import annotations

import base64
import hashlib
import json
import logging
import random
import re
import time
import urllib.error
import urllib.parse
import urllib.request
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Callable, Sequence

from niannian.storage.cloud_base import parse_remote_root_parts
from niannian.storage.upload_checkpoint import ChunkedUploadHooks, UploadCheckpointStore
from niannian.storage.upload_progress import UploadProgressTracker

logger = logging.getLogger(__name__)

DEFAULT_QUARK_API_BASE = "https://drive-pc.quark.cn/1/clouddrive"
QUARK_BROWSER_UA = (
    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
    "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/134.0.0.0 Safari/537.36"
)
MAX_SINGLE_UPLOAD = 5 * 1024 * 1024
DEFAULT_CHUNK_SIZE_BYTES = 5 * 1024 * 1024
DEFAULT_MAX_UPLOAD_BYTES = 512 * 1024 * 1024
QUARK_AUTH_ERROR_CODES = {31004, "31004"}
DEFAULT_PERSIST_INTERVAL_SECONDS = 60.0
HttpOpener = Callable[..., Any]


class QuarkAPIError(Exception):
    pass


def _split_chunks(data: bytes, chunk_size: int) -> list[bytes]:
    if chunk_size <= 0:
        raise ValueError("chunk_size must be positive")
    return [data[i : i + chunk_size] for i in range(0, len(data), chunk_size)]


class QuarkDriveClient:
    """夸克网盘 Cookie 鉴权 — 小文件 upload/download（pre + OSS 单分片）。"""

    def __init__(
        self,
        cookie: str,
        *,
        remote_root: str | Sequence[str] | None = None,
        api_base: str = DEFAULT_QUARK_API_BASE,
        timeout: float = 60.0,
        http_opener: HttpOpener | None = None,
        state_path: Path | None = None,
        credential_manager: Any | None = None,
        persist_interval_seconds: float = DEFAULT_PERSIST_INTERVAL_SECONDS,
        chunk_size_bytes: int = DEFAULT_CHUNK_SIZE_BYTES,
        max_upload_bytes: int = DEFAULT_MAX_UPLOAD_BYTES,
    ) -> None:
        self._cookie = cookie
        if isinstance(remote_root, str) or remote_root is None:
            self._root_parts = parse_remote_root_parts(remote_root)
        else:
            self._root_parts = [part for part in remote_root if part]
            if not self._root_parts:
                raise QuarkAPIError("remote_root must not be empty")
        self._api_base = api_base.rstrip("/")
        self._timeout = timeout
        self._open = http_opener or urllib.request.urlopen
        self._state_path = state_path
        self._credential_manager = credential_manager
        self._persist_interval_seconds = float(persist_interval_seconds)
        self._chunk_size_bytes = min(int(chunk_size_bytes), MAX_SINGLE_UPLOAD)
        self._max_upload_bytes = int(max_upload_bytes)
        self._last_persist_at = 0.0
        self._last_persisted_cookie = cookie
        self._root_fid: str | None = None
        self._dir_cache: dict[str, str] = {}
        self._session_token: str | None = None
        self._load_state()

    @property
    def cookie(self) -> str:
        return self._cookie

    def upload_object(
        self,
        rel_path: str,
        data: bytes,
        *,
        progress: UploadProgressTracker | None = None,
        checkpoint: UploadCheckpointStore | None = None,
        job_id: str | None = None,
    ) -> None:
        if len(data) > self._max_upload_bytes:
            logger.warning(
                "Quark upload rejected: %s exceeds max_upload_bytes (%d > %d)",
                rel_path,
                len(data),
                self._max_upload_bytes,
            )
            raise QuarkAPIError(
                f"file too large: {rel_path} ({len(data)} bytes > max {self._max_upload_bytes})"
            )
        parent_fid, file_name = self._ensure_parent_dir(rel_path)
        self._delete_if_exists(parent_fid, file_name)
        if len(data) <= MAX_SINGLE_UPLOAD:
            self._upload_single_part(parent_fid, file_name, data)
            return
        hooks = ChunkedUploadHooks(
            provider="quark",
            rel_path=rel_path,
            remote_root="/".join(self._root_parts),
            data=data,
            chunk_size=self._chunk_size_bytes,
            progress=progress,
            checkpoint=checkpoint,
            job_id=job_id,
        )
        if hooks.enabled:
            resume = hooks.begin()
            try:
                self._upload_multipart(parent_fid, file_name, data, hooks=hooks, resume=resume)
                hooks.done()
            except Exception as exc:
                hooks.failed(str(exc))
                raise
        else:
            self._upload_multipart(parent_fid, file_name, data)

    def download_object(self, rel_path: str) -> bytes | None:
        parent_fid, file_name = self._parent_parts(rel_path)
        parent = self._ensure_dir_chain(parent_fid)
        fid = self._find_child(parent, file_name, want_dir=False)
        if not fid:
            return None
        url = self._download_url(fid)
        req = urllib.request.Request(
            url,
            method="GET",
            headers={
                "User-Agent": QUARK_BROWSER_UA,
                "Referer": "https://pan.quark.cn/",
                "Cookie": self._cookie,
            },
        )
        with self._open(req, timeout=self._timeout) as resp:
            return resp.read()

    def _load_state(self) -> None:
        if not self._state_path or not self._state_path.is_file():
            return
        try:
            raw = json.loads(self._state_path.read_text(encoding="utf-8"))
            if raw.get("root_parts") != self._root_parts:
                return
            self._root_fid = raw.get("root_fid")
            self._dir_cache = dict(raw.get("dirs") or {})
        except (OSError, json.JSONDecodeError):
            pass

    def _save_state(self) -> None:
        if not self._state_path:
            return
        self._state_path.parent.mkdir(parents=True, exist_ok=True)
        payload = {
            "root_parts": self._root_parts,
            "root_fid": self._root_fid,
            "dirs": self._dir_cache,
        }
        self._state_path.write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")

    def _headers(self, *, content_type: str = "application/json") -> dict[str, str]:
        return {
            "Cookie": self._cookie,
            "User-Agent": QUARK_BROWSER_UA,
            "Accept": "application/json, text/plain, */*",
            "Origin": "https://pan.quark.cn",
            "Referer": "https://pan.quark.cn/",
            "Content-Type": content_type,
            "x-canary": "client=web,app=adrive,version=v2.3.1",
        }

    @staticmethod
    def _cookie_value(cookie: str, name: str) -> str:
        for part in cookie.split(";"):
            part = part.strip()
            if part.startswith(f"{name}="):
                return part.split("=", 1)[1].strip()
        return ""

    def _resolve_session_token(self) -> str:
        for key in ("ctoken", "__kps", "__kp"):
            value = self._cookie_value(self._cookie, key)
            if value:
                return value
        return ""

    def _anti_bot_params(self, extra: dict[str, str] | None = None) -> dict[str, str]:
        now_ms = str(int(time.time() * 1000))
        params = {
            "pr": "ucpro",
            "fr": "pc",
            "uc_param_str": "",
            "__dt": str(random.randint(100, 9999)),
            "__t": now_ms,
            "_timestamp": now_ms,
        }
        token = self._session_token or self._resolve_session_token()
        if token:
            params["_token"] = token
        if extra:
            params.update(extra)
        return params

    def _merge_set_cookie(self, headers: Any) -> bool:
        raw = headers.get("Set-Cookie") if headers is not None else None
        if not raw:
            return False
        changed = False
        for name in ("__puus", "__pus", "ctoken"):
            match = re.search(rf"{re.escape(name)}=([^;]+)", raw)
            if not match:
                continue
            value = match.group(1)
            if self._cookie_value(self._cookie, name):
                new_cookie = re.sub(
                    rf"{re.escape(name)}=[^;]*",
                    f"{name}={value}",
                    self._cookie,
                )
            else:
                new_cookie = f"{self._cookie.rstrip('; ')}; {name}={value}"
            if new_cookie != self._cookie:
                self._cookie = new_cookie
                changed = True
            if name == "ctoken":
                self._session_token = value
        return changed

    @staticmethod
    def _read_http_error_body(exc: urllib.error.HTTPError) -> str:
        if exc.fp is None:
            return str(exc)
        if isinstance(exc.fp, (bytes, bytearray)):
            return bytes(exc.fp).decode("utf-8", "replace")
        try:
            return exc.read().decode("utf-8", "replace")
        except Exception:
            return str(exc)

    def _apply_response_headers(self, resp_or_exc: Any, raw: str | None = None) -> None:
        headers = getattr(resp_or_exc, "headers", None)
        if headers is not None:
            self._merge_set_cookie(headers)
        if raw:
            self._apply_metadata(self._try_parse_json(raw))

    @staticmethod
    def _try_parse_json(raw: str) -> dict[str, Any]:
        if not raw:
            return {}
        try:
            payload = json.loads(raw)
        except json.JSONDecodeError:
            return {}
        return payload if isinstance(payload, dict) else {}

    def _apply_metadata(self, payload: dict[str, Any]) -> bool:
        metadata = payload.get("metadata")
        if not isinstance(metadata, dict):
            return False
        meta_token = metadata.get("_token")
        if not meta_token:
            return False
        token = str(meta_token)
        if self._session_token == token:
            return False
        self._session_token = token
        if self._cookie_value(self._cookie, "ctoken"):
            self._cookie = re.sub(
                r"\bctoken=[^;]*",
                f"ctoken={token}",
                self._cookie,
            )
        else:
            self._cookie = f"{self._cookie.rstrip('; ')}; ctoken={token}"
        return True

    @staticmethod
    def _is_auth_failure(
        *,
        http_code: int | None = None,
        payload: dict[str, Any] | None = None,
        detail: str = "",
    ) -> bool:
        if http_code == 401:
            return True
        text = detail.lower()
        if "st invalid" in text or "token invalid" in text or "invalid token" in text:
            return True
        if not payload:
            return False
        code = payload.get("code")
        if code in QUARK_AUTH_ERROR_CODES:
            return True
        msg = str(payload.get("message") or payload.get("msg") or "").lower()
        return "st invalid" in msg or "token invalid" in msg or "invalid token" in msg

    def _maybe_persist_session(self) -> None:
        if self._credential_manager is None:
            return
        if self._cookie == self._last_persisted_cookie:
            return
        now = time.monotonic()
        if (
            self._last_persist_at
            and now - self._last_persist_at < self._persist_interval_seconds
        ):
            return
        updater = getattr(self._credential_manager, "update_quark_cookie", None)
        if not callable(updater):
            return
        if updater(self._cookie, session_token=self._session_token):
            self._last_persisted_cookie = self._cookie
            self._last_persist_at = now

    def _mark_auth_failed(self, error: str) -> None:
        if self._credential_manager is None:
            return
        mark = getattr(self._credential_manager, "mark_needs_relogin", None)
        if callable(mark):
            mark("quark", error=error[:200])

    def _build_request(
        self,
        method: str,
        path: str,
        *,
        params: dict[str, str] | None = None,
        body: dict[str, Any] | None = None,
    ) -> urllib.request.Request:
        query = self._default_params()
        if params:
            query.update(params)
        url = f"{self._api_base}/{path.lstrip('/')}?{urllib.parse.urlencode(query)}"
        data = None
        headers = self._headers()
        if body is not None:
            data = json.dumps(body, ensure_ascii=False).encode("utf-8")
        return urllib.request.Request(url, data=data, method=method.upper(), headers=headers)

    def _default_params(self) -> dict[str, str]:
        return self._anti_bot_params()

    def _request_json(
        self,
        method: str,
        path: str,
        *,
        params: dict[str, str] | None = None,
        body: dict[str, Any] | None = None,
        auth_retry: bool = True,
    ) -> dict[str, Any]:
        last_exc: QuarkAPIError | None = None
        attempts = 2 if auth_retry else 1
        for attempt in range(attempts):
            req = self._build_request(method, path, params=params, body=body)
            try:
                with self._open(req, timeout=self._timeout) as resp:
                    raw = resp.read().decode("utf-8", "replace")
                    self._apply_response_headers(resp, raw)
            except urllib.error.HTTPError as exc:
                detail = self._read_http_error_body(exc)
                payload = self._try_parse_json(detail)
                self._apply_response_headers(exc, detail)
                if attempt + 1 < attempts and self._is_auth_failure(
                    http_code=exc.code, payload=payload, detail=detail
                ):
                    continue
                hint = ""
                if exc.code == 401 and "token" in detail.lower():
                    hint = (
                        " — Cookie/ctoken 可能过期；请执行 "
                        "./scripts/quark_cloud_login.sh 或 "
                        "python3 -m niannian.cli.main cloud login quark --browser chrome --probe"
                    )
                last_exc = QuarkAPIError(f"{method} {path} HTTP {exc.code}: {detail[:200]}{hint}")
                break
            try:
                payload = json.loads(raw) if raw else {}
            except json.JSONDecodeError as exc:
                raise QuarkAPIError(f"invalid JSON from {path}: {raw[:200]}") from exc
            if payload.get("code") not in (None, 0) and payload.get("status") not in (True, 200, "ok"):
                msg = payload.get("message") or payload.get("msg") or raw[:200]
                self._apply_metadata(payload)
                if attempt + 1 < attempts and self._is_auth_failure(payload=payload, detail=str(msg)):
                    continue
                last_exc = QuarkAPIError(f"{method} {path} failed: {msg}")
                break
            self._apply_metadata(payload)
            self._maybe_persist_session()
            return payload if isinstance(payload, dict) else {}
        if last_exc is not None:
            self._mark_auth_failed(str(last_exc))
            raise last_exc
        raise QuarkAPIError(f"{method} {path} failed")

    def _list_children(self, pdir_fid: str) -> list[dict[str, Any]]:
        payload = self._request_json(
            "GET",
            "file/sort",
            params=self._anti_bot_params(
                {
                    "pdir_fid": pdir_fid,
                    "_page": "1",
                    "_size": "200",
                    "_fetch_total": "1",
                    "_fetch_sub_dirs": "0",
                    "_sort": "file_type:asc,updated_at:desc",
                    "fetch_all_file": "1",
                    "fetch_risk_file_name": "1",
                }
            ),
        )
        data = payload.get("data") or {}
        items = data.get("list") if isinstance(data, dict) else None
        if isinstance(items, list):
            return items
        if isinstance(payload.get("data"), list):
            return payload["data"]
        return []

    def _find_child(self, pdir_fid: str, name: str, *, want_dir: bool) -> str | None:
        for item in self._list_children(pdir_fid):
            if item.get("file_name") != name:
                continue
            is_dir = bool(item.get("dir"))
            if want_dir == is_dir:
                return str(item.get("fid"))
        return None

    def _create_folder(self, pdir_fid: str, name: str) -> str:
        payload = self._request_json(
            "POST",
            "file",
            body={
                "pdir_fid": pdir_fid,
                "file_name": name,
                "dir_init_lock": False,
                "dir_path": "",
            },
        )
        data = payload.get("data") or {}
        fid = data.get("fid")
        if fid:
            return str(fid)
        existing = self._find_child(pdir_fid, name, want_dir=True)
        if existing:
            return existing
        raise QuarkAPIError(f"create folder failed: {name}")

    def _ensure_root(self) -> str:
        if self._root_fid:
            return self._root_fid
        cached = self._dir_cache.get("")
        if cached:
            self._root_fid = cached
            return cached
        fid = "0"
        key_parts: list[str] = []
        for part in self._root_parts:
            key_parts.append(part)
            key = "/".join(key_parts)
            cached_dir = self._dir_cache.get(key)
            if cached_dir:
                fid = cached_dir
                continue
            found = self._find_child(fid, part, want_dir=True)
            if not found:
                found = self._create_folder(fid, part)
            self._dir_cache[key] = found
            fid = found
        self._root_fid = fid
        self._dir_cache[""] = fid
        self._save_state()
        return fid

    def _ensure_dir_chain(self, parts: list[str]) -> str:
        fid = self._ensure_root()
        key_parts: list[str] = []
        for part in parts:
            key_parts.append(part)
            key = "/".join(key_parts)
            cached = self._dir_cache.get(key)
            if cached:
                fid = cached
                continue
            found = self._find_child(fid, part, want_dir=True)
            if not found:
                found = self._create_folder(fid, part)
            self._dir_cache[key] = found
            fid = found
        self._save_state()
        return fid

    def _parent_parts(self, rel_path: str) -> tuple[list[str], str]:
        clean = rel_path.strip("/").replace("\\", "/")
        parts = clean.split("/")
        if len(parts) == 1:
            return [], parts[0]
        return parts[:-1], parts[-1]

    def _ensure_parent_dir(self, rel_path: str) -> tuple[str, str]:
        dir_parts, file_name = self._parent_parts(rel_path)
        parent_fid = self._ensure_dir_chain(dir_parts)
        return parent_fid, file_name

    def _delete_if_exists(self, pdir_fid: str, file_name: str) -> None:
        fid = self._find_child(pdir_fid, file_name, want_dir=False)
        if not fid:
            return
        self._request_json(
            "POST",
            "file/delete",
            body={"action_type": 2, "filelist": [fid], "exclude_fids": []},
        )

    def _download_url(self, fid: str) -> str:
        payload = self._request_json("POST", "file/download", body={"fids": [fid]})
        data = payload.get("data") or []
        if not data:
            raise QuarkAPIError("download response empty")
        url = data[0].get("download_url")
        if not url:
            raise QuarkAPIError("download_url missing")
        return str(url)

    def _upload_single_part(self, parent_fid: str, file_name: str, data: bytes) -> None:
        now_ms = int(time.time() * 1000)
        mime = "application/octet-stream"
        md5 = hashlib.md5(data).hexdigest()
        sha1 = hashlib.sha1(data).hexdigest()

        pre = self._request_json(
            "POST",
            "file/upload/pre",
            body={
                "ccp_hash_update": True,
                "parallel_upload": True,
                "pdir_fid": parent_fid,
                "dir_name": "",
                "size": len(data),
                "file_name": file_name,
                "format_type": mime,
                "l_updated_at": now_ms,
                "l_created_at": now_ms,
            },
        )
        info = pre.get("data") or {}
        task_id = info.get("task_id")
        auth_info = info.get("auth_info", "")
        upload_id = info.get("upload_id", "")
        obj_key = info.get("obj_key", "")
        bucket = info.get("bucket", "ul-zb")
        upload_url = info.get("upload_url", "")
        callback = info.get("callback") or {}
        if not task_id:
            raise QuarkAPIError("pre upload missing task_id")
        if not upload_url:
            raise QuarkAPIError("pre upload missing upload_url")

        self._request_json(
            "POST",
            "file/update/hash",
            body={"task_id": task_id, "md5": md5, "sha1": sha1},
        )

        auth = self._upload_auth(
            task_id=task_id,
            auth_info=auth_info,
            upload_id=upload_id,
            obj_key=obj_key,
            bucket=bucket,
            upload_url=upload_url,
            mime=mime,
            part_number=1,
        )
        etag = self._oss_put(auth["upload_url"], auth["headers"], data)

        xml = (
            '<?xml version="1.0" encoding="UTF-8"?>\n'
            "<CompleteMultipartUpload>\n"
            f'<Part><PartNumber>1</PartNumber><ETag>"{etag}"</ETag></Part>\n'
            "</CompleteMultipartUpload>"
        )
        post_auth = self._complete_auth(
            task_id=task_id,
            auth_info=auth_info,
            upload_id=upload_id,
            obj_key=obj_key,
            bucket=bucket,
            upload_url=upload_url,
            mime="application/xml",
            xml_data=xml,
            callback=callback,
        )
        self._oss_post(post_auth["upload_url"], post_auth["headers"], xml.encode("utf-8"))

        finish_body: dict[str, Any] = {"task_id": task_id}
        if obj_key:
            finish_body["obj_key"] = obj_key
        self._request_json("POST", "file/upload/finish", body=finish_body)

    def _upload_multipart(
        self,
        parent_fid: str,
        file_name: str,
        data: bytes,
        *,
        hooks: ChunkedUploadHooks | None = None,
        resume: Any | None = None,
    ) -> None:
        now_ms = int(time.time() * 1000)
        mime = "application/octet-stream"
        md5 = hashlib.md5(data).hexdigest()
        sha1 = hashlib.sha1(data).hexdigest()
        chunks = _split_chunks(data, self._chunk_size_bytes)
        ps = dict((resume.provider_state if resume else {}) or {})

        task_id = ps.get("task_id")
        auth_info = ps.get("auth_info", "")
        upload_id = ps.get("upload_id", "")
        obj_key = ps.get("obj_key", "")
        bucket = ps.get("bucket", "ul-zb")
        upload_url = ps.get("upload_url", "")
        callback = ps.get("callback") or {}
        completed_parts = {int(x) for x in (ps.get("completed_part_numbers") or [])}
        part_entries: list[tuple[int, str]] = [
            (int(num), str(etag)) for num, etag in (ps.get("part_entries") or [])
        ]

        if not task_id:
            pre = self._request_json(
                "POST",
                "file/upload/pre",
                body={
                    "ccp_hash_update": True,
                    "parallel_upload": False,
                    "pdir_fid": parent_fid,
                    "dir_name": "",
                    "size": len(data),
                    "file_name": file_name,
                    "format_type": mime,
                    "l_updated_at": now_ms,
                    "l_created_at": now_ms,
                },
            )
            info = pre.get("data") or {}
            task_id = info.get("task_id")
            auth_info = info.get("auth_info", "")
            upload_id = info.get("upload_id", "")
            obj_key = info.get("obj_key", "")
            bucket = info.get("bucket", "ul-zb")
            upload_url = info.get("upload_url", "")
            callback = info.get("callback") or {}
            if not task_id:
                raise QuarkAPIError("pre upload missing task_id")
            if not upload_url:
                raise QuarkAPIError("pre upload missing upload_url")

            self._request_json(
                "POST",
                "file/update/hash",
                body={"task_id": task_id, "md5": md5, "sha1": sha1},
            )

        for part_number, chunk in enumerate(chunks, start=1):
            if part_number in completed_parts:
                continue
            auth = self._upload_auth(
                task_id=task_id,
                auth_info=auth_info,
                upload_id=upload_id,
                obj_key=obj_key,
                bucket=bucket,
                upload_url=upload_url,
                mime=mime,
                part_number=part_number,
            )
            etag = self._oss_put(auth["upload_url"], auth["headers"], chunk)
            part_entries.append((part_number, etag))
            completed_parts.add(part_number)
            if hooks is not None:
                hooks.after_chunk(
                    part_number - 1,
                    provider_state={
                        "task_id": task_id,
                        "auth_info": auth_info,
                        "upload_id": upload_id,
                        "obj_key": obj_key,
                        "bucket": bucket,
                        "upload_url": upload_url,
                        "callback": callback,
                        "completed_part_numbers": sorted(completed_parts),
                        "part_entries": [[n, e] for n, e in part_entries],
                    },
                )

        parts_xml = "".join(
            f'<Part><PartNumber>{num}</PartNumber><ETag>"{etag}"</ETag></Part>\n'
            for num, etag in sorted(part_entries, key=lambda x: x[0])
        )
        xml = (
            '<?xml version="1.0" encoding="UTF-8"?>\n'
            "<CompleteMultipartUpload>\n"
            f"{parts_xml}"
            "</CompleteMultipartUpload>"
        )
        post_auth = self._complete_auth(
            task_id=task_id,
            auth_info=auth_info,
            upload_id=upload_id,
            obj_key=obj_key,
            bucket=bucket,
            upload_url=upload_url,
            mime="application/xml",
            xml_data=xml,
            callback=callback,
        )
        self._oss_post(post_auth["upload_url"], post_auth["headers"], xml.encode("utf-8"))

        finish_body: dict[str, Any] = {"task_id": task_id}
        if obj_key:
            finish_body["obj_key"] = obj_key
        self._request_json("POST", "file/upload/finish", body=finish_body)

    @staticmethod
    def _oss_object_base(bucket: str, upload_url: str, obj_key: str) -> str:
        host = upload_url
        if host.startswith("http://"):
            host = host[7:]
        elif host.startswith("https://"):
            host = host[8:]
        host = host.rstrip("/")
        return f"https://{bucket}.{host}/{obj_key}"

    def _upload_auth(
        self,
        *,
        task_id: str,
        auth_info: str,
        upload_id: str,
        obj_key: str,
        bucket: str,
        upload_url: str,
        mime: str,
        part_number: int,
    ) -> dict[str, Any]:
        oss_date = datetime.now(timezone.utc).strftime("%a, %d %b %Y %H:%M:%S GMT")
        auth_meta = (
            f"PUT\n\n{mime}\n{oss_date}\n"
            f"x-oss-date:{oss_date}\n"
            "x-oss-user-agent:aliyun-sdk-js/1.0.0 Chrome Mobile 139.0.0.0 on Google Nexus 5 (Android 6.0)\n"
            f"/{bucket}/{obj_key}?partNumber={part_number}&uploadId={upload_id}"
        )
        payload = self._request_json(
            "POST",
            "file/upload/auth",
            body={"task_id": task_id, "auth_info": auth_info, "auth_meta": auth_meta},
        )
        auth_key = (payload.get("data") or {}).get("auth_key", "")
        headers = {
            "Content-Type": mime,
            "Referer": "https://pan.quark.cn/",
            "x-oss-date": oss_date,
            "x-oss-user-agent": "aliyun-sdk-js/1.0.0 Chrome Mobile 139.0.0.0 on Google Nexus 5 (Android 6.0)",
        }
        if auth_key:
            headers["Authorization"] = auth_key
        put_url = (
            f"{self._oss_object_base(bucket, upload_url, obj_key)}"
            f"?partNumber={part_number}&uploadId={upload_id}"
        )
        return {"upload_url": put_url, "headers": headers}

    def _complete_auth(
        self,
        *,
        task_id: str,
        auth_info: str,
        upload_id: str,
        obj_key: str,
        bucket: str,
        upload_url: str,
        mime: str,
        xml_data: str,
        callback: dict[str, Any],
    ) -> dict[str, Any]:
        oss_date = datetime.now(timezone.utc).strftime("%a, %d %b %Y %H:%M:%S GMT")
        xml_md5 = base64.b64encode(hashlib.md5(xml_data.encode("utf-8")).digest()).decode("utf-8")
        callback_b64 = base64.b64encode(json.dumps(callback, separators=(",", ":")).encode("utf-8")).decode("utf-8")
        auth_meta = (
            f"POST\n{xml_md5}\n{mime}\n{oss_date}\n"
            f"x-oss-callback:{callback_b64}\n"
            f"x-oss-date:{oss_date}\n"
            "x-oss-user-agent:aliyun-sdk-js/1.0.0 Chrome 139.0.0.0 on OS X 10.15.7 64-bit\n"
            f"/{bucket}/{obj_key}?uploadId={upload_id}"
        )
        payload = self._request_json(
            "POST",
            "file/upload/auth",
            body={"task_id": task_id, "auth_info": auth_info, "auth_meta": auth_meta},
        )
        auth_key = (payload.get("data") or {}).get("auth_key", "")
        headers = {
            "Content-Type": mime,
            "Referer": "https://pan.quark.cn/",
            "x-oss-date": oss_date,
            "x-oss-user-agent": "aliyun-sdk-js/1.0.0 Chrome 139.0.0.0 on OS X 10.15.7 64-bit",
            "x-oss-callback": callback_b64,
            "Content-MD5": xml_md5,
        }
        if auth_key:
            headers["Authorization"] = auth_key
        post_url = f"{self._oss_object_base(bucket, upload_url, obj_key)}?uploadId={upload_id}"
        return {"upload_url": post_url, "headers": headers}

    def _oss_put(self, url: str, headers: dict[str, str], data: bytes) -> str:
        req = urllib.request.Request(url, data=data, method="PUT", headers=headers)
        try:
            with self._open(req, timeout=self._timeout) as resp:
                etag = resp.headers.get("ETag") or resp.headers.get("etag") or ""
        except urllib.error.HTTPError as exc:
            detail = exc.read().decode("utf-8", "replace") if exc.fp else str(exc)
            raise QuarkAPIError(f"OSS PUT failed {exc.code}: {detail[:200]}") from exc
        etag = etag.strip('"')
        if not etag:
            raise QuarkAPIError("OSS PUT missing ETag")
        return etag

    def _oss_post(self, url: str, headers: dict[str, str], data: bytes) -> None:
        req = urllib.request.Request(url, data=data, method="POST", headers=headers)
        try:
            with self._open(req, timeout=self._timeout) as resp:
                status = getattr(resp, "status", 200)
                if status not in (200, 203, 204):
                    body = resp.read(200).decode("utf-8", "replace")
                    raise QuarkAPIError(f"OSS POST failed {status}: {body}")
        except urllib.error.HTTPError as exc:
            if exc.code in (200, 203, 204):
                return
            detail = exc.read().decode("utf-8", "replace") if exc.fp else str(exc)
            raise QuarkAPIError(f"OSS POST failed {exc.code}: {detail[:200]}") from exc
