#!/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_ / TIKTOK_SESSIONID_ 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 ('🎥 %s (@%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 ('🎥 %s (@%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 = '🎥 %s (@%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 = ('🔴 بدأ بث مباشر — جارٍ التسجيل\n\n' '🎥 %s (@%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 = ('✅ انتهى البث واكتمل الأرشيف\n\n' '🎥 %s (@%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('%s' % (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('%s' % (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('الجزء %d' % (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 = '🎬 البث الكامل — اضغط للمشاهدة:' else: stats = '⌛️ المدة: %d:%02d ساعة • 📼 الأجزاء: %d • 💾 %.1f MB' % ( mins // 60, mins % 60, sess.sent_count, total_bytes / 1048576) hdr = '🗂 فهرس الأجزاء — اضغط أي جزء للمشاهدة:' text = ('✅ انتهى البث واكتمل الأرشيف\n\n' '🎥 %s (@%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"""record — مسجل البثوث

🔴 record — مسجّل بثوث TikTok

الحالة: يعمل • يراقب {n_users} مذيعًا • يسجّل الآن {n_live} • السقف {MAX_RECORDERS} بثوث متزامنة

📺 الأرشيف الدائم: قناة شباجةtv

🔐 لوحة التحكم

""" @app.get('/panel', response_class=HTMLResponse) def home(request: Request): if not authed(request): return HTMLResponse('

🔒

', 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'@{u}{st}' f'
' f'' f'' f'
') done = [s for s in sessions.values() if s.get('ended')] done.sort(key=lambda s: s.get('ended') or '', reverse=True) srows = ''.join('@%s%s%s%s جزء' % (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'''record

🔴 record — مسجّل البثوث

يراقب {len(users)} • يسجل الآن {len(live)} • السقف {MAX_RECORDERS} • التنظيف بعد {RETENTION_DAYS} يوم

{rows}
المذيعالحالة

💡 قائمة المراقبة بلا حد — الصق كل الحسابات دفعة واحدة (الأساسية أولًا ثم الاحتياطية؛ الترتيب = أولوية المقاعد عند التزاحم)

آخر الجلسات المكتملة

{srows}
المذيعبدأانتهىالأجزاء
''' @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('') 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('') 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')))