From dd94f068e7e7d5a54c80733b3ba47b87d595b3be Mon Sep 17 00:00:00 2001 From: starmen1265 Date: Fri, 13 Mar 2026 18:12:16 +0800 Subject: [PATCH] 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 --- kb_pipeline/link_parser.py | 181 +++++++++++++++++++ tests/test_pipeline/test_link_parser.py | 231 ++++++++++++++++++++++++ 2 files changed, 412 insertions(+) create mode 100644 kb_pipeline/link_parser.py create mode 100644 tests/test_pipeline/test_link_parser.py diff --git a/kb_pipeline/link_parser.py b/kb_pipeline/link_parser.py new file mode 100644 index 0000000..cd5f779 --- /dev/null +++ b/kb_pipeline/link_parser.py @@ -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 diff --git a/tests/test_pipeline/test_link_parser.py b/tests/test_pipeline/test_link_parser.py new file mode 100644 index 0000000..5a9d8da --- /dev/null +++ b/tests/test_pipeline/test_link_parser.py @@ -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