import requests
|
import urllib.parse
|
import datetime
|
import random
|
import os
|
from io import BytesIO
|
|
from auto_media_publisher.config.conf_base import (
|
TRANSLATE_API_TEMPLATE,
|
POLLINATIONS_API_TEMPLATE,
|
POLLINATIONS_DEFAULT_WIDTH ,
|
POLLINATIONS_DEFAULT_HEIGHT ,
|
POLLINATIONS_DEFAULT_MODEL,
|
)
|
from .utils_images import resize_and_crop_center_any
|
|
|
class Text2ImageGenerator:
|
def __init__(self, width=POLLINATIONS_DEFAULT_WIDTH, height=POLLINATIONS_DEFAULT_HEIGHT, model=POLLINATIONS_DEFAULT_MODEL):
|
self.width = width
|
self.height = height
|
self.model = model
|
|
def translate_prompt(self, prompt: str) -> str:
|
"""
|
使用 MyMemory API 将中文提示词翻译为英文，失败则返回原文。
|
"""
|
|
encoded_prompt = urllib.parse.quote(prompt)
|
api_url = TRANSLATE_API_TEMPLATE.format(prompt=encoded_prompt)
|
try:
|
res = requests.get(api_url, timeout=30)
|
res.raise_for_status()
|
data = res.json()
|
if data.get("responseStatus") == 200 and "translatedText" in data.get("responseData", {}):
|
translated = data["responseData"]["translatedText"]
|
print(f"[✓] 翻译成功: {translated}")
|
return translated
|
except Exception as e:
|
print(f"[!] 翻译失败，使用原文: {e}")
|
return prompt
|
|
def build_url(self, prompt: str) -> str:
|
"""
|
构建 Pollinations 图片生成 URL。
|
"""
|
seed = random.randint(1, 1_000_000)
|
encoded_prompt = urllib.parse.quote(prompt)
|
return POLLINATIONS_API_TEMPLATE.format(
|
prompt=encoded_prompt,
|
width=self.width,
|
height=self.height,
|
seed=seed,
|
model=self.model
|
)
|
|
def _fetch_with_retry(self, url: str, max_retries: int = 3):
|
"""
|
最多尝试 max_retries 次请求 URL，成功则返回 response.content，否则返回 None。
|
"""
|
for attempt in range(1, max_retries + 1):
|
try:
|
response = requests.get(url, timeout=30)
|
response.raise_for_status()
|
return resize_and_crop_center_any(response.content,self.width,self.height)
|
except Exception as e:
|
print(f"[!] 请求第 {attempt} 次失败：{e}")
|
return None
|
|
def generate_image_with_trans(self, prompt: str, filename: str = None) -> str:
|
if not prompt.strip():
|
raise ValueError("提示词不能为空")
|
|
translated_prompt = self.translate_prompt(prompt)
|
url = self.build_url(translated_prompt)
|
|
content = self._fetch_with_retry(url)
|
if content is None:
|
return None
|
|
if not filename:
|
now = datetime.datetime.now()
|
filename = now.strftime("%Y%m%d%H%M%S") + ".png"
|
if not filename.lower().endswith(".png"):
|
filename += ".png"
|
|
with open(filename, "wb") as f:
|
f.write(content)
|
|
abs_path = os.path.abspath(filename)
|
print(f"[✓] 图片已保存：{abs_path}")
|
return abs_path
|
|
def generate_image_without_trans(self, prompt: str, filename: str = None) -> str:
|
if not prompt.strip():
|
raise ValueError("提示词不能为空")
|
|
url = self.build_url(prompt)
|
print(url)
|
|
content = self._fetch_with_retry(url)
|
if content is None:
|
return None
|
|
if not filename:
|
now = datetime.datetime.now()
|
filename = now.strftime("%Y%m%d%H%M%S") + ".png"
|
if not filename.lower().endswith(".png"):
|
filename += ".png"
|
|
with open(filename, "wb") as f:
|
f.write(content)
|
|
abs_path = os.path.abspath(filename)
|
print(f"[✓] 图片已保存：{abs_path}")
|
return abs_path
|
|
def generate_image_with_trans_to_stream(self, prompt: str) -> BytesIO:
|
if not prompt.strip():
|
raise ValueError("提示词不能为空")
|
|
translated_prompt = self.translate_prompt(prompt)
|
url = self.build_url(translated_prompt)
|
|
content = self._fetch_with_retry(url)
|
if content is None:
|
return None
|
|
image_stream = BytesIO(content)
|
return image_stream
|
