QUICKTEST: temporary manual-key relay harness for live validation (to be removed)

app.py

@@ -507,6 +507,69 @@ def safe_file(name):
 def download(name:str):return FileResponse(safe_file(name),filename=name)
 @app.delete('/api/files/{name}')
 def delete(name:str):safe_file(name).unlink();return {'ok':True}
+# ---- QUICKTEST — temporary live-validation harness (REMOVE after test) -----
+# Pushes the top ranked TikTok source of one account to a MANUAL rtmps key with
+# the exact same Sender/relay.py pipeline the manager uses (YOUTUBE_SEND_MODE
+# applies), so a real stream key can be validated before OAuth automation.
+QT={'running':False,'status':'','restarts':0,'started_at':None,'log':[],'sender':{}}
+_qt_stop=threading.Event()
+def qt_log(msg):
+    QT['log'].append('%s %s'%(time.strftime('%H:%M:%S'),msg))
+    QT['log']=QT['log'][-80:]
+    log('quicktest',msg)
+class QuickTestBody(BaseModel):
+    user:str
+    rtmp:str
+def qt_run(user,destination):
+    from relay import Sender
+    from flv_bridge import FormatChanged
+    sender=Sender(FFMPEG,qt_log,headers_for=tiktok_headers)
+    def reporter():
+        while not _qt_stop.is_set():
+            QT['sender']=sender.snapshot()
+            _qt_stop.wait(2)
+    threading.Thread(target=reporter,daemon=True).start()
+    while not _qt_stop.is_set():
+        try:
+            sources=extract(user)
+            if not sources:
+                QT['status']='المصدر غير مباشر حاليًا';_qt_stop.wait(15);continue
+            top=[s for s in sources if any(c in s['label'].lower() for c in ('h264','avc','h265','hevc'))][:1] or sources[:1]
+            QT['status']='الإرسال إلى يوتيوب'
+            sender.send(user,top[0],destination,_qt_stop)
+            if _qt_stop.is_set():break
+            QT['restarts']+=1
+            QT['status']='انقطع الإرسال؛ إعادة الاتصال'
+            _qt_stop.wait(2)
+        except FormatChanged:
+            QT['restarts']+=1
+            qt_log('تغيّرت أبعاد المصدر (بداية/نهاية جولة PK؟)؛ إعادة الاتصال بمقاس جديد')
+        except Exception as e:
+            QT['status']='تعثر: '+str(e)[:140]
+            _qt_stop.wait(5)
+    QT['running']=False
+    QT['status']='متوقف'
+@app.post('/api/quicktest')
+def quicktest_start(b:QuickTestBody):
+    if QT['running']:raise HTTPException(400,'التجربة تعمل بالفعل')
+    if not b.rtmp.startswith(('rtmp://','rtmps://')):raise HTTPException(400,'عنوان الإرسال غير صالح')
+    try:u=clean_user(b.user)
+    except ValueError as e:raise HTTPException(400,str(e))
+    _qt_stop.clear()
+    QT.update(running=True,status='بدء التجربة',restarts=0,started_at=time.time(),log=[],sender={})
+    threading.Thread(target=qt_run,args=(u,b.rtmp),daemon=True).start()
+    qt_log('بدأت التجربة السريعة نحو مفتاح يدوي للحساب '+u)
+    return {'ok':True}
+@app.get('/api/quicktest')
+def quicktest_state():
+    return QT
+@app.post('/api/quicktest/stop')
+def quicktest_stop():
+    _qt_stop.set()
+    qt_log('طلب إيقاف التجربة')
+    return {'ok':True}
+# ---- end QUICKTEST ----------------------------------------------------------
+
 @app.on_event('startup')
 def startup():
     load_tokens()