openclaw-skills/skills/kb-link-parser/link-parser.py

#!/usr/bin/env python3
"""
Link Parser - Extract and resolve IDs from various platform URLs.

Supports:
- Bilibili (video, article, up主, collection, series)
- YouTube (video, channel, playlist)
- Zhihu (question, answer, user, collection)
- Short links (b23.tv, etc.)

Usage:
    python link-parser.py --url "https://www.bilibili.com/video/BV1xx411c7mD"
    python link-parser.py --batch urls.txt --output results.json
"""

import argparse
import json
import re
import sys
from typing import Optional, Dict, Any
from urllib.parse import urlparse

from link_parser_shortlink import ShortLinkResolver


class LinkParser:
    """Main link parser class"""

    # Bilibili patterns
    BILIBILI_VIDEO_PATTERN = re.compile(r'https?://(?:www\.)?bilibili\.com/video/(BV[a-zA-Z0-9]+)')
    BILIBILI_VIDEO_SHORT_PATTERN = re.compile(r'https?://b23\.tv/([a-zA-Z0-9]+)')
    BILIBILI_ARTICLE_PATTERN = re.compile(r'https?://(?:www\.)?bilibili\.com/(?:article|read)/([a-zA-Z0-9]+)')
    BILIBILI_SPACE_PATTERN = re.compile(r'https?://(?:space\.)?bilibili\.com/(\d+)')
    BILIBILI_MOBILE_SPACE_PATTERN = re.compile(r'https?://m\.bilibili\.com/space/(\d+)')
    BILIBILI_MOBILE_OPUS_PATTERN = re.compile(r'https?://m\.bilibili\.com/opus/(\d+)')
    BILIBILI_FAVLIST_PATTERN = re.compile(r'https?://(?:space\.)?bilibili\.com/\d+/favlist\?fid=(\d+)')
    BILIBILI_OPUS_PATTERN = re.compile(r'https?://(?:www\.)?bilibili\.com/opus/(\d+)')
    BILIBILI_SERIES_PATTERN = re.compile(r'https?://(?:space\.)?bilibili\.com/\d+/lists/(\d+)\?type=series')
    BILIBILI_READLIST_PATTERN = re.compile(r'https?://(?:www\.)?bilibili\.com/read/readlist/(rl[a-zA-Z0-9]+)')

    # YouTube patterns
    YOUTUBE_VIDEO_PATTERN = re.compile(r'(?:https?://)?(?:www\.)?(?:youtube\.com/watch\?v=|youtu\.be/)([a-zA-Z0-9_-]{11})')
    YOUTUBE_CHANNEL_PATTERN = re.compile(r'(?:https?://)?(?:www\.)?youtube\.com/channel/([a-zA-Z0-9_-]+)')
    YOUTUBE_USER_PATTERN = re.compile(r'(?:https?://)?(?:www\.)?youtube\.com/user/([a-zA-Z0-9_-]+)')
    YOUTUBE_PLAYLIST_PATTERN = re.compile(r'(?:https?://)?(?:www\.)?youtube\.com/playlist\?list=([a-zA-Z0-9_-]+)')

    # Zhihu patterns
    ZHIHAU_QUESTION_PATTERN = re.compile(r'https?://(?:www\.)?zhihu\.com/question/(\d+)')
    ZHIHAU_ANSWER_PATTERN = re.compile(r'https?://(?:www\.)?zhihu\.com/question/\d+/answer/(\d+)')
    ZHIHAU_USER_PATTERN = re.compile(r'https?://(?:www\.)?zhihu\.com/people/([a-zA-Z0-9_-]+)')
    ZHIHAU_COLUMN_PATTERN = re.compile(r'https?://(?:www\.)?zhihu\.com/column/([a-zA-Z0-9_-]+)')
    ZHIHAU_COLLECTION_PATTERN = re.compile(r'https?://(?:www\.)?zhihu\.com/collection/(\d+)')

    def __init__(self):
        self.shortlink_resolver = ShortLinkResolver()

    def resolve_shortlink(self, url: str) -> Optional[str]:
        """
        Resolve short link to get final URL.

        Args:
            url: Short URL to resolve

        Returns:
            Final URL after resolution, or None if not resolvable
        """
        # Check if it's a known short link pattern
        if self.shortlink_resolver.is_bilibili_short_link(url):
            # Try to resolve using requests (follows redirects)
            try:
                import requests
                response = requests.head(url, allow_redirects=True, timeout=10)
                return response.url
            except Exception:
                # Fallback: return original URL
                return url

        return url

    def extract_bilibili_id(self, url: str) -> Optional[Dict[str, str]]:
        """
        Extract Bilibili ID from URL.

        Args:
            url: Bilibili URL

        Returns:
            Dict with platform, type, and id fields, or None if not matched
        """
        # Try series/playlist (must check before space pattern)
        match = self.BILIBILI_SERIES_PATTERN.match(url)
        if match:
            return {
                "platform": "bilibili",
                "type": "playlist",
                "id": match.group(1)
            }

        # Try readlist (专栏合集)
        match = self.BILIBILI_READLIST_PATTERN.match(url)
        if match:
            return {
                "platform": "bilibili",
                "type": "readlist",
                "id": match.group(1)
            }

        # Try favlist (收藏夹)
        match = self.BILIBILI_FAVLIST_PATTERN.match(url)
        if match:
            return {
                "platform": "bilibili",
                "type": "favlist",
                "id": match.group(1)
            }

        # Try opus (动态)
        match = self.BILIBILI_OPUS_PATTERN.match(url)
        if match:
            return {
                "platform": "bilibili",
                "type": "opus",
                "id": match.group(1)
            }

        # Try mobile opus (移动端动态)
        match = self.BILIBILI_MOBILE_OPUS_PATTERN.match(url)
        if match:
            return {
                "platform": "bilibili",
                "type": "opus",
                "id": match.group(1)
            }

        # Try video
        match = self.BILIBILI_VIDEO_PATTERN.match(url)
        if match:
            return {
                "platform": "bilibili",
                "type": "video",
                "id": match.group(1)
            }

        # Try article/read
        match = self.BILIBILI_ARTICLE_PATTERN.match(url)
        if match:
            return {
                "platform": "bilibili",
                "type": "article",
                "id": match.group(1)
            }

        # Try space (UP主) - must check after specific patterns
        match = self.BILIBILI_SPACE_PATTERN.match(url)
        if match:
            return {
                "platform": "bilibili",
                "type": "up主",
                "id": match.group(1)
            }

        # Try mobile space (移动端UP主空间)
        match = self.BILIBILI_MOBILE_SPACE_PATTERN.match(url)
        if match:
            return {
                "platform": "bilibili",
                "type": "up主",
                "id": match.group(1)
            }

        return None

    def extract_youtube_id(self, url: str) -> Optional[Dict[str, str]]:
        """
        Extract YouTube ID from URL.

        Args:
            url: YouTube URL

        Returns:
            Dict with platform, type, and id fields, or None if not matched
        """
        # Try video
        match = self.YOUTUBE_VIDEO_PATTERN.match(url)
        if match:
            return {
                "platform": "youtube",
                "type": "video",
                "id": match.group(1)
            }

        # Try channel
        match = self.YOUTUBE_CHANNEL_PATTERN.match(url)
        if match:
            return {
                "platform": "youtube",
                "type": "channel",
                "id": match.group(1)
            }

        # Try user
        match = self.YOUTUBE_USER_PATTERN.match(url)
        if match:
            return {
                "platform": "youtube",
                "type": "user",
                "id": match.group(1)
            }

        # Try playlist
        match = self.YOUTUBE_PLAYLIST_PATTERN.match(url)
        if match:
            return {
                "platform": "youtube",
                "type": "playlist",
                "id": match.group(1)
            }

        return None

    def extract_zhihu_id(self, url: str) -> Optional[Dict[str, str]]:
        """
        Extract Zhihu ID from URL.

        Args:
            url: Zhihu URL

        Returns:
            Dict with platform, type, and id fields, or None if not matched
        """
        # Try question
        match = self.ZHIHAU_QUESTION_PATTERN.match(url)
        if match:
            return {
                "platform": "zhihu",
                "type": "question",
                "id": match.group(1)
            }

        # Try answer
        match = self.ZHIHAU_ANSWER_PATTERN.match(url)
        if match:
            return {
                "platform": "zhihu",
                "type": "answer",
                "id": match.group(1)
            }

        # Try user
        match = self.ZHIHAU_USER_PATTERN.match(url)
        if match:
            return {
                "platform": "zhihu",
                "type": "user",
                "id": match.group(1)
            }

        # Try column
        match = self.ZHIHAU_COLUMN_PATTERN.match(url)
        if match:
            return {
                "platform": "zhihu",
                "type": "column",
                "id": match.group(1)
            }

        # Try collection
        match = self.ZHIHAU_COLLECTION_PATTERN.match(url)
        if match:
            return {
                "platform": "zhihu",
                "type": "collection",
                "id": match.group(1)
            }

        return None

    def parse(self, url: str) -> Dict[str, Any]:
        """
        Parse URL and extract platform-specific ID.

        Args:
            url: URL to parse

        Returns:
            Dict with platform, type, id, and original_url fields
        """
        result = {
            "platform": None,
            "type": None,
            "id": None,
            "original_url": url,
            "resolved_url": None,
            "success": False,
            "error": None
        }

        # First, resolve short links
        resolved = self.resolve_shortlink(url)
        if resolved:
            result["resolved_url"] = resolved
            url = resolved

        # Try each platform
        bilibili_result = self.extract_bilibili_id(url)
        if bilibili_result:
            result.update(bilibili_result)
            result["success"] = True
            return result

        youtube_result = self.extract_youtube_id(url)
        if youtube_result:
            result.update(youtube_result)
            result["success"] = True
            return result

        zhihu_result = self.extract_zhihu_id(url)
        if zhihu_result:
            result.update(zhihu_result)
            result["success"] = True
            return result

        # Check for short links that need resolution
        short_links = self.shortlink_resolver.detect_short_links(url)
        if short_links:
            result["error"] = "Short link detected. Please provide the full URL."
            return result

        result["error"] = "Unsupported URL format"
        return result


def format_output(data: Dict[str, Any], output_format: str = "json") -> str:
    """Format output based on specified format."""
    if output_format == "json":
        return json.dumps(data, ensure_ascii=False, indent=2)
    elif output_format == "compact":
        return json.dumps(data, ensure_ascii=False, separators=(',', ':'))
    elif output_format == "text":
        lines = []
        for key, value in data.items():
            if value is not None:
                lines.append(f"{key}: {value}")
        return "\n".join(lines)
    else:
        return json.dumps(data, ensure_ascii=False, indent=2)


def main():
    """Main entry point."""
    parser = argparse.ArgumentParser(
        description="Parse and extract IDs from various platform URLs",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  %(prog)s --url "https://www.bilibili.com/video/BV1xx411c7mD"
  %(prog)s --url "https://www.youtube.com/watch?v=dQw4w9WgXcQ"
  %(prog)s --url "https://www.zhihu.com/question/266309957"
  %(prog)s --batch urls.txt --output results.json
        """
    )

    parser.add_argument(
        "--url", "-u",
        help="Single URL to parse"
    )

    parser.add_argument(
        "--batch", "-b",
        help="File containing URLs to parse (one per line)"
    )

    parser.add_argument(
        "--output", "-o",
        help="Output file path (for batch mode)"
    )

    parser.add_argument(
        "--format", "-f",
        choices=["json", "compact", "text"],
        default="json",
        help="Output format (default: json)"
    )

    args = parser.parse_args()

    # Check if we're in shortlink help mode
    if len(sys.argv) == 2 and sys.argv[1] in ['--help', '-h']:
        parser.print_help()
        return 0

    if not args.url and not args.batch:
        parser.print_help()
        print("\nError: Please provide --url or --batch")
        return 1

    link_parser = LinkParser()

    if args.url:
        # Single URL mode
        result = link_parser.parse(args.url)
        print(format_output(result, args.format))
        return 0 if result["success"] else 1

    elif args.batch:
        try:
            with open(args.batch, 'r', encoding='utf-8') as f:
                urls = [line.strip() for line in f if line.strip()]
        except Exception as e:
            print(f"Error reading file: {e}", file=sys.stderr)
            return 1

        results = []
        for url in urls:
            result = link_parser.parse(url)
            results.append(result)

        if args.output:
            with open(args.output, 'w', encoding='utf-8') as f:
                json.dump(results, f, ensure_ascii=False, indent=2)
            print(f"Results written to {args.output}")
        else:
            for result in results:
                print(format_output(result, args.format))
                print("-" * 40)

        return 0


if __name__ == '__main__':
    sys.exit(main())