# -*- coding: utf-8 -*- import html from typing import Any, Dict, List def _escape(value: Any) -> str: return html.escape(str(value or "")) def _render_metric_card(label: str, value: Any, hint: str = "") -> str: return ( '

' f'

{_escape(label)}

' f'

{_escape(value)}

' f'

{_escape(hint)}

' "

" ) def _render_list(items: List[str], item_class: str = "bullet-list") -> str: if not items: return "" lis = "".join(f'

{_escape(item)}

' for item in items if str(item or "").strip()) return f'

{lis}' if lis else "" def _split_summary_blocks(danmu_summary: str) -> tuple[str, List[str]]: lead_parts = [] insight_items = [] for line in str(danmu_summary or "").splitlines(): stripped = line.strip() if not stripped: continue if stripped.startswith("- "): insight_items.append(stripped[2:].strip()) else: lead_parts.append(stripped) lead = " ".join(lead_parts).strip() return lead, insight_items def _render_insight_cards(items: List[str]) -> str: labels = ["主线", "情绪", "梗点", "节奏", "反馈", "补充"] blocks = [] for idx, item in enumerate(items[:6]): blocks.append( '

' f'

{_escape(labels[idx] if idx < len(labels) else "观察")}

' f'

{_escape(item)}

' "

" ) return "".join(blocks) def _render_badges(top_badges: List[Dict[str, Any]]) -> str: blocks = [] for item in top_badges[:6]: badge_name = str(item.get("badge_name") or "").strip() if not badge_name: continue blocks.append( '

' f'{_escape(badge_name)}' f'{_escape(item.get("user_count", 0))}人 / {_escape(item.get("message_count", 0))}条' "

" ) return "".join(blocks) def _render_hot_times(peak_buckets: List[Dict[str, Any]]) -> str: blocks = [] for item in peak_buckets[:3]: start_time = str(item.get("start_time") or "")[-8:-3] terms = [str(term.get("term") or "").strip() for term in (item.get("top_terms", []) or [])[:4]] terms = [term for term in terms if term] blocks.append( '

' f'

{_escape(start_time)}

' f'

{_escape(item.get("message_count", 0))} 条弹幕

' f'

{_escape(" / ".join(terms))}

' "

" ) return "".join(blocks) def render_daily_report_html( payload: Dict[str, Any], danmu_summary: str, operator_summary_lines: List[str], ) -> str: meta = payload.get("report_meta", {}) or {} operator = payload.get("operator_metrics", {}) or {} title_name = str(meta.get("nickname") or meta.get("room_name") or meta.get("room_id") or "主播") subtitle = ( f"{meta.get('anchor_day', '')} | 场次 {meta.get('session_count', 0)}" f" | 弹幕 {meta.get('message_count', 0)} | 活跃用户 {meta.get('unique_user_count', 0)}" ) metrics_html = "".join([ _render_metric_card("活跃用户", meta.get("unique_user_count", 0), "当天参与弹幕的去重人数"), _render_metric_card("带牌活跃", operator.get("fans_badge_user_count", 0), "带粉丝牌发言用户"), _render_metric_card("10+粉丝牌", operator.get("high_fans_level_user_count", 0), "高粘性活跃用户"), _render_metric_card("30+等级用户", operator.get("high_room_level_user_count", 0), "高等级老观众"), ]) merged_templates = payload.get("merged_templates", []) or [] template_items = [ f"{str(item.get('text') or '').strip()[:72]}（{int(item.get('count', 0) or 0)}次）" for item in merged_templates[:5] if str(item.get("text") or "").strip() ] top_active_users = payload.get("operator_metrics", {}).get("top_active_users", []) or [] active_user_items = [] for item in top_active_users[:10]: nickname = str(item.get("nickname") or item.get("uid") or "").strip() fans_name = str(item.get("fans_name") or "").strip() message_count = int(item.get("message_count", 0) or 0) if fans_name: active_user_items.append(f"{nickname} | {fans_name} | {message_count}条") else: active_user_items.append(f"{nickname} | {message_count}条") lead_summary, danmu_bullets = _split_summary_blocks(danmu_summary) html_doc = f"""

DOUYU DAILY REPORT

{_escape(title_name)}

{_escape(subtitle)}

{metrics_html}

弹幕总结

整体观察

{_escape(lead_summary)}

{_render_insight_cards(danmu_bullets)}

高频梗

{_render_list(template_items)}

{_render_hot_times(payload.get("peak_buckets", []) or [])}

运营数据总结

{_render_list(operator_summary_lines)}

活跃牌子

{_render_badges(operator.get("top_badges", []) or [])}

核心发言用户

{_render_list(active_user_items, "compact-user-list")}

""" return html_doc