openclaw-skills/process_video.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
处理B站视频并保存到知识库
"""
import asyncio
import sys
import os
from datetime import datetime

# 添加项目路径
sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'skills', 'obsidian-kb'))

from src.extractors.bilibili_extractor import BilibiliExtractor
from src.outputs.obsidian_writer import ObsidianWriter
from src.ai_client import AIClient

async def process_video(video_url):
    """处理视频并生成 Obsidian 笔记"""

    print(f"正在处理视频: {video_url}")
    print("=" * 60)

    # 1. 提取视频信息
    print("\n步骤 1/3: 提取视频信息...")
    extractor = BilibiliExtractor()
    try:
        result = await extractor._extract_async(video_url)
        print("✅ 视频信息提取成功")
    except Exception as e:
        print(f"❌ 提取失败: {e}")
        import traceback
        traceback.print_exc()
        return

    # 显示提取的信息
    print("\n视频信息:")
    print("-" * 60)
    print(f"标题: {result['metadata']['title']}")
    print(f"UP主: {result['metadata']['author']}")
    print(f"时长: {result['metadata']['duration']}")
    print(f"发布时间: {result['metadata']['pubtime']}")
    print(f"播放量: {result['stats']['view']:,}")
    print(f"点赞: {result['stats']['like']:,}")
    print(f"BV号: {result['metadata']['bvid']}")

    # 显示字幕信息
    subtitle = result['subtitle']
    if subtitle:
        lines = subtitle.split('\n')
        print(f"\n字幕: 找到 {len(lines)} 条字幕")
    else:
        print("\n字幕: 该视频没有官方字幕")

    # 2. 生成 Obsidian 笔记
    print("\n步骤 2/3: 生成 Obsidian 笔记...")

    # 创建笔记内容
    now = datetime.now().strftime('%Y-%m-%d %H:%M:%S')

    # 构建标签
    tags = result['metadata']['tags'].split(', ') if result['metadata']['tags'] else []
    tags_str = ' '.join([f"#{tag}" for tag in tags if tag])

    # 构建笔记
    note_content = f"""---
title: "{result['metadata']['title']}"
source: "bilibili"
url: "{result['metadata']['url']}"
author: "{result['metadata']['author']}"
duration: "{result['metadata']['duration']}"
date: "{result['metadata']['pubtime']}"
bvid: "{result['metadata']['bvid']}"
created: "{now}"
stats:
  view: {result['stats']['view']}
  like: {result['stats']['like']}
  coin: {result['stats']['coin']}
  favorite: {result['stats']['favorite']}
tags:
{chr(10).join([f'  - {tag}' for tag in tags if tag])}
---

{tags_str}

## 视频简介
{result['description'][:500]}{'...' if len(result['description']) > 500 else ''}

## 数据统计
- 播放量: {result['stats']['view']:,}
- 点赞: {result['stats']['like']:,}
- 投币: {result['stats']['coin']:,}
- 收藏: {result['stats']['favorite']:,}
- 分享: {result['stats']['share']:,}

"""

    # 如果有字幕，添加到笔记
    if subtitle:
        note_content += "## 字幕内容\n\n"
        note_content += f"共 {len(subtitle.split(chr(10)))} 条字幕\n\n"

        # 只显示前50条字幕，避免过长
        lines = subtitle.split('\n')
        if len(lines) > 50:
            note_content += '\n'.join(lines[:50])
            note_content += f"\n\n...(还有 {len(lines) - 50} 条字幕，已省略)\n"
        else:
            note_content += subtitle + '\n'

    print("✅ 笔记生成成功")

    # 3. 保存笔记文件
    print("\n步骤 3/3: 保存笔记文件...")

    # 确定保存目录
    kb_dir = os.path.expanduser("~/ObsidianVault")
    if not os.path.exists(kb_dir):
        # 如果用户的主目录没有 ObsidianVault，使用本地目录
        kb_dir = os.path.join(os.path.dirname(__file__), 'kb_output')
        if not os.path.exists(kb_dir):
            os.makedirs(kb_dir)
            print(f"创建输出目录: {kb_dir}")

    # 生成文件名（使用日期和标题）
    date_str = datetime.now().strftime('%Y%m%d')
    safe_title = result['metadata']['title'][:50].replace('/', '-').replace('\\', '-').replace(':', '-')
    filename = f"{date_str}-{safe_title}.md"

    filepath = os.path.join(kb_dir, filename)

    # 写入文件
    with open(filepath, 'w', encoding='utf-8') as f:
        f.write(note_content)

    print(f"✅ 笔记已保存到: {filepath}")

    print("\n" + "=" * 60)
    print("🎉 处理完成！")
    print("=" * 60)

if __name__ == '__main__':
    # 从URL中提取BV号
    import re

    video_url = "https://www.bilibili.com/video/BV1KtFzzqEUG/?spm_id_from=333.1007.tianma.2-1-4.click&vd_source=d82ff6409be4a2d868295c58516354a2"

    # 提取BV号
    bv_match = re.search(r'BV[a-zA-Z0-9]{10}', video_url)
    if bv_match:
        bvid = bv_match.group(0)
        print(f"识别到BV号: {bvid}\n")
        asyncio.run(process_video(bvid))
    else:
        print("❌ 无法从URL中提取BV号")