# -*- coding: utf-8 -*- """ 图绘平台上传服务 将处理好的图片上传到图绘平台，返回图片 URL """ import os import httpx import logging import mimetypes from dataclasses import dataclass from pathlib import Path from typing import Iterator, Optional 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_WEB_BASE_URL = os.getenv("TUHUI_WEB_BASE_URL", "https://tuhui.cloud").rstrip("/") 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", "设计素材") @dataclass class TuhuiUploadResult: """图绘上传结果。主返回 URL 为站内作品页，保留三元组解包兼容。""" success: bool download_url: str work_id: int image_url: str = "" thumbnail_url: str = "" watermarked_url: str = "" message: str = "" def __iter__(self) -> Iterator[object]: # 兼容历史调用：ok, download_url, work_id = result yield self.success yield self.download_url yield self.work_id def as_dict(self) -> dict: return { "success": self.success, "download_url": self.download_url, "work_id": self.work_id, "image_url": self.image_url, "thumbnail_url": self.thumbnail_url, "watermarked_url": self.watermarked_url, "message": self.message, } 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 _build_work_url(work_id: int) -> str: return f"{TUHUI_WEB_BASE_URL}/detail/{int(work_id)}" @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 = "", designer_name: str = "", ) -> TuhuiUploadResult: """ 上传图片到图绘平台 Args: image_path: 图片文件路径 title: 作品标题 description: 作品描述 price: 定价（元），默认使用 TUHUI_DEFAULT_PRICE category: 分类 Returns: TuhuiUploadResult - success: 是否上传成功 - download_url: 站内作品页地址 - image_url: 原图 URL（保留，便于需要时取用） - thumbnail_url: 缩略图 URL - watermarked_url: 水印图 URL - work_id: 作品 ID """ try: # 如果 token 过期，重新登录 if not self.access_token: if not await self.login(): return TuhuiUploadResult(False, "", 0, message="登录失败") # 准备上传数据 price = price or self.default_price # 读取图片文件 if not os.path.exists(image_path): logger.error(f"图片文件不存在：{image_path}") return TuhuiUploadResult(False, "", 0, message="文件不存在") 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 if designer_name: data["designer_name"] = str(designer_name).strip() 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 TuhuiUploadResult( False, "", 0, message=str(payload.get("message", "上传失败")), ) 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 "") thumbnail_url = str( payload.get("thumbnail_url") or payload.get("work", {}).get("thumbnail_image") or "" ) watermarked_url = str( payload.get("watermarked_url") or payload.get("work", {}).get("watermarked_image") or "" ) download_url = self._build_work_url(work_id) if work_id else "" logger.info( f"图绘平台上传成功，作品 ID: {work_id}, 站内地址: {download_url}, 原图: {image_url}" ) return TuhuiUploadResult( True, download_url, work_id, image_url=image_url, thumbnail_url=thumbnail_url, watermarked_url=watermarked_url, message=str(payload.get("message", "上传成功")), ) 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, tags, designer_name ) return TuhuiUploadResult(False, "", 0, message=f"上传失败：{response.text}") except Exception as e: logger.error(f"图绘平台上传异常：{e}") return TuhuiUploadResult(False, "", 0, message=f"上传异常：{e}") # 单例 _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 = "", designer_name: str = "", ) -> TuhuiUploadResult: """ 便捷函数：上传图片到图绘平台 Returns: TuhuiUploadResult """ service = get_tuhui_service() return await service.upload_image(image_path, title, description, price, category, tags, designer_name)