diff --git a/dsLightRag/Routes/QWenImageRoute.py b/dsLightRag/Routes/QWenImageRoute.py index 876cdcbd..6169eeeb 100644 --- a/dsLightRag/Routes/QWenImageRoute.py +++ b/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两种格式，二选一 - image_url: Optional[str] = Field(default=None, description="原始图片URL") + # 仅保留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, - negative_prompt=request.negative_prompt + prompt=request.prompt ) # 处理结果 diff --git a/dsLightRag/Routes/__pycache__/QWenImageRoute.cpython-310.pyc b/dsLightRag/Routes/__pycache__/QWenImageRoute.cpython-310.pyc index 1b64f4e6..e84167f8 100644 Binary files a/dsLightRag/Routes/__pycache__/QWenImageRoute.cpython-310.pyc and b/dsLightRag/Routes/__pycache__/QWenImageRoute.cpython-310.pyc differ diff --git a/dsLightRag/Test/TestQWen3ImageEdit.py b/dsLightRag/Test/TestQWen3ImageEdit.py index 5fce73ac..edc944f4 100644 --- a/dsLightRag/Test/TestQWen3ImageEdit.py +++ b/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}") - response = requests.get(url) + """测试编辑图片接口""" + # 1. 定义测试图片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: - result = response.json() - 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==" + logger.info(f"调用编辑图片接口: {api_url}") + logger.info(f"请求参数: prompt={prompt[:50]}..., image_url={test_image_url[:50]}...") - # 读取图片文件并转换为base64 - with open(image_path, "rb") as image_file: - 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 = { + # 构造请求数据 + payload = { "prompt": prompt, - "image_base64": image_base64, - "size": size, - "save_local": save_local + "image_url": test_image_url # 正确传递图片URL参数 } - - 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)) - - # 计算耗时 - elapsed_time = time.time() - start_time - logger.info(f"请求耗时: {elapsed_time:.2f}秒") - - if response.status_code == 200: - result = response.json() + # 发送请求到正确的API端点 + headers = {"Content-Type": "application/json"} + response = requests.post(api_url, headers=headers, data=json.dumps(payload, ensure_ascii=False)) + + # 解析响应结果 + result = response.json() + if result.get("code") == 200: + # 修复计数逻辑 - 从响应数据判断成功状态 + 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)}张图片") - - # 如果保存了本地文件，检查文件是否存在 - 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}") - + logger.info(f"成功编辑{success_count}张图片") + 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 = "将图片转换为水彩画风格，增加明亮度" - edit_success, edit_data = test_edit_image( - prompt=basic_prompt, - size="1328*1328", - save_local=True + basic_prompt = "把“霓裳汉服社”改成“通义实验室”" + test_edit_image( + prompt=basic_prompt ) - - # 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__":