# 推送邮件 HTML 模板 # 用于生成定时推送给用户的热点摘要邮件 # 邮件客户端不支持 Font Awesome,改用 Emoji 代替平台图标 PLATFORM_EMOJI: dict[str, str] = { "微博热搜": "🔴", "微博": "🔴", "知乎热榜": "🔵", "知乎": "🔵", "百度热搜": "🔍", "今日头条": "📰", "抖音热榜": "🎵", "抖音": "🎵", "bilibili 热搜": "📺", "B站热搜": "📺", "华尔街见闻": "📈", "澎湃新闻": "🌊", "财联社热门": "💰", "凤凰网": "🦅", "贴吧": "💬", } DIGEST_HTML_TEMPLATE = """\

聚势智见 · 热点快报

{delivery_time} · 为你精选了 {event_count} 条事件

{mode_label}
{event_cards_html}
""" EVENT_CARD_TEMPLATE = """\
{hot_label} {hot_score} {tags_html}
{title}
{summary_html} {platforms_html} {match_html}
""" def _hot_level(score: int) -> tuple[str, str, str]: """返回 (label, badge_class, hot_class)""" if score >= 10: return "全网沸腾", "badge-hot", " is-hot" if score >= 5: return "高度关注", "badge-warm", "" if score >= 3: return "上升中", "badge-normal", "" return "一般关注", "badge-tag", "" def _get_event_summary(ev) -> str: """ 兼容 ORM 字段名(ai_comprehensive_summary)和 schema 字段名(summary)。 """ return ( getattr(ev, "summary", None) or getattr(ev, "ai_comprehensive_summary", None) or "" ) def _build_platforms_html(platform_list: list[dict]) -> str: """ 将平台数据列表渲染为 HTML。 每条包含:emoji 图标 + 来源名 + 排名徽章 + 可点击标题链接。 """ if not platform_list: return "" rows = [] seen_sources: set[str] = set() for p in platform_list[:8]: source_name = p.get("source_name", "未知") # 同一来源只显示第一条(通常是排名最靠前的那条) if source_name in seen_sources: continue seen_sources.add(source_name) headline = p.get("headline", "") url = p.get("url", "") ranking = p.get("ranking") emoji = PLATFORM_EMOJI.get(source_name, "🔗") rank_html = "" if ranking: rank_html = f'TOP {ranking}' if url: title_html = ( f'{headline}' ) else: title_html = f'{headline}' rows.append( f'
  • ' f'
    {emoji} {source_name}{rank_html}
    ' f'{title_html}' f'
  • ' ) if not rows: return "" return '
    " def build_digest_html( items: list, delivery_time_str: str, platforms_map: dict[int, list[dict]] | None = None, app_url: str = "http://localhost:5173", is_default_push: bool = False, ) -> str: """ 根据事件列表生成推送邮件 HTML 正文。 items 元素可以是 MatchedEventResult 或 _DefaultEventItem, 二者均有 .event / .tags / .exact_hits / .semantic_hits / .match_score 属性。 platforms_map: event_id → [{source_name, headline, url, ranking}] """ if platforms_map is None: platforms_map = {} mode_label = "全网热点推送" if is_default_push else "个性化关键词匹配" mode_badge_class = "mode-default" if is_default_push else "mode-keyword" cards = [] for item in items: ev = item.event hot_label, badge_class, hot_class = _hot_level(ev.hot_score) tags_html = "".join( f'{t}' for t in item.tags[:4] ) summary_text = _get_event_summary(ev) summary_html = "" if summary_text: summary_html = ( f'
    AI 洞察:{summary_text}
    ' ) platform_list = platforms_map.get(ev.id, []) platforms_html = _build_platforms_html(platform_list) match_parts = [] # 仅个性化模式才显示匹配信息 if not getattr(item, "is_default", False): for h in item.exact_hits[:3]: match_parts.append(f'精确 {h}') for s in item.semantic_hits[:2]: sim_pct = int(s.get("similarity", 0) * 100) match_parts.append( f'语义 {s.get("topic_keyword", "")} {sim_pct}%' ) match_html = "" if match_parts: match_html = ( f'
    匹配度 {item.match_score:.0f} · ' + " ".join(match_parts) + "
    " ) cards.append( EVENT_CARD_TEMPLATE.format( hot_class=hot_class, badge_class=badge_class, hot_label=hot_label, hot_score=ev.hot_score, tags_html=tags_html, title=ev.unified_title, summary_html=summary_html, platforms_html=platforms_html, match_html=match_html, ) ) return DIGEST_HTML_TEMPLATE.format( delivery_time=delivery_time_str, event_count=len(items), event_cards_html="\n".join(cards), app_url=app_url, mode_label=mode_label, mode_badge_class=mode_badge_class, )