feat: 初始化 WordPress 自动发布系统(飞书机器人集成)

- 飞书消息接收与处理(文字、图片、Word 文档)
- WordPress REST API 文章发布
- 图片自动上传到媒体库
- Word 文档解析与发布
- HTML 格式化与分类自动匹配
- Python CLI 工具(避免 shell 引号冲突)
- Webhook 服务器(8080 端口)
- 完整日志系统
This commit is contained in:
wp-publish-bot 2026-05-12 15:09:30 +08:00
commit 1fb93e34c6
18 changed files with 4521 additions and 0 deletions

33
.gitignore vendored Normal file
View File

@ -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

193
BUGFIX_SUMMARY.md Normal file
View File

@ -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. 确保图片发布时不会覆盖文字内容
---
## 🔍 问题 3JSON 格式错误
### 原始问题
发布图片时报错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
```
---
## 📞 技术支持
如有问题,请查看日志文件获取详细错误信息,或联系技术支持。

129
FEISHU_SETUP.md Normal file
View File

@ -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
```
## 📞 技术支持
如有问题,请查看日志文件获取详细错误信息。

171
IMAGE_FIX.md Normal file
View File

@ -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
```
## 📞 技术支持
如有问题,请查看日志文件获取详细错误信息,或联系技术支持。

162
README.md Normal file
View File

@ -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
- 检查文件是否损坏
## 📞 技术支持
如有问题,请查看日志文件获取详细错误信息。

664
feishu_bot.py Normal file
View File

@ -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}")

5
modules/__init__.py Normal file
View File

@ -0,0 +1,5 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
WordPress 发布系统 - 模块包
"""

289
modules/feishu_api.py Normal file
View File

@ -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 客户端初始化失败")

359
modules/wp_api.py Normal file
View File

@ -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 <wp_url> <wp_user> <wp_password>")
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']}")

275
modules/wp_category.py Normal file
View File

@ -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 <wp_url> <wp_user> <wp_password> [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}")

291
modules/wp_formatter.py Normal file
View File

@ -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 src="{img["url"]}" alt="图片 {img_index}" style="max-width: 100%; height: auto; display: block; margin: 16px auto;">')
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'<h2>{self._escape_html(para)}</h2>')
# 检测是否为列表
elif self._is_list(para):
html_parts.append(self._format_list(para))
# 普通段落
else:
html_parts.append(f'<p>{self._format_text_styles(para)}</p>')
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'</p>\s*<p>', '</p>\n\n<p>', html)
# 移除空段落
html = re.sub(r'<p>\s*</p>', '', 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'<li>{self._format_text_styles(clean_text)}</li>')
# 判断列表类型
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</{list_tag}>'
def _format_text_styles(self, text):
"""格式化文本样式(加粗、斜体等)"""
# 处理 Markdown 风格的加粗
text = re.sub(r'\*\*(.+?)\*\*', r'<strong>\1</strong>', text)
text = re.sub(r'__(.+?)__', r'<strong>\1</strong>', text)
# 处理斜体
text = re.sub(r'\*(.+?)\*', r'<em>\1</em>', text)
text = re.sub(r'_(.+?)_', r'<em>\1</em>', text)
# 处理链接
text = re.sub(r'\[(.+?)\]\((.+?)\)', r'<a href="\2">\1</a>', 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)[^>]*>.*?</\1>', protect, text)
text = re.sub(r'<(strong|em|a)[^>]*/>', protect, text)
# 转义其他特殊字符
text = text.replace('&', '&amp;')
text = text.replace('<', '&lt;')
text = text.replace('>', '&gt;')
text = text.replace('"', '&quot;')
# 恢复保护的标签
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)

309
modules/wp_image_handler.py Normal file
View File

@ -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'<img src="{image_url}" alt="{alt_text}" style="{style}" loading="lazy">'
return f'<figure style="text-align: center;">{html}</figure>'
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 <wp_url> <wp_user> <wp_password> <image_path>")
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}")

173
modules/wp_logger.py Normal file
View File

@ -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')

445
modules/wp_parse_docx.py Normal file
View File

@ -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'<h2>{self._escape_html(text)}</h2>')
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'<h3>{self._escape_html(text)}</h3>')
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'<li>{self._format_run_styles(clean_text, paragraph)}</li>')
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'<p>{formatted_text}</p>')
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'<strong>{self._escape_html(run_text)}</strong>'
else:
run_text = self._escape_html(run_text)
# 斜体
if run.font.italic:
run_text = f'<em>{run_text}</em>'
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)}</{list_type}>']
def _escape_html(self, text):
"""转义 HTML 特殊字符"""
text = text.replace('&', '&amp;')
text = text.replace('<', '&lt;')
text = text.replace('>', '&gt;')
text = text.replace('"', '&quot;')
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 <word 文件路径>")
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'])}")

292
scripts/wp_cli.py Normal file
View File

@ -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()

249
scripts/wp_publish.py Normal file
View File

@ -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()

284
scripts/wp_publish_text.py Normal file
View File

@ -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('</p>')
new_html = []
img_index = 0
for para in paragraphs:
para = para.strip()
if not para:
continue
new_html.append(para + '</p>')
# 在段落间插入图片
if img_index < len(uploaded_images):
img = uploaded_images[img_index]
if 'url' in img:
img_html = f'<img src="{img["url"]}" alt="{img.get("title", "")}" style="max-width: 100%; height: auto; display: block; margin: 16px auto;">'
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()

198
webhook_server.py Normal file
View File

@ -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 = """
<html>
<head><title>飞书机器人 Webhook 服务</title></head>
<body>
<h1>🤖 飞书机器人 Webhook 服务</h1>
<p>服务运行正常 </p>
<p>时间{time}</p>
</body>
</html>
""".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)