Auto commit: 2026-03-28 14:57:14

2026-03-28 14:57:14 +08:00
parent fefc7239eb
commit 5f7cdac9be
6 changed files with 652 additions and 0 deletions
@@ -0,0 +1,158 @@
+"""Popiart AI OpenAPI 公共模块：HTTP 请求、鉴权、图片保存等工具函数"""
+
+import base64
+import datetime
+import json
+import os
+import re
+import struct
+import sys
+import time
+from typing import Optional, Tuple
+import urllib.request
+import urllib.error
+
+# ========== 配置 ==========
+POPI_OPENAPI_URL = os.environ.get("POPI_OPENAPI_URL", "https://llmapitest.popi.art")
+API_KEY = os.environ.get("POPI_OPENAPI_KEY", "")
+
+if not API_KEY:
+    print("错误：请设置 POPI_OPENAPI_KEY 环境变量", file=sys.stderr)
+    print("  export POPI_OPENAPI_KEY=sk-xxxx", file=sys.stderr)
+    sys.exit(1)
+
+# 轮询配置
+POLL_INTERVAL = 30   # 秒
+MAX_WAIT_TIME = 600  # 秒
+# ==========================
+
+
+def headers(content_type="application/json"):
+    h = {"Authorization": f"Bearer {API_KEY}"}
+    if content_type:
+        h["Content-Type"] = content_type
+    return h
+
+
+def post_json(path: str, body: dict, timeout: int = 120) -> dict:
+    url = f"{POPI_OPENAPI_URL.rstrip('/')}{path}"
+    data = json.dumps(body).encode("utf-8")
+    req = urllib.request.Request(url, data=data, method="POST", headers=headers())
+    try:
+        with urllib.request.urlopen(req, timeout=timeout) as resp:
+            return json.loads(resp.read().decode("utf-8"))
+    except urllib.error.HTTPError as e:
+        body_text = e.read().decode("utf-8") if e.fp else ""
+        print(f"❌ HTTP {e.code}: {body_text}", file=sys.stderr)
+        sys.exit(1)
+    except urllib.error.URLError as e:
+        print(f"❌ 网络错误: {e.reason}", file=sys.stderr)
+        sys.exit(1)
+
+
+def get_json(path: str, timeout: int = 30) -> dict:
+    url = f"{POPI_OPENAPI_URL.rstrip('/')}{path}"
+    req = urllib.request.Request(url, method="GET", headers=headers(content_type=None))
+    try:
+        with urllib.request.urlopen(req, timeout=timeout) as resp:
+            return json.loads(resp.read().decode("utf-8"))
+    except urllib.error.HTTPError as e:
+        body_text = e.read().decode("utf-8") if e.fp else ""
+        print(f"❌ HTTP {e.code}: {body_text}", file=sys.stderr)
+        sys.exit(1)
+    except urllib.error.URLError as e:
+        print(f"❌ 网络错误: {e.reason}", file=sys.stderr)
+        sys.exit(1)
+
+
+def post_multipart(path: str, fields: dict, file_fields: list = None, timeout: int = 30) -> dict:
+    """
+    multipart/form-data POST。
+    fields: {name: value}
+    file_fields: [(field_name, filename, bytes, mime_type), ...] 或 None，支持多个同名字段
+    """
+    import requests as _requests
+    url = f"{POPI_OPENAPI_URL.rstrip('/')}{path}"
+    # requests 支持 list of tuples 以传递同名多字段
+    files = [(k, (None, v)) for k, v in fields.items()]
+    for fname, filename, data, mime in (file_fields or []):
+        files.append((fname, (filename, data, mime)))
+    resp = _requests.post(url, headers={"Authorization": f"Bearer {API_KEY}"}, files=files, timeout=timeout)
+    if resp.status_code != 200:
+        print(f"❌ HTTP {resp.status_code}: {resp.text}", file=sys.stderr)
+        sys.exit(1)
+    return resp.json()
+
+
+def download_stream(path: str, output_path: str, timeout: int = 120) -> bool:
+    """流式下载文件到 output_path，返回是否成功"""
+    import requests as _requests
+    url = f"{POPI_OPENAPI_URL.rstrip('/')}{path}"
+    resp = _requests.get(url, headers={"Authorization": f"Bearer {API_KEY}"}, stream=True, timeout=timeout)
+    if resp.status_code != 200:
+        print(f"❌ 下载失败 HTTP {resp.status_code}: {resp.text}", file=sys.stderr)
+        return False
+    total = int(resp.headers.get("content-length", 0))
+    downloaded = 0
+    with open(output_path, "wb") as f:
+        for chunk in resp.iter_content(chunk_size=8192):
+            if chunk:
+                f.write(chunk)
+                downloaded += len(chunk)
+                if total:
+                    print(f"\r   {downloaded/total*100:.1f}%  ({downloaded}/{total} bytes)", end="")
+    print()
+    return True
+
+
+def save_base64_image(b64_data: str, prefix: str = "image") -> str:
+    """将 base64 图片数据保存为 PNG 文件，返回文件路径"""
+    ts = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
+    filename = f"{prefix}_{ts}.png"
+    script_dir = os.path.dirname(os.path.abspath(__file__))
+    generation_dir = os.path.dirname(script_dir)
+    filepath = os.path.join(generation_dir,"generation", filename)
+    with open(filepath, "wb") as f:
+        f.write(base64.b64decode(b64_data))
+    # 输出超链接格式（使用 file:// 协议）
+    print(f"\n🔗 点击查看图片: file://{filepath}")
+    return filepath
+
+
+
+def extract_image_urls(markdown_text: str) -> list:
+    """从 markdown 文本中提取图片 URL"""
+    return re.findall(r'!\[.*?\]\((https?://[^)]+)\)', markdown_text)
+
+
+def poll_video(video_id: str) -> Optional[str]:
+    """
+    轮询视频生成状态，返回最终 task_id（用于下载），失败返回 None。
+    网关返回结构：{ code, data: { status, progress, result_url, ... } }
+    """
+    url = f"/v1/video/generations/{video_id}"
+    start = time.time()
+    while True:
+        elapsed = int(time.time() - start)
+        if elapsed > MAX_WAIT_TIME:
+            print(f"⏱️  超时：已等待 {MAX_WAIT_TIME}s")
+            return None
+        result = get_json(url)
+        data = result.get("data") or result
+        status = data.get("status", "")
+        progress = data.get("progress", 0)
+        print(f"  [{elapsed:>4}s] status={status}  progress={progress}%")
+        if status == "SUCCESS":
+            url_result = data.get("result_url", "")
+            if url_result:
+                print(f"  result_url: {url_result}")
+            return data.get("task_id") or video_id
+        elif status == "FAILURE":
+            print(f"❌ 生成失败: {data.get('fail_reason', '')}")
+            return None
+        print(f"  等待 {POLL_INTERVAL}s...")
+        time.sleep(POLL_INTERVAL)
+
+
+def ts() -> str:
+    return datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
@@ -0,0 +1,245 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+import sys
+import io
+sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8')
+sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding='utf-8')
+
+"""
+图像生成 skill 脚本
+用法:
+  python3 generate_image.py <model> --prompt TEXT [选项]
+
+模型:
+  seedream       SeeDream 文生图（OpenAI Images API）
+  nano-pro       nano-banana-pro 文生图 / 图生图（Gemini 原生 API）
+  nano-2         nano-banana-2 文生图 / 图生图（Gemini 原生 API）
+  sora           Sora Image 文生图 / 图生图（Chat Completions API）
+  不传则默认 seedream
+
+图生图触发条件:
+  nano-pro/nano-2 传 --image-path（本地图片路径）
+  sora            传 --image-list（一张或多张在线图片 URL）
+
+必填参数:
+  --prompt TEXT          图像生成提示词
+
+可选参数:
+  --image-path PATH ...  本地图片路径列表（nano-pro / nano-2 图生图，支持多张）
+  --image-list URL ...   原图 URL 列表（sora 图生图，支持多张）
+  --n N                  生成数量，>=1，>1 时循环调用（sora，默认 1）
+  --size SIZE            图片尺寸，如 2K / 4K （默认 2K）
+  --quality Q            图片质量，standard/hd（seedream，默认 hd）
+  --ratio W:H            图片比例，如 2:3（sora 文生图，默认 2:3）
+  --aspect-ratio W:H     宽高比，如 16:9（nano，默认 16:9）
+
+示例:
+  python3 generate_image.py seedream --prompt "一只猫"  ✅ 完成
+  python3 generate_image.py seedream --prompt "一只猫" --n 2    ✅ 完成（n未生效，只返回一张图）
+  python3 generate_image.py nano-pro --prompt "一只猫"  ✅ 完成
+  python3 generate_image.py nano-pro --prompt "改成油画风格" --image-path ./cat.jpg ✅ 完成
+  python3 generate_image.py nano-pro --prompt "融合两张图" --image-path ./a.jpg ./b.jpg ✅ 完成
+  python3 generate_image.py nano-2 --prompt "一只猫" --aspect-ratio 1:1 ✅ 完成
+  python3 generate_image.py nano-2 --prompt "改成油画风格" --image-path ./cat.jpg   ✅ 完成
+  python3 generate_image.py nano-2 --prompt "融合两张图" --image-path ./cat.jpg ./dog.jpg   ✅ 完成
+  python3 generate_image.py sora --prompt "一只柴犬"    ✅ 完成
+  python3 generate_image.py sora --prompt "一只柴犬" --n 2  ✅ 完成
+  python3 generate_image.py sora --prompt "把毛色改成彩虹色" --image-list https://example.com/dog.jpg    ✅ 完成
+  python3 generate_image.py sora --prompt "融合两张图" --image-list https://a.com/1.jpg https://b.com/2.jpg ✅ 完成
+"""
+
+import argparse
+import base64
+import os
+import sys
+
+sys.path.insert(0, os.path.dirname(__file__))
+from _common_new import post_json, save_base64_image, extract_image_urls, ts, POPI_OPENAPI_URL, API_KEY
+
+DEFAULT_MODEL = "seedream"
+
+
+# ---------- seedream ----------
+
+def gen_seedream_text(prompt: str, size: str = "2K", quality: str = "hd", n: int = 1):
+    print(f"模型: seedream 4.5  模式: 文生图  尺寸: {size}  质量: {quality}  数量: {n}")
+    result = post_json("/v1/images/generations", {
+        "model": "seedream-4-5-251128",
+        "prompt": prompt,
+        "size": size,
+        "quality": quality,
+        "n": n,
+        "response_format": "url",
+    }, timeout=120)
+    if "data" in result:
+        for item in result["data"]:
+            if "url" in item:
+                print(f"🔗 {item['url']}")
+                return item['url']
+    print(f"❌ {result}")
+    return False
+
+
+
+# ---------- gemini ----------
+
+def gen_gemini_text(prompt: str, model_name: str, aspect_ratio: str = "16:9", size: str = "2K"):
+    print(f"模型: {model_name}  模式: 文生图  宽高比: {aspect_ratio}  尺寸: {size}")
+    result = post_json(f"/v1beta/models/{model_name}:generateContent", {
+        "contents": [{"parts": [{"text": prompt}]}],
+        "generationConfig": {
+            "responseModalities": ["IMAGE"],
+            "imageConfig": {"aspectRatio": aspect_ratio, "imageSize": size},
+        },
+    }, timeout=300)
+    if "candidates" in result:
+        try:
+            b64 = result["candidates"][0]["content"]["parts"][0]["inlineData"]["data"]
+            path = save_base64_image(b64, prefix="gemini")
+            print(f"🖼️  已保存: {path}")
+            # return True
+            return path
+        except (KeyError, IndexError) as e:
+            print(f"❌ 解析响应失败: {e}")
+    print(f"❌ {result}")
+    return False
+
+
+def gen_gemini_image(image_paths: list, prompt: str, model_name: str,
+                     aspect_ratio: str = "16:9", size: str = "2K"):
+    print(f"模型: {model_name}  模式: 图生图  图片: {image_paths}")
+    parts = [{"text": prompt}]
+    for image_path in image_paths:
+        try:
+            with open(image_path, "rb") as f:
+                img_b64 = base64.b64encode(f.read()).decode()
+        except FileNotFoundError:
+            print(f"❌ 文件不存在: {image_path}")
+            return False
+        mime = "image/jpeg" if image_path.lower().endswith((".jpg", ".jpeg")) else "image/png"
+        parts.append({"inlineData": {"mimeType": mime, "data": img_b64}})
+    result = post_json(f"/v1beta/models/{model_name}:generateContent", {
+        "contents": [{"parts": parts}],
+        "generationConfig": {
+            "responseModalities": ["IMAGE"],
+            "imageConfig": {"aspectRatio": aspect_ratio, "imageSize": size},
+        },
+    }, timeout=300)
+    if "candidates" in result:
+        try:
+            b64 = result["candidates"][0]["content"]["parts"][0]["inlineData"]["data"]
+            path = save_base64_image(b64, prefix="gemini_edited")
+            print(f"🖼️  已保存: {path}")
+            return {"path": path}
+        except (KeyError, IndexError) as e:
+            print(f"❌ 解析响应失败: {e}")
+    print(f"❌ {result}")
+    return False
+
+
+# ---------- sora ----------
+
+def gen_sora(prompt: str, ratio: str = "2:3", image_list: list = None, n: int = 1):
+    """
+    sora_image 文生图 / 图生图统一入口。
+    - image_list 为空 → 文生图，prompt 末尾附加比例标记
+    - image_list 非空 → 图生图，每张图作为 image_url 传入
+    - n >= 1，大于 1 时循环调用接口生成对应数量图片
+    """
+    if n < 1:
+        print("❌ n 必须大于等于 1")
+        return False
+
+    image_list = image_list or []
+    mode = "图生图" if image_list else "文生图"
+    print(f"模型: sora_image  模式: {mode}  比例: {ratio}  数量: {n}")
+    if image_list:
+        print(f"原图: {image_list}")
+
+    def _build_messages():
+        if image_list:
+            content = [{"type": "text", "text": prompt}]
+            for url in image_list:
+                content.append({"type": "image_url", "image_url": {"url": url}})
+            return [{"role": "user", "content": content}]
+        else:
+            p = f"{prompt}【{ratio}】" if ratio in ("2:3", "3:2", "1:1") else prompt
+            return [{"role": "user", "content": p}]
+
+    success = 0
+    for i in range(n):
+        if n > 1:
+            print(f"\n[{i+1}/{n}]")
+        result = post_json("/v1/chat/completions", {
+            "model": "sora_image",
+            "messages": _build_messages(),
+        }, timeout=120)
+        if "choices" in result:
+            content = result["choices"][0]["message"]["content"]
+            urls = extract_image_urls(content)
+            if urls:
+                print(f"🔗 {urls[0]}")
+            else:
+                print(f"内容: {content[:200]}")
+            success += 1
+        else:
+            print(f"❌ {result}")
+
+    return success == n
+
+
+# ---------- 主入口 ----------
+
+def main():
+    parser = argparse.ArgumentParser(
+        description="图像生成 skill",
+        formatter_class=argparse.RawDescriptionHelpFormatter,
+        epilog=__doc__,
+    )
+    parser.add_argument("model", nargs="?", default=DEFAULT_MODEL,
+                        choices=["seedream", "nano-pro", "nano-2", "sora"],
+                        help=f"模型别名，默认 {DEFAULT_MODEL}")
+    parser.add_argument("--prompt", required=True, help="图像生成提示词（必填）")
+    parser.add_argument("--image-path", nargs="+", default=[], metavar="PATH",
+                        help="本地图片路径列表（nano-pro / nano-2 图生图，支持多张）")
+    parser.add_argument("--image-list", nargs="+", default=[], metavar="URL",
+                        help="原图 URL 列表（sora 图生图，支持多张）")
+    parser.add_argument("--n", type=int, default=1, help="生成数量，>=1（sora，默认 1）")
+    parser.add_argument("--size", default="2K", help="图片尺寸，如 2K / 4K / 1024x1024（默认 2K）")
+    parser.add_argument("--quality", default="hd", help="图片质量 standard/hd（seedream，默认 hd）")
+    parser.add_argument("--ratio", default="2:3", help="图片比例（sora 文生图，默认 2:3）")
+    parser.add_argument("--aspect-ratio", default="16:9", help="宽高比（gemini，默认 16:9）")
+    args = parser.parse_args()
+
+    model = args.model.lower()
+    prompt = args.prompt
+
+    print(f"⏰ {ts()}  POPI_OPENAPI_URL={POPI_OPENAPI_URL}")
+    print(f"提示词: {prompt}\n")
+
+    ok = False
+    if model == "seedream":
+        ok = gen_seedream_text(prompt, size=args.size, quality=args.quality, n=args.n)
+    elif model == "nano-pro":
+        if args.image_path:
+            ok = gen_gemini_image(args.image_path, prompt, "gemini-3-pro-image-preview",
+                                  aspect_ratio=args.aspect_ratio, size=args.size)
+        else:
+            ok = gen_gemini_text(prompt, "gemini-3-pro-image-preview",
+                                 aspect_ratio=args.aspect_ratio, size=args.size)
+    elif model == "nano-2":
+        if args.image_path:
+            ok = gen_gemini_image(args.image_path, prompt, "gemini-3.1-flash-image-preview",
+                                  aspect_ratio=args.aspect_ratio, size=args.size)
+        else:
+            ok = gen_gemini_text(prompt, "gemini-3.1-flash-image-preview",
+                                 aspect_ratio=args.aspect_ratio, size=args.size)
+    elif model == "sora":
+        ok = gen_sora(prompt, ratio=args.ratio, image_list=args.image_list, n=args.n)
+
+    print(f"\n{'✅ 完成' if ok else '❌ 失败'}  {ts()}")
+    sys.exit(0 if ok else 1)
+
+
+if __name__ == "__main__":
+    main()