([^<]+)

# multi-writing-skills 多平台 Markdown 发布工具，支持微信公众号、知乎、今日头条。 ## 功能特性 - **Markdown 转换**: 支持基础模式、API 模式、AI 模式三种转换方式 - **多平台发布**: 一键发布到微信公众号、知乎、今日头条草稿箱 - **代码块格式化**: 自动处理缩进和换行，微信公众号完美展示代码 - **AI 写作助手**: 多种写作风格支持 - **AI 去痕**: 去除 AI 写作痕迹 - **图片生成**: 支持 OpenAI DALL-E、Gemini、ModelScope ## 触发条件当用户需要以下操作时自动触发： - 将 Markdown 文章发布到微信公众号、知乎或今日头条 - 转换 Markdown 为公众号格式 HTML - 修复微信公众号代码块显示问题 - 使用 AI 写作、改写或去痕 - 生成文章封面图片 ## 使用方式在 Claude Code 中直接用自然语言描述你的需求即可，例如： - "帮我把这篇 Markdown 发布到微信公众号" - "转换 article.md 为公众号格式" - "用 AI 写一篇关于 Python 的文章" - "给这篇文章去痕" 也可以手动调用： ``` /multi-writing-skills <任务描述> ``` ## openclaw CLI 如果需要使用命令行工具（openclaw）： ```bash # 初始化配置 openclaw config init # 转换并发布 openclaw convert article.md --draft --platform wechat # AI 写作 openclaw write "写一篇关于Python的文章" # AI 去痕 openclaw humanize article.md -o article_clean.md ``` ## 作者 yuesf ## 许可证 MIT FILE:CLAUDE.md # CLAUDE.md This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository. ## Project Overview `multi-writing-skills` 是一个多平台 Markdown 发布工具，支持微信公众号、知乎、今日头条的 Markdown 文章转换与发布。项目使用 Python 3.12+ 开发，约 4200 行代码。 ## Common Commands ```bash # 安装依赖 uv sync # 运行 CLI uv run multi-writing-skills --help # 运行测试 uv run pytest # 代码格式化 uv run black . # Linting uv run ruff check . # 类型检查 uv run mypy . ``` ## Architecture 项目采用模块化架构，主要模块如下： ``` src/multi_writing_skills/ ├── cli.py # CLI 入口，使用 typer 构建 ├── config.py # 配置管理（YAML 配置 + 环境变量） ├── platforms/ # 多平台发布模块 │ ├── base.py # 平台基类，定义发布接口 │ ├── wechat.py # 微信公众号发布 │ ├── zhihu.py # 知乎发布 │ └── toutiao.py # 今日头条发布 ├── converter/ # Markdown 转换模块 │ ├── wechat_style.py # 微信样式转换器 │ ├── themes.py # 内置主题管理 │ ├── css_theme.py # 自定义 CSS 主题 │ ├── api.py # API 模式（调用 mdnice 服务） │ └── ai.py # AI 模式转换 ├── image/ # AI 图片生成 │ └── providers/ # 支持 OpenAI DALL-E、Gemini、ModelScope ├── writer/ # AI 写作助手 └── humanizer/ # AI 去痕模块 ``` ## Key Design Patterns - **平台适配器模式**: `platforms/base.py` 定义抽象基类，各平台实现独立 - **配置分层**: 支持 YAML 配置文件 + 环境变量覆盖 - **多模式转换**: 内置模式/API 模式/AI 模式三种 Markdown 转换方式 FILE:README.md # multi-writing-skills 多平台 Markdown 发布工具，支持微信公众号、知乎、今日头条。 ## 功能特性 - **Markdown 转换**: 基础模式、API 模式、AI 模式 - **多平台发布**: 微信公众号、知乎、今日头条 - **代码块格式化**: 自动处理缩进和换行，微信公众号完美展示代码 - **图片上传**: 自动上传到各平台图床 - **AI 写作助手**: 5种写作风格支持 - **AI 去痕**: 去除 AI 写作痕迹 - **图片生成**: 支持 OpenAI DALL-E、Gemini、ModelScope ## 安装 ```bash pip install multi-writing-skills ``` 或使用 uv: ```bash uv tool install multi-writing-skills ``` ## 快速开始 ### 第一步配置凭证首次使用时，运行以下命令初始化配置： ```bash openclaw config init ``` 然后设置各平台的凭证： - **微信公众号**: 登录微信公众平台，获取 AppID 和 AppSecret - **知乎**: 登录知乎，从浏览器开发者工具复制 Cookie - **今日头条**: 登录头条号，从浏览器开发者工具复制 Cookie ```bash openclaw config set wechat.app_id <你的AppID> openclaw config set wechat.app_secret <你的AppSecret> ``` ### 第二步转换并发布把 Markdown 文章发布到微信公众号草稿箱： ```bash openclaw convert 你的文章.md --draft --platform wechat ``` 发布到知乎或今日头条： ```bash openclaw convert 你的文章.md --draft --platform zhihu openclaw convert 你的文章.md --draft --platform toutiao ``` ### 第三步发表文章打开对应平台的后台，就能看到文章已经在草稿箱里了。 ## AI 写作用自然语言描述你想写的内容： ```bash openclaw write "写一篇关于Python入门的教程" ``` 指定写作风格和长度： ```bash openclaw write "写一篇关于Python的文章" --style technical --length long ``` ## AI 去痕去除 AI 写作痕迹，让文章更自然： ```bash openclaw humanize article.md -o article_clean.md ``` 可选择轻度、中度、重度去痕强度。 ## 图片生成生成文章封面图片： ```bash openclaw generate-image "一只可爱的猫咪" --provider openai ``` 支持 OpenAI DALL-E、Gemini、ModelScope。 ## Claude Code 集成在 Claude Code 中，可以直接用自然语言描述需求： - "帮我把这篇 Markdown 发布到微信公众号" - "转换 article.md 为公众号格式" - "用 AI 写一篇关于 Python 的文章" - "给这篇文章去痕" ## 写作风格 | 风格 | 描述 | |------|------| | Dan Koe | 简洁、直接、实用 | | 技术风格 | 严谨、详细、专业 | | 随意风格 | 轻松、亲切、口语化 | | 正式风格 | 严肃、规范、学术 | | 故事风格 | 叙事、引人入胜 | ## 排版主题内置多种主题：默认、橙色、蓝色、绿色、紫色、简约。使用主题发布： ```bash openclaw convert article.md --draft --platform wechat --theme blue ``` ## 开发 ```bash git clone <仓库地址> cd wechat-publisher uv sync uv run openclaw --help ``` ## 发布到 clawhub ```bash clawhub publish ./multi-writing-skills \ --slug multi-writing-skills \ --name "multi-writing-skills" \ --version 1.1.1 \ --tags latest \ --changelog "新增 AI 图片生成功能说明" ``` ## 作者 yuesf ## 许可证 MIT FILE:pyproject.toml [project] name = "multi-writing-skills" version = "0.1.0" description = "多平台 Markdown 发布工具，支持微信公众号、知乎、今日头条" readme = "README.md" requires-python = ">=3.12" license = { text = "MIT" } authors = [ { name = "yuesf", email = "yuesf@example.com" } ] keywords = ["writing", "markdown", "publisher", "wechat", "zhihu", "toutiao"] classifiers = [ "Development Status :: 3 - Alpha", "Intended Audience :: Developers", "License :: OSI Approved :: MIT License", "Programming Language :: Python :: 3", "Programming Language :: Python :: 3.12", ] dependencies = [ "typer>=0.9.0", "pydantic>=2.0", "pydantic-settings>=2.0", "httpx>=0.27", "Pillow>=10.0", "markdown-it-py>=3.0", "jinja2>=3.1", "pyyaml>=6.0", "rich>=13.0", "linkify-it-py>=2.1.0", ] [project.optional-dependencies] dev = [ "pytest>=8.0", "pytest-asyncio>=0.23", "black>=24.0", "ruff>=0.3", "mypy>=1.8", ] [project.scripts] multi-writing-skills = "multi_writing_skills.cli:app" [build-system] requires = ["hatchling"] build-backend = "hatchling.build" [tool.hatch.build.targets.wheel] packages = ["src/multi_writing_skills"] [tool.ruff] line-length = 100 target-version = "py312" [tool.black] line-length = 100 target-version = ["py312"] [tool.mypy] python_version = "3.12" strict = true [tool.pytest.ini_options] asyncio_mode = "auto" FILE:src/multi_writing_skills/__init__.py """ multi-writing-skills - 多平台 Markdown 发布工具支持微信公众号、知乎、今日头条等多个平台。 """ __version__ = "0.1.0" from .cli import app __all__ = ["app"] FILE:src/multi_writing_skills/cli.py """ CLI 入口 """ import asyncio from pathlib import Path from typing import Optional import typer from rich.console import Console from rich.table import Table from .config import settings from .platforms import WeChatPlatform, ZhihuPlatform, ToutiaoPlatform, registry app = typer.Typer( name="multi-writing-skills", help="多平台 Markdown 发布工具，支持微信公众号、知乎、今日头条", ) console = Console() def init_registry() -> None: """初始化平台注册表""" settings.load() # 注册微信平台 if settings.is_wechat_configured(): wechat = WeChatPlatform( app_id=settings.wechat.app_id, app_secret=settings.wechat.app_secret, ) registry.register(wechat) # 注册知乎平台 if settings.is_zhihu_configured(): zhihu = ZhihuPlatform(cookie=settings.zhihu.cookie) registry.register(zhihu) # 注册今日头条平台 if settings.is_toutiao_configured(): toutiao = ToutiaoPlatform(cookie=settings.toutiao.cookie) registry.register(toutiao) @app.callback() def main() -> None: """初始化""" init_registry() # 配置命令组 config_app = typer.Typer(help="配置管理") app.add_typer(config_app, name="config") @config_app.command("init") def config_init() -> None: """初始化配置文件""" settings.save() console.print(f"[green]配置文件已创建: {settings.get_config_file()}[/green]") @config_app.command("show") def config_show() -> None: """显示当前配置""" settings.load() table = Table(title="配置信息") table.add_column("配置项", style="cyan") table.add_column("值", style="green") table.add_row("配置文件", str(settings.get_config_file())) table.add_row("微信 AppID", settings.wechat.app_id or "[red]未配置[/red]") table.add_row("微信 Secret", "***" if settings.wechat.app_secret else "[red]未配置[/red]") table.add_row("知乎", "已配置" if settings.is_zhihu_configured() else "[red]未配置[/red]") table.add_row("头条", "已配置" if settings.is_toutiao_configured() else "[red]未配置[/red]") table.add_row("AI Provider", settings.ai.provider or "[red]未配置[/red]") table.add_row("AI Model", settings.ai.model or "[red]未配置[/red]") table.add_row("API 端点", settings.api_endpoint or "[red]未配置[/red]") console.print(table) @config_app.command("set") def config_set( key: str = typer.Argument(..., help="配置键，格式: section.key"), value: str = typer.Argument(..., help="配置值"), ) -> None: """设置配置项示例: multi-writing-skills config set wechat.app_id your_app_id multi-writing-skills config set wechat.app_secret your_secret multi-writing-skills config set ai.api_key your_api_key """ settings.load() parts = key.split(".") if len(parts) != 2: console.print("[red]错误: 配置键格式应为 section.key[/red]") raise typer.Exit(1) section, subkey = parts if section == "wechat": if subkey == "app_id": settings.wechat.app_id = value elif subkey == "app_secret": settings.wechat.app_secret = value else: console.print(f"[red]未知配置项: {key}[/red]") raise typer.Exit(1) elif section == "zhihu": if subkey == "cookie": settings.zhihu.cookie = value else: console.print(f"[red]未知配置项: {key}[/red]") raise typer.Exit(1) elif section == "toutiao": if subkey == "cookie": settings.toutiao.cookie = value else: console.print(f"[red]未知配置项: {key}[/red]") raise typer.Exit(1) elif section == "ai": if subkey == "provider": settings.ai.provider = value elif subkey == "api_key": settings.ai.api_key = value elif subkey == "base_url": settings.ai.base_url = value elif subkey == "model": settings.ai.model = value else: console.print(f"[red]未知配置项: {key}[/red]") raise typer.Exit(1) else: console.print(f"[red]未知配置节: {section}[/red]") raise typer.Exit(1) settings.save() console.print(f"[green]已设置 {key}[/green]") # 转换命令 @app.command() def convert( file: Path = typer.Argument(..., help="Markdown 或 HTML 文件路径", exists=True), platform: str = typer.Option("wechat", "--platform", "-p", help="发布平台 (wechat/zhihu/toutiao)"), draft: bool = typer.Option(False, "--draft", "-d", help="发布到草稿箱"), preview: bool = typer.Option(False, "--preview", help="预览 HTML 输出"), cover: Optional[str] = typer.Option(None, "--cover", "-c", help="封面图片路径"), author: Optional[str] = typer.Option(None, "--author", "-a", help="作者"), output: Optional[Path] = typer.Option(None, "--output", "-o", help="输出文件路径"), theme: str = typer.Option("default", "--theme", "-t", help="排版主题名称"), css: Optional[str] = typer.Option(None, "--css", help="CSS 文件路径或 URL（兼容 wenyan-cli 格式）"), use_api: bool = typer.Option(False, "--api", help="使用 mdnice API 模式转换（推荐，排版更精美）"), use_ai: bool = typer.Option(False, "--ai", help="使用 AI 模式转换"), pdf: bool = typer.Option(False, "--pdf", help="输出为 PDF 文件（需要 ReportLab）"), ) -> None: """转换 Markdown 或直接发布 HTML 到指定平台文件格式支持： - .md/.markdown: Markdown 文件，会转换为 HTML - .html/.htm: HTML 文件，直接使用，不进行转换主题支持： - 内置主题: default/orange/blue/green/purple/simple - CSS 主题: 使用 --css 指定 CSS 文件（兼容 wenyan-cli 格式）转换模式说明（仅 Markdown）： - 默认模式：使用内置样式转换，速度快，适合离线使用 - --api 模式：调用 mdnice API，排版更精美，推荐用于正式发布 - --ai 模式：使用 AI 生成样式，需要配置 AI API Key """ if not file.exists(): console.print(f"[red]文件不存在: {file}[/red]") raise typer.Exit(1) # 读取文件内容 content = file.read_text(encoding="utf-8") title = file.stem # 判断文件类型 file_ext = file.suffix.lower() is_html = file_ext in (".html", ".htm") console.print(f"[cyan]处理文件: {file}[/cyan]") # HTML 文件直接使用 if is_html: console.print("[cyan]HTML 文件，直接使用...[/cyan]") html_content = content # 从 HTML 中提取标题 import re title_match = re.search(r"]*>([^<]+)", content) if title_match: title = title_match.group(1) else: title_match = re.search(r"([^<]+)", content) if title_match: title = title_match.group(1) else: # Markdown 文件尝试从内容提取标题（第一个 # 开头的行） import re # 匹配 # 标题，支持 # Title 或者 ## Title，取第一个 title_match = re.search(r'^\s*#+\s+(.+)$', content, re.MULTILINE) if title_match: title = title_match.group(1).strip() # 移除标题行，避免正文中重复显示标题 content = re.sub(r'^\s*#+\s+.+\n?', '', content, count=1, flags=re.MULTILINE) # Markdown 转换 html_content: str first_platform = platform.split(",")[0].strip() if platform else "default" # 处理主题：支持 CSS 文件 actual_theme = theme css_path = css if css_path: console.print(f"[cyan]使用 CSS 主题: {css_path}[/cyan]") if use_api: from .converter import MarkdownConverter console.print("[cyan]使用 mdnice API 模式转换...[/cyan]") converter = MarkdownConverter() result = converter.convert( content, title, platform=first_platform, theme=theme, use_api=True, api_endpoint=settings.api_endpoint or None, ) html_content = result.html elif use_ai and settings.ai.api_key: from .converter.ai import convert_with_ai console.print("[cyan]使用 AI 模式转换...[/cyan]") html_content = asyncio.run( convert_with_ai( content, api_key=settings.ai.api_key, provider=settings.ai.provider, base_url=settings.ai.base_url, model=settings.ai.model, ) ) else: from .converter import MarkdownConverter if css_path: console.print(f"[cyan]使用 CSS 主题转换: {css_path}[/cyan]") else: console.print(f"[cyan]使用主题 '{actual_theme}' 转换...[/cyan]") converter = MarkdownConverter() result = converter.convert( content, title, platform=first_platform, theme=actual_theme, css_path=css_path, ) html_content = result.html if preview: console.print("\n[yellow]预览 HTML:[/yellow]") console.print(html_content[:500] + "..." if len(html_content) > 500 else html_content) if output: output.write_text(html_content, encoding="utf-8") console.print(f"[green]已保存到: {output}[/green]") # PDF 输出（仅支持 Markdown 文件） if pdf: if is_html: console.print("[yellow]PDF 输出仅支持 Markdown 文件，HTML 文件请先转换回 Markdown[/yellow]") else: try: from .pdf.converter import markdown_to_pdf except ImportError: console.print("[red]PDF 支持需要安装 ReportLab：pip install reportlab[/red]") raise typer.Exit(1) pdf_output = output.with_suffix(".pdf") if output else Path(file.stem + ".pdf") ok = markdown_to_pdf(content, str(pdf_output), title=title) if ok: console.print(f"[green]PDF 生成成功：{pdf_output}[/green]") else: console.print(f"[red]PDF 生成失败[/red]") raise typer.Exit(1) if draft: # 支持多平台发布 platforms = platform.split(",") for plat_name in platforms: plat = registry.get(plat_name.strip()) if not plat: console.print(f"[red]未知平台: {plat_name}[/red]") continue if not plat.is_configured(): console.print(f"[red]平台 {plat_name} 未配置，请先运行 config set 设置凭证[/red]") continue from .platforms import PublishRequest request = PublishRequest( title=title, content=html_content, cover=cover, author=author, ) result = asyncio.run(plat.publish(request)) if result.success: console.print(f"[green]发布到 {plat.display_name} 成功！[/green]") if result.media_id: console.print(f" Media ID: {result.media_id}") if result.article_url: console.print(f" URL: {result.article_url}") else: console.print(f"[red]发布到 {plat.display_name} 失败: {result.message}[/red]") # 发布命令 @app.command() def publish( file: Path = typer.Argument(..., help="HTML 文件路径", exists=True), platform: str = typer.Option("wechat", "--platform", "-p", help="发布平台 (wechat/zhihu/toutiao)"), title: Optional[str] = typer.Option(None, "--title", "-t", help="文章标题（默认从文件名提取）"), cover: Optional[str] = typer.Option(None, "--cover", "-c", help="封面图片路径"), author: Optional[str] = typer.Option(None, "--author", "-a", help="作者"), ) -> None: """发布 HTML 文件到指定平台直接发布 HTML 文件，不进行任何转换。适用于已准备好的 HTML 内容。 """ if not file.exists(): console.print(f"[red]文件不存在: {file}[/red]") raise typer.Exit(1) # 读取 HTML 内容 html_content = file.read_text(encoding="utf-8") # 确定标题 article_title = title or file.stem if not title: # 尝试从 HTML 中提取标题 import re title_match = re.search(r"]*>([^<]+)", html_content) if title_match: article_title = title_match.group(1) else: title_match = re.search(r"([^<]+)", html_content) if title_match: article_title = title_match.group(1) console.print(f"[cyan]发布 HTML 文件: {file}[/cyan]") console.print(f"[cyan]标题: {article_title}[/cyan]") # 发布到平台 platforms = platform.split(",") for plat_name in platforms: plat = registry.get(plat_name.strip()) if not plat: console.print(f"[red]未知平台: {plat_name}[/red]") continue if not plat.is_configured(): console.print(f"[red]平台 {plat_name} 未配置，请先运行 config set 设置凭证[/red]") continue from .platforms import PublishRequest request = PublishRequest( title=article_title, content=html_content, cover=cover, author=author, ) result = asyncio.run(plat.publish(request)) if result.success: console.print(f"[green]发布到 {plat.display_name} 成功！[/green]") if result.media_id: console.print(f" Media ID: {result.media_id}") if result.article_url: console.print(f" URL: {result.article_url}") else: console.print(f"[red]发布到 {plat.display_name} 失败: {result.message}[/red]") # 写作命令 @app.command() def write( topic: str = typer.Argument(..., help="写作主题"), style: str = typer.Option("dan-koe", "--style", "-s", help="写作风格"), length: str = typer.Option("medium", "--length", "-l", help="文章长度 (short/medium/long)"), context: Optional[str] = typer.Option(None, "--context", help="额外上下文"), cover_prompt: bool = typer.Option(False, "--cover-prompt", help="生成封面提示词"), ) -> None: """AI 写作助手""" if not settings.ai.api_key: console.print("[red]请先配置 AI API Key: multi-writing-skills config set ai.api_key your_key[/red]") raise typer.Exit(1) from .writer import AIWriter writer = AIWriter( api_key=settings.ai.api_key, provider=settings.ai.provider, base_url=settings.ai.base_url, model=settings.ai.model, ) try: if cover_prompt: console.print(f"[cyan]生成封面提示词...[/cyan]") prompt = asyncio.run(writer.generate_cover_prompt(topic)) console.print(f"\n[green]封面提示词:[/green]") console.print(prompt) else: console.print(f"[cyan]使用 {style} 风格写作...[/cyan]") article = asyncio.run( writer.write( topic=topic, style=style, length=length, context=context, ) ) console.print(f"\n[green]生成的文章:[/green]") console.print(article) finally: asyncio.run(writer.close()) @app.command() def rewrite( file: Path = typer.Argument(..., help="Markdown 文件路径", exists=True), style: str = typer.Option("dan-koe", "--style", "-s", help="目标写作风格"), keep_structure: bool = typer.Option(True, "--keep-structure/--no-keep-structure", help="是否保留原有结构"), output: Optional[Path] = typer.Option(None, "--output", "-o", help="输出文件路径"), ) -> None: """使用指定风格改写文章""" if not settings.ai.api_key: console.print("[red]请先配置 AI API Key[/red]") raise typer.Exit(1) content = file.read_text(encoding="utf-8") from .writer import AIWriter writer = AIWriter( api_key=settings.ai.api_key, provider=settings.ai.provider, base_url=settings.ai.base_url, model=settings.ai.model, ) try: console.print(f"[cyan]使用 {style} 风格改写...[/cyan]") rewritten = asyncio.run( writer.rewrite( content=content, style=style, keep_structure=keep_structure, ) ) if output: output.write_text(rewritten, encoding="utf-8") console.print(f"[green]已保存到: {output}[/green]") else: console.print(f"\n[green]改写后的文章:[/green]") console.print(rewritten) finally: asyncio.run(writer.close()) # AI 去痕命令 @app.command() def humanize( file: Path = typer.Argument(..., help="Markdown 文件路径", exists=True), intensity: str = typer.Option("medium", "--intensity", "-i", help="处理强度 (light/medium/heavy)"), output: Optional[Path] = typer.Option(None, "--output", "-o", help="输出文件路径"), ) -> None: """AI 写作去痕""" if not settings.ai.api_key: console.print("[red]请先配置 AI API Key[/red]") raise typer.Exit(1) content = file.read_text(encoding="utf-8") from .humanizer import Humanizer humanizer = Humanizer( api_key=settings.ai.api_key, provider=settings.ai.provider, base_url=settings.ai.base_url, model=settings.ai.model, ) try: console.print(f"[cyan]执行 AI 去痕 (强度: {intensity})...[/cyan]") result = asyncio.run(humanizer.humanize(content, intensity)) if output: output.write_text(result.humanized, encoding="utf-8") console.print(f"[green]已保存到: {output}[/green]") else: console.print(f"\n[green]去痕后的文章:[/green]") console.print(result.humanized) if result.changes: console.print(f"\n[yellow]主要变化:[/yellow]") for change in result.changes: console.print(f" - {change}") finally: asyncio.run(humanizer.close()) # 图片生成命令 @app.command("generate-image") def generate_image( prompt: str = typer.Argument(..., help="图片描述"), provider: str = typer.Option("openai", "--provider", "-p", help="图片生成 Provider (openai/gemini/modelscope/minimax/zhipu/doubao)"), size: str = typer.Option("1024x1024", "--size", "-s", help="图片尺寸"), style: Optional[str] = typer.Option(None, "--style", help="图片风格"), output: Optional[Path] = typer.Option(None, "--output", "-o", help="输出文件路径"), ) -> None: """AI 图片生成""" import os # MiniMax 支持使用 MINIMAX_API_KEY 环境变量 if provider == "minimax": if not os.environ.get("MINIMAX_API_KEY") and not settings.ai.api_key: console.print("[red]请设置 MINIMAX_API_KEY 环境变量[/red]") raise typer.Exit(1) elif not settings.ai.api_key: console.print("[red]请先配置 AI API Key[/red]") raise typer.Exit(1) console.print(f"[cyan]使用 {provider} 生成图片...[/cyan]") async def generate() -> tuple[bool, str, Optional[str]]: if provider == "openai": from .image.providers.openai import OpenAIProvider p = OpenAIProvider(api_key=settings.ai.api_key, model="dall-e-3") result = await p.generate(prompt, size, style) await p.close() return result.success, result.message or "", result.image_url elif provider == "gemini": from .image.providers.gemini import GeminiProvider p = GeminiProvider(api_key=settings.ai.api_key) result = await p.generate(prompt, size, style) await p.close() return result.success, result.message or "", result.local_path or result.image_url elif provider == "modelscope": from .image.providers.modelscope import ModelScopeProvider p = ModelScopeProvider(api_key=settings.ai.api_key) result = await p.generate(prompt, size, style) await p.close() return result.success, result.message or "", result.image_url elif provider == "minimax": from .image.providers.minimax import MiniMaxProvider api_key = os.environ.get("MINIMAX_API_KEY") or settings.ai.api_key p = MiniMaxProvider(api_key=api_key) result = await p.generate(prompt, size, style) await p.close() return result.success, result.message or "", result.image_url elif provider == "zhipu": from .image.providers.zhipu import ZhipuProvider p = ZhipuProvider(api_key=settings.ai.api_key) result = await p.generate(prompt, size, style) await p.close() return result.success, result.message or "", result.image_url elif provider == "doubao": from .image.providers.doubao import DoubaoProvider p = DoubaoProvider(api_key=settings.ai.api_key) result = await p.generate(prompt, size, style) await p.close() return result.success, result.message or "", result.image_url else: return False, f"未知 Provider: {provider}", None success, message, image_url = asyncio.run(generate()) if success: console.print(f"[green]图片生成成功！[/green]") if image_url: console.print(f" URL: {image_url}") if output and image_url: # 下载图片到本地 import httpx async def download() -> None: async with httpx.AsyncClient() as client: resp = await client.get(image_url) output.write_bytes(resp.content) asyncio.run(download()) console.print(f" 已保存到: {output}") else: console.print(f"[red]图片生成失败: {message}[/red]") # 平台命令组 platform_app = typer.Typer(help="平台管理") app.add_typer(platform_app, name="platform") @platform_app.command("list") def platform_list() -> None: """列出所有支持的平台""" table = Table(title="支持的平台") table.add_column("平台", style="cyan") table.add_column("标识", style="yellow") table.add_column("状态", style="green") platforms_info = [ ("微信公众号", "wechat", settings.is_wechat_configured()), ("知乎", "zhihu", settings.is_zhihu_configured()), ("今日头条", "toutiao", settings.is_toutiao_configured()), ] for name, ident, configured in platforms_info: status = "[green]已配置[/green]" if configured else "[red]未配置[/red]" table.add_row(name, ident, status) console.print(table) @platform_app.command("test") def platform_test( platform: str = typer.Argument(..., help="平台标识 (wechat/zhihu/toutiao)"), ) -> None: """测试平台连接""" plat = registry.get(platform) if not plat: console.print(f"[red]未知平台: {platform}[/red]") raise typer.Exit(1) if not plat.is_configured(): console.print(f"[red]平台 {platform} 未配置[/red]") raise typer.Exit(1) console.print(f"[cyan]测试 {plat.display_name} 连接...[/cyan]") result = asyncio.run(plat.validate_credentials()) if result: console.print(f"[green]{plat.display_name} 连接成功！[/green]") else: console.print(f"[red]{plat.display_name} 连接失败[/red]") # 写作风格命令 @app.command("styles") def list_styles() -> None: """列出所有写作风格""" from .writer import AIWriter # 创建临时 writer 实例 writer = AIWriter( api_key=settings.ai.api_key or "temp", provider=settings.ai.provider, ) table = Table(title="写作风格") table.add_column("标识", style="cyan") table.add_column("名称", style="yellow") table.add_column("描述", style="green") for style in writer.list_styles(): table.add_row(style["name"], style["display_name"], style["description"]) console.print(table) # 排版主题命令 @app.command("themes") def list_themes() -> None: """列出所有排版主题（内置 + 自定义 CSS）""" from .converter.themes import list_themes as get_themes_list table = Table(title="排版主题") table.add_column("标识", style="cyan") table.add_column("名称", style="yellow") table.add_column("类型", style="magenta") table.add_column("描述", style="green") for theme in get_themes_list(): theme_type = theme.get("type", "builtin") type_display = "内置" if theme_type == "builtin" else "CSS" table.add_row(theme["name"], theme["display_name"], type_display, theme["description"]) console.print(table) console.print("\n[cyan]提示: 使用 --css 参数可指定自定义 CSS 文件（兼容 wenyan-cli 格式）[/cyan]") # 主题管理命令组 theme_app = typer.Typer(help="主题管理") app.add_typer(theme_app, name="theme") @theme_app.command("list") def theme_list() -> None: """列出所有主题""" from .converter.themes import list_themes as get_themes_list table = Table(title="排版主题") table.add_column("标识", style="cyan") table.add_column("名称", style="yellow") table.add_column("类型", style="magenta") table.add_column("描述", style="green") for theme in get_themes_list(): theme_type = theme.get("type", "builtin") type_display = "内置" if theme_type == "builtin" else "CSS" table.add_row(theme["name"], theme["display_name"], type_display, theme["description"]) console.print(table) @theme_app.command("add") def theme_add( name: str = typer.Option(..., "--name", "-n", help="主题名称（唯一标识）"), path: str = typer.Option(..., "--path", "-p", help="CSS 文件路径或 URL"), ) -> None: """添加自定义 CSS 主题（兼容 wenyan-cli 格式）""" from pathlib import Path from .converter.themes import register_theme, THEME_DIR import shutil # 确保主题目录存在 THEME_DIR.mkdir(parents=True, exist_ok=True) # 如果是本地文件，复制到主题目录 if not path.startswith(('http://', 'https://')): src_path = Path(path) if not src_path.exists(): console.print(f"[red]文件不存在: {path}[/red]") raise typer.Exit(1) dest_path = THEME_DIR / f"{name}.css" shutil.copy(src_path, dest_path) console.print(f"[green]已复制 CSS 文件到: {dest_path}[/green]") path = str(dest_path) # 注册主题 try: register_theme(path, name) console.print(f"[green]主题 '{name}' 已添加[/green]") except Exception as e: console.print(f"[red]添加主题失败: {e}[/red]") raise typer.Exit(1) @theme_app.command("remove") def theme_remove( name: str = typer.Argument(..., help="要删除的主题名称"), ) -> None: """删除自定义 CSS 主题""" from pathlib import Path from .converter.themes import THEMES, THEME_DIR # 检查是否是内置主题 if name in ["default", "orange", "blue", "green", "purple", "simple"]: console.print(f"[red]不能删除内置主题: {name}[/red]") raise typer.Exit(1) # 删除 CSS 文件 css_file = THEME_DIR / f"{name}.css" if css_file.exists(): css_file.unlink() console.print(f"[green]已删除主题文件: {css_file}[/green]") # 从注册表移除 if name in THEMES: del THEMES[name] console.print(f"[green]主题 '{name}' 已删除[/green]") else: console.print(f"[yellow]主题 '{name}' 不存在[/yellow]") if __name__ == "__main__": app() FILE:src/multi_writing_skills/config.py """ 配置管理 """ import os from pathlib import Path from typing import Optional import yaml from pydantic import BaseModel from pydantic_settings import BaseSettings def is_openclaw_env() -> bool: """检测是否在 openclaw 环境中运行""" openclaw_dir = Path.home() / ".openclaw" return openclaw_dir.exists() and openclaw_dir.is_dir() def load_openclaw_env() -> dict[str, str]: """从 ~/.openclaw/.env 加载环境变量""" env_file = Path.home() / ".openclaw" / ".env" env_vars = {} if env_file.exists(): with open(env_file, encoding="utf-8") as f: for line in f: line = line.strip() if line and not line.startswith("#"): if "=" in line: key, value = line.split("=", 1) env_vars[key.strip()] = value.strip() return env_vars class WeChatConfig(BaseModel): """微信公众号配置""" app_id: str = "" app_secret: str = "" class ZhihuConfig(BaseModel): """知乎配置""" cookie: str = "" class ToutiaoConfig(BaseModel): """今日头条配置""" cookie: str = "" class AIConfig(BaseModel): """AI 配置""" provider: str = "openai" api_key: str = "" base_url: str = "" model: str = "gpt-4" class Settings(BaseSettings): """应用配置""" # 配置文件路径 config_dir: Path = Path.home() / ".multi-writing-skills" # 平台配置 wechat: WeChatConfig = WeChatConfig() zhihu: ZhihuConfig = ZhihuConfig() toutiao: ToutiaoConfig = ToutiaoConfig() # AI 配置 ai: AIConfig = AIConfig() # Markdown 转换配置 api_endpoint: str = "" # API 模式的端点 default_theme: str = "default" class Config: env_prefix = "WRITING_PUBLISHER_" env_nested_delimiter = "_" def get_config_file(self) -> Path: """获取配置文件路径""" return self.config_dir / "config.yaml" def load(self) -> None: """从文件和环境变量加载配置""" # 如果在 openclaw 环境中，先加载 ~/.openclaw/.env openclaw_env = {} if is_openclaw_env(): openclaw_env = load_openclaw_env() # 先从配置文件加载 config_file = self.get_config_file() if config_file.exists(): with open(config_file, encoding="utf-8") as f: data = yaml.safe_load(f) or {} if "wechat" in data: self.wechat = WeChatConfig(**data["wechat"]) if "zhihu" in data: self.zhihu = ZhihuConfig(**data["zhihu"]) if "toutiao" in data: self.toutiao = ToutiaoConfig(**data["toutiao"]) if "ai" in data: self.ai = AIConfig(**data["ai"]) if "api_endpoint" in data: self.api_endpoint = data["api_endpoint"] if "default_theme" in data: self.default_theme = data["default_theme"] # 环境变量优先级更高（先检查 openclaw 环境变量，再检查系统环境变量） # 微信公众号配置 wechat_app_id = openclaw_env.get("WECHAT_APP_ID") or os.environ.get("WECHAT_APP_ID") if wechat_app_id is not None: self.wechat.app_id = wechat_app_id wechat_app_secret = openclaw_env.get("WECHAT_APP_SECRET") or os.environ.get("WECHAT_APP_SECRET") if wechat_app_secret is not None: self.wechat.app_secret = wechat_app_secret # 知乎配置 zhihu_cookie = openclaw_env.get("ZHIHU_COOKIE") or os.environ.get("ZHIHU_COOKIE") if zhihu_cookie is not None: self.zhihu.cookie = zhihu_cookie # 今日头条配置 toutiao_cookie = openclaw_env.get("TOUTIAO_COOKIE") or os.environ.get("TOUTIAO_COOKIE") if toutiao_cookie is not None: self.toutiao.cookie = toutiao_cookie # AI 配置 ai_provider = openclaw_env.get("AI_PROVIDER") or os.environ.get("AI_PROVIDER") if ai_provider is not None: self.ai.provider = ai_provider ai_api_key = openclaw_env.get("AI_API_KEY") or os.environ.get("AI_API_KEY") if ai_api_key is not None: self.ai.api_key = ai_api_key ai_base_url = openclaw_env.get("AI_BASE_URL") or os.environ.get("AI_BASE_URL") if ai_base_url is not None: self.ai.base_url = ai_base_url ai_model = openclaw_env.get("AI_MODEL") or os.environ.get("AI_MODEL") if ai_model is not None: self.ai.model = ai_model def save(self) -> None: """保存配置到文件""" self.config_dir.mkdir(parents=True, exist_ok=True) config_file = self.get_config_file() data = { "wechat": self.wechat.model_dump(), "zhihu": self.zhihu.model_dump(), "toutiao": self.toutiao.model_dump(), "ai": self.ai.model_dump(), "api_endpoint": self.api_endpoint, "default_theme": self.default_theme, } with open(config_file, "w", encoding="utf-8") as f: yaml.dump(data, f, allow_unicode=True, default_flow_style=False) def is_wechat_configured(self) -> bool: return bool(self.wechat.app_id and self.wechat.app_secret) def is_zhihu_configured(self) -> bool: return bool(self.zhihu.cookie) def is_toutiao_configured(self) -> bool: return bool(self.toutiao.cookie) # 全局配置实例 settings = Settings() FILE:src/multi_writing_skills/converter/__init__.py """ Markdown 转换器支持 API 模式和 AI 模式转换 Markdown 为各平台兼容的 HTML。支持 wenyan-cli 格式的 CSS 主题。 """ from pathlib import Path from typing import Optional from markdown_it import MarkdownIt from pydantic import BaseModel from .wechat_style import WeChatStyleConverter from .themes import get_theme, list_themes class ConvertOptions(BaseModel): """转换选项""" theme: str = "default" api_endpoint: Optional[str] = None highlight_code: bool = True image_host: Optional[str] = None platform: str = "default" css_path: Optional[str] = None class ConvertResult(BaseModel): """转换结果""" title: str html: str images: list[str] = [] class MarkdownConverter: """Markdown 转换器""" # mdnice API 端点 MDNICE_API = "https://api.mdnice.com/api/v1/markdown" def __init__(self, options: Optional[ConvertOptions] = None) -> None: self.options = options or ConvertOptions() self._md = MarkdownIt("gfm-like", {"html": True, "linkify": True}) def convert( self, content: str, title: Optional[str] = None, platform: str = "default", theme: str = "default", use_api: bool = False, api_endpoint: Optional[str] = None, css_path: Optional[str] = None, ) -> ConvertResult: """转换 Markdown 为 HTML Args: content: Markdown 内容 title: 文章标题 platform: 目标平台 (default/wechat) theme: 排版主题名称 use_api: 是否使用 API 模式 api_endpoint: API 端点 css_path: CSS 文件路径（兼容 wenyan-cli 格式） """ # 提取标题 extracted_title = title lines = content.split("\n") for line in lines: if line.startswith("# "): extracted_title = line[2:].strip() break if not extracted_title: extracted_title = "无标题文章" # 选择转换方式 if platform == "wechat": if use_api: # API 模式：调用 mdnice 或自定义 API html = self._convert_via_api(content, api_endpoint, theme) else: # 本地样式转换（支持 CSS 主题） converter = WeChatStyleConverter(theme=theme, css_path=css_path) html = converter.convert(content, extracted_title) else: # 基础转换 html = self._md.render(content) # 提取图片 images = self._extract_images(content) return ConvertResult( title=extracted_title, html=html, images=images, ) def _convert_via_api( self, markdown: str, api_endpoint: Optional[str], theme: str, ) -> str: """通过 API 转换 Markdown""" import httpx endpoint = api_endpoint or self.MDNICE_API try: # mdnice API 格式 payload = { "content": markdown, "theme": theme, "mobile": False, } with httpx.Client(timeout=60.0) as client: response = client.post(endpoint, json=payload) if response.status_code == 200: data = response.json() # mdnice 返回格式 if "data" in data and "content" in data["data"]: return data["data"]["content"] return data.get("content", data.get("html", markdown)) # 如果 mdnice 失败，回退到本地转换 converter = WeChatStyleConverter(theme=theme) return converter.convert(markdown) except Exception: # 出错时回退到本地转换 converter = WeChatStyleConverter(theme=theme) return converter.convert(markdown) def convert_file( self, file_path: Path, title: Optional[str] = None, platform: str = "default", theme: str = "default", use_api: bool = False, api_endpoint: Optional[str] = None, css_path: Optional[str] = None, ) -> ConvertResult: """转换 Markdown 文件""" content = file_path.read_text(encoding="utf-8") if not title: title = file_path.stem return self.convert(content, title, platform, theme, use_api, api_endpoint, css_path) def _extract_images(self, content: str) -> list[str]: """提取 Markdown 中的图片 URL""" import re pattern = r"!\[.*?\]$(.*?)$" return re.findall(pattern, content) FILE:src/multi_writing_skills/converter/ai.py """ AI 模式转换器使用 AI 模型进行 Markdown 转换和优化。 """ from typing import Optional import httpx from jinja2 import Environment, FileSystemLoader, select_autoescape from pydantic import BaseModel class AIConfig(BaseModel): """AI 配置""" provider: str = "openai" api_key: str = "" base_url: str = "" model: str = "gpt-4" class AIConverter: """AI 模式转换器""" DEFAULT_PROMPT = """你是一个专业的 Markdown 到 HTML 转换专家。请将以下 Markdown 内容转换为适合微信公众号发布的 HTML 格式。要求： 1. 保留原文结构和语义 2. 使用语义化的 HTML 标签 3. 代码块使用

 标签
4. 图片保留原始链接
5. 表格使用标准的  标签
6. 不添加额外的样式，保持简洁

Markdown 内容：
{markdown}

请直接输出转换后的 HTML 内容，不要添加任何解释。"""

    def __init__(self, config: AIConfig) -> None:
        self.config = config
        self._client = httpx.AsyncClient(timeout=120.0)

    async def convert(self, markdown: str, prompt: Optional[str] = None) -> str:
        """
        使用 AI 进行转换

        Args:
            markdown: Markdown 内容
            prompt: 自定义提示词

        Returns:
            转换后的 HTML
        """
        system_prompt = prompt or self.DEFAULT_PROMPT.format(markdown=markdown)

        # 根据不同的 Provider 调用不同的 API
        if self.config.provider == "openai":
            return await self._call_openai(system_prompt, markdown)
        elif self.config.provider == "anthropic":
            return await self._call_anthropic(system_prompt, markdown)
        elif self.config.provider == "gemini":
            return await self._call_gemini(system_prompt, markdown)
        else:
            # 默认使用 OpenAI 兼容格式
            return await self._call_openai(system_prompt, markdown)

    async def _call_openai(self, system_prompt: str, user_content: str) -> str:
        """调用 OpenAI API"""
        base_url = self.config.base_url or "https://api.openai.com/v1"

        response = await self._client.post(
            f"{base_url}/chat/completions",
            headers={
                "Authorization": f"Bearer {self.config.api_key}",
                "Content-Type": "application/json",
            },
            json={
                "model": self.config.model,
                "messages": [
                    {"role": "system", "content": system_prompt},
                    {"role": "user", "content": user_content},
                ],
                "temperature": 0.3,
            },
        )

        response.raise_for_status()
        data = response.json()
        return data["choices"][0]["message"]["content"]

    async def _call_anthropic(self, system_prompt: str, user_content: str) -> str:
        """调用 Anthropic API"""
        response = await self._client.post(
            "https://api.anthropic.com/v1/messages",
            headers={
                "x-api-key": self.config.api_key,
                "anthropic-version": "2023-06-01",
                "Content-Type": "application/json",
            },
            json={
                "model": self.config.model or "claude-3-opus-20240229",
                "max_tokens": 4096,
                "system": system_prompt,
                "messages": [{"role": "user", "content": user_content}],
            },
        )

        response.raise_for_status()
        data = response.json()
        return data["content"][0]["text"]

    async def _call_gemini(self, system_prompt: str, user_content: str) -> str:
        """调用 Gemini API"""
        response = await self._client.post(
            f"https://generativelanguage.googleapis.com/v1beta/models/{self.config.model or 'gemini-pro'}:generateContent",
            headers={"Content-Type": "application/json"},
            params={"key": self.config.api_key},
            json={
                "contents": [{"parts": [{"text": f"{system_prompt}\n\n{user_content}"}]}],
            },
        )

        response.raise_for_status()
        data = response.json()
        return data["candidates"][0]["content"]["parts"][0]["text"]

    async def close(self) -> None:
        """关闭客户端连接"""
        await self._client.aclose()


async def convert_with_ai(
    markdown: str,
    api_key: str,
    provider: str = "openai",
    base_url: str = "",
    model: str = "gpt-4",
    prompt: Optional[str] = None,
) -> str:
    """
    使用 AI 转换 Markdown

    Args:
        markdown: Markdown 内容
        api_key: API 密钥
        provider: AI 提供商 (openai/anthropic/gemini)
        base_url: API 基础 URL
        model: 模型名称
        prompt: 自定义提示词

    Returns:
        转换后的 HTML
    """
    config = AIConfig(
        provider=provider,
        api_key=api_key,
        base_url=base_url,
        model=model,
    )
    converter = AIConverter(config)
    try:
        return await converter.convert(markdown, prompt)
    finally:
        await converter.close()
FILE:src/multi_writing_skills/converter/api.py
"""
API 模式转换器

调用外部 API 服务进行 Markdown 到 HTML 的转换，适用于微信公众号样式优化。
"""

import httpx
from pydantic import BaseModel


class APIConvertOptions(BaseModel):
    """API 转换选项"""

    endpoint: str
    timeout: float = 60.0
    theme: str = "default"


class APIConverter:
    """API 模式转换器"""

    def __init__(self, options: APIConvertOptions) -> None:
        self.options = options
        self._client = httpx.AsyncClient(timeout=options.timeout)

    async def convert(self, markdown: str, title: str = "") -> dict:
        """
        调用 API 进行转换

        Args:
            markdown: Markdown 内容
            title: 文章标题

        Returns:
            转换结果，包含 html, images 等字段
        """
        try:
            response = await self._client.post(
                self.options.endpoint,
                json={
                    "markdown": markdown,
                    "title": title,
                    "theme": self.options.theme,
                },
            )

            response.raise_for_status()
            return response.json()

        except httpx.HTTPStatusError as e:
            raise RuntimeError(f"API 请求失败: {e.response.status_code}") from e
        except httpx.RequestError as e:
            raise RuntimeError(f"API 请求异常: {str(e)}") from e

    async def convert_with_images(self, markdown: str, title: str = "") -> dict:
        """
        转换并处理图片

        Args:
            markdown: Markdown 内容
            title: 文章标题

        Returns:
            转换结果，包含处理后的 html 和图片信息
        """
        result = await self.convert(markdown, title)

        # 如果 API 返回了图片列表，返回原始结果
        if "images" in result:
            return result

        # 否则尝试从 HTML 中提取图片
        import re

        html = result.get("html", "")
        images = re.findall(r']+src=["\']([^"\']+)["\']', html)

        result["images"] = images
        return result

    async def close(self) -> None:
        """关闭客户端连接"""
        await self._client.aclose()


# 预设的 API 端点
BUILTIN_ENDPOINTS = {
    "mdnice": "https://api.mdnice.com/api/v1/markdown",
    "wechat": "https://api.weixin.qq.com/cgi-bin/media/upload",  # 示例
}


def get_converter(endpoint: str, theme: str = "default") -> APIConverter:
    """
    获取 API 转换器实例

    Args:
        endpoint: API 端点 URL 或预设名称 (如 "mdnice")
        theme: 主题名称

    Returns:
        APIConverter 实例
    """
    # 检查是否是预设端点
    actual_endpoint = BUILTIN_ENDPOINTS.get(endpoint, endpoint)

    options = APIConvertOptions(endpoint=actual_endpoint, theme=theme)
    return APIConverter(options)
FILE:src/multi_writing_skills/converter/css_theme.py
"""
CSS 主题解析器

解析 wenyan-cli 格式的 CSS 文件，转换为微信公众号可用的内联样式。
"""

import re
from pathlib import Path
from typing import Dict, Optional
from dataclasses import dataclass

from .theme_types import ThemeStyles, ThemeColors


@dataclass
class CSSRule:
    """CSS 规则"""
    selector: str
    properties: Dict[str, str]


def parse_css(css_content: str) -> list[CSSRule]:
    """解析 CSS 内容，提取规则"""
    rules = []

    # 移除注释
    css_content = re.sub(r'/\*.*?\*/', '', css_content, flags=re.DOTALL)

    # 匹配选择器和属性块
    pattern = r'([^{]+)\{([^}]+)\}'

    for match in re.finditer(pattern, css_content, re.DOTALL):
        selector = match.group(1).strip()
        properties_str = match.group(2).strip()

        # 解析属性
        properties = {}
        for prop in properties_str.split(';'):
            prop = prop.strip()
            if ':' in prop:
                key, value = prop.split(':', 1)
                properties[key.strip()] = value.strip()

        if properties:
            rules.append(CSSRule(selector=selector, properties=properties))

    return rules


def css_to_inline_style(rules: list[CSSRule]) -> Dict[str, str]:
    """将 CSS 规则转换为元素样式映射

    wenyan-cli 格式：
    - #wenyan h1 -> h1
    - #wenyan p -> p
    - #wenyan pre code -> code_block
    - #wenyan p code -> code_inline
    - #wenyan pre -> code_block_container
    """
    styles = {}

    # 选择器映射：wenyan-cli 选择器 -> 元素类型
    selector_map = {
        '#wenyan h1': 'h1',
        '#wenyan h2': 'h2',
        '#wenyan h3': 'h3',
        '#wenyan h4': 'h4',
        '#wenyan h5': 'h5',
        '#wenyan h6': 'h6',
        '#wenyan p': 'p',
        '#wenyan blockquote': 'blockquote',
        '#wenyan pre': 'pre',
        '#wenyan pre code': 'code_block',
        '#wenyan code': 'code_inline',
        '#wenyan p code': 'code_inline',
        '#wenyan ul': 'ul',
        '#wenyan ol': 'ol',
        '#wenyan li': 'li',
        '#wenyan table': 'table',
        '#wenyan table th': 'th',
        '#wenyan table td': 'td',
        '#wenyan table tr': 'tr',
        '#wenyan img': 'img',
        '#wenyan a': 'a',
        '#wenyan hr': 'hr',
        '#wenyan strong': 'strong',
        '#wenyan p strong': 'strong',
        '#wenyan em': 'em',
        '#wenyan': 'container',
    }

    for rule in rules:
        # 处理 #wenyan 前缀的选择器
        for wenyan_selector, element in selector_map.items():
            if rule.selector == wenyan_selector:
                if element not in styles:
                    styles[element] = {}

                # 合并属性，后面的覆盖前面的
                styles[element].update(rule.properties)

                # 处理特殊选择器
                if element == 'pre':
                    # pre::before 用于代码块头部
                    styles['pre_before'] = rule.properties.copy()

    # 转换为内联样式字符串
    result = {}
    for element, props in styles.items():
        if props:
            result[element] = props_to_string(props)

    return result


def props_to_string(props: Dict[str, str]) -> str:
    """将属性字典转换为内联样式字符串"""
    # CSS 属性名转换为 camelCase（用于某些特殊情况）
    # 但微信公众号支持 kebab-case，所以直接使用
    parts = []
    for key, value in props.items():
        # 跳过伪元素和不支持的属性
        if key in ('content', 'display', 'position', 'z-index', 'transform'):
            # 这些属性在内联样式中可能不生效或不需要
            if key != 'display':  # display 通常是有用的
                continue
        parts.append(f'{key}: {value};')

    return ' '.join(parts)


def load_css_theme(css_path: str, theme_name: str = "custom") -> ThemeStyles:
    """从 CSS 文件加载主题

    Args:
        css_path: CSS 文件路径（本地路径或 URL）
        theme_name: 主题名称

    Returns:
        ThemeStyles 对象
    """
    import httpx

    # 读取 CSS 内容
    if css_path.startswith(('http://', 'https://')):
        # 网络 URL
        with httpx.Client(timeout=30.0) as client:
            response = client.get(css_path)
            css_content = response.text
    else:
        # 本地文件
        path = Path(css_path)
        if not path.exists():
            raise FileNotFoundError(f"CSS 文件不存在: {css_path}")
        css_content = path.read_text(encoding='utf-8')

    # 解析 CSS
    rules = parse_css(css_content)
    styles = css_to_inline_style(rules)

    # 创建 ThemeStyles
    return ThemeStyles(
        name=theme_name,
        display_name=theme_name.capitalize(),
        description=f"从 {css_path} 加载的主题",
        colors=ThemeColors(),
        h1_style=styles.get('h1', ''),
        h2_style=styles.get('h2', ''),
        h3_style=styles.get('h3', ''),
        h4_style=styles.get('h4', ''),
        p_style=styles.get('p', ''),
        blockquote_style=styles.get('blockquote', ''),
        code_inline_style=styles.get('code_inline', ''),
        code_block_style=styles.get('code_block', styles.get('pre', '')),
        code_header_style=styles.get('pre_before', ''),
        ul_style=styles.get('ul', ''),
        ol_style=styles.get('ol', ''),
        li_style=styles.get('li', ''),
        table_style=styles.get('table', ''),
        th_style=styles.get('th', ''),
        td_style=styles.get('td', ''),
        img_style=styles.get('img', ''),
        a_style=styles.get('a', ''),
        hr_style=styles.get('hr', ''),
        strong_style=styles.get('strong', ''),
        em_style=styles.get('em', ''),
        container_style=styles.get('container', ''),
    )


def register_css_theme(css_path: str, theme_name: str, themes_dict: Optional[dict] = None) -> ThemeStyles:
    """注册 CSS 主题

    Args:
        css_path: CSS 文件路径
        theme_name: 主题名称（唯一标识）
        themes_dict: 主题字典（可选，用于注册主题）

    Returns:
        ThemeStyles 对象
    """
    theme = load_css_theme(css_path, theme_name)
    if themes_dict is not None:
        themes_dict[theme_name] = theme
    return theme


def list_css_themes(theme_dir: Path) -> list[dict]:
    """列出目录中的所有 CSS 主题

    Args:
        theme_dir: 主题目录

    Returns:
        主题信息列表
    """
    themes = []

    if not theme_dir.exists():
        return themes

    for css_file in theme_dir.glob('*.css'):
        theme_name = css_file.stem
        themes.append({
            'name': theme_name,
            'path': str(css_file),
            'display_name': theme_name.capitalize(),
            'description': f"CSS 主题: {css_file.name}",
        })

    return themes
FILE:src/multi_writing_skills/converter/theme_types.py
"""
主题数据类型定义

定义主题的数据结构，避免循环导入。
"""

from dataclasses import dataclass, field


@dataclass
class ThemeColors:
    """主题颜色配置"""

    primary: str = "#333333"
    secondary: str = "#666666"
    accent: str = "#3f3f3f"
    background: str = "#ffffff"
    code_background: str = "#282c34"
    code_text: str = "#abb2bf"
    quote_background: str = "#f7f7f7"
    quote_border: str = "#ddd"
    link: str = "#576b95"
    table_header_bg: str = "#f5f5f5"


@dataclass
class ThemeStyles:
    """主题样式配置 - 兼容 wenyan-cli 的 #wenyan 容器格式"""

    name: str = "default"
    display_name: str = "默认主题"
    description: str = "简洁清爽的默认主题"
    colors: ThemeColors = field(default_factory=ThemeColors)

    # 标题样式
    h1_style: str = ""
    h2_style: str = ""
    h3_style: str = ""
    h4_style: str = ""

    # 段落样式
    p_style: str = ""

    # 引用样式
    blockquote_style: str = ""

    # 代码样式
    code_inline_style: str = ""
    code_block_style: str = ""
    code_header_style: str = ""  # 代码块头部样式

    # 列表样式
    ul_style: str = ""
    ol_style: str = ""
    li_style: str = ""

    # 表格样式
    table_style: str = ""
    th_style: str = ""
    td_style: str = ""
    tr_odd_style: str = ""  # 奇数行样式
    tr_even_style: str = ""  # 偶数行样式

    # 图片样式
    img_style: str = ""

    # 其他样式
    a_style: str = ""
    hr_style: str = ""
    strong_style: str = ""
    em_style: str = ""

    # 整体容器样式 - 使用 #wenyan 选择器格式
    container_style: str = ""
FILE:src/multi_writing_skills/converter/themes.py
"""
微信公众号主题样式配置

支持多种主题，每个主题包含完整的样式配置。
兼容 wenyan-cli 的 CSS 主题格式，使用 #wenyan 容器选择器。
"""

from pathlib import Path
from typing import Optional

from .theme_types import ThemeStyles, ThemeColors
from .css_theme import load_css_theme, register_css_theme


# 主题存储目录
THEME_DIR = Path.home() / ".multi-writing-skills" / "themes"


# ============ 预设主题 ============

# 默认主题 - 深色代码块 + 美化表格
DEFAULT_THEME = ThemeStyles(
    name="default",
    display_name="默认主题",
    description="简洁清爽的默认主题",
    colors=ThemeColors(),
    h1_style="font-size: 24px; font-weight: bold; color: #333; margin: 20px 0 10px; padding-bottom: 10px; border-bottom: 2px solid #333;",
    h2_style="font-size: 20px; font-weight: bold; color: #333; margin: 18px 0 8px; padding-left: 12px; border-left: 4px solid #333; background: linear-gradient(90deg, #f5f5f5 0%, transparent 100%);",
    h3_style="font-size: 18px; font-weight: bold; color: #333; margin: 16px 0 6px; padding-bottom: 5px; border-bottom: 1px dashed #ccc;",
    h4_style="font-size: 16px; font-weight: bold; color: #333; margin: 14px 0 4px;",
    p_style="font-size: 15px; line-height: 1.8; color: #3f3f3f; margin: 10px 0; letter-spacing: 0.5px;",
    blockquote_style="background-color: #f8f8f8; border-left: 4px solid #555; padding: 12px 16px; margin: 16px 0; color: #555; font-size: 14px; border-radius: 0 6px 6px 0;",
    code_inline_style="background-color: #f0f0f0; color: #d14; padding: 3px 8px; border-radius: 4px; font-family: 'SF Mono', Consolas, 'Liberation Mono', Menlo, monospace; font-size: 14px;",
    code_block_style="background-color: #282c34; color: #abb2bf; border-radius: 8px; padding: 16px; margin: 16px 0; overflow-x: auto; font-family: 'SF Mono', Consolas, 'Liberation Mono', Menlo, monospace; font-size: 14px; line-height: 1.7; box-shadow: 0 2px 10px rgba(0,0,0,0.1);",
    code_header_style="background-color: #21252b; color: #abb2bf; padding: 8px 16px; font-size: 12px; border-radius: 8px 8px 0 0; border-bottom: 1px solid #333; font-family: 'SF Mono', Consolas, monospace;",
    ul_style="padding-left: 20px; margin: 10px 0; list-style-type: disc;",
    ol_style="padding-left: 20px; margin: 10px 0;",
    li_style="font-size: 15px; line-height: 1.8; color: #3f3f3f; margin: 6px 0;",
    table_style="width: 100%; border-collapse: collapse; margin: 16px 0; font-size: 14px; border-radius: 8px; overflow: hidden; box-shadow: 0 1px 3px rgba(0,0,0,0.1);",
    th_style="background-color: #333; color: #fff; font-weight: bold; padding: 12px 15px; text-align: left; border: none;",
    td_style="padding: 12px 15px; border: none; border-bottom: 1px solid #eee;",
    tr_odd_style="background-color: #fff;",
    tr_even_style="background-color: #f9f9f9;",
    img_style="max-width: 100%; height: auto; display: block; margin: 15px auto; border-radius: 8px; box-shadow: 0 2px 8px rgba(0,0,0,0.1);",
    a_style="color: #333; text-decoration: none; border-bottom: 1px solid #333;",
    hr_style="border: none; height: 2px; background: linear-gradient(90deg, transparent, #333, transparent); margin: 24px 0;",
    strong_style="font-weight: bold; color: #333;",
    em_style="font-style: italic; color: #555;",
    container_style='padding: 20px; font-family: -apple-system, BlinkMacSystemFont, "Helvetica Neue", "PingFang SC", "Hiragino Sans GB", "Microsoft YaHei UI", "Microsoft YaHei", Arial, sans-serif;',
)

# 橙心主题 - 温暖活力
ORANGE_THEME = ThemeStyles(
    name="orange",
    display_name="橙心主题",
    description="温暖活力的橙色主题",
    colors=ThemeColors(
        primary="#d9534f",
        secondary="#f0ad4e",
        accent="#ff6b6b",
        code_background="#2d2d2d",
        code_text="#f8f8f2",
        quote_border="#d9534f",
        link="#d9534f",
    ),
    h1_style="font-size: 24px; font-weight: bold; color: #d9534f; margin: 20px 0 10px; text-align: center;",
    h2_style="font-size: 20px; font-weight: bold; color: #fff; margin: 18px 0 8px; padding: 10px 16px; background: linear-gradient(135deg, #d9534f 0%, #f0ad4e 100%); border-radius: 6px;",
    h3_style="font-size: 18px; font-weight: bold; color: #d9534f; margin: 16px 0 6px; padding-left: 12px; border-left: 3px solid #f0ad4e;",
    h4_style="font-size: 16px; font-weight: bold; color: #333; margin: 14px 0 4px;",
    p_style="font-size: 15px; line-height: 1.8; color: #3f3f3f; margin: 10px 0; letter-spacing: 0.5px;",
    blockquote_style="background: linear-gradient(135deg, #fff5f5 0%, #fff 100%); border-left: 4px solid #d9534f; padding: 12px 16px; margin: 16px 0; color: #666; font-size: 14px; border-radius: 0 8px 8px 0;",
    code_inline_style="background-color: #fff5f5; color: #d9534f; padding: 3px 8px; border-radius: 4px; font-family: 'SF Mono', Consolas, monospace; font-size: 14px;",
    code_block_style="background-color: #2d2d2d; color: #f8f8f2; border-radius: 8px; padding: 16px; margin: 16px 0; overflow-x: auto; font-family: 'SF Mono', Consolas, monospace; font-size: 14px; line-height: 1.7; box-shadow: 0 4px 12px rgba(217, 83, 79, 0.15);",
    code_header_style="background-color: #252525; color: #f8f8f2; padding: 8px 16px; font-size: 12px; border-radius: 8px 8px 0 0; border-bottom: 1px solid #3a3a3a;",
    ul_style="padding-left: 20px; margin: 10px 0; list-style-type: disc;",
    ol_style="padding-left: 20px; margin: 10px 0;",
    li_style="font-size: 15px; line-height: 1.8; color: #3f3f3f; margin: 6px 0;",
    table_style="width: 100%; border-collapse: collapse; margin: 16px 0; font-size: 14px; border-radius: 8px; overflow: hidden; box-shadow: 0 2px 8px rgba(217, 83, 79, 0.1);",
    th_style="background: linear-gradient(135deg, #d9534f 0%, #f0ad4e 100%); color: #fff; font-weight: bold; padding: 12px 15px; text-align: left; border: none;",
    td_style="padding: 12px 15px; border: none; border-bottom: 1px solid #f0f0f0;",
    tr_odd_style="background-color: #fff;",
    tr_even_style="background-color: #fffbf7;",
    img_style="max-width: 100%; height: auto; display: block; margin: 15px auto; border-radius: 8px; box-shadow: 0 4px 12px rgba(217, 83, 79, 0.15);",
    a_style="color: #d9534f; text-decoration: none; border-bottom: 1px dashed #d9534f;",
    hr_style="border: none; height: 2px; background: linear-gradient(90deg, #d9534f, #f0ad4e, #d9534f); margin: 24px 0;",
    strong_style="font-weight: bold; color: #d9534f;",
    em_style="font-style: italic; color: #f0ad4e;",
    container_style='padding: 20px; font-family: -apple-system, BlinkMacSystemFont, "Helvetica Neue", "PingFang SC", "Hiragino Sans GB", "Microsoft YaHei UI", "Microsoft YaHei", Arial, sans-serif;',
)

# 蓝色主题 - 清新专业
BLUE_THEME = ThemeStyles(
    name="blue",
    display_name="蓝皓主题",
    description="清新专业的蓝色主题",
    colors=ThemeColors(
        primary="#3b8bba",
        secondary="#5bc0de",
        accent="#2a6496",
        code_background="#1e3a5f",
        code_text="#e6f3ff",
        quote_border="#3b8bba",
        link="#3b8bba",
        table_header_bg="#3b8bba",
    ),
    h1_style="font-size: 24px; font-weight: bold; color: #3b8bba; margin: 20px 0 10px; text-align: center; border-bottom: 2px solid #3b8bba; padding-bottom: 10px;",
    h2_style="font-size: 20px; font-weight: bold; color: #fff; margin: 18px 0 8px; padding: 10px 16px; background: #3b8bba; border-radius: 4px; box-shadow: 0 2px 8px rgba(59, 139, 186, 0.3);",
    h3_style="font-size: 18px; font-weight: bold; color: #3b8bba; margin: 16px 0 6px; border-bottom: 2px dotted #3b8bba; padding-bottom: 5px;",
    h4_style="font-size: 16px; font-weight: bold; color: #3b8bba; margin: 14px 0 4px;",
    p_style="font-size: 15px; line-height: 1.8; color: #3f3f3f; margin: 10px 0; letter-spacing: 0.5px;",
    blockquote_style="background: linear-gradient(135deg, #f0f7ff 0%, #fff 100%); border-left: 4px solid #3b8bba; padding: 12px 16px; margin: 16px 0; color: #555; font-size: 14px; border-radius: 0 8px 8px 0;",
    code_inline_style="background-color: #e8f4fc; color: #2a6496; padding: 3px 8px; border-radius: 4px; font-family: 'SF Mono', Consolas, monospace; font-size: 14px;",
    code_block_style="background-color: #1e3a5f; color: #e6f3ff; border-radius: 8px; padding: 16px; margin: 16px 0; overflow-x: auto; font-family: 'SF Mono', Consolas, monospace; font-size: 14px; line-height: 1.7; box-shadow: 0 4px 12px rgba(30, 58, 95, 0.3);",
    code_header_style="background-color: #163252; color: #7eb8e2; padding: 8px 16px; font-size: 12px; border-radius: 8px 8px 0 0; border-bottom: 1px solid #2a4a6a;",
    ul_style="padding-left: 20px; margin: 10px 0; list-style-type: disc;",
    ol_style="padding-left: 20px; margin: 10px 0;",
    li_style="font-size: 15px; line-height: 1.8; color: #3f3f3f; margin: 6px 0;",
    table_style="width: 100%; border-collapse: collapse; margin: 16px 0; font-size: 14px; border-radius: 8px; overflow: hidden; box-shadow: 0 2px 8px rgba(59, 139, 186, 0.15);",
    th_style="background: #3b8bba; color: #fff; font-weight: bold; padding: 12px 15px; text-align: left; border: none;",
    td_style="padding: 12px 15px; border: none; border-bottom: 1px solid #e8f4fc;",
    tr_odd_style="background-color: #fff;",
    tr_even_style="background-color: #f5fafd;",
    img_style="max-width: 100%; height: auto; display: block; margin: 15px auto; border-radius: 8px; box-shadow: 0 2px 8px rgba(59, 139, 186, 0.2);",
    a_style="color: #3b8bba; text-decoration: none; border-bottom: 1px solid #3b8bba;",
    hr_style="border: none; height: 1px; background: linear-gradient(90deg, transparent, #3b8bba, transparent); margin: 24px 0;",
    strong_style="font-weight: bold; color: #3b8bba;",
    em_style="font-style: italic; color: #5bc0de;",
    container_style='padding: 20px; font-family: -apple-system, BlinkMacSystemFont, "Helvetica Neue", "PingFang SC", "Hiragino Sans GB", "Microsoft YaHei UI", "Microsoft YaHei", Arial, sans-serif;',
)

# 绿色主题 - 清新自然
GREEN_THEME = ThemeStyles(
    name="green",
    display_name="绿意主题",
    description="清新自然的绿色主题",
    colors=ThemeColors(
        primary="#27ae60",
        secondary="#2ecc71",
        accent="#1e8449",
        code_background="#1a3d2e",
        code_text="#d4edda",
        quote_border="#27ae60",
        link="#27ae60",
    ),
    h1_style="font-size: 24px; font-weight: bold; color: #27ae60; margin: 20px 0 10px; text-align: center;",
    h2_style="font-size: 20px; font-weight: bold; color: #27ae60; margin: 18px 0 8px; padding-left: 12px; border-left: 4px solid #27ae60; background: linear-gradient(90deg, #e8f8ee 0%, #fff 100%);",
    h3_style="font-size: 18px; font-weight: bold; color: #27ae60; margin: 16px 0 6px;",
    h4_style="font-size: 16px; font-weight: bold; color: #333; margin: 14px 0 4px;",
    p_style="font-size: 15px; line-height: 1.8; color: #3f3f3f; margin: 10px 0; letter-spacing: 0.5px;",
    blockquote_style="background-color: #e8f8ee; border-left: 4px solid #27ae60; padding: 12px 16px; margin: 16px 0; color: #555; font-size: 14px; border-radius: 0 8px 8px 0;",
    code_inline_style="background-color: #e8f8ee; color: #1e8449; padding: 3px 8px; border-radius: 4px; font-family: 'SF Mono', Consolas, monospace; font-size: 14px;",
    code_block_style="background-color: #1a3d2e; color: #d4edda; border-radius: 8px; padding: 16px; margin: 16px 0; overflow-x: auto; font-family: 'SF Mono', Consolas, monospace; font-size: 14px; line-height: 1.7; box-shadow: 0 4px 12px rgba(39, 174, 96, 0.15);",
    code_header_style="background-color: #143024; color: #7dcea0; padding: 8px 16px; font-size: 12px; border-radius: 8px 8px 0 0; border-bottom: 1px solid #2a5a3e;",
    ul_style="padding-left: 20px; margin: 10px 0; list-style-type: disc;",
    ol_style="padding-left: 20px; margin: 10px 0;",
    li_style="font-size: 15px; line-height: 1.8; color: #3f3f3f; margin: 6px 0;",
    table_style="width: 100%; border-collapse: collapse; margin: 16px 0; font-size: 14px; border-radius: 8px; overflow: hidden; box-shadow: 0 2px 8px rgba(39, 174, 96, 0.1);",
    th_style="background: #27ae60; color: #fff; font-weight: bold; padding: 12px 15px; text-align: left; border: none;",
    td_style="padding: 12px 15px; border: none; border-bottom: 1px solid #e8f8ee;",
    tr_odd_style="background-color: #fff;",
    tr_even_style="background-color: #f5fdf8;",
    img_style="max-width: 100%; height: auto; display: block; margin: 15px auto; border-radius: 8px;",
    a_style="color: #27ae60; text-decoration: none;",
    hr_style="border: none; height: 2px; background: linear-gradient(90deg, transparent, #27ae60, transparent); margin: 24px 0;",
    strong_style="font-weight: bold; color: #27ae60;",
    em_style="font-style: italic;",
    container_style='padding: 20px; font-family: -apple-system, BlinkMacSystemFont, "Helvetica Neue", "PingFang SC", "Hiragino Sans GB", "Microsoft YaHei UI", "Microsoft YaHei", Arial, sans-serif;',
)

# 紫色主题 - 优雅神秘
PURPLE_THEME = ThemeStyles(
    name="purple",
    display_name="紫韵主题",
    description="优雅神秘的紫色主题",
    colors=ThemeColors(
        primary="#9b59b6",
        secondary="#8e44ad",
        accent="#6c3483",
        code_background="#2d1f3d",
        code_text="#e8daef",
        quote_border="#9b59b6",
        link="#9b59b6",
    ),
    h1_style="font-size: 24px; font-weight: bold; color: #9b59b6; margin: 20px 0 10px; text-align: center;",
    h2_style="font-size: 20px; font-weight: bold; color: #fff; margin: 18px 0 8px; padding: 10px 16px; background: linear-gradient(135deg, #9b59b6 0%, #8e44ad 100%); border-radius: 20px; display: inline-block;",
    h3_style="font-size: 18px; font-weight: bold; color: #9b59b6; margin: 16px 0 6px; padding-bottom: 5px; border-bottom: 2px dotted #9b59b6;",
    h4_style="font-size: 16px; font-weight: bold; color: #8e44ad; margin: 14px 0 4px;",
    p_style="font-size: 15px; line-height: 1.8; color: #3f3f3f; margin: 10px 0; letter-spacing: 0.5px;",
    blockquote_style="background-color: #f9f5fc; border-left: 4px solid #9b59b6; padding: 12px 16px; margin: 16px 0; color: #555; font-size: 14px; border-radius: 0 8px 8px 0;",
    code_inline_style="background-color: #f9f5fc; color: #9b59b6; padding: 3px 8px; border-radius: 4px; font-family: 'SF Mono', Consolas, monospace; font-size: 14px;",
    code_block_style="background-color: #2d1f3d; color: #e8daef; border-radius: 8px; padding: 16px; margin: 16px 0; overflow-x: auto; font-family: 'SF Mono', Consolas, monospace; font-size: 14px; line-height: 1.7; box-shadow: 0 4px 12px rgba(155, 89, 182, 0.2);",
    code_header_style="background-color: #231730; color: #bb8fce; padding: 8px 16px; font-size: 12px; border-radius: 8px 8px 0 0; border-bottom: 1px solid #3d2a50;",
    ul_style="padding-left: 20px; margin: 10px 0; list-style-type: disc;",
    ol_style="padding-left: 20px; margin: 10px 0;",
    li_style="font-size: 15px; line-height: 1.8; color: #3f3f3f; margin: 6px 0;",
    table_style="width: 100%; border-collapse: collapse; margin: 16px 0; font-size: 14px; border-radius: 8px; overflow: hidden; box-shadow: 0 2px 8px rgba(155, 89, 182, 0.15);",
    th_style="background: linear-gradient(135deg, #9b59b6 0%, #8e44ad 100%); color: #fff; font-weight: bold; padding: 12px 15px; text-align: left; border: none;",
    td_style="padding: 12px 15px; border: none; border-bottom: 1px solid #f3e5f5;",
    tr_odd_style="background-color: #fff;",
    tr_even_style="background-color: #fcf5fe;",
    img_style="max-width: 100%; height: auto; display: block; margin: 15px auto; border-radius: 8px;",
    a_style="color: #9b59b6; text-decoration: none;",
    hr_style="border: none; height: 1px; background: linear-gradient(90deg, transparent, #9b59b6, transparent); margin: 24px 0;",
    strong_style="font-weight: bold; color: #9b59b6;",
    em_style="font-style: italic; color: #8e44ad;",
    container_style='padding: 20px; font-family: -apple-system, BlinkMacSystemFont, "Helvetica Neue", "PingFang SC", "Hiragino Sans GB", "Microsoft YaHei UI", "Microsoft YaHei", Arial, sans-serif;',
)

# 简约主题 - 极简风格
SIMPLE_THEME = ThemeStyles(
    name="simple",
    display_name="简约主题",
    description="极简风格，专注内容",
    colors=ThemeColors(
        primary="#333333",
        secondary="#666666",
        code_background="#fafafa",
        code_text="#333",
        quote_border="#e0e0e0",
        link="#333333",
    ),
    h1_style="font-size: 22px; font-weight: bold; color: #333; margin: 25px 0 15px; text-align: center;",
    h2_style="font-size: 18px; font-weight: bold; color: #333; margin: 20px 0 10px;",
    h3_style="font-size: 16px; font-weight: bold; color: #333; margin: 15px 0 8px;",
    h4_style="font-size: 15px; font-weight: bold; color: #333; margin: 12px 0 6px;",
    p_style="font-size: 15px; line-height: 2; color: #333; margin: 12px 0;",
    blockquote_style="border-left: 3px solid #e0e0e0; padding-left: 15px; margin: 15px 0; color: #666; font-size: 14px;",
    code_inline_style="color: #c7254e; font-family: 'SF Mono', Consolas, monospace; font-size: 14px;",
    code_block_style="background-color: #fafafa; padding: 16px; margin: 16px 0; overflow-x: auto; font-family: 'SF Mono', Consolas, monospace; font-size: 14px; line-height: 1.7; border: 1px solid #eee; border-radius: 4px;",
    code_header_style="background-color: #f5f5f5; color: #666; padding: 6px 16px; font-size: 12px; border-bottom: 1px solid #eee; font-family: 'SF Mono', Consolas, monospace;",
    ul_style="padding-left: 20px; margin: 10px 0; list-style-type: disc;",
    ol_style="padding-left: 20px; margin: 10px 0;",
    li_style="font-size: 15px; line-height: 2; color: #333; margin: 3px 0;",
    table_style="width: 100%; border-collapse: collapse; margin: 16px 0; font-size: 14px;",
    th_style="font-weight: bold; border-bottom: 2px solid #333; padding: 12px; text-align: left; background: none;",
    td_style="border-bottom: 1px solid #eee; padding: 12px;",
    tr_odd_style="background-color: #fff;",
    tr_even_style="background-color: #fafafa;",
    img_style="max-width: 100%; height: auto; display: block; margin: 20px auto;",
    a_style="color: #333; text-decoration: underline;",
    hr_style="border: none; border-top: 1px solid #e0e0e0; margin: 25px 0;",
    strong_style="font-weight: bold;",
    em_style="font-style: italic;",
    container_style='padding: 20px; font-family: -apple-system, BlinkMacSystemFont, "Helvetica Neue", "PingFang SC", "Hiragino Sans GB", "Microsoft YaHei UI", "Microsoft YaHei", Arial, sans-serif;',
)

# 主题注册表
THEMES: dict[str, ThemeStyles] = {
    "default": DEFAULT_THEME,
    "orange": ORANGE_THEME,
    "blue": BLUE_THEME,
    "green": GREEN_THEME,
    "purple": PURPLE_THEME,
    "simple": SIMPLE_THEME,
}


def get_theme(name: str, css_path: Optional[str] = None) -> ThemeStyles:
    """获取主题配置

    Args:
        name: 主题名称
        css_path: 可选的 CSS 文件路径（本地或 URL）

    Returns:
        ThemeStyles 对象
    """
    # 如果指定了 CSS 文件，加载 CSS 主题
    if css_path:
        return load_css_theme(css_path, name)

    # 如果是内置主题，直接返回
    if name in THEMES:
        return THEMES[name]

    # 检查是否是用户自定义 CSS 主题
    css_file = THEME_DIR / f"{name}.css"
    if css_file.exists():
        return load_css_theme(str(css_file), name)

    # 返回默认主题
    return DEFAULT_THEME


def register_theme(css_path: str, name: Optional[str] = None) -> ThemeStyles:
    """注册自定义 CSS 主题

    Args:
        css_path: CSS 文件路径（本地或 URL）
        name: 主题名称，默认从文件名提取

    Returns:
        ThemeStyles 对象
    """
    global THEMES
    if not name:
        name = Path(css_path).stem

    theme = register_css_theme(css_path, name, THEMES)
    return theme


def list_themes() -> list[dict]:
    """列出所有主题（内置 + 自定义 CSS）"""
    result = []

    # 内置主题
    for theme in THEMES.values():
        result.append({
            "name": theme.name,
            "display_name": theme.display_name,
            "description": theme.description,
            "type": "builtin",
        })

    # 自定义 CSS 主题
    if THEME_DIR.exists():
        for css_file in THEME_DIR.glob("*.css"):
            name = css_file.stem
            result.append({
                "name": name,
                "display_name": name.capitalize(),
                "description": f"CSS 主题: {css_file.name}",
                "type": "css",
                "path": str(css_file),
            })

    return result
FILE:src/multi_writing_skills/converter/wechat_style.py
"""
微信公众号样式转换器

完全重写列表和代码块转换逻辑，确保格式正确。
支持 wenyan-cli 格式的 CSS 主题。
"""

import re
from typing import Optional, List, Tuple
from dataclasses import dataclass

from .theme_types import ThemeStyles
from .themes import get_theme, THEMES


@dataclass
class CodeBlock:
    """代码块"""
    lang: str
    code: str
    start: int
    end: int


@dataclass
class ListItem:
    """列表项"""
    level: int
    ordered: bool
    content: str
    number: int = 0


class WeChatStyleConverter:
    """微信公众号样式转换器"""

    def __init__(self, theme: str = "default", css_path: Optional[str] = None):
        self.theme_name = theme
        self.css_path = css_path
        self.theme: ThemeStyles = get_theme(theme, css_path)

    def set_theme(self, theme: str, css_path: Optional[str] = None) -> None:
        self.theme_name = theme
        self.css_path = css_path
        self.theme = get_theme(theme, css_path)

    def convert(self, markdown: str, title: Optional[str] = None) -> str:
        """转换 Markdown 为带样式的 HTML"""
        # 1. 先处理代码块（避免被其他处理干扰）
        markdown, code_blocks = self._extract_code_blocks(markdown)

        # 2. 处理表格
        html = self._convert_tables(markdown)

        # 3. 处理标题
        html = self._convert_headers(html)

        # 4. 处理引用块
        html = self._convert_blockquotes(html)

        # 5. 处理列表（完全重写）
        html = self._convert_lists(html)

        # 6. 处理分割线
        html = self._convert_hr(html)

        # 7. 处理图片
        html = self._convert_images(html)

        # 8. 处理链接
        html = self._convert_links(html)

        # 9. 处理行内代码
        html = self._convert_inline_code(html)

        # 10. 处理粗体和斜体
        html = self._convert_strong_em(html)

        # 11. 处理段落
        html = self._convert_paragraphs(html)

        # 12. 恢复代码块
        html = self._restore_code_blocks(html, code_blocks)

        # 后处理
        html = self._postprocess(html)

        # 添加标题
        if title:
            html = f'{title}{html}'

        return html

    # ============ 代码块处理 ============

    def _extract_code_blocks(self, text: str) -> Tuple[str, List[CodeBlock]]:
        """提取代码块，用占位符替换"""
        code_blocks = []
        pattern = r'```(\w*)\n(.*?)```'

        def replace(match):
            lang = match.group(1) or "text"
            code = match.group(2)
            index = len(code_blocks)
            code_blocks.append(CodeBlock(lang=lang, code=code, start=match.start(), end=match.end()))
            return f'\x00CODE_BLOCK_{index}\x00'

        text = re.sub(pattern, replace, text, flags=re.DOTALL)
        return text, code_blocks

    def _restore_code_blocks(self, html: str, code_blocks: List[CodeBlock]) -> str:
        """恢复代码块并渲染"""
        for i, block in enumerate(code_blocks):
            placeholder = f'\x00CODE_BLOCK_{i}\x00'
            rendered = self._render_code_block(block)
            html = html.replace(placeholder, rendered)
        return html

    def _render_code_block(self, block: CodeBlock) -> str:
        """渲染单个代码块"""
        code = block.code

        # 1. 格式化代码：统一处理缩进，移除首尾空行，确保一致的行尾
        code = self._format_code(code)

        # 2. 先完全转义 HTML（此时代码是纯净的原始代码）
        code = code.replace('&', '&')
        code = code.replace('<', '<')
        code = code.replace('>', '>')

        # 3. 然后在转义后的代码上进行高亮
        code = self._highlight_code_escaped(code, block.lang)

        # 4. 处理缩进：将前导空格转换为  ，同时处理制表符
        # 微信公众号需要用 
 强制换行，仅靠 \n 在 pre 标签中不生效
        lines = code.split('\n')
        processed_lines = []
        for line in lines:
            # 将制表符转换为 4 个空格
            line = line.expandtabs(4)
            # 统计前导空格
            leading_spaces = len(line) - len(line.lstrip(' '))
            if leading_spaces > 0:
                line = ' ' * leading_spaces + line.lstrip(' ')
            processed_lines.append(line)

        code = '
'.join(processed_lines)

        # 语言名称和图标颜色
        lang_names = {
            'python': ('Python', '#3572A5'),
            'py': ('Python', '#3572A5'),
            'javascript': ('JavaScript', '#f7df1e'),
            'js': ('JavaScript', '#f7df1e'),
            'typescript': ('TypeScript', '#3178c6'),
            'ts': ('TypeScript', '#3178c6'),
            'java': ('Java', '#b07219'),
            'go': ('Go', '#00ADD8'),
            'rust': ('Rust', '#dea584'),
            'cpp': ('C++', '#f34b7d'),
            'c': ('C', '#555555'),
            'sql': ('SQL', '#e38c00'),
            'bash': ('Bash', '#4EAA25'),
            'shell': ('Shell', '#4EAA25'),
            'json': ('JSON', '#292929'),
            'yaml': ('YAML', '#cb171e'),
            'yml': ('YAML', '#cb171e'),
            'html': ('HTML', '#e34c26'),
            'css': ('CSS', '#563d7c'),
            'xml': ('XML', '#0060ac'),
            'markdown': ('Markdown', '#083fa1'),
            'md': ('Markdown', '#083fa1'),
            'text': ('Code', '#666666'),
            '': ('Code', '#666666'),
        }
        lang_info = lang_names.get(block.lang.lower(), (block.lang.upper() if block.lang else 'Code', '#666666'))
        lang_display = lang_info[0]
        lang_color = lang_info[1]

        # 样式 - 优化代码块视觉效果
        code_style = (
            "background-color: #282c34; "
            "color: #abb2bf; "
            "border-radius: 0 0 8px 8px; "
            "padding: 16px; "
            "margin: 0; "
            "overflow-x: auto; "
            "font-family: 'SF Mono', 'Fira Code', Consolas, 'Liberation Mono', Menlo, monospace; "
            "font-size: 13px; "
            "line-height: 1.6; "
            "box-shadow: 0 4px 12px rgba(0,0,0,0.15); "
        )

        # 代码块头部样式 - Mac 风格窗口
        header_style = (
            "background-color: #21252b; "
            "color: #abb2bf; "
            "padding: 12px 16px; "
            "font-size: 12px; "
            "border-radius: 8px 8px 0 0; "
            "border-bottom: 1px solid #181a1f; "
            "font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', sans-serif; "
            "display: flex; "
            "align-items: center; "
            "gap: 8px;"
        )

        return f'''




{lang_display}

{code}
'''

    def _format_code(self, code: str) -> str:
        """格式化代码块：
        - 移除首尾空行
        - 统一换行符
        - 移除尾部多余空格
        """
        # 统一换行符
        code = code.replace('\r\n', '\n').replace('\r', '\n')
        # 按行分割
        lines = code.split('\n')
        # 处理每一行：移除行尾空格
        lines = [line.rstrip() for line in lines]
        # 移除开头空行
        while lines and lines[0].strip() == '':
            lines.pop(0)
        # 移除结尾空行
        while lines and lines[-1].strip() == '':
            lines.pop()
        # 重新连接
        return '\n'.join(lines)

    def _highlight_code(self, code: str, lang: str) -> str:
        """语法高亮 - 在原始代码上应用高亮（已废弃，请使用 _highlight_code_escaped）"""
        return code

    def _highlight_code_escaped(self, code: str, lang: str) -> str:
        """语法高亮 - 在已转义的代码上应用高亮

        由于代码已经被 HTML 转义（< -> <, > -> >），
        可以安全地使用 span 标签进行高亮，不会与 HTML 标签冲突
        """
        # 颜色定义 (One Dark 主题)
        colors = {
            'keyword': '#c678dd',      # 紫色 - 关键字
            'string': '#98c379',       # 绿色 - 字符串
            'number': '#d19a66',       # 橙色 - 数字
            'comment': '#75715e',      # 灰色 - 注释
            'class': '#e5c07b',        # 黄色 - 类名
            'decorator': '#d19a66',    # 橙色 - 装饰器
            'builtin': '#e06c75',      # 红色 - 内置函数
        }

        # 关键字列表
        keywords = ['def', 'class', 'if', 'else', 'elif', 'for', 'while', 'return',
            'import', 'from', 'as', 'try', 'except', 'finally', 'with',
            'lambda', 'yield', 'raise', 'pass', 'break', 'continue',
            'True', 'False', 'None', 'and', 'or', 'not', 'in', 'is',
            'async', 'await', 'global', 'nonlocal', 'del', 'assert',
            'const', 'let', 'var', 'function', 'export', 'new', 'this',
            'super', 'extends', 'interface', 'type', 'struct', 'map',
            'func', 'package', 'go', 'defer', 'chan', 'range', 'select',
            'switch', 'case', 'default', 'catch', 'throw', 'typeof',
            'instanceof', 'void', 'null', 'undefined', 'true', 'false',
            'nil', 'make', 'goto', 'fallthrough']

        # 1. 字符串高亮 - 转义后双引号变成 "
        code = re.sub(r'"([^&]*)"', rf'"\1"', code)
        code = re.sub(r''([^&]*)'', rf''\1'', code)

        # 2. 注释高亮
        code = re.sub(r'(?<=\n|\s)#(.*?)(?=\n|$)', rf'#\1', code)

        # 3. 数字高亮
        code = re.sub(r'\b(\d+\.?\d*)\b', rf'\1', code)

        # 4. 装饰器
        code = re.sub(r'(@\w+)', rf'\1', code)

        # 5. 关键字
        for kw in keywords:
            code = re.sub(
                rf'(?{kw}',
                code,
                flags=re.IGNORECASE
            )

        # 6. 类名 (PascalCase)
        code = re.sub(r'\b([A-Z][a-zA-Z0-9]+)\b', rf'\1', code)

        return code

    # ============ 列表处理（完全重写）============

    def _convert_lists(self, text: str) -> str:
        """转换列表 - 完全重写，支持嵌套"""
        lines = text.split('\n')
        result = []

        i = 0
        while i < len(lines):
            line = lines[i]

            # 检测无序列表项（允许内容为空）
            ul_match = re.match(r'^(\s*)[-*+]\s*(.*)$', line)
            # 检测有序列表项（允许内容为空）
            ol_match = re.match(r'^(\s*)(\d+)\.\s*(.*)$', line)

            if ul_match or ol_match:
                # 找到了列表开始，收集整个列表块
                list_html, consumed = self._parse_list_block(lines, i)
                result.append(list_html)
                i += consumed
            else:
                result.append(line)
                i += 1

        return '\n'.join(result)

    def _parse_list_block(self, lines: List[str], start: int) -> Tuple[str, int]:
        """解析整个列表块（空行不中断列表，允许列表项之间有空行）"""
        items = []
        i = start
        first_is_ordered = None  # 第一个列表项是否是有序列表
        first_number = 1  # 有序列表的起始编号

        while i < len(lines):
            line = lines[i]

            # 检测无序列表项（允许内容为空）
            ul_match = re.match(r'^(\s*)[-*+]\s*(.*)$', line)
            # 检测有序列表项（允许内容为空）
            ol_match = re.match(r'^(\s*)(\d+)\.\s*(.*)$', line)

            # 跳过纯空行（检查后面是否还有列表项）
            if line.strip() == '':
                # 查找下一个非空行
                next_non_empty = i + 1
                while next_non_empty < len(lines) and lines[next_non_empty].strip() == '':
                    next_non_empty += 1

                if next_non_empty >= len(lines):
                    # 后面没有内容了，结束列表
                    i += 1
                    break

                next_line = lines[next_non_empty]
                next_ul = re.match(r'^(\s*)[-*+]\s*(.*)$', next_line)
                next_ol = re.match(r'^(\s*)(\d+)\.\s*(.*)$', next_line)

                # 如果下一个非空行还是列表项，继续
                if next_ul or (next_ol and first_is_ordered):
                    i += 1
                    continue
                else:
                    # 下一个非空行不是列表项，结束
                    i += 1
                    break

            # 跳过不匹配的行
            if not ul_match and not ol_match:
                break

            if ul_match:
                indent = len(ul_match.group(1))
                content = ul_match.group(2).strip()
                if first_is_ordered is None:
                    first_is_ordered = False
                # 跳过空内容的无序列表项（单独的 - * +）
                if content:
                    items.append((indent, False, content, 0))
                i += 1
            elif ol_match:
                indent = len(ol_match.group(1))
                number = int(ol_match.group(2))
                content = ol_match.group(3).strip()
                if first_is_ordered is None:
                    first_is_ordered = True
                    first_number = number  # 记录第一个有序列表项的编号
                # 跳过空内容的有序列表项（单独的 1. 2. 3.）
                if content:
                    # 如果第一个有内容的项编号是 first_number+1 且第一个空项编号是 first_number
                    # 说明列表应该从 first_number 开始
                    items.append((indent, True, content, number))
                i += 1

        # 调整有序列表的编号：如果第一个列表项为空且编号为 1，后续项编号减 1
        if first_is_ordered and items:
            # 检查第一个有内容的项的编号是否比 first_number 大 1
            if items[0][3] == first_number + 1:
                # 列表应该从 first_number 开始，调整所有项的编号
                adjusted_items = [(indent, ordered, content, number - 1) for indent, ordered, content, number in items]
                items = adjusted_items

        # 渲染列表
        html = self._render_list_items(items)
        return html, i - start

    def _render_list_items(self, items: List[Tuple[int, bool, str, int]]) -> str:
        """渲染列表项为 HTML - 使用 div 模拟列表，避免微信重新解析"""
        if not items:
            return ''

        # 判断列表类型
        first_ordered = items[0][1]

        # 使用 div 模拟列表，避免微信编辑器重新解析
        # 无序列表：使用圆点字符 •
        # 有序列表：使用数字编号

        html_parts = []

        if first_ordered:
            # 有序列表
            list_style = self.theme.ol_style
            # 提取 padding-left 值用于缩进
            padding_match = re.search(r'padding-left:\s*(\d+)px', list_style)
            padding_left = int(padding_match.group(1)) if padding_match else 20

            for indent, ordered, content, number in items:
                # 处理内联格式
                content = self._process_inline(content)
                # 使用 section 模拟有序列表项
                html_parts.append(
                    f''
                    f'{number}.'
                    f'{content}'
                    f''
                )
        else:
            # 无序列表
            list_style = self.theme.ul_style
            # 提取 padding-left 值用于缩进
            padding_match = re.search(r'padding-left:\s*(\d+)px', list_style)
            padding_left = int(padding_match.group(1)) if padding_match else 20

            for indent, ordered, content, number in items:
                # 处理内联格式
                content = self._process_inline(content)
                # 使用 section 模拟无序列表项，使用 • 作为项目符号
                html_parts.append(
                    f''
                    f'•'
                    f'{content}'
                    f''
                )

        return '\n'.join(html_parts)

    def _process_inline(self, text: str) -> str:
        """处理行内格式（粗体、斜体、行内代码）"""
        # 行内代码
        text = re.sub(r'`([^`]+)`', f'\\1', text)
        # 粗体
        text = re.sub(r'\*\*([^*]+)\*\*', f'\\1', text)
        text = re.sub(r'__([^_]+)__', f'\\1', text)
        # 斜体
        text = re.sub(r'\*([^*]+)\*', f'\\1', text)
        # 下划线斜体：不处理包含大写字母的（避免误处理代码块占位符）
        text = re.sub(r'_([a-z][^_]*[a-z])_', f'\\1', text)
        return text

    # ============ 其他元素处理 ============

    def _convert_headers(self, text: str) -> str:
        """转换标题"""
        lines = text.split('\n')
        result = []

        for line in lines:
            if line.startswith('###### '):
                result.append(f'{line[7:]}')
            elif line.startswith('##### '):
                result.append(f'{line[6:]}')
            elif line.startswith('#### '):
                result.append(f'{line[5:]}')
            elif line.startswith('### '):
                result.append(f'{line[4:]}')
            elif line.startswith('## '):
                result.append(f'{line[3:]}')
            elif line.startswith('# '):
                result.append(f'{line[2:]}')
            else:
                result.append(line)

        return '\n'.join(result)

    def _convert_blockquotes(self, text: str) -> str:
        """转换引用块"""
        lines = text.split('\n')
        result = []
        quote_lines = []

        for line in lines:
            if line.startswith('> '):
                quote_lines.append(line[2:])
            elif line.startswith('>'):
                quote_lines.append(line[1:])
            else:
                if quote_lines:
                    content = '
'.join(quote_lines)
                    result.append(f'{content}')
                    quote_lines = []
                result.append(line)

        if quote_lines:
            content = '
'.join(quote_lines)
            result.append(f'{content}')

        return '\n'.join(result)

    def _convert_tables(self, text: str) -> str:
        """转换表格"""
        lines = text.split('\n')
        result = []
        in_table = False
        is_header = True
        row_index = 0

        for line in lines:
            if '|' in line and line.strip().startswith('|'):
                cells = [c.strip() for c in line.split('|')[1:-1]]

                if not in_table:
                    result.append(f'')
                    result.append('')
                    in_table = True
                    is_header = True

                # 跳过分隔行
                if all(set(c) <= {'-', ':', ' '} for c in cells):
                    result.append('')
                    is_header = False
                    continue

                if is_header:
                    cells_html = ''.join([f'' for c in cells])
                    result.append(f'{cells_html}')
                else:
                    row_index += 1
                    row_style = self.theme.tr_odd_style if row_index % 2 == 1 else self.theme.tr_even_style
                    cells_html = ''.join([f'' for c in cells])
                    result.append(f'{cells_html}')
            else:
                if in_table:
                    result.append('
{c}
{c}
')
                    in_table = False
                result.append(line)

        if in_table:
            result.append('')

        return '\n'.join(result)

    def _convert_hr(self, text: str) -> str:
        """转换分割线"""
        return re.sub(r'^---+$', f'', text, flags=re.MULTILINE)

    def _convert_images(self, text: str) -> str:
        """转换图片"""
        return re.sub(r'!\[([^\]]*)\]\(([^)]+)\)',
                      f'', text)

    def _convert_links(self, text: str) -> str:
        """转换链接"""
        return re.sub(r'\[([^\]]+)\]\(([^)]+)\)',
                      f'\\1', text)

    def _convert_inline_code(self, text: str) -> str:
        """转换行内代码"""
        return re.sub(r'`([^`]+)`', f'\\1', text)

    def _convert_strong_em(self, text: str) -> str:
        """转换粗体和斜体"""
        text = re.sub(r'\*\*([^*]+)\*\*', f'\\1', text)
        text = re.sub(r'__([^_]+)__', f'\\1', text)
        text = re.sub(r'\*([^*]+)\*', f'\\1', text)
        # 下划线斜体：不处理包含大写字母的（避免误处理代码块占位符如 CODE_BLOCK_0）
        text = re.sub(r'_([a-z][^_]*[a-z])_', f'\\1', text)
        return text

    def _convert_paragraphs(self, text: str) -> str:
        """转换段落"""
        lines = text.split('\n')
        result = []

        for line in lines:
            stripped = line.strip()
            # 跳过已转换的块元素
            if stripped.startswith(('$', stripped):
                result.append(line)
            elif stripped:
                result.append(f'{stripped}')

        return '\n'.join(result)

    def _postprocess(self, text: str) -> str:
        """后处理 - 使用 #wenyan 容器格式（兼容 wenyan-cli）"""
        text = re.sub(r'\n{3,}', '\n\n', text)
        # 使用 id="wenyan" 容器，兼容 wenyan-cli 的 CSS 主题格式
        text = f'{text}'
        return text


def get_available_themes() -> List[str]:
    """获取所有可用主题名称"""
    return list(THEMES.keys())
FILE:src/multi_writing_skills/humanizer/__init__.py
"""
AI 去痕模块

去除 AI 写作痕迹，使文章更加自然人性化。

支持的 Provider:
- openai: OpenAI GPT 系列
- qwen: 通义千问 (阿里云 DashScope)
- doubao: 豆包 (字节跳动火山引擎)
- minimax: MiniMax
- moonshot: Moonshot (月之暗面)
- zhipu: 智谱 GLM
- hunyuan: 腾讯混元
- yi: 零一万物 Yi 系列
"""

from dataclasses import dataclass
from enum import Enum
from typing import Optional

import httpx


class Intensity(str, Enum):
    """处理强度"""

    LIGHT = "light"  # 轻度：保留大部分原文
    MEDIUM = "medium"  # 中度：适度调整
    HEAVY = "heavy"  # 重度：大幅改写


class Provider(str, Enum):
    """支持的 AI 提供商"""

    OPENAI = "openai"
    QWEN = "qwen"
    DOUBAO = "doubao"
    MINIMAX = "minimax"
    MOONSHOT = "moonshot"
    ZHIPU = "zhipu"
    HUNYUAN = "hunyuan"
    YI = "yi"
    CUSTOM = "custom"  # 自定义 provider


# Provider 配置映射
PROVIDER_CONFIG = {
    Provider.OPENAI: {
        "name": "OpenAI",
        "base_url": "https://api.openai.com/v1",
        "default_model": "gpt-4",
    },
    Provider.QWEN: {
        "name": "通义千问",
        "base_url": "https://dashscope.aliyuncs.com/compatible-mode/v1",
        "default_model": "qwen-plus",
    },
    Provider.DOUBAO: {
        "name": "豆包",
        "base_url": "https://ark.cn-beijing.volces.com/api/v3",
        "default_model": "doubao-pro-32k",
    },
    Provider.MINIMAX: {
        "name": "MiniMax",
        "base_url": "https://api.minimax.chat/v1",
        "default_model": "MiniMax-Text-01",
    },
    Provider.MOONSHOT: {
        "name": "Moonshot",
        "base_url": "https://api.moonshot.cn/v1",
        "default_model": "moonshot-v1-8k",
    },
    Provider.ZHIPU: {
        "name": "智谱 GLM",
        "base_url": "https://open.bigmodel.cn/api/paas/v4",
        "default_model": "glm-4-flash",
    },
    Provider.HUNYUAN: {
        "name": "腾讯混元",
        "base_url": "https://hunyuan.cloud.tencent.com/hunyuan-vision-api",
        "default_model": "hunyuan",
    },
    Provider.YI: {
        "name": "零一万物 Yi",
        "base_url": "https://api.lingyiwanwu.com/v1",
        "default_model": "yi-medium",
    },
}


@dataclass
class HumanizeResult:
    """去痕结果"""

    original: str
    humanized: str
    changes: list[str]


class Humanizer:
    """AI 去痕处理器"""

    SYSTEM_PROMPT = """你是一个专业的文本人性化专家。你的任务是重写 AI 生成的内容，使其更加自然、人性化。

AI 写作的常见特征包括：
1. 过于规整的句式结构
2. 频繁使用"首先"、"其次"、"最后"等过渡词
3. 缺乏个人观点和情感
4. 过度使用被动语态
5. 句子长度过于均匀
6. 缺少口语化表达
7. 过于完美的逻辑结构

人性化改写的策略：
1. 变化句式长度和结构
2. 添加个人观点和感受
3. 使用更口语化的表达
4. 加入适当的语气词
5. 打破过于规整的结构
6. 添加一些"不完美"但更真实的表达
7. 保留核心信息，调整表达方式

请根据指定的强度进行改写。"""

    INTENSITY_GUIDES = {
        "light": """轻度改写：保持原文大部分内容，只做轻微调整。
- 调整个别句式
- 添加少量口语化表达
- 保持原有结构不变""",
        "medium": """中度改写：适度调整，保留核心内容。
- 重新组织部分段落
- 添加个人观点
- 调整句式多样性
- 加入一些口语化表达""",
        "heavy": """重度改写：大幅调整，使文章焕然一新。
- 重新组织文章结构
- 大量使用口语化表达
- 添加个人经历和感受
- 打破原有的规整模式
- 让文章更有"人味" """,
    }

    def __init__(
        self,
        api_key: str,
        provider: str = "openai",
        base_url: str = "",
        model: str = "",
    ) -> None:
        self.api_key = api_key
        self.provider = Provider(provider) if isinstance(provider, str) else provider
        self._config = PROVIDER_CONFIG.get(self.provider, PROVIDER_CONFIG[Provider.OPENAI])
        # 如果指定了 base_url 或 model，使用自定义值
        self.base_url = base_url or self._config["base_url"]
        self.model = model or self._config["default_model"]
        self._client = httpx.AsyncClient(timeout=120.0)

    async def humanize(
        self,
        content: str,
        intensity: str = "medium",
        style_hint: Optional[str] = None,
    ) -> HumanizeResult:
        """
        对内容进行去痕处理

        Args:
            content: 原始内容
            intensity: 处理强度 (light/medium/heavy)
            style_hint: 风格提示

        Returns:
            HumanizeResult 包含处理结果
        """
        intensity_guide = self.INTENSITY_GUIDES.get(intensity, self.INTENSITY_GUIDES["medium"])

        prompt = f"""{self.SYSTEM_PROMPT}

改写强度：{intensity}
{intensity_guide}

{f"风格提示：{style_hint}" if style_hint else ""}

原文内容：
{content}

请对上述内容进行人性化改写，使其读起来更像是真人写的内容。
直接输出改写后的内容，使用 Markdown 格式。"""

        humanized = await self._call_ai(prompt)

        # 分析变化
        changes = await self._analyze_changes(content, humanized)

        return HumanizeResult(
            original=content,
            humanized=humanized,
            changes=changes,
        )

    async def _analyze_changes(self, original: str, humanized: str) -> list[str]:
        """分析改写的变化"""
        prompt = f"""对比原文和改写后的内容，列出主要的变化点。

原文：
{original[:1000]}

改写后：
{humanized[:1000]}

请用简洁的语言列出 3-5 个主要变化点，每点一句话。格式：
1. xxx
2. xxx
..."""

        try:
            result = await self._call_ai(prompt)
            # 解析变化点
            changes = []
            for line in result.split("\n"):
                line = line.strip()
                if line and (line[0].isdigit() or line.startswith("-")):
                    # 移除序号
                    change = line.lstrip("0123456789.-) ")
                    if change:
                        changes.append(change)
            return changes[:5]
        except Exception:
            return ["内容已人性化改写"]

    async def _call_ai(self, prompt: str) -> str:
        """调用 AI API"""
        # 所有 provider 都使用 OpenAI 兼容格式
        return await self._call_openai_compatible(prompt)

    def _is_custom_config(self) -> bool:
        """检查是否使用了自定义配置"""
        default_config = PROVIDER_CONFIG.get(self.provider, PROVIDER_CONFIG[Provider.OPENAI])
        return (
            self.base_url != default_config["base_url"]
            or self.model != default_config["default_model"]
        )

    async def _call_openai_compatible(self, prompt: str) -> str:
        """调用 OpenAI 兼容格式的 API"""
        response = await self._client.post(
            f"{self.base_url}/chat/completions",
            headers={
                "Authorization": f"Bearer {self.api_key}",
                "Content-Type": "application/json",
            },
            json={
                "model": self.model,
                "messages": [
                    {"role": "system", "content": self.SYSTEM_PROMPT},
                    {"role": "user", "content": prompt},
                ],
                "temperature": 0.8,
            },
        )

        response.raise_for_status()
        data = response.json()

        # 兼容不同 provider 的响应格式
        if "choices" in data:
            return data["choices"][0]["message"]["content"]
        elif "output" in data:
            # 某些 provider 可能使用 output 字段
            return data["output"]["choices"][0]["message"]["content"]
        else:
            raise ValueError(f"Unexpected response format: {data}")

    async def close(self) -> None:
        """关闭客户端连接"""
        await self._client.aclose()


async def humanize_content(
    content: str,
    api_key: str,
    intensity: str = "medium",
    provider: str = "openai",
    base_url: str = "",
    model: str = "",
) -> HumanizeResult:
    """
    快捷函数：对内容进行去痕处理

    Args:
        content: 原始内容
        api_key: API 密钥
        intensity: 处理强度
        provider: AI 提供商 (openai/qwen/doubao/minimax/moonshot/zhipu/hunyuan/yi/custom)
        base_url: API 基础 URL (仅在 custom provider 时使用)
        model: 模型名称 (留空则使用 provider 默认模型)

    Returns:
        HumanizeResult
    """
    humanizer = Humanizer(
        api_key=api_key,
        provider=provider,
        base_url=base_url,
        model=model,
    )
    try:
        return await humanizer.humanize(content, intensity)
    finally:
        await humanizer.close()
FILE:src/multi_writing_skills/image/__init__.py
"""
图片处理模块
"""

from pathlib import Path
from typing import Optional

from PIL import Image


class ImageProcessor:
    """图片处理器"""

    @staticmethod
    def resize(
        image_path: str,
        output_path: str,
        width: Optional[int] = None,
        height: Optional[int] = None,
        max_size: Optional[int] = None,
    ) -> str:
        """
        调整图片大小

        Args:
            image_path: 原图路径
            output_path: 输出路径
            width: 目标宽度
            height: 目标高度
            max_size: 最大边长（保持比例）

        Returns:
            输出文件路径
        """
        img = Image.open(image_path)

        if max_size:
            # 按最长边缩放，保持比例
            ratio = min(max_size / img.width, max_size / img.height)
            new_size = (int(img.width * ratio), int(img.height * ratio))
            img = img.resize(new_size, Image.Resampling.LANCZOS)
        elif width and height:
            img = img.resize((width, height), Image.Resampling.LANCZOS)
        elif width:
            ratio = width / img.width
            new_size = (width, int(img.height * ratio))
            img = img.resize(new_size, Image.Resampling.LANCZOS)
        elif height:
            ratio = height / img.height
            new_size = (int(img.width * ratio), height)
            img = img.resize(new_size, Image.Resampling.LANCZOS)

        img.save(output_path, quality=95)
        return output_path

    @staticmethod
    def optimize(image_path: str, output_path: str, quality: int = 85) -> str:
        """
        优化图片大小

        Args:
            image_path: 原图路径
            output_path: 输出路径
            quality: 质量 (1-100)

        Returns:
            输出文件路径
        """
        img = Image.open(image_path)

        # 转换为 RGB（如果需要）
        if img.mode in ("RGBA", "P"):
            img = img.convert("RGB")

        img.save(output_path, "JPEG", quality=quality, optimize=True)
        return output_path

    @staticmethod
    def get_info(image_path: str) -> dict:
        """
        获取图片信息

        Args:
            image_path: 图片路径

        Returns:
            图片信息字典
        """
        img = Image.open(image_path)
        return {
            "width": img.width,
            "height": img.height,
            "format": img.format,
            "mode": img.mode,
            "size_bytes": Path(image_path).stat().st_size,
        }
FILE:src/multi_writing_skills/image/providers/__init__.py
"""
图片生成 Provider 基类
"""

from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Optional


@dataclass
class ImageGenerateResult:
    """图片生成结果"""

    success: bool
    image_url: Optional[str] = None
    local_path: Optional[str] = None
    message: Optional[str] = None


class ImageProvider(ABC):
    """图片生成 Provider 基类"""

    @property
    @abstractmethod
    def name(self) -> str:
        """Provider 名称"""
        pass

    @abstractmethod
    async def generate(
        self,
        prompt: str,
        size: str = "1024x1024",
        style: Optional[str] = None,
    ) -> ImageGenerateResult:
        """
        生成图片

        Args:
            prompt: 图片描述
            size: 图片尺寸
            style: 风格

        Returns:
            ImageGenerateResult
        """
        pass

    @abstractmethod
    def is_configured(self) -> bool:
        """检查是否已配置"""
        pass
FILE:src/multi_writing_skills/image/providers/doubao.py
"""
豆包（ByteDance Doubao）图片生成 Provider
支持 FLUX 模型，国内直连
"""

from typing import Optional

import httpx

from . import ImageGenerateResult, ImageProvider


class DoubaoProvider(ImageProvider):
    """豆包图片生成"""

    BASE_URL = "https://ark.cn-beijing.volces.com/api/v3/images/generations"

    def __init__(
        self,
        api_key: str,
        model: str = "doubao-flux-1",
    ) -> None:
        self.api_key = api_key
        self.model = model
        self._client = httpx.AsyncClient(timeout=120.0)

    @property
    def name(self) -> str:
        return "doubao"

    def is_configured(self) -> bool:
        return bool(self.api_key)

    async def generate(
        self,
        prompt: str,
        size: str = "1024x1024",
        style: Optional[str] = None,
    ) -> ImageGenerateResult:
        """
        使用豆包 FLUX 模型生成图片
        """
        try:
            # 解析尺寸
            size_map = {
                "1024x1024": "1024x1024",
                "720x1280": "720x1280",
                "1280x720": "1280x720",
                "1024x1792": "1024x1792",
                "1792x1024": "1792x1024",
            }
            model_size = size_map.get(size, "1024x1024")

            response = await self._client.post(
                self.BASE_URL,
                headers={
                    "Authorization": f"Bearer {self.api_key}",
                    "Content-Type": "application/json",
                },
                json={
                    "model": self.model,
                    "prompt": prompt,
                    "size": model_size,
                    "n": 1,
                    "style": style,
                },
            )

            response.raise_for_status()
            data = response.json()

            if "data" in data and len(data["data"]) > 0:
                image_url = data["data"][0].get("url") or data["data"][0].get("b64_json")
                if image_url:
                    return ImageGenerateResult(
                        success=True,
                        image_url=image_url,
                        message="生成成功",
                    )

            return ImageGenerateResult(
                success=False,
                message=f"生成失败: {data}",
            )

        except httpx.HTTPStatusError as e:
            return ImageGenerateResult(
                success=False,
                message=f"API 错误: {e.response.status_code}",
            )
        except Exception as e:
            return ImageGenerateResult(
                success=False,
                message=f"生成异常: {str(e)}",
            )

    async def close(self) -> None:
        await self._client.aclose()

FILE:src/multi_writing_skills/image/providers/gemini.py
"""
Google Gemini 图片生成 Provider
"""

from typing import Optional

import httpx

from . import ImageGenerateResult, ImageProvider


class GeminiProvider(ImageProvider):
    """Google Gemini 图片生成"""

    def __init__(self, api_key: str, model: str = "gemini-2.0-flash-exp") -> None:
        self.api_key = api_key
        self.model = model
        self._client = httpx.AsyncClient(timeout=120.0)

    @property
    def name(self) -> str:
        return "gemini"

    def is_configured(self) -> bool:
        return bool(self.api_key)

    async def generate(
        self,
        prompt: str,
        size: str = "1024x1024",
        style: Optional[str] = None,
    ) -> ImageGenerateResult:
        """
        使用 Gemini 生成图片

        注意：Gemini 的图片生成能力可能需要特定的模型版本
        """
        try:
            url = f"https://generativelanguage.googleapis.com/v1beta/models/{self.model}:generateContent"

            response = await self._client.post(
                url,
                headers={"Content-Type": "application/json"},
                params={"key": self.api_key},
                json={
                    "contents": [
                        {
                            "parts": [
                                {"text": f"Generate an image: {prompt}"},
                            ]
                        }
                    ],
                    "generationConfig": {
                        "responseModalities": ["image", "text"],
                    },
                },
            )

            response.raise_for_status()
            data = response.json()

            # 解析响应，提取图片
            if "candidates" in data:
                for candidate in data["candidates"]:
                    if "content" in candidate and "parts" in candidate["content"]:
                        for part in candidate["content"]["parts"]:
                            if "inlineData" in part:
                                # Base64 图片数据
                                import base64
                                import tempfile

                                image_data = base64.b64decode(part["inlineData"]["data"])
                                temp_file = tempfile.NamedTemporaryFile(
                                    suffix=".png", delete=False
                                )
                                temp_file.write(image_data)
                                temp_file.close()

                                return ImageGenerateResult(
                                    success=True,
                                    local_path=temp_file.name,
                                    message="生成成功",
                                )

            return ImageGenerateResult(
                success=False,
                message=f"生成失败: {data}",
            )

        except httpx.HTTPStatusError as e:
            return ImageGenerateResult(
                success=False,
                message=f"API 错误: {e.response.status_code}",
            )
        except Exception as e:
            return ImageGenerateResult(
                success=False,
                message=f"生成异常: {str(e)}",
            )

    async def close(self) -> None:
        await self._client.aclose()
FILE:src/multi_writing_skills/image/providers/minimax.py
"""
MiniMax 图片生成 Provider
"""

from typing import Optional

import httpx

from . import ImageGenerateResult, ImageProvider


class MiniMaxProvider(ImageProvider):
    """MiniMax 图片生成"""

    BASE_URL = "https://api.minimaxi.com/v1/image_generation"

    def __init__(
        self,
        api_key: str,
        model: str = "image-01",
    ) -> None:
        self.api_key = api_key
        self.model = model
        self._client = httpx.AsyncClient(timeout=120.0)

    @property
    def name(self) -> str:
        return "minimax"

    def is_configured(self) -> bool:
        return bool(self.api_key)

    async def generate(
        self,
        prompt: str,
        size: str = "1024x1024",
        style: Optional[str] = None,
    ) -> ImageGenerateResult:
        """
        使用 MiniMax 生成图片
        """
        try:
            # 解析尺寸
            aspect_ratio_map = {
                "1:1": "1:1",
                "16:9": "16:9",
                "9:16": "9:16",
                "1024x1024": "1:1",
                "1024x1792": "9:16",
                "1792x1024": "16:9",
            }
            aspect_ratio = aspect_ratio_map.get(size, "1:1")

            response = await self._client.post(
                self.BASE_URL,
                headers={
                    "Authorization": f"Bearer {self.api_key}",
                    "Content-Type": "application/json",
                },
                json={
                    "model": self.model,
                    "prompt": prompt,
                    "aspect_ratio": aspect_ratio,
                    "n": 1,
                },
            )

            response.raise_for_status()
            data = response.json()

            if data.get("data") and data["data"].get("image_urls"):
                image_urls = data["data"]["image_urls"]
                if image_urls and len(image_urls) > 0:
                    return ImageGenerateResult(
                        success=True,
                        image_url=image_urls[0],
                        message="生成成功",
                    )

            return ImageGenerateResult(
                success=False,
                message=f"生成失败: {data}",
            )

        except httpx.HTTPStatusError as e:
            return ImageGenerateResult(
                success=False,
                message=f"API 错误: {e.response.status_code}",
            )
        except Exception as e:
            return ImageGenerateResult(
                success=False,
                message=f"生成异常: {str(e)}",
            )

    async def close(self) -> None:
        await self._client.aclose()

FILE:src/multi_writing_skills/image/providers/modelscope.py
"""
ModelScope 图片生成 Provider
"""

from typing import Optional

import httpx

from . import ImageGenerateResult, ImageProvider


class ModelScopeProvider(ImageProvider):
    """ModelScope 图片生成"""

    BASE_URL = "https://dashscope.aliyuncs.com/api/v1/services/aigc/text2image/image-synthesis"

    def __init__(
        self,
        api_key: str,
        model: str = "wanx-v1",
    ) -> None:
        self.api_key = api_key
        self.model = model
        self._client = httpx.AsyncClient(timeout=120.0)

    @property
    def name(self) -> str:
        return "modelscope"

    def is_configured(self) -> bool:
        return bool(self.api_key)

    async def generate(
        self,
        prompt: str,
        size: str = "1024x1024",
        style: Optional[str] = None,
    ) -> ImageGenerateResult:
        """
        使用 ModelScope (通义万相) 生成图片
        """
        try:
            # 解析尺寸
            size_map = {
                "1024x1024": "1024*1024",
                "720x1280": "720*1280",
                "1280x720": "1280*720",
            }
            model_size = size_map.get(size, "1024*1024")

            response = await self._client.post(
                self.BASE_URL,
                headers={
                    "Authorization": f"Bearer {self.api_key}",
                    "Content-Type": "application/json",
                    "X-DashScope-Async": "enable",
                },
                json={
                    "model": self.model,
                    "input": {
                        "prompt": prompt,
                    },
                    "parameters": {
                        "size": model_size,
                        "n": 1,
                        "style": style or "",
                    },
                },
            )

            response.raise_for_status()
            data = response.json()

            # 获取任务 ID，轮询结果
            if "output" in data and "task_id" in data["output"]:
                task_id = data["output"]["task_id"]
                return await self._poll_result(task_id)

            return ImageGenerateResult(
                success=False,
                message=f"生成失败: {data}",
            )

        except httpx.HTTPStatusError as e:
            return ImageGenerateResult(
                success=False,
                message=f"API 错误: {e.response.status_code}",
            )
        except Exception as e:
            return ImageGenerateResult(
                success=False,
                message=f"生成异常: {str(e)}",
            )

    async def _poll_result(self, task_id: str) -> ImageGenerateResult:
        """轮询获取结果"""
        import asyncio

        url = f"{self.BASE_URL}/{task_id}"

        for _ in range(60):  # 最多等待 60 次
            await asyncio.sleep(2)

            response = await self._client.get(
                url,
                headers={"Authorization": f"Bearer {self.api_key}"},
            )

            data = response.json()

            if data["output"].get("task_status") == "SUCCEEDED":
                results = data["output"].get("results", [])
                if results and "url" in results[0]:
                    return ImageGenerateResult(
                        success=True,
                        image_url=results[0]["url"],
                        message="生成成功",
                    )

            elif data["output"].get("task_status") == "FAILED":
                return ImageGenerateResult(
                    success=False,
                    message=f"任务失败: {data['output'].get('message', '未知错误')}",
                )

        return ImageGenerateResult(
            success=False,
            message="生成超时",
        )

    async def close(self) -> None:
        await self._client.aclose()
FILE:src/multi_writing_skills/image/providers/openai.py
"""
OpenAI DALL-E 图片生成 Provider
"""

from typing import Optional

import httpx

from . import ImageGenerateResult, ImageProvider


class OpenAIProvider(ImageProvider):
    """OpenAI DALL-E 图片生成"""

    def __init__(
        self,
        api_key: str,
        base_url: str = "",
        model: str = "dall-e-3",
    ) -> None:
        self.api_key = api_key
        self.base_url = base_url or "https://api.openai.com/v1"
        self.model = model
        self._client = httpx.AsyncClient(timeout=120.0)

    @property
    def name(self) -> str:
        return "openai"

    def is_configured(self) -> bool:
        return bool(self.api_key)

    async def generate(
        self,
        prompt: str,
        size: str = "1024x1024",
        style: Optional[str] = None,
    ) -> ImageGenerateResult:
        """
        使用 DALL-E 生成图片

        Args:
            prompt: 图片描述
            size: 尺寸 (256x256, 512x512, 1024x1024, 1792x1024, 1024x1792)
            style: 风格 (vivid, natural)

        Returns:
            ImageGenerateResult
        """
        try:
            response = await self._client.post(
                f"{self.base_url}/images/generations",
                headers={
                    "Authorization": f"Bearer {self.api_key}",
                    "Content-Type": "application/json",
                },
                json={
                    "model": self.model,
                    "prompt": prompt,
                    "n": 1,
                    "size": size,
                    "style": style or "vivid",
                    "response_format": "url",
                },
            )

            response.raise_for_status()
            data = response.json()

            if "data" in data and len(data["data"]) > 0:
                image_url = data["data"][0]["url"]
                return ImageGenerateResult(
                    success=True,
                    image_url=image_url,
                    message="生成成功",
                )

            return ImageGenerateResult(
                success=False,
                message=f"生成失败: {data}",
            )

        except httpx.HTTPStatusError as e:
            return ImageGenerateResult(
                success=False,
                message=f"API 错误: {e.response.status_code}",
            )
        except Exception as e:
            return ImageGenerateResult(
                success=False,
                message=f"生成异常: {str(e)}",
            )

    async def close(self) -> None:
        await self._client.aclose()
FILE:src/multi_writing_skills/image/providers/zhipu.py
"""
智谱 AI（Zhipu）图片生成 Provider
支持 CogView-3 模型，国内直连
"""

from typing import Optional

import httpx

from . import ImageGenerateResult, ImageProvider


class ZhipuProvider(ImageProvider):
    """智谱 AI 图片生成"""

    BASE_URL = "https://open.bigapi.cn/v1/images/generations"

    def __init__(
        self,
        api_key: str,
        model: str = "cogview-3",
    ) -> None:
        self.api_key = api_key
        self.model = model
        self._client = httpx.AsyncClient(timeout=120.0)

    @property
    def name(self) -> str:
        return "zhipu"

    def is_configured(self) -> bool:
        return bool(self.api_key)

    async def generate(
        self,
        prompt: str,
        size: str = "1024x1024",
        style: Optional[str] = None,
    ) -> ImageGenerateResult:
        """
        使用智谱 CogView-3 生成图片
        """
        try:
            # 解析尺寸
            size_map = {
                "1024x1024": "1024x1024",
                "720x1280": "720x1280",
                "1280x720": "1280x720",
                "1024x1792": "1024x1792",
                "1792x1024": "1792x1024",
            }
            model_size = size_map.get(size, "1024x1024")

            response = await self._client.post(
                self.BASE_URL,
                headers={
                    "Authorization": f"Bearer {self.api_key}",
                    "Content-Type": "application/json",
                },
                json={
                    "model": self.model,
                    "prompt": prompt,
                    "size": model_size,
                    "n": 1,
                    "style": style or "auto",
                },
            )

            response.raise_for_status()
            data = response.json()

            if "data" in data and len(data["data"]) > 0:
                image_url = data["data"][0].get("url") or data["data"][0].get("b64_json")
                if image_url:
                    return ImageGenerateResult(
                        success=True,
                        image_url=image_url,
                        message="生成成功",
                    )

            return ImageGenerateResult(
                success=False,
                message=f"生成失败: {data}",
            )

        except httpx.HTTPStatusError as e:
            return ImageGenerateResult(
                success=False,
                message=f"API 错误: {e.response.status_code}",
            )
        except Exception as e:
            return ImageGenerateResult(
                success=False,
                message=f"生成异常: {str(e)}",
            )

    async def close(self) -> None:
        await self._client.aclose()

FILE:src/multi_writing_skills/pdf/converter.py
"""
Markdown → PDF 转换模块
将 Markdown 内容转换为精美的 PDF 文档
"""

import re
import sys
from pathlib import Path
from typing import Optional

from reportlab.lib import colors
from reportlab.lib.pagesizes import A4
from reportlab.lib.styles import ParagraphStyle
from reportlab.lib.units import cm
from reportlab.lib.utils import ImageReader
from reportlab.platypus import (
    Paragraph,
    SimpleDocTemplate,
    Spacer,
    Table,
    TableStyle,
    PageBreak,
)
from reportlab.pdfbase import pdfmetrics
from reportlab.pdfbase.ttfonts import TTFont

# 页面宽度（A4，左右各 2cm 边距）
PW = A4[0]
MG = 2.0 * cm
CW = PW - 2 * MG  # 内容宽度

# 检测CJK字体
FONT = "Helvetica"
CJKFONT = None
for p in [
    "/usr/share/fonts/google-noto-cjk/NotoSansCJK-DemiLight.ttc,0",
    "/usr/share/fonts/google-noto-cjk/NotoSansCJK-Light.ttc,0",
    "/usr/share/fonts/google-droid/DroidSansFallback.ttf",
]:
    if p.split(",")[0] if "," in p else p:
        path = p.split(",")[0]
        if Path(path).exists():
            try:
                if path.endswith(".ttc"):
                    pdfmetrics.registerFont(TTFont("CJK", path, subfontIndex=0))
                else:
                    pdfmetrics.registerFont(TTFont("CJK", path))
                CJKFONT = "CJK"
                FONT = "CJK"
                break
            except Exception:
                pass

# 颜色
white = colors.white
dark = colors.HexColor("#1E293B")
gray = colors.HexColor("#64748B")
border = colors.HexColor("#E2E8F0")
light_bg = colors.HexColor("#F8FAFC")
code_bg = colors.HexColor("#1E1E2E")
accent = colors.HexColor("#F97316")


def _S(name, **kw):
    """创建段落样式"""
    defaults = dict(fontName=FONT, fontSize=10, leading=14, textColor=dark)
    defaults.update(kw)
    return ParagraphStyle(name, **defaults)


ST = {
    "h1": _S("h1", fontSize=22, leading=28, textColor=dark, fontName=FONT, spaceAfter=6),
    "h2": _S("h2", fontSize=15, leading=20, textColor=dark, fontName=FONT, spaceBefore=14, spaceAfter=4),
    "h3": _S("h3", fontSize=12, leading=16, textColor=dark, fontName=FONT, spaceBefore=10, spaceAfter=3),
    "body": _S("body", fontSize=10, leading=15, textColor=dark),
    "small": _S("small", fontSize=9, leading=13, textColor=gray),
    "code": _S("code", fontName="Courier", fontSize=8.5, leading=13, textColor=colors.HexColor("#E2E8F0")),
    "code_cjk": _S("code_cjk", fontName=CJKFONT or FONT, fontSize=8.5, leading=13, textColor=colors.HexColor("#E2E8F0")),
    "th": _S("th", fontName=FONT, fontSize=9, textColor=white, leading=13),
    "td": _S("td", fontName=FONT, fontSize=9, textColor=dark, leading=13),
}


def _pick_font(line: str) -> str:
    """判断一行文本是否包含中文，选择合适字体"""
    return CJKFONT if CJKFONT and any(ord(c) > 0x3000 for c in line) else "Courier"


def _escape(text: str) -> str:
    """转义 HTML 特殊字符"""
    return (text
        .replace("&", "&")
        .replace("<", "<")
        .replace(">", ">")
        .replace('"', """))


def _render_inline(text: str) -> str:
    """简单渲染 inline markdown（粗体、斜体、行内代码）"""
    # 行内代码 `code`
    text = re.sub(r"`([^`]+)`", r"\1", text)
    # 粗体 **text**
    text = re.sub(r"\*\*([^*]+)\*\*", r"\1", text)
    # 斜体 *text*
    text = re.sub(r"\*([^*]+)\*", r"\1", text)
    # 图片 ![alt](url) → 显示alt文本
    text = re.sub(r"!\[([^\]]*)\]\([^)]+\)", r"[\1]", text)
    # 链接 [text](url) → 显示text
    text = re.sub(r"\[([^\]]+)\]\([^)]+\)", r"\1", text)
    return text


def _build_table(html_content: str, styles) -> Optional[Table]:
    """解析 HTML 表格并构建 ReportLab Table"""
    # 简单处理：匹配 | col1 | col2 | 行
    rows_match = re.findall(r"]*>(.*?)", html_content, re.DOTALL | re.IGNORECASE)
    if not rows_match:
        return None

    table_data = []
    for row_html in rows_match:
        cells = re.findall(r"]*>(.*?)", row_html, re.DOTALL | re.IGNORECASE)
        if cells:
            # 清理 HTML 标签
            clean_cells = []
            for cell in cells:
                cell = re.sub(r"<[^>]+>", "", cell)
                cell = _escape(cell.strip())
                clean_cells.append(Paragraph(cell, ST["td"]))
            table_data.append(clean_cells)

    if not table_data:
        return None

    col_count = max(len(row) for row in table_data)
    col_widths = [CW / col_count] * col_count

    tbl = Table(table_data, colWidths=col_widths)
    tbl.setStyle(TableStyle([
        ("BACKGROUND", (0, 0), (-1, 0), accent),
        ("TEXTCOLOR", (0, 0), (-1, 0), white),
        ("FONTNAME", (0, 0), (-1, 0), FONT),
        ("FONTSIZE", (0, 0), (-1, -1), 9),
        ("GRID", (0, 0), (-1, -1), 0.5, border),
        ("ROWBACKGROUNDS", (0, 1), (-1, -1), [white, light_bg]),
        ("TOPPADDING", (0, 0), (-1, -1), 6),
        ("BOTTOMPADDING", (0, 0), (-1, -1), 6),
        ("LEFTPADDING", (0, 0), (-1, -1), 8),
        ("RIGHTPADDING", (0, 0), (-1, -1), 8),
    ]))
    return tbl


def markdown_to_pdf(
    markdown_content: str,
    output_path: str,
    title: Optional[str] = None,
) -> bool:
    """
    将 Markdown 内容转换为 PDF

    Args:
        markdown_content: Markdown 文本
        output_path: 输出 PDF 路径
        title: 文章标题（从内容中提取或手动指定）

    Returns:
        True 成功，False 失败
    """
    try:
        doc = SimpleDocTemplate(
            output_path,
            pagesize=A4,
            leftMargin=MG,
            rightMargin=MG,
            topMargin=MG,
            bottomMargin=MG,
        )

        story = []
        lines = markdown_content.split("\n")
        i = 0

        # 提取标题
        if not title:
            for line in lines:
                m = re.match(r"^#\s+(.+)", line)
                if m:
                    title = m.group(1).strip()
                    break

        # 封面
        if title:
            story.append(Spacer(1, 3 * cm))
            story.append(Paragraph(title, _S("cover", fontSize=26, leading=32, textColor=dark, fontName=FONT)))
            story.append(Spacer(1, 0.4 * cm))
            story.append(Table([[""]], colWidths=[CW], rowHeights=[3],
                               style=TableStyle([("BACKGROUND", (0, 0), (0, 0), accent)])))
            story.append(Spacer(1, 0.5 * cm))

        while i < len(lines):
            line = lines[i]

            # 跳过 HTML 标签行
            if re.match(r"^\s*<[^>]+>\s*$", line):
                i += 1
                continue

            # 跳过图片行
            if re.match(r"^\s*!\[", line):
                i += 1
                continue

            # H1
            m = re.match(r"^#\s+(.+)", line)
            if m:
                # 封面已经在开头处理了，跳过第一个 H1
                if i > 0:
                    story.append(PageBreak())
                    story.append(Paragraph(m.group(1), ST["h1"]))
                i += 1
                continue

            # H2
            m = re.match(r"^##\s+(.+)", line)
            if m:
                story.append(Spacer(1, 0.3 * cm))
                story.append(Paragraph(m.group(1), ST["h2"]))
                i += 1
                continue

            # H3
            m = re.match(r"^###\s+(.+)", line)
            if m:
                story.append(Paragraph(m.group(1), ST["h3"]))
                i += 1
                continue

            # 水平线
            if re.match(r"^---+$", line.strip()) or re.match(r"^\*\*\*+$", line.strip()):
                story.append(Spacer(1, 0.2 * cm))
                i += 1
                continue

            # 无序列表
            m = re.match(r"^[\-\*]\s+(.+)", line)
            if m:
                item_text = _render_inline(m.group(1).strip())
                story.append(Paragraph(f"• {item_text}", ST["body"]))
                i += 1
                continue

            # 有序列表
            m = re.match(r"^\d+\.\s+(.+)", line)
            if m:
                item_text = _render_inline(m.group(1).strip())
                story.append(Paragraph(f"1. {item_text}", ST["body"]))
                i += 1
                continue

            # 引用块
            if line.strip().startswith(">"):
                # 收集整个引用块
                quote_lines = []
                while i < len(lines) and lines[i].strip().startswith(">"):
                    q = lines[i].strip()[1:].strip()
                    if q:
                        quote_lines.append(q)
                    i += 1
                if quote_lines:
                    quote_text = " ".join(_render_inline(q) for q in quote_lines)
                    q_tbl = Table([[Paragraph(quote_text, _S("quote", fontSize=10, leading=15, textColor=gray, fontName=FONT))]],
                                  colWidths=[CW])
                    q_tbl.setStyle(TableStyle([
                        ("BACKGROUND", (0, 0), (-1, -1), light_bg),
                        ("LEFTPADDING", (0, 0), (-1, -1), 12),
                        ("RIGHTPADDING", (0, 0), (-1, -1), 12),
                        ("TOPPADDING", (0, 0), (-1, -1), 8),
                        ("BOTTOMPADDING", (0, 0), (-1, -1), 8),
                        ("LINEBEFORE", (0, 0), (0, -1), 3, accent),
                    ]))
                    story.append(q_tbl)
                continue

            # 代码块 ```...```
            if line.strip().startswith("```"):
                lang = line.strip()[3:]
                code_lines = []
                i += 1
                while i < len(lines) and not lines[i].strip().startswith("```"):
                    code_lines.append(lines[i])
                    i += 1
                i += 1  # 跳过结束 ```

                if code_lines:
                    code_text = "\n".join(code_lines)
                    fn = _pick_font(code_text)
                    items = [Paragraph(code_text, _S("code_cjk" if fn == CJKFONT else "code",
                                                    fontName=fn, fontSize=8.5, leading=13,
                                                    textColor=colors.HexColor("#E2E8F0")))]
                    code_tbl = Table([[items]], colWidths=[CW])
                    code_tbl.setStyle(TableStyle([
                        ("BACKGROUND", (0, 0), (-1, -1), code_bg),
                        ("TOPPADDING", (0, 0), (-1, -1), 10),
                        ("BOTTOMPADDING", (0, 0), (-1, -1), 10),
                        ("LEFTPADDING", (0, 0), (-1, -1), 14),
                        ("RIGHTPADDING", (0, 0), (-1, -1), 14),
                    ]))
                    story.append(code_tbl)
                    story.append(Spacer(1, 0.15 * cm))
                continue

            # 表格（简单 HTML 表格支持）
            if "
                i += 1
                tbl_html = "\n".join(tbl_lines)
                tbl = _build_table(tbl_html, ST)
                if tbl:
                    story.append(tbl)
                    story.append(Spacer(1, 0.2 * cm))
                continue

            # 空行
            if not line.strip():
                i += 1
                continue

            # 普通文本段落
            rendered = _render_inline(line.strip())
            if rendered:
                story.append(Paragraph(rendered, ST["body"]))
            i += 1

        doc.build(story)
        return True

    except Exception as e:
        print(f"PDF 生成失败: {e}", file=sys.stderr)
        return False

FILE:src/multi_writing_skills/platforms/__init__.py
"""
平台适配器

支持微信公众号、知乎、今日头条等多平台发布。
"""

from .base import Platform, PlatformType, PlatformRegistry, PublishRequest, PublishResult, ImageUploadResult, registry
from .wechat import WeChatPlatform
from .zhihu import ZhihuPlatform
from .toutiao import ToutiaoPlatform

__all__ = [
    "Platform",
    "PlatformType",
    "PlatformRegistry",
    "PublishRequest",
    "PublishResult",
    "ImageUploadResult",
    "WeChatPlatform",
    "ZhihuPlatform",
    "ToutiaoPlatform",
    "registry",
]
FILE:src/multi_writing_skills/platforms/base.py
"""
平台抽象层

定义多平台发布的统一接口。
"""

from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from enum import Enum
from typing import Optional


class PlatformType(str, Enum):
    """支持的平台类型"""

    WECHAT = "wechat"
    ZHIHU = "zhihu"
    TOUTIAO = "toutiao"


@dataclass
class PublishRequest:
    """发布请求"""

    title: str
    content: str  # HTML 格式
    cover: Optional[str] = None
    author: Optional[str] = None
    digest: Optional[str] = None
    source_url: Optional[str] = None
    tags: list[str] = field(default_factory=list)
    category: Optional[str] = None
    column_id: Optional[str] = None  # 知乎专栏 ID


@dataclass
class PublishResult:
    """发布结果"""

    success: bool
    platform: str
    media_id: Optional[str] = None
    article_url: Optional[str] = None
    draft_url: Optional[str] = None
    message: Optional[str] = None
    title: Optional[str] = None


@dataclass
class ImageUploadResult:
    """图片上传结果"""

    success: bool
    url: Optional[str] = None
    media_id: Optional[str] = None
    message: Optional[str] = None


class Platform(ABC):
    """平台抽象基类"""

    @property
    @abstractmethod
    def name(self) -> str:
        """平台标识"""
        pass

    @property
    @abstractmethod
    def display_name(self) -> str:
        """平台显示名称"""
        pass

    @abstractmethod
    async def publish(self, request: PublishRequest) -> PublishResult:
        """
        发布文章

        Args:
            request: 发布请求

        Returns:
            发布结果
        """
        pass

    @abstractmethod
    async def upload_image(self, image_path: str) -> ImageUploadResult:
        """
        上传图片

        Args:
            image_path: 图片路径（本地路径或 URL）

        Returns:
            上传结果
        """
        pass

    @abstractmethod
    async def validate_credentials(self) -> bool:
        """
        验证凭证是否有效

        Returns:
            凭证是否有效
        """
        pass

    @abstractmethod
    def is_configured(self) -> bool:
        """
        检查是否已配置凭证

        Returns:
            是否已配置
        """
        pass


class PlatformRegistry:
    """平台注册表"""

    def __init__(self) -> None:
        self._platforms: dict[str, Platform] = {}

    def register(self, platform: Platform) -> None:
        """注册平台"""
        self._platforms[platform.name] = platform

    def get(self, name: str) -> Optional[Platform]:
        """获取平台"""
        return self._platforms.get(name)

    def list_platforms(self) -> list[Platform]:
        """列出所有平台"""
        return list(self._platforms.values())

    def list_configured(self) -> list[Platform]:
        """列出已配置的平台"""
        return [p for p in self._platforms.values() if p.is_configured()]


# 全局注册表
registry = PlatformRegistry()
FILE:src/multi_writing_skills/platforms/toutiao.py
"""
今日头条平台适配器
"""

import json
import re
import time
from pathlib import Path
from typing import Optional

import httpx

from .base import ImageUploadResult, Platform, PublishRequest, PublishResult


class ToutiaoPlatform(Platform):
    """今日头条平台"""

    BASE_URL = "https://mp.toutiao.com/mp_v3"

    def __init__(self, cookie: str) -> None:
        self.cookie = cookie
        self._client = httpx.AsyncClient(timeout=30.0)

    @property
    def name(self) -> str:
        return "toutiao"

    @property
    def display_name(self) -> str:
        return "今日头条"

    def is_configured(self) -> bool:
        return bool(self.cookie)

    def _get_headers(self) -> dict[str, str]:
        """获取请求头"""
        return {
            "Cookie": self.cookie,
            "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36",
            "Content-Type": "application/json",
            "Referer": "https://mp.toutiao.com/",
        }

    async def validate_credentials(self) -> bool:
        """验证凭证是否有效"""
        try:
            url = f"{self.BASE_URL}/user/info"
            resp = await self._client.get(url, headers=self._get_headers())
            data = resp.json()
            return data.get("ret") == "success"
        except Exception:
            return False

    async def upload_image(self, image_path: str) -> ImageUploadResult:
        """上传图片到头条图床"""
        try:
            # 判断是 URL 还是本地文件
            if image_path.startswith(("http://", "https://")):
                resp = await self._client.get(image_path)
                image_data = resp.content
                filename = image_path.split("/")[-1].split("?")[0]
            else:
                path = Path(image_path)
                if not path.exists():
                    return ImageUploadResult(
                        success=False, message=f"图片文件不存在: {image_path}"
                    )
                image_data = path.read_bytes()
                filename = path.name

            # 头条图片上传
            url = f"{self.BASE_URL}/image/upload"
            headers = self._get_headers()
            del headers["Content-Type"]  # multipart 上传需要删除这个

            files = {"image": (filename, image_data, "image/jpeg")}
            resp = await self._client.post(url, headers=headers, files=files)
            data = resp.json()

            if data.get("ret") == "success" and "url" in data.get("data", {}):
                return ImageUploadResult(
                    success=True,
                    url=data["data"]["url"],
                    message="上传成功",
                )

            return ImageUploadResult(
                success=False, message=f"上传失败: {data.get('msg', '未知错误')}"
            )

        except Exception as e:
            return ImageUploadResult(success=False, message=f"上传异常: {str(e)}")

    async def publish(self, request: PublishRequest) -> PublishResult:
        """发布文章到今日头条"""
        try:
            # 如果有封面图片，先上传
            cover_url = None
            if request.cover:
                result = await self.upload_image(request.cover)
                if result.success and result.url:
                    cover_url = result.url

            # 构建文章数据
            url = f"{self.BASE_URL}/article/create"
            headers = self._get_headers()

            payload = {
                "title": request.title,
                "content": request.content,
                "cover_url": cover_url or "",
                "abstract": request.digest or "",
                "source_url": request.source_url or "",
                "tag": ",".join(request.tags) if request.tags else "",
            }

            resp = await self._client.post(url, headers=headers, json=payload)
            data = resp.json()

            if data.get("ret") == "success":
                article_id = data.get("data", {}).get("article_id")
                article_url = f"https://www.toutiao.com/article/{article_id}"

                return PublishResult(
                    success=True,
                    platform=self.name,
                    article_url=article_url,
                    message="发布成功",
                )

            return PublishResult(
                success=False,
                platform=self.name,
                message=f"发布失败: {data.get('msg', '未知错误')}",
            )

        except Exception as e:
            return PublishResult(
                success=False, platform=self.name, message=f"发布异常: {str(e)}"
            )

    async def close(self) -> None:
        """关闭客户端连接"""
        await self._client.aclose()
FILE:src/multi_writing_skills/platforms/wechat.py
"""
微信公众号平台适配器
"""

import hashlib
import hmac
import time
from pathlib import Path
from typing import Optional

import httpx

from .base import ImageUploadResult, Platform, PublishRequest, PublishResult


class WeChatPlatform(Platform):
    """微信公众号平台"""

    BASE_URL = "https://api.weixin.qq.com/cgi-bin"

    def __init__(self, app_id: str, app_secret: str) -> None:
        self.app_id = app_id
        self.app_secret = app_secret
        self._access_token: Optional[str] = None
        self._token_expires_at: float = 0
        self._client = httpx.AsyncClient(timeout=30.0)

    @property
    def name(self) -> str:
        return "wechat"

    @property
    def display_name(self) -> str:
        return "微信公众号"

    def is_configured(self) -> bool:
        return bool(self.app_id and self.app_secret)

    async def _get_access_token(self) -> str:
        """获取 access_token"""
        # 检查缓存是否有效
        if self._access_token and time.time() < self._token_expires_at:
            return self._access_token

        url = f"{self.BASE_URL}/token"
        params = {
            "grant_type": "client_credential",
            "appid": self.app_id,
            "secret": self.app_secret,
        }

        resp = await self._client.get(url, params=params)
        data = resp.json()

        if "errcode" in data:
            raise RuntimeError(f"获取 access_token 失败: {data.get('errmsg')}")

        self._access_token = data["access_token"]
        # 提前 5 分钟过期
        self._token_expires_at = time.time() + data["expires_in"] - 300

        return self._access_token

    async def validate_credentials(self) -> bool:
        """验证凭证是否有效"""
        try:
            await self._get_access_token()
            return True
        except Exception:
            return False

    async def upload_image(self, image_path: str) -> ImageUploadResult:
        """上传图片到微信素材库"""
        token = await self._get_access_token()

        # 判断是 URL 还是本地文件
        if image_path.startswith(("http://", "https://")):
            # 下载远程图片
            resp = await self._client.get(image_path)
            image_data = resp.content
            filename = image_path.split("/")[-1].split("?")[0]
        else:
            # 读取本地文件
            path = Path(image_path)
            if not path.exists():
                return ImageUploadResult(
                    success=False, message=f"图片文件不存在: {image_path}"
                )
            image_data = path.read_bytes()
            filename = path.name

        # 上传到微信
        url = f"{self.BASE_URL}/material/add_material"
        params = {"access_token": token, "type": "image"}

        files = {"media": (filename, image_data, "image/jpeg")}
        resp = await self._client.post(url, params=params, files=files)
        data = resp.json()

        if "errcode" in data and data["errcode"] != 0:
            return ImageUploadResult(
                success=False, message=f"上传失败: {data.get('errmsg')}"
            )

        return ImageUploadResult(
            success=True, url=data.get("url"), media_id=data.get("media_id")
        )

    async def publish(self, request: PublishRequest) -> PublishResult:
        """发布文章到微信草稿箱"""
        try:
            token = await self._get_access_token()

            # 上传封面图片（必须有封面）
            thumb_media_id = None
            if request.cover:
                result = await self.upload_image(request.cover)
                if result.success and result.media_id:
                    thumb_media_id = result.media_id
                else:
                    return PublishResult(
                        success=False,
                        platform=self.name,
                        message=f"封面上传失败: {result.message}",
                    )
            else:
                # 没有封面时，生成默认封面
                thumb_media_id = await self._upload_default_cover(token)
                if not thumb_media_id:
                    return PublishResult(
                        success=False,
                        platform=self.name,
                        message="默认封面上传失败",
                    )

            # 创建草稿
            url = f"{self.BASE_URL}/draft/add"
            params = {"access_token": token}

            articles = [
                {
                    "title": request.title,
                    "author": request.author or "",
                    "digest": request.digest or "",
                    "content": request.content,
                    "content_source_url": request.source_url or "",
                    "thumb_media_id": thumb_media_id,
                    "need_open_comment": 0,
                    "only_fans_can_comment": 0,
                }
            ]

            payload = {"articles": articles}

            resp = await self._client.post(url, params=params, json=payload)
            data = resp.json()

            if "errcode" in data and data["errcode"] != 0:
                return PublishResult(
                    success=False,
                    platform=self.name,
                    message=f"创建草稿失败: {data.get('errmsg')}",
                )

            media_id = data["media_id"]

            return PublishResult(
                success=True,
                platform=self.name,
                media_id=media_id,
                draft_url=f"https://mp.weixin.qq.com",
                message="草稿创建成功",
            )

        except Exception as e:
            return PublishResult(
                success=False, platform=self.name, message=f"发布失败: {str(e)}"
            )

    async def _upload_default_cover(self, token: str) -> Optional[str]:
        """生成并上传默认封面图片"""
        try:
            import io

            from PIL import Image, ImageDraw, ImageFont

            # 创建一个简单的渐变封面
            width, height = 900, 500
            img = Image.new("RGB", (width, height), color="#4A90D9")

            # 添加一些装饰
            draw = ImageDraw.Draw(img)

            # 绘制渐变效果
            for i in range(height):
                r = int(74 + (i / height) * 30)
                g = int(144 - (i / height) * 50)
                b = int(217 - (i / height) * 30)
                draw.line([(0, i), (width, i)], fill=(r, g, b))

            # 保存到内存
            buffer = io.BytesIO()
            img.save(buffer, format="JPEG", quality=95)
            image_data = buffer.getvalue()

            # 上传到微信
            url = f"{self.BASE_URL}/material/add_material"
            params = {"access_token": token, "type": "image"}

            files = {"media": ("cover.jpg", image_data, "image/jpeg")}
            resp = await self._client.post(url, params=params, files=files)
            data = resp.json()

            if "media_id" in data:
                return data["media_id"]
            return None

        except Exception:
            return None

    async def close(self) -> None:
        """关闭客户端连接"""
        await self._client.aclose()
FILE:src/multi_writing_skills/platforms/zhihu.py
"""
知乎平台适配器
"""

import json
import re
import time
from pathlib import Path
from typing import Optional

import httpx

from .base import ImageUploadResult, Platform, PublishRequest, PublishResult


class ZhihuPlatform(Platform):
    """知乎平台"""

    BASE_URL = "https://www.zhihu.com/api/v4"
    WEB_URL = "https://zhuanlan.zhihu.com"

    def __init__(self, cookie: str) -> None:
        self.cookie = cookie
        self._client = httpx.AsyncClient(timeout=30.0)
        self._x_zse_96: Optional[str] = None
        self._x_xsrf_token: Optional[str] = None

    @property
    def name(self) -> str:
        return "zhihu"

    @property
    def display_name(self) -> str:
        return "知乎"

    def is_configured(self) -> bool:
        return bool(self.cookie)

    def _parse_xsrf_token(self) -> Optional[str]:
        """从 Cookie 中解析 x_xsrf_token"""
        # 尝试 x_xsrf_token 和 _xsrf 两种格式
        match = re.search(r"x_xsrf_token=([^;]+)", self.cookie)
        if not match:
            match = re.search(r"_xsrf=([^;]+)", self.cookie)
        if match:
            return match.group(1)
        return None

    def _get_headers(self, api_version: str = "v4") -> dict[str, str]:
        """获取请求头"""
        xsrf = self._parse_xsrf_token() or ""
        if api_version == "v3":
            # v3 API 需要的请求头
            return {
                "Cookie": self.cookie,
                "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/133.0.0.0 Safari/537.36",
                "Content-Type": "application/json",
                "Referer": "https://www.zhihu.com/composer",
                "X-Xsrftoken": xsrf,
                "Origin": "https://www.zhihu.com",
            }
        return {
            "Cookie": self.cookie,
            "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36",
            "Content-Type": "application/json",
            "x-xsrf-token": xsrf,
            "x-zse-83": "403_3.0",
        }

    def _get_headers_for_zhuanlan(self) -> dict[str, str]:
        """获取知乎专栏请求头"""
        xsrf = self._parse_xsrf_token() or ""
        return {
            "Cookie": self.cookie,
            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/133.0.0.0 Safari/537.36",
            "Content-Type": "application/json",
            "Referer": "https://zhuanlan.zhihu.com/",
            "Origin": "https://zhuanlan.zhihu.com",
            "X-Xsrftoken": xsrf,
        }

    def _get_headers_for_publish(self) -> dict[str, str]:
        """获取发布文章请求头"""
        xsrf = self._parse_xsrf_token() or ""
        return {
            "Cookie": self.cookie,
            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/133.0.0.0 Safari/537.36",
            "Content-Type": "application/json",
            "Referer": "https://www.zhihu.com/",
            "Origin": "https://www.zhihu.com",
            "X-Xsrftoken": xsrf,
        }

    async def validate_credentials(self) -> bool:
        """验证凭证是否有效"""
        try:
            url = "https://www.zhihu.com/api/v4/me"
            resp = await self._client.get(url, headers=self._get_headers())
            data = resp.json()
            return "id" in data
        except Exception:
            return False

    async def upload_image(self, image_path: str) -> ImageUploadResult:
        """上传图片到知乎图床"""
        try:
            # 判断是 URL 还是本地文件
            if image_path.startswith(("http://", "https://")):
                resp = await self._client.get(image_path)
                image_data = resp.content
                filename = image_path.split("/")[-1].split("?")[0]
            else:
                path = Path(image_path)
                if not path.exists():
                    return ImageUploadResult(
                        success=False, message=f"图片文件不存在: {image_path}"
                    )
                image_data = path.read_bytes()
                filename = path.name

            # 知乎图片上传
            url = f"{self.BASE_URL}/images"
            headers = self._get_headers()
            headers["Content-Type"] = "image/jpeg"

            params = {"image_source": "bubian_m", "image_name": filename}

            resp = await self._client.post(
                url, headers=headers, params=params, content=image_data
            )
            data = resp.json()

            if "upload_url" in data:
                # 获取上传结果
                upload_resp = await self._client.put(
                    data["upload_url"],
                    content=image_data,
                    headers={"Content-Type": "image/jpeg"},
                )
                if upload_resp.status_code == 200:
                    return ImageUploadResult(
                        success=True,
                        url=data.get("original_src", ""),
                        message="上传成功",
                    )

            if "src" in data or "original_src" in data:
                return ImageUploadResult(
                    success=True,
                    url=data.get("src") or data.get("original_src", ""),
                    message="上传成功",
                )

            return ImageUploadResult(
                success=False, message=f"上传失败: {data.get('message', '未知错误')}"
            )

        except Exception as e:
            return ImageUploadResult(success=False, message=f"上传异常: {str(e)}")

    async def publish(self, request: PublishRequest) -> PublishResult:
        """发布文章到知乎专栏"""
        try:
            # 如果有封面图片，先上传
            cover_image_id = None
            if request.cover:
                result = await self.upload_image(request.cover)
                if result.success and result.url:
                    cover_image_id = result.url

            # 构建文章数据
            url = f"{self.BASE_URL}/articles"
            headers = self._get_headers()

            payload = {
                "title": request.title,
                "content": request.content,
                "excerpt": request.digest or "",
                "source_url": request.source_url or "",
                "can_comment": True,
                "image_url": cover_image_id or "",
                "disclaimer_type": 0,
                "disclaimer_status": "disclaimer_none",
            }

            # 如果指定了专栏
            if request.column_id:
                payload["column_id"] = request.column_id

            resp = await self._client.post(url, headers=headers, json=payload)
            data = resp.json()

            if "id" in data:
                article_id = data["id"]
                article_url = f"https://zhuanlan.zhihu.com/p/{article_id}"

                return PublishResult(
                    success=True,
                    platform=self.name,
                    article_url=article_url,
                    message="发布成功",
                )

            if "error" in data:
                return PublishResult(
                    success=False,
                    platform=self.name,
                    message=f"发布失败: {data['error'].get('message', '未知错误')}",
                )

            return PublishResult(
                success=False, platform=self.name, message=f"发布失败: {data}"
            )

        except Exception as e:
            return PublishResult(
                success=False, platform=self.name, message=f"发布异常: {str(e)}"
            )

    async def create_draft(self, request: PublishRequest) -> PublishResult:
        """创建文章草稿到知乎"""
        try:
            # 如果有封面图片，先上传
            cover_image_id = None
            if request.cover:
                result = await self.upload_image(request.cover)
                if result.success and result.url:
                    cover_image_id = result.url

            # 使用专栏 API 创建草稿
            url = "https://zhuanlan.zhihu.com/api/articles/drafts"
            headers = self._get_headers_for_zhuanlan()

            # 构建草稿数据
            payload = {
                "title": request.title,
                "content": request.content,
                "excerpt": request.digest or "",
                "image_url": cover_image_id or "",
            }

            resp = await self._client.post(url, headers=headers, json=payload)
            data = resp.json()

            if "id" in data:
                draft_id = data["id"]
                # 草稿没有公开 URL，返回内部 ID
                return PublishResult(
                    success=True,
                    platform=self.name,
                    media_id=str(draft_id),
                    title=request.title,
                    message=f"草稿创建成功，ID: {draft_id}",
                )

            if "error" in data:
                return PublishResult(
                    success=False,
                    platform=self.name,
                    message=f"创建草稿失败: {data['error'].get('message', '未知错误')}",
                )

            return PublishResult(
                success=False, platform=self.name, message=f"创建草稿失败: {data}"
            )

        except Exception as e:
            return PublishResult(
                success=False, platform=self.name, message=f"创建草稿异常: {str(e)}"
            )

    async def publish_draft(
        self,
        draft_id: str,
        title: str = "",
        content: str = "",
        comment_permission: str = "anyone",
    ) -> PublishResult:
        """发布知乎草稿"""
        try:
            import uuid
            import time

            url = "https://www.zhihu.com/api/v4/content/publish"
            headers = self._get_headers_for_publish()

            # 构建发布请求
            trace_id = f"{int(time.time() * 1000)},{uuid.uuid4()}"

            pc_business_params = {
                "commentPermission": comment_permission,
                "disclaimer_type": "none",
                "disclaimer_status": "close",
                "table_of_contents_enabled": False,
                "content": content,
                "title": title,
                "commercial_report_info": {"commercial_types": []},
                "commercial_zhitask_bind_info": None,
                "canReward": False,
            }

            import json

            publish_payload = {
                "action": "article",
                "data": {
                    "publish": {"traceId": trace_id},
                    "extra_info": {
                        "publisher": "pc",
                        "pc_business_params": json.dumps(pc_business_params),
                    },
                    "draft": {
                        "disabled": 1,
                        "id": draft_id,
                        "isPublished": False,
                    },
                    "commentsPermission": {"comment_permission": comment_permission},
                    "creationStatement": {
                        "disclaimer_type": "none",
                        "disclaimer_status": "close",
                    },
                    "contentsTables": {"table_of_contents_enabled": False},
                    "commercialReportInfo": {"isReport": 0},
                    "appreciate": {"can_reward": False, "tagline": ""},
                    "hybridInfo": {},
                    "hybrid": {"html": content, "textLength": len(content)},
                    "title": {"title": title},
                },
            }

            resp = await self._client.post(url, headers=headers, json=publish_payload)
            data = resp.json()

            if data.get("code") == 0 or "success" in data.get("message", "").lower():
                # 解析返回的文章 ID
                result_data = data.get("data", {}).get("result", "{}")
                if isinstance(result_data, str):
                    import json

                    result_data = json.loads(result_data)
                article_id = result_data.get("publish", {}).get("id", draft_id)
                article_url = f"https://zhuanlan.zhihu.com/p/{article_id}"
                return PublishResult(
                    success=True,
                    platform=self.name,
                    article_url=article_url,
                    media_id=article_id,
                    message=f"发布成功: {article_url}",
                )

            return PublishResult(
                success=False,
                platform=self.name,
                message=f"发布草稿失败: {data.get('message', str(data))}",
            )

        except Exception as e:
            return PublishResult(
                success=False, platform=self.name, message=f"发布草稿异常: {str(e)}"
            )

    async def get_columns(self) -> list[dict]:
        """获取用户的专栏列表"""
        try:
            # 先获取用户信息
            me_resp = await self._client.get(
                "https://www.zhihu.com/api/v4/me", headers=self._get_headers()
            )
            me_data = me_resp.json()

            if "url_token" not in me_data:
                return []

            # 获取专栏列表
            url = f"{self.BASE_URL}/people/{me_data['url_token']}/columns"
            resp = await self._client.get(url, headers=self._get_headers())
            data = resp.json()

            if "data" in data:
                return [{"id": c["id"], "title": c["title"]} for c in data["data"]]
            return []

        except Exception:
            return []

    async def close(self) -> None:
        """关闭客户端连接"""
        await self._client.aclose()
FILE:src/multi_writing_skills/prompts/styles.yaml
# 写作风格提示词模板

dan_koe:
  name: "Dan Koe 风格"
  description: "简洁、直接、实用，适合个人成长和效率类文章"
  system_prompt: |
    你是一个擅长 Dan Koe 风格写作的专家。你的写作特点是：

    1. 简洁有力：每句话都有价值，不写废话
    2. 直接表达：开门见山，直奔主题
    3. 实用优先：提供可操作的建议和方法
    4. 个人视角：用第一人称分享经验和见解
    5. 结构清晰：使用列表、小标题等提高可读性

    请用这种风格改写用户的内容。

technical:
  name: "技术风格"
  description: "严谨、详细、专业，适合技术教程和深度文章"
  system_prompt: |
    你是一个技术写作专家。你的写作特点是：

    1. 严谨准确：使用准确的技术术语
    2. 逻辑清晰：按步骤、层次展开
    3. 代码示例：提供可运行的代码片段
    4. 注重细节：解释"为什么"而不只是"怎么做"
    5. 结构化：使用清晰的章节和小标题

    请用这种风格改写用户的内容。

casual:
  name: "随意风格"
  description: "轻松、亲切、口语化，适合生活类和个人博客"
  system_prompt: |
    你是一个擅长轻松写作的专家。你的写作特点是：

    1. 口语化：像和朋友聊天一样
    2. 亲切感：用"我们"、"你"拉近距离
    3. 轻松幽默：适当加入轻松元素
    4. 真实感：分享真实经历和感受
    5. 互动性：经常提问，引发思考

    请用这种风格改写用户的内容。
FILE:src/multi_writing_skills/writer/__init__.py
"""
AI 写作助手

支持多种写作风格的 AI 写作辅助工具。
"""

from dataclasses import dataclass
from enum import Enum
from pathlib import Path
from typing import Optional

import httpx
import yaml


class WritingStyle(str, Enum):
    """写作风格"""

    DAN_KOE = "dan-koe"  # Dan Koe 风格：简洁、直接、实用
    TECHNICAL = "technical"  # 技术风格：严谨、详细、专业
    CASUAL = "casual"  # 随意风格：轻松、亲切、口语化
    FORMAL = "formal"  # 正式风格：严肃、规范、学术
    STORYTELLING = "storytelling"  # 故事风格：叙事、引人入胜


@dataclass
class StyleConfig:
    """风格配置"""

    name: str
    description: str
    system_prompt: str
    tone: str
    example: str


# 预设写作风格
BUILTIN_STYLES: dict[str, StyleConfig] = {
    "dan-koe": StyleConfig(
        name="Dan Koe 风格",
        description="简洁、直接、实用，适合个人成长和效率类文章",
        system_prompt="""你是一个擅长 Dan Koe 风格写作的专家。你的写作特点是：

1. 简洁有力：每句话都有价值，不写废话
2. 直接表达：开门见山，直奔主题
3. 实用优先：提供可操作的建议和方法
4. 个人视角：用第一人称分享经验和见解
5. 结构清晰：使用列表、小标题等提高可读性

请用这种风格改写用户的内容。""",
        tone="专业但亲切，直接但不生硬",
        example="很多人想提高效率，但不知道从何开始。其实方法很简单：先做最重要的事。",
    ),
    "technical": StyleConfig(
        name="技术风格",
        description="严谨、详细、专业，适合技术教程和深度文章",
        system_prompt="""你是一个技术写作专家。你的写作特点是：

1. 严谨准确：使用准确的技术术语
2. 逻辑清晰：按步骤、层次展开
3. 代码示例：提供可运行的代码片段
4. 注重细节：解释"为什么"而不只是"怎么做"
5. 结构化：使用清晰的章节和小标题

请用这种风格改写用户的内容。""",
        tone="专业、客观、深入",
        example="首先，我们需要理解核心概念。然后，通过代码示例演示具体实现。",
    ),
    "casual": StyleConfig(
        name="随意风格",
        description="轻松、亲切、口语化，适合生活类和个人博客",
        system_prompt="""你是一个擅长轻松写作的专家。你的写作特点是：

1. 口语化：像和朋友聊天一样
2. 亲切感：用"我们"、"你"拉近距离
3. 轻松幽默：适当加入轻松元素
4. 真实感：分享真实经历和感受
5. 互动性：经常提问，引发思考

请用这种风格改写用户的内容。""",
        tone="轻松、亲切、有趣",
        example="说实话，我也曾遇到过这个问题。后来发现，其实解决起来没那么难。",
    ),
    "formal": StyleConfig(
        name="正式风格",
        description="严肃、规范、学术，适合专业论文和正式报告",
        system_prompt="""你是一个正式写作专家。你的写作特点是：

1. 规范用语：使用标准的书面语
2. 客观中立：避免主观情感表达
3. 逻辑严密：论证有理有据
4. 引用出处：标注信息来源
5. 格式规范：符合学术或商务写作规范

请用这种风格改写用户的内容。""",
        tone="正式、专业、客观",
        example="根据研究表明，该方法在实践中具有较高的可行性。",
    ),
    "storytelling": StyleConfig(
        name="故事风格",
        description="叙事、引人入胜，适合情感类和案例分享",
        system_prompt="""你是一个故事写作专家。你的写作特点是：

1. 场景感：描述具体的时间、地点、人物
2. 情节性：有起承转合
3. 情感共鸣：引发读者的情感反应
4. 人物刻画：生动描述人物特点
5. 意义升华：从故事中提炼经验和道理

请用这种风格改写用户的内容。""",
        tone="生动、感人、有感染力",
        example="那是一个平凡的下午，我收到了改变人生的那封邮件...",
    ),
}


class AIWriter:
    """AI 写作助手"""

    def __init__(
        self,
        api_key: str,
        provider: str = "openai",
        base_url: str = "",
        model: str = "gpt-4",
    ) -> None:
        self.api_key = api_key
        self.provider = provider
        self.base_url = base_url
        self.model = model
        self._client = httpx.AsyncClient(timeout=120.0)

    def get_style(self, style_name: str) -> Optional[StyleConfig]:
        """获取写作风格配置"""
        return BUILTIN_STYLES.get(style_name)

    def list_styles(self) -> list[dict]:
        """列出所有可用的写作风格"""
        return [
            {
                "name": name,
                "display_name": config.name,
                "description": config.description,
            }
            for name, config in BUILTIN_STYLES.items()
        ]

    async def write(
        self,
        topic: str,
        style: str = "dan-koe",
        length: str = "medium",
        context: Optional[str] = None,
    ) -> str:
        """
        根据主题生成文章

        Args:
            topic: 写作主题
            style: 写作风格
            length: 文章长度 (short/medium/long)
            context: 额外上下文

        Returns:
            生成的文章内容
        """
        style_config = self.get_style(style)
        if not style_config:
            raise ValueError(f"未知的写作风格: {style}")

        length_guide = {
            "short": "300-500字",
            "medium": "800-1200字",
            "long": "2000-3000字",
        }

        prompt = f"""{style_config.system_prompt}

主题：{topic}

要求：
- 文章长度：{length_guide.get(length, "800-1200字")}
- 写作风格：{style_config.tone}
{f"- 补充信息：{context}" if context else ""}

请直接输出文章内容，使用 Markdown 格式。"""

        return await self._call_ai(prompt)

    async def rewrite(
        self,
        content: str,
        style: str = "dan-koe",
        keep_structure: bool = True,
    ) -> str:
        """
        用指定风格改写内容

        Args:
            content: 原始内容
            style: 目标写作风格
            keep_structure: 是否保留原有结构

        Returns:
            改写后的内容
        """
        style_config = self.get_style(style)
        if not style_config:
            raise ValueError(f"未知的写作风格: {style}")

        prompt = f"""{style_config.system_prompt}

原始内容：
{content}

要求：
- 保持原文的核心观点和信息
{'- 保留原有的文章结构' if keep_structure else '- 可以重新组织文章结构'}
- 使用目标写作风格进行改写

请直接输出改写后的内容，使用 Markdown 格式。"""

        return await self._call_ai(prompt)

    async def generate_cover_prompt(
        self,
        title: str,
        content: Optional[str] = None,
    ) -> str:
        """
        生成封面图片的提示词

        Args:
            title: 文章标题
            content: 文章内容（可选）

        Returns:
            图片生成提示词
        """
        prompt = f"""请根据以下文章信息，生成一个适合作为封面图片的英文提示词。

标题：{title}
{f"内容摘要：{content[:500]}..." if content else ""}

要求：
1. 提示词用英文描述
2. 适合生成简洁、专业的封面图
3. 避免文字和复杂场景
4. 突出主题，视觉冲击力强

请只输出提示词，不要其他内容。"""

        return await self._call_ai(prompt)

    async def _call_ai(self, prompt: str) -> str:
        """调用 AI API"""
        base_url = self.base_url or "https://api.openai.com/v1"

        response = await self._client.post(
            f"{base_url}/chat/completions",
            headers={
                "Authorization": f"Bearer {self.api_key}",
                "Content-Type": "application/json",
            },
            json={
                "model": self.model,
                "messages": [{"role": "user", "content": prompt}],
                "temperature": 0.7,
            },
        )

        response.raise_for_status()
        data = response.json()
        return data["choices"][0]["message"]["content"]

    async def close(self) -> None:
        """关闭客户端连接"""
        await self._client.aclose()
FILE:test_article.md
# OpenClaw 架构拆解：深入解析现代爬虫框架的设计与实现

## 引言

在当今数据驱动的时代，爬虫技术已成为获取互联网数据的重要手段。OpenClaw 作为一款新兴的爬虫框架，以其模块化设计和强大的扩展性受到了广泛关注。本文将深入拆解 OpenClaw 的架构设计，帮助读者理解其核心设计理念。

![OpenClaw Logo](https://example.com/openclaw-logo.png)

> "优秀的框架不是功能的堆砌，而是对复杂性的优雅封装。" —— OpenClaw 设计哲学

---

## 一、整体架构概览

OpenClaw 采用分层架构设计，主要分为以下几个核心层次：

### 1.1 架构图

```
┌─────────────────────────────────────────────────────────┐
│                    Application Layer                     │
│  (CLI, API Server, Scheduler)                           │
├─────────────────────────────────────────────────────────┤
│                    Core Engine Layer                     │
│  (Request Manager, Response Handler, Pipeline)          │
├─────────────────────────────────────────────────────────┤
│                    Extractor Layer                       │
│  (CSS Selector, XPath, Regex, JSON Path)                │
├─────────────────────────────────────────────────────────┤
│                    Infrastructure Layer                  │
│  (Storage, Cache, Proxy, Middleware)                    │
└─────────────────────────────────────────────────────────┘
```

### 1.2 核心模块说明

| 模块名称 | 功能描述 | 重要程度 |
|---------|---------|---------|
| Request Manager | 请求调度与并发控制 | ⭐⭐⭐⭐⭐ |
| Response Handler | 响应解析与预处理 | ⭐⭐⭐⭐ |
| Extractor | 数据提取与清洗 | ⭐⭐⭐⭐⭐ |
| Pipeline | 数据存储与后处理 | ⭐⭐⭐⭐ |
| Middleware | 请求/响应拦截扩展 | ⭐⭐⭐ |

---

## 二、核心设计模式

### 2.1 装饰器模式

OpenClaw 大量使用装饰器模式来实现功能的灵活扩展：

```python
@spider(name="example")
class ExampleSpider:
    @on_request(url="https://example.com/*")
    def parse_list(self, response):
        # 解析列表页
        pass

    @on_response(selector=".article")
    def parse_detail(self, response):
        # 解析详情页
        pass

    @pipeline(item=ArticleItem)
    def save_article(self, item):
        # 保存文章
        db.articles.insert(item)
```

### 2.2 中间件模式

通过中间件可以实现请求/响应的拦截和处理：

```python
class ProxyMiddleware:
    def process_request(self, request):
        request.meta['proxy'] = get_proxy()
        return request

    def process_response(self, response):
        if response.status == 407:
            # 代理认证失败，更换代理
            return retry_request(response.request)
        return response
```

---

## 三、关键技术实现

### 3.1 异步并发控制

OpenClaw 基于 asyncio 实现高效的并发控制：

```python
class RequestScheduler:
    def __init__(self, max_concurrent=10):
        self.semaphore = asyncio.Semaphore(max_concurrent)
        self.queue = asyncio.Queue()

    async def fetch(self, request):
        async with self.semaphore:
            return await self.downloader.download(request)
```

### 3.2 分布式支持

通过 Redis 实现分布式任务调度：

```python
class DistributedScheduler:
    def __init__(self, redis_url):
        self.redis = await aioredis.create_redis_pool(redis_url)

    async def schedule(self, spider_name, requests):
        for req in requests:
            await self.redis.lpush(f"queue:{spider_name}", req)
```

---

## 四、性能优化策略

### 4.1 连接池管理

- HTTP 连接复用：使用 aiohttp 连接池
- 数据库连接池：预建立连接，减少创建开销

### 4.2 缓存策略

```
┌──────────────┬─────────────┬──────────────┐
│   缓存类型    │   存储介质   │    适用场景   │
├──────────────┼─────────────┼──────────────┤
│ 响应缓存     │ Redis/Memcached │ 重复请求 │
│ DNS 缓存     │    内存      │  域名解析   │
│ 请求去重     │    Redis    │  防止重复   │
└──────────────┴─────────────┴──────────────┘
```

### 4.3 增量爬取

通过时间戳或版本号实现增量更新：

```python
async def should_fetch(self, url, last_update):
    remote_time = await self.get_remote_time(url)
    return remote_time > last_update
```

---

## 五、扩展性设计

### 5.1 插件系统

OpenClaw 支持插件扩展，开发者可以通过以下方式自定义功能：

1. **自定义提取器** - 实现 `Extractor` 接口
2. **自定义存储** - 实现 `Storage` 接口
3. **自定义中间件** - 继承 `Middleware` 基类

### 5.2 配置驱动

通过 YAML/JSON 配置文件灵活控制爬虫行为：

```yaml
spider:
  name: my_spider
  settings:
    concurrent: 10
    retry: 3
    timeout: 30

  pipelines:
    - type: file
      path: ./data.json
    - type: mongodb
      uri: mongodb://localhost:27017
```

---

## 六、最佳实践

### 6.1 项目结构推荐

```
my_spider/
├── spiders/
│   ├── __init__.py
│   └── my_spider.py
├── pipelines/
│   ├── __init__.py
│   └── data_pipeline.py
├── middlewares/
│   ├── __init__.py
│   └── proxy_middleware.py
├── settings.yaml
└── main.py
```

### 6.2 错误处理建议

1. **重试机制**：对网络错误自动重试
2. **异常隔离**：单个请求失败不影响整体
3. **日志记录**：详细记录爬取过程便于排查

---

## 七、总结

OpenClaw 通过模块化、插件化的架构设计，为爬虫开发提供了灵活且强大的解决方案。其核心优势包括：

- 🚀 **高性能**：基于 asyncio 的异步架构
- 🔌 **高扩展**：插件化的中间件系统
- 📦 **易使用**：简洁的 API 设计
- 🌍 **分布式**：支持大规模分布式爬取

随着互联网数据价值的不断提升，OpenClaw 这类现代化爬虫框架将发挥越来越重要的作用。

---

**作者**：Yuesf
**首发平台**：微信公众号
**发布日期**：2024年

---

*如果觉得文章对你有帮助，欢迎关注、点赞、转发！*
FILE:test_output.html
OpenClaw 架构拆解：深入解析现代爬虫框架的设计与实现
OpenClaw 架构拆解：深入解析现代爬虫框架的设计与实现

引言

在当今数据驱动的时代，爬虫技术已成为获取互联网数据的重要手段。OpenClaw 作为一款新兴的爬虫框架，以其模块化设计和强大的扩展性受到了广泛关注。本文将深入拆解 OpenClaw 的架构设计，帮助读者理解其核心设计理念。



"优秀的框架不是功能的堆砌，而是对复杂性的优雅封装。" —— OpenClaw 设计哲学

•--
一、整体架构概览

OpenClaw 采用分层架构设计，主要分为以下几个核心层次：

1.1 架构图






Code

┌─────────────────────────────────────────────────────────┐
│                    Application Layer                     │
│  (CLI, API Server, Scheduler)                           │
├─────────────────────────────────────────────────────────┤
│                    Core Engine Layer                     │
│  (Request Manager, Response Handler, Pipeline)          │
├─────────────────────────────────────────────────────────┤
│                    Extractor Layer                       │
│  (CSS Selector, XPath, Regex, JSON Path)                │
├─────────────────────────────────────────────────────────┤
│                    Infrastructure Layer                  │
│  (Storage, Cache, Proxy, Middleware)                    │
└─────────────────────────────────────────────────────────┘


1.2 核心模块说明



模块名称 功能描述 重要程度

Request Manager 请求调度与并发控制 ⭐⭐⭐⭐⭐
Response Handler 响应解析与预处理 ⭐⭐⭐⭐
Extractor 数据提取与清洗 ⭐⭐⭐⭐⭐
Pipeline 数据存储与后处理 ⭐⭐⭐⭐
Middleware 请求/响应拦截扩展 ⭐⭐⭐


•--
二、核心设计模式

2.1 装饰器模式

OpenClaw 大量使用装饰器模式来实现功能的灵活扩展：






Python

@spider(name="example")
class ExampleSpider:
    @on_request(url="https://example.com/*")
    def parse_list(self, response):
        # 解析列表页
        pass

    @on_response(selector=".article")
    def parse_detail(self, response):
        # 解析详情页
        pass

    @pipeline(item=ArticleItem)
    def save_article(self, item):
        # 保存文章
        db.articles.insert(item)


2.2 中间件模式

通过中间件可以实现请求/响应的拦截和处理：






Python

class ProxyMiddleware:
    def process_request(self, request):
        request.meta['proxy'] = get_proxy()
        return request

    def process_response(self, response):
        if response.status == 407:
            # 代理认证失败，更换代理
            return retry_request(response.request)
        return response


•--
三、关键技术实现

3.1 异步并发控制

OpenClaw 基于 asyncio 实现高效的并发控制：






Python

class RequestScheduler:
    def __init__(self, max_concurrent=10):
        self.semaphore = asyncio.Semaphore(max_concurrent)
        self.queue = asyncio.Queue()

    async def fetch(self, request):
        async with self.semaphore:
            return await self.downloader.download(request)


3.2 分布式支持

通过 Redis 实现分布式任务调度：






Python

class DistributedScheduler:
    def __init__(self, redis_url):
        self.redis = await aioredis.create_redis_pool(redis_url)

    async def schedule(self, spider_name, requests):
        for req in requests:
            await self.redis.lpush(f"queue:{spider_name}", req)


•--
四、性能优化策略

4.1 连接池管理

•HTTP 连接复用：使用 aiohttp 连接池
•数据库连接池：预建立连接，减少创建开销
4.2 缓存策略






Code

┌──────────────┬─────────────┬──────────────┐
│   缓存类型    │   存储介质   │    适用场景   │
├──────────────┼─────────────┼──────────────┤
│ 响应缓存     │ Redis/Memcached │ 重复请求 │
│ DNS 缓存     │    内存      │  域名解析   │
│ 请求去重     │    Redis    │  防止重复   │
└──────────────┴─────────────┴──────────────┘


4.3 增量爬取

通过时间戳或版本号实现增量更新：






Python

async def should_fetch(self, url, last_update):
    remote_time = await self.get_remote_time(url)
    return remote_time > last_update


•--
五、扩展性设计

5.1 插件系统

OpenClaw 支持插件扩展，开发者可以通过以下方式自定义功能：

1.自定义提取器 - 实现 Extractor 接口
2.自定义存储 - 实现 Storage 接口
3.自定义中间件 - 继承 Middleware 基类
5.2 配置驱动

通过 YAML/JSON 配置文件灵活控制爬虫行为：






YAML

spider:
  name: my_spider
  settings:
    concurrent: 10
    retry: 3
    timeout: 30

  pipelines:
    - type: file
      path: ./data.json
    - type: mongodb
      uri: mongodb://localhost:27017


•--
六、最佳实践

6.1 项目结构推荐






Code

my_spider/
├── spiders/
│   ├── __init__.py
│   └── my_spider.py
├── pipelines/
│   ├── __init__.py
│   └── data_pipeline.py
├── middlewares/
│   ├── __init__.py
│   └── proxy_middleware.py
├── settings.yaml
└── main.py


6.2 错误处理建议

1.重试机制：对网络错误自动重试
2.异常隔离：单个请求失败不影响整体
3.日志记录：详细记录爬取过程便于排查
0.--
七、总结

OpenClaw 通过模块化、插件化的架构设计，为爬虫开发提供了灵活且强大的解决方案。其核心优势包括：

•🚀 高性能：基于 asyncio 的异步架构
•🔌 高扩展：插件化的中间件系统
•📦 易使用：简洁的 API 设计
•🌍 分布式：支持大规模分布式爬取
随着互联网数据价值的不断提升，OpenClaw 这类现代化爬虫框架将发挥越来越重要的作用。

•--
•作者：Yuesf

•首发平台：微信公众号
•发布日期：2024年

•--
•如果觉得文章对你有帮助，欢迎关注、点赞、转发！
FILE:test_output_v2.html
OpenClaw 架构拆解：深入解析现代爬虫框架的设计与实现
OpenClaw 架构拆解：深入解析现代爬虫框架的设计与实现

引言

在当今数据驱动的时代，爬虫技术已成为获取互联网数据的重要手段。OpenClaw 作为一款新兴的爬虫框架，以其模块化设计和强大的扩展性受到了广泛关注。本文将深入拆解 OpenClaw 的架构设计，帮助读者理解其核心设计理念。



"优秀的框架不是功能的堆砌，而是对复杂性的优雅封装。" —— OpenClaw 设计哲学

•--
一、整体架构概览

OpenClaw 采用分层架构设计，主要分为以下几个核心层次：

1.1 架构图






Code

┌─────────────────────────────────────────────────────────┐
│                    Application Layer                     │
│  (CLI, API Server, Scheduler)                           │
├─────────────────────────────────────────────────────────┤
│                    Core Engine Layer                     │
│  (Request Manager, Response Handler, Pipeline)          │
├─────────────────────────────────────────────────────────┤
│                    Extractor Layer                       │
│  (CSS Selector, XPath, Regex, JSON Path)                │
├─────────────────────────────────────────────────────────┤
│                    Infrastructure Layer                  │
│  (Storage, Cache, Proxy, Middleware)                    │
└─────────────────────────────────────────────────────────┘



1.2 核心模块说明



模块名称 功能描述 重要程度

Request Manager 请求调度与并发控制 ⭐⭐⭐⭐⭐
Response Handler 响应解析与预处理 ⭐⭐⭐⭐
Extractor 数据提取与清洗 ⭐⭐⭐⭐⭐
Pipeline 数据存储与后处理 ⭐⭐⭐⭐
Middleware 请求/响应拦截扩展 ⭐⭐⭐


•--
二、核心设计模式

2.1 装饰器模式

OpenClaw 大量使用装饰器模式来实现功能的灵活扩展：






Python

@spider(name="example")
class ExampleSpider:
    @on_request(url="https:">//example.com/*")
    def parse_list(self, response):
        "># 解析列表页
        pass

    @on_response(selector=".article")
    def parse_detail(self, response):
        "># 解析详情页
        pass

    @pipeline(item=ArticleItem)
    def save_article(self, item):
        # 保存文章
        db.articles.insert(item)



2.2 中间件模式

通过中间件可以实现请求/响应的拦截和处理：






Python

class ProxyMiddleware:
    def process_request(self, request):
        request.meta['proxy'] = get_proxy()
        return request

    def process_response(self, response):
        if response.status == 407:
            # 代理认证失败，更换代理
            return retry_request(response.request)
        return response



•--
三、关键技术实现

3.1 异步并发控制

OpenClaw 基于 asyncio 实现高效的并发控制：






Python

class RequestScheduler:
    def __init__(self, max_concurrent=10):
        self.semaphore = asyncio.Semaphore(max_concurrent)
        self.queue = asyncio.Queue()

    async def fetch(self, request):
        async with self.semaphore:
            return await self.downloader.download(request)



3.2 分布式支持

通过 Redis 实现分布式任务调度：






Python

class DistributedScheduler:
    def __init__(self, redis_url):
        self.redis = await aioredis.create_redis_pool(redis_url)

    async def schedule(self, spider_name, requests):
        for req in requests:
            await self.redis.lpush(f"queue:{spider_name}", req)



•--
四、性能优化策略

4.1 连接池管理

•HTTP 连接复用：使用 aiohttp 连接池
•数据库连接池：预建立连接，减少创建开销
4.2 缓存策略






Code

┌──────────────┬─────────────┬──────────────┐
│   缓存类型    │   存储介质   │    适用场景   │
├──────────────┼─────────────┼──────────────┤
│ 响应缓存     │ Redis/Memcached │ 重复请求 │
│ DNS 缓存     │    内存      │  域名解析   │
│ 请求去重     │    Redis    │  防止重复   │
└──────────────┴─────────────┴──────────────┘



4.3 增量爬取

通过时间戳或版本号实现增量更新：






Python

async def should_fetch(self, url, last_update):
    remote_time = await self.get_remote_time(url)
    return remote_time > last_update



•--
五、扩展性设计

5.1 插件系统

OpenClaw 支持插件扩展，开发者可以通过以下方式自定义功能：

1.自定义提取器 - 实现 Extractor 接口
2.自定义存储 - 实现 Storage 接口
3.自定义中间件 - 继承 Middleware 基类
5.2 配置驱动

通过 YAML/JSON 配置文件灵活控制爬虫行为：






YAML

spider:
  name: my_spider
  settings:
    concurrent: 10
    retry: 3
    timeout: 30

  pipelines:
    - type: file
      path: ./data.json
    - type: mongodb
      uri: mongodb://localhost:27017



•--
六、最佳实践

6.1 项目结构推荐






Code

my_spider/
├── spiders/
│   ├── __init__.py
│   └── my_spider.py
├── pipelines/
│   ├── __init__.py
│   └── data_pipeline.py
├── middlewares/
│   ├── __init__.py
│   └── proxy_middleware.py
├── settings.yaml
└── main.py



6.2 错误处理建议

1.重试机制：对网络错误自动重试
2.异常隔离：单个请求失败不影响整体
3.日志记录：详细记录爬取过程便于排查
0.--
七、总结

OpenClaw 通过模块化、插件化的架构设计，为爬虫开发提供了灵活且强大的解决方案。其核心优势包括：

•🚀 高性能：基于 asyncio 的异步架构
•🔌 高扩展：插件化的中间件系统
•📦 易使用：简洁的 API 设计
•🌍 分布式：支持大规模分布式爬取
随着互联网数据价值的不断提升，OpenClaw 这类现代化爬虫框架将发挥越来越重要的作用。

•--
•作者：Yuesf

•首发平台：微信公众号
•发布日期：2024年

•--
•如果觉得文章对你有帮助，欢迎关注、点赞、转发！
FILE:test_output_v3.html
OpenClaw 架构拆解：深入解析现代爬虫框架的设计与实现
OpenClaw 架构拆解：深入解析现代爬虫框架的设计与实现

引言

在当今数据驱动的时代，爬虫技术已成为获取互联网数据的重要手段。OpenClaw 作为一款新兴的爬虫框架，以其模块化设计和强大的扩展性受到了广泛关注。本文将深入拆解 OpenClaw 的架构设计，帮助读者理解其核心设计理念。



"优秀的框架不是功能的堆砌，而是对复杂性的优雅封装。" —— OpenClaw 设计哲学

•--
一、整体架构概览

OpenClaw 采用分层架构设计，主要分为以下几个核心层次：

1.1 架构图






Code

┌─────────────────────────────────────────────────────────┐
│                    Application Layer                     │
│  (CLI, API Server, Scheduler)                           │
├─────────────────────────────────────────────────────────┤
│                    Core Engine Layer                     │
│  (Request Manager, Response Handler, Pipeline)          │
├─────────────────────────────────────────────────────────┤
│                    Extractor Layer                       │
│  (CSS Selector, XPath, Regex, JSON Path)                │
├─────────────────────────────────────────────────────────┤
│                    Infrastructure Layer                  │
│  (Storage, Cache, Proxy, Middleware)                    │
└─────────────────────────────────────────────────────────┘



1.2 核心模块说明



模块名称 功能描述 重要程度

Request Manager 请求调度与并发控制 ⭐⭐⭐⭐⭐
Response Handler 响应解析与预处理 ⭐⭐⭐⭐
Extractor 数据提取与清洗 ⭐⭐⭐⭐⭐
Pipeline 数据存储与后处理 ⭐⭐⭐⭐
Middleware 请求/响应拦截扩展 ⭐⭐⭐


•--
二、核心设计模式

2.1 装饰器模式

OpenClaw 大量使用装饰器模式来实现功能的灵活扩展：






Python

@spider(name="example")
class ExampleSpider:
    @on_request(url="https:">//example.com/*")
    def parse_list(self, response):
        # 解析列表页
        pass

    @on_response(selector=".article")
    def parse_detail(self, response):
        # 解析详情页
        pass

    @pipeline(item=ArticleItem)
    def save_article(self, item):
        # 保存文章
        db.articles.insert(item)



2.2 中间件模式

通过中间件可以实现请求/响应的拦截和处理：






Python

class ProxyMiddleware:
    def process_request(self, request):
        request.meta['proxy'] = get_proxy()
        return request

    def process_response(self, response):
        if response.status == 407:
            # 代理认证失败，更换代理
            return retry_request(response.request)
        return response



•--
三、关键技术实现

3.1 异步并发控制

OpenClaw 基于 asyncio 实现高效的并发控制：






Python

class RequestScheduler:
    def __init__(self, max_concurrent=10):
        self.semaphore = asyncio.Semaphore(max_concurrent)
        self.queue = asyncio.Queue()

    async def fetch(self, request):
        async with self.semaphore:
            return await self.downloader.download(request)



3.2 分布式支持

通过 Redis 实现分布式任务调度：






Python

class DistributedScheduler:
    def __init__(self, redis_url):
        self.redis = await aioredis.create_redis_pool(redis_url)

    async def schedule(self, spider_name, requests):
        for req in requests:
            await self.redis.lpush(f"queue:{spider_name}", req)



•--
四、性能优化策略

4.1 连接池管理

•HTTP 连接复用：使用 aiohttp 连接池
•数据库连接池：预建立连接，减少创建开销
4.2 缓存策略






Code

┌──────────────┬─────────────┬──────────────┐
│   缓存类型    │   存储介质   │    适用场景   │
├──────────────┼─────────────┼──────────────┤
│ 响应缓存     │ Redis/Memcached │ 重复请求 │
│ DNS 缓存     │    内存      │  域名解析   │
│ 请求去重     │    Redis    │  防止重复   │
└──────────────┴─────────────┴──────────────┘



4.3 增量爬取

通过时间戳或版本号实现增量更新：






Python

async def should_fetch(self, url, last_update):
    remote_time = await self.get_remote_time(url)
    return remote_time > last_update



•--
五、扩展性设计

5.1 插件系统

OpenClaw 支持插件扩展，开发者可以通过以下方式自定义功能：

1.自定义提取器 - 实现 Extractor 接口
2.自定义存储 - 实现 Storage 接口
3.自定义中间件 - 继承 Middleware 基类
5.2 配置驱动

通过 YAML/JSON 配置文件灵活控制爬虫行为：






YAML

spider:
  name: my_spider
  settings:
    concurrent: 10
    retry: 3
    timeout: 30

  pipelines:
    - type: file
      path: ./data.json
    - type: mongodb
      uri: mongodb://localhost:27017



•--
六、最佳实践

6.1 项目结构推荐






Code

my_spider/
├── spiders/
│   ├── __init__.py
│   └── my_spider.py
├── pipelines/
│   ├── __init__.py
│   └── data_pipeline.py
├── middlewares/
│   ├── __init__.py
│   └── proxy_middleware.py
├── settings.yaml
└── main.py



6.2 错误处理建议

1.重试机制：对网络错误自动重试
2.异常隔离：单个请求失败不影响整体
3.日志记录：详细记录爬取过程便于排查
0.--
七、总结

OpenClaw 通过模块化、插件化的架构设计，为爬虫开发提供了灵活且强大的解决方案。其核心优势包括：

•🚀 高性能：基于 asyncio 的异步架构
•🔌 高扩展：插件化的中间件系统
•📦 易使用：简洁的 API 设计
•🌍 分布式：支持大规模分布式爬取
随着互联网数据价值的不断提升，OpenClaw 这类现代化爬虫框架将发挥越来越重要的作用。

•--
•作者：Yuesf

•首发平台：微信公众号
•发布日期：2024年

•--
•如果觉得文章对你有帮助，欢迎关注、点赞、转发！
FILE:test_output_v4.html
OpenClaw 架构拆解：深入解析现代爬虫框架的设计与实现
OpenClaw 架构拆解：深入解析现代爬虫框架的设计与实现

引言

在当今数据驱动的时代，爬虫技术已成为获取互联网数据的重要手段。OpenClaw 作为一款新兴的爬虫框架，以其模块化设计和强大的扩展性受到了广泛关注。本文将深入拆解 OpenClaw 的架构设计，帮助读者理解其核心设计理念。



"优秀的框架不是功能的堆砌，而是对复杂性的优雅封装。" —— OpenClaw 设计哲学

•--
一、整体架构概览

OpenClaw 采用分层架构设计，主要分为以下几个核心层次：

1.1 架构图






Code

┌─────────────────────────────────────────────────────────┐
│                    Application Layer                     │
│  (CLI, API Server, Scheduler)                           │
├─────────────────────────────────────────────────────────┤
│                    Core Engine Layer                     │
│  (Request Manager, Response Handler, Pipeline)          │
├─────────────────────────────────────────────────────────┤
│                    Extractor Layer                       │
│  (CSS Selector, XPath, Regex, JSON Path)                │
├─────────────────────────────────────────────────────────┤
│                    Infrastructure Layer                  │
│  (Storage, Cache, Proxy, Middleware)                    │
└─────────────────────────────────────────────────────────┘



1.2 核心模块说明



模块名称 功能描述 重要程度

Request Manager 请求调度与并发控制 ⭐⭐⭐⭐⭐
Response Handler 响应解析与预处理 ⭐⭐⭐⭐
Extractor 数据提取与清洗 ⭐⭐⭐⭐⭐
Pipeline 数据存储与后处理 ⭐⭐⭐⭐
Middleware 请求/响应拦截扩展 ⭐⭐⭐


•--
二、核心设计模式

2.1 装饰器模式

OpenClaw 大量使用装饰器模式来实现功能的灵活扩展：






Python

@spider(name="example")
class ExampleSpider:
    @on_request(url="https://example.com/*")
    def parse_list(self, response):
        "color: #5c6370; font-style: italic;"># 解析列表页
        pass

    @on_response(selector=".article")
    def parse_detail(self, response):
        "color: #5c6370; font-style: italic;"># 解析详情页
        pass

    @pipeline(item=ArticleItem)
    def save_article(self, item):
        "color: #5c6370; font-style: italic;"># 保存文章
        db.articles.insert(item)



2.2 中间件模式

通过中间件可以实现请求/响应的拦截和处理：






Python

class ProxyMiddleware:
    def process_request(self, request):
        request.meta['proxy'] = get_proxy()
        return request

    def process_response(self, response):
        if response.status == 407:
            "color: #5c6370; font-style: italic;"># 代理认证失败，更换代理
            return retry_request(response.request)
        return response



•--
三、关键技术实现

3.1 异步并发控制

OpenClaw 基于 asyncio 实现高效的并发控制：






Python

class RequestScheduler:
    def __init__(self, max_concurrent=10):
        self.semaphore = asyncio.Semaphore(max_concurrent)
        self.queue = asyncio.Queue()

    async def fetch(self, request):
        async with self.semaphore:
            return await self.downloader.download(request)



3.2 分布式支持

通过 Redis 实现分布式任务调度：






Python

class DistributedScheduler:
    def __init__(self, redis_url):
        self.redis = await aioredis.create_redis_pool(redis_url)

    async def schedule(self, spider_name, requests):
        for req in requests:
            await self.redis.lpush(f"queue:{spider_name}", req)



•--
四、性能优化策略

4.1 连接池管理

•HTTP 连接复用：使用 aiohttp 连接池
•数据库连接池：预建立连接，减少创建开销
4.2 缓存策略






Code

┌──────────────┬─────────────┬──────────────┐
│   缓存类型    │   存储介质   │    适用场景   │
├──────────────┼─────────────┼──────────────┤
│ 响应缓存     │ Redis/Memcached │ 重复请求 │
│ DNS 缓存     │    内存      │  域名解析   │
│ 请求去重     │    Redis    │  防止重复   │
└──────────────┴─────────────┴──────────────┘



4.3 增量爬取

通过时间戳或版本号实现增量更新：






Python

async def should_fetch(self, url, last_update):
    remote_time = await self.get_remote_time(url)
    return remote_time > last_update



•--
五、扩展性设计

5.1 插件系统

OpenClaw 支持插件扩展，开发者可以通过以下方式自定义功能：

1.自定义提取器 - 实现 Extractor 接口
2.自定义存储 - 实现 Storage 接口
3.自定义中间件 - 继承 Middleware 基类
5.2 配置驱动

通过 YAML/JSON 配置文件灵活控制爬虫行为：






YAML

spider:
  name: my_spider
  settings:
    concurrent: 10
    retry: 3
    timeout: 30

  pipelines:
    - type: file
      path: ./data.json
    - type: mongodb
      uri: mongodb://localhost:27017



•--
六、最佳实践

6.1 项目结构推荐






Code

my_spider/
├── spiders/
│   ├── __init__.py
│   └── my_spider.py
├── pipelines/
│   ├── __init__.py
│   └── data_pipeline.py
├── middlewares/
│   ├── __init__.py
│   └── proxy_middleware.py
├── settings.yaml
└── main.py



6.2 错误处理建议

1.重试机制：对网络错误自动重试
2.异常隔离：单个请求失败不影响整体
3.日志记录：详细记录爬取过程便于排查
0.--
七、总结

OpenClaw 通过模块化、插件化的架构设计，为爬虫开发提供了灵活且强大的解决方案。其核心优势包括：

•🚀 高性能：基于 asyncio 的异步架构
•🔌 高扩展：插件化的中间件系统
•📦 易使用：简洁的 API 设计
•🌍 分布式：支持大规模分布式爬取
随着互联网数据价值的不断提升，OpenClaw 这类现代化爬虫框架将发挥越来越重要的作用。

•--
•作者：Yuesf

•首发平台：微信公众号
•发布日期：2024年

•--
•如果觉得文章对你有帮助，欢迎关注、点赞、转发！
FILE:test_output_v5.html
OpenClaw 架构拆解：深入解析现代爬虫框架的设计与实现
OpenClaw 架构拆解：深入解析现代爬虫框架的设计与实现

引言

在当今数据驱动的时代，爬虫技术已成为获取互联网数据的重要手段。OpenClaw 作为一款新兴的爬虫框架，以其模块化设计和强大的扩展性受到了广泛关注。本文将深入拆解 OpenClaw 的架构设计，帮助读者理解其核心设计理念。



"优秀的框架不是功能的堆砌，而是对复杂性的优雅封装。" —— OpenClaw 设计哲学

•--
一、整体架构概览

OpenClaw 采用分层架构设计，主要分为以下几个核心层次：

1.1 架构图






Code

┌─────────────────────────────────────────────────────────┐
│                    Application Layer                     │
│  (CLI, API Server, Scheduler)                           │
├─────────────────────────────────────────────────────────┤
│                    Core Engine Layer                     │
│  (Request Manager, Response Handler, Pipeline)          │
├─────────────────────────────────────────────────────────┤
│                    Extractor Layer                       │
│  (CSS Selector, XPath, Regex, JSON Path)                │
├─────────────────────────────────────────────────────────┤
│                    Infrastructure Layer                  │
│  (Storage, Cache, Proxy, Middleware)                    │
└─────────────────────────────────────────────────────────┘



1.2 核心模块说明



模块名称 功能描述 重要程度

Request Manager 请求调度与并发控制 ⭐⭐⭐⭐⭐
Response Handler 响应解析与预处理 ⭐⭐⭐⭐
Extractor 数据提取与清洗 ⭐⭐⭐⭐⭐
Pipeline 数据存储与后处理 ⭐⭐⭐⭐
Middleware 请求/响应拦截扩展 ⭐⭐⭐


•--
二、核心设计模式

2.1 装饰器模式

OpenClaw 大量使用装饰器模式来实现功能的灵活扩展：






Python

@spider(name=#98c379;">"example")
class ExampleSpider:
    @on_request(url=#98c379;">"https://example.com/*")
    def parse_list(self, response):
        # 解析列表页
        pass

    @on_response(selector=#98c379;">".article")
    def parse_detail(self, response):
        # 解析详情页
        pass

    @pipeline(item=ArticleItem)
    def save_article(self, item):
        # 保存文章
        db.articles.insert(item)



2.2 中间件模式

通过中间件可以实现请求/响应的拦截和处理：






Python

class ProxyMiddleware:
    def process_request(self, request):
        request.meta[#98c379;\">'proxy'] = get_proxy()
        return request

    def process_response(self, response):
        if response.status == 407:
            # 代理认证失败，更换代理
            return retry_request(response.request)
        return response



•--
三、关键技术实现

3.1 异步并发控制

OpenClaw 基于 asyncio 实现高效的并发控制：






Python

class RequestScheduler:
    def __init__(self, max_concurrent=10):
        self.semaphore = asyncio.Semaphore(max_concurrent)
        self.queue = asyncio.Queue()

    async def fetch(self, request):
        async with self.semaphore:
            return await self.downloader.download(request)



3.2 分布式支持

通过 Redis 实现分布式任务调度：






Python

class DistributedScheduler:
    def __init__(self, redis_url):
        self.redis = await aioredis.create_redis_pool(redis_url)

    async def schedule(self, spider_name, requests):
        for req in requests:
            await self.redis.lpush(f#98c379;">"queue:{spider_name}", req)



•--
四、性能优化策略

4.1 连接池管理

•HTTP 连接复用：使用 aiohttp 连接池
•数据库连接池：预建立连接，减少创建开销
4.2 缓存策略






Code

┌──────────────┬─────────────┬──────────────┐
│   缓存类型    │   存储介质   │    适用场景   │
├──────────────┼─────────────┼──────────────┤
│ 响应缓存     │ Redis/Memcached │ 重复请求 │
│ DNS 缓存     │    内存      │  域名解析   │
│ 请求去重     │    Redis    │  防止重复   │
└──────────────┴─────────────┴──────────────┘



4.3 增量爬取

通过时间戳或版本号实现增量更新：






Python

async def should_fetch(self, url, last_update):
    remote_time = await self.get_remote_time(url)
    return remote_time > last_update



•--
五、扩展性设计

5.1 插件系统

OpenClaw 支持插件扩展，开发者可以通过以下方式自定义功能：

1.自定义提取器 - 实现 Extractor 接口
2.自定义存储 - 实现 Storage 接口
3.自定义中间件 - 继承 Middleware 基类
5.2 配置驱动

通过 YAML/JSON 配置文件灵活控制爬虫行为：






YAML

spider:
  name: my_spider
  settings:
    concurrent: 10
    retry: 3
    timeout: 30

  pipelines:
    - type: file
      path: ./data.json
    - type: mongodb
      uri: mongodb://localhost:27017



•--
六、最佳实践

6.1 项目结构推荐






Code

my_spider/
├── spiders/
│   ├── __init__.py
│   └── my_spider.py
├── pipelines/
│   ├── __init__.py
│   └── data_pipeline.py
├── middlewares/
│   ├── __init__.py
│   └── proxy_middleware.py
├── settings.yaml
└── main.py



6.2 错误处理建议

1.重试机制：对网络错误自动重试
2.异常隔离：单个请求失败不影响整体
3.日志记录：详细记录爬取过程便于排查
0.--
七、总结

OpenClaw 通过模块化、插件化的架构设计，为爬虫开发提供了灵活且强大的解决方案。其核心优势包括：

•🚀 高性能：基于 asyncio 的异步架构
•🔌 高扩展：插件化的中间件系统
•📦 易使用：简洁的 API 设计
•🌍 分布式：支持大规模分布式爬取
随着互联网数据价值的不断提升，OpenClaw 这类现代化爬虫框架将发挥越来越重要的作用。

•--
•作者：Yuesf

•首发平台：微信公众号
•发布日期：2024年

•--
•如果觉得文章对你有帮助，欢迎关注、点赞、转发！
FILE:test_output_v6.html
OpenClaw 架构拆解：深入解析现代爬虫框架的设计与实现
OpenClaw 架构拆解：深入解析现代爬虫框架的设计与实现

引言

在当今数据驱动的时代，爬虫技术已成为获取互联网数据的重要手段。OpenClaw 作为一款新兴的爬虫框架，以其模块化设计和强大的扩展性受到了广泛关注。本文将深入拆解 OpenClaw 的架构设计，帮助读者理解其核心设计理念。



"优秀的框架不是功能的堆砌，而是对复杂性的优雅封装。" —— OpenClaw 设计哲学

•--
一、整体架构概览

OpenClaw 采用分层架构设计，主要分为以下几个核心层次：

1.1 架构图






Code

┌─────────────────────────────────────────────────────────┐
│                    Application Layer                     │
│  (CLI, API Server, Scheduler)                           │
├─────────────────────────────────────────────────────────┤
│                    Core Engine Layer                     │
│  (Request Manager, Response Handler, Pipeline)          │
├─────────────────────────────────────────────────────────┤
│                    Extractor Layer                       │
│  (CSS Selector, XPath, Regex, JSON Path)                │
├─────────────────────────────────────────────────────────┤
│                    Infrastructure Layer                  │
│  (Storage, Cache, Proxy, Middleware)                    │
└─────────────────────────────────────────────────────────┘



1.2 核心模块说明



模块名称 功能描述 重要程度

Request Manager 请求调度与并发控制 ⭐⭐⭐⭐⭐
Response Handler 响应解析与预处理 ⭐⭐⭐⭐
Extractor 数据提取与清洗 ⭐⭐⭐⭐⭐
Pipeline 数据存储与后处理 ⭐⭐⭐⭐
Middleware 请求/响应拦截扩展 ⭐⭐⭐


•--
二、核心设计模式

2.1 装饰器模式

OpenClaw 大量使用装饰器模式来实现功能的灵活扩展：






Python

@spider(name=#98c379;">"example")</span>
class ExampleSpider:
    @on_request(url=#98c379;">"https://example.com/*")</span>
    def parse_list(self, response):
        # 解析列表页
        pass

    @on_response(selector=#98c379;">".article")</span>
    def parse_detail(self, response):
        # 解析详情页
        pass

    @pipeline(item=ArticleItem)
    def save_article(self, item):
        # 保存文章
        db.articles.insert(item)



2.2 中间件模式

通过中间件可以实现请求/响应的拦截和处理：






Python

class ProxyMiddleware:
    def process_request(self, request):
        request.meta[#98c379;\">'proxy'] = get_proxy()</span>
        return request

    def process_response(self, response):
        if response.status == 407:
            # 代理认证失败，更换代理
            return retry_request(response.request)
        return response



•--
三、关键技术实现

3.1 异步并发控制

OpenClaw 基于 asyncio 实现高效的并发控制：






Python

class RequestScheduler:
    def __init__(self, max_concurrent=10):
        self.semaphore = asyncio.Semaphore(max_concurrent)
        self.queue = asyncio.Queue()

    async def fetch(self, request):
        async with self.semaphore:
            return await self.downloader.download(request)



3.2 分布式支持

通过 Redis 实现分布式任务调度：






Python

class DistributedScheduler:
    def __init__(self, redis_url):
        self.redis = await aioredis.create_redis_pool(redis_url)

    async def schedule(self, spider_name, requests):
        for req in requests:
            await self.redis.lpush(f#98c379;">"queue:{spider_name}", req)</span>



•--
四、性能优化策略

4.1 连接池管理

•HTTP 连接复用：使用 aiohttp 连接池
•数据库连接池：预建立连接，减少创建开销
4.2 缓存策略






Code

┌──────────────┬─────────────┬──────────────┐
│   缓存类型    │   存储介质   │    适用场景   │
├──────────────┼─────────────┼──────────────┤
│ 响应缓存     │ Redis/Memcached │ 重复请求 │
│ DNS 缓存     │    内存      │  域名解析   │
│ 请求去重     │    Redis    │  防止重复   │
└──────────────┴─────────────┴──────────────┘



4.3 增量爬取

通过时间戳或版本号实现增量更新：






Python

async def should_fetch(self, url, last_update):
    remote_time = await self.get_remote_time(url)
    return remote_time > last_update



•--
五、扩展性设计

5.1 插件系统

OpenClaw 支持插件扩展，开发者可以通过以下方式自定义功能：

1.自定义提取器 - 实现 Extractor 接口
2.自定义存储 - 实现 Storage 接口
3.自定义中间件 - 继承 Middleware 基类
5.2 配置驱动

通过 YAML/JSON 配置文件灵活控制爬虫行为：






YAML

spider:
  name: my_spider
  settings:
    concurrent: 10
    retry: 3
    timeout: 30

  pipelines:
    - type: file
      path: ./data.json
    - type: mongodb
      uri: mongodb://localhost:27017



•--
六、最佳实践

6.1 项目结构推荐






Code

my_spider/
├── spiders/
│   ├── __init__.py
│   └── my_spider.py
├── pipelines/
│   ├── __init__.py
│   └── data_pipeline.py
├── middlewares/
│   ├── __init__.py
│   └── proxy_middleware.py
├── settings.yaml
└── main.py



6.2 错误处理建议

1.重试机制：对网络错误自动重试
2.异常隔离：单个请求失败不影响整体
3.日志记录：详细记录爬取过程便于排查
0.--
七、总结

OpenClaw 通过模块化、插件化的架构设计，为爬虫开发提供了灵活且强大的解决方案。其核心优势包括：

•🚀 高性能：基于 asyncio 的异步架构
•🔌 高扩展：插件化的中间件系统
•📦 易使用：简洁的 API 设计
•🌍 分布式：支持大规模分布式爬取
随着互联网数据价值的不断提升，OpenClaw 这类现代化爬虫框架将发挥越来越重要的作用。

•--
•作者：Yuesf

•首发平台：微信公众号
•发布日期：2024年

•--
•如果觉得文章对你有帮助，欢迎关注、点赞、转发！
FILE:test_output_v7.html
OpenClaw 架构拆解：深入解析现代爬虫框架的设计与实现
OpenClaw 架构拆解：深入解析现代爬虫框架的设计与实现

引言

在当今数据驱动的时代，爬虫技术已成为获取互联网数据的重要手段。OpenClaw 作为一款新兴的爬虫框架，以其模块化设计和强大的扩展性受到了广泛关注。本文将深入拆解 OpenClaw 的架构设计，帮助读者理解其核心设计理念。



"优秀的框架不是功能的堆砌，而是对复杂性的优雅封装。" —— OpenClaw 设计哲学

•--
一、整体架构概览

OpenClaw 采用分层架构设计，主要分为以下几个核心层次：

1.1 架构图






Code

┌─────────────────────────────────────────────────────────┐
│                    Application</span> Layer</span>                     │
│  (CLI</span>, API</span> Server</span>, Scheduler</span>)                           │
├─────────────────────────────────────────────────────────┤
│                    Core</span> Engine</span> Layer</span>                     │
│  (Request</span> Manager</span>, Response</span> Handler</span>, Pipeline</span>)          │
├─────────────────────────────────────────────────────────┤
│                    Extractor</span> Layer</span>                       │
│  (CSS</span> Selector</span>, XPath</span>, Regex</span>, JSON</span> Path</span>)                │
├─────────────────────────────────────────────────────────┤
│                    Infrastructure</span> Layer</span>                  │
│  (Storage</span>, Cache</span>, Proxy</span>, Middleware</span>)                    │
└─────────────────────────────────────────────────────────┘



1.2 核心模块说明



模块名称 功能描述 重要程度

Request Manager 请求调度与并发控制 ⭐⭐⭐⭐⭐
Response Handler 响应解析与预处理 ⭐⭐⭐⭐
Extractor 数据提取与清洗 ⭐⭐⭐⭐⭐
Pipeline 数据存储与后处理 ⭐⭐⭐⭐
Middleware 请求/响应拦截扩展 ⭐⭐⭐


•--
二、核心设计模式

2.1 装饰器模式

OpenClaw 大量使用装饰器模式来实现功能的灵活扩展：






Python

@spider</span>(name=#98c379;">"example"</span>)</span>
class</span> ExampleSpider</span>:
    @on_request</span>(url=#98c379;">"https://example.com/*"</span>)</span>
    def</span> parse_list(self, response):
        # 解析列表页</span>
        pass</span>

    @on_response</span>(selector=#98c379;">".article"</span>)</span>
    def</span> parse_detail(self, response):
        # 解析详情页</span>
        pass</span>

    @pipeline</span>(item=ArticleItem</span>)
    def</span> save_article(self, item):
        # 保存文章</span>
        db.articles.insert(item)



2.2 中间件模式

通过中间件可以实现请求/响应的拦截和处理：






Python

class</span> ProxyMiddleware</span>:
    def</span> process_request(self, request):
        request.meta[#98c379;\">'proxy'</span>] = get_proxy()</span>
        return</span> request

    def</span> process_response(self, response):
        if</span> response.status == 407</span>:
            # 代理认证失败，更换代理</span>
            return</span> retry_request(response.request)
        return</span> response



•--
三、关键技术实现

3.1 异步并发控制

OpenClaw 基于 asyncio 实现高效的并发控制：






Python

class</span> RequestScheduler</span>:
    def</span> __init__(self, max_concurrent=10</span>):
        self.semaphore = asyncio.Semaphore</span>(max_concurrent)
        self.queue = asyncio.Queue</span>()

    async</span> def</span> fetch(self, request):
        async</span> with</span> self.semaphore:
            return</span> await</span> self.downloader.download(request)



3.2 分布式支持

通过 Redis 实现分布式任务调度：






Python

class</span> DistributedScheduler</span>:
    def</span> __init__(self, redis_url):
        self.redis = await</span> aioredis.create_redis_pool(redis_url)

    async</span> def</span> schedule(self, spider_name, requests):
        for</span> req in</span> requests:
            await</span> self.redis.lpush(f#98c379;">"queue:{spider_name}"</span>, req)</span>



•--
四、性能优化策略

4.1 连接池管理

•HTTP 连接复用：使用 aiohttp 连接池
•数据库连接池：预建立连接，减少创建开销
4.2 缓存策略






Code

┌──────────────┬─────────────┬──────────────┐
│   缓存类型    │   存储介质   │    适用场景   │
├──────────────┼─────────────┼──────────────┤
│ 响应缓存     │ Redis</span>/Memcached</span> │ 重复请求 │
│ DNS</span> 缓存     │    内存      │  域名解析   │
│ 请求去重     │    Redis</span>    │  防止重复   │
└──────────────┴─────────────┴──────────────┘



4.3 增量爬取

通过时间戳或版本号实现增量更新：






Python

async</span> def</span> should_fetch(self, url, last_update):
    remote_time = await</span> self.get_remote_time(url)
    return</span> remote_time > last_update



•--
五、扩展性设计

5.1 插件系统

OpenClaw 支持插件扩展，开发者可以通过以下方式自定义功能：

1.自定义提取器 - 实现 Extractor 接口
2.自定义存储 - 实现 Storage 接口
3.自定义中间件 - 继承 Middleware 基类
5.2 配置驱动

通过 YAML/JSON 配置文件灵活控制爬虫行为：






YAML

spider:
  name: my_spider
  settings:
    concurrent: 10</span>
    retry: 3</span>
    timeout: 30</span>

  pipelines:
    - type</span>: file
      path: ./data.json
    - type</span>: mongodb
      uri: mongodb://localhost:27017</span>



•--
六、最佳实践

6.1 项目结构推荐






Code

my_spider/
├── spiders/
│   ├── __init__.py
│   └── my_spider.py
├── pipelines/
│   ├── __init__.py
│   └── data_pipeline.py
├── middlewares/
│   ├── __init__.py
│   └── proxy_middleware.py
├── settings.yaml
└── main.py



6.2 错误处理建议

1.重试机制：对网络错误自动重试
2.异常隔离：单个请求失败不影响整体
3.日志记录：详细记录爬取过程便于排查
0.--
七、总结

OpenClaw 通过模块化、插件化的架构设计，为爬虫开发提供了灵活且强大的解决方案。其核心优势包括：

•🚀 高性能：基于 asyncio 的异步架构
•🔌 高扩展：插件化的中间件系统
•📦 易使用：简洁的 API 设计
•🌍 分布式：支持大规模分布式爬取
随着互联网数据价值的不断提升，OpenClaw 这类现代化爬虫框架将发挥越来越重要的作用。

•--
•作者：Yuesf

•首发平台：微信公众号
•发布日期：2024年

•--
•如果觉得文章对你有帮助，欢迎关注、点赞、转发！
FILE:test_output_v9.html
OpenClaw 架构拆解：深入解析现代爬虫框架的设计与实现
OpenClaw 架构拆解：深入解析现代爬虫框架的设计与实现

引言

在当今数据驱动的时代，爬虫技术已成为获取互联网数据的重要手段。OpenClaw 作为一款新兴的爬虫框架，以其模块化设计和强大的扩展性受到了广泛关注。本文将深入拆解 OpenClaw 的架构设计，帮助读者理解其核心设计理念。



"优秀的框架不是功能的堆砌，而是对复杂性的优雅封装。" —— OpenClaw 设计哲学

•--
一、整体架构概览

OpenClaw 采用分层架构设计，主要分为以下几个核心层次：

1.1 架构图






Code

┌─────────────────────────────────────────────────────────┐
│                    Application Layer                     │
│  (CLI, API Server, Scheduler)                           │
├─────────────────────────────────────────────────────────┤
│                    Core Engine Layer                     │
│  (Request Manager, Response Handler, Pipeline)          │
├─────────────────────────────────────────────────────────┤
│                    Extractor Layer                       │
│  (CSS Selector, XPath, Regex, JSON Path)                │
├─────────────────────────────────────────────────────────┤
│                    Infrastructure Layer                  │
│  (Storage, Cache, Proxy, Middleware)                    │
└─────────────────────────────────────────────────────────┘



1.2 核心模块说明



模块名称 功能描述 重要程度

Request Manager 请求调度与并发控制 ⭐⭐⭐⭐⭐
Response Handler 响应解析与预处理 ⭐⭐⭐⭐
Extractor 数据提取与清洗 ⭐⭐⭐⭐⭐
Pipeline 数据存储与后处理 ⭐⭐⭐⭐
Middleware 请求/响应拦截扩展 ⭐⭐⭐


•--
二、核心设计模式

2.1 装饰器模式

OpenClaw 大量使用装饰器模式来实现功能的灵活扩展：






Python

@spider(name="example")
class ExampleSpider:
    @on_request(url="https://example.com/*")
    def parse_list(self, response):
        # 解析列表页
        pass

    @on_response(selector=".article")
    def parse_detail(self, response):
        # 解析详情页
        pass

    @pipeline(item=ArticleItem)
    def save_article(self, item):
        # 保存文章
        db.articles.insert(item)



2.2 中间件模式

通过中间件可以实现请求/响应的拦截和处理：






Python

class ProxyMiddleware:
    def process_request(self, request):
        request.meta['proxy'] = get_proxy()
        return request

    def process_response(self, response):
        if response.status == 407:
            # 代理认证失败，更换代理
            return retry_request(response.request)
        return response



•--
三、关键技术实现

3.1 异步并发控制

OpenClaw 基于 asyncio 实现高效的并发控制：






Python

class RequestScheduler:
    def __init__(self, max_concurrent=10):
        self.semaphore = asyncio.Semaphore(max_concurrent)
        self.queue = asyncio.Queue()

    async def fetch(self, request):
        async with self.semaphore:
            return await self.downloader.download(request)



3.2 分布式支持

通过 Redis 实现分布式任务调度：






Python

class DistributedScheduler:
    def __init__(self, redis_url):
        self.redis = await aioredis.create_redis_pool(redis_url)

    async def schedule(self, spider_name, requests):
        for req in requests:
            await self.redis.lpush(f"queue:{spider_name}", req)



•--
四、性能优化策略

4.1 连接池管理

•HTTP 连接复用：使用 aiohttp 连接池
•数据库连接池：预建立连接，减少创建开销
4.2 缓存策略






Code

┌──────────────┬─────────────┬──────────────┐
│   缓存类型    │   存储介质   │    适用场景   │
├──────────────┼─────────────┼──────────────┤
│ 响应缓存     │ Redis/Memcached │ 重复请求 │
│ DNS 缓存     │    内存      │  域名解析   │
│ 请求去重     │    Redis    │  防止重复   │
└──────────────┴─────────────┴──────────────┘



4.3 增量爬取

通过时间戳或版本号实现增量更新：






Python

async def should_fetch(self, url, last_update):
    remote_time = await self.get_remote_time(url)
    return remote_time > last_update



•--
五、扩展性设计

5.1 插件系统

OpenClaw 支持插件扩展，开发者可以通过以下方式自定义功能：

1.自定义提取器 - 实现 Extractor 接口
2.自定义存储 - 实现 Storage 接口
3.自定义中间件 - 继承 Middleware 基类
5.2 配置驱动

通过 YAML/JSON 配置文件灵活控制爬虫行为：






YAML

spider:
  name: my_spider
  settings:
    concurrent: 10
    retry: 3
    timeout: 30

  pipelines:
    - type: file
      path: ./data.json
    - type: mongodb
      uri: mongodb://localhost:27017



•--
六、最佳实践

6.1 项目结构推荐






Code

my_spider/
├── spiders/
│   ├── __init__.py
│   └── my_spider.py
├── pipelines/
│   ├── __init__.py
│   └── data_pipeline.py
├── middlewares/
│   ├── __init__.py
│   └── proxy_middleware.py
├── settings.yaml
└── main.py



6.2 错误处理建议

1.重试机制：对网络错误自动重试
2.异常隔离：单个请求失败不影响整体
3.日志记录：详细记录爬取过程便于排查
0.--
七、总结

OpenClaw 通过模块化、插件化的架构设计，为爬虫开发提供了灵活且强大的解决方案。其核心优势包括：

•🚀 高性能：基于 asyncio 的异步架构
•🔌 高扩展：插件化的中间件系统
•📦 易使用：简洁的 API 设计
•🌍 分布式：支持大规模分布式爬取
随着互联网数据价值的不断提升，OpenClaw 这类现代化爬虫框架将发挥越来越重要的作用。

•--
•作者：Yuesf

•首发平台：微信公众号
•发布日期：2024年

•--
•如果觉得文章对你有帮助，欢迎关注、点赞、转发！
FILE:test_zhihu_draft.py
"""测试完整的创建草稿和发布流程"""
import asyncio
import os
from multi_writing_skills.platforms import ZhihuPlatform
from multi_writing_skills.platforms.base import PublishRequest

async def test_full_flow():
    cookie = os.environ.get("ZHIHU_COOKIE", "")
    if not cookie:
        print("未找到 ZHIHU_COOKIE")
        return

    platform = ZhihuPlatform(cookie=cookie)

    # 1. 创建草稿
    request = PublishRequest(
        title="自动化测试文章标题",
        content="这是自动化测试的内容",
        digest="这是摘要",
    )

    draft_result = await platform.create_draft(request)
    print(f"创建草稿: {draft_result}")

    if draft_result.success and draft_result.media_id:
        # 2. 发布草稿
        publish_result = await platform.publish_draft(
            draft_id=draft_result.media_id,
            title=request.title,
            content=request.content,
        )
        print(f"发布结果: {publish_result}")

    await platform.close()

if __name__ == "__main__":
    asyncio.run(test_full_flow())
FILE:txt.txt

wechat-publisher
https://github.com/0731coderlee-sudo/wechat-publisher
wenyan-cli
https://github.com/caol64/wenyan-cli
md2wechat-skill
https://github.com/geekjourneyx/md2wechat-skill

模块名称	功能描述	重要程度
Request Manager	请求调度与并发控制	⭐⭐⭐⭐⭐
Response Handler	响应解析与预处理	⭐⭐⭐⭐
Extractor	数据提取与清洗	⭐⭐⭐⭐⭐
Pipeline	数据存储与后处理	⭐⭐⭐⭐
Middleware	请求/响应拦截扩展	⭐⭐⭐

Multi Writing Skills Main

Comments (0)