feat: 添加 Spider 基类 + Pipeline 配置 + 测试

This commit is contained in:
2026-05-09 13:37:40 +08:00
parent a994dc5f53
commit d9af819d2e
6 changed files with 146 additions and 0 deletions
View File
+13
View File
@@ -0,0 +1,13 @@
import pytest
@pytest.fixture
def sample_announcement_data():
return {
"title": "测试公告标题",
"publish_date": "2026-05-09",
"purchase_name": "测试单位",
"content_url": "https://example.com/detail/123",
"source_code": "ZcyAnnouncement1",
"source_name": "采购公告",
}
View File
+76
View File
@@ -0,0 +1,76 @@
import pytest
from datetime import datetime
from app.crawler.base import BaseSpider, CrawlResult, PipelineConfig, PipelineResult
class FakeSpider(BaseSpider):
name = "test_spider"
source_code = "test_source"
source_name = "测试来源"
async def crawl(self) -> CrawlResult:
return CrawlResult(
source_code=self.source_code,
source_name=self.source_name,
total_count=5,
new_count=3,
crawled_at=datetime.now(),
)
def get_pipeline_config(self) -> PipelineConfig:
return PipelineConfig(
filter_enabled=True,
keywords=["测试"],
dedup_enabled=True,
notify_mode="filtered",
mark_sent=False,
)
@pytest.mark.asyncio
async def test_base_spider_crawl():
spider = FakeSpider()
result = await spider.crawl()
assert result.source_code == "test_source"
assert result.total_count == 5
assert result.new_count == 3
assert result.success is True
def test_pipeline_config_defaults():
config = PipelineConfig()
assert config.filter_enabled is True
assert config.keywords == []
assert config.dedup_enabled is True
assert config.notify_mode == "filtered"
assert config.mark_sent is False
def test_pipeline_result_defaults():
result = PipelineResult()
assert result.stored == 0
assert result.filtered == 0
assert result.notified == 0
assert result.markdown_generated is False
def test_generate_content_hash():
h = BaseSpider.generate_content_hash(
"title", "2026-05-09", "unit", "https://x.com", "source"
)
assert len(h) == 64
assert all(c in "0123456789abcdef" for c in h)
def test_generate_content_hash_deterministic():
h1 = BaseSpider.generate_content_hash("t", "d", "p", "u", "s")
h2 = BaseSpider.generate_content_hash("t", "d", "p", "u", "s")
assert h1 == h2
def test_crawl_result_failure():
result = CrawlResult(
source_code="s", source_name="n",
error_message="连接超时"
)
assert result.success is False