13315423919/LandPPT
Template
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Add File

Browse Source
This commit is contained in:
13315423919
2025-11-07 09:05:14 +08:00
parent 9d37b0695d
commit 475fd82f58
1 changed files with 448 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

448
src/landppt/services/image/config/image_config.py Normal file
View File

@@ -0,0 +1,448 @@
"""
图片服务配置管理
"""
import os
from typing import Dict, Any, Optional, List
from pathlib import Path
import logging
logger = logging.getLogger(__name__)
class ImageServiceConfig:
"""图片服务配置管理器"""
def __init__(self):
self._config = self._load_default_config()
self._load_env_config()
def _load_default_config(self) -> Dict[str, Any]:
"""加载默认配置"""
return {
# DALL-E配置
'dalle': {
'api_key': '',
'api_base': 'https://api.openai.com/v1',
'model': 'dall-e-3',
'default_size': '1792x1024', # 16:9比例，适合PPT
'default_quality': 'standard', # standard, hd
'default_style': 'natural', # vivid, natural
'rate_limit_requests': 50, # 每分钟请求数
'rate_limit_window': 60, # 时间窗口（秒）
'timeout': 180 # 请求超时（秒）
},
# Stable Diffusion配置
'stable_diffusion': {
'api_key': '',
'api_base': 'https://api.stability.ai/v1',
'engine_id': 'stable-diffusion-xl-1024-v1-0',
'default_width': 1024,
'default_height': 576, # 16:9比例
'default_steps': 30,
'default_cfg_scale': 7.0,
'default_sampler': 'K_DPM_2_ANCESTRAL',
'rate_limit_requests': 150, # 每分钟请求数
'rate_limit_window': 60, # 时间窗口（秒）
'timeout': 180 # 请求超时（秒）
},
# SiliconFlow配置
'siliconflow': {
'api_key': '',
'api_base': 'https://api.siliconflow.cn/v1',
'model': 'Kwai-Kolors/Kolors',
'default_size': '1024x1024',
'default_batch_size': 1,
'default_steps': 20,
'default_guidance_scale': 7.5,
'rate_limit_requests': 60, # 每分钟请求数
'rate_limit_window': 60, # 时间窗口（秒）
'timeout': 120 # 请求超时（秒）
},
# Pollinations配置
'pollinations': {
'api_base': 'https://image.pollinations.ai',
'api_token': '', # API token（可选，用于认证和移除logo）
'referrer': '', # 推荐人标识符（可选，用于认证）
'model': 'flux', # flux, turbo, gptimage
'default_width': 1024,
'default_height': 1024,
'default_enhance': False, # 是否增强提示词
'default_safe': False, # 是否启用安全过滤
'default_nologo': False, # 是否移除logo（需要注册用户或token）
'default_private': False, # 是否私有模式
'default_transparent': False, # 是否生成透明背景（仅gptimage模型）
'rate_limit_requests': 60, # 每分钟请求数
'rate_limit_window': 60, # 时间窗口（秒）
'timeout': 300 # 请求超时（秒，增加以适应图片生成时间）
},
# Unsplash配置（网络搜索）
'unsplash': {
'api_key': '',
'api_base': 'https://api.unsplash.com',
'per_page': 20,
'rate_limit_requests': 50,
'rate_limit_window': 3600, # 1小时
'timeout': 30
},
# Pixabay配置（网络搜索）- 根据官方API文档
'pixabay': {
'api_key': '',
'api_base': 'https://pixabay.com/api',
'per_page': 20, # 默认20，最大200
'rate_limit_requests': 100, # 官方文档：100请求/60秒
'rate_limit_window': 60, # 官方文档：60秒窗口
'timeout': 30,
'cache_duration': 86400 # 官方要求：缓存24小时
},
# SearXNG配置（网络搜索）
'searxng': {
'host': '', # SearXNG实例的主机地址，如 http://209.135.170.113:8888
'per_page': 20,
'rate_limit_requests': 60, # 每分钟请求限制
'rate_limit_window': 60, # 60秒窗口
'timeout': 30,
'language': 'auto', # 搜索语言
'theme': 'simple' # 主题
},
# 缓存配置 - 简化配置，图片永久有效
'cache': {
'base_dir': 'temp/images_cache',
'max_size_gb': 5.0, # 默认最大缓存大小5GB
'cleanup_interval_hours': 24 # 默认24小时清理一次（虽然图片永久有效，但保留配置项）
},
# 图片处理配置
'processing': {
'max_file_size_mb': 50,
'supported_formats': ['jpg', 'jpeg', 'png', 'webp', 'gif'],
'default_quality': 85,
'thumbnail_size': (300, 200),
'watermark_enabled': False,
'watermark_text': 'LandPPT',
'watermark_opacity': 0.3
},
# 智能匹配配置
'matching': {
'similarity_threshold': 0.3,
'max_suggestions': 10,
'keyword_weight': 0.4,
'tag_weight': 0.3,
'usage_weight': 0.2,
'freshness_weight': 0.1
},
# PPT适配器配置
'ppt_adapter': {
'default_size': '1792x1024',
'quality_mode': 'standard',
'style_preference': 'natural',
'enable_negative_prompts': True,
'prompt_enhancement': True
}
}
def _load_env_config(self):
"""从环境变量加载配置"""
# DALL-E配置
if os.getenv('OPENAI_API_KEY'):
self._config['dalle']['api_key'] = os.getenv('OPENAI_API_KEY')
if os.getenv('DALLE_API_BASE'):
self._config['dalle']['api_base'] = os.getenv('DALLE_API_BASE')
# Stable Diffusion配置
if os.getenv('STABILITY_API_KEY'):
self._config['stable_diffusion']['api_key'] = os.getenv('STABILITY_API_KEY')
if os.getenv('STABILITY_API_BASE'):
self._config['stable_diffusion']['api_base'] = os.getenv('STABILITY_API_BASE')
# SiliconFlow配置
if os.getenv('SILICONFLOW_API_KEY'):
self._config['siliconflow']['api_key'] = os.getenv('SILICONFLOW_API_KEY')
if os.getenv('SILICONFLOW_API_BASE'):
self._config['siliconflow']['api_base'] = os.getenv('SILICONFLOW_API_BASE')
# 从配置服务加载SiliconFlow配置
try:
from ...config_service import get_config_service
config_service = get_config_service()
all_config = config_service.get_all_config()
# 加载SiliconFlow API密钥
siliconflow_api_key = all_config.get('siliconflow_api_key')
if siliconflow_api_key:
self._config['siliconflow']['api_key'] = siliconflow_api_key
# 加载SiliconFlow生成参数
siliconflow_size = all_config.get('siliconflow_image_size')
if siliconflow_size:
self._config['siliconflow']['default_size'] = siliconflow_size
siliconflow_steps = all_config.get('siliconflow_steps')
if siliconflow_steps:
self._config['siliconflow']['default_steps'] = int(siliconflow_steps)
siliconflow_guidance = all_config.get('siliconflow_guidance_scale')
if siliconflow_guidance:
self._config['siliconflow']['default_guidance_scale'] = float(siliconflow_guidance)
# 加载Unsplash配置
unsplash_access_key = all_config.get('unsplash_access_key')
if unsplash_access_key:
self._config['unsplash']['api_key'] = unsplash_access_key
# 加载Pixabay配置
pixabay_api_key = all_config.get('pixabay_api_key')
if pixabay_api_key:
self._config['pixabay']['api_key'] = pixabay_api_key
except Exception as e:
logger.warning(f"Failed to load config from config service: {e}")
# Unsplash配置（环境变量）
if os.getenv('UNSPLASH_ACCESS_KEY'):
self._config['unsplash']['api_key'] = os.getenv('UNSPLASH_ACCESS_KEY')
# Pixabay配置（环境变量）
if os.getenv('PIXABAY_API_KEY'):
self._config['pixabay']['api_key'] = os.getenv('PIXABAY_API_KEY')
# SearXNG配置（环境变量）
if os.getenv('SEARXNG_HOST'):
self._config['searxng']['host'] = os.getenv('SEARXNG_HOST')
# Pollinations配置（环境变量）
if os.getenv('POLLINATIONS_API_BASE'):
self._config['pollinations']['api_base'] = os.getenv('POLLINATIONS_API_BASE')
if os.getenv('POLLINATIONS_API_TOKEN'):
self._config['pollinations']['api_token'] = os.getenv('POLLINATIONS_API_TOKEN')
if os.getenv('POLLINATIONS_REFERRER'):
self._config['pollinations']['referrer'] = os.getenv('POLLINATIONS_REFERRER')
if os.getenv('POLLINATIONS_MODEL'):
self._config['pollinations']['model'] = os.getenv('POLLINATIONS_MODEL')
# 从配置服务加载Pollinations配置
try:
from ...config_service import config_service
all_config = config_service.get_all_config()
# 加载Pollinations配置
pollinations_token = all_config.get('pollinations_api_token')
if pollinations_token:
self._config['pollinations']['api_token'] = pollinations_token
pollinations_referrer = all_config.get('pollinations_referrer')
if pollinations_referrer:
self._config['pollinations']['referrer'] = pollinations_referrer
pollinations_model = all_config.get('pollinations_model')
if pollinations_model:
self._config['pollinations']['model'] = pollinations_model
pollinations_enhance = all_config.get('pollinations_enhance')
if pollinations_enhance is not None:
# 处理布尔值或字符串
if isinstance(pollinations_enhance, bool):
self._config['pollinations']['default_enhance'] = pollinations_enhance
else:
self._config['pollinations']['default_enhance'] = str(pollinations_enhance).lower() == 'true'
pollinations_safe = all_config.get('pollinations_safe')
if pollinations_safe is not None:
if isinstance(pollinations_safe, bool):
self._config['pollinations']['default_safe'] = pollinations_safe
else:
self._config['pollinations']['default_safe'] = str(pollinations_safe).lower() == 'true'
pollinations_nologo = all_config.get('pollinations_nologo')
if pollinations_nologo is not None:
if isinstance(pollinations_nologo, bool):
self._config['pollinations']['default_nologo'] = pollinations_nologo
else:
self._config['pollinations']['default_nologo'] = str(pollinations_nologo).lower() == 'true'
pollinations_private = all_config.get('pollinations_private')
if pollinations_private is not None:
if isinstance(pollinations_private, bool):
self._config['pollinations']['default_private'] = pollinations_private
else:
self._config['pollinations']['default_private'] = str(pollinations_private).lower() == 'true'
pollinations_transparent = all_config.get('pollinations_transparent')
if pollinations_transparent is not None:
if isinstance(pollinations_transparent, bool):
self._config['pollinations']['default_transparent'] = pollinations_transparent
else:
self._config['pollinations']['default_transparent'] = str(pollinations_transparent).lower() == 'true'
except Exception as e:
logger.debug(f"Could not load Pollinations config from config service: {e}")
# 缓存目录配置
if os.getenv('IMAGE_CACHE_DIR'):
self._config['cache']['base_dir'] = os.getenv('IMAGE_CACHE_DIR')
def get_config(self) -> Dict[str, Any]:
"""获取完整配置"""
return self._config.copy()
def get_provider_config(self, provider: str) -> Dict[str, Any]:
"""获取特定提供者的配置"""
return self._config.get(provider, {}).copy()
def update_config(self, updates: Dict[str, Any]):
"""更新配置"""
def deep_update(base_dict, update_dict):
for key, value in update_dict.items():
if key in base_dict and isinstance(base_dict[key], dict) and isinstance(value, dict):
deep_update(base_dict[key], value)
else:
base_dict[key] = value
deep_update(self._config, updates)
def is_provider_configured(self, provider: str) -> bool:
"""检查提供者是否已配置"""
provider_config = self._config.get(provider, {})
# SearXNG使用host而不是api_key
if provider == 'searxng':
host = provider_config.get('host', '')
return bool(host and host.strip())
# Pollinations不需要API密钥，只要配置存在就认为已配置
if provider == 'pollinations':
return True # Pollinations是免费服务，不需要API密钥
# 其他提供者检查API密钥是否存在
api_key = provider_config.get('api_key', '')
return bool(api_key and api_key.strip())
def should_enable_search_provider(self, provider: str) -> bool:
"""检查是否应该启用指定的搜索提供者"""
# 首先检查API密钥是否配置
if not self.is_provider_configured(provider):
return False
# 获取默认网络搜索提供商配置
try:
from ...config_service import get_config_service
config_service = get_config_service()
all_config = config_service.get_all_config()
default_provider = all_config.get('default_network_search_provider', 'unsplash')
# 如果是默认提供商，则启用
return provider.lower() == default_provider.lower()
except Exception as e:
logger.warning(f"Failed to get default network search provider config: {e}")
# 如果获取配置失败，回退到检查API密钥
return self.is_provider_configured(provider)
def get_configured_providers(self) -> List[str]:
"""获取已配置的提供者列表"""
providers = []
for provider in ['dalle', 'stable_diffusion', 'siliconflow', 'pollinations', 'unsplash', 'pixabay', 'searxng']:
if self.is_provider_configured(provider):
providers.append(provider)
return providers
def get_all_config(self) -> Dict[str, Any]:
"""获取所有配置"""
return self._config
def validate_config(self) -> Dict[str, List[str]]:
"""验证配置并返回错误信息"""
errors = {}
# 验证缓存配置
cache_config = self._config.get('cache', {})
cache_errors = []
# 检查max_size_gb（如果存在的话）
max_size_gb = cache_config.get('max_size_gb')
if max_size_gb is not None and max_size_gb <= 0:
cache_errors.append("max_size_gb must be greater than 0")
# 检查cleanup_interval_hours（如果存在的话）
cleanup_interval = cache_config.get('cleanup_interval_hours')
if cleanup_interval is not None and cleanup_interval <= 0:
cache_errors.append("cleanup_interval_hours must be greater than 0")
# 检查base_dir是否存在
if not cache_config.get('base_dir'):
cache_errors.append("base_dir is required")
if cache_errors:
errors['cache'] = cache_errors
# 验证处理配置
processing_config = self._config.get('processing', {})
processing_errors = []
if processing_config.get('max_file_size_mb', 0) <= 0:
processing_errors.append("max_file_size_mb must be greater than 0")
if not processing_config.get('supported_formats'):
processing_errors.append("supported_formats cannot be empty")
if processing_errors:
errors['processing'] = processing_errors
return errors
def save_to_file(self, file_path: str):
"""保存配置到文件"""
import json
config_path = Path(file_path)
config_path.parent.mkdir(parents=True, exist_ok=True)
with open(config_path, 'w', encoding='utf-8') as f:
json.dump(self._config, f, indent=2, ensure_ascii=False)
def load_from_file(self, file_path: str):
"""从文件加载配置"""
import json
config_path = Path(file_path)
if config_path.exists():
with open(config_path, 'r', encoding='utf-8') as f:
file_config = json.load(f)
self.update_config(file_config)
# 创建全局配置实例
image_config = ImageServiceConfig()
def get_image_config() -> ImageServiceConfig:
"""获取图片服务配置实例"""
return image_config
def update_image_config(updates: Dict[str, Any]):
"""更新图片服务配置"""
image_config.update_config(updates)
def get_provider_config(provider: str) -> Dict[str, Any]:
"""获取特定提供者的配置"""
return image_config.get_provider_config(provider)
def is_provider_configured(provider: str) -> bool:
"""检查提供者是否已配置"""
return image_config.is_provider_configured(provider)