288 lines
9.4 KiB
Python
Executable File
288 lines
9.4 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
"""
|
||
飞书小说同步工具
|
||
将本地章节同步到飞书文档
|
||
"""
|
||
|
||
import os
|
||
import json
|
||
import time
|
||
from datetime import datetime
|
||
import requests
|
||
|
||
class FeishuNovelSync:
|
||
def __init__(self, config_path="configs/feishu_config.json"):
|
||
"""初始化同步工具"""
|
||
self.config = self.load_config(config_path)
|
||
self.base_url = "https://open.feishu.cn/open-apis"
|
||
self.headers = {
|
||
"Authorization": f"Bearer {self.config.get('access_token')}",
|
||
"Content-Type": "application/json"
|
||
}
|
||
|
||
def load_config(self, config_path):
|
||
"""加载配置文件"""
|
||
if os.path.exists(config_path):
|
||
with open(config_path, 'r', encoding='utf-8') as f:
|
||
return json.load(f)
|
||
else:
|
||
# 默认配置
|
||
return {
|
||
"app_id": "",
|
||
"app_secret": "",
|
||
"access_token": "",
|
||
"refresh_token": "",
|
||
"expire_time": 0
|
||
}
|
||
|
||
def save_config(self, config_path="configs/feishu_config.json"):
|
||
"""保存配置文件"""
|
||
os.makedirs(os.path.dirname(config_path), exist_ok=True)
|
||
with open(config_path, 'w', encoding='utf-8') as f:
|
||
json.dump(self.config, f, ensure_ascii=False, indent=2)
|
||
|
||
def get_access_token(self):
|
||
"""获取访问令牌"""
|
||
if self.config.get('expire_time', 0) > time.time():
|
||
return self.config.get('access_token')
|
||
|
||
url = f"{self.base_url}/auth/v3/tenant_access_token/internal"
|
||
data = {
|
||
"app_id": self.config['app_id'],
|
||
"app_secret": self.config['app_secret']
|
||
}
|
||
|
||
try:
|
||
response = requests.post(url, json=data)
|
||
result = response.json()
|
||
|
||
if result.get('code') == 0:
|
||
self.config['access_token'] = result['tenant_access_token']
|
||
self.config['expire_time'] = time.time() + result['expire'] - 60
|
||
self.save_config()
|
||
return self.config['access_token']
|
||
else:
|
||
print(f"获取token失败: {result}")
|
||
return None
|
||
except Exception as e:
|
||
print(f"请求失败: {e}")
|
||
return None
|
||
|
||
def create_doc(self, title, folder_token=None):
|
||
"""创建飞书文档"""
|
||
token = self.get_access_token()
|
||
if not token:
|
||
return None
|
||
|
||
url = f"{self.base_url}/drive/v1/files/create"
|
||
headers = self.headers.copy()
|
||
headers["Authorization"] = f"Bearer {token}"
|
||
|
||
data = {
|
||
"title": title,
|
||
"type": "doc",
|
||
"folder_token": folder_token or ""
|
||
}
|
||
|
||
try:
|
||
response = requests.post(url, headers=headers, json=data)
|
||
result = response.json()
|
||
|
||
if result.get('code') == 0:
|
||
doc_token = result['data']['token']
|
||
print(f"文档创建成功: {title} (token: {doc_token})")
|
||
return doc_token
|
||
else:
|
||
print(f"创建文档失败: {result}")
|
||
return None
|
||
except Exception as e:
|
||
print(f"创建文档请求失败: {e}")
|
||
return None
|
||
|
||
def update_doc(self, doc_token, content):
|
||
"""更新文档内容"""
|
||
token = self.get_access_token()
|
||
if not token:
|
||
return False
|
||
|
||
url = f"{self.base_url}/drive/v1/files/{doc_token}/content"
|
||
headers = self.headers.copy()
|
||
headers["Authorization"] = f"Bearer {token}"
|
||
|
||
# 构建文档内容
|
||
doc_content = {
|
||
"title": "小说章节",
|
||
"body": {
|
||
"blocks": [
|
||
{
|
||
"type": "paragraph",
|
||
"paragraph": {
|
||
"elements": [
|
||
{
|
||
"type": "text",
|
||
"text": content
|
||
}
|
||
]
|
||
}
|
||
}
|
||
]
|
||
}
|
||
}
|
||
|
||
try:
|
||
response = requests.put(url, headers=headers, json=doc_content)
|
||
result = response.json()
|
||
|
||
if result.get('code') == 0:
|
||
print(f"文档更新成功: {doc_token}")
|
||
return True
|
||
else:
|
||
print(f"更新文档失败: {result}")
|
||
return False
|
||
except Exception as e:
|
||
print(f"更新文档请求失败: {e}")
|
||
return False
|
||
|
||
def sync_chapter(self, chapter_path, doc_token=None, title=None):
|
||
"""同步单个章节"""
|
||
if not os.path.exists(chapter_path):
|
||
print(f"章节文件不存在: {chapter_path}")
|
||
return None
|
||
|
||
# 读取章节内容
|
||
with open(chapter_path, 'r', encoding='utf-8') as f:
|
||
content = f.read()
|
||
|
||
# 获取章节标题
|
||
if not title:
|
||
title = os.path.basename(chapter_path).replace('.md', '')
|
||
|
||
# 如果没有文档token，先创建文档
|
||
if not doc_token:
|
||
doc_token = self.create_doc(title)
|
||
if not doc_token:
|
||
return None
|
||
|
||
# 更新文档内容
|
||
if self.update_doc(doc_token, content):
|
||
return doc_token
|
||
else:
|
||
return None
|
||
|
||
def sync_project(self, project_path, folder_token=None):
|
||
"""同步整个项目"""
|
||
if not os.path.exists(project_path):
|
||
print(f"项目路径不存在: {project_path}")
|
||
return False
|
||
|
||
chapters_dir = os.path.join(project_path, "chapters")
|
||
if not os.path.exists(chapters_dir):
|
||
print(f"章节目录不存在: {chapters_dir}")
|
||
return False
|
||
|
||
# 获取所有章节文件
|
||
chapter_files = []
|
||
for root, dirs, files in os.walk(chapters_dir):
|
||
for file in files:
|
||
if file.endswith('.md'):
|
||
chapter_files.append(os.path.join(root, file))
|
||
|
||
if not chapter_files:
|
||
print("没有找到章节文件")
|
||
return False
|
||
|
||
print(f"找到 {len(chapter_files)} 个章节文件")
|
||
|
||
# 按文件名排序
|
||
chapter_files.sort()
|
||
|
||
# 同步每个章节
|
||
results = []
|
||
for i, chapter_file in enumerate(chapter_files, 1):
|
||
print(f"[{i}/{len(chapter_files)}] 同步: {os.path.basename(chapter_file)}")
|
||
|
||
chapter_title = os.path.basename(chapter_file).replace('.md', '')
|
||
doc_token = self.sync_chapter(chapter_file, None, chapter_title)
|
||
|
||
if doc_token:
|
||
results.append({
|
||
"file": chapter_file,
|
||
"title": chapter_title,
|
||
"doc_token": doc_token,
|
||
"status": "success"
|
||
})
|
||
print(f" 成功: https://example.feishu.cn/docx/{doc_token}")
|
||
else:
|
||
results.append({
|
||
"file": chapter_file,
|
||
"title": chapter_title,
|
||
"doc_token": None,
|
||
"status": "failed"
|
||
})
|
||
print(f" 失败")
|
||
|
||
# 避免请求过快
|
||
time.sleep(1)
|
||
|
||
# 保存同步结果
|
||
sync_log = {
|
||
"project": project_path,
|
||
"sync_time": datetime.now().isoformat(),
|
||
"total_chapters": len(chapter_files),
|
||
"success": len([r for r in results if r['status'] == 'success']),
|
||
"failed": len([r for r in results if r['status'] == 'failed']),
|
||
"details": results
|
||
}
|
||
|
||
log_file = os.path.join(project_path, "sync", "sync_log.json")
|
||
os.makedirs(os.path.dirname(log_file), exist_ok=True)
|
||
|
||
with open(log_file, 'w', encoding='utf-8') as f:
|
||
json.dump(sync_log, f, ensure_ascii=False, indent=2)
|
||
|
||
print(f"\n同步完成:")
|
||
print(f" 总章节: {len(chapter_files)}")
|
||
print(f" 成功: {sync_log['success']}")
|
||
print(f" 失败: {sync_log['failed']}")
|
||
print(f" 日志: {log_file}")
|
||
|
||
return sync_log['success'] > 0
|
||
|
||
def main():
|
||
"""主函数"""
|
||
import argparse
|
||
|
||
parser = argparse.ArgumentParser(description="飞书小说同步工具")
|
||
parser.add_argument("--project", help="项目路径", default=".")
|
||
parser.add_argument("--config", help="配置文件路径", default="configs/feishu_config.json")
|
||
parser.add_argument("--chapter", help="单个章节文件路径")
|
||
parser.add_argument("--init", action="store_true", help="初始化配置")
|
||
|
||
args = parser.parse_args()
|
||
|
||
sync_tool = FeishuNovelSync(args.config)
|
||
|
||
if args.init:
|
||
# 初始化配置
|
||
app_id = input("请输入 App ID: ")
|
||
app_secret = input("请输入 App Secret: ")
|
||
|
||
sync_tool.config.update({
|
||
"app_id": app_id,
|
||
"app_secret": app_secret
|
||
})
|
||
sync_tool.save_config()
|
||
|
||
print("配置已保存")
|
||
return
|
||
|
||
if args.chapter:
|
||
# 同步单个章节
|
||
sync_tool.sync_chapter(args.chapter)
|
||
else:
|
||
# 同步整个项目
|
||
sync_tool.sync_project(args.project)
|
||
|
||
if __name__ == "__main__":
|
||
main()
|