# -*- 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'' 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 _normalize_summary_bullets(payload: Dict[str, Any], items: List[str], target_count: int = 5) -> List[str]: normalized = [str(item or "").strip() for item in items if str(item or "").strip()] if len(normalized) >= target_count: return normalized[:target_count] top_terms = [str(item.get("term") or "").strip() for item in (payload.get("top_terms", []) or []) if str(item.get("term") or "").strip()] merged_templates = [str(item.get("text") or "").strip() for item in (payload.get("merged_templates", []) or []) if str(item.get("text") or "").strip()] peak_buckets = payload.get("peak_buckets", []) or [] representative_messages = payload.get("representative_messages", []) or [] supplements: List[str] = [] if top_terms: supplements.append(f"讨论焦点比较集中，弹幕反复围绕 {'、'.join(top_terms[:5])} 展开。") if merged_templates: sample_templates = "；".join(text[:24] for text in merged_templates[:3]) supplements.append(f"复读和共识梗比较强，重复内容主要集中在 {sample_templates}。") if peak_buckets: top_bucket = peak_buckets[0] bucket_terms = [str(term.get('term') or '').strip() for term in (top_bucket.get("top_terms", []) or []) if str(term.get('term') or '').strip()] if bucket_terms: supplements.append( f"高峰时段出现在 {str(top_bucket.get('start_time') or '')[-8:-3]} 前后，话题明显偏向 {'、'.join(bucket_terms[:4])}。" ) if representative_messages: supplements.append("代表性发言里既有操作反馈，也有玩梗调侃和情绪宣泄，互动意愿比较强。") existing = set(normalized) for item in supplements: if item not in existing: normalized.append(item) existing.add(item) if len(normalized) >= target_count: break return normalized[:target_count] def _build_template_items(payload: Dict[str, Any], limit: int = 8) -> List[str]: items: List[str] = [] seen = set() def push(text: str, suffix: str = "") -> None: value = str(text or "").strip() if not value: return normalized_key = value if normalized_key in seen: return seen.add(normalized_key) items.append(f"{value}{suffix}".strip()) for item in (payload.get("merged_templates", []) or [])[:6]: text = str(item.get("text") or "").strip() count = int(item.get("count", 0) or 0) if text: push(text[:72], f"（{count}次）") for item in (payload.get("repeated_messages", []) or [])[:6]: text = str(item.get("text") or "").strip() count = int(item.get("count", 0) or 0) if text: push(text[:72], f"（{count}次）") for item in (payload.get("burst_terms", []) or [])[:6]: text = str(item.get("text") or "").strip() count = int(item.get("count", 0) or 0) if text: push(text[:36], f"（爆发 {count} 次）") for item in (payload.get("top_terms", []) or [])[:6]: term = str(item.get("term") or "").strip() count = int(item.get("count", 0) or 0) if term: push(term, f"（{count}次提及）") return items[:limit] def _render_insight_cards(items: List[str]) -> str: labels = ["主线", "情绪", "梗点", "节奏", "反馈", "补充"] blocks = [] for idx, item in enumerate(items[:6]): extra_class = " full-span" if len(items[:6]) % 2 == 1 and idx == len(items[:6]) - 1 else "" blocks.append( f'
    ' 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), "高等级老观众"), ]) template_items = _build_template_items(payload) 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) danmu_bullets = _normalize_summary_bullets(payload, danmu_bullets, target_count=5) 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")}
    ABOT · Douyu Report Template
    """ return html_doc