| |
| """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: |
| cr = None |
|
|
| |
| 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 |
| TG_ALBUM_MAX_WAIT = 1800 |
| TG_ALBUM_MAX_BYTES = 44 * 1024 * 1024 |
| 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_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)) |
|
|
| |
| 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: |
| 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 |
|
|
| |
| |
| |
| |
| |
| |
| 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 |
|
|
| |
| 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 = [] |
| 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 = [] |
| 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 = [] |
| 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) |
|
|
| |
| TG_BIG_MAX_BYTES = 1900 * 1024 * 1024 |
| _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: |
| |
| |
| |
| 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 |
|
|
| |
| |
| |
| |
| |
| |
| |
| |
| |
| 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 |
| self.acache = None; self.acfg = None |
| 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 |
|
|
| |
| 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 |
|
|
| |
| 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: |
| 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: |
| 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: |
| 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 |
|
|
| |
| 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]: |
| 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]}') |
| |
| 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) |
| |
| 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() |
| |
| 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) |
|
|
| |
| 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() |
|
|
| |
| def cleanup(): |
| while True: |
| try: |
| horizon = now_utc() - timedelta(days=RETENTION_DAYS) |
| |
| 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}') |
| |
| 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) |
|
|
| |
| 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'))) |
|
|
