feat(maizone/ai-image): 添加多提供商 AI 图像支持

此更改在 MaiZone 插件中引入了对多个 AI 图像生成提供商的强大支持,即 NovelAI 和 SiliconFlow。整个 AI 图像生成工作流程已被重新设计,以允许 LLM 为图像服务提供详细的提示,包括 NovelAI 的负面提示和纵横比。

重大更改:已移除本地图像发布功能。所有相关配置字段(`send.enable_image`、`send.image_number`、`send.image_directory`)已被移除。AI 图像生成配置已完全重建,并移动到新的专用部分(`ai_image`、`siliconflow`、`novelai`)。
This commit is contained in:
tt-P607
2025-12-16 15:16:56 +08:00
parent 1730a62363
commit 1c0f143225
5 changed files with 698 additions and 106 deletions

View File

@@ -43,19 +43,26 @@ class MaiZoneRefactoredPlugin(BasePlugin):
"plugin": {"enable": ConfigField(type=bool, default=True, description="是否启用插件")},
"models": {
"text_model": ConfigField(type=str, default="maizone", description="生成文本的模型名称"),
"siliconflow_apikey": ConfigField(type=str, default="", description="硅基流动AI生图API密钥"),
},
"ai_image": {
"enable_ai_image": ConfigField(type=bool, default=False, description="是否启用AI生成配图"),
"provider": ConfigField(type=str, default="siliconflow", description="AI生图服务提供商siliconflow/novelai"),
"image_number": ConfigField(type=int, default=1, description="生成图片数量1-4张"),
},
"siliconflow": {
"api_key": ConfigField(type=str, default="", description="硅基流动API密钥"),
},
"novelai": {
"api_key": ConfigField(type=str, default="", description="NovelAI官方API密钥"),
"character_prompt": ConfigField(type=str, default="", description="Bot角色外貌描述AI判断需要bot出镜时插入"),
"base_negative_prompt": ConfigField(type=str, default="nsfw, nude, explicit, sexual content, lowres, bad anatomy, bad hands, missing fingers, extra digit, fewer digits, cropped, worst quality, low quality", description="基础负面提示词(禁止不良内容)"),
"proxy_host": ConfigField(type=str, default="", description="代理服务器地址127.0.0.1"),
"proxy_port": ConfigField(type=int, default=0, description="代理服务器端口7890"),
},
"send": {
"permission": ConfigField(type=list, default=[], description="发送权限QQ号列表"),
"permission_type": ConfigField(type=str, default="whitelist", description="权限类型"),
"enable_image": ConfigField(type=bool, default=False, description="是否启用说说配图"),
"enable_ai_image": ConfigField(type=bool, default=False, description="是否启用AI生成配图"),
"enable_reply": ConfigField(type=bool, default=True, description="完成后是否回复"),
"ai_image_number": ConfigField(type=int, default=1, description="AI生成图片数量1-4张"),
"image_number": ConfigField(type=int, default=1, description="本地配图数量1-9张"),
"image_directory": ConfigField(
type=str, default=(Path(__file__).parent / "images").as_posix(), description="图片存储目录"
),
},
"read": {
"permission": ConfigField(type=list, default=[], description="阅读权限QQ号列表"),

View File

@@ -54,9 +54,10 @@ class ContentService:
logger.error("未配置LLM模型")
return ""
# 获取机器人信息
bot_personality = config_api.get_global_config("personality.personality_core", "一个机器人")
bot_expression = config_api.get_global_config("personality.reply_style", "内容积极向上")
# 获取机器人信息(核心人格配置)
bot_personality_core = config_api.get_global_config("personality.personality_core", "一个机器人")
bot_personality_side = config_api.get_global_config("personality.personality_side", "")
bot_reply_style = config_api.get_global_config("personality.reply_style", "内容积极向上")
qq_account = config_api.get_global_config("bot.qq_account", "")
# 获取当前时间信息
@@ -65,13 +66,20 @@ class ContentService:
weekday_names = ["星期一", "星期二", "星期三", "星期四", "星期五", "星期六", "星期日"]
weekday = weekday_names[now.weekday()]
# 构建人设描述
personality_desc = f"你的核心人格:{bot_personality_core}"
if bot_personality_side:
personality_desc += f"\n你的人格侧面:{bot_personality_side}"
personality_desc += f"\n\n你的表达方式:{bot_reply_style}"
# 构建提示词
prompt_topic = f"主题是'{topic}'" if topic else "主题不限"
prompt = f"""
你是'{bot_personality}',现在是{current_time}{weekday}),你想写一条{prompt_topic}的说说发表在qq空间上。
{bot_expression}
{personality_desc}
请严格遵守以下规则:
现在是{current_time}{weekday}),你想写一条{prompt_topic}的说说发表在qq空间上。
请严格遵守以下规则:
1. **绝对禁止**在说说中直接、完整地提及当前的年月日或几点几分。
2. 你应该将当前时间作为创作的背景,用它来判断现在是“清晨”、“傍晚”还是“深夜”。
3. 使用自然、模糊的词语来暗示时间,例如“刚刚”、“今天下午”、“夜深啦”等。
@@ -112,7 +120,244 @@ class ContentService:
logger.error(f"生成说说内容时发生异常: {e}")
return ""
async def generate_comment(self, content: str, target_name: str, rt_con: str = "", images: list = []) -> str:
async def generate_story_with_image_info(
self, topic: str, context: str | None = None
) -> tuple[str, dict]:
"""
生成说说内容并同时生成NovelAI图片提示词信息
:param topic: 说说的主题
:param context: 可选的聊天上下文
:return: (说说文本, 图片信息字典)
图片信息字典格式: {
"prompt": str, # NovelAI提示词英文
"negative_prompt": str, # 负面提示词(英文)
"include_character": bool, # 画面是否包含bot自己true时插入角色外貌提示词
"aspect_ratio": str # 画幅(方图/横图/竖图)
}
"""
try:
# 获取模型配置
models = llm_api.get_available_models()
text_model = str(self.get_config("models.text_model", "replyer"))
model_config = models.get(text_model)
if not model_config:
logger.error("未配置LLM模型")
return "", {"has_image": False}
# 获取机器人信息(核心人格配置)
bot_personality_core = config_api.get_global_config("personality.personality_core", "一个机器人")
bot_personality_side = config_api.get_global_config("personality.personality_side", "")
bot_reply_style = config_api.get_global_config("personality.reply_style", "内容积极向上")
qq_account = config_api.get_global_config("bot.qq_account", "")
# 获取角色外貌描述用于告知LLM
character_prompt = self.get_config("novelai.character_prompt", "")
# 获取当前时间信息
now = datetime.datetime.now()
current_time = now.strftime("%Y年%m月%d%H:%M")
weekday_names = ["星期一", "星期二", "星期三", "星期四", "星期五", "星期六", "星期日"]
weekday = weekday_names[now.weekday()]
# 构建提示词
prompt_topic = f"主题是'{topic}'" if topic else "主题不限"
# 构建人设描述
personality_desc = f"你的核心人格:{bot_personality_core}"
if bot_personality_side:
personality_desc += f"\n你的人格侧面:{bot_personality_side}"
personality_desc += f"\n\n你的表达方式:{bot_reply_style}"
# 检查是否启用AI配图统一开关
ai_image_enabled = self.get_config("ai_image.enable_ai_image", False)
provider = self.get_config("ai_image.provider", "siliconflow")
# NovelAI配图指引内置
novelai_guide = ""
output_format = '{"text": "说说正文内容"}'
if ai_image_enabled and provider == "novelai":
# 构建角色信息提示
character_info = ""
if character_prompt:
character_info = f"""
**角色特征锚点**当include_character=true时会插入以下基础特征
```
{character_prompt}
```
📌 重要说明:
- 这只是角色的**基础外貌特征**(发型、眼睛、耳朵等固定特征),用于锚定角色身份
- 你可以**自由描述**:衣服、动作、表情、姿势、装饰、配饰等所有可变元素
- 例如可以让角色穿不同风格的衣服casual, formal, sportswear, dress等
- 例如可以设计各种动作sitting, standing, walking, running, lying down等
- 例如可以搭配各种表情smile, laugh, serious, thinking, surprised等
- **鼓励创意**:根据说说内容自由发挥,让画面更丰富生动!
"""
novelai_guide = f"""
**配图说明:**
这条说说会使用NovelAI Diffusion模型二次元风格生成配图。
{character_info}
**提示词生成要求(非常重要):**
你需要生成一段详细的英文图片提示词,必须包含以下要素:
1. **画质标签**(必需):
- 开头必须加masterpiece, best quality, detailed, high resolution
2. **主体元素**(自由发挥):
- 人物描述:表情、动作、姿态(**完全自由**,不受角色锚点限制)
- 服装搭配casual clothing, dress, hoodie, school uniform, sportswear等**任意选择**
- 配饰装饰hat, glasses, ribbon, jewelry, bag等**随意添加**
- 物体/场景:具体的物品、建筑、自然景观等
3. **场景与环境**(必需):
- 地点indoor/outdoor, cafe, park, bedroom, street, beach, forest等
- 背景描述背景的细节sky, trees, buildings, ocean, mountains等
4. **氛围与风格**(必需):
- 光线sunlight, sunset, golden hour, soft lighting, dramatic lighting, night
- 天气/时间sunny day, rainy, cloudy, starry night, dawn, dusk
- 整体氛围peaceful, cozy, romantic, energetic, melancholic, playful
5. **色彩与细节**(推荐):
- 主色调warm colors, cool tones, pastel colors, vibrant colors
- 特殊细节falling petals, sparkles, lens flare, depth of field, bokeh
6. **include_character字段**
- true画面中包含"你自己"(自拍、你在画面中的场景)
- false画面中不包含你风景、物品、他人
7. **negative_prompt负面提示词**
- **严格禁止**以下内容nsfw, nude, explicit, sexual content, violence, gore, blood
- 排除质量问题lowres, bad anatomy, bad hands, deformed, mutilated, ugly
- 排除瑕疵blurry, poorly drawn, worst quality, low quality, jpeg artifacts
- 可以自行补充其他不需要的元素
8. **aspect_ratio画幅**
- 方图:适合头像、特写、正方形构图
- 横图:适合风景、全景、宽幅场景
- 竖图:适合人物全身、纵向构图
**内容审核规则(必须遵守)**
- 🚫 严禁生成NSFW、色情、裸露、性暗示内容
- 🚫 严禁生成暴力、血腥、恐怖、惊悚内容
- 🚫 严禁生成肢体畸形、器官变异、恶心画面
- ✅ 提示词必须符合健康、积极、美好的审美标准
- ✅ 专注于日常生活、自然风景、温馨场景等正面内容
**创意自由度**
- 💡 **衣服搭配**:可以自由设计各种服装风格(休闲、正式、运动、可爱、时尚等)
- 💡 **动作姿势**:站、坐、躺、走、跑、跳、伸展等任意动作
- 💡 **表情情绪**:微笑、大笑、思考、惊讶、温柔、调皮等丰富表情
- 💡 **场景创意**:根据说说内容自由发挥,让画面更贴合心情和主题
**示例提示词(展示多样性)**
- 休闲风:"masterpiece, best quality, 1girl, casual clothing, white t-shirt, jeans, sitting on bench, outdoor park, reading book, afternoon sunlight, relaxed atmosphere"
- 运动风:"masterpiece, best quality, 1girl, sportswear, running in park, energetic, morning light, trees background, dynamic pose, healthy lifestyle"
- 咖啡馆:"masterpiece, best quality, 1girl, sitting in cozy cafe, holding coffee cup, warm lighting, wooden table, books beside, peaceful atmosphere"
"""
output_format = '''{"text": "说说正文内容", "image": {"prompt": "详细的英文提示词(包含画质+主体+场景+氛围+光线+色彩)", "negative_prompt": "负面词", "include_character": true/false, "aspect_ratio": "方图/横图/竖图"}}'''
elif ai_image_enabled and provider == "siliconflow":
novelai_guide = """
**配图说明:**
这条说说会使用AI生成配图。
**提示词生成要求(非常重要):**
你需要生成一段详细的英文图片描述,必须包含以下要素:
1. **主体内容**:画面的核心元素(人物/物体/场景)
2. **具体场景**:地点、环境、背景细节
3. **氛围与风格**:整体感觉、光线、天气、色调
4. **细节描述**:补充的视觉细节(动作、表情、装饰等)
**示例提示词**
- "a girl sitting in a modern cafe, warm afternoon lighting, wooden furniture, coffee cup on table, books beside her, cozy and peaceful atmosphere, soft focus background"
- "sunset over the calm ocean, golden hour, orange and purple sky, gentle waves, peaceful and serene mood, wide angle view"
- "cherry blossoms in spring, soft pink petals falling, blue sky, sunlight filtering through branches, peaceful park scene, gentle breeze"
"""
output_format = '''{"text": "说说正文内容", "image": {"prompt": "详细的英文描述(主体+场景+氛围+光线+细节)"}}'''
prompt = f"""
{personality_desc}
现在是{current_time}{weekday}),你想写一条{prompt_topic}的说说发表在qq空间上。
**说说文本规则:**
1. **绝对禁止**在说说中直接、完整地提及当前的年月日或几点几分。
2. 你应该将当前时间作为创作的背景,用它来判断现在是"清晨""傍晚"还是"深夜"
3. 使用自然、模糊的词语来暗示时间,例如"刚刚""今天下午""夜深啦"等。
4. **内容简短**总长度严格控制在100字以内。
5. **禁止表情**严禁使用任何Emoji表情符号。
6. **严禁重复**:下方会提供你最近发过的说说历史,你必须创作一条全新的、与历史记录内容和主题都不同的说说。
7. 不要刻意突出自身学科背景,不要浮夸,不要夸张修辞。
{novelai_guide}
**输出格式JSON**
{output_format}
只输出JSON格式不要有其他内容。
"""
# 如果有上下文则加入到prompt中
if context:
prompt += f"\n\n作为参考,这里有一些最近的聊天记录:\n---\n{context}\n---"
# 添加历史记录以避免重复
prompt += "\n\n---历史说说记录---\n"
history_block = await get_send_history(qq_account)
if history_block:
prompt += history_block
# 调用LLM生成内容
success, response, _, _ = await llm_api.generate_with_model(
prompt=prompt,
model_config=model_config,
request_type="story.generate_with_image",
temperature=0.3,
max_tokens=1500,
)
if success:
# 解析JSON响应
import json5
try:
# 提取JSON部分去除可能的markdown代码块标记
json_text = response.strip()
if json_text.startswith("```json"):
json_text = json_text[7:]
if json_text.startswith("```"):
json_text = json_text[3:]
if json_text.endswith("```"):
json_text = json_text[:-3]
json_text = json_text.strip()
data = json5.loads(json_text)
story_text = data.get("text", "")
image_info = data.get("image", {})
# 确保图片信息完整
if not isinstance(image_info, dict):
image_info = {}
logger.info(f"成功生成说说:'{story_text}'")
logger.info(f"配图信息: {image_info}")
return story_text, image_info
except Exception as e:
logger.error(f"解析JSON失败: {e}, 原始响应: {response[:200]}")
# 降级处理:只返回文本,空配图信息
return response, {}
else:
logger.error("生成说说内容失败")
return "", {}
except Exception as e:
logger.error(f"生成说说内容时发生异常: {e}")
return "", {}
"""
针对一条具体的说说内容生成评论。
"""

View File

@@ -31,18 +31,48 @@ class ImageService:
"""
self.get_config = get_config
async def generate_image_from_prompt(self, prompt: str, save_dir: str | None = None) -> tuple[bool, Path | None]:
"""
直接使用提示词生成图片(硅基流动)
:param prompt: 图片提示词(英文)
:param save_dir: 图片保存目录None使用默认
:return: (是否成功, 图片路径)
"""
try:
api_key = str(self.get_config("siliconflow.api_key", ""))
image_num = self.get_config("ai_image.image_number", 1)
if not api_key:
logger.warning("硅基流动API未配置跳过图片生成")
return False, None
# 图片目录
if save_dir:
image_dir = Path(save_dir)
else:
plugin_dir = Path(__file__).parent.parent
image_dir = plugin_dir / "images"
image_dir.mkdir(parents=True, exist_ok=True)
logger.info(f"正在生成 {image_num} 张AI配图...")
success, img_path = await self._call_siliconflow_api(api_key, prompt, str(image_dir), image_num)
return success, img_path
except Exception as e:
logger.error(f"生成AI配图时发生异常: {e}")
return False, None
async def generate_images_for_story(self, story: str) -> bool:
"""
根据说说内容判断是否需要生成AI配图并执行生成任务。
根据说说内容判断是否需要生成AI配图并执行生成任务(硅基流动)
:param story: 说说内容。
:return: 图片是否成功生成(或不需要生成)。
"""
try:
enable_ai_image = bool(self.get_config("send.enable_ai_image", False))
api_key = str(self.get_config("models.siliconflow_apikey", ""))
image_dir = str(self.get_config("send.image_directory", "./data/plugins/maizone_refactored/images"))
image_num_raw = self.get_config("send.ai_image_number", 1)
api_key = str(self.get_config("siliconflow.api_key", ""))
image_num_raw = self.get_config("ai_image.image_number", 1)
# 安全地处理图片数量配置并限制在API允许的范围内
try:
@@ -52,15 +82,14 @@ class ImageService:
logger.warning(f"无效的图片数量配置: {image_num_raw}使用默认值1")
image_num = 1
if not enable_ai_image:
return True # 未启用AI配图视为成功
if not api_key:
logger.error("启用了AI配图但未填写SiliconFlow API密钥")
return False
logger.warning("硅基流动API未配置跳过图片生成")
return True
# 确保图片目录存在
Path(image_dir).mkdir(parents=True, exist_ok=True)
# 图片目录(使用统一配置)
plugin_dir = Path(__file__).parent.parent
image_dir = plugin_dir / "images"
image_dir.mkdir(parents=True, exist_ok=True)
# 生成图片提示词
image_prompt = await self._generate_image_prompt(story)
@@ -69,7 +98,8 @@ class ImageService:
return False
logger.info(f"正在为说说生成 {image_num} 张AI配图...")
return await self._call_siliconflow_api(api_key, image_prompt, image_dir, image_num)
success, _ = await self._call_siliconflow_api(api_key, image_prompt, str(image_dir), image_num)
return success
except Exception as e:
logger.error(f"处理AI配图时发生异常: {e}")
@@ -127,7 +157,7 @@ class ImageService:
logger.error(f"生成图片提示词时发生异常: {e}")
return ""
async def _call_siliconflow_api(self, api_key: str, image_prompt: str, image_dir: str, batch_size: int) -> bool:
async def _call_siliconflow_api(self, api_key: str, image_prompt: str, image_dir: str, batch_size: int) -> tuple[bool, Path | None]:
"""
调用硅基流动SiliconFlow的API来生成图片。
@@ -135,7 +165,7 @@ class ImageService:
:param image_prompt: 用于生成图片的提示词。
:param image_dir: 图片保存目录。
:param batch_size: 生成图片的数量1-4
:return: API调用是否成功
:return: (API调用是否成功, 第一张图片路径)
"""
url = "https://api.siliconflow.cn/v1/images/generations"
headers = {
@@ -175,12 +205,13 @@ class ImageService:
error_text = await response.text()
logger.error(f"生成图片出错,错误码[{response.status}]")
logger.error(f"错误响应: {error_text}")
return False
return False, None
json_data = await response.json()
image_urls = [img["url"] for img in json_data["images"]]
success_count = 0
first_img_path = None
# 下载并保存图片
for i, img_url in enumerate(image_urls):
try:
@@ -194,7 +225,7 @@ class ImageService:
image = Image.open(BytesIO(img_data))
# 保存图片为PNG格式确保兼容性
filename = f"image_{i}.png"
filename = f"siliconflow_{i}.png"
save_path = Path(image_dir) / filename
# 转换为RGB模式如果必要避免RGBA等模式的问题
@@ -206,21 +237,25 @@ class ImageService:
image.save(save_path, format="PNG")
logger.info(f"图片已保存至: {save_path}")
success_count += 1
# 记录第一张图片路径
if first_img_path is None:
first_img_path = save_path
except Exception as e:
logger.error(f"处理图片失败: {e!s}")
continue
except Exception as e:
logger.error(f"下载{i+1}图片失败: {e!s}")
logger.error(f"下载图片失败: {e!s}")
continue
# 只要至少有一张图片成功就返回True
return success_count > 0
# 至少有一张图片成功就返回True
return success_count > 0, first_img_path
except Exception as e:
logger.error(f"调用AI生图API时发生异常: {e}")
return False
return False, None
def _encode_image_to_base64(self, img: Image.Image) -> str:
"""

View File

@@ -0,0 +1,286 @@
"""
NovelAI图片生成服务 - 空间插件专用
独立实现,不依赖其他插件
"""
import asyncio
import base64
import random
import uuid
import zipfile
import io
from pathlib import Path
from typing import Optional
import aiohttp
from PIL import Image
from src.common.logger import get_logger
logger = get_logger("MaiZone.NovelAIService")
class MaiZoneNovelAIService:
"""空间插件的NovelAI图片生成服务独立实现"""
def __init__(self, get_config):
self.get_config = get_config
# NovelAI配置
self.api_key = self.get_config("novelai.api_key", "")
self.base_url = "https://image.novelai.net/ai/generate-image"
self.model = "nai-diffusion-4-5-full"
# 代理配置
proxy_host = self.get_config("novelai.proxy_host", "")
proxy_port = self.get_config("novelai.proxy_port", 0)
self.proxy = f"http://{proxy_host}:{proxy_port}" if proxy_host and proxy_port else ""
# 生成参数
self.steps = 28
self.scale = 5.0
self.sampler = "k_euler"
self.noise_schedule = "karras"
# 角色提示词当LLM决定包含角色时使用
self.character_prompt = self.get_config("novelai.character_prompt", "")
self.base_negative_prompt = self.get_config("novelai.base_negative_prompt", "nsfw, nude, explicit, sexual content, lowres, bad anatomy, bad hands")
# 图片保存目录(使用统一配置)
plugin_dir = Path(__file__).parent.parent
self.image_dir = plugin_dir / "images"
self.image_dir.mkdir(parents=True, exist_ok=True)
if self.api_key:
logger.info(f"NovelAI图片生成已配置模型: {self.model}")
def is_available(self) -> bool:
"""检查NovelAI服务是否可用"""
return bool(self.api_key)
async def generate_image_from_prompt_data(
self,
prompt: str,
negative_prompt: Optional[str] = None,
include_character: bool = False,
width: int = 1024,
height: int = 1024
) -> tuple[bool, Optional[Path], str]:
"""根据提示词生成图片
Args:
prompt: NovelAI格式的英文提示词
negative_prompt: LLM生成的负面提示词可选
include_character: 是否包含角色形象
width: 图片宽度
height: 图片高度
Returns:
(是否成功, 图片路径, 消息)
"""
if not self.api_key:
return False, None, "NovelAI API Key未配置"
try:
# 处理角色提示词
final_prompt = prompt
if include_character and self.character_prompt:
final_prompt = f"{self.character_prompt}, {prompt}"
logger.info(f"包含角色形象,添加角色提示词")
# 合并负面提示词
final_negative = self.base_negative_prompt
if negative_prompt:
if final_negative:
final_negative = f"{final_negative}, {negative_prompt}"
else:
final_negative = negative_prompt
logger.info(f"🎨 开始生成图片...")
logger.info(f" 尺寸: {width}x{height}")
logger.info(f" 正面提示词: {final_prompt[:100]}...")
logger.info(f" 负面提示词: {final_negative[:100]}...")
# 构建请求payload
payload = self._build_payload(final_prompt, final_negative, width, height)
# 发送请求
image_data = await self._call_novelai_api(payload)
if not image_data:
return False, None, "API请求失败"
# 保存图片
image_path = await self._save_image(image_data)
if not image_path:
return False, None, "图片保存失败"
logger.info(f"✅ 图片生成成功: {image_path}")
return True, image_path, "生成成功"
except Exception as e:
logger.error(f"生成图片时出错: {e}", exc_info=True)
return False, None, f"生成失败: {str(e)}"
def _build_payload(self, prompt: str, negative_prompt: str, width: int, height: int) -> dict:
"""构建NovelAI API请求payload"""
is_v4_model = "diffusion-4" in self.model
is_v3_model = "diffusion-3" in self.model
parameters = {
"width": width,
"height": height,
"scale": self.scale,
"steps": self.steps,
"sampler": self.sampler,
"seed": random.randint(0, 9999999999),
"n_samples": 1,
"ucPreset": 0,
"qualityToggle": True,
"sm": False,
"sm_dyn": False,
"noise_schedule": self.noise_schedule if is_v4_model else "native",
}
# V4.5模型使用新格式
if is_v4_model:
parameters.update({
"params_version": 3,
"cfg_rescale": 0,
"autoSmea": False,
"legacy": False,
"legacy_v3_extend": False,
"legacy_uc": False,
"add_original_image": True,
"controlnet_strength": 1,
"dynamic_thresholding": False,
"prefer_brownian": True,
"normalize_reference_strength_multiple": True,
"use_coords": True,
"inpaintImg2ImgStrength": 1,
"deliberate_euler_ancestral_bug": False,
"skip_cfg_above_sigma": None,
"characterPrompts": [],
"stream": "msgpack",
"v4_prompt": {
"caption": {
"base_caption": prompt,
"char_captions": []
},
"use_coords": True,
"use_order": True
},
"v4_negative_prompt": {
"caption": {
"base_caption": negative_prompt,
"char_captions": []
},
"legacy_uc": False
},
"negative_prompt": negative_prompt,
"reference_image_multiple": [],
"reference_information_extracted_multiple": [],
"reference_strength_multiple": []
})
# V3使用negative_prompt字段
elif is_v3_model:
parameters["negative_prompt"] = negative_prompt
payload = {
"input": prompt,
"model": self.model,
"action": "generate",
"parameters": parameters
}
# V4.5需要额外字段
if is_v4_model:
payload["use_new_shared_trial"] = True
return payload
async def _call_novelai_api(self, payload: dict) -> Optional[bytes]:
"""调用NovelAI API"""
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json",
}
connector = None
request_kwargs = {
"json": payload,
"headers": headers,
"timeout": aiohttp.ClientTimeout(total=120)
}
if self.proxy:
request_kwargs["proxy"] = self.proxy
connector = aiohttp.TCPConnector()
logger.info(f"使用代理: {self.proxy}")
try:
async with aiohttp.ClientSession(connector=connector) as session:
async with session.post(self.base_url, **request_kwargs) as resp:
if resp.status != 200:
error_text = await resp.text()
logger.error(f"API请求失败 ({resp.status}): {error_text[:200]}")
return None
img_data = await resp.read()
logger.info(f"收到响应数据: {len(img_data)} bytes")
# 检查是否是ZIP文件
if img_data[:4] == b'PK\x03\x04':
logger.info("检测到ZIP格式解压中...")
return self._extract_from_zip(img_data)
elif img_data[:4] == b'\x89PNG':
logger.info("检测到PNG格式")
return img_data
else:
logger.warning(f"未知文件格式前4字节: {img_data[:4].hex()}")
return img_data
except Exception as e:
logger.error(f"API调用失败: {e}", exc_info=True)
return None
def _extract_from_zip(self, zip_data: bytes) -> Optional[bytes]:
"""从ZIP中提取PNG"""
try:
with zipfile.ZipFile(io.BytesIO(zip_data)) as zf:
for filename in zf.namelist():
if filename.lower().endswith('.png'):
img_data = zf.read(filename)
logger.info(f"从ZIP提取: {filename} ({len(img_data)} bytes)")
return img_data
logger.error("ZIP中未找到PNG文件")
return None
except Exception as e:
logger.error(f"解压ZIP失败: {e}")
return None
async def _save_image(self, image_data: bytes) -> Optional[Path]:
"""保存图片到本地"""
try:
filename = f"novelai_{uuid.uuid4().hex[:12]}.png"
filepath = self.image_dir / filename
# 写入文件
with open(filepath, "wb") as f:
f.write(image_data)
f.flush()
import os
os.fsync(f.fileno())
# 验证图片
try:
with Image.open(filepath) as img:
img.verify()
with Image.open(filepath) as img:
logger.info(f"图片验证成功: {img.format} {img.size}")
except Exception as e:
logger.warning(f"图片验证失败: {e}")
return filepath
except Exception as e:
logger.error(f"保存图片失败: {e}")
return None

View File

@@ -83,21 +83,93 @@ class QZoneService:
return context
async def send_feed(self, topic: str, stream_id: str | None) -> dict[str, Any]:
"""发送一条说说"""
"""发送一条说说支持AI配图"""
cross_context = await self._get_cross_context()
story = await self.content_service.generate_story(topic, context=cross_context)
if not story:
return {"success": False, "message": "生成说说内容失败"}
await self.image_service.generate_images_for_story(story)
# 检查是否启用AI配图
ai_image_enabled = self.get_config("ai_image.enable_ai_image", False)
provider = self.get_config("ai_image.provider", "siliconflow")
image_path = None
if ai_image_enabled:
# 启用AI配图文本模型生成说说+图片提示词
story, image_info = await self.content_service.generate_story_with_image_info(topic, context=cross_context)
if not story:
return {"success": False, "message": "生成说说内容失败"}
# 根据provider调用对应的生图服务
if provider == "novelai":
try:
from .novelai_service import MaiZoneNovelAIService
novelai_service = MaiZoneNovelAIService(self.get_config)
if novelai_service.is_available():
# 解析画幅
aspect_ratio = image_info.get("aspect_ratio", "方图")
size_map = {
"方图": (1024, 1024),
"横图": (1216, 832),
"竖图": (832, 1216),
}
width, height = size_map.get(aspect_ratio, (1024, 1024))
logger.info(f"🎨 开始生成NovelAI配图...")
success, img_path, msg = await novelai_service.generate_image_from_prompt_data(
prompt=image_info.get("prompt", ""),
negative_prompt=image_info.get("negative_prompt"),
include_character=image_info.get("include_character", False),
width=width,
height=height
)
if success and img_path:
image_path = img_path
logger.info(f"✅ NovelAI配图生成成功")
else:
logger.warning(f"⚠️ NovelAI配图生成失败: {msg}")
else:
logger.warning("NovelAI服务不可用未配置API Key")
except Exception as e:
logger.error(f"NovelAI配图生成出错: {e}", exc_info=True)
elif provider == "siliconflow":
try:
# 调用硅基流动生成图片
success, img_path = await self.image_service.generate_image_from_prompt(
prompt=image_info.get("prompt", ""),
save_dir=None # 使用默认images目录
)
if success and img_path:
image_path = img_path
logger.info(f"✅ 硅基流动配图生成成功")
else:
logger.warning(f"⚠️ 硅基流动配图生成失败")
except Exception as e:
logger.error(f"硅基流动配图生成出错: {e}", exc_info=True)
else:
# 不使用AI配图只生成说说文本
story = await self.content_service.generate_story(topic, context=cross_context)
if not story:
return {"success": False, "message": "生成说说内容失败"}
qq_account = config_api.get_global_config("bot.qq_account", "")
api_client = await self._get_api_client(qq_account, stream_id)
if not api_client:
return {"success": False, "message": "获取QZone API客户端失败"}
image_dir = self.get_config("send.image_directory")
images_bytes = self._load_local_images(image_dir)
# 加载图片
images_bytes = []
# 使用AI生成的图片
if image_path and image_path.exists():
try:
with open(image_path, "rb") as f:
images_bytes.append(f.read())
logger.info(f"添加AI配图到说说")
except Exception as e:
logger.error(f"读取AI配图失败: {e}")
try:
success, _ = await api_client["publish"](story, images_bytes)
@@ -115,19 +187,16 @@ class QZoneService:
if not story:
return {"success": False, "message": "根据活动生成说说内容失败"}
await self.image_service.generate_images_for_story(story)
if self.get_config("send.enable_ai_image", False):
await self.image_service.generate_images_for_story(story)
qq_account = config_api.get_global_config("bot.qq_account", "")
# 注意:定时任务通常在后台运行,没有特定的用户会话,因此 stream_id 为 None
api_client = await self._get_api_client(qq_account, stream_id=None)
if not api_client:
return {"success": False, "message": "获取QZone API客户端失败"}
image_dir = self.get_config("send.image_directory")
images_bytes = self._load_local_images(image_dir)
try:
success, _ = await api_client["publish"](story, images_bytes)
success, _ = await api_client["publish"](story, [])
if success:
return {"success": True, "message": story}
return {"success": False, "message": "发布说说至QQ空间失败"}
@@ -434,7 +503,12 @@ class QZoneService:
logger.debug(f"锁定待评论说说: {comment_key}")
self.processing_comments.add(comment_key)
try:
comment_text = await self.content_service.generate_comment(content, target_name, rt_con, images)
# 使用content_service生成评论相当于回复好友的说说
comment_text = await self.content_service.generate_comment_reply(
story_content=content or rt_con or "说说内容",
comment_content="", # 评论说说时没有评论内容
commenter_name=target_name
)
if comment_text:
success = await api_client["comment"](target_qq, fid, comment_text)
if success:
@@ -465,61 +539,6 @@ class QZoneService:
return result
def _load_local_images(self, image_dir: str) -> list[bytes]:
"""随机加载本地图片(不删除文件)"""
images = []
if not image_dir or not os.path.exists(image_dir):
logger.warning(f"图片目录不存在或未配置: {image_dir}")
return images
try:
# 获取所有图片文件
all_files = [
f
for f in os.listdir(image_dir)
if os.path.isfile(os.path.join(image_dir, f))
and f.lower().endswith((".jpg", ".jpeg", ".png", ".gif", ".bmp"))
]
if not all_files:
logger.warning(f"图片目录中没有找到图片文件: {image_dir}")
return images
# 检查是否启用配图
enable_image = bool(self.get_config("send.enable_image", False))
if not enable_image:
logger.info("说说配图功能已关闭")
return images
# 根据配置选择图片数量
config_image_number = self.get_config("send.image_number", 1)
try:
config_image_number = int(config_image_number)
except (ValueError, TypeError):
config_image_number = 1
logger.warning("配置项 image_number 值无效,使用默认值 1")
max_images = min(min(config_image_number, 9), len(all_files)) # 最多9张最少1张
selected_count = max(1, max_images) # 确保至少选择1张
selected_files = random.sample(all_files, selected_count)
logger.info(f"{len(all_files)} 张图片中随机选择了 {selected_count} 张配图")
for filename in selected_files:
full_path = os.path.join(image_dir, filename)
try:
with open(full_path, "rb") as f:
image_data = f.read()
images.append(image_data)
logger.info(f"加载图片: {filename} ({len(image_data)} bytes)")
except Exception as e:
logger.error(f"加载图片 {filename} 失败: {e}")
return images
except Exception as e:
logger.error(f"加载本地图片失败: {e}")
return []
def _generate_gtk(self, skey: str) -> str:
hash_val = 5381
for char in skey: