"""TikTok Live Monitor → FFmpeg relay → YouTube RTMPS. No local recording."""
import os, re, json, time, threading, shutil, secrets, logging
from concurrent.futures import ThreadPoolExecutor, as_completed
from pathlib import Path
from urllib.parse import urlparse
from fastapi import FastAPI, Depends, HTTPException, Request
from fastapi.security import HTTPBasic, HTTPBasicCredentials
from fastapi.responses import HTMLResponse, JSONResponse
from pydantic import BaseModel
from curl_cffi import requests
import tiktok_auth
from ingest import hub, env_rtmp_destination, rtmp_configured, format_session_title, APP_VERSION, max_concurrent_relays

MSG=('تيك توك يحجب روابط البث لهذا الحساب (غرفة مقيدة بعمر 18+ أو جلسة تحتاج تحققًا)؛ '
     'الصق كوكيز الجلسة الكامل في لوحة «توكن تيك توك» ثم أعد المحاولة')

STORAGE_WAIT=max(0,float(os.getenv('STORAGE_WAIT_SECONDS','30')))
def _mounted(path):
    try:
        with open('/proc/mounts','rb') as fh:
            names={l.decode().split()[1] for l in fh.read().split(b'\n') if len(l.split())>=2}
        return path in names
    except OSError:return False

DATA_DIR=os.getenv('DATA_DIR','').strip()
if DATA_DIR:
    ROOT=Path(DATA_DIR).resolve()
else:
    ROOT=Path('/data/tiktok-relay')
    if Path('/data').is_dir():
        deadline=time.monotonic()+STORAGE_WAIT
        warned=False
        while not _mounted('/data') and time.monotonic()<deadline:
            if not warned:
                logging.warning('awaiting the /data storage bucket before choosing the data root')
                warned=True
            time.sleep(2)
        if not _mounted('/data'):
            logging.warning('/data is not a mount after %.0fs — state may not survive a reboot', STORAGE_WAIT)
        # Keep reading an older recorder folder if this Space already has one.
        legacy=Path('/data/tiktok-recorder')
        if (legacy/'youtube-api-state.json').exists() or (legacy/'tasks.json').exists():
            ROOT=legacy
    else:
        logging.warning('no /data mount — using ./storage; YouTube journal and tokens will be lost on reboot')
        ROOT=Path('./storage').resolve()
try:
    ROOT.mkdir(parents=True,exist_ok=True)
except OSError as e:
    logging.error('cannot create %s (%s) — starting read-only', ROOT, type(e).__name__)

def _ensure_root():
    try:
        ROOT.mkdir(parents=True,exist_ok=True)
        return True
    except OSError:
        return False

TOKENS=ROOT/'tokens.json'
tiktok_auth.configure(ROOT/'tiktok-tokens.json')
PASSWORD=os.getenv('PASSWORD','')
if not PASSWORD: raise RuntimeError('Set PASSWORD before starting the app')
FFMPEG=os.getenv('FFMPEG_PATH') or shutil.which('ffmpeg')
if not FFMPEG: raise RuntimeError('ffmpeg is required')
# ffprobe is optional: used only as a codec hint, never required to relay.
FFPROBE=os.getenv('FFPROBE_PATH') or shutil.which('ffprobe') or ''

INTERVAL=max(8,int(os.getenv('CHECK_INTERVAL','20')))
_LIMIT=int(os.getenv('MAX_ACCOUNTS','0') or 0)
MAX_ACCOUNTS=_LIMIT if _LIMIT>0 else None
SOURCE_STALL_TIMEOUT=max(0.0,float(os.getenv('SOURCE_STALL_TIMEOUT') or os.getenv('YOUTUBE_SOURCE_STALL_SECONDS','12')))
os.environ.setdefault('YOUTUBE_SOURCE_STALL_SECONDS',str(int(SOURCE_STALL_TIMEOUT) if SOURCE_STALL_TIMEOUT else 0))
os.environ.setdefault('YOUTUBE_SCAN_SECONDS',os.getenv('CHECK_INTERVAL','20'))

print('[BOOT] TikTok Live Monitor → Direct Relay → YouTube  version=%s CHECK_INTERVAL=%s SOURCE_STALL_TIMEOUT=%s rtmp=%s'%(APP_VERSION,INTERVAL,SOURCE_STALL_TIMEOUT,'yes' if rtmp_configured() else 'no'),flush=True)

lock=threading.RLock(); tasks={}; events=[]
security=HTTPBasic()
_monitor_alive=threading.Event()
_last_monitor_cycle=0.0
_monitor_stop=threading.Event()
_states=('OFFLINE','CHECKING','LIVE_DETECTED','STARTING','LIVE','STALLED','RECONNECTING','STOPPING','ERROR','BACKOFF')

def auth(c:HTTPBasicCredentials=Depends(security)):
    if not (secrets.compare_digest(c.username.encode(),b'admin') and secrets.compare_digest(c.password.encode(),PASSWORD.encode())):
        raise HTTPException(401,'Unauthorized',headers={'WWW-Authenticate':'Basic'})

app=FastAPI(docs_url=None,redoc_url=None,openapi_url=None)

@app.middleware('http')
async def _auth_mw(request:Request,call_next):
    # /health is public so the container/orchestrator can probe liveness without secrets.
    if request.url.path in ('/health','/healthz'):
        return await call_next(request)
    header=request.headers.get('authorization') or ''
    if header.lower().startswith('basic '):
        import base64
        try:
            raw=base64.b64decode(header.split(None,1)[1]).decode()
            user,sep,pw=raw.partition(':')
            if sep and secrets.compare_digest(user.encode(),b'admin') and secrets.compare_digest(pw.encode(),PASSWORD.encode()):
                return await call_next(request)
        except Exception:
            pass
    return JSONResponse({'detail':'Unauthorized'},status_code=401,headers={'WWW-Authenticate':'Basic'})

def log(user,msg):
    with lock:
        events.append({'time':time.strftime('%H:%M:%S'),'user':user,'message':msg})
        del events[:-100]

def update(user,**kw):
    with lock:
        if user in tasks: tasks[user].update(kw)

_failure_notes={}
def throttle_failure(user,msg,now=None):
    now=time.monotonic() if now is None else now
    st=_failure_notes.get(user)
    if st and st['msg']==msg and now-st['at']<120:
        st['suppressed']+=1;return None
    out=msg+((' (كُتمت %d رسالة مطابقة خلال آخر دقيقتين)'%st['suppressed']) if st and st['suppressed'] else '')
    _failure_notes[user]={'msg':msg,'at':now,'suppressed':0}
    return out

def save():
    with lock:
        tmp=ROOT/'tasks.tmp';tmp.write_text(json.dumps({u:t['enabled'] for u,t in tasks.items()}));tmp.replace(ROOT/'tasks.json')

def cleanup_entry(user):
    _stop_direct(user, 'delete')
    with lock:
        tasks.pop(user,None)
        _tokens.pop(user,None)
        try:save_tokens()
        except OSError:pass
    try:youtube.destinations.pop(user,None)
    except Exception:pass
    with _observation_guard:
        _observation_cache.pop(user,None)
        _observation_locks.pop(user,None)
    log(user,'[MONITOR] @%s removed'%user)
    return None

_tokens={}
def load_tokens():
    global _tokens
    try:_tokens=json.loads(TOKENS.read_text())
    except (FileNotFoundError,ValueError,OSError):_tokens={}
    for u,t in list(_tokens.items()):
        try:
            if t and t.strip():tiktok_auth.save(u,t)
        except (ValueError,OSError):pass
def save_tokens():
    with lock:
        tmp=ROOT/'tokens.tmp';tmp.write_text(json.dumps(_tokens));tmp.replace(TOKENS)
def set_token(user,token):
    token=(token or '').strip()
    with lock:
        if token:
            _tokens[user]=token
            try:tiktok_auth.save(user,token)
            except ValueError:pass
        else:
            _tokens.pop(user,None)
            tiktok_auth.delete(user)
        save_tokens()
def get_token(user):
    """Effective TikTok session: HF secrets beat the UI store (tiktok_auth.session_for)."""
    sess=tiktok_auth.session_for(user)
    return (sess.get('cookie') or sess.get('sessionid') or '') if sess else ''
def _cookie_headers(tok):
    if not tok:return {}
    sessionid,pairs=tiktok_auth.parse_token(tok)
    if not sessionid:return {}
    return {'Cookie':'; '.join('%s=%s'%(k,v) for k,v in pairs.items())}
def tiktok_headers(user):
    h=tiktok_auth.headers(user)
    return {'Cookie':h['Cookie']} if h.get('Cookie') else {}
def clean_user(raw):
    raw=raw.strip()
    if raw.startswith('https://'):
        p=urlparse(raw)
        if p.hostname not in ('www.tiktok.com','tiktok.com','vm.tiktok.com','vt.tiktok.com'):raise ValueError('الرابط يجب أن يكون من TikTok')
        m=re.fullmatch(r'/@([\w.]{1,32})(?:/live)?/?',p.path)
        if not m:raise ValueError('استخدم اسم الحساب أو رابطه الكامل، وليس رابط المشاركة المختصر')
        raw=m.group(1)
    raw=raw.lstrip('@')
    if not re.fullmatch(r'[A-Za-z0-9_.]{1,32}',raw):raise ValueError('اسم حساب غير صالح')
    return raw.lower()

def _tiktok_http_status(e):
    resp=getattr(e,'response',None)
    for cand in (getattr(resp,'status_code',None),getattr(e,'status_code',None)):
        try:cand=int(cand)
        except (TypeError,ValueError):continue
        if 100<=cand<600:return cand
    return 0

def _live_has_playable_sources(j):
    live=((j or {}).get('data') or {}).get('liveRoom') or {}
    if live.get('status') not in (2,'2'):
        return True
    blocks=[live[k] for k in ('streamData','hevcStreamData') if isinstance(live.get(k),dict)]
    try:
        return bool(rank_sources({'blocks':blocks}))
    except Exception:
        return False

def get_json(url,params,user=None):
    """Public lives first (no session). Global TikTok cookies only if the room is age-gated."""
    def _pull(headers):
        r=requests.get(url,params=params,impersonate='chrome',headers=headers,timeout=20)
        r.raise_for_status();return r.json()
    public={'Referer':'https://www.tiktok.com/'}
    try:
        j=_pull(public)
        if _live_has_playable_sources(j):
            return j
    except Exception as e:
        status=_tiktok_http_status(e)
        if status==404:
            raise RuntimeError('تيك توك لا يعرف هذا الحساب (404) — تحقق من اسم المستخدم') from None
        if status==429:
            raise RuntimeError('تيك توك حدّ الطلبات مؤقتًا (429)؛ سيُعاد المحاولة في الدورة التالية') from None
        if 500<=status<600:
            raise RuntimeError('تيك توك أعاد %s؛ سيُعاد المحاولة لاحقًا'%status) from None
        if status not in (0,401,403):
            raise RuntimeError('تعذر الوصول إلى تيك توك (%s%s)'%(type(e).__name__,'، HTTP '+str(status) if status else '')) from None
        j=None
    # Age-gated / empty public payload: one global session for every account.
    try:
        j=_pull(tiktok_auth.headers(user))
        return j
    except Exception as e:
        status=_tiktok_http_status(e)
        if status in (401,403):
            raise RuntimeError('تيك توك رفض الجلسة (%s) — الغرفة مقيدة (18+) والتوكن العام لا يغطيها؛ حدّث TIKTOK_SESSIONID'%status) from None
        if status==404:
            raise RuntimeError('تيك توك لا يعرف هذا الحساب (404) — تحقق من اسم المستخدم') from None
        if status==429:
            raise RuntimeError('تيك توك حدّ الطلبات مؤقتًا (429)؛ سيُعاد المحاولة في الدورة التالية') from None
        if 500<=status<600:
            raise RuntimeError('تيك توك أعاد %s؛ سيُعاد المحاولة لاحقًا'%status) from None
        raise RuntimeError('تعذر الوصول إلى تيك توك (%s%s)'%(type(e).__name__,'، HTTP '+str(status) if status else '')) from None

_observation_cache={}
_observation_locks={}
_observation_guard=threading.Lock()

def _as_dict(x):
    return x if isinstance(x, dict) else {}

def _offline_obs(room='', nickname='', live_title=''):
    rid=str(room or '')
    if not rid.isdigit() or rid=='0':rid=''
    return {'status':'offline','room_id':rid,'sources':[],'nickname':nickname or '','live_title':live_title or ''}

def observe_session(user,force=False):
    with _observation_guard:guard=_observation_locks.setdefault(user,threading.Lock())
    with guard:
        cached=_observation_cache.get(user)
        if cached and not force and time.monotonic()-cached[0]<15:return cached[1]
        result=_observe_session(user)
        _observation_cache[user]=(time.monotonic(),result)
        return result

def _observe_session(user):
    """TikTok room lookup. Null liveRoom/user/data must be Offline, not a crash.

    Production 67c9415: several accounts returned liveRoom=null. `.get` on None
    became AttributeError and the monitor stored the generic 'فشل طلب بيانات البث'
    even while one relay was healthy.
    """
    j=_as_dict(get_json('https://www.tiktok.com/api-live/user/room',{'aid':1988,'uniqueId':user,'sourceType':54,'staleTime':0},user))
    d=_as_dict(j.get('data')); live=_as_dict(d.get('liveRoom')); person=_as_dict(d.get('user'))
    status=live.get('status')
    room=str(person.get('roomId') or person.get('room_id') or live.get('roomId') or live.get('room_id') or '')
    nickname=str(person.get('nickname') or live.get('nickname') or '').strip()
    live_title=str(live.get('title') or '').strip()
    if status in (4,'4') or (not live and (not room or room=='0')):
        return _offline_obs(room, nickname, live_title)
    blocks=[live[k] for k in ('streamData','hevcStreamData') if isinstance(live.get(k),dict)]
    sources=rank_sources({'blocks':blocks}) if status in (2,'2') else []
    if status in (2,'2') and not sources and tiktok_auth.classify_restriction(j):
        raise RuntimeError(MSG)
    if not room or room=='0':
        try:
            r=requests.get(f'https://www.tiktok.com/@{user}/live',impersonate='chrome',headers={'Referer':'https://www.tiktok.com/',**tiktok_headers(user)},timeout=20)
            r.raise_for_status()
            m=re.search(r'\"roomId\"\s*:\s*\"(\d+)\"',r.text)
            room=m.group(1) if m else ''
        except Exception:
            room=room if str(room).isdigit() else ''
    if not room.isdigit() or room=='0':
        if status not in (2,'2'):
            return _offline_obs('', nickname, live_title)
        raise RuntimeError('تعذر تحديد معرّف جلسة البث؛ لن ننشئ حدث يوتيوب بالتخمين')
    if not sources:
        j=_as_dict(get_json('https://webcast.tiktok.com/webcast/room/info/',{'aid':1988,'room_id':room},user))
        info=_as_dict(j.get('data')); status=info.get('status'); owner=_as_dict(info.get('owner'))
        nickname=str(owner.get('nickname') or '').strip() or nickname
        live_title=str(info.get('title') or '').strip() or live_title
        if status in (4,'4'):return _offline_obs(room, nickname, live_title)
        if status not in (2,'2'):
            if tiktok_auth.classify_restriction(j):raise RuntimeError(MSG)
            raise RuntimeError('حالة تيك توك غير مؤكدة؛ لا نعتبرها انتهاء للبث')
        sources=rank_sources(_as_dict(info.get('stream_url')))
        if not sources and tiktok_auth.classify_restriction(j):raise RuntimeError(MSG)
    return {'status':'live','room_id':room,'sources':sources,'nickname':nickname,'live_title':live_title}

def extract(user):
    observed=observe_session(user)
    if observed['status']=='offline':return []
    if not observed['sources']:raise RuntimeError('البث مباشر لكن روابط الفيديو غير متاحة — '+MSG)
    return observed['sources']

def source_brief(item):
    """Human label with no URL/token: 'origin 1080p H264'."""
    q=str(item.get('quality_name') or (str(item.get('label') or '').split('·')[0]) or 'src').strip()
    q=re.sub(r'\s+','',q) or 'src'
    res=str(item.get('resolution') or '')
    nums=[int(n) for n in re.findall(r'\d+',res)]
    pretty=''
    if len(nums)>=2:
        h=max(nums[0],nums[1])
        pretty='%dp'%h if h>=144 else '%sx%s'%(nums[0],nums[1])
    elif nums:
        pretty='%dp'%nums[0]
    codec=str(item.get('codec') or '').strip().lower()
    if any(k in codec for k in ('h264','avc')):codec='H264'
    elif any(k in codec for k in ('h265','hevc','hvc1')):codec='HEVC'
    else:codec=codec.upper() or ''
    media=str(item.get('media') or '').upper()
    parts=[q]
    if pretty:parts.append(pretty)
    if codec:parts.append(codec)
    if media and media!='FLV':parts.append(media)
    return ' '.join(parts)

def rank_sources(source):
    """Every usable pull URL, ranked for *relay*, not for archive quality.

    Priority: (1) actually relayable, (2) YouTube-compatible codec, (3) quality,
    (4) sane bitrate. H264 is preferred over HEVC of the same session because
    YouTube ingest + stream-copy is H264; HEVC is kept as a fallback, never dropped.
    """
    candidates=[]
    priorities={'origin':100,'original':100,'origion':100,'uhd':90,'uhd_60':95,'hd_60':75,
                'full_hd1':90,'full_hd':90,'hd1':70,'hd':70,'hd5':65,'sd2':40,'sd':40,'sd1':30,'ld':20,'ld5':15}
    def add(url,label,media,params=None,assumed=False):
        if not isinstance(url,str) or not url.startswith('https://'):return
        if str(label).lower() in ('ao','audio','audio_only'):return
        params=params or {};res=str(params.get('resolution',''))
        nums=re.findall(r'\d+',res)
        width=int(nums[0]) if nums else 0
        height=int(nums[1]) if len(nums)>1 else 0
        pixels=width*height if height else (width*width if width else 0)
        try:rate=float(params.get('vbitrate') or params.get('bitrate') or 0)
        except (ValueError,TypeError):rate=0
        quality=priorities.get(str(label).lower(),50)
        codec=str(params.get('VCodec') or params.get('v_codec') or params.get('codec') or '')
        cl=codec.lower()
        h264=any(k in cl for k in ('h264','avc'))
        hevc=any(k in cl for k in ('h265','hevc','hvc1'))
        # FLV is the pipe we control; HLS is a fallback (FFmpeg fetches it itself).
        flv=1 if media=='flv' else 0
        # Compatible copy path first — never assume HEVC is "better" because it is sharper.
        compat=2 if h264 else (1 if not hevc else 0)
        # Extreme bitrates often stall FFmpeg/YouTube on small hosts.
        rate_score=rate if rate<=6500 else max(0,13000-rate)
        score=(flv, compat, quality, pixels, rate_score, not assumed)
        item={'url':url,'media':media,'codec':codec,'quality_name':str(label),
              'resolution':res,'width':width,'height':height,'bitrate':rate,
              'label':str(label)+((' · '+res) if res else '')+' · '+(codec or 'غير محدد')+((' · '+media.upper()) if media!='flv' else ''),
              'score':score}
        item['brief']=source_brief(item)
        candidates.append(item)
    blocks=list(source.get('blocks',[]))
    if source.get('live_core_sdk_data'):blocks.append(source['live_core_sdk_data'])
    for block in blocks:
        try:
            payload=block['pull_data']['stream_data']
            stream=json.loads(payload) if isinstance(payload,str) else payload
            for quality,item in stream.get('data',{}).items():
                main=item.get('main',{})
                try:
                    params=main.get('sdk_params') or {}
                    if isinstance(params,str):params=json.loads(params)
                except (ValueError,TypeError):params={}
                add(main.get('flv'),quality,'flv',params)
                add(main.get('hls'),quality,'hls',params)
        except (KeyError,TypeError,ValueError):pass
    for quality,url in (source.get('flv_pull_url') or {}).items():
        try:params=json.loads((source.get('flv_pull_url_params') or {}).get(quality) or '{}')
        except (ValueError,TypeError):params={}
        add(url,quality,'flv',params)
    for quality,url in (source.get('hls_pull_url') or {}).items():
        add(url,quality,'hls')
    add(source.get('rtmp_pull_url'),'غير مصنفة','flv',assumed=True)
    result=[];seen=set()
    for item in sorted(candidates,key=lambda c:c['score'],reverse=True):
        if item['url'] not in seen:result.append(item);seen.add(item['url'])
    return result

from youtube_manager import ManagedYouTube
from relay import Sender, SourceStalled, RelayStartError
from flv_bridge import FormatChanged
def enabled_accounts():
    with lock:return [u for u,t in tasks.items() if t.get('enabled')]
youtube=ManagedYouTube(FFMPEG,observe_session,log,enabled_accounts,ROOT/'youtube-api-state.json',headers_for=tiktok_headers)
if FFPROBE:
    youtube.ffprobe=FFPROBE

_direct_handles={}  # user -> {thread,stop,sender}
_direct_backoff={}
_direct_unstable={}
_direct_guard=threading.RLock()

def _oauth_automation_on():
    try:return bool(youtube.journal.read().get('enabled'))
    except Exception:return False

def _stop_direct(user, reason='stop'):
    with _direct_guard:
        h=_direct_handles.pop(user, None)
    if not h:return
    h['stop'].set()
    try:
        if h.get('sender'):h['sender'].terminate()
    except Exception:pass
    th=h.get('thread')
    if th and th.is_alive():
        th.join(timeout=4)
        if th.is_alive():
            try:h['sender'].terminate()
            except Exception:pass
    hub.release(user, 'direct')
    log(user, '[RELAY] @%s stopping (%s)'%(user, reason))
    log(user, '[CLEANUP] @%s complete'%user)

def _direct_loop(user, dest, stop):
    """TikTok → FFmpeg → RTMPS without requiring OAuth. One session at a time."""
    sender=Sender(FFMPEG, log, tiktok_headers)
    if FFPROBE:
        try:sender.set_probe(youtube.codec_probe)
        except Exception:pass
    hub.update(user, sender=sender, thread=threading.current_thread())
    with _direct_guard:
        if user in _direct_handles:_direct_handles[user]['sender']=sender
    room=''
    try:
        while not stop.is_set() and not _monitor_stop.is_set():
            try:
                obs=observe_session(user, force=True)
            except Exception as e:
                msg=str(e) if isinstance(e,RuntimeError) else type(e).__name__
                log(user,'[MONITOR] @%s observe failed during relay: %s'%(user, msg[:120]))
                if stop.wait(min(30, _direct_backoff.get(user, 2))):return
                continue
            if obs.get('status')!='live':
                log(user,'[MONITOR] @%s OFFLINE'%user)
                return
            if room and obs.get('room_id') and obs['room_id']!=room:
                log(user,'[MONITOR] @%s room changed; clean source transition'%user)
                _direct_unstable.pop(user, None)
            room=str(obs.get('room_id') or room)
            name=str(obs.get('nickname') or user)
            title, date, tm=format_session_title(name)
            hub.update(user, room_id=room, title=title)
            if _oauth_automation_on():
                pass
            else:
                log(user,'[RELAY] metadata automation unavailable — title would be %s; video continues via RTMPS'%title)
            sources=list(obs.get('sources') or [])
            bad=_direct_unstable.get(user) or set()
            ordered=[c for c in sources if (c.get('brief') or c.get('label')) not in bad]+[c for c in sources if (c.get('brief') or c.get('label')) in bad]
            if not ordered:
                log(user,'[SOURCE] @%s no sources; waiting'%user)
                if stop.wait(3):return
                continue
            log(user,'[SOURCE] @%s selecting source'%user)
            sent=False
            for cand in ordered:
                if stop.is_set():return
                brief=cand.get('brief') or cand.get('label') or 'src'
                log(user,'[SOURCE] @%s selected %s'%(user, brief))
                tagged={**cand,'user':user}
                try:
                    log(user,'[RELAY] @%s starting'%user)
                    update(user, state='LIVE', source=brief, codec=cand.get('codec') or '', quality=cand.get('resolution') or '')
                    ok=sender.send(user, tagged, dest, stop)
                    sent=bool(ok)
                    if sent:break
                except RelayStartError as e:
                    log(user,'[SOURCE] @%s source %s unstable (start failed)'%(user, brief))
                    _direct_unstable.setdefault(user,set()).add(brief)
                    log(user,'[SOURCE] @%s trying fallback'%user)
                    continue
                except SourceStalled as e:
                    log(user,'[WATCHDOG] @%s data stalled'%user)
                    _direct_unstable.setdefault(user,set()).add(brief)
                    update(user, state='STALLED', last_error='source stalled')
                    log(user,'[SOURCE] @%s trying fallback'%user)
                    continue
                except FormatChanged:
                    log(user,'[SOURCE] @%s format changed; reconnect'%user)
                    sent=True
                    break
                except Exception as e:
                    log(user,'[RELAY] @%s error %s'%(user, type(e).__name__))
                    continue
            wait=2 if sent else min(60, max(2, (_direct_backoff.get(user,2))*2))
            _direct_backoff[user]=2 if sent else wait
            if stop.wait(wait):return
    finally:
        try:sender.terminate()
        except Exception:pass
        hub.release(user,'direct')
        with _direct_guard:_direct_handles.pop(user, None)
        log(user,'[MONITOR] @%s waiting'%user)

def _maybe_start_direct(user, obs):
    """Start RTMPS relay when OAuth automation is off but an ingest destination exists."""
    if _oauth_automation_on():
        return
    dest=env_rtmp_destination()
    if not dest:
        update(user, youtube_status='no destination (set YOUTUBE_RTMP_URL + YOUTUBE_STREAM_KEY or enable OAuth)')
        return
    if hub.busy(user):
        return
    # One env stream key = one publisher. Do not collide.
    if hub.env_key_busy():
        update(user, state='BACKOFF', status='وجهة RTMPS مشغولة بمذيع آخر (مفتاح واحد)')
        log(user,'[RELAY] @%s waiting: single env stream key already in use'%user)
        return
    if hub.at_capacity(except_user=user):
        update(user, state='BACKOFF', status='بلغنا حد MAX_CONCURRENT_RELAYS')
        log(user,'[RELAY] @%s waiting: at MAX_CONCURRENT_RELAYS=%d'%(user, max_concurrent_relays()))
        return
    stop=threading.Event()
    title=format_session_title(obs.get('nickname') or user)[0]
    if not hub.occupy(user,'direct', room_id=obs.get('room_id') or '', title=title):
        return
    th=threading.Thread(target=_direct_loop, args=(user, dest, stop), daemon=True, name='direct-'+user)
    hub.update(user, thread=th, stop=stop)
    with _direct_guard:
        _direct_handles[user]={'thread':th,'stop':stop,'sender':None}
    log(user,'[RELAY] @%s starting (env RTMPS, OAuth not required)'%user)
    update(user, state='STARTING', status='بدء Relay')
    th.start()

class Account(BaseModel):
    username:str
    token:str=''
class TokenBody(BaseModel):
    token:str=''
class YouTubeStart(BaseModel):
    channel_id:str
class YouTubeSettings(BaseModel):
    name:str=''
    privacy:str
    title:str
    description:str
    send_mode:str='auto'

def _blank_task():
    return {'enabled':True,'status':'بدء التشغيل','state':'OFFLINE','quality':'','source':'',
            'codec':'','live_since':None,'room_id':'','last_error':'','nickname':''}

def _set_state(user,state,status, **extra):
    kw={'state':state,'status':status}
    kw.update(extra)
    update(user,**kw)

def _relay_snapshot(user):
    slot=hub.snapshot(user)
    sender=slot.get('sender') if slot else None
    snap=sender.snapshot() if sender and hasattr(sender,'snapshot') else {}
    pid=None
    try:
        proc=getattr(sender,'process',None)
        pid=int(proc.pid) if proc and proc.poll() is None else None
    except Exception:
        pid=None
    return {
        'relay_owner':slot.get('owner') if slot else '',
        'relay_pid':pid,
        'last_data_time':snap.get('last_source_at'),
        'bytes_received':snap.get('source_bytes') or 0,
        'source_age':snap.get('source_age'),
        'output_seconds':snap.get('output_seconds') or 0,
        'relay_alive':bool(pid),
        'session_title':slot.get('title') if slot else '',
        'relay_dest':slot.get('dest') if slot else '',
        'restarts':hub.restart_count(user),
        'send_mode':snap.get('send_mode') or '',
    }

def _cgroup_usage_ns():
    try:
        with open('/sys/fs/cgroup/cpu.stat') as f:
            for line in f:
                if line.startswith('usage_usec'):
                    return int(line.split()[1])*1000
    except Exception:
        pass
    for cand in ('/sys/fs/cgroup/cpuacct/cpuacct.usage','/sys/fs/cgroup/cpu,cpuacct/cpuacct.usage'):
        try:
            return int(open(cand).read().strip())
        except Exception:
            continue
    return None

def _cgroup_quota_cpus():
    try:
        parts=open('/sys/fs/cgroup/cpu.max').read().split()
        if not parts:
            return None
        if parts[0]=='max':
            return float(os.cpu_count() or 1)
        return float(parts[0])/float(parts[1])
    except Exception:
        pass
    try:
        q=int(open('/sys/fs/cgroup/cpu/cpu.cfs_quota_us').read())
        p=int(open('/sys/fs/cgroup/cpu/cpu.cfs_period_us').read())
        if q>0 and p>0:
            return q/p
    except Exception:
        pass
    return None

def _cgroup_mem():
    """حدّ الذاكرة والمستهلك منها **لهذه الحاوية** بالميغابايت (لا للخادم المشترك).

    /proc/meminfo يصف الآلة المضيفة كاملة (16 نواة و126 غيغا) بينما لوحة Hugging
    Face تعرض حصّة الـSpace نفسها (2 vCPU و16 غيغا) — فكان الرقمان يصفان شيئين
    مختلفين. cgroup v2 يعطي ما تراه الحاوية فعلًا، فيتطابق الرقمان.
    """
    def _read(path):
        try:
            return open(path).read().strip()
        except Exception:
            return ''
    limit=(_read('/sys/fs/cgroup/memory.max') or _read('/sys/fs/cgroup/memory.high')
           or _read('/sys/fs/cgroup/memory/memory.limit_in_bytes'))
    if limit=='max':
        limit=_read('/sys/fs/cgroup/memory.high')
    used=_read('/sys/fs/cgroup/memory.current') or _read('/sys/fs/cgroup/memory/memory.usage_in_bytes')
    try:
        used_mb=int(used)//1048576 if used else None
    except ValueError:
        used_mb=None
    try:
        limit_mb=int(limit)//1048576 if limit and limit!='max' else None
    except ValueError:
        limit_mb=None
    # قيمة خيالية (أكبر من ذاكرة المضيف) تعني «بلا حدّ» على هذا النظام.
    if limit_mb and limit_mb>1024*1024:
        limit_mb=None
    return limit_mb,used_mb


_cpu_lock=threading.Lock()
_cpu_sample={'t':None,'ns':None}
# قراءة واحدة يتشاركها كل القرّاء خلال هذه النافذة. السبب: العيّنة ذات حالة
# (فرقٌ عن الاستدعاء السابق)، وكان /health و/api/state يصفّر كلٌّ منهما أساس
# الآخر ⇒ رقمان مختلفان للحظة واحدة، أو «—» إذا قلّ الفاصل عن 0.4ث.
_RES_TTL=2.0
_res_cache={'t':None,'data':None}

def resource_snapshot():
    """حصّة هذا الـSpace من المعالج والذاكرة (cgroup)، لا عتاد الخادم المشترك.

    القفل يغطّي الدالة كاملة: استدعاءان متزامنان يعيدان نفس القراءة بدل أن
    يُفسد كلٌّ منهما عيّنة الآخر.
    """
    with _cpu_lock:
        now=time.monotonic()
        cached=_res_cache['data']
        if cached is not None and _res_cache['t'] is not None and now-_res_cache['t']<_RES_TTL:
            return dict(cached)
        out={'load1':None,'cpus':os.cpu_count() or 1,'mem_total_mb':None,'mem_avail_mb':None,
             'quota_cpus':None,'cpu_cores_used':None,'cpu_pct_of_quota':None}
        try:
            out['load1']=float(open('/proc/loadavg').read().split()[0])
        except Exception:
            pass
        lim,used=_cgroup_mem()
        if lim:
            out['mem_total_mb']=lim
            out['mem_avail_mb']=max(0,lim-(used or 0))
        else:
            try:
                info={}
                for line in open('/proc/meminfo'):
                    parts=line.split()
                    if len(parts)>=2:info[parts[0].rstrip(':')]=int(parts[1])
                out['mem_total_mb']=info.get('MemTotal',0)//1024
                out['mem_avail_mb']=info.get('MemAvailable', info.get('MemFree',0))//1024
            except Exception:
                pass
        out['quota_cpus']=_cgroup_quota_cpus()
        if out['quota_cpus']:
            # «عدد الأنوية» في اللوحة هو حصّة الـSpace لا أنوية الخادم المشترك.
            out['cpus']=out['quota_cpus']
        ns=_cgroup_usage_ns()
        prev=_cpu_sample
        if prev['t'] is not None and prev['ns'] is not None and ns is not None and now-prev['t']>=0.4:
            cores=(ns-prev['ns'])/1e9/(now-prev['t'])
            if cores<0: cores=0.0
            out['cpu_cores_used']=round(cores,4)
            q=out['quota_cpus']
            if q and q>0:
                out['cpu_pct_of_quota']=round(min(999.0, max(0.0, 100.0*cores/q)),2)
        elif cached is not None:
            # نافذة أقصر من 0.4ث: أعد آخر نسبة معروفة بدل «—».
            out['cpu_cores_used']=cached.get('cpu_cores_used')
            out['cpu_pct_of_quota']=cached.get('cpu_pct_of_quota')
        _cpu_sample['t']=now
        _cpu_sample['ns']=ns
        _res_cache['t']=now
        _res_cache['data']=dict(out)
        return dict(out)

def _check_account(user):
    """One monitor cycle. Starts DirectRelay when OAuth automation is off."""
    with lock:
        tsk=tasks.get(user)
        if not tsk or not tsk.get('enabled'):return
    relaying=hub.busy(user)
    if not relaying:
        _set_state(user,'CHECKING','فحص البث')
    try:
        obs=observe_session(user, force=not relaying)
    except Exception as e:
        msg=str(e) if isinstance(e,RuntimeError) else 'فشل طلب بيانات البث (%s)'%type(e).__name__
        if relaying:
            out=throttle_failure(user, msg)
            if out:log(user,'[MONITOR] @%s observe failed; relay continues — %s'%(user,out[:120]))
            return
        _set_state(user,'ERROR' if 'رفض' in msg else 'BACKOFF', msg, last_error=msg)
        out=throttle_failure(user,msg)
        if out:log(user,'[MONITOR] @%s %s'%(user,out))
        return
    if obs.get('status')!='live':
        prev=None
        with lock:prev=(tasks.get(user) or {}).get('state')
        if relaying:
            # small confirmation before killing a live relay
            try:
                again=observe_session(user, force=True)
            except Exception:
                again=obs
            if again.get('status')=='live':
                obs=again
            else:
                _stop_direct(user, 'tiktok offline')
                _set_state(user,'OFFLINE','Offline',quality='',source='',codec='',live_since=None,
                           room_id=obs.get('room_id') or '', last_error='')
                log(user,'[MONITOR] @%s OFFLINE'%user)
                log(user,'[MONITOR] @%s waiting'%user)
                return
        else:
            _set_state(user,'OFFLINE','Offline',quality='',source='',codec='',live_since=None,
                       room_id=obs.get('room_id') or '', last_error='')
            if prev!='OFFLINE':
                log(user,'[MONITOR] @%s OFFLINE'%user)
            return
    src=obs.get('sources') or []
    top=src[0] if src else None
    brief=top.get('brief') if top else ''
    codec=(top.get('codec') if top else '') or ''
    quality=(top.get('resolution') if top else '') or ''
    live_since=None
    with lock:
        cur=tasks.get(user) or {}
        live_since=cur.get('live_since') or time.time()
        prev_state=cur.get('state')
        prev_room=cur.get('room_id')
    extra={'quality':quality,'source':brief,'codec':codec,'live_since':live_since,
           'room_id':obs.get('room_id') or '','nickname':obs.get('nickname') or '','last_error':''}
    extra.update(_relay_snapshot(user))
    if prev_state!='LIVE' or (prev_room and prev_room!=obs.get('room_id')):
        log(user,'[MONITOR] @%s LIVE detected'%user)
        if top:log(user,'[SOURCE] @%s selected %s'%(user,brief))
        if prev_room and prev_room!=obs.get('room_id') and relaying:
            log(user,'[MONITOR] @%s session/room changed; source will be refreshed'%user)
    _set_state(user,'LIVE','LIVE',**extra)
    _maybe_start_direct(user, obs)
    if _oauth_automation_on() and not hub.busy(user) and (obs.get('sources') or []):
        try:
            youtube.nudge_live(user, obs)
        except Exception:
            log(user,'[RELAY] @%s youtube nudge failed'%user)

def monitor_loop():
    """Global monitor: every enabled account in parallel. One fault never blocks the rest."""
    global _last_monitor_cycle
    while not _monitor_stop.is_set():
        _monitor_alive.set()
        started=time.monotonic()
        try:
            with lock:users=[u for u,t in tasks.items() if t.get('enabled')]
            if users:
                workers=min(8,len(users))
                with ThreadPoolExecutor(max_workers=workers,thread_name_prefix='mon') as pool:
                    futs=[pool.submit(_check_account,u) for u in users]
                    for f in as_completed(futs,timeout=max(15,INTERVAL*2)):
                        try:f.result()
                        except Exception as e:
                            log('system','[MONITOR] worker fault: %s'%type(e).__name__)
            _last_monitor_cycle=time.time()
        except Exception as e:
            log('system','[MONITOR] cycle error: %s'%type(e).__name__)
        # Keep YouTube manager alive independently of this loop.
        try:
            d=youtube.journal.read()
            if (d.get('enabled') or d.get('events')) and (not youtube.manager_thread or not youtube.manager_thread.is_alive()):
                log('system','[MONITOR] YouTube manager thread died; restarting')
                youtube.launch()
        except Exception:
            pass
        wait=max(1.0,INTERVAL-(time.monotonic()-started))
        _monitor_stop.wait(wait)
    _monitor_alive.clear()

@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():
    with lock:
        n=len(tasks);enabled=sum(1 for t in tasks.values() if t.get('enabled'))
        last_error, last_error_user=_pick_last_error(tasks)
        users=list(tasks.items())
    active=hub.active_count()
    yt_health={}
    try:
        for ev in (youtube.snapshot().get('events') or []):
            if ev.get('username'):
                yt_health[ev['username']]=ev
    except Exception:
        yt_health={}
    per=[]
    for u,tsk in users:
        rs=_relay_snapshot(u)
        ye=yt_health.get(u) or {}
        per.append({'username':u,'status':tsk.get('state') or '',
                    'last_data_time':rs.get('last_data_time'),
                    'bytes_received':rs.get('bytes_received') or 0,
                    'relay_pid':rs.get('relay_pid'),'relay_alive':bool(rs.get('relay_alive')),
                    'room_id':tsk.get('room_id') or '','source':tsk.get('source') or '',
                    'restarts':rs.get('restarts') or 0,'dest':rs.get('relay_dest') or '',
                    'last_error':tsk.get('last_error') or '','source_age':rs.get('source_age'),
                    'output_seconds':rs.get('output_seconds') or 0,
                    'send_mode':rs.get('send_mode') or '',
                    'youtube_health':str(ye.get('youtube_health') or ''),
                    'youtube_issues':str(ye.get('youtube_issues') or '')[:120],
                    'youtube_status':str(ye.get('youtube_status') or '')})
    res=resource_snapshot()
    mgr=bool(youtube.manager_thread and youtube.manager_thread.is_alive())
    y_events=0; y_hist=0; y_senders=0; y_status=''; y_skip=''
    try:
        jd=youtube.journal.read()
        y_events=len(jd.get('events') or {})
        y_hist=len(jd.get('history') or [])
        y_status=str(getattr(youtube,'status_text','') or '')[:160]
        y_senders=sum(1 for e in (youtube.senders or {}).values() if e.get('thread') and e['thread'].is_alive())
        skip=getattr(youtube,'_skip_last',None) or {}
        if skip:
            y_skip='; '.join('%s: %s'%(u,str(t)[:80]) for u,t in list(skip.items())[-3:])[:200]
        try:
            acc=list(youtube.accounts() or [])
            y_skip=(y_skip+(' | accounts='+','.join(acc)[:80])).strip(' |')
        except Exception:
            pass
    except Exception:
        pass
    return {'application':'running','monitor':'running' if _monitor_alive.is_set() else 'stopped',
            'accounts':n,'enabled':enabled,'active_relays':active,
            'max_concurrent_relays':max_concurrent_relays(),
            'source_stall_timeout':SOURCE_STALL_TIMEOUT,
            'load1':res.get('load1'),'cpus':res.get('cpus'),
            'mem_total_mb':res.get('mem_total_mb'),'mem_avail_mb':res.get('mem_avail_mb'),
            'quota_cpus':res.get('quota_cpus'),'cpu_cores_used':res.get('cpu_cores_used'),
            'cpu_pct_of_quota':res.get('cpu_pct_of_quota'),
            'last_monitor_cycle':_last_monitor_cycle,'last_error':last_error,
            'last_error_user':last_error_user,
            'version':APP_VERSION,'rtmp_configured':rtmp_configured(),
            'metadata_automation':_oauth_automation_on(),
            'youtube_manager':'running' if mgr else 'stopped',
            'youtube_status':y_status,'youtube_events':y_events,'youtube_history':y_hist,
            'youtube_senders':y_senders,'youtube_skip':y_skip,
            'broadcasters':per}

@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,'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.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/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'),
                '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()