from __future__ import annotations import os import tempfile import uuid from typing import Any import requests from .config import AUTO_DRAW_TIMEOUT_SECONDS async def auto_draw_preview( image_url: str, customer_id: str, requirement: str = "", ) -> dict[str, Any]: """ 统一自动作图入口(直调本地链路): 1) 下载客户图 2) 调 Gemini 生成 3) 上传图绘,返回可外发 URL """ try: from services.service_gemini import GeminiExtractV2Service # type: ignore from services.service_tuhui_upload import upload_to_tuhui # type: ignore except Exception as e: return {"ok": False, "error": f"import_failed:{e}"} prompt = requirement.strip() or "按原图做高清修复,保留主体细节,输出清晰可用版本" input_path = os.path.join(tempfile.gettempdir(), f"qjcs_in_{uuid.uuid4().hex}.jpg") output_path = os.path.join(tempfile.gettempdir(), f"qjcs_out_{uuid.uuid4().hex}.jpg") try: headers = { "User-Agent": ( "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " "AppleWebKit/537.36 (KHTML, like Gecko) " "Chrome/122.0.0.0 Safari/537.36" ), "Referer": "https://www.taobao.com/", "Accept": "image/avif,image/webp,image/apng,image/*,*/*;q=0.8", } resp = requests.get(image_url, headers=headers, timeout=AUTO_DRAW_TIMEOUT_SECONDS) if resp.status_code != 200: return {"ok": False, "error": f"download_http_{resp.status_code}"} with open(input_path, "wb") as f: f.write(resp.content) service = GeminiExtractV2Service() ok_extract, msg_extract, _ = await service.extract_pattern( input_path=input_path, output_path=output_path, custom_prompt=prompt, aspect_ratio="1:1", ) if not ok_extract: return {"ok": False, "error": f"extract_failed:{msg_extract}"} if not os.path.exists(output_path): return {"ok": False, "error": "extract_no_output_file"} ok_upload, link, _ = await upload_to_tuhui( output_path, title=f"客户{customer_id[-4:]}-预览图" if customer_id else "预览图", description="AI自动作图预览", price=1, ) if not ok_upload: return {"ok": False, "error": f"upload_failed:{link}"} return {"ok": True, "url": str(link)} except Exception as e: return {"ok": False, "error": str(e)} finally: try: if os.path.exists(input_path): os.remove(input_path) except Exception: pass try: if os.path.exists(output_path): os.remove(output_path) except Exception: pass