spb/uqo-imm1003

1"""videos.py — séances vidéo (enregistrements Zoom) : transcription synchronisée, chapitres, storyboard des diapositives.23Sources : _Site_web/videos/<code>/<nn>/4    video.mp4          enregistrement (remuxé faststart), gitignoré5    audio.m4a          piste audio seule (option « audio seulement »), gitignorée6    transcript.json    sortie whisper (segments + mots)7    meta.json          rédigé à la main : titre, date, chapitres, résumé, à retenir, questions8    chat.txt           clavardage Zoom (optionnel)9    scenes.txt         horodatages des changements d'image (ffmpeg select=scene), optionnel1011Sortie : dist/<code>/videos/<nn>/ (page + médias + captions.vtt + chapters.vtt + transcript.{json,txt,srt} + sb/*.webp)12"""13from __future__ import annotations1415import glob16import json17import os18import re19import shutil20import subprocess2122HALLUCINATIONS = re.compile(r"^(sous-titr(age|es)|merci d'avoir regardé|abonnez-vous|amara\.org|❤️|…)", re.I)232425def hms(t: float, always_h: bool = False) -> str:26    t = max(0, int(round(t)))27    h, m, s = t // 3600, (t % 3600) // 60, t % 6028    return f"{h}:{m:02d}:{s:02d}" if h or always_h else f"{m}:{s:02d}"293031def ts_vtt(t: float) -> str:32    ms = int(round(t * 1000))33    h, m, s, ms = ms // 3600000, (ms % 3600000) // 60000, (ms % 60000) // 1000, ms % 100034    return f"{h:02d}:{m:02d}:{s:02d}.{ms:03d}"353637def ts_srt(t: float) -> str:38    return ts_vtt(t).replace(".", ",")394041def human_dur(sec: float) -> str:42    m = int(round(sec / 60))43    return f"{m // 60} h {m % 60:02d}" if m >= 60 else f"{m} min"444546def clean_text(s: str) -> str:47    s = s.strip()48    s = re.sub(r"\s+", " ", s)49    s = re.sub(r"\s+([,.;:!?])", r"\1", s)50    s = re.sub(r"([«])\s*", r"\1 ", s)51    s = re.sub(r"\s*([»])", r" \1", s)52    s = re.sub(r"\s*([;:!?])", " \\1", s)  # espace fine française53    return s545556def load_segments(transcript: dict):57    """Segments whisper nettoyés : [{start, end, text, words}]"""58    out = []59    prev = ""60    for s in transcript.get("segments", []):61        text = s.get("text", "").strip()62        if not text or HALLUCINATIONS.search(text):63            continue64        if s.get("no_speech_prob", 0) > 0.85 and s.get("avg_logprob", 0) < -1.0:65            continue66        if text == prev and len(text) < 40:  # boucles de répétition67            continue68        prev = text69        words = [{"w": w["word"], "s": float(w["start"]), "e": float(w["end"])} for w in s.get("words", []) if w.get("start") is not None]70        out.append({"start": float(s["start"]), "end": float(s["end"]), "text": text, "words": words})71    return out727374def make_cues(segments, max_chars: int = 84, max_dur: float = 6.0, max_gap: float = 1.6):75    """Sous-titres : flux de mots (tous segments confondus) regroupé en cues d'une à deux lignes, coupées à la ponctuation."""76    words = []77    for seg in segments:78        if seg["words"]:79            words.extend(seg["words"])80        else:81            words.append({"w": " " + seg["text"], "s": seg["start"], "e": seg["end"]})82    cues, cur, t0 = [], [], None83    for i, w in enumerate(words):84        if not cur:85            t0 = w["s"]86        cur.append(w)87        text = "".join(x["w"] for x in cur).strip()88        last = i == len(words) - 189        nxt = None if last else words[i + 1]90        tok = w["w"].strip()91        ends_sent = bool(re.search(r"[.!?…]$", tok))92        ends_clause = bool(re.search(r"[,;:]$", tok))93        gap = (nxt["s"] - w["e"]) if nxt else 094        nxt_len = len(text) + (len(nxt["w"]) if nxt else 0)95        dur = w["e"] - t096        cut = last or gap > max_gap or nxt_len > max_chars or dur >= max_dur or (ends_sent and len(text) > 28) or (ends_clause and len(text) > max_chars * .65)97        if cut:98            cues.append((t0, w["e"], clean_text(text)))99            cur = []100    fixed = []101    for i, (a, b, t) in enumerate(cues):102        if i + 1 < len(cues):103            b = min(b, cues[i + 1][0] - 0.02)104        if b - a < 0.8:105            b = a + 0.8106        fixed.append((a, b, t))107    return fixed108109110def make_blocks(segments, target_chars: int = 260, max_gap: float = 2.5):111    """Paragraphes de lecture : fusion des segments jusqu'à ~260 caractères et fin de phrase."""112    blocks = []113    cur = None114    for seg in segments:115        t = clean_text(seg["text"])116        if cur and (seg["start"] - cur["end"] > max_gap and len(cur["text"]) > 60):117            blocks.append(cur)118            cur = None119        if cur is None:120            cur = {"start": seg["start"], "end": seg["end"], "text": t}121            continue122        cur["text"] = (cur["text"] + " " + t).strip()123        cur["end"] = seg["end"]124        if len(cur["text"]) >= target_chars and re.search(r"[.!?…]$", cur["text"]):125            blocks.append(cur)126            cur = None127    if cur:128        blocks.append(cur)129    for b in blocks:130        b["text"] = b["text"][:1].upper() + b["text"][1:]131    return blocks132133134def parse_chat(path: str):135    rows = []136    if not os.path.exists(path):137        return rows138    for line in open(path, encoding="utf-8", errors="replace"):139        m = re.match(r"\s*(\d{1,2}:\d{2}:\d{2})\s+From\s+(.*?)\s*(?:to\s+.*?)?\s*:\s*(.*)$", line.strip())140        if m:141            rows.append({"time": m.group(1), "who": m.group(2).strip(), "text": m.group(3).strip()})142    return rows143144145def parse_scenes(path: str, min_gap: float = 2.5):146    if not os.path.exists(path):147        return []148    ts = []149    for line in open(path):150        line = line.strip()151        if not line:152            continue153        try:154            t = float(line)155        except ValueError:156            continue157        if not ts or t - ts[-1] >= min_gap:158            ts.append(t)159    return ts160161162def frame(video: str, t: float, out_path: str, width: int, quality: int = 72):163    """Extrait une image (ffmpeg → PNG) puis l'encode en WebP avec Pillow (ffmpeg Homebrew sans libwebp)."""164    if os.path.exists(out_path):165        return True166    tmp = out_path + ".png"167    p = subprocess.run(["ffmpeg", "-v", "error", "-ss", f"{t:.2f}", "-i", video, "-frames:v", "1", "-vf", f"scale={width}:-2", "-y", tmp],168                       stdout=subprocess.DEVNULL, stderr=subprocess.PIPE)169    if p.returncode != 0 or not os.path.exists(tmp):170        return False171    try:172        from PIL import Image173        Image.open(tmp).convert("RGB").save(out_path, "WEBP", quality=quality, method=4)174    except Exception:  # noqa: BLE001175        shutil.move(tmp, out_path)  # repli : PNG sous extension webp (le serveur sert selon l'extension, éviter)176        return False177    finally:178        if os.path.exists(tmp):179            os.remove(tmp)180    return os.path.exists(out_path)181182183def duration_of(video: str) -> float:184    try:185        out = subprocess.run(["ffprobe", "-v", "error", "-show_entries", "format=duration", "-of", "csv=p=0", video], stdout=subprocess.PIPE, text=True, timeout=30).stdout.strip()186        return float(out)187    except Exception:  # noqa: BLE001188        return 0.0189190191def build_videos(code: str, site_dir: str, out: str, cache_dir: str, chapters, log, warn):192    """Construit dist/<code>/videos/<nn>/… ; retourne la liste des séances vidéo (dicts pour les gabarits)."""193    src_root = os.path.join(site_dir, "videos", code)194    if not os.path.isdir(src_root):195        return []196    videos = []197    for d in sorted(glob.glob(os.path.join(src_root, "[0-9][0-9]"))):198        nn = os.path.basename(d)199        n = int(nn)200        meta_p, tr_p, vid_p = os.path.join(d, "meta.json"), os.path.join(d, "transcript.json"), os.path.join(d, "video.mp4")201        if not os.path.exists(vid_p):202            warn(f"vidéo {nn} : video.mp4 manquant")203            continue204        meta = json.load(open(meta_p, encoding="utf-8")) if os.path.exists(meta_p) else {}205        transcript = json.load(open(tr_p, encoding="utf-8")) if os.path.exists(tr_p) else {"segments": []}206        segments = load_segments(transcript)207        cues = make_cues(segments)208        blocks = make_blocks(segments)209        dur = duration_of(vid_p) or (segments[-1]["end"] if segments else 0)210        ch_title = chapters[n - 1].title_text if 0 < n <= len(chapters) else meta.get("title", f"Séance {n}")211        ch_title_html = chapters[n - 1].title_html if 0 < n <= len(chapters) else ch_title212213        vdir = os.path.join(out, "videos", nn)214        os.makedirs(os.path.join(vdir, "sb"), exist_ok=True)215        # médias (copie ou lien dur pour éviter de dupliquer 130 Mo)216        for name in ("video.mp4", "audio.m4a", "chat.txt"):217            sp = os.path.join(d, name)218            if os.path.exists(sp):219                dp = os.path.join(vdir, name)220                try:221                    os.link(sp, dp)222                except OSError:223                    shutil.copy2(sp, dp)224        # sous-titres + chapitres WebVTT + SRT + TXT225        vtt = ["WEBVTT", "Kind: captions", "Language: fr", ""]226        for i, (a, b, t) in enumerate(cues, 1):227            vtt += [str(i), f"{ts_vtt(a)} --> {ts_vtt(b)}", t, ""]228        open(os.path.join(vdir, "captions.vtt"), "w", encoding="utf-8").write("\n".join(vtt))229        srt = []230        for i, (a, b, t) in enumerate(cues, 1):231            srt += [str(i), f"{ts_srt(a)} --> {ts_srt(b)}", t, ""]232        open(os.path.join(vdir, "transcript.srt"), "w", encoding="utf-8").write("\n".join(srt))233        chaps = meta.get("chapters", [])234        for i, c in enumerate(chaps):235            c["end"] = chaps[i + 1]["t"] if i + 1 < len(chaps) else dur236            c["id"] = f"c{i + 1}"237        cvtt = ["WEBVTT", "Kind: chapters", ""]238        for c in chaps:239            cvtt += [f"{ts_vtt(c['t'])} --> {ts_vtt(c['end'])}", c["title"], ""]240        open(os.path.join(vdir, "chapters.vtt"), "w", encoding="utf-8").write("\n".join(cvtt))241        txt = [f"{code.upper()} — Séance {n} — {ch_title}", f"Enregistrement Zoom du {meta.get('date_text', meta.get('date', ''))} · durée {human_dur(dur)}", "Transcription automatique (whisper large-v3-turbo), non corrigée.", ""]242        for c in chaps:243            txt.append(f"[{hms(c['t'], True)}] {c['title']}")244        txt.append("")245        for b in blocks:246            txt.append(f"[{hms(b['start'], True)}] {b['text']}")247            txt.append("")248        open(os.path.join(vdir, "transcript.txt"), "w", encoding="utf-8").write("\n".join(txt))249        # storyboard (changements d'image)250        scenes = parse_scenes(os.path.join(d, "scenes.txt"))251        cdir = os.path.join(cache_dir, "video-frames", f"{code}-{nn}")252        os.makedirs(cdir, exist_ok=True)253        sb = []254        for i, t in enumerate(scenes):255            name = f"{i:03d}.webp"256            cp = os.path.join(cdir, name)257            if frame(vid_p, t + 0.6, cp, 360):258                shutil.copy2(cp, os.path.join(vdir, "sb", name))259                sb.append({"t": round(t, 2), "src": f"/videos/{nn}/sb/{name}"})260        # affiche261        poster_t = meta.get("poster_t", chaps[0]["t"] + 2 if chaps else 60)262        poster_c = os.path.join(cdir, f"poster-{int(poster_t)}.webp")263        if frame(vid_p, poster_t, poster_c, 1280, 78):264            shutil.copy2(poster_c, os.path.join(vdir, "poster.webp"))265        # données client266        client = {"n": nn, "title": ch_title, "duration": dur, "chapters": [{"t": c["t"], "end": c["end"], "title": c["title"], "id": c["id"]} for c in chaps],267                  "blocks": [[round(b["start"], 2), round(b["end"], 2), b["text"]] for b in blocks],268                  "cues": [[round(a, 2), round(b, 2), t] for a, b, t in cues],269                  "storyboard": sb}270        json.dump(client, open(os.path.join(vdir, "transcript.json"), "w", encoding="utf-8"), ensure_ascii=False)271        words = sum(len(b["text"].split()) for b in blocks)272        sizes = {name: os.path.getsize(os.path.join(vdir, name)) for name in ("video.mp4", "audio.m4a") if os.path.exists(os.path.join(vdir, name))}273        videos.append({"n": n, "nn": nn, "title": ch_title, "title_html": ch_title_html, "meta": meta, "duration": dur, "duration_text": human_dur(dur),274                       "chapters": chaps, "blocks": blocks, "n_cues": len(cues), "words": words, "chat": parse_chat(os.path.join(d, "chat.txt")),275                       "storyboard": sb, "sizes": sizes, "has_audio": "audio.m4a" in sizes, "url": f"/videos/{nn}/"})276        log(f"  vidéo {nn} : {human_dur(dur)}, {len(segments)} segments → {len(cues)} sous-titres, {len(blocks)} paragraphes, {len(chaps)} chapitres, {len(sb)} vignettes")277    return videos278