Spaces:
Running
Running
| """تشخيص مباشر لحالة مصدر تيك توك أثناء البث (يُشغَّل من داخل الحاوية أو محليًا). | |
| الغرض: عندما يقول حدث يوتيوب «الجلسة مستمرة؛ انتظار المصدر المحدد للجودة» نريد أن نعرف | |
| بالضبط أيُّ شكلٍ رجّعتْه تيك توك في تلك اللحظة، بدل تخمين سبب فقدان المرشّحات عند الدخول | |
| إلى جولات PK أو الخروج منها. لا يكتب أي توكن ولا أي رابط كامل — فقط البنية والحالة. | |
| الاستعمال: | |
| python3 probe_live.py <username> [<username> ...] | |
| python3 probe_live.py --relay [url] # طرف يوتيوب/المشرف من اللوحة نفسها | |
| """ | |
| import json, os, sys | |
| def relay_state(url=None, user=None, password=None): | |
| """يطبع ما يراه المشرف من طرف يوتيوب: عمر آخر بايت، حدّ الحارس، سبب القفل، ونافذة التأجيل. | |
| لا يستورد app ولا يلمس أي ملف: قراءة HTTP واحدة للوحة، بكلمة سرّ من البيئة فقط — حتى تُشغَّل | |
| من داخل الحاوية (`python3 probe_live.py omarko89 --relay`) وهي تشخّص الطرفين في نفس الوقت. | |
| """ | |
| import base64, urllib.request | |
| url = url or os.getenv('PROBE_RELAY_URL', 'http://127.0.0.1:7860/api/youtube') | |
| user = user or os.getenv('USER_NAME', 'admin') | |
| password = password or os.getenv('PASSWORD', '') | |
| req = urllib.request.Request(url) | |
| if password: | |
| req.add_header('Authorization', 'Basic ' + base64.b64encode(('%s:%s' % (user, password)).encode()).decode()) | |
| try: | |
| with urllib.request.urlopen(req, timeout=15) as resp: | |
| d = json.loads(resp.read().decode('utf-8')) | |
| except Exception as e: | |
| print('\n=== relay | EXC %s: %s' % (type(e).__name__, str(e)[:150])) | |
| return | |
| print('\n=== طرف المشرف (يوتيوب) | send_mode=%s · %s' % (d.get('send_mode'), d.get('status'))) | |
| for ev in d.get('events') or []: | |
| print(' @%s phase=%s yt=%s health=%s' % (ev.get('username'), ev.get('phase'), | |
| ev.get('youtube_status'), ev.get('youtube_health'))) | |
| print(' مصدر: %s | خرج %.1f ث | آخر بايت %s ث | حدّ الحارس %s ث | صمت عند آخر قفل %s ث' % ( | |
| ev.get('source'), float(ev.get('output_seconds') or 0), ev.get('source_age_seconds'), | |
| ev.get('source_stall_limit'), ev.get('source_gap'))) | |
| if ev.get('retry_in'): | |
| print(' ⏳ إعادة الاتصال مؤجّلة %.0f ث (عقاب الوجهة) — إيقاف ثم تفعيل الأتمتة يُلغيها' % ev['retry_in']) | |
| if ev.get('note'): | |
| print(' note: %s' % str(ev['note'])[:140]) | |
| diag = list(ev.get('ffmpeg_diagnostics') or []) | |
| raw = list(ev.get('ffmpeg_errors') or []) | |
| if diag or raw: | |
| # «لماذا أقفلت الوجهة؟» من فم FFmpeg نفسه، بعد إخفاء الرابط والمفتاح في relay.py | |
| print(' تشخيص FFmpeg: %s' % (' | '.join(diag) if diag else 'سطر غير مصنَّف')) | |
| if raw: | |
| print(' آخر رسالة: %s' % str(raw[-1])[:160]) | |
| if not (d.get('events') or []): | |
| print(' لا أحداث جارية (المذيع غير مباشر أو الأتمتة موقوفة)') | |
| def brief(url): | |
| if not isinstance(url, str): | |
| return type(url).__name__ | |
| host = url.split('/', 3)[2] if url.startswith('http') else '(not-http)' | |
| return '%s…%s len=%d' % (url[:14], host, len(url)) | |
| def shape(node, depth=0, path=''): | |
| """وصف مختصر للشجرة: المفاتيح وأنواعها، مع تفريغ ما يخصّ stream_data فقط.""" | |
| out = [] | |
| pad = ' ' * depth | |
| if isinstance(node, dict): | |
| for k in sorted(node.keys()): | |
| v = node[k] | |
| if isinstance(v, (dict, list)) and depth < 4: | |
| out.append('%s%s: %s(%d)' % (pad, k, type(v).__name__, len(v))) | |
| if k in ('stream_data', 'data', 'main', 'stream_url', 'pull_data', 'live_core_sdk_data', 'sdk_params'): | |
| out += shape(v, depth + 1, path + '/' + k) | |
| else: | |
| t = type(v).__name__ | |
| if t == 'str' and len(v) > 60: | |
| out.append('%s%s: url %s' % (pad, k, brief(v))) | |
| elif t in ('str', 'int', 'float', 'bool') or v is None: | |
| out.append('%s%s: %s=%r' % (pad, k, t, v if t != 'str' else v[:60])) | |
| else: | |
| out.append('%s%s: %s' % (pad, k, t)) | |
| elif isinstance(node, list): | |
| out.append('%s[list of %d]' % (pad, len(node))) | |
| if node: | |
| out += shape(node[0], depth + 1, path + '[0]') | |
| return out | |
| def main(): | |
| sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)) or '.') | |
| from curl_cffi import requests | |
| users = sys.argv[1:] or ['viggoo313'] | |
| for user in users: | |
| tok = os.getenv('TIKTOK_SESSIONID_' + user.upper().replace('.', '_'), '') or os.getenv('TIKTOK_SESSIONID', '') | |
| sid = tok.split(';')[0].strip() | |
| sid = sid.split('=', 1)[1] if '=' in sid else sid | |
| headers = {'Referer': 'https://www.tiktok.com/'} | |
| if sid: | |
| headers['Cookie'] = 'sessionid=%s; sessionid_ss=%s; sid_tt=%s' % (sid, sid, sid) | |
| url = 'https://www.tiktok.com/api-live/user/room' | |
| try: | |
| r = requests.get(url, params={'aid': 1988, 'uniqueId': user, 'sourceType': 54, 'staleTime': 0}, | |
| impersonate='chrome', headers=headers, timeout=25) | |
| print('\n=== %s | HTTP %s | bytes %d' % (user, r.status_code, len(r.content))) | |
| if r.status_code != 200: | |
| continue | |
| j = r.json() | |
| d = j.get('data', {}) or {} | |
| live = d.get('liveRoom', {}) or {} | |
| status = live.get('status') | |
| person = d.get('user', {}) or {} | |
| print(' liveRoom.status=%r roomId=%r title=%r' % (status, person.get('roomId'), str(live.get('title'))[:40])) | |
| for key in ('streamData', 'hevcStreamData', 'pull_stream_url', 'live_core_sdk_data'): | |
| blk = live.get(key) | |
| if not isinstance(blk, dict): | |
| print(' %-20s: %s' % (key, type(blk).__name__)) | |
| continue | |
| sd = blk.get('pull_data', {}).get('stream_data') | |
| if isinstance(sd, str): | |
| try: | |
| sd = json.loads(sd) | |
| except Exception: | |
| print(' %-20s: stream_data نص غير قابل للتحليل' % key) | |
| continue | |
| data = (sd or {}).get('data') | |
| print(' %-20s: data=%s' % (key, type(data).__name__)) | |
| items = data.items() if isinstance(data, dict) else list(enumerate(data or [])) | |
| for label, item in items: | |
| if not isinstance(item, dict): | |
| print(' %s: %s' % (label, type(item).__name__)) | |
| continue | |
| main = item.get('main', {}) or {} | |
| keys = sorted(main.keys()) | |
| params = main.get('sdk_params') or {} | |
| if isinstance(params, str): | |
| try: | |
| params = json.loads(params) | |
| except Exception: | |
| params = {} | |
| print(' %-8s main=%s vcodec=%r res=%r fps=%r' % ( | |
| label, keys[:9], params.get('VCodec'), params.get('resolution'), params.get('expectedFps'))) | |
| for media in ('flv', 'hls', 'lls'): | |
| if media in main: | |
| print(' %s → %s' % (media, brief(main[media]))) | |
| except Exception as e: | |
| print('\n=== %s | EXC %s: %s' % (user, type(e).__name__, str(e)[:160])) | |
| def main_relay(): | |
| argv = [a for a in sys.argv[1:] if not a.startswith('--')] | |
| relay_state(argv[0] if argv else None) | |
| if __name__ == '__main__': | |
| if '--relay' in sys.argv: | |
| main_relay() | |
| else: | |
| main() | |