windowed_summary.py

"""Windowed summarization.

A single LLM call over a long transcript reliably under-covers: the model
writes what it considers a complete summary and stops, leaving the tail of
the video unrepresented. Splitting the transcript into fixed time windows
and summarizing each independently makes coverage a property of the
construction rather than something the model has to sustain.

A final pass writes the overview paragraphs, which no individual window
can see well enough to produce.
"""

import re

WINDOW_MINUTES = 15


def fmt_label(total_seconds):
    """[M:SS] under an hour, [H:MM:SS] at or past it."""
    t = int(total_seconds)
    h, rem = divmod(t, 3600)
    m, s = divmod(rem, 60)
    if h:
        return f"[{h}:{m:02d}:{s:02d}]"
    return f"[{m}:{s:02d}]"


def split_windows(segments, window_minutes=WINDOW_MINUTES):
    """Split segments into consecutive time windows.

    Returns a list of (start_sec, end_sec, [segments]) with empty windows
    dropped. A video shorter than one window yields a single window, which
    makes this behave identically to a single call for short videos.
    """
    if not segments:
        return []
    end = segments[-1]["start"] + segments[-1]["duration"]
    span = window_minutes * 60
    n = max(1, int((end + span - 1) // span))
    out = []
    for i in range(n):
        lo = i * span
        hi = end if i == n - 1 else (i + 1) * span
        sel = [s for s in segments if lo <= s["start"] < hi]
        if sel:
            out.append((lo, hi, sel))
    return out


def window_transcript_text(segs, block_seconds=30):
    """Render a window's segments as '[MM:SS] text' lines grouped into blocks.

    MM is total minutes and may exceed 59; the prompt explains this so the
    model can compute the correct t= seconds value.
    """
    lines = []
    buf = []
    bstart = None
    for s in segs:
        if bstart is None:
            bstart = s["start"]
        buf.append(s["text"])
        if s["start"] - bstart >= block_seconds:
            t = int(bstart)
            lines.append(f"[{t // 60:02d}:{t % 60:02d}] " + " ".join(buf))
            buf = []
            bstart = None
    if buf:
        t = int(bstart)
        lines.append(f"[{t // 60:02d}:{t % 60:02d}] " + " ".join(buf))
    return "\n".join(lines)


def build_window_prompt(video_id, lo, hi, body, context_hint="", is_last=False):
    hint = f"\nCONTEXT: {context_hint}\n" if context_hint else ""
    closing = (
        "\n- This is the final stretch of the video. If there is closing guidance "
        "or a summation, make sure it is captured."
        if is_last else ""
    )
    return f"""You are summarizing ONE EXCERPT of a longer video (ID: {video_id}).
This excerpt covers minutes {int(lo)//60} to {int(hi)//60}.
{hint}
Summarize THIS EXCERPT ONLY. Do not speculate about content outside it.

Transcript lines look like `[MM:SS] text`, where MM is TOTAL MINUTES and may
exceed 59 (e.g. `[96:30]` means 96 minutes 30 seconds).

Output format, exactly:

<h2>Descriptive Section Title <a href="https://www.youtube.com/watch?v={video_id}&t=SECONDSs" target="yt-player">[LABEL]</a></h2>
<ul>
<li><a href="https://www.youtube.com/watch?v={video_id}&t=SECONDSs" target="yt-player">[LABEL]</a> A substantive point, using <strong>bold</strong> for key terms.</li>
</ul>

Rules:
- SECONDS = MM*60 + SS from the transcript timestamp. Integer.
- LABEL = [M:SS] when total minutes < 60, else [H:MM:SS].
  Example: 96m30s -> t=5790s and label [1:36:30].
- Produce 2-3 <h2> sections, each followed by one <ul> of 3-5 <li> bullets.
- Every <h2> and <li> carries a timestamp link matching where the point is made.
- Be specific and substantive. Prefer concrete claims the speaker actually makes.
- Output ONLY the HTML fragment. No markdown fences, no preamble.{closing}

TRANSCRIPT:
{body}"""


def build_intro_prompt(video_id, title, outline, context_hint="", transcript_excerpt="",
                       duration_seconds=0):
    hint = f"\nCONTEXT: {context_hint}\n" if context_hint else ""
    mins = int(duration_seconds // 60)
    brevity = ""
    if duration_seconds and duration_seconds < 300:
        brevity = (
            f"\nThis video is only {mins} minute(s) long. A clip this short cannot "
            "support a multi-step workflow description or a list of themes. Write ONE "
            "short paragraph. Do not pad.\n"
        )
    return f"""Below is the section outline and the opening transcript of a video
titled "{title}" ({mins} minutes long).
{hint}{brevity}
Write the opening of the summary page:

1. An <h1> giving a plain, accurate title for this video.
2. Then 1-3 <p> paragraphs describing what the video covers.

HARD RULES - these matter more than being interesting:
- State ONLY what is supported by the outline and transcript below. Every tool,
  product, company, ticker symbol, technique and number you mention must appear
  in the source. Do not add plausible domain detail that "would fit".
- Do not name a speaker, or say how many speakers there are, unless the source
  makes it clear.
- If you have little to work with, write less. A short accurate paragraph is
  correct; an padded one is a defect.
- Prefer the speaker's own framing over an impressive-sounding one.

Output ONLY the <h1> and <p> tags. No markdown fences, no section headings,
no commentary.

SECTION OUTLINE:
{outline}

TRANSCRIPT (opening):
{transcript_excerpt}"""


def normalize_labels(html):
    """Recompute every timestamp label from its own t= value.

    The model is asked to compute SECONDS = MM*60 + SS and to render a matching
    label. It occasionally slips by a few seconds on one or the other. The t=
    value drives where the link actually lands, so it is treated as the source
    of truth and the visible label is regenerated from it. This makes label
    drift structurally impossible rather than something to audit for later.
    """
    def repl(m):
        return m.group(1) + fmt_label(int(m.group(2)))
    html = re.sub(r'((?:[?&])t=(\d+)s"[^>]*>)\[[\d:]+\]', repl, html)
    html = re.sub(r'((?:#)t=(\d+)"[^>]*>)\[[\d:]+\]', repl, html)
    return html


def fix_mojibake(text):
    """Repair UTF-8 bytes that were decoded as latin-1 somewhere upstream.

    Applied at generation time, not just as a post-hoc repair, so a rebuild
    cannot reintroduce garbled em dashes and accented characters.
    """
    def repl(m):
        try:
            return m.group(0).encode('latin-1').decode('utf-8')
        except (UnicodeEncodeError, UnicodeDecodeError):
            return m.group(0)
    return re.sub(r'[\xC2-\xF4][\x80-\xBF]{1,3}', repl, text)


def strip_fences(text):
    text = re.sub(r'^\s*```(?:html)?\s*\n?', '', text)
    text = re.sub(r'\n?\s*```\s*$', '', text)
    return text.strip()


def build_dropdowns(section_seconds, segments, edited_paragraphs, fn_edited, fn_raw):
    """Build transcript excerpt dropdowns for each section, programmatically.

    Deterministic on purpose: the transcript text is quoted from the actual
    transcript rather than regenerated by the model, so it cannot drift.
    """
    video_end = int(segments[-1]["start"] + segments[-1]["duration"]) if segments else 0
    edited = []
    for para in edited_paragraphs:
        m = re.search(r'id="t(\d+)"', para)
        if m:
            edited.append((int(m.group(1)), para))

    out = {}
    for i, secs in enumerate(section_seconds):
        nxt = section_seconds[i + 1] if i + 1 < len(section_seconds) else video_end
        ed = [html for t, html in edited if secs <= t < nxt][:3]
        ed_excerpt = "\n".join(ed)
        raw_segs = [s for s in segments if secs <= s["start"] < min(secs + 90, nxt)]
        raw_text = " ".join(s["text"] for s in raw_segs)
        if len(raw_text) > 500:
            raw_text = raw_text[:497] + "..."
        out[secs] = (
            "<details>\n<summary>Transcript excerpt - edited</summary>\n"
            + ed_excerpt
            + f'\n<p><a href="{fn_edited}#t{secs}">Read full edited transcript at this point &rarr;</a></p>\n'
            + "</details>\n<details>\n<summary>Transcript excerpt - raw</summary>\n"
            + f"<p>{raw_text}</p>\n"
            + f'<p><a href="{fn_raw}#t{secs}">Read full raw transcript at this point &rarr;</a></p>\n'
            + "</details>"
        )
    return out


def splice_dropdowns(body, dropdowns):
    """Insert each section's dropdowns immediately after its </ul>."""
    parts = re.split(r'(?=<h2>)', body)
    out = []
    for part in parts:
        if not part.strip().startswith("<h2>"):
            out.append(part)
            continue
        m = re.search(r't=(\d+)s', part)
        if not m:
            out.append(part)
            continue
        secs = int(m.group(1))
        block = dropdowns.get(secs, "")
        idx = part.rfind("</ul>")
        if idx == -1:
            out.append(part + "\n" + block)
        else:
            cut = idx + len("</ul>")
            out.append(part[:cut] + "\n" + block + part[cut:])
    return "".join(out)


def coverage_ratio(body, segments):
    """max(section/bullet timestamp) / video duration. 1.0 means full span."""
    if not segments:
        return 0.0
    end = segments[-1]["start"] + segments[-1]["duration"]
    ts = [int(x) for x in re.findall(r't=(\d+)s', body)]
    if not ts or end <= 0:
        return 0.0
    return max(ts) / end


def generate_windowed_summary(video_id, title, segments, config, call_llm,
                              progress_callback=None, context_hint="",
                              edited_paragraphs=None,
                              fn_edited="transcript_edited.html",
                              fn_raw="transcript_full.html",
                              window_minutes=WINDOW_MINUTES):
    """Summarize a video window by window, then write the overview.

    call_llm(messages, max_tokens, purpose, progress_label) -> (content, usage)
    Injected so this module stays free of transport concerns.

    Returns (html_body, total_usage).
    """
    windows = split_windows(segments, window_minutes)
    total_in = total_out = 0
    fragments = []

    for i, (lo, hi, segs) in enumerate(windows):
        if progress_callback:
            progress_callback(f"Summarizing window {i+1}/{len(windows)} ({int(lo)//60}-{int(hi)//60}m)")
        prompt = build_window_prompt(
            video_id, lo, hi, window_transcript_text(segs),
            context_hint, is_last=(i == len(windows) - 1),
        )
        content, usage = call_llm(
            [{"role": "user", "content": prompt}],
            config.get("max_clean_tokens", 6000),
            f"summary_window_{i+1}",
            f"Window {i+1}/{len(windows)}",
        )
        total_in += usage.get("prompt_tokens", 0)
        total_out += usage.get("completion_tokens", 0)
        fragments.append(strip_fences(content))

    body = fix_mojibake(normalize_labels("\n\n".join(fragments)))

    # Overview pass: no single window can see the whole video.
    outline = "\n".join(
        re.sub(r'<[^>]+>', '', t).strip()
        for t in re.findall(r'<h2>(.*?)</h2>', body, re.S)
    )
    if progress_callback:
        progress_callback("Writing overview")
    intro, usage = call_llm(
        [{"role": "user", "content": build_intro_prompt(
            video_id, title, outline, context_hint,
            transcript_excerpt=window_transcript_text(segments[:120]),
            duration_seconds=(segments[-1]["start"] + segments[-1]["duration"]) if segments else 0,
        )}],
        2000,
        "summary_intro",
        "Overview",
    )
    total_in += usage.get("prompt_tokens", 0)
    total_out += usage.get("completion_tokens", 0)
    intro = fix_mojibake(strip_fences(intro))

    if edited_paragraphs:
        secs_list = [int(x) for x in re.findall(r'<h2>.*?t=(\d+)s', body, re.S)]
        body = splice_dropdowns(
            body, build_dropdowns(secs_list, segments, edited_paragraphs, fn_edited, fn_raw)
        )

    return intro + "\n\n" + body, {"prompt_tokens": total_in, "completion_tokens": total_out}