Obsidian上传到语雀

import os
import requests
import frontmatter
from PyInquirer import prompt

class ObsidianToYuque:
    def __init__(self, obsidian_dir, yuque_token, yuque_namespace):
        self.obsidian_dir = obsidian_dir
        self.yuque_token = yuque_token
        self.yuque_namespace = yuque_namespace
        self.headers = {
            "Content-Type": "application/json",
            "X-Auth-Token": self.yuque_token
        }
        self.api_base_url = "https://www.yuque.com/api/v2"

    def get_all_md_files(self):
        """遍历 Obsidian 文件夹获取所有 Markdown 文件路径"""
        md_files = []
        for root, _, files in os.walk(self.obsidian_dir):
            for file in files:
                if file.endswith(".md"):
                    md_files.append(os.path.join(root, file))
        return md_files

    def create_repo(self, repo_name):
        """创建语雀知识库"""
        url = f"{self.api_base_url}/users/{self.yuque_namespace}/repos"
        payload = {
            "name": repo_name,
            "slug": repo_name.lower().replace(" ", "-"),
            "type": "Book",
            "description": f"Imported from Obsidian: {repo_name}",
            "public": 0  # 0为私密，1为公开
        }
        response = requests.post(url, json=payload, headers=self.headers)
        if response.status_code == 200:
            print(f"知识库 '{repo_name}' 创建成功")
            return response.json()["data"]["id"]
        elif response.status_code == 400 and "已存在" in response.text:
            print(f"知识库 '{repo_name}' 已存在，继续使用")
            return None  # 已存在则无需新建
        else:
            print(f"知识库创建失败: {response.text}")
            return None

    def upload_doc(self, repo_slug, title, body):
        """上传文档到语雀"""
        url = f"{self.api_base_url}/repos/{self.yuque_namespace}/{repo_slug}/docs"
        payload = {
            "title": title,
            "slug": title.lower().replace(" ", "-"),
            "body": body
        }
        response = requests.post(url, json=payload, headers=self.headers)
        if response.status_code == 200:
            print(f"文档 '{title}' 上传成功")
        else:
            print(f"文档 '{title}' 上传失败: {response.text}")

    def process_files(self):
        """处理所有 Markdown 文件"""
        md_files = self.get_all_md_files()
        repo_name = os.path.basename(self.obsidian_dir)
        repo_slug = repo_name.lower().replace(" ", "-")
        
        # 创建知识库（如果不存在）
        self.create_repo(repo_name)

        # 上传每个 Markdown 文件
        for file_path in md_files:
            with open(file_path, "r", encoding="utf-8") as f:
                content = f.read()
                metadata, body = frontmatter.parse(content)
                title = metadata.get("title", os.path.basename(file_path).replace(".md", ""))
                self.upload_doc(repo_slug, title, body)

def main():
    # 用户输入部分
    questions = [
        {
            "type": "input",
            "name": "obsidian_dir",
            "message": "请输入 Obsidian 文件夹路径："
        },
        {
            "type": "input",
            "name": "yuque_token",
            "message": "请输入语雀 API Token："
        },
        {
            "type": "input",
            "name": "yuque_namespace",
            "message": "请输入语雀 Namespace (个人或团队空间标识)："
        }
    ]
    answers = prompt(questions)

    obsidian_dir = answers["obsidian_dir"]
    yuque_token = answers["yuque_token"]
    yuque_namespace = answers["yuque_namespace"]

    if not os.path.exists(obsidian_dir):
        print("Obsidian 文件夹路径不存在，请检查后重试！")
        return

    # 执行上传
    uploader = ObsidianToYuque(obsidian_dir, yuque_token, yuque_namespace)
    uploader.process_files()

if __name__ == "__main__":
    main()