Delta Lab中国 · 广州

本地笔记到公开博客的无缝衔接

• 20 分钟阅读 •

我希望实现这样的工作流：

  1. 在本地 Notes 文件夹中使用 Obsidian 写作。
  2. 将 Obsidian 笔记提交至 GitHub 私有仓库 Notes。
  3. 当笔记的 YAML 中 draft: false 时，自动同步到另一个私有仓库 Blogs。
  4. 通过 Cloudflare Pages 抓取 Blogs 仓库，使用 Hugo 构建并部署到个人域名 linqy.tech。

最终效果：
只需将本地笔记中的 draft 字段改为 false，即可自动发布为博客文章，全程无需手动操作。

一、选择并魔改博客模板

1. 模板来源

2. 获取源码

git clone https://github.com/koobai/blog.git

3. 安装 Hugo

从官方发布页下载最新版 Hugo（如 v0.148.2）：
🔗 Hugo Releases

在项目根目录运行：

hugo server

即可本地预览博客效果。

二、个性化模板修改

使用 AI 工具（如 Cursor、Tare）辅助修改主题，主要调整如下：

  1. 评论系统：集成 Giscus，需使用一个公开的 GitHub 仓库存储评论。
  2. “唠里唠叨”栏目：用博文样式重构，保持风格统一。
  3. 增加文章置顶功能：通过 isTop: true 控制。
  4. 美化渲染效果：优化表格、数学公式、代码块的显示样式。
  5. 清理无用功能：仅保留 douban 相关功能。

✅ 建议边开发边预览，逐步迭代至满意为止。

三、部署博客到 Cloudflare Pages

1. 创建私有仓库 Blogs

将修改后的博客代码推送到 GitHub 私有仓库。

2. 触发 GitHub Action

首次推送后，检查是否生成了 csv 数据文件（如豆瓣导出数据）。
若失败，可先回退到初始版本测试，成功后再逐步替换为自己的 ID。

3. 配置 Cloudflare Workers & Pages

进入 Cloudflare Dashboard，创建 Pages 项目：

构建配置

配置项 内容
构建命令 hugo --gc --minify
部署命令 npx wrangler deploy
版本命令 npx wrangler versions upload
根目录 /

构建成功后，可通过 blog.linqiyuan2021.workers.dev 访问。

4. 绑定自定义域名

  1. 在阿里云购买域名（如 linqy.tech）
  2. 将域名的 DNS 解析服务器改为 Cloudflare 提供的 NS 地址
  3. 在 Cloudflare 中删除阿里云生成的 DNS 记录
  4. 添加 CNAME 记录，将 www 和 @ 指向 .workers.dev 地址

✅ 配置完成后，每次 Blogs 仓库更新，网站将自动重新构建并上线。

四、从 Obsidian 到博客的自动化同步

1. 自动化工具链

工具 作用
Templater 新建笔记时自动生成 YAML 头部
Linter 保存时自动格式化 Markdown
QuickAdd 调用外部脚本，注入 AI 生成内容
Obsidian Git 提交笔记到私有仓库 Notes
Git Hook + Python 脚本 监听 draft 状态，自动同步到 Blogs

2. 使用 AI 自动生成 YAML 元信息

通过 QuickAdd 调用 Python 脚本，利用 DeepSeek API 自动生成：

✅ QuickAdd JS 脚本

// QuickAdd 脚本：AI生成YAML前置内容
module.exports = async (params) => {
    const { quickAddApi: { inputPrompt }, app } = params;
    
    const activeFile = app.workspace.getActiveFile();
    if (!activeFile || activeFile.extension !== 'md') {
        new Notice("请打开一个 Markdown 文件");
        return;
    }
    
    const filePath = activeFile.path;
    const fullPath = app.vault.adapter.path.join(app.vault.adapter.basePath, filePath);
    
    new Notice("正在调用AI生成YAML内容...", 3000);
    
    try {
        const { exec } = require('child_process');
        const pythonScript = app.vault.adapter.path.join(app.vault.adapter.basePath, 'ai_yaml_generator.py');
        const command = `python "${pythonScript}" "${fullPath}"`;
        
        exec(command, (error, stdout, stderr) => {
            if (error) {
                new Notice(`执行失败: ${error.message}`);
                return;
            }
            if (stderr) new Notice(`警告: ${stderr}`);
            
            app.vault.adapter.read(filePath).then(() => {
                new Notice("YAML内容已更新！");
                app.workspace.getActiveViewOfType(MarkdownView)?.previewMode.rerender(true);
            });
        });
    } catch (err) {
        new Notice(`脚本执行失败: ${err.message}`);
    }
};

✅ Python AI 脚本（ai_yaml_generator.py）

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
使用 DeepSeek API 自动生成 slug 和 description
"""

import os
import sys
import re
import requests
import json
from datetime import datetime

API_KEY = "sk-xxx"
API_URL = "https://api.deepseek.com/v1/chat/completions"

def read_file_content(file_path):
    try:
        with open(file_path, 'r', encoding='utf-8') as f:
            return f.read()
    except Exception as e:
        print(f"读取文件失败: {e}")
        return None

def extract_yaml_and_content(content):
    yaml_pattern = r'^---\s*\n(.*?)\n---\s*\n(.*)$'
    match = re.match(yaml_pattern, content, re.DOTALL)
    return (match.group(1), match.group(2)) if match else ("", content)

def call_deepseek_api(content):
    prompt = f"""
请为以下文档生成：
1. 一个英文 slug（小写连字符分隔）
2. 一段50-100字的中文摘要（以作者口吻）

内容：
{content[:2000]}

请以 JSON 格式返回：
{{"slug": "title-here", "description": "摘要内容"}}
"""

    headers = {"Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json"}
    data = {
        "model": "deepseek-chat",
        "messages": [{"role": "user", "content": prompt}],
        "temperature": 0.7,
        "max_tokens": 500
    }

    try:
        response = requests.post(API_URL, headers=headers, json=data, timeout=30)
        response.raise_for_status()
        ai_response = response.json()['choices'][0]['message']['content']
        
        # 提取 JSON
        json_match = re.search(r'\{[^}]*"slug"[^}]*\}', ai_response)
        if json_match:
            return json.loads(json_match.group(0))
        else:
            slug = re.search(r'"slug"\s*:\s*"([^"]+)"', ai_response)
            desc = re.search(r'"description"\s*:\s*"([^"]+)"', ai_response)
            if slug and desc:
                return {"slug": slug.group(1), "description": desc.group(1)}
    except Exception as e:
        print(f"API 调用失败: {e}")
        return None

def update_yaml_frontmatter(yaml_content, ai_result):
    yaml_dict = {}
    if yaml_content:
        for line in yaml_content.split('\n'):
            if ':' in line:
                k, v = line.split(':', 1)
                yaml_dict[k.strip()] = v.strip().strip('"')

    yaml_dict.update({
        'slug': ai_result['slug'],
        'description': ai_result['description'],
        'authors': 'kean',
        'draft': 'false',
        'hideInList': 'false',
        'isTop': 'false',
        'toc': 'true',
        'modified': datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    })

    if 'date' not in yaml_dict:
        yaml_dict['date'] = yaml_dict['modified']

    return '\n'.join(f"{k}: {v}" for k, v in yaml_dict.items())

def write_file_content(file_path, yaml_content, main_content):
    try:
        with open(file_path, 'w', encoding='utf-8') as f:
            f.write(f"---\n{yaml_content}\n---\n{main_content}")
        return True
    except Exception as e:
        print(f"写入失败: {e}")
        return False

def main():
    if len(sys.argv) < 2:
        print("请提供文件路径")
        return

    file_path = sys.argv[1]
    if not os.path.exists(file_path):
        print(f"文件不存在: {file_path}")
        return

    content = read_file_content(file_path)
    if not content: return

    yaml_content, main_content = extract_yaml_and_content(content)
    ai_result = call_deepseek_api(main_content)

    if not ai_result: return

    new_yaml = update_yaml_frontmatter(yaml_content, ai_result)
    if write_file_content(file_path, new_yaml, main_content):
        print("✅ YAML 更新成功")

if __name__ == "__main__":
    main()

3. Git Hook 同步机制

通过 pre-commit 钩子，在提交 Notes 仓库时自动同步 draft: false 的文章到 Blogs。

核心逻辑

✅ Python 同步脚本（pre-commit-hook.py）

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import os
import sys
import subprocess
import logging
import yaml
from pathlib import Path

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
logger = logging.getLogger(__name__)

def extract_yaml_front_matter(content: str) -> dict:
    if not content.startswith('---'): return {}
    lines = content.split('\n')
    end_index = next((i for i, line in enumerate(lines[1:], 1) if line.strip() == '---'), -1)
    if end_index == -1: return {}
    try:
        return yaml.safe_load('\n'.join(lines[1:end_index])) or {}
    except yaml.YAMLError:
        return {}

def get_staged_files():
    try:
        result = subprocess.run(
            ["git", "diff", "--cached", "--name-only", "--diff-filter=ACM"],
            capture_output=True, text=True, check=True, encoding='utf-8', errors='ignore'
        )
        return [Path(f) for f in result.stdout.strip().split('\n') if f.endswith('.md')]
    except subprocess.CalledProcessError:
        return []

def get_file_draft_status(file_path: Path):
    current_draft = previous_draft = None
    try:
        with open(file_path, 'r', encoding='utf-8') as f:
            yaml_data = extract_yaml_front_matter(f.read())
        current_draft = yaml_data.get('draft')
    except Exception as e:
        logger.warning(f"读取失败 {file_path}: {e}")

    try:
        result = subprocess.run(
            ["git", "show", f"HEAD:{file_path}"],
            capture_output=True, text=True, encoding='utf-8', errors='ignore'
        )
        if result.returncode == 0:
            yaml_data = extract_yaml_front_matter(result.stdout)
            previous_draft = yaml_data.get('draft')
    except Exception:
        previous_draft = None

    return current_draft, previous_draft

def get_target_folder(filepath: Path) -> str:
    mapping = {
        '01_表征处理': 'characterization',
        '02_材料合成': 'synthesis',
        '03_软件代码': 'programming',
        '04_文献笔记': 'literature',
        'AI': 'artificial_intelligence',
        '06 Memos': 'memos'
    }
    for part in filepath.parts:
        if part in mapping:
            return mapping[part]
    if any(kw in str(filepath) for kw in ['表征', 'XRD', 'SEM', 'XPS']): return 'characterization'
    if any(kw in str(filepath) for kw in ['合成', '制备']): return 'synthesis'
    if any(kw in str(filepath) for kw in ['代码', '软件', '编程']): return 'programming'
    if 'AI' in str(filepath): return 'artificial_intelligence'
    if 'Memos' in str(filepath): return 'memos'
    return 'literature'

def copy_file_to_blog(source_file: Path, blog_dir: Path) -> bool:
    target_folder = get_target_folder(source_file)
    target_dir = (blog_dir.parent / 'memos') if target_folder == 'memos' else (blog_dir / target_folder)
    target_dir.mkdir(parents=True, exist_ok=True)
    try:
        import shutil
        shutil.copy2(source_file, target_dir / source_file.name)
        logger.info(f"✅ 复制: {source_file.name} -> {target_folder}/")
        return True
    except Exception as e:
        logger.error(f"❌ 复制失败: {e}")
        return False

def delete_file_from_blog(source_file: Path, blog_dir: Path) -> bool:
    target_folder = get_target_folder(source_file)
    target_file = (blog_dir.parent / 'memos' / source_file.name) if target_folder == 'memos' else (blog_dir / target_folder / source_file.name)
    if target_file.exists():
        target_file.unlink()
        logger.info(f"🗑️ 删除: {source_file.name}")
    return True

def analyze_draft_changes():
    files_to_copy, files_to_delete = [], []
    for file_path in get_staged_files():
        curr, prev = get_file_draft_status(file_path)
        if curr is None: continue

        if prev is None and curr == False:
            files_to_copy.append(file_path)
        elif prev == True and curr == False:
            files_to_copy.append(file_path)
        elif prev == False and curr == True:
            files_to_delete.append(file_path)
    return files_to_copy, files_to_delete

def sync_draft_changes():
    blog_dir = Path(r"E:\Sync Drive\Blogs\Work\blog\content\posts")
    files_to_copy, files_to_delete = analyze_draft_changes()

    if not files_to_copy and not files_to_delete:
        logger.info("✅ 无变化")
        return True

    for f in files_to_copy: copy_file_to_blog(f, blog_dir)
    for f in files_to_delete: delete_file_from_blog(f, blog_dir)
    return True

def sync_to_blog():
    blog_root = Path(r"E:\Sync Drive\Blogs\Work\blog")
    os.chdir(blog_root)
    
    status = subprocess.run(["git", "status", "--porcelain"], capture_output=True, text=True)
    if not status.stdout.strip():
        logger.info("✅ 博客无变更")
        return True

    subprocess.run(["git", "add", "."])
    commit_msg = f"Auto-sync: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"
    subprocess.run(["git", "commit", "-m", commit_msg])
    subprocess.run(["git", "push", "origin", "main"])
    logger.info("✅ 博客同步完成")
    return True

def main():
    if not sync_draft_changes():
        sys.exit(1)
    if not sync_to_blog():
        logger.error("⚠️ 博客同步失败，但笔记提交继续")
    logger.info("🚀 全部完成")

if __name__ == "__main__":
    main()

五、总结

这套系统实现了：

✅ 完全自动化发布：改 draft: false 即发布
✅ AI 辅助写作：自动生成 slug 与摘要
✅ 双向隔离：笔记私密，博客公开
✅ 实时部署：GitHub → Cloudflare → 自定义域名

只需专注写作，剩下的交给系统。

文章标签:

上一篇 : Zotero Short Title Generator
下一篇 : AI 管家贾维斯
留言