Byted Kickart Video Analyzer

                                 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 click
import logging
import sys
import time
import json

import math

import jsonpath
from typing import Any
from core import Result
from core.api.iccp.service import IccpService
from core.api.meida.media import SimpleMediaService

def format_for_origin(data: dict) -> dict:
    """
    格式化数据为原始格式，仅保留指定字段
    
    保留字段：
    1. 视频链接、视频时长
    2. 商品名称、商品描述
    3. 分镜编号、开始/结束时间，镜头语言、镜头主体、营销意图、口播、旁白、BGM、花字&字幕
    
    Args:
        data: 输入的字典数据
        
    Returns:
        格式化后的字典数据
    """
    result = {}
    
    # 使用jsonpath提取字段，提取成功时返回列表，失败返回None
    def extract_field(path: str) -> Any:
        """使用jsonpath提取字段值"""
        value = jsonpath.jsonpath(data, path)
        return value[0] if (value and len(value) > 0) else None
    
    # 提取video_info，仅保留video_duration
    video_info = extract_field('$.video_info')
    video_info_copy = {}
    if video_info and isinstance(video_info, dict):
        video_info_copy['video_duration'] = video_info.get('video_duration', '')
    
    # 提取并清理shot_breakdown，只保留需要的字段
    shot_breakdown = extract_field('$.video_info.shot_breakdown')
    cleaned_shots = []
    if isinstance(shot_breakdown, list):
        for shot in shot_breakdown:
            if isinstance(shot, dict):
                cleaned_shot = {
                    'shot_number': shot.get('shot_number', ''),
                    'start_time': shot.get('start_time', ''),
                    'end_time': shot.get('end_time', ''),
                    'camera_language': shot.get('camera_language', ''),
                    'main_subject': shot.get('main_subject', ''),
                    'marketing_intent': shot.get('marketing_intent', ''),
                    'on_camera_speech': shot.get('on_camera_speech', ''),
                    'voiceover_text': shot.get('voiceover_text', ''),
                    'bgm': shot.get('bgm', ''),
                    'stickers': shot.get('stickers', ''),
                    'text_style': shot.get('text_style', '')
                }
                cleaned_shots.append(cleaned_shot)
    
    video_info_copy['shot_breakdown'] = cleaned_shots
    result['video_info'] = video_info_copy
    
    # 提取product_info，仅保留product_title和product_description
    product_info = extract_field('$.product_info')
    product_info_copy = {}
    if product_info and isinstance(product_info, dict):
        product_info_copy['product_title'] = product_info.get('product_title', '')
        product_info_copy['product_description'] = product_info.get('product_description', '')
    result['product_info'] = product_info_copy
    
    return result

def format_for_seedance(data: dict) -> dict:
    """
    格式化数据为Seedance格式，仅保留指定字段
    
    保留字段：
    1. 视频链接、视频时长
    2. 商品名称、商品描述
    3. 分镜编号、开始/结束时间，镜头语言、镜头主体、营销意图、口播、旁白、BGM、花字&字幕
    4. 全片级音频基调
    
    格式规范：
    - 台词: 需用 {} 包裹
    - 画外音、字幕语义: 可用 【】
    - 音乐: 需用 () 包裹
    - 音效: 需用 <> 包裹
    - 字幕: 需用 【】 包裹
    
    Args:
        data: 输入的字典数据
        
    Returns:
        格式化后的字典数据
    """
    result = {}
    
    # 使用jsonpath提取字段
    def extract_field(path: str) -> Any:
        """使用jsonpath提取字段值"""
        value = jsonpath.jsonpath(data, path)
        return value[0] if (value and len(value) > 0) else None
    
    # ===== 视频信息 =====
    result['video_duration'] = extract_field('$.video_info.video_duration') or ''
    
    # ===== 商品信息 =====
    product_info = extract_field('$.product_info')
    if product_info and isinstance(product_info, dict):
        result['product_title'] = product_info.get('product_title', '')
        result['product_description'] = product_info.get('product_description', '')
    
    # ===== 分镜信息 =====
    shot_breakdown = extract_field('$.video_info.shot_breakdown')
    shots = []
    
    if isinstance(shot_breakdown, list):
        for shot in shot_breakdown:
            if isinstance(shot, dict):
                # 按照Seedance格式规范格式化字段
                on_camera_speech = shot.get('on_camera_speech', '')
                voiceover_text = shot.get('voiceover_text', '')
                bgm = shot.get('bgm', '')
                stickers = shot.get('stickers', '')
                
                shots.append({
                    'shot_number': shot.get('shot_number', ''),
                    'start_time': shot.get('start_time', ''),
                    'end_time': shot.get('end_time', ''),
                    'camera_language': shot.get('camera_language', ''),
                    'main_subject': shot.get('main_subject', ''),
                    'marketing_intent': shot.get('marketing_intent', ''),
                    # 台词用 {} 包裹
                    'on_camera_speech': "{" + on_camera_speech + "}",
                    # 画外音用 【】 包裹
                    'voiceover_text': "【" + voiceover_text + "】",
                    # 音乐用 () 包裹
                    'bgm_music': "(" + bgm + ")",
                    # 音效用 <> 包裹
                    'bgm_sound_effect': "<" + bgm + ">",
                    # 字幕用 【】 包裹
                    'stickers': "【" + stickers + "】"
                })
    
    result['shots'] = shots
    
    # ===== 全片级音频基调 =====
    result['audio_tone'] = extract_field('$.video_info.audio_tone') or ''

    # ===== 音色参数 =====
    role_list = extract_field('$.scene_info.role_list')
    if role_list and isinstance(role_list, list):
        result['role_list'] = list(zip([role.get('id', '') for role in role_list], [role.get('vocal_attributes', '') for role in role_list]))
    
    return result

@click.command()
@click.option("--media-id", required=True, type=str, help="输入视频的媒资ID")
@click.option("--output", required=True, type=str, help="输出结果所在的json文件路径")
def main(media_id, output):
    """本地视频文件解析工具，提取视频文件的元数据信息"""
    logging.info(f"[tool] >>> python3 {' '.join(sys.argv)}")

    try:

        media_service = SimpleMediaService()
        material = media_service.get_media(media_id)

        body = json.dumps({
            "video_url": material["url"],
            "video_duration": 1 + math.floor(material["duration"])
        }, ensure_ascii=False)

        iccp_service = IccpService()
        submit_res = iccp_service.submit(115997442, body)
        click.echo(submit_res.model_dump_json())
        if submit_res.code != "0": exit(1)
        click.echo(f"提交任务成功，任务ID: {submit_res.data}")

        for _ in range(2 * 5):
            time.sleep(30)
            poll_res = iccp_service.query(submit_res.data) # type: ignore

            if poll_res.code == "1000": 
                continue
            if poll_res.code != "0": 
                click.echo(poll_res.model_dump_json(), err=True)
                exit(1)
            
            result = json.loads(poll_res.data) # type: ignore
            
            # 输出格式化后的JSON文件（原始格式）
            with open(output, "w") as f:
                cleaned_result = format_for_origin(result)
                json.dump(cleaned_result, f, ensure_ascii=False, indent=2)
            
            # 输出符合seedance格式的JSON文件
            seedance_output = output.replace('.json', '_seedance.json')
            with open(seedance_output, "w") as f:
                seedance_result = format_for_seedance(result)
                json.dump(seedance_result, f, ensure_ascii=False, indent=2)
            
            click.echo(Result(code="0", message="success", data={"output": output, "seedance_output": seedance_output}).model_dump_json())
            click.echo(f"任务完成，结果已保存到 {output} 和 {seedance_output}")
            return
        
        click.echo(f"任务正在执行中，请通过任务ID:{submit_res.data}查询任务状态")
    except Exception as e:
        click.echo(Result(code="-1", message=str(e)), err=True)
        exit(1)

if __name__ == "__main__":
    # main()
    with open("input/1.json", "r") as f:
        data = json.load(f)
        result = format_for_seedance(data)
        with open("output/1_seedance.json", "w") as f:
            json.dump(result, f, ensure_ascii=False, indent=2)

# 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-video-analyzer/media"

@cache
def init():
    """只执行一次的初始化方法，用于配置日志和目录"""
    log_dir = "/tmp/openclaw/byted-kickart-video-analyzer/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"] = self.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
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]:
        return {
            "ServiceName": service,
            "Authorization": f"Bearer {self.token}",
            "Content-Type": "application/octet-stream" if is_binary else "application/json"
        }

    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}")

        return {
            "X-Date": date,
            "X-Content-Sha256": payload_hash,
            "Content-Type": content_type,
            "Authorization": authorization
        }

    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 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
from core import MediaConfig
from utils.exception import SkillException



class RemoteUploader(ABC):
    """远程上传器接口 (策略模式)"""
    @abstractmethod
    def upload(self, file_path: 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_path: str) -> Any:
        owner_id = self.uploader.iam.get_admin_user_id()
        file_md5, file_crc32, file_size = HashUtils.file_hash(file_path)
        
        file_name = os.path.splitext(os.path.basename(file_path))[0]
        file_ext = os.path.splitext(file_path)[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_path, 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 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):
            raise SkillException("A0402", validation_result.get('errors', []))

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

        # 补全缺失的媒体信息
        matriel.width = getattr(metadata_result, 'width', 0)
        matriel.height = getattr(metadata_result, 'height', 0)
        matriel.duration = getattr(metadata_result, 'duration', 0)
        matriel.size = getattr(metadata_result, 'size', 0)

        # 将媒体信息保存到本地缓存
        media_info = {
            'id': matriel.id,
            'url': matriel.url,
            'type': matriel.type,
            'width': matriel.width,
            'height': matriel.height,
            'duration': matriel.duration,
            'size': matriel.size,
            'timestamp': time.time()
        }
        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)")

# 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.

"""
SKILL内部异常类定义模块
"""
class SkillException(Exception):
    """SKILL基础异常类
    
    所有SKILL内部异常的基类，提供统一的异常处理接口
    """
    
    def __init__(self, code: str, message: str):
        """
        Args:
            code: 错误码，用于标识具体错误类型
            message: 错误描述信息
            detail: 错误详情，可选
        """
        super().__init__(message)
        self.code = code
        self.message = message

    def __str__(self):
        return f"[{self.code}] {self.message}"
    
    def to_dict(self):
        """将异常转换为字典格式，便于序列化输出"""
        return {
            "code": self.code,
            "message": self.message
        }

# 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 math
from abc import ABC, abstractmethod

from PIL import Image
import cv2
from core.utils.matriel import Matriel, ImageMatriel, VideoMatriel

class MetadataExtractor(ABC):
    """元数据提取器接口 (策略模式接口)"""
    @abstractmethod
    def extract(self, file_path: str) -> Matriel:
        """提取文件的元数据，返回 Matriel 对象（ImageMatriel 或 VideoMatriel）"""
        pass


class ImageMetadataExtractor(MetadataExtractor):
    """图片元数据提取器 (具体策略)"""
    
    def extract(self, file_path: str) -> Matriel:
        file_size = os.path.getsize(file_path)
        with Image.open(file_path) as img:
            width, height = img.size
        return ImageMatriel(id="", type="image", url="", size=file_size, width=width, height=height)


class VideoMetadataExtractor(MetadataExtractor):
    """视频元数据提取器 (具体策略)"""
    
    def extract(self, file_path: str) -> Matriel:
        file_size = os.path.getsize(file_path)
        
        cap = cv2.VideoCapture(file_path)
        if not cap.isOpened():
            return VideoMatriel(id="", type="video", url="", size=file_size, width=0, height=0, duration=0.0)
        try:
            width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
            height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
            fps = cap.get(cv2.CAP_PROP_FPS)
            frame_count = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
            duration = math.floor(frame_count / fps if fps > 0 else 0.0) + 1
            
            return VideoMatriel(id="", type="video", url="", size=file_size, width=width, height=height, duration=duration)
        finally:
            cap.release()



__all__ = [
    "MetadataExtractor",
    "ImageMetadataExtractor", 
    "VideoMetadataExtractor"
]