318 lines
12 KiB
Python
318 lines
12 KiB
Python
# -*- 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)
|