1fb93e34c6
- 飞书消息接收与处理（文字、图片、Word 文档） - WordPress REST API 文章发布 - 图片自动上传到媒体库 - Word 文档解析与发布 - HTML 格式化与分类自动匹配 - Python CLI 工具（避免 shell 引号冲突） - Webhook 服务器（8080 端口） - 完整日志系统
360 lines
11 KiB
Python
360 lines
11 KiB
Python
#!/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']}")
|