Spaces:
Paused
Paused
| """TikTok Live Monitor → FFmpeg relay → YouTube RTMPS. No local recording.""" | |
| import os, re, json, time, threading, shutil, secrets, logging | |
| from concurrent.futures import ThreadPoolExecutor, as_completed | |
| from pathlib import Path | |
| from urllib.parse import urlparse | |
| from fastapi import FastAPI, Depends, HTTPException, Request | |
| from fastapi.security import HTTPBasic, HTTPBasicCredentials | |
| from fastapi.responses import HTMLResponse, JSONResponse | |
| from pydantic import BaseModel | |
| from curl_cffi import requests | |
| import tiktok_auth | |
| from ingest import hub, env_rtmp_destination, rtmp_configured, format_session_title, APP_VERSION, max_concurrent_relays | |
| MSG=('تيك توك يحجب روابط البث لهذا الحساب (غرفة مقيدة بعمر 18+ أو جلسة تحتاج تحققًا)؛ ' | |
| 'الصق كوكيز الجلسة الكامل في لوحة «توكن تيك توك» ثم أعد المحاولة') | |
| STORAGE_WAIT=max(0,float(os.getenv('STORAGE_WAIT_SECONDS','30'))) | |
| def _mounted(path): | |
| try: | |
| with open('/proc/mounts','rb') as fh: | |
| names={l.decode().split()[1] for l in fh.read().split(b'\n') if len(l.split())>=2} | |
| return path in names | |
| except OSError:return False | |
| DATA_DIR=os.getenv('DATA_DIR','').strip() | |
| if DATA_DIR: | |
| ROOT=Path(DATA_DIR).resolve() | |
| else: | |
| ROOT=Path('/data/tiktok-relay') | |
| if Path('/data').is_dir(): | |
| deadline=time.monotonic()+STORAGE_WAIT | |
| warned=False | |
| while not _mounted('/data') and time.monotonic()<deadline: | |
| if not warned: | |
| logging.warning('awaiting the /data storage bucket before choosing the data root') | |
| warned=True | |
| time.sleep(2) | |
| if not _mounted('/data'): | |
| logging.warning('/data is not a mount after %.0fs — state may not survive a reboot', STORAGE_WAIT) | |
| # Keep reading an older recorder folder if this Space already has one. | |
| legacy=Path('/data/tiktok-recorder') | |
| if (legacy/'youtube-api-state.json').exists() or (legacy/'tasks.json').exists(): | |
| ROOT=legacy | |
| else: | |
| logging.warning('no /data mount — using ./storage; YouTube journal and tokens will be lost on reboot') | |
| ROOT=Path('./storage').resolve() | |
| try: | |
| ROOT.mkdir(parents=True,exist_ok=True) | |
| except OSError as e: | |
| logging.error('cannot create %s (%s) — starting read-only', ROOT, type(e).__name__) | |
| def _ensure_root(): | |
| try: | |
| ROOT.mkdir(parents=True,exist_ok=True) | |
| return True | |
| except OSError: | |
| return False | |
| 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') | |
| # ffprobe is optional: used only as a codec hint, never required to relay. | |
| FFPROBE=os.getenv('FFPROBE_PATH') or shutil.which('ffprobe') or '' | |
| INTERVAL=max(8,int(os.getenv('CHECK_INTERVAL','20'))) | |
| _LIMIT=int(os.getenv('MAX_ACCOUNTS','0') or 0) | |
| MAX_ACCOUNTS=_LIMIT if _LIMIT>0 else None | |
| SOURCE_STALL_TIMEOUT=max(0.0,float(os.getenv('SOURCE_STALL_TIMEOUT') or os.getenv('YOUTUBE_SOURCE_STALL_SECONDS','12'))) | |
| os.environ.setdefault('YOUTUBE_SOURCE_STALL_SECONDS',str(int(SOURCE_STALL_TIMEOUT) if SOURCE_STALL_TIMEOUT else 0)) | |
| os.environ.setdefault('YOUTUBE_SCAN_SECONDS',os.getenv('CHECK_INTERVAL','20')) | |
| print('[BOOT] TikTok Live Monitor → Direct Relay → YouTube version=%s CHECK_INTERVAL=%s SOURCE_STALL_TIMEOUT=%s rtmp=%s'%(APP_VERSION,INTERVAL,SOURCE_STALL_TIMEOUT,'yes' if rtmp_configured() else 'no'),flush=True) | |
| lock=threading.RLock(); tasks={}; events=[] | |
| security=HTTPBasic() | |
| _monitor_alive=threading.Event() | |
| _last_monitor_cycle=0.0 | |
| _monitor_stop=threading.Event() | |
| _states=('OFFLINE','CHECKING','LIVE_DETECTED','STARTING','LIVE','STALLED','RECONNECTING','STOPPING','ERROR','BACKOFF') | |
| 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(docs_url=None,redoc_url=None,openapi_url=None) | |
| async def _auth_mw(request:Request,call_next): | |
| # /health is public so the container/orchestrator can probe liveness without secrets. | |
| if request.url.path in ('/health','/healthz'): | |
| return await call_next(request) | |
| header=request.headers.get('authorization') or '' | |
| if header.lower().startswith('basic '): | |
| import base64 | |
| try: | |
| raw=base64.b64decode(header.split(None,1)[1]).decode() | |
| user,sep,pw=raw.partition(':') | |
| if sep and secrets.compare_digest(user.encode(),b'admin') and secrets.compare_digest(pw.encode(),PASSWORD.encode()): | |
| return await call_next(request) | |
| except Exception: | |
| pass | |
| return JSONResponse({'detail':'Unauthorized'},status_code=401,headers={'WWW-Authenticate':'Basic'}) | |
| def log(user,msg): | |
| with lock: | |
| events.append({'time':time.strftime('%H:%M:%S'),'user':user,'message':msg}) | |
| del events[:-100] | |
| def update(user,**kw): | |
| with lock: | |
| if user in tasks: tasks[user].update(kw) | |
| _failure_notes={} | |
| def throttle_failure(user,msg,now=None): | |
| now=time.monotonic() if now is None else now | |
| st=_failure_notes.get(user) | |
| if st and st['msg']==msg and now-st['at']<120: | |
| st['suppressed']+=1;return None | |
| out=msg+((' (كُتمت %d رسالة مطابقة خلال آخر دقيقتين)'%st['suppressed']) if st and st['suppressed'] else '') | |
| _failure_notes[user]={'msg':msg,'at':now,'suppressed':0} | |
| return out | |
| 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 cleanup_entry(user): | |
| _stop_direct(user, 'delete') | |
| with lock: | |
| tasks.pop(user,None) | |
| _tokens.pop(user,None) | |
| try:save_tokens() | |
| except OSError:pass | |
| try:youtube.destinations.pop(user,None) | |
| except Exception:pass | |
| with _observation_guard: | |
| _observation_cache.pop(user,None) | |
| _observation_locks.pop(user,None) | |
| log(user,'[MONITOR] @%s removed'%user) | |
| return None | |
| _tokens={} | |
| def load_tokens(): | |
| global _tokens | |
| try:_tokens=json.loads(TOKENS.read_text()) | |
| except (FileNotFoundError,ValueError,OSError):_tokens={} | |
| 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 | |
| else: | |
| _tokens.pop(user,None) | |
| tiktok_auth.delete(user) | |
| save_tokens() | |
| def get_token(user): | |
| """Effective TikTok session: HF secrets beat the UI store (tiktok_auth.session_for).""" | |
| sess=tiktok_auth.session_for(user) | |
| return (sess.get('cookie') or sess.get('sessionid') or '') if sess else '' | |
| def _cookie_headers(tok): | |
| 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 tiktok_headers(user): | |
| h=tiktok_auth.headers(user) | |
| return {'Cookie':h['Cookie']} if h.get('Cookie') else {} | |
| 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') | |
| 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 _tiktok_http_status(e): | |
| resp=getattr(e,'response',None) | |
| for cand in (getattr(resp,'status_code',None),getattr(e,'status_code',None)): | |
| try:cand=int(cand) | |
| except (TypeError,ValueError):continue | |
| if 100<=cand<600:return cand | |
| return 0 | |
| def _live_has_playable_sources(j): | |
| live=((j or {}).get('data') or {}).get('liveRoom') or {} | |
| if live.get('status') not in (2,'2'): | |
| return True | |
| blocks=[live[k] for k in ('streamData','hevcStreamData') if isinstance(live.get(k),dict)] | |
| try: | |
| return bool(rank_sources({'blocks':blocks})) | |
| except Exception: | |
| return False | |
| def get_json(url,params,user=None): | |
| """Public lives first (no session). Global TikTok cookies only if the room is age-gated.""" | |
| def _pull(headers): | |
| r=requests.get(url,params=params,impersonate='chrome',headers=headers,timeout=20) | |
| r.raise_for_status();return r.json() | |
| public={'Referer':'https://www.tiktok.com/'} | |
| try: | |
| j=_pull(public) | |
| if _live_has_playable_sources(j): | |
| return j | |
| except Exception as e: | |
| status=_tiktok_http_status(e) | |
| if status==404: | |
| raise RuntimeError('تيك توك لا يعرف هذا الحساب (404) — تحقق من اسم المستخدم') from None | |
| if status==429: | |
| raise RuntimeError('تيك توك حدّ الطلبات مؤقتًا (429)؛ سيُعاد المحاولة في الدورة التالية') from None | |
| if 500<=status<600: | |
| raise RuntimeError('تيك توك أعاد %s؛ سيُعاد المحاولة لاحقًا'%status) from None | |
| if status not in (0,401,403): | |
| raise RuntimeError('تعذر الوصول إلى تيك توك (%s%s)'%(type(e).__name__,'، HTTP '+str(status) if status else '')) from None | |
| j=None | |
| # Age-gated / empty public payload: one global session for every account. | |
| try: | |
| j=_pull(tiktok_auth.headers(user)) | |
| return j | |
| except Exception as e: | |
| status=_tiktok_http_status(e) | |
| if status in (401,403): | |
| raise RuntimeError('تيك توك رفض الجلسة (%s) — الغرفة مقيدة (18+) والتوكن العام لا يغطيها؛ حدّث TIKTOK_SESSIONID'%status) from None | |
| if status==404: | |
| raise RuntimeError('تيك توك لا يعرف هذا الحساب (404) — تحقق من اسم المستخدم') from None | |
| if status==429: | |
| raise RuntimeError('تيك توك حدّ الطلبات مؤقتًا (429)؛ سيُعاد المحاولة في الدورة التالية') from None | |
| if 500<=status<600: | |
| raise RuntimeError('تيك توك أعاد %s؛ سيُعاد المحاولة لاحقًا'%status) from None | |
| raise RuntimeError('تعذر الوصول إلى تيك توك (%s%s)'%(type(e).__name__,'، HTTP '+str(status) if status else '')) from None | |
| _observation_cache={} | |
| _observation_locks={} | |
| _observation_guard=threading.Lock() | |
| def _as_dict(x): | |
| return x if isinstance(x, dict) else {} | |
| def _offline_obs(room='', nickname='', live_title=''): | |
| rid=str(room or '') | |
| if not rid.isdigit() or rid=='0':rid='' | |
| return {'status':'offline','room_id':rid,'sources':[],'nickname':nickname or '','live_title':live_title or ''} | |
| def observe_session(user,force=False): | |
| with _observation_guard:guard=_observation_locks.setdefault(user,threading.Lock()) | |
| with guard: | |
| cached=_observation_cache.get(user) | |
| if cached and not force 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): | |
| """TikTok room lookup. Null liveRoom/user/data must be Offline, not a crash. | |
| Production 67c9415: several accounts returned liveRoom=null. `.get` on None | |
| became AttributeError and the monitor stored the generic 'فشل طلب بيانات البث' | |
| even while one relay was healthy. | |
| """ | |
| j=_as_dict(get_json('https://www.tiktok.com/api-live/user/room',{'aid':1988,'uniqueId':user,'sourceType':54,'staleTime':0},user)) | |
| d=_as_dict(j.get('data')); live=_as_dict(d.get('liveRoom')); person=_as_dict(d.get('user')) | |
| status=live.get('status') | |
| room=str(person.get('roomId') or person.get('room_id') or live.get('roomId') or live.get('room_id') or '') | |
| nickname=str(person.get('nickname') or live.get('nickname') or '').strip() | |
| live_title=str(live.get('title') or '').strip() | |
| if status in (4,'4') or (not live and (not room or room=='0')): | |
| return _offline_obs(room, nickname, 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 [] | |
| if status in (2,'2') and not sources and tiktok_auth.classify_restriction(j): | |
| raise RuntimeError(MSG) | |
| if not room or room=='0': | |
| try: | |
| 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 '' | |
| except Exception: | |
| room=room if str(room).isdigit() else '' | |
| if not room.isdigit() or room=='0': | |
| if status not in (2,'2'): | |
| return _offline_obs('', nickname, live_title) | |
| raise RuntimeError('تعذر تحديد معرّف جلسة البث؛ لن ننشئ حدث يوتيوب بالتخمين') | |
| if not sources: | |
| j=_as_dict(get_json('https://webcast.tiktok.com/webcast/room/info/',{'aid':1988,'room_id':room},user)) | |
| info=_as_dict(j.get('data')); status=info.get('status'); owner=_as_dict(info.get('owner')) | |
| 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 _offline_obs(room, nickname, live_title) | |
| if status not in (2,'2'): | |
| if tiktok_auth.classify_restriction(j):raise RuntimeError(MSG) | |
| raise RuntimeError('حالة تيك توك غير مؤكدة؛ لا نعتبرها انتهاء للبث') | |
| sources=rank_sources(_as_dict(info.get('stream_url'))) | |
| 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 source_brief(item): | |
| """Human label with no URL/token: 'origin 1080p H264'.""" | |
| q=str(item.get('quality_name') or (str(item.get('label') or '').split('·')[0]) or 'src').strip() | |
| q=re.sub(r'\s+','',q) or 'src' | |
| res=str(item.get('resolution') or '') | |
| nums=[int(n) for n in re.findall(r'\d+',res)] | |
| pretty='' | |
| if len(nums)>=2: | |
| h=max(nums[0],nums[1]) | |
| pretty='%dp'%h if h>=144 else '%sx%s'%(nums[0],nums[1]) | |
| elif nums: | |
| pretty='%dp'%nums[0] | |
| codec=str(item.get('codec') or '').strip().lower() | |
| if any(k in codec for k in ('h264','avc')):codec='H264' | |
| elif any(k in codec for k in ('h265','hevc','hvc1')):codec='HEVC' | |
| else:codec=codec.upper() or '' | |
| media=str(item.get('media') or '').upper() | |
| parts=[q] | |
| if pretty:parts.append(pretty) | |
| if codec:parts.append(codec) | |
| if media and media!='FLV':parts.append(media) | |
| return ' '.join(parts) | |
| def rank_sources(source): | |
| """Every usable pull URL, ranked for *relay*, not for archive quality. | |
| Priority: (1) actually relayable, (2) YouTube-compatible codec, (3) quality, | |
| (4) sane bitrate. H264 is preferred over HEVC of the same session because | |
| YouTube ingest + stream-copy is H264; HEVC is kept as a fallback, never dropped. | |
| """ | |
| 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,'hd5':65,'sd2':40,'sd':40,'sd1':30,'ld':20,'ld5':15} | |
| def add(url,label,media,params=None,assumed=False): | |
| if not isinstance(url,str) or not url.startswith('https://'):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) | |
| width=int(nums[0]) if nums else 0 | |
| height=int(nums[1]) if len(nums)>1 else 0 | |
| pixels=width*height if height else (width*width if width else 0) | |
| try:rate=float(params.get('vbitrate') or params.get('bitrate') or 0) | |
| except (ValueError,TypeError):rate=0 | |
| quality=priorities.get(str(label).lower(),50) | |
| codec=str(params.get('VCodec') or params.get('v_codec') or params.get('codec') or '') | |
| cl=codec.lower() | |
| h264=any(k in cl for k in ('h264','avc')) | |
| hevc=any(k in cl for k in ('h265','hevc','hvc1')) | |
| # FLV is the pipe we control; HLS is a fallback (FFmpeg fetches it itself). | |
| flv=1 if media=='flv' else 0 | |
| # Compatible copy path first — never assume HEVC is "better" because it is sharper. | |
| compat=2 if h264 else (1 if not hevc else 0) | |
| # Extreme bitrates often stall FFmpeg/YouTube on small hosts. | |
| rate_score=rate if rate<=6500 else max(0,13000-rate) | |
| score=(flv, compat, quality, pixels, rate_score, not assumed) | |
| item={'url':url,'media':media,'codec':codec,'quality_name':str(label), | |
| 'resolution':res,'width':width,'height':height,'bitrate':rate, | |
| 'label':str(label)+((' · '+res) if res else '')+' · '+(codec or 'غير محدد')+((' · '+media.upper()) if media!='flv' else ''), | |
| 'score':score} | |
| item['brief']=source_brief(item) | |
| candidates.append(item) | |
| 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,'flv',params) | |
| add(main.get('hls'),quality,'hls',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,'flv',params) | |
| for quality,url in (source.get('hls_pull_url') or {}).items(): | |
| add(url,quality,'hls') | |
| add(source.get('rtmp_pull_url'),'غير مصنفة','flv',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 | |
| from relay import Sender, SourceStalled, RelayStartError | |
| from flv_bridge import FormatChanged | |
| 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) | |
| if FFPROBE: | |
| youtube.ffprobe=FFPROBE | |
| _direct_handles={} # user -> {thread,stop,sender} | |
| _direct_backoff={} | |
| _direct_unstable={} | |
| _direct_guard=threading.RLock() | |
| def _oauth_automation_on(): | |
| try:return bool(youtube.journal.read().get('enabled')) | |
| except Exception:return False | |
| def _stop_direct(user, reason='stop'): | |
| with _direct_guard: | |
| h=_direct_handles.pop(user, None) | |
| if not h:return | |
| h['stop'].set() | |
| try: | |
| if h.get('sender'):h['sender'].terminate() | |
| except Exception:pass | |
| th=h.get('thread') | |
| if th and th.is_alive(): | |
| th.join(timeout=4) | |
| if th.is_alive(): | |
| try:h['sender'].terminate() | |
| except Exception:pass | |
| hub.release(user, 'direct') | |
| log(user, '[RELAY] @%s stopping (%s)'%(user, reason)) | |
| log(user, '[CLEANUP] @%s complete'%user) | |
| def _direct_loop(user, dest, stop): | |
| """TikTok → FFmpeg → RTMPS without requiring OAuth. One session at a time.""" | |
| sender=Sender(FFMPEG, log, tiktok_headers) | |
| if FFPROBE: | |
| try:sender.set_probe(youtube.codec_probe) | |
| except Exception:pass | |
| hub.update(user, sender=sender, thread=threading.current_thread()) | |
| with _direct_guard: | |
| if user in _direct_handles:_direct_handles[user]['sender']=sender | |
| room='' | |
| try: | |
| while not stop.is_set() and not _monitor_stop.is_set(): | |
| try: | |
| obs=observe_session(user, force=True) | |
| except Exception as e: | |
| msg=str(e) if isinstance(e,RuntimeError) else type(e).__name__ | |
| log(user,'[MONITOR] @%s observe failed during relay: %s'%(user, msg[:120])) | |
| if stop.wait(min(30, _direct_backoff.get(user, 2))):return | |
| continue | |
| if obs.get('status')!='live': | |
| log(user,'[MONITOR] @%s OFFLINE'%user) | |
| return | |
| if room and obs.get('room_id') and obs['room_id']!=room: | |
| log(user,'[MONITOR] @%s room changed; clean source transition'%user) | |
| _direct_unstable.pop(user, None) | |
| room=str(obs.get('room_id') or room) | |
| name=str(obs.get('nickname') or user) | |
| title, date, tm=format_session_title(name) | |
| hub.update(user, room_id=room, title=title) | |
| if _oauth_automation_on(): | |
| pass | |
| else: | |
| log(user,'[RELAY] metadata automation unavailable — title would be %s; video continues via RTMPS'%title) | |
| sources=list(obs.get('sources') or []) | |
| bad=_direct_unstable.get(user) or set() | |
| ordered=[c for c in sources if (c.get('brief') or c.get('label')) not in bad]+[c for c in sources if (c.get('brief') or c.get('label')) in bad] | |
| if not ordered: | |
| log(user,'[SOURCE] @%s no sources; waiting'%user) | |
| if stop.wait(3):return | |
| continue | |
| log(user,'[SOURCE] @%s selecting source'%user) | |
| sent=False | |
| for cand in ordered: | |
| if stop.is_set():return | |
| brief=cand.get('brief') or cand.get('label') or 'src' | |
| log(user,'[SOURCE] @%s selected %s'%(user, brief)) | |
| tagged={**cand,'user':user} | |
| try: | |
| log(user,'[RELAY] @%s starting'%user) | |
| update(user, state='LIVE', source=brief, codec=cand.get('codec') or '', quality=cand.get('resolution') or '') | |
| ok=sender.send(user, tagged, dest, stop) | |
| sent=bool(ok) | |
| if sent:break | |
| except RelayStartError as e: | |
| log(user,'[SOURCE] @%s source %s unstable (start failed)'%(user, brief)) | |
| _direct_unstable.setdefault(user,set()).add(brief) | |
| log(user,'[SOURCE] @%s trying fallback'%user) | |
| continue | |
| except SourceStalled as e: | |
| log(user,'[WATCHDOG] @%s data stalled'%user) | |
| _direct_unstable.setdefault(user,set()).add(brief) | |
| update(user, state='STALLED', last_error='source stalled') | |
| log(user,'[SOURCE] @%s trying fallback'%user) | |
| continue | |
| except FormatChanged: | |
| log(user,'[SOURCE] @%s format changed; reconnect'%user) | |
| sent=True | |
| break | |
| except Exception as e: | |
| log(user,'[RELAY] @%s error %s'%(user, type(e).__name__)) | |
| continue | |
| wait=2 if sent else min(60, max(2, (_direct_backoff.get(user,2))*2)) | |
| _direct_backoff[user]=2 if sent else wait | |
| if stop.wait(wait):return | |
| finally: | |
| try:sender.terminate() | |
| except Exception:pass | |
| hub.release(user,'direct') | |
| with _direct_guard:_direct_handles.pop(user, None) | |
| log(user,'[MONITOR] @%s waiting'%user) | |
| def _maybe_start_direct(user, obs): | |
| """Start RTMPS relay when OAuth automation is off but an ingest destination exists.""" | |
| if _oauth_automation_on(): | |
| return | |
| dest=env_rtmp_destination() | |
| if not dest: | |
| update(user, youtube_status='no destination (set YOUTUBE_RTMP_URL + YOUTUBE_STREAM_KEY or enable OAuth)') | |
| return | |
| if hub.busy(user): | |
| return | |
| # One env stream key = one publisher. Do not collide. | |
| if hub.env_key_busy(): | |
| update(user, state='BACKOFF', status='وجهة RTMPS مشغولة بمذيع آخر (مفتاح واحد)') | |
| log(user,'[RELAY] @%s waiting: single env stream key already in use'%user) | |
| return | |
| if hub.at_capacity(except_user=user): | |
| update(user, state='BACKOFF', status='بلغنا حد MAX_CONCURRENT_RELAYS') | |
| log(user,'[RELAY] @%s waiting: at MAX_CONCURRENT_RELAYS=%d'%(user, max_concurrent_relays())) | |
| return | |
| stop=threading.Event() | |
| title=format_session_title(obs.get('nickname') or user)[0] | |
| if not hub.occupy(user,'direct', room_id=obs.get('room_id') or '', title=title): | |
| return | |
| th=threading.Thread(target=_direct_loop, args=(user, dest, stop), daemon=True, name='direct-'+user) | |
| hub.update(user, thread=th, stop=stop) | |
| with _direct_guard: | |
| _direct_handles[user]={'thread':th,'stop':stop,'sender':None} | |
| log(user,'[RELAY] @%s starting (env RTMPS, OAuth not required)'%user) | |
| update(user, state='STARTING', status='بدء Relay') | |
| th.start() | |
| class Account(BaseModel): | |
| username:str | |
| token:str='' | |
| class TokenBody(BaseModel): | |
| token:str='' | |
| class YouTubeStart(BaseModel): | |
| channel_id:str | |
| class YouTubeSettings(BaseModel): | |
| name:str='' | |
| privacy:str | |
| title:str | |
| description:str | |
| send_mode:str='auto' | |
| def _blank_task(): | |
| return {'enabled':True,'status':'بدء التشغيل','state':'OFFLINE','quality':'','source':'', | |
| 'codec':'','live_since':None,'room_id':'','last_error':'','nickname':''} | |
| def _set_state(user,state,status, **extra): | |
| kw={'state':state,'status':status} | |
| kw.update(extra) | |
| update(user,**kw) | |
| def _relay_snapshot(user): | |
| slot=hub.snapshot(user) | |
| sender=slot.get('sender') if slot else None | |
| snap=sender.snapshot() if sender and hasattr(sender,'snapshot') else {} | |
| pid=None | |
| try: | |
| proc=getattr(sender,'process',None) | |
| pid=int(proc.pid) if proc and proc.poll() is None else None | |
| except Exception: | |
| pid=None | |
| return { | |
| 'relay_owner':slot.get('owner') if slot else '', | |
| 'relay_pid':pid, | |
| 'last_data_time':snap.get('last_source_at'), | |
| 'bytes_received':snap.get('source_bytes') or 0, | |
| 'source_age':snap.get('source_age'), | |
| 'output_seconds':snap.get('output_seconds') or 0, | |
| 'relay_alive':bool(pid), | |
| 'session_title':slot.get('title') if slot else '', | |
| 'relay_dest':slot.get('dest') if slot else '', | |
| 'restarts':hub.restart_count(user), | |
| 'send_mode':snap.get('send_mode') or '', | |
| } | |
| def _cgroup_usage_ns(): | |
| try: | |
| with open('/sys/fs/cgroup/cpu.stat') as f: | |
| for line in f: | |
| if line.startswith('usage_usec'): | |
| return int(line.split()[1])*1000 | |
| except Exception: | |
| pass | |
| for cand in ('/sys/fs/cgroup/cpuacct/cpuacct.usage','/sys/fs/cgroup/cpu,cpuacct/cpuacct.usage'): | |
| try: | |
| return int(open(cand).read().strip()) | |
| except Exception: | |
| continue | |
| return None | |
| def _cgroup_quota_cpus(): | |
| try: | |
| parts=open('/sys/fs/cgroup/cpu.max').read().split() | |
| if not parts: | |
| return None | |
| if parts[0]=='max': | |
| return float(os.cpu_count() or 1) | |
| return float(parts[0])/float(parts[1]) | |
| except Exception: | |
| pass | |
| try: | |
| q=int(open('/sys/fs/cgroup/cpu/cpu.cfs_quota_us').read()) | |
| p=int(open('/sys/fs/cgroup/cpu/cpu.cfs_period_us').read()) | |
| if q>0 and p>0: | |
| return q/p | |
| except Exception: | |
| pass | |
| return None | |
| def _cgroup_mem(): | |
| """حدّ الذاكرة والمستهلك منها **لهذه الحاوية** بالميغابايت (لا للخادم المشترك). | |
| /proc/meminfo يصف الآلة المضيفة كاملة (16 نواة و126 غيغا) بينما لوحة Hugging | |
| Face تعرض حصّة الـSpace نفسها (2 vCPU و16 غيغا) — فكان الرقمان يصفان شيئين | |
| مختلفين. cgroup v2 يعطي ما تراه الحاوية فعلًا، فيتطابق الرقمان. | |
| """ | |
| def _read(path): | |
| try: | |
| return open(path).read().strip() | |
| except Exception: | |
| return '' | |
| limit=(_read('/sys/fs/cgroup/memory.max') or _read('/sys/fs/cgroup/memory.high') | |
| or _read('/sys/fs/cgroup/memory/memory.limit_in_bytes')) | |
| if limit=='max': | |
| limit=_read('/sys/fs/cgroup/memory.high') | |
| used=_read('/sys/fs/cgroup/memory.current') or _read('/sys/fs/cgroup/memory/memory.usage_in_bytes') | |
| try: | |
| used_mb=int(used)//1048576 if used else None | |
| except ValueError: | |
| used_mb=None | |
| try: | |
| limit_mb=int(limit)//1048576 if limit and limit!='max' else None | |
| except ValueError: | |
| limit_mb=None | |
| # قيمة خيالية (أكبر من ذاكرة المضيف) تعني «بلا حدّ» على هذا النظام. | |
| if limit_mb and limit_mb>1024*1024: | |
| limit_mb=None | |
| return limit_mb,used_mb | |
| _cpu_lock=threading.Lock() | |
| _cpu_sample={'t':None,'ns':None} | |
| # قراءة واحدة يتشاركها كل القرّاء خلال هذه النافذة. السبب: العيّنة ذات حالة | |
| # (فرقٌ عن الاستدعاء السابق)، وكان /health و/api/state يصفّر كلٌّ منهما أساس | |
| # الآخر ⇒ رقمان مختلفان للحظة واحدة، أو «—» إذا قلّ الفاصل عن 0.4ث. | |
| _RES_TTL=2.0 | |
| _res_cache={'t':None,'data':None} | |
| def resource_snapshot(): | |
| """حصّة هذا الـSpace من المعالج والذاكرة (cgroup)، لا عتاد الخادم المشترك. | |
| القفل يغطّي الدالة كاملة: استدعاءان متزامنان يعيدان نفس القراءة بدل أن | |
| يُفسد كلٌّ منهما عيّنة الآخر. | |
| """ | |
| with _cpu_lock: | |
| now=time.monotonic() | |
| cached=_res_cache['data'] | |
| if cached is not None and _res_cache['t'] is not None and now-_res_cache['t']<_RES_TTL: | |
| return dict(cached) | |
| out={'load1':None,'cpus':os.cpu_count() or 1,'mem_total_mb':None,'mem_avail_mb':None, | |
| 'quota_cpus':None,'cpu_cores_used':None,'cpu_pct_of_quota':None} | |
| try: | |
| out['load1']=float(open('/proc/loadavg').read().split()[0]) | |
| except Exception: | |
| pass | |
| lim,used=_cgroup_mem() | |
| if lim: | |
| out['mem_total_mb']=lim | |
| out['mem_avail_mb']=max(0,lim-(used or 0)) | |
| else: | |
| try: | |
| info={} | |
| for line in open('/proc/meminfo'): | |
| parts=line.split() | |
| if len(parts)>=2:info[parts[0].rstrip(':')]=int(parts[1]) | |
| out['mem_total_mb']=info.get('MemTotal',0)//1024 | |
| out['mem_avail_mb']=info.get('MemAvailable', info.get('MemFree',0))//1024 | |
| except Exception: | |
| pass | |
| out['quota_cpus']=_cgroup_quota_cpus() | |
| if out['quota_cpus']: | |
| # «عدد الأنوية» في اللوحة هو حصّة الـSpace لا أنوية الخادم المشترك. | |
| out['cpus']=out['quota_cpus'] | |
| ns=_cgroup_usage_ns() | |
| prev=_cpu_sample | |
| if prev['t'] is not None and prev['ns'] is not None and ns is not None and now-prev['t']>=0.4: | |
| cores=(ns-prev['ns'])/1e9/(now-prev['t']) | |
| if cores<0: cores=0.0 | |
| out['cpu_cores_used']=round(cores,4) | |
| q=out['quota_cpus'] | |
| if q and q>0: | |
| out['cpu_pct_of_quota']=round(min(999.0, max(0.0, 100.0*cores/q)),2) | |
| elif cached is not None: | |
| # نافذة أقصر من 0.4ث: أعد آخر نسبة معروفة بدل «—». | |
| out['cpu_cores_used']=cached.get('cpu_cores_used') | |
| out['cpu_pct_of_quota']=cached.get('cpu_pct_of_quota') | |
| _cpu_sample['t']=now | |
| _cpu_sample['ns']=ns | |
| _res_cache['t']=now | |
| _res_cache['data']=dict(out) | |
| return dict(out) | |
| def _check_account(user): | |
| """One monitor cycle. Starts DirectRelay when OAuth automation is off.""" | |
| with lock: | |
| tsk=tasks.get(user) | |
| if not tsk or not tsk.get('enabled'):return | |
| relaying=hub.busy(user) | |
| if not relaying: | |
| _set_state(user,'CHECKING','فحص البث') | |
| try: | |
| obs=observe_session(user, force=not relaying) | |
| except Exception as e: | |
| msg=str(e) if isinstance(e,RuntimeError) else 'فشل طلب بيانات البث (%s)'%type(e).__name__ | |
| if relaying: | |
| out=throttle_failure(user, msg) | |
| if out:log(user,'[MONITOR] @%s observe failed; relay continues — %s'%(user,out[:120])) | |
| return | |
| _set_state(user,'ERROR' if 'رفض' in msg else 'BACKOFF', msg, last_error=msg) | |
| out=throttle_failure(user,msg) | |
| if out:log(user,'[MONITOR] @%s %s'%(user,out)) | |
| return | |
| if obs.get('status')!='live': | |
| prev=None | |
| with lock:prev=(tasks.get(user) or {}).get('state') | |
| if relaying: | |
| # small confirmation before killing a live relay | |
| try: | |
| again=observe_session(user, force=True) | |
| except Exception: | |
| again=obs | |
| if again.get('status')=='live': | |
| obs=again | |
| else: | |
| _stop_direct(user, 'tiktok offline') | |
| _set_state(user,'OFFLINE','Offline',quality='',source='',codec='',live_since=None, | |
| room_id=obs.get('room_id') or '', last_error='') | |
| log(user,'[MONITOR] @%s OFFLINE'%user) | |
| log(user,'[MONITOR] @%s waiting'%user) | |
| return | |
| else: | |
| _set_state(user,'OFFLINE','Offline',quality='',source='',codec='',live_since=None, | |
| room_id=obs.get('room_id') or '', last_error='') | |
| if prev!='OFFLINE': | |
| log(user,'[MONITOR] @%s OFFLINE'%user) | |
| return | |
| src=obs.get('sources') or [] | |
| top=src[0] if src else None | |
| brief=top.get('brief') if top else '' | |
| codec=(top.get('codec') if top else '') or '' | |
| quality=(top.get('resolution') if top else '') or '' | |
| live_since=None | |
| with lock: | |
| cur=tasks.get(user) or {} | |
| live_since=cur.get('live_since') or time.time() | |
| prev_state=cur.get('state') | |
| prev_room=cur.get('room_id') | |
| extra={'quality':quality,'source':brief,'codec':codec,'live_since':live_since, | |
| 'room_id':obs.get('room_id') or '','nickname':obs.get('nickname') or '','last_error':''} | |
| extra.update(_relay_snapshot(user)) | |
| if prev_state!='LIVE' or (prev_room and prev_room!=obs.get('room_id')): | |
| log(user,'[MONITOR] @%s LIVE detected'%user) | |
| if top:log(user,'[SOURCE] @%s selected %s'%(user,brief)) | |
| if prev_room and prev_room!=obs.get('room_id') and relaying: | |
| log(user,'[MONITOR] @%s session/room changed; source will be refreshed'%user) | |
| _set_state(user,'LIVE','LIVE',**extra) | |
| _maybe_start_direct(user, obs) | |
| if _oauth_automation_on() and not hub.busy(user) and (obs.get('sources') or []): | |
| try: | |
| youtube.nudge_live(user, obs) | |
| except Exception: | |
| log(user,'[RELAY] @%s youtube nudge failed'%user) | |
| def monitor_loop(): | |
| """Global monitor: every enabled account in parallel. One fault never blocks the rest.""" | |
| global _last_monitor_cycle | |
| while not _monitor_stop.is_set(): | |
| _monitor_alive.set() | |
| started=time.monotonic() | |
| try: | |
| with lock:users=[u for u,t in tasks.items() if t.get('enabled')] | |
| if users: | |
| workers=min(8,len(users)) | |
| with ThreadPoolExecutor(max_workers=workers,thread_name_prefix='mon') as pool: | |
| futs=[pool.submit(_check_account,u) for u in users] | |
| for f in as_completed(futs,timeout=max(15,INTERVAL*2)): | |
| try:f.result() | |
| except Exception as e: | |
| log('system','[MONITOR] worker fault: %s'%type(e).__name__) | |
| _last_monitor_cycle=time.time() | |
| except Exception as e: | |
| log('system','[MONITOR] cycle error: %s'%type(e).__name__) | |
| # Keep YouTube manager alive independently of this loop. | |
| try: | |
| d=youtube.journal.read() | |
| if (d.get('enabled') or d.get('events')) and (not youtube.manager_thread or not youtube.manager_thread.is_alive()): | |
| log('system','[MONITOR] YouTube manager thread died; restarting') | |
| youtube.launch() | |
| except Exception: | |
| pass | |
| wait=max(1.0,INTERVAL-(time.monotonic()-started)) | |
| _monitor_stop.wait(wait) | |
| _monitor_alive.clear() | |
| def index():return Path(__file__).with_name('index.html').read_text() | |
| def _pick_last_error(tasks_map): | |
| """Prefer a LIVE/STALLED/ERROR fault over a stale BACKOFF from an idle account.""" | |
| last_error=''; last_error_user='' | |
| for u,t in tasks_map.items(): | |
| err=t.get('last_error') or '' | |
| if err and t.get('state') in ('LIVE','STALLED','ERROR','STARTING'): | |
| return err, u | |
| for u,t in tasks_map.items(): | |
| err=t.get('last_error') or '' | |
| if err: | |
| return err, u | |
| return last_error, last_error_user | |
| def health_public(): | |
| with lock: | |
| n=len(tasks);enabled=sum(1 for t in tasks.values() if t.get('enabled')) | |
| last_error, last_error_user=_pick_last_error(tasks) | |
| users=list(tasks.items()) | |
| active=hub.active_count() | |
| yt_health={} | |
| try: | |
| for ev in (youtube.snapshot().get('events') or []): | |
| if ev.get('username'): | |
| yt_health[ev['username']]=ev | |
| except Exception: | |
| yt_health={} | |
| per=[] | |
| for u,tsk in users: | |
| rs=_relay_snapshot(u) | |
| ye=yt_health.get(u) or {} | |
| per.append({'username':u,'status':tsk.get('state') or '', | |
| 'last_data_time':rs.get('last_data_time'), | |
| 'bytes_received':rs.get('bytes_received') or 0, | |
| 'relay_pid':rs.get('relay_pid'),'relay_alive':bool(rs.get('relay_alive')), | |
| 'room_id':tsk.get('room_id') or '','source':tsk.get('source') or '', | |
| 'restarts':rs.get('restarts') or 0,'dest':rs.get('relay_dest') or '', | |
| 'last_error':tsk.get('last_error') or '','source_age':rs.get('source_age'), | |
| 'output_seconds':rs.get('output_seconds') or 0, | |
| 'send_mode':rs.get('send_mode') or '', | |
| 'youtube_health':str(ye.get('youtube_health') or ''), | |
| 'youtube_issues':str(ye.get('youtube_issues') or '')[:120], | |
| 'youtube_status':str(ye.get('youtube_status') or '')}) | |
| res=resource_snapshot() | |
| mgr=bool(youtube.manager_thread and youtube.manager_thread.is_alive()) | |
| y_events=0; y_hist=0; y_senders=0; y_status=''; y_skip='' | |
| try: | |
| jd=youtube.journal.read() | |
| y_events=len(jd.get('events') or {}) | |
| y_hist=len(jd.get('history') or []) | |
| y_status=str(getattr(youtube,'status_text','') or '')[:160] | |
| y_senders=sum(1 for e in (youtube.senders or {}).values() if e.get('thread') and e['thread'].is_alive()) | |
| skip=getattr(youtube,'_skip_last',None) or {} | |
| if skip: | |
| y_skip='; '.join('%s: %s'%(u,str(t)[:80]) for u,t in list(skip.items())[-3:])[:200] | |
| try: | |
| acc=list(youtube.accounts() or []) | |
| y_skip=(y_skip+(' | accounts='+','.join(acc)[:80])).strip(' |') | |
| except Exception: | |
| pass | |
| except Exception: | |
| pass | |
| return {'application':'running','monitor':'running' if _monitor_alive.is_set() else 'stopped', | |
| 'accounts':n,'enabled':enabled,'active_relays':active, | |
| 'max_concurrent_relays':max_concurrent_relays(), | |
| 'source_stall_timeout':SOURCE_STALL_TIMEOUT, | |
| 'load1':res.get('load1'),'cpus':res.get('cpus'), | |
| 'mem_total_mb':res.get('mem_total_mb'),'mem_avail_mb':res.get('mem_avail_mb'), | |
| 'quota_cpus':res.get('quota_cpus'),'cpu_cores_used':res.get('cpu_cores_used'), | |
| 'cpu_pct_of_quota':res.get('cpu_pct_of_quota'), | |
| 'last_monitor_cycle':_last_monitor_cycle,'last_error':last_error, | |
| 'last_error_user':last_error_user, | |
| 'version':APP_VERSION,'rtmp_configured':rtmp_configured(), | |
| 'metadata_automation':_oauth_automation_on(), | |
| 'youtube_manager':'running' if mgr else 'stopped', | |
| 'youtube_status':y_status,'youtube_events':y_events,'youtube_history':y_hist, | |
| 'youtube_senders':y_senders,'youtube_skip':y_skip, | |
| 'broadcasters':per} | |
| def health_detail(): | |
| with lock: | |
| n=len(tasks) | |
| last_error, last_error_user=_pick_last_error(tasks) | |
| users=list(tasks.items()) | |
| per=[] | |
| for u,tsk in users: | |
| rs=_relay_snapshot(u) | |
| per.append({'username':u,'status':tsk.get('state'),'last_error':tsk.get('last_error') or '', | |
| 'room_id':tsk.get('room_id') or '','last_data_time':rs.get('last_data_time'), | |
| 'bytes_received':rs.get('bytes_received') or 0,'relay_pid':rs.get('relay_pid'), | |
| 'relay_alive':rs.get('relay_alive'),'source':tsk.get('source') or ''}) | |
| mgr=bool(youtube.manager_thread and youtube.manager_thread.is_alive()) | |
| return {'application':'running','monitor':'running' if _monitor_alive.is_set() else 'stopped', | |
| 'youtube_manager':'running' if mgr else 'stopped', | |
| 'accounts':n,'active_relays':hub.active_count(),'last_monitor_cycle':_last_monitor_cycle, | |
| 'last_error':last_error,'last_error_user':last_error_user,'version':APP_VERSION,'rtmp_configured':rtmp_configured(), | |
| 'oauth_configured':youtube.api.configured(),'metadata_automation':_oauth_automation_on(), | |
| 'broadcasters':per} | |
| def youtube_state():return youtube.snapshot() | |
| def youtube_check(): | |
| try:return {'channel':youtube.check()} | |
| except Exception as e: | |
| from youtube_api import APIError | |
| msg=str(e) if isinstance(e,(APIError,RuntimeError,ValueError)) else 'تعذر فحص تفويض يوتيوب' | |
| raise HTTPException(400,msg) | |
| def youtube_settings(a:YouTubeSettings): | |
| try:youtube.settings(a.model_dump()) | |
| except ValueError as e:raise HTTPException(400,str(e)) | |
| return {'ok':True} | |
| def youtube_start(a:YouTubeStart): | |
| try:youtube.enable(a.channel_id) | |
| except Exception as e: | |
| from youtube_api import APIError | |
| msg=str(e) if isinstance(e,(APIError,RuntimeError,ValueError)) else 'تعذر بدء أتمتة يوتيوب' | |
| raise HTTPException(400,msg) | |
| return {'ok':True} | |
| def youtube_stop(): | |
| youtube.disable();return {'ok':True} | |
| def state(): | |
| with lock: | |
| yt_health={} | |
| try: | |
| for ev in (youtube.snapshot().get('events') or []): | |
| if ev.get('username'): | |
| yt_health[ev['username']]=ev | |
| except Exception: | |
| yt_health={} | |
| task_list=[] | |
| for u,t in tasks.items(): | |
| live_since=t.get('live_since') | |
| live_for=int(time.time()-live_since) if live_since else 0 | |
| rs=_relay_snapshot(u) | |
| ye=yt_health.get(u) or {} | |
| task_list.append({'username':u,'enabled':t.get('enabled'),'status':t.get('status') or '', | |
| 'state':t.get('state') or 'OFFLINE','quality':t.get('quality') or '', | |
| 'source':t.get('source') or '','codec':t.get('codec') or '', | |
| 'live_since':live_since,'live_for':live_for,'room_id':t.get('room_id') or '', | |
| 'last_error':t.get('last_error') or '','nickname':t.get('nickname') or '', | |
| 'has_token':bool(get_token(u)), | |
| 'relay_pid':rs.get('relay_pid'),'bytes_received':rs.get('bytes_received') or 0, | |
| 'last_data_time':rs.get('last_data_time'),'relay_alive':rs.get('relay_alive'), | |
| 'session_title':rs.get('session_title') or '','restarts':rs.get('restarts') or 0, | |
| 'source_age':rs.get('source_age'), | |
| 'youtube_health':str(ye.get('youtube_health') or ''), | |
| 'youtube_issues':str(ye.get('youtube_issues') or '')[:120], | |
| 'youtube_status':str(ye.get('youtube_status') or ('relay' if rs.get('relay_alive') else (t.get('state') or '')))}) | |
| tiktok_state=tiktok_auth.describe([t['username'] for t in task_list]) | |
| active=0 | |
| try:active=sum(1 for e in youtube.senders.values() if e.get('thread') and e['thread'].is_alive()) | |
| except Exception:pass | |
| last_error, last_error_user=_pick_last_error(tasks) | |
| _res=resource_snapshot() | |
| return {'tasks':task_list,'logs':events[-80:],'version':APP_VERSION,'last_error_user':last_error_user, | |
| 'max_accounts':MAX_ACCOUNTS if MAX_ACCOUNTS is not None else 'بلا حد', | |
| 'tiktok':tiktok_state,'check_interval':INTERVAL, | |
| 'source_stall_timeout':SOURCE_STALL_TIMEOUT, | |
| 'monitor_running':_monitor_alive.is_set(), | |
| 'active_relays':hub.active_count(),'last_monitor_cycle':_last_monitor_cycle, | |
| 'max_concurrent_relays':max_concurrent_relays(), | |
| **{k:_res.get(k) for k in ('load1','cpus','mem_total_mb','mem_avail_mb','quota_cpus','cpu_cores_used','cpu_pct_of_quota')}, | |
| 'last_error':last_error,'rtmp_configured':rtmp_configured(), | |
| 'oauth_configured':youtube.api.configured(), | |
| 'metadata_automation':_oauth_automation_on(), | |
| 'volume_ok':_ensure_root()} | |
| def add(a:Account): | |
| try:u=clean_user(a.username) | |
| except ValueError as e:raise HTTPException(400,str(e)) | |
| with lock: | |
| if MAX_ACCOUNTS is not None and u not in tasks and len(tasks)>=MAX_ACCOUNTS: | |
| raise HTTPException(400,'بلغت حد الموارد المحدد؛ يمكنك تغييره عبر MAX_ACCOUNTS') | |
| tasks.setdefault(u,_blank_task());tasks[u]['enabled']=True;save() | |
| if a.token and a.token.strip():set_token(u,a.token) | |
| log(u,'[MONITOR] @%s added'%u) | |
| return {'ok':True} | |
| 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])} | |
| def tiktok_token_state(): | |
| with lock:users=list(tasks.keys()) | |
| return tiktok_auth.describe(users) | |
| def tiktok_token_validate(b:TokenBody): | |
| raw=(b.token or '').strip() | |
| if not raw:raise HTTPException(400,'الصق التوكن أولًا') | |
| report=tiktok_auth.validate(raw) | |
| if report.get('ok'): | |
| 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 | |
| def tiktok_token_check(user:str): | |
| try:return tiktok_auth.unlock_check(user) | |
| except Exception as e: | |
| raise HTTPException(400,str(e) if isinstance(e,(ValueError,RuntimeError)) else 'تعذر اختبار فتح الحساب') | |
| def tiktok_token_delete(user:str): | |
| with lock: | |
| _tokens.pop(user,None);save_tokens() | |
| return {'ok':tiktok_auth.delete(user)} | |
| def action(user:str,action:str): | |
| do_stop_direct=False | |
| with lock: | |
| if user not in tasks:raise HTTPException(404) | |
| if action=='start': | |
| tasks[user]['enabled']=True;tasks[user]['state']='OFFLINE';tasks[user]['status']='بدء التشغيل' | |
| elif action=='stop': | |
| tasks[user]['enabled']=False | |
| tasks[user]['state']='STOPPING';tasks[user]['status']='متوقف' | |
| try: | |
| for key,ev in list(youtube.journal.read()['events'].items()): | |
| if ev.get('username')==user: | |
| youtube.event_update(key,closing=True) | |
| except Exception:pass | |
| do_stop_direct=True | |
| elif action=='delete': | |
| cleanup_entry(user) | |
| do_stop_direct=True | |
| else:raise HTTPException(400) | |
| save() | |
| if do_stop_direct: | |
| _stop_direct(user, action) | |
| return {'ok':True} | |
| def startup(): | |
| load_tokens() | |
| 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={} | |
| saved_items=list(saved.items()) | |
| if MAX_ACCOUNTS is not None:saved_items=saved_items[:MAX_ACCOUNTS] | |
| for u,enabled in saved_items: | |
| try:u=clean_user(u) | |
| except ValueError:continue | |
| tasks[u]=_blank_task();tasks[u]['enabled']=bool(enabled) | |
| if not enabled:tasks[u]['status']='متوقف' | |
| threading.Thread(target=monitor_loop,daemon=True,name='global-monitor').start() | |
| youtube.restore() | |
| log('system','[MONITOR] waiting') | |
| def shutdown(): | |
| _monitor_stop.set() | |
| for u in list(_direct_handles): | |
| _stop_direct(u, 'shutdown') | |
| youtube.stop() | |