OpenClaw Agent
7a1ad6088a
- 更新 SKILL.md：添加 frontmatter 元数据和自动触发规则说明 - 创建 handler.py：实现链接检测和预览生成 - 支持多种 URL 模式：B站（完整/BV号/短链接）、YouTube、知乎 - 智能去重避免重复显示 - 短链接显示友好提示 - 支持批量检测多个链接 - 添加 AUTO_TRIGGER.md 实现文档 - 更新任务文件：标记任务1已完成 测试通过： ✅ B站完整 URL、BV号、短链接 ✅ YouTube 视频 ✅ 知乎文章 ✅ 批量检测和去重
333 lines
9.6 KiB
Python
Executable File
333 lines
9.6 KiB
Python
Executable File
#!/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'))
|
||
|
||
# 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() -> str:
|
||
"""
|
||
Format help message for short links.
|
||
|
||
Returns:
|
||
Help message
|
||
"""
|
||
return """🔗 检测到短链接
|
||
|
||
⚠️ 由于反爬虫保护，短链接无法自动解析
|
||
|
||
📱 手机用户：
|
||
1. 打开链接
|
||
2. 在浏览器地址栏复制完整 URL
|
||
3. 发送给助手
|
||
|
||
💻 桌面用户：
|
||
1. 发送 BV 号（如：BV1jcFqz7EFW）
|
||
2. 或使用完整 URL
|
||
|
||
💡 提示：完整 URL 格式为：https://www.bilibili.com/video/BVxxxxx"""
|
||
|
||
|
||
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
|
||
"""
|
||
# Detect links
|
||
detected = detect_links(text)
|
||
|
||
if not detected:
|
||
return None
|
||
|
||
# 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()
|