feat(02-01): add LinkParser and ParsedLink for unified URL parsing
- Create ParsedLink dataclass with platform, content_type, content_id fields - Implement LinkParser class with regex-based URL parsing - Support Bilibili video/article URLs (with/without www) - Support Zhihu answer/zhuanlan URLs - Support b23.tv short link resolution with requests.head - Return None for unrecognized URLs (no exceptions) - Add 15 unit tests covering all supported URL types Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,181 @@
|
||||
"""Link Parser module
|
||||
|
||||
Provides unified URL parsing for various content platforms.
|
||||
Supports Bilibili (video, article) and Zhihu (answer, zhuanlan) URLs.
|
||||
"""
|
||||
|
||||
import re
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Optional
|
||||
|
||||
import requests
|
||||
|
||||
from kb_pipeline.retry import retry_on_network_error
|
||||
|
||||
|
||||
@dataclass
|
||||
class ParsedLink:
|
||||
"""Represents a parsed link with platform and content information
|
||||
|
||||
Attributes:
|
||||
platform: Platform name (bilibili, zhihu)
|
||||
content_type: Content type (video, article, answer, zhuanlan)
|
||||
content_id: Unique content identifier
|
||||
original_url: Original URL provided by user
|
||||
resolved_url: Resolved URL after short link expansion (defaults to original_url)
|
||||
"""
|
||||
platform: str
|
||||
content_type: str
|
||||
content_id: str
|
||||
original_url: str
|
||||
resolved_url: str = field(default="")
|
||||
|
||||
def __post_init__(self):
|
||||
"""Set resolved_url to original_url if not provided"""
|
||||
if not self.resolved_url:
|
||||
self.resolved_url = self.original_url
|
||||
|
||||
|
||||
class LinkParser:
|
||||
"""Unified URL parser for multiple platforms
|
||||
|
||||
Parses URLs from Bilibili and Zhihu, extracting platform, content type,
|
||||
and content ID. Supports short link resolution for b23.tv URLs.
|
||||
"""
|
||||
|
||||
# Bilibili patterns
|
||||
BILIBILI_VIDEO_PATTERN = re.compile(
|
||||
r'https?://(?:www\.)?bilibili\.com/video/(BV[a-zA-Z0-9]+)',
|
||||
re.IGNORECASE
|
||||
)
|
||||
BILIBILI_ARTICLE_PATTERN = re.compile(
|
||||
r'https?://(?:www\.)?bilibili\.com/(?:read|article)/(cv[a-zA-Z0-9]+)',
|
||||
re.IGNORECASE
|
||||
)
|
||||
|
||||
# Zhihu patterns
|
||||
ZHIHU_ANSWER_PATTERN = re.compile(
|
||||
r'https?://(?:www\.)?zhihu\.com/question/\d+/answer/(\d+)',
|
||||
re.IGNORECASE
|
||||
)
|
||||
ZHIHU_ZHUANLAN_PATTERN = re.compile(
|
||||
r'https?://zhuanlan\.zhihu\.com/p/(\d+)',
|
||||
re.IGNORECASE
|
||||
)
|
||||
|
||||
# Short link patterns
|
||||
B23_TV_PATTERN = re.compile(
|
||||
r'https?://(?:www\.)?b23\.tv/([a-zA-Z0-9]+)',
|
||||
re.IGNORECASE
|
||||
)
|
||||
|
||||
# Timeout for short link resolution
|
||||
SHORT_LINK_TIMEOUT = 10
|
||||
|
||||
@retry_on_network_error
|
||||
def _resolve_shortlink(self, url: str) -> Optional[str]:
|
||||
"""Resolve short link to its final URL
|
||||
|
||||
Uses HTTP HEAD request with redirects to get the final URL.
|
||||
|
||||
Args:
|
||||
url: Short link URL to resolve
|
||||
|
||||
Returns:
|
||||
Final URL after redirects, or None if resolution failed
|
||||
"""
|
||||
try:
|
||||
response = requests.head(
|
||||
url,
|
||||
allow_redirects=True,
|
||||
timeout=self.SHORT_LINK_TIMEOUT
|
||||
)
|
||||
if response.status_code == 200:
|
||||
return response.url
|
||||
return None
|
||||
except (requests.RequestException, OSError):
|
||||
return None
|
||||
|
||||
def parse(self, url: str) -> Optional[ParsedLink]:
|
||||
"""Parse a URL and extract platform, content type, and ID
|
||||
|
||||
Args:
|
||||
url: URL to parse
|
||||
|
||||
Returns:
|
||||
ParsedLink with extracted information, or None if URL is not recognized
|
||||
"""
|
||||
if not url or not isinstance(url, str):
|
||||
return None
|
||||
|
||||
url = url.strip()
|
||||
|
||||
# Try b23.tv short link first
|
||||
if self.B23_TV_PATTERN.match(url):
|
||||
resolved_url = self._resolve_shortlink(url)
|
||||
if resolved_url:
|
||||
# Recursively parse the resolved URL
|
||||
result = self._parse_standard_url(resolved_url)
|
||||
if result:
|
||||
result.original_url = url
|
||||
result.resolved_url = resolved_url
|
||||
return result
|
||||
return None
|
||||
|
||||
return self._parse_standard_url(url)
|
||||
|
||||
def _parse_standard_url(self, url: str) -> Optional[ParsedLink]:
|
||||
"""Parse a standard (non-short) URL
|
||||
|
||||
Args:
|
||||
url: URL to parse
|
||||
|
||||
Returns:
|
||||
ParsedLink with extracted information, or None if URL is not recognized
|
||||
"""
|
||||
# Try Bilibili video
|
||||
match = self.BILIBILI_VIDEO_PATTERN.match(url)
|
||||
if match:
|
||||
return ParsedLink(
|
||||
platform="bilibili",
|
||||
content_type="video",
|
||||
content_id=match.group(1),
|
||||
original_url=url,
|
||||
resolved_url=url
|
||||
)
|
||||
|
||||
# Try Bilibili article
|
||||
match = self.BILIBILI_ARTICLE_PATTERN.match(url)
|
||||
if match:
|
||||
return ParsedLink(
|
||||
platform="bilibili",
|
||||
content_type="article",
|
||||
content_id=match.group(1),
|
||||
original_url=url,
|
||||
resolved_url=url
|
||||
)
|
||||
|
||||
# Try Zhihu answer
|
||||
match = self.ZHIHU_ANSWER_PATTERN.match(url)
|
||||
if match:
|
||||
return ParsedLink(
|
||||
platform="zhihu",
|
||||
content_type="answer",
|
||||
content_id=match.group(1),
|
||||
original_url=url,
|
||||
resolved_url=url
|
||||
)
|
||||
|
||||
# Try Zhihu zhuanlan
|
||||
match = self.ZHIHU_ZHUANLAN_PATTERN.match(url)
|
||||
if match:
|
||||
return ParsedLink(
|
||||
platform="zhihu",
|
||||
content_type="zhuanlan",
|
||||
content_id=match.group(1),
|
||||
original_url=url,
|
||||
resolved_url=url
|
||||
)
|
||||
|
||||
# URL not recognized
|
||||
return None
|
||||
@@ -0,0 +1,231 @@
|
||||
"""Tests for LinkParser module
|
||||
|
||||
Tests URL parsing for various platforms:
|
||||
- Bilibili video URLs
|
||||
- Bilibili article URLs
|
||||
- Zhihu answer URLs
|
||||
- Zhihu zhuanlan URLs
|
||||
- Short link resolution (b23.tv)
|
||||
- Unknown URL handling
|
||||
"""
|
||||
|
||||
import pytest
|
||||
from unittest.mock import patch, MagicMock
|
||||
|
||||
|
||||
class TestLinkParserBilibiliVideo:
|
||||
"""Tests for Bilibili video URL parsing"""
|
||||
|
||||
def test_parse_bilibili_video_url_returns_correct_parsed_link(self):
|
||||
"""Test 1: 解析B站视频URL返回正确的 ParsedLink (platform=bilibili, type=video, id=BVxxx)"""
|
||||
from kb_pipeline.link_parser import LinkParser, ParsedLink
|
||||
|
||||
parser = LinkParser()
|
||||
result = parser.parse("https://www.bilibili.com/video/BV1GJ411x7Qw")
|
||||
|
||||
assert result is not None
|
||||
assert result.platform == "bilibili"
|
||||
assert result.content_type == "video"
|
||||
assert result.content_id == "BV1GJ411x7Qw"
|
||||
assert result.original_url == "https://www.bilibili.com/video/BV1GJ411x7Qw"
|
||||
|
||||
def test_parse_bilibili_video_url_without_www(self):
|
||||
"""B站视频URL不带www也能正确解析"""
|
||||
from kb_pipeline.link_parser import LinkParser
|
||||
|
||||
parser = LinkParser()
|
||||
result = parser.parse("https://bilibili.com/video/BV1GJ411x7Qw")
|
||||
|
||||
assert result is not None
|
||||
assert result.platform == "bilibili"
|
||||
assert result.content_type == "video"
|
||||
assert result.content_id == "BV1GJ411x7Qw"
|
||||
|
||||
def test_parse_bilibili_video_url_with_http(self):
|
||||
"""B站视频URL使用http也能正确解析"""
|
||||
from kb_pipeline.link_parser import LinkParser
|
||||
|
||||
parser = LinkParser()
|
||||
result = parser.parse("http://www.bilibili.com/video/BV1GJ411x7Qw")
|
||||
|
||||
assert result is not None
|
||||
assert result.platform == "bilibili"
|
||||
assert result.content_type == "video"
|
||||
|
||||
|
||||
class TestLinkParserBilibiliArticle:
|
||||
"""Tests for Bilibili article URL parsing"""
|
||||
|
||||
def test_parse_bilibili_article_url_returns_correct_parsed_link(self):
|
||||
"""Test 2: 解析B站专栏URL返回正确的 ParsedLink (platform=bilibili, type=article, id=cvxxx)"""
|
||||
from kb_pipeline.link_parser import LinkParser
|
||||
|
||||
parser = LinkParser()
|
||||
result = parser.parse("https://www.bilibili.com/read/cv12345678")
|
||||
|
||||
assert result is not None
|
||||
assert result.platform == "bilibili"
|
||||
assert result.content_type == "article"
|
||||
assert result.content_id == "cv12345678"
|
||||
|
||||
def test_parse_bilibili_article_url_with_article_prefix(self):
|
||||
"""B站专栏URL使用article前缀也能正确解析"""
|
||||
from kb_pipeline.link_parser import LinkParser
|
||||
|
||||
parser = LinkParser()
|
||||
result = parser.parse("https://www.bilibili.com/article/cv12345678")
|
||||
|
||||
assert result is not None
|
||||
assert result.platform == "bilibili"
|
||||
assert result.content_type == "article"
|
||||
assert result.content_id == "cv12345678"
|
||||
|
||||
|
||||
class TestLinkParserZhihuAnswer:
|
||||
"""Tests for Zhihu answer URL parsing"""
|
||||
|
||||
def test_parse_zhihu_answer_url_returns_correct_parsed_link(self):
|
||||
"""Test 3: 解析知乎问答URL返回正确的 ParsedLink (platform=zhihu, type=answer, id=xxx)"""
|
||||
from kb_pipeline.link_parser import LinkParser
|
||||
|
||||
parser = LinkParser()
|
||||
result = parser.parse("https://www.zhihu.com/question/12345678/answer/87654321")
|
||||
|
||||
assert result is not None
|
||||
assert result.platform == "zhihu"
|
||||
assert result.content_type == "answer"
|
||||
assert result.content_id == "87654321"
|
||||
|
||||
def test_parse_zhihu_answer_url_without_www(self):
|
||||
"""知乎问答URL不带www也能正确解析"""
|
||||
from kb_pipeline.link_parser import LinkParser
|
||||
|
||||
parser = LinkParser()
|
||||
result = parser.parse("https://zhihu.com/question/12345678/answer/87654321")
|
||||
|
||||
assert result is not None
|
||||
assert result.platform == "zhihu"
|
||||
assert result.content_type == "answer"
|
||||
|
||||
|
||||
class TestLinkParserZhihuZhuanlan:
|
||||
"""Tests for Zhihu zhuanlan URL parsing"""
|
||||
|
||||
def test_parse_zhihu_zhuanlan_url_returns_correct_parsed_link(self):
|
||||
"""Test 4: 解析知乎专栏URL返回正确的 ParsedLink (platform=zhihu, type=zhuanlan, id=xxx)"""
|
||||
from kb_pipeline.link_parser import LinkParser
|
||||
|
||||
parser = LinkParser()
|
||||
result = parser.parse("https://zhuanlan.zhihu.com/p/12345678")
|
||||
|
||||
assert result is not None
|
||||
assert result.platform == "zhihu"
|
||||
assert result.content_type == "zhuanlan"
|
||||
assert result.content_id == "12345678"
|
||||
|
||||
|
||||
class TestLinkParserShortLink:
|
||||
"""Tests for short link resolution"""
|
||||
|
||||
@patch("kb_pipeline.link_parser.requests.head")
|
||||
def test_parse_b23tv_shortlink_resolves_to_real_url(self, mock_head):
|
||||
"""Test 5: 解析b23.tv短链返回解析后的真实URL"""
|
||||
from kb_pipeline.link_parser import LinkParser
|
||||
|
||||
# Mock the redirect response
|
||||
mock_response = MagicMock()
|
||||
mock_response.status_code = 200
|
||||
mock_response.url = "https://www.bilibili.com/video/BV1GJ411x7Qw"
|
||||
mock_head.return_value = mock_response
|
||||
|
||||
parser = LinkParser()
|
||||
result = parser.parse("https://b23.tv/abc123")
|
||||
|
||||
assert result is not None
|
||||
assert result.platform == "bilibili"
|
||||
assert result.content_type == "video"
|
||||
assert result.content_id == "BV1GJ411x7Qw"
|
||||
assert result.resolved_url == "https://www.bilibili.com/video/BV1GJ411x7Qw"
|
||||
|
||||
@patch("kb_pipeline.link_parser.requests.head")
|
||||
def test_shortlink_resolution_uses_allow_redirects(self, mock_head):
|
||||
"""短链解析应该使用 allow_redirects=True"""
|
||||
from kb_pipeline.link_parser import LinkParser
|
||||
|
||||
mock_response = MagicMock()
|
||||
mock_response.status_code = 200
|
||||
mock_response.url = "https://www.bilibili.com/video/BV1GJ411x7Qw"
|
||||
mock_head.return_value = mock_response
|
||||
|
||||
parser = LinkParser()
|
||||
parser.parse("https://b23.tv/abc123")
|
||||
|
||||
mock_head.assert_called_once()
|
||||
call_kwargs = mock_head.call_args[1]
|
||||
assert call_kwargs.get("allow_redirects") is True
|
||||
|
||||
|
||||
class TestLinkParserUnknownURL:
|
||||
"""Tests for unknown URL handling"""
|
||||
|
||||
def test_parse_unknown_url_returns_none(self):
|
||||
"""Test 6: 无法识别的URL返回None（不抛异常）"""
|
||||
from kb_pipeline.link_parser import LinkParser
|
||||
|
||||
parser = LinkParser()
|
||||
result = parser.parse("https://www.youtube.com/watch?v=dQw4w9WgXcQ")
|
||||
|
||||
assert result is None
|
||||
|
||||
def test_parse_invalid_url_returns_none(self):
|
||||
"""无效URL返回None"""
|
||||
from kb_pipeline.link_parser import LinkParser
|
||||
|
||||
parser = LinkParser()
|
||||
result = parser.parse("not a valid url")
|
||||
|
||||
assert result is None
|
||||
|
||||
def test_parse_empty_string_returns_none(self):
|
||||
"""空字符串返回None"""
|
||||
from kb_pipeline.link_parser import LinkParser
|
||||
|
||||
parser = LinkParser()
|
||||
result = parser.parse("")
|
||||
|
||||
assert result is None
|
||||
|
||||
|
||||
class TestParsedLink:
|
||||
"""Tests for ParsedLink dataclass"""
|
||||
|
||||
def test_parsed_link_has_all_required_fields(self):
|
||||
"""ParsedLink 应包含所有必需字段"""
|
||||
from kb_pipeline.link_parser import ParsedLink
|
||||
|
||||
link = ParsedLink(
|
||||
platform="bilibili",
|
||||
content_type="video",
|
||||
content_id="BV1GJ411x7Qw",
|
||||
original_url="https://www.bilibili.com/video/BV1GJ411x7Qw",
|
||||
resolved_url="https://www.bilibili.com/video/BV1GJ411x7Qw"
|
||||
)
|
||||
|
||||
assert link.platform == "bilibili"
|
||||
assert link.content_type == "video"
|
||||
assert link.content_id == "BV1GJ411x7Qw"
|
||||
assert link.original_url == "https://www.bilibili.com/video/BV1GJ411x7Qw"
|
||||
assert link.resolved_url == "https://www.bilibili.com/video/BV1GJ411x7Qw"
|
||||
|
||||
def test_parsed_link_resolved_url_defaults_to_original(self):
|
||||
"""ParsedLink 的 resolved_url 默认等于 original_url"""
|
||||
from kb_pipeline.link_parser import ParsedLink
|
||||
|
||||
link = ParsedLink(
|
||||
platform="bilibili",
|
||||
content_type="video",
|
||||
content_id="BV1GJ411x7Qw",
|
||||
original_url="https://www.bilibili.com/video/BV1GJ411x7Qw"
|
||||
)
|
||||
|
||||
assert link.resolved_url == link.original_url
|
||||
Reference in New Issue
Block a user