'commit'

dsLightRag/Routes/QWenImageRoute.py

@@ -7,7 +7,7 @@ import uuid
 from typing import Optional
 
 from fastapi import APIRouter, HTTPException
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, Field, root_validator
 
 from Config.Config import OBS_PREFIX, OBS_BUCKET, OBS_SERVER
 from QWenImage.QWenImageEditKit import QwenImageEditor
@@ -36,10 +36,9 @@ class GenerateImageRequest(BaseModel):
 class EditImageRequest(BaseModel):
     """编辑图片请求模型"""
     prompt: str = Field(..., description="编辑提示词")
-    size: str = Field(default="1328*1328", description="图片尺寸")
     api_key: Optional[str] = Field(default=None, description="自定义API密钥")
-    # 支持URL或Base64两种格式，二选一
+    # 仅保留URL方式，设为必填项
-    image_url: Optional[str] = Field(default=None, description="原始图片URL")
+    image_url: str = Field(..., description="原始图片URL")
     image_base64: Optional[str] = Field(default=None, description="原始图片Base64编码")
 
     @root_validator(pre=True)
@@ -145,16 +144,16 @@ async def edit_image(request: EditImageRequest):
         dict: 包含编辑结果的字典
     """
     try:
+        # 恢复为仅使用image_url的日志记录
         logger.info(f"接收到图片编辑请求: image_url={request.image_url[:50]}..., prompt={request.prompt[:50]}...")
         
         # 如果提供了自定义API密钥，创建新的编辑器实例
         editor = QwenImageEditor(api_key=request.api_key) if request.api_key else image_editor
 
-        # 调用图片编辑API
+        # 使用image_url参数调用编辑接口
         result = editor.edit_image(
             image_url=request.image_url,
-            prompt=request.prompt,
+            prompt=request.prompt
-            negative_prompt=request.negative_prompt
         )
         
         # 处理结果

dsLightRag/Test/TestQWen3ImageEdit.py

@@ -16,99 +16,45 @@ logger = logging.getLogger("TestQWenImageEdit")
 base_url = "http://localhost:8200/api/qwenImage"
 
 
-def test_get_config():
+def test_edit_image(prompt):
-    """测试获取配置接口"""
     try:
-        url = f"{base_url}/config"
+        """测试编辑图片接口"""
-        logger.info(f"调用获取配置接口: {url}")
+        # 1. 定义测试图片URL
-        response = requests.get(url)
+        test_image_url = "https://help-static-aliyun-doc.aliyuncs.com/assets/img/zh-CN/7219965571/p1000025.png"
+        # 2. 定义API端点URL（修复核心问题）
+        api_url = f"{base_url}/edit"
         
-        if response.status_code == 200:
+        logger.info(f"调用编辑图片接口: {api_url}")
-            result = response.json()
+        logger.info(f"请求参数: prompt={prompt[:50]}...,  image_url={test_image_url[:50]}...")
-            logger.info(f"获取配置成功: {json.dumps(result, ensure_ascii=False, indent=2)}")
-            return True, result
-        else:
-            logger.error(f"获取配置失败: HTTP状态码={response.status_code}, 响应内容={response.text}")
-            return False, None
-    except Exception as e:
-        logger.exception(f"获取配置时发生异常: {str(e)}")
-        return False, None
-
-
-def get_test_image_base64(image_path=None):
-    """获取测试图片的base64编码"""
-    try:
-        # 默认使用测试目录下的示例图片
-        if not image_path:
-            test_dir = os.path.dirname(os.path.abspath(__file__))
-            image_path = os.path.join(test_dir, "test_image.jpg")
-            
-            # 如果默认图片不存在，创建一个简单的base64字符串
-            if not os.path.exists(image_path):
-                logger.warning(f"测试图片不存在，使用默认base64字符串: {image_path}")
-                return "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mNk+M9QDwADhgGAWjR9awAAAABJRU5ErkJggg=="
         
-        # 读取图片文件并转换为base64
+        # 构造请求数据
-        with open(image_path, "rb") as image_file:
+        payload = {
-            return base64.b64encode(image_file.read()).decode('utf-8')
-    except Exception as e:
-        logger.exception(f"获取图片base64编码时发生异常: {str(e)}")
-        return "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mNk+M9QDwADhgGAWjR9awAAAABJRU5ErkJggg=="
-
-
-def test_edit_image(prompt, image_base64=None, size='1328*1328', save_local=True):
-    """测试编辑图片接口（仅支持base64流式上传）"""
-    try:
-        url = f"{base_url}/edit"
-        headers = {"Content-Type": "application/json"}
-        
-        # 如果未提供base64，使用默认测试图片
-        if not image_base64:
-            image_base64 = get_test_image_base64()
-            logger.info("使用默认测试图片base64数据")
-        
-        data = {
             "prompt": prompt,
-            "image_base64": image_base64,
+            "image_url": test_image_url  # 正确传递图片URL参数
-            "size": size,
-            "save_local": save_local
         }
-        
-        logger.info(f"调用编辑图片接口: {url}")
-        logger.info(f"请求参数: prompt={prompt[:50]}..., size={size}, save_local={save_local}")
-        
         # 记录开始时间
-        start_time = time.time()
+        success_count=0
-        
         # 发送请求
-        response = requests.post(url, headers=headers, data=json.dumps(data, ensure_ascii=False))
+        # 发送请求到正确的API端点
-        
+        headers = {"Content-Type": "application/json"}
-        # 计算耗时
+        response = requests.post(api_url, headers=headers, data=json.dumps(payload, ensure_ascii=False))
-        elapsed_time = time.time() - start_time
+
-        logger.info(f"请求耗时: {elapsed_time:.2f}秒")
+        # 解析响应结果
-        
+        result = response.json()
-        if response.status_code == 200:
+        if result.get("code") == 200:
-            result = response.json()
+            # 修复计数逻辑 - 从响应数据判断成功状态
+            success_count += 1
             logger.info(f"编辑图片成功: {json.dumps(result, ensure_ascii=False, indent=2)}")
-            
+        else:
+            logger.error(f"编辑图片失败: {json.dumps(result, ensure_ascii=False)}")
+
             # 检查返回的数据
             if result.get("code") == 200 and "data" in result:
                 images = result["data"].get("images", [])
-                logger.info(f"成功编辑{len(images)}张图片")
+                logger.info(f"成功编辑{success_count}张图片")
-                
+
-                # 如果保存了本地文件，检查文件是否存在
-                if save_local and "local_file_paths" in result["data"]:
-                    for file_path in result["data"]["local_file_paths"]:
-                        full_path = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))), file_path.lstrip('.'))
-                        if os.path.exists(full_path):
-                            logger.info(f"本地文件已保存: {full_path}")
-                        else:
-                            logger.warning(f"本地文件不存在: {full_path}")
-                
             return True, result
-        else:
+
-            logger.error(f"编辑图片失败: HTTP状态码={response.status_code}, 响应内容={response.text}")
-            return False, None
     except Exception as e:
         logger.exception(f"编辑图片时发生异常: {str(e)}")
         return False, None
@@ -118,36 +64,13 @@ def main():
     """主函数，运行单元测试"""
     logger.info("===== 开始测试QWenImage编辑接口 ====")
     
-    # 1. 测试获取配置接口
-    logger.info("\n1. 测试获取配置接口")
-    config_success, config_data = test_get_config()
-    
     # 2. 测试编辑图片接口 - 基本测试
     logger.info("\n2. 测试编辑图片接口 - 基本测试")
-    basic_prompt = "将图片转换为水彩画风格，增加明亮度"
+    basic_prompt = "把“霓裳汉服社”改成“通义实验室”"
-    edit_success, edit_data = test_edit_image(
+    test_edit_image(
-        prompt=basic_prompt,
+        prompt=basic_prompt
-        size="1328*1328",
-        save_local=True
     )
-    
-    # 3. 测试编辑图片接口 - 不同参数
-    if config_success:
-        supported_sizes = config_data["data"].get("supported_sizes", ["1328*1328"])
 
-        logger.info(f"\n3. 测试编辑图片接口 - 不同参数(size={supported_sizes[0]})")
-        different_prompt = "添加下雪效果，转为冷色调"
-        test_edit_image(
-            prompt=different_prompt,
-            size=supported_sizes[0],
-            save_local=True
-        )
-    
-    logger.info("\n===== QWenImage编辑接口测试完成 =====")
-    
-    # 输出测试结果摘要
-    success_count = sum([config_success, edit_success])
-    logger.info(f"测试结果: {success_count} 接口测试成功")
 
 
 if __name__ == "__main__":