Guest detection via PirateTok host-room events (GUEST_WATCH_HOSTS)

- guest_watch.py: keyless WebSocket watcher on host rooms (piratetok-live-py 0.3.0).
A guest is active while its userId keeps appearing in link_mic_method events.
- app.py: when an enabled account is offline in its own room but is a guest on a
watched host, relay the host stream and show LIVE on host. Switch back when the
guest signal expires. Own-room behaviour is unchanged otherwise.
- Dockerfile: GUEST_WATCH_HOSTS=talolil, GUEST_ACTIVE_SECONDS=120.
- test_guest_watch.py: unit checks.

Dockerfile

@@ -5,8 +5,8 @@ RUN ffmpeg -hide_banner -h protocol=rtmps 2>&1 | grep -q rtmp_enhanced_codecs
 WORKDIR /app
 COPY requirements.txt .
 RUN pip install --no-cache-dir -U -r requirements.txt
-COPY app.py relay.py youtube_api.py youtube_manager.py flv_bridge.py stream_monitor.py tiktok_auth.py ingest.py index.html ./
-COPY verify.sh verify_tokens.py probe_live.py test_watchdog.py test_lifecycle.py test_relay.py test_source_reconnect.py test_ingest_routes.py test_transition_ingest.py test_concurrency.py test_flv_bridge.py test_adaptive_format.py test_regressions.py test_completion.py test_channel_switch.py test_dashboard.js ./
+COPY app.py relay.py youtube_api.py youtube_manager.py flv_bridge.py stream_monitor.py tiktok_auth.py ingest.py guest_watch.py index.html ./
+COPY verify.sh verify_tokens.py probe_live.py test_guest_watch.py test_watchdog.py test_lifecycle.py test_relay.py test_source_reconnect.py test_ingest_routes.py test_transition_ingest.py test_concurrency.py test_flv_bridge.py test_adaptive_format.py test_regressions.py test_completion.py test_channel_switch.py test_dashboard.js ./
 
 # TikTok Live Monitor → Direct Relay → YouTube. No local video files.
 ENV TZ=Asia/Baghdad \
@@ -60,7 +60,9 @@ ENV TZ=Asia/Baghdad \
     HLS_START_INDEX=-3 \
     PORT=7860 \
     PYTHONUNBUFFERED=1 \
-    YOUTUBE_VIDEO_MODE=original
+    YOUTUBE_VIDEO_MODE=original \
+    GUEST_WATCH_HOSTS=talolil \
+    GUEST_ACTIVE_SECONDS=120
 
 EXPOSE 7860
 CMD ["sh", "-c", "exec uvicorn app:app --host 0.0.0.0 --port ${PORT:-7860} --timeout-graceful-shutdown 65 --no-access-log"]

app.py

@@ -9,6 +9,7 @@ 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
+from guest_watch import GuestWatch, parse_hosts
 
 MSG=('تيك توك يحجب روابط البث لهذا الحساب (غرفة مقيدة بعمر 18+ أو جلسة تحتاج تحققًا)؛ '
      'الصق كوكيز الجلسة الكامل في لوحة «توكن تيك توك» ثم أعد المحاولة')
@@ -454,6 +455,21 @@ _direct_backoff={}
 _direct_unstable={}
 _direct_guard=threading.RLock()
 
+GUEST=GuestWatch(parse_hosts(os.getenv('GUEST_WATCH_HOSTS','')), log=log)
+
+def _guest_source(user, own):
+    """(obs, host) when @user is a guest on a watched host and that host is live; else None."""
+    if not GUEST.running:return None
+    host=GUEST.active_host(user)
+    if not host:return None
+    try:hobs=observe_session(host, force=True)
+    except Exception as e:
+        log(user,'[GUEST] @%s host observe failed: %s'%(user,type(e).__name__))
+        return None
+    if hobs.get('status')!='live' or not hobs.get('sources'):return None
+    nick=str((own or {}).get('nickname') or hobs.get('nickname') or user)
+    return {**hobs,'nickname':nick}, host
+
 def _oauth_automation_on():
     try:return bool(youtube.journal.read().get('enabled'))
     except Exception:return False
@@ -495,9 +511,15 @@ def _direct_loop(user, dest, stop):
                 log(user,'[MONITOR] @%s observe failed during relay: %s'%(user, msg[:120]))
                 if stop.wait(min(30, _direct_backoff.get(user, 2))):return
                 continue
+            src_host=''
             if obs.get('status')!='live':
-                log(user,'[MONITOR] @%s OFFLINE'%user)
-                return
+                gs=_guest_source(user, obs)
+                if not gs:
+                    log(user,'[MONITOR] @%s OFFLINE'%user)
+                    return
+                obs,src_host=gs
+            with _direct_guard:
+                if user in _direct_handles:_direct_handles[user]['src']=src_host
             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)
@@ -553,7 +575,7 @@ def _direct_loop(user, dest, stop):
         with _direct_guard:_direct_handles.pop(user, None)
         log(user,'[MONITOR] @%s waiting'%user)
 
-def _maybe_start_direct(user, obs):
+def _maybe_start_direct(user, obs, src_host=''):
     """Start RTMPS relay when OAuth automation is off but an ingest destination exists."""
     if _oauth_automation_on():
         return
@@ -579,7 +601,7 @@ def _maybe_start_direct(user, obs):
     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}
+        _direct_handles[user]={'thread':th,'stop':stop,'sender':None,'src':src_host}
     log(user,'[RELAY] @%s starting (env RTMPS, OAuth not required)'%user)
     update(user, state='STARTING', status='بدء Relay')
     th.start()
@@ -778,6 +800,11 @@ def _check_account(user):
         out=throttle_failure(user,msg)
         if out:log(user,'[MONITOR] @%s %s'%(user,out))
         return
+    src_host=''
+    if obs.get('status')!='live':
+        gs=_guest_source(user, obs)
+        if gs:
+            obs,src_host=gs
     if obs.get('status')!='live':
         prev=None
         with lock:prev=(tasks.get(user) or {}).get('state')
@@ -821,8 +848,15 @@ def _check_account(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 relaying:
+        with _direct_guard:cur_src=(_direct_handles.get(user) or {}).get('src','')
+        if cur_src!=src_host:
+            _stop_direct(user, 'guest source switch')
+            log(user,'[GUEST] @%s source switch: %s'%(user, ('guest on @%s'%src_host) if src_host else 'own room'))
+    if src_host:
+        log(user,'[GUEST] @%s is a guest on @%s; relaying host stream'%(user,src_host))
+    _set_state(user,'LIVE','LIVE' if not src_host else 'LIVE · ضيف عند @%s'%src_host,**extra)
+    _maybe_start_direct(user, obs, src_host)
     if _oauth_automation_on() and not hub.busy(user) and (obs.get('sources') or []):
         try:
             youtube.nudge_live(user, obs)
@@ -1082,6 +1116,7 @@ def startup():
         except ValueError:continue
         tasks[u]=_blank_task();tasks[u]['enabled']=bool(enabled)
         if not enabled:tasks[u]['status']='متوقف'
+    GUEST.start()
     threading.Thread(target=monitor_loop,daemon=True,name='global-monitor').start()
     youtube.restore()
     log('system','[MONITOR] waiting')

guest_watch.py

@@ -0,0 +1,142 @@
+"""Guest (LinkMic / co-host) detection from host rooms.
+
+Uses the public room WebSocket through piratetok-live-py (no API key, no signing
+server). A guest is treated as "on stage" while the guest's userId keeps appearing
+in link_mic_method events of a watched host room. The signal expires after
+GUEST_ACTIVE_SECONDS of silence.
+
+Hosts to watch come from GUEST_WATCH_HOSTS (comma separated). Guests are matched
+against the accounts that are enabled in this project. A guest does not need to be
+listed as a host; the host room where the guest appears must be listed.
+"""
+import os, re, threading, time
+
+DEFAULT_ACTIVE_SECONDS = 120.0
+
+
+def parse_hosts(raw):
+    out = []
+    for h in re.split(r'[,\s;]+', raw or ''):
+        h = h.strip().lstrip('@').lower()
+        if h and h not in out:
+            out.append(h)
+    return out
+
+
+def _active_seconds():
+    try:
+        return max(30.0, float(os.getenv('GUEST_ACTIVE_SECONDS', str(DEFAULT_ACTIVE_SECONDS))))
+    except ValueError:
+        return DEFAULT_ACTIVE_SECONDS
+
+
+class GuestWatch:
+    def __init__(self, hosts, log=None, active_seconds=None):
+        self.hosts = list(hosts or [])
+        self.active_seconds = float(active_seconds) if active_seconds else _active_seconds()
+        self._log = log or (lambda user, msg: None)
+        self._lock = threading.Lock()
+        self._seen = {}     # guest uid -> (host, last_ts)
+        self._uids = {}     # username -> (uid, looked_up_ts)
+        self._names = {}    # uid -> username
+        self._stop = threading.Event()
+        self._threads = []
+        self._profiles = None
+        self.running = False
+
+    # ---- identity -------------------------------------------------------
+    def _profile_cache(self):
+        if self._profiles is None:
+            from piratetok_live import ProfileCache
+            self._profiles = ProfileCache()
+        return self._profiles
+
+    def uid_for(self, username):
+        """TikTok numeric userId for a username (cached). Empty string when unknown."""
+        username = (username or '').strip().lstrip('@').lower()
+        if not username:
+            return ''
+        now = time.time()
+        with self._lock:
+            hit = self._uids.get(username)
+            if hit and (hit[0] or now - hit[1] < 600):
+                return hit[0]
+        uid = ''
+        try:
+            uid = str(self._profile_cache().fetch(username).user_id or '')
+        except Exception as e:
+            self._log('system', '[GUEST] profile lookup failed for @%s: %s' % (username, type(e).__name__))
+        with self._lock:
+            self._uids[username] = (uid, now)
+            if uid:
+                self._names[uid] = username
+        return uid
+
+    # ---- signal ---------------------------------------------------------
+    def _label(self, uid):
+        with self._lock:
+            return self._names.get(uid) or ('id…' + uid[-4:])
+
+    def _mark(self, uid, host):
+        now = time.time()
+        with self._lock:
+            prev = self._seen.get(uid)
+            self._seen[uid] = (host, now)
+        if not prev or now - prev[1] > self.active_seconds:
+            self._log('system', '[GUEST] @%s guest activity detected on @%s' % (self._label(uid), host))
+
+    def active_host(self, username):
+        """Host name if @username is currently a guest on a watched host, else ''."""
+        if not self.running:
+            return ''
+        uid = self.uid_for(username)
+        if not uid:
+            return ''
+        with self._lock:
+            hit = self._seen.get(uid)
+        if hit and time.time() - hit[1] <= self.active_seconds:
+            return hit[0]
+        return ''
+
+    # ---- watchers -------------------------------------------------------
+    def _run_host(self, host):
+        from piratetok_live import TikTokLiveClient, EventType
+        host_uid = ''
+        while not self._stop.is_set():
+            try:
+                if not host_uid:
+                    host_uid = self.uid_for(host)
+                client = TikTokLiveClient(host).cdn_eu()
+
+                def on_link(evt, host=host, host_uid=host_uid):
+                    d = evt.data if isinstance(evt.data, dict) else {}
+                    uid = str(d.get('userId') or '')
+                    if uid and uid != '0' and uid != host_uid:
+                        self._mark(uid, host)
+
+                client.on(EventType.link_mic_method)(on_link)
+                self._log('system', '[GUEST] watching host @%s' % host)
+                client.run()
+                self._log('system', '[GUEST] @%s watcher ended; reconnecting' % host)
+            except Exception as e:
+                self._log('system', '[GUEST] @%s watcher error: %s' % (host, type(e).__name__))
+            self._stop.wait(30)
+
+    def start(self):
+        if not self.hosts:
+            self._log('system', '[GUEST] GUEST_WATCH_HOSTS empty; guest detection off')
+            return False
+        try:
+            import piratetok_live  # noqa: F401
+        except Exception as e:
+            self._log('system', '[GUEST] piratetok unavailable (%s); guest detection off' % type(e).__name__)
+            return False
+        self.running = True
+        for h in self.hosts:
+            t = threading.Thread(target=self._run_host, args=(h,), daemon=True, name='guest-' + h[:24])
+            t.start()
+            self._threads.append(t)
+        return True
+
+    def stop(self):
+        self._stop.set()

requirements.txt

@@ -3,3 +3,4 @@ fastapi==0.142.2
 uvicorn==0.54.0
 curl_cffi==0.16.3
 pydantic==2.13.5
+piratetok-live-py==0.3.0

test_guest_watch.py

@@ -0,0 +1,44 @@
+#!/usr/bin/env python3
+"""Unit checks for guest_watch.GuestWatch (no network)."""
+import sys, time, pathlib
+
+ROOT = pathlib.Path(sys.argv[1] if len(sys.argv) > 1 else '.').resolve()
+sys.path.insert(0, str(ROOT))
+import guest_watch
+from guest_watch import GuestWatch, parse_hosts
+
+assert parse_hosts(' @Talolil, muhajer_iraqi ;talolil\n') == ['talolil', 'muhajer_iraqi'], 'host parsing'
+assert parse_hosts('') == []
+
+logs = []
+g = GuestWatch(['talolil'], log=lambda u, m: logs.append(m), active_seconds=60)
+g.running = True
+uids = {'ahmadfamouss': '7577974697217475597', 'talolil': '6923684579661169666'}
+g.uid_for = lambda name: uids.get((name or '').lstrip('@').lower(), '')
+
+# Not seen yet
+assert g.active_host('ahmadfamouss') == ''
+
+# Guest seen on talolil -> active
+g._mark('7577974697217475597', 'talolil')
+assert g.active_host('ahmadfamouss') == 'talolil', 'guest should be active after event'
+assert g.active_host('@AhmadFamouss') == 'talolil', 'username normalisation'
+
+# Unrelated account stays inactive
+assert g.active_host('someone_else') == ''
+
+# Silence longer than active_seconds -> inactive
+with g._lock:
+    host, _ = g._seen['7577974697217475597']
+    g._seen['7577974697217475597'] = (host, time.time() - 61)
+assert g.active_host('ahmadfamouss') == '', 'guest should expire after silence'
+
+# Not running -> never active
+g.running = False
+g._mark('7577974697217475597', 'talolil')
+assert g.active_host('ahmadfamouss') == ''
+
+# Host userId must be ignored by the watcher callback logic (checked via uid compare)
+assert GuestWatch(['x']).start() in (False, True)  # must not crash when piratetok is missing or present
+
+print('PASS  guest_watch: join/expire/normalise/hosts')