Fix PDF: strip emoji/unicode chars that Helvetica can't render

ml/report_generator.py

@@ -7,12 +7,298 @@ Uses fpdf2 — no system dependencies required.
 from __future__ import annotations
 
 import io
+import re
 from datetime import datetime
 from collections import Counter
 
 from fpdf import FPDF
 
 
+# ── Colour palette (matches dashboard theme) ──────────────────────────────────
+_C_BG        = (7,   7,  15)
+_C_CARD      = (15,  15,  30)
+_C_ACCENT    = (124, 58, 237)
+_C_POS       = (34, 197,  94)
+_C_NEU       = (234, 179,   8)
+_C_NEG       = (239,  68,  68)
+_C_TEXT1     = (241, 245, 249)
+_C_TEXT2     = (148, 163, 184)
+_C_WHITE     = (255, 255, 255)
+_C_DIVIDER   = (30,  30,  50)
+
+TOPIC_COLORS = {
+    "Appreciation":    (245, 158,  11),
+    "Question":        ( 59, 130, 246),
+    "Request/Feedback":(139,  92, 246),
+    "Promo":           (236,  72, 153),
+    "Spam":            (239,  68,  68),
+    "General":         (107, 114, 128),
+    "MCQ Answer":      ( 16, 185, 129),
+}
+
+
+def _safe(text: str, max_len: int = 80) -> str:
+    """Strip emoji and non-Latin-1 characters so Helvetica can render them."""
+    if not text:
+        return ""
+    # Remove emoji and symbols outside Latin-1
+    cleaned = re.sub(r'[^\x00-\xFF]', '', str(text))
+    # Collapse multiple spaces
+    cleaned = re.sub(r'\s+', ' ', cleaned).strip()
+    return cleaned[:max_len]
+
+
+class LivePulsePDF(FPDF):
+
+    def __init__(self):
+        super().__init__(orientation="P", unit="mm", format="A4")
+        self.set_auto_page_break(auto=True, margin=15)
+        self.set_margins(15, 15, 15)
+
+    def header(self):
+        self.set_fill_color(*_C_BG)
+        self.rect(0, 0, 210, 20, "F")
+        self.set_font("Helvetica", "B", 11)
+        self.set_text_color(*_C_ACCENT)
+        self.set_y(6)
+        self.cell(0, 8, "LivePulse  |  YouTube Live Chat Analytics", align="L")
+        self.set_font("Helvetica", "", 8)
+        self.set_text_color(*_C_TEXT2)
+        self.cell(0, 8, f"Generated {datetime.now().strftime('%d %b %Y  %H:%M')}", align="R")
+        self.ln(4)
+
+    def footer(self):
+        self.set_y(-12)
+        self.set_font("Helvetica", "", 8)
+        self.set_text_color(*_C_TEXT2)
+        self.cell(0, 8, f"Page {self.page_no()}", align="C")
+
+    def section_title(self, title: str, pill: str = "") -> None:
+        self.set_fill_color(*_C_CARD)
+        self.set_draw_color(*_C_ACCENT)
+        self.set_line_width(0.5)
+        self.rect(15, self.get_y(), 180, 9, "FD")
+        self.set_font("Helvetica", "B", 10)
+        self.set_text_color(*_C_TEXT1)
+        self.set_x(17)
+        self.cell(140, 9, _safe(title), ln=0)
+        if pill:
+            self.set_font("Helvetica", "", 8)
+            self.set_text_color(*_C_ACCENT)
+            self.cell(0, 9, _safe(pill), align="R")
+        self.ln(11)
+
+    def stat_box(self, label: str, value: str, color: tuple, x: float, y: float, w: float = 42, h: float = 18) -> None:
+        self.set_fill_color(*_C_CARD)
+        self.set_draw_color(*color)
+        self.set_line_width(0.4)
+        self.rect(x, y, w, h, "FD")
+        self.set_fill_color(*color)
+        self.rect(x, y, w, 1.5, "F")
+        self.set_font("Helvetica", "B", 14)
+        self.set_text_color(*color)
+        self.set_xy(x, y + 3)
+        self.cell(w, 7, _safe(value, 12), align="C")
+        self.set_font("Helvetica", "", 7)
+        self.set_text_color(*_C_TEXT2)
+        self.set_xy(x, y + 10)
+        self.cell(w, 5, _safe(label.upper(), 20), align="C")
+
+    def h_bar(self, label: str, value: int, max_val: int, color: tuple, bar_w: float = 100) -> None:
+        y = self.get_y()
+        self.set_font("Helvetica", "", 8)
+        self.set_text_color(*_C_TEXT1)
+        self.set_x(17)
+        self.cell(55, 6, _safe(label, 35), ln=0)
+        self.set_fill_color(*_C_DIVIDER)
+        self.rect(73, y + 1, bar_w, 4, "F")
+        fill_w = (value / max(max_val, 1)) * bar_w
+        self.set_fill_color(*color)
+        self.rect(73, y + 1, fill_w, 4, "F")
+        self.set_font("Helvetica", "B", 8)
+        self.set_text_color(*_C_TEXT2)
+        self.set_xy(175, y)
+        self.cell(20, 6, str(value), align="R")
+        self.ln(7)
+
+    def table_header(self, cols: list[tuple[str, float]]) -> None:
+        self.set_fill_color(*_C_CARD)
+        self.set_font("Helvetica", "B", 8)
+        self.set_text_color(*_C_ACCENT)
+        for label, w in cols:
+            self.cell(w, 7, _safe(label), border=0, fill=True, align="L")
+        self.ln(7)
+        self.set_draw_color(*_C_ACCENT)
+        self.set_line_width(0.3)
+        self.line(15, self.get_y(), 195, self.get_y())
+        self.ln(1)
+
+    def table_row(self, values: list[tuple[str, float]], alt: bool = False) -> None:
+        if alt:
+            self.set_fill_color(20, 20, 35)
+        else:
+            self.set_fill_color(*_C_BG)
+        self.set_font("Helvetica", "", 8)
+        self.set_text_color(*_C_TEXT1)
+        for val, w in values:
+            self.cell(w, 6, _safe(str(val), 40), border=0, fill=True, align="L")
+        self.ln(6)
+
+
+def generate_report(
+    all_data: list[dict],
+    stream_title: str = "LivePulse Stream",
+    msg_limit: int = 100,
+) -> bytes:
+    pdf = LivePulsePDF()
+    pdf.add_page()
+
+    # Cover
+    pdf.set_fill_color(*_C_BG)
+    pdf.rect(0, 20, 210, 40, "F")
+    pdf.set_font("Helvetica", "B", 20)
+    pdf.set_text_color(*_C_TEXT1)
+    pdf.set_y(28)
+    pdf.cell(0, 10, "Dashboard Report", align="C", ln=True)
+    pdf.set_font("Helvetica", "", 11)
+    pdf.set_text_color(*_C_ACCENT)
+    pdf.cell(0, 8, _safe(stream_title, 80), align="C", ln=True)
+    pdf.set_font("Helvetica", "", 9)
+    pdf.set_text_color(*_C_TEXT2)
+    pdf.cell(0, 6, f"Total messages analysed: {len(all_data)}", align="C", ln=True)
+    pdf.ln(8)
+
+    if not all_data:
+        pdf.set_font("Helvetica", "", 11)
+        pdf.set_text_color(*_C_NEG)
+        pdf.cell(0, 10, "No data available.", align="C")
+        return bytes(pdf.output())
+
+    # Pre-compute
+    sentiments   = [m.get("sentiment", "Neutral") for m in all_data]
+    topics       = [m.get("topic", "General")     for m in all_data]
+    action_types = [m.get("action_type", "N/A")   for m in all_data]
+    authors      = [m.get("author", "Unknown")     for m in all_data]
+
+    c_pos   = sentiments.count("Positive")
+    c_neu   = sentiments.count("Neutral")
+    c_neg   = sentiments.count("Negative")
+    c_total = max(len(all_data), 1)
+
+    topic_counts  = Counter(topics)
+    action_counts = Counter(a for a in action_types if a not in ("N/A", "", None))
+    author_counts = Counter(authors)
+
+    try:
+        from datetime import datetime as _dt
+        recent = all_data[-50:]
+        n = len(recent)
+        t0 = _dt.fromisoformat(recent[0]["time"])
+        t1 = _dt.fromisoformat(recent[-1]["time"])
+        elapsed = max((t1 - t0).total_seconds() / 60, 0.1)
+        rate = round(n / elapsed, 1)
+        pos_ratio = sum(1 for m in recent if m.get("sentiment") == "Positive") / max(n, 1)
+        q_density = sum(1 for m in recent if m.get("topic") == "Question") / max(n, 1)
+        rate_norm = min(rate / 60, 1.0)
+        eng_score = round((rate_norm * 0.4 + pos_ratio * 0.4 + q_density * 0.2) * 100)
+    except Exception:
+        eng_score = 0; rate = 0.0; pos_ratio = 0.0; q_density = 0.0
+
+    # Section 1: Engagement Summary
+    pdf.section_title("Engagement Summary", "Live")
+    y0 = pdf.get_y()
+    pdf.stat_box("Engagement",  str(eng_score),                         _C_ACCENT, 15,  y0)
+    pdf.stat_box("Positive",    f"{c_pos} ({c_pos/c_total*100:.0f}%)",  _C_POS,    59,  y0)
+    pdf.stat_box("Neutral",     f"{c_neu} ({c_neu/c_total*100:.0f}%)",  _C_NEU,    103, y0)
+    pdf.stat_box("Negative",    f"{c_neg} ({c_neg/c_total*100:.0f}%)",  _C_NEG,    147, y0)
+    pdf.set_y(y0 + 22)
+    y1 = pdf.get_y()
+    pdf.stat_box("Total Msgs",  str(c_total),          _C_TEXT2,  15,  y1)
+    pdf.stat_box("Msgs/min",    f"{rate:.1f}",          _C_ACCENT, 59,  y1)
+    pdf.stat_box("Pos ratio",   f"{pos_ratio*100:.0f}%",_C_POS,    103, y1)
+    pdf.stat_box("Q density",   f"{q_density*100:.0f}%",_C_NEU,    147, y1)
+    pdf.set_y(y1 + 22)
+    pdf.ln(4)
+
+    # Section 2: Topic Distribution
+    pdf.section_title("Topic Distribution", "All Time")
+    max_topic = max(topic_counts.values(), default=1)
+    for topic in ["Appreciation", "Question", "Request/Feedback", "Promo", "Spam", "General", "MCQ Answer"]:
+        pdf.h_bar(topic, topic_counts.get(topic, 0), max_topic, TOPIC_COLORS.get(topic, _C_TEXT2))
+    pdf.ln(4)
+
+    # Section 3: Action Types
+    if action_counts:
+        pdf.section_title("Top Action Types", "Questions & Requests")
+        max_action = max(action_counts.values(), default=1)
+        for action, count in action_counts.most_common(15):
+            pdf.h_bar(action[:40], count, max_action, _C_ACCENT)
+        pdf.ln(4)
+
+    # Section 4: Top Contributors
+    pdf.section_title("Top Contributors", "All Time")
+    cols = [("Author", 60), ("Messages", 25), ("Positive%", 30), ("Neutral%", 30), ("Negative%", 30)]
+    pdf.table_header(cols)
+    for i, (author, count) in enumerate(author_counts.most_common(15)):
+        author_msgs = [m for m in all_data if m.get("author") == author]
+        total_a = max(len(author_msgs), 1)
+        pos_p = round(sum(1 for m in author_msgs if m.get("sentiment") == "Positive") / total_a * 100)
+        neu_p = round(sum(1 for m in author_msgs if m.get("sentiment") == "Neutral")  / total_a * 100)
+        neg_p = round(sum(1 for m in author_msgs if m.get("sentiment") == "Negative") / total_a * 100)
+        pdf.table_row([
+            (_safe(author, 28), 60), (str(count), 25),
+            (f"{pos_p}%", 30), (f"{neu_p}%", 30), (f"{neg_p}%", 30),
+        ], alt=(i % 2 == 1))
+    pdf.ln(4)
+
+    # Section 5: Recent Comments
+    pdf.add_page()
+    recent_msgs = all_data[-msg_limit:]
+    pdf.section_title("Recent Comments", f"Last {len(recent_msgs)} messages")
+    cols_c = [("Author", 40), ("Message", 90), ("Sentiment", 22), ("Topic", 28)]
+    pdf.table_header(cols_c)
+    sent_colors = {"Positive": _C_POS, "Negative": _C_NEG, "Neutral": _C_NEU}
+    for i, msg in enumerate(reversed(recent_msgs)):
+        author = _safe(msg.get("author", ""), 18)
+        text   = _safe(msg.get("text",   ""), 55)
+        sent   = msg.get("sentiment", "Neutral")
+        topic  = _safe(msg.get("topic", "General"), 14)
+        alt = (i % 2 == 1)
+        pdf.set_fill_color(20, 20, 35) if alt else pdf.set_fill_color(*_C_BG)
+        pdf.set_font("Helvetica", "", 7.5)
+        pdf.set_text_color(*_C_TEXT1)
+        pdf.cell(40, 5.5, author, border=0, fill=True)
+        pdf.cell(90, 5.5, text,   border=0, fill=True)
+        pdf.set_text_color(*sent_colors.get(sent, _C_TEXT2))
+        pdf.cell(22, 5.5, sent,  border=0, fill=True)
+        pdf.set_text_color(*TOPIC_COLORS.get(topic, _C_TEXT2))
+        pdf.cell(28, 5.5, topic, border=0, fill=True)
+        pdf.ln(5.5)
+
+    # Section 6: Questions Log
+    questions = [m for m in all_data if m.get("topic") == "Question"]
+    if questions:
+        pdf.add_page()
+        pdf.section_title("Questions Asked", f"{len(questions)} total")
+        cols_q = [("Author", 40), ("Question", 115), ("Action Type", 40)]
+        pdf.table_header(cols_q)
+        for i, msg in enumerate(reversed(questions[-100:])):
+            author = _safe(msg.get("author", ""), 18)
+            text   = _safe(msg.get("text",   ""), 65)
+            action = _safe(msg.get("action_type", "N/A"), 22)
+            alt = (i % 2 == 1)
+            pdf.set_fill_color(20, 20, 35) if alt else pdf.set_fill_color(*_C_BG)
+            pdf.set_font("Helvetica", "", 7.5)
+            pdf.set_text_color(*_C_TEXT1)
+            pdf.cell(40,  5.5, author, border=0, fill=True)
+            pdf.cell(115, 5.5, text,   border=0, fill=True)
+            pdf.set_text_color(*_C_ACCENT)
+            pdf.cell(40,  5.5, action, border=0, fill=True)
+            pdf.ln(5.5)
+
+    return bytes(pdf.output())
+
+
 # ── Colour palette (matches dashboard theme) ──────────────────────────────────
 _C_BG        = (7,   7,  15)   # dark background
 _C_CARD      = (15,  15,  30)