Gemini Web Quicker Skill

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

# Python
__pycache__/
*.pyc

# Output
output/
generated_assets/

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):
        self.config = self._load_config()
        self.base_url = self.config.get("base_url", "").rstrip("/")
        self.api_key = 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}"

        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')):
            mime_type = mime_type if "video" in mime_type else "video/mp4"

        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': (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": 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", "gemini-3-flash")
        
        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)
            
        files_payload = []
        
        for path in paths:
            if not os.path.exists(path): continue
            
            # 智能优化：如果视频太大，先压缩。
            is_video = path.lower().endswith(('.mp4', '.mov', '.webm'))
            file_size = os.path.getsize(path)
            
            working_path = path
            # [生产模式]: 视频超过 100MB 时自动执行 AI 优化压缩
            if is_video and file_size > 100 * 1024 * 1024:
                working_path = self._optimize_video(path)
            
            try:
                mime_type, _ = mimetypes.guess_type(working_path)
                # 强制修正常见视频格式的 mime_type
                if is_video: mime_type = "video/mp4"
                mime_type = mime_type or "application/octet-stream"
                
                print(f"[*] Encoding media for files-payload: {os.path.basename(working_path)}", file=sys.stderr)
                with open(working_path, "rb") as f:
                    b64_data = base64.b64encode(f.read()).decode("utf-8")
                
                files_payload.append({
                    "filename": os.path.basename(path),
                    "mime_type": mime_type,
                    "file_data": f"data:{mime_type};base64,{b64_data}"
                })
            except Exception as e:
                print(f"[-] Failed to process {path}: {e}", file=sys.stderr)

        payload = {
            "model": model,
            "messages": messages,
            "temperature": temperature,
            "stream": True
        }
        
        # 如果有文件，注入顶级 files 字段 (Antigravity 专有协议)
        if files_payload:
            payload["files"] = files_payload

        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json",
            "User-Agent": "Antigravity/4.0.6" 
        }
        
        try:
            # 打印负载概览
            file_size_total = sum(len(f["file_data"]) for f in files_payload) / 1024 / 1024 if files_payload else 0
            print(f"[*] Sending Payload ({file_size_total:.1f}MB media) to {url}...", file=sys.stderr)
            
            response = s.post(url, headers=headers, json=payload, stream=True, timeout=900) 
            
            # --- 自动降级逻辑 (Fallback) ---
            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)
            
            if response.status_code != 200:
                print(f"[-] Request failed with status {response.status_code}: {response.text}", 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):
        """
        [协议修正]: 完全采用原生 urllib 进行文件上传和流式处理，修复 requests 库在此场景下的边界问题。
        """
        url = f"{self.base_url}/chat/completions"
        model = self.config.get("default_image_model", "gemini-3-flash-image")
        
        # 统一使用顶级 files 字段传输文件 (Antigravity 专有协议)
        files_payload = []
        if image_path and os.path.exists(image_path):
            try:
                mime_type, _ = mimetypes.guess_type(image_path)
                mime_type = mime_type or "image/png"
                print(f"[*] Encoding reference image for files-payload: {os.path.basename(image_path)}", file=sys.stderr)
                with open(image_path, "rb") as f:
                    b64_data = base64.b64encode(f.read()).decode("ascii")
                
                # 完全按照 test_image_upload.py 的 payload.files 格式
                files_payload.append({
                    "filename": os.path.basename(image_path),
                    "mime_type": mime_type,
                    "file_data": f"data:{mime_type};base64,{b64_data}"
                })
            except Exception as e:
                print(f"[-] Failed to process reference image {image_path}: {e}", file=sys.stderr)

        payload = {
            "model": model,
            "messages": [{"role": "user", "content": prompt}],
            "size": size,
            "stream": True # 强制流式
        }
        
        if files_payload:
            payload["files"] = files_payload
            print(f"[*] Injected reference image into payload['files']", file=sys.stderr)
        
        print(f"[*] Sending Image Request (urllib Native Protocol) to {model}...", file=sys.stderr)
        
        import urllib.request
        import urllib.error
        
        data = json.dumps(payload).encode("utf-8")
        req = urllib.request.Request(
            url=url,
            method="POST",
            data=data,
            headers={
                "Authorization": f"Bearer {self.api_key}",
                "Content-Type": "application/json",
                "Accept": "text/event-stream"
            }
        )
        
        try:
            with urllib.request.urlopen(req, timeout=180) as resp:
                content_type = (resp.headers.get("Content-Type") or "").lower()
                full_content = ""
                
                if "text/event-stream" not in content_type:
                    raw = resp.read().decode("utf-8", errors="replace")
                    try:
                        obj = json.loads(raw)
                        full_content = obj.get("choices", [{}])[0].get("message", {}).get("content", "")
                    except Exception:
                        pass
                else:
                    for raw_line in resp:
                        line = raw_line.decode("utf-8", errors="replace").rstrip("\r\n")
                        if not line or not line.startswith("data:"):
                            continue
                            
                        data_str = line[5:].strip()
                        if data_str == "[DONE]": 
                            break

                        try:
                            obj = json.loads(data_str)
                            delta = obj.get("choices", [{}])[0].get("delta", {}).get("content", "")
                            if delta:
                                full_content += delta
                        except Exception:
                            continue
                
                print(f"[*] Response content received (Length: {len(full_content)})", file=sys.stderr)
                if len(full_content) < 500:
                    print(f"[*] Response details: {full_content}", file=sys.stderr)
                
                return {
                    "choices": [{
                        "message": {"content": full_content}
                    }]
                }
        except urllib.error.HTTPError as e:
            err_body = e.read().decode("utf-8", errors="replace")
            print(f"[-] HTTP Error {e.code}: {err_body[:500]}", file=sys.stderr)
            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:55557/v1",
    "api_key": "YOUR_API_KEY_HERE",
    "default_chat_model": "gemini-3-flash",
    "default_image_model": "gemini-3-pro-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
from PIL import Image

# 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 create_black_reference_image(size_str, output_dir):
    """
    根据尺寸字符串 (如 1280x720) 创建一张纯黑图作为参考。
    """
    try:
        w, h = map(int, size_str.split('x'))
        img = Image.new('RGB', (w, h), color='black')
        output_dir.mkdir(parents=True, exist_ok=True)
        path = output_dir / f"ref_black_{size_str}.png"
        img.save(path)
        return path
    except Exception as e:
        print(f"[-] Failed to create reference image: {e}")
        return None

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", 
        "4:3": "1024x768",
        "3:4": "768x1024",
        "1:1": "1024x1024"
    }
    
    target_size = ratio_map.get(size_arg, size_arg)
    
    # [优化]: 将尺寸控制逻辑结合进提示词 (Prompt Injection)
    # Gemini 网页版对 "16:9 aspect ratio", "widescreen", "portrait" 等词汇更敏感
    enhanced_prompt = prompt
    if size_arg == "16:9":
        if "16:9" not in prompt and "widescreen" not in prompt.lower():
            enhanced_prompt = f"{prompt}, widescreen, cinematic wide shot, 16:9 aspect ratio"
    elif size_arg == "9:16":
        if "9:16" not in prompt and "portrait" not in prompt.lower():
            enhanced_prompt = f"{prompt}, portrait, vertical poster, phone wallpaper format, 9:16 aspect ratio"
    elif size_arg == "4:3":
        if "4:3" not in prompt and "landscape" not in prompt.lower():
            enhanced_prompt = f"{prompt}, standard landscape, 4:3 aspect ratio"
    elif size_arg == "3:4":
        if "3:4" not in prompt and "portrait" not in prompt.lower():
            enhanced_prompt = f"{prompt}, standard portrait, 3:4 aspect ratio"
    elif size_arg:
        if size_arg not in prompt:
            enhanced_prompt = f"{prompt}, {size_arg} aspect ratio"
    
    def create_blank_reference_image(size_str, output_dir):
        """
        根据尺寸字符串创建一个纯白画布。使用白底（代表空白画板）或随机噪点更不容易影响最终成图的色调。
        """
        try:
            output_dir.mkdir(parents=True, exist_ok=True)
            path = output_dir / f"ref_blank_{size_str}.png"
            # [持久化缓存]: 如果该比例的白底图已存在，直接复用，不重复创建
            if not path.exists():
                w, h = map(int, size_str.split('x'))
                img = Image.new('RGB', (w, h), color='white')
                img.save(path)
            return path
        except Exception as e:
            print(f"[-] Failed to create reference image: {e}")
            return None

    ref_image_path = image_path
    if not ref_image_path and size_arg in ratio_map and size_arg != "1:1":
        # 将参考白底图持久化存储在技能的 resources 目录中
        ref_dir = current_dir.parent / "resources" / "aspect_ratios"
        ref_image_path = create_blank_reference_image(target_size, ref_dir)
        print(f"[*] Using blank reference image for ratio {size_arg} ({target_size}): {ref_image_path}")
        # 添加防污染提示词，告诉模型仅仅使用这张图作为宽高比模板
        if ref_image_path:
            enhanced_prompt += " (IMPORTANT STRICT INSTRUCTION: The attached image is just a blank white canvas. IGNORING its color and content entirely. Generate the image purely based on the text prompt and fill the entire scene with appropriate, rich, and vibrant colors. ONLY use the attached image to MATCH the ASPECT RATIO)."
    
    client = AntigravityClient()
    print(f"[*] Final Prompt: {enhanced_prompt}")
    res = client.generate_image(enhanced_prompt, size=target_size, image_path=str(ref_image_path) if ref_image_path else None)
    
    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. 优先从 Markdown 语法中寻找图片 URL: ![](URL)
        markdown_urls = re.findall(r"!\[.*?\]\((http[s]?://[^\s\)]+)\)", content)
        
        # 2. 如果没找到，再尝试寻找纯文本中的 URL
        if not markdown_urls:
            plain_urls = re.findall(r"http[s]?://(?:[a-zA-Z]|[0-9]|[$-_@.&+=]|[!*\\(\\),]|(?:%[0-9a-fA-F][0-9a-fA-F]))+", content)
            # 过滤掉末尾可能被误抓的括号（如果是成对出现的 URL 内部括号除外，但通常末尾的是包装括号）
            urls = [u.rstrip(')') for u in plain_urls]
        else:
            urls = markdown_urls

        if urls:
            for i, url in enumerate(urls):
                try:
                    # 清理 URL (移除末尾可能的 Markdown 干扰字符)
                    url = url.split(')')[0] if ')' in url and '(' not in url else url
                    
                    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
                    else:
                        print(f"[-] Download failed (Status {img_resp.status_code}) for URL: {url}")
                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)

import json
import requests
import sys
from pathlib import Path

def test():
    config_path = Path(r"f:\Desktop\kaifa\jianying-editor-skill2\.agent\skills\antigravity-api-skill\libs\data\config.json")
    if not config_path.exists():
        print(f"[-] Config not found at {config_path}")
        return

    config = json.loads(config_path.read_text(encoding='utf-8'))
    base_url = config.get("base_url", "").rstrip("/")
    api_key = config.get("api_key", "sk-antigravity") # Default if placeholder
    
    print(f"[*] Testing Endpoint: {base_url}")
    print(f"[*] API Key: {api_key[:8]}...")

    payload = {
        "model": "banana",
        "messages": [{"role": "user", "content": "Generate a beautiful 4k image of a futuristic city"}],
        "size": "1024x1024",
        "stream": False
    }
    
    headers = {
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json"
    }
    
    try:
        print("[*] Sending request to /chat/completions...")
        # Use a shorter timeout for testing
        response = requests.post(
            f"{base_url}/chat/completions", 
            headers=headers, 
            json=payload, 
            timeout=180, # Image generation can be slow
            proxies={"http": None, "https": None} # Disable system proxies for localhost
        )
        print(f"[+] Status Code: {response.status_code}")
        print("\n[+] Response Body:")
        print(response.text)
        
        if response.status_code == 200:
            try:
                data = response.json()
                content = data.get("choices", [{}])[0].get("message", {}).get("content", "")
                print("\n[*] Extracted Content:")
                print(content)
            except Exception as eje:
                print(f"[-] JSON Parse Error: {eje}")
    except requests.exceptions.Timeout:
        print("[-] Request Timed Out (Manager might be processing or stuck)")
    except Exception as e:
        print(f"[-] Request failed: {e}")

if __name__ == "__main__":
    test()

import json
import requests
import sys
from pathlib import Path

def test_stream():
    config_path = Path(r"f:\Desktop\kaifa\jianying-editor-skill2\.agent\skills\antigravity-api-skill\libs\data\config.json")
    if not config_path.exists():
        print("[-] Config not found")
        return
        
    config = json.loads(config_path.read_text(encoding='utf-8'))
    base_url = config.get("base_url", "").rstrip("/")
    api_key = config.get("api_key", "YOUR_API_KEY_HERE")
    
    payload = {
        "model": "banana",
        "messages": [{"role": "user", "content": "Generate a beautiful 4k image of a futuristic city"}],
        "size": "1024x1024",
        "stream": True 
    }
    
    headers = {
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json"
    }
    
    print(f"[*] Starting Stream Test on {base_url}...")
    try:
        # 使用较长的超时时间，因为生图比较慢
        response = requests.post(
            f"{base_url}/chat/completions", 
            headers=headers, 
            json=payload, 
            timeout=180, 
            stream=True,
            proxies={"http": None, "https": None}
        )
        
        print(f"[+] Status Code: {response.status_code}")
        
        full_content = ""
        for line in response.iter_lines():
            if line:
                decoded_line = line.decode('utf-8')
                # 打印所有原始数据，不放过任何信息
                print(f"RAW: {decoded_line}")
                
                if decoded_line.startswith("data: "):
                    data_str = decoded_line[6:].strip()
                    if data_str == "[DONE]":
                        print("[*] Stream Ended with [DONE]")
                        break
                    try:
                        data_json = json.loads(data_str)
                        # 打印完整 JSON 结构，用于诊断非标准字段
                        print(f"DEBUG FULL DATA: {json.dumps(data_json)}")
                        
                        choice = data_json.get("choices", [{}])[0]
                        # 尝试从各种可能的地方获取文本
                        content = (choice.get("delta", {}).get("content", "") or 
                                  choice.get("message", {}).get("content", "") or
                                  data_json.get("url", "") or  # 尝试顶级字段
                                  data_json.get("image_url", "")) # 尝试顶级字段
                        full_content += content
                    except Exception as e:
                        pass
        
        print("\n--- Final Aggregated Content ---")
        if full_content:
            print(full_content)
        else:
            print("[!] Warning: Aggregated content is empty!")
            
    except Exception as e:
        print(f"[-] Stream failed: {e}")

if __name__ == "__main__":
    test_stream()

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

def test_video_analysis():
    config_path = Path(r"f:\Desktop\kaifa\jianying-editor-skill2\.agent\skills\antigravity-api-skill\libs\data\config.json")
    if not config_path.exists():
        print("[-] Config not found")
        return
        
    config = json.loads(config_path.read_text(encoding='utf-8'))
    base_url = config.get("base_url", "").rstrip("/")
    api_key = config.get("api_key", "YOUR_API_KEY_HERE")
    
    video_path = r"F:\Desktop\test_output.mp4"
    if not os.path.exists(video_path):
        print(f"[-] Video not found at {video_path}")
        return

    print(f"[*] Reading and encoding video: {video_path}")
    with open(video_path, "rb") as f:
        video_b64 = base64.b64encode(f.read()).decode("utf-8")
    
    # 按照用户给出的结构构造 Payload
    payload = {
      "model": "gemini-3-pro",
      "stream": True,
      "messages": [
        {
          "role": "user",
          "content": "请总结这个视频的内容"
        }
      ],
      "files": [
        {
          "filename": "test_output.mp4",
          "mime_type": "video/mp4",
          "file_data": f"data:video/mp4;base64,{video_b64}"
        }
      ]
    }
    
    headers = {
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json"
    }
    
    print(f"[*] Sending Video Analysis Request (Streaming) to {base_url}...")
    try:
        response = requests.post(
            f"{base_url}/chat/completions", 
            headers=headers, 
            json=payload, 
            timeout=300, 
            stream=True,
            proxies={"http": None, "https": None}
        )
        
        print(f"[+] Status Code: {response.status_code}")
        
        full_content = ""
        for line in response.iter_lines():
            if line:
                decoded_line = line.decode('utf-8').strip()
                if decoded_line.startswith("data: "):
                    data_str = decoded_line[6:].strip()
                    if data_str == "[DONE]":
                        print("\n[*] Stream Ended with [DONE]")
                        break
                    try:
                        data_json = json.loads(data_str)
                        delta = data_json.get("choices", [{}])[0].get("delta", {})
                        content = delta.get("content", "")
                        print(content, end="", flush=True)
                        full_content += content
                    except:
                        pass
        
        if not full_content:
            print("\n[!] Warning: No content returned from server.")
            
    except Exception as e:
        print(f"\n[-] Request failed: {e}")

if __name__ == "__main__":
    test_video_analysis()