Claude Code自动化开发实战
·
Claude Code：AI驱动的自动化开发与内容发布实战指南
Claude Code 是 Anthropic 公司推出的 AI 编程助手，能够理解自然语言指令并生成、执行代码，特别在自动化任务处理方面展现出强大能力。结合 MCP（Model Context Protocol）等扩展模块，Claude Code 可以实现从代码编写到内容发布的完整自动化流程。
一、Claude Code 核心能力与架构解析
1. 技术架构与工作原理
Claude Code 基于 Claude 3系列模型构建，通过专门的代码训练和优化，具备以下核心能力：
| 能力维度 | 具体功能 | 技术实现 |
|---|---|---|
| 代码理解 | 解析现有代码库、理解复杂逻辑 | 基于 Transformer 的代码表示学习 |
| 代码生成 | 根据需求生成完整函数、类或脚本 | 代码补全、函数级生成、文件级生成 |
| 代码执行 | 在安全沙箱中运行生成代码 | Docker 容器隔离、资源限制 |
| 错误调试 | 分析错误信息并提供修复方案 | 静态分析、动态追踪 |
| 自动化集成 | 与开发工具链无缝集成 | API 调用、CLI 工具封装 |
# Claude Code 基础交互示例
import anthropic
class ClaudeCodeClient:
def __init__(self, api_key):
self.client = anthropic.Anthropic(api_key=api_key)
def generate_code(self, prompt, language="python"):
"""请求 Claude Code 生成代码"""
system_prompt = f"""你是一个专业的{language}开发助手。
请根据用户需求生成完整、可运行的代码。
代码应包含必要的注释和错误处理。"""
response = self.client.messages.create(
model="claude-35-sonnet-20241022",
max_tokens=4000,
system=system_prompt,
messages=[
{"role": "user", "content": prompt}
]
)
return response.content[0].text
def execute_code(self, code, timeout=30):
"""在安全环境中执行生成的代码"""
import subprocess import tempfile
with tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False) as f:
f.write(code)
temp_file = f.name try:
result = subprocess.run(
['python', temp_file],
capture_output=True,
text=True,
timeout=timeout )
return {
'success': result.returncode == 0,
'stdout': result.stdout,
'stderr': result.stderr,
'returncode': result.returncode
}
except subprocess.TimeoutExpired:
return {'success': False, 'error': '执行超时'}
finally:
import os os.unlink(temp_file)
# 使用示例
client = ClaudeCodeClient(api_key="your-api-key")
code_prompt = "编写一个Python函数，从CSDN博客页面提取文章标题和内容"
generated_code = client.generate_code(code_prompt)
execution_result = client.execute_code(generated_code)
print(f"代码执行结果: {execution_result}")
2. MCP（Model Context Protocol）集成
MCP 是 Claude Code 的核心扩展协议，允许模型访问外部工具和资源，实现真正的自动化操作。
# MCP 服务器示例 文件系统访问
from mcp import Serverimport json
class FileSystemMCP:
def __init__(self):
self.server = Server()
self._register_tools()
def _register_tools(self):
"""注册文件系统操作工具"""
@self.server.tool()
def read_file(filepath: str) -> str:
"""读取文件内容"""
try:
with open(filepath, 'r', encoding='utf-8') as f:
return f.read()
except Exception as e:
return f"读取文件失败: {str(e)}"
@self.server.tool()
def write_file(filepath: str, content: str) -> str:
"""写入文件内容"""
try:
with open(filepath, 'w', encoding='utf-8') as f:
f.write(content)
return "文件写入成功"
except Exception as e:
return f"写入文件失败: {str(e)}"
@self.server.tool()
def list_directory(dirpath: str) -> list:
"""列出目录内容"""
import os try:
return os.listdir(dirpath)
except Exception as e:
return [f"列出目录失败: {str(e)}"]
def run(self):
"""启动MCP服务器"""
self.server.run()
# Claude Code 通过MCP调用工具
mcp_instruction = """
我需要读取 /articles 目录下的所有Markdown文件，
提取标题和标签，然后生成发布脚本。
请使用MCP工具完成这个任务。
"""
二、CSDN 自动化发布全流程实现
1. 环境配置与依赖安装
# 1. 安装 Claude Code CLIpip install anthropic
npm install -g @anthropic-ai/claude-code
# 2. 安装 Playwright 用于浏览器自动化
pip install playwright
playwright install chromium
# 3. 安装必要的Python库
pip install pyyaml beautifulsoup4 markdown
# 4. 配置环境变量
export CLAUDE_API_KEY="your-claude-api-key"
export CSDN_USERNAME="your-csdn-username"
export CSDN_PASSWORD="your-csdn-password"
2. 核心发布工具实现
# csdn_auto_publisher.pyimport os
import yaml
import markdown
import json
import time
from pathlib import Path
from typing import Dict, List, Optional
from playwright.sync_api import sync_playwright, Page, BrowserContext
class CSDNAutoPublisher:
"""CSDN自动化发布工具"""
def __init__(self, headless: bool = False):
self.headless = headless self.cookie_path = Path.home() / ".csdn_cookies.json"
self.session = None def load_markdown_with_frontmatter(self, filepath: str) -> Dict:
"""解析带Front Matter的Markdown文件"""
content = Path(filepath).read_text(encoding='utf-8')
# 解析YAML Front Matter
if content.startswith('---'):
parts = content.split('---', 2)
if len(parts) >= 3:
frontmatter = yaml.safe_load(parts[1])
markdown_content = parts[2].strip()
return {
'metadata': frontmatter,
'content': markdown_content,
'html': self._markdown_to_html(markdown_content)
}
# 无Front Matter的情况
return {
'metadata': {'title': Path(filepath).stem},
'content': content,
'html': self._markdown_to_html(content)
}
def _markdown_to_html(self, markdown_text: str) -> str:
"""将Markdown转换为HTML"""
# 扩展支持代码高亮、表格等
extensions = [
'markdown.extensions.extra',
'markdown.extensions.codehilite',
'markdown.extensions.tables',
'markdown.extensions.toc'
]
return markdown.markdown(markdown_text, extensions=extensions)
def login_csdn(self, page: Page, username: str, password: str) -> bool:
"""登录CSDN并保存Cookie"""
try:
page.goto("https://passport.csdn.net/login")
page.wait_for_load_state("networkidle")
# 切换到账号密码登录
page.click("text=账号登录")
time.sleep(1)
# 输入用户名密码 page.fill("#all", username)
page.fill("#password-number", password)
# 处理验证码（如有）
if page.is_visible(".geetest_holder"):
print("检测到验证码，请手动处理...")
page.pause() # 暂停等待手动处理 # 点击登录 page.click(".btn.btn-primary")
page.wait_for_url("**://blog.csdn.net/**", timeout=10000)
# 保存Cookie cookies = page.context.cookies()
with open(self.cookie_path, 'w') as f:
json.dump(cookies, f)
print("登录成功并保存Cookie")
return True
except Exception as e:
print(f"登录失败: {e}")
return False
def load_cookies(self, context: BrowserContext) -> bool:
"""加载保存的Cookie"""
if self.cookie_path.exists():
with open(self.cookie_path, 'r') as f:
cookies = json.load(f)
context.add_cookies(cookies)
print(f"已加载 {len(cookies)} 个Cookie")
return True
return False
def publish_article(self, article_data: Dict, tags: List[str] = None,
category: str = "原创") -> Dict:
"""发布文章到CSDN"""
with sync_playwright() as p:
# 启动浏览器
browser = p.chromium.launch(headless=self.headless)
context = browser.new_context(
viewport={'width': 1920, 'height': 1080},
user_agent='Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
)
# 尝试加载Cookie if not self.load_cookies(context):
# 需要重新登录
page = context.new_page()
if not self.login_csdn(page,
os.getenv('CSDN_USERNAME'),
os.getenv('CSDN_PASSWORD')):
return {'success': False, 'error': '登录失败'}
page = context.new_page()
page.goto("https://mp.csdn.net/mp_blog/manage/article?spm=1001.2014.3001.5352")
page.wait_for_load_state("networkidle")
# 点击写文章按钮 page.click("text=写文章")
page.wait_for_selector(".editor-container", timeout=10000)
# 输入标题 title_input = page.locator("input[placeholder='请输入文章标题']")
title_input.fill(article_data['metadata'].get('title', '无标题'))
# 输入内容（处理富文本编辑器）
self._fill_editor_content(page, article_data['html'])
# 设置标签 if tags:
self._set_tags(page, tags)
# 设置分类
self._set_category(page, category)
# 发布文章
publish_button = page.locator("button:has-text('发布')")
publish_button.click()
# 等待发布完成 try:
page.wait_for_selector("text=发布成功", timeout=30000)
# 获取文章URL article_url = page.url
if "mp.csdn.net" in article_url:
# 跳转到文章页面获取真实URL page.goto(article_url.replace("mp.csdn.net", "blog.csdn.net"))
article_url = page.url
return {
'success': True,
'url': article_url,
'title': article_data['metadata'].get('title'),
'message': '发布成功'
}
except Exception as e:
return {
'success': False,
'error': f'发布失败: {str(e)}',
'screenshot': self._take_screenshot(page)
}
finally:
browser.close()
def _fill_editor_content(self, page: Page, html_content: str):
"""向富文本编辑器填充内容"""
# CSDN编辑器有多种实现方式，这里提供两种策略 try:
# 策略1：直接设置iframe内容 editor_frame = page.frame_locator(".editoriframe")
if editor_frame:
editor_body = editor_frame.locator("body")
editor_body.click()
page.keyboard.press("Control+A")
page.keyboard.press("Delete")
editor_body.evaluate(f"(element) => element.innerHTML = `{html_content}`")
return except:
pass # 策略2：使用execCommand（备用方案）
page.evaluate(f"""
const editor = document.querySelector('.editor-container [contenteditable="true"]');
if (editor) {{
editor.innerHTML = `{html_content}`;
editor.dispatchEvent(new Event('input', {{ bubbles: true }}));
}}
""")
def _set_tags(self, page: Page, tags: List[str]):
"""设置文章标签"""
tags_input = page.locator("input[placeholder='输入标签，按回车确认']")
for tag in tags[:5]: # CSDN最多5个标签
tags_input.fill(tag)
tags_input.press("Enter")
time.sleep(0.5)
def _set_category(self, page: Page, category: str):
"""设置文章分类"""
try:
category_dropdown = page.locator("text=选择分类")
category_dropdown.click()
page.click(f"text={category}")
except:
print("设置分类失败，使用默认分类")
def _take_screenshot(self, page: Page) -> str:
"""截图保存"""
timestamp = time.strftime("%Y%m%d_%H%M%S")
screenshot_path = f"screenshot_error_{timestamp}.png"
page.screenshot(path=screenshot_path, full_page=True)
return screenshot_path
# 使用示例
if __name__ == "__main__":
publisher = CSDNAutoPublisher(headless=False)
# 加载文章
article = publisher.load_markdown_with_frontmatter("article.md")
# 发布文章
result = publisher.publish_article(
article_data=article,
tags=article['metadata'].get('tags', ['AI', '自动化', 'Python']),
category="原创"
)
print(f"发布结果: {result}")
3. 多平台分发工具实现
# multi_platform_publisher.py
import asyncio
from typing import Dict, List
from abc import ABC, abstractmethod
class BlogPlatform(ABC):
"""博客平台抽象基类"""
def __init__(self, platform_name: str):
self.platform_name = platform_name self.cookies = {}
@abstractmethod async def login(self, credentials: Dict) -> bool:
"""登录平台"""
pass
@abstractmethod async def publish(self, article: Dict) -> Dict:
"""发布文章"""
pass @abstractmethod
async def check_status(self, article_url: str) -> Dict:
"""检查文章状态"""
pass
class CSDNPlatform(BlogPlatform):
"""CSDN平台实现"""
def __init__(self):
super().__init__("CSDN")
self.base_url = "https://blog.csdn.net"
async def login(self, credentials: Dict) -> bool:
"""CSDN登录实现"""
# 使用Playwright进行登录 from playwright.async_api import async_playwright async with async_playwright() as p:
browser = await p.chromium.launch(headless=True)
context = await browser.new_context()
page = await context.new_page()
await page.goto(f"{self.base_url}/login")
# ... 登录逻辑
# 保存cookies self.cookies = await context.cookies()
await browser.close()
return True
async def publish(self, article: Dict) -> Dict:
"""CSDN发布实现"""
# 实现CSDN特定的发布逻辑 return {
'platform': self.platform_name,
'success': True,
'url': 'https://blog.csdn.net/...',
'article_id': '123456'
}
class JuejinPlatform(BlogPlatform):
"""掘金平台实现"""
def __init__(self):
super().__init__("掘金")
self.base_url = "https://juejin.cn"
async def publish(self, article: Dict) -> Dict:
"""掘金发布实现"""
# 掘金API发布逻辑 return {
'platform': self.platform_name,
'success': True,
'url': 'https://juejin.cn/post/...'
}
class MultiPlatformPublisher:
"""多平台发布管理器"""
def __init__(self):
self.platforms = {
'csdn': CSDNPlatform(),
'juejin': JuejinPlatform(),
# 可以扩展更多平台 # 'blogcn': BlogCNPlatform(),
# '51cto': Platform51CTO(),
# 'jianshu': JianShuPlatform()
}
self.results = []
async def publish_to_all(self, article: Dict, platforms: List[str] = None) -> List[Dict]:
"""发布到所有指定平台"""
if platforms is None:
platforms = list(self.platforms.keys())
tasks = []
for platform_name in platforms:
if platform_name in self.platforms:
platform = self.platforms[platform_name]
task = self._publish_with_retry(platform, article)
tasks.append(task)
# 并发发布 results = await asyncio.gather(*tasks, return_exceptions=True)
# 处理结果
processed_results = []
for i, result in enumerate(results):
platform_name = platforms[i]
if isinstance(result, Exception):
processed_results.append({
'platform': platform_name,
'success': False,
'error': str(result)
})
else:
processed_results.append(result)
self.results = processed_results
return processed_results async def _publish_with_retry(self, platform: BlogPlatform, article: Dict, max_retries: int = 3) -> Dict:
"""带重试的发布"""
for attempt in range(max_retries):
try:
result = await platform.publish(article)
return result except Exception as e:
if attempt == max_retries - 1:
raise
await asyncio.sleep(2 ** attempt) # 指数退避
return {'platform': platform.platform_name, 'success': False, 'error': '重试失败'}
# 使用Claude Code生成发布脚本
claude_prompt = """
我需要一个Python脚本，能够将Markdown文章发布到多个技术博客平台。
要求：
1. 支持CSDN、掘金、博客园、51CTO、简书
2. 每个平台使用独立的登录和发布逻辑
3. 支持并发发布以提高效率
4. 包含错误处理和重试机制
5. 生成发布报告
请生成完整的代码实现。
"""
# Claude Code会生成相应的多平台发布代码
三、Claude Code 自动化工作流集成
1. 基于OpenCLI的自动化发布
# opencli-config.yaml
name: csdn-auto-publisher
version: 1.0.0
description: 使用Claude Code自动发布文章到CSDN
commands:
publish:
description: 发布Markdown文章到CSDN
parameters:
- name: file type: string
required: true description: Markdown文件路径
name: tags
type: array description: 文章标签
name: category
type: string default: "原创"
description: 文章分类
steps:
- name: 解析文章 run: |
pythonc "
import yaml, markdown
# 解析Front Matter和内容 "
name: 登录CSDN
run: |
python scripts/login_csdn.py
--username ${CSDN_USERNAME}
--password ${CSDN_PASSWORD}
- name: 发布文章 run: |
python scripts/publish_article.py
--file ${file}
--tags ${tags}
--category ${category}
name: 验证发布
run: |
python scripts/verify_publish.py --url ${published_url}
hooks:
pre-publish:
name: 检查文章质量 run: python scripts/quality_check.py --file ${file}
name: 生成摘要 run: python scripts/generate_summary.py --file ${file}
post-publish:
name: 分享到社交媒体 run: python scripts/share_social.py --url ${published_url}
- name: 更新索引 run: python scripts/update_index.py --url ${published_url}
# opencli_integration.py
import subprocess
import json
from pathlib import Path
class OpenCLIIntegration:
"""OpenCLI与Claude Code集成"""
def __init__(self, cli_config_path: str):
self.config_path = Path(cli_config_path)
self.load_config()
def load_config(self):
"""加载OpenCLI配置"""
import yaml with open(self.config_path, 'r', encoding='utf-8') as f:
self.config = yaml.safe_load(f)
def execute_command(self, command_name: str, **kwargs):
"""执行OpenCLI命令"""
if command_name not in self.config.get('commands', {}):
raise ValueError(f"命令 {command_name} 不存在")
command_config = self.config['commands'][command_name]
# 构建命令参数 args = []
for param in command_config.get('parameters', []):
param_name = param['name']
if param_name in kwargs:
if param['type'] == 'array':
for item in kwargs[param_name]:
args.extend([f"--{param_name}", str(item)])
else:
args.extend([f"--{param_name}", str(kwargs[param_name])])
elif param.get('required', False):
raise ValueError(f"缺少必需参数: {param_name}")
# 执行前置钩子 self._run_hooks('pre-' + command_name, kwargs)
# 执行命令步骤 results = []
for step in command_config.get('steps', []):
cmd = step['run']
# 替换变量
for key, value in kwargs.items():
cmd = cmd.replace(f'${{{key}}}', str(value))
result = subprocess.run(cmd, shell=True, capture_output=True, text=True)
results.append({
'step': step['name'],
'success': result.returncode == 0,
'output': result.stdout,
'error': result.stderr
})
# 执行后置钩子
self._run_hooks('post-' + command_name, kwargs)
return results
def _run_hooks(self, hook_type: str, context: dict):
"""运行钩子脚本"""
hooks = self.config.get('hooks', {}).get(hook_type, [])
for hook in hooks:
cmd = hook['run']
for key, value in context.items():
cmd = cmd.replace(f'${{{key}}}', str(value))
subprocess.run(cmd, shell=True)
# 使用示例
opencli = OpenCLIIntegration("opencli-config.yaml")
# 发布文章
results = opencli.execute_command(
"publish",
file="article.md",
tags=["AI", "自动化", "Python"],
category="原创"
)
for result in results:
print(f"{result['step']}: {'成功' if result['success'] else '失败'}")
2.自动化测试与验证
# automated_testing.py
import pytest
from playwright.sync_api import Page, expectclass CSDNAutoTest:
"""CSDN自动化测试套件"""
def test_login(self, page: Page):
"""测试登录功能"""
page.goto("https://passport.csdn.net/login")
# 验证页面元素
expect(page).to_have_title("登录CSDN")
expect(page.locator("text=账号登录")).to_be_visible()
# 测试登录流程 page.click("text=账号登录")
page.fill("#all", "test_user")
page.fill("#password-number", "test_password")
# 验证登录按钮状态 login_button = page.locator(".btn.btn-primary")
expect(login_button).to_be_enabled()
print("登录测试通过")
def test_article_editor(self, page: Page):
"""测试文章编辑器"""
# 先登录
self._login(page)
# 进入编辑器
page.goto("https://mp.csdn.net/mp_blog/manage/article")
page.click("text=写文章")
# 验证编辑器加载 expect(page.locator(".editor-container")).to_be_visible(timeout=10000)
# 测试标题输入
title = "自动化测试文章"
title_input = page.locator("input[placeholder='请输入文章标题']")
title_input.fill(title)
expect(title_input).to_have_value(title)
# 测试内容输入 test_content = "# 测试标题\
\
这是测试内容"
self._fill_editor(page, test_content)
print("编辑器测试通过")
def test_publish_flow(self, page: Page):
"""测试完整发布流程"""
self._login(page)
# 创建测试文章
article_data = {
'title': '自动化测试文章 ' + str(time.time()),
'content': '# 测试内容\
\
这是由自动化测试创建的文章。',
'tags': ['测试', '自动化'],
'category': '原创'
}
# 使用发布工具 publisher = CSDNAutoPublisher()
result = publisher.publish_article(article_data)
# 验证发布结果
assert result['success'] == True assert 'csdn.net' in result['url']
# 访问发布后的文章
page.goto(result['url'])
expect(page).to_have_title(article_data['title'])
print("发布流程测试通过")
def _login(self, page: Page):
"""辅助登录方法"""
# 使用Cookie或测试账号登录 pass
def _fill_editor(self, page: Page, content: str):
"""辅助填充编辑器方法"""
# 实现编辑器内容填充
pass
# 使用pytest运行测试
if __name__ == "__main__":
import sys sys.exit(pytest.main([__file__, "-v"]))
四、生产环境部署与优化
1. Docker容器化部署
# Dockerfile
FROM python:3.9-slim
# 安装系统依赖
RUN apt-get update && apt-get install -y \
wget \
gnupg \
unzip \
&& rm -rf /var/lib/apt/lists/*
# 安装Chrome和Playwright
RUN wget -q -O - https://dl-ssl.google.com/linux/linux_signing_key.pub | apt-key add - \
&& echo "deb [arch=amd64] http://dl.google.com/linux/chrome/deb/ stable main" >> /etc/apt/sources.list.d/google.list \
&& apt-get update \
&& apt-get install -y google-chrome-stable \
&& rm -rf /var/lib/apt/lists/*
# 安装Python依赖
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
# 安装Playwright浏览器
RUN playwright install chromium
# 复制应用代码
WORKDIR /app
COPY . .
# 创建非root用户
RUN useradd -m -u 1000 claudeuserUSER claudeuser# 启动应用
CMD ["python", "main.py"]
# docker-compose.yml
version: '3.8'
services:
claude-code-publisher:
build: .
container_name: csdn-auto-publisher
environment:
- CLAUDE_API_KEY=${CLAUDE_API_KEY}
CSDN_USERNAME=${CSDN_USERNAME}
CSDN_PASSWORD=${CSDN_PASSWORD}
REDIS_URL=redis://redis:6379 volumes:
./articles:/app/articles
./config:/app/config
./logs:/app/logs depends_on:
- redis restart: unless-stopped
redis:
image: redis:7-alpine container_name: csdn-redis
volumes:
- redis-data:/data command: redis-server --appendonly yes
restart: unless-stopped
monitor:
image: grafana/grafana:latest
container_name: csdn-monitor ports:
"3000:3000"
volumes:
./monitoring/grafana:/var/lib/grafana
environment:
- GF_SECURITY_ADMIN_PASSWORD=admin restart: unless-stopped
volumes:
redis-data:
2. 监控与日志系统
# monitoring.py
import logging
import time
from datetime import datetime
from typing import Dict, Any
import redis
from prometheus_client import Counter, Histogram, Gauge, start_http_server
class PublisherMonitor:
"""发布监控系统"""
def __init__(self, redis_host='localhost', redis_port=6379):
# 初始化Redis连接 self.redis = redis.Redis(
host=redis_host, port=redis_port,
decode_responses=True
)
# Prometheus指标 self.publish_requests = Counter(
'csdn_publish_requests_total',
'Total publish requests'
)
self.publish_errors = Counter(
'csdn_publish_errors_total',
'Total publish errors',
['error_type']
)
self.publish_duration = Histogram(
'csdn_publish_duration_seconds',
'Publish duration in seconds'
)
self.articles_published = Gauge(
'csdn_articles_published_total',
'Total articles published'
)
# 日志配置
self.setup_logging()
def setup_logging(self):
"""配置结构化日志"""
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('publisher.log'),
logging.StreamHandler()
]
)
self.logger = logging.getLogger(__name__)
def track_publish(self, article_id: str, platform: str):
"""跟踪发布过程"""
start_time = time.time()
@self.publish_duration.time()
def _track():
# 记录发布开始
self.publish_requests.inc()
self.logger.info(f"开始发布文章 {article_id} 到 {platform}")
# 存储发布状态到Redis
publish_key = f"publish:{article_id}:{platform}"
self.redis.hset(publish_key, mapping={
'start_time': datetime.now().isoformat(),
'platform': platform,
'status': 'processing'
})
return publish_key publish_key = _track()
return publish_key def record_success(self, publish_key: str, result: Dict):
"""记录发布成功"""
duration = time.time() - float(self.redis.hget(publish_key, 'start_time'))
self.redis.hset(publish_key, mapping={
'end_time': datetime.now().isoformat(),
'status': 'success',
'duration': duration,
'url': result.get('url', ''),
'article_id': result.get('article_id', '')
})
self.articles_published.inc()
self.logger.info(f"发布成功: {publish_key}, 耗时: {duration:.2f}s")
def record_error(self, publish_key: str, error: Exception):
"""记录发布错误"""
error_type = type(error).__name__
self.publish_errors.labels(error_type=error_type).inc()
self.redis.hset(publish_key, mapping={
'end_time': datetime.now().isoformat(),
'status': 'failed',
'error': str(error),
'error_type': error_type })
self.logger.error(f"发布失败: {publish_key}, 错误: {error}")
def get_stats(self) -> Dict[str, Any]:
"""获取发布统计"""
stats = {
'total_requests': self.publish_requests._value.get(),
'total_errors': sum(
self.publish_errors.labels(error_type=et)._value.get()
for et in ['NetworkError', 'LoginError', 'PublishError']
),
'success_rate': self.calculate_success_rate(),
'avg_duration': self.calculate_avg_duration(),
'recent_publishes': self.get_recent_publishes(limit=10)
}
return stats def calculate_success_rate(self) -> float:
"""计算成功率"""
total = self.publish_requests._value.get()
errors = sum(
self.publish_errors.labels(error_type=et)._value.get()
for et in ['NetworkError', 'LoginError', 'PublishError']
)
if total == 0:
return 0.0 return (total - errors) / total * 100
def calculate_avg_duration(self) -> float:
"""计算平均发布时长"""
# 从Redis获取最近100次发布的时长
publish_keys = self.redis.keys("publish:*")
durations = []
for key in publish_keys[:100]:
duration = self.redis.hget(key, 'duration')
if duration:
durations.append(float(duration))
return sum(durations) / len(durations) if durations else 0.0
def get_recent_publishes(self, limit: int = 10) -> list:
"""获取最近发布记录"""
publish_keys = self.redis.keys("publish:*")
recent = []
for key in sorted(publish_keys, reverse=True)[:limit]:
data = self.redis.hgetall(key)
recent.append({
'key': key,
'status': data.get('status'),
'duration': data.get('duration'),
'url': data.get('url'),
'error': data.get('error')
})
return recent
# 启动监控服务器
if __name__ == "__main__":
monitor = PublisherMonitor()
# 启动Prometheus HTTP服务器 start_http_server(8000)
print("监控服务器已启动，访问 http://localhost:8000 查看指标")
# 保持运行 import time while True:
stats = monitor.get_stats()
print(f"当前统计: {stats}")
time.sleep(60)
3. 安全与最佳实践
# security.py
import hashlib
import hmac
import base64
import json
from datetime import datetime, timedelta
from typing import Dict, Optional
import secrets
class SecurityManager:
"""安全管理器"""
def __init__(self, secret_key: str):
self.secret_key = secret_key.encode()
self.token_expiry = timedelta(hours=24)
def encrypt_password(self, password: str) -> str:
"""加密密码"""
salt = secrets.token_bytes(16)
key = hashlib.pbkdf2_hmac(
'sha256',
password.encode(),
salt,
100000 )
return base64.b64encode(salt + key).decode()
def verify_password(self, password: str, encrypted: str) -> bool:
"""验证密码"""
decoded = base64.b64decode(encrypted)
salt, stored_key = decoded[:16], decoded[16:]
key = hashlib.pbkdf2_hmac(
'sha256',
password.encode(),
salt,
100000
)
return hmac.compare_digest(key, stored_key)
def generate_token(self, user_id: str, data: Dict = None) -> str:
"""生成JWT Token"""
header = {
"alg": "HS256",
"typ": "JWT"
}
payload = {
"user_id": user_id,
"exp": (datetime.utcnow() + self.token_expiry).timestamp(),
"iat": datetime.utcnow().timestamp()
}
if data:
payload.update(data)
# 编码Header和Payload encoded_header = base64.urlsafe_b64encode(
json.dumps(header).encode()
).decode().rstrip('=')
encoded_payload = base64.urlsafe_b64encode(
json.dumps(payload).encode()
).decode().rstrip('=')
# 生成签名 message = f"{encoded_header}.{encoded_payload}"
signature = hmac.new(
self.secret_key,
message.encode(),
hashlib.sha256 ).digest()
encoded_signature = base64.urlsafe_b64encode(
signature ).decode().rstrip('=')
return f"{encoded_header}.{encoded_payload}.{encoded_signature}"
def verify_token(self, token: str) -> Optional[Dict]:
"""验证JWT Token"""
try:
parts = token.split('.')
if len(parts) != 3:
return None encoded_header, encoded_payload, encoded_signature = parts # 验证签名 message = f"{encoded_header}.{encoded_payload}"
expected_signature = hmac.new(
self.secret_key,
message.encode(),
hashlib.sha256 ).digest()
signature = base64.urlsafe_b64decode(
encoded_signature + '=' * (4 - len(encoded_signature) % 4)
)
if not hmac.compare_digest(signature, expected_signature):
return None # 解码Payload payload_json = base64.urlsafe_b64decode(
encoded_payload + '=' * (4 - len(encoded_payload) % 4)
)
payload = json.loads(payload_json)
# 检查过期时间
if payload['exp'] < datetime.utcnow().timestamp():
return None return payload except Exception:
return None def sanitize_input(self, text: str) -> str:
"""清理用户输入，防止XSS攻击"""
import html # 转义HTML特殊字符 sanitized = html.escape(text)
# 移除危险标签和属性
import re
dangerous_patterns = [
r'<script.*?>.*?</script>',
r'javascript:',
r'on\w+=\".*?\"',
r'data:.*?;base64,'
]
for pattern in dangerous_patterns:
sanitized = re.sub(pattern, '', sanitized, flags=re.IGNORECASE)
return sanitized def validate_article(self, article: Dict) -> tuple[bool, str]:
"""验证文章内容安全性"""
errors = []
# 检查标题长度
title = article.get('title', '')
if len(title) < 5 or len(title) > 100:
errors.append("标题长度应在5-100字符之间")
# 检查内容长度
content = article.get('content', '')
if len(content) < 100:
errors.append("内容太短，至少100字符")
if len(content) > 50000:
errors.append("内容太长，最多50000字符")
# 检查标签数量 tags = article.get('tags', [])
if len(tags) > 5:
errors.append("最多只能设置5个标签")
# 检查敏感词
sensitive_words = self._load_sensitive_words()
for word in sensitive_words:
if word in title.lower() or word in content.lower():
errors.append(f"包含敏感词: {word}")
break
if errors:
return False, "; ".join(errors)
return True, "验证通过"
def _load_sensitive_words(self) -> list:
"""加载敏感词列表"""
# 可以从文件或数据库加载
return ["敏感词1", "敏感词2", "违禁词"]
# 使用示例
security = SecurityManager(secret_key="your-secret-key")
# 加密密码
encrypted_password = security.encrypt_password("user_password")
print(f"加密后的密码: {encrypted_password}")
# 生成Token
token = security.generate_token("user123", {"role": "admin"})
print(f"生成的Token: {token}")
# 验证Token
payload = security.verify_token(token)
if payload:
print(f"Token有效，用户: {payload['user_id']}")
else:
print("Token无效或已过期")
# 验证文章
article = {
"title": "测试文章",
"content": "这是一篇测试文章内容。",
"tags": ["测试", "Python"]
}
is_valid, message = security.validate_article(article)
print(f"文章验证: {is_valid}, 消息: {message}")
五、实际应用场景与案例
1. 技术博客自动化发布流水线
# blog_pipeline.py
import schedule
import time
from datetime import datetime
from pathlib import Path
import yaml
class BlogAutomationPipeline:
"""博客自动化发布流水线"""
def __init__(self, config_path: str = "config/pipeline.yaml"):
self.config = self.load_config(config_path)
self.publisher = CSDNAutoPublisher(
headless=self.config.get('headless', True)
)
self.monitor = PublisherMonitor()
self.security = SecurityManager(
self.config.get('secret_key', 'default-secret')
)
def load_config(self, config_path: str) -> dict:
"""加载流水线配置"""
with open(config_path, 'r', encoding='utf-8') as f:
return yaml.safe_load(f)
def discover_articles(self) -> list:
"""发现待发布的文章"""
articles_dir = Path(self.config['articles_dir'])
articles = []
for md_file in articles_dir.glob("**/*.md"):
# 检查Front Matter中的发布状态
content = md_file.read_text(encoding='utf-8')
if content.startswith('---'):
parts = content.split('---', 2)
if len(parts) >= 3:
frontmatter = yaml.safe_load(parts[1])
status = frontmatter.get('status', 'draft')
if status == 'ready':
articles.append({
'path': str(md_file),
'frontmatter': frontmatter,
'content': parts[2].strip()
})
return articles def process_article(self, article_info: dict) -> dict:
"""处理单篇文章"""
article_path = article_info['path']
frontmatter = article_info['frontmatter']
print(f"开始处理文章: {frontmatter.get('title', '无标题')}")
# 1. 安全验证
is_valid, message = self.security.validate_article({
'title': frontmatter.get('title', ''),
'content': article_info['content'],
'tags': frontmatter.get('tags', [])
})
if not is_valid:
return {
'success': False,
'error': f'安全验证失败: {message}',
'article': frontmatter.get('title')
}
# 2. 加载文章数据 article_data = self.publisher.load_markdown_with_frontmatter(article_path)
# 3. 开始监控 publish_key = self.monitor.track_publish(
article_id=frontmatter.get('id', str(hash(article_path))),
platform='csdn'
)
try:
# 4. 发布文章
result = self.publisher.publish_article(
article_data=article_data,
tags=frontmatter.get('tags', []),
category=frontmatter.get('category', '原创')
)
if result['success']:
# 5. 记录成功
self.monitor.record_success(publish_key, result)
# 6. 更新文章状态
self._update_article_status(article_path, 'published', result['url'])
return {
'success': True,
'url': result['url'],
'article': frontmatter.get('title')
}
else:
# 记录失败 self.monitor.record_error(publish_key, Exception(result.get('error', '未知错误')))
return {
'success': False,
'error': result.get('error'),
'article': frontmatter.get('title')
}
except Exception as e:
# 记录异常 self.monitor.record_error(publish_key, e)
return {
'success': False,
'error': str(e),
'article': frontmatter.get('title')
}
def _update_article_status(self, article_path: str, status: str, url: str = None):
"""更新文章发布状态"""
content = Path(article_path).read_text(encoding='utf-8')
if content.startswith('---'):
parts = content.split('---', 2)
frontmatter = yaml.safe_load(parts[1])
# 更新状态和URL
frontmatter['status'] = status
if url:
frontmatter['published_url'] = url frontmatter['last_published'] = datetime.now().isoformat()
# 重新写入文件 new_content = f"---\
{yaml.dump(frontmatter, allow_unicode=True)}---\
{parts[2]}"
Path(article_path).write_text(new_content, encoding='utf-8')
def run_pipeline(self):
"""运行发布流水线"""
print(f"开始运行博客发布流水线: {datetime.now()}")
# 发现待发布文章
articles = self.discover_articles()
print(f"发现 {len(articles)} 篇待发布文章")
results = []
for article_info in articles:
result = self.process_article(article_info)
results.append(result)
# 避免请求过于频繁
time.sleep(self.config.get('delay_between_posts', 10))
# 生成报告
report = self.generate_report(results)
self.send_notification(report)
return results def generate_report(self, results: list) -> dict:
"""生成发布报告"""
total = len(results)
successful = sum(1 for r in results if r['success'])
failed = total - successful
report = {
'timestamp': datetime.now().isoformat(),
'total_articles': total,
'successful': successful,
'failed': failed,
'success_rate': (successful / total * 100) if total > 0 else 0,
'details': results
}
# 保存报告 report_dir = Path(self.config.get('report_dir', 'reports'))
report_dir.mkdir(exist_ok=True)
report_file = report_dir / f"report_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
with open(report_file, 'w', encoding='utf-8') as f:
json.dump(report, f, ensure_ascii=False, indent=2)
return report
def send_notification(self, report: dict):
"""发送通知"""
notification_config = self.config.get('notifications', {})
if notification_config.get('email', {}).get('enabled', False):
self._send_email_notification(report, notification_config['email'])
if notification_config.get('webhook', {}).get('enabled', False):
self._send_webhook_notification(report, notification_config['webhook'])
def _send_email_notification(self, report: dict, config: dict):
"""发送邮件通知"""
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
msg = MIMEMultipart()
msg['From'] = config['from']
msg['To'] = ', '.join(config['to'])
msg['Subject'] = f"博客发布报告 - {datetime.now().strftime('%Y-%m-%d %H:%M')}"
# 构建邮件内容 content = f"""
博客发布完成报告：
总计文章：{report['total_articles']}
发布成功：{report['successful']}
发布失败：{report['failed']}
成功率：{report['success_rate']:.1f}%
详细结果：
{json.dumps(report['details'], ensure_ascii=False, indent=2)}
"""
msg.attach(MIMEText(content, 'plain'))
# 发送邮件
with smtplib.SMTP(config['smtp_server'], config['smtp_port']) as server:
server.starttls()
server.login(config['username'], config['password'])
server.send_message(msg)
def _send_webhook_notification(self, report: dict, config: dict):
"""发送Webhook通知"""
import requests
webhook_data = {
'text': f"博客发布完成: {report['successful']}/{report['total_articles']} 成功",
'attachments': [{
'title': '发布详情',
'text': json.dumps(report['details'], ensure_ascii=False, indent=2),
'color': 'good' if report['success_rate'] > 90 else 'warning'
}]
}
try:
response = requests.post(config['url'], json=webhook_data)
response.raise_for_status()
except Exception as e:
print(f"Webhook通知发送失败: {e}")
# 定时任务调度
def schedule_pipeline():
"""调度发布流水线"""
pipeline = BlogAutomationPipeline()
# 每天凌晨2点运行 schedule.every().day.at("02:00").do(pipeline.run_pipeline)
# 每小时检查一次
schedule.every().hour.do(lambda: print(f"定时检查: {datetime.now()}"))
print("博客自动化流水线已启动，按Ctrl+C停止")
while True:
schedule.run_pending()
time.sleep(60)
if __name__ == "__main__":
# 立即运行一次 pipeline = BlogAutomationPipeline()
results = pipeline.run_pipeline()
print(f"发布完成，结果: {results}")
# 或者启动定时任务 # schedule_pipeline()
2. 与CI/CD流水线集成
# .github/workflows/blog-publish.yml
name: Blog Auto Publish
on:
push:
branches: [ main ]
paths:
'articles/**'
schedule:
# 每天凌晨3点运行 - cron: '0 3 * * *'
workflow_dispatch: # 手动触发
jobs:
publish:
runs-on: ubuntu-latest
steps:
- name: Checkout code uses: actions/checkout@v3
- name: Set up Python uses: actions/setup-python@v4
with:
python-version: '3.9'
name: Install dependencies run: |
python -m pip install --upgrade pip
pip install -r requirements.txt
playwright install chromium - name: Configure environment
run: |
echo "CLAUDE_API_KEY=${{ secrets.CLAUDE_API_KEY }}" >> $GITHUB_ENV echo "CSDN_USERNAME=${{ secrets.CSDN_USERNAME }}" >> $GITHUB_ENV echo "CSDN_PASSWORD=${{ secrets.CSDN_PASSWORD }}" >> $GITHUB_ENV - name: Run tests run: |
python -m pytest tests/ -v
name: Discover articles id: discover
run: |
python scripts/discover_articles.py echo "::set-output name=article_count::$(cat article_count.txt)"
name: Publish articles
if: steps.discover.outputs.article_count > 0 run: |
python scripts/run_pipeline.py - name: Generate report
if: always()
run: |
python scripts/generate_report.py - name: Upload report
if: always()
uses: actions/upload-artifact@v3 with:
name: publish-report
path: reports/
name: Send notification if: always()
run: |
python scripts/send_notification.py
六、性能优化与故障处理
1. 性能优化策略
# performance_optimizer.py
import asyncio
import aiohttp
import concurrent.futures
from typing import List, Dict
import time
from functools import lru_cache
class PerformanceOptimizer:
"""性能优化器"""
def __init__(self, max_workers: int = 5):
self.max_workers = max_workers
self.cache = {}
async def batch_publish(self, articles: List[Dict]) -> List[Dict]:
"""批量并发发布文章"""
async with aiohttp.ClientSession() as session:
tasks = []
for article in articles:
task = self._publish_article_async(session, article)
tasks.append(task)
results = await asyncio.gather(*tasks, return_exceptions=True)
return self._process_results(results)
async def _publish_article_async(self, session: aiohttp.ClientSession, article: Dict) -> Dict:
"""异步发布单篇文章"""
# 实现异步发布逻辑 pass
def parallel_process_articles(self, articles: List[Dict]) -> List[Dict]:
"""并行处理文章（CPU密集型任务）"""
with concurrent.futures.ProcessPoolExecutor(
max_workers=self.max_workers ) as executor:
futures = []
for article in articles:
future = executor.submit(self._process_article_sync, article)
futures.append(future)
results = []
for future in concurrent.futures.as_completed(futures):
try:
result = future.result(timeout=300) # 5分钟超时
results.append(result)
except concurrent.futures.TimeoutError:
results.append({'error': '处理超时'})
except Exception as e:
results.append({'error': str(e)})
return results
@lru_cache(maxsize=100)
def _process_article_sync(self, article_content: str) -> Dict:
"""处理文章内容（可缓存）"""
# CPU密集型处理逻辑 pass
def optimize_images(self, html_content: str) -> str:
"""优化HTML中的图片"""
import re from io import BytesIO
from PIL import Image
# 查找图片标签
img_pattern = r'<img[^>]+src="([^">]+)"[^>]*>'
images = re.findall(img_pattern, html_content)
for img_url in images:
try:
# 下载并优化图片
optimized_url = self._optimize_single_image(img_url)
# 替换原URL
html_content = html_content.replace(img_url, optimized_url)
except Exception as e:
print(f"优化图片失败 {img_url}: {e}")
return html_content
def _optimize_single_image(self, image_url: str) -> str:
"""优化单张图片"""
# 实现图片压缩、格式转换等 pass
def monitor_performance(self):
"""性能监控"""
import psutil
import GPUtil
metrics = {
'timestamp': time.time(),
'cpu_percent': psutil.cpu_percent(interval=1),
'memory_percent': psutil.virtual_memory().percent,
'disk_usage': psutil.disk_usage('/').percent,
'network_io': psutil.net_io_counters()._asdict()
}
# GPU监控（如果可用）
try:
gpus = GPUtil.getGPUs()
metrics['gpu_usage'] = [gpu.load * 100 for gpu in gpus]
except:
metrics['gpu_usage'] = None return metricsclass ConnectionPool:
"""连接池管理"""
def __init__(self, max_size: int = 10):
self.max_size = max_size
self.pool = []
self.lock = asyncio.Lock()
async def get_connection(self):
"""获取连接"""
async with self.lock:
if self.pool:
return self.pool.pop()
elif len(self.pool) < self.max_size:
return await self._create_connection()
else:
# 等待连接释放
while not self.pool:
await asyncio.sleep(0.1)
return self.pool.pop()
async def release_connection(self, conn):
"""释放连接"""
async with self.lock:
self.pool.append(conn)
async def _create_connection(self):
"""创建新连接"""
# 实现连接创建逻辑
pass
2. 故障处理与恢复
# fault_tolerance.py
import time
from typing import Callable, Anyfrom functools import wraps
import logging
class FaultTolerance:
"""故障容错处理"""
def __init__(self, max_retries: int = 3, backoff_factor: float = 1.5,
exceptions_to_catch: tuple = (Exception,)):
self.max_retries = max_retries
self.backoff_factor = backoff_factor
self.exceptions_to_catch = exceptions_to_catch self.logger = logging.getLogger(__name__)
def retry(self, func: Callable) -> Callable:
"""重试装饰器"""
@wraps(func)
def wrapper(*args, **kwargs):
last_exception = None
for attempt in range(self.max_retries):
try:
return func(*args, **kwargs)
except self.exceptions_to_catch as e:
last_exception = e wait_time = self.backoff_factor ** attempt
self.logger.warning(
f"函数 {func.__name__} 第{attempt + 1}次尝试失败: {e}. "
f"{wait_time:.1f}秒后重试..."
)
if attempt < self.max_retries - 1:
time.sleep(wait_time)
else:
self.logger.error(
f"函数 {func.__name__} 达到最大重试次数"
)
raise last_exception raise last_exception
return wrapper
def circuit_breaker(self, failure_threshold: int = 5, recovery_timeout: int = 60):
"""熔断器装饰器"""
def decorator(func: Callable) -> Callable:
failures = 0
last_failure_time = 0 state = "CLOSED" # CLOSED, OPEN, HALF_OPEN @wraps(func)
def wrapper(*args, **kwargs):
nonlocal failures, last_failure_time, state current_time = time.time()
# 检查是否需要恢复
if state == "OPEN" and \
current_time - last_failure_time > recovery_timeout:
state = "HALF_OPEN"
self.logger.info(f"熔断器进入半开状态: {func.__name__}")
# 如果熔断器打开，直接失败
if state == "OPEN":
raise Exception(f"熔断器打开: {func.__name__}")
try:
result = func(*args, **kwargs)
# 成功调用，重置状态
if state == "HALF_OPEN":
state = "CLOSED"
failures = 0 self.logger.info(f"熔断器恢复关闭状态: {func.__name__}")
return result
except Exception as e:
failures += 1 last_failure_time = current_time
self.logger.error(
f"函数 {func.__name__} 调用失败: {e}. "
f"失败次数: {failures}/{failure_threshold}"
)
# 达到失败阈值，打开熔断器
if failures >= failure_threshold:
state = "OPEN"
self.logger.error(f"熔断器打开: {func.__name__}")
raise e
return wrapper
return decorator def fallback(self, fallback_func: Callable) -> Callable:
"""降级装饰器"""
def decorator(func: Callable) -> Callable:
@wraps(func)
def wrapper(*args, **kwargs):
try:
return func(*args, **kwargs)
except Exception as e:
self.logger.warning(
f"主函数 {func.__name__} 失败，使用降级函数: {e}"
)
return fallback_func(*args, **kwargs)
return wrapper
return decorator def timeout(self, timeout_seconds: int) -> Callable:
"""超时装饰器"""
def decorator(func: Callable) -> Callable:
@wraps(func)
def wrapper(*args, **kwargs):
import signal class TimeoutError(Exception):
pass def timeout_handler(signum, frame):
raise TimeoutError(f"函数 {func.__name__} 执行超时")
# 设置信号处理器
signal.signal(signal.SIGALRM, timeout_handler)
signal.alarm(timeout_seconds)
try:
result = func(*args, **kwargs)
signal.alarm(0) # 取消闹钟
return result
except TimeoutError as e:
self.logger.error(str(e))
raise e finally:
signal.alarm(0) # 确保取消闹钟 return wrapper return decorator
# 使用示例
fault_tolerance = FaultTolerance(max_retries=3)
class CSDNAPI:
"""带故障容错的CSDN API客户端"""
def __init__(self):
self.ft = fault_tolerance
@fault_tolerance.retry
@fault_tolerance.circuit_breaker(failure_threshold=3, recovery_timeout=30)
@fault_tolerance.timeout(30)
def publish_article(self, article: Dict) -> Dict:
"""发布文章（带重试、熔断、超时）"""
# 实际的发布逻辑
pass def _fallback_publish(self, article: Dict) -> Dict:
"""降级发布逻辑（如保存到本地）"""
self.logger.info("使用降级发布逻辑")
# 保存到本地文件
import json
filename = f"fallback_{int(time.time())}.json"
with open(filename, 'w', encoding='utf-8') as f:
json.dump(article, f, ensure_ascii=False, indent=2)
return {
'success': False,
'error': '发布失败，已保存到本地',
'fallback_file': filename }
@fault_tolerance.fallback(_fallback_publish)
def publish_with_fallback(self, article: Dict) -> Dict:
"""带降级的发布"""
return self.publish_article(article)
class RecoveryManager:
"""恢复管理器"""
def __init__(self, checkpoint_dir: str = "checkpoints"):
self.checkpoint_dir = Path(checkpoint_dir)
self.checkpoint_dir.mkdir(exist_ok=True)
self.logger = logging.getLogger(__name__)
def save_checkpoint(self, task_id: str, data: Dict):
"""保存检查点"""
checkpoint_file = self.checkpoint_dir / f"{task_id}.json"
checkpoint_data = {
'task_id': task_id,
'data': data,
'timestamp': time.time(),
'status': 'in_progress'
}
with open(checkpoint_file, 'w', encoding='utf-8') as f:
json.dump(checkpoint_data, f, ensure_ascii=False, indent=2)
self.logger.info(f"保存检查点: {task_id}")
def load_checkpoint(self, task_id: str) -> Optional[Dict]:
"""加载检查点"""
checkpoint_file = self.checkpoint_dir / f"{task_id}.json"
if checkpoint_file.exists():
with open(checkpoint_file, 'r', encoding='utf-8') as f:
checkpoint_data = json.load(f)
self.logger.info(f"加载检查点: {task_id}")
return checkpoint_data['data']
return None def mark_completed(self, task_id: str):
"""标记任务完成"""
checkpoint_file = self.checkpoint_dir / f"{task_id}.json"
if checkpoint_file.exists():
with open(checkpoint_file, 'r', encoding='utf-8') as f:
checkpoint_data = json.load(f)
checkpoint_data['status'] = 'completed'
checkpoint_data['completed_at'] = time.time()
with open(checkpoint_file, 'w', encoding='utf-8') as f:
json.dump(checkpoint_data, f, ensure_ascii=False, indent=2)
self.logger.info(f"标记任务完成: {task_id}")
def recover_failed_tasks(self):
"""恢复失败的任务"""
failed_tasks = []
for checkpoint_file in self.checkpoint_dir.glob("*.json"):
with open(checkpoint_file, 'r', encoding='utf-8') as f:
checkpoint_data = json.load(f)
if checkpoint_data['status'] == 'in_progress':
# 检查是否超时（超过1小时）
if time.time() - checkpoint_data['timestamp'] > 3600:
failed_tasks.append(checkpoint_data)
return failed_tasks
通过上述完整实现，Claude Code 结合自动化工具能够构建稳定、高效的 CSDN 内容发布系统。该系统不仅实现了基本的发布功能，还包含了错误处理、性能优化、监控告警等生产级特性，为技术博客的自动化管理提供了完整解决方案。
汇聚全球AI编程工具，助力开发者即刻编程。
更多推荐
9
0- 0
已为社区贡献3条内容
所有评论(0)