全异步 Yahoo 热门新闻爬虫：模块化架构与三路数据输出实践

1
async def scrape(self) -> list[TrendingItem]:
2
    """
3
    优先使用 Playwright（完整 JS 渲染），
4
    若 Playwright 失败则回退到 httpx（快速但内容不完整）。
5
    """
6
    if self.config.force_playwright:
7
        try:
8
            html = await self._fetch_with_playwright()
9
        except Exception as e:
10
            logger.warning(f"Playwright 失败，回退到 httpx: {e}")
11
            html = await self._fetch_with_httpx()
12
    else:
13
        html = await self._fetch_with_httpx()
14

15
    return self.parser.parse(html)

1
SELECTOR_STRATEGIES = [
2
    # 策略 1：最精确，基于稳定的 story ID 属性
3
    {"container": "div[id^='storyline-']", "title": "h3", "link": "a"},
4
    # 策略 2：基于语义化 article 标签
5
    {"container": "article.story", "title": "h3.title", "link": "a.title-link"},
6
    # 策略 3：基于 stream 容器
7
    {"container": "li.stream-item", "title": "h3", "link": "a"},
8
    # 策略 4：通用卡片选择器
9
    {"container": "div.Ov\\(h\\)", "title": "h3", "link": "a"},
10
    # 策略 5：基于 data 属性
11
    {"container": "[data-test-locator='stream-item']", "title": "h3", "link": "a"},
12
    # 策略 6：最宽泛的 article 标签
13
    {"container": "article", "title": "h3,h2", "link": "a[href*='yahoo.com']"},
14
    # 策略 7：终极兜底
15
    {"container": "li", "title": "h3,h2,h1", "link": "a[href*='yahoo.com']"},
16
]
17

18
def parse(self, html: str) -> list[TrendingItem]:
19
    soup = BeautifulSoup(html, 'lxml')
20
    for strategy in SELECTOR_STRATEGIES:
21
        items = self._try_parse(soup, strategy)
22
        if len(items) >= 5:   # 至少解析到 5 条才认为成功
23
            logger.info(f"选择器策略成功: {strategy['container']}, 解析到 {len(items)} 条")
24
            return items
25
    logger.error("所有选择器策略均失败，返回空列表")
26
    return []

1
class TrendingScheduler:
2
    def __init__(self, scrape_fn, output_fns: list):
3
        self.scheduler = AsyncIOScheduler()
4
        self.scheduler.add_job(
5
            func=self._scrape_and_publish,
6
            trigger=CronTrigger.from_crontab("0 * * * *"),  # 每小时整点
7
            id='yahoo_trending_hourly',
8
            max_instances=1,
9
            misfire_grace_time=300,
10
        )
11

12
    async def _scrape_and_publish(self):
13
        items = await self.scrape_fn()
14
        # 三路并行输出
15
        await asyncio.gather(
16
            self.file_writer.save(items),
17
            self._push_to_kafka(items),
18
            self._push_to_ods_async(items),
19
        )

1
async def main():
2
    stop_event = asyncio.Event()
3

4
    def _handle_signal():
5
        logger.info("收到停止信号，正在优雅停机...")
6
        stop_event.set()
7

8
    loop = asyncio.get_event_loop()
9
    for sig in (signal.SIGINT, signal.SIGTERM):
10
        loop.add_signal_handler(sig, _handle_signal)
11

12
    scheduler = TrendingScheduler(...)
13
    scheduler.start()
14

15
    await stop_event.wait()   # 阻塞直到收到信号
16

17
    scheduler.shutdown(wait=False)
18
    kafka_producer.close()    # flush 未发送消息
19
    logger.info("停机完成")