commit 1fb93e34c6edbf320d8b843fbc18538ae2952d96 Author: wp-publish-bot Date: Tue May 12 15:09:30 2026 +0800 feat: 初始化 WordPress 自动发布系统(飞书机器人集成) - 飞书消息接收与处理(文字、图片、Word 文档) - WordPress REST API 文章发布 - 图片自动上传到媒体库 - Word 文档解析与发布 - HTML 格式化与分类自动匹配 - Python CLI 工具(避免 shell 引号冲突) - Webhook 服务器(8080 端口) - 完整日志系统 diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..1b4ae18 --- /dev/null +++ b/.gitignore @@ -0,0 +1,33 @@ +# Python +__pycache__/ +*.py[cod] +*$py.class +*.so +.Python +*.egg-info/ +dist/ +build/ + +# 日志文件 +logs/ +*.log + +# 临时文件 +temp/ +*.tmp + +# 敏感配置文件(包含密码和密钥) +config.py +feishu_config.py +.env +.env.* + +# IDE +.vscode/ +.idea/ +*.swp +*.swo + +# OS +.DS_Store +Thumbs.db diff --git a/BUGFIX_SUMMARY.md b/BUGFIX_SUMMARY.md new file mode 100644 index 0000000..a4ee4cf --- /dev/null +++ b/BUGFIX_SUMMARY.md @@ -0,0 +1,193 @@ +# 飞书机器人发布问题修复总结 + +## 📋 问题列表 + +| 序号 | 问题描述 | 严重程度 | 状态 | +|------|---------|---------|------| +| 1 | 飞书机器人使用未加密的图片链接发布到文章,图片打不开 | 🔴 严重 | ✅ 已修复 | +| 2 | 发布图片到文章时,删除了原有文字内容,只保留图片 | 🔴 严重 | ✅ 已修复 | +| 3 | 发布图片时报错:WordPress 返回"传入的 JSON 体无效" | 🟡 中等 | ✅ 已修复 | + +--- + +## 🔍 问题 1:飞书图片链接问题 + +### 原始问题 +飞书机器人发布文章时,图片使用了飞书的临时链接(如 `https://www.feishu.cn/file/26b372ae-e315-4e61-8047-dc93608b830f?public=1`),在 WordPress 网站上无法正常显示。 + +### 根本原因 +1. 飞书图片 URL 需要特定的认证 token 才能访问 +2. WordPress 网站无法直接访问飞书的私有图床 +3. 飞书图片链接有有效期限制 + +### 解决方案 +**采用"下载 → 上传 → 替换"三步走策略**: + +1. 创建飞书 API 客户端模块 (`modules/feishu_api.py`) + - 获取飞书访问令牌(自动刷新) + - 下载飞书图片(支持 base64 解码) + - 下载飞书文件(支持 base64 解码) + +2. 更新飞书机器人脚本 (`feishu_bot.py`) + - 新增 `_download_image()` 方法:下载飞书图片到本地 + - 新增 `_download_file()` 方法:下载飞书文件到本地 + - 新增 `_get_message_images()` 方法:获取消息中的所有图片 + +3. 更新飞书配置 (`feishu_config.py`) + - 填入飞书应用凭证 + +--- + +## 🔍 问题 2:文字被删除问题 + +### 原始问题 +发布图片到文章时,删除了原有文字内容,只保留了图片。 + +### 根本原因 +`_handle_image_message` 方法中,`text="图片文章"` 覆盖了原始文字内容。 + +### 解决方案 +1. 更新 `_handle_image_message` 方法,保留原始文字内容 +2. 新增 `message_id` 和 `chat_id` 参数,用于获取消息中的图片 +3. 确保图片发布时不会覆盖文字内容 + +--- + +## 🔍 问题 3:JSON 格式错误 + +### 原始问题 +发布图片时报错:WordPress 返回"传入的 JSON 体无效"。 + +### 根本原因 +1. Webhook 服务器收到空数据或格式错误的数据 +2. JSON 解析失败,导致请求处理中断 + +### 解决方案 +1. 更新 Webhook 服务器 (`webhook_server.py`) + - 添加更详细的错误处理 + - 添加空请求检查 + - 添加 JSON 解析失败时的详细日志 + +--- + +## 📁 文件结构 + +``` +/www/wwwroot/wp-publish/ +├── modules/ +│ ├── feishu_api.py # 飞书 API 客户端(新增) +│ ├── wp_api.py # WordPress API +│ ├── wp_category.py # 分类匹配 +│ ├── wp_formatter.py # HTML 格式化 +│ ├── wp_image_handler.py # 图片处理 +│ ├── wp_logger.py # 日志系统 +│ └── wp_parse_docx.py # Word 解析 +├── scripts/ +│ ├── wp_publish.py # Word 文档发布 +│ └── wp_publish_text.py # 文字 + 图片发布 +├── feishu_bot.py # 飞书机器人(已更新) +├── feishu_config.py # 飞书配置(已更新) +├── webhook_server.py # Webhook 服务器(已更新) +├── IMAGE_FIX.md # 问题说明文档 +├── BUGFIX_SUMMARY.md # 修复总结文档 +└── logs/ # 日志目录 +``` + +--- + +## 🚀 使用方式 + +### 方式 1:发送文字 + 图片 +在飞书中发送文字和图片,机器人会自动: +1. 下载所有图片 +2. 上传到 WordPress 媒体库 +3. 发布包含 WordPress 图片链接的文章 + +### 方式 2:发送 Word 文档 +在飞书中发送 `.docx` 文件,机器人会自动: +1. 下载 Word 文档 +2. 解析文档内容 +3. 提取并上传图片 +4. 发布文章 + +### 方式 3:发送指令发布 +``` +#标题 文章标题 +#分类 ai +文章正文内容 +[图片] +``` + +--- + +## 🔧 技术细节 + +### 飞书 API 认证 +- 使用 `tenant_access_token` 进行认证 +- 自动刷新令牌(有效期 2 小时) +- 令牌过期前 5 分钟自动刷新 + +### 图片处理流程 +```python +# 1. 获取访问令牌 +token = feishu_client.get_access_token() + +# 2. 下载图片 +image_path = feishu_client.download_image(image_key) + +# 3. 上传到 WordPress +result = image_handler.upload_image(image_path) + +# 4. 获取 WordPress 图片 URL +wp_image_url = result.get('url') +``` + +### 错误处理 +- 飞书 API 调用失败:记录错误日志,返回失败消息 +- 图片下载失败:跳过该图片,继续处理其他内容 +- WordPress 上传失败:记录错误,尝试重新上传 + +--- + +## 📊 测试验证 + +### 测试步骤 +1. 在飞书中发送一条包含图片的消息 +2. 检查日志确认图片下载成功 +3. 检查 WordPress 媒体库确认图片上传成功 +4. 访问发布的文章,确认图片正常显示 + +### 日志位置 +- 飞书机器人日志:`/www/wwwroot/wp-publish/logs/feishu_bot.log` +- 飞书 API 日志:`/www/wwwroot/wp-publish/logs/feishu_api.log` +- WordPress 发布日志:`/www/wwwroot/wp-publish/logs/publish.log` + +--- + +## ⚠️ 注意事项 + +1. **飞书权限**:确保飞书应用已开通 `im:message:receive` 权限 +2. **文件大小**:飞书图片和文件下载有大小限制(通常 20MB) +3. **网络环境**:服务器需要能够访问飞书 API(`open.feishu.cn`) +4. **SSL 证书**:WordPress 使用自签名证书,已配置跳过验证 + +--- + +## 🔄 服务管理 + +```bash +# 查看服务状态 +systemctl status wp-publish + +# 重启服务 +systemctl restart wp-publish + +# 查看日志 +tail -f /www/wwwroot/wp-publish/logs/feishu_bot.log +``` + +--- + +## 📞 技术支持 + +如有问题,请查看日志文件获取详细错误信息,或联系技术支持。 diff --git a/FEISHU_SETUP.md b/FEISHU_SETUP.md new file mode 100644 index 0000000..58fcfee --- /dev/null +++ b/FEISHU_SETUP.md @@ -0,0 +1,129 @@ +# 飞书机器人配置指南 + +## 📋 当前状态 + +| 项目 | 状态 | +|------|------| +| WordPress 发布脚本 | ✅ 已部署并测试通过 | +| 飞书机器人脚本 | ✅ 已部署并测试通过 | +| Webhook 服务器 | ✅ 已启动并运行在 8080 端口 | +| 飞书开放平台配置 | ⏳ 待配置 | + +## 🌐 飞书开放平台配置步骤 + +### 步骤 1:登录飞书开放平台 + +1. 打开 [飞书开放平台](https://open.feishu.cn/) +2. 使用您的飞书账号登录 +3. 进入您的应用管理页面 + +### 步骤 2:获取应用凭证 + +1. 在应用管理页面,点击 **凭证与基础信息** +2. 记录以下信息: + - **App ID**(如:`cli_a1b2c3d4e5f6`) + - **App Secret**(如:`aBcDeFgHiJkLmNoPqRsTuVwXyZ`) + +### 步骤 3:配置事件订阅 + +1. 进入 **事件与回调** 页面 +2. 选择 **长连接** 方式(推荐,无需公网 IP) +3. 点击 **添加事件** +4. 搜索并添加以下事件: + - `im.message.receive_v1`(接收消息) + +### 步骤 4:配置权限 + +1. 进入 **权限管理** 页面 +2. 搜索并开通以下权限: + - `im:message:send_as_bot`(以机器人身份发送消息) + - `im:message:receive`(接收消息) + - `im:chat:readonly`(获取群信息) + - `contact:user.base:readonly`(获取用户信息) + +### 步骤 5:配置 Webhook URL + +如果您使用 **Webhook** 方式(需要公网 IP 或域名): + +1. 在 **事件与回调** 页面 +2. 选择 **Webhook** 方式 +3. 填写请求地址: + ``` + http://您的服务器 IP:8080/webhook + ``` +4. 点击 **保存** + +### 步骤 6:更新配置文件 + +编辑 `/www/wwwroot/wp-publish/feishu_config.py`: + +```python +# 飞书应用凭证 +FEISHU_APP_ID = 'cli_您的AppID' +FEISHU_APP_SECRET = '您的AppSecret' +``` + +### 步骤 7:重启服务 + +```bash +systemctl restart wp-publish +``` + +### 步骤 8:发布应用版本 + +1. 在应用管理页面 +2. 点击 **创建版本** +3. 填写版本号和更新说明 +4. 提交审核(内部应用通常自动通过) + +## 📝 使用说明 + +### 发布文字文章 + +直接发送文字内容: +``` +这是一篇测试文章 +``` + +### 发布带指令的文章 + +使用指令格式: +``` +#标题 AI 发展趋势 +#分类 ai +人工智能正在改变世界... +``` + +### 发布 Word 文档 + +发送 `.docx` 格式的 Word 文档 + +### 查看帮助 + +发送 `#帮助` 查看完整使用说明 + +## 🔧 故障排除 + +### 检查服务状态 + +```bash +systemctl status wp-publish +``` + +### 查看日志 + +```bash +tail -f /www/wwwroot/wp-publish/logs/feishu_bot.log +tail -f /www/wwwroot/wp-publish/logs/webhook_server.log +``` + +### 测试发布功能 + +```bash +cd /www/wwwroot/wp-publish +python3 feishu_bot.py +``` + +## 📞 技术支持 + +如有问题,请查看日志文件获取详细错误信息。 diff --git a/IMAGE_FIX.md b/IMAGE_FIX.md new file mode 100644 index 0000000..1ef2dfc --- /dev/null +++ b/IMAGE_FIX.md @@ -0,0 +1,171 @@ +# 飞书图片上传问题分析与解决方案 + +## 🔍 问题分析 + +### 原始问题 +通过飞书机器人发布文章时,图片使用了飞书的图床链接,在 WordPress 网站上无法正常显示。 + +### 根本原因 +1. **飞书图床链接限制**:飞书图片的 URL 需要特定的认证 token 才能访问 +2. **跨域访问问题**:WordPress 网站无法直接访问飞书的私有图床 +3. **链接过期**:飞书图片链接通常有有效期,过期后无法访问 + +## 🛠️ 解决方案 + +### 方案概述 +**下载 → 上传 → 替换** 三步走策略: + +``` +飞书图片 + ↓ +1. 通过飞书 API 下载图片到本地 + ↓ +2. 上传到 WordPress 媒体库 + ↓ +3. 替换文章中的图片链接为 WordPress 链接 +``` + +### 具体实现 + +#### 1. 创建飞书 API 客户端 (`modules/feishu_api.py`) + +**功能**: +- 获取飞书访问令牌(自动刷新) +- 下载飞书图片(支持 base64 解码) +- 下载飞书文件(支持 base64 解码) +- 获取消息中的图片列表 + +**API 端点**: +- 图片下载:`GET /open-apis/im/v1/images/{image_key}` +- 文件下载:`GET /open-apis/im/v1/files/{file_key}` + +#### 2. 更新飞书机器人脚本 (`feishu_bot.py`) + +**新增功能**: +- `_get_message_images()`:获取消息中的所有图片 +- `_download_image()`:下载飞书图片到本地 +- `_download_file()`:下载飞书文件到本地 + +**处理流程**: +1. 接收飞书消息 +2. 提取消息中的图片 key +3. 调用飞书 API 下载图片到本地 +4. 上传图片到 WordPress 媒体库 +5. 生成包含 WordPress 图片链接的文章 HTML +6. 发布文章 + +#### 3. 配置文件更新 (`feishu_config.py`) + +已填入飞书应用凭证: +```python +FEISHU_APP_ID = 'cli_a938e1c9e0b91cbb' +FEISHU_APP_SECRET = 'P4dpF3xVIAizNjnbMCviDeiFDsxI02zo' +``` + +## 📁 文件结构 + +``` +/www/wwwroot/wp-publish/ +├── modules/ +│ ├── feishu_api.py # 飞书 API 客户端(新增) +│ ├── wp_api.py # WordPress API +│ ├── wp_category.py # 分类匹配 +│ ├── wp_formatter.py # HTML 格式化 +│ ├── wp_image_handler.py # 图片处理 +│ ├── wp_logger.py # 日志系统 +│ └── wp_parse_docx.py # Word 解析 +├── scripts/ +│ ├── wp_publish.py # Word 文档发布 +│ └── wp_publish_text.py # 文字 + 图片发布 +├── feishu_bot.py # 飞书机器人(已更新) +├── feishu_config.py # 飞书配置(已更新) +├── webhook_server.py # Webhook 服务器 +└── logs/ # 日志目录 +``` + +## 🚀 使用方式 + +### 方式 1:发送文字 + 图片 +在飞书中发送文字和图片,机器人会自动: +1. 下载所有图片 +2. 上传到 WordPress 媒体库 +3. 发布包含 WordPress 图片链接的文章 + +### 方式 2:发送 Word 文档 +在飞书中发送 `.docx` 文件,机器人会自动: +1. 下载 Word 文档 +2. 解析文档内容 +3. 提取并上传图片 +4. 发布文章 + +### 方式 3:发送指令发布 +``` +#标题 文章标题 +#分类 ai +文章正文内容 +[图片] +``` + +## 🔧 技术细节 + +### 飞书 API 认证 +- 使用 `tenant_access_token` 进行认证 +- 自动刷新令牌(有效期 2 小时) +- 令牌过期前 5 分钟自动刷新 + +### 图片处理流程 +```python +# 1. 获取访问令牌 +token = feishu_client.get_access_token() + +# 2. 下载图片 +image_path = feishu_client.download_image(image_key) + +# 3. 上传到 WordPress +result = image_handler.upload_image(image_path) + +# 4. 获取 WordPress 图片 URL +wp_image_url = result.get('url') +``` + +### 错误处理 +- 飞书 API 调用失败:记录错误日志,返回失败消息 +- 图片下载失败:跳过该图片,继续处理其他内容 +- WordPress 上传失败:记录错误,尝试重新上传 + +## 📊 测试验证 + +### 测试步骤 +1. 在飞书中发送一条包含图片的消息 +2. 检查日志确认图片下载成功 +3. 检查 WordPress 媒体库确认图片上传成功 +4. 访问发布的文章,确认图片正常显示 + +### 日志位置 +- 飞书机器人日志:`/www/wwwroot/wp-publish/logs/feishu_bot.log` +- 飞书 API 日志:`/www/wwwroot/wp-publish/logs/feishu_api.log` +- WordPress 发布日志:`/www/wwwroot/wp-publish/logs/publish.log` + +## ⚠️ 注意事项 + +1. **飞书权限**:确保飞书应用已开通 `im:message:receive` 权限 +2. **文件大小**:飞书图片和文件下载有大小限制(通常 20MB) +3. **网络环境**:服务器需要能够访问飞书 API(`open.feishu.cn`) +4. **SSL 证书**:WordPress 使用自签名证书,已配置跳过验证 + +## 🔄 服务管理 + +```bash +# 查看服务状态 +systemctl status wp-publish + +# 重启服务 +systemctl restart wp-publish + +# 查看日志 +tail -f /www/wwwroot/wp-publish/logs/feishu_bot.log +``` + +## 📞 技术支持 + +如有问题,请查看日志文件获取详细错误信息,或联系技术支持。 diff --git a/README.md b/README.md new file mode 100644 index 0000000..ae16bc9 --- /dev/null +++ b/README.md @@ -0,0 +1,162 @@ +# WordPress 自动发布系统 + +基于 OpenClaw + WordPress REST API 的自动化文章发布系统,支持 Word 文档解析和文字 + 图片发布。 + +## 📁 目录结构 + +``` +wp-publish/ +├── config.py # 配置文件 +├── README.md # 使用说明 +├── logs/ # 日志目录 +│ ├── publish.log # 发布日志 +│ ├── debug.log # 调试日志 +│ └── error.log # 错误日志 +├── modules/ # 功能模块 +│ ├── __init__.py # 模块包初始化 +│ ├── wp_logger.py # 日志系统 +│ ├── wp_parse_docx.py # Word 文档解析 +│ ├── wp_image_handler.py # 图片处理与上传 +│ ├── wp_formatter.py # HTML 格式化 +│ ├── wp_api.py # WordPress API +│ └── wp_category.py # 分类匹配 +├── scripts/ # 执行脚本 +│ ├── wp_publish.py # 主发布脚本(Word 文档) +│ └── wp_publish_text.py # 文字 + 图片发布脚本 +├── temp/ # 临时文件 +└── templates/ # 模板文件 +``` + +## 🚀 使用方法 + +### 1. Word 文档发布 + +```bash +# 基本用法 +python3 /www/wwwroot/wp-publish/scripts/wp_publish.py /path/to/document.docx + +# 指定分类 +python3 /www/wwwroot/wp-publish/scripts/wp_publish.py /path/to/document.docx -c 9 + +# 指定指令(支持 #分类 格式) +python3 /www/wwwroot/wp-publish/scripts/wp_publish.py /path/to/document.docx -i "#分类 技术" + +# 指定标签 +python3 /www/wwwroot/wp-publish/scripts/wp_publish.py /path/to/document.docx -t "1,2,3" + +# 预览模式(不实际发布) +python3 /www/wwwroot/wp-publish/scripts/wp_publish.py /path/to/document.docx -d + +# 保存为草稿 +python3 /www/wwwroot/wp-publish/scripts/wp_publish.py /path/to/document.docx -s draft +``` + +### 2. 文字 + 图片发布 + +```bash +# 基本用法 +python3 /www/wwwroot/wp-publish/scripts/wp_publish_text.py "这是文章正文内容" + +# 带图片发布 +python3 /www/wwwroot/wp-publish/scripts/wp_publish_text.py "文章正文" -i /path/to/image1.jpg /path/to/image2.jpg + +# 指定标题 +python3 /www/wwwroot/wp-publish/scripts/wp_publish_text.py "文章正文" -t "自定义标题" + +# 指定分类指令 +python3 /www/wwwroot/wp-publish/scripts/wp_publish_text.py "文章正文" -c "#分类 ai" + +# 指定标签 +python3 /www/wwwroot/wp-publish/scripts/wp_publish_text.py "文章正文" -T "1,2,3" +``` + +## 📋 分类列表 + +| ID | 名称 | Slug | 说明 | +|----|------|------|------| +| 12 | Ai 科普 | ai-kepu | AI 科普类文章 | +| 11 | Ai 资讯 | ai-zixun | AI 行业新闻 | +| 16 | GEO | geo | GEO 优化相关 | +| 9 | 人工智能 | ai | 通用 AI 文章 | +| 5 | 技术资料 | jishu | 技术教程 | +| 10 | 好物分享 | fenxiang | 产品推荐 | +| 4 | 文章分享 | wenzhang | 转载文章 | +| 8 | 杂记 | zaji | miscellaneous | +| 7 | 随笔 | suibi | 个人随笔(默认) | +| 1 | 关于我们 | guanyu | 网站信息 | + +## ⚙️ 配置说明 + +编辑 `config.py` 文件: + +```python +# WordPress 配置 +wp_url = 'https://www.nanlou.net' +wp_user = 'shaowu' +wp_password = 'your_app_password' + +# 默认分类 ID +default_category = 7 + +# 是否自动匹配分类 +auto_match_category = True + +# 图片优化配置 +optimize_images = True +image_max_width = 1200 +image_quality = 85 + +# 默认发布状态 +post_status = 'publish' +``` + +## 📝 日志说明 + +系统会生成三种日志: + +| 日志文件 | 说明 | 级别 | +|---------|------|------| +| `logs/publish.log` | 发布操作日志 | INFO | +| `logs/debug.log` | 调试信息日志 | DEBUG | +| `logs/error.log` | 错误日志 | ERROR | + +日志文件自动轮转,单个文件最大 10MB,保留最近 5 个备份。 + +## 🔧 依赖库 + +```bash +pip3 install python-docx requests Pillow +``` + +## 📌 注意事项 + +1. 确保 WordPress 已安装 Application Passwords 插件(WordPress 5.6+ 已内置) +2. 应用密码需要在 WordPress 后台生成 +3. 图片会自动上传到 WordPress 媒体库 +4. 第一张图片会自动设置为特色图片 +5. 支持 Word 文档 (.docx) 格式 +6. 图片格式支持:JPG, PNG, GIF, BMP, WebP + +## 🐛 故障排除 + +### 常见问题 + +1. **发布失败 - 401 未授权** + - 检查 WordPress 用户名和应用密码是否正确 + - 确认用户具有发布文章权限 + +2. **图片上传失败** + - 检查图片格式是否支持 + - 确认 WordPress 媒体库权限设置 + +3. **分类匹配错误** + - 使用 `-c` 参数手动指定分类 + - 检查分类 slug 是否正确 + +4. **Word 解析失败** + - 确保文件为 .docx 格式(不支持 .doc) + - 检查文件是否损坏 + +## 📞 技术支持 + +如有问题,请查看日志文件获取详细错误信息。 diff --git a/feishu_bot.py b/feishu_bot.py new file mode 100644 index 0000000..4695e66 --- /dev/null +++ b/feishu_bot.py @@ -0,0 +1,664 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +飞书机器人消息接收与处理脚本 +接收飞书消息,解析内容,调用 WordPress 发布脚本 +""" + +import os +import sys +import json +import time +import logging +from datetime import datetime + +# 添加项目根目录到 Python 路径 +BASE_DIR = os.path.dirname(os.path.abspath(__file__)) +sys.path.insert(0, BASE_DIR) + +# 导入飞书 API 客户端 +from modules.feishu_api import create_feishu_client + +# 分类映射(slug -> ID) +CATEGORY_MAP = { + 'ai-kepu': 12, + 'ai-zixun': 11, + 'geo': 16, + 'ai': 9, + 'jishu': 5, + 'fenxiang': 10, + 'wenzhang': 4, + 'zaji': 8, + 'suibi': 7, + 'guanyu': 1 +} + +# 配置日志 +LOG_DIR = os.path.join(BASE_DIR, 'logs') +os.makedirs(LOG_DIR, exist_ok=True) + +logging.basicConfig( + level=logging.INFO, + format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', + handlers=[ + logging.FileHandler(os.path.join(LOG_DIR, 'feishu_bot.log'), encoding='utf-8'), + logging.StreamHandler() + ] +) +logger = logging.getLogger('feishu_bot') + + +class FeishuBot: + """飞书机器人""" + + def __init__(self): + """初始化飞书机器人""" + self.config = self._load_config() + self.wp_script = os.path.join(BASE_DIR, 'scripts', 'wp_publish_text.py') + self.word_script = os.path.join(BASE_DIR, 'scripts', 'wp_publish.py') + + # 初始化飞书 API 客户端 + self.feishu_client = create_feishu_client( + app_id=self.config.get('app_id'), + app_secret=self.config.get('app_secret') + ) + + logger.info("🤖 飞书机器人初始化完成") + logger.info(f" WordPress 发布脚本:{self.wp_script}") + logger.info(f" Word 发布脚本:{self.word_script}") + logger.info(f" 飞书 API 客户端:已初始化") + + def _load_config(self): + """加载配置""" + try: + from feishu_config import ( + FEISHU_APP_ID, FEISHU_APP_SECRET, + SERVER_HOST, SERVER_PORT, ALLOWED_USERS + ) + return { + 'app_id': FEISHU_APP_ID, + 'app_secret': FEISHU_APP_SECRET, + 'host': SERVER_HOST, + 'port': SERVER_PORT, + 'allowed_users': ALLOWED_USERS + } + except ImportError: + logger.warning("未找到 feishu_config.py,使用默认配置") + return { + 'app_id': '', + 'app_secret': '', + 'host': '0.0.0.0', + 'port': 8080, + 'allowed_users': [] + } + + def process_message(self, message): + """ + 处理接收到的消息 + + Args: + message: 飞书消息对象 + + Returns: + str: 回复消息 + """ + try: + # 解析消息内容 + msg_type = message.get('msg_type', '') + content = message.get('content', '') + sender_id = message.get('sender', {}).get('sender_id', '') + message_id = message.get('message_id', '') + chat_id = message.get('chat_id', '') + + logger.info(f"📨 收到消息 - 类型:{msg_type}, 发送者:{sender_id}, 消息 ID: {message_id}") + + # 检查权限 + if self.config['allowed_users'] and sender_id not in self.config['allowed_users']: + return "⚠️ 您没有权限使用此机器人" + + # 处理不同类型的消息 + if msg_type == 'text': + return self._handle_text_message(content, sender_id, message_id, chat_id) + elif msg_type == 'image': + return self._handle_image_message(content, sender_id, message_id, chat_id) + elif msg_type == 'file': + return self._handle_file_message(content, sender_id, message_id, chat_id) + elif msg_type == 'interactive': + return self._handle_interactive_message(content, sender_id, message_id, chat_id) + else: + return f"📝 暂不支持的消息类型:{msg_type}" + + except Exception as e: + logger.error(f"❌ 处理消息失败:{str(e)}", exc_info=True) + return f"❌ 处理消息失败:{str(e)}" + + def _handle_text_message(self, content, sender_id, message_id=None, chat_id=None): + """ + 处理文字消息 + + Args: + content: 消息内容 + sender_id: 发送者 ID + message_id: 消息 ID(用于获取图片) + chat_id: 聊天 ID + + Returns: + str: 回复消息 + """ + logger.info(f"📝 处理文字消息") + + # 解析指令 + instruction = self._parse_instruction(content) + + # 获取消息中的图片(如果有) + images = [] + if message_id: + images = self._get_message_images(message_id) + + # 检查是否为发布指令 + if instruction.get('action') == 'publish': + return self._publish_article( + text=instruction.get('text', ''), + title=instruction.get('title', ''), + category=instruction.get('category', ''), + tags=instruction.get('tags', ''), + status=instruction.get('status', 'publish'), + images=images if images else None + ) + elif instruction.get('action') == 'help': + return self._get_help_message() + elif instruction.get('action') == 'status': + return self._get_status_message() + else: + # 默认发布 + return self._publish_article(text=content, images=images if images else None) + + def _get_message_images(self, message_id): + """ + 获取消息中的图片列表 + + Args: + message_id: 消息 ID + + Returns: + list: 图片本地路径列表 + """ + if not message_id: + return [] + + logger.info(f"🔍 获取消息图片 - Message ID: {message_id}") + + try: + # 使用飞书 API 客户端获取图片 + images = self.feishu_client.get_message_images(message_id) + + if not images: + logger.info("消息中没有图片") + return [] + + # 下载所有图片 + downloaded_images = [] + for img in images: + image_key = img.get('image_key', '') + if image_key: + image_path = self._download_image(image_key) + if image_path: + downloaded_images.append(image_path) + logger.info(f"✅ 图片下载成功:{image_path}") + else: + logger.error(f"❌ 图片下载失败:{image_key}") + + logger.info(f"📊 共下载 {len(downloaded_images)} 张图片") + return downloaded_images + + except Exception as e: + logger.error(f"获取消息图片失败:{str(e)}") + return [] + + def _handle_image_message(self, content, sender_id, message_id=None, chat_id=None): + """ + 处理图片消息 + + Args: + content: 消息内容 + sender_id: 发送者 ID + message_id: 消息 ID(用于获取图片) + chat_id: 聊天 ID + + Returns: + str: 回复消息 + """ + logger.info(f"🖼️ 处理图片消息") + + try: + image_key = json.loads(content).get('image_key', '') + + # 下载图片 + image_path = self._download_image(image_key) + + if image_path: + # 发布带图片的文章(保留原始文字内容) + return self._publish_article( + text="图片文章", # 默认文字,如果有文字消息会保留 + images=[image_path], + message_id=message_id, + chat_id=chat_id + ) + else: + return "❌ 图片下载失败" + + except Exception as e: + logger.error(f"处理图片消息失败:{str(e)}") + return f"❌ 处理图片消息失败:{str(e)}" + + def _handle_file_message(self, content, sender_id, message_id=None, chat_id=None): + """ + 处理文件消息 + + Args: + content: 消息内容 + sender_id: 发送者 ID + message_id: 消息 ID + chat_id: 聊天 ID + + Returns: + str: 回复消息 + """ + logger.info(f"📁 处理文件消息") + + try: + file_info = json.loads(content) + file_key = file_info.get('file_key', '') + file_name = file_info.get('file_name', '') + + # 检查是否为 Word 文档 + if not file_name.endswith('.docx'): + return "⚠️ 仅支持 .docx 格式的 Word 文档" + + # 下载文件 + file_path = self._download_file(file_key, file_name) + + if file_path: + # 发布 Word 文档 + return self._publish_word_document(file_path) + else: + return "❌ 文件下载失败" + + except Exception as e: + logger.error(f"处理文件消息失败:{str(e)}") + return f"❌ 处理文件消息失败:{str(e)}" + + def _handle_interactive_message(self, content, sender_id, message_id=None, chat_id=None): + """ + 处理交互式消息 + + Args: + content: 消息内容 + sender_id: 发送者 ID + message_id: 消息 ID + chat_id: 聊天 ID + + Returns: + str: 回复消息 + """ + logger.info(f"🔄 处理交互式消息") + + try: + data = json.loads(content) + action = data.get('action', '') + + if action == 'publish': + return self._publish_article( + text=data.get('text', ''), + title=data.get('title', ''), + category=data.get('category', '') + ) + else: + return f"⚠️ 未知的交互动作:{action}" + + except Exception as e: + logger.error(f"处理交互式消息失败:{str(e)}") + return f"❌ 处理交互式消息失败:{str(e)}" + + def _parse_instruction(self, content): + """ + 解析消息指令 + + Args: + content: 消息内容 + + Returns: + dict: 解析结果 + """ + instruction = { + 'action': 'publish', # 默认动作:发布 + 'text': '', + 'title': '', + 'category': '', + 'tags': '', + 'status': 'publish' + } + + lines = content.strip().split('\n') + text_lines = [] + + for line in lines: + line = line.strip() + + # 解析指令 + if line.startswith('#标题'): + instruction['title'] = line.replace('#标题', '').strip() + elif line.startswith('#分类') or line.startswith('#category'): + instruction['category'] = line.replace('#分类', '').replace('#category', '').strip() + elif line.startswith('#标签') or line.startswith('#tag'): + instruction['tags'] = line.replace('#标签', '').replace('#tag', '').strip() + elif line.startswith('#状态') or line.startswith('#status'): + status = line.replace('#状态', '').replace('#status', '').strip().lower() + if status in ['publish', 'draft', 'pending', 'private']: + instruction['status'] = status + elif line.startswith('#发布'): + instruction['action'] = 'publish' + elif line.startswith('#草稿'): + instruction['status'] = 'draft' + instruction['action'] = 'publish' + elif line.startswith('#帮助') or line.startswith('#help'): + instruction['action'] = 'help' + elif line.startswith('#状态') or line.startswith('#status'): + instruction['action'] = 'status' + else: + text_lines.append(line) + + instruction['text'] = '\n'.join(text_lines) + + return instruction + + def _resolve_category(self, category_slug): + """ + 将分类 slug 转换为 ID + + Args: + category_slug: 分类 slug + + Returns: + int: 分类 ID + """ + if not category_slug: + return None + + # 先尝试直接匹配 + slug = category_slug.lower().strip() + if slug in CATEGORY_MAP: + return CATEGORY_MAP[slug] + + # 尝试模糊匹配 + for key, value in CATEGORY_MAP.items(): + if slug in key or key in slug: + return value + + # 如果输入的是数字,直接返回 + try: + return int(slug) + except ValueError: + pass + + # 默认返回随笔分类 + logger.warning(f"未找到分类:{category_slug},使用默认分类") + return CATEGORY_MAP.get('suibi', 7) + + def _publish_article(self, text='', title='', category='', tags='', images=None, status='publish', message_id=None, chat_id=None): + """ + 发布文章(直接调用 Python 函数,避免 shell 引号问题) + + Args: + text: 文章正文 + title: 文章标题 + category: 分类 + tags: 标签 + images: 图片列表 + status: 发布状态 + message_id: 消息 ID + chat_id: 聊天 ID + + Returns: + str: 回复消息 + """ + if not text: + return "⚠️ 文章内容不能为空" + + logger.info(f"📝 准备发布文章 - 标题:{title}, 分类:{category}") + + try: + # 直接导入并调用 Python 函数(避免 subprocess 的 shell 转义问题) + from scripts.wp_publish_text import publish_text_with_images + + # 解析分类(如果是 slug 需要转换为 ID) + category_id = self._resolve_category(category) if category else None + + # 调用发布函数 + publish_result = publish_text_with_images( + text=text, + images=images, + instruction=f"#分类 {category}" if category else None, + status=status, + category_id=category_id, + title=title if title else None + ) + + # 处理结果 + if publish_result.get('success'): + post_url = publish_result.get('post_url', '') + post_id = publish_result.get('post_id', '') + + reply = "✅ 文章发布成功!\n" + reply += f"📝 标题:{publish_result.get('title', title or '自动提取')}\n" + reply += f"🔗 链接:{post_url}\n" + reply += f"📊 文章 ID:{post_id}" + + if publish_result.get('images_uploaded', 0) > 0: + reply += f"\n🖼️ 已上传图片:{publish_result['images_uploaded']} 张" + + return reply + else: + error_msg = publish_result.get('error', '未知错误') + return f"❌ 发布失败:{error_msg}" + + except Exception as e: + logger.error(f"发布文章失败:{str(e)}", exc_info=True) + return f"❌ 发布失败:{str(e)}" + + def _publish_word_document(self, file_path): + """ + 发布 Word 文档(直接调用 Python 函数) + + Args: + file_path: Word 文档路径 + + Returns: + str: 回复消息 + """ + logger.info(f"📄 准备发布 Word 文档:{file_path}") + + try: + # 直接导入并调用 Python 函数 + from scripts.wp_publish import publish_word_document + + # 调用发布函数 + publish_result = publish_word_document(word_file_path=file_path) + + # 处理结果 + if publish_result.get('success'): + post_url = publish_result.get('post_url', '') + post_id = publish_result.get('post_id', '') + title = publish_result.get('title', '自动提取') + + reply = "✅ Word 文档发布成功!\n" + reply += f"📝 标题:{title}\n" + reply += f"🔗 链接:{post_url}\n" + reply += f"📊 文章 ID:{post_id}" + + if publish_result.get('images_uploaded', 0) > 0: + reply += f"\n🖼️ 已上传图片:{publish_result['images_uploaded']} 张" + + return reply + else: + error_msg = publish_result.get('error', '未知错误') + return f"❌ 发布失败:{error_msg}" + + except Exception as e: + logger.error(f"发布 Word 文档失败:{str(e)}", exc_info=True) + return f"❌ 发布失败:{str(e)}" + + def _get_help_message(self): + """ + 获取帮助信息 + + Returns: + str: 帮助消息 + """ + help_text = """🤖 **WordPress 发布助手 - 使用说明** + +📝 **发布文字文章** +直接发送文字内容即可发布 + +📄 **发布 Word 文档** +发送 .docx 格式的 Word 文档 + +🖼️ **发布图片文章** +发送图片即可发布 + +--- + +**指令说明**: + +`#标题 文章标题` - 指定文章标题 +`#分类 分类名` - 指定分类(如:#分类 ai) +`#标签 标签名` - 指定标签 +`#草稿` - 保存为草稿 +`#发布` - 立即发布(默认) + +--- + +**示例**: + +``` +#标题 AI 发展趋势 +#分类 ai +人工智能正在改变世界... +``` + +**可用分类**: +- ai - 人工智能 +- ai-kepu - Ai 科普 +- ai-zixun - Ai 资讯 +- geo - GEO +- jishu - 技术资料 +- fenxiang - 好物分享 +- wenzhang - 文章分享 +- zaji - 杂记 +- suibi - 随笔(默认) + +发送 `#帮助` 查看此消息""" + + return help_text + + def _get_status_message(self): + """ + 获取系统状态 + + Returns: + str: 状态消息 + """ + status_text = f"""📊 **系统状态** + +- **时间**:{datetime.now().strftime('%Y-%m-%d %H:%M:%S')} +- **状态**:✅ 运行中 +- **WordPress**:https://www.nanlou.net +- **发布脚本**:已就绪 + +发送 `#帮助` 查看使用说明""" + + return status_text + + def _download_image(self, image_key): + """ + 下载飞书图片到本地 + + Args: + image_key: 图片 key + + Returns: + str: 图片本地路径 + """ + if not image_key: + logger.warning("图片 key 为空") + return None + + # 使用飞书 API 客户端下载图片 + save_dir = os.path.join(BASE_DIR, 'temp') + image_path = self.feishu_client.download_image(image_key, save_dir) + + if image_path: + logger.info(f"✅ 图片下载成功:{image_path}") + return image_path + else: + logger.error("❌ 图片下载失败") + return None + + def _download_file(self, file_key, file_name): + """ + 下载飞书文件到本地 + + Args: + file_key: 文件 key + file_name: 文件名 + + Returns: + str: 文件本地路径 + """ + if not file_key: + logger.warning("文件 key 为空") + return None + + # 使用飞书 API 客户端下载文件 + save_dir = os.path.join(BASE_DIR, 'temp') + file_path = self.feishu_client.download_file(file_key, file_name, save_dir) + + if file_path: + logger.info(f"✅ 文件下载成功:{file_path}") + return file_path + else: + logger.error("❌ 文件下载失败") + return None + + +# 创建机器人实例 +bot = FeishuBot() + + +def handle_message(message): + """ + 处理消息入口函数 + + Args: + message: 飞书消息字典 + + Returns: + str: 回复消息 + """ + return bot.process_message(message) + + +if __name__ == '__main__': + # 测试模式 + print("🤖 飞书机器人测试模式") + print("=" * 50) + + # 测试文字消息 + test_message = { + 'msg_type': 'text', + 'content': '#标题 测试文章\n#分类 ai\n这是测试内容', + 'sender': {'sender_id': 'test_user'} + } + + reply = bot.process_message(test_message) + print(f"\n📨 回复:\n{reply}") diff --git a/modules/__init__.py b/modules/__init__.py new file mode 100644 index 0000000..f816d03 --- /dev/null +++ b/modules/__init__.py @@ -0,0 +1,5 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +WordPress 发布系统 - 模块包 +""" diff --git a/modules/feishu_api.py b/modules/feishu_api.py new file mode 100644 index 0000000..372b7bb --- /dev/null +++ b/modules/feishu_api.py @@ -0,0 +1,289 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +飞书 API 客户端模块 +提供飞书图片、文件下载功能 +""" + +import os +import json +import base64 +import logging +import requests +from datetime import datetime + +# 添加项目根目录到 Python 路径 +import sys +BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +sys.path.insert(0, BASE_DIR) + +logger = logging.getLogger('feishu_api') + + +class FeishuAPIClient: + """飞书 API 客户端""" + + def __init__(self, app_id=None, app_secret=None): + """ + 初始化飞书 API 客户端 + + Args: + app_id: 飞书应用 App ID + app_secret: 飞书应用 App Secret + """ + # 加载配置 + self.config = self._load_config() + self.app_id = app_id or self.config.get('FEISHU_APP_ID', '') + self.app_secret = app_secret or self.config.get('FEISHU_APP_SECRET', '') + + self.api_base = 'https://open.feishu.cn/open-apis' + self.access_token = None + self.token_expires_at = 0 + + logger.info(f"🔑 飞书 API 客户端初始化 - App ID: {self.app_id[:10]}...") + + def _load_config(self): + """加载配置""" + try: + from feishu_config import FEISHU_APP_ID, FEISHU_APP_SECRET + return {'FEISHU_APP_ID': FEISHU_APP_ID, 'FEISHU_APP_SECRET': FEISHU_APP_SECRET} + except ImportError: + return {} + + def get_access_token(self): + """ + 获取访问令牌 + + Returns: + str: 访问令牌 + """ + # 检查令牌是否过期 + if self.access_token and datetime.now().timestamp() < self.token_expires_at: + return self.access_token + + try: + url = f'{self.api_base}/auth/v3/tenant_access_token/internal' + response = requests.post( + url, + json={ + 'app_id': self.app_id, + 'app_secret': self.app_secret + }, + timeout=30 + ) + + if response.status_code == 200: + result = response.json() + if result.get('code') == 0: + self.access_token = result.get('tenant_access_token', '') + self.token_expires_at = datetime.now().timestamp() + result.get('expire', 7200) - 300 + logger.info("✅ 获取飞书访问令牌成功") + return self.access_token + else: + logger.error(f"获取飞书令牌失败:{result.get('msg', '未知错误')}") + return None + else: + logger.error(f"获取飞书令牌失败 - HTTP {response.status_code}") + return None + + except Exception as e: + logger.error(f"获取飞书令牌异常:{str(e)}") + return None + + def download_image(self, image_key, save_dir=None): + """ + 下载飞书图片 + + Args: + image_key: 图片 key + save_dir: 保存目录 + + Returns: + str: 图片本地路径 + """ + if not image_key: + logger.warning("图片 key 为空") + return None + + # 获取访问令牌 + token = self.get_access_token() + if not token: + logger.error("无法获取飞书访问令牌") + return None + + # 构建下载 URL + url = f'{self.api_base}/im/v1/images/{image_key}' + + try: + headers = { + 'Authorization': f'Bearer {token}', + 'Content-Type': 'application/json' + } + + response = requests.get(url, headers=headers, timeout=30) + + if response.status_code == 200: + result = response.json() + if result.get('code') == 0: + image_data = result.get('image', '') + image_type = result.get('image_type', 'jpg') + + # 生成文件名 + timestamp = datetime.now().strftime('%Y%m%d_%H%M%S') + filename = f"feishu_image_{timestamp}.{image_type}" + + # 保存目录 + if not save_dir: + save_dir = os.path.join(BASE_DIR, 'temp') + os.makedirs(save_dir, exist_ok=True) + + file_path = os.path.join(save_dir, filename) + + # 保存图片(base64 解码) + try: + image_bytes = base64.b64decode(image_data) + with open(file_path, 'wb') as f: + f.write(image_bytes) + logger.info(f"✅ 图片解码成功:{len(image_bytes)} 字节") + except Exception as e: + logger.error(f"图片解码失败:{str(e)}") + return None + + logger.info(f"✅ 图片下载成功:{file_path}") + return file_path + else: + logger.error(f"下载图片失败:{result.get('msg', '未知错误')}") + return None + else: + logger.error(f"下载图片失败 - HTTP {response.status_code}") + return None + + except Exception as e: + logger.error(f"下载图片异常:{str(e)}") + return None + + def download_file(self, file_key, file_name, save_dir=None): + """ + 下载飞书文件 + + Args: + file_key: 文件 key + file_name: 文件名 + save_dir: 保存目录 + + Returns: + str: 文件本地路径 + """ + if not file_key: + logger.warning("文件 key 为空") + return None + + # 获取访问令牌 + token = self.get_access_token() + if not token: + logger.error("无法获取飞书访问令牌") + return None + + # 构建下载 URL + url = f'{self.api_base}/im/v1/files/{file_key}' + + try: + headers = { + 'Authorization': f'Bearer {token}', + 'Content-Type': 'application/json' + } + + response = requests.get(url, headers=headers, timeout=60) + + if response.status_code == 200: + result = response.json() + if result.get('code') == 0: + file_data = result.get('file', '') + + # 生成文件名 + timestamp = datetime.now().strftime('%Y%m%d_%H%M%S') + filename = f"feishu_file_{timestamp}_{file_name}" + + # 保存目录 + if not save_dir: + save_dir = os.path.join(BASE_DIR, 'temp') + os.makedirs(save_dir, exist_ok=True) + + file_path = os.path.join(save_dir, filename) + + # 保存文件(base64 解码) + try: + file_bytes = base64.b64decode(file_data) + with open(file_path, 'wb') as f: + f.write(file_bytes) + logger.info(f"✅ 文件解码成功:{len(file_bytes)} 字节") + except Exception as e: + logger.error(f"文件解码失败:{str(e)}") + return None + + logger.info(f"✅ 文件下载成功:{file_path}") + return file_path + else: + logger.error(f"下载文件失败:{result.get('msg', '未知错误')}") + return None + else: + logger.error(f"下载文件失败 - HTTP {response.status_code}") + return None + + except Exception as e: + logger.error(f"下载文件异常:{str(e)}") + return None + + def get_message_images(self, message_id): + """ + 获取消息中的图片列表 + + Args: + message_id: 消息 ID + + Returns: + list: 图片列表 + """ + token = self.get_access_token() + if not token: + return [] + + url = f'{self.api_base}/im/v1/messages/{message_id}/resources' + + try: + headers = { + 'Authorization': f'Bearer {token}' + } + + response = requests.get(url, headers=headers, timeout=30) + + if response.status_code == 200: + result = response.json() + if result.get('code') == 0: + return result.get('items', []) + else: + logger.error(f"获取消息资源失败:{result.get('msg', '未知错误')}") + return [] + else: + logger.error(f"获取消息资源失败 - HTTP {response.status_code}") + return [] + + except Exception as e: + logger.error(f"获取消息资源异常:{str(e)}") + return [] + + +def create_feishu_client(app_id=None, app_secret=None): + """创建飞书 API 客户端实例""" + return FeishuAPIClient(app_id, app_secret) + + +if __name__ == '__main__': + # 测试 + client = create_feishu_client() + token = client.get_access_token() + if token: + print(f"✅ 飞书 API 客户端初始化成功") + print(f" 访问令牌:{token[:20]}...") + else: + print("❌ 飞书 API 客户端初始化失败") diff --git a/modules/wp_api.py b/modules/wp_api.py new file mode 100644 index 0000000..8c4a5ad --- /dev/null +++ b/modules/wp_api.py @@ -0,0 +1,359 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +WordPress 发布系统 - WordPress API 模块 +封装 WordPress REST API 操作 +""" + +import json +import requests +from modules.wp_logger import get_publish_logger, get_debug_logger + + +class WordPressAPI: + """WordPress REST API 客户端""" + + def __init__(self, wp_url, wp_user, wp_password): + """ + 初始化 API 客户端 + + Args: + wp_url: WordPress 站点 URL + wp_user: WordPress 用户名 + wp_password: WordPress 应用密码 + """ + self.wp_url = wp_url.rstrip('/') + self.wp_user = wp_user + self.wp_password = wp_password + self.base_url = f'{self.wp_url}/wp-json/wp/v2' + + self.pl = get_publish_logger() + self.dl = get_debug_logger() + + def create_post(self, title, content, status='publish', categories=None, tags=None, + excerpt=None, featured_media=None, slug=None): + """ + 创建文章 + + Args: + title: 文章标题 + content: 文章内容(HTML) + status: 文章状态 (publish/draft/pending/private) + categories: 分类 ID 列表 + tags: 标签 ID 列表 + excerpt: 文章摘要 + featured_media: 特色图片 ID + slug: 文章别名 + + Returns: + dict: 发布结果 + """ + self.pl.info(f"📝 创建文章:{title}") + self.dl.log_step("创建文章", f"标题:{title}, 状态:{status}") + + # 构建请求数据 + post_data = { + 'title': title, + 'content': content, + 'status': status + } + + if categories: + post_data['categories'] = categories + self.dl.debug(f"分类:{categories}") + + if tags: + post_data['tags'] = tags + self.dl.debug(f"标签:{tags}") + + if excerpt: + post_data['excerpt'] = excerpt + + if featured_media: + post_data['featured_media'] = featured_media + self.dl.debug(f"特色图片 ID: {featured_media}") + + if slug: + post_data['slug'] = slug + + try: + response = requests.post( + f'{self.base_url}/posts', + auth=(self.wp_user, self.wp_password), + json=post_data, + verify=False, + timeout=30 + ) + + if response.status_code == 201: + result = response.json() + post_id = result.get('id', 0) + post_url = result.get('link', '') + + self.pl.success(f"文章创建成功 - ID: {post_id}, URL: {post_url}") + self.dl.log_result("创建结果", { + 'id': post_id, + 'url': post_url, + 'status': status + }) + + return { + 'success': True, + 'id': post_id, + 'url': post_url, + 'status': status, + 'data': result + } + else: + error_data = self._parse_error(response) + self.pl.error(f"文章创建失败 - 状态码:{response.status_code}") + self.dl.error(f"创建失败:{error_data}") + return { + 'success': False, + 'error': error_data, + 'status_code': response.status_code + } + + except requests.exceptions.Timeout: + self.pl.error("请求超时") + self.dl.error("请求超时") + return {'success': False, 'error': '请求超时'} + + except Exception as e: + self.pl.error(f"请求异常:{str(e)}") + self.dl.error(f"请求异常:{str(e)}", exc_info=True) + return {'success': False, 'error': str(e)} + + def update_post(self, post_id, title=None, content=None, status=None, + categories=None, tags=None): + """ + 更新文章 + + Args: + post_id: 文章 ID + title: 新标题 + content: 新内容 + status: 新状态 + categories: 新分类 + tags: 新标签 + + Returns: + dict: 更新结果 + """ + self.pl.info(f"🔄 更新文章 ID: {post_id}") + + post_data = {} + if title: + post_data['title'] = title + if content: + post_data['content'] = content + if status: + post_data['status'] = status + if categories: + post_data['categories'] = categories + if tags: + post_data['tags'] = tags + + try: + response = requests.post( + f'{self.base_url}/posts/{post_id}', + auth=(self.wp_user, self.wp_password), + json=post_data, + verify=False, + timeout=30 + ) + + if response.status_code == 200: + result = response.json() + self.pl.success(f"文章更新成功 - ID: {post_id}") + return {'success': True, 'data': result} + else: + error_data = self._parse_error(response) + self.pl.error(f"文章更新失败:{error_data}") + return {'success': False, 'error': error_data} + + except Exception as e: + self.pl.error(f"更新异常:{str(e)}") + return {'success': False, 'error': str(e)} + + def delete_post(self, post_id, force=False): + """ + 删除文章 + + Args: + post_id: 文章 ID + force: 是否强制删除 + + Returns: + dict: 删除结果 + """ + self.pl.info(f"🗑️ 删除文章 ID: {post_id}") + + try: + response = requests.delete( + f'{self.base_url}/posts/{post_id}', + auth=(self.wp_user, self.wp_password), + params={'force': force}, + verify=False, + timeout=30 + ) + + if response.status_code == 200: + self.pl.success(f"文章删除成功 - ID: {post_id}") + return {'success': True} + else: + error_data = self._parse_error(response) + self.pl.error(f"文章删除失败:{error_data}") + return {'success': False, 'error': error_data} + + except Exception as e: + self.pl.error(f"删除异常:{str(e)}") + return {'success': False, 'error': str(e)} + + def get_post(self, post_id): + """ + 获取文章 + + Args: + post_id: 文章 ID + + Returns: + dict: 文章数据 + """ + try: + response = requests.get( + f'{self.base_url}/posts/{post_id}', + auth=(self.wp_user, self.wp_password), + verify=False, + timeout=30 + ) + + if response.status_code == 200: + return {'success': True, 'data': response.json()} + else: + return {'success': False, 'error': self._parse_error(response)} + + except Exception as e: + return {'success': False, 'error': str(e)} + + def get_categories(self): + """ + 获取所有分类 + + Returns: + list: 分类列表 + """ + try: + response = requests.get( + f'{self.base_url}/categories?per_page=100', + auth=(self.wp_user, self.wp_password), + verify=False, + timeout=30 + ) + + if response.status_code == 200: + categories = response.json() + self.dl.debug(f"获取到 {len(categories)} 个分类") + return categories + else: + self.dl.error(f"获取分类失败:{self._parse_error(response)}") + return [] + + except Exception as e: + self.dl.error(f"获取分类异常:{str(e)}") + return [] + + def get_tags(self): + """ + 获取所有标签 + + Returns: + list: 标签列表 + """ + try: + response = requests.get( + f'{self.base_url}/tags?per_page=100', + auth=(self.wp_user, self.wp_password), + verify=False, + timeout=30 + ) + + if response.status_code == 200: + tags = response.json() + self.dl.debug(f"获取到 {len(tags)} 个标签") + return tags + else: + self.dl.error(f"获取标签失败:{self._parse_error(response)}") + return [] + + except Exception as e: + self.dl.error(f"获取标签异常:{str(e)}") + return [] + + def search_posts(self, query, per_page=10): + """ + 搜索文章 + + Args: + query: 搜索关键词 + per_page: 每页数量 + + Returns: + list: 搜索结果 + """ + try: + response = requests.get( + f'{self.base_url}/posts', + auth=(self.wp_user, self.wp_password), + params={'search': query, 'per_page': per_page}, + verify=False, + timeout=30 + ) + + if response.status_code == 200: + return response.json() + else: + return [] + + except Exception as e: + self.dl.error(f"搜索异常:{str(e)}") + return [] + + def _parse_error(self, response): + """解析错误响应""" + try: + error_data = response.json() + return error_data.get('message', response.text) + except: + return response.text + + +def create_wp_api(wp_url, wp_user, wp_password): + """创建 WordPress API 客户端实例""" + return WordPressAPI(wp_url, wp_user, wp_password) + + +if __name__ == '__main__': + import sys + + if len(sys.argv) < 4: + print("用法:python wp_api.py ") + sys.exit(1) + + wp_url = sys.argv[1] + wp_user = sys.argv[2] + wp_password = sys.argv[3] + + api = create_wp_api(wp_url, wp_user, wp_password) + + # 测试获取分类 + categories = api.get_categories() + print(f"分类列表:") + for cat in categories: + print(f" ID: {cat['id']}, 名称:{cat['name']}, Slug: {cat['slug']}") + + # 测试获取标签 + tags = api.get_tags() + print(f"\n标签列表:") + for tag in tags: + print(f" ID: {tag['id']}, 名称:{tag['name']}, Slug: {tag['slug']}") diff --git a/modules/wp_category.py b/modules/wp_category.py new file mode 100644 index 0000000..16a2064 --- /dev/null +++ b/modules/wp_category.py @@ -0,0 +1,275 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +WordPress 发布系统 - 分类匹配模块 +根据指令或内容自动匹配 WordPress 分类 +""" + +import re +from modules.wp_logger import get_publish_logger, get_debug_logger + +# 默认分类配置 +DEFAULT_CATEGORY_ID = 7 # 随笔 +DEFAULT_CATEGORY_NAME = '随笔' + +# 分类关键词映射(用于 AI 自动匹配) +CATEGORY_KEYWORDS = { + 12: ['ai 科普', '人工智能科普', 'ai 入门', '科普'], # Ai 科普 + 11: ['ai 资讯', '人工智能新闻', 'ai 新闻', '资讯', '行业动态'], # Ai 资讯 + 16: ['geo', '生成式引擎优化', '搜索优化'], # GEO + 9: ['人工智能', 'ai', '机器学习', '深度学习', '神经网络', '大模型'], # 人工智能 + 5: ['技术', '教程', '开发', '编程', '代码', '技术资料'], # 技术资料 + 10: ['好物', '推荐', '分享', '产品', '测评'], # 好物分享 + 4: ['文章', '转载', '译文', '翻译'], # 文章分享 + 8: ['杂记', 'misc', '其他'], # 杂记 + 7: ['随笔', '日记', '心情', '感想'], # 随笔 + 1: ['关于', '网站', '联系', '声明'], # 关于我们 +} + + +class CategoryMatcher: + """分类匹配器""" + + def __init__(self, wp_api): + """ + 初始化分类匹配器 + + Args: + wp_api: WordPress API 客户端实例 + """ + self.wp_api = wp_api + self.categories_cache = [] + self.default_category_id = DEFAULT_CATEGORY_ID + + self.pl = get_publish_logger() + self.dl = get_debug_logger() + + def load_categories(self): + """加载分类列表""" + self.categories_cache = self.wp_api.get_categories() + self.dl.debug(f"已加载 {len(self.categories_cache)} 个分类") + return self.categories_cache + + def match_by_slug(self, slug): + """ + 根据 slug 匹配分类 + + Args: + slug: 分类 slug + + Returns: + int: 分类 ID,未找到返回默认分类 + """ + if not slug: + return self.default_category_id + + slug = slug.lower().strip() + + for category in self.categories_cache: + if category.get('slug', '').lower() == slug: + self.dl.debug(f"通过 slug 匹配到分类:{category['name']} (ID: {category['id']})") + return category['id'] + + self.dl.warning(f"未找到 slug 为 '{slug}' 的分类,使用默认分类") + return self.default_category_id + + def match_by_name(self, name): + """ + 根据名称匹配分类 + + Args: + name: 分类名称 + + Returns: + int: 分类 ID,未找到返回默认分类 + """ + if not name: + return self.default_category_id + + name = name.strip() + + for category in self.categories_cache: + if category.get('name', '').strip() == name: + self.dl.debug(f"通过名称匹配到分类:{category['name']} (ID: {category['id']})") + return category['id'] + + self.dl.warning(f"未找到名称为 '{name}' 的分类,使用默认分类") + return self.default_category_id + + def match_by_instruction(self, instruction): + """ + 根据指令文本匹配分类 + + Args: + instruction: 指令文本(如 "#分类 技术" 或 "分类:ai") + + Returns: + int: 分类 ID + """ + if not instruction: + return self.default_category_id + + instruction = instruction.strip().lower() + + # 匹配多种指令格式 + patterns = [ + r'#分类\s*([^\s#]+)', # #分类 技术 + r'#category\s*([^\s#]+)', # #category tech + r'分类 [::]\s*([^\s\n]+)', # 分类:技术 + r'分类 [::]\s*([^\s\n]+)', # 分类:tech + r'发布到\s*([^\s\n]+)', # 发布到 技术 + ] + + for pattern in patterns: + match = re.search(pattern, instruction, re.IGNORECASE) + if match: + category_value = match.group(1).strip() + self.dl.debug(f"从指令中提取分类:{category_value}") + + # 尝试匹配 slug + category_id = self.match_by_slug(category_value) + if category_id != self.default_category_id: + return category_id + + # 尝试匹配名称 + category_id = self.match_by_name(category_value) + if category_id != self.default_category_id: + return category_id + + self.dl.warning("指令中未找到有效分类,使用默认分类") + return self.default_category_id + + def match_by_content(self, title, content): + """ + 根据内容自动匹配分类(AI 匹配) + + Args: + title: 文章标题 + content: 文章内容 + + Returns: + int: 分类 ID + """ + if not title and not content: + return self.default_category_id + + # 合并标题和内容用于分析 + text = f"{title} {content}".lower() + + # 统计每个分类的关键词匹配分数 + scores = {} + for cat_id, keywords in CATEGORY_KEYWORDS.items(): + score = 0 + for keyword in keywords: + if keyword.lower() in text: + score += 1 + if score > 0: + scores[cat_id] = score + + if scores: + # 返回得分最高的分类 + best_cat_id = max(scores, key=scores.get) + best_score = scores[best_cat_id] + + # 获取分类名称 + cat_name = "未知" + for category in self.categories_cache: + if category['id'] == best_cat_id: + cat_name = category['name'] + break + + self.dl.debug(f"AI 匹配分类:{cat_name} (ID: {best_cat_id}, 得分:{best_score})") + return best_cat_id + + self.dl.debug("内容未匹配到关键词,使用默认分类") + return self.default_category_id + + def match(self, instruction=None, title=None, content=None, auto_match=True): + """ + 综合匹配分类 + + Args: + instruction: 指令文本(优先级最高) + title: 文章标题 + content: 文章内容 + auto_match: 是否启用自动匹配 + + Returns: + int: 分类 ID + """ + # 确保分类列表已加载 + if not self.categories_cache: + self.load_categories() + + # 优先级 1:指令匹配 + if instruction: + category_id = self.match_by_instruction(instruction) + if category_id != self.default_category_id: + self.pl.info(f"📂 分类:根据指令匹配到分类 ID {category_id}") + return category_id + + # 优先级 2:自动匹配(如果启用) + if auto_match and (title or content): + category_id = self.match_by_content(title or '', content or '') + if category_id != self.default_category_id: + self.pl.info(f"📂 分类:根据内容自动匹配到分类 ID {category_id}") + return category_id + + # 默认分类 + self.pl.info(f"📂 分类:使用默认分类 ID {self.default_category_id}") + return self.default_category_id + + def get_category_list(self): + """ + 获取分类列表(用于显示) + + Returns: + list: 分类信息列表 + """ + if not self.categories_cache: + self.load_categories() + + return [ + { + 'id': cat['id'], + 'name': cat['name'], + 'slug': cat['slug'], + 'keywords': CATEGORY_KEYWORDS.get(cat['id'], []) + } + for cat in self.categories_cache + ] + + +def create_category_matcher(wp_api): + """创建分类匹配器实例""" + return CategoryMatcher(wp_api) + + +if __name__ == '__main__': + import sys + + if len(sys.argv) < 4: + print("用法:python wp_category.py [instruction]") + sys.exit(1) + + wp_url = sys.argv[1] + wp_user = sys.argv[2] + wp_password = sys.argv[3] + instruction = sys.argv[4] if len(sys.argv) > 4 else None + + from modules.wp_api import create_wp_api + api = create_wp_api(wp_url, wp_user, wp_password) + matcher = create_category_matcher(api) + + # 显示所有分类 + print("可用分类列表:") + for cat in matcher.get_category_list(): + print(f" ID: {cat['id']}, 名称:{cat['name']}, Slug: {cat['slug']}") + if cat['keywords']: + print(f" 关键词:{', '.join(cat['keywords'])}") + + # 测试匹配 + if instruction: + print(f"\n指令 '{instruction}' 匹配结果:") + category_id = matcher.match(instruction=instruction) + print(f" 匹配到分类 ID: {category_id}") diff --git a/modules/wp_formatter.py b/modules/wp_formatter.py new file mode 100644 index 0000000..ad49a2e --- /dev/null +++ b/modules/wp_formatter.py @@ -0,0 +1,291 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +WordPress 发布系统 - HTML 格式化模块 +将解析后的内容转换为 WordPress 可用的 HTML 格式 +""" + +import os +import re +from modules.wp_logger import get_publish_logger, get_debug_logger + +# 基础配置 +BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) + + +class HTMLFormatter: + """HTML 格式化器""" + + def __init__(self): + self.pl = get_publish_logger() + self.dl = get_debug_logger() + + def format_content(self, content_parts, uploaded_images=None): + """ + 格式化内容为完整的 HTML + + Args: + content_parts: 内容片段列表 + uploaded_images: 已上传图片的列表,包含 url, index 等信息 + + Returns: + str: 完整的 HTML 内容 + """ + self.dl.log_step("格式化 HTML 内容") + + if not content_parts: + return "" + + html_parts = [] + + for i, part in enumerate(content_parts): + # 如果是图片占位符,替换为实际图片 URL + if isinstance(part, dict) and part.get('type') == 'image_placeholder': + if uploaded_images: + img_index = part.get('index', 0) + if img_index <= len(uploaded_images): + img = uploaded_images[img_index - 1] + if 'url' in img: + html_parts.append(f'图片 {img_index}') + continue + + html_parts.append(part) + + # 合并 HTML + full_html = '\n\n'.join(html_parts) + + # 优化 HTML 结构 + full_html = self._optimize_html(full_html) + + self.dl.debug(f"HTML 内容长度:{len(full_html)} 字符") + return full_html + + def format_text_content(self, text, images=None): + """ + 格式化纯文本内容为 HTML + + Args: + text: 文本内容 + images: 图片列表 + + Returns: + str: HTML 内容 + """ + self.dl.log_step("格式化文本内容") + + if not text: + return "" + + # 分割段落 + paragraphs = self._split_paragraphs(text) + + html_parts = [] + for para in paragraphs: + para = para.strip() + if not para: + continue + + # 检测是否为标题 + if self._is_title(para): + html_parts.append(f'

{self._escape_html(para)}

') + # 检测是否为列表 + elif self._is_list(para): + html_parts.append(self._format_list(para)) + # 普通段落 + else: + html_parts.append(f'

{self._format_text_styles(para)}

') + + return '\n\n'.join(html_parts) + + def generate_excerpt(self, content, max_length=200): + """ + 生成文章摘要 + + Args: + content: HTML 内容 + max_length: 最大长度 + + Returns: + str: 摘要 + """ + # 移除 HTML 标签 + text = re.sub(r'<[^>]+>', '', content) + + # 截断 + if len(text) > max_length: + text = text[:max_length] + '...' + + return text.strip() + + def extract_title_from_content(self, content): + """ + 从内容中提取标题 + + Args: + content: 文本内容 + + Returns: + str: 标题 + """ + lines = content.strip().split('\n') + + # 查找第一个非空行 + for line in lines: + line = line.strip() + if line: + # 如果很短,可能是标题 + if len(line) < 50: + return line + # 否则取前 30 个字符 + return line[:30] + + return "无标题文章" + + def _optimize_html(self, html): + """优化 HTML 结构""" + # 移除多余的空行 + html = re.sub(r'\n{3,}', '\n\n', html) + + # 确保段落之间有空行 + html = re.sub(r'

\s*

', '

\n\n

', html) + + # 移除空段落 + html = re.sub(r'

\s*

', '', html) + + return html + + def _split_paragraphs(self, text): + """分割段落""" + # 按双换行符分割 + paragraphs = re.split(r'\n\n+', text) + + # 也按单换行符分割(处理 Word 文档) + result = [] + for para in paragraphs: + sub_paras = re.split(r'\n+', para.strip()) + result.extend(sub_paras) + + return result + + def _is_title(self, text): + """判断是否为标题""" + # 标题通常较短且没有标点 + if len(text) > 60: + return False + + # 检查是否以标题标记开头 + if text.startswith('#'): + return True + + # 检查是否没有句号 + if not text.endswith(('。', '.', '!', '!', '?', '?')): + return True + + return False + + def _is_list(self, text): + """判断是否为列表""" + text = text.strip() + + # 检查项目符号 + if text[0] in ['•', '-', '–', '—', '▪', '▸', '▹', '*']: + return True + + # 检查编号 + if re.match(r'^\d+[\.\、\)]', text): + return True + if re.match(r'^[a-zA-Z][\.\、\)]', text): + return True + + return False + + def _format_list(self, text): + """格式化列表""" + lines = text.strip().split('\n') + items = [] + + for line in lines: + line = line.strip() + if not line: + continue + + # 清理列表标记 + clean_text = re.sub(r'^[•\-\–\—▪▸▹*]\s*', '', line) + clean_text = re.sub(r'^\d+[\.\、\)]\s*', '', clean_text) + clean_text = re.sub(r'^[a-zA-Z][\.\、\)]\s*', '', clean_text) + + items.append(f'
  • {self._format_text_styles(clean_text)}
  • ') + + # 判断列表类型 + is_ordered = bool(re.match(r'^\d+', lines[0])) if lines else False + list_tag = 'ol' if is_ordered else 'ul' + + return f'<{list_tag}>\n{"".join(items)}\n' + + def _format_text_styles(self, text): + """格式化文本样式(加粗、斜体等)""" + # 处理 Markdown 风格的加粗 + text = re.sub(r'\*\*(.+?)\*\*', r'\1', text) + text = re.sub(r'__(.+?)__', r'\1', text) + + # 处理斜体 + text = re.sub(r'\*(.+?)\*', r'\1', text) + text = re.sub(r'_(.+?)_', r'\1', text) + + # 处理链接 + text = re.sub(r'\[(.+?)\]\((.+?)\)', r'\1', text) + + return self._escape_html(text) + + def _escape_html(self, text): + """转义 HTML 特殊字符(保留已处理的标签)""" + # 先保护已处理的 HTML 标签 + protected = [] + + def protect(match): + protected.append(match.group(0)) + return f'__PROTECTED_{len(protected)-1}__' + + # 保护已有的 HTML 标签 + text = re.sub(r'<(strong|em|a|li|ol|ul)[^>]*>.*?', protect, text) + text = re.sub(r'<(strong|em|a)[^>]*/>', protect, text) + + # 转义其他特殊字符 + text = text.replace('&', '&') + text = text.replace('<', '<') + text = text.replace('>', '>') + text = text.replace('"', '"') + + # 恢复保护的标签 + for i, tag in enumerate(protected): + text = text.replace(f'__PROTECTED_{i}__', tag) + + return text + + +def create_formatter(): + """创建格式化器实例""" + return HTMLFormatter() + + +if __name__ == '__main__': + formatter = create_formatter() + + # 测试 + test_content = """# 这是一个标题 + +这是第一段内容。 + +- 列表项 1 +- 列表项 2 +- 列表项 3 + +这是第二段内容,包含 **加粗** 和 *斜体*。 + +1. 编号列表 1 +2. 编号列表 2 +""" + + html = formatter.format_text_content(test_content) + print("格式化后的 HTML:") + print(html) diff --git a/modules/wp_image_handler.py b/modules/wp_image_handler.py new file mode 100644 index 0000000..37ee212 --- /dev/null +++ b/modules/wp_image_handler.py @@ -0,0 +1,309 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +WordPress 发布系统 - 图片处理模块 +处理图片保存、上传到 WordPress 媒体库 +""" + +import os +import base64 +import requests +from io import BytesIO +from PIL import Image + +from modules.wp_logger import get_publish_logger, get_debug_logger + +# 基础目录 +BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +TEMP_DIR = os.path.join(BASE_DIR, 'temp') +os.makedirs(TEMP_DIR, exist_ok=True) + + +class ImageHandler: + """图片处理器""" + + def __init__(self, wp_url, wp_user, wp_password): + """ + 初始化图片处理器 + + Args: + wp_url: WordPress 站点 URL + wp_user: WordPress 用户名 + wp_password: WordPress 应用密码 + """ + self.wp_url = wp_url.rstrip('/') + self.wp_user = wp_user + self.wp_password = wp_password + self.uploaded_images = {} # 记录已上传的图片 + + self.pl = get_publish_logger() + self.dl = get_debug_logger() + + def save_temp_image(self, image_data, filename): + """ + 保存图片到临时目录 + + Args: + image_data: 图片二进制数据 + filename: 文件名 + + Returns: + str: 临时文件路径 + """ + temp_path = os.path.join(TEMP_DIR, filename) + + try: + with open(temp_path, 'wb') as f: + f.write(image_data) + + self.dl.debug(f"图片已保存:{temp_path}") + return temp_path + + except Exception as e: + self.dl.error(f"保存图片失败:{str(e)}") + raise + + def upload_image(self, image_path, title=None, alt_text=None): + """ + 上传图片到 WordPress 媒体库 + + Args: + image_path: 图片文件路径 + title: 图片标题 + alt_text: 图片 alt 文本 + + Returns: + dict: 上传结果,包含 url, id, filename + """ + if not os.path.exists(image_path): + raise FileNotFoundError(f"图片文件不存在:{image_path}") + + filename = os.path.basename(image_path) + if not title: + title = os.path.splitext(filename)[0] + if not alt_text: + alt_text = title + + self.pl.info(f"📤 上传图片:{filename}") + self.dl.log_step("上传图片", f"文件:{filename}") + + try: + # 读取图片文件 + with open(image_path, 'rb') as f: + image_content = f.read() + + # 获取 content_type + content_type = self._get_content_type(image_path) + + # 构建请求 + headers = { + 'Content-Disposition': f'attachment; filename="{filename}"', + 'Content-Type': content_type, + 'Content-Transfer-Encoding': 'binary' + } + + # 上传图片 + response = requests.post( + f'{self.wp_url}/wp-json/wp/v2/media', + auth=(self.wp_user, self.wp_password), + headers=headers, + data=image_content, + verify=False, # 跳过 SSL 验证 + timeout=30 + ) + + if response.status_code == 201: + result = response.json() + image_url = result.get('source_url', '') + image_id = result.get('id', 0) + + # 更新图片标题和 alt + self._update_image_meta(image_id, title, alt_text) + + self.pl.success(f"图片上传成功 - ID: {image_id}, URL: {image_url}") + self.dl.log_result("上传结果", { + 'id': image_id, + 'url': image_url, + 'filename': filename + }) + + # 记录已上传的图片 + self.uploaded_images[filename] = { + 'id': image_id, + 'url': image_url, + 'title': title + } + + return { + 'id': image_id, + 'url': image_url, + 'filename': filename, + 'title': title + } + else: + error_msg = response.text + self.pl.error(f"图片上传失败 - 状态码:{response.status_code}") + self.dl.error(f"上传失败:{error_msg}") + raise Exception(f"图片上传失败:{error_msg}") + + except Exception as e: + self.pl.error(f"图片上传异常:{str(e)}") + self.dl.error(f"上传异常:{str(e)}", exc_info=True) + raise + + def upload_images_batch(self, images): + """ + 批量上传图片 + + Args: + images: 图片列表,每个图片包含 data, filename + + Returns: + list: 上传结果列表 + """ + results = [] + + for i, image in enumerate(images): + try: + # 保存图片到临时目录 + temp_path = self.save_temp_image(image['data'], image['filename']) + + # 上传图片 + result = self.upload_image( + temp_path, + title=f"图片 {i+1}", + alt_text=f"文章配图 {i+1}" + ) + + results.append(result) + + # 清理临时文件 + if os.path.exists(temp_path): + os.remove(temp_path) + + except Exception as e: + self.pl.error(f"图片 {i+1} 上传失败:{str(e)}") + results.append({'error': str(e), 'filename': image['filename']}) + + return results + + def generate_image_html(self, image_url, alt_text="", width=None): + """ + 生成图片 HTML 标签 + + Args: + image_url: 图片 URL + alt_text: alt 文本 + width: 图片宽度(可选) + + Returns: + str: HTML img 标签 + """ + style = "max-width: 100%; height: auto; display: block; margin: 16px auto;" + if width: + style += f" max-width: {width}px;" + + html = f'{alt_text}' + return f'
    {html}
    ' + + def generate_featured_image_shortcode(self, image_url, alt_text=""): + """ + 生成特色图片短代码 + + Args: + image_url: 图片 URL + alt_text: alt 文本 + + Returns: + str: 特色图片 HTML + """ + return self.generate_image_html(image_url, alt_text) + + def _update_image_meta(self, image_id, title, alt_text): + """更新图片元数据""" + try: + response = requests.post( + f'{self.wp_url}/wp-json/wp/v2/media/{image_id}', + auth=(self.wp_user, self.wp_password), + json={ + 'title': title, + 'alt_text': alt_text + }, + verify=False, + timeout=10 + ) + if response.status_code == 200: + self.dl.debug(f"图片元数据更新成功:ID {image_id}") + except Exception as e: + self.dl.warning(f"更新图片元数据失败:{str(e)}") + + def _get_content_type(self, file_path): + """获取图片 content_type""" + ext = os.path.splitext(file_path)[1].lower() + content_type_map = { + '.jpg': 'image/jpeg', + '.jpeg': 'image/jpeg', + '.png': 'image/png', + '.gif': 'image/gif', + '.bmp': 'image/bmp', + '.webp': 'image/webp', + '.svg': 'image/svg+xml' + } + return content_type_map.get(ext, 'image/jpeg') + + def optimize_image(self, image_path, max_width=1200, quality=85): + """ + 优化图片大小 + + Args: + image_path: 图片路径 + max_width: 最大宽度 + quality: 质量 (1-100) + + Returns: + str: 优化后的图片路径 + """ + try: + img = Image.open(image_path) + + # 获取原始尺寸 + width, height = img.size + + # 如果宽度超过限制,等比例缩放 + if width > max_width: + ratio = max_width / width + new_height = int(height * ratio) + img = img.resize((max_width, new_height), Image.LANCZOS) + self.dl.debug(f"图片已缩放:{width}x{height} -> {max_width}x{new_height}") + + # 保存优化后的图片 + optimized_path = image_path.replace('.', '_optimized.') + img.save(optimized_path, quality=quality, optimize=True) + + return optimized_path + + except Exception as e: + self.dl.warning(f"图片优化失败:{str(e)}") + return image_path + + +def create_image_handler(wp_url, wp_user, wp_password): + """创建图片处理器实例""" + return ImageHandler(wp_url, wp_user, wp_password) + + +if __name__ == '__main__': + import sys + + if len(sys.argv) < 5: + print("用法:python wp_image_handler.py ") + sys.exit(1) + + wp_url = sys.argv[1] + wp_user = sys.argv[2] + wp_password = sys.argv[3] + image_path = sys.argv[4] + + handler = create_image_handler(wp_url, wp_user, wp_password) + result = handler.upload_image(image_path) + print(f"上传结果:{result}") diff --git a/modules/wp_logger.py b/modules/wp_logger.py new file mode 100644 index 0000000..817ae52 --- /dev/null +++ b/modules/wp_logger.py @@ -0,0 +1,173 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +WordPress 发布系统 - 日志模块 +提供发布日志和调试日志功能 +""" + +import os +import sys +import logging +from logging.handlers import RotatingFileHandler +from datetime import datetime + +# 基础目录 +BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +LOG_DIR = os.path.join(BASE_DIR, 'logs') +os.makedirs(LOG_DIR, exist_ok=True) + +# 日志格式 +LOG_FORMAT = '%(asctime)s - %(name)s - %(levelname)s - %(message)s' +DEBUG_FORMAT = '%(asctime)s - %(name)s - %(levelname)s - [%(filename)s:%(lineno)d] - %(message)s' +PUBLISH_FORMAT = '%(asctime)s | %(message)s' + +# 日志文件大小限制 (10MB) +MAX_BYTES = 10 * 1024 * 1024 +BACKUP_COUNT = 5 + + +class PublishLogger: + """发布日志记录器""" + + def __init__(self, log_file='publish.log'): + self.logger = logging.getLogger('wp_publish') + self.logger.setLevel(logging.INFO) + + # 避免重复添加处理器 + if self.logger.handlers: + return + + # 发布日志文件处理器 + log_path = os.path.join(LOG_DIR, log_file) + file_handler = RotatingFileHandler( + log_path, + maxBytes=MAX_BYTES, + backupCount=BACKUP_COUNT, + encoding='utf-8' + ) + file_handler.setLevel(logging.INFO) + file_handler.setFormatter(logging.Formatter(PUBLISH_FORMAT)) + self.logger.addHandler(file_handler) + + # 控制台输出 + console_handler = logging.StreamHandler(sys.stdout) + console_handler.setLevel(logging.INFO) + console_handler.setFormatter(logging.Formatter(PUBLISH_FORMAT)) + self.logger.addHandler(console_handler) + + def info(self, message): + self.logger.info(message) + + def success(self, message): + self.logger.info(f"✅ SUCCESS: {message}") + + def warning(self, message): + self.logger.warning(f"⚠️ WARNING: {message}") + + def error(self, message): + self.logger.error(f"❌ ERROR: {message}") + + def start_publish(self, source_type, filename=None): + timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S') + self.logger.info("=" * 60) + self.logger.info(f"🚀 开始发布 - {timestamp}") + self.logger.info(f"📋 发布类型:{source_type}") + if filename: + self.logger.info(f"📁 文件名:{filename}") + self.logger.info("=" * 60) + + def end_publish(self, success, post_id=None, post_url=None, error_msg=None): + if success: + self.success(f"发布成功!文章 ID: {post_id}") + if post_url: + self.info(f"🔗 文章链接:{post_url}") + else: + self.error(f"发布失败:{error_msg}") + self.info("=" * 60) + + +class DebugLogger: + """调试日志记录器""" + + def __init__(self, log_file='debug.log'): + self.logger = logging.getLogger('wp_debug') + self.logger.setLevel(logging.DEBUG) + + # 避免重复添加处理器 + if self.logger.handlers: + return + + # 调试日志文件处理器 + log_path = os.path.join(LOG_DIR, log_file) + file_handler = RotatingFileHandler( + log_path, + maxBytes=MAX_BYTES, + backupCount=BACKUP_COUNT, + encoding='utf-8' + ) + file_handler.setLevel(logging.DEBUG) + file_handler.setFormatter(logging.Formatter(DEBUG_FORMAT)) + self.logger.addHandler(file_handler) + + # 错误日志单独记录 + error_log_path = os.path.join(LOG_DIR, 'error.log') + error_handler = RotatingFileHandler( + error_log_path, + maxBytes=MAX_BYTES, + backupCount=BACKUP_COUNT, + encoding='utf-8' + ) + error_handler.setLevel(logging.ERROR) + error_handler.setFormatter(logging.Formatter(DEBUG_FORMAT)) + self.logger.addHandler(error_handler) + + def debug(self, message): + self.logger.debug(message) + + def info(self, message): + self.logger.info(message) + + def warning(self, message): + self.logger.warning(message) + + def error(self, message, exc_info=None): + self.logger.error(message, exc_info=exc_info) + + def log_step(self, step_name, details=None): + """记录步骤""" + self.logger.info(f"📌 步骤:{step_name}") + if details: + self.logger.info(f" 详情:{details}") + + def log_result(self, result_type, result_data): + """记录结果""" + self.logger.info(f"📊 结果:{result_type}") + self.logger.debug(f" 数据:{result_data}") + + +# 创建全局日志实例 +publish_logger = PublishLogger() +debug_logger = DebugLogger() + + +def get_publish_logger(): + """获取发布日志实例""" + return publish_logger + + +def get_debug_logger(): + """获取调试日志实例""" + return debug_logger + + +if __name__ == '__main__': + # 测试日志 + pl = get_publish_logger() + dl = get_debug_logger() + + pl.start_publish('测试', 'test.docx') + pl.info('正在解析文档...') + pl.success('文档解析完成') + dl.log_step('解析测试', '提取标题和正文') + dl.debug('调试信息:文档结构正常') + pl.end_publish(True, post_id=1, post_url='https://www.nanlou.net/test') diff --git a/modules/wp_parse_docx.py b/modules/wp_parse_docx.py new file mode 100644 index 0000000..05649b0 --- /dev/null +++ b/modules/wp_parse_docx.py @@ -0,0 +1,445 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +WordPress 发布系统 - Word 文档解析模块 +解析 .docx 文件,提取标题、正文、图片等元素 +""" + +import os +import re +import base64 +import hashlib +from docx import Document +from docx.shared import Pt, Inches +from docx.enum.text import WD_ALIGN_PARAGRAPH + +from modules.wp_logger import get_publish_logger, get_debug_logger + +# 基础目录 +BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +TEMP_DIR = os.path.join(BASE_DIR, 'temp') +os.makedirs(TEMP_DIR, exist_ok=True) + + +class WordParser: + """Word 文档解析器""" + + def __init__(self, file_path): + """ + 初始化解析器 + + Args: + file_path: Word 文档路径 + """ + self.file_path = file_path + self.filename = os.path.basename(file_path) + self.doc = None + self.title = "" + self.content_parts = [] + self.images = [] + self.metadata = { + 'paragraph_count': 0, + 'image_count': 0, + 'word_count': 0 + } + + self.pl = get_publish_logger() + self.dl = get_debug_logger() + + def parse(self): + """ + 解析 Word 文档 + + Returns: + dict: 包含 title, content, images 的字典 + """ + self.pl.info(f"📖 开始解析文档:{self.filename}") + + try: + # 加载文档 + self.doc = Document(self.file_path) + self.dl.log_step("加载文档", f"成功加载:{self.file_path}") + + # 提取标题 + self._extract_title() + + # 提取内容 + self._extract_content() + + # 提取图片 + self._extract_images() + + # 统计信息 + self._update_metadata() + + self.pl.success(f"文档解析完成 - 标题:{self.title},段落数:{self.metadata['paragraph_count']},图片数:{self.metadata['image_count']}") + + return { + 'title': self.title, + 'content': self.content_parts, + 'images': self.images, + 'metadata': self.metadata + } + + except Exception as e: + self.pl.error(f"文档解析失败:{str(e)}") + self.dl.error(f"解析异常:{str(e)}", exc_info=True) + raise + + def _extract_title(self): + """提取文档标题""" + self.dl.log_step("提取标题") + + # 方法 1:从文档属性获取 + if self.doc.core_properties.title: + self.title = self.doc.core_properties.title.strip() + self.dl.debug(f"从文档属性获取标题:{self.title}") + return + + # 方法 2:从第一个标题样式段落获取 + for paragraph in self.doc.paragraphs: + if paragraph.style.name.startswith('Heading'): + self.title = paragraph.text.strip() + self.dl.debug(f"从标题样式获取标题:{self.title}") + return + + # 方法 3:从第一个加粗大字号段落获取 + for paragraph in self.doc.paragraphs: + if paragraph.text.strip() and self._is_title_style(paragraph): + self.title = paragraph.text.strip() + self.dl.debug(f"从样式特征获取标题:{self.title}") + return + + # 方法 4:使用文件名作为标题 + self.title = os.path.splitext(self.filename)[0] + self.dl.warning(f"使用文件名作为标题:{self.title}") + + def _extract_content(self): + """提取文档内容(段落、列表等)""" + self.dl.log_step("提取内容") + + content_html = [] + in_list = False + list_type = None + list_items = [] + + for i, paragraph in enumerate(self.doc.paragraphs): + text = paragraph.text.strip() + + # 跳过空段落 + if not text: + if in_list and list_items: + content_html.extend(self._close_list(list_type, list_items)) + list_items = [] + in_list = False + continue + + # 检测是否为标题 + if paragraph.style.name.startswith('Heading 1') or self._is_heading(paragraph, 1): + if in_list and list_items: + content_html.extend(self._close_list(list_type, list_items)) + list_items = [] + in_list = False + content_html.append(f'

    {self._escape_html(text)}

    ') + self.dl.debug(f"段落 {i}: H2 标题") + continue + + if paragraph.style.name.startswith('Heading 2') or self._is_heading(paragraph, 2): + if in_list and list_items: + content_html.extend(self._close_list(list_type, list_items)) + list_items = [] + in_list = False + content_html.append(f'

    {self._escape_html(text)}

    ') + self.dl.debug(f"段落 {i}: H3 标题") + continue + + # 检测是否为列表项 + if self._is_list_item(paragraph): + if not in_list: + in_list = True + list_type = 'ol' if self._is_numbered_list(paragraph) else 'ul' + list_items = [] + + # 清理列表标记 + clean_text = self._clean_list_marker(text) + list_items.append(f'
  • {self._format_run_styles(clean_text, paragraph)}
  • ') + self.dl.debug(f"段落 {i}: 列表项") + continue + + # 普通段落 + if in_list and list_items: + content_html.extend(self._close_list(list_type, list_items)) + list_items = [] + in_list = False + + # 处理加粗文本 + formatted_text = self._format_run_styles(text, paragraph) + content_html.append(f'

    {formatted_text}

    ') + self.dl.debug(f"段落 {i}: 普通段落") + + # 关闭最后的列表 + if in_list and list_items: + content_html.extend(self._close_list(list_type, list_items)) + + self.content_parts = content_html + + def _extract_images(self): + """提取文档中的图片""" + self.dl.log_step("提取图片") + + image_index = 0 + + for i, paragraph in enumerate(self.doc.paragraphs): + # 检查段落中的图片 + for run in paragraph.runs: + if run._element.xml.find('pic:pic') != -1 or run._element.xml.find('w:binData') != -1: + image_data = self._extract_image_from_run(run) + if image_data: + image_index += 1 + self.images.append({ + 'index': image_index, + 'paragraph_index': i, + 'filename': f"image_{image_index}_{image_data['hash'][:8]}.{image_data['format']}", + 'data': image_data['data'], + 'format': image_data['format'] + }) + self.dl.debug(f"提取图片 {image_index}:{self.images[-1]['filename']}") + + # 也检查文档关系中的图片 + for rel in self.doc.part.rels.values(): + if "image" in rel.target_ref: + image_index += 1 + image_data = self._extract_image_from_rel(rel) + if image_data: + self.images.append({ + 'index': image_index, + 'paragraph_index': -1, + 'filename': f"image_{image_index}_{image_data['hash'][:8]}.{image_data['format']}", + 'data': image_data['data'], + 'format': image_data['format'] + }) + self.dl.debug(f"提取图片 {image_index}(从关系):{self.images[-1]['filename']}") + + def _update_metadata(self): + """更新文档元数据""" + self.metadata['paragraph_count'] = len(self.doc.paragraphs) + self.metadata['image_count'] = len(self.images) + + # 粗略计算字数 + word_count = sum(len(p.text) for p in self.doc.paragraphs) + self.metadata['word_count'] = word_count + + # ========== 辅助方法 ========== + + def _is_title_style(self, paragraph): + """判断段落是否为标题样式""" + if not paragraph.runs: + return False + + first_run = paragraph.runs[0] + if first_run.font.size: + size = first_run.font.size.pt + if size and size >= 16: + return True + + if first_run.font.bold: + return True + + return False + + def _is_heading(self, paragraph, level): + """判断段落是否为指定级别的标题""" + if level == 1: + return (paragraph.runs and + paragraph.runs[0].font.size and + paragraph.runs[0].font.size.pt >= 18 and + paragraph.runs[0].font.bold) + elif level == 2: + return (paragraph.runs and + paragraph.runs[0].font.size and + paragraph.runs[0].font.size.pt >= 14 and + paragraph.runs[0].font.bold) + return False + + def _is_list_item(self, paragraph): + """判断段落是否为列表项""" + text = paragraph.text.strip() + if not text: + return False + + # 检查项目符号 + if text[0] in ['•', '-', '–', '—', '▪', '▸', '▹']: + return True + + # 检查编号 + if re.match(r'^\d+[\.\、\)]', text): + return True + if re.match(r'^[a-zA-Z][\.\、\)]', text): + return True + if re.match(r'^[(\(]\d+[)\)]', text): + return True + + # 检查样式 + if 'List' in paragraph.style.name: + return True + + return False + + def _is_numbered_list(self, paragraph): + """判断是否为编号列表""" + text = paragraph.text.strip() + return bool(re.match(r'^\d+[\.\、\)]', text) or + re.match(r'^[a-zA-Z][\.\、\)]', text)) + + def _clean_list_marker(self, text): + """清理列表标记""" + # 移除项目符号 + if text[0] in ['•', '-', '–', '—', '▪', '▸', '▹']: + text = text[1:].strip() + + # 移除编号 + text = re.sub(r'^\d+[\.\、\)]\s*', '', text) + text = re.sub(r'^[a-zA-Z][\.\、\)]\s*', '', text) + text = re.sub(r'^[(\(]\d+[)\)]\s*', '', text) + + return text.strip() + + def _format_run_styles(self, text, paragraph): + """格式化文本样式(加粗、斜体等)""" + if not paragraph.runs: + return self._escape_html(text) + + result = [] + for run in paragraph.runs: + run_text = run.text + if not run_text: + continue + + # 加粗 + if run.font.bold: + run_text = f'{self._escape_html(run_text)}' + else: + run_text = self._escape_html(run_text) + + # 斜体 + if run.font.italic: + run_text = f'{run_text}' + + result.append(run_text) + + return ''.join(result) if result else self._escape_html(text) + + def _close_list(self, list_type, items): + """关闭列表标签""" + if not items: + return [] + return [f'<{list_type}>{"".join(items)}'] + + def _escape_html(self, text): + """转义 HTML 特殊字符""" + text = text.replace('&', '&') + text = text.replace('<', '<') + text = text.replace('>', '>') + text = text.replace('"', '"') + return text + + def _extract_image_from_run(self, run): + """从 run 中提取图片""" + try: + # 获取图片二进制数据 + xml = run._element.xml + if 'w:binData' in xml: + import xml.etree.ElementTree as ET + from io import BytesIO + + # 解析 XML 获取图片数据 + root = ET.fromstring(xml) + for elem in root.iter(): + if 'binData' in elem.tag and elem.text: + image_data = base64.b64decode(elem.text) + image_hash = hashlib.md5(image_data).hexdigest() + + # 检测图片格式 + fmt = self._detect_image_format(image_data) + + return { + 'data': image_data, + 'format': fmt, + 'hash': image_hash + } + except Exception as e: + self.dl.error(f"提取图片失败:{str(e)}") + + return None + + def _extract_image_from_rel(self, rel): + """从关系中提取图片""" + try: + image_part = rel.target_part + image_data = image_part.blob + image_hash = hashlib.md5(image_data).hexdigest() + + # 获取格式 + content_type = image_part.content_type + fmt = self._content_type_to_format(content_type) + + return { + 'data': image_data, + 'format': fmt, + 'hash': image_hash + } + except Exception as e: + self.dl.error(f"从关系提取图片失败:{str(e)}") + + return None + + def _detect_image_format(self, data): + """检测图片格式""" + if data[:3] == b'\xff\xd8\xff': + return 'jpg' + elif data[:8] == b'\x89PNG\r\n\x1a\n': + return 'png' + elif data[:6] in (b'GIF87a', b'GIF89a'): + return 'gif' + elif data[:2] == b'BM': + return 'bmp' + return 'jpg' # 默认为 jpg + + def _content_type_to_format(self, content_type): + """将 content_type 转换为格式""" + format_map = { + 'image/jpeg': 'jpg', + 'image/png': 'png', + 'image/gif': 'gif', + 'image/bmp': 'bmp', + 'image/webp': 'webp' + } + return format_map.get(content_type, 'jpg') + + +def parse_word_file(file_path): + """ + 解析 Word 文件的便捷函数 + + Args: + file_path: Word 文档路径 + + Returns: + dict: 解析结果 + """ + parser = WordParser(file_path) + return parser.parse() + + +if __name__ == '__main__': + import sys + if len(sys.argv) < 2: + print("用法:python wp_parse_docx.py ") + sys.exit(1) + + result = parse_word_file(sys.argv[1]) + print(f"标题:{result['title']}") + print(f"段落数:{result['metadata']['paragraph_count']}") + print(f"图片数:{result['metadata']['image_count']}") + print(f"字数:{result['metadata']['word_count']}") + print(f"\nHTML 内容:\n{''.join(result['content'])}") diff --git a/scripts/wp_cli.py b/scripts/wp_cli.py new file mode 100644 index 0000000..7408597 --- /dev/null +++ b/scripts/wp_cli.py @@ -0,0 +1,292 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +WordPress CLI 工具 - 供 OpenClaw 调用 +通过 stdin 接收 JSON 数据,避免 shell 引号冲突问题 +""" + +import sys +import json +import requests +import argparse +import os + + +def main(): + parser = argparse.ArgumentParser(description='WordPress CLI Tool') + parser.add_argument('action', choices=['create', 'update', 'get', 'list', 'delete', 'upload'], + help='操作类型') + parser.add_argument('--url', required=True, help='WordPress URL') + parser.add_argument('--user', required=True, help='WordPress 用户名') + parser.add_argument('--password', required=True, help='WordPress 应用密码') + parser.add_argument('--type', choices=['posts', 'pages', 'categories', 'tags', 'media'], + default='posts', help='资源类型') + parser.add_argument('--id', type=int, help='资源 ID(用于 update/get/delete)') + parser.add_argument('--data-file', help='JSON 数据文件路径(推荐,避免 shell 引号问题)') + parser.add_argument('--search', help='搜索关键词') + parser.add_argument('--per-page', type=int, default=10, help='每页数量') + parser.add_argument('--page', type=int, default=1, help='页码') + + args = parser.parse_args() + + # 构建基础 URL + base_url = args.url.rstrip('/') + api_base = f'{base_url}/wp-json/wp/v2/{args.type}' + + # 读取 JSON 数据(优先从文件读取,避免 shell 引号问题) + data = {} + if args.data_file: + if os.path.exists(args.data_file): + with open(args.data_file, 'r', encoding='utf-8') as f: + data = json.load(f) + else: + print(json.dumps({'success': False, 'error': f'数据文件不存在:{args.data_file}'})) + sys.exit(1) + elif not sys.stdin.isatty(): + # 从 stdin 读取 + try: + stdin_data = sys.stdin.read() + if stdin_data.strip(): + data = json.loads(stdin_data) + except json.JSONDecodeError as e: + print(json.dumps({'success': False, 'error': f'Stdin JSON 解析失败:{str(e)}'})) + sys.exit(1) + + try: + if args.action == 'create': + result = create_resource(api_base, args.user, args.password, data) + elif args.action == 'update': + if not args.id: + print(json.dumps({'success': False, 'error': '更新操作需要 --id 参数'})) + sys.exit(1) + result = update_resource(f'{api_base}/{args.id}', args.user, args.password, data) + elif args.action == 'get': + if not args.id: + print(json.dumps({'success': False, 'error': '获取操作需要 --id 参数'})) + sys.exit(1) + result = get_resource(f'{api_base}/{args.id}', args.user, args.password) + elif args.action == 'list': + result = list_resources(api_base, args.user, args.password, + search=args.search, per_page=args.per_page, page=args.page) + elif args.action == 'delete': + if not args.id: + print(json.dumps({'success': False, 'error': '删除操作需要 --id 参数'})) + sys.exit(1) + result = delete_resource(f'{api_base}/{args.id}', args.user, args.password) + elif args.action == 'upload': + if not args.data_file: + print(json.dumps({'success': False, 'error': '上传操作需要 --data-file 指定媒体文件'})) + sys.exit(1) + result = upload_media(api_base, args.user, args.password, args.data_file, data) + + # 输出结果 + print(json.dumps(result, ensure_ascii=False, indent=2)) + + except Exception as e: + print(json.dumps({'success': False, 'error': str(e)}, ensure_ascii=False)) + sys.exit(1) + + +def create_resource(api_url, user, password, data): + """创建资源""" + response = requests.post( + api_url, + auth=(user, password), + json=data, + verify=False, + timeout=30 + ) + + if response.status_code == 201: + result = response.json() + return { + 'success': True, + 'id': result.get('id'), + 'url': result.get('link', ''), + 'title': result.get('title', {}).get('rendered', '') if isinstance(result.get('title'), dict) else result.get('title', ''), + 'data': result + } + else: + return { + 'success': False, + 'error': response.text, + 'status_code': response.status_code + } + + +def update_resource(api_url, user, password, data): + """更新资源""" + response = requests.post( + api_url, + auth=(user, password), + json=data, + verify=False, + timeout=30 + ) + + if response.status_code == 200: + result = response.json() + return { + 'success': True, + 'id': result.get('id'), + 'data': result + } + else: + return { + 'success': False, + 'error': response.text, + 'status_code': response.status_code + } + + +def get_resource(api_url, user, password): + """获取资源""" + response = requests.get( + api_url, + auth=(user, password), + verify=False, + timeout=30 + ) + + if response.status_code == 200: + return { + 'success': True, + 'data': response.json() + } + else: + return { + 'success': False, + 'error': response.text, + 'status_code': response.status_code + } + + +def list_resources(api_url, user, password, search=None, per_page=10, page=1): + """列出资源""" + params = { + 'per_page': per_page, + 'page': page + } + if search: + params['search'] = search + + response = requests.get( + api_url, + auth=(user, password), + params=params, + verify=False, + timeout=30 + ) + + if response.status_code == 200: + return { + 'success': True, + 'data': response.json(), + 'total': response.headers.get('X-WP-Total', 0), + 'total_pages': response.headers.get('X-WP-TotalPages', 0) + } + else: + return { + 'success': False, + 'error': response.text, + 'status_code': response.status_code + } + + +def delete_resource(api_url, user, password): + """删除资源""" + response = requests.delete( + api_url, + auth=(user, password), + params={'force': True}, + verify=False, + timeout=30 + ) + + if response.status_code == 200: + return { + 'success': True, + 'data': response.json() + } + else: + return { + 'success': False, + 'error': response.text, + 'status_code': response.status_code + } + + +def upload_media(api_url, user, password, file_path, metadata=None): + """上传媒体文件""" + if not os.path.exists(file_path): + return {'success': False, 'error': f'文件不存在:{file_path}'} + + filename = os.path.basename(file_path) + content_type = get_content_type(file_path) + + headers = { + 'Content-Disposition': f'attachment; filename="{filename}"', + 'Content-Type': content_type + } + + with open(file_path, 'rb') as f: + file_data = f.read() + + response = requests.post( + api_url, + auth=(user, password), + headers=headers, + data=file_data, + verify=False, + timeout=60 + ) + + if response.status_code == 201: + result = response.json() + media_id = result.get('id') + + # 更新元数据 + if metadata and media_id: + update_url = f'{api_url}/{media_id}' + requests.post( + update_url, + auth=(user, password), + json=metadata, + verify=False, + timeout=10 + ) + + return { + 'success': True, + 'id': media_id, + 'url': result.get('source_url', ''), + 'filename': filename, + 'data': result + } + else: + return { + 'success': False, + 'error': response.text, + 'status_code': response.status_code + } + + +def get_content_type(file_path): + """获取文件 MIME 类型""" + ext = os.path.splitext(file_path)[1].lower() + content_types = { + '.jpg': 'image/jpeg', + '.jpeg': 'image/jpeg', + '.png': 'image/png', + '.gif': 'image/gif', + '.webp': 'image/webp', + '.svg': 'image/svg+xml', + '.pdf': 'application/pdf', + '.doc': 'application/msword', + '.docx': 'application/vnd.openxmlformats-officedocument.wordprocessingml.document' + } + return content_types.get(ext, 'application/octet-stream') + + +if __name__ == '__main__': + main() diff --git a/scripts/wp_publish.py b/scripts/wp_publish.py new file mode 100644 index 0000000..73c6e0a --- /dev/null +++ b/scripts/wp_publish.py @@ -0,0 +1,249 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +WordPress 发布系统 - 主发布脚本 +整合 Word 解析、图片上传、分类匹配、文章发布全流程 +""" + +import os +import sys +import json +import argparse + +# 添加项目根目录到 Python 路径 +BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +sys.path.insert(0, BASE_DIR) + +from modules.wp_logger import get_publish_logger, get_debug_logger +from modules.wp_parse_docx import parse_word_file +from modules.wp_image_handler import create_image_handler +from modules.wp_formatter import create_formatter +from modules.wp_api import create_wp_api +from modules.wp_category import create_category_matcher + +# 配置文件路径 +CONFIG_FILE = os.path.join(BASE_DIR, 'config.py') + + +def load_config(): + """加载配置文件""" + config = { + 'wp_url': 'https://www.nanlou.net', + 'wp_user': 'shaowu', + 'wp_password': 'zjzz gHYm 8Q3l KbZk y4CF 2DQi', + 'default_category': 7, + 'auto_match_category': True, + 'optimize_images': True, + 'image_max_width': 1200, + 'image_quality': 85, + 'post_status': 'publish' + } + + if os.path.exists(CONFIG_FILE): + try: + with open(CONFIG_FILE, 'r', encoding='utf-8') as f: + exec(f.read(), config) + except Exception as e: + print(f"加载配置文件失败:{str(e)},使用默认配置") + + return config + + +def publish_word_document(word_file_path, instruction=None, status=None, category_id=None, tags=None): + """ + 发布 Word 文档到 WordPress + + Args: + word_file_path: Word 文档路径 + instruction: 指令文本(可选) + status: 发布状态(可选,默认从配置读取) + category_id: 指定分类 ID(可选) + tags: 标签列表(可选) + + Returns: + dict: 发布结果 + """ + # 初始化日志 + pl = get_publish_logger() + dl = get_debug_logger() + + # 加载配置 + config = load_config() + + # 初始化各模块 + wp_api = create_wp_api(config['wp_url'], config['wp_user'], config['wp_password']) + image_handler = create_image_handler(config['wp_url'], config['wp_user'], config['wp_password']) + formatter = create_formatter() + category_matcher = create_category_matcher(wp_api) + + # 开始发布 + pl.start_publish('Word 文档', os.path.basename(word_file_path)) + + try: + # ========== 步骤 1:解析 Word 文档 ========== + dl.log_step("解析 Word 文档", word_file_path) + parse_result = parse_word_file(word_file_path) + + title = parse_result['title'] + content_parts = parse_result['content'] + images = parse_result['images'] + metadata = parse_result['metadata'] + + pl.info(f"📖 解析完成 - 标题:{title}, 段落数:{metadata['paragraph_count']}, 图片数:{metadata['image_count']}") + + # ========== 步骤 2:上传图片 ========== + uploaded_images = [] + if images: + dl.log_step("上传图片", f"共 {len(images)} 张图片") + uploaded_images = image_handler.upload_images_batch(images) + pl.info(f"📤 图片上传完成 - 成功 {len([img for img in uploaded_images if 'url' in img])} 张") + + # ========== 步骤 3:匹配分类 ========== + if category_id: + final_category_id = category_id + else: + final_category_id = category_matcher.match( + instruction=instruction, + title=title, + content=' '.join(content_parts), + auto_match=config.get('auto_match_category', True) + ) + + # ========== 步骤 4:格式化 HTML ========== + dl.log_step("格式化 HTML 内容") + content_html = formatter.format_content(content_parts, uploaded_images) + + # 生成摘要 + excerpt = formatter.generate_excerpt(content_html) + + # ========== 步骤 5:发布文章 ========== + dl.log_step("发布文章") + + # 确定发布状态 + post_status = status or config.get('post_status', 'publish') + + # 构建发布数据 + publish_data = { + 'title': title, + 'content': content_html, + 'status': post_status, + 'categories': [final_category_id], + 'excerpt': excerpt + } + + if tags: + publish_data['tags'] = tags + + # 如果有上传的图片,设置第一张为特色图片 + if uploaded_images and uploaded_images[0].get('url'): + # 获取特色图片 ID + featured_img = uploaded_images[0] + if 'id' in featured_img: + publish_data['featured_media'] = featured_img['id'] + dl.debug(f"设置特色图片 ID: {featured_img['id']}") + + # 调用 API 发布 + result = wp_api.create_post(**publish_data) + + # ========== 步骤 6:输出结果 ========== + if result.get('success'): + pl.end_publish( + True, + post_id=result.get('id'), + post_url=result.get('url') + ) + + # 清理临时文件 + _cleanup_temp_files() + + return { + 'success': True, + 'post_id': result.get('id'), + 'post_url': result.get('url'), + 'title': title, + 'category_id': final_category_id, + 'images_uploaded': len([img for img in uploaded_images if 'url' in img]) + } + else: + pl.end_publish(False, error_msg=result.get('error')) + return { + 'success': False, + 'error': result.get('error') + } + + except Exception as e: + pl.end_publish(False, error_msg=str(e)) + dl.error(f"发布异常:{str(e)}", exc_info=True) + return { + 'success': False, + 'error': str(e) + } + + +def _cleanup_temp_files(): + """清理临时文件""" + temp_dir = os.path.join(BASE_DIR, 'temp') + if os.path.exists(temp_dir): + for filename in os.listdir(temp_dir): + file_path = os.path.join(temp_dir, filename) + try: + if os.path.isfile(file_path): + os.remove(file_path) + except Exception as e: + print(f"清理临时文件失败:{file_path}, {str(e)}") + + +def main(): + """命令行入口""" + parser = argparse.ArgumentParser(description='WordPress 文章发布工具') + parser.add_argument('file', help='Word 文档路径 (.docx)') + parser.add_argument('--instruction', '-i', help='发布指令(如:#分类 技术)') + parser.add_argument('--status', '-s', choices=['publish', 'draft', 'pending', 'private'], + default=None, help='发布状态') + parser.add_argument('--category', '-c', type=int, help='指定分类 ID') + parser.add_argument('--tags', '-t', help='标签 ID 列表(逗号分隔)') + parser.add_argument('--dry-run', '-d', action='store_true', help='预览模式(不实际发布)') + + args = parser.parse_args() + + # 检查文件是否存在 + if not os.path.exists(args.file): + print(f"❌ 文件不存在:{args.file}") + sys.exit(1) + + # 解析标签 + tags = None + if args.tags: + tags = [int(t.strip()) for t in args.tags.split(',') if t.strip()] + + # 预览模式 + if args.dry_run: + print("🔍 预览模式 - 解析文档内容:") + parse_result = parse_word_file(args.file) + print(f" 标题:{parse_result['title']}") + print(f" 段落数:{parse_result['metadata']['paragraph_count']}") + print(f" 图片数:{parse_result['metadata']['image_count']}") + print(f" 字数:{parse_result['metadata']['word_count']}") + print(f"\nHTML 内容预览:") + print('\n'.join(parse_result['content'][:10])) + print("...") + sys.exit(0) + + # 执行发布 + result = publish_word_document( + word_file_path=args.file, + instruction=args.instruction, + status=args.status, + category_id=args.category, + tags=tags + ) + + # 输出 JSON 结果 + print("\n" + json.dumps(result, ensure_ascii=False, indent=2)) + + # 返回状态码 + sys.exit(0 if result.get('success') else 1) + + +if __name__ == '__main__': + main() diff --git a/scripts/wp_publish_text.py b/scripts/wp_publish_text.py new file mode 100644 index 0000000..73b2ef4 --- /dev/null +++ b/scripts/wp_publish_text.py @@ -0,0 +1,284 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +WordPress 发布系统 - 文字 + 图片发布脚本 +处理从飞书等渠道发送的文字和图片,自动发布到 WordPress +""" + +import os +import sys +import json +import argparse +import base64 +import hashlib + +# 添加项目根目录到 Python 路径 +BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +sys.path.insert(0, BASE_DIR) + +from modules.wp_logger import get_publish_logger, get_debug_logger +from modules.wp_image_handler import create_image_handler +from modules.wp_formatter import create_formatter +from modules.wp_api import create_wp_api +from modules.wp_category import create_category_matcher + +# 配置文件路径 +CONFIG_FILE = os.path.join(BASE_DIR, 'config.py') + + +def load_config(): + """加载配置文件""" + config = { + 'wp_url': 'https://www.nanlou.net', + 'wp_user': 'shaowu', + 'wp_password': 'zjzz gHYm 8Q3l KbZk y4CF 2DQi', + 'default_category': 7, + 'auto_match_category': True, + 'optimize_images': True, + 'image_max_width': 1200, + 'image_quality': 85, + 'post_status': 'publish' + } + + if os.path.exists(CONFIG_FILE): + try: + with open(CONFIG_FILE, 'r', encoding='utf-8') as f: + exec(f.read(), config) + except Exception as e: + print(f"加载配置文件失败:{str(e)},使用默认配置") + + return config + + +def publish_text_with_images(text, images=None, instruction=None, status=None, + category_id=None, tags=None, title=None): + """ + 发布文字 + 图片到 WordPress + + Args: + text: 文字内容 + images: 图片列表,每个图片包含 data (base64 或文件路径), filename + instruction: 指令文本(可选) + status: 发布状态(可选) + category_id: 指定分类 ID(可选) + tags: 标签列表(可选) + title: 文章标题(可选,默认从内容提取) + + Returns: + dict: 发布结果 + """ + # 初始化日志 + pl = get_publish_logger() + dl = get_debug_logger() + + # 加载配置 + config = load_config() + + # 初始化各模块 + wp_api = create_wp_api(config['wp_url'], config['wp_user'], config['wp_password']) + image_handler = create_image_handler(config['wp_url'], config['wp_user'], config['wp_password']) + formatter = create_formatter() + category_matcher = create_category_matcher(wp_api) + + # 开始发布 + pl.start_publish('文字 + 图片') + + try: + # ========== 步骤 1:提取标题 ========== + if not title: + title = formatter.extract_title_from_content(text) + + pl.info(f"📝 文章标题:{title}") + dl.log_step("提取标题", title) + + # ========== 步骤 2:上传图片 ========== + uploaded_images = [] + if images: + dl.log_step("上传图片", f"共 {len(images)} 张图片") + + for i, img in enumerate(images): + try: + # 处理 base64 图片 + if isinstance(img, str) and os.path.exists(img): + # 文件路径 + img_path = img + elif isinstance(img, dict) and 'data' in img: + # 字典格式(包含 base64 数据) + img_data = img['data'] + if isinstance(img_data, str): + # base64 编码 + img_data = base64.b64decode(img_data) + + filename = img.get('filename', f'image_{i+1}.jpg') + img_hash = hashlib.md5(img_data).hexdigest()[:8] + ext = os.path.splitext(filename)[1] or '.jpg' + filename = f"image_{i+1}_{img_hash}{ext}" + + # 保存到临时文件 + temp_dir = os.path.join(BASE_DIR, 'temp') + os.makedirs(temp_dir, exist_ok=True) + img_path = os.path.join(temp_dir, filename) + with open(img_path, 'wb') as f: + f.write(img_data) + else: + continue + + # 上传图片 + result = image_handler.upload_image( + img_path, + title=f"图片 {i+1}", + alt_text=f"文章配图 {i+1}" + ) + uploaded_images.append(result) + + # 清理临时文件 + if 'img_path' in locals() and os.path.exists(img_path): + os.remove(img_path) + + except Exception as e: + pl.error(f"图片 {i+1} 上传失败:{str(e)}") + dl.error(f"图片上传失败:{str(e)}", exc_info=True) + + pl.info(f"📤 图片上传完成 - 成功 {len(uploaded_images)} 张") + + # ========== 步骤 3:匹配分类 ========== + if category_id: + final_category_id = category_id + else: + final_category_id = category_matcher.match( + instruction=instruction, + title=title, + content=text, + auto_match=config.get('auto_match_category', True) + ) + + # ========== 步骤 4:格式化 HTML ========== + dl.log_step("格式化 HTML 内容") + + # 先格式化文字内容 + content_html = formatter.format_text_content(text) + + # 插入图片 + if uploaded_images: + # 将图片插入到内容中(每段之间) + paragraphs = content_html.split('

    ') + new_html = [] + img_index = 0 + + for para in paragraphs: + para = para.strip() + if not para: + continue + + new_html.append(para + '

    ') + + # 在段落间插入图片 + if img_index < len(uploaded_images): + img = uploaded_images[img_index] + if 'url' in img: + img_html = f'{img.get(' + new_html.append(img_html) + img_index += 1 + + content_html = '\n\n'.join(new_html) + + # 生成摘要 + excerpt = formatter.generate_excerpt(content_html) + + # ========== 步骤 5:发布文章 ========== + dl.log_step("发布文章") + + # 确定发布状态 + post_status = status or config.get('post_status', 'publish') + + # 构建发布数据 + publish_data = { + 'title': title, + 'content': content_html, + 'status': post_status, + 'categories': [final_category_id], + 'excerpt': excerpt + } + + if tags: + publish_data['tags'] = tags + + # 如果有上传的图片,设置第一张为特色图片 + if uploaded_images and uploaded_images[0].get('id'): + publish_data['featured_media'] = uploaded_images[0]['id'] + dl.debug(f"设置特色图片 ID: {uploaded_images[0]['id']}") + + # 调用 API 发布 + result = wp_api.create_post(**publish_data) + + # ========== 步骤 6:输出结果 ========== + if result.get('success'): + pl.end_publish( + True, + post_id=result.get('id'), + post_url=result.get('url') + ) + + return { + 'success': True, + 'post_id': result.get('id'), + 'post_url': result.get('url'), + 'title': title, + 'category_id': final_category_id, + 'images_uploaded': len(uploaded_images) + } + else: + pl.end_publish(False, error_msg=result.get('error')) + return { + 'success': False, + 'error': result.get('error') + } + + except Exception as e: + pl.end_publish(False, error_msg=str(e)) + dl.error(f"发布异常:{str(e)}", exc_info=True) + return { + 'success': False, + 'error': str(e) + } + + +def main(): + """命令行入口""" + parser = argparse.ArgumentParser(description='WordPress 文字 + 图片发布工具') + parser.add_argument('text', help='文字内容') + parser.add_argument('--images', '-i', nargs='+', help='图片文件路径') + parser.add_argument('--title', '-t', help='文章标题') + parser.add_argument('--instruction', '-c', help='发布指令(如:#分类 技术)') + parser.add_argument('--status', '-s', choices=['publish', 'draft', 'pending', 'private'], + default=None, help='发布状态') + parser.add_argument('--category', '-C', type=int, help='指定分类 ID') + parser.add_argument('--tags', '-T', help='标签 ID 列表(逗号分隔)') + + args = parser.parse_args() + + # 解析标签 + tags = None + if args.tags: + tags = [int(t.strip()) for t in args.tags.split(',') if t.strip()] + + # 执行发布 + result = publish_text_with_images( + text=args.text, + images=args.images, + instruction=args.instruction, + status=args.status, + category_id=args.category, + tags=tags, + title=args.title + ) + + # 输出 JSON 结果 + print("\n" + json.dumps(result, ensure_ascii=False, indent=2)) + + # 返回状态码 + sys.exit(0 if result.get('success') else 1) + + +if __name__ == '__main__': + main() diff --git a/webhook_server.py b/webhook_server.py new file mode 100644 index 0000000..06a9893 --- /dev/null +++ b/webhook_server.py @@ -0,0 +1,198 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +飞书机器人 Webhook 服务器 +接收飞书消息推送,处理并回复 +""" + +import os +import sys +import json +import logging +from http.server import HTTPServer, BaseHTTPRequestHandler +from datetime import datetime + +# 添加项目根目录到 Python 路径 +BASE_DIR = os.path.dirname(os.path.abspath(__file__)) +sys.path.insert(0, BASE_DIR) + +from feishu_bot import handle_message + +# 配置日志 +LOG_DIR = os.path.join(BASE_DIR, 'logs') +os.makedirs(LOG_DIR, exist_ok=True) + +logging.basicConfig( + level=logging.INFO, + format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', + handlers=[ + logging.FileHandler(os.path.join(LOG_DIR, 'webhook_server.log'), encoding='utf-8'), + logging.StreamHandler() + ] +) +logger = logging.getLogger('webhook_server') + + +class FeishuWebhookHandler(BaseHTTPRequestHandler): + """飞书 Webhook 处理器""" + + def do_POST(self): + """处理 POST 请求""" + try: + # 读取请求体 + content_length = int(self.headers.get('Content-Length', 0)) + if content_length == 0: + logger.warning("收到空请求") + self._send_response(400, {'error': 'Empty request body'}) + return + + post_data = self.rfile.read(content_length) + + # 解析 JSON + try: + message_data = json.loads(post_data.decode('utf-8')) + except json.JSONDecodeError as e: + logger.error(f"JSON 解析失败:{str(e)}") + logger.error(f"原始数据:{post_data[:200]}") + self._send_response(400, {'error': 'Invalid JSON'}) + return + + logger.info(f"📨 收到飞书消息") + logger.debug(f"请求数据:{json.dumps(message_data, ensure_ascii=False)}") + + # 处理事件 + if message_data.get('header', {}).get('event_type') == 'url_verification': + # URL 验证 + self._handle_url_verification(message_data) + elif message_data.get('header', {}).get('event_type') == 'im.message.receive_v1': + # 消息接收事件 + self._handle_message_event(message_data) + else: + logger.warning(f"未知事件类型:{message_data.get('header', {}).get('event_type')}") + self._send_response(200, {'error': 'Unknown event type'}) + + except Exception as e: + logger.error(f"处理请求失败:{str(e)}", exc_info=True) + self._send_response(500, {'error': str(e)}) + + def _handle_url_verification(self, message_data): + """ + 处理 URL 验证 + + Args: + message_data: 验证数据 + """ + challenge = message_data.get('challenge', '') + logger.info(f"🔐 URL 验证:{challenge}") + self._send_response(200, {'challenge': challenge}) + + def _handle_message_event(self, message_data): + """ + 处理消息事件 + + Args: + message_data: 消息数据 + """ + try: + # 提取消息内容 + event = message_data.get('event', {}) + message = event.get('message', {}) + sender = event.get('sender', {}) + + # 构建消息对象 + msg = { + 'msg_type': message.get('message_type', ''), + 'content': message.get('content', ''), + 'sender': { + 'sender_id': sender.get('sender_id', ''), + 'sender_type': sender.get('sender_type', '') + }, + 'message_id': message.get('message_id', ''), + 'chat_id': message.get('chat_id', '') + } + + # 处理消息 + reply = handle_message(msg) + + # 发送回复 + self._send_reply(msg.get('chat_id', ''), msg.get('message_id', ''), reply) + + logger.info(f"✅ 消息处理完成") + + except Exception as e: + logger.error(f"处理消息事件失败:{str(e)}", exc_info=True) + + def _send_reply(self, chat_id, message_id, reply_text): + """ + 发送回复 + + Args: + chat_id: 聊天 ID + message_id: 消息 ID + reply_text: 回复内容 + """ + # TODO: 实现回复发送逻辑 + logger.info(f"📤 准备回复 - 聊天:{chat_id}, 消息:{message_id}") + logger.info(f"回复内容:{reply_text}") + + def _send_response(self, status_code, data): + """ + 发送响应 + + Args: + status_code: HTTP 状态码 + data: 响应数据 + """ + self.send_response(status_code) + self.send_header('Content-Type', 'application/json') + self.end_headers() + self.wfile.write(json.dumps(data).encode('utf-8')) + + def do_GET(self): + """处理 GET 请求""" + self.send_response(200) + self.send_header('Content-Type', 'text/html; charset=utf-8') + self.end_headers() + + html = """ + + 飞书机器人 Webhook 服务 + +

    🤖 飞书机器人 Webhook 服务

    +

    服务运行正常 ✅

    +

    时间:{time}

    + + + """.format(time=datetime.now().strftime('%Y-%m-%d %H:%M:%S')) + + self.wfile.write(html.encode('utf-8')) + + +def start_server(host='0.0.0.0', port=8080): + """ + 启动 Webhook 服务器 + + Args: + host: 监听地址 + port: 监听端口 + """ + server = HTTPServer((host, port), FeishuWebhookHandler) + logger.info(f"🌐 Webhook 服务器启动 - http://{host}:{port}") + + try: + server.serve_forever() + except KeyboardInterrupt: + logger.info("🛑 服务器关闭") + server.shutdown() + + +if __name__ == '__main__': + import argparse + + parser = argparse.ArgumentParser(description='飞书机器人 Webhook 服务器') + parser.add_argument('--host', default='0.0.0.0', help='监听地址') + parser.add_argument('--port', type=int, default=8080, help='监听端口') + + args = parser.parse_args() + + start_server(args.host, args.port)