| """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, HTTPException, Request |
| 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) |
| |
| 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__) |
| os.environ.setdefault('STREAM_EVENTS_PATH',str(ROOT/'logs'/'stream_events.json')) |
|
|
| 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=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=[] |
| _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') |
|
|
| app=FastAPI(docs_url=None,redoc_url=None,openapi_url=None) |
|
|
| @app.middleware('http') |
| async def _auth_mw(request:Request,call_next): |
| |
| |
| |
| public=request.url.path in ('/','/health','/healthz') |
| allowed=public |
| header=request.headers.get('authorization') or '' |
| if not allowed and header.lower().startswith('basic '): |
| import base64 |
| try: |
| raw=base64.b64decode(header.split(None,1)[1],validate=True).decode() |
| user,sep,pw=raw.partition(':') |
| allowed=bool(sep and secrets.compare_digest(user.encode(),b'admin') and |
| secrets.compare_digest(pw.encode(),PASSWORD.encode())) |
| except Exception: |
| allowed=False |
| if allowed: |
| response=await call_next(request) |
| else: |
| |
| |
| response=JSONResponse({'detail':'Unauthorized'},status_code=401) |
| |
| response.headers['Cache-Control']='no-store' |
| response.headers['X-Content-Type-Options']='nosniff' |
| response.headers['Referrer-Policy']='no-referrer' |
| response.headers['Permissions-Policy']='camera=(), microphone=(), geolocation=()' |
| response.headers['Content-Security-Policy']=( |
| "default-src 'self'; img-src 'self' data:; style-src 'self' 'unsafe-inline'; " |
| "script-src 'self' 'unsafe-inline'; connect-src 'self'; object-src 'none'; " |
| "base-uri 'none'; form-action 'self'; frame-ancestors 'self' https://huggingface.co https://*.huggingface.co") |
| return response |
|
|
| def log(user,msg): |
| text=str(msg or '').replace('\r',' ').replace('\n',' ') |
| with lock: |
| events.append({'time':time.strftime('%H:%M:%S'),'user':user,'message':text}) |
| del events[:-100] |
| |
| |
| |
| stdout_mode=str(os.getenv('LOG_EVENTS_STDOUT','important')).strip().lower() |
| important=any(tag in text for tag in ('[WATCHDOG]','[RESTART]','[CLEANUP]','[OFFLINE]','[WARN]')) or \ |
| any(word in text.lower() for word in ('error','failed','stalled','dead','died','quota','traceback')) or \ |
| any(word in text for word in ('فشل','تعذر','رفض','مات','انتهت مهلة')) or \ |
| ('[RELAY]' in text and any(word in text for word in (' starting','بدء FFmpeg',' stopping'))) |
| if stdout_mode not in ('0','false','off','no') and (stdout_mode in ('1','all','true','on','yes') or important): |
| safe=re.sub(r'\b(?:rtmps?|rtsp|https?)://\S+','«رابط-مخفي»',text) |
| safe=re.sub(r'(?i)(sessionid|token|key|auth)\s*[=:]\s*[^\s;&]+',r'\1=«مخفي»',safe) |
| print('[EVENT] @%s %s'%(str(user or 'system')[:48],safe[:700]),flush=True) |
|
|
| 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)) |
| |
| |
| try:tmp.chmod(0o600) |
| except OSError:pass |
| tmp.replace(TOKENS) |
| try:TOKENS.chmod(0o600) |
| except OSError:pass |
| 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 |
| |
| 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=1 if media=='flv' else 0 |
| |
| compat=2 if h264 else (1 if not hevc else 0) |
| |
| 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) |
| hls=source.get('hls_pull_url') or source.get('hls_pull_url_map') or {} |
| if isinstance(hls,dict): |
| for quality,url in hls.items():add(url,quality,'hls') |
| elif isinstance(hls,str): |
| |
| add(hls,'origin','hls',assumed=True) |
| 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 stream_monitor import read_events |
| from telemetry import read_records as telemetry_records, status as telemetry_status |
| 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={} |
| _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 '') |
| sender.telemetry_context={'room_id':room,'path':'direct'} |
| 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 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 |
| |
| 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} |
| |
| |
| |
| _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']: |
| |
| 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: |
| |
| 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: |
| |
| 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__) |
| |
| 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() |
|
|
| @app.get('/',response_class=HTMLResponse) |
| 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 |
|
|
| @app.get('/health') |
| @app.get('/healthz') |
| def health_public(): |
| |
| |
| return {'status':'ok','application':'running', |
| 'monitor':'running' if _monitor_alive.is_set() else 'starting', |
| 'version':APP_VERSION} |
|
|
| @app.get('/api/health') |
| 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,'telemetry':telemetry_status(),'rtmp_configured':rtmp_configured(), |
| 'oauth_configured':youtube.api.configured(),'metadata_automation':_oauth_automation_on(), |
| 'broadcasters':per} |
|
|
| @app.get('/api/youtube') |
| def youtube_state():return youtube.snapshot() |
| @app.get('/api/stream-events') |
| def stream_events(limit:int=200,broadcaster:str=''): |
| rows=read_events(limit,broadcaster.strip().lstrip('@') or None) |
| return {'count':len(rows),'events':rows,'file':'logs/stream_events.json'} |
| @app.get('/api/telemetry') |
| def telemetry_data(limit:int=200,broadcaster:str=''): |
| rows=telemetry_records(limit,broadcaster.strip().lstrip('@') or None) |
| return {'count':len(rows),'records':rows,'file':'telemetry/telemetry.jsonl', |
| 'status':telemetry_status(),'retention':'manual_only_no_auto_delete'} |
| @app.post('/api/youtube/check') |
| 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) |
| @app.post('/api/youtube/channel') |
| def youtube_confirm_channel(a:YouTubeStart): |
| |
| try:return {'ok':True,'channel':youtube.confirm_channel(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) |
| @app.post('/api/youtube/settings') |
| def youtube_settings(a:YouTubeSettings): |
| try:youtube.settings(a.model_dump()) |
| except ValueError as e:raise HTTPException(400,str(e)) |
| return {'ok':True} |
| @app.post('/api/youtube/start') |
| 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} |
| @app.post('/api/youtube/stop') |
| def youtube_stop(): |
| youtube.disable();return {'ok':True} |
|
|
| @app.get('/api/state') |
| 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'),'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 ('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()} |
|
|
| @app.post('/api/accounts') |
| 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} |
| @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(): |
| with lock:users=list(tasks.keys()) |
| return tiktok_auth.describe(users) |
|
|
| @app.post('/api/tiktok/token/validate') |
| 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 |
|
|
| @app.post('/api/tiktok/token/check/{user}') |
| 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 'تعذر اختبار فتح الحساب') |
|
|
| @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): |
| 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} |
|
|
| @app.on_event('startup') |
| 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') |
|
|
| @app.on_event('shutdown') |
| def shutdown(): |
| _monitor_stop.set() |
| for u in list(_direct_handles): |
| _stop_direct(u, 'shutdown') |
| youtube.stop() |
|
|
