效果展示:
(1)系统环境:
(2)步骤:
- 创建虚拟环境:
source epubtowordpress/bin/activate
- 执行命令:python3 epubtowordpress.py
(3)epubtowordpress.py 代码
#!/usr/bin/env python3
"""
EPUB到WordPress发布脚本 - 修复版
解决tuple indices must be integers or slices, not str错误
"""
import os
import glob
import zipfile
import shutil
import tempfile
import time
import json
from ebooklib import epub
from wordpress_xmlrpc import Client, WordPressPost
from wordpress_xmlrpc.methods.posts import NewPost
from wordpress_xmlrpc.methods.media import UploadFile
from wordpress_xmlrpc.compat import xmlrpc_client
class EpubToWordPress:
def __init__(self, wp_url, wp_username, wp_password):
"""
初始化发布器
Args:
wp_url: WordPress网站地址
wp_username: WordPress用户名
wp_password: WordPress密码
"""
self.wp_url = wp_url.rstrip('/')
self.wp_username = wp_username
self.wp_password = wp_password
self.client = Client(f"{wp_url}/xmlrpc.php", wp_username, wp_password)
def safe_extract_metadata(self, book, namespace, name):
"""
安全提取元数据,修复tuple indices错误
"""
try:
metadata = book.get_metadata(namespace, name)
if not metadata:
return ""
# 处理 [(value, attributes)] 格式,取元组第一个元素
if isinstance(metadata, list) and len(metadata) > 0:
item = metadata[0] # 取列表第一个元素(元组)
if isinstance(item, (tuple, list)) and len(item) > 0:
return str(item[0]) # 取元组第一个元素(实际值)
elif isinstance(item, str):
return item
else:
return str(item)
return ""
except Exception as e:
print(f"元数据提取警告 ({namespace}/{name}): {e}")
return ""
def safe_extract_metadata_list(self, book, namespace, name):
"""
安全提取元数据列表
"""
results = []
try:
metadata = book.get_metadata(namespace, name)
if not metadata:
return results
for item in metadata:
if isinstance(item, (tuple, list)) and len(item) > 0:
results.append(str(item[0])) # 修复:取元组第一个元素
elif isinstance(item, str):
results.append(item)
else:
results.append(str(item))
return results
except Exception as e:
print(f"元数据列表提取警告: {e}")
return []
def extract_complete_metadata(self, epub_path):
"""
提取EPUB文件的完整元数据
"""
try:
print(f"正在解析EPUB文件: {os.path.basename(epub_path)}")
if not self.check_epub_integrity(epub_path):
print("文件完整性检查失败")
return None
book = epub.read_epub(epub_path)
metadata = {
'title': self.safe_extract_metadata(book, 'DC', 'title') or
os.path.basename(epub_path).replace('.epub', ''),
'authors': self.safe_extract_metadata_list(book, 'DC', 'creator'),
'publisher': self.safe_extract_metadata(book, 'DC', 'publisher'),
'language': self.safe_extract_metadata(book, 'DC', 'language'),
'description': self.safe_extract_metadata(book, 'DC', 'description'),
'dates': self.safe_extract_metadata_list(book, 'DC', 'date'),
'subjects': self.safe_extract_metadata_list(book, 'DC', 'subject'),
'identifiers': [],
'filename': os.path.basename(epub_path)
}
# 特殊处理标识符(修复元组取值)
try:
identifiers = book.get_metadata('DC', 'identifier')
if identifiers:
for identifier in identifiers:
if isinstance(identifier, (tuple, list)) and len(identifier) > 0:
id_value = str(identifier[0]) # 修复:取元组第一个元素
id_type = 'unknown'
if len(identifier) > 1 and hasattr(identifier[1], 'get'):
id_type = identifier[1].get('id', 'unknown')
metadata['identifiers'].append({
'type': id_type,
'value': id_value
})
except Exception as e:
print(f"标识符提取警告: {e}")
print(f"✓ 成功提取元数据: {metadata['title']}")
return metadata
except Exception as e:
print(f"✗ 解析EPUB文件失败: {e}")
return None
def check_epub_integrity(self, epub_path):
"""检查EPUB文件完整性"""
try:
with zipfile.ZipFile(epub_path, 'r') as zip_ref:
return zip_ref.testzip() is None
except Exception as e:
print(f"文件完整性检查错误: {e}")
return False
def upload_epub_file(self, epub_path):
"""上传EPUB文件到WordPress媒体库"""
try:
print(f"正在上传文件: {os.path.basename(epub_path)}")
data = {
'name': os.path.basename(epub_path),
'type': 'application/epub+zip'
}
with open(epub_path, 'rb') as f:
data['bits'] = xmlrpc_client.Binary(f.read())
response = self.client.call(UploadFile(data))
print(f"✓ 文件上传成功: {response['url']}")
return {
'url': response['url'],
'id': response['id']
}
except Exception as e:
print(f"✗ 文件上传失败: {e}")
return None
def create_post(self, metadata, file_url):
"""创建WordPress文章"""
try:
post = WordPressPost()
post.title = metadata['title']
post.content = self.generate_post_content(metadata, file_url)
post.post_status = 'publish'
if metadata['description']:
post.excerpt = metadata['description'][:300]
else:
post.excerpt = f"{metadata['title']} - 电子书下载"
tags = ['EPUB', '电子书' ]
if metadata['subjects']:
tags.extend(metadata['subjects'][:3])
categories = ['书籍分享', 'sjfx']
if metadata['publisher']:
tags.append(metadata['publisher'])
post.terms_names = {
'post_tag': tags,
'category': categories
}
result = self.client.call(NewPost(post))
if result:
print(f"✓ 文章发布成功: {metadata['title']}")
return True
else:
print(f"✗ 文章发布失败")
return False
except Exception as e:
print(f"✗ 文章创建失败: {e}")
return False
def generate_post_content(self, metadata, file_url):
"""生成文章内容HTML"""
content = f"""
<div class="epub-book-details">
<h2>📖 图书详细信息</h2>
<table class="book-info">
<tr><th>书名:</th><td>{metadata['title']}</td></tr>
<tr><th>作者:</th><td>{', '.join(metadata['authors']) if metadata['authors'] else '未知'}</td></tr>
<tr><th>出版商:</th><td>{metadata['publisher'] or '未知'}</td></tr>
<tr><th>语言:</th><td>{metadata['language'] or '未知'}</td></tr>
<tr><th>出版日期:</th><td>{', '.join(metadata['dates']) if metadata['dates'] else '未知'}</td></tr>
</table>
"""
if metadata['identifiers']:
content += "<h3>🔍 标识符</h3><ul>"
for identifier in metadata['identifiers']:
content += f'<li><strong>{identifier["type"]}:</strong> {identifier["value"]}</li>'
content += "</ul>"
if metadata['description']:
content += f'<h3>📝 内容简介</h3><div class="description">{metadata["description"]}</div>'
if metadata['subjects']:
content += f'<h3>🏷️ 主题分类</h3><p>{", ".join(metadata["subjects"])}</p>'
content += f"""
<div class="download-section">
<h3>⬇️ 下载链接</h3>
<p><a href="{file_url}" class="download-button" download="{metadata['filename']}">
📥 下载 EPUB 文件 ({metadata['filename']})
</a></p>
</div>
<style>
.book-info {{
border-collapse: collapse;
width: 100%;
margin: 20px 0;
font-size: 14px;
}}
.book-info th, .book-info td {{
border: 1px solid #ddd;
padding: 12px;
text-align: left;
}}
.book-info th {{
background-color: #f5f5f5;
width: 120px;
font-weight: bold;
}}
.download-button {{
display: inline-block;
padding: 12px 24px;
background-color: #0073aa;
color: white;
text-decoration: none;
border-radius: 5px;
font-weight: bold;
font-size: 16px;
}}
.download-button:hover {{
background-color: #005a87;
}}
.description {{
line-height: 1.6;
padding: 15px;
background-color: #f8f9fa;
border-radius: 5px;
margin: 15px 0;
}}
</style>
"""
return content
def process_single_epub(self, epub_path):
"""处理单个EPUB文件"""
print(f"\n{'='*60}")
print(f"处理文件: {os.path.basename(epub_path)}")
metadata = self.extract_complete_metadata(epub_path)
if not metadata:
return {'file': epub_path, 'status': 'metadata_failed'}
upload_result = self.upload_epub_file(epub_path)
if not upload_result:
return {'file': epub_path, 'status': 'upload_failed'}
if self.create_post(metadata, upload_result['url']):
return {
'file': epub_path,
'status': 'success',
'title': metadata['title'],
'url': upload_result['url']
}
else:
return {'file': epub_path, 'status': 'publish_failed'}
def process_directory(self, directory='.'):
"""处理目录中的所有EPUB文件"""
epub_files = glob.glob(os.path.join(directory, "*.epub"))
if not epub_files:
print("在当前目录下未找到EPUB文件")
return []
print(f"找到 {len(epub_files)} 个EPUB文件,开始处理...")
results = []
for epub_path in epub_files:
result = self.process_single_epub(epub_path)
results.append(result)
time.sleep(2)
self.print_summary(results)
return results
def print_summary(self, results):
"""打印处理结果摘要"""
success_count = sum(1 for r in results if r['status'] == 'success')
failed_count = len(results) - success_count
print(f"\n{'='*60}")
print(f"处理完成摘要:")
print(f" ✅ 成功: {success_count} 个文件")
print(f" ❌ 失败: {failed_count} 个文件")
if success_count > 0:
print(f"\n成功发布的文章:")
for result in results:
if result['status'] == 'success':
print(f" - {result['title']}")
if failed_count > 0:
print(f"\n失败的文件:")
for result in results:
if result['status'] != 'success':
status_map = {
'metadata_failed': '元数据提取失败',
'upload_failed': '文件上传失败',
'publish_failed': '文章发布失败'
}
print(f" - {os.path.basename(result['file'])}: {status_map[result['status']]}")
print(f"{'='*60}")
def main():
"""主函数 - 配置和使用发布器"""
# WordPress配置信息
WP_CONFIG = {
'url': 'http://ip地址:端口号', # 替换为你的WordPress地址
'username': '用户名', # 替换为你的用户名
'password': '密码!', # 替换为你的密码或应用密码
}
# 创建发布器实例
publisher = EpubToWordPress(
wp_url=WP_CONFIG['url'],
wp_username=WP_CONFIG['username'],
wp_password=WP_CONFIG['password']
)
print("开始处理EPUB文件...")
results = publisher.process_directory('.')
# 保存处理结果到JSON文件
with open('epub_publishing_results.json', 'w', encoding='utf-8') as f:
json.dump(results, f, ensure_ascii=False, indent=2)
print("\n详细结果已保存到: epub_publishing_results.json")
if __name__ == "__main__":
main()