openclaw-skills/skills/obsidian-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
)

# 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+)'),
}


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 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 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
    """
    # First, check for short links and provide help
    has_short, short_msg = detect_and_format(text)

    # Detect all links (including short ones)
    detected = detect_links(text)

    if not detected:
        return None

    # 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 'text to test'")
        print("\nExamples:")
        print('  python3 handler.py "https://www.bilibili.com/video/BV1xx411c7mD"')
        print('  python3 handler.py "Check this video: https://b23.tv/xxxxx"')
        sys.exit(1)

    text = sys.argv[1]
    response = handle_message(text)

    if response:
        print(response)
    else:
        print("No content links detected")


if __name__ == '__main__':
    main()