Multimodal Gen

#!/usr/bin/env python3
"""
图像生成工具 - 调用性价比 API 的多模态模型
"""

import sys
import json
import base64
import subprocess
import re
from datetime import datetime
from pathlib import Path
import requests

# 配置
API_BASE = "https://your-provider.example.com/v1"
DEFAULT_MODEL = "gemini-3-pro-image-preview"
OUTPUT_DIR = Path.home() / "clawd" / "output" / "images"

# 可用模型
MODELS = {
    "gemini": "gemini-3-pro-image-preview",
    "flux": "flux-pro-max",
    "flux-ultra": "flux-pro-1.1-ultra",
    "imagen": "google/imagen-4-ultra",
    "dalle": "gpt-image-1",
    "kling": "kling-image",
    "seedream": "doubao-seedream-4-5-251128",
}

def get_api_key():
    """从 pass 获取 API key"""
    result = subprocess.run(["pass", "api/your-provider"], capture_output=True, text=True)
    return result.stdout.strip()

def generate_image(prompt: str, model: str = None, output_path: str = None) -> dict:
    """生成图片"""
    api_key = get_api_key()
    model = model or DEFAULT_MODEL
    
    # 支持别名
    if model in MODELS:
        model = MODELS[model]
    
    headers = {
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json"
    }
    
    data = {
        "model": model,
        "messages": [{"role": "user", "content": f"Generate an image: {prompt}"}]
    }
    
    try:
        response = requests.post(f"{API_BASE}/chat/completions", headers=headers, json=data, timeout=120)
        result = response.json()
    except Exception as e:
        return {"error": str(e)}
    
    if "error" in result:
        return {"error": result["error"]}
    
    # 提取图片
    content = result.get("choices", [{}])[0].get("message", {}).get("content", "")
    
    # 查找 base64 图片
    if "data:image" in content:
        match = re.search(r'data:image/(\w+);base64,([A-Za-z0-9+/=]+)', content)
        if match:
            img_format = match.group(1)
            img_data = match.group(2)
            
            # 保存图片
            OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
            timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
            
            if output_path:
                filepath = Path(output_path)
            else:
                filepath = OUTPUT_DIR / f"gen_{timestamp}.{img_format}"
            
            with open(filepath, "wb") as f:
                f.write(base64.b64decode(img_data))
            
            return {
                "success": True,
                "path": str(filepath),
                "format": img_format,
                "model": model,
                "prompt": prompt
            }
    
    # 检查是否有 inline_data (Gemini 格式)
    if isinstance(content, list):
        for part in content:
            if isinstance(part, dict) and "inline_data" in part:
                img_data = part["inline_data"].get("data", "")
                mime_type = part["inline_data"].get("mime_type", "image/jpeg")
                img_format = mime_type.split("/")[-1]
                
                OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
                timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
                filepath = OUTPUT_DIR / f"gen_{timestamp}.{img_format}"
                
                with open(filepath, "wb") as f:
                    f.write(base64.b64decode(img_data))
                
                return {
                    "success": True,
                    "path": str(filepath),
                    "format": img_format,
                    "model": model,
                    "prompt": prompt
                }
    
    # 检查是否有 URL 链接 (Flux 等模型返回 URL)
    url_match = re.search(r'https://[^\s\)]+\.(png|jpg|jpeg|webp)', content)
    if url_match:
        img_url = url_match.group(0)
        img_format = url_match.group(1)
        
        OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        
        if output_path:
            filepath = Path(output_path)
        else:
            filepath = OUTPUT_DIR / f"gen_{timestamp}.{img_format}"
        
        # 下载图片
        img_response = requests.get(img_url, timeout=60)
        with open(filepath, "wb") as f:
            f.write(img_response.content)
        
        return {
            "success": True,
            "path": str(filepath),
            "format": img_format,
            "model": model,
            "prompt": prompt,
            "url": img_url
        }
    
    # 检查 s3.ffire.cc 链接
    ffire_match = re.search(r'https://s3\.ffire\.cc/[^\s\)\]]+', content)
    if ffire_match:
        img_url = ffire_match.group(0)
        
        OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        
        if output_path:
            filepath = Path(output_path)
        else:
            filepath = OUTPUT_DIR / f"gen_{timestamp}.png"
        
        # 下载图片
        img_response = requests.get(img_url, timeout=60)
        with open(filepath, "wb") as f:
            f.write(img_response.content)
        
        return {
            "success": True,
            "path": str(filepath),
            "format": "png",
            "model": model,
            "prompt": prompt,
            "url": img_url
        }
    
    return {"error": "No image found in response", "raw": str(content)[:500]}

def main():
    if len(sys.argv) < 2:
        print("Usage: generate_image.py <prompt> [model] [output_path]")
        print(f"Available models: {', '.join(MODELS.keys())}")
        sys.exit(1)
    
    prompt = sys.argv[1]
    model = sys.argv[2] if len(sys.argv) > 2 and not sys.argv[2].startswith("/") else None
    output_path = sys.argv[-1] if len(sys.argv) > 2 and sys.argv[-1].startswith("/") else None
    
    result = generate_image(prompt, model, output_path)
    print(json.dumps(result, ensure_ascii=False, indent=2))

if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
视频生成工具 - 调用性价比 API 的视频生成模型（异步）
"""

import sys
import json
import subprocess
import time
from datetime import datetime
from pathlib import Path
import requests

API_BASE = "https://your-provider.example.com/v1"
DEFAULT_MODEL = "veo3.1"
OUTPUT_DIR = Path.home() / "clawd" / "output" / "videos"

# 可用模型
MODELS = {
    "veo3.1": "veo3.1",
    "veo3.1-4k": "veo3.1-4k",
    "veo3.1-pro": "veo3.1-pro",
    "veo3.1-pro-4k": "veo3.1-pro-4k",
    "veo3": "veo3",
    "sora2": "sora-2-all",
    "sora": "sora-2-all",
    "kling": "kling-video",
    "hailuo": "MiniMax-Hailuo-2.3",
    "runway": "runwayml-gen4_turbo-10",
    "grok": "grok-video-3",
}

def get_api_key():
    result = subprocess.run(["pass", "api/your-provider"], capture_output=True, text=True)
    return result.stdout.strip()

def submit_video_task(prompt: str, model: str = None) -> dict:
    """提交视频生成任务（异步）"""
    api_key = get_api_key()
    model_id = model or DEFAULT_MODEL
    
    if model_id in MODELS:
        model_id = MODELS[model_id]
    
    headers = {
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json"
    }
    
    data = {
        "model": model_id,
        "prompt": prompt
    }
    
    try:
        response = requests.post(f"{API_BASE}/video/generations", headers=headers, json=data, timeout=60)
        result = response.json()
        
        if result.get("code") == 500:
            return {"error": result.get("message", "服务器繁忙"), "status": "error"}
        
        task_id = result.get("data", {}).get("task_id") or result.get("task_id")
        if task_id:
            return {
                "status": "submitted",
                "task_id": task_id,
                "model": model_id,
                "prompt": prompt,
                "message": "视频生成任务已提交，请稍后查询结果"
            }
        
        return {"error": "未获取到任务ID", "raw": result}
    except Exception as e:
        return {"error": str(e)}

def query_video_task(task_id: str) -> dict:
    """查询视频生成任务状态"""
    api_key = get_api_key()
    
    headers = {
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json"
    }
    
    try:
        response = requests.get(f"{API_BASE}/video/generations/{task_id}", headers=headers, timeout=30)
        result = response.json()
        
        status = result.get("data", {}).get("task_status") or result.get("task_status")
        
        if status == "completed" or status == "succeed":
            video_url = result.get("data", {}).get("video_url") or result.get("video_url")
            if video_url:
                # 下载视频
                OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
                timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
                filepath = OUTPUT_DIR / f"gen_{timestamp}.mp4"
                
                video_response = requests.get(video_url, timeout=120)
                with open(filepath, "wb") as f:
                    f.write(video_response.content)
                
                return {
                    "status": "completed",
                    "task_id": task_id,
                    "path": str(filepath),
                    "url": video_url
                }
        
        return {
            "status": status or "unknown",
            "task_id": task_id,
            "raw": result
        }
    except Exception as e:
        return {"error": str(e), "task_id": task_id}

def generate_video(prompt: str, model: str = None, wait: bool = False, max_wait: int = 300) -> dict:
    """生成视频（可选等待完成）"""
    # 提交任务
    submit_result = submit_video_task(prompt, model)
    
    if submit_result.get("error"):
        return submit_result
    
    task_id = submit_result.get("task_id")
    if not task_id:
        return submit_result
    
    if not wait:
        return submit_result
    
    # 等待完成
    start_time = time.time()
    while time.time() - start_time < max_wait:
        time.sleep(10)  # 每10秒查询一次
        query_result = query_video_task(task_id)
        
        status = query_result.get("status")
        if status == "completed":
            return query_result
        elif status in ["failed", "error"]:
            return query_result
        
        print(f"状态: {status}, 已等待 {int(time.time() - start_time)}s...", file=sys.stderr)
    
    return {"status": "timeout", "task_id": task_id, "message": f"等待超时 ({max_wait}s)"}

def main():
    if len(sys.argv) < 2:
        print("Usage:")
        print("  generate_video.py submit <prompt> [model]  - 提交任务")
        print("  generate_video.py query <task_id>          - 查询任务")
        print("  generate_video.py <prompt> [model]         - 提交并等待")
        print(f"\nAvailable models: {', '.join(MODELS.keys())}")
        sys.exit(1)
    
    action = sys.argv[1]
    
    if action == "submit":
        prompt = sys.argv[2]
        model = sys.argv[3] if len(sys.argv) > 3 else None
        result = submit_video_task(prompt, model)
    elif action == "query":
        task_id = sys.argv[2]
        result = query_video_task(task_id)
    else:
        # 默认：提交并等待
        prompt = sys.argv[1]
        model = sys.argv[2] if len(sys.argv) > 2 else None
        result = generate_video(prompt, model, wait=True)
    
    print(json.dumps(result, ensure_ascii=False, indent=2))

if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
统一生成入口 - 自动优化 prompt 后调用生成模型
"""

import sys
import json
import subprocess
from pathlib import Path

SKILL_DIR = Path(__file__).parent

def run_script(script: str, *args) -> dict:
    """运行脚本并返回 JSON 结果"""
    cmd = ["python3", str(SKILL_DIR / script)] + list(args)
    result = subprocess.run(cmd, capture_output=True, text=True)
    try:
        return json.loads(result.stdout)
    except:
        return {"error": result.stderr or result.stdout}

def generate(prompt: str, media_type: str = "image", model: str = None, skip_optimize: bool = False) -> dict:
    """
    生成图像或视频
    1. 先优化 prompt
    2. 再调用生成模型
    """
    results = {"original_prompt": prompt}
    
    # Step 1: 优化 prompt
    if not skip_optimize:
        opt_result = run_script("prompt_optimizer.py", prompt, media_type)
        if opt_result.get("success"):
            optimized_prompt = opt_result["optimized"]
            results["optimized_prompt"] = optimized_prompt
            results["optimizer_model"] = opt_result.get("model")
        else:
            # 优化失败，使用原始 prompt
            optimized_prompt = prompt
            results["optimize_error"] = opt_result.get("error")
    else:
        optimized_prompt = prompt
    
    # Step 2: 生成
    if media_type == "image":
        model = model or "gemini"  # 默认 gemini
        gen_result = run_script("generate_image.py", optimized_prompt, model)
    else:  # video
        model = model or "veo3.1"  # 默认 veo3.1
        gen_result = run_script("generate_video.py", optimized_prompt, model)
    
    results.update(gen_result)
    return results

def main():
    if len(sys.argv) < 2:
        print("Usage: generate.py <prompt> [image|video] [model] [--no-optimize]")
        print("\nImage models: gemini (default), flux, flux-ultra, imagen, dalle, kling")
        print("Video models: veo3.1 (default), veo3, sora2, kling, hailuo")
        sys.exit(1)
    
    prompt = sys.argv[1]
    media_type = "image"
    model = None
    skip_optimize = "--no-optimize" in sys.argv
    
    for arg in sys.argv[2:]:
        if arg in ["image", "video"]:
            media_type = arg
        elif arg != "--no-optimize":
            model = arg
    
    result = generate(prompt, media_type, model, skip_optimize)
    print(json.dumps(result, ensure_ascii=False, indent=2))

if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Prompt 优化器 - 调用文本模型优化图像/视频生成提示词
"""

import sys
import json
import subprocess
import requests

API_BASE = "https://your-provider.example.com/v1"
OPTIMIZER_MODEL = "deepseek-v3.2"  # 性价比高，中文理解好

def get_api_key():
    result = subprocess.run(["pass", "api/your-provider"], capture_output=True, text=True)
    return result.stdout.strip()

def optimize_prompt(user_prompt: str, media_type: str = "image") -> dict:
    """优化用户提示词"""
    api_key = get_api_key()
    
    if media_type == "image":
        system_prompt = """你是一个专业的AI图像生成提示词优化专家。

用户会给你一个简单的图像描述，你需要将其优化为高质量的英文提示词。

优化原则：
1. 翻译成英文（如果是中文）
2. 添加艺术风格描述（如 digital art, oil painting, anime style 等）
3. 添加画面质量词（如 high quality, detailed, 4k, masterpiece）
4. 添加光影氛围描述（如 dramatic lighting, soft glow, golden hour）
5. 添加构图描述（如 close-up, wide shot, dynamic pose）
6. 保持核心主题不变
7. 避免敏感/违规内容：
   - 将"萝莉/loli/lolita"改为"young girl"或"little girl"或"child"
   - 避免任何可能被误解为不当内容的描述
   - 确保输出对所有AI图像生成模型都是安全的

只输出优化后的英文提示词，不要解释。"""
    else:  # video
        system_prompt = """你是一个专业的AI视频生成提示词优化专家。

用户会给你一个简单的视频描述，你需要将其优化为高质量的英文提示词。

优化原则：
1. 翻译成英文（如果是中文）
2. 描述动作和运动（如 walking slowly, camera panning, zooming in）
3. 添加时间/节奏描述（如 in slow motion, timelapse）
4. 添加环境氛围（如 cinematic, dramatic, peaceful）
5. 添加画面质量词（如 4K, high quality, professional）
6. 描述镜头运动（如 tracking shot, aerial view, first person）
7. 保持核心主题不变
8. 避免敏感/违规内容，如有需要进行适当修改

只输出优化后的英文提示词，不要解释。"""

    headers = {
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json"
    }
    
    data = {
        "model": OPTIMIZER_MODEL,
        "messages": [
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_prompt}
        ],
        "max_tokens": 500,
        "temperature": 0.7
    }
    
    try:
        response = requests.post(f"{API_BASE}/chat/completions", headers=headers, json=data, timeout=30)
        result = response.json()
        
        if "error" in result:
            return {"error": result["error"], "original": user_prompt}
        
        optimized = result["choices"][0]["message"]["content"].strip()
        
        return {
            "success": True,
            "original": user_prompt,
            "optimized": optimized,
            "model": OPTIMIZER_MODEL,
            "type": media_type
        }
    except Exception as e:
        return {"error": str(e), "original": user_prompt}

def main():
    if len(sys.argv) < 2:
        print("Usage: prompt_optimizer.py <prompt> [image|video]")
        sys.exit(1)
    
    prompt = sys.argv[1]
    media_type = sys.argv[2] if len(sys.argv) > 2 else "image"
    
    result = optimize_prompt(prompt, media_type)
    print(json.dumps(result, ensure_ascii=False, indent=2))

if __name__ == "__main__":
    main()