Spaces:

Reoo88
/

Prayer-Time-Bot

Running

Exact copy of Reoo88/On @ 826bfe0 (relay-v2.32-gap-filler)
ce0ef24 verified 26 days ago
52.2 kB
"""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)
        # 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__)
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 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=[]
_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):
    # The HTML shell is public so an embedded Space can show its own login form. Browsers
    # commonly suppress HTTP-Basic prompts in cross-origin iframes; every data/mutation API
    # remains protected. /health stays minimal and public for the platform probe.
    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:
        # No WWW-Authenticate challenge here: browsers may suppress or mishandle native Basic
        # prompts inside the Hugging Face iframe. The in-page form sends the same Basic header.
        response=JSONResponse({'detail':'Unauthorized'},status_code=401)
    # The panel carries credentials and operational data: never cache it in an intermediary.
    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]
    # Container logs used to contain only 5-second dashboard access lines, so a production
    # noData incident had no diagnostic trace.  Emit the already operator-facing event stream,
    # with a final defensive redaction for URLs and credential-shaped query fields.
    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))
        # This is a legacy compatibility store containing live TikTok cookies.  Keep it
        # owner-only just like tiktok_auth.py's primary store, including after replace().
        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
    # 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)
    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):
        # webcast/room/info currently returns one signed HLS URL as a string, not a map.
        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={}  # 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 '')
                    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
    # 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()

@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():
    # Liveness is intentionally non-secret.  Detailed usernames, room IDs, errors and
    # YouTube state belong to the authenticated /api/health and /api/state endpoints.
    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):
    # Explicit channel adoption is separate from starting any public broadcast.
    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()