288 lines
9.4 KiB
Python
288 lines
9.4 KiB
Python
|
#!/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()
|