9.1 KiB
9.1 KiB
API文档
本文档提供了ArticleReplaceBatch系统的API参考。
目录
配置管理器 API
ConfigManager
配置管理器单例类,用于统一管理应用配置。
类方法
__new__(cls) -> ConfigManager
创建或返回ConfigManager单例实例。
返回值:
ConfigManager: 单例实例
示例:
from config_manager import config_manager
# 获取单例实例
manager = config_manager
get(section: str, option: str, fallback: str = '') -> str
获取配置值。
参数:
section: 配置节名称option: 配置项名称fallback: 默认值(默认为空字符串)
返回值:
str: 配置值
示例:
workflow_id = config_manager.get('Coze', 'workflow_id')
get_int(section: str, option: str, fallback: int = 0) -> int
获取整数配置值。
参数:
section: 配置节名称option: 配置项名称fallback: 默认值(默认为0)
返回值:
int: 整数配置值
示例:
max_threads = config_manager.get_int('General', 'max_threads')
set(section: str, option: str, value: Any) -> None
设置配置值。
参数:
section: 配置节名称option: 配置项名称value: 配置值
示例:
config_manager.set('General', 'max_threads', '5')
save() -> None
保存配置到文件。
示例:
config_manager.save()
reload() -> None
重新加载配置文件。
示例:
config_manager.reload()
服务层 API
WebScrapingService
网页抓取服务,提供异步网页内容提取功能。
类方法
__init__(self, max_workers: int = 3)
初始化网页抓取服务。
参数:
max_workers: 最大工作线程数(默认为3)
示例:
from src.services.web_scraping import WebScrapingService
service = WebScrapingService(max_workers=5)
extract_content_async(self, links: List[str], max_retries: int = 3) -> List[Tuple[str, str, List[str]]]
异步提取多个链接的内容。
参数:
links: 链接列表max_retries: 最大重试次数(默认为3)
返回值:
List[Tuple[str, str, List[str]]]: 提取结果列表,每个元组包含(标题、内容、图片URLs)
示例:
links = [
"https://www.toutiao.com/article/123",
"https://www.toutiao.com/article/456"
]
results = service.extract_content_async(links)
for title, content, images in results:
print(f"标题: {title}")
print(f"内容长度: {len(content)}")
print(f"图片数量: {len(images)}")
ImageProcessingService
图片处理服务,提供异步图片下载和处理功能。
类方法
__init__(self, max_workers: int = 5)
初始化图片处理服务。
参数:
max_workers: 最大工作线程数(默认为5)
示例:
from src.services.image_processing import ImageProcessingService
service = ImageProcessingService(max_workers=5)
process_images_async(self, image_urls: List[str], base_filename: str, save_dir: str) -> bool
异步处理多个图片。
参数:
image_urls: 图片URL列表base_filename: 基础文件名save_dir: 保存目录
返回值:
bool: 是否全部成功
示例:
image_urls = [
"https://example.com/img1.jpg",
"https://example.com/img2.jpg"
]
success = service.process_images_async(image_urls, "test", "output")
AIService
AI服务基类,提供AI调用功能。
CozeAIService
Coze AI服务实现。
类方法
__init__(self)
初始化Coze AI服务。
示例:
from src.services.ai_service import CozeAIService
service = CozeAIService()
call_article_workflow(self, parameters: Dict[str, Any]) -> str
调用Coze文章工作流。
参数:
parameters: 参数字典
返回值:
str: AI生成的结果
示例:
result = service.call_article_workflow({"article": "测试文章内容"})
get_stats(self) -> Dict[str, Any]
获取服务统计信息。
返回值:
Dict[str, Any]: 包含调用次数、总时间、平均时间的统计信息
示例:
stats = service.get_stats()
print(f"调用次数: {stats['call_count']}")
print(f"平均时间: {stats['avg_time']:.2f}s")
工具 API
Validator
数据验证器,提供各种数据验证功能。
静态方法
validate_url(url: str) -> bool
验证URL格式。
参数:
url: URL字符串
返回值:
bool: URL是否有效
示例:
from src.utils.validation import Validator
is_valid = Validator.validate_url("https://www.example.com")
validate_article_url(url: str) -> bool
验证文章URL(头条、微信、网易等)。
参数:
url: 文章URL
返回值:
bool: URL是否为支持的文章链接
示例:
is_valid = Validator.validate_article_url("https://www.toutiao.com/article/123")
validate_string(value: Any, min_length: int = 0, max_length: Optional[int] = None) -> bool
验证字符串。
参数:
value: 待验证的值min_length: 最小长度(默认为0)max_length: 最大长度(可选)
返回值:
bool: 字符串是否有效
示例:
is_valid = Validator.validate_string("测试", min_length=1, max_length=10)
ArticleValidator
文章数据验证器。
静态方法
validate_title(title: str) -> bool
验证文章标题。
参数:
title: 文章标题
返回值:
bool: 标题是否有效
示例:
from src.utils.validation import ArticleValidator
is_valid = ArticleValidator.validate_title("测试标题")
validate_content(content: str, min_length: int = 100) -> bool
验证文章内容。
参数:
content: 文章内容min_length: 最小长度(默认为100)
返回值:
bool: 内容是否有效
示例:
is_valid = ArticleValidator.validate_content("测试内容...", min_length=50)
UI组件 API
LogTextHandler
自定义日志处理器,将日志输出到CustomTkinter文本框。
类方法
__init__(self, text_widget, level=logging.NOTSET)
初始化日志处理器。
参数:
text_widget: CustomTkinter文本框组件level: 日志级别(默认为NOTSET)
示例:
from src.ui.log_handler import LogTextHandler
handler = LogTextHandler(text_widget)
命令行接口
CLI Usage
使用cli.py提供的命令行接口。
基本用法
python cli.py --excel 文章链接.xlsx --threads 3
参数说明
--excel,-e: Excel文件路径(包含文章链接)--link,-l: 单个文章链接--threads,-t: 线程数(默认为1)--service,-s: AI服务(默认为coze)--type,-T: 生成类型("短篇"或"文章")--template: 使用的模板名称--verbose,-v: 显示详细日志--config: 配置文件路径
示例
处理Excel文件:
python cli.py --excel 文章链接.xlsx --threads 3 --type 文章
处理单个链接:
python cli.py --link https://www.toutiao.com/article/123
使用详细日志:
python cli.py --excel 文章链接.xlsx --verbose
使用自定义配置:
python cli.py --excel 文章链接.xlsx --config custom_config.ini
异常处理
ValidationError
数据验证错误异常。
示例:
from src.utils.validation import validate_and_raise, ArticleValidator
try:
validate_and_raise(data, ArticleValidator.validate_article_data, "文章数据验证失败")
except ValidationError as e:
print(f"验证失败: {e}")
最佳实践
1. 使用配置管理器
# 推荐:使用ConfigManager
from config_manager import config_manager
workflow_id = config_manager.get('Coze', 'workflow_id')
# 不推荐:直接访问CONFIG
from config import CONFIG
workflow_id = CONFIG['Coze']['workflow_id']
2. 使用服务层
# 推荐:使用服务层
from src.services.web_scraping import web_scraping_service
results = web_scraping_service.extract_content_async(links)
# 不推荐:直接调用底层函数
from get_web_content import extract_content_with_retry
results = [extract_content_with_retry(link) for link in links]
3. 验证输入数据
# 推荐:使用验证器
from src.utils.validation import Validator
if Validator.validate_article_url(url):
process_url(url)
# 不推荐:不验证直接使用
process_url(url)
版本信息
- 当前版本: 1.0.0
- Python版本: 3.10+
- 最后更新: 2026-03-07
联系方式
如有问题或建议,请通过以下方式联系:
- 提交Issue
- 发送Pull Request
- 联系项目维护者
文档版本: v1.0
维护者: opencode