Byted Kickart Marketing Material Generator

                                 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.

import os
import time
import logging
from pydantic import BaseModel
from dotenv import load_dotenv

__all__ = ["filename", "Result", "authentication"]
### 日志配置
filename = f"/tmp/openclaw/kickart/info.{time.strftime('%Y%m%d', time.localtime())}.log"
logging.basicConfig(
    level=logging.INFO,
    filename=filename,
    format="%(asctime)s - %(levelname)s - %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S",
)
load_dotenv()


### 结果模型
class Result(BaseModel):
    code: str
    message: str
    data: object = None


def perror(res: Result):
    if res.code != "0":
        print(res.model_dump_json())
    exit(1)


### 判断使用那种鉴权方式
def authentication():
    ADDR = os.getenv("ARK_SKILL_API_BASE")
    TOKEN = os.getenv("ARK_SKILL_API_KEY")

    if ADDR is not None and TOKEN is not None:
        return 2

    ACCESS_KEY_ID = os.getenv("ACCESS_KEY_ID")
    SECRET_ACCESS_KEY = os.getenv("SECRET_ACCESS_KEY")
    if ACCESS_KEY_ID is not None and SECRET_ACCESS_KEY is not None:
        return 1
    return 0

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
智能创作云 aPaaS 媒资上传完整流程脚本 (含 ListUsers 获取 Admin ID)

流程：
  0. ListUsers        — 查找角色为 admin 的用户，获取其 Uid 作为 owner-id
  1. GetUploadState   — 查询文件是否已上传（支持断点续传）
  2. StreamUploadData — 分片上传文件（支持并发、幂等）
  3. CreateMaterial   — 创建媒资，获取 MediaId
  4. GetMediaInfo     — 轮询媒资详情，获取处理状态
  5. 提取下载链接

用法：
  pip install requests
  # 使用 ArkClaw Token
  export ARKCLAW_TOKEN="your-token"
  python upload_material.py --host <host> --file /path/to/file.mp4 \
      --owner-type user --title "我的视频" --category video
  
  # 使用 AK/SK
  export ACCESS_KEY_ID="your-ak"
  export SECRET_ACCESS_KEY="your-sk"
  python upload_material.py --host <host> --file /path/to/file.mp4 \
      --owner-type user --title "我的视频" --category video
"""

import argparse
import hashlib
import json
import os
import sys
import time
import hmac
import jsonpath
import requests
from pydantic import BaseModel

from urllib.parse import urlencode, urlparse
from typing import List, TypedDict
from base import Result

__all__ = ["Matriel", "upload"]
# ─── 类型定义 ──────────────────────────────────────────


class RangeDict(TypedDict):
    Start: int
    End: int


class UploadStateResult(TypedDict):
    SkipDataComplete: bool
    PartSize: int
    Ranges: List[RangeDict]


### 素材
class Matriel(BaseModel):
    type: str
    url: str
    width: int
    height: int


class ImageMatriel(Matriel):
    pass


class VideoMatriel(Matriel):
    duration: float


# ─── 全局常量 ──────────────────────────────────────────

# 从环境变量读取配置
ACCESS_KEY_ID = os.getenv("ACCESS_KEY_ID") or ""
SECRET_ACCESS_KEY = os.getenv("SECRET_ACCESS_KEY") or ""
ARKCLAW_TOKEN = os.getenv("ARK_SKILL_API_KEY") or ""

# 默认域名（ArkClaw 方式）
DEFAULT_HOST = os.getenv("ARK_SKILL_API_BASE") or ""
# AK/SK 方式的默认域名
DEFAULT_ICP_HOST = "https://icp.volcengineapi.com"

REGION = "cn-north"
VERSION = "2022-02-01"

# 不同接口可能属于不同的 Service
SERVICE_MUSE = "iccloud_muse"  # 媒资/任务相关
SERVICE_IAM = "ic_iam"  # 用户/权限相关

POLL_MAX_ATTEMPTS = 60  # 最多轮询 60 次
POLL_INTERVAL = 5  # 每次间隔 5 秒

# 文件类型与后缀映射
IMAGE_EXTENSIONS = {"jpg", "jpeg", "png", "gif", "bmp", "webp", "tiff", "tif"}
VIDEO_EXTENSIONS = {"mp4", "avi", "mov", "wmv", "flv", "mkv", "webm", "m4v", "3gp"}

# Debug 开关
DEBUG = False


def debug_print(*args, **kwargs):
    if DEBUG:
        print(*args, **kwargs)


# ─── AK/SK 鉴权函数（参考 icp.py）─────────────────────


def hmac_sha256(key: bytes, content: str) -> bytes:
    """HMAC-SHA256加密"""
    h = hmac.new(key, content.encode("utf-8"), hashlib.sha256)
    return h.digest()


def get_signed_key(secret_key: str, date: str, region: str, service: str) -> bytes:
    """生成签名密钥链"""
    k_date = hmac_sha256(secret_key.encode("utf-8"), date)
    k_region = hmac_sha256(k_date, region)
    k_service = hmac_sha256(k_region, service)
    k_signing = hmac_sha256(k_service, "request")
    return k_signing


def hash_sha256(data: bytes) -> bytes:
    """SHA256哈希"""
    h = hashlib.sha256()
    h.update(data)
    return h.digest()


# ─── 通用请求 ──────────────────────────────────────────


def _request_with_aksk(
    action: str,
    service: str,
    host: str | None,
    body: dict | None = None,
    extra_query: dict | None = None,
) -> dict:
    """使用 AK/SK 鉴权发起请求"""
    extra_query = extra_query or {}
    body_bytes = json.dumps(body or {}, ensure_ascii=False).encode()

    queries = extra_query.copy()
    queries["Action"] = action
    queries["Version"] = VERSION

    query_string = urlencode(sorted(queries.items()))
    query_string = query_string.replace("+", "%20")
    url = f"{host}?{query_string}"

    date = time.strftime("%Y%m%dT%H%M%SZ", time.gmtime(time.time()))
    auth_date = date[:8]

    payload = hash_sha256(body_bytes).hex()

    signed_headers = ["host", "x-date", "x-content-sha256", "content-type"]
    parsed_url = urlparse(host)
    host_name = parsed_url.netloc

    header_list = []
    for header in signed_headers:
        if header == "host":
            header_list.append(f"{header}:{host_name}")
        elif header == "x-date":
            header_list.append(f"{header}:{date}")
        elif header == "x-content-sha256":
            header_list.append(f"{header}:{payload}")
        elif header == "content-type":
            header_list.append(f"{header}:application/json")
    header_string = "\n".join(header_list)

    canonical_string = "\n".join(
        [
            "POST",
            "/",
            query_string,
            f"{header_string}\n",
            ";".join(signed_headers),
            payload,
        ]
    )
    hashed_canonical_string = hash_sha256(canonical_string.encode("utf-8")).hex()
    credential_scope = f"{auth_date}/{REGION}/{service}/request"
    sign_string = "\n".join(
        ["HMAC-SHA256", date, credential_scope, hashed_canonical_string]
    )
    signed_key = get_signed_key(SECRET_ACCESS_KEY, auth_date, REGION, service)
    signature = hmac_sha256(signed_key, sign_string).hex()
    authorization = (
        f"HMAC-SHA256 Credential={ACCESS_KEY_ID}/{credential_scope},"
        f" SignedHeaders={';'.join(signed_headers)},"
        f" Signature={signature}"
    )

    headers = {
        "X-Date": date,
        "X-Content-Sha256": payload,
        "Content-Type": "application/json",
        "Authorization": authorization,
    }

    debug_print(f"result is {url}")
    debug_print(f"headers is {headers}")
    debug_print(f"url is {url}, body is {body_bytes}")
    debug_print(f"canonical_string: {repr(canonical_string)}")
    debug_print(f"sign_string: {repr(sign_string)}")

    resp = requests.post(url, data=body_bytes, headers=headers, timeout=30)

    try:
        result = resp.json()
    except Exception as e:
        print(f"json parse error, resp is {resp.text}, error is {e}")
        sys.exit(1)

    _check_resp(result, action)

    return result


def _request_with_arkclaw(
    action: str,
    service: str,
    host: str | None,
    arkclaw_token: str | None,
    body: dict | None,
    extra_query: dict | None,
) -> dict:
    """使用 ArkClaw Token 鉴权发起请求"""
    extra_query = extra_query or {}
    body_bytes = json.dumps(body or {}, ensure_ascii=False).encode()
    headers = {
        "ServiceName": service,
        "Authorization": f"Bearer {arkclaw_token}",
        "Content-Type": "application/json",
    }
    url = f"{host}/?Action={action}&Version={VERSION}"
    if extra_query:
        url += "&" + urlencode(extra_query)
    debug_print(f"result is {url}")
    debug_print(f"headers is {headers}")
    debug_print(f"url is {url}, body is {body_bytes}")
    resp = requests.post(url, data=body_bytes, headers=headers, timeout=30)

    try:
        result = resp.json()
    except Exception as e:
        print(f"json parse error, resp is {resp.text}, error is {e}")
        sys.exit(1)

    _check_resp(result, action)

    return result


def _request_binary_with_arkclaw(
    action: str,
    service: str,
    host: str | None,
    arkclaw_token: str | None,
    extra_query: dict | None,
    data: bytes,
) -> dict:
    """使用 ArkClaw Token 鉴权发起二进制请求"""
    headers = {
        "ServiceName": service,
        "Authorization": f"Bearer {arkclaw_token}",
        "Content-Type": "application/octet-stream",
    }

    url = f"{host}/?Action={action}&Version={VERSION}"
    if extra_query:
        url += "&" + urlencode(extra_query)
    debug_print(f"url is {url}, data size is {len(data)}")
    resp = requests.post(url, data=data, headers=headers, timeout=60)

    try:
        result = resp.json()
    except Exception as e:
        print(f"json parse error, resp is {resp.text}, error is {e}")
        sys.exit(1)

    _check_resp(result, action)

    return result


def _request_binary_with_aksk(
    action: str, service: str, host: str | None, extra_query: dict, data: bytes
) -> dict:
    """使用 AK/SK 鉴权发起二进制请求"""
    queries = extra_query.copy()
    queries["Action"] = action
    queries["Version"] = VERSION

    query_string = urlencode(sorted(queries.items()))
    query_string = query_string.replace("+", "%20")
    url = f"{host}?{query_string}"

    date = time.strftime("%Y%m%dT%H%M%SZ", time.gmtime(time.time()))
    auth_date = date[:8]

    payload = hash_sha256(data).hex()

    signed_headers = ["host", "x-date", "x-content-sha256", "content-type"]
    parsed_url = urlparse(host)
    host_name = parsed_url.netloc

    header_list = []
    for header in signed_headers:
        if header == "host":
            header_list.append(f"{header}:{host_name}")
        elif header == "x-date":
            header_list.append(f"{header}:{date}")
        elif header == "x-content-sha256":
            header_list.append(f"{header}:{payload}")
        elif header == "content-type":
            header_list.append(f"{header}:application/octet-stream")
    header_string = "\n".join(header_list)

    canonical_string = "\n".join(
        [
            "POST",
            "/",
            query_string,
            f"{header_string}\n",
            ";".join(signed_headers),
            payload,
        ]
    )
    hashed_canonical_string = hash_sha256(canonical_string.encode("utf-8")).hex()
    credential_scope = f"{auth_date}/{REGION}/{service}/request"
    sign_string = "\n".join(
        ["HMAC-SHA256", date, credential_scope, hashed_canonical_string]
    )
    signed_key = get_signed_key(SECRET_ACCESS_KEY, auth_date, REGION, service)
    signature = hmac_sha256(signed_key, sign_string).hex()
    authorization = (
        f"HMAC-SHA256 Credential={ACCESS_KEY_ID}/{credential_scope},"
        f" SignedHeaders={';'.join(signed_headers)},"
        f" Signature={signature}"
    )

    headers = {
        "X-Date": date,
        "X-Content-Sha256": payload,
        "Content-Type": "application/octet-stream",
        "Authorization": authorization,
    }

    debug_print(f"url is {url}, data size is {len(data)}")
    debug_print(f"canonical_string: {repr(canonical_string)}")
    debug_print(f"sign_string: {repr(sign_string)}")
    resp = requests.post(url, data=data, headers=headers, timeout=60)

    try:
        result = resp.json()
    except Exception as e:
        print(f"json parse error, resp is {resp.text}, error is {e}")
        sys.exit(1)

    _check_resp(result, action)

    return result


# 全局变量，存储授权方式和参数
_auth_mode = None
_arkclaw_token = None
_host = None


def init_auth(
    auth_mode: str, arkclaw_token: str | None = None, host: str | None = None
):
    """初始化授权配置"""
    global _auth_mode, _arkclaw_token, _host
    _auth_mode = auth_mode
    _arkclaw_token = arkclaw_token
    _host = host


def _request(
    action: str,
    service: str = SERVICE_MUSE,
    body: dict | None = None,
    extra_query: dict | None = None,
) -> dict:
    """通用请求函数，根据授权模式选择"""
    if _auth_mode == "arkclaw":
        return _request_with_arkclaw(
            action, service, _host, _arkclaw_token, body, extra_query
        )
    else:
        return _request_with_aksk(action, service, _host, body, extra_query)


def _request_binary(action: str, service: str, extra_query: dict, data: bytes) -> dict:
    """通用二进制请求函数，根据授权模式选择"""
    if _auth_mode == "arkclaw":
        return _request_binary_with_arkclaw(
            action, service, _host, _arkclaw_token, extra_query, data
        )
    else:
        return _request_binary_with_aksk(action, service, _host, extra_query, data)


# ─── Step 0: ListUsers ────────────────────────────────


def get_admin_user_id() -> int:
    """调用 ListUsers 接口，从用户列表中寻找 admin 用户。"""
    debug_print("[0/5] ListUsers — 正在寻找 admin 用户...")
    result = _request(action="ListUsers", service=SERVICE_IAM, body={"UserType": "All"})
    _check_resp(result, "ListUsers")

    users = result.get("Result", {}).get("Users", [])

    debug_print(f"\n📋 调试信息：共找到 {len(users)} 个用户")
    for i, user in enumerate(users, 1):
        debug_print(f"\n用户 {i}:")
        debug_print(f"  Id: {user.get('Id')}")
        debug_print(f"  IsAdmin: {user.get('IsAdmin')}")
        debug_print(f"  Permitted: {user.get('Permitted')}")
        debug_print(f"  DisplayName: {user.get('DisplayName')}")
        debug_print(f"  VolcUserName: {user.get('VolcUserName')}")
        debug_print(f"  VolcUserId: {user.get('VolcUserId')}")

    if not users:
        print("❌ 未获取到任何用户信息")
        sys.exit(1)

    for user in users:
        if user.get("IsAdmin"):
            uid = user.get("Id")
            name = user.get("DisplayName") or user.get("VolcUserName") or ""
            if uid:
                debug_print(f"    找到 admin 用户: {name} (Id={uid}) ✅")
                return uid

    first_uid = users[0].get("Id")
    first_name = users[0].get("DisplayName") or users[0].get("VolcUserName") or ""
    debug_print(
        f"⚠️ 未找到 IsAdmin=true 的用户，使用第一个用户代替: {first_name} (Id={first_uid})"
    )
    return first_uid


# ─── Step 1: GetUploadState ────────────────────────────


def get_upload_state(
    file_md5: str, file_size: int, file_crc32: int, owner_id: int
) -> UploadStateResult:
    """查询文件是否已上传（支持断点续传）"""
    debug_print(
        f"[1/5] GetUploadState — md5={file_md5}, size={file_size}, crc32={file_crc32}"
    )

    body = {
        "Owner": {"Id": owner_id, "Type": "PERSON"},
        "Md5": file_md5,
        "Size": file_size,
        "Start": 0,
        "End": file_size - 1,
        "Crc": file_crc32,
    }

    result = _request(action="GetUploadState", service=SERVICE_MUSE, body=body)
    _check_resp(result, "GetUploadState")

    raw_state = result.get("Result", {})
    state: UploadStateResult = {
        "SkipDataComplete": bool(raw_state.get("SkipDataComplete", False)),
        "PartSize": int(raw_state.get("PartSize", 0)),
        "Ranges": raw_state.get("Ranges", []),
    }
    debug_print(f"文件上传状态：{state}")
    return state


def upload_part(
    owner_id: int, chunk: bytes, offset: int, part_size: int, chunk_md5: str
) -> dict:
    """上传文件分片"""
    debug_print(
        f"    上传分片 {offset} 到 {offset + part_size - 1} (大小 {part_size} 字节)"
    )

    query = {
        "Md5": chunk_md5,
        "Size": part_size,
        "Offset": offset,
        "OwnerId": owner_id,
        "OwnerType": "PERSON",
    }
    resp = _request_binary("StreamUploadData", SERVICE_MUSE, query, chunk)
    _check_resp(resp, f"StreamUploadData，分片 {offset} 到 {offset + part_size - 1}")

    return resp


# ─── Step 2: StreamUploadData ──────────────────────────


def stream_upload_data(
    file_path: str,
    file_md5: str,
    file_size: int,
    file_crc32: int,
    owner_id: int,
    upload_state: UploadStateResult,
) -> UploadStateResult:
    if upload_state["SkipDataComplete"]:
        debug_print("[2/5] 文件已存在且上传完成，跳过上传 ✅")
        return upload_state

    with open(file_path, "rb") as f:
        data = f.read()

    current_state = upload_state
    chunk_count = 0
    max_chunks = 1000
    offset = 0

    debug_print("[2/5] StreamUploadData — 开始上传...")

    while not current_state["SkipDataComplete"] and chunk_count < max_chunks:
        skip_data_complete = current_state["SkipDataComplete"]
        part_size = current_state.get("PartSize", 0)

        if skip_data_complete:
            break

        if part_size == 0:
            debug_print("    PartSize 为 0，上传整个文件")
            chunk = data
            chunk_size = file_size
        else:
            debug_print(f"    使用服务端返回的 PartSize: {part_size} bytes")
            debug_print(f"    当前上传位置: offset={offset}")

            if offset + part_size * 2 <= file_size:
                chunk = data[offset : offset + part_size]
                chunk_size = part_size
            else:
                chunk = data[offset:file_size]
                chunk_size = file_size - offset

        debug_print(
            f"[2/5] StreamUploadData — 上传第 {chunk_count + 1} 个分片，offset={offset}, size={chunk_size}"
        )
        upload_part(owner_id, chunk, offset, chunk_size, file_md5)

        chunk_count += 1
        offset += chunk_size

        current_state = get_upload_state(file_md5, file_size, file_crc32, owner_id)
        debug_print(f"current_state is {current_state}")

        if (
            current_state["SkipDataComplete"]
            or current_state["Ranges"] == []
            or offset >= file_size
        ):
            debug_print(
                "    上传完成，SkipDataComplete=true 或 Ranges 为空 或 offset 已超过文件大小 ✅"
            )
            return current_state

    if chunk_count >= max_chunks and not current_state["SkipDataComplete"]:
        print(
            f"❌ 上传失败：已上传 {max_chunks} 个分片，但 SkipDataComplete 仍为 false"
        )
        sys.exit(1)

    return current_state


# ─── Step 3: CreateMaterial ────────────────────────────


def create_material(
    file_md5: str,
    file_size: int,
    file_name: str,
    file_ext: str,
    skip_data_complete: bool,
    owner_id: int,
    owner_type: str,
    title: str,
    category: str,
) -> str:
    debug_print(
        f"[3/5] CreateMaterial — title={title}, category={category}, skip_data_complete={skip_data_complete}"
    )

    body = {
        "Owner": {"Id": owner_id, "Type": "PERSON"},
        "StoreItem": {
            "Md5": file_md5,
            "Size": file_size,
            "SkipDataComplete": skip_data_complete,
            "Filename": file_name,
            "FileExtension": file_ext,
        },
        "CreateMaterialInfo": {
            "Visibility": 0,
            "Title": title,
            "MediaType": 1,
            "MediaFirstCategory": category,
            "Tags": [],
            "MediaExtension": file_ext,
        },
    }

    debug_print(
        f"    [DEBUG] CreateMaterial Body: {json.dumps(body, ensure_ascii=False)}"
    )

    result = _request(action="CreateMaterial", service=SERVICE_MUSE, body=body)
    _check_resp(result, "CreateMaterial")
    media_id = result.get("Result", {}).get("MediaId")
    debug_print(f"    MediaId={media_id} ✅")
    return media_id


# ─── Step 4: GetMediaInfo ──────────────────────────────


def get_media_info(media_id: str, owner_id: int, owner_type: str) -> dict:
    debug_print(f"[4/5] GetMediaInfo 轮询 — MediaId={media_id}")

    for attempt in range(1, POLL_MAX_ATTEMPTS + 1):
        result = _request(
            action="GetMediaInfo",
            service=SERVICE_MUSE,
            body={"MediaIds": [media_id], "MediaType": 1},
        )
        _check_resp(result, "GetMediaInfo")
        media_infos = result.get("Result", {}).get("MediaInfos", [])
        if not media_infos:
            debug_print(f"    [{attempt}/{POLL_MAX_ATTEMPTS}] 未获取到媒资信息...")
            time.sleep(POLL_INTERVAL)
            continue

        media_info = media_infos[0]
        status = media_info.get("BasicInfo", {}).get("MediaStatus")

        if status >= 2:
            debug_print(f"    处理成功 ✅, media_info={media_info}")
            return media_info
        if status == 5 or status == 1:
            print("❌ 处理失败")
            sys.exit(1)

        debug_print(
            f"    [{attempt}/{POLL_MAX_ATTEMPTS}] 处理中 (MediaStatus={status})..."
        )
        time.sleep(POLL_INTERVAL)
    sys.exit(1)


# ─── Step 5: 提取下载链接 ──────────────────────────────


def extract_url(media_info: dict) -> str:
    media_first_category = media_info.get("BasicInfo", {}).get("MediaFirstCategory", "")

    if media_first_category == "image":
        image_media = media_info.get("ImageMedia", {})
        download_url = image_media.get("DownloadUrl", "")
        if download_url:
            return download_url
        transcode_urls = image_media.get("TranscodeDownloadUrls", {})
        if transcode_urls:
            for quality in ["origin", "jpeg_1080p", "jpeg_480p"]:
                if quality in transcode_urls:
                    return transcode_urls[quality]
    elif media_first_category == "video":
        video_media = media_info.get("VideoMedia", {})
        download_url = video_media.get("DownloadUrl", "")
        if download_url:
            return download_url
        play_info = video_media.get("PlayInfo", [])
        if play_info and play_info[0].get("Url"):
            return play_info[0]["Url"]
    elif media_first_category == "audio":
        audio_media = media_info.get("AudioMedia", {})
        download_url = audio_media.get("DownloadUrl", "")
        if download_url:
            return download_url
        play_info = audio_media.get("PlayInfo", [])
        if play_info and play_info[0].get("Url"):
            return play_info[0]["Url"]

    return ""


# ─── 辅助函数 ──────────────────────────────────────────


def _file_md5(file_path: str) -> str:
    h = hashlib.md5()
    with open(file_path, "rb") as f:
        for chunk in iter(lambda: f.read(1 << 20), b""):
            h.update(chunk)
    return h.hexdigest()


def _check_resp(resp: dict, action: str):
    debug_print(f"resp is {resp}")
    meta = resp.get("ResponseMetadata", {})

    error_obj = meta.get("Error")
    if error_obj:
        code = error_obj.get("Code") or error_obj.get("CodeN")
        if code == 500 or code == "500":
            print(
                Result(
                    code="x01010",
                    message="阅读SKILL.md中的「错误处理规范」，向用户推荐处理意见～",
                ).model_dump_json()
            )
            sys.exit(1)
        msg = error_obj.get("Message", "")
        print(f"❌ {action} 失败: code={code}, msg={msg}")
        sys.exit(1)

    code = meta.get("Code")
    if code is not None and str(code) not in ("0", "Success", "200"):
        if code == 500 or code == "500":
            print(
                Result(
                    code="x01010",
                    message="阅读SKILL.md中的「错误处理规范」，向用户推荐处理意见～",
                ).model_dump_json()
            )
            sys.exit(1)
        msg = meta.get("Message") or ""
        print(f"❌ {action} 失败: code={code}, msg={msg}")
        sys.exit(1)

    debug_print(f"✅ {action} 成功")


# ─── 主流程 ────────────────────────────────────────────


def list_users_debug():
    """调试接口：单独列出所有用户信息"""
    print("🔍 调试模式：获取用户列表\n")
    result = _request(action="ListUsers", service=SERVICE_IAM, body={"UserType": "All"})
    _check_resp(result, "ListUsers")

    users = result.get("Result", {}).get("Users", [])
    print(f"共找到 {len(users)} 个用户\n")

    for i, user in enumerate(users, 1):
        print(f"{'=' * 60}")
        print(f"用户 {i}:")
        print(f"{'=' * 60}")
        print(f"  Id:            {user.get('Id')}")
        print(f"  IsAdmin:       {user.get('IsAdmin')}")
        print(f"  Permitted:     {user.get('Permitted')}")
        print(f"  DisplayName:   {user.get('DisplayName')}")
        print(f"  Description:   {user.get('Description')}")
        print(f"  VolcUserName:  {user.get('VolcUserName')}")
        print(f"  VolcUserId:    {user.get('VolcUserId')}")
        print(f"  TeamInfos:     {user.get('TeamInfos')}")
        print(f"  RoleInfos:     {user.get('RoleInfos')}")
        print(f"  完整信息:      {json.dumps(user, ensure_ascii=False, indent=4)}")
        print()

    return users


def main():
    parser = argparse.ArgumentParser(description="智能创作云媒资上传工具")
    parser.add_argument("--host", help="API 域名地址")
    parser.add_argument(
        "--list-users", action="store_true", help="调试模式：仅列出所有用户信息"
    )
    parser.add_argument("--file", help="本地文件路径")
    parser.add_argument(
        "--owner-id", type=int, help="Owner Id (可选，若不提供则自动查找 admin)"
    )
    parser.add_argument("--owner-type", default="user", help="Owner Type (user/team)")
    parser.add_argument("--title", help="标题（可选，默认自动生成）")
    parser.add_argument("--category", default="video", help="类型: video/image/audio")
    parser.add_argument("--debug", action="store_true", help="开启调试日志")
    args = parser.parse_args()

    if args.debug:
        global DEBUG
        DEBUG = True

    auth_mode = None
    arkclaw_token = None
    host = None

    if ARKCLAW_TOKEN:
        auth_mode = "arkclaw"
        arkclaw_token = ARKCLAW_TOKEN
        host = args.host or DEFAULT_HOST
        debug_print(f"使用 ArkClaw Token 方式，域名: {host}")
    elif ACCESS_KEY_ID and SECRET_ACCESS_KEY:
        auth_mode = "aksk"
        host = args.host or DEFAULT_ICP_HOST
        debug_print(f"使用 AK/SK 方式，域名: {host}")
    else:
        print("❌ 请设置环境变量 ARKCLAW_TOKEN 或 ACCESS_KEY_ID/SECRET_ACCESS_KEY")
        sys.exit(1)

    init_auth(auth_mode, arkclaw_token, host)

    if args.list_users:
        list_users_debug()
        return

    if not args.file:
        parser.error("上传模式需要 --file 参数")

    if not args.title:
        args.title = f"artclaw-material-{int(time.time())}"

    if not os.path.exists(args.file):
        print(f"❌ 文件不存在: {args.file}")
        sys.exit(1)
    if not os.path.isfile(args.file):
        print(f"❌ 路径不是文件: {args.file}")
        sys.exit(1)

    owner_id = args.owner_id
    if owner_id is None:
        owner_id = get_admin_user_id()

    import hashlib
    import zlib

    file_md5_obj = hashlib.md5()
    file_crc32 = 0
    file_size = 0

    with open(args.file, "rb") as f:
        while chunk := f.read(8192 * 1024):
            file_md5_obj.update(chunk)
            file_crc32 = zlib.crc32(chunk, file_crc32)
            file_size += len(chunk)

    file_md5 = file_md5_obj.hexdigest()
    file_crc32 = file_crc32 & 0xFFFFFFFF

    file_name = os.path.splitext(os.path.basename(args.file))[0]
    file_ext = os.path.splitext(args.file)[1].lstrip(".")

    if not args.category or args.category == "video":
        file_ext_lower = file_ext.lower()
        if file_ext_lower in IMAGE_EXTENSIONS:
            args.category = "image"
        elif file_ext_lower in VIDEO_EXTENSIONS:
            args.category = "video"

    state = get_upload_state(file_md5, file_size, file_crc32, owner_id)

    state = stream_upload_data(
        args.file, file_md5, file_size, file_crc32, owner_id, state
    )

    media_id = create_material(
        file_md5,
        file_size,
        file_name,
        file_ext,
        state["SkipDataComplete"],
        owner_id,
        args.owner_type,
        args.title,
        args.category,
    )
    media_info = get_media_info(media_id, owner_id, args.owner_type)
    url = extract_url(media_info)

    print(f"\n✅ 流程结束！\nMediaId: {media_id}\nDownloadUrl: {url}")


def simplify(media_info: dict):
    media_first_category = media_info.get("BasicInfo", {}).get("MediaFirstCategory", "")
    if media_first_category == "image":
        image_media = media_info.get("ImageMedia", {})
        download_url = image_media.get("DownloadUrl", None)
        if download_url:
            image_media["DownloadUrl"] = download_url

        transcode_urls = image_media.get("TranscodeDownloadUrls", {})
        for quality in ["origin", "jpeg_1080p", "jpeg_480p"]:
            if quality in transcode_urls:
                transcode_urls[quality] = transcode_urls[quality]

    if media_first_category == "video":
        video_media = media_info.get("VideoMedia", {})
        download_url = video_media.get("DownloadUrl", None)
        if download_url:
            video_media["DownloadUrl"] = download_url
        play_info = video_media.get("PlayInfo", [])
        if play_info and play_info[0].get("Url"):
            play_info[0]["Url"] = play_info[0]["Url"]

    if media_first_category == "audio":
        audio_media = media_info.get("AudioMedia", {})
        download_url = audio_media.get("DownloadUrl", None)
        if download_url:
            audio_media["DownloadUrl"] = download_url
        play_info = audio_media.get("PlayInfo", [])
        if play_info and play_info[0].get("Url"):
            play_info[0]["Url"] = play_info[0]["Url"]

    return media_info


def format(media_info: dict) -> Matriel:
    image_media = jsonpath.jsonpath(media_info, "$.ImageMedia")
    if image_media:
        matriel = ImageMatriel(type="image", url="", height=0, width=0)
        url = jsonpath.jsonpath(media_info, "$.ImageMedia.DownloadUrl")
        matriel.url = url[0] if url else ""

        width = jsonpath.jsonpath(media_info, "$.ImageMedia.Width")
        matriel.width = width[0] if width else 0

        height = jsonpath.jsonpath(media_info, "$.ImageMedia.Height")
        matriel.height = height[0] if height else 0
        return matriel

    video_media = jsonpath.jsonpath(media_info, "$.VideoMedia")
    if video_media:
        matriel = VideoMatriel(type="video", url="", height=0, width=0, duration=0)
        url = jsonpath.jsonpath(media_info, "$.VideoMedia.DownloadUrl")
        matriel.url = url[0] if url else ""

        width = jsonpath.jsonpath(media_info, "$.VideoMedia.MediaMetaInfo.Width")
        matriel.width = width[0] if width else 0

        height = jsonpath.jsonpath(media_info, "$.VideoMedia.MediaMetaInfo.Height")
        matriel.height = height[0] if height else 0

        duration = jsonpath.jsonpath(media_info, "$.VideoMedia.MediaMetaInfo.Duration")
        matriel.duration = duration[0] / 1000 if duration else 0.0
        return matriel
    return Matriel(type="", url="", height=0, width=0)


def upload(args: dict) -> Matriel:
    # 自动生成 title
    args["title"] = f"artclaw-material-{int(time.time())}"
    args["owner_type"] = "user"

    auth_mode = None
    arkclaw_token = None
    host = None

    if ARKCLAW_TOKEN:
        auth_mode = "arkclaw"
        arkclaw_token = ARKCLAW_TOKEN
        host = DEFAULT_HOST
        debug_print(f"使用 ArkClaw Token 方式，域名: {host}")
    elif ACCESS_KEY_ID and SECRET_ACCESS_KEY:
        auth_mode = "aksk"
        host = DEFAULT_ICP_HOST
        debug_print(f"使用 AK/SK 方式，域名: {host}")
    else:
        print("❌ 请设置环境变量 ARKCLAW_TOKEN 或 ACCESS_KEY_ID/SECRET_ACCESS_KEY")
        sys.exit(1)

    init_auth(auth_mode, arkclaw_token, host)
    # 如果没有提供 owner-id，则自动通过 ListUsers 获取
    owner_id = args["owner_id"] if "owner_id" in args else get_admin_user_id()

    import hashlib
    import zlib

    file_md5_obj = hashlib.md5()
    file_crc32 = 0
    file_size = 0

    with open(args["file"], "rb") as f:
        while chunk := f.read(8192 * 1024):
            file_md5_obj.update(chunk)
            file_crc32 = zlib.crc32(chunk, file_crc32)
            file_size += len(chunk)

    file_md5 = file_md5_obj.hexdigest()
    file_crc32 = file_crc32 & 0xFFFFFFFF

    file_name = os.path.splitext(os.path.basename(args["file"]))[0]
    file_ext = os.path.splitext(args["file"])[1].lstrip(".")

    # 根据文件后缀自动识别 category
    file_ext_lower = file_ext.lower()
    if file_ext_lower in IMAGE_EXTENSIONS:
        args["category"] = "image"
    elif file_ext_lower in VIDEO_EXTENSIONS:
        args["category"] = "video"

    state = get_upload_state(file_md5, file_size, file_crc32, owner_id)

    state = stream_upload_data(
        args["file"], file_md5, file_size, file_crc32, owner_id, state
    )

    media_id = create_material(
        file_md5,
        file_size,
        file_name,
        file_ext,
        state["SkipDataComplete"],
        owner_id,
        args["owner_type"],
        args["title"],
        args["category"],
    )
    media_info = get_media_info(media_id, owner_id, args["owner_type"])

    return format(simplify(media_info))


if __name__ == "__main__":
    main()

import os
import sys
import json
import logging
import subprocess
from subprocess import DEVNULL
import click
from base import Result, perror
from task import submit


@click.command()
@click.option("--storyboard", default=0, type=int, help="用于成片任务的故事板编号")
@click.option(
    "--session",
    required=True,
    type=str,
    help="当前Session的UUID，可用于openclaw agent指令的--session-id参数",
)
@click.option("--metadata", required=True, type=str, help="当前消息的完整未修改元信息")
@click.option("--input", required=True, type=str, help="故事板创作结果JSON文件路径")
@click.option("--output", required=True, type=str, help="输出结果所在的json文件路径")
def main(storyboard, session, metadata, input, output):
    """消费成片任务"""
    logging.info(f"[tool] >>> python3 {' '.join(sys.argv)}")

    with open(input, "r") as f:
        data = json.load(f)

    storyboards = data["storyboards"]
    if storyboard >= len(storyboards):
        perror(
            Result(
                code="-1",
                message=f"故事板编号超出范围，总故事板数为{len(storyboards)}, 请指定一个有效的故事板编号（1-{len(storyboards)}）",
            )
        )

    data["storyboard"] = storyboards[storyboard]
    data["storyboards"] = None

    submit_res = submit(2935355633875543, json.dumps(data, ensure_ascii=False))
    if submit_res.code != "0":
        perror(submit_res)
    print(submit_res.model_dump_json(), flush=True)

    ### 创建后台轮询任务
    workspace = os.path.dirname(os.path.abspath(__file__))
    metadata = json.loads(metadata)
    cmd = [
        "bash",
        "poll.sh",
        submit_res.message,
        output,
        session,
        metadata["channel"],
        metadata["chat_id"],
    ]
    process = subprocess.Popen(
        cmd,
        cwd=workspace,
        text=True,
        start_new_session=True,
        stdin=DEVNULL,
        stdout=DEVNULL,
        stderr=DEVNULL,
    )
    logging.info(f"[cron] >>> {submit_res.message} to {process.pid}")


if __name__ == "__main__":
    main()

import sys
import json
import logging
import click
from collections import defaultdict

from media import media_list
from base import Result, perror
from task import submit, poll


def concat(input=None, session_id=None, duration=15, prompt="无人物出镜") -> dict:
    material = defaultdict()
    if input:
        with open(input, "r") as f:
            material = json.load(f)
    material["video_duration"] = duration
    material["user_prompt"] = prompt

    if session_id:
        # Create a namespace-like object for media.list
        materials = media_list(session_id)
        material["user_images"] = [m for m in materials if m["type"] == "image"]
        material["user_videos"] = [m for m in materials if m["type"] == "video"]
    return material


def check(duration) -> Result:
    if duration < 0 or duration > 60:
        return Result(code="-1", message="视频时长必须在0-60秒之间", data=None)
    return Result(code="0", message="success", data=None)


@click.command()
@click.option("--input", type=str, help="素材分析结果JSON文件路径")
@click.option("--session-id", type=str, help="已上传的远程素材列表对应的会话ID")
@click.option("--duration", default=15, type=int, help="自定义时长，默认15秒")
@click.option(
    "--prompt", default="无人物出镜", type=str, help="自定义提示，默认无人物出镜"
)
@click.option("--output", required=True, type=str, help="输出结果所在的json文件路径")
def main(input, session_id, duration, prompt, output):
    """创意分析服务"""
    logging.info(f"[tool] >>> python3 {' '.join(sys.argv)}")

    material = json.dumps(
        concat(input, session_id, duration, prompt), ensure_ascii=False
    )

    check_res = check(duration)
    if check_res.code != "0":
        perror(check_res)

    submit_res = submit(3339986037439799, material)
    if submit_res.code != "0":
        perror(submit_res)
    print(f"提交任务成功，任务ID: {submit_res.message}", flush=True)

    poll_res = poll(submit_res.message)
    if poll_res.code != "0":
        perror(poll_res)
    with open(output, "w") as f:
        result = json.loads(poll_res.message)
        json.dump(result, f, ensure_ascii=False, indent=2)
    print(Result(code="0", message=output).model_dump_json())


if __name__ == "__main__":
    main()

import sys
import json
import click
import logging
import requests
from urllib.parse import urlparse, parse_qs, urlencode
from base import Result, perror
from task import submit, poll


def simplify(url: str, keys: list) -> Result:
    try:
        # 校验域名
        parsed_original = urlparse(url)
        original_domain = parsed_original.netloc
        allowed_domains = ["haohuo.jinritemai.com", "v.douyin.com"]

        if original_domain not in allowed_domains:
            return Result(
                code="-1",
                message=f"URL域名不支持，仅支持以下域名：{', '.join(allowed_domains)}",
            )

        response = requests.head(url, allow_redirects=True)
        parsed = urlparse(response.url)
        query = {k: v for k, v in parse_qs(parsed.query).items() if k in keys}
        simplified_url = parsed._replace(query=urlencode(query, doseq=True)).geturl()
        return Result(code="0", message=simplified_url)
    except Exception as e:
        return Result(code="-1", message=f"简化URL失败：{str(e)}")


@click.command()
@click.option("--url", required=True, type=str, help="抖店商品链接")
@click.option("--output", required=True, type=str, help="输出结果所在的json文件路径")
def main(url, output):
    """调用创作云服务"""
    logging.info(f"[tool] >>> python3 {' '.join(sys.argv)}")

    simplify_res = simplify(url, ["id"])
    if simplify_res.code != "0":
        perror(simplify_res)
    print(f"简化URL成功，简化后的URL: {simplify_res.message}")

    submit_res = submit(
        3296206833079096, json.dumps({"url": simplify_res.message}, ensure_ascii=False)
    )
    if submit_res.code != "0":
        perror(submit_res)
    print(f"提交任务成功，任务ID: {submit_res.message}", flush=True)

    poll_res = poll(submit_res.message)
    if poll_res.code != "0":
        perror(poll_res)
    with open(output, "w") as f:
        result = json.loads(poll_res.message)
        json.dump(result, f, ensure_ascii=False, indent=2)
    print(Result(code="0", message=output).model_dump_json())


if __name__ == "__main__":
    main()

import json
import os
import sys
import collections
from typing import Dict, Any
from pathlib import Path
import pandas as pd
from PIL import Image
import logging
import time
import click
from chunks import upload

__all__ = ["media_list"]

COLUMNS = ["session_id", "path", "material", "timestamp"]
IMAGE_EXTENSIONS = {".jpg", ".jpeg", ".png"}
VIDEO_EXTENSIONS = {".mp4", ".avi", ".mov"}

IMAGE_MAX_SIZE, VIDEO_MAX_SIZE = 8 * 1024 * 1024, 50 * 1024 * 1024
IMAGE_MIN_WIDTH, IMAGE_MIN_HEIGHT, IMAGE_MAX_PIXELS = 300, 300, 36_000_000


def validate(file_path: str) -> Dict[str, Any]:
    """
    校验图片/视频文件是否合法。

    返回示例：
    {
        "valid": True,
        "file_type": "image",
        "errors": [],
        "warnings": []
    }
    """
    result = {"valid": False, "file_type": None, "errors": [], "warnings": []}

    if not os.path.isfile(file_path):
        result["errors"].append("文件不存在")
        return result

    _, ext = os.path.splitext(file_path)
    ext = ext.lower()

    file_size = os.path.getsize(file_path)

    # 图片校验
    if ext in IMAGE_EXTENSIONS:
        result["file_type"] = "image"

        if file_size > IMAGE_MAX_SIZE:
            result["warnings"].append("图片单张大小建议≤8MB")
            return result

        try:
            with Image.open(file_path) as img:
                width, height = img.size
                total_pixels = width * height

                if width < IMAGE_MIN_WIDTH or height < IMAGE_MIN_HEIGHT:
                    result["errors"].append(
                        f"图片分辨率不足，当前为 {width}x{height}，要求至少 300x300"
                    )

                if total_pixels > IMAGE_MAX_PIXELS:
                    result["errors"].append(
                        f"图片总像素过大，当前为 {total_pixels}，要求≤36,000,000"
                    )
                result["valid"] = True
                return result
        except Exception as e:
            result["errors"].append(f"无法读取图片文件: {e}")
            return result

    # 视频校验
    if ext in VIDEO_EXTENSIONS:
        result["file_type"] = "video"

        if file_size > VIDEO_MAX_SIZE:
            result["errors"].append("视频文件大小超过 50MB")
            return result

        result["valid"] = True
        return result

    result["errors"].append(
        "不支持的文件格式，仅支持图片(jpg/jpeg/png)或视频(mp4/avi/mov)"
    )
    return result


def load(session_id: str) -> pd.DataFrame:
    path = Path(f"/tmp/kickart/material_state_{session_id}.csv")
    if not path.exists():
        return pd.DataFrame(columns=COLUMNS)
    return pd.read_csv(path, header=None, names=COLUMNS)


def save(session_id: str, df: pd.DataFrame):
    path = Path(f"/tmp/kickart/material_state_{session_id}.csv")
    os.makedirs(path.parent, exist_ok=True)
    df.to_csv(path, index=False, header=False)


def remove(session_id: str):
    path = Path(f"/tmp/kickart/material_state_{session_id}.csv")
    os.remove(path)

def media_list(session_id):
    """列出所有已上传的抖音营销素材"""
    df = load(session_id)
    df = df["material"].map(lambda x: json.loads(x))
    return df.to_list()

@click.group()
def main():
    """抖音营销素材上传工具"""
    logging.info(f"[tool] >>> python3 {' '.join(sys.argv)}")


@main.command()
@click.argument("file")
@click.option("--session-id", "-s", required=True, help="会话ID")
def add(file, session_id):
    """上传抖音营销素材到远程服务器，返还上传后的素材URL"""
    # 文件校验 图片8M，视频50M
    result = validate(file)
    if not result["valid"]:
        click.echo(result)
        return
    # upload file to remote server
    matriel = upload({"file": file})

    row = collections.defaultdict()
    row["session_id"] = session_id
    row["path"] = file
    row["material"] = matriel.model_dump_json()
    row["timestamp"] = str(time.time())

    df = load(session_id)
    df.loc[len(df)] = row
    save(session_id, df)
    click.echo(matriel)


@main.command()
@click.option("--session-id", "-s", required=True, help="会话ID")
def list(session_id):
    """列出所有已上传的抖音营销素材"""
    df = load(session_id)
    df = df["material"].map(lambda x: json.loads(x))
    click.echo(df.to_list())
    return df.to_list()


@main.command()
@click.option("--session-id", "-s", required=True, help="会话ID")
def clear(session_id):
    """清空当前会话中的所有已上传的抖音营销素材"""
    remove(session_id)


if __name__ == "__main__":
    main()