# -*- 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 urllib.parse import urlparse from dotenv import load_dotenv logger = logging.getLogger(__name__) load_dotenv() # 图绘平台配置 TUHUI_BASE_URL = os.getenv("TUHUI_BASE_URL", "https://aidg168.uk") TUHUI_FALLBACK_BASE_URL = "https://aidg168.uk" TUHUI_DIRECT_BASE_URL = os.getenv("TUHUI_DIRECT_BASE_URL", "http://156.226.181.204:8002") TUHUI_WEB_BASE_URL = os.getenv("TUHUI_WEB_BASE_URL", "https://aidg168.uk").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("/"), TUHUI_DIRECT_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 _normalize_asset_url(raw_url: str) -> str: url = str(raw_url or "").strip() if not url: return "" if url.startswith("/"): return f"{TUHUI_WEB_BASE_URL}{url}" parsed = urlparse(url) if not parsed.scheme or not parsed.netloc: return url host = (parsed.netloc or "").lower() if host in { "tuhui.cloud", "www.tuhui.cloud", "aidg168.uk", "www.aidg168.uk", "156.226.181.204:8002", "1.12.50.92:8002", "127.0.0.1:8002", }: path = parsed.path or "" if parsed.query: path = f"{path}?{parsed.query}" if parsed.fragment: path = f"{path}#{parsed.fragment}" return f"{TUHUI_WEB_BASE_URL}{path}" return url @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 = self.default_price if price is None else 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 = self._normalize_asset_url( payload.get("image_url") or payload.get("work", {}).get("original_image") or "" ) thumbnail_url = self._normalize_asset_url( payload.get("thumbnail_url") or payload.get("work", {}).get("thumbnail_image") or "" ) watermarked_url = self._normalize_asset_url( 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)