# coding=utf-8 """ HTML 报告渲染模块 提供 HTML 格式的热点新闻报告生成功能 """ from datetime import datetime from typing import Dict, List, Optional, Callable from trendradar.report.helpers import html_escape def render_html_content( report_data: Dict, total_titles: int, is_daily_summary: bool = False, mode: str = "daily", update_info: Optional[Dict] = None, *, reverse_content_order: bool = False, get_time_func: Optional[Callable[[], datetime]] = None, rss_items: Optional[List[Dict]] = None, rss_new_items: Optional[List[Dict]] = None, display_mode: str = "keyword", ) -> str: """渲染HTML内容 Args: report_data: 报告数据字典,包含 stats, new_titles, failed_ids, total_new_count total_titles: 新闻总数 is_daily_summary: 是否为当日汇总 mode: 报告模式 ("daily", "current", "incremental") update_info: 更新信息(可选) reverse_content_order: 是否反转内容顺序(新增热点在前) get_time_func: 获取当前时间的函数(可选,默认使用 datetime.now) rss_items: RSS 统计条目列表(可选) rss_new_items: RSS 新增条目列表(可选) display_mode: 显示模式 ("keyword"=按关键词分组, "platform"=按平台分组) Returns: 渲染后的 HTML 字符串 """ html = """ 热点新闻分析
热点新闻分析
报告类型 """ # 处理报告类型显示 if is_daily_summary: if mode == "current": html += "当前榜单" elif mode == "incremental": html += "增量模式" else: html += "当日汇总" else: html += "实时分析" html += """
新闻总数 """ html += f"{total_titles} 条" # 计算筛选后的热点新闻数量 hot_news_count = sum(len(stat["titles"]) for stat in report_data["stats"]) html += """
热点新闻 """ html += f"{hot_news_count} 条" html += """
生成时间 """ # 使用提供的时间函数或默认 datetime.now if get_time_func: now = get_time_func() else: now = datetime.now() html += now.strftime("%m-%d %H:%M") html += """
""" # 处理失败ID错误信息 if report_data["failed_ids"]: html += """
⚠️ 请求失败的平台
    """ for id_value in report_data["failed_ids"]: html += f'
  • {html_escape(id_value)}
  • ' html += """
""" # 生成热点词汇统计部分的HTML stats_html = "" if report_data["stats"]: total_count = len(report_data["stats"]) for i, stat in enumerate(report_data["stats"], 1): count = stat["count"] # 确定热度等级 if count >= 10: count_class = "hot" elif count >= 5: count_class = "warm" else: count_class = "" escaped_word = html_escape(stat["word"]) stats_html += f"""
{escaped_word}
{count} 条
{i}/{total_count}
""" # 处理每个词组下的新闻标题,给每条新闻标上序号 for j, title_data in enumerate(stat["titles"], 1): is_new = title_data.get("is_new", False) new_class = "new" if is_new else "" stats_html += f"""
{j}
""" # 根据 display_mode 决定显示来源还是关键词 if display_mode == "keyword": # keyword 模式:显示来源 stats_html += f'{html_escape(title_data["source_name"])}' else: # platform 模式:显示关键词 matched_keyword = title_data.get("matched_keyword", "") if matched_keyword: stats_html += f'[{html_escape(matched_keyword)}]' # 处理排名显示 ranks = title_data.get("ranks", []) if ranks: min_rank = min(ranks) max_rank = max(ranks) rank_threshold = title_data.get("rank_threshold", 10) # 确定排名等级 if min_rank <= 3: rank_class = "top" elif min_rank <= rank_threshold: rank_class = "high" else: rank_class = "" if min_rank == max_rank: rank_text = str(min_rank) else: rank_text = f"{min_rank}-{max_rank}" stats_html += f'{rank_text}' # 处理时间显示 time_display = title_data.get("time_display", "") if time_display: # 简化时间显示格式,将波浪线替换为~ simplified_time = ( time_display.replace(" ~ ", "~") .replace("[", "") .replace("]", "") ) stats_html += ( f'{html_escape(simplified_time)}' ) # 处理出现次数 count_info = title_data.get("count", 1) if count_info > 1: stats_html += f'{count_info}次' stats_html += """
""" # 处理标题和链接 escaped_title = html_escape(title_data["title"]) link_url = title_data.get("mobile_url") or title_data.get("url", "") if link_url: escaped_url = html_escape(link_url) stats_html += f'{escaped_title}' else: stats_html += escaped_title stats_html += """
""" stats_html += """
""" # 生成新增新闻区域的HTML new_titles_html = "" if report_data["new_titles"]: new_titles_html += f"""
本次新增热点 (共 {report_data['total_new_count']} 条)
""" for source_data in report_data["new_titles"]: escaped_source = html_escape(source_data["source_name"]) titles_count = len(source_data["titles"]) new_titles_html += f"""
{escaped_source} · {titles_count}条
""" # 为新增新闻也添加序号 for idx, title_data in enumerate(source_data["titles"], 1): ranks = title_data.get("ranks", []) # 处理新增新闻的排名显示 rank_class = "" if ranks: min_rank = min(ranks) if min_rank <= 3: rank_class = "top" elif min_rank <= title_data.get("rank_threshold", 10): rank_class = "high" if len(ranks) == 1: rank_text = str(ranks[0]) else: rank_text = f"{min(ranks)}-{max(ranks)}" else: rank_text = "?" new_titles_html += f"""
{idx}
{rank_text}
""" # 处理新增新闻的链接 escaped_title = html_escape(title_data["title"]) link_url = title_data.get("mobile_url") or title_data.get("url", "") if link_url: escaped_url = html_escape(link_url) new_titles_html += f'{escaped_title}' else: new_titles_html += escaped_title new_titles_html += """
""" new_titles_html += """
""" new_titles_html += """
""" # 生成 RSS 统计内容 def render_rss_stats_html(items: List[Dict], title: str = "RSS 订阅更新") -> str: if not items: return "" rss_html = "" rss_count = len(items) rss_html += f"""
{title}
{rss_count} 条
""" # 按 feed_id 分组 feeds_grouped = {} for item in items: feed_id = item.get("feed_id", "unknown") if feed_id not in feeds_grouped: feeds_grouped[feed_id] = { "name": item.get("feed_name", feed_id), "items": [] } feeds_grouped[feed_id]["items"].append(item) # 渲染每个 feed 分组 for feed_id, feed_data in feeds_grouped.items(): feed_name = feed_data["name"] feed_items = feed_data["items"] feed_item_count = len(feed_items) rss_html += f"""
{html_escape(feed_name)}
{feed_item_count} 条
""" for item in feed_items: item_title = item.get("title", "") url = item.get("url", "") published_at = item.get("published_at", "") author = item.get("author", "") # 格式化发布时间 time_str = "" if published_at: if isinstance(published_at, datetime): time_str = published_at.strftime("%m-%d %H:%M") else: time_str = str(published_at)[:16] if len(str(published_at)) > 16 else str(published_at) rss_html += """
""" if time_str: rss_html += f'{html_escape(time_str)}' if author: rss_html += f'by {html_escape(author)}' rss_html += """
""" escaped_title = html_escape(item_title) if url: escaped_url = html_escape(url) rss_html += f'{escaped_title}' else: rss_html += escaped_title rss_html += """
""" rss_html += """
""" rss_html += """
""" return rss_html # 生成 RSS 统计和新增 HTML rss_stats_html = render_rss_stats_html(rss_items, "RSS 订阅更新") if rss_items else "" rss_new_html = render_rss_stats_html(rss_new_items, "RSS 新增更新") if rss_new_items else "" # 根据配置决定内容顺序(与推送逻辑一致) if reverse_content_order: # 新增在前,统计在后 # 顺序:热榜新增 → RSS新增 → 热榜统计 → RSS统计 html += new_titles_html + rss_new_html + stats_html + rss_stats_html else: # 默认:统计在前,新增在后 # 顺序:热榜统计 → RSS统计 → 热榜新增 → RSS新增 html += stats_html + rss_stats_html + new_titles_html + rss_new_html html += """
""" return html