import os, re, json, time, threading, subprocess, shutil, secrets, logging from pathlib import Path from urllib.parse import urlparse from concurrent.futures import ThreadPoolExecutor from fastapi import FastAPI, Depends, HTTPException from fastapi.security import HTTPBasic, HTTPBasicCredentials from fastapi.responses import HTMLResponse, FileResponse from pydantic import BaseModel from curl_cffi import requests import tiktok_auth ROOT=Path(os.getenv('DATA_DIR','/data/tiktok-recorder' if Path('/data').is_dir() else './storage')).resolve() ROOT.mkdir(parents=True,exist_ok=True) FILES=ROOT/'recordings'; FILES.mkdir(exist_ok=True) TOKENS=ROOT/'tokens.json' tiktok_auth.configure(ROOT/'tiktok-tokens.json') PASSWORD=os.getenv('PASSWORD','') if not PASSWORD: raise RuntimeError('Set PASSWORD before starting the app') FFMPEG=os.getenv('FFMPEG_PATH') or shutil.which('ffmpeg') if not FFMPEG: raise RuntimeError('ffmpeg is required') MKVMERGE=os.getenv('MKVMERGE_PATH') or shutil.which('mkvmerge') FFPROBE=os.getenv('FFPROBE_PATH') or shutil.which('ffprobe') if not MKVMERGE or not FFPROBE: raise RuntimeError('mkvmerge and ffprobe are required') FINALIZE_LOCK=threading.Lock() INTERVAL=max(30,int(os.getenv('CHECK_INTERVAL','60'))) SEGMENT=max(0,int(os.getenv('SEGMENT_SECONDS','0'))) # 0 = no scheduled split # MAX_ACCOUNTS=0 means unlimited. The old default of 3 blocked adding # legitimate accounts even though the rest of the app can handle them. _MAX_ACCOUNTS=int(os.getenv('MAX_ACCOUNTS','0')) MAX_ACCOUNTS=_MAX_ACCOUNTS if _MAX_ACCOUNTS>0 else None MIN_FREE_MB=int(os.getenv('MIN_FREE_MB','1024')) # SMART CLEANUP: guarantee free space for continuous broadcast KEEP_LAST_FILES=max(1,int(os.getenv('KEEP_LAST_FILES','2'))) # keep newest N files always ENSURE_FREE_MB=max(MIN_FREE_MB*2,int(os.getenv('ENSURE_FREE_MB','2048'))) # target free before/after finalize AUTO_DELETE_MKV=True # never keep .mkv if .mp4 succeeded (user request) lock=threading.RLock(); tasks={}; events=[]; handles={} security=HTTPBasic() def auth(c:HTTPBasicCredentials=Depends(security)): if not (secrets.compare_digest(c.username.encode(),b'admin') and secrets.compare_digest(c.password.encode(),PASSWORD.encode())): raise HTTPException(401,'Unauthorized',headers={'WWW-Authenticate':'Basic'}) app=FastAPI(dependencies=[Depends(auth)],docs_url=None,redoc_url=None,openapi_url=None) def log(user,msg): with lock: events.append({'time':time.strftime('%H:%M:%S'),'user':user,'message':msg}) del events[:-200] def update(user,**kw): with lock: tasks[user].update(kw) def save(): with lock: tmp=ROOT/'tasks.tmp';tmp.write_text(json.dumps({u:t['enabled'] for u,t in tasks.items()}));tmp.replace(ROOT/'tasks.json') def smart_cleanup(reason="periodic"): """Intelligent deletion: always keep KEEP_LAST_FILES newest, delete oldest until ENSURE_FREE_MB is free. Never touches .partial. files (currently recording/finalizing). Deletes .mkv first if MP4 exists.""" try: # Never delete partial files all_completed = [p for p in FILES.iterdir() if p.is_file() and '.partial.' not in p.name and p.suffix in ('.mp4','.ts','.mkv','.flv') and p.name != '.gitkeep'] if not all_completed: return 0 # Sort by mtime: newest first all_completed.sort(key=lambda p: p.stat().st_mtime, reverse=True) # Always keep newest KEEP_LAST_FILES keep = set(all_completed[:KEEP_LAST_FILES]) candidates = all_completed[KEEP_LAST_FILES:] # oldest first will be at end, but we iterate oldest first for deletion candidates.sort(key=lambda p: p.stat().st_mtime) # oldest first freed=0 deleted=[] # Phase 1: delete .mkv files that have a corresponding .mp4 (redundant) - user requested no mkv if AUTO_DELETE_MKV: for p in list(candidates): if p.suffix == '.mkv': # if same stem has mp4, it's redundant stem = p.stem # stem may be like user_20260917_... - check if any mp4 with similar prefix exists in keep or candidates # simpler: just delete all mkv older than keep try: sz=p.stat().st_size p.unlink() freed+=sz deleted.append(p.name) candidates.remove(p) except OSError: pass # also check keep set for mkv that have mp4 equivalent? Keep as is if user wants to keep 2 newest mkv? but user said no need mkv for p in list(keep): if p.suffix == '.mkv': # if there's an mp4 with same date prefix, delete mkv even from keep? But respect KEEP_LAST_FILES: keep mp4 instead # Find if there's a newer mp4 that covers same period - just keep it for now, but log pass # Phase 2: ensure free space free_mb = shutil.disk_usage(FILES).free / (1024*1024) # Also consider if total size of completed files > 80% of disk, proactively clean for p in candidates: if free_mb >= ENSURE_FREE_MB: break try: sz=p.stat().st_size p.unlink() freed+=sz free_mb += sz/(1024*1024) deleted.append(p.name) except OSError: continue if deleted: log('system', f'تنظيف ذكي ({reason}): حذف {len(deleted)} ملف قديم لتوفير {round(freed/1048576,1)}MB — ضمنا {int(free_mb)}MB فراغ للبث المستمر (احتفظنا بآخر {KEEP_LAST_FILES})') return len(deleted) except Exception as e: log('system', f'تعذر التنظيف الذكي: {type(e).__name__}') return 0 # TikTok session tokens (sessionid) — required for age-restricted (18+) lives. # Stored per account; a global fallback can be set via the TIKTOK_SESSIONID secret. _tokens={} def load_tokens(): global _tokens try:_tokens=json.loads(TOKENS.read_text()) except (FileNotFoundError,ValueError,OSError):_tokens={} # One-way migration so tokens saved by earlier versions gain expiry tracking. for u,t in list(_tokens.items()): try: if t and t.strip():tiktok_auth.save(u,t) except (ValueError,OSError):pass def save_tokens(): with lock: tmp=ROOT/'tokens.tmp';tmp.write_text(json.dumps(_tokens));tmp.replace(TOKENS) def set_token(user,token): token=(token or '').strip() with lock: if token: _tokens[user]=token try:tiktok_auth.save(user,token) except ValueError:pass # keep the raw value; the UI reports format problems separately else: _tokens.pop(user,None) tiktok_auth.delete(user) save_tokens() def get_token(user): with lock:return _tokens.get(user) or os.getenv('TIKTOK_SESSIONID','').strip() def tiktok_headers(user): # Accepts a bare sessionid value or a full cookie string. Never logged or echoed back. # tiktok_auth expands it to the cookie set TikTok's own web client sends for # age-restricted rooms: sessionid + sessionid_ss + sid_tt + tt-target-idc. tok=get_token(user) if user else '' if not tok:return {} sessionid,pairs=tiktok_auth.parse_token(tok) if not sessionid:return {} return {'Cookie':'; '.join('%s=%s'%(k,v) for k,v in pairs.items())} def clean_user(raw): raw=raw.strip() if raw.startswith('https://'): p=urlparse(raw) if p.hostname not in ('www.tiktok.com','tiktok.com','vm.tiktok.com','vt.tiktok.com'):raise ValueError('الرابط يجب أن يكون من TikTok') # Only direct profile/live URLs: avoids following arbitrary redirects server-side. m=re.fullmatch(r'/@([\w.]{1,32})(?:/live)?/?',p.path) if not m:raise ValueError('استخدم اسم الحساب أو رابطه الكامل، وليس رابط المشاركة المختصر') raw=m.group(1) raw=raw.lstrip('@') if not re.fullmatch(r'[A-Za-z0-9_.]{1,32}',raw):raise ValueError('اسم حساب غير صالح') return raw.lower() def get_json(url,params,user=None): r=requests.get(url,params=params,impersonate='chrome',headers={'Referer':'https://www.tiktok.com/',**tiktok_headers(user)},timeout=20) r.raise_for_status();return r.json() _observation_cache={} _observation_locks={} _observation_guard=threading.Lock() def observe_session(user): with _observation_guard:guard=_observation_locks.setdefault(user,threading.Lock()) with guard: cached=_observation_cache.get(user) if cached and time.monotonic()-cached[0]<15:return cached[1] result=_observe_session(user) _observation_cache[user]=(time.monotonic(),result) return result def _observe_session(user): j=get_json('https://www.tiktok.com/api-live/user/room',{'aid':1988,'uniqueId':user,'sourceType':54,'staleTime':0},user) d=j.get('data',{});live=d.get('liveRoom',{});status=live.get('status');person=d.get('user',{}) room=str(person.get('roomId') or person.get('room_id') or '') nickname=str(person.get('nickname') or '').strip() live_title=str(live.get('title') or '').strip() if status in (4,'4'):return {'status':'offline','room_id':room,'sources':[],'nickname':nickname,'live_title':live_title} blocks=[live[k] for k in ('streamData','hevcStreamData') if isinstance(live.get(k),dict)] sources=rank_sources({'blocks':blocks}) if status in (2,'2') else [] # Live but every stream block is empty: the classic age-restricted (18+) shape. if status in (2,'2') and not sources and tiktok_auth.classify_restriction(j): raise RuntimeError(MSG) if not room or room=='0': r=requests.get(f'https://www.tiktok.com/@{user}/live',impersonate='chrome',headers={'Referer':'https://www.tiktok.com/',**tiktok_headers(user)},timeout=20) r.raise_for_status();m=re.search(r'"roomId"\s*:\s*"(\d+)"',r.text) room=m.group(1) if m else '' if not room.isdigit() or room=='0':raise RuntimeError('تعذر تحديد معرّف جلسة البث؛ لن ننشئ حدث يوتيوب بالتخمين') if not sources: j=get_json('https://webcast.tiktok.com/webcast/room/info/',{'aid':1988,'room_id':room},user) info=j.get('data',{});status=info.get('status');owner=info.get('owner') or {} nickname=str(owner.get('nickname') or '').strip() or nickname live_title=str(info.get('title') or '').strip() or live_title if status in (4,'4'):return {'status':'offline','room_id':room,'sources':[],'nickname':nickname,'live_title':live_title} if status not in (2,'2'): if tiktok_auth.classify_restriction(j):raise RuntimeError(MSG) raise RuntimeError('حالة تيك توك غير مؤكدة؛ لا نعتبرها انتهاء للبث') sources=rank_sources(info.get('stream_url') or {}) if not sources and tiktok_auth.classify_restriction(j):raise RuntimeError(MSG) return {'status':'live','room_id':room,'sources':sources,'nickname':nickname,'live_title':live_title} def extract(user): observed=observe_session(user) if observed['status']=='offline':return [] if not observed['sources']:raise RuntimeError('البث مباشر لكن روابط الفيديو غير متاحة — '+MSG) return observed['sources'] def rank_sources(source): candidates=[] priorities={'origin':100,'original':100,'origion':100,'uhd':90,'uhd_60':95,'hd_60':75,'full_hd1':90,'full_hd':90,'hd1':70,'hd':70,'sd2':40,'sd':40,'sd1':30,'ld':20} def add(url,label,params=None,assumed=False): if not isinstance(url,str) or not url.startswith('https://') or '.m3u8' in url:return if str(label).lower() in ('ao','audio','audio_only'):return params=params or {};res=str(params.get('resolution','')) nums=re.findall(r'\d+',res) pixels=int(nums[0])*int(nums[1]) if len(nums)>1 else (int(nums[0])**2 if nums else 0) try:rate=float(params.get('vbitrate') or 0) except (ValueError,TypeError):rate=0 quality=priorities.get(str(label).lower(),50) # Explicit original labels first; use declared dimensions/rate when supplied. score=(quality==100, pixels,rate,quality,not assumed) candidates.append({'url':url,'label':str(label)+((' · '+res) if res else '')+' · '+str(params.get('VCodec') or params.get('v_codec') or 'غير محدد'),'score':score}) blocks=list(source.get('blocks',[])) if source.get('live_core_sdk_data'):blocks.append(source['live_core_sdk_data']) for block in blocks: try: payload=block['pull_data']['stream_data'] stream=json.loads(payload) if isinstance(payload,str) else payload for quality,item in stream.get('data',{}).items(): main=item.get('main',{}) try: params=main.get('sdk_params') or {} if isinstance(params,str):params=json.loads(params) except (ValueError,TypeError):params={} add(main.get('flv'),quality,params) except (KeyError,TypeError,ValueError):pass for quality,url in (source.get('flv_pull_url') or {}).items(): try:params=json.loads((source.get('flv_pull_url_params') or {}).get(quality) or '{}') except (ValueError,TypeError):params={} add(url,quality,params) # Do not assume the generic pull URL is the original feed. add(source.get('rtmp_pull_url'),'غير مصنفة',assumed=True) result=[];seen=set() for item in sorted(candidates,key=lambda c:c['score'],reverse=True): if item['url'] not in seen:result.append(item);seen.add(item['url']) return result from youtube_manager import ManagedYouTube def enabled_accounts(): with lock:return [u for u,t in tasks.items() if t.get('enabled')] youtube=ManagedYouTube(FFMPEG,observe_session,log,enabled_accounts,ROOT/'youtube-api-state.json',headers_for=tiktok_headers) class FLVWriter: """Write complete FLV tags only; arbitrary network chunk boundaries are not packet boundaries.""" def __init__(self,out):self.out=out;self.buf=bytearray();self.header=False;self.written=0 def feed(self,chunk): self.buf.extend(chunk) if not self.header: if len(self.buf)<9:return if self.buf[:3]!=b'FLV':raise RuntimeError('المصدر لا يعيد FLV صالحًا') size=int.from_bytes(self.buf[5:9],'big')+4 if not 13<=size<=1048576:raise RuntimeError('ترويسة FLV غير صالحة') if len(self.buf)=11: size=11+int.from_bytes(self.buf[1:4],'big')+4 if len(self.buf)0),None) if not video:raise RuntimeError('لم يوجد مسار فيديو صالح') return video def finalize(raw,stem,user): mkv=FILES/(stem+'.partial.mkv');mp4=FILES/(stem+'.partial.mp4') # SMART: ensure space before heavy remux (needs ~2x file size temporarily) smart_cleanup(reason="قبل البناء") # Serialize remuxes: recording threads continue, but concurrent disk copies are limited. with FINALIZE_LOCK: # Double-check free after waiting for lock if shutil.disk_usage(FILES).free < ENSURE_FREE_MB*1024*1024: smart_cleanup(reason="قبل البناء - طارئ") r=subprocess.run([MKVMERGE,'-o',str(mkv),str(raw)],stdout=subprocess.DEVNULL,stderr=subprocess.DEVNULL,timeout=3600) if r.returncode not in (0,1) or not mkv.exists():raise RuntimeError('تعذر تغليف FLV إلى MKV؛ احتفظنا بالأصل') info=inspect_media(mkv) codec_args=['-tag:v','hvc1'] if info.get('codec_name')=='hevc' else [] r=subprocess.run([FFMPEG,'-v','error','-y','-i',str(mkv),'-map','0:v:0','-map','0:a:0?','-c','copy',*codec_args,str(mp4)],stdout=subprocess.DEVNULL,stderr=subprocess.DEVNULL,timeout=3600) if r.returncode==0 and mp4.exists() and mp4.stat().st_size>20000: measured=inspect_media(mp4) mp4.replace(FILES/(stem+'.mp4'));mkv.unlink(missing_ok=True);raw.unlink(missing_ok=True) quality=f"{measured['width']}×{measured['height']} · {measured.get('codec_name','')}" if user in tasks:update(user,quality=quality) log(user,'تم حفظ MP4 دون إعادة ترميز؛ قياس الملف: '+quality+' (ليس فحص فك ترميز كاملًا)') # SMART: never keep mkv if mp4 succeeded (user request) # Already deleted mkv above. Also ensure we keep only KEEP_LAST_FILES smart_cleanup(reason="بعد البناء") else: mp4.unlink(missing_ok=True) # If mp4 failed but mkv succeeded, keep mkv but user said no need mkv for youtube. Keep mkv only as fallback # To respect request "ماله داعي نحفظ بث اليوتيوب الي امتداده mkv" -> delete mkv if we prefer mp4 only? Keep mkv only if mp4 failed if AUTO_DELETE_MKV and mkv.exists() and mkv.stat().st_size < 50*1024*1024: # small mkv with no mp4? Keep it but log pass mkv.replace(FILES/(stem+'.mkv'));raw.unlink(missing_ok=True) log(user,'تعذر MP4؛ تم حفظ MKV صالح حسب فحص المسارات') smart_cleanup(reason="بعد بناء MKV") def record(user,candidate,stop): # SMART: pre-check and clean before starting new segment if shutil.disk_usage(FILES).free < ENSURE_FREE_MB*1024*1024: smart_cleanup(reason="قبل التسجيل") if shutil.disk_usage(FILES).free < MIN_FREE_MB*1024**2: raise RuntimeError('مساحة نظام الملفات منخفضة — تم التنظيف الذكي وما زالت ممتلئة') stem=f'{user}_{time.strftime("%Y%m%d_%H%M%S")}_{secrets.token_hex(2)}' raw=FILES/(stem+'.partial.flv');start=time.monotonic();last_update=start if user in tasks:update(user,status='جاري الاتصال',quality='مصدر: '+candidate['label'],received_mb=0,elapsed_seconds=0) response=requests.get(candidate['url'],impersonate='chrome',headers={'Referer':'https://www.tiktok.com/',**tiktok_headers(user)},stream=True,timeout=20) try: response.raise_for_status() with open(raw,'wb') as out: writer=FLVWriter(out) try: for chunk in response.iter_content(): if stop.is_set() or (SEGMENT and time.monotonic()-start>=SEGMENT):break if chunk:writer.feed(chunk) if time.monotonic()-last_update>=2: out.flush();last_update=time.monotonic() if user in tasks:update(user,status='تسجيل — وصول بيانات',received_mb=round(writer.written/1048576,1),elapsed_seconds=int(last_update-start)) if shutil.disk_usage(FILES).free=MAX_ACCOUNTS: raise HTTPException(400,'بلغت حد الموارد المحدد؛ يمكنك تغييره عبر MAX_ACCOUNTS') tasks.setdefault(u,{'enabled':True,'status':'بدء التشغيل'});tasks[u]['enabled']=True;save();start(u) if a.token and a.token.strip():set_token(u,a.token) return {'ok':True} @app.post('/api/accounts/{user}/token') def account_token(user:str,b:TokenBody): with lock: if user not in tasks:raise HTTPException(404) set_token(user,b.token) return {'ok':True,'has_token':bool(get_token(user)),'tiktok':tiktok_auth.describe([user])} @app.get('/api/tiktok/token') def tiktok_token_state(): """Masked token state only — never the token itself.""" with lock:users=list(tasks.keys()) return tiktok_auth.describe(users) @app.post('/api/tiktok/token/validate') def tiktok_token_validate(b:TokenBody): """Ask TikTok whether the pasted token is alive. A rejected token is not saved.""" raw=(b.token or '').strip() if not raw:raise HTTPException(400,'الصق التوكن أولًا') report=tiktok_auth.validate(raw) if report.get('ok'): # Only a token TikTok accepted reaches the store. try:tiktok_auth.save('',raw) except ValueError as e:raise HTTPException(400,str(e)) log('system','حُفظ توكن تيك توك %s — سيُستخدم للرصد والتنزيل'%report.get('sessionid_masked','')) else: log('system','تيك توك رفض التوكن؛ لم يُحفظ') return report @app.post('/api/tiktok/token/check/{user}') def tiktok_token_check(user:str): """Live proof: does this account's room yield video URLs with vs without the token?""" try:return tiktok_auth.unlock_check(user) except Exception as e: raise HTTPException(400,str(e) if isinstance(e,(ValueError,RuntimeError)) else 'تعذر اختبار فتح الحساب') @app.delete('/api/tiktok/token/{user}') def tiktok_token_delete(user:str): with lock: _tokens.pop(user,None);save_tokens() return {'ok':tiktok_auth.delete(user)} @app.post('/api/accounts/{user}/{action}') def action(user:str,action:str): with lock: if user not in tasks:raise HTTPException(404) if action=='start': if user in handles and handles[user]['thread'].is_alive() and handles[user]['stop'].is_set():raise HTTPException(409,'انتظر اكتمال إيقاف المقطع') tasks[user]['enabled']=True;start(user) elif action=='stop': tasks[user]['enabled']=False if user in handles:handles[user]['stop'].set() elif action=='delete': if user in handles and handles[user]['thread'].is_alive():raise HTTPException(409,'أوقف الحساب وانتظر انتهاء الحفظ قبل حذفه') del tasks[user] _tokens.pop(user,None);save_tokens() else:raise HTTPException(400) save() return {'ok':True} def safe_file(name): p=(FILES/name).resolve() if p.parent!=FILES or '.partial.' in p.name or p.suffix not in ('.ts','.mp4','.mkv','.flv') or not p.is_file():raise HTTPException(404) return p @app.get('/api/files/{name}') def download(name:str):return FileResponse(safe_file(name),filename=name) @app.delete('/api/files/{name}') def delete(name:str):safe_file(name).unlink();return {'ok':True} @app.on_event('startup') def startup(): load_tokens() # smart cleanup on startup to guarantee space try: smart_cleanup(reason="بدء التشغيل") except Exception: pass try:saved=json.loads((ROOT/'tasks.json').read_text()) except FileNotFoundError:saved={} except (OSError,ValueError): logging.exception('Unable to restore task settings') log('system','تعذر قراءة الحسابات المحفوظة؛ تحقق من التخزين قبل إضافة حسابات') saved={} if not isinstance(saved,dict): log('system','صيغة ملف الحسابات غير صالحة');saved={} restored=list(saved.items()) if MAX_ACCOUNTS is not None: restored=restored[:MAX_ACCOUNTS] for u,enabled in restored: try:u=clean_user(u) except ValueError:continue tasks[u]={'enabled':bool(enabled),'status':'متوقف'} if enabled:start(u) youtube.restore() @app.on_event('shutdown') def shutdown(): youtube.stop() for h in list(handles.values()):h['stop'].set() deadline=time.monotonic()+50 for h in list(handles.values()):h['thread'].join(timeout=max(0,deadline-time.monotonic())) # periodic smart janitor every 10 minutes def janitor_loop(): while True: time.sleep(600) try: smart_cleanup(reason="دوري كل 10 دقايق") except Exception: pass threading.Thread(target=janitor_loop, daemon=True).start()