Reoo88 commited on
Commit
e4951d8
·
verified ·
1 Parent(s): 36f588a

Revert Prayer-Time-Bot to 0be4ca683e2a (user: 0pe4ca6) — H264 FLV copy, no HEVC origin fallthrough

Browse files
Files changed (6) hide show
  1. Dockerfile +2 -2
  2. app.py +2 -10
  3. index.html +1 -1
  4. ingest.py +11 -31
  5. relay.py +15 -105
  6. youtube_manager.py +41 -247
Dockerfile CHANGED
@@ -13,8 +13,8 @@ ENV TZ=Asia/Baghdad \
13
  CHECK_INTERVAL=20 \
14
  SOURCE_STALL_TIMEOUT=12 \
15
  YOUTUBE_SOURCE_STALL_SECONDS=12 \
16
- MAX_CONCURRENT_RELAYS=5 \
17
- YOUTUBE_MAX_EVENTS=5 \
18
  YOUTUBE_TITLE_TEMPLATE="{broadcaster} | {date} {time}" \
19
  YOUTUBE_SEND_MODE=auto \
20
  YOUTUBE_AUTO_TRANSCODE=1 \
 
13
  CHECK_INTERVAL=20 \
14
  SOURCE_STALL_TIMEOUT=12 \
15
  YOUTUBE_SOURCE_STALL_SECONDS=12 \
16
+ MAX_CONCURRENT_RELAYS=2 \
17
+ YOUTUBE_MAX_EVENTS=2 \
18
  YOUTUBE_TITLE_TEMPLATE="{broadcaster} | {date} {time}" \
19
  YOUTUBE_SEND_MODE=auto \
20
  YOUTUBE_AUTO_TRANSCODE=1 \
app.py CHANGED
@@ -537,7 +537,7 @@ def _maybe_start_direct(user, obs):
537
  return
538
  if hub.at_capacity(except_user=user):
539
  update(user, state='BACKOFF', status='بلغنا حد MAX_CONCURRENT_RELAYS')
540
- log(user,'[RELAY] @%s waiting: at MAX_CONCURRENT_RELAYS=%s'%(user, max_concurrent_relays() or 'open'))
541
  return
542
  stop=threading.Event()
543
  title=format_session_title(obs.get('nickname') or user)[0]
@@ -786,15 +786,9 @@ def _pick_last_error(tasks_map):
786
  return err, u
787
  return last_error, last_error_user
788
 
789
- _health_cache={'t':0.0,'v':None}
790
-
791
  @app.get('/health')
792
  @app.get('/healthz')
793
  def health_public():
794
- now=time.monotonic()
795
- cached=_health_cache.get('v')
796
- if cached is not None and now-float(_health_cache.get('t') or 0)<2.0:
797
- return cached
798
  with lock:
799
  n=len(tasks);enabled=sum(1 for t in tasks.values() if t.get('enabled'))
800
  last_error, last_error_user=_pick_last_error(tasks)
@@ -842,7 +836,7 @@ def health_public():
842
  pass
843
  except Exception:
844
  pass
845
- payload={'application':'running','monitor':'running' if _monitor_alive.is_set() else 'stopped',
846
  'accounts':n,'enabled':enabled,'active_relays':active,
847
  'max_concurrent_relays':max_concurrent_relays(),
848
  'source_stall_timeout':SOURCE_STALL_TIMEOUT,
@@ -858,8 +852,6 @@ def health_public():
858
  'youtube_status':y_status,'youtube_events':y_events,'youtube_history':y_hist,
859
  'youtube_senders':y_senders,'youtube_skip':y_skip,
860
  'broadcasters':per}
861
- _health_cache['t']=time.monotonic(); _health_cache['v']=payload
862
- return payload
863
 
864
  @app.get('/api/health')
865
  def health_detail():
 
537
  return
538
  if hub.at_capacity(except_user=user):
539
  update(user, state='BACKOFF', status='بلغنا حد MAX_CONCURRENT_RELAYS')
540
+ log(user,'[RELAY] @%s waiting: at MAX_CONCURRENT_RELAYS=%d'%(user, max_concurrent_relays()))
541
  return
542
  stop=threading.Event()
543
  title=format_session_title(obs.get('nickname') or user)[0]
 
786
  return err, u
787
  return last_error, last_error_user
788
 
 
 
789
  @app.get('/health')
790
  @app.get('/healthz')
791
  def health_public():
 
 
 
 
792
  with lock:
793
  n=len(tasks);enabled=sum(1 for t in tasks.values() if t.get('enabled'))
794
  last_error, last_error_user=_pick_last_error(tasks)
 
836
  pass
837
  except Exception:
838
  pass
839
+ return {'application':'running','monitor':'running' if _monitor_alive.is_set() else 'stopped',
840
  'accounts':n,'enabled':enabled,'active_relays':active,
841
  'max_concurrent_relays':max_concurrent_relays(),
842
  'source_stall_timeout':SOURCE_STALL_TIMEOUT,
 
852
  'youtube_status':y_status,'youtube_events':y_events,'youtube_history':y_hist,
853
  'youtube_senders':y_senders,'youtube_skip':y_skip,
854
  'broadcasters':per}
 
 
855
 
856
  @app.get('/api/health')
857
  def health_detail():
index.html CHANGED
@@ -181,7 +181,7 @@ async function refresh(){
181
  const lastErr=tasks.map(t=>t.last_error).filter(Boolean)[0] || (s0.logs||[]).filter(l=>/error|فشل|stall|خطأ/i.test(l.message||'')).slice(-1)[0]?.message || 'لا يوجد';
182
  $('mon').textContent=s0.monitor_running?'يعمل':'متوقف';
183
  $('nacc').textContent=String(tasks.length);
184
- const _nrel=String(s0.active_relays||evs.length||0);const _cap=s0.max_concurrent_relays;$('nrel').textContent=_nrel+' / '+((_cap&&_cap>0)?_cap:'مفتوح');
185
  $('ncpu').textContent=(s0.load1!=null?('load '+s0.load1+' / '+ (s0.cpus||'?')+' CPU'):'—');
186
  $('nram').textContent=(s0.mem_avail_mb!=null?(s0.mem_avail_mb+' MB متاح / '+s0.mem_total_mb+' MB'):'—');
187
  $('lasterr').textContent=lastErr;
 
181
  const lastErr=tasks.map(t=>t.last_error).filter(Boolean)[0] || (s0.logs||[]).filter(l=>/error|فشل|stall|خطأ/i.test(l.message||'')).slice(-1)[0]?.message || 'لا يوجد';
182
  $('mon').textContent=s0.monitor_running?'يعمل':'متوقف';
183
  $('nacc').textContent=String(tasks.length);
184
+ $('nrel').textContent=String(s0.active_relays||evs.length||0)+' / '+(s0.max_concurrent_relays||'?');
185
  $('ncpu').textContent=(s0.load1!=null?('load '+s0.load1+' / '+ (s0.cpus||'?')+' CPU'):'—');
186
  $('nram').textContent=(s0.mem_avail_mb!=null?(s0.mem_avail_mb+' MB متاح / '+s0.mem_total_mb+' MB'):'—');
187
  $('lasterr').textContent=lastErr;
ingest.py CHANGED
@@ -28,36 +28,19 @@ def rtmp_configured():
28
  return bool(env_rtmp_destination())
29
 
30
 
31
- def _open_cap(raw, default='0'):
32
- """0 / open / unlimited = no global cap. Hub still 0-or-1 per broadcaster."""
33
- s = str(raw if raw is not None else default).strip().lower()
34
- if s in ('', '0', 'unlimited', 'open', 'none', 'off', 'no', 'false'):
35
- return 0
36
- try:
37
- n = int(s)
38
- except (TypeError, ValueError):
39
- return 0
40
- return n if n > 0 else 0
41
-
42
-
43
  def max_concurrent_relays():
44
- """Simultaneous FFmpeg publishers. 0 = every LIVE broadcaster may copy.
45
 
46
- Hub still enforces 0 or 1 relay per username. A positive MAX_CONCURRENT_RELAYS
47
- is an optional extra cap; there is no hidden ceiling of 8.
 
48
  """
49
- raw = os.getenv('MAX_CONCURRENT_RELAYS')
50
- if raw is None or str(raw).strip() == '':
51
- raw = os.getenv('YOUTUBE_MAX_EVENTS') or '0'
52
- return _open_cap(raw, '0')
53
-
54
-
55
- def youtube_max_events():
56
- """Simultaneous YouTube events. 0 = no extra cap beyond one event per user."""
57
- raw = os.getenv('YOUTUBE_MAX_EVENTS')
58
- if raw is None or str(raw).strip() == '':
59
- return max_concurrent_relays()
60
- return _open_cap(raw, '0')
61
 
62
 
63
  def format_session_title(broadcaster, when=None):
@@ -162,12 +145,9 @@ class RelayHub:
162
 
163
  def at_capacity(self, except_user=None):
164
  with self.lock:
165
- cap = max_concurrent_relays()
166
- if cap <= 0:
167
- return False
168
  n = sum(1 for u, v in self.slots.items()
169
  if u != except_user and self._ffmpeg_alive(v))
170
- return n >= cap
171
 
172
  def env_key_busy(self, except_user=None):
173
  """A single env stream key can only accept one publisher (any owner)."""
 
28
  return bool(env_rtmp_destination())
29
 
30
 
 
 
 
 
 
 
 
 
 
 
 
 
31
  def max_concurrent_relays():
32
+ """Cap on simultaneous FFmpeg publishers. Not a quality guarantee.
33
 
34
+ Default 2: cpu-basic has carried 1 comfortably and 2 at high quota; 3+ is
35
+ not a pass. Operators may raise MAX_CONCURRENT_RELAYS; health reports the
36
+ value actually in force. Ceiling 8 is only a fork-bomb guard.
37
  """
38
+ raw = os.getenv('MAX_CONCURRENT_RELAYS') or os.getenv('YOUTUBE_MAX_EVENTS') or '2'
39
+ try:
40
+ n = int(str(raw).strip())
41
+ except (TypeError, ValueError):
42
+ n = 2
43
+ return max(1, min(n, 8))
 
 
 
 
 
 
44
 
45
 
46
  def format_session_title(broadcaster, when=None):
 
145
 
146
  def at_capacity(self, except_user=None):
147
  with self.lock:
 
 
 
148
  n = sum(1 for u, v in self.slots.items()
149
  if u != except_user and self._ffmpeg_alive(v))
150
+ return n >= max_concurrent_relays()
151
 
152
  def env_key_busy(self, except_user=None):
153
  """A single env stream key can only accept one publisher (any owner)."""
relay.py CHANGED
@@ -117,18 +117,9 @@ class Sender:
117
  stamp=d.get('last_source_at')
118
  if stamp:d['source_age']=round(max(0.0,time.monotonic()-stamp),1)
119
  try:
120
- d['send_mode']=getattr(self,'last_mode',None) or self.sticky_mode or self.mode_override or ''
121
  except Exception:
122
  d['send_mode']=''
123
- # Hide leftover TLS/EOF from a previous handshake while this FFmpeg is actually sending.
124
- try:
125
- alive=self.alive()
126
- except Exception:
127
- alive=False
128
- if alive and float(d.get('output_seconds') or 0)>=15 and float(d.get('source_age') or 99)<3:
129
- d['ffmpeg_errors']=[]
130
- d['ffmpeg_diagnostics']=[]
131
- d['source_gap']=0
132
  return d
133
  def wait_until(self,condition,stop,poll=1.0,timeout=None):
134
  """Block while `condition()` holds; returns as soon as it stops or a stop is requested."""
@@ -163,13 +154,12 @@ class Sender:
163
  FLV is exactly the "picture froze, viewers must leave the app and come
164
  back" symptom, because the player latches onto the first configuration.
165
  copy -> forward the source untouched (needs a proven H.264 source)
166
- match -> re-encode at the source size (never upscale); GOP 2s for YouTube
167
- fixed1080 -> constant 1080x1920 canvas, re-encodes (upscale — operator only)
168
- fixed720 -> constant 720x1280 canvas, re-encodes
169
  """
170
  mode=(self.mode_override or os.getenv('YOUTUBE_SEND_MODE','auto')).strip().lower()
171
- if mode not in ('auto','copy','match','fixed1080','fixed720'):
172
- raise ValueError('وضع الإرسال يجب أن يكون auto أو copy أو match أو fixed1080 أو fixed720')
173
  # Sticky transcode is for HEVC / layout flap only. Measured 2026-09-27 on cpu-basic:
174
  # two fixed720 publishers used 1.67–1.85 of 2.0 quota (peak 92.6%) with YouTube
175
  # health still good — no headroom for a third. If this candidate is H264, copy.
@@ -196,20 +186,14 @@ class Sender:
196
  now=time.monotonic()
197
  while flap and now-flap[0]>600:flap.popleft()
198
  if len(flap)>=int(os.getenv('YOUTUBE_FLAP_LIMIT','4')):
199
- fb=os.getenv('YOUTUBE_FLAP_FALLBACK_MODE','').strip().lower()
200
- # copy-first: resolution flap must not switch to transcode unless operator set fixed*.
201
- if fb in ('match','fixed1080','fixed720'):
202
- self.sticky_mode=fb
203
- mode=fb
204
  return mode
205
  def note_format_change(self):
206
  self.flap.append(time.monotonic())
207
  if len(self.flap)>=int(os.getenv('YOUTUBE_FLAP_LIMIT','4')):
208
- fb=os.getenv('YOUTUBE_FLAP_FALLBACK_MODE','').strip().lower()
209
- if fb in ('fixed1080','fixed720'):
210
- self.log('system','تغيّرت أبعاد المصدر عدة مرات خلال 10 دقائق؛ التحول إلى كانفس ثابت (%s) بطلب المشغّل'%fb)
211
- else:
212
- self.log('system','تغيّرت أبعاد المصدر عدة مرات؛ نُبقي copy ونعيد توصيل تيك توك على نفس جلسة يوتيوب')
213
  @staticmethod
214
  def copy_is_safe(declared,probed):
215
  """Stream copy is allowed only on positive proof of H.264, never on absence of evidence.
@@ -245,8 +229,8 @@ class Sender:
245
  label=str(cand.get('label') or cand.get('brief') or '')
246
  if any(k in str(declared).lower() for k in ('h264','avc')) or any(k in label.lower() for k in ('h264','avc')):
247
  return None
248
- mode=os.getenv('YOUTUBE_CODEC_FALLBACK_MODE','match').strip().lower()
249
- if mode not in ('match','fixed1080','fixed720'):mode='match'
250
  self.sticky_mode=mode
251
  why=('ffprobe: '+str(probed)) if probed else (('مُعلَن: '+declared) if declared else 'ترميز مجهول والفحص لم يأتِ بنتيجة')
252
  self.log('system','لا يمكن نسخ المصدر كما هو ('+why+') — تحويل تلقائي إل�� '+mode+
@@ -335,64 +319,15 @@ class Sender:
335
  return []
336
  return ['-rtmp_enhanced_codecs', 'hvc1'] if self.hevc_payload(candidate) else []
337
 
338
- def _match_geometry(self, candidate):
339
- """Same size as the broadcaster. Never upscale. Bitrate follows the source."""
340
- c=candidate or {}
341
- try:
342
- w=int(c.get('width') or 0); h=int(c.get('height') or 0)
343
- except (TypeError, ValueError):
344
- w=h=0
345
- if w<2 or h<2:
346
- # label like hd 1280p / origin 1920p when TikTok omitted width
347
- text=' '.join(str(c.get(k) or '') for k in ('brief','label','quality_name','resolution'))
348
- nums=[int(n) for n in __import__('re').findall(r'\d+', text)]
349
- tall=max(nums) if nums else 1280
350
- if tall>=1800: w,h=1080,1920
351
- elif tall>=1400: w,h=720,1440
352
- elif tall>=1000: w,h=720,1280
353
- else: w,h=720,1280
354
- w-=w%2; h-=h%2
355
- # YouTube useful cap; decrease only, never grow.
356
- if min(w,h)>1080 or max(w,h)>1920:
357
- if h>=w:
358
- scale=min(1080/min(w,h), 1920/max(w,h))
359
- else:
360
- scale=min(1920/max(w,h), 1080/min(w,h))
361
- w=int(w*scale); h=int(h*scale)
362
- w-=w%2; h-=h%2
363
- env=(os.getenv('VIDEO_BITRATE') or '').strip()
364
- if env:
365
- br=env
366
- else:
367
- raw=c.get('bitrate') or c.get('bitrate_kbps') or 0
368
- try:
369
- n=int(str(raw).lower().replace('k','').split('.')[0] or 0)
370
- except (TypeError, ValueError):
371
- n=0
372
- if n>20000:
373
- n=n//1000
374
- if n<400:
375
- n=int(max(w*h, 720*1280)*30*0.08/1000)
376
- n=max(1200, min(n, 6000))
377
- br='%dk'%n
378
- return w,h,br
379
  def command(self,destination,candidate=None):
380
  # إعلان Enhanced-RTMP يُقاس على الترميز المدفوع فعلًا؛ ملفات الاختبار المحلية لا بروتوكول لها.
381
  mode=self.send_mode()
382
  network=destination.startswith(('rtmp://','rtmps://'))
383
  enhanced=self.enhanced_advertisement(mode,candidate) if network else []
384
- self.last_mode=mode
385
  if mode=='copy':
386
  video=['-c:v','copy']
387
- # Repeat SPS/PPS on every IDR so a YouTube viewer who hits loading can
388
- # resync without leaving the app. Video remains copy (no re-encode).
389
- if not self.hevc_payload(candidate):
390
- video += ['-bsf:v','dump_extra=freq=keyframe']
391
  else:
392
- if mode=='match':
393
- width,height,bitrate=self._match_geometry(candidate)
394
- else:
395
- width,height,bitrate=(1080,1920,'6000k') if mode=='fixed1080' else (720,1280,'4000k')
396
  # VIDEO_BITRATE was documented in the README but never read anywhere; it now caps the
397
  # transcode path (copy mode never re-encodes, so it has no bitrate to set).
398
  bitrate=os.getenv('VIDEO_BITRATE','').strip() or bitrate
@@ -654,8 +589,6 @@ class Sender:
654
  except Exception as e:
655
  self.log(user, '[RELAY] تعذر رصد مصدر بديل دون قطع يوتيوب (%s)' % type(e).__name__)
656
  return None
657
- if not cand or not cand.get('url'):
658
- cand = getattr(self, 'current_candidate', None)
659
  if not cand or cand.get('media') == 'hls' or not cand.get('url'):
660
  return None
661
  try:
@@ -877,17 +810,10 @@ class Sender:
877
  # في HLS يستحيل القياس إلا على FFmpeg (هو من يفتح القائمة)، فنقبل تقدّمه.
878
  _fresh = (sess['ff_last'] if response is None else sess['last'])
879
  if stall_limit and not sess.get('swapping') and now - _fresh > stall_limit:
880
- attached=None
881
- for _try in range(3):
882
- attached = self._try_reattach(user, p, bridge, sess, stop, refresh,
883
- read_timeout, retries, response, pump)
884
- if attached:
885
- response, pump = attached
886
- break
887
- if stop.is_set() or (p.poll() is not None):
888
- break
889
- time.sleep(1.0)
890
  if attached:
 
891
  continue
892
  # نُفيق الخيط أولًا (قد يكون محبوسًا في write على أنبوب ممتلئ)، ثم نرفع:
893
  # إغلاق stdin يجعل FFmpeg يرى EOF ويتوقف بلطف بدل SIGKILL.
@@ -899,22 +825,6 @@ class Sender:
899
  pass
900
  raise SourceStalled('انقطع تدفق المصدر منذ %.0f ث (FFmpeg ما زال حيًا، '
901
  'ويوتيوب يبقى على آخر ثانية وصلته)' % (now - _fresh))
902
- if sess.get('format_changed') and p.poll() is None:
903
- sess['format_changed'] = False
904
- self.log(user, '[RELAY] تغيّر شكل المصدر؛ إعادة توصيل تيك توك على نفس FFmpeg — لا قطع يوتيوب')
905
- attached=None
906
- for _try in range(3):
907
- attached = self._try_reattach(user, p, bridge, sess, stop, refresh,
908
- read_timeout, retries, response, pump)
909
- if attached:
910
- response, pump = attached
911
- break
912
- if stop.is_set() or p.poll() is not None:
913
- break
914
- time.sleep(1.0)
915
- if attached:
916
- continue
917
- raise FormatChanged()
918
  rc = p.poll()
919
  if rc is not None:
920
  if sess['format_changed']:
 
117
  stamp=d.get('last_source_at')
118
  if stamp:d['source_age']=round(max(0.0,time.monotonic()-stamp),1)
119
  try:
120
+ d['send_mode']=self.sticky_mode or self.mode_override or ''
121
  except Exception:
122
  d['send_mode']=''
 
 
 
 
 
 
 
 
 
123
  return d
124
  def wait_until(self,condition,stop,poll=1.0,timeout=None):
125
  """Block while `condition()` holds; returns as soon as it stops or a stop is requested."""
 
154
  FLV is exactly the "picture froze, viewers must leave the app and come
155
  back" symptom, because the player latches onto the first configuration.
156
  copy -> forward the source untouched (needs a proven H.264 source)
157
+ fixed1080 -> constant 1080x1920 canvas, re-encodes
158
+ fixed720 -> constant 720x1280 canvas, re-encodes (lighter on CPU)
 
159
  """
160
  mode=(self.mode_override or os.getenv('YOUTUBE_SEND_MODE','auto')).strip().lower()
161
+ if mode not in ('auto','copy','fixed1080','fixed720'):
162
+ raise ValueError('وضع الإرسال يجب أن يكون auto أو copy أو fixed1080 أو fixed720')
163
  # Sticky transcode is for HEVC / layout flap only. Measured 2026-09-27 on cpu-basic:
164
  # two fixed720 publishers used 1.67–1.85 of 2.0 quota (peak 92.6%) with YouTube
165
  # health still good — no headroom for a third. If this candidate is H264, copy.
 
186
  now=time.monotonic()
187
  while flap and now-flap[0]>600:flap.popleft()
188
  if len(flap)>=int(os.getenv('YOUTUBE_FLAP_LIMIT','4')):
189
+ mode=os.getenv('YOUTUBE_FLAP_FALLBACK_MODE','fixed720').strip().lower()
190
+ if mode not in ('fixed1080','fixed720'):mode='fixed720'
191
+ self.sticky_mode=mode
 
 
192
  return mode
193
  def note_format_change(self):
194
  self.flap.append(time.monotonic())
195
  if len(self.flap)>=int(os.getenv('YOUTUBE_FLAP_LIMIT','4')):
196
+ self.log('system','تغيّرت أبعاد المصدر عدة مرات خلال 10 دقائق؛ التحول تلقائيًا إلى كانفس ثابت (%s) لمنع تكرار الانقطاع'%os.getenv('YOUTUBE_FLAP_FALLBACK_MODE','fixed720'))
 
 
 
 
197
  @staticmethod
198
  def copy_is_safe(declared,probed):
199
  """Stream copy is allowed only on positive proof of H.264, never on absence of evidence.
 
229
  label=str(cand.get('label') or cand.get('brief') or '')
230
  if any(k in str(declared).lower() for k in ('h264','avc')) or any(k in label.lower() for k in ('h264','avc')):
231
  return None
232
+ mode=os.getenv('YOUTUBE_CODEC_FALLBACK_MODE','fixed720').strip().lower()
233
+ if mode not in ('fixed1080','fixed720'):mode='fixed720'
234
  self.sticky_mode=mode
235
  why=('ffprobe: '+str(probed)) if probed else (('مُعلَن: '+declared) if declared else 'ترميز مجهول والفحص لم يأتِ بنتيجة')
236
  self.log('system','لا يمكن نسخ المصدر كما هو ('+why+') — تحويل تلقائي إل�� '+mode+
 
319
  return []
320
  return ['-rtmp_enhanced_codecs', 'hvc1'] if self.hevc_payload(candidate) else []
321
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
322
  def command(self,destination,candidate=None):
323
  # إعلان Enhanced-RTMP يُقاس على الترميز المدفوع فعلًا؛ ملفات الاختبار المحلية لا بروتوكول لها.
324
  mode=self.send_mode()
325
  network=destination.startswith(('rtmp://','rtmps://'))
326
  enhanced=self.enhanced_advertisement(mode,candidate) if network else []
 
327
  if mode=='copy':
328
  video=['-c:v','copy']
 
 
 
 
329
  else:
330
+ width,height,bitrate=(1080,1920,'6000k') if mode=='fixed1080' else (720,1280,'4000k')
 
 
 
331
  # VIDEO_BITRATE was documented in the README but never read anywhere; it now caps the
332
  # transcode path (copy mode never re-encodes, so it has no bitrate to set).
333
  bitrate=os.getenv('VIDEO_BITRATE','').strip() or bitrate
 
589
  except Exception as e:
590
  self.log(user, '[RELAY] تعذر رصد مصدر بديل دون قطع يوتيوب (%s)' % type(e).__name__)
591
  return None
 
 
592
  if not cand or cand.get('media') == 'hls' or not cand.get('url'):
593
  return None
594
  try:
 
810
  # في HLS يستحيل القياس إلا على FFmpeg (هو من يفتح القائمة)، فنقبل تقدّمه.
811
  _fresh = (sess['ff_last'] if response is None else sess['last'])
812
  if stall_limit and not sess.get('swapping') and now - _fresh > stall_limit:
813
+ attached = self._try_reattach(user, p, bridge, sess, stop, refresh,
814
+ read_timeout, retries, response, pump)
 
 
 
 
 
 
 
 
815
  if attached:
816
+ response, pump = attached
817
  continue
818
  # نُفيق الخيط أولًا (قد يكون محبوسًا في write على أنبوب ممتلئ)، ثم نرفع:
819
  # إغلاق stdin يجعل FFmpeg يرى EOF ويتوقف بلطف بدل SIGKILL.
 
825
  pass
826
  raise SourceStalled('انقطع تدفق المصدر منذ %.0f ث (FFmpeg ما زال حيًا، '
827
  'ويوتيوب يبقى على آخر ثانية وصلته)' % (now - _fresh))
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
828
  rc = p.poll()
829
  if rc is not None:
830
  if sess['format_changed']:
youtube_manager.py CHANGED
@@ -12,7 +12,7 @@ from flv_bridge import FormatChanged
12
  import inspect
13
 
14
  from youtube_api import GoogleAPI, Journal, APIError
15
- from ingest import hub, env_rtmp_destination, rtmp_configured, format_session_title, APP_VERSION, max_concurrent_relays, youtube_max_events
16
 
17
  class ManagedYouTube:
18
  def __init__(self,ffmpeg,observe,log,accounts,path,headers_for=None):
@@ -45,9 +45,8 @@ class ManagedYouTube:
45
  # إيقاع إعادة المحاولة عندما تكون تيك توك «مباشرة» بلا مرشّحات (لحظات انتقال جولات PK):
46
  # كان الحدث ينتظر دورة المشغّل كاملة (30ث) مع كاش رصد قديم، فبدا «معلّقًا على يوتيوب».
47
  self.source_retry=max(5,int(os.getenv('YOUTUBE_SOURCE_RETRY',str(min(15,self.scan)))))
48
- self.max_events=youtube_max_events()
49
  self.last_logged=None;self.status_text='متوقف'
50
- self._match_lock=set();self._gop_bad={}
51
  # Written in _select_source / _skip_note; declared here so a fresh manager never has to
52
  # rely on getattr() defaults (and so the reason survives the first skip).
53
  self._probe_note='';self._skip_last={}
@@ -56,7 +55,7 @@ class ManagedYouTube:
56
  # حتى لو كانت روابطه الموقّعة ما زالت صالحة لساعات. YOUTUBE_OBS_HOLD_SECONDS يضبط
57
  # السقف، و0 يطفئ الاحتفاظ.
58
  self.hold_seconds=max(0.0,float(os.getenv('YOUTUBE_OBS_HOLD_SECONDS','900')))
59
- self._held={};self._last_force_at={};self.hammer={};self.retry_at={};self._offline_logged=set();self._gate_note_at={};self.auth_pause={};self.starve={};self._hold_notes={};self.last_seen_status={};self._blocked403={};self._skip_probe_once={};self._skip_hold=set();self._loop_miss={};self._ingest_bad_since={};self._good_copy={};self._spawn_at={}
60
  # هل تقبل دالة الرصد المحقونة force=؟ (app.py يفعّلها؛ الاختبارات قد لا تفعل)
61
  try:self._observe_force='force' in inspect.signature(self.observe).parameters
62
  except (TypeError,ValueError):self._observe_force=False
@@ -179,19 +178,6 @@ class ManagedYouTube:
179
  light=[c for c in out if not _too_heavy(c)]
180
  if light:
181
  out=light
182
- def _short(c):
183
- try:
184
- return min(int(c.get('width') or 0), int(c.get('height') or 0))
185
- except (TypeError, ValueError):
186
- return 0
187
- # copy-first: never chase a 640/320 PK blip when a 720-class H264 FLV exists.
188
- hi=[c for c in out if _short(c)>=700]
189
- if hi:
190
- out=hi+ [c for c in out if c not in hi]
191
- out=sorted(out, key=lambda c: (
192
- 0 if str(c.get('media') or '')=='flv' else 1,
193
- -_short(c),
194
- ))
195
  return out[:limit]
196
  @staticmethod
197
  def _adult_account(user):
@@ -274,66 +260,20 @@ class ManagedYouTube:
274
  if self.is_fault_note(ev.get('note')):
275
  self.event_update(key, note='')
276
 
277
- def _403_is_hard(self,reason):
278
- s=(reason or '').lower()
279
- return any(k in s for k in (
280
- 'livestreamingnotenabled','insufficientlivepermissions','livepermissionblocked',
281
- 'youtubesignuprequired','accessnotconfigured','autherror','invalid_grant','forbidden'))
282
-
283
- def _403_info(self,key):
284
- info=self._blocked403.get(key)
285
- if info is None:
286
- return None
287
- if isinstance(info,(int,float)):
288
- return {'t':float(info),'reason':'','hard':True}
289
- return info
290
-
291
- def _403_blocked(self,key,user):
292
- info=self._403_info(key)
293
- if not info:
294
- return False
295
- hard=bool(info.get('hard'))
296
- limit=float(os.getenv('YOUTUBE_403_BLOCK_SECONDS') or (600 if hard else 90))
297
- left=int(limit-(time.monotonic()-float(info.get('t') or 0)))
298
- if left>0:
299
- why=(info.get('reason') or '?')
300
- self._skip_note(user,'يوتيوب رفض إنشاء حدث لهذه الجلسة (403/%s)؛ ننتظر %d ث قبل المحاولة من جديد'%(why,left))
301
- return True
302
- self._blocked403.pop(key,None)
303
- return False
304
-
305
- def _release_soft_403(self,except_user=None):
306
- """A finished sibling event may free a YouTube concurrent slot — retry copy, do not transcode."""
307
- drop=[]
308
- for k,info in list(self._blocked403.items()):
309
- meta=info if isinstance(info,dict) else {'hard':True}
310
- if meta.get('hard'):
311
- continue
312
- u=k.split(':',1)[0]
313
- if except_user and u==except_user:
314
- continue
315
- drop.append(k)
316
- for k in drop:
317
- self._blocked403.pop(k,None)
318
- self.log(k.split(':',1)[0],'[MONITOR] أُخلي قفل 403 المؤقت بعد تحرّر مقعد يوتيوب؛ سيُعاد فتح copy')
319
-
320
  def _youtube_rejected(self,user,room_id,reason):
321
- """N16/N16b: رفض يوتيوب 403 عند إنشاء الحدث أو تحديثه.
322
-
323
- صلاحية/حساب = قفل طويل. سقف متزامن أو خطأ عابر = قفل قصير يُفرَّغ عندما ينتهي
324
- حدث شقيق حتى يُفتح copy للمذيع المنتظر دون ترميز.
325
- """
326
  _keys=[k for k,ev in self.journal.read()['events'].items() if ev.get('username')==user and not ev.get('closing')]
327
  for _k in _keys:
328
  self.event_note(_k,'يوتيوب رفض إنشاء/تحديث الحدث (%s): تحقّق من صلاحية البث المباشر وسقف الأحداث في حسابك' % reason)
329
  try:self.finish_event(_k)
330
  except Exception:self.archive_event(_k,'local_cancelled_'+str(reason))
331
  _rid=str(room_id or '')
332
- hard=self._403_is_hard(reason)
333
- limit=float(os.getenv('YOUTUBE_403_BLOCK_SECONDS') or (600 if hard else 90))
334
- if _rid:
335
- self._blocked403[user+':'+_rid]={'t':time.monotonic(),'reason':str(reason or ''),'hard':hard}
336
- self.log(user,'[MONITOR] يوتيوب رفض إنشاء الحدث (403/%s hard=%s)؛ قفل %d ث — copy سيُعاد بعد النافذة أو بعد تحرّر مقعد'%(reason,hard,int(limit)))
337
 
338
  # ---------- channel / settings ----------
339
  def check_channel(self,allow_switch=False,expected=None,report_only=False):
@@ -380,8 +320,8 @@ class ManagedYouTube:
380
  with self.operations:
381
  d=self.journal.read()
382
  clean={k:str(values.get(k,'')).strip() for k in ('name','privacy','title','description','send_mode')}
383
- if clean['send_mode'] and clean['send_mode'] not in ('auto','copy','match','fixed1080','fixed720'):
384
- raise ValueError('وضع الإرسال يجب أن يكون auto أو copy أو match أو fixed1080 أو fixed720')
385
  if clean['privacy'] not in ('public','unlisted','private'):raise ValueError('خصوصية غير صالحة')
386
  if 'send_mode' in values:
387
  # The ingestion mode is safe to change mid-broadcast: it only decides what the
@@ -561,7 +501,9 @@ class ManagedYouTube:
561
  d=self.journal.read();key=user+':'+observation['room_id']
562
  # قفل رفض 403 يُفحص أولًا: بعد الإغلاق يبقى المفتاح في history والفحص اللاحق كان
563
  # يسمح بلوغ الحدث مرة أخرى كل دورة رفض متكررة (الرفض حصل لأن المفتاح في history).
564
- if self._403_blocked(key,user):
 
 
565
  return False
566
  if key in d['events'] or any(h['session_key']==key for h in d['history']):return False
567
  dt=datetime.now(ZoneInfo('Asia/Baghdad'));settings=d['settings']
@@ -598,7 +540,7 @@ class ManagedYouTube:
598
  body={'snippet':{'title':ev['title'],'description':ev['description'],
599
  'scheduledStartTime':(datetime.now(timezone.utc)+timedelta(seconds=30)).isoformat()},
600
  'status':dict({'privacyStatus':ev['privacy']},**({'selfDeclaredMadeForKids':False} if self._adult_account(ev['username']) else {})),
601
- 'contentDetails':{'enableAutoStart':True,'enableAutoStop':False,'enableDvr':True,
602
  'recordFromStart':True,'monitorStream':{'enableMonitorStream':False}}}
603
  try:r=self.api.call('liveBroadcasts','POST',{'part':'snippet,status,contentDetails'},body)
604
  except APIError as e:
@@ -632,9 +574,6 @@ class ManagedYouTube:
632
  def finish_event(self,key):
633
  ev=self.journal.read()['events'].get(key)
634
  if not ev:return True
635
- self._match_lock.discard(ev.get('username'))
636
- self._gop_bad.pop(key, None)
637
- self._release_soft_403(except_user=ev.get('username'))
638
  self.event_update(key,closing=True,phase='closing')
639
  if not self.stop_sender(key):return False
640
  ev=self.journal.read()['events'].get(key)
@@ -701,25 +640,11 @@ class ManagedYouTube:
701
  self.log(user or 'system', '[WATCHDOG] أُلغي تأجيل إعادة الاتصال (%d حدث) — %s' % (dropped, why))
702
  return dropped
703
 
704
-
705
- def _yt_public(self, ev):
706
- """Viewers can already open the watch URL — any delay is a YouTube loading spinner."""
707
- if not ev:
708
- return False
709
- if ev.get('was_live'):
710
- return True
711
- return str(ev.get('youtube_status') or '') in ('live', 'testing', 'liveStarting')
712
  def start_sender(self,key):
713
- ev=self.journal.read()['events'].get(key)
714
- public=self._yt_public(ev)
715
- # عقاب الوتيرة يجب أن يصمد أمام نقرات tick() قبل أن يصبح الحدث علنيًا.
716
- # بعد live أي تأخير = دائرة تحميل أمام المشاهد — نلغي القفل ونعيد copy فورًا.
717
  until=self.retry_at.get(key)
718
- if public:
719
- if until:
720
- self.retry_at.pop(key, None)
721
- self._gate_note_at.pop(key, None)
722
- until=None
723
  if until and time.monotonic()<until:
724
  left=int(round(until-time.monotonic()))
725
  # الصمت هنا هو بالضبط «بالانتظار بلا سبب» الذي شكا منه المستخدم: القفل صحيح لكن يجب
@@ -779,12 +704,10 @@ class ManagedYouTube:
779
  self.event_note(key,'مفتاح RTMPS البيئي مشغول بمذيع آخر؛ لن نخلط بثّين على نفس المفتاح')
780
  return
781
  if hub.at_capacity(except_user=ev['username']):
782
- self.log(ev['username'],'[RELAY] @%s waiting: at MAX_CONCURRENT_RELAYS=%s'%(ev['username'], max_concurrent_relays() or 'open'))
783
  return
784
  if hub.busy(ev['username']):
785
  return
786
- if (not public) and time.monotonic()-float(self._spawn_at.get(ev['username']) or 0)<8:
787
- return
788
  # لا نلمس entry هنا: بعد موت خيط الإرسال تُبقي tick() القيد الميت في self.senders،
789
  # و`entry['candidate']=None` على قيد ميت كان يرمي TypeError عند كل دورة مشغّل
790
  # (هذا تحديدًا ما جمّد حدث viggoo313 على phase=bound مع output_seconds=0)،
@@ -794,9 +717,7 @@ class ManagedYouTube:
794
  sender.set_probe(self.codec_probe)
795
  # حين تبدأ الحافة بقبولنا فعلًا يُمسح العقاب وأثره — لا يُتركان حتى أول انقطاع لاحق
796
  sender.on_stable=lambda u=ev['username']: self.mark_stream_stable(u)
797
- sender.mode_override=(os.getenv('YOUTUBE_SEND_MODE') or '').strip() or self.send_mode_setting() or None
798
- if ev.get('username') in getattr(self,'_match_lock',set()):
799
- sender.sticky_mode='match'
800
  thread=threading.Thread(target=self.sender_loop,
801
  args=(key,ev['username'],ev['room_id'],destination,stop,sender),daemon=True)
802
  self.senders[key]={'thread':thread,'stop':stop,'sender':sender}
@@ -804,7 +725,6 @@ class ManagedYouTube:
804
  self.senders.pop(key,None)
805
  self.log(ev['username'],'[RELAY] @%s skipped duplicate session'%ev['username'])
806
  return
807
- self._spawn_at[ev['username']]=time.monotonic()
808
  self.log(ev['username'], '[RESTART] جلسة إرسال جديدة للحساب (خيط إشراف أُقلع من الصفر)')
809
  thread.start()
810
  def restart_dead_senders(self):
@@ -831,8 +751,7 @@ class ManagedYouTube:
831
  if not since:
832
  entry['idle_since']=time.monotonic()
833
  continue
834
- grace=2.0 if self._yt_public(ev) else 20.0
835
- if time.monotonic()-since<grace:
836
  continue
837
  self.log(ev.get('username') or key,
838
  '[RELAY] @%s sender thread without FFmpeg; recycle'%ev.get('username'))
@@ -979,11 +898,7 @@ class ManagedYouTube:
979
  # Skip the idle wait right after a resolution change; otherwise keep polling.
980
  if restart_at and time.monotonic()-restart_at<30:restart_at=0.0;continue
981
  gap=float(getattr(sender,'last_source_gap',0) or 0)
982
- ev_now=(self.journal.read()['events'].get(key) or {})
983
- if self._yt_public(ev_now):
984
- wait=0.5
985
- else:
986
- wait=0.5 if gap>=8 else self._backoff(user)
987
  if stop.wait(wait):return
988
  def mark_stream_stable(self, user):
989
  """الحافة تقبل التدفق: تُرمى نافذة العقاب وملاحظتها وعدّادها. تُستدعى من المرسل عند أول
@@ -1005,15 +920,6 @@ class ManagedYouTube:
1005
  cleared += 1
1006
  if self.hammer.pop(user, None) is not None:
1007
  cleared += 1
1008
- try:
1009
- for e in (self.senders or {}).values():
1010
- snd=e.get('sender')
1011
- cand=getattr(snd,'current_candidate',None) if snd else None
1012
- if cand and cand.get('url'):
1013
- self._good_copy[user]=dict(cand)
1014
- break
1015
- except Exception:
1016
- pass
1017
  if cleared:
1018
  self.retry[user] = self.retry_min
1019
  self.log(user, '[WATCHDOG] الاستقبال يقبل التدفق؛ مُسح تأجيل العقاب وأثره من اللوحة (النوافذ: %d)' % cleared)
@@ -1078,21 +984,15 @@ class ManagedYouTube:
1078
  except Exception as e:
1079
  self.log(user, '[WATCHDOG] تعذر تجديد عنوان الاستقبال: ' + self.safe_error(e)[:120])
1080
  if n >= 3:
1081
- public=any(self._yt_public(ev) for ev in _evs)
1082
- if public:
1083
- # الحدث أمام المشاهد: أي تأجيل = دائرة تحميل. نعيد copy على نفس المفتاح فورًا.
1084
- for k in [k for k, ev in self.journal.read()['events'].items() if ev.get('username') == user]:
1085
- self.retry_at.pop(k, None)
1086
- self.retry[user] = self.retry_min
1087
- self.log(user, '[WATCHDOG] يوتيوب أغلق RTMPS والحدث ما زال live — إعادة copy فورًا على نفس المفتاح (بلا تأجيل %d×)'%n)
1088
- else:
1089
- wait = min(120.0, max(20.0, self.retry_min) * (2 ** (n - 3)))
1090
- self.retry[user] = max(self.retry.get(user, 0) or 0, wait)
1091
- for k in [k for k, ev in self.journal.read()['events'].items() if ev.get('username') == user]:
1092
- self.retry_at[k] = time.monotonic() + wait
1093
- self.log(user, '[WATCHDOG] الوجهة أغلقت الاتصال بعد %.0f ث (رمز %s) — المرة %d: تأجيل إعادة الاتصال إلى %d ث بدل النقر على حافة ترفض%s'
1094
- % (lasted, code, n, int(self._backoff(user)),
1095
- '؛ السبب المعلوم: الحدث لم يُشغَّل بعد' if not_started else ''))
1096
  return
1097
 
1098
  def _backoff(self,user):
@@ -1173,25 +1073,10 @@ class ManagedYouTube:
1173
  return None
1174
  if (obs or {}).get('status')!='live':
1175
  return None
1176
- good=getattr(self,'_good_copy',{}).get(user) or {}
1177
- try:
1178
- gh=min(int(good.get('width') or 0), int(good.get('height') or 0))
1179
- except (TypeError, ValueError):
1180
- gh=0
1181
- fallback=None
1182
  for c in self._select_source(user, obs) or []:
1183
- if c.get('media')=='hls' or not c.get('url'):
1184
- continue
1185
- try:
1186
- sh=min(int(c.get('width') or 0), int(c.get('height') or 0))
1187
- except (TypeError, ValueError):
1188
- sh=0
1189
- if gh>=700 and sh and sh<500:
1190
- if fallback is None:
1191
- fallback=c
1192
- continue
1193
- return c
1194
- return fallback
1195
  def _select_source(self,user,obs):
1196
  """All relayable renditions, stable first. Never drop a working lower-quality source."""
1197
  candidates=self.video_sources(obs.get('sources') or [])
@@ -1348,7 +1233,7 @@ class ManagedYouTube:
1348
  return False
1349
  if any(e.get('username')==user for e in d['events'].values()):
1350
  return False
1351
- if self.max_events>0 and len(d['events'])>=self.max_events:
1352
  return False
1353
  key=user+':'+str(obs['room_id'])
1354
  entry=next((h for h in d['history'] if h.get('session_key')==key),None)
@@ -1443,7 +1328,7 @@ class ManagedYouTube:
1443
  # 2) Discover new live sessions, one event per account, up to the concurrency cap.
1444
  cap=self.max_events;active=len(d['events']);capped=False
1445
  for user in accounts:
1446
- if cap>0 and active>=cap:capped=True;break
1447
  if any(e['username']==user for e in d['events'].values()):continue
1448
  # رفض الهوية (توكن منتهٍ) لا يُعالَج بوتيرة الرصد: كل دورة كانت تضرب تيك توك
1449
  # برفض مؤكد (~10/د على خمسة حسابات) وتُغرق السجل حتى وهو مكبوت. إيقاف لكل
@@ -1712,34 +1597,13 @@ class ManagedYouTube:
1712
 
1713
  if life in ('complete','revoked'):
1714
  self.event_update(key,closing=True);self.finish_event(key);return
1715
- if life in ('live','liveStarting'):
1716
- if life=='live' and not ev['was_live']:
1717
  self.event_update(key,was_live=True,phase='live')
1718
  self.log(user,'[LIVE] يوتيوب أكد أن الحدث مباشر: '+ev['broadcast_id'])
1719
- # liveStarting without FFmpeg = frozen/loading picture. Keep pushing copy.
1720
  self.start_sender(key)
1721
  self.stream_health(key,user,ev)
1722
- ev=self.journal.read()['events'].get(key) or ev
1723
- self._ingest_copy_watchdog(key,user,ev)
1724
- # Measured Prayer-Time-Bot: life=liveStarting, health=ok, out≈wall for 10+ min,
1725
- # public never left «starting». Re-request live while ingest is healthy.
1726
- if life=='liveStarting' and needs_poll:
1727
- healthy=(str(ev.get('youtube_health') or '') in ('good','ok'))
1728
- try:
1729
- snap=(self.senders.get(key) or {}).get('sender')
1730
- out=float((snap.snapshot() if snap else {}).get('output_seconds') or 0)
1731
- except Exception:
1732
- out=0.0
1733
- if healthy or out>=30:
1734
- if not ev.get('was_live'):
1735
- self.event_update(key,was_live=True,phase='starting')
1736
- try:
1737
- self.api.call('liveBroadcasts/transition','POST',{'part':'id,status','id':ev['broadcast_id'],'broadcastStatus':'live'})
1738
- self.event_note(key,'يوتيوب يستقبل؛ أُعيد طلب التحويل من liveStarting إلى مباشر')
1739
- self.log(user,'[LIVE] إعادة طلب live من liveStarting '+str(ev.get('broadcast_id')))
1740
- except APIError as e:
1741
- # already transitioning / not allowed — keep copy, do not tear down
1742
- self.log(user,'[LIVE] تحويل liveStarting: '+self.safe_error(e)[:120])
1743
  return
1744
 
1745
  # life is expected to be 'ready' or 'testing' for a newly bound event
@@ -1771,82 +1635,13 @@ class ManagedYouTube:
1771
  self.event_update(key, note='انتظار تأكيد استقبال يوتيوب قبل بدء الحدث')
1772
  elif not ev.get('note'):
1773
  self.event_note(key,'انتظار تأكيد استقبال يوتيوب قبل بدء الحدث')
1774
-
1775
- def _ingest_copy_watchdog(self,key,user,ev):
1776
- """Restart copy onto the same broadcast when YouTube ingest is dead but TikTok still flows.
1777
-
1778
- FFmpeg alive is not proof of ingest. noData/bad + fresh source_age → new copy sender,
1779
- same broadcast_id. Never switches to transcode.
1780
- """
1781
- health=(ev.get('youtube_health') or '')
1782
- if health in ('good','ok'):
1783
- self._ingest_bad_since.pop(key,None)
1784
- return
1785
- if health not in ('noData','bad','noStream'):
1786
- return
1787
- if time.time()-float(ev.get('created_at') or 0)<90:
1788
- return
1789
- entry=self.senders.get(key) or {}
1790
- snd=entry.get('sender')
1791
- try:
1792
- alive=bool(snd and snd.alive())
1793
- except Exception:
1794
- alive=False
1795
- if not alive:
1796
- self._ingest_bad_since.pop(key,None)
1797
- return
1798
- try:
1799
- age=(snd.snapshot() or {}).get('source_age')
1800
- except Exception:
1801
- age=None
1802
- if age is None or float(age)>3:
1803
- return
1804
- now=time.monotonic()
1805
- since=self._ingest_bad_since.get(key)
1806
- if since is None:
1807
- self._ingest_bad_since[key]=now
1808
- return
1809
- wait=float(os.getenv('YOUTUBE_NODATA_RESTART_SECONDS','90'))
1810
- if now-since<wait:
1811
- return
1812
- self._ingest_bad_since[key]=now
1813
- self.log(user,'[WATCHDOG] يوتيوب ingest=%s والمصدر يتدفق (age=%s ث)؛ إعادة copy على نفس الحدث %s'%(health,age,ev.get('broadcast_id') or ''))
1814
- self.stop_sender(key)
1815
-
1816
- def _maybe_match_from_health(self,key,user,ev):
1817
- """Copy first. One switch to source-sized transcode if YouTube GOP/ingest cannot play copy.
1818
-
1819
- No upscale. No oscillation: lock match for the rest of this process/event.
1820
- bitrateLow alone is not a reason to transcode.
1821
- """
1822
- if user in getattr(self,'_match_lock',set()):
1823
- return
1824
- issues=str(ev.get('youtube_issues') or '').lower()
1825
- health=str(ev.get('youtube_health') or '')
1826
- gop=any(k in issues for k in ('gopsize','opengop','videoingestionstarved','videocodec','videoresolution'))
1827
- if not gop and health not in ('bad','noData','noStream'):
1828
- self._gop_bad.pop(key, None)
1829
- return
1830
- if not gop:
1831
- return
1832
- n=int(self._gop_bad.get(key) or 0)+1
1833
- self._gop_bad[key]=n
1834
- need=max(2, int(os.getenv('YOUTUBE_MATCH_AFTER_HEALTH','2')))
1835
- if n<need:
1836
- return
1837
- self._match_lock.add(user)
1838
- self.log(user,'[RELAY] يوتيوب لا يقبل copy (GOP/ingest)؛ ترميز بنفس مقاس المصدر — بلا تكبير')
1839
- self.stop_sender(key)
1840
  def stream_health(self,key,user,ev):
1841
  """Read YouTube's own verdict on the inbound stream. The Live Streaming API reports
1842
  exactly why viewers buffer (videoIngestionStarved), why the picture can be rejected
1843
  (videoResolutionUnsupported / videoCodec) and why playback stalls (gopSizeLong /
1844
  openGop). Surfacing it replaces guesswork with ground truth. Polled on a slower
1845
  cadence than the life-cycle check so it stays inside the quota budget."""
1846
- prev_h=(ev.get('youtube_health') or '')
1847
- slow=max(60,int(os.getenv('YOUTUBE_HEALTH_INTERVAL','180')))
1848
- fast=max(30,int(os.getenv('YOUTUBE_HEALTH_FAST_INTERVAL','45')))
1849
- interval=fast if prev_h in ('','noData','bad','noStream') else slow
1850
  if time.time()-float(ev.get('last_health') or 0)<interval:return
1851
  stream_id=(self.journal.read()['streams'].get(user) or {}).get('id')
1852
  if not stream_id:return
@@ -1864,7 +1659,6 @@ class ManagedYouTube:
1864
  self.event_update(key,youtube_health=health.get('status',''),youtube_issues=text,last_issues=text)
1865
  if text and text!=prev:
1866
  self.log(user,'يوتيوب يشخّص البث: '+text)
1867
- self._maybe_match_from_health(key,user,self.journal.read()['events'].get(key) or ev)
1868
  def run_manager(self):
1869
  checked=False
1870
  last_tick=0.0
 
12
  import inspect
13
 
14
  from youtube_api import GoogleAPI, Journal, APIError
15
+ from ingest import hub, env_rtmp_destination, rtmp_configured, format_session_title, APP_VERSION, max_concurrent_relays
16
 
17
  class ManagedYouTube:
18
  def __init__(self,ffmpeg,observe,log,accounts,path,headers_for=None):
 
45
  # إيقاع إعادة المحاولة عندما تكون تيك توك «مباشرة» بلا مرشّحات (لحظات انتقال جولات PK):
46
  # كان الحدث ينتظر دورة المشغّل كاملة (30ث) مع كاش رصد قديم، فبدا «معلّقًا على يوتيوب».
47
  self.source_retry=max(5,int(os.getenv('YOUTUBE_SOURCE_RETRY',str(min(15,self.scan)))))
48
+ self.max_events=max(1,int(os.getenv('YOUTUBE_MAX_EVENTS') or max_concurrent_relays()))
49
  self.last_logged=None;self.status_text='متوقف'
 
50
  # Written in _select_source / _skip_note; declared here so a fresh manager never has to
51
  # rely on getattr() defaults (and so the reason survives the first skip).
52
  self._probe_note='';self._skip_last={}
 
55
  # حتى لو كانت روابطه الموقّعة ما زالت صالحة لساعات. YOUTUBE_OBS_HOLD_SECONDS يضبط
56
  # السقف، و0 يطفئ الاحتفاظ.
57
  self.hold_seconds=max(0.0,float(os.getenv('YOUTUBE_OBS_HOLD_SECONDS','900')))
58
+ self._held={};self._last_force_at={};self.hammer={};self.retry_at={};self._offline_logged=set();self._gate_note_at={};self.auth_pause={};self.starve={};self._hold_notes={};self.last_seen_status={};self._blocked403={};self._skip_probe_once={};self._skip_hold=set();self._loop_miss={}
59
  # هل تقبل دالة الرصد المحقونة force=؟ (app.py يفعّلها؛ الاختبارات قد لا تفعل)
60
  try:self._observe_force='force' in inspect.signature(self.observe).parameters
61
  except (TypeError,ValueError):self._observe_force=False
 
178
  light=[c for c in out if not _too_heavy(c)]
179
  if light:
180
  out=light
 
 
 
 
 
 
 
 
 
 
 
 
 
181
  return out[:limit]
182
  @staticmethod
183
  def _adult_account(user):
 
260
  if self.is_fault_note(ev.get('note')):
261
  self.event_update(key, note='')
262
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
263
  def _youtube_rejected(self,user,room_id,reason):
264
+ """N16/N16b: رفض يوتيوب 403 (صلاحية/سقف) عند إنشاء الحدث أو تحديثه.
265
+ ليست خطأً عابرًا: حدث planned بلا مفتاح يعني مقعدًا محروقًا من السقف وحالة
266
+ مربكة على اللوحة، وإعادة المحاولة كل دورة مسح تكرر الرفض بلا نهاية. نكتب
267
+ ملاحظة صادقة، نُنهي كل أحداث المستخدم غير المغلقة، ونقفل مفتاح الجلسة داخل
268
+ نافذة زمنية — يُعاد الاكتشاف تلقائيًا بعد معالجة الصلاحية أو انقضاء النافذة."""
269
  _keys=[k for k,ev in self.journal.read()['events'].items() if ev.get('username')==user and not ev.get('closing')]
270
  for _k in _keys:
271
  self.event_note(_k,'يوتيوب رفض إنشاء/تحديث الحدث (%s): تحقّق من صلاحية البث المباشر وسقف الأحداث في حسابك' % reason)
272
  try:self.finish_event(_k)
273
  except Exception:self.archive_event(_k,'local_cancelled_'+str(reason))
274
  _rid=str(room_id or '')
275
+ if _rid:self._blocked403[user+':'+_rid]=time.monotonic()
276
+ self.log(user,'[MONITOR] يوتيوب رفض إنشاء الحدث (403/%s)؛ أُغلق حدثه ويُمنع تكرار المفتاح %d ث — يُعاد الاكتشاف بعد معالجة الصلاحية'%(reason,int(float(os.getenv('YOUTUBE_403_BLOCK_SECONDS','600')))))
 
 
 
277
 
278
  # ---------- channel / settings ----------
279
  def check_channel(self,allow_switch=False,expected=None,report_only=False):
 
320
  with self.operations:
321
  d=self.journal.read()
322
  clean={k:str(values.get(k,'')).strip() for k in ('name','privacy','title','description','send_mode')}
323
+ if clean['send_mode'] and clean['send_mode'] not in ('auto','copy','fixed1080','fixed720'):
324
+ raise ValueError('وضع الإرسال يجب أن يكون auto أو copy أو fixed1080 أو fixed720')
325
  if clean['privacy'] not in ('public','unlisted','private'):raise ValueError('خصوصية غير صالحة')
326
  if 'send_mode' in values:
327
  # The ingestion mode is safe to change mid-broadcast: it only decides what the
 
501
  d=self.journal.read();key=user+':'+observation['room_id']
502
  # قفل رفض 403 يُفحص أولًا: بعد الإغلاق يبقى المفتاح في history والفحص اللاحق كان
503
  # يسمح بلوغ الحدث مرة أخرى كل دورة رفض متكررة (الرفض حصل لأن المفتاح في history).
504
+ block=self._blocked403.get(key)
505
+ if block and time.monotonic()-block<float(os.getenv('YOUTUBE_403_BLOCK_SECONDS','600')):
506
+ self._skip_note(user,'يوتيوب رفض إنشاء حدث لهذه الجلسة (403) مؤخرًا؛ ننتظر %d ث قبل المحاولة من جديد'%int(float(os.getenv('YOUTUBE_403_BLOCK_SECONDS','600'))-(time.monotonic()-block)))
507
  return False
508
  if key in d['events'] or any(h['session_key']==key for h in d['history']):return False
509
  dt=datetime.now(ZoneInfo('Asia/Baghdad'));settings=d['settings']
 
540
  body={'snippet':{'title':ev['title'],'description':ev['description'],
541
  'scheduledStartTime':(datetime.now(timezone.utc)+timedelta(seconds=30)).isoformat()},
542
  'status':dict({'privacyStatus':ev['privacy']},**({'selfDeclaredMadeForKids':False} if self._adult_account(ev['username']) else {})),
543
+ 'contentDetails':{'enableAutoStart':False,'enableAutoStop':False,'enableDvr':True,
544
  'recordFromStart':True,'monitorStream':{'enableMonitorStream':False}}}
545
  try:r=self.api.call('liveBroadcasts','POST',{'part':'snippet,status,contentDetails'},body)
546
  except APIError as e:
 
574
  def finish_event(self,key):
575
  ev=self.journal.read()['events'].get(key)
576
  if not ev:return True
 
 
 
577
  self.event_update(key,closing=True,phase='closing')
578
  if not self.stop_sender(key):return False
579
  ev=self.journal.read()['events'].get(key)
 
640
  self.log(user or 'system', '[WATCHDOG] أُلغي تأجيل إعادة الاتصال (%d حدث) — %s' % (dropped, why))
641
  return dropped
642
 
 
 
 
 
 
 
 
 
643
  def start_sender(self,key):
644
+ # عقاب الوتيرة يجب أن يصمد أمام نقرات tick(): أي أن إعادة الاتصال المؤجّلة بسبب رفض
645
+ # الوجهة (187) لا تُبطلها دورة رصد جاءت بعد 15 ث. بدون هذا القفل يبقى التأجيل شكليًا
646
+ # لأن tick/drive_event يستدعيان start_sender كل دورة.
 
647
  until=self.retry_at.get(key)
 
 
 
 
 
648
  if until and time.monotonic()<until:
649
  left=int(round(until-time.monotonic()))
650
  # الصمت هنا هو بالضبط «بالانتظار بلا سبب» الذي شكا منه المستخدم: القفل صحيح لكن يجب
 
704
  self.event_note(key,'مفتاح RTMPS البيئي مشغول بمذيع آخر؛ لن نخلط بثّين على نفس المفتاح')
705
  return
706
  if hub.at_capacity(except_user=ev['username']):
707
+ self.log(ev['username'],'[RELAY] @%s waiting: at MAX_CONCURRENT_RELAYS=%d'%(ev['username'], max_concurrent_relays()))
708
  return
709
  if hub.busy(ev['username']):
710
  return
 
 
711
  # لا نلمس entry هنا: بعد موت خيط الإرسال تُبقي tick() القيد الميت في self.senders،
712
  # و`entry['candidate']=None` على قيد ميت كان يرمي TypeError عند كل دورة مشغّل
713
  # (هذا تحديدًا ما جمّد حدث viggoo313 على phase=bound مع output_seconds=0)،
 
717
  sender.set_probe(self.codec_probe)
718
  # حين تبدأ الحافة بقبولنا فعلًا يُمسح العقاب وأثره — لا يُتركان حتى أول انقطاع لاحق
719
  sender.on_stable=lambda u=ev['username']: self.mark_stream_stable(u)
720
+ sender.mode_override=self.send_mode_setting() or None
 
 
721
  thread=threading.Thread(target=self.sender_loop,
722
  args=(key,ev['username'],ev['room_id'],destination,stop,sender),daemon=True)
723
  self.senders[key]={'thread':thread,'stop':stop,'sender':sender}
 
725
  self.senders.pop(key,None)
726
  self.log(ev['username'],'[RELAY] @%s skipped duplicate session'%ev['username'])
727
  return
 
728
  self.log(ev['username'], '[RESTART] جلسة إرسال جديدة للحساب (خيط إشراف أُقلع من الصفر)')
729
  thread.start()
730
  def restart_dead_senders(self):
 
751
  if not since:
752
  entry['idle_since']=time.monotonic()
753
  continue
754
+ if time.monotonic()-since<20:
 
755
  continue
756
  self.log(ev.get('username') or key,
757
  '[RELAY] @%s sender thread without FFmpeg; recycle'%ev.get('username'))
 
898
  # Skip the idle wait right after a resolution change; otherwise keep polling.
899
  if restart_at and time.monotonic()-restart_at<30:restart_at=0.0;continue
900
  gap=float(getattr(sender,'last_source_gap',0) or 0)
901
+ wait=0.5 if gap>=8 else self._backoff(user)
 
 
 
 
902
  if stop.wait(wait):return
903
  def mark_stream_stable(self, user):
904
  """الحافة تقبل التدفق: تُرمى نافذة العقاب وملاحظتها وعدّادها. تُستدعى من المرسل عند أول
 
920
  cleared += 1
921
  if self.hammer.pop(user, None) is not None:
922
  cleared += 1
 
 
 
 
 
 
 
 
 
923
  if cleared:
924
  self.retry[user] = self.retry_min
925
  self.log(user, '[WATCHDOG] الاستقبال يقبل التدفق؛ مُسح تأجيل العقاب وأثره من اللوحة (النوافذ: %d)' % cleared)
 
984
  except Exception as e:
985
  self.log(user, '[WATCHDOG] تعذر تجديد عنوان الاستقبال: ' + self.safe_error(e)[:120])
986
  if n >= 3:
987
+ # أرضية 20 ث × تصاعد هندسي: retry_min=1 في الصورة كان يجعل الطلعة الثالثة «2 ث»
988
+ # (لا شيء يتغيّر فعليًا)، لذا الحدّ الأدنى 20 ث ثم 40 و80 حتى 120.
989
+ wait = min(120.0, max(20.0, self.retry_min) * (2 ** (n - 3)))
990
+ self.retry[user] = max(self.retry.get(user, 0) or 0, wait)
991
+ for k in [k for k, ev in self.journal.read()['events'].items() if ev.get('username') == user]:
992
+ self.retry_at[k] = time.monotonic() + wait # يقفل start_sender من tick() نفسه
993
+ self.log(user, '[WATCHDOG] الوجهة أغلقت الاتصال بعد %.0f ث (رمز %s) — المرة %d: تأجيل إعادة الاتصال إلى %d ث بدل النقر على حافة ترفض%s'
994
+ % (lasted, code, n, int(self._backoff(user)),
995
+ '؛ السبب المعلوم: الحدث لم يُشغَّل بعد' if not_started else ''))
 
 
 
 
 
 
996
  return
997
 
998
  def _backoff(self,user):
 
1073
  return None
1074
  if (obs or {}).get('status')!='live':
1075
  return None
 
 
 
 
 
 
1076
  for c in self._select_source(user, obs) or []:
1077
+ if c.get('media')!='hls' and c.get('url'):
1078
+ return c
1079
+ return None
 
 
 
 
 
 
 
 
 
1080
  def _select_source(self,user,obs):
1081
  """All relayable renditions, stable first. Never drop a working lower-quality source."""
1082
  candidates=self.video_sources(obs.get('sources') or [])
 
1233
  return False
1234
  if any(e.get('username')==user for e in d['events'].values()):
1235
  return False
1236
+ if len(d['events'])>=self.max_events:
1237
  return False
1238
  key=user+':'+str(obs['room_id'])
1239
  entry=next((h for h in d['history'] if h.get('session_key')==key),None)
 
1328
  # 2) Discover new live sessions, one event per account, up to the concurrency cap.
1329
  cap=self.max_events;active=len(d['events']);capped=False
1330
  for user in accounts:
1331
+ if active>=cap:capped=True;break
1332
  if any(e['username']==user for e in d['events'].values()):continue
1333
  # رفض الهوية (توكن منتهٍ) لا يُعالَج بوتيرة الرصد: كل دورة كانت تضرب تيك توك
1334
  # برفض مؤكد (~10/د على خمسة حسابات) وتُغرق السجل حتى وهو مكبوت. إيقاف لكل
 
1597
 
1598
  if life in ('complete','revoked'):
1599
  self.event_update(key,closing=True);self.finish_event(key);return
1600
+ if life=='live':
1601
+ if not ev['was_live']:
1602
  self.event_update(key,was_live=True,phase='live')
1603
  self.log(user,'[LIVE] يوتيوب أكد أن الحدث مباشر: '+ev['broadcast_id'])
1604
+ # Now safe to ensure sender is running; viewer already watching
1605
  self.start_sender(key)
1606
  self.stream_health(key,user,ev)
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1607
  return
1608
 
1609
  # life is expected to be 'ready' or 'testing' for a newly bound event
 
1635
  self.event_update(key, note='انتظار تأكيد استقبال يوتيوب قبل بدء الحدث')
1636
  elif not ev.get('note'):
1637
  self.event_note(key,'انتظار تأكيد استقبال يوتيوب قبل بدء الحدث')
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1638
  def stream_health(self,key,user,ev):
1639
  """Read YouTube's own verdict on the inbound stream. The Live Streaming API reports
1640
  exactly why viewers buffer (videoIngestionStarved), why the picture can be rejected
1641
  (videoResolutionUnsupported / videoCodec) and why playback stalls (gopSizeLong /
1642
  openGop). Surfacing it replaces guesswork with ground truth. Polled on a slower
1643
  cadence than the life-cycle check so it stays inside the quota budget."""
1644
+ interval=max(60,int(os.getenv('YOUTUBE_HEALTH_INTERVAL','180')))
 
 
 
1645
  if time.time()-float(ev.get('last_health') or 0)<interval:return
1646
  stream_id=(self.journal.read()['streams'].get(user) or {}).get('id')
1647
  if not stream_id:return
 
1659
  self.event_update(key,youtube_health=health.get('status',''),youtube_issues=text,last_issues=text)
1660
  if text and text!=prev:
1661
  self.log(user,'يوتيوب يشخّص البث: '+text)
 
1662
  def run_manager(self):
1663
  checked=False
1664
  last_tick=0.0