Antigravity Api Skill

# Ignore configuration files with secrets
libs/data/*.json
!libs/data/*.example.json

# Python
__pycache__/
*.pyc

# Output
output/
generated_assets/
video_cache/
**/video_cache/

import json
import os
import sys
from pathlib import Path
import requests
import base64
import mimetypes
import subprocess
import tempfile
import time

# Globally disable proxies to prevent localhost connection issues
s = requests.Session()
s.trust_env = False


class AntigravityClient:
    def __init__(self, api_key=None, base_url=None):
        self.config = self._load_config()
        self.base_url = base_url or self.config.get("base_url", "").rstrip("/")
        self.api_key = api_key or self.config.get("api_key", "")
        
        if not self.base_url or not self.api_key:
            print("[-] Error: Configuration missing base_url or api_key", file=sys.stderr)
            sys.exit(1)
            
    def _load_config(self):
        # [Fix] 支持 PyInstaller 打包后的路径
        paths_to_check = []
        if getattr(sys, 'frozen', False):
            exe_dir = Path(sys.executable).parent
            paths_to_check.append(exe_dir / "data" / "config.json")
            if hasattr(sys, '_MEIPASS'):
                paths_to_check.append(Path(sys._MEIPASS) / "data" / "config.json")

        current_dir = Path(__file__).parent
        paths_to_check.append(current_dir / "data" / "config.json")
        paths_to_check.append(Path.cwd() / "data" / "config.json")

        # 增加对 example 配置的回退支持 (实现零配置启动)
        paths_to_check.append(current_dir / "data" / "config.example.json")

        for p in paths_to_check:
            if p and p.exists():
                try:
                    config = json.loads(p.read_text(encoding='utf-8'))
                    if p.name.endswith(".example.json"):
                        print(f"[*] Config not found, using default template: {p.name}", file=sys.stderr)
                    return config
                except:
                    continue

        print(f"[-] Warning: No config or example found.", file=sys.stderr)
        return {}

    def _optimize_video(self, input_path, mute=False):
        """
        Use FFmpeg to compress large videos to a manageable size for AI.
        Target: 360P at low bitrate, keeping timing intact.
        """
        # Save to current working directory cache instead of temp
        cache_dir = Path("video_cache")
        cache_dir.mkdir(parents=True, exist_ok=True)

        # Consistent naming for caching based on modification time, name and mute status
        mtime = int(os.path.getmtime(input_path))
        safe_name = os.path.basename(input_path).replace(" ", "_")
        mute_suffix = "_muted" if mute else ""
        output_path = cache_dir / f"optimized_{mtime}{mute_suffix}_{safe_name}.mp4"

        if output_path.exists() and output_path.stat().st_size > 0:
            if mute:
                print(f"[*] 使用已缓存的压缩视频 (已静音): {output_path}", file=sys.stderr)
            return str(output_path)

        print(f"[*] 正在为 AI 分析优化视频: {os.path.basename(input_path)}...", file=sys.stderr)
        if mute:
            print("[!] 提示：为了极速上传，本次压缩已移除音频数据。", file=sys.stderr)
        else:
            print("[*] 提示：正在尝试保留原声压缩，如上传过慢可尝试在指令中要求“静音分析”。", file=sys.stderr)

        audio_opt = ['-an'] if mute else ['-c:a', 'aac', '-b:a', '64k']

        try:
            # 优先尝试 GPU 加速
            try:
                print(f"[*] 尝试硬件加速 (NVENC) 压缩...", file=sys.stderr)
                gpu_cmd = [
                    'ffmpeg', '-y', '-i', input_path,
                    '-c:v', 'h264_nvenc', '-preset', 'fast', '-cq', '38',
                    '-vf', 'scale=-2:360,fps=10'
                ] + audio_opt + [str(output_path)]
                subprocess.run(gpu_cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=True)
            except Exception:
                # 回退到 CPU
                print(f"[*] 硬件加速不可用，切换到 CPU (Ultrafast) 压缩...", file=sys.stderr)
                cpu_cmd = [
                    'ffmpeg', '-y', '-i', input_path,
                    '-vcodec', 'libx264', '-crf', '35', '-preset', 'ultrafast',
                    '-vf', 'scale=-2:360,fps=10'
                ] + audio_opt + [str(output_path)]
                subprocess.run(cpu_cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=True)

            new_size = os.path.getsize(output_path)
            print(f"[+] 优化完成: {new_size/1024/1024:.2f}MB", file=sys.stderr)
            return str(output_path)
        except Exception as e:
            print(f"[-] 优化失败 (FFmpeg 可能未安装或文件损坏): {e}", file=sys.stderr)
            return input_path # Fallback to original

    def upload_file(self, file_path):
        """
        Stream large files to the server using the /files endpoint.
        Try multiple formats and endpoints for compatibility.
        """
        if not os.path.exists(file_path):
            return None
            
        file_name = os.path.basename(file_path)
        file_size = os.path.getsize(file_path)
        mime_type, _ = mimetypes.guess_type(file_path)
        mime_type = mime_type or "application/octet-stream"
        
        if file_path.lower().endswith(('.mp4', '.mov', '.webm', '.ts')):
            mime_type = mime_type if "video" in mime_type else "video/mp4"

        # 安全处理文件名：Header 中不能包含非 ASCII 字符
        from urllib.parse import quote
        safe_file_name = quote(file_name)

        print(f"[*] Uploading {file_name} ({file_size/1024/1024:.2f}MB)...", file=sys.stderr)
        
        # Try a few common endpoints
        endpoints = [f"{self.base_url}/files"]
        if "/v1" in self.base_url:
            endpoints.append(self.base_url.replace("/v1", "") + "/files")
            endpoints.append(self.base_url.replace("/v1", "/upload/v1") + "/files")
            endpoints.append(self.base_url.replace("/v1", "/upload/v1beta") + "/files")
            
        for url in endpoints:
            try:
                # Mode 1: Multipart (Standard OpenAI compatible)
                with open(file_path, "rb") as f:
                    files = {
                        'file': (safe_file_name, f, mime_type),
                        'purpose': (None, 'fine-tune')
                    }
                    headers = {"Authorization": f"Bearer {self.api_key}"}
                    response = s.post(url, headers=headers, files=files, timeout=600)
                
                if response.status_code == 200:
                    result = response.json()
                    file_uri = result.get("file_uri") or result.get("id") or result.get("uri")
                    if file_uri:
                        print(f"[+] Upload success: {file_uri}")
                        return {"uri": file_uri, "mime_type": mime_type}
                else:
                    print(f"[-] Mode 1 failed ({response.status_code}) for {url}: {response.text[:100]}", file=sys.stderr)
                
                # Mode 2: Octet-stream
                with open(file_path, "rb") as f:
                    headers = {
                        "Authorization": f"Bearer {self.api_key}",
                        "X-File-Name": safe_file_name,
                        "X-File-Type": mime_type,
                        "Content-Type": "application/octet-stream"
                    }
                    response = s.post(url, headers=headers, data=f, timeout=600)
                
                if response.status_code == 200:
                    result = response.json()
                    file_uri = result.get("file_uri") or result.get("uri")
                    if file_uri:
                        print(f"[+] Upload success: {file_uri}")
                        return {"uri": file_uri, "mime_type": mime_type}
                else:
                    print(f"[-] Mode 2 failed ({response.status_code}) for {url}: {response.text[:100]}", file=sys.stderr)
                        
            except Exception as e:
                print(f"[-] Attempt failed for {url}: {e}", file=sys.stderr)
                
        return None

    def chat_completion(self, messages, model=None, temperature=0.7, file_paths=None, file_path=None):
        url = f"{self.base_url}/chat/completions"
        model = model or self.config.get("default_chat_model", "claude-sonnet-4-5")
        
        paths = []
        if file_path: paths.append(file_path)
        if file_paths:
            if isinstance(file_paths, list): paths.extend(file_paths)
            else: paths.append(file_paths)
            
        multimodal_content = []
        
        for path in paths:
            if not os.path.exists(path): continue
            
            # Smart optimization: if it's a video and > 10MB, compress it first
            is_video = path.lower().endswith(('.mp4', '.mov', '.webm', '.ts'))
            file_size = os.path.getsize(path)
            
            working_path = path
            if is_video and file_size > 10 * 1024 * 1024:
                working_path = self._optimize_video(path)
            
            # All media sent via Base64 for maximum compatibility
            try:
                mime_type, _ = mimetypes.guess_type(working_path)
                mime_type = mime_type or "application/octet-stream"
                if is_video: mime_type = "video/mp4" # Ensure video mime type
                
                print(f"[*] Encoding media (Base64): {os.path.basename(working_path)}", file=sys.stderr)
                with open(working_path, "rb") as f:
                    b64_data = base64.b64encode(f.read()).decode("utf-8")
                
                multimodal_content.append({
                    "type": "image_url",
                    "image_url": {"url": f"data:{mime_type};base64,{b64_data}"}
                })
            except Exception as e:
                print(f"[-] Failed to process {path}: {e}", file=sys.stderr)

        if multimodal_content and messages and messages[-1]['role'] == 'user':
            original_text = messages[-1]['content']
            # Avoid nesting if it's already a list from a previous attempt
            if isinstance(original_text, list):
                # Check if it already has multimodal items to avoid duplicates
                existing_types = [item.get("type") for item in original_text if isinstance(item, dict)]
                if "image_url" not in existing_types:
                    original_text.extend(multimodal_content)
            else:
                new_content = [{"type": "text", "text": original_text}]
                new_content.extend(multimodal_content)
                messages[-1]['content'] = new_content

        payload = {
            "model": model,
            "messages": messages,
            "temperature": temperature,
            "stream": True
        }
        
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json",
            "User-Agent": "Antigravity/4.0.6" 
        }
        
        try:
            print(f"[*] Sending Payload ({len(json.dumps(payload))/1024/1024:.1f}MB) to {url}...", file=sys.stderr)
            print("[*] Please wait, this may take a minute for large videos...", file=sys.stderr)
            response = s.post(url, headers=headers, json=payload, stream=True, timeout=900) 
            
            # --- 自动降级逻辑 (Fallback) ---
            # 如果请求的是 gemini-3-pro 且返回 503 (账号故障/负载过高)
            if response.status_code == 503 and model == "gemini-3-pro":
                print(f"[!] gemini-3-pro 返回 503 (繁忙/账号故障)，正在自动切换到 gemini-3-flash 进行重试...", file=sys.stderr)
                payload["model"] = "gemini-3-flash"
                response = s.post(url, headers=headers, json=payload, stream=True, timeout=900)
            
            print(f"[*] Response received: {response.status_code}", file=sys.stderr)
            return response
        except Exception as e:
            print(f"[-] Request failed: {e}", file=sys.stderr)
            return None

    def generate_image(self, prompt, size="1024x1024", image_path=None, quality="standard", n=1):
        # According to the provided SDK, images are generated via the /chat/completions endpoint
        url = f"{self.base_url}/chat/completions"
        model = "gemini-3.1-flash-image"
        
        messages = []
        if image_path and os.path.exists(image_path):
            print(f"[*] Encoding reference image: {image_path}", file=sys.stderr)
            try:
                mime_type, _ = mimetypes.guess_type(image_path)
                mime_type = mime_type or "image/png"
                img_data = base64.b64encode(open(image_path, "rb").read()).decode("utf-8")
                
                messages.append({
                    "role": "user",
                    "content": [
                        {"type": "text", "text": prompt},
                        {
                            "type": "image_url",
                            "image_url": {"url": f"data:{mime_type};base64,{img_data}"}
                        }
                    ]
                })
            except Exception as e:
                print(f"[-] Failed to read image: {e}", file=sys.stderr)
                messages.append({"role": "user", "content": prompt})
        else:
            messages.append({"role": "user", "content": prompt})

        # Mapping size for the chat endpoint structure
        payload = {
            "model": model,
            "messages": messages,
            "size": size, # Using extra_body logic as top level for simplicity in REST
            "stream": False # Getting full response usually contains the URL/Image Markdown
        }
        
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json",
            "User-Agent": "Antigravity/4.0.6"
        }
        
        print(f"[*] Sending Image Request via Chat API to {model}...", file=sys.stderr)
        try:
            response = s.post(url, headers=headers, json=payload, timeout=120)
            if response.status_code == 200:
                return response.json()
            else:
                print(f"[-] API Error {response.status_code}: {response.text}")
                return None
        except Exception as e:
            print(f"[-] Image Request failed: {e}", file=sys.stderr)
            return None

    def get_models(self):
        url = f"{self.base_url}/models"
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "User-Agent": "Antigravity/4.0.6"
        }
        
        try:
            response = s.get(url, headers=headers, timeout=10)
            if response.status_code == 200:
                data = response.json()
                # Unified parsing: some APIs return {'data': [...]}, some return list directly
                if isinstance(data, dict) and 'data' in data:
                    return data['data']
                elif isinstance(data, list):
                    return data
                return []
            else:
                print(f"[-] API Error {response.status_code}: {response.text}")
                return []
        except Exception as e:
            print(f"[-] Models Request failed: {e}", file=sys.stderr)
            return []

{
    "base_url": "http://127.0.0.1:8045/v1",
    "api_key": "YOUR_API_KEY_HERE",
    "default_chat_model": "gemini-3-flash",
    "default_image_model": "gemini-3.1-flash-image"
}

import sys
import os
import json
from pathlib import Path

# 强制设置标准输出为 UTF-8，解决 Windows 乱码问题
if sys.stdout.encoding != 'utf-8':
    try:
        sys.stdout.reconfigure(encoding='utf-8')
    except AttributeError:
        # 兼容旧版本 Python
        import io
        sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8')

# Add libs to path
current_dir = Path(__file__).parent
libs_path = current_dir.parent / "libs"
sys.path.append(str(libs_path))

try:
    from api_client import AntigravityClient
except ImportError:
    print("[-] Error: libs module not found")
    sys.exit(1)

def main():
    if len(sys.argv) < 2:
        print("Usage: python chat.py \"Your prompt here\" [model_name] [media_path]")
        return

    prompt = sys.argv[1]
    # Try to find file path in args
    media_paths = []
    # Collect all existing file paths from arguments
    for arg in sys.argv[2:]:
        if os.path.exists(arg):
            media_paths.append(arg)
            
    # Set model if it was provided and isn't a file path
    model = None
    if len(sys.argv) > 2 and not os.path.exists(sys.argv[2]):
        model = sys.argv[2]
    
    client = AntigravityClient()
    
    messages = [{"role": "user", "content": prompt}]
    
    print(f"[*] Asking {model or client.config.get('default_chat_model')}...")
    
    response = client.chat_completion(messages, model=model, file_paths=media_paths)
    
    if not response or response.status_code != 200:
        if response:
            print(f"[-] AI Request failed ({response.status_code}): {response.text}")
        return

    full_content = ""
    print("\nStarting response stream:\n" + "-"*30)
    
    # Simple SSE parser
    for line in response.iter_lines():
        if not line: continue
        line_str = line.decode('utf-8')
        if line_str.startswith("data: "):
            data_str = line_str[6:]
            if data_str.strip() == "[DONE]":
                break
            try:
                data = json.loads(data_str)
                delta = data.get("choices", [{}])[0].get("delta", {})
                content = delta.get("content", "")
                if content:
                    print(content, end="", flush=True)
                    full_content += content
            except:
                pass
                
    print("\n" + "-"*30 + "\n[Done]")

if __name__ == "__main__":
    main()

import sys
import os
import time
import base64
import re
import requests
from pathlib import Path

# Add libs to path
current_dir = Path(__file__).parent
libs_path = current_dir.parent / "libs"
sys.path.append(str(libs_path))

try:
    from api_client import AntigravityClient
except ImportError:
    print("[-] Error: libs module not found")
    sys.exit(1)

def main():
    if len(sys.argv) < 2:
        print("Usage: python generate_image.py \"Prompt\" [size] [image_path]")
        return

    prompt = sys.argv[1]
    size_arg = sys.argv[2] if len(sys.argv) > 2 else "1024x1024"
    image_path = sys.argv[3] if len(sys.argv) > 3 else None
    
    ratio_map = {
        "16:9": "1280x720",
        "9:16": "720x1280", 
        "1:1": "1024x1024"
    }
    
    target_size = ratio_map.get(size_arg, size_arg)
    
    client = AntigravityClient()
    res = client.generate_image(prompt, size=target_size, image_path=image_path)
    
    if res and "choices" in res:
        content = res["choices"][0].get("message", {}).get("content", "")
        print(f"[*] Response content received (Length: {len(content)})")
        
        save_dir = Path(os.getcwd()) / "generated_assets"
        save_dir.mkdir(parents=True, exist_ok=True)
        saved_any = False

        # 1. Look for plain URLs
        urls = re.findall(r"http[s]?://(?:[a-zA-Z]|[0-9]|[$-_@.&+]|[!*\\(\\),]|(?:%[0-9a-fA-F][0-9a-fA-F]))+", content)
        if urls:
            for i, url in enumerate(urls):
                try:
                    print(f"[*] Downloading image from {url}...")
                    img_resp = requests.get(url, timeout=30)
                    if img_resp.status_code == 200:
                        fname = f"antigravity_{int(time.time())}_url_{i}.png"
                        save_path = save_dir / fname
                        save_path.write_bytes(img_resp.content)
                        print(f"[+] Image saved: {save_path}")
                        saved_any = True
                except Exception as e:
                    print(f"[-] Download failed: {e}")

        # 2. Look for Base64 Data (common in Markdown or raw)
        # Pattern: data:image/png;base64,xxxx or just long base64 string inside parentheses
        b64_matches = re.findall(r"data:image\/[a-zA-Z]+;base64,([a-zA-Z0-9+/=]+)", content)
        if not b64_matches:
            # Try to find base64-like blobs in Markdown image syntax ![alt](data:...)
            b64_matches = re.findall(r"base64,([a-zA-Z0-9+/=]{100,})", content)

        if b64_matches:
            for i, b64_str in enumerate(b64_matches):
                try:
                    print(f"[*] Decoding Base64 image {i}...")
                    img_data = base64.b64decode(b64_str)
                    fname = f"antigravity_{int(time.time())}_b64_{i}.png"
                    save_path = save_dir / fname
                    save_path.write_bytes(img_data)
                    print(f"[+] Image saved: {save_path}")
                    saved_any = True
                except Exception as e:
                    print(f"[-] Base64 decode failed: {e}")

        if not saved_any:
            print("[-] No image URL or Base64 data found in response")
            if len(content) > 200:
                print(f"[*] Content snippet: {content[:200]}...")
    else:
        print("[-] Generation failed")

if __name__ == "__main__":
    main()

import sys
from pathlib import Path

# Add libs to path
current_dir = Path(__file__).parent
libs_path = current_dir.parent / "libs"
sys.path.append(str(libs_path))

try:
    from api_client import AntigravityClient
except ImportError:
    print("[-] Error: libs module not found")
    sys.exit(1)

def main():
    client = AntigravityClient()
    print("[*] Fetching available models...")
    
    models = client.get_models()
    
    if not models:
        print("[-] No models found or request failed.")
        return

    print(f"\n[+] Found {len(models)} models:\n")
    
    # Categorize models for better readability
    chat_models = []
    image_models = []
    other_models = []
    
    for m in models:
        mid = m['id'] if isinstance(m, dict) else str(m)
        if "image" in mid or "paint" in mid:
            image_models.append(mid)
        elif "claude" in mid or "gpt" in mid or "gemini" in mid:
            chat_models.append(mid)
        else:
            other_models.append(mid)
            
    if chat_models:
        print("--- Chat / Text Models ---")
        for m in sorted(chat_models):
            print(f"  {m}")
        print("")
        
    if image_models:
        print("--- Image / Vision Models ---")
        for m in sorted(image_models):
            print(f"  {m}")
        print("")

    if other_models:
        print("--- Other Models ---")
        for m in sorted(other_models):
            print(f"  {m}")

if __name__ == "__main__":
    main()

import sys
import os
import base64
import requests
import mimetypes
from pathlib import Path

# Add libs to path
current_dir = Path(__file__).parent
libs_path = current_dir.parent / "libs"
sys.path.append(str(libs_path))

try:
    from api_client import AntigravityClient
except ImportError:
    print("[-] Error: libs module not found")
    sys.exit(1)

def main():
    if len(sys.argv) < 3:
        print("Usage: python test_video_upload.py \"Prompt\" \"Video Path\"")
        return

    prompt = sys.argv[1]
    video_path = sys.argv[2]
    
    if not os.path.exists(video_path):
        print(f"[-] Video file not found: {video_path}")
        return

    print(f"[*] Reading video file: {video_path}...")
    try:
        video_data = open(video_path, "rb").read()
        b64_video = base64.b64encode(video_data).decode("utf-8")
        
        # Simple mime guessing, default to mp4
        mime_type, _ = mimetypes.guess_type(video_path)
        mime_type = mime_type or "video/mp4"
        
        print(f"[*] Video size: {len(video_data)/1024/1024:.2f} MB")
        print(f"[*] MIME type: {mime_type}")
    except Exception as e:
        print(f"[-] Failed to read video: {e}")
        return

    client = AntigravityClient()
    
    # Constructing payload with video (Google/Gemini style or OpenAI Vision style experiment)
    # Gemini 1.5 Pro/Flash supports video input. The structure usually is a list of parts.
    # For OpenAI compatibility layers, it might be treated as an image_url with video mime type, 
    # or a specific 'video_url' block depending on the backend implementation.
    # We will try the standard "image_url" block first but with video mime type, 
    # as some adapters use this for all media.
    
    messages = [
        {
            "role": "user",
            "content": [
                {"type": "text", "text": prompt},
                {
                    "type": "image_url", # Trying image_url first as a generic media container
                    "image_url": {"url": f"data:{mime_type};base64,{b64_video}"}
                }
            ]
        }
    ]
    
    # We use a chat model that is likely to support multimodal (Gemini 1.5 Pro / Flash)
    model = "gemini-3-flash" 
    
    print(f"[*] Sending Video Chat Request to {model}...")
    
    # We use chat_completion method but manually override the payload if needed within the method, 
    # or just call it directly since we constructed the messages.
    
    try:
        # Re-using the client logic but injecting our multimodal message
        response = client.chat_completion(messages, model=model)
        
        if response:
            print("\n" + "="*30)
            print(f"Status Code: {response.status_code}")
            # Stream the response
            for line in response.iter_lines():
                if not line: continue
                line_str = line.decode('utf-8')
                if line_str.startswith("data: "):
                    data_str = line_str[6:]
                    if data_str.strip() == "[DONE]":
                        break
                    try:
                        data = json.loads(data_str)
                        delta = data.get("choices", [{}])[0].get("delta", {})
                        content = delta.get("content", "")
                        if content:
                            print(content, end="", flush=True)
                    except:
                        pass
            print("\n" + "="*30)
        else:
            print("[-] No response received")
            
    except Exception as e:
        print(f"[-] Request failed: {e}")

if __name__ == "__main__":
    import json
    main()

import sys
import os
import json
from pathlib import Path

# 自动寻找库文件路径
current_dir = Path(__file__).parent
libs_path = current_dir.parent / "libs"
sys.path.append(str(libs_path))

try:
    from api_client import AntigravityClient
except ImportError:
    print("[-] 错误: 找不到 libs 模块，请检查目录结构。")
    sys.exit(1)

def analyze_video(video_path, custom_prompt=None):
    if not os.path.exists(video_path):
        print(f"[-] 错误: 找不到视频文件 {video_path}")
        return

    # 1. 实例化客户端 (自动从 config.json 获取端口和 key)
    client = AntigravityClient()
    
    # 2. 默认的高精度分析提示词
    default_prompt = (
        "请拆解视频的镜头。分析每一个镜头的开始时间、持续秒数、以及内容描述（包含景别、动作）。\n"
        "请严格按照以下 JSON 数组格式输出，不要包含 Markdown 代码块标记或任何其他多余文本：\n"
        "[\n"
        "  {\"start\": \"HH:MM:SS\", \"duration\": 5, \"text\": \"分镜分析描述\"},\n"
        "  ...\n"
        "]\n"
    )
    prompt = custom_prompt or default_prompt
    
    # 3. 指定最适合视频分析的模型
    # 优先使用 gemini-3-flash
    model = "gemini-3-flash"
    
    print(f"[*] 正在分析视频: {os.path.basename(video_path)}", file=sys.stderr)
    print(f"[*] 正在请求模型: {model} (连接地址: {client.base_url})", file=sys.stderr)
    
    messages = [{"role": "user", "content": prompt}]
    
    # 获取响应流
    try:
        response = client.chat_completion(messages, model=model, file_paths=[video_path])
    except Exception as e:
        print(f"[-] 连接服务失败: {e}", file=sys.stderr)
        return
    
    if not response or response.status_code != 200:
        if response:
            print(f"[-] API 请求失败 ({response.status_code}): {response.text}", file=sys.stderr)
        else:
            print("[-] 未能收到有效响应，请确认服务是否开启。", file=sys.stderr)
        return

    # 4. 获取完整 JSON 响应
    full_content = ""
    for line in response.iter_lines():
        if not line: continue
        line_str = line.decode('utf-8')
        if line_str.startswith("data: "):
            data_str = line_str[6:]
            if data_str.strip() == "[DONE]": break
            try:
                data = json.loads(data_str)
                content = data.get("choices", [{}])[0].get("delta", {}).get("content", "")
                if content:
                    full_content += content
            except: pass
    
    # 清理 Markdown 代码块包裹
    clean_json = full_content.strip()
    if clean_json.startswith("```"):
        clean_json = clean_json.split("\n", 1)[1]
    if clean_json.endswith("```"):
        clean_json = clean_json.rsplit("\n", 1)[0]
    
    print(clean_json.strip())

if __name__ == "__main__":
    if len(sys.argv) < 2:
        print("用法: python video_analyzer.py \"视频绝对路径\" [可选自定义提示词]")
    else:
        # 处理可能的双引号包裹
        path = sys.argv[1].strip('"').strip("'")
        p = sys.argv[2] if len(sys.argv) > 2 else None
        analyze_video(path, p)