Claude (homelab agent)andClaude Opus 5.5
56eb5af0ff
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
66 lines
2.2 KiB
Python
66 lines
2.2 KiB
Python
"""Share links.
|
|
|
|
/s/<token>: one recording and the conversation it belongs to, open without the site password (Caddy lets
|
|
/s/* through). The token is random and maps to DAY/FILE here only: no path ever appears in the link or page.
|
|
"""
|
|
import os
|
|
import re
|
|
import secrets
|
|
import threading
|
|
import time
|
|
from datetime import datetime
|
|
|
|
from radio.config import CONV_GAP_S, REC_DIR, SHARES_FILE
|
|
from radio.recordings import FILE_RE, arch_index, find_item
|
|
from radio.util import read_json, write_json_atomic
|
|
|
|
SHARE_RE = re.compile(r"^[A-Za-z0-9_-]{16,32}$")
|
|
_share_lock = threading.Lock()
|
|
|
|
|
|
def share_token(rid):
|
|
with _share_lock:
|
|
shares = read_json(SHARES_FILE, {})
|
|
for t, v in shares.items():
|
|
if v.get("id") == rid:
|
|
return t
|
|
t = secrets.token_urlsafe(12)
|
|
shares[t] = {"id": rid, "created": int(time.time())}
|
|
write_json_atomic(SHARES_FILE, shares)
|
|
return t
|
|
|
|
|
|
def share_conv(token):
|
|
"""[items of the conversation, oldest first], index of the shared one; None if unknown / gone."""
|
|
if not SHARE_RE.match(token or ""):
|
|
return None
|
|
sh = read_json(SHARES_FILE, {}).get(token)
|
|
it = find_item(sh["id"]) if sh else None
|
|
if not it:
|
|
return None
|
|
day, name = it["id"].split("/")
|
|
freq = FILE_RE.match(name)["f"]
|
|
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
|
|
names.update(arch_index(day).keys())
|
|
same = []
|
|
for n in names:
|
|
m = FILE_RE.match(n)
|
|
if not m or m["f"] != freq:
|
|
continue
|
|
x = it if n == name else find_item(f"{day}/{n}")
|
|
if x and (x is it or x.get("class") != "noise"):
|
|
same.append(x)
|
|
ts = lambda x: datetime.fromisoformat(x["ts"]).timestamp()
|
|
same.sort(key=ts)
|
|
i = next(k for k, x in enumerate(same) if x["id"] == it["id"])
|
|
lo = hi = i
|
|
while lo > 0 and ts(same[lo]) - (ts(same[lo - 1]) + (same[lo - 1].get("duration_s") or 0)) < CONV_GAP_S:
|
|
lo -= 1
|
|
while hi < len(same) - 1 and ts(same[hi + 1]) - (ts(same[hi]) + (same[hi].get("duration_s") or 0)) < CONV_GAP_S:
|
|
hi += 1
|
|
return same[lo:hi + 1], i - lo
|