Byted Byteplus Vod Precision Erasure

                                 Apache License
                           Version 2.0, January 2004
                        http://www.apache.org/licenses/

   TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION

   1. Definitions.

      "License" shall mean the terms and conditions for use, reproduction,
      and distribution as defined by Sections 1 through 9 of this document.

      "Licensor" shall mean the copyright owner or entity authorized by
      the copyright owner that is granting the License.

      "Legal Entity" shall mean the union of the acting entity and all
      other entities that control, are controlled by, or are under common
      control with that entity. For the purposes of this definition,
      "control" means (i) the power, direct or indirect, to cause the
      direction or management of such entity, whether by contract or
      otherwise, or (ii) ownership of fifty percent (50%) or more of the
      outstanding shares, or (iii) beneficial ownership of such entity.

      "You" (or "Your") shall mean an individual or Legal Entity
      exercising permissions granted by this License.

      "Source" form shall mean the preferred form for making modifications,
      including but not limited to software source code, documentation
      source, and configuration files.

      "Object" form shall mean any form resulting from mechanical
      transformation or translation of a Source form, including but
      not limited to compiled object code, generated documentation,
      and conversions to other media types.

      "Work" shall mean the work of authorship, whether in Source or
      Object form, made available under the License, as indicated by a
      copyright notice that is included in or attached to the work
      (an example is provided in the Appendix below).

      "Derivative Works" shall mean any work, whether in Source or Object
      form, that is based on (or derived from) the Work and for which the
      editorial revisions, annotations, elaborations, or other modifications
      represent, as a whole, an original work of authorship. For the purposes
      of this License, Derivative Works shall not include works that remain
      separable from, or merely link (or bind by name) to the interfaces of,
      the Work and Derivative Works thereof.

      "Contribution" shall mean any work of authorship, including
      the original version of the Work and any modifications or additions
      to that Work or Derivative Works thereof, that is intentionally
      submitted to Licensor for inclusion in the Work by the copyright owner
      or by an individual or Legal Entity authorized to submit on behalf of
      the copyright owner. For the purposes of this definition, "submitted"
      means any form of electronic, verbal, or written communication sent
      to the Licensor or its representatives, including but not limited to
      communication on electronic mailing lists, source code control systems,
      and issue tracking systems that are managed by, or on behalf of, the
      Licensor for the purpose of discussing and improving the Work, but
      excluding communication that is conspicuously marked or otherwise
      designated in writing by the copyright owner as "Not a Contribution."

      "Contributor" shall mean Licensor and any individual or Legal Entity
      on behalf of whom a Contribution has been received by Licensor and
      subsequently incorporated within the Work.

   2. Grant of Copyright License. Subject to the terms and conditions of
      this License, each Contributor hereby grants to You a perpetual,
      worldwide, non-exclusive, no-charge, royalty-free, irrevocable
      copyright license to reproduce, prepare Derivative Works of,
      publicly display, publicly perform, sublicense, and distribute the
      Work and such Derivative Works in Source or Object form.

   3. Grant of Patent License. Subject to the terms and conditions of
      this License, each Contributor hereby grants to You a perpetual,
      worldwide, non-exclusive, no-charge, royalty-free, irrevocable
      (except as stated in this section) patent license to make, have made,
      use, offer to sell, sell, import, and otherwise transfer the Work,
      where such license applies only to those patent claims licensable
      by such Contributor that are necessarily infringed by their
      Contribution(s) alone or by combination of their Contribution(s)
      with the Work to which such Contribution(s) was submitted. If You
      institute patent litigation against any entity (including a
      cross-claim or counterclaim in a lawsuit) alleging that the Work
      or a Contribution incorporated within the Work constitutes direct
      or contributory patent infringement, then any patent licenses
      granted to You under this License for that Work shall terminate
      as of the date such litigation is filed.

   4. Redistribution. You may reproduce and distribute copies of the
      Work or Derivative Works thereof in any medium, with or without
      modifications, and in Source or Object form, provided that You
      meet the following conditions:

      (a) You must give any other recipients of the Work or
          Derivative Works a copy of this License; and

      (b) You must cause any modified files to carry prominent notices
          stating that You changed the files; and

      (c) You must retain, in the Source form of any Derivative Works
          that You distribute, all copyright, patent, trademark, and
          attribution notices from the Source form of the Work,
          excluding those notices that do not pertain to any part of
          the Derivative Works; and

      (d) If the Work includes a "NOTICE" text file as part of its
          distribution, then any Derivative Works that You distribute must
          include a readable copy of the attribution notices contained
          within such NOTICE file, excluding those notices that do not
          pertain to any part of the Derivative Works, in at least one
          of the following places: within a NOTICE text file distributed
          as part of the Derivative Works; within the Source form or
          documentation, if provided along with the Derivative Works; or,
          within a display generated by the Derivative Works, if and
          wherever such third-party notices normally appear. The contents
          of the NOTICE file are for informational purposes only and
          do not modify the License. You may add Your own attribution
          notices within Derivative Works that You distribute, alongside
          or as an addendum to the NOTICE text from the Work, provided
          that such additional attribution notices cannot be construed
          as modifying the License.

      You may add Your own copyright statement to Your modifications and
      may provide additional or different license terms and conditions
      for use, reproduction, or distribution of Your modifications, or
      for any such Derivative Works as a whole, provided Your use,
      reproduction, and distribution of the Work otherwise complies with
      the conditions stated in this License.

   5. Submission of Contributions. Unless You explicitly state otherwise,
      any Contribution intentionally submitted for inclusion in the Work
      by You to the Licensor shall be under the terms and conditions of
      this License, without any additional terms or conditions.
      Notwithstanding the above, nothing herein shall supersede or modify
      the terms of any separate license agreement you may have executed
      with Licensor regarding such Contributions.

   6. Trademarks. This License does not grant permission to use the trade
      names, trademarks, service marks, or product names of the Licensor,
      except as required for reasonable and customary use in describing the
      origin of the Work and reproducing the content of the NOTICE file.

   7. Disclaimer of Warranty. Unless required by applicable law or
      agreed to in writing, Licensor provides the Work (and each
      Contributor provides its Contributions) on an "AS IS" BASIS,
      WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
      implied, including, without limitation, any warranties or conditions
      of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
      PARTICULAR PURPOSE. You are solely responsible for determining the
      appropriateness of using or redistributing the Work and assume any
      risks associated with Your exercise of permissions under this License.

   8. Limitation of Liability. In no event and under no legal theory,
      whether in tort (including negligence), contract, or otherwise,
      unless required by applicable law (such as deliberate and grossly
      negligent acts) or agreed to in writing, shall any Contributor be
      liable to You for damages, including any direct, indirect, special,
      incidental, or consequential damages of any character arising as a
      result of this License or out of the use or inability to use the
      Work (including but not limited to damages for loss of goodwill,
      work stoppage, computer failure or malfunction, or any and all
      other commercial damages or losses), even if such Contributor
      has been advised of the possibility of such damages.

   9. Accepting Warranty or Additional Liability. While redistributing
      the Work or Derivative Works thereof, You may choose to offer,
      and charge a fee for, acceptance of support, warranty, indemnity,
      or other liability obligations and/or rights consistent with this
      License. However, in accepting such obligations, You may act only
      on Your own behalf and on Your sole responsibility, not on behalf
      of any other Contributor, and only if You agree to indemnify,
      defend, and hold each Contributor harmless for any liability
      incurred by, or claims asserted against, such Contributor by reason
      of your accepting any such warranty or additional liability.

# BytePlus authentication (preferred)
BYTEPLUS_ACCESSKEY=
BYTEPLUS_SECRETKEY=

# Legacy names (still supported by scripts)
# VOLCENGINE_ACCESS_KEY=
# VOLCENGINE_SECRET_KEY=

# VOD space name (required)
VOD_SPACE_NAME=

#!/usr/bin/env python3
# Copyright (c) 2025 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""
poll_execution.py — resume polling a precision erasure job by RunId

Usage:
  uv run python scripts/poll_execution.py <RunId> [space_name]
"""

import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent))
from vod_client import get_client, get_space_name, out, bail
from precise_erase import poll_erase


def main():
    if len(sys.argv) < 2:
        bail("Usage: uv run python scripts/poll_execution.py <RunId> [space_name]")

    run_id = sys.argv[1].strip()
    if not run_id:
        bail("RunId must not be empty")

    space_name = get_space_name(argv_pos=2)
    client = get_client()
    out(poll_erase(client, run_id, space_name))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
# Copyright (c) 2025 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""
precise_erase.py — BytePlus VOD precision erasure (StartExecution Task Type Erase)

Submits OperationTaskErase jobs and polls GetExecution until completion.

Usage:
  uv run python scripts/precise_erase.py '<json_args>'
  uv run python scripts/precise_erase.py @params.json

See references/precise_erase.md (skill: byted-byteplus-vod-precision-erasure).
"""

from __future__ import annotations

import json
import os
import sys
import time
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent))
from vod_client import get_client, get_space_name, build_media_input, get_play_url_by_filename, out, log, bail

# ── Polling configuration ──────────────────────────────────────────────────
POLL_INTERVAL = float(os.environ.get("VOD_POLL_INTERVAL", "5"))
POLL_MAX = int(os.environ.get("VOD_POLL_MAX", "360"))  # 360 × 5s = 30 minutes

# ── VOD API constants ──────────────────────────────────────────────────────
_ACTION_START = "StartExecution"
_ACTION_GET = "GetExecution"
_VERSION_EXEC = "2025-07-01"

_PENDING = {"", "PendingStart", "Running"}
_TERMINAL_FAIL = {"Failed", "Terminated"}

_CLIP_MODE_API = {
    "skip": "Skip",
    "selected": "Selected",
}


def _truthy(v) -> bool:
    if isinstance(v, bool):
        return v
    if isinstance(v, (int, float)):
        return v != 0
    if isinstance(v, str):
        return v.strip().lower() in ("1", "true", "yes", "on")
    return False


def _parse_clip_filter(args: dict) -> dict | None:
    if "clip_filter" not in args:
        return None
    cf = args.get("clip_filter")
    if cf is None:
        return None
    if not isinstance(cf, dict):
        bail("precise_erase: 'clip_filter' must be an object or null")
    mode_raw = cf.get("mode")
    if mode_raw is None or (isinstance(mode_raw, str) and not mode_raw.strip()):
        bail("precise_erase: when 'clip_filter' is set, 'clip_filter.mode' is required (skip or selected)")
    mode_key = str(mode_raw).strip().lower()
    if mode_key not in _CLIP_MODE_API:
        bail("precise_erase: 'clip_filter.mode' must be 'skip' or 'selected'")
    clips_raw = cf.get("clips")
    if clips_raw is None:
        bail("precise_erase: when using clip_filter, 'clip_filter.clips' is required (non-empty array)")
    if not isinstance(clips_raw, list) or len(clips_raw) < 1:
        bail("precise_erase: 'clip_filter.clips' must be a non-empty array")

    clips_out: list[dict] = []
    for idx, c in enumerate(clips_raw):
        if not isinstance(c, dict):
            bail(f"precise_erase: clip_filter.clips[{idx}] must be an object")
        start = c.get("start", c.get("Start"))
        end = c.get("end", c.get("End"))
        try:
            start_f = float(start)
            end_f = float(end)
        except (TypeError, ValueError):
            bail(f"precise_erase: clip_filter.clips[{idx}] needs numeric 'start' and 'end' (seconds)")
        if end_f <= start_f:
            bail(f"precise_erase: clip_filter.clips[{idx}]: end must be greater than start")
        clips_out.append({"Start": start_f, "End": end_f})

    return {
        "Mode": _CLIP_MODE_API[mode_key],
        "Clips": clips_out,
    }


def _parse_include_erase_detail(args: dict) -> bool:
    """Maps to API WithEraseInfo (default true)."""
    if "with_erase_info" not in args:
        return True
    return _truthy(args.get("with_erase_info"))


def _parse_erase_type_subtitle_vs_text(args: dict) -> tuple[str, dict]:
    """
    Default subtitle-only OCR; optional broader 'text' type.
    Returns (Auto.Type api value, auto_extra dict merged into Auto).
    """
    if _truthy(args.get("text")):
        type_api = "Text"
        extra: dict = {}
    elif _truthy(args.get("all_text")):
        type_api = "Text"
        extra = {}
    else:
        type_api = "Subtitle"
        extra = {"SubtitleFilter": {}}
    return type_api, extra


def build_erase_operation(args: dict) -> dict:
    """Construct OperationTaskErase payload (always Auto; NewVid always true)."""
    include_detail = _parse_include_erase_detail(args)
    clip_filter = _parse_clip_filter(args)
    type_api, extra = _parse_erase_type_subtitle_vs_text(args)

    auto_block: dict = {"Type": type_api, **extra}

    erase: dict = {
        "Mode": "Auto",
        "Auto": auto_block,
        "WithEraseInfo": include_detail,
        "NewVid": True,
    }
    if clip_filter is not None:
        erase["EraseOption"] = {"ClipFilter": clip_filter}

    return erase


def _start_execution(client, payload: dict) -> str:
    resp = client.post(_ACTION_START, _VERSION_EXEC, payload)
    result = resp.get("Result", {}) or {}
    run_id = result.get("RunId", "")
    if not run_id:
        bail(f"StartExecution did not return a RunId, response: {resp}")
    return run_id


def _get_execution(client, run_id: str, *, include_erase_detail: bool) -> dict:
    resp = client.get(_ACTION_GET, _VERSION_EXEC, {"RunId": run_id})
    result = resp.get("Result", {}) or {}
    status = result.get("Status", "")
    space_name = (result.get("Meta", {}) or {}).get("SpaceName", "")

    if status != "Success":
        return {
            "Status": status,
            "Code": result.get("Code", ""),
            "SpaceName": space_name,
        }

    output = ((result.get("Output", {}) or {}).get("Task", {}) or {})
    erase = output.get("Erase", {}) or {}
    file_info = erase.get("File", {}) or {}
    vid = (file_info.get("Vid") or "").strip()
    file_name = (file_info.get("FileName") or "").strip()
    file_id = (file_info.get("FileId") or file_info.get("StoreId") or vid or "").strip()

    direct_url = file_name
    url = ""
    if direct_url and space_name:
        url = get_play_url_by_filename(
            client,
            space_name,
            direct_url,
            expired_minutes=int(os.environ.get("VOD_URL_EXPIRE_MINUTES", "60")),
        )

    source = ""
    if vid:
        source = f"vid://{vid}"
    elif direct_url:
        source = f"directurl://{direct_url}"

    out_obj: dict = {
        "Status": "Success",
        "SpaceName": space_name,
        "VideoUrls": [
            {
                "FileId": file_id,
                "Vid": vid,
                "DirectUrl": direct_url,
                "Source": source,
                "Url": url,
            }
        ],
        "AudioUrls": [],
        "Texts": [],
    }
    if include_erase_detail:
        out_obj["EraseMeta"] = {
            "Duration": erase.get("Duration"),
            "Info": erase.get("Info"),
        }
    else:
        out_obj["EraseMeta"] = {}

    return out_obj


def poll_erase(
    client,
    run_id: str,
    space_name: str,
    *,
    include_erase_detail: bool = True,
) -> dict:
    for i in range(1, POLL_MAX + 1):
        log(f"Polling precise_erase job [{i}/{POLL_MAX}] RunId={run_id} ...")
        try:
            result = _get_execution(client, run_id, include_erase_detail=include_erase_detail)
        except Exception as exc:
            log(f"  query exception: {exc}")
            time.sleep(POLL_INTERVAL)
            continue

        status = result.get("Status", "")

        if status in _PENDING:
            log(f"  status={status!r}, waiting {POLL_INTERVAL}s ...")
            time.sleep(POLL_INTERVAL)
            continue

        if status in _TERMINAL_FAIL:
            ret = {
                "Status": status,
                "Code": result.get("Code", ""),
                "SpaceName": result.get("SpaceName", space_name),
            }
            if status == "Failed":
                ret["resume_hint"] = {
                    "description": "The job failed; check arguments and resubmit, or resume polling with the command below",
                    "command": f"uv run python scripts/poll_execution.py '{run_id}' {space_name}",
                }
            else:
                ret["note"] = "The job was terminated; please resubmit."
            return ret

        if status == "Success":
            return result

        log(f"  unknown status={status!r}, continuing to wait ...")
        time.sleep(POLL_INTERVAL)

    return {
        "error": f"Polling timed out ({POLL_MAX} attempts × {POLL_INTERVAL}s); the job is still processing",
        "resume_hint": {
            "description": "The job has not finished yet; resume polling with the command below",
            "command": f"uv run python scripts/poll_execution.py '{run_id}' {space_name}",
        },
    }


def main():
    if len(sys.argv) < 2:
        bail("Usage: uv run python scripts/precise_erase.py '<json_args>'")

    raw = sys.argv[1]
    if raw.startswith("@"):
        fpath = raw[1:]
        if not Path(fpath).is_file():
            bail(f"Parameter file does not exist: {fpath}")
        with open(fpath, "r", encoding="utf-8") as f:
            raw = f.read()

    try:
        args = json.loads(raw)
    except json.JSONDecodeError as e:
        bail(f"Failed to parse JSON arguments: {e}")

    asset_type = args.get("type", "Vid")
    video = args.get("video", "")
    if not video:
        bail("precise_erase: the 'video' field must not be empty")

    space_name = get_space_name(argv_pos=2)
    client = get_client()

    media_input = build_media_input(asset_type, video, space_name)
    erase_body = build_erase_operation(args)

    payload = {
        "Input": media_input,
        "Operation": {
            "Type": "Task",
            "Task": {
                "Type": "Erase",
                "Erase": erase_body,
            },
        },
    }

    flags: list[str] = []
    auto = erase_body.get("Auto") or {}
    flags.append("text" if auto.get("Type") == "Text" else "subtitle")
    if erase_body.get("EraseOption"):
        flags.append("clip_filter")
    if not erase_body.get("WithEraseInfo", True):
        flags.append("no_erase_detail")
    log(
        "Submitting precise_erase job, "
        f"video={video} type={asset_type} opts={'+'.join(flags)}"
    )
    try:
        run_id = _start_execution(client, payload)
    except SystemExit:
        raise
    except Exception as exc:
        bail(f"Failed to submit precise_erase job: {exc}")

    log(f"Job submitted, RunId={run_id}, starting polling ...")
    out(
        poll_erase(
            client,
            run_id,
            space_name,
            include_erase_detail=bool(erase_body.get("WithEraseInfo", True)),
        )
    )


if __name__ == "__main__":
    main()

[project]
name = "byted-byteplus-vod-precision-erasure"
version = "1.0.0"
description = "BytePlus VOD upload and precision erasure (subtitles / on-screen text)"
requires-python = ">=3.10"
dependencies = [
    "requests>=2.31.0",
    "python-dotenv>=1.0.0",
]

# Copyright (c) 2025 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""
TOS media upload — aligned with the official VodService.upload_tob behaviour.

Handles the binary upload step between ApplyUploadInfo and CommitUploadInfo:
  1. VPC pre-signed path  (vpc direct / vpc part)
  2. Candidate upload addresses  (main → backup → fallback)
  3. Default UploadAddress fallback

Small files (< chunk_size) use direct PUT; larger files use chunked upload
(init → upload parts → merge).
"""

from __future__ import annotations

import json
import os
import time
from typing import Any, Callable, Dict, List, Optional, Tuple
from zlib import crc32

import requests

MIN_CHUNK_SIZE = 1024 * 1024 * 20  # 20 MiB


# ---------------------------------------------------------------------------
# helpers
# ---------------------------------------------------------------------------

class _FileSectionReader:
    """Readable wrapper over a slice of a file object (for chunked PUT)."""

    def __init__(self, fobj, size: int, init_offset: Optional[int] = None):
        self.fobj = fobj
        self.size = size
        self.offset = 0
        if init_offset is not None:
            self.fobj.seek(init_offset, os.SEEK_SET)

    def read(self, amt=None):
        if self.offset >= self.size:
            return b""
        if amt is None or amt < 0 or amt + self.offset >= self.size:
            data = self.fobj.read(self.size - self.offset)
            self.offset = self.size
            return data
        self.offset += amt
        return self.fobj.read(amt)

    @property
    def len(self) -> int:
        return self.size


def _norm_headers(raw: Any) -> Dict[str, str]:
    if isinstance(raw, dict):
        return {str(k): str(v) for k, v in raw.items()}
    return {}


def _retry(fn: Callable[[], Any], tries: int = 3, delay: float = 1.0, backoff: float = 2.0) -> Any:
    last_err: Optional[BaseException] = None
    d = delay
    for attempt in range(tries):
        try:
            return fn()
        except Exception as exc:
            last_err = exc
            if attempt == tries - 1:
                break
            time.sleep(d)
            d *= backoff
    assert last_err is not None
    raise last_err


# ---------------------------------------------------------------------------
# TOS uploader
# ---------------------------------------------------------------------------

class TosUploader:
    """PUT files to the TOS gateway using credentials returned by ApplyUploadInfo."""

    def __init__(self, session: Optional[requests.Session] = None):
        self._s = session or requests.Session()

    # -- low-level helpers --------------------------------------------------

    def _put_file(self, url: str, path: str, headers: Dict[str, str]) -> Tuple[bool, bytes]:
        with open(path, "rb") as f:
            r = self._s.put(url, headers=headers, data=f)
        headers["X-Tt-Logid"] = r.headers.get("X-Tt-Logid", "")
        return r.status_code == 200, r.content

    def _put_data(self, url: str, data: Optional[bytes], headers: Dict[str, str]) -> Tuple[bool, bytes]:
        r = self._s.put(url, headers=headers, data=data)
        headers["X-Tt-Logid"] = r.headers.get("X-Tt-Logid", "")
        return r.status_code == 200, r.content

    def _storage_headers(self, headers: Dict[str, str], sc: int) -> None:
        if sc == 2:
            headers["X-Upload-Storage-Class"] = "archive"
        elif sc == 3:
            headers["X-Upload-Storage-Class"] = "ia"

    # -- direct upload (small files) ----------------------------------------

    def direct_upload(self, host: str, oid: str, auth: str, path: str, sc: int) -> None:
        def _once():
            with open(path, "rb") as f:
                data = f.read()
            crc = "%08x" % (crc32(data) & 0xFFFFFFFF)
            url = f"https://{host}/{oid}"
            hdrs: Dict[str, str] = {"Content-CRC32": crc, "Authorization": auth}
            self._storage_headers(hdrs, sc)
            ok, body = self._put_file(url, path, hdrs)
            text = body.decode()
            if not ok:
                raise RuntimeError(f"direct upload error: {text}, logid: {hdrs.get('X-Tt-Logid', '')}")
            j = json.loads(text)
            if j.get("success") not in (0,):
                raise RuntimeError(f"direct upload error: {text}, logid: {hdrs.get('X-Tt-Logid', '')}")
        _retry(_once)

    # -- chunked upload (large files) ---------------------------------------

    def _init_part(self, host: str, oid: str, auth: str, large: bool, sc: int) -> str:
        def _once():
            url = f"https://{host}/{oid}?uploads"
            hdrs: Dict[str, str] = {"Authorization": auth}
            if large:
                hdrs["X-Storage-Mode"] = "gateway"
            self._storage_headers(hdrs, sc)
            ok, body = self._put_data(url, None, hdrs)
            text = body.decode()
            if not ok:
                raise RuntimeError(f"init upload error: {text}")
            return json.loads(text)["payload"]["uploadID"]
        return _retry(_once)

    def _upload_part(self, host: str, oid: str, auth: str, uid: str,
                     pn: int, data: bytes, large: bool, sc: int) -> Tuple[str, Any]:
        def _once():
            url = f"https://{host}/{oid}?partNumber={pn}&uploadID={uid}"
            crc = "%08x" % (crc32(data) & 0xFFFFFFFF)
            hdrs: Dict[str, str] = {"Content-CRC32": crc, "Authorization": auth}
            if large:
                hdrs["X-Storage-Mode"] = "gateway"
            self._storage_headers(hdrs, sc)
            ok, body = self._put_data(url, data, hdrs)
            text = body.decode()
            if not ok:
                raise RuntimeError(f"upload part error: {text}")
            j = json.loads(text)
            if j.get("success") not in (0,):
                raise RuntimeError(f"upload part error: {text}")
            return crc, j["payload"]
        return _retry(_once)

    def _merge_parts(self, host: str, oid: str, auth: str, uid: str,
                     crcs: List[str], large: bool, sc: int, meta: Optional[Dict]) -> None:
        def _once():
            occ = ""
            if meta and meta.get("ObjectContentType"):
                occ = meta["ObjectContentType"]
            url = f"https://{host}/{oid}?uploadID={uid}&ObjectContentType={occ}"
            merge = ",".join(f"{i}:{crcs[i]}" for i in range(len(crcs)))
            hdrs: Dict[str, str] = {"Authorization": auth}
            if large:
                hdrs["X-Storage-Mode"] = "gateway"
            self._storage_headers(hdrs, sc)
            ok, body = self._put_data(url, merge.encode(), hdrs)
            text = body.decode()
            if not ok:
                raise RuntimeError(f"merge upload error: {text}")
            j = json.loads(text)
            if j.get("success") not in (0,):
                raise RuntimeError(f"merge upload error: {text}")
        _retry(_once)

    def chunk_upload(self, path: str, host: str, oid: str, auth: str,
                     size: int, large: bool, sc: int, chunk: int) -> None:
        uid = self._init_part(host, oid, auth, large, sc)
        n = size // chunk
        last = n - 1
        crcs: List[str] = []
        meta: Dict[str, Any] = {}
        with open(path, "rb") as f:
            for i in range(last):
                pn = i + 1 if large else i
                c, payload = self._upload_part(host, oid, auth, uid, pn, f.read(chunk), large, sc)
                if pn == 1:
                    meta = payload.get("meta") or {}
                crcs.append(c)
            if large:
                last = last + 1
            c, payload = self._upload_part(host, oid, auth, uid, last, f.read(), large, sc)
            if last == 1:
                meta = payload.get("meta") or {}
            crcs.append(c)
        self._merge_parts(host, oid, auth, uid, crcs, large, sc, meta)

    # -- VPC pre-signed paths -----------------------------------------------

    def vpc_upload(self, addr: Dict[str, Any], path: str, size: int) -> None:
        if addr.get("QuickCompleteMode") == "enable":
            return
        mode = addr.get("UploadMode") or ""
        if mode == "direct":
            put_url = addr.get("PutUrl") or ""
            put_hdrs = _norm_headers(addr.get("PutUrlHeaders"))
            with open(path, "rb") as f:
                r = self._s.put(put_url, headers=put_hdrs, data=f)
            if r.status_code != 200:
                raise RuntimeError(f"vpc put error, logId: {r.headers.get('x-tos-request-id', '')}")
        elif mode == "part":
            self._vpc_part_upload(addr.get("PartUploadInfo") or {}, path, size)

    def _vpc_part_upload(self, info: Dict[str, Any], path: str, size: int) -> None:
        chunk = int(info.get("PartSize") or 0)
        urls = info.get("PartPutUrls") or []
        total = size // chunk
        if size % chunk == 0:
            total -= 1
        if len(urls) != total + 1:
            raise RuntimeError("mismatch part upload")
        offset = 0
        etags: List[str] = []
        with open(path, "rb") as f:
            for i in range(total):
                sr = _FileSectionReader(f, chunk, init_offset=offset)
                r = self._s.put(urls[i], data=sr)
                if r.status_code != 200:
                    raise RuntimeError(f"vpc part put error, logId: {r.headers.get('x-tos-request-id', '')}")
                etags.append(r.headers.get("ETag", ""))
                offset += chunk
            sr = _FileSectionReader(f, size - offset, init_offset=offset)
            r = self._s.put(urls[total], data=sr)
            if r.status_code != 200:
                raise RuntimeError(f"vpc part put error, logId: {r.headers.get('x-tos-request-id', '')}")
            etags.append(r.headers.get("ETag", ""))
        parts = ",".join(
            '{' + f'"PartNumber": {i + 1}, "ETag": {etags[i]}' + '}'
            for i in range(len(etags))
        )
        body = f'{{"Parts":[{parts}]}}'.encode()
        comp_url = info.get("CompletePartUrl") or ""
        comp_hdrs = _norm_headers(info.get("CompleteUrlHeaders"))
        r = self._s.post(comp_url, data=body, headers=comp_hdrs)
        if r.status_code != 200:
            raise RuntimeError(f"vpc post error, logId: {r.headers.get('x-tos-request-id', '')}")


# ---------------------------------------------------------------------------
# High-level: pick the best upload path from ApplyUploadInfo data
# ---------------------------------------------------------------------------

def upload_to_tos(
    data: Dict[str, Any],
    file_path: str,
    storage_class: int = 1,
    chunk_size: int = 0,
    log_fn: Callable[[str], None] = print,
) -> str:
    """
    Given the Result.Data from ApplyUploadInfo, upload *file_path* to TOS and
    return the SessionKey needed for CommitUploadInfo.
    """
    if not os.path.isfile(file_path):
        raise RuntimeError(f"file not found: {file_path}")
    if chunk_size < MIN_CHUNK_SIZE:
        chunk_size = MIN_CHUNK_SIZE
    fsize = os.path.getsize(file_path)
    uploader = TosUploader()

    # 1) VPC pre-signed
    vpc = data.get("VpcTosUploadAddress")
    if vpc and (vpc.get("UploadMode") or ""):
        sk = (data.get("UploadAddress") or {}).get("SessionKey") or ""
        uploader.vpc_upload(vpc, file_path, fsize)
        return sk

    # 2) Candidate addresses (main → backup → fallback)
    cand = data.get("CandidateUploadAddresses")
    addrs: List[Dict[str, Any]] = []
    if cand:
        addrs.extend(cand.get("MainUploadAddresses") or [])
        addrs.extend(cand.get("BackupUploadAddresses") or [])
        addrs.extend(cand.get("FallbackUploadAddresses") or [])
    if addrs:
        for addr in addrs:
            hosts = addr.get("UploadHosts") or []
            stores = addr.get("StoreInfos") or []
            if not hosts or not stores or not stores[0]:
                continue
            host = hosts[0]
            sk = addr.get("SessionKey") or ""
            auth = stores[0].get("Auth") or ""
            oid = stores[0].get("StoreUri") or ""
            try:
                if fsize < chunk_size:
                    uploader.direct_upload(host, oid, auth, file_path, storage_class)
                else:
                    uploader.chunk_upload(file_path, host, oid, auth, fsize, True, storage_class, chunk_size)
            except Exception as exc:
                log_fn(f"upload failed on {host}, switching host… ({exc})")
                continue
            return sk
        raise RuntimeError("upload failed on all candidate hosts")

    # 3) Default UploadAddress
    ua = data.get("UploadAddress") or {}
    stores = ua.get("StoreInfos") or []
    hosts = ua.get("UploadHosts") or []
    if not stores or not hosts:
        raise RuntimeError("ApplyUploadInfo: UploadAddress missing StoreInfos or UploadHosts")
    oid = stores[0].get("StoreUri") or ""
    sk = ua.get("SessionKey") or ""
    auth = stores[0].get("Auth") or ""
    host = hosts[0]
    if fsize < chunk_size:
        uploader.direct_upload(host, oid, auth, file_path, storage_class)
    else:
        uploader.chunk_upload(file_path, host, oid, auth, fsize, True, storage_class, chunk_size)
    return sk

#!/usr/bin/env python3
# Copyright (c) 2025 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""
upload.py — upload a media asset to BytePlus VOD and return the Vid

Supported inputs:
  1) Local file path  → ApplyUploadInfo + TOS direct/chunked PUT + CommitUploadInfo
  2) http/https link  → UploadMediaByUrl (async) + poll QueryUploadTaskInfo

Usage:
  uv run python scripts/upload.py "<local_path_or_url>" [space_name]

Output (JSON on stdout):
  {"Vid":"vxxxx","Source":"vid://vxxxx","PlayURL":"...","PosterUri":"","FileName":"...","SpaceName":"...","SourceUrl":"..."}

  PlayURL is built via get_play_url_by_filename (storage host / play domain + path).
"""

import os
import sys
import time
import uuid
from pathlib import Path
from urllib.parse import urlparse

sys.path.insert(0, str(Path(__file__).resolve().parent))
from vod_client import (
    get_client,
    get_space_name,
    get_play_url_by_filename,
    apply_upload_info,
    commit_upload_info,
    out,
    log,
    bail,
)
from tos_upload import upload_to_tos

# ── Polling configuration ──────────────────────────────────────────────────
POLL_INTERVAL = float(os.environ.get("VOD_POLL_INTERVAL", "5"))
POLL_MAX = int(os.environ.get("VOD_POLL_MAX", "360"))  # 360 × 5s = 30 minutes

# ── VOD API constants ──────────────────────────────────────────────────────
_ACTION_UPLOAD_BY_URL = "UploadMediaByUrl"
_ACTION_QUERY_TASK = "QueryUploadTaskInfo"
_VERSION = "2023-01-01"


# ══════════════════════════════════════════════════════════════════════════
# Path safety: restrict which local paths may be uploaded
# ══════════════════════════════════════════════════════════════════════════

_ALLOWED_PREFIXES: list[str] | None = None


def _get_allowed_prefixes() -> list[str]:
    global _ALLOWED_PREFIXES
    if _ALLOWED_PREFIXES is not None:
        return _ALLOWED_PREFIXES

    prefixes: list[str] = []

    # 1) WORKSPACE (or cwd) — the project the agent is operating in
    ws = os.environ.get("WORKSPACE", os.getcwd())
    prefixes.append(os.path.realpath(ws))

    # 2) sibling userdata/ directory (Cursor sandbox convention)
    ud = os.path.join(os.path.dirname(ws), "userdata") if ws else ""
    if ud:
        prefixes.append(os.path.realpath(ud))

    # 3) /tmp — commonly used for scratch files
    prefixes.append("/tmp")

    # 4) VOD_UPLOAD_ALLOWED_DIRS — comma-separated extra directories
    extra = os.environ.get("VOD_UPLOAD_ALLOWED_DIRS", "")
    for d in extra.split(","):
        d = d.strip()
        if d:
            prefixes.append(os.path.realpath(d))

    _ALLOWED_PREFIXES = prefixes
    return _ALLOWED_PREFIXES


def _validate_local_path(file_path: str) -> None:
    """Ensure the resolved path falls under an allowed prefix (symlink-safe)."""
    real = os.path.realpath(file_path)
    for prefix in _get_allowed_prefixes():
        if real == prefix or real.startswith(prefix + os.sep):
            return
    bail(
        f"Path not allowed. Only files under workspace/, userdata/, /tmp, "
        f"or VOD_UPLOAD_ALLOWED_DIRS may be uploaded. Rejected: {file_path}"
    )


def _guess_ext(path_str: str) -> str:
    _, ext = os.path.splitext(path_str or "")
    ext = ext.strip()
    if not ext:
        bail("The file must carry a file extension (e.g. .mp4 / .mov / .mp3)")
    if not ext.startswith("."):
        ext = "." + ext
    return ext


def _is_url(s: str) -> bool:
    return s.startswith("http://") or s.startswith("https://")


# ══════════════════════════════════════════════════════════════════════════
# URL upload (async pull)
# ══════════════════════════════════════════════════════════════════════════

def _submit_url_upload(client, space_name: str, source_url: str, file_ext: str) -> list:
    file_name = f"{uuid.uuid4().hex}{file_ext}"
    body = {
        "SpaceName": space_name,
        "URLSets": [
            {
                "SourceUrl": source_url,
                "FileExtension": file_ext,
                "FileName": file_name,
            }
        ],
    }
    log(f"Submitting URL upload: {source_url}")
    resp = client.post(_ACTION_UPLOAD_BY_URL, _VERSION, body)
    result = resp.get("Result", {}) or {}
    data = result.get("Data", []) or []
    job_ids = [item["JobId"] for item in data if isinstance(item, dict) and item.get("JobId")]
    if not job_ids:
        bail(f"UploadMediaByUrl did not return a JobId, response: {resp}")
    return job_ids


def _poll_upload_task(client, job_ids_str: str) -> dict:
    last_state = ""
    for i in range(1, POLL_MAX + 1):
        log(f"Polling upload task [{i}/{POLL_MAX}] JobIds={job_ids_str}")
        try:
            resp = client.get(_ACTION_QUERY_TASK, _VERSION, {"JobIds": job_ids_str})
        except Exception as exc:
            log(f"  query exception: {exc}")
            time.sleep(POLL_INTERVAL)
            continue

        result = resp.get("Result", {}) or {}
        data = result.get("Data", {}) or {}
        media_list = data.get("MediaInfoList", []) or []

        if not media_list:
            time.sleep(POLL_INTERVAL)
            continue

        item = media_list[0]
        state = item.get("State", "")
        last_state = state or last_state
        vid = item.get("Vid", "")

        if vid:
            source_info = item.get("SourceInfo", {}) or {}
            return {
                "Vid": vid,
                "FileName": source_info.get("FileName", ""),
                "State": state,
                "SpaceName": item.get("SpaceName", ""),
                "JobId": item.get("JobId", ""),
            }

        if state.lower() in {"fail", "failed", "error"}:
            bail(f"URL pull upload failed: State={state!r}, JobIds={job_ids_str}")

        time.sleep(POLL_INTERVAL)

    return {
        "error": f"Polling timed out ({POLL_MAX} attempts × {POLL_INTERVAL}s); the URL pull upload is still processing",
        "resume_hint": {
            "description": "The URL upload has not finished yet; retry with the command below",
            "command": 'uv run python scripts/upload.py "<original URL>" [space_name]',
        },
        "JobIds": job_ids_str,
        "State": last_state,
    }


def _do_url_upload(client, space_name: str, source_url: str) -> None:
    file_ext = _guess_ext(urlparse(source_url).path)
    try:
        job_ids = _submit_url_upload(client, space_name, source_url, file_ext)
    except SystemExit:
        raise
    except Exception as exc:
        bail(f"Failed to submit URL upload: {exc}")

    job_ids_str = ",".join(job_ids)
    log(f"Upload job submitted, JobIds={job_ids_str}")

    info = _poll_upload_task(client, job_ids_str)
    if "error" in info:
        out(info)
        return

    vid = info.get("Vid", "")
    file_name = (info.get("FileName") or "").strip()
    play_url = _build_play_url(client, space_name, file_name, vid)
    out({
        "Vid": vid,
        "Source": f"vid://{vid}",
        "PlayURL": play_url,
        "PosterUri": "",
        "FileName": file_name,
        "SpaceName": space_name,
        "SourceUrl": source_url,
        "JobId": job_ids_str,
    })


# ══════════════════════════════════════════════════════════════════════════
# Local file upload (ApplyUploadInfo → TOS → CommitUploadInfo)
# ══════════════════════════════════════════════════════════════════════════

def _do_local_upload(client, space_name: str, file_path: str) -> None:
    _validate_local_path(file_path)
    if not os.path.isfile(file_path):
        bail(f"Local file not found: {file_path}")

    file_ext = _guess_ext(file_path)
    file_size = os.path.getsize(file_path)
    file_name = f"{uuid.uuid4().hex}{file_ext}"
    log(f"Local upload: {file_path} ({file_size} bytes) → FileName={file_name}")

    # Step 1: ApplyUploadInfo
    try:
        apply_data = apply_upload_info(
            client, space_name,
            file_size=file_size, file_name=file_name, file_ext=file_ext,
        )
    except SystemExit:
        raise
    except Exception as exc:
        bail(f"ApplyUploadInfo failed: {exc}")

    # Step 2: upload binary to TOS
    try:
        session_key = upload_to_tos(apply_data, file_path, log_fn=log)
    except SystemExit:
        raise
    except Exception as exc:
        bail(f"TOS upload failed: {exc}")

    # Step 3: CommitUploadInfo
    try:
        commit_data = commit_upload_info(client, space_name, session_key)
    except SystemExit:
        raise
    except Exception as exc:
        bail(f"CommitUploadInfo failed: {exc}")

    vid = commit_data.get("Vid", "")
    if not vid:
        bail("CommitUploadInfo succeeded but returned no Vid")
    source_info = commit_data.get("SourceInfo") or {}
    returned_file_name = source_info.get("FileName", "")

    play_url = _build_play_url(client, space_name, returned_file_name, vid)
    out({
        "Vid": vid,
        "Source": f"vid://{vid}",
        "PlayURL": play_url,
        "PosterUri": commit_data.get("PosterUri", ""),
        "FileName": returned_file_name,
        "SpaceName": space_name,
        "SourceUrl": file_path,
    })


# ══════════════════════════════════════════════════════════════════════════
# Shared: build a play URL from FileName (fallback to Vid as path key)
# ══════════════════════════════════════════════════════════════════════════

def _build_play_url(client, space_name: str, file_name: str, vid: str) -> str:
    vid_key = vid[len("vid://"):] if vid.startswith("vid://") else vid
    path_for_url = (file_name or vid_key).strip()
    if not path_for_url:
        return ""
    expired = int(os.environ.get("VOD_URL_EXPIRE_MINUTES", "60"))
    if not file_name and vid_key:
        log("No FileName; trying get_play_url_by_filename with Vid as path key")
    return get_play_url_by_filename(client, space_name, path_for_url, expired_minutes=expired)


# ══════════════════════════════════════════════════════════════════════════
# main
# ══════════════════════════════════════════════════════════════════════════

def main():
    if len(sys.argv) < 2:
        bail('Usage: uv run python scripts/upload.py "<local_path_or_url>" [space_name]')

    source = sys.argv[1].strip()
    space_name = get_space_name(argv_pos=2)
    client = get_client()

    if _is_url(source):
        _do_url_upload(client, space_name, source)
    else:
        _do_local_upload(client, space_name, source)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
# Copyright (c) 2025 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""
vod_client.py — shared VOD transport layer and utility helpers

Authentication:
  - Direct OpenAPI HMAC-SHA256
    - Preferred env: BYTEPLUS_ACCESSKEY + BYTEPLUS_SECRETKEY
    - Backward-compatible env: VOLCENGINE_ACCESS_KEY + VOLCENGINE_SECRET_KEY
"""

from __future__ import annotations

import hashlib
import hmac
import json
import os
import secrets
import sys
import time
from datetime import datetime, timedelta, timezone
from functools import reduce
from pathlib import Path
from typing import NoReturn
from urllib.parse import quote, urlencode

import requests
from dotenv import load_dotenv

# ── .env loading ───────────────────────────────────────────────────────────
# Look for .env in the current working directory first, then in the script's
# own directory (so the skill still works when invoked from elsewhere).
_SCRIPT_DIR = Path(__file__).resolve().parent
for _base in (Path.cwd(), _SCRIPT_DIR):
    _env = _base / ".env"
    if _env.is_file():
        load_dotenv(_env, override=False)
        break

# ── VOD host ───────────────────────────────────────────────────────────────
# Defaults to the BytePlus (overseas) endpoint. Set VOD_HOST to override,
# e.g. "vod.volcengineapi.com" for the Volcengine (mainland China) endpoint.
_VOD_HOST = (os.environ.get("VOD_HOST") or "vod.byteplusapi.com").strip()

_HTTP_TIMEOUT = float(os.environ.get("VOD_HTTP_TIMEOUT", "20"))


# ══════════════════════════════════════════════════════════════════════════
# Output helpers
# ══════════════════════════════════════════════════════════════════════════

def out(data: dict):
    """Print the result as JSON to stdout."""
    print(json.dumps(data, ensure_ascii=False), flush=True)


def log(msg: str):
    """Write a debug message to stderr."""
    print(f"[byted-byteplus-vod-precision-erasure] {msg}", file=sys.stderr, flush=True)


def bail(msg: str) -> NoReturn:
    """Print an error JSON and exit."""
    out({"error": msg})
    sys.exit(1)


# ══════════════════════════════════════════════════════════════════════════
# Volcengine OpenAPI client (HMAC-SHA256 signing)
# ══════════════════════════════════════════════════════════════════════════

class VolcClient:
    _REGION = (os.environ.get("VOD_REGION") or "ap-southeast-1").strip()
    _SERVICE = "vod"

    def __init__(self, ak: str, sk: str):
        self._ak = ak
        self._sk = sk

    def post(self, action: str, version: str, body: dict) -> dict:
        body_str = json.dumps(body, ensure_ascii=False)
        url, headers = self._sign("POST", action, version, {}, body_str)
        r = requests.post(url, headers=headers, data=body_str.encode(), timeout=_HTTP_TIMEOUT)
        self._check(r)
        return r.json()

    def get(self, action: str, version: str, params: dict) -> dict:
        url, headers = self._sign("GET", action, version, params or {}, "")
        r = requests.get(url, headers=headers, timeout=_HTTP_TIMEOUT)
        self._check(r)
        return r.json()

    def _sign(self, method: str, action: str, version: str,
              query_extra: dict, body_str: str) -> tuple[str, dict]:
        qp = {"Action": action, "Version": version}
        qp.update(query_extra)
        canonical_query = urlencode(sorted(qp.items()), quote_via=quote, safe="-_.~")
        url = f"https://{_VOD_HOST}/?{canonical_query}"

        ts = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
        body_hash = hashlib.sha256(body_str.encode()).hexdigest()

        h = {
            "content-type": "application/json; charset=utf-8",
            "host": _VOD_HOST,
            "x-content-sha256": body_hash,
            "x-date": ts,
        }
        signed_keys = sorted(h.keys())
        canonical_headers = "".join(f"{k}:{h[k]}\n" for k in signed_keys)
        signed_headers_str = ";".join(signed_keys)
        canonical_request = (
            f"{method}\n/\n{canonical_query}\n"
            f"{canonical_headers}\n{signed_headers_str}\n{body_hash}"
        )

        credential_scope = f"{ts[:8]}/{self._REGION}/{self._SERVICE}/request"
        string_to_sign = (
            f"HMAC-SHA256\n{ts}\n{credential_scope}\n"
            f"{hashlib.sha256(canonical_request.encode()).hexdigest()}"
        )
        signing_key = reduce(
            lambda k, v: hmac.new(k, v.encode(), hashlib.sha256).digest(),
            [ts[:8], self._REGION, self._SERVICE, "request"],
            self._sk.encode(),
        )
        signature = hmac.new(signing_key, string_to_sign.encode(),
                             hashlib.sha256).hexdigest()

        headers = {k.title().replace("X-C", "X-c"): v for k, v in h.items()}
        headers["Authorization"] = (
            f"HMAC-SHA256 Credential={self._ak}/{credential_scope}, "
            f"SignedHeaders={signed_headers_str}, Signature={signature}"
        )
        return url, headers

    @staticmethod
    def _check(r):
        if r.status_code != 200:
            raise RuntimeError(f"HTTP {r.status_code}: {r.text}")


# ══════════════════════════════════════════════════════════════════════════
# Factory: pick a client based on environment variables
# ══════════════════════════════════════════════════════════════════════════

def get_client() -> VolcClient:
    """Build a direct OpenAPI client from environment variables."""
    ak = (os.environ.get("BYTEPLUS_ACCESSKEY") or "").strip() or (os.environ.get("VOLCENGINE_ACCESS_KEY") or "").strip()
    sk = (os.environ.get("BYTEPLUS_SECRETKEY") or "").strip() or (os.environ.get("VOLCENGINE_SECRET_KEY") or "").strip()
    if not (ak and sk):
        bail(
            "Missing credentials. Set BYTEPLUS_ACCESSKEY and BYTEPLUS_SECRETKEY "
            "(preferred) or VOLCENGINE_ACCESS_KEY and VOLCENGINE_SECRET_KEY "
            "(legacy) in the environment or in .env."
        )
    return VolcClient(ak, sk)


# ══════════════════════════════════════════════════════════════════════════
# Space name resolution
# ══════════════════════════════════════════════════════════════════════════

def get_space_name(argv_pos: int = 2) -> str:
    """Resolve space_name with priority: CLI argument > VOD_SPACE_NAME env."""
    if len(sys.argv) > argv_pos:
        v = sys.argv[argv_pos].strip()
        if v:
            return v
    sp = (os.environ.get("VOD_SPACE_NAME") or "").strip()
    if sp:
        return sp
    bail("VOD space name not specified: pass it as a CLI argument or set VOD_SPACE_NAME.")


# ══════════════════════════════════════════════════════════════════════════
# Media input builder (used by StartExecution)
# ══════════════════════════════════════════════════════════════════════════

def build_media_input(asset_type: str, asset_value: str, space_name: str) -> dict:
    """
    Build the Input field for StartExecution.

    asset_type: "Vid" or "DirectUrl"
    asset_value: the vid value (without the vid:// prefix) or the VOD FileName
    """
    if asset_type not in ("Vid", "DirectUrl"):
        bail(f"type must be Vid or DirectUrl, got: {asset_type!r}")
    if not asset_value:
        bail("media asset value must not be empty")
    if not space_name:
        bail("space_name must not be empty")

    # Strip protocol prefix
    value = asset_value
    if value.startswith("vid://"):
        value = value[len("vid://"):]
    elif value.startswith("directurl://"):
        value = value[len("directurl://"):]

    media_input: dict = {"Type": asset_type}
    if asset_type == "Vid":
        media_input["Vid"] = value
    else:
        media_input["DirectUrl"] = {"FileName": value, "SpaceName": space_name}
    return media_input


# ══════════════════════════════════════════════════════════════════════════
# Playback URL signing (turn a DirectUrl/FileName into an accessible URL)
# Reference: vod-media-kit/volcengine-ai-mediakit/scripts/api_manage.py
# ══════════════════════════════════════════════════════════════════════════

_VOD_ACTION_APPLY_UPLOAD_INFO = "ApplyUploadInfo"
_VOD_ACTION_COMMIT_UPLOAD_INFO = "CommitUploadInfo"
_VOD_ACTION_LIST_DOMAIN = "ListDomain"
_VOD_ACTION_DESCRIBE_DOMAIN_CONFIG = "DescribeDomainConfig"
_VOD_ACTION_GET_STORAGE_CONFIG = "GetStorageConfig"
_VOD_ACTION_GET_PLAY_INFO = "GetPlayInfo"
_VOD_ACTION_UPDATE_MEDIA_PUBLISH_STATUS = "UpdateMediaPublishStatus"

_VOD_VERSION = "2023-01-01"

_CACHE: dict = {"available_domains": {}, "storage_config": {}}


def _encode_path_str(s: str = "") -> str:
    return quote(s, safe="-_.~$&+,/:;=@")


def _encode_rfc3986_uri_component(s: str) -> str:
    return quote(s, safe=":/?&=%-_.~")


def _random_string(length: int) -> str:
    alphabet = "0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ"
    return "".join(secrets.choice(alphabet) for _ in range(length))


def _parse_time(value):
    if isinstance(value, (int, float)):
        try:
            return datetime.fromtimestamp(float(value), tz=timezone.utc)
        except Exception:
            return None
    if isinstance(value, str):
        try:
            v = value.replace("Z", "+00:00") if "Z" in value else value
            return datetime.fromisoformat(v)
        except Exception:
            return None
    return None


def _is_https_available(certificate: dict) -> bool:
    if certificate and certificate.get("HttpsStatus") == "enable":
        exp = _parse_time(certificate.get("ExpiredAt"))
        if exp:
            return exp > datetime.now(timezone.utc)
    return False


def _get_domain_config(client: VolcClient, domain: str, space_name: str, domain_type: str = "play") -> dict:
    detail = client.get(
        _VOD_ACTION_DESCRIBE_DOMAIN_CONFIG,
        _VOD_VERSION,
        {"SpaceName": space_name, "Domain": domain, "DomainType": domain_type},
    )
    result = detail.get("Result", {}) if isinstance(detail, dict) else {}
    cdn_config = result.get("Config") or {}
    signed_url_auth_control = cdn_config.get("SignedUrlAuthControl") or {}
    signed_url_auth_rules = (signed_url_auth_control.get("SignedUrlAuth") or {}).get("SignedUrlAuthRules", [])
    if not signed_url_auth_rules:
        return {}
    signed_url_auth_action = (signed_url_auth_rules[0] or {}).get("SignedUrlAuthAction", {}) or {}
    base_domain = result.get("Domain", {}) or {}
    status = "enable" if base_domain.get("ConfigStatus") == "online" else base_domain.get("ConfigStatus")
    return {
        "AuthType": signed_url_auth_action.get("URLAuthType"),
        "AuthKey": signed_url_auth_action.get("MasterSecretKey")
        or signed_url_auth_action.get("BackupSecretKey")
        or "",
        "Status": status,
        "Domain": base_domain.get("Domain", ""),
    }


def _get_available_domain(client: VolcClient, space_name: str) -> list[dict]:
    cached = (_CACHE.get("available_domains") or {}).get(space_name) or []
    if cached:
        return cached

    offset = 0
    total = 1
    domain_list: list[dict] = []
    while offset < total:
        data = client.get(
            _VOD_ACTION_LIST_DOMAIN,
            _VOD_VERSION,
            {"SpaceName": space_name, "SourceStationType": 1, "DomainType": "play", "Offset": offset},
        )
        offset = int(data.get("Offset", 0) or 0)
        total = int(data.get("Total", 0) or 0)
        result = data.get("Result", {}) or {}
        instances = ((result.get("PlayInstanceInfo") or {}).get("ByteInstances") or [])
        for item in instances:
            domains = item.get("Domains") or []
            for domain in domains:
                d = dict(domain)
                d["SourceStationType"] = 1
                d["DomainType"] = "play"
                domain_list.append(d)

    domain_list = [d for d in domain_list if d.get("CdnStatus") == "enable"]
    enriched: list[dict] = []
    for d in domain_list:
        auth_info = _get_domain_config(client, d.get("Domain", ""), space_name, d.get("DomainType", "play"))
        d2 = dict(d)
        d2["AuthInfo"] = auth_info
        enriched.append(d2)

    available = [d for d in enriched if (not d.get("AuthInfo")) or ((d.get("AuthInfo") or {}).get("AuthType") == "typea")]
    _CACHE["available_domains"] = {**(_CACHE.get("available_domains") or {}), space_name: available}
    return available


def _gen_url(domain_obj: dict, path: str, expired_minutes: int) -> str:
    is_https = _is_https_available(domain_obj.get("Certificate") or {})
    file_name = f"/{path}"
    auth_info = domain_obj.get("AuthInfo") or {}
    if auth_info.get("AuthType") == "typea":
        expire_ts = int((datetime.now(timezone.utc) + timedelta(minutes=expired_minutes)).timestamp())
        rand_str = _random_string(16)
        key = auth_info.get("AuthKey") or ""
        md5_input = f"{_encode_path_str(file_name)}-{expire_ts}-{rand_str}-0-{key}".encode("utf-8")
        md5_str = hashlib.md5(md5_input).hexdigest()
        url = (
            f"{'https' if is_https else 'http'}://{domain_obj.get('Domain')}{file_name}"
            f"?auth_key={expire_ts}-{rand_str}-0-{md5_str}"
        )
        return _encode_rfc3986_uri_component(url)
    url = f"{'https' if is_https else 'http'}://{domain_obj.get('Domain')}{file_name}"
    return _encode_rfc3986_uri_component(url)


def _get_storage_config(client: VolcClient, space_name: str) -> dict:
    cached = (_CACHE.get("storage_config") or {}).get(space_name) or {}
    if cached:
        return cached
    reqs = client.get(_VOD_ACTION_GET_STORAGE_CONFIG, _VOD_VERSION, {"SpaceName": space_name})
    storage_config = reqs.get("Result") or {}
    _CACHE["storage_config"] = {**(_CACHE.get("storage_config") or {}), space_name: storage_config}
    return storage_config


def _gen_wild_url(storage_config: dict, file_name: str) -> str:
    file_path = f"/{file_name}"
    conf = storage_config.get("StorageUrlAuthConfig") or {}
    if (
        storage_config.get("StorageType") == "volc"
        and conf.get("Type") == "cdn_typea"
        and conf.get("Status") == "enable"
    ):
        type_a = conf.get("TypeAConfig") or {}
        expire_seconds = int(type_a.get("ExpireTime") or 0)
        expire_ts = int((datetime.now(timezone.utc) + timedelta(seconds=expire_seconds)).timestamp())
        rand_str = _random_string(16)
        key = type_a.get("MasterKey") or type_a.get("BackupKey") or ""
        md5_input = f"{_encode_path_str(file_path)}-{expire_ts}-{rand_str}-0-{key}".encode("utf-8")
        md5_str = hashlib.md5(md5_input).hexdigest()
        sig_arg = type_a.get("SignatureArgs") or "auth_key"
        signed = f"{storage_config.get('StorageHost')}{file_path}?{sig_arg}={expire_ts}-{rand_str}-0-{md5_str}&preview=1"
        return _encode_rfc3986_uri_component(signed)
    if storage_config.get("StorageType") == "volc" and conf.get("Status") == "disable":
        signed = f"{storage_config.get('StorageHost')}{file_path}?preview=1"
        return _encode_rfc3986_uri_component(signed)
    return ""


def get_play_url_by_filename(
    client: VolcClient,
    space_name: str,
    file_name: str,
    *,
    expired_minutes: int = 60,
) -> str:
    """
    Turn a DirectUrl/FileName into an accessible URL (possibly carrying typea auth params).

    Priority:
    1) env `VOD_PLAY_DOMAIN` (build the URL directly from this domain; use to force a specific domain)
    2) ListDomain + DescribeDomainConfig (obtain the CDN domain and auth rules)
    3) GetStorageConfig fallback (use StorageHost + optional cdn_typea)
    """
    if not file_name:
        return ""
    cleaned = file_name
    if cleaned.startswith("directurl://"):
        cleaned = cleaned[len("directurl://"):]

    forced_domain = (os.environ.get("VOD_PLAY_DOMAIN") or "").strip()
    if forced_domain:
        domain = forced_domain.rstrip("/")
        scheme = "https" if "://" not in domain else ""
        if scheme:
            domain = f"https://{domain}"
        return _encode_rfc3986_uri_component(f"{domain}/{cleaned.lstrip('/')}")

    try:
        available = _get_available_domain(client, space_name)
        if available:
            return _gen_url(available[0], cleaned.lstrip("/"), expired_minutes)
    except Exception as exc:
        log(f"Failed to fetch play domain; falling back to storage: {exc}")

    try:
        storage_config = _get_storage_config(client, space_name)
        return _gen_wild_url(storage_config, cleaned.lstrip("/"))
    except Exception as exc:
        log(f"Storage fallback signing failed: {exc}")
        return ""


# ══════════════════════════════════════════════════════════════════════════
# Local file upload: ApplyUploadInfo → TOS PUT → CommitUploadInfo
# Docs: https://docs.byteplus.com/en/byteplus-vod/reference/applyuploadinfo
#       https://docs.byteplus.com/en/byteplus-vod/reference/commituploadinfo
# ══════════════════════════════════════════════════════════════════════════

def apply_upload_info(client: VolcClient, space_name: str, *,
                      file_size: int, file_name: str, file_ext: str) -> dict:
    """
    Call ApplyUploadInfo and return Result.Data (contains UploadAddress, SessionKey, etc.).
    """
    resp = client.get(
        _VOD_ACTION_APPLY_UPLOAD_INFO,
        _VOD_VERSION,
        {
            "SpaceName": space_name,
            "FileSize": file_size,
            "FileType": "",
            "FileName": file_name,
            "FileExtension": file_ext,
            "StorageClass": 1,
            "NeedFallback": True,
        },
    )
    _raise_for_vod(resp)
    return ((resp.get("Result") or {}).get("Data")) or {}


def commit_upload_info(client: VolcClient, space_name: str, session_key: str) -> dict:
    """
    Call CommitUploadInfo and return Result.Data (contains Vid, SourceInfo, etc.).
    """
    resp = client.get(
        _VOD_ACTION_COMMIT_UPLOAD_INFO,
        _VOD_VERSION,
        {"SpaceName": space_name, "SessionKey": session_key},
    )
    _raise_for_vod(resp)
    return ((resp.get("Result") or {}).get("Data")) or {}


def _raise_for_vod(resp: dict) -> None:
    meta = resp.get("ResponseMetadata") or {}
    err = meta.get("Error") or {}
    code = err.get("Code", "") or err.get("code", "")
    if code not in ("", None, 0, "0"):
        rid = meta.get("RequestId", "")
        raise RuntimeError(f"VOD API error: {err} request_id={rid}")


# ══════════════════════════════════════════════════════════════════════════
# Publish + GetPlayInfo helpers
# ══════════════════════════════════════════════════════════════════════════

def _update_media_publish_status(client: VolcClient, vid: str, status: str = "Published") -> None:
    """
    Publish (or unpublish) a media asset.

    Docs: https://docs.byteplus.com/en/docs/byteplus-vod/reference-updatemediapublishstatus
    Required before GetPlayInfo will return a playable URL.
    """
    client.get(
        _VOD_ACTION_UPDATE_MEDIA_PUBLISH_STATUS,
        _VOD_VERSION,
        {"Vid": vid, "Status": status},
    )


def _get_play_info(client: VolcClient, vid: str) -> str:
    """
    Call GetPlayInfo once and return the first playable URL, or "" if none.

    Docs: https://docs.byteplus.com/en/docs/byteplus-vod/reference-getplayinfo
    Response shape: Result.PlayInfoList[*].{MainPlayUrl, BackupPlayUrl}.
    Ssl=1 requests HTTPS playback URLs.
    """
    resp = client.get(
        _VOD_ACTION_GET_PLAY_INFO,
        _VOD_VERSION,
        {"Vid": vid, "Ssl": "1"},
    )
    result = resp.get("Result", {}) if isinstance(resp, dict) else {}
    play_list = result.get("PlayInfoList") or []
    for entry in play_list:
        if not isinstance(entry, dict):
            continue
        url = entry.get("MainPlayUrl") or entry.get("BackupPlayUrl") or ""
        if url:
            return url
    return ""


def get_play_url_by_vid(client: VolcClient, vid: str) -> str:
    """
    Fetch a playable URL by Vid.

    Per the BytePlus docs, GetPlayInfo will only return URLs after the asset
    has been published. UpdateMediaPublishStatus is idempotent, so we always
    publish first and then query once:
      1. UpdateMediaPublishStatus(Published)
      2. GetPlayInfo
    """
    if not vid:
        return ""
    v = vid[len("vid://"):] if vid.startswith("vid://") else vid

    try:
        _update_media_publish_status(client, v, "Published")
    except Exception as exc:
        # Publishing failures are logged but not fatal — we still attempt
        # GetPlayInfo in case the asset was already Published.
        log(f"UpdateMediaPublishStatus failed (vid={v}): {exc}")
    else:
        # Small delay to let the publish propagate before querying.
        time.sleep(0.35)

    try:
        return _get_play_info(client, v)
    except Exception as exc:
        log(f"GetPlayInfo failed (vid={v}): {exc}")
        return ""