openclaw-skills/skills/kb-workflow/kb_handler.py

#!/usr/bin/env python3
"""
Auto-trigger handler for obsidian-kb skill.

Detects video/article links in user messages and automatically processes them.
Supports Bilibili, YouTube, and Zhihu.

## 触发词
- "保存到知识库" + 链接 → 添加到收件箱并处理
- "订阅到知识库" + 链接 → 添加订阅
"""

import re
import sys
import os
from typing import List, Dict, Optional, Tuple
from urllib.parse import urlparse

# Add path for imports
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), 'src'))

# Import short link resolver
from shortlink_resolver import (
    ShortLinkResolver,
    detect_and_format,
    PlaywrightResolver
)

# 导入主流程
try:
    from main_pipeline import MainPipeline
    MAIN_PIPELINE_AVAILABLE = True
except ImportError:
    MAIN_PIPELINE_AVAILABLE = False

# Pattern definitions for auto-detection
PATTERNS = {
    'bilibili_full': re.compile(r'https?://(?:www\.)?bilibili\.com/video/(BV\w+)'),
    'bilibili_short': re.compile(r'https?://b23\.tv/(\w+)'),
    'bilibili_bv': re.compile(r'\b(BV[a-zA-Z0-9]{10})\b'),
    'youtube_full': re.compile(r'https?://(?:www\.)?youtube\.com/watch\?v=([a-zA-Z0-9_-]{11})'),
    'youtube_short': re.compile(r'https?://youtu\.be/([a-zA-Z0-9_-]{11})'),
    'zhihu_article': re.compile(r'https?://zhuanlan\.zhihu\.com/p/(\d+)'),
    'zhihu_answer': re.compile(r'https?://(?:www\.)?zhihu\.com/question/\d+/answer/(\d+)'),
}

# 触发词模式
TRIGGER_PATTERNS = {
    'save': re.compile(r'(保存到知识库|加入知识库|添加到知识库)', re.IGNORECASE),
    'subscribe': re.compile(r'(订阅到知识库|订阅知识库|关注到知识库)', re.IGNORECASE),
}


def detect_links(text: str) -> List[Dict[str, str]]:
    """
    Detect video/article links in text.

    Args:
        text: Input text to search

    Returns:
        List of detected links with metadata:
        [
            {
                'url': 'https://...',
                'platform': 'bilibili',
                'type': 'video',
                'id': 'BVxxx',
                'is_short': False
            },
            ...
        ]
    """
    detected = []
    seen_ids = set()  # Track IDs to avoid duplicates

    for pattern_name, pattern in PATTERNS.items():
        matches = pattern.finditer(text)

        for match in matches:
            # Determine platform and type
            if 'bilibili' in pattern_name:
                platform = 'bilibili'
                content_type = 'video'
                if 'short' in pattern_name:
                    is_short = True
                    url = match.group(0)
                    content_id = None  # Need to resolve short URL
                else:
                    is_short = False
                    if pattern_name == 'bilibili_bv':
                        url = f"https://www.bilibili.com/video/{match.group(1)}"
                    else:
                        url = match.group(0)
                    content_id = match.group(1)

            elif 'youtube' in pattern_name:
                platform = 'youtube'
                content_type = 'video'
                is_short = 'short' in pattern_name
                url = match.group(0)
                content_id = match.group(1)

            elif 'zhihu' in pattern_name:
                platform = 'zhihu'
                content_type = 'article'
                is_short = False
                url = match.group(0)
                content_id = match.group(1)

            else:
                continue

            # Skip duplicates by URL or ID
            unique_key = url if (not content_id or is_short) else content_id
            if unique_key in seen_ids:
                continue
            seen_ids.add(unique_key)

            detected.append({
                'url': url,
                'platform': platform,
                'type': content_type,
                'id': content_id,
                'is_short': is_short,
                'pattern': pattern_name
            })

    return detected


def format_preview(detected: List[Dict[str, str]]) -> str:
    """
    Format a preview message for detected links.

    Args:
        detected: List of detected links

    Returns:
        Formatted preview message
    """
    if not detected:
        return ""

    lines = ["🔍 检测到内容链接:\n"]

    for i, item in enumerate(detected, 1):
        platform_emoji = {
            'bilibili': '📺',
            'youtube': '▶️',
            'zhihu': '📝'
        }.get(item['platform'], '🔗')

        platform_name = {
            'bilibili': 'B站视频',
            'youtube': 'YouTube视频',
            'zhihu': '知乎文章'
        }.get(item['platform'], item['platform'])

        lines.append(f"{platform_emoji} [{i}] {platform_name}")
        lines.append(f"    URL: {item['url']}")

        if item['is_short']:
            lines.append(f"    ⚠️  短链接，需要用户提供完整URL或ID")
        elif item['id']:
            lines.append(f"    ID: {item['id']}")

        lines.append("")

    return "\n".join(lines)


def format_short_link_help(short_link: Optional[str] = None) -> str:
    """
    Format help message for short links.

    Args:
        short_link: Optional short link URL to mention

    Returns:
        Help message
    """
    return ShortLinkResolver.format_short_link_help(short_link)


def get_action_prompt(detected: List[Dict[str, str]]) -> str:
    """
    Generate prompt asking user what to do.

    Args:
        detected: List of detected links

    Returns:
        Action prompt
    """
    has_short = any(item['is_short'] for item in detected)

    if has_short:
        return f"{format_short_link_help()}\n\n其他链接是否立即处理？(y/n)"

    count = len(detected)
    if count == 1:
        return "是否添加到收件箱并处理？(y/n)"
    else:
        return f"是否将上述 {count} 个内容添加到收件箱？(y/n)"


def detect_triggers(text: str) -> Optional[str]:
    """
    检测用户消息中的触发词。

    Args:
        text: 用户消息

    Returns:
        'save' | 'subscribe' | None
    """
    # 检测"保存到知识库"
    if TRIGGER_PATTERNS['save'].search(text):
        return 'save'

    # 检测"订阅到知识库"
    if TRIGGER_PATTERNS['subscribe'].search(text):
        return 'subscribe'

    return None


def handle_save_to_kb(detected: List[Dict[str, str]]) -> str:
    """
    处理"保存到知识库"命令：将链接添加到收件箱并处理。

    Args:
        detected: 检测到的链接列表

    Returns:
        处理结果消息
    """
    if not detected:
        return "❌ 未检测到任何链接，请提供要保存的链接"

    results = []
    for item in detected:
        if item['is_short']:
            results.append(f"⚠️  短链接无法处理: {item['url']}")
            continue

        # 调用保存处理
        success, msg = process_link(item['url'], item['platform'], auto_process=True)
        results.append(msg)

    return "\n".join(results)


def handle_subscribe_to_kb(detected: List[Dict[str, str]]) -> str:
    """
    处理"订阅到知识库"命令：将链接添加到订阅列表。

    Args:
        detected: 检测到的链接列表

    Returns:
        处理结果消息
    """
    if not detected:
        return "❌ 未检测到任何链接，请提供要订阅的链接"

    results = []
    for item in detected:
        if item['platform'] != 'bilibili':
            results.append(f"⚠️  暂不支持订阅 {item['platform']}: {item['url']}")
            continue

        if item['is_short']:
            results.append(f"⚠️  短链接无法订阅: {item['url']}")
            continue

        try:
            # 调用订阅命令
            import subprocess
            cmd = [
                sys.executable,
                os.path.join(os.path.dirname(__file__), 'kb_workflow.py'),
                '--trigger', 'subscriptions',
                '--add', 'bilibili', 'video', item['id'], item.get('url', '')
            ]
            result = subprocess.run(cmd, capture_output=True, text=True)
            if result.returncode == 0:
                results.append(f"✅ 已订阅: {item['id']}")
            else:
                results.append(f"❌ 订阅失败: {result.stderr}")
        except Exception as e:
            results.append(f"❌ 订阅失败: {e}")

    return "\n".join(results)


def should_auto_process(text: str) -> bool:
    """
    Check if user wants to auto-process based on their message.

    Args:
        text: User's message

    Returns:
        True if user indicates they want immediate processing
    """
    auto_keywords = ['立即处理', '处理', 'process', '--process', '马上']
    return any(keyword in text.lower() for keyword in auto_keywords)


def process_link(url: str, platform: str, auto_process: bool = False) -> Tuple[bool, str]:
    """
    Process a single link (extract metadata, add to inbox).

    Args:
        url: Content URL
        platform: Platform name (bilibili/youtube/zhihu)
        auto_process: Whether to immediately process after adding

    Returns:
        (success, message) tuple
    """
    try:
        # Import here to avoid issues if dependencies not available
        from sources.bilibili import BilibiliAdapter
        from models import Content, ContentStatus
        from storage import Storage

        # Only support Bilibili for now
        if platform != 'bilibili':
            return False, f"暂不支持 {platform}，请手动添加"

        # Handle short links
        if 'b23.tv' in url:
            return False, "短链接无法自动处理，请提供完整URL或BV号"

        # Create adapter
        adapter = BilibiliAdapter()

        # Extract metadata
        content = adapter.extract_content(url)
        if not content:
            return False, "无法提取视频信息"

        # Get transcript
        transcript = adapter.get_transcript(content)
        if transcript:
            content.transcript = transcript

        # Save to inbox
        content.status = ContentStatus.CONFIRMED
        storage = Storage()
        storage.save_content(content)

        # Build success message
        msg = f"✅ 已添加到收件箱\n"
        msg += f"   标题: {content.title}\n"
        msg += f"   UP主: {content.author}\n"
        msg += f"   时长: {content.duration}秒\n"

        if auto_process:
            # Process immediately
            try:
                from src.pipeline import ProcessingPipeline
                pipeline = ProcessingPipeline()
                content_text = content.transcript or content.summary or ""
                result = pipeline.process_full(content, content_text)
                storage.save_content(result)

                msg += f"\n🔄 处理完成\n"
                msg += f"   📄 笔记: {result.output_path or 'N/A'}\n"
                msg += f"   🏷️  标签: {', '.join(result.tags)}"
            except Exception as e:
                msg += f"\n⚠️  处理失败: {e}"

        return True, msg

    except ImportError as e:
        return False, f"依赖模块未安装: {e}"
    except Exception as e:
        return False, f"处理失败: {e}"


def process_video_full(source_id: str, enable_git_sync: bool = False) -> Tuple[bool, str]:
    """
    使用主流程完整处理视频（新接口）

    完整流程: 提取 → 处理 → 输出 → 同步

    Args:
        source_id: 视频ID（支持BV号、URL）
        enable_git_sync: 是否启用Git同步

    Returns:
        (success, message) tuple
    """
    if not MAIN_PIPELINE_AVAILABLE:
        return False, "主流程模块未可用，请检查依赖"

    try:
        # 导入配置
        from src.config import get_config

        # 获取配置
        config = get_config()

        # 创建主流程
        pipeline = MainPipeline(config=config, enable_git_sync=enable_git_sync)

        # 处理视频
        result = pipeline.process_video(source_id)

        # 返回结果
        if result['success']:
            return True, result['message']
        else:
            return False, result.get('error', 'Unknown error')

    except Exception as e:
        return False, f"处理失败: {e}"


def handle_message(text: str) -> Optional[str]:
    """
    Main entry point: handle incoming message and return response.

    ## 触发条件
    - 检测到"保存到知识库"/"订阅到知识库"触发词 + 链接 → 自动处理
    - 仅检测到链接 → 询问用户是否处理

    Args:
        text: User's message text

    Returns:
        Response message, or None if no action needed
    """
    # 首先检测触发词
    trigger = detect_triggers(text)

    # 检测链接
    detected = detect_links(text)

    # 如果没有链接，也没有触发词，返回 None
    if not detected and not trigger:
        return None

    # 如果有触发词（save/subscribe），但没有链接
    if trigger and not detected:
        if trigger == 'save':
            return "请提供要保存到知识库的链接"
        elif trigger == 'subscribe':
            return "请提供要订阅到知识库的链接"

    # 处理"保存到知识库"触发
    if trigger == 'save':
        return handle_save_to_kb(detected)

    # 处理"订阅到知识库"触发
    if trigger == 'subscribe':
        return handle_subscribe_to_kb(detected)

    # 以下是原有的仅链接检测逻辑
    # First, check for short links and provide help
    has_short, short_msg = detect_and_format(text)

    # If we have short links, show help message first
    if has_short:
        return short_msg

    # Check if this is a confirmation (y/yes)
    text_lower = text.lower().strip()
    if text_lower in ('y', 'yes', '是', '好的', 'ok'):
        # This would need state management - for now return a message
        # about using explicit command
        return "请使用 /kb-add <URL> 命令添加内容，或重新发送链接"

    # Show preview
    preview = format_preview(detected)
    prompt = get_action_prompt(detected)

    return f"{preview}\n{prompt}"


# CLI interface for testing
def main():
    """CLI for testing the handler."""
    if len(sys.argv) < 2:
        print("Usage: python3 handler.py <command> [options]")
        print("\nCommands:")
        print("  detect 'text'              - 检测文本中的内容链接")
        print("  process <BV/URL>           - 使用主流程处理视频")
        print("  process <BV/URL> --git     - 处理视频并同步到Git")
        print("\nExamples:")
        print('  python3 handler.py detect "https://www.bilibili.com/video/BV1xx411c7mD"')
        print('  python3 handler.py process BV1xx411c7mD')
        print('  python3 handler.py process BV1xx411c7mD --git')
        sys.exit(1)

    command = sys.argv[1].lower()

    if command == 'detect':
        # 检测链接
        if len(sys.argv) < 3:
            print("Error: Please provide text to detect")
            sys.exit(1)
        text = ' '.join(sys.argv[2:])
        response = handle_message(text)
        if response:
            print(response)
        else:
            print("No content links detected")

    elif command == 'process':
        # 使用主流程处理视频
        if len(sys.argv) < 3:
            print("Error: Please provide BV number or URL")
            sys.exit(1)

        source_id = sys.argv[2]
        enable_git = '--git' in sys.argv

        print(f"Processing video: {source_id}")
        if enable_git:
            print("Git sync: enabled")
        else:
            print("Git sync: disabled")
        print()

        success, message = process_video_full(source_id, enable_git_sync=enable_git)

        if success:
            print(message)
            sys.exit(0)
        else:
            print(f"\n❌ Failed: {message}")
            sys.exit(1)

    else:
        # 兼容旧格式：直接传入文本进行检测
        text = sys.argv[1]
        response = handle_message(text)
        if response:
            print(response)
        else:
            print("No content links detected")


if __name__ == '__main__':
    main()