import os
import time
import json
from openai import OpenAI
from datetime import datetime
import logging




class QwenBatchProcessor:
    """Qwen批量處理器完整實現"""


    def __init__(self, api_key, base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"):
        self.client = OpenAI(api_key=api_key, base_url=base_url)
        self.active_batches = {}


    def create_input_file(self, messages_list, output_path="batch_input.jsonl"):
        """創建批量處理輸入文件 (JSONL格式)"""
        with open(output_path, 'w', encoding='utf-8') as f:
            for i, messages in enumerate(messages_list):
                request_data = {
                    "custom_id": f"request_{i+1}_{int(time.time())}",
                    "method": "POST",
                    "url": "/v1/chat/completions",
                    "body": {
                        "model": "qwen-max",
                        "messages": messages,
                        "max_tokens": 2000
                    }
                }
                f.write(json.dumps(request_data, ensure_ascii=False) + '\n')
        return output_path


    def upload_file(self, file_path, purpose="batch"):
        """步驟1: 上傳文件到Qwen服務"""
        with open(file_path, "rb") as file:
            file_object = self.client.files.create(
                file=file,
                purpose=purpose
            )
        return file_object


    def create_batch_job(self, input_file_id, endpoint="/v1/chat/completions", metadata=None):
        """步驟2: 創建批量處理任務"""
        batch_config = {
            "input_file_id": input_file_id,
            "endpoint": endpoint,
            "completion_window": "24h"
        }


        if metadata:
            batch_config["metadata"] = metadata


        batch_object = self.client.batches.create(**batch_config)
        self.active_batches[batch_object.id] = batch_object
        return batch_object


    def check_job_status(self, batch_id):
        """步驟3: 檢查任務狀態"""
        return self.client.batches.retrieve(batch_id)


    def monitor_batch_status(self, batch_id, poll_interval=30, timeout=3600):
        """監控批量任務狀態 (自動輪詢)"""
        start_time = time.time()


        status_mapping = {
            "validating": {"icon": "??", "desc": "驗證中"},
            "in_progress": {"icon": "??", "desc": "處理中"},
            "completed": {"icon": "?", "desc": "已完成"},
            "failed": {"icon": "?", "desc": "失敗"},
            "expired": {"icon": "?", "desc": "已過期"},
            "cancelled": {"icon": "??", "desc": "已取消"}
        }


        while time.time() - start_time < timeout:
            batch_status = self.check_job_status(batch_id)
            current_status = batch_status.status
            status_info = status_mapping.get(current_status, {"icon": "?", "desc": "未知"})


            print(f"{status_info['icon']} 任務狀態: {status_info['desc']} ({current_status})")


            if current_status in ["completed", "failed", "expired", "cancelled"]:
                return batch_status


            time.sleep(poll_interval)


        raise TimeoutError(f"任務監控超時，已等待 {timeout} 秒")


    def get_output_file_id(self, batch_id):
        """步驟4: 獲取輸出文件ID"""
        batch_status = self.check_job_status(batch_id)
        return batch_status.output_file_id


    def get_error_file_id(self, batch_id):
        """步驟5: 獲取錯誤文件ID"""
        batch_status = self.check_job_status(batch_id)
        return getattr(batch_status, 'error_file_id', None)


    def download_results(self, file_id, output_path=None):
        """步驟6: 下載結果文件"""
        file_content = self.client.files.content(file_id).text


        if output_path:
            with open(output_path, 'w', encoding='utf-8') as f:
                f.write(file_content)


        return file_content


    def download_errors(self, file_id, output_path=None):
        """步驟7: 下載錯誤文件"""
        if not file_id:
            return None
        return self.download_results(file_id, output_path)


    def safe_batch_operation(self, operation_func, max_retries=3, initial_delay=1):
        """安全批量操作包裝器 - 帶重試機制"""
        last_exception = None


        for attempt in range(max_retries + 1):
            try:
                return operation_func()
            except Exception as e:
                last_exception = e
                if attempt < max_retries:
                    delay = initial_delay * (2 ** attempt)
                    print(f"?? 操作失敗，{delay}秒后重試 (嘗試 {attempt + 1}/{max_retries}): {str(e)}")
                    time.sleep(delay)
                else:
                    raise last_exception


# 使用示例
if __name__ == "__main__":
    API_KEY = "your-dashscope-api-key-here"
    processor = QwenBatchProcessor(api_key=API_KEY)


    test_messages = [
        [{"role": "user", "content": "請用一句話介紹人工智能"}],
        [{"role": "user", "content": "解釋一下機器學習的基本概念"}],
    ]


    # 執行完整工作流
    input_file_path = processor.create_input_file(test_messages)
    input_file = processor.upload_file(input_file_path)
    batch_job = processor.create_batch_job(input_file.id)
    final_status = processor.monitor_batch_status(batch_job.id)


    if final_status.status == 'completed':
        output_file_id = processor.get_output_file_id(batch_job.id)
        results = processor.download_results(output_file_id, "output.jsonl")
        print("批量處理完成！")

{"custom_id": "request_1", "method": "POST", "url": "/v1/chat/completions", "body": {"model": "qwen-max", "messages": [{"role": "user", "content": "你好"}]}}
{"custom_id": "request_2", "method": "POST", "url": "/v1/chat/completions", "body": {"model": "qwen-max", "messages": [{"role": "user", "content": "介紹一下AI"}]}}

def monitor_batch_status(batch_id, poll_interval=30):
    """實時監控任務狀態"""
    client = OpenAI(api_key="your-api-key")


    while True:
        batch_status = client.batches.retrieve(batch_id)
        current_status = batch_status.status


        status_info = {
            "validating": "?? 驗證中",
            "in_progress": "?? 處理中", 
            "completed": "? 已完成",
            "failed": "? 失敗"
        }.get(current_status, "未知狀態")


        print(f"{status_info} - 當前狀態: {current_status}")


        if current_status in ["completed", "failed", "expired", "cancelled"]:
            break


        time.sleep(poll_interval)

def safe_batch_operation(operation_func, max_retries=3, initial_delay=1):
    """指數退避重試機制"""
    for attempt in range(max_retries + 1):
        try:
            return operation_func()
        except Exception as e:
            if attempt < max_retries:
                delay = initial_delay * (2 ** attempt)  # 指數退避
                time.sleep(delay)
            else:
                raise e

from concurrent.futures import ThreadPoolExecutor


def concurrent_batch_processing(file_paths, max_workers=3):
    """并發執行批量處理"""
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = {
            executor.submit(process_single_batch, file_path): file_path 
            for file_path in file_paths
        }
        results = []
        for future in as_completed(futures):
            try:
                result = future.result()
                results.append(result)
            except Exception as e:
                print(f"處理失敗: {e}")
    return results

import psutil


def monitor_system_resources():
    """實時監控系統資源"""
    memory_usage = psutil.virtual_memory()
    cpu_percent = psutil.cpu_percent(interval=1)


    print(f"內存使用率: {memory_usage.percent}%")
    print(f"CPU使用率: {cpu_percent}%")


    # 設置資源限制
    if memory_usage.percent > 80:
        print("?? 內存使用過高，建議優化批量大小")