# -*- coding: utf-8 -*- """ 图绘平台上传服务 将处理好的图片上传到图绘平台，返回图片 URL """ import os import httpx import logging import mimetypes from pathlib import Path from typing import Optional, Tuple from dotenv import load_dotenv logger = logging.getLogger(__name__) load_dotenv() # 图绘平台配置 TUHUI_BASE_URL = os.getenv("TUHUI_BASE_URL", "https://tuhui.cloud") TUHUI_FALLBACK_BASE_URL = "https://tuhui.cloud" TUHUI_PHONE = os.getenv("TUHUI_PHONE", "17520145271") # 图绘账号手机号 TUHUI_PASSWORD = os.getenv("TUHUI_PASSWORD", "zuowei1216") # 图绘账号密码 TUHUI_DEFAULT_PRICE = int(os.getenv("TUHUI_DEFAULT_PRICE", "20")) # 默认定价（元） TUHUI_DEFAULT_CATEGORY = os.getenv("TUHUI_DEFAULT_CATEGORY", "设计素材") class TuhuiUploadService: """图绘平台上传服务""" def __init__(self): self.base_url = TUHUI_BASE_URL.rstrip("/") self.base_urls = [] for candidate in (TUHUI_FALLBACK_BASE_URL.rstrip("/"), self.base_url): if candidate and candidate not in self.base_urls: self.base_urls.append(candidate) if self.base_urls: self.base_url = self.base_urls[0] self.phone = TUHUI_PHONE self.password = TUHUI_PASSWORD self.default_price = TUHUI_DEFAULT_PRICE self.access_token = None self.user_id = None @staticmethod def _build_api_url(base_url: str, path: str) -> str: normalized = path if path.startswith("/") else f"/{path}" if base_url.endswith("/api"): return f"{base_url}{normalized}" return f"{base_url}/api{normalized}" def _api_url(self, path: str) -> str: return self._build_api_url(self.base_url, path) @staticmethod def _guess_file_meta(image_path: str) -> tuple[str, str]: path = Path(image_path) filename = path.name or "image.jpg" mime_type, _ = mimetypes.guess_type(filename) return filename, mime_type or "application/octet-stream" async def login(self) -> bool: """登录图绘平台获取 token""" last_error = "" for base_url in self.base_urls: try: async with httpx.AsyncClient() as client: response = await client.post( self._build_api_url(base_url, "/auth/login"), json={ "phone": self.phone, "password": self.password }, timeout=10.0 ) if response.status_code == 200: data = response.json() self.access_token = data.get("access_token") user = data.get("user", {}) self.user_id = user.get("id") self.base_url = base_url logger.info(f"图绘平台登录成功，用户 ID: {self.user_id}，base={self.base_url}") return True last_error = f"{response.status_code} {response.text}" logger.warning(f"图绘平台登录失败，base={base_url}：{last_error}") except Exception as e: last_error = str(e) logger.warning(f"图绘平台登录异常，base={base_url}：{type(e).__name__}: {e!r}") logger.error(f"图绘平台登录失败：{last_error}") return False async def upload_image( self, image_path: str, title: str, description: str = "", price: Optional[int] = None, category: str = TUHUI_DEFAULT_CATEGORY, tags: str = "", ) -> Tuple[bool, str, int]: """ 上传图片到图绘平台 Args: image_path: 图片文件路径 title: 作品标题 description: 作品描述 price: 定价（元），默认使用 TUHUI_DEFAULT_PRICE category: 分类 Returns: (success, image_url, work_id) - success: 是否上传成功 - image_url: 图片 URL - work_id: 作品 ID """ try: # 如果 token 过期，重新登录 if not self.access_token: if not await self.login(): return False, "登录失败", 0 # 准备上传数据 price = price or self.default_price # 读取图片文件 if not os.path.exists(image_path): logger.error(f"图片文件不存在：{image_path}") return False, "文件不存在", 0 filename, mime_type = self._guess_file_meta(image_path) with open(image_path, "rb") as f: files = { "file": (filename, f, mime_type) } data = { "title": title, "description": description, "price": str(price), "category": category, } if tags: data["tags"] = tags headers = { "Authorization": f"Bearer {self.access_token}" } async with httpx.AsyncClient() as client: response = await client.post( self._api_url("/upload"), files=files, data=data, headers=headers, timeout=30.0 ) if response.status_code in [200, 201]: payload = response.json() if not payload.get("success", False): logger.error(f"图绘平台上传返回失败：{payload}") return False, payload.get("message", "上传失败"), 0 work_id = int(payload.get("work_id") or payload.get("work", {}).get("id") or 0) image_url = str(payload.get("image_url") or payload.get("work", {}).get("original_image") or "") logger.info(f"图绘平台上传成功，作品 ID: {work_id}, URL: {image_url}") return True, image_url, work_id else: logger.error(f"图绘平台上传失败：{response.status_code} {response.text}") # 如果 token 过期，尝试重新登录后再上传 if response.status_code == 401: logger.info("Token 可能过期，尝试重新登录...") self.access_token = None if await self.login(): # 重新上传 return await self.upload_image( image_path, title, description, price, category ) return False, f"上传失败：{response.text}", 0 except Exception as e: logger.error(f"图绘平台上传异常：{e}") return False, f"上传异常：{e}", 0 # 单例 _tuhui_service: Optional[TuhuiUploadService] = None def get_tuhui_service() -> TuhuiUploadService: """获取图绘上传服务单例""" global _tuhui_service if _tuhui_service is None: _tuhui_service = TuhuiUploadService() return _tuhui_service # 便捷函数 async def upload_to_tuhui( image_path: str, title: str, description: str = "", price: int = 20, category: str = TUHUI_DEFAULT_CATEGORY, tags: str = "", ) -> Tuple[bool, str, int]: """ 便捷函数：上传图片到图绘平台 Returns: (success, image_url, work_id) """ service = get_tuhui_service() return await service.upload_image(image_path, title, description, price, category, tags)