remove browser eye (operator request); relay-v2.42-guestwatch-pro

Dockerfile

@@ -5,10 +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
-# متصفح حقيقي للعين الاحترافية (guestwatch) — يُفعَّل فقط بـ GUEST_BROWSER=1
-RUN pip install --no-cache-dir playwright && playwright install --with-deps chromium-headless-shell
-COPY app.py relay.py youtube_api.py youtube_manager.py flv_bridge.py stream_monitor.py telemetry.py tiktok_auth.py ingest.py guest_probe.py tt_sign.py browser_eye.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 test_telemetry.py test_smart_mode.py test_guest_probe.py test_tt_sign.py test_browser_eye.py ./
+COPY app.py relay.py youtube_api.py youtube_manager.py flv_bridge.py stream_monitor.py telemetry.py tiktok_auth.py ingest.py guest_probe.py tt_sign.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 test_telemetry.py test_smart_mode.py test_guest_probe.py test_tt_sign.py ./
 
 # TikTok Live Monitor → Direct Relay → YouTube. No local video files.
 ENV TZ=Asia/Baghdad \

app.py

@@ -490,33 +490,6 @@ def _guest_pro_eye():
     except Exception:
         _guest_pro=None
     return _guest_pro
-_guest_browser=False
-def _guest_browser_eye():
-    global _guest_browser
-    if _guest_browser not in (False,'building'):return _guest_browser
-    if os.getenv('GUEST_BROWSER','').strip().lower() not in ('1','true','yes','on'):
-        _guest_browser=None;return None
-    if _guest_browser=='building':return None
-    _guest_browser='building'
-    def _build():
-        global _guest_browser
-        try:
-            import browser_eye
-            if not browser_eye.ensure_installed(lambda m:log('system',m)):
-                log('system','[GUEST] browser eye unavailable (playwright install failed)')
-                _guest_browser=None;return
-            cookie=''
-            try:
-                us=_guest_users()
-                if us:cookie=tiktok_auth.headers(us[0]).get('Cookie','') or ''
-            except Exception:cookie=''
-            _guest_browser=browser_eye.BrowserEye(cookie_header=cookie)
-            log('system','[GUEST] browser eye ready (cookies=%s)'%('yes' if cookie else 'no'))
-        except Exception as e:
-            log('system','[GUEST] browser eye build failed: %s %s'%(type(e).__name__,str(e)[:120]))
-            _guest_browser=None
-    import threading;threading.Thread(target=_build,name='browser-eye-build',daemon=True).start()
-    return None
 _guest_diag={'n':0}
 def _guest_cycle():
     global _guest_watcher
@@ -528,7 +501,7 @@ def _guest_cycle():
         import guest_probe as _gp
         _hosts=_guest_hosts()
         _scan=(lambda uids:_gp.scan_host_rooms(uids,_hosts)) if _hosts else None
-        written=_guest_watcher.cycle(users,probe=lambda u:_gp.probe_user(u,browser=_guest_browser_eye(),pro=_guest_pro_eye(),**_guest_fetchers(u)),host_scan=_scan)
+        written=_guest_watcher.cycle(users,probe=lambda u:_gp.probe_user(u,pro=_guest_pro_eye(),**_guest_fetchers(u)),host_scan=_scan)
     except Exception as e:
         _guest_diag['n']+=1
         if _guest_diag['n']<=5 or _guest_diag['n']%30==1:

browser_eye.py

@@ -1,302 +0,0 @@
-"""العين المتصفحية: متصفح Chromium حقيقي (Playwright) ينفّذ طلبات تيك توك من
-داخل صفحة tiktok.com الأصلية، فيوقّعها webmssdk تلقائيًا (X-Bogus/X-Gnarly/
-msToken حقيقية) وتُنفَّذ باتصال متصفح كامل — نفس نمط خوادم التوقيع المجانية
-المنشورة (carcabot/tiktok-signature وأمثالها).
-
-لماذا؟ قياس 2026-10-17: طلبات /api/* تُفرَّغ (200 بطول 0) لكل عميل غير
-متصفحي مهما كان التوقيع سليمًا، و webcast/room/info_by_user المنفّذة داخل
-متصفح حقيقي هي أقوى سياق مجاني متاح. مع كوكيز جلسة المستخدم نفسه (اختياري)
-تصبح العين «مُصادقة + موقّعة + متصفحية» معًا.
-
-البنية: خيط خلفي واحد يملك حلقة asyncio ومتصفحًا كسول الإقلاع يُعاد تدويره
-كل recycle_seconds. كل resolve ≤ ~40 ثانية؛ أي فشل يرفع استثناء والمستدعي
-(guest_probe) يسجّله ويكمل بالقنوات الأخرى. بلا أي حذف/تدوير لأي سجل.
-"""
-from __future__ import annotations
-
-import asyncio
-import json
-import re
-import threading
-import time
-
-UA = ('Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 '
-      '(KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36')
-HOME_URL = 'https://www.tiktok.com/'
-LIVE_URL = 'https://www.tiktok.com/@%s/live'
-BY_USER_URL = ('https://webcast.tiktok.com/webcast/room/info_by_user/'
-               '?aid=1988&unique_id=%s')
-DETAIL_URL = 'https://www.tiktok.com/api/user/detail/?uniqueId=%s'
-_ROOM_RE = re.compile(r'"roomId"\s*:\s*"(\d{10,})"')
-
-
-def parse_cookie_header(header):
-    """'a=1; b=2' -> قائمة كوكيز Playwright لنطاق تيك توك."""
-    out = []
-    for part in str(header or '').split(';'):
-        part = part.strip()
-        if not part or '=' not in part:
-            continue
-        name, value = part.split('=', 1)
-        name = name.strip()
-        if not name:
-            continue
-        out.append({'name': name, 'value': value.strip(),
-                    'domain': '.tiktok.com', 'path': '/'})
-    return out
-
-
-def _json_obj_at(text, start):
-    depth = 0
-    in_str = False
-    esc = False
-    for i in range(start, len(text)):
-        ch = text[i]
-        if in_str:
-            if esc:
-                esc = False
-            elif ch == '\\':
-                esc = True
-            elif ch == '"':
-                in_str = False
-        elif ch == '"':
-            in_str = True
-        elif ch == '{':
-            depth += 1
-        elif ch == '}':
-            depth -= 1
-            if depth == 0:
-                try:
-                    return json.loads(text[start:i + 1])
-                except ValueError:
-                    return None
-    return None
-
-
-def parse_by_user_body(text):
-    """رد info_by_user -> {owner, room, status, message}."""
-    out = {}
-    try:
-        j = json.loads(text or '')
-    except ValueError:
-        return out
-    data = (j or {}).get('data') or {}
-    if not isinstance(data, dict):
-        return out
-    if data.get('message'):
-        out['message'] = str(data.get('message'))[:80]
-    owner = data.get('owner') or {}
-    handle = str(owner.get('display_id') or owner.get('uniqueId') or owner.get('unique_id') or '')
-    if handle:
-        out['owner'] = handle
-    for k in ('id_str', 'room_id', 'roomId', 'id'):
-        v = data.get(k)
-        if isinstance(v, int) and v > 10 ** 15:
-            out['room'] = str(v)
-            break
-        if isinstance(v, str) and v.isdigit() and len(v) >= 15:
-            out['room'] = v
-            break
-    if data.get('status') is not None:
-        out['status'] = data.get('status')
-    return out
-
-
-def parse_detail_body(text):
-    """رد user/detail -> أدلة الحقول الحيّة فقط (بلا روابط)."""
-    out = {'bytes': len(text or '')}
-    if not text:
-        return out
-    try:
-        j = json.loads(text)
-    except ValueError:
-        out['unparsed'] = True
-        return out
-    if isinstance(j, dict):
-        usr = ((j.get('userInfo') or {}).get('user')) or {}
-        fields = {}
-        for k, v in usr.items():
-            if 'live' in str(k).lower() or 'room' in str(k).lower():
-                fields[k] = v if not isinstance(v, (dict, list)) else '<obj>'
-        if fields:
-            out['live_fields'] = fields
-    ids = sorted(set(_ROOM_RE.findall(text)))
-    if ids:
-        out['room_ids'] = ids
-    m = re.search(r'"liveRoom"\s*:\s*\{', text)
-    if m:
-        node = _json_obj_at(text, m.end() - 1)
-        if isinstance(node, dict) and str(node.get('status')) == '2':
-            for k in ('id_str', 'room_id', 'roomId', 'id'):
-                v = node.get(k)
-                if isinstance(v, int) and v > 10 ** 15:
-                    out['guest_room'] = str(v)
-                    break
-                if isinstance(v, str) and v.isdigit() and len(v) >= 15:
-                    out['guest_room'] = v
-                    break
-            ow = node.get('owner') or {}
-            if isinstance(ow, dict):
-                handle = str(ow.get('display_id') or ow.get('uniqueId') or '')
-                if handle:
-                    out['guest_owner'] = handle
-    return out
-
-
-def ensure_installed(log=lambda *a: None):
-    """ثبّت Playwright + Chromium عند غيابهما (يُستدعى من خيط الإقلاع فقط)."""
-    try:
-        import playwright  # noqa: F401
-        from playwright.sync_api import sync_playwright
-        with sync_playwright() as pw:
-            exe = pw.chromium.executable_path
-        import os
-        if exe and os.path.exists(exe):
-            return True
-    except Exception:
-        pass
-    import subprocess
-    import sys
-    log('[BROWSER] installing playwright chromium (one-time per container)')
-    subprocess.run([sys.executable, '-m', 'pip', 'install', '-q', 'playwright'],
-                   check=False, timeout=600)
-    subprocess.run([sys.executable, '-m', 'playwright', 'install', 'chromium-headless-shell'],
-                   check=False, timeout=900)
-    try:
-        import os
-        from playwright.sync_api import sync_playwright
-        with sync_playwright() as pw:
-            return bool(pw.chromium.executable_path and os.path.exists(pw.chromium.executable_path))
-    except Exception:
-        return False
-
-
-class BrowserEye:
-    """واجهة متزامنة فوق متصفح خلفي. resolve(user) يعيد dict بالأدلة."""
-
-    def __init__(self, cookie_header='', recycle_seconds=1500.0):
-        self._cookies = parse_cookie_header(cookie_header)
-        self._recycle = float(recycle_seconds)
-        self._loop = None
-        self._thread = None
-        self._browser = None
-        self._context = None
-        self._born = 0.0
-        self._started = False
-        self._lock = threading.Lock()
-
-    # ---------- البنية الخلفية ----------
-    def _ensure_thread(self):
-        if self._thread and self._thread.is_alive():
-            return
-        ready = threading.Event()
-
-        def run():
-            self._loop = asyncio.new_event_loop()
-            asyncio.set_event_loop(self._loop)
-            ready.set()
-            self._loop.run_forever()
-
-        self._thread = threading.Thread(target=run, name='browser-eye', daemon=True)
-        self._thread.start()
-        ready.wait(10)
-
-    def _run(self, coro, timeout):
-        self._ensure_thread()
-        fut = asyncio.run_coroutine_threadsafe(coro, self._loop)
-        return fut.result(timeout=timeout)
-
-    async def _get_context(self):
-        from playwright.async_api import async_playwright
-        now = time.time()
-        if (self._context is not None and self._browser is not None
-                and now - self._born < self._recycle):
-            return self._context
-        await self._close_async()
-        self._pw = await async_playwright().start()
-        self._browser = await self._pw.chromium.launch(args=['--no-sandbox'])
-        self._context = await self._browser.new_context(user_agent=UA, locale='en-US')
-        if self._cookies:
-            try:
-                await self._context.add_cookies(self._cookies)
-            except Exception:
-                pass
-        page = await self._context.new_page()
-        try:
-            await page.goto(HOME_URL, wait_until='domcontentloaded', timeout=60000)
-            await page.wait_for_timeout(5000)
-        finally:
-            await page.close()
-        self._born = now
-        return self._context
-
-    async def _close_async(self):
-        for attr in ('_context', '_browser'):
-            obj = getattr(self, attr, None)
-            if obj is not None:
-                try:
-                    await obj.close()
-                except Exception:
-                    pass
-                setattr(self, attr, None)
-        pw = getattr(self, '_pw', None)
-        if pw is not None:
-            try:
-                await pw.stop()
-            except Exception:
-                pass
-            self._pw = None
-
-    # ---------- القياس ----------
-    async def _resolve_async(self, user):
-        import guest_probe
-        ctx = await self._get_context()
-        page = await ctx.new_page()
-        captured = {}
-
-        def on_req(req):
-            u = req.url
-            if 'user/detail' in u:
-                captured['detail'] = u
-            elif 'info_by_user' in u:
-                captured['by_user'] = u
-
-        page.on('request', on_req)
-        out = {}
-        try:
-            await page.goto(LIVE_URL % user, wait_until='domcontentloaded', timeout=60000)
-            await page.wait_for_timeout(7000)
-            html = await page.content()
-            out['ssr'] = guest_probe.parse_live_page(html)
-            out['final_url'] = page.url[:120]
-            await page.evaluate(
-                """(u) => {
-                  fetch('/api/user/detail/?uniqueId=' + u).catch(() => {});
-                  fetch('https://webcast.tiktok.com/webcast/room/info_by_user/?aid=1988&unique_id=' + u,
-                        {credentials: 'include'}).catch(() => {});
-                  return 1;
-                }""", user)
-            await page.wait_for_timeout(4000)
-        finally:
-            page.remove_listener('request', on_req)
-            await page.close()
-        for name, parser in (('by_user', parse_by_user_body), ('detail', parse_detail_body)):
-            url = captured.get(name)
-            if not url:
-                continue
-            p2 = await ctx.new_page()
-            try:
-                resp = await p2.goto(url, wait_until='domcontentloaded', timeout=45000)
-                body = await p2.inner_text('body')
-                parsed = parser(body)
-                parsed['http'] = resp.status if resp else 0
-                out[name] = parsed
-            except Exception as e:
-                out[name] = {'err': type(e).__name__}
-            finally:
-                await p2.close()
-        return out
-
-    def resolve(self, user, timeout=120.0):
-        with self._lock:
-            return self._run(self._resolve_async(user), timeout)

guest_probe.py

@@ -546,70 +546,8 @@ def _euler_view(user, rec, pro, get_json, req):
     return None
 
 
-def _browser_view(user, rec, browser, get_json, req):
-    """قرار العين المتصفحية: Chromium حقيقي + (اختياريًا) جلسة المستخدم نفسه.
-    نفس قواعد _pro_view لكن بمصادر browser_*، والأدلة تحت rec['browser']."""
-    ev = browser.resolve(user) or {}
-    bv = {}
-    promoted = None
-
-    def _decide(owner, owner_id, room, status, source):
-        nonlocal promoted
-        if promoted is not None or not owner or not room or str(status) != '2':
-            return
-        same_person = owner.lower() == str(user).lower() or (
-            owner_id and rec.get('uid') and str(owner_id) == str(rec.get('uid')))
-        if same_person:
-            promoted = ('hosting', user, room, source)
-        else:
-            promoted = ('guest', owner, room, source)
-
-    ssr = ev.get('ssr') or {}
-    keep = {k: ssr.get(k) for k in ('owner', 'owner_id', 'room', 'status')
-            if ssr.get(k) not in (None, '')}
-    if keep:
-        bv['ssr'] = keep
-    _decide(str(ssr.get('owner') or ''), str(ssr.get('owner_id') or ''),
-            str(ssr.get('room') or ''), ssr.get('status'), 'browser_ssr')
-    bu = ev.get('by_user') or {}
-    keep = {k: bu.get(k) for k in ('owner', 'room', 'status', 'message', 'http')
-            if bu.get(k) not in (None, '')}
-    if keep:
-        bv['by_user'] = keep
-    _decide(str(bu.get('owner') or ''), '', str(bu.get('room') or ''),
-            bu.get('status'), 'browser_by_user')
-    dt = ev.get('detail') or {}
-    if dt:
-        bv['detail'] = dt
-    if promoted is None:
-        groom = str(dt.get('guest_room') or '')
-        if groom and groom != str(rec.get('own_room') or ''):
-            promoted = ('guest', str(dt.get('guest_owner') or ''), groom, 'browser_detail')
-    rec['browser'] = bv
-    if promoted:
-        kind, host_user, host_room, src = promoted
-        rec.update(state=kind, host_user=host_user, host_room=host_room,
-                   host_status=2, evidence_source=src)
-        if kind == 'guest':
-            try:
-                if get_json is not None:
-                    info = (get_json(INFO_API, {'aid': 1988, 'room_id': host_room}) or {}).get('data') or {}
-                else:
-                    rr = req().get(INFO_API, params={'aid': 1988, 'room_id': host_room},
-                                   impersonate='chrome',
-                                   headers={'Referer': 'https://www.tiktok.com/'}, timeout=20)
-                    rr.raise_for_status()
-                    info = (rr.json() or {}).get('data') or {}
-                rec['watched_present'] = _watched_present(info, rec.get('uid', ''), user)
-                rec['evidence'] = _participants_view(info)
-                rec['host_title'] = str(info.get('title') or '')[:80]
-            except Exception as e:
-                rec['err'] = (rec['err'] + ' ' if rec['err'] else '') + 'browser_confirm:' + type(e).__name__
-    return promoted
-
-
 def probe_user(user, client=None, get_json=None, get_page=None,
-               get_json_authed=None, get_page_authed=None, pro=None, browser=None):
+               get_json_authed=None, get_page_authed=None, pro=None):
     """قياس كامل لمستخدم واحد الآن. يعيد سجلًّا جاهزًا للإلحاق (لا يكتب بنفسه)."""
     req = client
     def _req():
@@ -687,11 +625,6 @@ def probe_user(user, client=None, get_json=None, get_page=None,
     rec['state'] = classify(watched=user, own_status=rec['own_status'],
                             page_user=rec['page_user'], page_room=rec['page_room'],
                             host_status=rec['host_status'])
-    if rec['state'] in ('offline', 'unknown') and browser is not None:
-        try:
-            _browser_view(user, rec, browser, get_json, _req)
-        except Exception as e:
-            rec['browser'] = {'err': type(e).__name__}
     if rec['state'] in ('offline', 'unknown') and pro is not None:
         try:
             _pro_view(user, rec, pro, get_json, _req)

ingest.py

@@ -6,7 +6,7 @@ import os, threading, time
 from datetime import datetime
 from zoneinfo import ZoneInfo
 
-APP_VERSION = 'relay-v2.41-guestwatch-browser'
+APP_VERSION = 'relay-v2.42-guestwatch-pro'
 
 
 def env_rtmp_destination():

test_browser_eye.py

@@ -1,86 +0,0 @@
-"""اختبارات نقية للعين المتصفحية: تحليل الكوكيز والردود دون متصفح فعلي."""
-import json
-import unittest
-
-import browser_eye as be
-import guest_probe as gp
-
-
-class CookieParseTests(unittest.TestCase):
-    def test_pairs(self):
-        out = be.parse_cookie_header('sessionid=abc; ttwid=xyz; ; msToken = tok ')
-        names = [c['name'] for c in out]
-        self.assertEqual(names, ['sessionid', 'ttwid', 'msToken'])
-        self.assertTrue(all(c['domain'] == '.tiktok.com' for c in out))
-
-    def test_empty(self):
-        self.assertEqual(be.parse_cookie_header(''), [])
-        self.assertEqual(be.parse_cookie_header(None), [])
-
-
-class BodyParseTests(unittest.TestCase):
-    def test_by_user_finished(self):
-        out = be.parse_by_user_body('{"data":{"message":"room has finished"},"status_code":30003}')
-        self.assertEqual(out, {'message': 'room has finished'})
-
-    def test_by_user_live_host(self):
-        body = json.dumps({'data': {'id_str': '777666555444333222', 'status': 2,
-                                    'owner': {'display_id': 'bob'}}})
-        out = be.parse_by_user_body(body)
-        self.assertEqual(out['owner'], 'bob')
-        self.assertEqual(out['room'], '777666555444333222')
-        self.assertEqual(out['status'], 2)
-
-    def test_detail_empty(self):
-        self.assertEqual(be.parse_detail_body(''), {'bytes': 0})
-
-    def test_detail_guest_room(self):
-        body = json.dumps({'userInfo': {'user': {'uniqueId': 'alice', 'roomId': ''}},
-                           'liveRoom': {'status': 2, 'id_str': '777666555444333222',
-                                        'owner': {'display_id': 'bob'}}})
-        out = be.parse_detail_body(body)
-        self.assertEqual(out['guest_room'], '777666555444333222')
-        self.assertEqual(out['guest_owner'], 'bob')
-
-
-class _FakeBrowser:
-    def __init__(self, result):
-        self._result = result
-
-    def resolve(self, user):
-        return dict(self._result)
-
-
-class BrowserViewTests(unittest.TestCase):
-    def _offline_json(self, url, params):
-        if 'user/room' in url:
-            return {'data': {'liveRoom': {'status': 4, 'roomId': '555'},
-                             'user': {'id': '4242', 'roomId': '555', 'nickname': 'Alice'}}}
-        return {'data': {'status': 2, 'title': 'Host LIVE', 'owner': {'display_id': 'bob'},
-                         'link_mic': {'linked_user_list': []},
-                         'linker_map': {'4242': {'user_id': '4242'}}}}
-
-    def _page_self(self, url):
-        return 'https://www.tiktok.com/@alice/live', '"roomId":"555"'
-
-    def test_browser_by_user_guest(self):
-        br = _FakeBrowser({'by_user': {'owner': 'bob', 'room': '777666555444333222', 'status': 2, 'http': 200}})
-        rec = gp.probe_user('alice', get_json=self._offline_json,
-                            get_page=self._page_self, browser=br)
-        self.assertEqual(rec['state'], 'guest')
-        self.assertEqual(rec['host_user'], 'bob')
-        self.assertEqual(rec['evidence_source'], 'browser_by_user')
-        self.assertIs(rec['watched_present'], True)
-
-    def test_browser_ssr_offline_stays_offline(self):
-        br = _FakeBrowser({'ssr': {'owner': 'alice', 'owner_id': '4242', 'room': '555', 'status': 4},
-                           'by_user': {'message': 'room has finished', 'http': 200}})
-        rec = gp.probe_user('alice', get_json=self._offline_json,
-                            get_page=self._page_self, browser=br)
-        self.assertEqual(rec['state'], 'offline')
-        self.assertEqual(rec['browser']['by_user']['message'], 'room has finished')
-
-
-if __name__ == '__main__':
-    import sys
-    unittest.main(argv=[sys.argv[0]])

verify.sh

@@ -36,7 +36,6 @@ python3 test_telemetry.py "$ROOT"
 python3 test_smart_mode.py "$ROOT"
 python3 test_guest_probe.py "$ROOT"
 python3 test_tt_sign.py "$ROOT"
-python3 test_browser_eye.py "$ROOT"
 # Optional heavy tests when ffmpeg exists
 if command -v ffmpeg >/dev/null 2>&1; then
   python3 test_watchdog.py "$ROOT"