Update app.py

app.py

@@ -2,13 +2,15 @@ import gradio as gr
 import random
 import os
 import requests
+import base64
+from PIL import Image
+from io import BytesIO
 
 HF_TOKEN = os.environ.get("girlToken")
 
 API_BASE = "https://prithivmlmods-qwen-image-edit-2511-loras-fast.hf.space"
-UPLOAD_URL = f"{API_BASE}/gradio_api/upload"
-CALL_URL = f"{API_BASE}/gradio_api/call/v2/infer"
-POLL_URL = f"{API_BASE}/gradio_api/call/infer"  # need event_id
+INFER_URL = f"{API_BASE}/gradio/infer"  # Not used, for clarity
+NAMED_API_URL = f"{API_BASE}/gradio_api/call/v2/infer"
 
 LORA_STYLES = [
     'Multiple-Angles', 'Photo-to-Anime', 'Anime-V2', 'Light-Migration',
@@ -20,54 +22,66 @@ LORA_STYLES = [
 ]
 MAX_SEED = 2**31 - 1
 
-def upload_file(image_path):
-    files = {'files': (os.path.basename(image_path), open(image_path, "rb"), "application/octet-stream")}
-    headers = {'Authorization': f'Bearer {HF_TOKEN}'}
-    resp = requests.post(UPLOAD_URL, files=files, headers=headers)
-    resp.raise_for_status()
-    # 新增: 先检查返回内容类型，如果非JSON直接返回字符串（文件路径？）包为dict以兼容API异常返回
+def encode_image_file_to_b64_json(image_path):
     try:
-        res = resp.json()
-    except Exception as e:
-        # 检查是不是直接给了一个路径字符串
+        with open(image_path, "rb") as f:
+            image_bytes = f.read()
+        # 检查格式，如果不是 jpeg/png 尝试转码
         try:
-            raw_text = resp.text
-            # 可能是图片路径？直接包装为dict
-            if raw_text and raw_text.strip().startswith("/"):
-                return {"path": raw_text.strip()}
-            else:
-                raise RuntimeError("解析API响应的JSON失败且内容非预期: %s" % raw_text)
-        except Exception as e2:
-            raise RuntimeError("解析API响应的JSON失败: %s / 文本: %s" % (e, resp.text))
-
-    if isinstance(res, list) and len(res) > 0:
-        if isinstance(res[0], dict):
-            return res[0]
-        # 允许直接为字符串则强制包装
-        elif isinstance(res[0], str) and res[0].startswith("/"):
-            return {"path": res[0]}
-        else:
-            raise RuntimeError("返回结果不是字典类型: %s" % str(res[0]))
-    elif isinstance(res, dict):
-        return res
-    elif isinstance(res, str) and res.startswith("/"):
-        return {"path": res}
-    else:
-        # 返回格式不明, 直接异常
-        raise RuntimeError("API响应格式异常，期望是dict或list: %s" % str(res))
-
-def call_infer(payload):
-    headers = {'Authorization': f'Bearer {HF_TOKEN}'}
-    resp = requests.post(CALL_URL, json=payload, headers=headers)
+            img = Image.open(BytesIO(image_bytes))
+            buffered = BytesIO()
+            img.save(buffered, format="JPEG")
+            image_bytes = buffered.getvalue()
+        except Exception:
+            pass
+        im_b64 = base64.b64encode(image_bytes).decode("utf-8")
+        # 和 gradio_client 一致，图片用 json 数组包裹并带类型
+        payload = [
+            {
+                "data": im_b64,
+                "mime_type": "image/jpeg",
+                "orig_name": os.path.basename(image_path),
+            }
+        ]
+        import json
+        return json.dumps(payload)
+    except Exception as e:
+        raise RuntimeError(f"图片编码失败: {e}")
+
+def call_named_infer(
+    images_b64_json,
+    prompt,
+    lora_adapter,
+    seed,
+    randomize_seed,
+    guidance_scale,
+    steps
+):
+    headers = {
+        'Authorization': f'Bearer {HF_TOKEN}',
+        'Content-Type': 'application/json'
+    }
+    payload = {
+        "images_b64_json": images_b64_json,
+        "prompt": prompt,
+        "lora_adapter": lora_adapter,
+        "seed": int(seed),
+        "randomize_seed": bool(randomize_seed),
+        "guidance_scale": float(guidance_scale), # 注意类型
+        "steps": int(steps),
+    }
+    print("准备调用/infer:", payload)
+    resp = requests.post(NAMED_API_URL, json=payload, headers=headers)
     resp.raise_for_status()
+    # gradio api新版返回 event_id，然后/POLL 获取最终结果
     job = resp.json()
     event_id = job.get("event_id")
     return event_id
 
 def poll_infer(event_id):
-    url = f"{POLL_URL}/{event_id}"
+    # 必须用 gradio_api/call/infer/{event_id}
+    url = f"{API_BASE}/gradio_api/call/infer/{event_id}"
     headers = {'Authorization': f'Bearer {HF_TOKEN}'}
-    # 简单轮询直到拿到结果（可优化为更好策略）
     import time
     for _ in range(60):
         resp = requests.get(url, headers=headers)
@@ -77,7 +91,7 @@ def poll_infer(event_id):
             return result.get("data"), result.get("outputs")
         elif result.get("status") == "error":
             raise Exception(result.get("error"))
-        time.sleep(2)  # 2秒再次轮询
+        time.sleep(2)
     raise TimeoutError("等候API返回超时")
 
 def infer(
@@ -101,59 +115,45 @@ def infer(
     if randomize_seed:
         seed = random.randint(0, MAX_SEED)
 
-    # 1. 上传文件
     try:
-        uploaded_info = upload_file(image)
-        # 检查uploaded_info类型
-        if not isinstance(uploaded_info, dict) or "path" not in uploaded_info:
-            print(f"[图片上传] 失败: 返回结果无'path'字段: {uploaded_info}")
-            return None, seed
-        img_obj = {
-            "path": uploaded_info["path"],
-            "meta": {"_type": "gradio.FileData"},
-            "orig_name": os.path.basename(image)
-        }
+        images_b64_json = encode_image_file_to_b64_json(image)
     except Exception as e:
-        print(f"[图片上传] 失败: {e}")
+        print(f"[图片 base64编码失败] {e}")
         return None, seed
 
-    # 2. 准备参数并调用推理
-    payload = {
-        "images": [{"image": img_obj, "caption": None}],
-        "prompt": prompt,
-        "lora_adapter": lora_adapter,
-        "seed": int(seed),
-        "randomize_seed": bool(randomize_seed),
-        "guidance_scale": int(guidance_scale),
-        "steps": int(steps),
-    }
-
-    print("准备调用远端API:", payload)
     try:
-        event_id = call_infer(payload)
-        print("API返回event_id: ", event_id)
-        # 3. 结果轮询
+        event_id = call_named_infer(
+            images_b64_json,
+            prompt,
+            lora_adapter,
+            seed,
+            randomize_seed,
+            guidance_scale,
+            steps
+        )
+        print("API返回event_id:", event_id)
         data, outputs = poll_infer(event_id)
         print("[API 完成] data:", data, "outputs:", outputs)
-        # 解析输出（通常outputs可能就是结果图片的路径）
-        # 兼容dict或list
-        image_info = None
+        # 适配返回结构：outputs 是 dict（含 path/url/seed 等），或者直接 path
+        img_out = None
         seed_used = seed
         if outputs:
-            if isinstance(outputs, list) and len(outputs) == 2:
-                image_info, seed_used = outputs
-            elif isinstance(outputs, dict):
-                image_info = outputs.get("path") or outputs.get("url")
+            if isinstance(outputs, dict):
+                img_out = outputs.get("url") or outputs.get("path")
                 seed_used = outputs.get("seed", seed)
+            elif isinstance(outputs, str) and outputs.startswith("/"):
+                img_out = API_BASE + outputs
             else:
-                image_info = outputs
+                img_out = outputs
         elif data:
-            image_info = data
-        if isinstance(image_info, dict):
-            img_out = image_info.get("path") or image_info.get("url")
-        else:
-            img_out = image_info
-        # 补成绝对地址
+            if isinstance(data, dict):
+                img_out = data.get("url") or data.get("path")
+                seed_used = data.get("seed", seed)
+            elif isinstance(data, str) and data.startswith("/"):
+                img_out = API_BASE + data
+            else:
+                img_out = data
+        # url 补全为完整远端URL（部分 gradio 返回 path 不是http开头）
         if img_out and isinstance(img_out, str) and not img_out.startswith("http"):
             img_out = API_BASE + img_out
         return img_out, int(seed_used)
@@ -206,7 +206,7 @@ with gr.Blocks(css=css) as demo:
             randomize_seed = gr.Checkbox(label="随机化种子", value=True)
             guidance_scale = gr.Slider(
                 label="引导强度 (Guidance Scale)",
-                minimum=0.1,
+                minimum=1.0,
                 maximum=10.0,
                 step=0.1,
                 value=1.0,