starmen1265/kb-skills
Archived
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

docs(08-03): complete anti-crawl measures plan

Browse Source 
- Proxy configuration from HTTP_PROXY/HTTPS_PROXY
- Random delay (1-3s) between requests
- Exponential backoff retry (3 attempts)
- x-zse-96 signature handled by browser

All 44 crawler tests passing

Requirements: NET-01, NET-02, NET-03 completed

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
starmen1265
2026-03-17 22:20:38 +08:00
parent 9e1452d2a9
commit 123e307522
3 changed files with 151 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_crawler/test_comment_crawler.py
+57 -53
View File
@@ -5,6 +5,7 @@ from unittest.mock import Mock, patch, MagicMock
from kb_workflow.crawler.zhihu_crawler import ZhihuCrawler
from kb_workflow.crawler.models import CommentData, CommentsResult
from kb_workflow.crawler.config import CrawlerConfig
class TestCommentCrawler:
@@ -13,7 +14,8 @@ class TestCommentCrawler:
@pytest.fixture
def crawler(self):
"""Create ZhihuCrawler instance for testing."""
return ZhihuCrawler(headless=True)
config = CrawlerConfig(proxy_required=False)
return ZhihuCrawler(headless=True, config=config)
@pytest.fixture
def mock_comment_response(self):
@@ -48,12 +50,14 @@ class TestCommentCrawler:
def test_crawl_comments_returns_comments_result(self, crawler, mock_comment_response):
"""crawl_comments returns CommentsResult."""
with patch.object(crawler, '_get_cookie', return_value="test_cookie"):
with patch('kb_workflow.crawler.zhihu_crawler.PlaywrightBrowser') as MockBrowser:
mock_browser = MockBrowser.return_value
with patch.object(crawler, '_start_browser_with_proxy'):
# Mock _page.request.get
mock_response = MagicMock()
mock_response.status = 200
mock_response.json.return_value = mock_comment_response
mock_browser.page.request.get.return_value = mock_response
crawler._page = MagicMock()
crawler._page.request.get.return_value = mock_response
result = crawler.crawl_comments("answers", 12345)
@@ -64,28 +68,30 @@ class TestCommentCrawler:
def test_comments_are_hot_order(self, crawler, mock_comment_response):
"""Comments use hot (popular) sorting."""
with patch.object(crawler, '_get_cookie', return_value="test_cookie"):
with patch('kb_workflow.crawler.zhihu_crawler.PlaywrightBrowser') as MockBrowser:
mock_browser = MockBrowser.return_value
with patch.object(crawler, '_start_browser_with_proxy'):
mock_response = MagicMock()
mock_response.status = 200
mock_response.json.return_value = mock_comment_response
mock_browser.page.request.get.return_value = mock_response
crawler._page = MagicMock()
crawler._page.request.get.return_value = mock_response
crawler.crawl_comments("answers", 12345)
# Verify API URL contains order=hot
call_args = mock_browser.page.request.get.call_args
call_args = crawler._page.request.get.call_args
assert "order=hot" in call_args[0][0]
def test_comment_has_required_fields(self, crawler, mock_comment_response):
"""Each comment contains required fields."""
with patch.object(crawler, '_get_cookie', return_value="test_cookie"):
with patch('kb_workflow.crawler.zhihu_crawler.PlaywrightBrowser') as MockBrowser:
mock_browser = MockBrowser.return_value
with patch.object(crawler, '_start_browser_with_proxy'):
mock_response = MagicMock()
mock_response.status = 200
mock_response.json.return_value = mock_comment_response
mock_browser.page.request.get.return_value = mock_response
crawler._page = MagicMock()
crawler._page.request.get.return_value = mock_response
result = crawler.crawl_comments("answers", 12345)
@@ -98,12 +104,13 @@ class TestCommentCrawler:
def test_no_nested_replies(self, crawler, mock_comment_response):
"""Do not expand nested replies - only first-level comments."""
with patch.object(crawler, '_get_cookie', return_value="test_cookie"):
with patch('kb_workflow.crawler.zhihu_crawler.PlaywrightBrowser') as MockBrowser:
mock_browser = MockBrowser.return_value
with patch.object(crawler, '_start_browser_with_proxy'):
mock_response = MagicMock()
mock_response.status = 200
mock_response.json.return_value = mock_comment_response
mock_browser.page.request.get.return_value = mock_response
crawler._page = MagicMock()
crawler._page.request.get.return_value = mock_response
result = crawler.crawl_comments("answers", 12345)
@@ -130,49 +137,46 @@ class TestCommentCrawler:
"voteup_count": 10,
"comment_count": 5
}
mock_comment_response = {
"data": [{"id": 1, "content": "Comment", "author": {"name": "User"}}],
"paging": {"totals": 1}
}
with patch.object(crawler, '_get_cookie', return_value="test_cookie"):
with patch('kb_workflow.crawler.zhihu_crawler.PlaywrightBrowser') as MockBrowser:
mock_browser = MockBrowser.return_value
# Setup for answer crawl
mock_browser.page.route = MagicMock()
mock_browser.page.request.get.return_value.status = 200
mock_browser.page.request.get.return_value.json.return_value = mock_comment_response
mock_browser.goto = MagicMock()
mock_browser.close = MagicMock()
# Create interceptor that will capture answer data
with patch('kb_workflow.crawler.zhihu_crawler.ZhihuApiInterceptor') as MockInterceptor:
mock_interceptor = MockInterceptor.return_value
with patch.object(crawler, '_start_browser_with_proxy'):
with patch.object(crawler, '_apply_random_delay'):
# Mock ZhihuApiInterceptor
from kb_workflow.crawler.zhihu_crawler import ZhihuApiInterceptor
mock_interceptor = MagicMock(spec=ZhihuApiInterceptor)
mock_interceptor.captured_data = {"answer": mock_answer_data}
# Mock crawl_comments to return test data
with patch.object(crawler, 'crawl_comments') as mock_comments:
mock_comments.return_value = CommentsResult(
content_type="answers",
content_id=456,
comments=[CommentData(id=1, content="Test comment")],
total_count=1
)
with patch('kb_workflow.crawler.zhihu_crawler.ZhihuApiInterceptor', return_value=mock_interceptor):
# Mock page operations
crawler._page = MagicMock()
crawler._page.goto = MagicMock()
crawler._page.route = MagicMock()
crawler._page.wait_for_timeout = MagicMock()
result = crawler.crawl_answer(
"https://www.zhihu.com/question/123/answer/456",
include_comments=True
)
# Mock crawl_comments
with patch.object(crawler, 'crawl_comments') as mock_comments:
mock_comments.return_value = CommentsResult(
content_type="answers",
content_id=456,
comments=[CommentData(id=1, content="Test comment")],
total_count=1
)
# Verify crawl_comments was called with correct params
mock_comments.assert_called_once_with(
content_type="answers",
content_id=456,
include_comments=True
)
with patch.object(crawler, '_close_browser'):
result = crawler.crawl_answer(
"https://www.zhihu.com/question/123/answer/456",
include_comments=True
)
# Verify result includes comments
assert result.success
assert result.comments is not None
assert len(result.comments) == 1
assert result.comments[0].content == "Test comment"
# Verify crawl_comments was called with correct params
mock_comments.assert_called_once_with(
content_type="answers",
content_id=456,
include_comments=True
)
# Verify result includes comments
assert result.success
assert result.comments is not None
assert len(result.comments) == 1
assert result.comments[0].content == "Test comment"