homelab-radio/sdr/radio/recordings.py

"""Recording days on disk, file-name regex, per-day index json, durations, and the item/list/count queries.

Layout: REC_DIR/YYYY-MM-DD/<base>_<YYYYMMDD>_<HHMMSS>_<freq_hz>.mp3, noise in DAY/noise/, classifier index
DAY/.cls.json, archive index DAY/.archived.json (files moved to the home-server HDD by sdr-archive).
"""
import json
import os
import re
import threading
import time
from datetime import datetime
from urllib.parse import quote

from radio.channels import channels
from radio.config import ARCH_INDEX, CLS_INDEX, HOLD_SINCE, REC_DIR, TR_DIR
from radio.mp3 import mp3_duration

DAY_RE = re.compile(r"^\d{4}-\d{2}-\d{2}$")
FILE_RE = re.compile(r"^(?P<base>[A-Za-z0-9-]+)_(?P<d>\d{8})_(?P<t>\d{6})_(?P<f>\d+)\.mp3$")

_dur_cache = {}
_lock = threading.Lock()
_cls = {}          # (day, name) -> (mtime, dict)


def _day_json(day, name):
    p = os.path.join(REC_DIR, day, name)
    try:
        m = os.stat(p).st_mtime
    except OSError:
        return {}
    key = (day, name)
    with _lock:
        c = _cls.get(key)
        if c and c[0] == m:
            return c[1]
    try:
        with open(p, encoding="utf-8") as fh:
            d = json.load(fh)
    except (OSError, ValueError):
        d = {}
    with _lock:
        _cls[key] = (m, d)
    return d


def cls_index(day):
    return _day_json(day, CLS_INDEX)


def arch_index(day):
    """Recordings moved to the home-server HDD by sdr-archive (index written before local delete)."""
    return _day_json(day, ARCH_INDEX)



_tr = {}           # day -> (mtime, dict)


def tr_index(day):
    """Transcripts by sdr-transcribe: TR_DIR/DAY.json {file: {"t": text, "l": lang} | {"err": ...}}."""
    p = os.path.join(TR_DIR, day + ".json")
    try:
        m = os.stat(p).st_mtime
    except OSError:
        return {}
    with _lock:
        c = _tr.get(day)
        if c and c[0] == m:
            return c[1]
    try:
        with open(p, encoding="utf-8") as fh:
            d = json.load(fh)
    except (OSError, ValueError):
        d = {}
    with _lock:
        _tr[day] = (m, d)
    return d


_trs = {"m": None, "v": "9999"}


def tr_since():
    """First day sdr-transcribe works on (status.json "since"); days before it are never queued."""
    p = os.path.join(TR_DIR, "status.json")
    try:
        m = os.stat(p).st_mtime
        if m != _trs["m"]:
            with open(p, encoding="utf-8") as fh:
                _trs.update(m=m, v=json.load(fh).get("since") or "9999")
    except (OSError, ValueError):
        pass
    return _trs["v"]


def duration(path, st):
    key = (path, st.st_mtime, st.st_size)
    with _lock:
        if key in _dur_cache:
            return _dur_cache[key]
    try:
        d = round(mp3_duration(path), 1)
    except OSError:
        d = None
    with _lock:
        if len(_dur_cache) > 20000:
            _dur_cache.clear()
        _dur_cache[key] = d
    return d


def list_days():
    try:
        return sorted((d for d in os.listdir(REC_DIR) if DAY_RE.match(d)), reverse=True)
    except OSError:
        return []


def scan(limit, voice_only=True, group=None, base="", with_peaks=False, groups=None, before=None, chan=None):
    """Newest-first recordings. Returns (items, pending_count).

    groups: only these channel groups; before: an id "DAY/FILE" -> only recordings older than it (paging).

    voice_only: classified voice only (pending + noise hidden); otherwise everything incl. DAY/noise/.
    """
    chans = channels().get("channels", {})
    out, pending = [], 0
    bkey = None
    if before:
        bm = FILE_RE.match(before.split("/")[-1])
        if bm:
            bkey = (bm["d"] + bm["t"], before.split("/")[-1])
    for day in list_days():
        if bkey and day.replace("-", "") > bkey[0][:8]:
            continue   # whole day is newer than the cursor
        ddir = os.path.join(REC_DIR, day)
        idx = cls_index(day)
        entries = []
        try:
            for n in os.listdir(ddir):
                if FILE_RE.match(n):
                    entries.append((n, ""))
        except OSError:
            continue
        arch = arch_index(day)
        if arch:
            local = {n for n, _ in entries}
            entries += [(n, "archive") for n in arch if n not in local and FILE_RE.match(n)]
            idx = dict(idx)
            for n, e in arch.items():
                idx.setdefault(n, e)
        if not voice_only:
            try:
                for n in os.listdir(os.path.join(ddir, "noise")):
                    if FILE_RE.match(n):
                        entries.append((n, "noise/"))
            except OSError:
                pass
        rows = []
        for n, sub in entries:
            m = FILE_RE.match(n)
            info = chans.get(m["f"], {})
            g = info.get("group", "other")
            if group and g != group:
                continue
            if groups and g not in groups:
                continue
            if chan and m["f"] != chan:
                continue
            if bkey and (m["d"] + m["t"], n) >= bkey:
                continue
            ci = idx.get(n)
            cls = ci.get("c") if ci else None
            if sub == "noise/":
                cls = "noise"
            if cls is None:
                pending += 1
                if voice_only:
                    continue
                cls = "pending"
            elif voice_only and cls != "voice":
                continue
            rows.append((m["d"] + m["t"], n, sub, info, g, m, cls, ci))
        rows.sort(key=lambda r: (r[0], r[1]), reverse=True)
        for key, n, sub, info, g, m, cls, ci in rows:
            it = make_item(day, n, sub, cls, ci, base, with_peaks)
            if it is None:
                continue
            out.append(it)
            if len(out) >= limit:
                return out, pending
    return out, pending


def make_item(day, n, sub, cls, ci, base="", with_peaks=True):
    """Recording dict; id = canonical DAY/FILE (stable even after a move into noise/)."""
    m = FILE_RE.match(n)
    archived = sub == "archive"
    if archived:
        sub = ""
        size = (ci or {}).get("size")
    else:
        p = os.path.join(REC_DIR, day, sub + n)
        try:
            size = os.stat(p).st_size
        except OSError:
            return None
    info = channels().get("channels", {}).get(m["f"], {})
    ts = datetime.strptime(m["d"] + m["t"], "%Y%m%d%H%M%S")
    f = (ci or {}).get("f") or {}
    dur = f.get("dur")
    if dur is None and not archived:
        try:
            dur = duration(p, os.stat(p))
        except OSError:
            dur = None
    fhz = int(m["f"])
    it = {
        "id": f"{day}/{n}",
        "file": f"{day}/{sub}{n}",
        "url": f"{base}/audio/{day}/{sub}{quote(n)}",
        "channel": info.get("label", f"{fhz / 1e6:.4f} МГц"),
        "group": info.get("group", "other"),
        "freq_mhz": round(fhz / 1e6, 5),
        "ts": ts.astimezone().isoformat(timespec="seconds"),
        "duration_s": dur,
        "size": size,
        "class": cls,
    }
    if archived:
        it["archived"] = True
    tr = tr_index(day).get(n)
    if tr and tr.get("t"):
        it["tr"] = tr["t"]
        it["tr_l"] = tr.get("l") or ""
        if tr.get("lc"):
            it["tr_lc"] = 1
    elif cls == "voice" and (not tr or ("err" in tr and tr.get("n", 0) < 3)) and day >= tr_since():
        it["tr_q"] = 1          # waiting in the sdr-transcribe queue (desktop off or busy)
    elif tr and tr.get("h"):
        it["tr_none"] = 1       # Whisper heard no words
    if with_peaks:
        it["peaks"] = (ci or {}).get("pk") or []
        it["voiced"] = f.get("vf")
        it["act"] = f.get("act")
        # per-frequency long hold exists only since the patched scanner started (2026-10-20 06:01:58)
        if info.get("hold_ms") and m["d"] + m["t"] >= HOLD_SINCE:
            it["hold_ms"] = info["hold_ms"]
        ac = (ci or {}).get("ac")
        if ac:
            it["ac"] = ac
    return it


def find_item(rid):
    """Look up one recording by DAY/FILE (in DAY/ or DAY/noise/). None if gone."""
    parts = rid.split("/")
    if len(parts) == 3 and parts[1] == "noise":
        parts = [parts[0], parts[2]]
    if len(parts) != 2 or not DAY_RE.match(parts[0]) or not FILE_RE.match(parts[1]):
        return None
    day, n = parts
    ci = cls_index(day).get(n)
    for sub in ("", "noise/"):
        if os.path.exists(os.path.join(REC_DIR, day, sub + n)):
            cls = "noise" if sub else ((ci or {}).get("c") or "pending")
            return make_item(day, n, sub, cls, ci)
    ae = arch_index(day).get(n)
    if ae:
        return make_item(day, n, "archive", ae.get("c") or "voice", ae)
    return None


_counts = {}


def counts(voice_only=True):
    """Recordings per channel group over all days (archived ones too), cached for 20 s."""
    hit = _counts.get(voice_only)
    if hit and time.time() - hit[0] < 20:
        return hit[1]
    chans = channels().get("channels", {})
    out = {"total": 0, "groups": {}, "days": 0, "chans": {}}
    def bump(n, g):
        f = FILE_RE.match(n)["f"]
        e = out["chans"].setdefault(f, {"f": int(f), "n": 0, "group": g,
                                         "label": chans.get(f, {}).get("label") or f"{int(f) / 1e6:.4f} МГц"})
        e["n"] += 1
    for day in list_days():
        out["days"] += 1
        idx = dict(cls_index(day))
        names = set()
        try:
            names.update(n for n in os.listdir(os.path.join(REC_DIR, day)) if FILE_RE.match(n))
        except OSError:
            pass
        for n, e in arch_index(day).items():
            if FILE_RE.match(n):
                names.add(n)
                idx.setdefault(n, e)
        for n in names:
            ci = idx.get(n)
            cls = ci.get("c") if ci else None
            if voice_only and cls != "voice":
                continue
            g = chans.get(FILE_RE.match(n)["f"], {}).get("group", "other")
            out["groups"][g] = out["groups"].get(g, 0) + 1
            out["total"] += 1
            bump(n, g)
        if not voice_only:
            try:
                for n in os.listdir(os.path.join(REC_DIR, day, "noise")):
                    if FILE_RE.match(n):
                        g = chans.get(FILE_RE.match(n)["f"], {}).get("group", "other")
                        out["groups"][g] = out["groups"].get(g, 0) + 1
                        out["total"] += 1
                        bump(n, g)
            except OSError:
                pass
    out["chans"] = sorted(out["chans"].values(), key=lambda e: (-e["n"], e["f"]))   # deterministic on ties
    _counts[voice_only] = (time.time(), out)
    return out