Byted Kickart Ai Beauty

                                 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 os
import zipfile
import tarfile
import shutil
from concurrent.futures import ThreadPoolExecutor, as_completed

from core import Result
from core.api.iccp.service import IccpService
from core.api.meida.media import MuseRemoteUploader

AI_BEAUTY_SERVICE_ID = 2456212295650178


from core.utils.validator import DefaultValidator


def get_image_url(file_input):
    """
    获取图片URL，支持本地文件或URL
    
    Args:
        file_input: 文件路径或URL
        
    Returns:
        image_url: 图片URL
    """
    if file_input.startswith('http://') or file_input.startswith('https://'):
        # 对于 URL，先下载到临时文件进行校验
        import tempfile
        
        # 创建一个临时文件
        fd, temp_path = tempfile.mkstemp(suffix='.jpg')
        os.close(fd)
        
        try:
            # 复用已有的 download_image 方法下载图片
            success = download_image(file_input, temp_path)
            if not success:
                raise ValueError(f"图片下载失败，请检查URL是否有效")
            
            # 进行图片 size 校验
            val_result = DefaultValidator.validate(temp_path)
            if not val_result.get("valid"):
                err_msgs = "; ".join(val_result.get("errors", []))
                raise ValueError(f"图片校验失败: {err_msgs}")
                
            return file_input
        except Exception as e:
            if isinstance(e, ValueError):
                raise e
            raise ValueError(f"下载或校验URL图片失败: {str(e)}")
        finally:
            # 清理临时文件
            if os.path.exists(temp_path):
                os.remove(temp_path)
    else:
        if not os.path.isfile(file_input):
            raise ValueError(f"文件不存在: {file_input}")
            
        # 进行图片 size 校验
        val_result = DefaultValidator.validate(file_input)
        if not val_result.get("valid"):
            err_msgs = "; ".join(val_result.get("errors", []))
            raise ValueError(f"图片校验失败: {err_msgs}")
            
        uploader = MuseRemoteUploader()
        material = uploader.upload(file_input)
        return material.url


def wait_for_task_completion(iccp_service, task_id):
    """
    轮询等待任务完成
    
    Args:
        iccp_service: ICCP服务实例
        task_id: 任务ID
        
    Returns:
        Result: 包含 code, message 的 Result 对象
            - code: "0" 表示成功；其他错误码直接沿用 iccp_service.query 返回的错误码；
                    "-1" 表示解析任务结果失败；"-2" 表示任务超时
            - message: 成功时是美化后的图片链接，失败时是错误原因
    """
    for _ in range(2 * 5):
        time.sleep(30)
        poll_res = iccp_service.query(task_id)

        if poll_res.code == "1000": 
            continue
        if poll_res.code != "0": 
            return Result(
                code=poll_res.code,
                message=poll_res.message
            )
        
        try:
            result_data = json.loads(poll_res.message)
            return Result(
                code="0",
                message=result_data.get("result_url", "")
            )
        except Exception as e:
            return Result(
                code="-1",
                message=f"解析任务结果失败: {str(e)}"
            )
    
    return Result(
        code="-2",
        message=f"任务超时，请通过任务ID:{task_id}查询任务状态"
    )


def process_single_image(file_input, iccp_service=None, uploader=None):
    """
    处理单张图片
    
    Args:
        file_input: 图片路径或URL
        iccp_service: ICCP服务实例（可选，用于复用）
        uploader: 上传器实例（可选，用于复用）
        
    Returns:
        result: 处理结果字典
            - 成功: {"image_name": "...", "result_url": "..."}
            - 失败: {"image_name": "...", "error_code": "...", "error_msg": "..."}
    """
    try:
        # 获取原始文件名
        if file_input.startswith('http'):
            image_name = file_input.split('/')[-1].split('?')[0] or f"image_{int(time.time())}.jpg"
        else:
            image_name = os.path.basename(file_input)
        
        image_url = get_image_url(file_input)
        body = json.dumps({
            "beautyToolConfigKey": "beautyPro_481"
        }, ensure_ascii=False)

        if iccp_service is None:
            iccp_service = IccpService()

        submit_res = iccp_service.submit(AI_BEAUTY_SERVICE_ID, image_url, body)
        click.echo(submit_res.model_dump_json())
        if submit_res.code != "0": 
            return {
                "image_name": image_name,
                "error_code": submit_res.code,
                "error_msg": f"任务提交失败: {submit_res.message}"
            }
        click.echo(f"提交任务成功，任务ID: {submit_res.message}")

        result = wait_for_task_completion(iccp_service, submit_res.message)
        if result.code == "0":
            return {
                "image_name": image_name,
                "result_url": result.message
            }
        else:
            return {
                "image_name": image_name,
                "error_code": result.code,
                "error_msg": result.message
            }
    except Exception as e:
        error_msg = str(e)
        click.echo(f"处理图片失败 {file_input}: {error_msg}", err=True)
        
        # 获取原始文件名（即使失败也需要）
        if file_input.startswith('http'):
            image_name = file_input.split('/')[-1].split('?')[0] or f"image_{int(time.time())}.jpg"
        else:
            image_name = os.path.basename(file_input)
        
        return {
            "image_name": image_name,
            "error_code": "-999",
            "error_msg": error_msg
        }


def process_single_image_task(item, index, total, output_dir):
    """
    处理单张图片的任务函数（用于并发）
    
    Args:
        item: 图片路径或URL
        index: 图片索引
        total: 总图片数
        output_dir: 输出目录
        
    Returns:
        result: 处理结果字典
            - 成功: {"image_name": "...", "result_url": "...", "success": True}
            - 失败: {"image_name": "...", "error_code": "...", "error_msg": "...", "success": False}
    """
    # 增加随机延迟，避免所有线程同时发起请求导致服务端并发限制(触发1400等异常)
    import random
    time.sleep(random.uniform(0.1, 1.0))
    
    max_retries = 2
    retry_delay = 3
    
    for attempt in range(max_retries + 1):
        try:
            item_str = item[:60] + "..." if len(item) > 60 else item
            click.echo(f"[并发] 正在处理第 {index}/{total} 张(尝试 {attempt+1}/{max_retries+1}): {item_str}")
            
            # 获取原始文件名
            if item.startswith('http'):
                image_name = item.split('/')[-1].split('?')[0] or f"image_{index}.jpg"
            else:
                image_name = os.path.basename(item)
            
            image_url = get_image_url(item)
            
            # 防御性检查：确保image_url不为空
            if not image_url or not isinstance(image_url, str) or len(image_url.strip()) == 0:
                error_msg = f"图片URL为空或无效: {image_url}"
                click.echo(f"[并发] 第 {index}/{total} 张 {error_msg}")
                return {
                    "image_name": image_name,
                    "success": False,
                    "error_code": "-100",
                    "error_msg": error_msg
                }
            
            body = json.dumps({
                "beautyToolConfigKey": "beautyPro_481"
            }, ensure_ascii=False)

            # 每次请求新建服务实例，避免多线程复用同一个实例产生状态混乱
            iccp_service = IccpService()
            submit_res = iccp_service.submit(AI_BEAUTY_SERVICE_ID, image_url, body)
            
            if submit_res.code != "0": 
                # 针对 1400 (参数错误/缺失，可能是服务端并发限制导致) 进行重试
                if submit_res.code == "1400" and attempt < max_retries:
                    click.echo(f"[并发] 第 {index}/{total} 张提交失败，准备重试: {submit_res.message}")
                    time.sleep(retry_delay * (attempt + 1))
                    continue
                    
                error_msg = f"提交失败: {submit_res.message}"
                click.echo(f"[并发] 第 {index}/{total} 张 {error_msg}")
                return {
                    "image_name": image_name,
                    "success": False,
                    "error_code": submit_res.code,
                    "error_msg": error_msg
                }
            
            task_id = submit_res.message
            click.echo(f"[并发] 第 {index}/{total} 张提交成功，任务ID: {task_id}")

            result = wait_for_task_completion(iccp_service, task_id)
            if result.code == "0":
                # 下载美颜后的图片
                result_url = result.message
                if not image_name.lower().endswith(('.jpg', '.jpeg', '.png')):
                    image_name = f"image_{index}.jpg"
                output_img_path = os.path.join(output_dir, f"beauty_{image_name}")
                if output_img_path.lower().endswith('.png'):
                    output_img_path = output_img_path[:-4] + '.jpg'
                download_image(result_url, output_img_path)
                
                return {
                    "image_name": image_name,
                    "success": True,
                    "result_url": result_url
                }
            else:
                # 针对轮询阶段返回的 1400 错误也进行重试
                if result.code == "1400" and attempt < max_retries:
                    click.echo(f"[并发] 第 {index}/{total} 张任务轮询失败，准备重试: {result.message}")
                    time.sleep(retry_delay * (attempt + 1))
                    continue
                    
                return {
                    "image_name": image_name,
                    "success": False,
                    "error_code": result.code,
                    "error_msg": result.message
                }
        except BaseException as e:
            error_msg = str(e)
            if not error_msg and isinstance(e, SystemExit):
                error_msg = "组件内部退出(SystemExit)"
            click.echo(f"[并发] 处理第 {index}/{total} 张失败(尝试 {attempt+1}/{max_retries+1}): {type(e).__name__} - {error_msg}", err=True)
            
            # 对系统异常也进行重试
            if attempt < max_retries:
                time.sleep(retry_delay * (attempt + 1))
                continue
                
            # 获取原始文件名（即使失败也需要）
            if item.startswith('http'):
                image_name = item.split('/')[-1].split('?')[0] or f"image_{index}.jpg"
            else:
                image_name = os.path.basename(item)
            
            return {
                "image_name": image_name,
                "success": False,
                "error_code": "-999",
                "error_msg": error_msg
            }


def extract_archive_file(archive_file, extract_dir):
    """
    解压压缩包（支持 zip, tar, tar.gz, tar.bz2 格式）
    
    Args:
        archive_file: 压缩包文件路径
        extract_dir: 解压目录
        
    Returns:
        success: 是否成功
    """
    if os.path.exists(extract_dir):
        shutil.rmtree(extract_dir)
    os.makedirs(extract_dir, exist_ok=True)
    
    # 常见的非 UTF-8 编码列表（按优先级尝试）
    common_encodings = ['utf-8', 'gbk', 'gb2312', 'gb18030', 'big5', 'shift_jis', 'euc_kr']
    
    file_lower = archive_file.lower()
    try:
        if file_lower.endswith('.zip'):
            with zipfile.ZipFile(archive_file, 'r') as zip_ref:
                extract_dir_real = os.path.realpath(extract_dir) + os.sep
                for member in zip_ref.infolist():
                    # 检查 ZIP 通用比特标志第 11 位是否为 1 (表示 UTF-8 编码)
                    is_utf8 = (member.flag_bits & 0x800) != 0
                    
                    filename = member.filename
                    if not is_utf8:
                        # 如果不是 UTF-8 标记，zipfile 会错误地用 cp437 解码。我们将其还原为原始字节
                        try:
                            raw_bytes = filename.encode('cp437')
                            # 尝试使用常见编码进行解码
                            for enc in common_encodings:
                                try:
                                    filename = raw_bytes.decode(enc)
                                    break
                                except UnicodeDecodeError:
                                    continue
                        except Exception:
                            pass # 还原原始字节失败则保持原样
                    
                    member_path = os.path.join(extract_dir, filename)
                    if not os.path.realpath(member_path).startswith(extract_dir_real):
                        raise ValueError(f"压缩包包含非法路径: {filename}")
                    
                    # 确保目标目录存在
                    if member.is_dir():
                        os.makedirs(member_path, exist_ok=True)
                        continue
                        
                    os.makedirs(os.path.dirname(member_path), exist_ok=True)
                    
                    # 写入文件
                    with zip_ref.open(member) as source, open(member_path, 'wb') as target:
                        shutil.copyfileobj(source, target)
            return True
        elif file_lower.endswith(('.tar', '.tar.gz', '.tgz', '.tar.bz2', '.tbz2', '.tar.xz', '.txz')):
            with tarfile.open(archive_file, 'r:*') as tar_ref:
                extract_dir_real = os.path.realpath(extract_dir) + os.sep
                for member in tar_ref.getmembers():
                    # tarfile 中文件名编码问题处理
                    filename = member.name
                    if isinstance(member.name, bytes):
                        for enc in common_encodings:
                            try:
                                filename = member.name.decode(enc)
                                break
                            except UnicodeDecodeError:
                                continue
                                
                    member_path = os.path.join(extract_dir, filename)
                    if not os.path.realpath(member_path).startswith(extract_dir_real):
                        raise ValueError(f"压缩包包含非法路径: {filename}")
                        
                    # 更新 member 的 name 属性，确保 extract 时使用修正后的名称
                    original_name = member.name
                    member.name = filename
                    try:
                        tar_ref.extract(member, extract_dir)
                    finally:
                        member.name = original_name
            return True
        else:
            click.echo(Result(code="-1", message=f"不支持的压缩格式: {archive_file}"), err=True)
            return False
    except Exception as e:
        click.echo(Result(code="-1", message=f"解压失败: {str(e)}"), err=True)
        return False


def get_image_files_from_dir(dir_path):
    """
    从目录中获取所有支持的图片文件
    
    Args:
        dir_path: 目录路径
        
    Returns:
        image_files: 图片文件路径列表
    """
    supported_exts = ('.jpg', '.jpeg', '.png', '.webp')
    image_files = []
    for root, dirs, files in os.walk(dir_path):
        for f in files:
            if f.lower().endswith(supported_exts) and not f.startswith('._'):
                image_files.append(os.path.join(root, f))
    return image_files


def download_image(url, output_path):
    """
    下载图片
    
    Args:
        url: 图片URL
        output_path: 输出路径
        
    Returns:
        success: 是否成功
    """
    try:
        import urllib.parse
        import urllib.request
        
        # 对 URL 中的路径部分进行正确的编码，处理中文等非 ASCII 字符
        parsed_url = urllib.parse.urlparse(url)
        # 对路径部分单独进行 quote，保留域名等其他部分的原样
        quoted_path = urllib.parse.quote(parsed_url.path, safe='/')
        # 重新拼接 URL
        encoded_url = urllib.parse.urlunparse((
            parsed_url.scheme,
            parsed_url.netloc,
            quoted_path,
            parsed_url.params,
            parsed_url.query,
            parsed_url.fragment
        ))
        
        # 添加 User-Agent 头，避免被部分服务器拒绝
        headers = {
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36'
        }
        req = urllib.request.Request(encoded_url, headers=headers)
        
        with urllib.request.urlopen(req) as response, open(output_path, 'wb') as out_file:
            out_file.write(response.read())
            
        click.echo(f"已下载: {output_path}")
        return True
    except Exception as e:
        click.echo(f"下载失败: {str(e)}")
        return False


def create_zip_from_dir(source_dir, output_zip):
    """
    从目录创建zip文件
    
    Args:
        source_dir: 源目录
        output_zip: 输出zip路径
    """
    try:
        with zipfile.ZipFile(output_zip, 'w', zipfile.ZIP_DEFLATED) as zipf:
            for root, dirs, files in os.walk(source_dir):
                for f in files:
                    if f.lower().endswith(('.jpg', '.jpeg', '.png')):
                        file_path = os.path.join(root, f)
                        zipf.write(file_path, os.path.basename(file_path))
        click.echo(f"批量处理完成，所有美颜后图片已打包到: {output_zip}")
    except Exception as e:
        click.echo(f"打包zip失败: {str(e)}")


def process_multiple_urls(url_list, output, max_workers=5):
    """
    批量处理多张图片URL（并发处理）
    
    Args:
        url_list: 图片URL列表
        output: 输出JSON文件路径
        max_workers: 最大并发数，默认5
    """
    output_dir = "/tmp/ai_beauty_batch_output"
    # 清理临时目录，避免历史文件累积
    if os.path.exists(output_dir):
        shutil.rmtree(output_dir)

    os.makedirs(output_dir, exist_ok=True)
    
    # 过滤掉空URL
    valid_urls = [url.strip() for url in url_list if url.strip()]
    total = len(valid_urls)
    
    if total == 0:
        click.echo(Result(code="-1", message="没有有效的图片URL"), err=True)
        exit(1)
    
    click.echo(f"开始并发处理 {total} 张图片，最大并发数: {max_workers}")
    
    # 并发处理图片
    all_results = []
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        # 提交所有任务
        futures = []
        for i, url in enumerate(valid_urls, 1):
            futures.append(executor.submit(
                process_single_image_task, url, i, total, output_dir
            ))
        
        # 收集结果
        for future in as_completed(futures):
            result = future.result()
            if result:
                all_results.append(result)
    
    # 区分成功和失败结果
    success_results = [r for r in all_results if r.get("success")]
    failed_results = [r for r in all_results if not r.get("success")]
    
    click.echo(f"并发处理完成，成功处理 {len(success_results)}/{total} 张图片")
    if failed_results:
        click.echo(f"失败 {len(failed_results)} 张图片:")
        for r in failed_results:
            click.echo(f"  - {r['image_name']}: {r.get('error_msg', '未知错误')}")
    
    # 打包处理好的图片
    zip_output_path = output.replace('.json', '.zip')
    create_zip_from_dir(output_dir, zip_output_path)
    
    # 保存结果（同时包含成功和失败信息）
    with open(output, "w") as f:
        json.dump({
            "success_count": len(success_results),
            "failed_count": len(failed_results),
            "total_count": total,
            "results": all_results,
            "zip_path": zip_output_path
        }, f, ensure_ascii=False, indent=2)
    click.echo(Result(code="0", message=output).model_dump_json())


def process_archive_file(archive_file, output, max_workers=5):
    """
    批量处理压缩包中的图片（支持 zip, tar, tar.gz, tar.bz2 格式，并发处理）
    
    Args:
        archive_file: 压缩包文件路径
        output: 输出JSON文件路径
        max_workers: 最大并发数，默认5
    """
    output_dir = "/tmp/ai_beauty_batch_output"
    extract_dir = "/tmp/ai_beauty_extract"

    # 清理临时目录，避免历史文件累积
    if os.path.exists(output_dir):
        shutil.rmtree(output_dir)

    os.makedirs(output_dir, exist_ok=True)
    
    # 解压压缩包
    if not extract_archive_file(archive_file, extract_dir):
        exit(1)
    
    # 获取图片文件
    image_files = get_image_files_from_dir(extract_dir)
    total = len(image_files)
    if total == 0:
        click.echo(Result(code="-1", message="压缩包中未找到支持的图片文件"), err=True)
        exit(1)
    
    click.echo(f"开始并发处理 {total} 张图片，最大并发数: {max_workers}")
    
    # 并发处理图片
    all_results = []
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        # 提交所有任务
        futures = []
        for i, img_path in enumerate(image_files, 1):
            futures.append(executor.submit(
                process_single_image_task, img_path, i, total, output_dir
            ))
        
        # 收集结果
        for future in as_completed(futures):
            result = future.result()
            if result:
                all_results.append(result)
    
    # 区分成功和失败结果
    success_results = [r for r in all_results if r.get("success")]
    failed_results = [r for r in all_results if not r.get("success")]
    
    click.echo(f"并发处理完成，成功处理 {len(success_results)}/{total} 张图片")
    if failed_results:
        click.echo(f"失败 {len(failed_results)} 张图片:")
        for r in failed_results:
            click.echo(f"  - {r['image_name']}: {r.get('error_msg', '未知错误')}")
    
    # 打包处理好的图片
    zip_output_path = output.replace('.json', '.zip')
    create_zip_from_dir(output_dir, zip_output_path)
    
    # 保存结果（同时包含成功和失败信息）
    with open(output, "w") as f:
        json.dump({
            "success_count": len(success_results),
            "failed_count": len(failed_results),
            "total_count": total,
            "results": all_results,
            "zip_path": zip_output_path
        }, f, ensure_ascii=False, indent=2)
    click.echo(Result(code="0", message=output).model_dump_json())


@click.command()
@click.option("--file", required=True, type=str, help="图片文件路径或图片URL地址，多个URL用逗号分隔")
@click.option("--output", required=True, type=str, help="输出结果所在的json文件路径")
def main(file, output):
    """AI美颜SKILL工具，对图片进行美颜处理
    
    支持的输入格式：
    - 单张本地图片路径
    - 单张图片URL
    - 多张图片URL（用逗号分隔）
    - zip/tar/tar.gz/tar.bz2压缩包路径（包含多张图片）
    """
    logging.info(f"[tool] >>> python3 {' '.join(sys.argv)}")

    # 判断是否为压缩包
    is_archive = file.lower().endswith(('.zip', '.tar', '.tar.gz', '.tgz', '.tar.bz2', '.tbz2', '.tar.xz', '.txz'))
    
    # 判断是否为多个URL（用逗号分隔）
    is_multiple_urls = ',' in file and ('http://' in file or 'https://' in file)
    
    if is_archive:
        # 压缩包 - 直接传给脚本自动处理
        click.echo(f"检测到压缩包({file})，自动解压并处理所有图片...")
        process_archive_file(file, output)
    elif is_multiple_urls:
        # 多张图片URL - 依次处理并打包为zip
        url_list = [url.strip() for url in file.split(',') if url.strip()]
        click.echo(f"检测到 {len(url_list)} 张图片URL，将依次处理并打包为zip...")
        process_multiple_urls(url_list, output)
    else:
        # 单张图片/URL
        result = process_single_image(file)
        with open(output, "w") as f:
            json.dump(result, f, ensure_ascii=False, indent=2)
        click.echo(Result(code="0", message=output).model_dump_json())
        click.echo(f"任务完成，结果已保存到 {output}")

if __name__ == "__main__":
    main()

# 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


@cache
def init():
    """只执行一次的初始化方法，用于配置日志和目录"""
    log_dir = "/tmp/openclaw/doudian-link-parser"
    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")) and (tt_env := os.getenv("X_TT_ENV")):
            headers.update({"X-TT-Env": tt_env, "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)
        # for k, v in response.headers.items():
        #     print(f"{k}: {v}")
        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"

        if (ppe_env := os.getenv("X_VOLC_ENV")) and (tt_env := os.getenv("X_TT_ENV")):
            headers.update({"X-TT-Env": tt_env, "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)
        self.template_id = "245621001"
        
    def submit(self, service_id: int, image_url: str, params: str) -> Result:
        try:
            payload = {
                "ResourceList": [
                    image_url
                ],
                "TemplateId": self.template_id,
                "Resolution": "2k",
                "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=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] == 2000: 
                msg = jsonpath.jsonpath(resp, "$.Result.Message")
                if "No face detected" in msg[0]:
                    # 提示用户未检测到人脸
                    return Result(code=str(result_code[0]), message="未检测到人脸")
                else:
                    return Result(code=str(result_code[0]), message="URL不合法")

            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=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 AuthStrategyFactory, 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 url := image_media.get("TranscodeDownloadUrls", {}).get("origin"): return url
            if dl := image_media.get("DownloadUrl"): return dl
            for q in ["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", {})
            for q in ["origin", "jpeg_1080p", "jpeg_480p"]:
                if url := im.get("TranscodeDownloadUrls", {}).get(q):
                    im["TranscodeDownloadUrls"][q] = url
            if dl := im.get("DownloadUrl"): im["DownloadUrl"] = dl
        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.TranscodeDownloadUrls.origin"): m.url = v[0]
            elif 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 core.api.meida.chunks import (
    AppConfig, ApiClientFactory, 
    MaterialUploader, MediaFormatter
)

from utils.validator import Validator, DefaultValidator
from utils.extractor import MetadataExtractor

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/replicator/media"


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 MediaService:
    """业务服务层：协调校验、提取、上传与存储 (依赖注入)"""
    
    def __init__(self, repository: MediaRepository, uploader: RemoteUploader = None): # type: ignore
        self.repository = repository
        self.uploader = uploader or MuseRemoteUploader()

    def add_media(self, file: str, group: str, metadata: dict, extractor: MetadataExtractor = None, validator: Validator = None) -> Any: # type: ignore
        if extractor and validator:
            metadata_result = extractor.extract(file)
            validation_result = validator.validate(metadata_result)
        else:
            validation_result = DefaultValidator.validate(file)
            
        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)