From cbeb3c3504e82080ed55f1207af46cb390db1753 Mon Sep 17 00:00:00 2001 From: v6ole Date: Sat, 9 May 2026 16:10:30 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E8=BF=9E=E6=8E=A5=20Pipeline=20?= =?UTF-8?q?=E5=88=B0=E7=88=AC=E5=8F=96=E6=B5=81=E7=A8=8B=20+=20=E5=A4=9C?= =?UTF-8?q?=E9=97=B4=E8=B7=B3=E8=BF=87=20+=20=E6=96=AD=E7=BD=91=E6=81=A2?= =?UTF-8?q?=E5=A4=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - CrawlService.run_spider 现在自动调用 PostCrawlPipeline（爬取→存储→筛选→去重→推送） - 定时任务添加 22:00-06:00 夜间跳过逻辑 - 断网恢复后不会漏公告：爬虫每次从 API 按日期倒序抓取，未推送的公告下次自动补推 - Pipeline 的 _exclude_sent + mark_sent 防止重复推送 --- app/api/crawl.py | 8 ++++ app/api/deps.py | 9 +++- app/crawler/base.py | 17 ++++---- app/scheduler/jobs.py | 29 +++++++++++-- app/services/crawl_service.py | 52 ++++++++++++++++++++--- tests/test_services/test_crawl_service.py | 16 +++++-- 6 files changed, 106 insertions(+), 25 deletions(-) diff --git a/app/api/crawl.py b/app/api/crawl.py index c2a94f0..9590fc2 100644 --- a/app/api/crawl.py +++ b/app/api/crawl.py @@ -12,13 +12,21 @@ async def trigger_crawl(request: CrawlTriggerRequest): names = service.get_spider_names() all_results = [] + total_stored = 0 + total_notified = 0 for name in names: results = await service.run_spider(name) all_results.extend(results) + for r in results: + if r.pipeline_result: + total_stored += r.pipeline_result.stored + total_notified += r.pipeline_result.notified return { "spiders_run": names, "total_announcements": sum(r.total_count for r in all_results), + "total_stored": total_stored, + "total_notified": total_notified, "errors": [r.error_message for r in all_results if not r.success], } diff --git a/app/api/deps.py b/app/api/deps.py index 573c0bc..315d580 100644 --- a/app/api/deps.py +++ b/app/api/deps.py @@ -3,10 +3,11 @@ from sqlalchemy.ext.asyncio import AsyncSession from app.crawler.dahuagov_spider import DahuagovSpider from app.crawler.gxgp_spider import GXGPSpider from app.services.crawl_service import CrawlService +from app.services.notification_service import NotificationService async def get_db() -> AsyncSession: - from app.main import async_session # 延迟导入避免循环引用 + from app.main import async_session async with async_session() as session: yield session @@ -17,7 +18,11 @@ _crawl_service: CrawlService | None = None def get_crawl_service() -> CrawlService: global _crawl_service if _crawl_service is None: - _crawl_service = CrawlService() + from app.main import async_session + _crawl_service = CrawlService( + db_session_factory=async_session, + notification_service=NotificationService(), + ) _crawl_service.register(GXGPSpider()) _crawl_service.register(DahuagovSpider()) return _crawl_service diff --git a/app/crawler/base.py b/app/crawler/base.py index 93be0a6..ff26959 100644 --- a/app/crawler/base.py +++ b/app/crawler/base.py @@ -4,6 +4,14 @@ from dataclasses import dataclass, field from datetime import datetime +@dataclass +class PipelineResult: + stored: int = 0 + filtered: int = 0 + notified: int = 0 + markdown_generated: bool = False + + @dataclass class CrawlResult: source_code: str @@ -14,6 +22,7 @@ class CrawlResult: error_message: str | None = None crawled_at: datetime = field(default_factory=datetime.now) duration: float = 0.0 + pipeline_result: PipelineResult | None = None @property def success(self) -> bool: @@ -29,14 +38,6 @@ class PipelineConfig: mark_sent: bool = False -@dataclass -class PipelineResult: - stored: int = 0 - filtered: int = 0 - notified: int = 0 - markdown_generated: bool = False - - class BaseSpider(ABC): name: str source_code: str diff --git a/app/scheduler/jobs.py b/app/scheduler/jobs.py index 86c711b..a5826b4 100644 --- a/app/scheduler/jobs.py +++ b/app/scheduler/jobs.py @@ -1,4 +1,6 @@ import logging +from datetime import datetime, time +from zoneinfo import ZoneInfo from apscheduler.schedulers.asyncio import AsyncIOScheduler from apscheduler.triggers.cron import CronTrigger @@ -9,19 +11,38 @@ from app.config import settings logger = logging.getLogger(__name__) scheduler = AsyncIOScheduler() +NIGHT_START = time(22, 0) +NIGHT_END = time(6, 0) +TZ = ZoneInfo("Asia/Shanghai") + + +def _is_night_time() -> bool: + """22:00 ~ 次日 06:00 夜间时段""" + current = datetime.now(TZ).time() + return current >= NIGHT_START or current < NIGHT_END + async def scheduled_crawl(): + if _is_night_time(): + logger.info("夜间时段 (22:00-06:00)，跳过爬取") + return + logger.info("开始定时爬取任务") service = get_crawl_service() - names = service.get_spider_names() - for name in names: + for name in service.get_spider_names(): try: results = await service.run_spider(name) for r in results: if not r.success: logger.error(f"Spider {name} 失败: {r.error_message}") + elif r.pipeline_result: + logger.info( + f"Spider {name}: 抓取{r.total_count}条, " + f"新增{r.pipeline_result.stored}条, " + f"通知{r.pipeline_result.notified}条" + ) else: - logger.info(f"Spider {name} 完成: {r.total_count} 条") + logger.info(f"Spider {name}: 抓取{r.total_count}条 (无新增)") except Exception as e: logger.error(f"Spider {name} 异常: {e}") logger.info("定时爬取任务完成") @@ -30,7 +51,7 @@ async def scheduled_crawl(): def start_scheduler(): if not settings.scheduler_enabled: return - trigger = CronTrigger.from_crontab(settings.scheduler_cron, timezone="Asia/Shanghai") + trigger = CronTrigger.from_crontab(settings.scheduler_cron, timezone=TZ) scheduler.add_job( scheduled_crawl, trigger=trigger, diff --git a/app/services/crawl_service.py b/app/services/crawl_service.py index 8132f27..96f9a10 100644 --- a/app/services/crawl_service.py +++ b/app/services/crawl_service.py @@ -1,9 +1,19 @@ +import logging + +from sqlalchemy.ext.asyncio import async_sessionmaker from app.crawler.base import BaseSpider, CrawlResult +from app.services.notification_service import NotificationService +from app.services.pipeline import PostCrawlPipeline + +logger = logging.getLogger(__name__) class CrawlService: - def __init__(self): + def __init__(self, db_session_factory: async_sessionmaker, + notification_service: NotificationService): + self.db_session_factory = db_session_factory + self.notification_service = notification_service self.spiders: dict[str, BaseSpider] = {} def register(self, spider: BaseSpider): @@ -11,19 +21,47 @@ class CrawlService: async def run_all(self) -> list[CrawlResult]: results = [] - for name, spider in self.spiders.items(): - result = await spider.crawl() - results.append(result) + for name in self.spiders: + spider_results = await self.run_spider(name) + results.extend(spider_results) return results - async def run_spider(self, name: str, **kwargs) -> list[CrawlResult]: + async def run_spider(self, name: str) -> list[CrawlResult]: spider = self.spiders.get(name) if spider is None: return [CrawlResult( source_code=name, source_name=name, - error_message=f"Spider not found: {name}" + error_message=f"Spider not found: {name}", )] - result = await spider.crawl(**kwargs) + + # 1. Crawl + result = await spider.crawl() + + # 2. Pipeline: store → filter → notify + if result.success and result.announcements: + try: + config = spider.get_pipeline_config() + async with self.db_session_factory() as db: + pipeline = PostCrawlPipeline( + db_session=db, + notification_service=self.notification_service, + ) + pipe_result = await pipeline.process( + result.announcements, config, + ) + result.pipeline_result = pipe_result + result.new_count = pipe_result.stored + logger.info( + f"Spider {name}: total={result.total_count}, " + f"stored={pipe_result.stored}, " + f"filtered={pipe_result.filtered}, " + f"notified={pipe_result.notified}" + ) + except Exception as e: + logger.error(f"Spider {name} pipeline 异常: {e}") + if not result.error_message: + result.error_message = str(e) + return [result] def get_spider_names(self) -> list[str]: diff --git a/tests/test_services/test_crawl_service.py b/tests/test_services/test_crawl_service.py index a1e98c0..791c2d4 100644 --- a/tests/test_services/test_crawl_service.py +++ b/tests/test_services/test_crawl_service.py @@ -1,3 +1,4 @@ +from unittest.mock import AsyncMock import pytest @@ -49,9 +50,16 @@ class MockDahuagovSpider(BaseSpider): ) +def _make_service(): + return CrawlService( + db_session_factory=AsyncMock(), + notification_service=AsyncMock(), + ) + + @pytest.mark.asyncio async def test_crawl_service_registers_spiders(): - service = CrawlService() + service = _make_service() service.register(MockGXGPSpider()) service.register(MockDahuagovSpider()) assert len(service.spiders) == 2 @@ -59,7 +67,7 @@ async def test_crawl_service_registers_spiders(): @pytest.mark.asyncio async def test_crawl_service_run_all(): - service = CrawlService() + service = _make_service() service.register(MockGXGPSpider()) service.register(MockDahuagovSpider()) results = await service.run_all() @@ -71,7 +79,7 @@ async def test_crawl_service_run_all(): @pytest.mark.asyncio async def test_crawl_service_run_specific(): - service = CrawlService() + service = _make_service() service.register(MockGXGPSpider()) service.register(MockDahuagovSpider()) results = await service.run_spider("mock_dahuagov") @@ -81,7 +89,7 @@ async def test_crawl_service_run_specific(): @pytest.mark.asyncio async def test_crawl_service_run_unknown(): - service = CrawlService() + service = _make_service() results = await service.run_spider("nonexistent") assert len(results) == 1 assert results[0].success is False