"""企微早报推送闸门。""" from __future__ import annotations import logging from dataclasses import dataclass, field from typing import Any from daily.config import force_push, skip_push_when_silent logger = logging.getLogger(__name__) @dataclass class PushGateResult: should_push: bool reasons: list[str] = field(default_factory=list) silent: bool = False def _has_board_moves(movement: dict[str, Any]) -> bool: keys = ( "skills_trending_moves", "skills_hot_moves", "github_trending_moves", "github_emerging_moves", "github_topic_moves", ) return any(movement.get(k) for k in keys) def evaluate_push_gate( *, movement: dict[str, Any], ai_news_items: list[dict[str, Any]], cn_ai_news_items: list[dict[str, Any]], featured_pick: dict[str, Any] | None, ) -> PushGateResult: if force_push(): logger.info("push_gate: force_push=on, 强制推送") return PushGateResult(should_push=True, reasons=["force_push"], silent=False) reasons: list[str] = [] if _has_board_moves(movement): reasons.append("board_moves") if ai_news_items: reasons.append("ai_news") if cn_ai_news_items: reasons.append("cn_ai_news") if featured_pick: reasons.append("featured_pick") should = bool(reasons) silent = not should and skip_push_when_silent() if should: logger.info("push_gate: 推送 (原因: %s)", ",".join(reasons)) elif silent: logger.info("push_gate: 静默日, 跳过推送 (无任何更新信号)") else: logger.info("push_gate: 无更新但 skip_push_when_silent=off, 仍推送") return PushGateResult(should_push=should, reasons=reasons, silent=silent)