Reset main to commit fad3e19 - restore static/yt_live.js version

dvr.py

@@ -1,452 +0,0 @@
-"""
-DVR / Timeshift cho VTV LIVE
-============================
-2 tính năng:
-  1) Timeshift "xem lại LIVE lùi tới 5 phút": ffmpeg ghi luồng m3u8 thành các
-     segment HLS 2s, playlist giữ ~5 phút gần nhất (omit_endlist) -> hls.js cho
-     phép tua lùi trong cửa sổ này.
-  2) Hẹn giờ record: đặt lịch (giờ bắt đầu + thời lượng), tới giờ ffmpeg ghi ra
-     file .mp4, ghi xong (tuỳ chọn) upload lên HF Dataset để giữ vĩnh viễn
-     (vì storage của Space là ephemeral, restart sẽ mất).
-
-Yêu cầu: ffmpeg (Dockerfile của Space đã cài sẵn).
-Tích hợp: trong app chính ->  app.include_router(dvr_router)
-          và gọi resume_schedules() lúc khởi động.
-"""
-import os, re, json, time, shutil, threading, subprocess, signal
-from datetime import datetime, timedelta, timezone
-from fastapi import APIRouter, Query, Body
-from fastapi.responses import JSONResponse, FileResponse, Response
-
-VN_TZ = timezone(timedelta(hours=7))
-router = APIRouter()
-
-# Thư mục làm việc. Trên HF Space dùng /data nếu có Persistent Storage, nếu không /tmp.
-BASE = "/data/dvr" if os.path.isdir("/data") and os.access("/data", os.W_OK) else "/tmp/dvr"
-BUF_DIR = os.path.join(BASE, "buffer")     # timeshift buffer (HLS segments)
-REC_DIR = os.path.join(BASE, "recordings") # file mp4 đã ghi
-os.makedirs(BUF_DIR, exist_ok=True)
-os.makedirs(REC_DIR, exist_ok=True)
-SCHED_FILE = os.path.join(BASE, "schedules.json")
-
-FFMPEG = shutil.which("ffmpeg") or os.environ.get("FFMPEG", "/usr/bin/ffmpeg")
-YTDP_PATH = shutil.which("yt-dlp") or os.environ.get("YTDLP", "/usr/bin/yt-dlp")
-
-# Cửa sổ timeshift: 5 phút = 150 segment * 2s
-SEG_SEC = 2
-TIMESHIFT_WINDOW_SEC = 5 * 60
-LIST_SIZE = TIMESHIFT_WINDOW_SEC // SEG_SEC  # 150
-
-# ---- lấy URL m3u8 thật của kênh (tái dùng backend vtv_api nếu có) ----
-def _resolve_stream(channel_id):
-    """Resolve stream URL với fallback chain. Trả về list các URLs thử."""
-    try:
-        from vtv_api import fetch_vtv_stream, XEMTV_US_ENDPOINTS, FPTPLAY_URLS, VTVGO_FAILOVER
-        # Primary URL từ API
-        primary = fetch_vtv_stream(channel_id)
-        candidates = []
-        if primary:
-            candidates.append(primary)
-        # Thêm fallback URLs trực tiếp
-        ch = channel_id.lower().strip()
-        if ch in FPTPLAY_URLS and FPTPLAY_URLS[ch] not in candidates:
-            candidates.append(FPTPLAY_URLS[ch])
-        if ch in VTVGO_FAILOVER and VTVGO_FAILOVER[ch] not in candidates:
-            candidates.append(VTVGO_FAILOVER[ch])
-        return candidates if candidates else None
-    except Exception:
-        return None
-
-# =====================================================================
-# 1) TIMESHIFT BUFFER
-# =====================================================================
-_buffers = {}  # channel_id -> Popen
-_buf_lock = threading.Lock()
-
-def _try_ffmpeg_stream(url):
-    """Test xem ffmpeg có mở được stream URL không (timeout 10s)."""
-    try:
-        cmd = [FFMPEG, "-hide_banner", "-loglevel", "error",
-                "-reconnect", "1", "-reconnect_streamed", "1", "-reconnect_delay_max", "5",
-                "-i", url, "-t", "1", "-c", "copy", "-f", "null", "-"]
-        r = subprocess.run(cmd, capture_output=True, timeout=15)
-        return r.returncode == 0
-    except Exception:
-        return False
-
-def _start_buffer(channel_id, stream_urls):
-    """Khởi động buffer, retry với nhiều sources nếu cần."""
-    if isinstance(stream_urls, str):
-        stream_urls = [stream_urls]
-    ch_dir = os.path.join(BUF_DIR, channel_id)
-    os.makedirs(ch_dir, exist_ok=True)
-    playlist = os.path.join(ch_dir, "index.m3u8")
-    seg = os.path.join(ch_dir, "seg_%05d.ts")
-
-    # Xóa segments cũ
-    for f in os.listdir(ch_dir):
-        if f.endswith(".ts"):
-            try: os.remove(os.path.join(ch_dir, f))
-            except: pass
-
-    # Chọn source đầu tiên hoạt động
-    working_url = None
-    for url in stream_urls:
-        print(f"[DVR] Testing source for {channel_id}: {url[:80]}...", flush=True)
-        if _try_ffmpeg_stream(url):
-            working_url = url
-            print(f"[DVR] Source OK: {url[:80]}", flush=True)
-            break
-        else:
-            print(f"[DVR] Source FAIL: {url[:80]}", flush=True)
-
-    if not working_url:
-        print(f"[DVR] ERROR: No working source for {channel_id}", flush=True)
-        return None
-
-    cmd = [
-        FFMPEG, "-hide_banner", "-loglevel", "warning", "-y",
-        "-reconnect", "1", "-reconnect_streamed", "1", "-reconnect_delay_max", "10",
-        "-i", working_url,
-        "-c", "copy", "-copyts", "-start_at_zero",
-        "-f", "hls",
-        "-hls_time", str(SEG_SEC),
-        "-hls_list_size", str(LIST_SIZE),
-        "-hls_flags", "delete_segments+omit_endlist+independent_segments+program_date_time",
-        "-hls_segment_type", "mpegts",
-        "-hls_segment_filename", seg,
-        playlist,
-    ]
-    p = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
-    # Log output trong background
-    def _log_output():
-        for line in iter(p.stdout.readline, b''):
-            line_s = line.decode("utf-8", errors="ignore").strip()
-            if line_s:
-                print(f"[DVR:{channel_id}] {line_s}", flush=True)
-    threading.Thread(target=_log_output, daemon=True).start()
-    return p
-
-@router.get("/api/dvr/buffer/start/{channel_id}")
-def buffer_start(channel_id: str):
-    channel_id = channel_id.lower().strip()
-    with _buf_lock:
-        existing = _buffers.get(channel_id)
-        if existing and existing.get("popen") and existing["popen"].poll() is None:
-            return JSONResponse({"status": "already_running", "channel": channel_id,
-                                 "playlist": f"/api/dvr/buffer/playlist/{channel_id}"})
-        urls = _resolve_stream(channel_id)
-        if not urls:
-            return JSONResponse({"error": "stream not found"}, status_code=404)
-        # Chạy buffer trong background thread
-        def _run():
-            p = _start_buffer(channel_id, urls)
-            with _buf_lock:
-                if p:
-                    _buffers[channel_id] = {"popen": p, "urls": urls, "error": None}
-                else:
-                    _buffers[channel_id] = {"popen": None, "urls": urls,
-                                             "error": "all_sources_failed",
-                                             "hint": "FPTPlay/VTVGo CDN block từ server IP. Cần proxy VN."}
-        threading.Thread(target=_run, daemon=True).start()
-    return JSONResponse({"status": "starting", "channel": channel_id,
-                         "window_sec": TIMESHIFT_WINDOW_SEC,
-                         "playlist": f"/api/dvr/buffer/playlist/{channel_id}",
-                         "note": "Buffer sẵn sau 5-10s. Nhấn DVR 5' lại để xem.",
-                         "warning": "Nếu buffer fail, FPTPlay/VTVGo CDN có thể block từ server IP."})
-
-@router.get("/api/dvr/buffer/stop/{channel_id}")
-def buffer_stop(channel_id: str):
-    channel_id = channel_id.lower().strip()
-    with _buf_lock:
-        entry = _buffers.pop(channel_id, None)
-    if entry:
-        p = entry.get("popen")
-        if p and p.poll() is None:
-            p.terminate()
-            try:
-                p.wait(timeout=5)
-            except:
-                p.kill()
-        return JSONResponse({"status": "stopped", "channel": channel_id})
-    return JSONResponse({"status": "not_running", "channel": channel_id})
-
-@router.get("/api/dvr/buffer/status/{channel_id}")
-def buffer_status(channel_id: str):
-    channel_id = channel_id.lower().strip()
-    with _buf_lock:
-        entry = _buffers.get(channel_id)
-    if not entry:
-        # Check if playlist exists from previous session
-        pl = os.path.join(BUF_DIR, channel_id, "index.m3u8")
-        if os.path.exists(pl):
-            return JSONResponse({"active": False, "channel": channel_id, "has_playlist": True,
-                                 "playlist": f"/api/dvr/buffer/playlist/{channel_id}"})
-        return JSONResponse({"active": False, "channel": channel_id, "has_playlist": False})
-    p = entry.get("popen")
-    err = entry.get("error")
-    pl = os.path.join(BUF_DIR, channel_id, "index.m3u8")
-    segs = []
-    if os.path.exists(pl):
-        segs = sorted([f for f in os.listdir(os.path.join(BUF_DIR, channel_id)) if f.endswith(".ts")])
-    return JSONResponse({
-        "active": p is not None and p.poll() is None,
-        "channel": channel_id,
-        "error": err,
-        "hint": entry.get("hint"),
-        "segments": len(segs),
-        "has_playlist": os.path.exists(pl),
-        "playlist": f"/api/dvr/buffer/playlist/{channel_id}" if os.path.exists(pl) else None,
-    })
-
-@router.get("/api/dvr/buffer/playlist/{channel_id}")
-def buffer_playlist(channel_id: str):
-    channel_id = channel_id.lower().strip()
-    pl = os.path.join(BUF_DIR, channel_id, "index.m3u8")
-    if not os.path.exists(pl):
-        return JSONResponse({"error": "buffer not ready"}, status_code=404)
-    txt = open(pl, "r", encoding="utf-8", errors="ignore").read()
-    # rewrite segment names -> absolute URLs for HLS.js
-    out = []
-    for line in txt.split("\n"):
-        s = line.strip()
-        if s and not s.startswith("#") and s.endswith(".ts"):
-            # Dùng absolute URL để HLS.js resolve đúng
-            out.append(f"/api/dvr/buffer/seg/{channel_id}/{s}")
-        else:
-            out.append(line)
-    # Thêm cache-bust để tránh browser cache cũ
-    return Response("\n".join(out), media_type="application/vnd.apple.mpegurl",
-                    headers={
-                        "Access-Control-Allow-Origin": "*",
-                        "Cache-Control": "no-cache, no-store, must-revalidate",
-                        "Pragma": "no-cache",
-                        "Expires": "0",
-                    })
-
-@router.get("/api/dvr/buffer/seg/{channel_id}/{seg}")
-def buffer_seg(channel_id: str, seg: str):
-    if not re.match(r"^seg_\d+\.ts$", seg):
-        return Response(status_code=400)
-    path = os.path.join(BUF_DIR, channel_id.lower().strip(), seg)
-    if not os.path.exists(path):
-        return Response(status_code=404)
-    return FileResponse(path, media_type="video/mp2t",
-                        headers={"Access-Control-Allow-Origin": "*"})
-
-# =====================================================================
-# 2) HẸN GIỜ RECORD
-# =====================================================================
-_sched_lock = threading.Lock()
-_active_recs = {}  # job_id -> Popen
-
-def _load_sched():
-    if os.path.exists(SCHED_FILE):
-        try:
-            return json.load(open(SCHED_FILE, encoding="utf-8"))
-        except Exception:
-            return []
-    return []
-
-def _save_sched(items):
-    json.dump(items, open(SCHED_FILE, "w", encoding="utf-8"), ensure_ascii=False, indent=2)
-
-def _do_record(job):
-    """Ghi 1 job: ưu tiên ffmpeg, fallback sang yt-dlp nếu ffmpeg fail."""
-    ch = job["channel"]
-    dur = int(job["duration_sec"])
-    fname = f"{ch}_{job['id']}.mp4"
-    out = os.path.join(REC_DIR, fname)
-    # Ưu tiên buffer timeshift nếu đang chạy => bắt được cả 5' "trước đó LIVE"
-    buf_pl = os.path.join(BUF_DIR, ch, "index.m3u8")
-    if os.path.exists(buf_pl):
-        src = buf_pl
-    else:
-        urls = _resolve_stream(ch)
-        src = urls[0] if urls else None
-    if not src:
-        job["status"] = "error"; job["error"] = "no source"; _save_all(); return
-
-    # Thử ffmpeg trước
-    cmd = [FFMPEG, "-hide_banner", "-loglevel", "warning", "-y"]
-    if os.path.exists(buf_pl):
-        cmd += ["-live_start_index", "0"]
-    cmd += ["-i", src, "-t", str(dur), "-c", "copy", out]
-    print(f"[REC] Recording {ch} for {dur}s from {src[:60]}...", flush=True)
-    job["status"] = "recording"; _save_all()
-
-    try:
-        p = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
-        _active_recs[job["id"]] = p
-        for line in iter(p.stdout.readline, b''):
-            line_s = line.decode("utf-8", errors="ignore").strip()
-            if line_s:
-                print(f"[REC:{ch}] {line_s}", flush=True)
-        p.wait()
-        _active_recs.pop(job["id"], None)
-        if os.path.exists(out) and os.path.getsize(out) > 0:
-            job["status"] = "done"
-            job["file"] = fname
-            print(f"[REC] Done: {out} ({os.path.getsize(out)} bytes)", flush=True)
-        else:
-            raise Exception("output empty")
-    except Exception as e:
-        # Fallback: thử yt-dlp nếu có
-        if YTDP_PATH:
-            print(f"[REC] ffmpeg failed ({e}), trying yt-dlp...", flush=True)
-            out_ytdlp = os.path.join(REC_DIR, f"{ch}_{job['id']}_ytdlp.mp4")
-            try:
-                cmd2 = [YTDP_PATH, "-o", out_ytdlp, "--no-part"]
-                # Nếu là HLS live stream, dùng --download-sections
-                if src.endswith(".m3u8") or ".m3u8" in src:
-                    # yt-dlp cần ffmpeg để merge, nhưng vẫn thử
-                    cmd2 += ["--downloader", "ffmpeg", "--downloader-args", "ffmpeg:-t " + str(dur)]
-                else:
-                    cmd2 += ["--downloader-args", "-t " + str(dur)]
-                cmd2.append(src)
-                p2 = subprocess.Popen(cmd2, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
-                _active_recs[job["id"]] = p2
-                for line in iter(p2.stdout.readline, b''):
-                    line_s = line.decode("utf-8", errors="ignore").strip()
-                    if line_s:
-                        print(f"[REC:{ch} yt-dlp] {line_s}", flush=True)
-                p2.wait()
-                _active_recs.pop(job["id"], None)
-                if os.path.exists(out_ytdlp) and os.path.getsize(out_ytdlp) > 0:
-                    os.rename(out_ytdlp, out)
-                    job["status"] = "done"
-                    job["file"] = fname
-                    print(f"[REC] yt-dlp Done: {out} ({os.path.getsize(out)} bytes)", flush=True)
-                else:
-                    raise Exception("yt-dlp output empty")
-            except Exception as e2:
-                _active_recs.pop(job["id"], None)
-                job["status"] = "error"
-                job["error"] = f"ffmpeg: {e}, yt-dlp: {e2}"
-        else:
-            _active_recs.pop(job["id"], None)
-            job["status"] = "error"
-            job["error"] = str(e)
-
-    # (tuỳ chọn) upload lên HF Dataset để giữ vĩnh viễn
-    if job["status"] == "done" and job.get("upload_repo"):
-        try:
-            from huggingface_hub import HfApi
-            HfApi(token=os.environ.get("HF_TOKEN")).upload_file(
-                path_or_fileobj=out, path_in_repo=f"recordings/{fname}",
-                repo_id=job["upload_repo"], repo_type="dataset")
-            job["uploaded"] = True
-        except Exception as e:
-            job["upload_error"] = str(e)
-    _save_all()
-
-def _schedule_job(job):
-    now = datetime.now(VN_TZ)
-    start = datetime.fromisoformat(job["start_iso"])
-    if start.tzinfo is None:
-        start = start.replace(tzinfo=VN_TZ)
-    delay = (start - now).total_seconds()
-    if delay < 0:
-        delay = 0
-    def _runner():
-        # auto bật buffer trước giờ ghi để có sẵn 5' lùi (nếu bật pre-buffer)
-        if job.get("prebuffer"):
-            try: buffer_start(job["channel"])
-            except Exception: pass
-        _do_record(job)
-    t = threading.Timer(delay, _runner)
-    t.daemon = True
-    t.start()
-    job["status"] = "scheduled"
-
-def _save_all():
-    with _sched_lock:
-        _save_sched(_SCHEDULES)
-
-_SCHEDULES = _load_sched()
-
-@router.post("/api/dvr/record/schedule")
-def record_schedule(payload: dict = Body(...)):
-    """body: {channel, start_iso (giờ VN), end_iso (giờ VN), duration_sec, prebuffer?, upload_repo?}"""
-    ch = str(payload.get("channel", "")).lower().strip()
-    if not ch:
-        return JSONResponse({"error": "missing channel"}, status_code=400)
-    start_iso = payload.get("start_iso")
-    end_iso = payload.get("end_iso")
-    if not start_iso:
-        return JSONResponse({"error": "missing start_iso"}, status_code=400)
-    # Parse và validate timezone
-    try:
-        start_dt = datetime.fromisoformat(start_iso)
-        if start_dt.tzinfo is None:
-            start_dt = start_dt.replace(tzinfo=VN_TZ)
-    except Exception:
-        return JSONResponse({"error": "invalid start_iso"}, status_code=400)
-    # Tính duration từ end_iso nếu có, nếu không dùng duration_sec
-    if end_iso:
-        try:
-            end_dt = datetime.fromisoformat(end_iso)
-            if end_dt.tzinfo is None:
-                end_dt = end_dt.replace(tzinfo=VN_TZ)
-            duration_from_end = int((end_dt - start_dt).total_seconds())
-            duration = max(5, min(3600, duration_from_end))
-        except Exception:
-            duration = int(payload.get("duration_sec", 600))
-    else:
-        duration = int(payload.get("duration_sec", 600))
-    duration = max(5, min(3600, duration))
-    job = {
-        "id": str(int(time.time() * 1000)),
-        "channel": ch,
-        "start_iso": start_dt.isoformat(),
-        "end_iso": end_iso or (start_dt + timedelta(seconds=duration)).isoformat(),
-        "duration_sec": duration,
-        "prebuffer": bool(payload.get("prebuffer", True)),
-        "upload_repo": payload.get("upload_repo"),
-        "status": "pending", "file": None,
-    }
-    with _sched_lock:
-        _SCHEDULES.append(job); _save_sched(_SCHEDULES)
-    _schedule_job(job); _save_all()
-    return JSONResponse({"status": "ok", "job": job})
-
-@router.get("/api/dvr/record/list")
-def record_list():
-    return JSONResponse({"jobs": _SCHEDULES})
-
-@router.get("/api/dvr/record/file/{job_id}")
-def record_file(job_id: str):
-    for j in _SCHEDULES:
-        if j["id"] == job_id and j.get("file"):
-            path = os.path.join(REC_DIR, j["file"])
-            if os.path.exists(path):
-                return FileResponse(path, media_type="video/mp4", filename=j["file"])
-    return JSONResponse({"error": "not found"}, status_code=404)
-
-@router.get("/api/dvr/record/cancel/{job_id}")
-def record_cancel(job_id: str):
-    p = _active_recs.get(job_id)
-    if p and p.poll() is None:
-        p.terminate()
-    for j in _SCHEDULES:
-        if j["id"] == job_id:
-            j["status"] = "cancelled"
-    _save_all()
-    return JSONResponse({"status": "cancelled", "job_id": job_id})
-
-def resume_schedules():
-    """Gọi lúc app khởi động: đặt lại lịch cho job còn ở tương lai."""
-    now = datetime.now(VN_TZ)
-    for j in _SCHEDULES:
-        if j.get("status") in ("pending", "scheduled"):
-            try:
-                start = datetime.fromisoformat(j["start_iso"])
-                if start.tzinfo is None:
-                    start = start.replace(tzinfo=VN_TZ)
-                if start + timedelta(seconds=j["duration_sec"]) > now:
-                    _schedule_job(j)
-                else:
-                    j["status"] = "missed"
-            except Exception:
-                pass
-    _save_all()

gitattributes

@@ -1,35 +0,0 @@
-*.7z filter=lfs diff=lfs merge=lfs -text
-*.arrow filter=lfs diff=lfs merge=lfs -text
-*.bin filter=lfs diff=lfs merge=lfs -text
-*.bz2 filter=lfs diff=lfs merge=lfs -text
-*.ckpt filter=lfs diff=lfs merge=lfs -text
-*.ftz filter=lfs diff=lfs merge=lfs -text
-*.gz filter=lfs diff=lfs merge=lfs -text
-*.h5 filter=lfs diff=lfs merge=lfs -text
-*.joblib filter=lfs diff=lfs merge=lfs -text
-*.lfs.* filter=lfs diff=lfs merge=lfs -text
-*.mlmodel filter=lfs diff=lfs merge=lfs -text
-*.model filter=lfs diff=lfs merge=lfs -text
-*.msgpack filter=lfs diff=lfs merge=lfs -text
-*.npy filter=lfs diff=lfs merge=lfs -text
-*.npz filter=lfs diff=lfs merge=lfs -text
-*.onnx filter=lfs diff=lfs merge=lfs -text
-*.ot filter=lfs diff=lfs merge=lfs -text
-*.parquet filter=lfs diff=lfs merge=lfs -text
-*.pb filter=lfs diff=lfs merge=lfs -text
-*.pickle filter=lfs diff=lfs merge=lfs -text
-*.pkl filter=lfs diff=lfs merge=lfs -text
-*.pt filter=lfs diff=lfs merge=lfs -text
-*.pth filter=lfs diff=lfs merge=lfs -text
-*.rar filter=lfs diff=lfs merge=lfs -text
-*.safetensors filter=lfs diff=lfs merge=lfs -text
-saved_model/**/* filter=lfs diff=lfs merge=lfs -text
-*.tar.* filter=lfs diff=lfs merge=lfs -text
-*.tar filter=lfs diff=lfs merge=lfs -text
-*.tflite filter=lfs diff=lfs merge=lfs -text
-*.tgz filter=lfs diff=lfs merge=lfs -text
-*.wasm filter=lfs diff=lfs merge=lfs -text
-*.xz filter=lfs diff=lfs merge=lfs -text
-*.zip filter=lfs diff=lfs merge=lfs -text
-*.zst filter=lfs diff=lfs merge=lfs -text
-*tfevents* filter=lfs diff=lfs merge=lfs -text

rebuild

@@ -1 +0,0 @@
-rebuilt at 2026-06-18T09:24:34.973630

restart_trigger

@@ -1 +0,0 @@
-# restart trigger