diff --git a/dsLightRag/QWenImage/QWenImageEditKit.py b/dsLightRag/QWenImage/QWenImageEditKit.py new file mode 100644 index 00000000..65369893 --- /dev/null +++ b/dsLightRag/QWenImage/QWenImageEditKit.py @@ -0,0 +1,167 @@ +from http import HTTPStatus +from urllib.parse import urlparse, unquote +from pathlib import PurePosixPath +import requests +from dashscope import MultiModalConversation +from Config.Config import ALY_LLM_API_KEY + +class QwenImageEditor: + """Qwen Image Editing Toolkit""" + def __init__(self, api_key=None): + """初始化图片编辑器 + + Args: + api_key: 阿里云API密钥，如果为None则使用配置文件中的密钥 + """ + self.api_key = api_key or ALY_LLM_API_KEY + self.model = "qwen-image-edit" + + def edit_image(self, image_url, prompt, negative_prompt="", stream=False, watermark=True): + """编辑图片 + + Args: + image_url: 原始图片URL + prompt: 编辑指令描述 + negative_prompt: 负面提示词 + stream: 是否流式返回 + watermark: 是否添加水印 + + Returns: + dict: 包含编辑结果的字典 + """ + try: + # 构建对话消息 + messages = [{ + "role": "user", + "content": [ + {"image": image_url}, + {"text": prompt} + ] + }] + + # 调用图片编辑API + response = MultiModalConversation.call( + api_key=self.api_key, + model=self.model, + messages=messages, + result_format='message', + stream=stream, + watermark=watermark, + negative_prompt=negative_prompt + ) + + # 处理响应结果 + if response.status_code == HTTPStatus.OK: + # 提取编辑后的图片URL + edited_image_url = response.output.choices[0].message.content[0].get("image") + + if not edited_image_url: + return { + 'success': False, + 'image_url': None, + 'error_msg': 'API返回空结果，图片编辑失败' + } + + return { + 'success': True, + 'image_url': edited_image_url, + 'error_msg': None + } + else: + error_msg = f'API调用失败: status_code={response.status_code}, code={response.code}, message={response.message}' + return { + 'success': False, + 'image_url': None, + 'error_msg': error_msg + } + except Exception as e: + return { + 'success': False, + 'image_url': None, + 'error_msg': f'发生异常: {str(e)}' + } + + def save_edited_image(self, image_url, save_dir='./'): + """保存编辑后的图片到本地 + + Args: + image_url: 编辑后的图片URL + save_dir: 保存目录 + + Returns: + dict: 包含保存结果的字典 + """ + try: + # 从URL解析文件名 + file_name = PurePosixPath(unquote(urlparse(image_url).path)).parts[-1] + save_path = f'{save_dir}{file_name}' + + # 下载并保存图片 + with open(save_path, 'wb+') as f: + f.write(requests.get(image_url).content) + + return { + 'success': True, + 'file_path': save_path, + 'error_msg': None + } + except Exception as e: + return { + 'success': False, + 'file_path': None, + 'error_msg': f'保存图片失败: {str(e)}' + } + + def edit_and_save_image(self, image_url, prompt, save_dir='./', negative_prompt=""): + """编辑图片并保存到本地 + + Args: + image_url: 原始图片URL + prompt: 编辑指令描述 + save_dir: 保存目录 + negative_prompt: 负面提示词 + + Returns: + dict: 包含编辑和保存结果的字典 + """ + # 编辑图片 + edit_result = self.edit_image(image_url, prompt, negative_prompt) + + if not edit_result['success']: + return edit_result + + # 保存编辑后的图片 + save_result = self.save_edited_image(edit_result['image_url'], save_dir) + + if save_result['success']: + return { + 'success': True, + 'image_url': edit_result['image_url'], + 'file_path': save_result['file_path'], + 'error_msg': None + } + else: + return { + 'success': False, + 'image_url': edit_result['image_url'], + 'file_path': None, + 'error_msg': save_result['error_msg'] + } + +# 示例使用 +if __name__ == "__main__": + # 创建编辑器实例 + editor = QwenImageEditor() + + # 示例参数 + image_url = "https://dashscope.oss-cn-beijing.aliyuncs.com/images/dog_and_girl.jpeg" + edit_prompt = "将图中的人物改为站立姿势，弯腰握住狗的前爪" + + print('----编辑图片，请等待任务执行----') + result = editor.edit_and_save_image(image_url, edit_prompt) + + if result['success']: + print(f'编辑成功，图片URL: {result["image_url"]}') + print(f'保存成功，文件路径: {result["file_path"]}') + else: + print(f'编辑失败: {result["error_msg"]}') diff --git a/dsLightRag/QWenImage/__pycache__/QWenImageEditKit.cpython-310.pyc b/dsLightRag/QWenImage/__pycache__/QWenImageEditKit.cpython-310.pyc new file mode 100644 index 00000000..58c6553e Binary files /dev/null and b/dsLightRag/QWenImage/__pycache__/QWenImageEditKit.cpython-310.pyc differ diff --git a/dsLightRag/Routes/QWenImageRoute.py b/dsLightRag/Routes/QWenImageRoute.py index c8d9a4b5..876cdcbd 100644 --- a/dsLightRag/Routes/QWenImageRoute.py +++ b/dsLightRag/Routes/QWenImageRoute.py @@ -10,7 +10,7 @@ from fastapi import APIRouter, HTTPException from pydantic import BaseModel, Field from Config.Config import OBS_PREFIX, OBS_BUCKET, OBS_SERVER -# 导入QwenImageGenerator类 +from QWenImage.QWenImageEditKit import QwenImageEditor from QWenImage.QwenImageKit import QwenImageGenerator from Util.ObsUtil import ObsUploader @@ -20,8 +20,9 @@ router = APIRouter(prefix="/api/qwenImage", tags=["千问生图"]) # 配置日志 logger = logging.getLogger(__name__) -# 初始化图片生成器 +# 初始化图片生成器和编辑器 image_generator = QwenImageGenerator() +image_editor = QwenImageEditor() class GenerateImageRequest(BaseModel): @@ -31,6 +32,23 @@ class GenerateImageRequest(BaseModel): size: str = Field(default="1328*1328", description="图片尺寸") api_key: Optional[str] = Field(default=None, description="自定义API密钥") +# 添加图片编辑请求模型 +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") + image_base64: Optional[str] = Field(default=None, description="原始图片Base64编码") + + @root_validator(pre=True) + def check_image_source(cls, values): + """验证图片来源，确保提供了URL或Base64中的一种""" + if not values.get("image_url") and not values.get("image_base64"): + raise ValueError("必须提供image_url或image_base64中的一种") + return values + @router.post("/generate") async def generate_image(request: GenerateImageRequest): @@ -116,6 +134,100 @@ async def generate_image(request: GenerateImageRequest): ) +@router.post("/edit") +async def edit_image(request: EditImageRequest): + """编辑图片API接口 + + Args: + request: 包含编辑图片参数的请求对象 + + Returns: + dict: 包含编辑结果的字典 + """ + try: + 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 + result = editor.edit_image( + image_url=request.image_url, + prompt=request.prompt, + negative_prompt=request.negative_prompt + ) + + # 处理结果 + if result['success']: + logger.info(f"图片编辑成功") + + # 上传到OBS + obs_urls = [] + uploader = ObsUploader() + + try: + # 下载编辑后的图片 + import requests + response_img = requests.get(result['image_url'], timeout=10) + response_img.raise_for_status() + bytes_data = response_img.content + + # 生成UUID文件名并上传 + jpg_file_name = f"{str(uuid.uuid4())}.jpg" + object_key = f"{OBS_PREFIX}/QWen3Image/edited/{jpg_file_name}" + success, upload_result = uploader.upload_base64_image(object_key, bytes_data) + + if success: + obs_url = f"https://{OBS_BUCKET}.{OBS_SERVER}/{object_key}" + obs_urls.append(obs_url) + logger.info(f"编辑图片上传OBS成功: {obs_url}") + else: + logger.error(f"编辑图片上传OBS失败: {upload_result}") + except Exception as e: + logger.error(f"处理编辑图片时出错: {str(e)}") + raise HTTPException( + status_code=500, + detail={ + "code": 500, + "message": "图片处理失败", + "error_detail": str(e) + } + ) + + # 构造返回响应 + response = { + "code": 200, + "message": "图片编辑成功", + "data": { + "original_image": request.image_url, + "edited_image": result['image_url'], + "obs_url": obs_urls[0] if obs_urls else None + } + } + return response + else: + logger.error(f"图片编辑失败: {result['error_msg']}") + raise HTTPException( + status_code=500, + detail={ + "code": 500, + "message": "图片编辑失败", + "error_detail": result['error_msg'] + } + ) + + except Exception as e: + logger.exception(f"处理图片编辑请求时发生异常: {str(e)}") + raise HTTPException( + status_code=500, + detail={ + "code": 500, + "message": "处理请求时发生异常", + "error_detail": str(e) + } + ) + + @router.get("/config") async def get_image_config(): """获取图片生成配置信息 diff --git a/dsLightRag/Routes/__pycache__/QWenImageRoute.cpython-310.pyc b/dsLightRag/Routes/__pycache__/QWenImageRoute.cpython-310.pyc index 4f2e2d94..1b64f4e6 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 new file mode 100644 index 00000000..5fce73ac --- /dev/null +++ b/dsLightRag/Test/TestQWen3ImageEdit.py @@ -0,0 +1,154 @@ +import requests +import json +import logging +import time +import os +import base64 + +# 配置日志 +logging.basicConfig( + level=logging.INFO, + format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' +) +logger = logging.getLogger("TestQWenImageEdit") + +# API基础URL +base_url = "http://localhost:8200/api/qwenImage" + + +def test_get_config(): + """测试获取配置接口""" + try: + url = f"{base_url}/config" + logger.info(f"调用获取配置接口: {url}") + response = requests.get(url) + + 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==" + + # 读取图片文件并转换为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 = { + "prompt": prompt, + "image_base64": image_base64, + "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() + + # 发送请求 + 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() + logger.info(f"编辑图片成功: {json.dumps(result, ensure_ascii=False, indent=2)}") + + # 检查返回的数据 + 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}") + + 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 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 + ) + + # 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__": + main() \ No newline at end of file