Byted Kickart Video Subtitler

                                 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.

# MIT License
#
# Copyright (c) 2026 ByteDance
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in all
# copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.

import os
import time
import logging
from functools import cache

from pydantic import BaseModel


class Result(BaseModel):
    code: str
    message: str
    data: object = None


class MediaConfig:
    """媒体相关配置与常量"""

    IMAGE_EXTENSIONS = {".jpg", ".jpeg", ".png"}
    VIDEO_EXTENSIONS = {".mp4", ".avi", ".mov"}

    IMAGE_MAX_SIZE = 8 * 1024 * 1024
    VIDEO_MAX_SIZE = 50 * 1024 * 1024

    IMAGE_MIN_WIDTH = 300
    IMAGE_MIN_HEIGHT = 300
    IMAGE_MAX_PIXELS = 36_000_000

    CSV_COLUMNS = ["group", "channel", "account", "path", "id", "material", "timestamp"]
    STORAGE_BASE_DIR = "/tmp/openclaw/byted-kickart-viral-replicator/media"


@cache
def init():
    """只执行一次的初始化方法，用于配置日志和目录"""
    log_dir = "/tmp/openclaw/byted-kickart-viral-replicator/logs"
    os.makedirs(log_dir, exist_ok=True)

    logging.basicConfig(
        level=logging.INFO,
        filename=f"{log_dir}/info.{time.strftime('%Y%m%d', time.localtime())}.log",
        format="%(asctime)s - %(levelname)s - %(message)s",
        datefmt="%Y-%m-%d %H:%M:%S",
    )


init()

__all__ = ["Result"]

# MIT License
#
# Copyright (c) 2026 ByteDance
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in all
# copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.

from abc import ABC, abstractmethod
import sys
import os
import time
import logging
import requests
from collections import defaultdict
from urllib.parse import urlencode, urlparse

# 动态加载项目根目录，以便于引入 core
sys.path.append(
    os.path.dirname(
        os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
    )
)
from core.utils.hash import HashUtils
from core.auth.strategy import AuthType, AuthStrategy


class IccpClient(ABC):
    @abstractmethod
    def do_request(self, method: str, queries: dict, body: bytes, action: str) -> dict:
        pass


class V1IccpClient(IccpClient):
    """基于 AK/SK 的请求客户端 (Strategy 实现)"""

    ADDR = "https://icp.volcengineapi.com"
    SERVICE = "iccloud_muse"
    REGION = "cn-north"
    VERSION = "2025-11-25"

    def __init__(self):
        self.ak = os.getenv("ACCESS_KEY_ID") or ""
        self.sk = os.getenv("SECRET_ACCESS_KEY") or ""

    def _get_signed_key(
        self, secret_key: str, date: str, region: str, service: str
    ) -> bytes:
        k_date = HashUtils.hmac_sha256(secret_key.encode("utf-8"), date)
        k_region = HashUtils.hmac_sha256(k_date, region)
        k_service = HashUtils.hmac_sha256(k_region, service)
        return HashUtils.hmac_sha256(k_service, "request")

    def do_request(self, method: str, queries: dict, body: bytes, action: str) -> dict:
        queries["Action"] = action
        queries["Version"] = self.VERSION

        query_string = urlencode(queries).replace("+", "%20")
        url = f"{self.ADDR}?{query_string}"

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

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

        header_list = [
            f"host:{host}",
            f"x-date:{date}",
            f"x-content-sha256:{payload}",
            "content-type:application/json",
        ]
        header_string = "\n".join(header_list)

        canonical_string = "\n".join(
            [
                method.upper(),
                "/",
                query_string,
                f"{header_string}\n",
                ";".join(signed_headers),
                payload,
            ]
        )
        hashed_canonical_string = HashUtils.hash_sha256(
            canonical_string.encode("utf-8")
        ).hex()

        credential_scope = f"{auth_date}/{self.REGION}/{self.SERVICE}/request"
        sign_string = "\n".join(
            ["HMAC-SHA256", date, credential_scope, hashed_canonical_string]
        )

        signed_key = self._get_signed_key(self.sk, auth_date, self.REGION, self.SERVICE)
        signature = HashUtils.hmac_sha256(signed_key, sign_string).hex()

        authorization = (
            f"HMAC-SHA256 Credential={self.ak}/{credential_scope},"
            f" SignedHeaders={';'.join(signed_headers)},"
            f" Signature={signature}"
        )

        headers = defaultdict(str)
        headers["X-Date"] = date
        headers["X-Content-Sha256"] = payload
        headers["Content-Type"] = "application/json"
        headers["Authorization"] = authorization

        if ppe_env := os.getenv("X_VOLC_ENV"):
            headers.update(
                {"X-TT-Env": "ppe_volcengine", "X-Volc-Env": ppe_env, "X-Use-Ppe": "1"}
            )

        logging.info(f">>> {method.upper()} {url} {headers} {body}")
        response = requests.request(
            method=method.upper(), url=url, headers=headers, data=body, timeout=30
        )
        logging.info(f"<<< {response.headers} {response.text}")
        return response.json()


class V2IccpClient(IccpClient):
    """基于 Ark Token 的请求客户端 (Strategy 实现)"""

    SERVICE = "iccloud_muse"
    REGION = "cn-north"
    VERSION = "2025-11-25"

    def __init__(self):
        self.addr = os.getenv("ARK_SKILL_API_BASE")
        self.token = os.getenv("ARK_SKILL_API_KEY") or ""

    def do_request(self, method: str, queries: dict, body: bytes, action: str) -> dict:
        queries["Action"] = action
        queries["Version"] = V2IccpClient.VERSION

        query_string = urlencode(queries).replace("+", "%20")
        url = f"{self.addr}?{query_string}"

        headers = defaultdict(str)
        headers["Authorization"] = f"Bearer {self.token}"
        headers["Content-Type"] = "application/json"
        headers["ServiceName"] = V2IccpClient.SERVICE

        if ppe_env := os.getenv("X_VOLC_ENV"):
            headers.update(
                {"X-TT-Env": "ppe_volcengine", "X-Volc-Env": ppe_env, "X-Use-Ppe": "1"}
            )

        logging.info(f">>> {method.upper()} {url} {headers} {body}")
        response = requests.request(
            method=method.upper(), url=url, headers=headers, data=body, timeout=30
        )
        logging.info(f"<<< {response.headers} {response.text}")
        return response.json()


class IccpClientFactory:
    @staticmethod
    def create(strategy: AuthStrategy) -> IccpClient:
        if strategy.strategy == AuthType.API_KEY:
            return V2IccpClient()
        if strategy.strategy == AuthType.AK_SK:
            return V1IccpClient()
        raise ValueError(f"不支持的认证策略类型: {strategy.strategy}")

# MIT License
#
# Copyright (c) 2026 ByteDance
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in all
# copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.

import os
import sys
import json
import jsonpath

# 动态加载根目录以便正确导入
sys.path.append(
    os.path.dirname(
        os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
    )
)
from core import Result
from core.auth.strategy import AuthStrategyFactory
from core.api.iccp.client import IccpClientFactory

# ─── 业务服务层 (Service Layer) ───────────────────────────────────


class IccpService:
    def __init__(self):
        strategy = AuthStrategyFactory.create()
        self.client = IccpClientFactory.create(strategy)

    def submit(self, service_id: int, params: str) -> Result:
        try:
            payload = {
                "ResourceList": [
                    "https://lf3-static.bytednsdoc.com/obj/eden-cn/jhteh7uhpxnult/test_image/woman/woman_4.png"
                ],
                "TemplateId": str(service_id),
                "Resolution": "1080p",
                "Extra": params,
            }

            submit_body = {
                "ServerId": service_id,
                "PayloadJson": json.dumps(payload, ensure_ascii=False),
            }
            submit_bytes = json.dumps(submit_body, ensure_ascii=False).encode("utf-8")

            response = self.client.do_request(
                "POST", {}, submit_bytes, action="SubmitAiTemplateTaskAsync"
            )

            code = jsonpath.jsonpath(response, "$.ResponseMetadata.Code")
            if not code:
                return Result(code="-1", message="提交任务失败, 响应内容为空")
            if code[0] != 0:
                return Result(
                    code=str(code[0]), message=f"提交任务失败, Code: {code[0]}"
                )

            task_id = jsonpath.jsonpath(response, "$.Result.TaskId")
            if not task_id or not task_id[0]:
                return Result(
                    code="-1", message=f"解析TaskId失败, 响应内容: {response}"
                )
            return Result(code="0", message="success", data=task_id[0])
        except Exception as e:
            return Result(code="-1", message=f"提交任务失败, 错误信息: {str(e)}")

    def query(self, task_id: str) -> Result:
        params = json.dumps({"TaskId": task_id}, ensure_ascii=False).encode("utf-8")
        try:
            resp = self.client.do_request(
                "POST", {}, params, action="QueryAiTemplateTaskResult"
            )

            code = jsonpath.jsonpath(resp, "$.ResponseMetadata.Code")
            if not code:
                return Result(code="-1", message="提交任务失败, 响应内容为空")
            if code[0] != 0:
                return Result(
                    code=str(code[0]), message=f"查询任务状态失败, Code: {code[0]}"
                )

            result_code = jsonpath.jsonpath(resp, "$.Result.Code")
            if not result_code:
                return Result(code="-1", message="提交任务失败, 响应内容为空")

            if result_code[0] in [1000, 1600]:
                return Result(code="1000", message="任务正在执行中")
            if result_code[0] != 0:
                msg = jsonpath.jsonpath(resp, "$.Result.Message")
                return Result(
                    code=str(result_code[0]), message=msg[0] if msg else "任务异常"
                )

            progress = jsonpath.jsonpath(resp, "$.Result.Progress")
            if not progress or progress[0] != 100:
                return Result(code="1000", message="任务正在执行中")

            result = jsonpath.jsonpath(resp, "$.Result.ResultExtra")
            if not result or not result[0]:
                return Result(code="-1", message="未获取到任务结果")

            return Result(code="0", message="success", data=result[0])
        except Exception as e:
            return Result(code="-1", message=f"查询任务状态失败: {str(e)}")

    def post(self, action: str, params: bytes) -> Result:
        try:
            resp = self.client.do_request("POST", {}, params, action=action)

            open_top_code = jsonpath.jsonpath(resp, "$.ResponseMetadata.Error.CodeN")
            if open_top_code and open_top_code[0] != 0:
                return Result(code=str(open_top_code[0]), message="")

            code = jsonpath.jsonpath(resp, "$.ResponseMetadata.Code")
            if code and code[0] != 0:
                return Result(code=str(code[0]), message="")

            if code and code[0] == 0:
                result = jsonpath.jsonpath(resp, "$.Result")
                if not result or not result[0]:
                    return Result(code="-1", message="接口返回值解析错误")
                expire = jsonpath.jsonpath(resp, "$.Result.expire_time")
                return Result(code="0", message=str(expire and expire[0]))

            return Result(code="-1", message="接口返回值解析错误")
        except Exception as e:
            return Result(code="-1", message=str(e))

# MIT License
#
# Copyright (c) 2026 ByteDance
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in all
# copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import json
import logging
import os
import sys
import time
import collections
from abc import ABC, abstractmethod
from typing import Dict, List, TypedDict
from urllib.parse import urlencode, urlparse

import jsonpath
import requests

# 动态加载项目根目录，以便于引入 utils
sys.path.append(
    os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
)
from utils.hash import HashUtils
from utils.matriel import Matriel, ImageMatriel, VideoMatriel
from auth.strategy import AuthType, AuthStrategy


# ─── 类型定义 ──────────────────────────────────────────


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


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


# ─── 配置管理 ──────────────────────────────────────────


class AppConfig:
    """全局配置管理"""

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

    SERVICE_MUSE = "iccloud_muse"
    SERVICE_IAM = "ic_iam"

    POLL_MAX_ATTEMPTS = 60
    POLL_INTERVAL = 5

    IMAGE_EXTENSIONS = {"jpg", "jpeg", "png", "gif", "bmp", "webp", "tiff", "tif"}
    VIDEO_EXTENSIONS = {"mp4", "avi", "mov", "wmv", "flv", "mkv", "webm", "m4v", "3gp"}


# ─── API 客户端 ────────────────────────────────────────


class ApiClient(ABC):
    """处理与后端的 HTTP 交互"""

    def __init__(self, host: str):
        self.host = host

    def _check_resp(self, resp: dict, action: str):
        meta = resp.get("ResponseMetadata", {})
        error_obj = meta.get("Error")
        if error_obj:
            code = error_obj.get("Code") or error_obj.get("CodeN")
            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"):
            msg = meta.get("Message") or ""
            print(f"❌ {action} 失败: code={code}, msg={msg}")
            sys.exit(1)

    def request(
        self, action: str, service: str, body: dict = None, extra_query: dict = None
    ) -> dict:  # type: ignore
        extra_query = extra_query or {}
        body_bytes = json.dumps(body or {}, ensure_ascii=False).encode()
        payload_hash = HashUtils.hash_sha256(body_bytes).hex()

        url = self.build_url(self.host, action, extra_query)
        query_string = urlparse(url).query
        headers = self.build_headers(
            service, self.host, query_string, payload_hash, is_binary=False
        )

        logging.info(
            f"[http] <<< {headers} {json.dumps(body or {}, ensure_ascii=False)}"
        )
        resp = requests.post(url, data=body_bytes, headers=headers, timeout=30)
        logging.info(f"[http] <<< {resp.headers} {resp.text}")

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

        self._check_resp(result, action)
        return result

    def request_binary(
        self, action: str, service: str, extra_query: dict, data: bytes
    ) -> dict:
        payload_hash = HashUtils.hash_sha256(data).hex()
        url = self.build_url(self.host, action, extra_query)
        query_string = urlparse(url).query
        headers = self.build_headers(
            service, self.host, query_string, payload_hash, is_binary=True
        )

        resp = requests.post(url, data=data, headers=headers, timeout=60)
        try:
            result = resp.json()
        except Exception:
            print(f"json parse error, resp is {resp.text}")
            sys.exit(1)

        self._check_resp(result, action)
        return result

    @abstractmethod
    def build_headers(
        self,
        service: str,
        host: str,
        query_string: str,
        payload_hash: str,
        is_binary: bool,
    ) -> Dict[str, str]:
        pass

    @abstractmethod
    def build_url(self, host: str, action: str, extra_query: dict) -> str:
        pass


class ArkClawApiClient(ApiClient):
    def __init__(self):
        super().__init__(os.getenv("ARK_SKILL_API_BASE", ""))
        self.token = os.getenv("ARK_SKILL_API_KEY", "")

    def build_headers(
        self,
        service: str,
        host: str,
        query_string: str,
        payload_hash: str,
        is_binary: bool,
    ) -> Dict[str, str]:
        headers = collections.defaultdict(str)
        headers["ServiceName"] = service
        headers["Authorization"] = f"Bearer {self.token}"
        headers["Content-Type"] = (
            "application/octet-stream" if is_binary else "application/json"
        )

        if ppe_env := os.getenv("X_VOLC_ENV"):
            headers.update(
                {"X-TT-Env": "ppe_volcengine", "X-Volc-Env": ppe_env, "X-Use-Ppe": "1"}
            )
        return headers

    def build_url(self, host: str, action: str, extra_query: dict) -> str:
        url = f"{host}/?Action={action}&Version={AppConfig.VERSION}"
        if extra_query:
            url += "&" + urlencode(extra_query)
        return url


class AkSkApiClient(ApiClient):
    def __init__(self):
        super().__init__("https://icp.volcengineapi.com")
        self.ak = os.getenv("ACCESS_KEY_ID", "")
        self.sk = os.getenv("SECRET_ACCESS_KEY", "")

    def build_headers(
        self,
        service: str,
        host: str,
        query_string: str,
        payload_hash: str,
        is_binary: bool,
    ) -> Dict[str, str]:
        date = time.strftime("%Y%m%dT%H%M%SZ", time.gmtime(time.time()))
        auth_date = date[:8]

        content_type = "application/octet-stream" if is_binary else "application/json"
        signed_headers = ["host", "x-date", "x-content-sha256", "content-type"]

        parsed_url = urlparse(host)
        host_name = parsed_url.netloc

        header_list = [
            f"host:{host_name}",
            f"x-date:{date}",
            f"x-content-sha256:{payload_hash}",
            f"content-type:{content_type}",
        ]
        header_string = "\n".join(header_list)

        canonical_string = "\n".join(
            [
                "POST",
                "/",
                query_string,
                f"{header_string}\n",
                ";".join(signed_headers),
                payload_hash,
            ]
        )
        hashed_canonical_string = HashUtils.hash_sha256(
            canonical_string.encode("utf-8")
        ).hex()

        credential_scope = f"{auth_date}/{AppConfig.REGION}/{service}/request"
        sign_string = "\n".join(
            ["HMAC-SHA256", date, credential_scope, hashed_canonical_string]
        )

        k_date = HashUtils.hmac_sha256(self.sk.encode("utf-8"), auth_date)
        k_region = HashUtils.hmac_sha256(k_date, AppConfig.REGION)
        k_service = HashUtils.hmac_sha256(k_region, service)
        signed_key = HashUtils.hmac_sha256(k_service, "request")

        signature = HashUtils.hmac_sha256(signed_key, sign_string).hex()
        authorization = (
            f"HMAC-SHA256 Credential={self.ak}/{credential_scope},"
            f" SignedHeaders={';'.join(signed_headers)},"
            f" Signature={signature}"
        )

        headers = collections.defaultdict(str)
        headers["X-Date"] = date
        headers["X-Content-Sha256"] = payload_hash
        headers["Content-Type"] = content_type
        headers["Authorization"] = authorization
        if ppe_env := os.getenv("X_VOLC_ENV"):
            headers.update(
                {"X-TT-Env": "ppe_volcengine", "X-Volc-Env": ppe_env, "X-Use-Ppe": "1"}
            )
        return headers

    def build_url(self, host: str, action: str, extra_query: dict) -> str:
        queries = extra_query.copy()
        queries["Action"] = action
        queries["Version"] = AppConfig.VERSION
        query_string = urlencode(sorted(queries.items())).replace("+", "%20")
        return f"{host}?{query_string}"


class ApiClientFactory:
    @staticmethod
    def create(strategy: AuthStrategy) -> ApiClient:
        if strategy.strategy == AuthType.API_KEY:
            return ArkClawApiClient()
        if strategy.strategy == AuthType.AK_SK:
            return AkSkApiClient()
        raise ValueError(f"不支持的认证策略类型: {strategy.strategy}")


# ─── 业务服务层 ────────────────────────────────────────


class IamService:
    def __init__(self, client: ApiClient):
        self.client = client

    def get_admin_user_id(self) -> int:
        result = self.client.request(
            action="ListUsers", service=AppConfig.SERVICE_IAM, body={"UserType": "All"}
        )
        users = result.get("Result", {}).get("Users", [])
        if not users:
            print("❌ 未获取到任何用户信息")
            sys.exit(1)

        for user in users:
            if user.get("IsAdmin") and user.get("Id"):
                return user.get("Id")
        return users[0].get("Id")


class MuseService:
    def __init__(self, client: ApiClient):
        self.client = client

    def get_upload_state(
        self, file_md5: str, file_size: int, file_crc32: int, owner_id: int
    ) -> UploadStateResult:
        body = {
            "Owner": {"Id": owner_id, "Type": "PERSON"},
            "Md5": file_md5,
            "Size": file_size,
            "Start": 0,
            "End": file_size - 1,
            "Crc": file_crc32,
        }
        result = self.client.request(
            action="GetUploadState", service=AppConfig.SERVICE_MUSE, body=body
        )
        raw_state = result.get("Result", {})
        return {
            "SkipDataComplete": bool(raw_state.get("SkipDataComplete", False)),
            "PartSize": int(raw_state.get("PartSize", 0)),
            "Ranges": raw_state.get("Ranges", []),
        }

    def upload_part(
        self, owner_id: int, chunk: bytes, offset: int, part_size: int, chunk_md5: str
    ) -> dict:
        query = {
            "Md5": chunk_md5,
            "Size": part_size,
            "Offset": offset,
            "OwnerId": owner_id,
            "OwnerType": "PERSON",
        }
        return self.client.request_binary(
            "StreamUploadData", AppConfig.SERVICE_MUSE, query, chunk
        )

    def create_material(
        self,
        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:
        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,
            },
        }
        result = self.client.request(
            action="CreateMaterial", service=AppConfig.SERVICE_MUSE, body=body
        )
        return result.get("Result", {}).get("MediaId")

    def poll_media_info(self, media_id: str, owner_id: int, owner_type: str) -> dict:
        for _ in range(AppConfig.POLL_MAX_ATTEMPTS):
            result = self.client.request(
                action="GetMediaInfo",
                service=AppConfig.SERVICE_MUSE,
                body={"MediaIds": [media_id], "MediaType": 1},
            )
            media_infos = result.get("Result", {}).get("MediaInfos", [])
            if media_infos:
                media_info = media_infos[0]
                status = media_info.get("BasicInfo", {}).get("MediaStatus")
                if status >= 2:
                    return media_info
                if status in (1, 5):
                    print("❌ 处理失败")
                    sys.exit(1)
            time.sleep(AppConfig.POLL_INTERVAL)
        sys.exit(1)


class KickartMuseService:
    def __init__(self, client: ApiClient):
        self.client = client

    def poll_media_info(self, media_id: str, owner_id: int, owner_type: str) -> dict:
        for _ in range(AppConfig.POLL_MAX_ATTEMPTS):
            time.sleep(AppConfig.POLL_INTERVAL)
            result = self.client.request(
                action="GetMediaInfo",
                service=AppConfig.SERVICE_MUSE,
                body={"MediaIds": [media_id], "MediaType": 3},
            )
            media_infos = result.get("Result", {}).get("MediaInfos", [])
            if not media_infos:
                continue

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

            if status == 4:
                return media_info

            elif status == 5 or status == 1:
                print("❌ 处理失败")
                sys.exit(1)

        sys.exit(1)


# ─── 编排与格式化层 ────────────────────────────────────


class MaterialUploader:
    def __init__(self, client: ApiClient):
        self.iam = IamService(client)
        self.muse = MuseService(client)

    def stream_upload(
        self,
        file_path: str,
        file_md5: str,
        file_size: int,
        file_crc32: int,
        owner_id: int,
        state: UploadStateResult,
    ) -> UploadStateResult:
        if state["SkipDataComplete"]:
            return state

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

        offset = 0
        for _ in range(1000):
            if state["SkipDataComplete"]:
                break
            part_size = state.get("PartSize", 0)

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

            self.muse.upload_part(owner_id, chunk, offset, chunk_size, file_md5)
            offset += chunk_size
            state = self.muse.get_upload_state(
                file_md5, file_size, file_crc32, owner_id
            )

            if state["SkipDataComplete"] or not state["Ranges"] or offset >= file_size:
                return state
        return state


class MediaFormatter:
    @staticmethod
    def extract_url(media_info: dict) -> str:
        cat = media_info.get("BasicInfo", {}).get("MediaFirstCategory", "")
        if cat == "image":
            image_media = media_info.get("ImageMedia", {})
            if dl := image_media.get("DownloadUrl"):
                return dl
            for q in ["origin", "jpeg_1080p", "jpeg_480p"]:
                if url := image_media.get("TranscodeDownloadUrls", {}).get(q):
                    return url
        elif cat in ("video", "audio"):
            media = media_info.get("VideoMedia" if cat == "video" else "AudioMedia", {})
            if dl := media.get("DownloadUrl"):
                return dl
            if play := media.get("PlayInfo", []):
                return play[0].get("Url", "")
        return ""

    @staticmethod
    def simplify(media_info: dict) -> dict:
        # 保持原逻辑的 simplify
        cat = media_info.get("BasicInfo", {}).get("MediaFirstCategory", "")
        if cat == "image":
            im = media_info.get("ImageMedia", {})
            if dl := im.get("DownloadUrl"):
                im["DownloadUrl"] = dl
            for q in ["origin", "jpeg_1080p", "jpeg_480p"]:
                if url := im.get("TranscodeDownloadUrls", {}).get(q):
                    im["TranscodeDownloadUrls"][q] = url
        elif cat in ("video", "audio"):
            vm = media_info.get("VideoMedia" if cat == "video" else "AudioMedia", {})
            if dl := vm.get("DownloadUrl"):
                vm["DownloadUrl"] = dl
            if play := vm.get("PlayInfo", []):
                play[0]["Url"] = play[0].get("Url")
        return media_info

    @staticmethod
    def format(media_info: dict) -> Matriel:
        if jsonpath.jsonpath(media_info, "$.ImageMedia"):
            m = ImageMatriel(id="", type="image", url="", size=0, height=0, width=0)
            if v := jsonpath.jsonpath(media_info, "$.BasicInfo.MediaId"):
                m.id = v[0]
            if v := jsonpath.jsonpath(media_info, "$.ImageMedia.DownloadUrl"):
                m.url = v[0]
            if v := jsonpath.jsonpath(media_info, "$.ImageMedia.Width"):
                m.width = v[0]
            if v := jsonpath.jsonpath(media_info, "$.ImageMedia.Height"):
                m.height = v[0]
            return m
        elif jsonpath.jsonpath(media_info, "$.VideoMedia"):
            m = VideoMatriel(
                id="", type="video", url="", size=0, height=0, width=0, duration=0
            )
            if v := jsonpath.jsonpath(media_info, "$.BasicInfo.MediaId"):
                m.id = v[0]
            if v := jsonpath.jsonpath(media_info, "$.VideoMedia.DownloadUrl"):
                m.url = v[0]
            if v := jsonpath.jsonpath(media_info, "$.VideoMedia.MediaMetaInfo.Width"):
                m.width = v[0]
            if v := jsonpath.jsonpath(media_info, "$.VideoMedia.MediaMetaInfo.Height"):
                m.height = v[0]
            if v := jsonpath.jsonpath(
                media_info, "$.VideoMedia.MediaMetaInfo.Duration"
            ):
                m.duration = v[0] / 1000
            return m
        return Matriel(id="", type="", url="", size=0, height=0, width=0)

# MIT License
#
# Copyright (c) 2026 ByteDance
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in all
# copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.

import json
import os
import sys
import time

from abc import ABC, abstractmethod
from typing import Dict, Any, List
from pathlib import Path
import pandas as pd

# 动态加载项目根目录，以便于引入 core.Result
sys.path.append(
    os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
)
from auth.strategy import AuthStrategyFactory
from utils.hash import HashUtils
from utils.validator import Validator
from utils.extractor import MetadataExtractor
from api.meida.chunks import (
    AppConfig,
    ApiClientFactory,
    MaterialUploader,
    MediaFormatter,
    IamService,
    KickartMuseService,
)
from core import MediaConfig


class RemoteUploader(ABC):
    """远程上传器接口 (策略模式)"""

    @abstractmethod
    def upload(self, file: str) -> Any:
        pass


class MuseRemoteUploader(RemoteUploader):
    """基于 Muse 的远程上传器具体实现"""

    def __init__(self):  # type: ignore
        strategy = AuthStrategyFactory.create()
        client = ApiClientFactory.create(strategy)
        self.uploader = MaterialUploader(client)

    def upload(self, file: str) -> Any:
        owner_id = self.uploader.iam.get_admin_user_id()
        file_md5, file_crc32, file_size = HashUtils.file_hash(file)

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

        cat = "image" if file_ext.lower() in AppConfig.IMAGE_EXTENSIONS else "video"
        title = f"artclaw-material-{int(time.time())}"
        owner_type = "user"

        state = self.uploader.muse.get_upload_state(
            file_md5, file_size, file_crc32, owner_id
        )
        state = self.uploader.stream_upload(
            file, file_md5, file_size, file_crc32, owner_id, state
        )

        media_id = self.uploader.muse.create_material(
            file_md5,
            file_size,
            file_name,
            file_ext,
            state["SkipDataComplete"],
            owner_id,
            owner_type,
            title,
            cat,
        )

        media_info = self.uploader.muse.poll_media_info(media_id, owner_id, owner_type)
        return MediaFormatter.format(MediaFormatter.simplify(media_info))


class KickartUploader(RemoteUploader):
    def __init__(self, source: str):
        self.strategy = AuthStrategyFactory.create()
        self.client = ApiClientFactory.create(self.strategy)
        self.iam = IamService(self.client)
        self.muse = KickartMuseService(self.client)
        self.source = source

    def upload(self, file: str) -> Any:
        """通过文件 上传媒资"""

        owner_id = self.iam.get_admin_user_id()
        owner_type = "user"
        title = f"artclaw-material-{int(time.time())}"

        body = {
            "Owner": {"Id": owner_id, "Type": "PERSON"},
            "CreateUrlFilmInfo": {
                "Title": title,
                "SourceFrom": self.source,
                "MediaFirstCategory": "video",
                "MaterialUrl": file,
                "Description": title,
            },
        }

        result = self.client.request(
            action="CreateUrlFilm", service=AppConfig.SERVICE_MUSE, body=body
        )
        media_id = result.get("Result", {}).get("MediaId")
        if not media_id:
            raise ValueError("CreateUrlFilm 未返回 MediaId")

        media_info = self.muse.poll_media_info(media_id, owner_id, owner_type)
        return MediaFormatter.format(MediaFormatter.simplify(media_info))


class MediaRepository:
    """仓储层：处理底层 CSV 数据的读写"""

    def __init__(self, base_dir: str = MediaConfig.STORAGE_BASE_DIR):
        self.base_dir = Path(base_dir)

    def _get_path(self, group: str) -> Path:
        return self.base_dir / f"{group}.csv"

    def load(self, group: str) -> pd.DataFrame:
        path = self._get_path(group)
        if not path.exists():
            return pd.DataFrame(columns=MediaConfig.CSV_COLUMNS)
        return pd.read_csv(path, header=None, names=MediaConfig.CSV_COLUMNS)

    def save(self, group: str, df: pd.DataFrame):
        path = self._get_path(group)
        os.makedirs(path.parent, exist_ok=True)
        df.to_csv(path, index=False, header=False)

    def clear(self, group: str):
        path = self._get_path(group)
        if path.exists():
            os.remove(path)


class SimpleMediaRepository:
    """媒体缓存仓储层：处理底层 JSON 文件的读写（仿照 MediaRepository 设计）"""

    def __init__(self, base_dir: str = None):  # type: ignore
        self.base_dir = Path(base_dir or MediaConfig.STORAGE_BASE_DIR)
        self.base_dir.mkdir(parents=True, exist_ok=True)

    def _get_path(self, media_id: str) -> Path:
        """获取媒体缓存文件路径"""
        return self.base_dir / f"{media_id}.json"

    def load(self, media_id: str) -> dict | None:
        """加载指定媒体ID的缓存数据"""
        path = self._get_path(media_id)
        if not path.exists():
            raise FileNotFoundError(f"文件不存在: {path}")
        with open(path, "r", encoding="utf-8") as f:
            return json.load(f)

    def save(self, media_id: str, data: dict):
        """保存媒体数据到缓存"""
        path = self._get_path(media_id)
        os.makedirs(path.parent, exist_ok=True)
        with open(path, "w", encoding="utf-8") as f:
            json.dump(data, f, ensure_ascii=False, indent=2)

    def clear(self, media_id: str):
        """清除指定媒体ID的缓存"""
        path = self._get_path(media_id)
        if path.exists():
            os.remove(path)

    def clear_all(self):
        """清除所有缓存"""
        for file in self.base_dir.glob("*.json"):
            file.unlink()


class MediaService:
    """业务服务层：协调校验、提取、上传与存储 (依赖注入)"""

    def __init__(
        self, repository: MediaRepository = None, uploader: RemoteUploader = None
    ):  # type: ignore
        self.repository = repository or MediaRepository()
        self.uploader = uploader or MuseRemoteUploader()

    def add_media(
        self,
        file: str,
        group: str,
        metadata: dict,
        extractor: MetadataExtractor,
        validator: Validator,
    ) -> Any:  # type: ignore
        metadata_result = extractor.extract(file)
        validation_result = validator.validate(metadata_result)

        if not validation_result.get("valid", False):
            return validation_result

        # 上传文件到远程服务器
        matriel = self.uploader.upload(file)

        # 补全缺失的媒体信息
        if hasattr(matriel, "type") and matriel.type == "":
            matriel.type = validation_result.get("file_type", "")

        # 构造存储记录
        row = {
            "group": group,
            "channel": metadata.get("channel", ""),
            "account": metadata.get("chat_id", ""),
            "path": file,
            "id": matriel.id,
            "material": matriel.model_dump_json(),
            "timestamp": str(time.time()),
        }

        # 持久化到仓储
        df = self.repository.load(group)
        df.loc[len(df)] = row
        self.repository.save(group, df)

        return matriel

    def list_media(self, group: str) -> List[Dict]:
        df = self.repository.load(group)
        if df.empty:
            return []
        return df["material"].map(lambda x: json.loads(x)).to_list()  # type: ignore

    def remove_media(self, media_id: str, group: str):
        df = self.repository.load(group)
        df.drop(df[df["id"].eq(media_id)].index, inplace=True)
        self.repository.save(group, df)

    def clear_media(self, group: str):
        self.repository.clear(group)


class SimpleMediaService:
    """简化版媒体服务：仅负责文件上传，不需要group参数，支持本地缓存"""

    def __init__(
        self, repository: SimpleMediaRepository = None, uploader: RemoteUploader = None
    ):  # type: ignore
        self.repository = repository or SimpleMediaRepository()
        self.uploader = uploader or MuseRemoteUploader()

    def add_media(
        self, file: str, extractor: MetadataExtractor, validator: Validator
    ) -> Any:  # type: ignore
        """
        上传单个文件到远程服务器，并将信息存储到本地缓存

        Args:
            file: 本地文件绝对路径
            extractor: 元数据提取器（可选）
            validator: 校验器（可选）

        Returns:
            Matriel 对象，包含上传后的媒体信息（id、url等）
        """
        metadata_result = extractor.extract(file)
        validation_result = validator.validate(metadata_result)

        if not validation_result.get("valid", False):
            return validation_result

        # 上传文件到远程服务器
        matriel = self.uploader.upload(file)

        # 补全缺失的媒体信息
        matriel.width = getattr(metadata_result, "width", 0)
        matriel.height = getattr(metadata_result, "height", 0)
        if hasattr(metadata_result, "duration"):
            matriel.duration = getattr(metadata_result, "duration", 0)
        if hasattr(metadata_result, "closest_ratio"):
            matriel.closest_ratio = getattr(metadata_result, "closest_ratio", "")
        matriel.size = getattr(metadata_result, "size", 0)

        # 将媒体信息保存到本地缓存
        media_info = {
            "id": matriel.id,
            "url": matriel.url,
            "type": matriel.type,
            "width": matriel.width,
            "height": matriel.height,
            "size": matriel.size,
            "timestamp": time.time(),
        }
        if hasattr(matriel, "duration"):
            media_info["duration"] = matriel.duration
        if hasattr(matriel, "closest_ratio"):
            media_info["closest_ratio"] = matriel.closest_ratio
        self.repository.save(matriel.id, media_info)
        return matriel

    def get_media(self, media_id: str) -> dict:
        """
        通过媒资ID获取媒体详细信息（优先从本地缓存读取）

        Args:
            media_id: 媒资ID

        Returns:
            媒体信息字典
        """
        return self.repository.load(media_id)  # type: ignore

# MIT License
#
# Copyright (c) 2026 ByteDance
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in all
# copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.

from .strategy import (
    AuthStrategy,
    AkSkAuthStrategy,
    ApiKeyAuthStrategy,
    AuthStrategyFactory,
    AuthType,
)

__all__ = [
    "AuthStrategy",
    "AkSkAuthStrategy",
    "ApiKeyAuthStrategy",
    "AuthStrategyFactory",
    "AuthType",
]

# MIT License
#
# Copyright (c) 2026 ByteDance
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in all
# copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.

import os
from abc import ABC, abstractmethod
from functools import cache
from enum import Enum


class AuthType(Enum):
    AK_SK = "ak_sk"
    API_KEY = "api_key"


class AuthStrategy(ABC):
    """鉴权策略接口 (Strategy Pattern)"""

    @property
    @abstractmethod
    def strategy(self) -> AuthType:
        """获取当前使用的鉴权策略类型"""
        pass


class AkSkAuthStrategy(AuthStrategy):
    """AK/SK 鉴权策略"""

    @property
    def strategy(self) -> AuthType:
        return AuthType.AK_SK

    def __init__(self):
        self.ak = os.getenv("ACCESS_KEY_ID")
        self.sk = os.getenv("SECRET_ACCESS_KEY")

        if not self.ak or not self.sk:
            raise ValueError(
                "AK/SK未提供，且环境变量中未找到 ACCESS_KEY_ID/SECRET_ACCESS_KEY"
            )


class ApiKeyAuthStrategy(AuthStrategy):
    """API Key 鉴权策略"""

    @property
    def strategy(self) -> AuthType:
        return AuthType.API_KEY

    def __init__(self):
        self.api_key = os.getenv("ARK_SKILL_API_KEY")
        self.base_url = os.getenv("ARK_SKILL_API_BASE")

        if not self.api_key or not self.base_url:
            raise ValueError(
                "API Key/Base URL 未提供，且环境变量中未找到 ARK_SKILL_API_KEY/ARK_SKILL_API_BASE"
            )


class AuthStrategyFactory:
    """鉴权策略工厂 (Factory Pattern)"""

    @staticmethod
    @cache
    def create() -> AuthStrategy:
        if os.getenv("ARK_SKILL_API_BASE") and os.getenv("ARK_SKILL_API_KEY"):
            return ApiKeyAuthStrategy()
        if os.getenv("ACCESS_KEY_ID") and os.getenv("SECRET_ACCESS_KEY"):
            return AkSkAuthStrategy()
        raise Exception("鉴权凭证未配置(缺少 AK/SK 或 Token)")

import os
import asyncio
import aiohttp
import logging
from abc import ABC, abstractmethod
from typing import List, Callable, Optional


class FilenameGenerator(ABC):
    """
    文件名生成策略接口（策略模式）
    """

    @abstractmethod
    def generate(self, url: str) -> str:
        """
        根据URL生成文件名

        Args:
            url: 文件URL

        Returns:
            生成的文件名
        """
        pass

    def modify(self, file_path: str) -> str:
        """
        修改文件路径（模板方法）

        默认直接返回原始文件路径，子类可以重写此方法实现文件修改逻辑

        Args:
            file_path: 原始文件路径

        Returns:
            修改后的文件路径
        """
        return file_path


class DefaultFilenameGenerator(FilenameGenerator):
    """
    默认文件名生成策略：从URL提取文件名
    """

    def generate(self, url: str) -> str:
        filename = os.path.basename(url).split("?")[0]
        if not filename or filename == ".":
            filename = f"download_{hash(url) % 10000}.tmp"
        return filename


class MagicFilenameGenerator(FilenameGenerator):
    """
    魔法文件名生成策略：根据文件内容生成文件名
    """

    # MIME类型到文件扩展名的映射
    MIME_TYPE_MAP = {
        "image/jpeg": ".jpg",
        "image/jpg": ".jpg",
        "image/png": ".png",
        "image/gif": ".gif",
        "image/webp": ".webp",
        "image/bmp": ".bmp",
        "image/tiff": ".tiff",
        "image/svg+xml": ".svg",
        "video/mp4": ".mp4",
        "video/mov": ".mov",
        "video/avi": ".avi",
        "video/mkv": ".mkv",
        "video/flv": ".flv",
        "video/webm": ".webm",
        "video/wmv": ".wmv",
        "video/mpeg": ".mpeg",
        "application/pdf": ".pdf",
        "application/json": ".json",
        "text/plain": ".txt",
        "application/octet-stream": ".bin",
    }

    def generate(self, url: str) -> str:
        filename = os.path.basename(url).split("?")[0]
        if not filename or filename == ".":
            filename = f"download_{hash(url) % 10000}"
        return filename

    def modify(self, file_path: str) -> str:
        """
        修改文件路径：使用magic库检测文件类型并添加正确的文件后缀

        Args:
            file_path: 原始文件路径

        Returns:
            添加正确后缀后的文件路径
        """
        import magic

        # 检查文件是否存在
        if not os.path.exists(file_path):
            return file_path

        # 使用magic库检测文件类型
        with open(file_path, "rb") as f:
            file_data = f.read(2048)
        mime_type = magic.from_buffer(file_data, mime=True)

        # 根据MIME类型获取扩展名
        ext = self.MIME_TYPE_MAP.get(mime_type, ".tmp")
        new_file_path = f"{file_path}{ext}"
        os.rename(file_path, new_file_path)
        return new_file_path


class SequentialFilenameGenerator(FilenameGenerator):
    """
    顺序文件名生成策略：按下载顺序生成文件名
    """

    def __init__(self, prefix: str = "file"):
        self.prefix = prefix
        self.counter = 0

    def generate(self, url: str) -> str:
        self.counter += 1
        ext = os.path.splitext(url.split("?")[0])[1] or ".tmp"
        return f"{self.prefix}_{self.counter}{ext}"


class DownloadResult:
    """
    下载结果数据类
    """

    def __init__(
        self,
        url: str,
        success: bool,
        file_path: Optional[str] = None,
        error: Optional[str] = None,
    ):
        self.url = url
        self.success = success
        self.file_path = file_path
        self.error = error


class BaseDownloader(ABC):
    """
    下载器抽象基类（模板方法模式）
    """

    def __init__(
        self, output: str, filename_generator: Optional[FilenameGenerator] = None
    ):
        """
        初始化下载器

        Args:
            output: 输出目录路径
            filename_generator: 文件名生成策略，默认为DefaultFilenameGenerator
        """
        self.output = output
        self.filename_generator = filename_generator or DefaultFilenameGenerator()
        self._ensure_output_dir()

    def _ensure_output_dir(self):
        """确保输出目录存在"""
        os.makedirs(self.output, exist_ok=True)

    @abstractmethod
    def download(self, urls: List[str]) -> int:
        """
        同步下载接口

        Args:
            urls: URL数组

        Returns:
            下载成功的文件数量
        """
        pass


class ParallelDownloader(BaseDownloader):
    """
    并行异步下载器（组合模式 + 策略模式）
    """

    def __init__(
        self,
        output: str,
        max_concurrent: int = 5,
        timeout: int = 30,
        filename_generator: Optional[FilenameGenerator] = None,
        on_progress: Optional[Callable[[int, int], None]] = None,
    ):
        """
        初始化并行下载器

        Args:
            output: 输出目录路径
            max_concurrent: 最大并发下载数，默认为5
            timeout: 单个请求超时时间（秒），默认为30
            filename_generator: 文件名生成策略
            on_progress: 进度回调函数 (current, total)
        """
        super().__init__(output, filename_generator)
        self.max_concurrent = max_concurrent
        self.timeout = timeout
        self.on_progress = on_progress
        self._downloaded_count = 0
        self._total_count = 0

    async def _download_single(
        self, session: aiohttp.ClientSession, url: str
    ) -> DownloadResult:
        """
        下载单个文件（私有方法）

        Args:
            session: aiohttp客户端会话
            url: 文件URL

        Returns:
            下载结果
        """
        try:
            timeout = aiohttp.ClientTimeout(total=self.timeout)
            async with session.get(url, timeout=timeout) as response:
                if response.status != 200:
                    return DownloadResult(
                        url=url, success=False, error=f"HTTP {response.status}"
                    )

                filename = self.filename_generator.generate(url)
                file_path = os.path.join(self.output, filename)

                with open(file_path, "wb") as f:
                    async for chunk in response.content.iter_chunked(8192):
                        f.write(chunk)
                file_path = self.filename_generator.modify(file_path)
                logging.info(f"下载成功: {url} -> {file_path}")
                return DownloadResult(url=url, success=True, file_path=file_path)
        except Exception as e:
            logging.error(f"下载失败 {url}: {str(e)}")
            return DownloadResult(url=url, success=False, error=str(e))

    async def _download_with_semaphore(
        self, session: aiohttp.ClientSession, url: str, semaphore: asyncio.Semaphore
    ) -> DownloadResult:
        """
        使用信号量控制并发的下载任务

        Args:
            session: aiohttp客户端会话
            url: 文件URL
            semaphore: 信号量对象

        Returns:
            下载结果
        """
        async with semaphore:
            result = await self._download_single(session, url)
            self._downloaded_count += 1
            if self.on_progress:
                self.on_progress(self._downloaded_count, self._total_count)
            return result

    async def _async_download(self, urls: List[str]) -> int:
        """
        异步下载核心方法

        Args:
            urls: URL数组

        Returns:
            下载成功的文件数量
        """
        if not urls:
            logging.warning("URL列表为空")
            return 0

        self._downloaded_count = 0
        self._total_count = len(urls)

        semaphore = asyncio.Semaphore(self.max_concurrent)

        async with aiohttp.ClientSession() as session:
            tasks = [
                self._download_with_semaphore(session, url, semaphore) for url in urls
            ]
            results = await asyncio.gather(*tasks)

        success_count = sum(1 for r in results if r.success)
        logging.info(f"下载完成: {success_count}/{len(urls)} 成功")
        return success_count

    def download(self, urls: List[str]) -> int:
        """
        同步下载接口（模板方法）

        Args:
            urls: URL数组

        Returns:
            下载成功的文件数量
        """
        return asyncio.run(self._async_download(urls))