Blame for app.py

#!/usr/bin/env python3
"""record — مسجّل بثوث TikTok متعدد + أرشفة HF Bucket + نشر Telegram.

Design (from fleet experience):
- Anonymous source resolution via api-live/user/room (same endpoint the
  fleet uses); optional TIKTOK_COOKIE secret unlocks age-restricted rooms.
- Tag-level FLV spool (original quality, near-zero CPU) in 5-minute
  configuration-pure segments: TikTok switches encoder config mid-broadcast,
  so segments rotate exactly at sequence-header changes and every archived
  part carries its own avcC. Finished segments are remuxed to MP4, archived
  to the bucket *while the
  broadcast continues*, and posted to Telegram as replies to a session
  card (start card -> parts -> closing summary). Recording never waits
  for uploads: separate worker queue.
- Up to MAX_RECORDERS simultaneous broadcasts; extra live users queue.
- Local files and bucket media are auto-cleaned RETENTION_DAYS after a
  session ends (manifests stay as a permanent index). Telegram is the
  permanent public archive.
"""
import asyncio, base64, hashlib, json, os, re, subprocess, threading, time
from concurrent.futures import ThreadPoolExecutor
from datetime import datetime, timezone, timedelta
from pathlib import Path

from fastapi import FastAPI, Request
from fastapi.responses import HTMLResponse, JSONResponse
import requests as rq

try:
    from curl_cffi import requests as cr
except Exception:  # pragma: no cover
    cr = None

# ---------------------------------------------------------------- config
BAGHDAD = timezone(timedelta(hours=3))
WORK = Path(os.getenv('WORK_DIR', '/app/work')); WORK.mkdir(parents=True, exist_ok=True)
BUCKET_ID = os.getenv('BUCKET_ID', 'Azizallawi96/record-storage')
BUCKET_TOKEN = os.getenv('BUCKET_TOKEN', '') or os.getenv('HF_TOKEN', '')
TG_TOKEN = os.getenv('TELEGRAM_BOT_TOKEN', '')
TG_CHAT = os.getenv('TG_CHAT_ID', '-1004351295913')
DASH_PASSWORD = os.getenv('DASH_PASSWORD', 'admin')
MAX_RECORDERS = max(1, int(os.getenv('MAX_RECORDERS', '4')))
POLL_SECONDS = max(10, int(os.getenv('POLL_SECONDS', '25')))
SEGMENT_SECONDS = max(60, int(os.getenv('SEGMENT_SECONDS', '300')))
RETENTION_DAYS = float(os.getenv('RETENTION_DAYS', '2'))
TG_MAX_BYTES = 48 * 1024 * 1024
TG_ALBUM_SIZE = 10      # Telegram media-group maximum: one post, one notification
TG_ALBUM_MAX_WAIT = 1800  # flush a partial album after 30 min so viewers aren't starved
TG_ALBUM_MAX_BYTES = 44 * 1024 * 1024  # cloud Bot API total-payload cap is ~50MB (413 above)
DEFAULT_USERS = [u.strip().lstrip('@').lower() for u in
                 os.getenv('WATCH_USERS', 'talolil,ahmadfamouss,strongarmor,muhajer_iraqi').split(',') if u.strip()]

def now_utc(): return datetime.now(timezone.utc)
def iso(dt=None): return (dt or now_utc()).isoformat().replace('+00:00', 'Z')
def bag(dt=None): return (dt or now_utc()).astimezone(BAGHDAD).strftime('%Y-%m-%d %H:%M')

def log(msg):
    print('[%s] %s' % (now_utc().strftime('%H:%M:%S'), msg), flush=True)

# ---------------------------------------------------------------- bucket
_bucket_api = None
def bucket():
    global _bucket_api
    if _bucket_api is None:
        from huggingface_hub import HfApi
        _bucket_api = HfApi(token=BUCKET_TOKEN)
    return _bucket_api

def bucket_put(local: Path, dest: str):
    bucket().batch_bucket_files(BUCKET_ID, add=[(str(local), dest)])

def bucket_put_bytes(data: bytes, dest: str):
    bucket().batch_bucket_files(BUCKET_ID, add=[(data, dest)])

def bucket_get(dest: str):
    try:
        local = WORK / '_dl' / dest
        local.parent.mkdir(parents=True, exist_ok=True)
        bucket().download_bucket_files(BUCKET_ID, [(dest, str(local))],
                                       raise_on_missing_files=True)
        return local.read_bytes()
    except Exception:
        return None

def bucket_list(prefix=''):
    try:
        out = []
        for t in bucket().list_bucket_tree(BUCKET_ID, path_in_repo=prefix, recursive=True):
            p = getattr(t, 'path', None)
            if p: out.append(p)
        return out
    except Exception:
        return []

def bucket_delete(paths):
    if paths: bucket().batch_bucket_files(BUCKET_ID, delete=list(paths))

# ---------------------------------------------------------------- telegram
TG_API = 'https://api.telegram.org/bot' + TG_TOKEN
def tg(method, data=None, files=None, timeout=120):
    for attempt in range(4):
        try:
            r = rq.post(f'{TG_API}/{method}', data=data or {}, files=files, timeout=timeout)
            j = r.json()
            if j.get('ok'): return j['result']
            if r.status_code == 429:
                wait = int(((j.get('parameters') or {}).get('retry_after') or 5))
                time.sleep(min(wait, 60)); continue
            log(f'TG {method} failed: {str(j)[:180]}')
            return None
        except Exception as e:
            log(f'TG {method} error: {type(e).__name__}')
            time.sleep(3 * (attempt + 1))
    return None

def tg_text(text, reply_to=None):
    data = {'chat_id': TG_CHAT, 'text': text, 'parse_mode': 'HTML',
            'disable_web_page_preview': 'true'}
    if reply_to: data['reply_to_message_id'] = reply_to
    r = tg('sendMessage', data, timeout=60)
    return (r or {}).get('message_id')

_cid = str(TG_CHAT)
TG_LINK_BASE = 'https://t.me/c/' + (_cid[4:] if _cid.startswith('-100') else _cid.lstrip('-'))

def tg_edit(text, message_id):
    r = tg('editMessageText', {'chat_id': TG_CHAT, 'message_id': message_id,
                               'text': text, 'parse_mode': 'HTML',
                               'disable_web_page_preview': 'true'}, timeout=60)
    return bool(r)

def tg_video(path: Path, caption, reply_to, meta):
    data = {'chat_id': TG_CHAT, 'caption': caption[:1024], 'parse_mode': 'HTML',
            'supports_streaming': 'true'}
    if reply_to: data['reply_to_message_id'] = reply_to
    if meta.get('duration'): data['duration'] = int(meta['duration'])
    if meta.get('width'): data['width'] = int(meta['width'])
    if meta.get('height'): data['height'] = int(meta['height'])
    with open(path, 'rb') as fh:
        r = tg('sendVideo', data, files={'video': (path.name, fh, 'video/mp4')}, timeout=600)
    if r:
        v = r.get('video') or {}
        return r.get('message_id'), {k: v.get(k) for k in
               ('file_id', 'file_size', 'duration', 'width', 'height', 'mime_type')}
    with open(path, 'rb') as fh:  # fallback: document keeps the file untouched
        r = tg('sendDocument', {'chat_id': TG_CHAT, 'caption': caption[:1024],
                                'parse_mode': 'HTML',
                                **({'reply_to_message_id': reply_to} if reply_to else {})},
               files={'document': (path.name, fh, 'video/mp4')}, timeout=600)
    return (r or {}).get('message_id'), None

# ---------------------------------------------------------------- resolver
# TikTok session cookies (optional) — same secret names as fleet projects:
#   TIKTOK_COOKIES_<USER> / TIKTOK_SESSIONID_<USER> per account (key = upper,
#   non-alnum -> '_'), then global TIKTOK_COOKIES / TIKTOK_SESSIONID, then
#   legacy TIKTOK_COOKIE. Values may be a Cookie header, a bare session id,
#   Netscape cookies.txt, or a JSON browser export.
SESSION_COOKIE_NAMES = ('sessionid', 'sessionid_ss', 'sid_tt')
_cookie_cache = {}

def _ck_norm(user):
    return re.sub(r'[^A-Z0-9]', '_', str(user or '').strip().upper())

def _json_cookie_pairs(text):
    try: exported = json.loads(text)
    except (ValueError, TypeError): return {}
    if isinstance(exported, dict): exported = exported.get('cookies')
    if not isinstance(exported, list): return {}
    pairs = {}; priority = {}; now = time.time()
    for item in exported:
        if not isinstance(item, dict): continue
        name = item.get('name'); value = item.get('value')
        if not isinstance(name, str) or not isinstance(value, str) or not value: continue
        if not re.fullmatch(r"[!#$%&'*+\-.^_`|~0-9A-Za-z]+", name): continue
        if any(c in value for c in ('\r', '\n', '\x00', ';')): continue
        domain = str(item.get('domain') or '').lower().lstrip('.')
        if domain and domain not in ('tiktok.com', 'www.tiktok.com'): continue
        if str(item.get('path') or '/') != '/': continue
        expiry = item.get('expirationDate', item.get('expires'))
        try: expired = not item.get('session') and expiry is not None and 0 < float(expiry) <= now
        except (TypeError, ValueError): expired = False
        if expired: continue
        rank = (int(domain == 'www.tiktok.com'), int(bool(item.get('hostOnly'))))
        if name not in pairs or rank >= priority[name]:
            pairs[name] = value; priority[name] = rank
    return pairs

def _parse_cookie_token(raw):
    text = str(raw or '').strip()
    if not text: return '', {}
    text = text.replace('\r', '\n')
    json_input = text.startswith(('[', '{'))
    pairs = _json_cookie_pairs(text) if json_input else {}
    sessionid = ''
    for line in (() if json_input else text.split('\n')):
        line = line.strip()
        if not line or line.startswith('#'): continue
        if '\t' in line and line.count('\t') >= 6:
            parts = line.split('\t')
            pairs[parts[5].strip()] = parts[6].strip()
            continue
        chunks = re.split(r'[;,]', line) if ('=' in line) else [line]
        for chunk in chunks:
            chunk = chunk.strip()
            if not chunk: continue
            if '=' in chunk:
                k, _, v = chunk.partition('=')
                pairs[k.strip()] = v.strip()
            elif not sessionid:
                sessionid = chunk
    for name in SESSION_COOKIE_NAMES:
        if pairs.get(name) and not sessionid: sessionid = pairs[name]
    if sessionid:
        for name in SESSION_COOKIE_NAMES:
            pairs.setdefault(name, sessionid)
        idc = pairs.get('tt-target-idc') or pairs.get('tt_target_idc') \
            or os.getenv('TIKTOK_TT_TARGET_IDC') or 'useast1a'
        pairs['tt-target-idc'] = idc.strip()
    return sessionid.strip(), {k: v for k, v in pairs.items() if v}

def cookie_header_for(user):
    """Effective Cookie header for a watched user ('' = anonymous). Cached."""
    if user in _cookie_cache: return _cookie_cache[user]
    key = _ck_norm(user)
    raw = ''
    for name in (f'TIKTOK_SESSIONID_{key}', f'TIKTOK_COOKIES_{key}',
                 'TIKTOK_SESSIONID', 'TIKTOK_COOKIES', 'TIKTOK_COOKIE'):
        raw = os.getenv(name, '')
        if raw: break
    header = ''
    if raw:
        sid, pairs = _parse_cookie_token(raw)
        if pairs: header = '; '.join(f'{k}={v}' for k, v in pairs.items())
        elif sid: header = f'sessionid={sid}'
    _cookie_cache[user] = header
    return header

def auth_users():
    return [u for u in STATE['users'] if cookie_header_for(u)]
_API_ROOM = 'https://www.tiktok.com/api-live/user/room'
_API_WEBCAST = 'https://webcast.tiktok.com/webcast/room/info/'
_last_err_log = {}

def _note_err(user, stage, e):
    now = time.time()
    if now - _last_err_log.get((user, stage), 0) < 300: return
    _last_err_log[(user, stage)] = now
    resp = getattr(e, 'response', None)
    st = getattr(resp, 'status_code', None) or getattr(e, 'status_code', None)
    log(f'resolve @{user} [{stage}] {type(e).__name__} http={st}')

def _note_badjson(user, stage, r):
    now = time.time()
    if now - _last_err_log.get((user, stage), 0) < 300: return
    _last_err_log[(user, stage)] = now
    try: body = (r.text or '')[:80].replace('\n', ' ')
    except Exception: body = '?'
    log(f'resolve @{user} [{stage}] non-JSON http={getattr(r, "status_code", "?")} '
        f'ctype={(r.headers or {}).get("content-type")} body={body!r}')

def _get_room(user, stage, headers, params):
    r = cr.get(_API_ROOM, params=params, impersonate='chrome', headers=headers, timeout=20)
    r.raise_for_status()
    try:
        return r.json()
    except Exception:
        _note_badjson(user, stage, r)
        raise

def _stream_map(block):
    """{quality: flv} from a streamData/hevcStreamData block."""
    if not isinstance(block, dict): return {}
    p = (block.get('pull_data') or {}).get('stream_data')
    try: s = json.loads(p) if isinstance(p, str) else (p or {})
    except ValueError: s = {}
    out = {}
    for q, item in (s.get('data') or {}).items():
        f = (item.get('main') or {}).get('flv')
        if f: out[q] = f
    return out

def _pick(avc, hevc):
    for q in ('origin', 'uhd', 'hd', 'sd', 'ld'):
        if q in avc: return avc[q], 'avc'
    for q in ('origin', 'uhd', 'hd', 'sd', 'ld'):
        if q in hevc: return hevc[q], 'hevc'
    return None, 'avc'

def _interpret(j, user):
    """-> ('offline', None) | ('live', info) | ('live_no_url', partial)"""
    d = (j or {}).get('data') or {}
    live = d.get('liveRoom') or {}
    person = d.get('user') or {}
    if live.get('status') not in (2, '2'):
        return ('offline', None)
    room = str(person.get('roomId') or live.get('id') or '')
    partial = {'title': live.get('title') or '',
               'nickname': person.get('nickname') or user, 'room_id': room}
    flv, codec = _pick(_stream_map(live.get('streamData')),
                       _stream_map(live.get('hevcStreamData')))
    if not flv: return ('live_no_url', partial)
    return ('live', dict(partial, flv=flv, codec=codec))

def _webcast_sources(room, headers):
    r = cr.get(_API_WEBCAST, params={'aid': 1988, 'room_id': room},
               impersonate='chrome', headers=headers, timeout=20)
    r.raise_for_status()
    d = (r.json() or {}).get('data') or {}
    if d.get('status') not in (2, '2'): return None
    su = d.get('stream_url') or {}
    flv, codec = None, 'avc'
    lcd = su.get('live_core_sdk_data') or {}
    pd = (lcd.get('pull_data') or {}).get('stream_data')
    try: s = json.loads(pd) if isinstance(pd, str) else (pd or {})
    except ValueError: s = {}
    m = {}
    for q, item in (s.get('data') or {}).items():
        f = (item.get('main') or {}).get('flv')
        if f: m[q] = f
    flv, codec = _pick(m, {})
    if not flv:
        fp = su.get('flv_pull_url') or {}
        if isinstance(fp, dict):
            for q in ('FULL_HD1', 'HD1', 'SD2', 'SD1', 'LD1'):
                if fp.get(q): flv = fp[q]; break
    if not flv and su.get('rtmp_pull_url'): flv = su['rtmp_pull_url']
    if not flv: return None
    return {'title': d.get('title') or '',
            'nickname': ((d.get('owner') or {}).get('nickname') or ''),
            'room_id': str(room), 'flv': flv, 'codec': codec}

def resolve(user):
    """Live info dict | {'live_no_url': ...} | None (offline). Fleet recipe:
    anonymous first; the account session only when sources are missing."""
    if cr is None: return None
    params = {'aid': 1988, 'uniqueId': user, 'sourceType': 54, 'staleTime': 0}
    pub = {'Referer': 'https://www.tiktok.com/'}
    j = None; st = ('offline', None)
    try:
        j = _get_room(user, 'anon', pub, params)
        st = _interpret(j, user)
    except Exception as e:
        _note_err(user, 'anon', e)
    if st[0] == 'live': return st[1]
    if st[0] == 'offline' and j is not None: return None
    partial = st[1] if st[0] == 'live_no_url' else None
    ck = cookie_header_for(user)
    auth = {'Referer': 'https://www.tiktok.com/', 'Cookie': ck} if ck else pub
    if ck:
        try:
            j2 = _get_room(user, 'auth', auth, params)
            st2 = _interpret(j2, user)
            if st2[0] == 'live': return st2[1]
            if st2[0] == 'live_no_url': partial = st2[1]
            elif partial is None and j is None: return None
        except Exception as e:
            _note_err(user, 'auth', e)
    room = (partial or {}).get('room_id')
    if room:
        try:
            info = _webcast_sources(room, auth)
            if info:
                if not info['nickname']: info['nickname'] = (partial or {}).get('nickname') or user
                if not info['title']: info['title'] = (partial or {}).get('title') or ''
                return info
        except Exception as e:
            _note_err(user, 'webcast', e)
    if partial is not None or st[0] == 'live_no_url':
        return {'live_no_url': True, 'title': (partial or {}).get('title', '')}
    return None

# ---------------------------------------------------------------- recorder
class Session:
    def __init__(self, user, info):
        self.user = user; self.info = info
        self.started = now_utc()
        self.id = '%s_%s' % (user, self.started.strftime('%Y%m%d-%H%M%S'))
        self.dir = WORK / 'recordings' / user / self.id
        self.dir.mkdir(parents=True, exist_ok=True)
        self.date = self.started.strftime('%Y-%m-%d')
        self.bucket_prefix = f'{user}/{self.date}/{self.id}'
        self.card_id = None; self.parts = []  # dicts: idx, mp4, secs, bytes, tg_id, bucket
        self.tg_pending = []; self.tg_pending_since = 0.0; self.albums = []
        self.tg_lock = threading.RLock()
        self.processed = set(); self.proc_lock = threading.Lock()
        self.published = []  # full-video messages: {'file','bytes','telegram','tg_file'}
        self.ended = None; self.restarts = 0; self.stop_flag = False
        self.use_ffmpeg = False
        self.proc = None; self.next_idx = 0; self.sent_count = 0

STATE = {'users': list(DEFAULT_USERS), 'sessions': {}, 'live': {}, 'queued': []}
LOCK = threading.RLock()
WORKQ = []  # (session, part_index) pending processing
QLOCK = threading.Lock()

import shutil
FFMPEG = shutil.which('ffmpeg') or ''
if not FFMPEG:
    try:
        import imageio_ffmpeg
        FFMPEG = imageio_ffmpeg.get_ffmpeg_exe()
    except Exception:
        FFMPEG = 'ffmpeg'

def ffprobe(path):
    try:
        r = subprocess.run([FFMPEG, '-hide_banner', '-i', str(path)],
                           capture_output=True, text=True, timeout=60)
        err = r.stderr or ''
        dur = 0.0; br = 0; w = h = None
        m = re.search(r'Duration: (\d+):(\d+):(\d+(?:\.\d+)?)(?:, start[^,]*)?, bitrate: (\d+)', err)
        if m:
            dur = int(m.group(1)) * 3600 + int(m.group(2)) * 60 + float(m.group(3))
            br = int(m.group(4)) * 1000
        m = re.search(r'Video:[^\n]*?, (\d{2,5})x(\d{2,5})[, ]', err)
        if m:
            w, h = int(m.group(1)), int(m.group(2))
        mc = re.search(r'Video: ([A-Za-z0-9_]+)', err)
        return {'duration': dur, 'width': w, 'height': h, 'bitrate': br,
                'codec': mc.group(1) if mc else None}
    except Exception:
        return {'duration': 0}

def split_if_big(mp4: Path, meta, max_bytes=None):
    """Return [paths] each <= max_bytes (stream-copy time chunks)."""
    cap = max_bytes or TG_MAX_BYTES
    size = mp4.stat().st_size
    if size <= cap: return [mp4]
    dur = meta.get('duration') or 0; br = meta.get('bitrate') or 0
    if not dur or not br: return [mp4]
    chunk = max(20, int(cap * 0.92 * 8 / br))
    out_pat = str(mp4.with_suffix('')) + '_c%02d.mp4'
    subprocess.run([FFMPEG, '-hide_banner', '-loglevel', 'error', '-i', str(mp4),
                    '-c', 'copy', '-f', 'segment', '-segment_time', str(chunk),
                    '-reset_timestamps', '1', out_pat], timeout=900)
    parts = sorted(mp4.parent.glob(mp4.stem + '_c*.mp4'))
    return parts or [mp4]

def caption_for(sess, part_no, meta, first):
    u = sess.user; info = sess.info
    tag = '#' + re.sub(r'[^A-Za-z0-9_]', '_', u)
    if first:
        return ('🎥 <b>%s</b> (@%s)\n🗓 %s (بغداد)\n📼 الجزء %s • %s\n%s'
                % (info.get('nickname') or u, u, bag(sess.started), part_no,
                   (info.get('title') or '')[:120], tag))
    return '📼 @%s • %s • الجزء %s\n%s' % (u, sess.date, part_no, tag)

def _album_caption(sess, batch):
    u = sess.user; info = sess.info
    tag = '#' + re.sub(r'[^A-Za-z0-9_]', '_', u)
    a, b = batch[0]['part_no'], batch[-1]['part_no']
    rng = ('الأجزاء %s–%s' % (a, b)) if str(a) != str(b) else ('الجزء %s' % a)
    title = ''
    if str(a) in ('1', '1.1'):
        title = '\n📝 ' + (info.get('title') or '')[:120]
    return ('🎥 <b>%s</b> (@%s)\n🗓 %s (بغداد)\n📼 %s من البث%s\n%s'
            % (info.get('nickname') or u, u, bag(sess.started), rng, title, tag))

def _part_rec(sess, idx):
    for p in sess.parts:
        if p.get('idx') == idx: return p
    return None

def _mark_sent(sess, item, message):
    rec = _part_rec(sess, item['idx'])
    if rec is None or not message: return
    mid = message.get('message_id')
    if mid and mid not in (rec.get('telegram') or []):
        rec.setdefault('telegram', []).append(mid)
    v = message.get('video') or {}
    if v and not rec.get('tg_file'):
        rec['tg_file'] = {k: v.get(k) for k in
                          ('file_id', 'file_size', 'duration', 'width', 'height', 'mime_type')}

def _send_album(sess, batch):
    media = []; files = {}; handles = []
    for i, it in enumerate(batch):
        key = 'm%d' % i
        m = it['meta'] or {}
        e = {'type': 'video', 'media': 'attach://' + key, 'supports_streaming': True}
        if m.get('duration'): e['duration'] = int(m['duration'])
        if m.get('width'): e['width'] = int(m['width'])
        if m.get('height'): e['height'] = int(m['height'])
        if i == 0:
            e['caption'] = _album_caption(sess, batch); e['parse_mode'] = 'HTML'
        media.append(e)
    data = {'chat_id': TG_CHAT, 'media': json.dumps(media)}
    if sess.card_id: data['reply_to_message_id'] = sess.card_id
    try:
        for i, it in enumerate(batch):
            fh = open(it['path'], 'rb'); handles.append(fh)
            files['m%d' % i] = (Path(it['path']).name, fh, 'video/mp4')
        res = tg('sendMediaGroup', data, files=files, timeout=1200)
    finally:
        for fh in handles:
            try: fh.close()
            except Exception: pass
    if not res or len(res) != len(batch): return False
    for it, msg in zip(batch, res): _mark_sent(sess, it, msg)
    sess.albums.append({'first': batch[0]['part_no'], 'last': batch[-1]['part_no'],
                        'msg': (res[0] or {}).get('message_id'), 'count': len(batch)})
    log(f'album sent @{sess.user} parts {batch[0]["part_no"]}..{batch[-1]["part_no"]} n={len(batch)}')
    return True

def _send_single(sess, item):
    cap = caption_for(sess, item['part_no'], item['meta'] or {}, first=(item['idx'] == 0))
    mid, vmeta = tg_video(item['path'], cap, sess.card_id, item['meta'] or {})
    if not mid: return False
    _mark_sent(sess, item, {'message_id': mid, 'video': None})
    rec = _part_rec(sess, item['idx'])
    if rec is not None and vmeta and not rec.get('tg_file'): rec['tg_file'] = vmeta
    sess.albums.append({'first': item['part_no'], 'last': item['part_no'],
                        'msg': mid, 'count': 1})
    return True

def flush_items(sess, items, force=False):
    """Publish the given pending items as albums: up to TG_ALBUM_SIZE videos
    and TG_ALBUM_MAX_BYTES in ONE channel post (one notification)."""
    items = list(items)
    with sess.tg_lock:
        while items:
            batch = []; total = 0
            for it in items:
                if len(batch) >= TG_ALBUM_SIZE: break
                try: b = Path(it['path']).stat().st_size
                except Exception: b = 0
                if batch and total + b > TG_ALBUM_MAX_BYTES: break
                batch.append(it); total += b
            ready = (force or len(batch) >= TG_ALBUM_SIZE
                     or (len(batch) >= 2 and len(batch) < len(items))
                     or (len(batch) >= 2 and total >= TG_ALBUM_MAX_BYTES))
            if not ready or not batch: break
            published = 0
            if len(batch) >= 2 and all(it['is_mp4'] for it in batch) \
                    and _send_album(sess, batch):
                published = len(batch)
            else:
                if len(batch) >= 2 and all(it['is_mp4'] for it in batch):
                    log(f'album send failed @{sess.user} n={len(batch)} -> singles')
                for it in batch:
                    if _send_single(sess, it): published += 1
                    else: break
            if published:
                done_ids = {id(x) for x in batch[:published]}
                sess.tg_pending = [x for x in sess.tg_pending if id(x) not in done_ids]
                sess.tg_pending_since = time.time() if sess.tg_pending else 0.0
                items = items[published:]
                save_state()
            if published < len(batch):
                log(f'album flush paused @{sess.user} ok={published}/{len(batch)} pending={len(sess.tg_pending)}')
                break

def flush_albums(sess, force=False):
    flush_items(sess, list(sess.tg_pending), force=force)

# ------------------------------------------------- full-video publishing
TG_BIG_MAX_BYTES = 1900 * 1024 * 1024  # MTProto bot ceiling is 2000 MB; keep margin
_MT_LOCK = threading.Lock()
_mt_warned = False

def mt_send_video(path, caption, reply_to, meta):
    """Upload a >50MB video via MTProto (Telethon, bot-token login): the bot
    stays a bot (no user account), and the HTTP Bot API keeps working.
    Returns (message_id, file_info) or (None, None)."""
    global _mt_warned
    api_id = os.getenv('TELEGRAM_API_ID', '').strip()
    api_hash = os.getenv('TELEGRAM_API_HASH', '').strip()
    if not api_id or not api_hash or not TG_TOKEN:
        if not _mt_warned:
            _mt_warned = True
            log('MTProto upload unavailable (TELEGRAM_API_ID/HASH unset)')
        return None, None
    import asyncio

    async def _go():
        from telethon import TelegramClient
        from telethon.tl.types import DocumentAttributeVideo
        client = TelegramClient(str(WORK / 'tg_mtproto'), int(api_id), api_hash)
        await client.start(bot_token=TG_TOKEN)
        try:
            attrs = [DocumentAttributeVideo(
                duration=int(meta.get('duration') or 0),
                w=int(meta.get('width') or 0), h=int(meta.get('height') or 0),
                supports_streaming=True)]
            return await client.send_file(int(TG_CHAT), str(path), caption=caption[:1024],
                                          parse_mode='html', reply_to=reply_to,
                                          attributes=attrs,
                                          progress_callback=lambda done, total: None)
        finally:
            await client.disconnect()

    try:
        with _MT_LOCK:
            msg = asyncio.run(_go())
    except Exception as e:
        log(f'MTProto upload failed {Path(path).name}: {type(e).__name__} {str(e)[:140]}')
        return None, None
    if not msg: return None, None
    info = {}
    try:
        f = msg.file
        info = {'size': getattr(f, 'size', None), 'mime_type': getattr(f, 'mime_type', None),
                'duration': getattr(f, 'duration', None)}
    except Exception:
        pass
    return msg.id, info

def _publish_one(sess, path, caption, meta):
    if path.stat().st_size <= TG_MAX_BYTES:
        return tg_video(path, caption, sess.card_id, meta)
    return mt_send_video(path, caption, sess.card_id, meta)

def _full_caption(sess, idx, total, meta):
    u = sess.user; info = sess.info
    tag = '#' + re.sub(r'[^A-Za-z0-9_]', '_', u)
    names = '🎥 <b>%s</b> (@%s)\n🗓 %s (بغداد)\n📝 %s' % (
        info.get('nickname') or u, u, bag(sess.started), (info.get('title') or '')[:120])
    if total == 1:
        dur = int(meta.get('duration') or 0)
        return '%s\n⌛️ البث كاملًا في فيديو واحد — %d:%02d:%02d\n%s' % (
            names, dur // 3600, (dur % 3600) // 60, dur % 60, tag)
    if idx == 0:
        return '%s\n🎬 البث كاملًا في %d مقاطع — هذا المقطع 1\n%s' % (names, total, tag)
    return '🎥 @%s • %s • مقطع %d من %d\n%s' % (u, sess.date, idx + 1, total, tag)

def _merge_run(sess, items, ri):
    if len(items) == 1:
        return sess.dir / items[0]['file']
    merged = sess.dir / ('full-%d.mp4' % ri)
    lst = sess.dir / ('concat-%d.txt' % ri)
    lst.write_text(''.join("file '%s'\n" % p['file'] for p in items), encoding='utf-8')
    r = subprocess.run([FFMPEG, '-hide_banner', '-loglevel', 'error', '-f', 'concat',
                        '-safe', '0', '-i', str(lst), '-c', 'copy',
                        '-movflags', '+faststart', str(merged)],
                       capture_output=True, text=True, timeout=1800, cwd=str(sess.dir))
    if r.returncode != 0 or not merged.exists() or merged.stat().st_size == 0:
        log(f'merge run{ri} failed @{sess.user}: {(r.stderr or "")[-200:]}')
        return None
    try: bucket_put(merged, f'{sess.bucket_prefix}/{merged.name}')
    except Exception as e: log(f'merged bucket put failed @{sess.user}: {type(e).__name__}')
    return merged

def _pick_canvas(recs):
    """Single output canvas for a normalized merge: the largest real picture
    seen in the broadcast (capped at 1080 short side), even dimensions."""
    best = None
    for p in recs:
        try: w, h = int(p.get('width') or 0), int(p.get('height') or 0)
        except (TypeError, ValueError): continue
        if w >= 16 and h >= 16 and (best is None or w * h > best[0] * best[1]):
            best = (w, h)
    if not best: return (720, 1280)
    w, h = best
    short = min(w, h)
    if short > 1080:
        w, h = int(w * 1080 / short), int(h * 1080 / short)
    return (max(2, w - w % 2), max(2, h - h % 2))

def _normalize_file(sess, src, idx, canvas, seconds):
    """Decode ONE uniform era onto the broadcast canvas/codec. Each era keeps
    its own decoder, so an HEVC era after AVC eras (or vice versa) is decoded
    by the right codec — the earlier single concat-demuxer re-encode dropped
    an HEVC era's video entirely (measured: audio 18.06s, video 361 frames
    of 540) because one demuxer/decoder chain cannot change codec mid-file."""
    W, H = canvas
    out = sess.dir / ('norm-%03d.mp4' % idx)
    vf = ('scale=%d:%d:force_original_aspect_ratio=decrease:flags=bicubic,'
          'pad=%d:%d:(ow-iw)/2:(oh-ih)/2:black,fps=30,setsar=1,format=yuv420p'
          % (W, H, W, H))
    timeout = int(min(21600, max(1800, (seconds or 0) * 3)))
    r = subprocess.run([FFMPEG, '-hide_banner', '-loglevel', 'error',
                        '-i', str(src), '-map', '0:v:0', '-map', '0:a:0?',
                        '-vf', vf, '-af', 'aresample=48000:async=1:first_pts=0',
                        '-c:v', 'libx264', '-preset', 'veryfast', '-crf', '17',
                        '-profile:v', 'high', '-c:a', 'aac', '-b:a', '160k',
                        '-ar', '48000', '-ac', '2',
                        '-avoid_negative_ts', 'make_zero',
                        '-video_track_timescale', '30000',
                        '-movflags', '+faststart', str(out)],
                       capture_output=True, text=True, timeout=timeout, cwd=str(sess.dir))
    if r.returncode != 0 or not out.exists() or out.stat().st_size == 0:
        log(f'normalize era{idx} failed @{sess.user}: {(r.stderr or "")[-200:]}')
        return None
    return out

def _concat_copy(sess, files, out_name):
    """Join files that are now byte-compatible (same codec/canvas/fps/audio)
    with stream copy. Safe precisely because _normalize_file made them so;
    never used on raw mixed-configuration parts."""
    out = sess.dir / out_name
    lst = sess.dir / ('concat-' + out.stem + '.txt')
    lst.write_text(''.join("file '%s'\n" % p.name for p in files), encoding='utf-8')
    r = subprocess.run([FFMPEG, '-hide_banner', '-loglevel', 'error',
                        '-f', 'concat', '-safe', '0', '-i', str(lst),
                        '-c', 'copy', '-movflags', '+faststart', str(out)],
                       capture_output=True, text=True, timeout=3600, cwd=str(sess.dir))
    if r.returncode != 0 or not out.exists() or out.stat().st_size == 0:
        log(f'normalized concat failed @{sess.user}: {(r.stderr or "")[-200:]}')
        return None
    return out

def _merge_normalized(sess, runs):
    """Merge ALL eras into ONE video: per-era decode onto a single canvas,
    then a stream-copy join of the now-uniform intermediates.

    Broadcasters who flip codec/resolution mid-broadcast (uplink wobble, PK
    layout hops) used to publish one video per configuration era — measured
    147 messages for a single muhajer_iraqi broadcast, exactly what the
    operator forbade ('no fragmented messages'). Stream-copy cannot join
    different avcC/configurations without corrupting every frame after a
    switch (proven 2026-10-20), and a single concat-demuxer re-encode drops
    an era whose codec differs from the first file's (proven locally:
    HEVC-era video vanished). So each era is decoded separately onto the
    broadcast's largest canvas (H.264/AAC, CFR 30), and only then joined.
    Uniform broadcasts never take this path (publish_full keeps their
    original-quality stream-copy merge). A/V sync: every archived part is
    internally in sync; per-era decode preserves its timing and
    aresample=async=1 absorbs boundary jitter instead of drifting."""
    recs = [p for _key, items in runs for p in items]
    canvas = _pick_canvas(recs)
    norms = []
    for ri, (_key, items) in enumerate(runs):
        merged = _merge_run(sess, items, ri)
        if not merged: return None
        norm = _normalize_file(sess, merged, ri, canvas,
                               sum((p.get('seconds') or 0) for p in items))
        if not norm: return None
        norms.append(norm)
    single = norms[0] if len(norms) == 1 else _concat_copy(sess, norms, 'full-single.mp4')
    if not single: return None
    try: bucket_put(single, f'{sess.bucket_prefix}/{single.name}')
    except Exception as e: log(f'merged bucket put failed @{sess.user}: {type(e).__name__}')
    log(f'normalized merge @{sess.user}: {len(recs)} parts / {len(runs)} eras '
        f'-> one {canvas[0]}x{canvas[1]} video')
    return single

def _decode_clean(path, seconds):
    """Publish gate for a merged file: decode EVERY video frame and audio
    sample (framemd5, -xerror) and demand zero error output plus non-empty
    frame/sample counts on both streams. This is the direct guard against
    the old failure class — a merged video that looks fine by duration but
    decodes into garbage. (Decoding to the null muxer was rejected as the
    gate: it emits benign non-monotonic-DTS muxer warnings on healthy
    concatenated MP4s while the decoder itself is silent.) Expensive on
    long broadcasts, deliberately: publishing a corrupt single video is
    worse than publishing late or falling back."""
    try:
        timeout = int(min(21600, max(600, (seconds or 0) * 2)))
        for stream in ('0:v', '0:a'):
            r = subprocess.run([FFMPEG, '-hide_banner', '-loglevel', 'error',
                                '-xerror', '-i', str(path), '-map', stream,
                                '-f', 'framemd5', '-'],
                               capture_output=True, text=True, timeout=timeout)
            frames = [l for l in (r.stdout or '').splitlines()
                      if l and not l.startswith('#')]
            if r.returncode != 0 or (r.stderr or '').strip() or not frames:
                log(f'decode check {Path(path).name} stream {stream}: '
                    f'rc={r.returncode} frames={len(frames)} '
                    f'err={(r.stderr or "")[-160:]}')
                return False
        return True
    except Exception as e:
        log(f'decode check failed {Path(path).name}: {type(e).__name__}')
        return False

def publish_full(sess):
    """Publish the ended session as ONE long video per uniform run (split
    only above ~1.9 GB) instead of one message per 5-minute part."""
    recs = [p for p in sorted(sess.parts, key=lambda x: x.get('idx') or 0)
            if str(p.get('file') or '').endswith('.mp4') and (sess.dir / p['file']).exists()]
    if not recs: return
    runs = []
    for p in recs:
        key = (p.get('codec') or '?', p.get('width') or 0, p.get('height') or 0,
               p.get('cfg') or '')
        if runs and runs[-1][0] == key: runs[-1][1].append(p)
        else: runs.append((key, [p]))
    plan = []
    if len(runs) > 1:
        # Mixed configurations: ONE normalized video for the whole broadcast.
        # Per-era videos are the 147-messages failure the operator rejected;
        # they survive below only as a never-lose-the-broadcast fallback.
        single = _merge_normalized(sess, runs)
        if single:
            meta = ffprobe(single)
            expect = sum((p.get('seconds') or 0) for p in recs)
            got = meta.get('duration') or 0
            if expect and got and abs(got - expect) > max(20, 0.1 * expect):
                log(f'normalized duration mismatch @{sess.user}: got {got:.0f}s expect {expect:.0f}s')
                single = None
            if single and not _decode_clean(single, expect):
                log(f'normalized decode check failed @{sess.user}; falling back to era videos')
                single = None
        if single:
            vids = [single]
            if single.stat().st_size > TG_BIG_MAX_BYTES:
                vids = split_if_big(single, meta, TG_BIG_MAX_BYTES)
            for v in vids:
                plan.append((v, ffprobe(v) if len(vids) > 1 else meta, recs))
    if not plan:
        for ri, (key, items) in enumerate(runs):
            merged = _merge_run(sess, items, ri)
            if not merged: continue
            meta = ffprobe(merged)
            expect = sum((p.get('seconds') or 0) for p in items)
            got = meta.get('duration') or 0
            if expect and got and abs(got - expect) > max(20, 0.1 * expect):
                log(f'merged duration mismatch @{sess.user} run{ri}: got {got:.0f}s expect {expect:.0f}s')
                continue
            vids = [merged]
            if merged.stat().st_size > TG_BIG_MAX_BYTES:
                vids = split_if_big(merged, meta, TG_BIG_MAX_BYTES)
            for v in vids:
                plan.append((v, ffprobe(v) if len(vids) > 1 else meta, items))
    if not plan: return
    log(f'publish @{sess.user}: {len(recs)} parts -> {len(plan)} video message(s)')
    for i, (v, vmeta, items) in enumerate(plan):
        mid, tgm = _publish_one(sess, v, _full_caption(sess, i, len(plan), vmeta), vmeta)
        if mid:
            sess.published.append({'file': v.name, 'bytes': v.stat().st_size,
                                   'telegram': mid, 'tg_file': tgm})
        else:
            log(f'publish failed @{sess.user} {v.name} -> album fallback for its parts')
            idxs = {p['idx'] for p in items}
            flush_items(sess, [it for it in sess.tg_pending if it['idx'] in idxs], force=True)

def process_part(sess, idx):
    raw = sess.dir / ('part-%03d.flv' % idx)
    with sess.proc_lock:
        if idx in sess.processed: return
        if not raw.exists(): return
        sess.processed.add(idx)
    mp4 = sess.dir / ('part-%03d.mp4' % idx)
    if not mp4.exists():
        r = subprocess.run([FFMPEG, '-hide_banner', '-loglevel', 'error', '-i', str(raw),
                            '-c', 'copy', '-movflags', '+faststart', str(mp4)], timeout=1800)
        if r.returncode != 0 or not mp4.exists(): mp4 = raw
    meta = ffprobe(mp4)
    cfg = mp4_cfg_hash(mp4) if mp4.suffix == '.mp4' else ''
    dest = f'{sess.bucket_prefix}/part-{idx:03d}.mp4'
    ok_bucket = False
    try:
        bucket_put(mp4, dest); ok_bucket = True
    except Exception as e:
        log(f'bucket put failed {dest}: {type(e).__name__}')
    part_no = idx + 1
    chunks = split_if_big(mp4, meta) if mp4.suffix == '.mp4' else [mp4]
    items = [{'idx': idx,
              'part_no': (part_no if len(chunks) == 1 else f'{part_no}.{ci+1}'),
              'path': ch, 'meta': meta, 'is_mp4': ch.suffix == '.mp4'}
             for ci, ch in enumerate(chunks)]
    sess.parts.append({'idx': idx, 'file': mp4.name, 'seconds': meta.get('duration'),
                       'bytes': mp4.stat().st_size, 'bucket': dest if ok_bucket else '',
                       'codec': meta.get('codec'), 'width': meta.get('width'),
                       'height': meta.get('height'), 'cfg': cfg,
                       'telegram': [], 'tg_file': None})
    with sess.tg_lock:
        sess.tg_pending.extend(items)
        if not sess.tg_pending_since: sess.tg_pending_since = time.time()
    sess.sent_count += 1
    with LOCK:
        if sess.id in STATE['sessions']:
            STATE['sessions'][sess.id]['parts'] = sess.sent_count
    save_state()

def worker():
    while True:
        job = None
        with QLOCK:
            if WORKQ: job = WORKQ.pop(0)
        if not job:
            time.sleep(1); continue
        sess, idx = job
        try: process_part(sess, idx)
        except Exception as e:
            with sess.proc_lock: sess.processed.discard(idx)
            log(f'process part {sess.id}/{idx}: {type(e).__name__} {e}')

def start_card(sess):
    info = sess.info
    text = ('🔴 <b>بدأ بث مباشر — جارٍ التسجيل</b>\n\n'
            '🎥 <b>%s</b> (@%s)\n📝 %s\n🗓 %s (بتوقيت بغداد)\n'
            '⏺ التسجيل من البداية وبأعلى جودة أصلية؛ وعند انتهاء البث يُنشر هنا كاملًا في فيديو واحد طويل — بلا رسائل مجزأة.'
            % (info.get('nickname') or sess.user, sess.user,
               (info.get('title') or 'بدون عنوان')[:160], bag(sess.started)))
    sess.card_id = tg_text(text)

def end_card(sess, total_secs, total_bytes):
    if sess.published:
        finish_card(sess, total_secs, total_bytes)
        return
    mins = int(total_secs // 60)
    text = ('✅ <b>انتهى البث واكتمل الأرشيف</b>\n\n'
            '🎥 <b>%s</b> (@%s)\n🗓 %s ← %s (بغداد)\n'
            '⌛️ المدة: %d:%02d ساعة • 📼 الأجزاء: %d • 💾 الحجم: %.1f MB\n'
            '🗄 نسخة الأرشيف الدائم: هذا المنشور وأجزاؤه أسفله.'
            % (sess.info.get('nickname') or sess.user, sess.user,
               bag(sess.started), bag(sess.ended), mins // 60, mins % 60,
               sess.sent_count, total_bytes / 1048576))
    tg_text(text, reply_to=sess.card_id)
    finish_card(sess, total_secs, total_bytes)

def finish_card(sess, total_secs, total_bytes):
    if not sess.card_id: return
    mins = int(total_secs // 60)
    links = []
    for i, pv in enumerate(sess.published or []):
        if not pv.get('telegram'): continue
        label = '🎬 مشاهدة البث كاملًا' if len(sess.published) == 1 else ('🎬 مقطع %d من %d' % (i + 1, len(sess.published)))
        links.append('<a href="%s/%s">%s</a>' % (TG_LINK_BASE, pv['telegram'], label))
    for a in (sess.albums or []):
        if not a.get('msg'): continue
        label = ('📼 أجزاء %s–%s' % (a['first'], a['last'])) if str(a['first']) != str(a['last']) \
            else ('📼 الجزء %s' % a['first'])
        links.append('<a href="%s/%s">%s</a>' % (TG_LINK_BASE, a['msg'], label))
    if not links:
        for p in sorted(sess.parts, key=lambda x: x.get('idx') or 0):
            ids = p.get('telegram') or []
            if ids:
                links.append('<a href="%s/%s">الجزء %d</a>'
                             % (TG_LINK_BASE, ids[0], (p.get('idx') or 0) + 1))
    if sess.published: links = links[:len(sess.published)]
    shown = links[:40]
    index = ' • '.join(shown) if shown else '—'
    if len(links) > len(shown):
        index += '\n… و%d جزءًا إضافيًا في الردود أسفل هذا المنشور' % (len(links) - len(shown))
    tag = '#' + re.sub(r'[^A-Za-z0-9_]', '_', sess.user)
    if sess.published:
        stats = '⌛️ المدة: %d:%02d ساعة • 🎬 فيديو كامل بلا تجزئة • 💾 %.1f MB' % (
            mins // 60, mins % 60, total_bytes / 1048576)
        hdr = '🎬 <b>البث الكامل</b> — اضغط للمشاهدة:'
    else:
        stats = '⌛️ المدة: %d:%02d ساعة • 📼 الأجزاء: %d • 💾 %.1f MB' % (
            mins // 60, mins % 60, sess.sent_count, total_bytes / 1048576)
        hdr = '🗂 <b>فهرس الأجزاء</b> — اضغط أي جزء للمشاهدة:'
    text = ('✅ <b>انتهى البث واكتمل الأرشيف</b>\n\n'
            '🎥 <b>%s</b> (@%s)\n🗓 %s ← %s (بغداد)\n'
            '%s\n\n%s\n%s\n\n%s'
            % (sess.info.get('nickname') or sess.user, sess.user,
               bag(sess.started), bag(sess.ended), stats, hdr, index, tag))
    if tg_edit(text, sess.card_id):
        log(f'card->index @{sess.user} msg={sess.card_id} parts={len(links)}')
    else:
        log(f'card edit failed @{sess.user} msg={sess.card_id}')

def write_manifest(sess):
    total_secs = sum((p.get('seconds') or 0) for p in sess.parts)
    total_bytes = sum((p.get('bytes') or 0) for p in sess.parts)
    man = {'username': sess.user, 'nickname': sess.info.get('nickname'),
           'title': sess.info.get('title'), 'room_id': sess.info.get('room_id'),
           'started_at': iso(sess.started), 'ended_at': iso(sess.ended) if sess.ended else None,
           'duration_seconds': total_secs, 'total_bytes': total_bytes,
           'telegram_card': sess.card_id, 'bucket_prefix': sess.bucket_prefix,
           'albums': sess.albums, 'published': sess.published, 'parts': sess.parts}
    try: bucket_put_bytes(json.dumps(man, ensure_ascii=False, indent=1).encode(),
                           f'{sess.bucket_prefix}/manifest.json')
    except Exception as e: log(f'manifest failed {sess.id}: {type(e).__name__}')
    return total_secs, total_bytes

# ---------------------------------------------------------------- FLV spool
# Tag-level pull (olived-style). TikTok switches the encoder configuration
# (SPS/PPS, sometimes resolution) in the middle of a broadcast whenever the
# broadcaster's uplink wobbles. The old pipeline (ffmpeg -c copy segmenter +
# per-part MP4 remux) froze the first avcC into every part, so all frames
# after a switch decoded with the wrong config -> colour garbage, sometimes
# for the rest of the broadcast. The spooler below keeps the FLV tag stream
# intact and rotates the segment file exactly at a sequence-header change,
# so every archived part is configuration-pure and carries its own avcC.
SPOOL_READ_TIMEOUT = 25

class _SpoolUnsupported(Exception):
    pass

def mp4_cfg_hash(path):
    """Config identity (SPS+PPS) of an MP4, for era-aware publish grouping."""
    try:
        with open(path, 'rb') as fh:
            data = fh.read(8 * 1024 * 1024)
    except Exception:
        return ''
    idx = 0
    while True:
        j = data.find(b'avcC', idx)
        if j < 4: return ''
        size = int.from_bytes(data[j-4:j], 'big')
        p = data[j+4:j+4+256]
        if 20 < size < 8192 and len(p) > 10 and p[0] == 1:
            try:
                sl = int.from_bytes(p[6:8], 'big'); sps = p[8:8+sl]; off = 8 + sl
                pl = int.from_bytes(p[off+1:off+3], 'big'); pps = p[off+3:off+3+pl]
                if sps and pps:
                    return hashlib.sha256(sps + b'|' + pps).hexdigest()[:16]
            except Exception:
                pass
        idx = j + 4

def _enqueue_part(sess, idx):
    with QLOCK: WORKQ.append((sess, idx))

class _Spool:
    """Incremental FLV parser/writer: one configuration-pure segment at a time."""

    def __init__(self, sess):
        self.sess = sess
        self.buf = bytearray()
        self.hdr_done = False
        self.header = b'FLV\x01\x05\x00\x00\x00\x09'
        self.vcache = None; self.vcfg = None   # latest video seq-header payload/config
        self.acache = None; self.acfg = None   # latest audio seq-header payload/config
        self.fh = None; self.seg_idx = -1
        self.seg_t0 = None; self.seg_wall0 = 0.0
        self.seg_vcfg = None; self.seg_acfg = None
        self.seg_vseq_written = False; self.seg_aseq_written = False
        self.seg_vframes = 0; self.seg_tags = 0
        self.next_idx = sess.next_idx
        for p in sorted(sess.dir.glob('part-*.flv')):
            try: i = int(p.stem.split('-')[1])
            except Exception: continue
            if i >= self.next_idx: self.next_idx = i + 1
            if i not in sess.processed: _enqueue_part(sess, i)
        sess.next_idx = self.next_idx

    # -- segment lifecycle -------------------------------------------------
    def _open_seg(self):
        self.seg_idx = self.next_idx; self.next_idx += 1
        self.sess.next_idx = self.next_idx
        self.fh = open(self.sess.dir / ('part-%03d.flv' % self.seg_idx), 'wb')
        self.fh.write(self.header + b'\x00\x00\x00\x00')
        self.seg_t0 = None; self.seg_wall0 = time.time()
        self.seg_vcfg = None; self.seg_acfg = None
        self.seg_vseq_written = False; self.seg_aseq_written = False
        self.seg_vframes = 0; self.seg_tags = 0

    def _close_seg(self):
        if self.fh is None: return
        try: self.fh.close()
        except Exception: pass
        self.fh = None
        if self.seg_tags > 0: _enqueue_part(self.sess, self.seg_idx)

    def _rotate(self):
        self._close_seg(); self._open_seg()

    def close_file(self):
        if self.fh is not None:
            try: self.fh.close()
            except Exception: pass
            self.fh = None

    def abandon(self):
        idx, frames = self.seg_idx, self.seg_vframes
        self.close_file()
        if frames == 0 and idx >= 0:
            try: (self.sess.dir / ('part-%03d.flv' % idx)).unlink()
            except Exception: pass

    # -- tag writing ---------------------------------------------------------
    def _write(self, ttype, ts, payload):
        rts = max(0, ts - (self.seg_t0 or 0))
        hdr = (bytes((ttype,)) + len(payload).to_bytes(3, 'big')
               + (rts & 0xFFFFFF).to_bytes(3, 'big')
               + bytes(((rts >> 24) & 0xFF,)) + b'\x00\x00\x00')
        self.fh.write(hdr + payload + (11 + len(payload)).to_bytes(4, 'big'))
        self.fh.flush()

    def _on_tag(self, ttype, ts, payload):
        if ttype == 9 and len(payload) >= 5:
            fb, pt = payload[0], payload[1]
            if (fb & 0x0F) != 7: raise _SpoolUnsupported()
            if pt == 0:  # AVC sequence header
                cfg = payload[5:]
                if (self.seg_vframes > 0 and self.seg_vcfg is not None
                        and cfg != self.seg_vcfg):
                    self._rotate()
                    log(f'REC cfg @{self.sess.user} part {self.seg_idx}: '
                        f'encoder config changed -> segment split')
                self.vcache, self.vcfg = payload, cfg
                self._write(9, ts, payload)
                self.seg_vseq_written = True; self.seg_vcfg = cfg
                return
            if pt == 1:  # video data
                key = (fb >> 4) == 1
                if self.seg_vframes > 0 and key and (
                        (self.seg_t0 is not None
                         and ts - self.seg_t0 >= SEGMENT_SECONDS * 1000)
                        or time.time() - self.seg_wall0 > SEGMENT_SECONDS + 120):
                    self._rotate()
                if self.seg_t0 is None: self.seg_t0 = ts
                if not self.seg_vseq_written and self.vcache is not None:
                    self._write(9, ts, self.vcache)
                    self.seg_vseq_written = True; self.seg_vcfg = self.vcfg
                self._write(9, ts, payload)
                self.seg_vframes += 1; self.seg_tags += 1
                return
            self._write(9, ts, payload); self.seg_tags += 1
            return
        if ttype == 8 and len(payload) >= 2:
            fmt, pt = payload[0] >> 4, payload[1]
            if fmt == 10 and pt == 0:  # AAC sequence header
                cfg = payload[2:]
                if (self.seg_vframes > 0 and self.seg_acfg is not None
                        and cfg != self.seg_acfg):
                    self._rotate()
                    log(f'REC cfg @{self.sess.user} part {self.seg_idx}: '
                        f'audio config changed -> segment split')
                self.acache, self.acfg = payload, cfg
                self._write(8, ts, payload)
                self.seg_aseq_written = True; self.seg_acfg = cfg
                return
            if self.seg_t0 is None: self.seg_t0 = ts
            if not self.seg_aseq_written and self.acache is not None:
                self._write(8, ts, self.acache)
                self.seg_aseq_written = True; self.seg_acfg = self.acfg
            self._write(8, ts, payload); self.seg_tags += 1
            return
        self._write(ttype, ts, payload); self.seg_tags += 1

    # -- parsing -------------------------------------------------------------
    def _next_tag(self):
        b = self.buf
        if len(b) < 11: return None
        dsize = int.from_bytes(b[1:4], 'big')
        if dsize > 32 * 1024 * 1024: raise ValueError('implausible FLV tag size')
        if len(b) < 11 + dsize + 4: return None
        ts = int.from_bytes(b[4:7], 'big') | (b[7] << 24)
        payload = bytes(b[11:11 + dsize])
        ttype = b[0]
        del b[:11 + dsize + 4]
        return ttype, ts, payload

    def feed(self, data):
        self.buf += data

    def drain(self):
        if not self.hdr_done:
            if len(self.buf) < 9: return
            if self.buf[:3] != b'FLV': raise _SpoolUnsupported()
            off = int.from_bytes(self.buf[5:9], 'big')
            if off < 9 or off > 1024: raise _SpoolUnsupported()
            if len(self.buf) < off + 4: return
            self.header = bytes(self.buf[:9])
            del self.buf[:off + 4]
            self.hdr_done = True
            self._open_seg()
        while True:
            tag = self._next_tag()
            if tag is None: break
            self._on_tag(*tag)
        if len(self.buf) > 16 * 1024 * 1024:
            raise ValueError('spool buffer overflow')

def pull_spool(sess):
    """Tag-level FLV pull (olived-style). Returns a short reason string when
    the pull dies; raises _SpoolUnsupported for non-AVC streams so the caller
    can fall back to the legacy ffmpeg pull for this session."""
    try:
        r = rq.get(sess.info.get('flv'), stream=True,
                   timeout=(10, SPOOL_READ_TIMEOUT), headers={
                       'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) '
                                     'AppleWebKit/537.36 (KHTML, like Gecko) '
                                     'Chrome/126.0 Safari/537.36',
                       'Referer': 'https://www.tiktok.com/'})
    except Exception as e:
        return f'connect {type(e).__name__}'
    if r.status_code != 200:
        try: r.close()
        except Exception: pass
        return f'http {r.status_code}'
    sp = _Spool(sess)
    reason = 'eof'
    try:
        for chunk in r.iter_content(chunk_size=65536):
            if sess.stop_flag:
                reason = 'stop'; break
            if not chunk: continue
            sp.feed(chunk); sp.drain()
    except _SpoolUnsupported:
        sp.abandon()
        raise
    except Exception as e:
        reason = type(e).__name__
    finally:
        try: r.close()
        except Exception: pass
        sp.close_file()
    return reason

def run_session(sess):
    user = sess.user
    start_card(sess)
    log(f'REC start @{user} room={sess.info.get("room_id")}')
    offline_votes = 0
    while not sess.stop_flag:
        if not sess.use_ffmpeg:
            try:
                reason = pull_spool(sess)
            except _SpoolUnsupported:
                sess.use_ffmpeg = True
                log(f'REC pull @{user}: non-AVC FLV -> legacy ffmpeg pull')
                continue
            if reason != 'stop':
                log(f'REC pull @{user} ended ({reason})')
        else:
            idx0 = sess.next_idx
            seg_pat = str(sess.dir / 'part-%03d.flv')
            cmd = [FFMPEG, '-hide_banner', '-loglevel', 'warning',
                   '-reconnect', '1', '-reconnect_streamed', '1', '-reconnect_delay_max', '5',
                   '-rw_timeout', '20000000', '-i', sess.info['flv'], '-c', 'copy',
                   '-f', 'segment', '-segment_time', str(SEGMENT_SECONDS),
                   '-segment_start_number', str(idx0),
                   '-reset_timestamps', '1', seg_pat]
            err_path = sess.dir / 'ffmpeg.err'
            err_fh = open(err_path, 'ab')
            sess.proc = subprocess.Popen(cmd, stdout=subprocess.DEVNULL, stderr=err_fh)
            seen = set()
            while sess.proc.poll() is None and not sess.stop_flag:
                time.sleep(2)
                parts = sorted(sess.dir.glob('part-*.flv'))
                for p in parts[:-1]:  # all but the currently-written one
                    i = int(p.stem.split('-')[1])
                    if i not in seen:
                        seen.add(i)
                        with QLOCK: WORKQ.append((sess, i))
                sess.next_idx = (int(parts[-1].stem.split('-')[1]) + 1) if parts else idx0
            try: sess.proc.wait(timeout=10)
            except Exception:
                sess.proc.kill()
            try: err_fh.close()
            except Exception: pass
            rc = sess.proc.returncode
            if rc not in (0, None):
                try:
                    tail = err_path.read_bytes()[-500:].decode(errors='replace').replace('\n', ' | ').strip()
                except Exception:
                    tail = ''
                log(f'ffmpeg @{user} exit rc={rc} parts={len(list(sess.dir.glob("part-*.flv")))}: {tail[:320]}')
        # input died: broadcaster still live?
        try: info = resolve(user)
        except Exception: info = None
        if info and not info.get('live_no_url'):
            sess.info = info; sess.restarts += 1; offline_votes = 0
            log(f'REC restart @{user} (#{sess.restarts}) — المصدر رجع')
            continue
        offline_votes += 1
        if offline_votes >= 2: break
        time.sleep(8)
    # finalize: enqueue last part(s)
    for p in sorted(sess.dir.glob('part-*.flv')):
        i = int(p.stem.split('-')[1])
        with QLOCK:
            with sess.proc_lock: already = i in sess.processed
            if not already and all(j != i for _, j in WORKQ):
                WORKQ.append((sess, i))
    sess.ended = now_utc()
    # wait until queue drains for this session (bounded)
    for _ in range(300):
        with QLOCK: pending = any(s is sess for s, _ in WORKQ)
        if not pending and len(sess.parts) >= len(list(sess.dir.glob('part-*.flv'))): break
        time.sleep(2)
    publish_full(sess)
    if not sess.published:
        flush_albums(sess, force=True)
    total_secs, total_bytes = write_manifest(sess)
    end_card(sess, total_secs, total_bytes)
    with LOCK:
        STATE['sessions'][sess.id] = {'user': user, 'started': iso(sess.started),
                                      'ended': iso(sess.ended), 'parts': sess.sent_count,
                                      'bucket': sess.bucket_prefix}
        STATE['live'].pop(user, None)
    save_state()
    log(f'REC end @{user} parts={sess.sent_count}')

def _safe_resolve(user):
    try: return resolve(user)
    except Exception as e:
        log(f'resolve @{user}: {type(e).__name__}'); return None

def manager():
    while True:
        try:
            with LOCK:
                users = list(STATE['users'])
                live_now = {k for k, v in STATE['live'].items() if v}
            active = len(live_now)
            cands = [u for u in users if u not in live_now]
            infos = {}
            if cands:
                with ThreadPoolExecutor(max_workers=8) as ex:
                    infos = dict(zip(cands, ex.map(_safe_resolve, cands)))
            queued = []
            for user in users:
                if user in live_now: continue
                info = infos.get(user)
                if not info: continue
                if info.get('live_no_url'):
                    log(f'@{user} live but no stream URL (restricted?)'); continue
                if active >= MAX_RECORDERS:
                    queued.append(user); continue
                sess = Session(user, info)
                with LOCK:
                    STATE['live'][user] = sess.id
                    STATE['sessions'][sess.id] = {'user': user, 'started': iso(sess.started),
                                                  'ended': None, 'parts': 0, 'bucket': sess.bucket_prefix}
                threading.Thread(target=run_session, args=(sess,), daemon=True).start()
                active += 1
            with LOCK: STATE['queued'] = queued
        except Exception as e:
            log(f'manager: {type(e).__name__} {e}')
        time.sleep(POLL_SECONDS)

# ---------------------------------------------------------------- state
def save_state():
    try:
        data = {'users': STATE['users'], 'sessions': STATE['sessions'],
                'saved_at': iso()}
        bucket_put_bytes(json.dumps(data, ensure_ascii=False).encode(), '_system/state.json')
    except Exception:
        pass

def load_state():
    raw = bucket_get('_system/users.json')
    if raw:
        try:
            users = json.loads(raw).get('users')
            if users: STATE['users'] = [str(u).lstrip('@').lower() for u in users]
        except Exception: pass
    raw = bucket_get('_system/state.json')
    if raw:
        try:
            old = json.loads(raw).get('sessions') or {}
            STATE['sessions'] = {k: v for k, v in old.items()}
        except Exception: pass

def save_users():
    try: bucket_put_bytes(json.dumps({'users': STATE['users']}).encode(), '_system/users.json')
    except Exception: pass
    save_state()

# ---------------------------------------------------------------- cleanup
def cleanup():
    while True:
        try:
            horizon = now_utc() - timedelta(days=RETENTION_DAYS)
            # local dirs of ended sessions older than retention
            root = WORK / 'recordings'
            if root.exists():
                for d in root.glob('*/*'):
                    try:
                        stamp = datetime.strptime(d.name.rsplit('_', 1)[1], '%Y%m%d-%H%M%S').replace(tzinfo=timezone.utc)
                    except Exception:
                        continue
                    if stamp < horizon:
                        subprocess.run(['rm', '-rf', str(d)])
                        log(f'cleanup local {d}')
            # bucket media older than retention (manifests kept as index)
            old = []
            for p in bucket_list():
                m = re.match(r'[^/]+/(\d{4}-\d{2}-\d{2})/', p)
                if m and not p.endswith('manifest.json'):
                    day = datetime.strptime(m.group(1), '%Y-%m-%d').replace(tzinfo=timezone.utc)
                    if day < horizon - timedelta(days=1): old.append(p)
            if old:
                bucket_delete(old); log(f'cleanup bucket media: {len(old)} files')
        except Exception as e:
            log(f'cleanup: {type(e).__name__}')
        time.sleep(3600)

# ---------------------------------------------------------------- web
app = FastAPI(docs_url=None, redoc_url=None, openapi_url=None)

def authed(request):
    h = request.headers.get('authorization') or ''
    if not h.lower().startswith('basic '): return False
    try:
        raw = base64.b64decode(h.split(None, 1)[1]).decode()
        u, _, p = raw.partition(':')
        return u == 'admin' and p == DASH_PASSWORD
    except Exception:
        return False

@app.get('/health')
def health():
    with LOCK:
        return {'status': 'ok', 'recording': sorted(STATE['live'].keys()),
                'queued': STATE['queued'], 'watching': len(STATE['users']),
                'max_recorders': MAX_RECORDERS, 'auth_users': auth_users()}


@app.get('/', response_class=HTMLResponse)
def landing():
    with LOCK:
        n_users = len(STATE['users']); n_live = sum(1 for v in STATE['live'].values() if v)
    return f"""<!doctype html><meta charset="utf-8"><title>record — مسجل البثوث</title>
<style>body{{font-family:system-ui;background:#0b0f14;color:#e6edf3;margin:2rem;text-align:center}}
a{{color:#58a6ff}}</style>
<h2>🔴 record — مسجّل بثوث TikTok</h2>
<p>الحالة: يعمل • يراقب {n_users} مذيعًا • يسجّل الآن {n_live} • السقف {MAX_RECORDERS} بثوث متزامنة</p>
<p>📺 الأرشيف الدائم: <a href="https://t.me/sh99bg">قناة شباجةtv</a></p>
<p><a href="/panel">🔐 لوحة التحكم</a></p>"""

@app.get('/panel', response_class=HTMLResponse)
def home(request: Request):
    if not authed(request):
        return HTMLResponse('<h3>🔒</h3>', status_code=401,
                            headers={'WWW-Authenticate': 'Basic'})
    with LOCK:
        users = list(STATE['users']); live = dict(STATE['live']); queued = list(STATE['queued'])
        sessions = dict(STATE['sessions'])
    rows = ''
    for u in users:
        st = '🔴 يسجل الآن' if u in live else ('⏳ بانتظار مقعد' if u in queued else '⚪️ غير مباشر')
        rows += (f'<tr><td>@{u}</td><td>{st}</td>'
                 f'<td><form method="post" action="/api/users">'
                 f'<input type="hidden" name="action" value="remove">'
                 f'<input type="hidden" name="username" value="{u}">'
                 f'<button>حذف</button></form></td></tr>')
    done = [s for s in sessions.values() if s.get('ended')]
    done.sort(key=lambda s: s.get('ended') or '', reverse=True)
    srows = ''.join('<tr><td>@%s</td><td>%s</td><td>%s</td><td>%s جزء</td></tr>'
                    % (s['user'], (s.get('started') or '')[:16].replace('T', ' '),
                       (s.get('ended') or '')[:16].replace('T', ' '), s.get('parts'))
                    for s in done[:15])
    return f'''<!doctype html><meta charset="utf-8"><title>record</title>
<style>body{{font-family:system-ui;background:#0b0f14;color:#e6edf3;margin:2rem}}
table{{border-collapse:collapse}}td,th{{border:1px solid #26303c;padding:.4rem .8rem}}
input,button{{padding:.35rem .6rem}}</style>
<h2>🔴 record — مسجّل البثوث</h2>
<p>يراقب {len(users)} • يسجل الآن {len(live)} • السقف {MAX_RECORDERS} • التنظيف بعد {RETENTION_DAYS} يوم</p>
<table><tr><th>المذيع</th><th>الحالة</th><th></th></tr>{rows}</table>
<form method="post" action="/api/users" style="margin-top:1rem">
<input type="hidden" name="action" value="add">
<input name="username" size="52" placeholder="أسماء مفصولة بمسافات أو فواصل — مثال: talolil, ahmadfamouss" required>
<button>➕ إضافة</button></form>
<p style="color:#8b98a5;font-size:.9em">💡 قائمة المراقبة بلا حد — الصق كل الحسابات دفعة واحدة (الأساسية أولًا ثم الاحتياطية؛ الترتيب = أولوية المقاعد عند التزاحم)</p>
<h3>آخر الجلسات المكتملة</h3>
<table><tr><th>المذيع</th><th>بدأ</th><th>انتهى</th><th>الأجزاء</th></tr>{srows}</table>'''

@app.post('/api/users')
async def users_api(request: Request):
    if not authed(request): return JSONResponse({'detail': 'Unauthorized'}, status_code=401)
    form = await request.form()
    action = form.get('action')
    names = [n.strip().lstrip('@').lower()
             for n in re.split(r'[\s,;]+', str(form.get('username') or ''))]
    names = [n for n in names if re.fullmatch(r'[a-z0-9_.]{1,32}', n or '')]
    if not names:
        return HTMLResponse('<script>history.back()</script>')
    with LOCK:
        for n in names:
            if action == 'add' and n not in STATE['users']: STATE['users'].append(n)
            if action == 'remove' and n in STATE['users']: STATE['users'].remove(n)
    save_users()
    return HTMLResponse('<script>location.href="/panel"</script>')

def boot():
    load_state()
    threading.Thread(target=manager, daemon=True).start()
    threading.Thread(target=worker, daemon=True).start()
    threading.Thread(target=cleanup, daemon=True).start()
    log(f'boot: watching={STATE["users"]} max={MAX_RECORDERS} auth={auth_users()} ffmpeg={FFMPEG}')

try:
    import gradio as gr
    demo = gr.Blocks(title='record')
    with demo:
        gr.Markdown('# 🔴 record — مسجّل البثوث يعمل. لوحة التحكم: /panel')
    app = gr.mount_gradio_app(app, demo, path='/gradio', ssr_mode=False)
except Exception:
    demo = None


try:
    import spaces

    @spaces.GPU
    def _zero_gpu_dummy():
        return 'ok'
except Exception:
    _zero_gpu_dummy = None


if __name__ == '__main__':
    import uvicorn
    try:
        import spaces.zero
        spaces.zero.startup()
        log('zerogpu startup report sent')
    except Exception as e:
        log(f'zerogpu startup skipped: {type(e).__name__}')
    boot()
    uvicorn.run(app, host='0.0.0.0', port=int(os.getenv('PORT', '7860')))