Spaces:
Running
Running
| """One FFmpeg push per live session. Multiple senders run concurrently, one per YouTube event.""" | |
| import fcntl, os, queue, re, select, threading, time, subprocess | |
| from collections import deque | |
| from urllib.parse import urlparse | |
| from curl_cffi import requests | |
| from flv_bridge import FLVBridge | |
| from stream_monitor import log_event, note_rtmp_open | |
| from telemetry import ( | |
| checkpoint as telemetry_checkpoint, | |
| finish as telemetry_finish, | |
| http_status_from_exception, | |
| observe as telemetry_observe, | |
| record_event as telemetry_event, | |
| start_session as telemetry_start, | |
| ) | |
| def source_http_headers(url, user=None, headers_for=None): | |
| """Headers for a signed media URL without leaking TikTok login cookies to a CDN. | |
| Browsers do not send a tiktok.com session cookie to tiktokcdn-us.com/byteoversea | |
| hosts. The relay used to do so manually, which exposed the cookie cross-domain and | |
| could also change how a CDN edge handles an otherwise-public signed media URL. | |
| Only a real tiktok.com host may receive the account headers; signed CDN URLs get the | |
| Referer alone. | |
| """ | |
| out={'Referer':'https://www.tiktok.com/'} | |
| try:host=(urlparse(str(url or '')).hostname or '').lower().rstrip('.') | |
| except Exception:host='' | |
| if host=='tiktok.com' or host.endswith('.tiktok.com'): | |
| extra=(headers_for or (lambda _u:{}))(user) or {} | |
| out.update({str(k):str(v) for k,v in extra.items() if v and str(k).lower()!='host'}) | |
| return out | |
| # أخطاء FFmpeg ← وصف عربي. قائمة واحدة على مستوى الوحدة حتى تُختبر على سطور FFmpeg حقيقية | |
| # بدل أن تُدفن في إغلاق داخل send(): الإغلاق هو ما أخفى أن كلمة المفتاح كُتبت بالخطأ. | |
| FFMPEG_ERROR_LABELS = [ | |
| ('Application defined reason', 'يوتيوب أرسل سبب الإغلاق من خادمه (رسالة RTMP)'), | |
| ('NetStream.Publish', 'يوتيوب رفض النشر على هذا المفتاح (NetStream.Publish)'), | |
| ('Publish stream', 'رفض خادم الاستقبال أمر النشر'), | |
| ('Connection refused', 'رفض اتصال خادم الاستقبال'), | |
| ('Connection timed out', 'انتهت مهلة اتصال خادم الاستقبال'), | |
| ('Connection reset by peer', 'خادم الاستقبال قطع الاتصال أثناء الإرسال'), | |
| ('Broken pipe', 'انقطع أنبوب الإرسال إلى خادم الاستقبال'), | |
| ('End of file', 'خادم الاستقبال أغلق الاتصال'), | |
| ('Immediate exit requested', 'طُلب إنهاء FFmpeg قبل اكتمال الإرسال'), | |
| ('Packet mismatch', 'عدم تطابق حزم الفيديو — تغيّر مقاس المصدر داخل الجلسة'), | |
| ('Input/output error', 'خطأ إدخال/إخراج في الإرسال'), | |
| ('Publishing stream at bitrates', 'يوتيوب قَبِل النشر فعلًا — فالقطع لاحقًا ليس رفض مفتاح'), | |
| ('Server error, Publish', 'خادم يوتيوب رفض أمر النشر على هذا المفتاح'), | |
| ('Server error', 'خادم الاستقبال أعاد خطأ'), | |
| ('Invalid data found', 'بيانات المصدر غير صالحة للقراءة'), | |
| ('Video codec', 'عدم توافق ترميز الفيديو مع الحاوية'), | |
| ('Invalid argument', 'أحد إعدادات FFmpeg غير مقبول'), | |
| # FFmpeg يكتب Non-monotonous لا Non-monotonic؛ النمط الخاطئ كان يجعل «لم يظهر سبب مصنّف» | |
| # عبارة كاذبة، لأن مشكلة التوقيت كانت تُرمى بصمت. | |
| ('Non-monotono', 'توقيت الفيديو غير متزايد (DTS) في التدفق الخارج'), | |
| ('input buffer overflow', 'امتلأ مخزن FFmpeg: المصدر أسرع من الإرسال أو الاتصال أبطأ'), | |
| ('Error writing packet', 'تعذّرت كتابة حزمة على الاتصال'), | |
| ('Unknown encoder', 'مرمز غير متوفر'), | |
| ('Option not found', 'خيار FFmpeg غير مدعوم'), | |
| ] | |
| INFO_KEEP = ('publish', 'reason', 'handshak', 'connect', '_error', 'reject', 'closing', | |
| 'input/output', 'broken pipe', 'reset by peer', 'timed out', 'refused', 'server error', | |
| 'unable to open') | |
| def _error_interesting(text): | |
| """ما يستحق البقاء من سطور FFmpeg: كل خطأ أو تحذير، أو سطر اتصال مفيد أثناء التشخيص | |
| القائمة بيضاء عمدًا: مع `info` تفيض سطور FFmpeg بالترويسات والروابط، والسجل يجب أن يبقى | |
| مقروءًا. وكل ما يبقى يمرّ بـ redact_stderr أولًا (يُمحى الرابط والمفتاح). | |
| """ | |
| low = (text or '').lower() | |
| if 'error' in low or 'warning' in low: | |
| return True | |
| return any(w in low for w in INFO_KEEP) | |
| def classify_ffmpeg_error(text): | |
| """أول وصف ينطبق على سطر من stderr، أو None. تصنيف نصي خالص بلا أي روابط.""" | |
| low = (text or '').lower() | |
| if 'netstream.publish.start' in low or 'fcpublish stream...' in low: | |
| return None # Successful/debug publish status is not a rejection. | |
| for needle, label in FFMPEG_ERROR_LABELS: | |
| if needle.lower() in low: | |
| return label | |
| return None | |
| def ffmpeg_output_failure(text): | |
| """True only for stderr evidence that FFmpeg's publishing side failed. | |
| This distinction matters: once RTMPS stops consuming, FFmpeg stops reading stdin and | |
| the source meter ages too. Calling that a TikTok stall creates a reconnect storm and | |
| never refreshes the YouTube destination or normalizes the feed. | |
| """ | |
| low=(text or '').lower() | |
| if 'netstream.publish.start' in low or 'fcpublish stream...' in low: | |
| return False | |
| return (any(x in low for x in ( | |
| 'error opening output', 'error writing', 'broken pipe', 'netstream.publish', | |
| 'server error, publish')) or | |
| (('[out#' in low or 'publish stream' in low) and | |
| any(x in low for x in ('error', 'failed', 'rejected', 'denied', 'unable', 'broken pipe'))) or | |
| # FFmpeg 7.1 can report an RTMPS write timeout through the output filter | |
| # instead of [out#]. The exact production line was: | |
| # [vf#0:0] Error sending frames to consumers: Connection timed out | |
| # Missing this mark made sender_loop blame/switch a healthy TikTok FLV | |
| # to HLS even though the failed side was YouTube. Keep the filter/output | |
| # prefix requirement so an HTTP/HLS *input* timeout is not misclassified. | |
| (('[out#' in low or '[vf#' in low or '[af#' in low) and | |
| any(x in low for x in ('error sending frames to consumers', | |
| 'connection timed out', 'connection reset by peer')))) | |
| def redact_stderr(text): | |
| """يُبقي سبب الخطأ ويخفي كل ما لا يجوز قراءته في سجل عام: الروابط والمفاتيح والرموز.""" | |
| t = re.sub(r'\b(?:rtmps?|rtsp|srt|https?)://\S+', '«رابط-مخفي»', text or '') | |
| # FFmpeg debug prints the RTMP playpath WITHOUT its URL scheme. | |
| t = re.sub(r"(?i)((?:publish|play) command for\s+)'[^']*'", r"\1'«مفتاح-مخفي»'", t) | |
| t = re.sub(r'(?i)(\b(?:fname|playpath)\s*=\s*)[^,\s]+', r'\1«مفتاح-مخفي»', t) | |
| t = re.sub(r'(?i)([?&=]|^|\s)((?:key|token|auth|sessionid|sign)\s*=\s*)[^\s&]+', r'\1\2«مخفي»', t) | |
| return t.strip()[:240] | |
| class SourceStalled(RuntimeError): | |
| """المصدر صامت: لا بايت من تيك توك منذ مهلة الحارس، أو لم يتقدّم FFmpeg. | |
| يُرفع بدل التعليق إلى الأبد لأن FFmpeg الحيّ لا يعني بثًا حيًا: العملية تنتظر stdin | |
| ويوتيوب يبقى مستقبِلًا على آخر ما وصل. يلتقطه sender_loop فيعيد رصد LIVE ويبني جلسة | |
| جديدة من الصفر. | |
| """ | |
| class RelayStartError(RuntimeError): | |
| """لم تبدأ الجلسة أصلًا (FFmpeg لم يولد أو المصدر رفض قبل التدفق). | |
| تُميَّز عن SourceStalled لأن العلاج مختلف: هنا يجب أن تُجرَّب تندرة أخرى/وضع آخر | |
| بدل إعادة رصد نفس القائمة فورًا — وبكل حال لا تبقى عملية نصفية. | |
| """ | |
| class Sender: | |
| """Single YouTube RTMPS push for one session. Process and progress are instance-scoped, | |
| so several senders can run at the same time without sharing state.""" | |
| def __init__(self,ffmpeg,log,headers_for=None): | |
| self.ffmpeg=ffmpeg;self.log=log | |
| self.headers_for=headers_for or (lambda user:{}) | |
| self.lock=threading.RLock();self.process=None | |
| self.data={'status':'','source':'','output_seconds':0,'last_output_at':None} | |
| self.last_exit_code=None # آخر رمز خروج FFmpeg (من آخر جلسة)، يقرؤه المشرف | |
| self.last_errors=[] # آخر سطور stderr مصنّفة ومُخفًى منها المفتاح: لماذا أقفلت الوجهة؟ | |
| self.on_stable=None # مستقبِل المدير: أول ≥25 ث مُسلَّمة = العقاب انتهى | |
| self.on_sticky=None # Persist a per-event format fallback before this FFmpeg is restarted. | |
| self.on_sticky_clear=None # Persist the verified return to copy (journal must not re-apply transcode). | |
| self.sticky_operator_locked=False # An operator-forced transcode is never auto-probed back to copy. | |
| self.on_sticky_pending=None | |
| self.pending_sticky=None # (mode, reason) while the bounded encoder slots are occupied. | |
| self.last_diagnostics=[] # آخر الأوصاف العربية المصنَّفة، تُعرض في اللوحة كما في السجل | |
| self.last_source_gap=0.0 # صمت المصدر لحظة التفكيك: يفصل جوع المصدر عن رفض الوجهة | |
| self.last_failure_side='' # 'source' or 'output'; stderr/backpressure outranks an aged meter | |
| self.format_unstable=False # رأى الجسر فاصل إطارات مفتاحية مضطربًا: لا COPY على هذا المصدر حتى يبرد الترميز | |
| self.trace_rtmp=False # One failed startup enables redacted handshake diagnostics. | |
| self.active_mode='' # exact FFmpeg mode, safe to expose in operational logs/state | |
| self.telemetry_key=None # Append-only diagnostics only; never used for control decisions. | |
| self.telemetry_context={} # Safe manager context (event/room labels), set before send(). | |
| # Timestamps of source resolution changes seen by this sender (anti-flap window). | |
| self.flap=deque() | |
| self._format_fallback_last_attempt=0.0 | |
| # Once the source proved to flap, stay on a constant canvas for the rest of the | |
| # event. Without this the flap window expires after 10 quiet minutes and the | |
| # sender silently drops back to copy mode, restarting the whole failure loop. | |
| self.sticky_mode=None | |
| self.sticky_since=0.0 | |
| self.copy_recovery_fails=0 | |
| self._recovering_copy=False | |
| self._last_light_output=0.0 | |
| self._gop_w=0 | |
| self._gop_h=0 | |
| self._no_copy_recovery=False | |
| # Candidate handed to send_mode() so the codec rule can judge the real source. | |
| self.current_candidate=None | |
| # Journal setting wins over the environment so the mode can be changed from the UI | |
| # without a factory rebuild (env vars cannot be written from inside the Space). | |
| self.mode_override=None | |
| # حالة الجلسة الجارية: يقرأها terminate() ليفرض إنهاءً فوريًا بدل انتظار مهلة، | |
| # ويقرأها snapshot() لتظهر «منذ كم ثانية وصل آخر بايت» في الواجهة. | |
| self.session=None | |
| self.data.update(source_bytes=0,last_source_at=None,stalled=False) | |
| def set(self,**kw): | |
| with self.lock:self.data.update(kw) | |
| def snapshot(self): | |
| with self.lock: | |
| d=dict(self.data) | |
| stamp=d.get('last_source_at') | |
| if stamp:d['source_age']=round(max(0.0,time.monotonic()-stamp),1) | |
| try: | |
| # active_mode is the exact mode used by this sender's current/last FFmpeg, | |
| # unlike the UI preference ("auto") or a global default (which can differ per account). | |
| d['send_mode']=self.active_mode or self.sticky_mode or self.mode_override or '' | |
| d['pending_send_mode']=self.pending_sticky[0] if self.pending_sticky else '' | |
| except Exception: | |
| d['send_mode']='' | |
| d['pending_send_mode']='' | |
| return d | |
| def wait_until(self,condition,stop,poll=1.0,timeout=None): | |
| """Block while `condition()` holds; returns as soon as it stops or a stop is requested.""" | |
| started=time.monotonic() | |
| while not stop.is_set(): | |
| try: | |
| if not condition():return True | |
| except Exception:return True | |
| if timeout is not None and time.monotonic()-started>timeout:return False | |
| time.sleep(poll) | |
| return True | |
| def alive(self): | |
| """True while a live FFmpeg process belongs to this sender.""" | |
| with self.lock:p=self.process | |
| return bool(p and p.poll() is None) | |
| def terminate(self): | |
| with self.lock: | |
| p=self.process;sess=self.session | |
| # إن كان هناك من معلّق على read من تيك توك فإنهاء FFmpeg وحده لا يوقظه: نغلق | |
| # الاستجابة ونُعلم الجلسة أن الإيقاف مقصود (لا إعادة محاولة داخل الخيط القارئ). | |
| if sess is not None:sess['force_stop']=True | |
| if p and p.poll() is None: | |
| try:p.terminate() | |
| except OSError:pass | |
| def send_mode(self): | |
| """Policy for what YouTube receives: | |
| auto -> copy a positively identified H.264 or HEVC source. HEVC is carried as | |
| Enhanced RTMP hvc1; no video decoder/encoder is inserted. | |
| copy -> forward supported encoded video access units untouched | |
| fixed1080 -> constant 1080x1920 canvas, re-encodes | |
| fixed720 -> constant 720x1280 canvas, re-encodes (lighter on CPU) | |
| """ | |
| mode=(self.mode_override or os.getenv('YOUTUBE_SEND_MODE','auto')).strip().lower() | |
| if mode not in ('auto','copy','fixed1080','fixed720','matchgop'): | |
| raise ValueError('وضع الإرسال يجب أن يكون auto أو copy أو fixed1080 أو fixed720 أو matchgop') | |
| # Optional legacy transcode fallback remains operator-selectable, but ordinary AVC/HEVC | |
| # decoder-record replacement never selects it; the image defaults to byte-only copy. | |
| if self.sticky_mode: | |
| # Do NOT drop LIGHT just because the next TikTok label says h264. | |
| # That path produced COPY with output_seconds=0 on the same ingest | |
| # (LIGHT healthy ~45s → copy restart → out=0). Recovery is gated. | |
| if self._may_attempt_copy_recovery(): | |
| self._recovering_copy=True | |
| return 'copy' | |
| return self.sticky_mode | |
| # Stream copy supports legacy AVC and Enhanced-RTMP HEVC. Before trusting TikTok's | |
| # label, the caller may hand us the candidate it is about to send (current_candidate). | |
| if mode in ('auto','copy') and self.current_candidate is not None: | |
| # Copy still needs positive codec evidence — see copy_is_safe(). | |
| fb=self.allow_codec_fallback() | |
| if fb:return fb | |
| if mode=='auto': | |
| # مصدر رأينا فاصل إطاراته يتأرجح (0.4-4 ث) أو خطه الزمني يقفز: النسخ يمرّر | |
| # هذا الاضطراب إلى يوتيوب كما هو (شرطه الرسمي GOP ~2 ث)، والتطبيع يزيل | |
| # المتغيّر كله. القفل لجلسة هذا الناشر؛ لا يمس اختيارًا صريحًا من المشغّل. | |
| if self.format_unstable: | |
| return 'matchgop' | |
| # Proven H.264 and a stable source: forward it untouched. | |
| return 'copy' if not self.sticky_mode else self.sticky_mode | |
| return mode | |
| def _copy_supported(candidate): | |
| """Cheap, local evidence that byte copy is even possible for this source. | |
| Supported means positively AVC or HEVC (Enhanced RTMP carries hvc1); | |
| an unknown/other codec is never probed blind. This reads only the | |
| candidate's own labels — no network probe is spent on a mode decision. | |
| """ | |
| cand=candidate or {} | |
| text=(str(cand.get('codec') or '')+' '+str(cand.get('label') or '')+' '+ | |
| str(cand.get('brief') or '')) | |
| return Sender.copy_is_safe(text,'') | |
| def _may_attempt_copy_recovery(self): | |
| """LIGHT→COPY probe after a stable light run, for AVC *and* HEVC sources. | |
| The probe is verified in send(): output that survives the threshold | |
| clears the sticky mode everywhere (including the manager journal); | |
| a probe that produces nothing re-enters the transcode and counts a | |
| failure, so a bad source cannot flap the publisher forever. | |
| Disable only explicitly with YOUTUBE_COPY_RECOVERY=0. | |
| """ | |
| if getattr(self,'sticky_operator_locked',False): | |
| return False | |
| if getattr(self,'format_unstable',False): | |
| # GOP/DTS مضطرب موثّق بالجسر: فحص COPY هنا يعيد فتح جرح أغلقه التطبيع. | |
| # الاستثناء الوحيد: حاكم البطء يمسح القفل بنفسه قبل طلب الفحص (انظر send). | |
| return False | |
| if str(os.getenv('YOUTUBE_COPY_RECOVERY','1')).strip().lower() in ('0','false','off','no'): | |
| return False | |
| try:max_fails=max(0,int(os.getenv('YOUTUBE_COPY_RECOVERY_MAX_FAILS','2'))) | |
| except (TypeError,ValueError):max_fails=2 | |
| if self.copy_recovery_fails>=max_fails: | |
| return False | |
| try:need=max(30.0, float(os.getenv('YOUTUBE_COPY_RECOVERY_SECONDS','60'))) | |
| except (TypeError,ValueError):need=60.0 | |
| if not self.sticky_since or (time.monotonic()-self.sticky_since)<need: | |
| return False | |
| if float(self._last_light_output or 0)<20: | |
| return False | |
| return self._copy_supported(self.current_candidate) | |
| def request_sticky(self, mode, user, reason): | |
| """Per-broadcaster: stick LIGHT/matchgop and end this FFmpeg only — not the Space.""" | |
| mode=(mode or 'matchgop').strip().lower() | |
| if mode not in ('matchgop','fixed720','fixed1080'): mode='matchgop' | |
| if self.sticky_mode==mode and self.session: | |
| return False | |
| if not self.sticky_mode: | |
| from ingest import hub | |
| if not hub.reserve_light(user): | |
| self.log(user, '[HEALTH] skip sticky %s — transcode slots full (%s)'%(mode, reason[:120])) | |
| return False | |
| self.sticky_mode=mode | |
| self.sticky_since=time.monotonic() | |
| self._recovering_copy=False | |
| self._no_copy_recovery=True | |
| self.pending_sticky=None | |
| cb=getattr(self,'on_sticky',None) | |
| if cb: | |
| try:cb(user,mode,int(self._gop_w or 0),int(self._gop_h or 0)) | |
| except Exception as e:self.log(user,'[HEALTH] could not persist sticky mode: '+str(e)[:80]) | |
| self.log(user, '[HEALTH] %s — sticky %s (this publisher only)'%(reason[:160], mode)) | |
| sess=self.session | |
| if isinstance(sess, dict): | |
| sess['force_stop']=True | |
| p=self.process | |
| if p is not None and p.stdin: | |
| try: p.stdin.close() | |
| except (OSError, ValueError): pass | |
| return True | |
| def _bridge_event(self,user,event,details): | |
| """Persist a redacted source event, then let the sender policy react to dimensions.""" | |
| try:log_event(user,event,details) | |
| except Exception:pass | |
| # Telemetry mirrors the already-redacted operational event only. Its | |
| # public functions are fail-open and non-blocking; no result is read | |
| # back into any streaming decision. | |
| telemetry_event(self.telemetry_key,user,event,details) | |
| if event=='BITRATE_CHANGE' and isinstance(details,dict): | |
| telemetry_observe(self.telemetry_key,{'source_bitrate_kbps':details.get('new_kbps')}) | |
| elif event=='STREAM_CONFIG' and isinstance(details,dict): | |
| telemetry_observe(self.telemetry_key,{ | |
| 'source_width':details.get('width'),'source_height':details.get('height')}) | |
| if event=='RESOLUTION_CHANGE': | |
| self.note_format_change(user,details) | |
| elif event=='KEYFRAME_CHANGE': | |
| # الجسر قاس فاصل الإطارات المفتاحية يتغيّر جوهريًا: مصدر غير مستقر الصيغة. | |
| # تُقفل عودة COPY لهذا الناشر ويُطبَّع الخرج — يوتيوب يريد GOP ثابتًا ~2 ث. | |
| self.format_unstable=True | |
| def note_format_change(self,user='system',details=None): | |
| """A bounded anti-flap policy: copy remains default until real resolution changes recur. | |
| The old counter was dead code: FLVBridge emitted RESOLUTION_CHANGE but no sender ever | |
| called this method. Once the configured threshold is reached, only that broadcaster's | |
| copy publisher is restarted into a constant-canvas encoder; the YouTube event is reused. | |
| """ | |
| now=time.monotonic() | |
| try:window=max(30.0,float(os.getenv('YOUTUBE_FLAP_WINDOW_SECONDS','600'))) | |
| except (TypeError,ValueError):window=600.0 | |
| while self.flap and now-self.flap[0]>window:self.flap.popleft() | |
| self.flap.append(now) | |
| try:limit=max(1,int(os.getenv('YOUTUBE_FLAP_LIMIT','4'))) | |
| except (TypeError,ValueError):limit=4 | |
| if len(self.flap)<limit or self.sticky_mode: | |
| return False | |
| active=(self.active_mode or self.sticky_mode or self.mode_override or | |
| os.getenv('YOUTUBE_SEND_MODE','auto')).strip().lower() | |
| if active not in ('auto','copy'): | |
| return False | |
| # If transcode slots are full, preserve the current live copy stream and retry on a later | |
| # genuine resolution transition; do not evict another broadcaster or stop this event. | |
| if now-self._format_fallback_last_attempt<5.0: | |
| return False | |
| self._format_fallback_last_attempt=now | |
| mode=os.getenv('YOUTUBE_FLAP_FALLBACK_MODE','matchgop').strip().lower() | |
| if mode not in ('fixed1080','fixed720','matchgop'):mode='matchgop' | |
| d=details or {} | |
| reason=('source resolution changed %d times within %.0fs (last %sx%s→%sx%s)'%( | |
| len(self.flap),window,d.get('old_width','?'),d.get('old_height','?'), | |
| d.get('new_width','?'),d.get('new_height','?'))) | |
| if not self.request_sticky(mode,user,reason): | |
| if not self.sticky_mode and self.pending_sticky is None: | |
| self.pending_sticky=(mode,reason) | |
| cb=getattr(self,'on_sticky_pending',None) | |
| if cb: | |
| try:cb(user,mode,reason) | |
| except Exception as e:self.log(user,'[HEALTH] could not persist pending format fallback: '+str(e)[:80]) | |
| return False | |
| log_event(user,'FORMAT_FALLBACK',{'mode':mode,'changes_in_window':len(self.flap), | |
| 'window_seconds':int(window),'reason':'repeated_resolution_change'}) | |
| return True | |
| def apply_pending_sticky(self,user): | |
| """Take a free encoder slot later without stopping/evicting another broadcaster.""" | |
| pending=self.pending_sticky | |
| if not pending:return False | |
| mode,reason=pending | |
| from ingest import hub | |
| if not hub.reserve_light(user):return False | |
| if self.request_sticky(mode,user,reason): | |
| log_event(user,'FORMAT_FALLBACK',{'mode':mode,'reason':'resolution_change_pending_slot'}) | |
| return True | |
| if not self.sticky_mode: | |
| hub.release_light(user) | |
| else: | |
| self.pending_sticky=None | |
| return False | |
| def copy_is_safe(declared,probed): | |
| """Allow byte copy only with positive, non-conflicting AVC/HEVC evidence. | |
| FFmpeg 7's FLV path preserves replacement decoder records for both codecs. HEVC output | |
| additionally uses the Enhanced-RTMP ``hvc1`` advertisement; unknown codecs and a label / | |
| probe disagreement remain unsafe rather than being guessed. | |
| """ | |
| def norm(x): | |
| x=str(x or '').strip().lower() | |
| if any(c in x for c in ('h264','avc')):return 'h264' | |
| if any(c in x for c in ('h265','hevc','hvc1')):return 'h265' | |
| return 'other' if x else '' | |
| d,q=norm(declared),norm(probed) | |
| if q in ('h264','h265'):return d in ('',q) | |
| if q=='other':return False | |
| return d in ('h264','h265') | |
| def allow_codec_fallback(self): | |
| """Decide from evidence once, then stick to it for the whole event.""" | |
| if str(os.getenv('YOUTUBE_AUTO_TRANSCODE','0')).strip().lower() in ('0','false','off','no'):return None | |
| cand=self.current_candidate | |
| if cand is None:return None | |
| found=self.live_probe(cand) | |
| probed=(found or {}).get('video') | |
| declared=str(cand.get('codec') or '') | |
| if self.copy_is_safe(declared,probed):return None | |
| label=str(cand.get('label') or cand.get('brief') or '') | |
| if any(k in str(declared).lower() for k in ('h264','avc')) or any(k in label.lower() for k in ('h264','avc')): | |
| return None | |
| try: | |
| from ingest import hub | |
| if not hub.light_available(): | |
| self.log('system','المصدر يحتاج ترميز لكن خانات CPU ممتلئة — نترك هذا المرشّح') | |
| return None | |
| except Exception: | |
| pass | |
| mode=os.getenv('YOUTUBE_CODEC_FALLBACK_MODE','matchgop').strip().lower() | |
| if mode not in ('fixed1080','fixed720','matchgop'):mode='matchgop' | |
| self.sticky_mode=mode | |
| self.sticky_since=time.monotonic() | |
| self._recovering_copy=False | |
| why=('ffprobe: '+str(probed)) if probed else (('مُعلَن: '+declared) if declared else 'ترميز مجهول والفحص لم يأتِ بنتيجة') | |
| self.log('system','لا يمكن نسخ المصدر كما هو ('+why+') — تحويل تلقائي إلى '+mode+ | |
| '؛ هذا يمنع تجمّد الصورة عند المشاهدين عند تغيّر المقاس أو في جولات PK') | |
| return mode | |
| def set_probe(self,probe): | |
| """Injected by the manager: probe(candidate) -> {'video':..,'audio':..} or None.""" | |
| self.probe_source=probe | |
| def live_probe(self,candidate): | |
| """FFmpeg's own view of the source, not the label TikTok sent us. | |
| None means "no opinion" (ffprobe unavailable, network dead): the caller must then trust | |
| the declared codec rather than treat an unknown as a mismatch and burn CPU needlessly. | |
| """ | |
| probe=getattr(self,'probe_source',None) | |
| if not probe:return None | |
| try:return probe(candidate) | |
| except Exception:return None | |
| def lock_gold(self, candidate=None, w=0, h=0): | |
| """Pin the YouTube output size to the first picture the player already latched. | |
| A later TikTok PK/layout hop must pad into this canvas. Changing width/height on | |
| an already-open live event is exactly the flash that only a YouTube-app restart | |
| clears for existing viewers. | |
| """ | |
| if int(getattr(self, '_gop_w', 0) or 0) >= 32: | |
| return (self._gop_w, self._gop_h) | |
| if candidate and not (w and h): | |
| try: | |
| # _select_source attaches dimensions parsed from the real FLV decoder record. | |
| # Prefer them over TikTok's rendition label, which can remain at an earlier | |
| # SD size while the served picture is already HD. Explicit w/h still wins. | |
| pw,ph=int(candidate.get('probed_width') or 0),int(candidate.get('probed_height') or 0) | |
| if pw>=32 and ph>=32: | |
| w,h=pw,ph | |
| else: | |
| w = int(candidate.get('width') or 0) | |
| h = int(candidate.get('height') or 0) | |
| except (TypeError, ValueError): | |
| w = h = 0 | |
| try: | |
| w, h = int(w or 0), int(h or 0) | |
| except (TypeError, ValueError): | |
| w = h = 0 | |
| w, h = w - w % 2, h - h % 2 | |
| if w >= 32 and h >= 32: | |
| w, h = self.youtube_safe_size(w, h) | |
| self._gop_w, self._gop_h = w, h | |
| # v2.43: tall pictures (h/w >= 1.7, e.g. 1:2 phone 640x1280 / 720x1440 / 960x1920) | |
| # are filled+cropped into the 9:16 canvas; others keep the letterbox pad. | |
| self._gop_fill = bool(h >= w and (h / float(w)) >= 1.7) | |
| return (int(getattr(self, '_gop_w', 0) or 0), int(getattr(self, '_gop_h', 0) or 0)) | |
| def youtube_safe_size(w, h): | |
| """Map 720-class sizes to a YouTube-healthy 720×1280 canvas. | |
| 720×1440 is 720-class (~1 Mbps) but YouTube treats height 1440 as 1440p | |
| and then reports videoIngestionStarved (measured a3dam_111). 640×1280 | |
| copy/matchgop was Studio black. True 1080 (short side ≥1080) is kept. | |
| """ | |
| try: | |
| w, h = int(w or 0), int(h or 0) | |
| except (TypeError, ValueError): | |
| return 0, 0 | |
| if w < 32 or h < 32: | |
| return 0, 0 # TikTok may begin with a 16x16 placeholder configuration. | |
| short, long_ = min(w, h), max(w, h) | |
| portrait = h >= w | |
| if short <= 720 and long_ >= 1280: | |
| return (720, 1280) if portrait else (1280, 720) | |
| return w - w % 2, h - h % 2 | |
| def canvas(width,height,fill=False): | |
| # eval=frame: when TikTok hops size mid-session the filter keeps the *output* | |
| # WxH (and therefore the x264 SPS) unchanged. Default eval=init would reinit | |
| # the encoder → new SPS → YouTube app flash until reopen. | |
| if fill: | |
| # v2.43: a tall 1:2 phone picture in a 9:16 canvas was pillarboxed to 640 of 720 | |
| # px. Fill the width and centre-crop the excess height instead (output WxH unchanged). | |
| return ('scale=w=%d:h=%d:force_original_aspect_ratio=increase:flags=lanczos:eval=frame,' | |
| 'crop=%d:%d:(iw-ow)/2:(ih-oh)/2,setsar=1,format=yuv420p'%(width,height,width,height)) | |
| return ('scale=w=%d:h=%d:force_original_aspect_ratio=decrease:flags=lanczos:eval=frame,' | |
| 'pad=%d:%d:(ow-iw)/2:(oh-ih)/2:color=black:eval=frame,setsar=1,format=yuv420p'%(width,height,width,height)) | |
| def input_args(self,candidate): | |
| """How FFmpeg *reads* the source — two transports, one sender. | |
| An FLV rendition is pushed through our stdin (FLVBridge validates and rewrites the | |
| header), while an HLS rendition is opened by FFmpeg itself, which is what makes the | |
| .m3u8 sources that rank_sources() used to throw away actually sendable: a playlist is | |
| text, so piping its bytes into `-i pipe:0` could only ever fail. FFmpeg's hls demuxer | |
| also fetches each segment on its own, so a CDN edge hiccup no longer kills the encoder. | |
| """ | |
| # Drop corrupt packets and regenerate PTS at the demuxer instead of feeding the decoder | |
| # broken frames: a viewer that hits a source hiccup recovers on its own instead of | |
| # freezing until they reconnect. (Checked on the image's FFmpeg 7.1.5: `-resync` is NOT | |
| # available — "Unrecognized option 'resync'" — so it must not be used; `discardcorrupt` | |
| # is the supported equivalent and is verified working in verify.sh step 17.) | |
| resync=['-fflags','+genpts+discardcorrupt'] if str(os.getenv('YOUTUBE_INPUT_RECOVERY','1')).strip().lower() in ('1','on','true','yes') else [] | |
| if (candidate or {}).get('media')=='hls': | |
| url=(candidate or {}).get('url') or '' | |
| # -reconnect belongs to the HTTP protocol layer: FFmpeg refuses it outright on a | |
| # non-network input ("Option reconnect not found"), so it is applied only where it | |
| # is both legal and wanted. | |
| net=['-reconnect','1','-reconnect_streamed','1','-reconnect_delay_max','5'] if url.startswith(('http://','https://')) else [] | |
| return {'stdin':None,'input':[ | |
| *net,*resync,'-live_start_index',os.getenv('HLS_START_INDEX','-3'), | |
| *(('-headers',self.headers_blob(candidate)) if candidate.get('user') else ()), | |
| '-i',url]} | |
| return {'stdin':subprocess.PIPE,'input':['-rw_timeout','20000000', | |
| '-fflags','+genpts','-i','pipe:0']} | |
| def headers_blob(self,candidate): | |
| """FFmpeg HTTP headers, applying the same cross-domain cookie boundary as FLV.""" | |
| c=candidate or {} | |
| items=source_http_headers(c.get('url'),c.get('user'),self.headers_for) | |
| return ''.join('%s: %s\r\n'%(k,v) for k,v in items.items() if v) | |
| def codec_unusable(candidate): | |
| """True when the declared codec is neither AVC nor Enhanced-RTMP HEVC.""" | |
| codec=str((candidate or {}).get('codec') or '').strip().lower() | |
| if not codec:return False | |
| return not any(c in codec for c in ('h264','avc','264','h265','hevc','hvc1','265')) | |
| def hevc_payload(candidate): | |
| """هل ما سيدفعه FFmpeg فعلًا HEVC؟ الإعلان عن Enhanced-RTMP يُقاس على هذا لا على الوضع.""" | |
| codec = str((candidate or {}).get('codec') or '').strip().lower() | |
| label = str((candidate or {}).get('label') or '').strip().lower() | |
| return any(k in codec for k in ('hevc', 'h265', 'hvc1')) or any(k in label for k in ('hevc', 'h265')) | |
| def enhanced_advertisement(self, mode, candidate): | |
| """قائمة خيارات FFmpeg لإعلان fourCcLive (Enhanced RTMP) — فارغة على تدفق H.264. | |
| `rtmp_enhanced_codecs` advertises fourCcLive in the RTMP connect command. It is emitted | |
| only when the bytes actually carried are HEVC; advertising hvc1 on an AVC-only source | |
| previously caused receiver-side rejection. ``YOUTUBE_ENHANCED_RTMP=off`` remains an | |
| emergency compatibility override. | |
| """ | |
| switch = os.getenv('YOUTUBE_ENHANCED_RTMP', 'auto').strip().lower() | |
| if switch in ('0', 'off', 'no', 'legacy'): | |
| return [] | |
| if mode != 'copy': | |
| return [] # إعادة ترميز إلى h264 دائمًا: لا شيء معزّز لنُعلنه | |
| if os.getenv('YOUTUBE_VIDEO_MODE', 'original').strip().lower() != 'original': | |
| return [] | |
| return ['-rtmp_enhanced_codecs', 'hvc1'] if self.hevc_payload(candidate) else [] | |
| def _dual_legs(destination,extra_destinations): | |
| """وجهات النشر المتزامن الآمنة للمروحة: الأساسي ثم الاحتياطي، بلا مكرّر ولا | |
| فواصل tee داخل أي عنوان. النشر المزدوج هو آلية يوتيوب الرسمية للوفرة: | |
| بثّان متزامنان لنفس الحدث، فإذا مات الأساسي أكمل الاحتياطي أمام المشاهدين | |
| بدل أن يبدأ الناشر من الصفر فيُرفض عند الفتح (187) لدقائق — حادثة | |
| 2026-10-21 21:19Z: موت 152 ثم ست جلسات 187 وبطاقة noData لـ3د42ث.""" | |
| if str(os.getenv('YOUTUBE_DUAL_INGEST','1')).strip().lower() in ('0','off','false','no'): | |
| return [] | |
| legs=[] | |
| for url in [destination]+list(extra_destinations or []): | |
| url=str(url or '') | |
| if (url.startswith(('rtmp://','rtmps://')) and url not in legs | |
| and '|' not in url and ']' not in url): | |
| legs.append(url) | |
| return legs if len(legs)>=2 else [] | |
| def _tee_spec(legs,enhanced): | |
| """مواصفة tee بساق FLV لكل وجهة: onfail=ignore يعني أن ساقًا ميتة (رفض فتح | |
| أو قطع منتصف البث) تُسقَط وحدها وتكمل أختها — مُختبَر محليًا على ffmpeg | |
| 7.0.2: قتل خادم RTMP بعد 4ث من بث 8ث فأكملت ساق الملف 8.02ث كاملة ورجع | |
| الناشر 0. مهلة كتابة قصيرة لكل ساق (8ث بدل 20) حتى لا تُسمّم ساق محجوبة | |
| أختها عبر خيط المروحة المشترك أكثر من ثوانٍ قبل أن تُسقَط.""" | |
| try:rw=str(int(os.getenv('YOUTUBE_DUAL_RW_TIMEOUT','8000000'))) | |
| except (TypeError,ValueError):rw='8000000' | |
| opts='f=flv:onfail=ignore:flvflags=no_duration_filesize' | |
| if enhanced:opts+=':rtmp_enhanced_codecs=hvc1' | |
| opts+=':rw_timeout='+rw | |
| return '|'.join('[%s]%s'%(opts,url) for url in legs) | |
| def command(self,destination,candidate=None,extra_destinations=None): | |
| # إعلان Enhanced-RTMP يُقاس على الترميز المدفوع فعلًا؛ ملفات الاختبار المحلية لا بروتوكول لها. | |
| mode=self.send_mode() | |
| self.active_mode=mode | |
| network=destination.startswith(('rtmp://','rtmps://')) | |
| enhanced=self.enhanced_advertisement(mode,candidate) if network else [] | |
| if mode=='copy': | |
| video=['-c:v','copy'] | |
| else: | |
| if mode=='matchgop': | |
| self.lock_gold(candidate) | |
| if not self._gop_w or not self._gop_h: | |
| # A 16x16 startup header is not the player's real canvas. Unknown | |
| # dimensions still need a constant output, not a reinitializing scale. | |
| self._gop_w,self._gop_h=720,1280 | |
| # CPU guard: re-encoding a 960x1920 (or larger) picture at full | |
| # size is what saturates a cpu-basic Space while the source is | |
| # only 720-class in bitrate. The fallback canvas is capped at | |
| # 720p-class; an explicit fixed1080 operator mode is not capped. | |
| if str(os.getenv('YOUTUBE_TRANSCODE_CAP','1')).strip().lower() not in ('0','off','false','no'): | |
| try:cap_short=max(144,int(os.getenv('YOUTUBE_TRANSCODE_MAX_SHORT','720'))) | |
| except (TypeError,ValueError):cap_short=720 | |
| try:cap_long=max(256,int(os.getenv('YOUTUBE_TRANSCODE_MAX_LONG','1280'))) | |
| except (TypeError,ValueError):cap_long=1280 | |
| w,h=int(self._gop_w),int(self._gop_h) | |
| if min(w,h)>cap_short or max(w,h)>cap_long: | |
| scale=min(cap_short/min(w,h),cap_long/max(w,h)) | |
| self._gop_w=max(2,int(w*scale)//2*2);self._gop_h=max(2,int(h*scale)//2*2) | |
| vf=self.canvas(self._gop_w,self._gop_h,fill=bool(getattr(self,'_gop_fill',False))) | |
| bitrate=os.getenv('VIDEO_BITRATE','').strip() or '4500k' | |
| else: | |
| width,height,bitrate=(1080,1920,'6000k') if mode=='fixed1080' else (720,1280,'4000k') | |
| vf=self.canvas(width,height) | |
| bitrate=os.getenv('VIDEO_BITRATE','').strip() or bitrate | |
| bufsize=str(int(int(str(bitrate).rstrip('kK'))*1.5))+'k' | |
| video=['-vf',vf,'-c:v','libx264','-preset',os.getenv('YOUTUBE_X264_PRESET','veryfast'), | |
| '-tune','zerolatency','-profile:v','high', | |
| # Live Control Room requires CBR. `-crf 22 -b:v/-maxrate` was only capped | |
| # CRF: a measured static 720x1280 input produced 14 kbit/s although FFmpeg's | |
| # stream header claimed 4500 kbit/s, and YouTube reported | |
| # videoIngestionStarved for both live relays. minrate + nal-hrd=cbr makes | |
| # x264 emit filler when a simple frame needs few bits (measured 4444 kbit/s), | |
| # keeping the ingest clock fed rather than merely raising a nominal field. | |
| '-b:v',bitrate,'-minrate',bitrate,'-maxrate',bitrate,'-bufsize',bufsize, | |
| # IDR at least every 2 s: a viewer who stalls (or joins) re-syncs on the next | |
| # keyframe instead of waiting for the source's own GOP, which is what made | |
| # people leave the YouTube app and come back to get a picture again. | |
| '-force_key_frames','expr:gte(t,n_forced*2)', | |
| '-g',os.getenv('YOUTUBE_GOP','60'),'-keyint_min','30','-sc_threshold','0','-bf','0', | |
| '-x264-params','nal-hrd=cbr:force-cfr=1:open-gop=0:scenecut=0:keyint=%s:min-keyint=30'%os.getenv('YOUTUBE_GOP','60'), | |
| '-pix_fmt','yuv420p','-fps_mode','cfr', | |
| '-colorspace','bt709','-color_primaries','bt709','-color_trc','bt709'] | |
| # Network output goes through the fifo pseudo-muxer instead of straight to flv. | |
| # fifo separates encoding from the RTMPS connection and reconnects transparently | |
| # when YouTube drops it, rather than killing the whole FFmpeg process. That is | |
| # exactly why the local recording kept running while the broadcast died: writing | |
| # to disk never fails, a network blip does, and with -f flv it was fatal. | |
| # Documented: https://ffmpeg.org/ffmpeg-formats.html#fifo | |
| # recover_any_error is deliberately NOT set, so permanent faults (bad stream key, | |
| # rejected codec) still fail fast and stay visible in the log instead of retrying | |
| # forever in silence. | |
| if network: | |
| # timeshift يؤخّر كل شيء عن الوجهة بمقدار ثابت. فائدته: سكّات المصدر تُبتلع بدل أن | |
| # تُجيع حافة يوتيوب، والحزم تبقى محفوظة عبر إعادة الاتصال. كلفته — وهي ما لم يكن | |
| # موثّقًا: أول بايت لا يصل قبل انقضاء المهلة كلها. مقيسٌ على أمر الشبكة نفسه بوجهة | |
| # ملف وتغذية 1 ميجابايت/ث: ts=10 ⇒ أول نمو للمخرج عند 10.0 ث، وts=0 ⇒ عند 1.4 ث. | |
| # وفي الإنتاج الحافة تقفل بعد 13-20 ث، فلا تبقى نافذة صالحة تكفي لتقول «good». | |
| # الافتراضي 0 إذًا. وتاريخًا (N26): كنّا نغلف الخرج بـfifo حتى مع ts=0 اعتمادًا على | |
| # استعادة الاتصال الشفافة — لكن ffmpeg 7.1 (صورة trixie الإنتاجية) غيّر دلالة | |
| # attempt_recovery: فشل فتح الوجهة (مفتاح فاسد/رفض/انقطاع DNS) صار يُعاد محاولته | |
| # إلى ما لا نهاية ⇒ عملية زومبي تلتهم المصدر وتُبلغ out_time وهميًا من المخزن | |
| # (مقيس: 84ث «مُسلَّمة» إلى rtmps://example.invalid في 40ث — ولا CLEANUP أبدًا). | |
| # لذا مع ts=0 خرج مباشر -f flv: يفشل بصدق فورًا على كل الإصدارات، والمدير يعيد | |
| # بناء الجلسة بدورة [RESTART] المجرَّبة. من يريد الدرع ضد سكّات المصدر يرفع | |
| # YOUTUBE_FIFO_TIMESHIFT=10 ويعرف أنه يدفعه تأخيرًا ويبقى معه الـfifo بكل أدواته. | |
| ts=int(os.getenv('YOUTUBE_FIFO_TIMESHIFT','0')) | |
| if ts>0: | |
| # queue_size MUST grow with timeshift. Its default is 60 packets, which holds | |
| # under a second at ~77 pkt/s (30 fps video + ~47 fps AAC); once timeshift | |
| # delays the output the queue fills and, with drop_pkts_on_overflow=1, the | |
| # stream is destroyed rather than delayed. Measured: ts=3 with the default | |
| # queue yields 1 frame out of 480; ts=3 with 5000 and ts=10 with 20000 both | |
| # yield byte-identical output to the no-fifo case. | |
| q=max(500,int(os.getenv('YOUTUBE_FIFO_QUEUE','0')) or (ts*250+500)) | |
| output=['-f','fifo','-fifo_format','flv','-format_opts','flvflags=no_duration_filesize', | |
| # الإسقاط عند الامتلاء يبقى مفعّلًا: البث حيّ، وإعادة إرسال مخزن قديم أسوأ | |
| # من فجوة صغيرة — والاحتباس هنا يرتدّ إلى stdin فيتوقف ضخّ المصدر ويصمت | |
| # مؤقّت الحارس، فتُتهم تيك توك بجرم الشبكة. | |
| '-drop_pkts_on_overflow','1','-attempt_recovery','1', | |
| '-restart_with_keyframe','1', | |
| '-recovery_wait_time',os.getenv('YOUTUBE_RECOVERY_WAIT','1'), | |
| '-queue_size',str(q),'-timeshift',str(ts)] | |
| else: | |
| # v2.41: بلا timeshift ومع وجهة احتياطية معروفة، الخرج مروحة tee إلى | |
| # الاستقبالين معًا (انظر _dual_legs) — الوفرة عند يوتيوب نفسه لا عند | |
| # إعادة فتحنا. بلا احتياطي يبقى خرج N26 المباشر كما هو: تقدم out_time | |
| # يعني تسليمًا فعليًا، بلا مخزن يكذب على الحارس. | |
| legs=self._dual_legs(destination,extra_destinations) | |
| if legs: | |
| # Each tee leg opens its own FLV muxer. With x264, global extradata | |
| # makes both YouTube ingests receive the same SPS/PPS from the start; | |
| # without it YouTube can report resolutionMismatch / Bad secondary | |
| # stream and change the displayed aspect ratio (FFmpeg ticket #11527). | |
| # Copy already carries the bridge's decoder record and is unchanged. | |
| if mode!='copy':video+=['-flags:v','+global_header'] | |
| output=['-f','tee',self._tee_spec(legs,enhanced)] | |
| else: | |
| output=['-flvflags','no_duration_filesize','-f','flv'] | |
| else: | |
| output=['-flvflags','no_duration_filesize','-f','flv'] | |
| ta=self.input_args(candidate) | |
| # مع المروحة تسكن الوجهات وإعلان hvc1 ومهلة الكتابة داخل مواصفة كل ساق؛ | |
| # إلحاق الوجهة المفردة أو الخيارات العامة بعد tee يُفسد الأمر كله. | |
| tail=list(output) if (network and output[:1]==['-f'] and 'tee' in output) else \ | |
| [*enhanced,'-rw_timeout','20000000',*output,destination] | |
| return [self.ffmpeg,'-hide_banner','-loglevel',('debug' if network and self.trace_rtmp else os.getenv('YOUTUBE_FFMPEG_LOGLEVEL','error')), | |
| '-nostats','-progress','pipe:1', | |
| *ta['input'],'-map','0:v:0','-map','0:a:0?', | |
| # TikTok streams HE-AAC (SBR) at 64 kbit/s — measured on the live source. Copied | |
| # into FLV it is what players "recover" from by reconnecting, and YouTube's | |
| # transcoder re-uses whatever AAC config the first packet declared. Re-encoding audio | |
| # costs ~nothing and removes the whole class of stalls; video stays untouched in copy. | |
| *video,'-c:a','aac','-profile:a',os.getenv('YOUTUBE_AUDIO_PROFILE','aac_low'), | |
| '-b:a',os.getenv('YOUTUBE_AUDIO_BITRATE','128k'),'-ar','48000','-ac',os.getenv('YOUTUBE_AUDIO_CHANNELS','2'), | |
| '-af','aresample=async=1:first_pts=0','-avoid_negative_ts','make_zero', | |
| *tail] | |
| # ------- مهل المصدر (كلها من البيئة؛ القيم الافتراضية مبرَّرة أدناه) ------- | |
| def stall_limit(self): | |
| """حدّ صمت المصدر بالثواني (0 = الحارس مطفأ) — تُعرض في اللوحة مع عمر آخر بايت.""" | |
| return self._stall_seconds() | |
| def _stall_seconds(self): | |
| """كم ثانية بلا بايت واحد من المصدر = المصدر متوقف (يوتيوب يبقى موصولًا واهمًا). | |
| 12ث: حافة يوتيوب تقفل بعد ~14-20ث صمت؛ عتبة 45 كانت تستيقظ بعد أن يتجمّد المشاهد. | |
| سكّات PK أقصر من هذا. 0 = بلا حارس. | |
| """ | |
| # 20 ث لا 45: مقيسٌ أن حافة يوتيوب تقفل النشر بعد 14-20 ث من الصمت، فعتبة أعلى تعني | |
| # أن نستيقظ بعد أن يكون هو قطع الجلسة (ونحصل على 187 بدل سبب حقيقي). أدنى من هذا تبدأ | |
| # إعادة اتصال كل سكّة صغيرة، وصمت تيك توك حول جولة PK يقيس ثوانٍ لا عشرات الثواني. | |
| raw = os.getenv('SOURCE_STALL_TIMEOUT') | |
| if raw is None or raw == '': | |
| raw = os.getenv('YOUTUBE_SOURCE_STALL_SECONDS', '12') | |
| return max(0.0, float(raw)) | |
| def _fill_output_grace(): | |
| """كم ثانية بلا تقدّم خرج فعلي أثناء الحشو قبل إعلان انسداد RTMPS.""" | |
| try: | |
| return min(45.0, max(10.0, float(os.getenv('YOUTUBE_FILL_OUTPUT_GRACE_SECONDS', '25')))) | |
| except (TypeError, ValueError): | |
| return 25.0 | |
| def _fill_output_stale(sess, now, grace, output_seconds=-1): | |
| """True when a fill is running but FFmpeg output has stopped advancing. | |
| A healthy frozen fill still makes the encoder/muxer emit progress, so | |
| no progress for the whole grace window means the planned silence is | |
| hiding a blocked RTMPS output rather than keeping a live leg warm. | |
| Sessions that have not produced any output yet are left to the normal | |
| startup/progress watchdog. | |
| """ | |
| if not sess or not sess.get('filling') or output_seconds < 0: | |
| return False | |
| since = float(sess.get('fill_since') or 0) | |
| ff_last = float(sess.get('ff_last') or 0) | |
| if not since or not ff_last or now - since < grace: | |
| return False | |
| if now - ff_last > grace: | |
| return True | |
| # كاتب الحشو نفسه شاهد: لو لم تُكتب قطعة حشو كاملة منذ السماحية فالأنبوب | |
| # محجوب/الخيط ميت، حتى لو بدت «ثواني» FFmpeg حديثة بسبب قفزة طوابع. | |
| emit = float(sess.get('fill_emit_at') or 0) | |
| if emit and now - emit > grace: | |
| return True | |
| # تقدّم خرج أسرع بكثير من الزمن الحقيقي أثناء الحشو ليس تسليمًا حيًا: | |
| # إنه قفزة خطّ زمني (خياطة فاسدة) يتجاهلها يوتيوب كـ noData بينما يبدو | |
| # العدّاد المحلي سعيدًا. لا نمنحه حصانة الحشو. | |
| start = sess.get('fill_out_start') | |
| if start is not None and float(output_seconds) - float(start) > (now - since) + 15.0: | |
| return True | |
| return False | |
| def _fill_write_renew_allowed(sess): | |
| """Whether a blocked stdin write may renew its deadline during fill.""" | |
| if not sess or not sess.get('filling') or sess.get('force_stop'): | |
| return False | |
| now = time.monotonic() | |
| until = float(sess.get('fill_until') or 0) | |
| if until and now >= until: | |
| return False | |
| grace = Sender._fill_output_grace() | |
| ff_last = float(sess.get('ff_last') or 0) | |
| if ff_last and now - ff_last > grace: | |
| return False | |
| emit = float(sess.get('fill_emit_at') or 0) | |
| if emit and now - emit > grace: | |
| return False | |
| return True | |
| def _deadline_base(sess): | |
| """مرجع مهلة كتابة stdin: آخر بايت مصدر عادةً. أثناء حشو فجوة أو تبديل مصدر | |
| فالصمت مخطّط له والضخّ قد يكون من خيط الحشو — المرجع يصير آخر تقدّم فعلي | |
| من FFmpeg (ff_last)، فلا يُقتل خرج يعمل بتهمة «عالق» بينما المُرمِّز يطحن | |
| إطارات مجمّدة أو يخيط مصدرًا جديدًا (مقيس: talolil 00:21:50 و00:30:49). | |
| خرج محجوب حقًا يجمّد التقدّم أيضًا، فتمضي المهلة ويموت بنظافة كالعادة.""" | |
| base=float(sess.get('last') or 0) | |
| now=time.monotonic() | |
| # لحام ما بعد الحشو/الخياطة: طحن FFmpeg لخياطة مصدر يمتد عبر اللحظة الفاصلة | |
| # بين نهاية حشو وبداية التالي (مقيس 2026-10-14 13:06:13: انتهى حشو ناجح، | |
| # عاد المصدر ثانيتين، صمت، فماتت المضخّة بمهلة من وصول القطعة لا من آخر | |
| # دليل ابتلاع). طالما آخر كتابة حشو ناجحة أو خياطة ضمن مهلة التوقّف، فآخر | |
| # دليل ابتلاع هو المرجع — وخرج محجوب حقًا يجمّد الأدلة كلها فتمضي المهلة. | |
| seam=max(float(sess.get('fill_emit_at') or 0),float(sess.get('reattach_at') or 0)) | |
| in_seam=bool(seam and now-seam<12.0) | |
| if sess.get('filling') or sess.get('swapping') or in_seam: | |
| base=max(base,float(sess.get('ff_last') or 0)) | |
| if sess.get('filling') or in_seam: | |
| base=max(base,float(sess.get('fill_emit_at') or 0)) | |
| return base | |
| def _stream_tuning(self): | |
| """ما يعادله في هذه الشجرة لخيارات Streamlink (المشروع لا يستعمل Streamlink أصلًا): | |
| --stream-timeout → read_timeout (curl_cffi يرمي ReadTimeout بدل التعليق) | |
| --stream-segment-attempts → retries (محاولات لكل قراءة قبل إعلان التوقف) | |
| --stream-segment-timeout → total_timeout (سقف الجلسة كاملة: لا تعليق بلا نهاية) | |
| لا تُمرَّر أي خيار لبرنامج غير مثبت؛ لا streamlink في requirements.txt ولا في الصورة. | |
| """ | |
| return (max(5.0, float(os.getenv('YOUTUBE_SOURCE_READ_TIMEOUT', '15'))), | |
| max(1, int(os.getenv('YOUTUBE_SOURCE_READ_ATTEMPTS', '3'))), | |
| max(30.0, float(os.getenv('YOUTUBE_SESSION_MAX_SECONDS', '21600')))) | |
| def _write_deadline(stdin, buf, deadline, sess=None): | |
| """write على stdin مع مهلة: أنبوب ممتلئ يجب ألّا يُعلّق الخيط إلى الأبد. | |
| الوضع المحجوز وحده يجعل «المصدر صامت + FFmpeg لا يقرأ» حالة لا تنتهي: الخيط يبقى | |
| داخل write، والحارس لا يرى جديدًا لأن آخر بايت سُجِّل قبل الانسداد. هنا نكتب بقدر ما | |
| يقبل الأنبوب، وننتظر بفارغ الصبر حتى deadline، ثم نعلن التوقف فيستيقظ الحارس. | |
| """ | |
| fd = stdin.fileno() | |
| flags = fcntl.fcntl(fd, fcntl.F_GETFL) | |
| try: | |
| fcntl.fcntl(fd, fcntl.F_SETFL, flags | os.O_NONBLOCK) | |
| view = memoryview(buf) | |
| while view: | |
| try: | |
| n = stdin.write(view) | |
| except BlockingIOError: | |
| n = 0 | |
| if n: | |
| view = view[n:] | |
| continue | |
| left = deadline - time.monotonic() | |
| if left <= 0: | |
| # أثناء حشوٍ نشط: العملية حية (موت خرج حقيقي يقتلها ونراه بـpoll) | |
| # ونحن نغذّي إطارات مجمّدة قصدًا — انسداد قطعة مصدر أثناء طحن | |
| # خياطة ليس دليل موت خرج (مقيس: 13:06:13 و13:21:46 و10:24:57). | |
| # نجدّد بمهل قصيرة ما دام الحشو قائمًا والخرج يتقدّم فعليًا، وسقف الحشو (90 ث) نفسه | |
| # يحدّ الصبر الكلي، فلا تعليق أبديًا ولا قتل لجلسة تُنقَذ. بلا تقدّم خرج | |
| # فالحشو يخفي انسداد RTMPS؛ لا تجديد عندها ويستيقظ الحارس فورًا. | |
| if Sender._fill_write_renew_allowed(sess): | |
| deadline = time.monotonic() + 2.0 | |
| continue | |
| raise SourceStalled('أنبوب FFmpeg ممتلئ ولا يقرأ منذ %.0f ث' % max(0.0, -left)) | |
| select.select([], [fd], [], min(0.5, left)) | |
| finally: | |
| try: | |
| fcntl.fcntl(fd, fcntl.F_SETFL, flags) # نُنظّف: نُبقي الوضع محجوزًا كسابق عهده | |
| except OSError: | |
| pass | |
| def _is_timeout(e): | |
| n = type(e).__name__.lower() | |
| return 'timeout' in n or 'timed out' in str(e).lower() | |
| def _chunk_reader(self, it, out): | |
| """يقرأ قطعة واحدة من الاستجابة في خيط منفصل ويضعها في طابور بحجم 1. | |
| الضرورة: `next()` على مقبس صامت ينام حتى read_timeout (15ث افتراضيًا)، وقد تكون | |
| أطول من مهلة الحارس نفسها، فيبقى خيط الضخّ حيًا بعد teardown وتُترك العملية وحيدة. | |
| """ | |
| try: | |
| out.put(next(it)) | |
| except StopIteration: | |
| out.put(None) | |
| except BaseException as e: # تُعاد رفعها في خيط الضخّ لتُصنَّف هناك | |
| out.put(e) | |
| def _pump_source(self, response, p, bridge, stop, sess, read_timeout, retries, cancel=None): | |
| """نقل بايتات تيك توك إلى FFmpeg في خيط مستقل، مع تسجيل زمن آخر وصول فعلي. | |
| القراءة هنا هي مصدر الحقيقة: `poll()` الخاص بـ FFmpeg يبقى ناجحًا وهو جالس ينتظر | |
| stdin، ويوتيوب يبقى مستقبِلًا على آخر ما وصل — لذا يُقاس تدفق البيانات لا حالة العملية. | |
| """ | |
| sess.setdefault('stdin_lock', threading.Lock()) | |
| sess.setdefault('filling', False); sess.setdefault('filling_at', 0.0); sess.setdefault('fill_emit_at', 0.0); sess.setdefault('reattach_at', 0.0); sess.setdefault('fill_since', 0.0); sess.setdefault('fill_until', 0.0) | |
| it = response.iter_content(None).__iter__() | |
| deadline = time.monotonic() + sess['max_seconds'] | |
| chunk_wait = min(read_timeout, max(1.0, self._stall_seconds() / 3.0)) if self._stall_seconds() else read_timeout | |
| pending = None | |
| reader = None | |
| while (not stop.is_set() and not sess['force_stop'] and not sess.get('swap_source') | |
| and not (cancel and cancel.is_set())): | |
| if time.monotonic() > deadline: | |
| sess['error'] = 'انتهت مهلة الجلسة القصوى (%ds)' % int(sess['max_seconds']) | |
| return | |
| if pending is None: | |
| if reader is not None and reader.is_alive(): | |
| # لا قراءتين متوازيتين على نفس المولّد (ليس خيط-آمن): ننتظر دوره قليلًا | |
| time.sleep(0.2) | |
| continue | |
| box = queue.Queue(1) | |
| reader = threading.Thread(target=self._chunk_reader, args=(it, box), daemon=True) | |
| reader.start() | |
| pending = box | |
| try: | |
| got = pending.get(timeout=chunk_wait) | |
| except queue.Empty: | |
| if stop.is_set() or sess['force_stop'] or sess.get('swap_source') or (cancel and cancel.is_set()): | |
| return | |
| silent = time.monotonic() - sess['last'] | |
| if self._stall_seconds() and silent > self._stall_seconds(): | |
| sess['error'] = 'صمت من تيك توك (%.0f ث بلا بايت)' % silent | |
| return | |
| continue | |
| pending = None | |
| # A late result from the old HTTP reader cannot mutate the replacement pump. | |
| if stop.is_set() or sess['force_stop'] or sess.get('swap_source') or (cancel and cancel.is_set()): | |
| return | |
| if isinstance(got, BaseException): | |
| e = got | |
| if self._is_timeout(e) and not sess['force_stop']: | |
| if retries and sess['timeouts'] < retries: | |
| sess['timeouts'] += 1 | |
| continue # timeout قصير: نعيد المحاولة بدل قتل بثٍّ يعمل | |
| sess['error'] = 'انتهت مهلة قراءة المصدر (%ds × %d)' % (int(read_timeout), retries) | |
| return | |
| sess['error'] = 'خطأ قراءة من تيك توك (%s: %s)' % (type(e).__name__, str(e)[:120]) | |
| return | |
| if got is None: | |
| sess['ended'] = True # المصدر أغلق الاستجابة = انتهى البث/تجدد الرابط | |
| return | |
| chunk = got | |
| sess['timeouts'] = 0 | |
| if not chunk: | |
| continue | |
| sess['last'] = time.monotonic() | |
| with self.lock: | |
| self.data['source_bytes'] += len(chunk) | |
| self.data['last_source_at'] = sess['last'] | |
| if p.stdin is None: | |
| continue # لا أنبوب (HLS يقرأ بنفسه): الزمن وحده هو ما يُسجَّل | |
| stall_limit = self._stall_seconds() | |
| try: | |
| buf = bridge.feed(chunk) | |
| if not buf: | |
| continue | |
| sess['writing'] = True | |
| try: | |
| with sess['stdin_lock']: | |
| if stall_limit: | |
| self._write_deadline(p.stdin, buf, self._deadline_base(sess) + stall_limit, sess) | |
| else: | |
| view = memoryview(buf) | |
| while view and not stop.is_set() and not sess['force_stop']: | |
| n = p.stdin.write(view) | |
| if not n: | |
| raise BrokenPipeError() | |
| view = view[n:] | |
| finally: | |
| sess['writing'] = False | |
| except SourceStalled as e: | |
| # A full stdin pipe means FFmpeg stopped consuming because its demux/output side | |
| # is stuck. It is not evidence that TikTok stopped sending this chunk. | |
| self.last_failure_side = 'output' | |
| sess['output_failure'] = True | |
| sess['error'] = str(e) | |
| return | |
| except (BrokenPipeError, OSError, ValueError): | |
| return # FFmpeg هو من مات؛ الحلقة الرئيسية تتصرف | |
| def _try_reattach(self, user, p, bridge, sess, stop, refresh, read_timeout, retries, response, pump): | |
| """Close only the TikTok HTTP; keep FFmpeg stdin so YouTube does not see a new ingest. | |
| Returns (new_response, new_pump) or None. Viewer cuts come from killing FFmpeg, not from | |
| a 1–2 s TikTok URL refresh into the same process. | |
| """ | |
| if (not refresh or p is None or p.stdin is None or p.poll() is not None | |
| or stop.is_set() or sess.get('force_stop')): | |
| return None | |
| if int(sess.get('reattaches') or 0) >= 24: | |
| return None | |
| sess['swapping'] = True | |
| sess['swap_source'] = True | |
| sess['swap_started'] = time.monotonic() | |
| old_cancel = sess.get('pump_cancel') | |
| if old_cancel is not None: | |
| old_cancel.set() | |
| try: | |
| if response: | |
| try: | |
| response.close() | |
| except (OSError, ValueError): | |
| pass | |
| if pump is not None: | |
| pump.join(timeout=float(os.getenv('YOUTUBE_PUMP_JOIN_SECONDS', '2'))) | |
| if pump.is_alive(): | |
| self.log(user, '[RELAY] قارئ المصدر السابق لم يتوقف؛ لا نشارك معه الجسر أو أنبوب FFmpeg') | |
| return None | |
| if stop.is_set() or sess.get('force_stop'): | |
| return None | |
| try: | |
| cand = refresh() | |
| except Exception as e: | |
| self.log(user, '[RELAY] تعذر رصد مصدر بديل دون قطع يوتيوب (%s)' % type(e).__name__) | |
| return None | |
| if (not cand or cand.get('media') == 'hls' or not cand.get('url') | |
| or stop.is_set() or sess.get('force_stop') or p.poll() is not None): | |
| return None | |
| try: | |
| new_resp = requests.get(cand['url'], | |
| headers=source_http_headers(cand.get('url'),user,self.headers_for), | |
| impersonate='chrome', stream=True, | |
| timeout=(5, min(8, read_timeout or 8))) | |
| new_resp.raise_for_status() | |
| except Exception as e: | |
| telemetry_event(self.telemetry_key,user,'SOURCE_OPEN_ERROR',{ | |
| 'stage':'reattach','error_type':type(e).__name__, | |
| 'http_status':http_status_from_exception(e)}) | |
| self.log(user, '[RELAY] فشل فتح مصدر تيك توك البديل (%s)' % type(e).__name__) | |
| return None | |
| if stop.is_set() or sess.get('force_stop') or p.poll() is not None: | |
| new_resp.close() | |
| return None | |
| bridge.reconnect() | |
| sess['ended'] = False | |
| sess['error'] = '' | |
| sess['stall_reason'] = '' | |
| sess['timeouts'] = 0 | |
| # Do not bump sess['last'] here: that hid 40–57s of silence from the | |
| # 12s stall watchdog (measured @vimtoo.2: out frozen, age 42–57, yh still good). | |
| sess['reattaches'] = int(sess.get('reattaches') or 0) + 1 | |
| sess['reattach_at'] = time.monotonic() | |
| self.current_candidate = cand | |
| self.set(source=cand.get('label') or cand.get('brief') or '', stalled=False) | |
| try: | |
| grace = max(0.5, min(5.0, float(os.getenv('YOUTUBE_REATTACH_GRACE_SECONDS', '2')))) | |
| except (TypeError, ValueError): | |
| grace = 2.0 | |
| limit = self._stall_seconds() | |
| if limit: | |
| grace = min(grace, max(0.5, limit / 2.0)) | |
| sess['reattach_not_before'] = time.monotonic() + grace | |
| new_cancel = threading.Event() | |
| sess['pump_cancel'] = new_cancel | |
| # Clear the OLD reader's handover flag BEFORE starting its replacement. If | |
| # start() schedules the new target immediately, a true flag kills it unread. | |
| sess['swap_source'] = False | |
| new_pump = threading.Thread(target=self._pump_source, | |
| args=(new_resp, p, bridge, stop, sess, read_timeout, retries, new_cancel), | |
| daemon=True) | |
| new_pump.start() | |
| self.log(user, '[RELAY] أُعيد توصيل تيك توك على نفس جلسة يوتيوب — بلا قتل FFmpeg') | |
| return new_resp, new_pump | |
| finally: | |
| sess['swap_source'] = False | |
| sess['swapping'] = False | |
| def _close_reason(self,sess,p): | |
| """Describe intentional stops and output failures truthfully in the persisted audit.""" | |
| reason=sess.get('stall_reason') or sess.get('error') | |
| if reason:return str(reason)[:300] | |
| if self.last_failure_side=='output': | |
| try:code=p.poll() if p is not None else self.last_exit_code | |
| except Exception:code=self.last_exit_code | |
| reason='فشل خرج RTMPS (FFmpeg exit code %s)'%(code if code is not None else 'unknown') | |
| if self.last_diagnostics: | |
| reason+=' — '+str(self.last_diagnostics[-1])[:120] | |
| return reason | |
| return 'إيقاف عادي' | |
| def _teardown(self, user, p, response, sess, stop, reason): | |
| """إنهاء مرتب — FFmpeg أولًا ثم الأنبوب ثم المصدر، مع wait محدود ثم kill. | |
| الترتيب مقصود: إغلاق stdin يخبر FFmpeg أن الإدخال انتهى فيتوقف بلطف ويُنهي ما في | |
| مخزنه؛ لو قتلنا FFmpeg أولًا وبقيت الاستجابة مفتوحة بقي الخيط القارئ معلقًا (وهو | |
| بالضبط ما كان يُبقي جلسة ميتة محتلةً لمكان الحدث). | |
| """ | |
| sess['force_stop'] = True | |
| # Do NOT set the supervisor `stop` Event here. That Event means "end the | |
| # whole account thread" (user stop / finish_event). Setting it on every | |
| # stall/format-change made sender_loop return, occupy a new FFmpeg, and | |
| # flash the YouTube spinner (measured on On @aboodi1994: age≈12s → rst++). | |
| # The pump watches sess['force_stop'] + stdin close. | |
| if p is not None: | |
| try: | |
| if p.stdin: | |
| p.stdin.close() | |
| except (OSError, ValueError): | |
| pass | |
| try: | |
| if response: | |
| response.close() # يفكّ التعليق داخل iter_content/read | |
| except (OSError, ValueError): | |
| pass | |
| codes = [] | |
| if p is not None: | |
| try: | |
| if p.poll() is None: | |
| p.terminate() | |
| except OSError: | |
| pass | |
| # عند صمت المصدر لا فائدة من انتظار «إنهاء لطيف» طويل: stdin مغلق ولا بيانات تُدفَّق. | |
| grace = float(os.getenv('YOUTUBE_KILL_GRACE_SECONDS', '10')) | |
| if sess.get('stall_reason') or sess.get('ended'): | |
| grace = min(grace, 2.0) | |
| try: | |
| codes.append(p.wait(timeout=grace)) | |
| except subprocess.TimeoutExpired: | |
| try: | |
| p.kill() | |
| except OSError: | |
| pass | |
| try: | |
| # SIGKILL لا يُقاوَم إلا في IO غير قابل للإيقاف؛ 3 ث تكفي والحارس يوثّق ما بعدها. | |
| codes.append(p.wait(timeout=3)) | |
| except subprocess.TimeoutExpired: | |
| sess['orphan'] = True # عملية لم تُقْبَر: لا يجوز أن تُنسى بصمت | |
| self.log(user, '[CLEANUP] ⚠ FFmpeg لم ينتهِ بعد kill؛ ستُراجع العملية في الدورة التالية (%s)' % reason) | |
| with self.lock: | |
| self.process = None | |
| # الرمز يبقى في الحالة بعد انتهاء الجلسة: المشرف يحتاجه ليميز «يوتيوب رفض الاستقبال» | |
| # (187/قطع من الخادم) عن صمت المصدر — الأول يستحق إبطاء الوتيرة لا تسريعها. | |
| self.last_exit_code = codes[0] if codes else None | |
| # الأخطاء تُنسخ في قارئ stderr أثناء الجلسة لا هنا: raw_tail/diagnostics محليّان داخل | |
| # send()، وأي إشارة لهما من _teardown ترفع NameError تُسقط بقية التفكيك والـset(). | |
| # كم كان المصدر صامتًا لحظة النهاية؟ بلا هذا القياس يُحسب كل قفل ذنبًا للوجهة، بينما | |
| # أكثر ما يحدث أن يوتيوب أقفل لأنه لم يعد تصله بايتات (مقيس: 13-14 ث بعد صمت المصدر). | |
| # لا معنى للقياس في HLS (لا استجابة عندنا يقرأها الحارس) فيبقى صفرا. | |
| if self.sticky_mode and not self._recovering_copy: | |
| try: | |
| self._last_light_output=max(float(self._last_light_output or 0), float(self.data.get('output_seconds') or 0)) | |
| except Exception: | |
| pass | |
| self.last_source_gap = round(max(0.0, time.monotonic() - sess['last']), 1) \ | |
| if (sess.get('src_metered') and sess.get('last')) else 0.0 | |
| self.set(status='', stalled=bool(sess.get('stall_reason') or sess.get('error')), note=reason, | |
| source_gap=self.last_source_gap, failure_side=self.last_failure_side, | |
| ffmpeg_errors=list(self.last_errors), | |
| ffmpeg_diagnostics=list(self.last_diagnostics)[-3:]) | |
| codes = [c for c in codes if c] | |
| self.log(user, '[CLEANUP] أُوقف FFmpeg (%s%s)' % (reason, '؛ رمز الخروج ' + ','.join(map(str, codes)) if codes else '')) | |
| def send(self, user, candidate, destination, stop, refresh=None, extra_destinations=None): | |
| """Push one rendition until it dies, the source stalls, or `stop` is set. | |
| الإشراف هنا على **البيانات** لا على العمليات: FFmpeg يبقى حيًا ومسرورًا وهو جالس | |
| ينتظر stdin، ويوتيوب يبقى مستقبِلًا على آخر ثانية وصلته، فيبدو البث «معلقًا» للأبد. | |
| لذا: خيط قراءة يسجّل آخر بايت، وحارس يقارن الزمن، وteardown مرتب، ثم رجوع إلى حلقة | |
| الإشراف في `sender_loop` التي تُعيد رصد LIVE وتبني جلسة جديدة من الصفر. | |
| """ | |
| self.current_candidate = candidate | |
| self.last_failure_side = '' | |
| self.last_errors=[] | |
| self.last_diagnostics=[] | |
| self.lock_gold(candidate) | |
| hls = (candidate or {}).get('media') == 'hls' | |
| # Replacement SPS/PPS or VPS/SPS/PPS records are forwarded through this bridge and the | |
| # same FFmpeg/RTMPS publisher. They are not a reason to select x264 or restart output. | |
| bridge = FLVBridge(user=user,event_logger=self._bridge_event) | |
| response = None | |
| p = None | |
| error_reader = None | |
| done = threading.Event() | |
| progress = {'last': time.monotonic(), 'seconds': -1, 'slow_since': None} | |
| pump = None | |
| diagnostics = [] | |
| raw_tail = [] # آخر سطور stderr بعد الإخفاء: تُطبع عند الخروج، لا تُخزَّن خامًا | |
| read_timeout, retries, max_seconds = self._stream_tuning() | |
| sess = {'last': time.monotonic(), 'ended': False, 'error': '', | |
| 'src_metered': False, | |
| 'force_stop': False, 'timeouts': 0, 'orphan': False, 'stable_notified': False, 'max_seconds': max_seconds, | |
| 'stall_reason': '', 'spawned': False, 'spawn_error': '', | |
| 'writing': False, 'output_failure': False, 'connection_event': False, | |
| 'ff_last': time.monotonic(), | |
| 'stdin_lock': threading.Lock(), 'filling': False, 'filling_at': 0.0, 'fill_emit_at': 0.0, 'fill_since': 0.0, 'fill_until': 0.0, 'reattach_at': 0.0} | |
| self.set(status='الاتصال بيوتيوب', source=candidate['label'], output_seconds=0, | |
| last_output_at=None, stalled=False, last_source_at=sess['last'], source_bytes=0, | |
| rtmp_phase='') | |
| with self.lock: | |
| self.session = sess | |
| # Telemetry session boundaries are observational only. The key is | |
| # never consulted by mode, watchdog, reconnect, or quality decisions. | |
| tele_meta={'candidate_label':candidate.get('label'),'candidate_brief':candidate.get('brief'), | |
| 'media':candidate.get('media'),'codec':candidate.get('codec'), | |
| 'quality_name':candidate.get('quality_name'),'width':candidate.get('width'), | |
| 'height':candidate.get('height'),'probed_width':candidate.get('probed_width'), | |
| 'probed_height':candidate.get('probed_height'), | |
| 'declared_bitrate_kbps':candidate.get('bitrate')} | |
| if isinstance(self.telemetry_context,dict): | |
| tele_meta.update(self.telemetry_context) | |
| self.telemetry_key=telemetry_start(user,tele_meta) | |
| tele_key=self.telemetry_key | |
| tele_fields={} | |
| tele_last_observe=time.monotonic() | |
| tele_last_checkpoint=time.monotonic() | |
| tele_last_source_bytes=0 | |
| tele_finished=False | |
| def tele_finish(reason,extra=None): | |
| nonlocal tele_finished | |
| if tele_finished:return | |
| tele_finished=True | |
| telemetry_finish(tele_key,reason,extra) | |
| if self.telemetry_key==tele_key:self.telemetry_key=None | |
| try: | |
| sess['src_metered'] = not hls # HLS: FFmpeg يقرأ القائمة، فلا عدّاد مصدر لنا | |
| args = self.command(destination, candidate, extra_destinations=extra_destinations) | |
| telemetry_observe(tele_key,{'mode':self.active_mode or self.send_mode(), | |
| 'source_width':candidate.get('width'), | |
| 'source_height':candidate.get('height'), | |
| 'declared_bitrate_kbps':candidate.get('bitrate')}) | |
| if hls: # noqa: E999 — HLS: بلا أنبوب stdin | |
| # FFmpeg يفتح القائمة بنفسه: لا HTTP عندنا ولا ضخّ على stdin ⇒ الحارس هنا | |
| # هو تقدّم FFmpeg نفسه (out_time)، لأن لا مصدر بايتات ثانٍ نقيسه. | |
| p = subprocess.Popen(args, stdin=None, stdout=subprocess.PIPE, stderr=subprocess.PIPE) | |
| else: | |
| try: | |
| response = requests.get(candidate['url'], | |
| headers=source_http_headers(candidate.get('url'),user,self.headers_for), | |
| impersonate='chrome', stream=True, | |
| timeout=(10, read_timeout)) | |
| response.raise_for_status() | |
| except Exception as e: | |
| telemetry_event(tele_key,user,'SOURCE_OPEN_ERROR',{ | |
| 'stage':'initial_open','error_type':type(e).__name__, | |
| 'http_status':http_status_from_exception(e)}) | |
| raise | |
| p = subprocess.Popen(args, stdin=subprocess.PIPE, stdout=subprocess.PIPE, | |
| stderr=subprocess.PIPE, bufsize=0) | |
| sess['spawned'] = True | |
| sess['spawned_at'] = time.monotonic() | |
| if destination.startswith(('rtmp://','rtmps://')): | |
| note_rtmp_open(user,{'mode':self.active_mode or self.send_mode(), | |
| 'codec':'h265' if self.hevc_payload(candidate) else 'h264'}) | |
| telemetry_event(tele_key,user,'RTMP_OPEN',{ | |
| 'mode':self.active_mode or self.send_mode(), | |
| 'codec':'h265' if self.hevc_payload(candidate) else 'h264'}) | |
| with self.lock: | |
| self.process = p | |
| self.log(user, '[RELAY] بدء FFmpeg ← %s (%s؛ mode=%s)' % (candidate['label'], | |
| 'إعلان Enhanced-RTMP: hvc1' if 'rtmp_enhanced_codecs' in ' '.join(args) | |
| else 'FLV تقليدي: لا إعلان hvc1', self.active_mode or 'unknown')) # لا إعادة ترميز في هذا الوضع: انقطاع copy لا يترك عملية ورائها | |
| if extra_destinations and '-f' in args and 'tee' in args: | |
| self.log(user, '[RELAY] نشر مزدوج متزامن: استقبال يوتيوب الأساسي والاحتياطي معًا على نفس الحدث — موت ساق واحدة لا يوقف البث أمام المشاهدين') | |
| def read_errors(): | |
| # يصنّف أخطاء FFmpeg المختارة؛ لا يُسرّب stderr خامًا ولا روابط البث — الذيل | |
| # يُخفى مفتاحه ورابطه قبل أي تخزين أو طباعة. | |
| try: | |
| for line in iter(p.stderr.readline, b''): | |
| text = line.decode('utf-8', 'replace') | |
| low = text.lower() | |
| if 'handshak' in low: | |
| self.set(rtmp_phase='handshake') | |
| elif 'fcpublish stream' in low or 'sending publish command' in low: | |
| self.set(rtmp_phase='publish') | |
| elif 'netstream.publish.start' in low or 'publishing stream at bitrates' in low: | |
| self.set(rtmp_phase='accepted') | |
| if ffmpeg_output_failure(text): | |
| sess['output_failure'] = True | |
| self.last_failure_side = 'output' | |
| label = classify_ffmpeg_error(text) | |
| if label and label not in diagnostics: | |
| diagnostics.append(label) | |
| self.last_diagnostics = list(diagnostics) | |
| if label or _error_interesting(text): | |
| safe = redact_stderr(text) | |
| if safe: | |
| raw_tail.append(safe) | |
| self.last_errors = list(raw_tail)[-4:] | |
| if len(raw_tail) > 60: | |
| # جلسة طويلة بإعادة اتصال متكرر: الذيل لا ينمو بلا سقف، وما | |
| # يهمّ المشرف هو آخر الرسائل لا أقدمها. | |
| del raw_tail[:40] | |
| except (OSError, ValueError): | |
| pass | |
| error_reader = threading.Thread(target=read_errors, daemon=True) | |
| error_reader.start() | |
| def read_progress(): | |
| try: | |
| for line in iter(p.stdout.readline, b''): | |
| if b'=' in line: | |
| raw_name, raw_value = line.split(b'=', 1) | |
| name = raw_name.decode('ascii', 'ignore').strip() | |
| value = raw_value.decode('ascii', 'ignore').strip() | |
| # The same progress stream FFmpeg already emits is | |
| # parsed for diagnostics; no extra probe/process is | |
| # introduced and telemetry only updates memory here. | |
| try: | |
| if name == 'frame': tele_fields['ffmpeg_frame'] = int(value) | |
| elif name == 'fps': tele_fields['ffmpeg_fps'] = float(value) | |
| elif name == 'bitrate' and value not in ('', 'N/A'): | |
| tele_fields['ffmpeg_bitrate_kbps'] = float(value.split('kbits/s', 1)[0]) | |
| elif name == 'total_size': tele_fields['ffmpeg_total_size'] = int(value) | |
| elif name == 'dup_frames': tele_fields['ffmpeg_dup_frames'] = int(value) | |
| elif name == 'drop_frames': tele_fields['ffmpeg_drop_frames'] = int(value) | |
| elif name == 'speed' and value not in ('', 'N/A'): | |
| tele_fields['ffmpeg_speed'] = float(value.rstrip('x')) | |
| if tele_fields['ffmpeg_speed'] < 0.90: | |
| if not progress.get('slow_since'): | |
| progress['slow_since'] = time.monotonic() | |
| elif tele_fields['ffmpeg_speed'] >= 0.95: | |
| progress['slow_since'] = None | |
| elif name == 'progress': | |
| snapshot = dict(tele_fields) | |
| snapshot['output_seconds'] = max(0.0, progress['seconds']) | |
| telemetry_observe(tele_key, snapshot) | |
| except (TypeError, ValueError): | |
| pass | |
| if line.startswith(b'out_time_us='): | |
| try: | |
| seconds = int(line.split(b'=', 1)[1]) / 1000000 | |
| except ValueError: | |
| continue | |
| if seconds > progress['seconds']: | |
| progress['seconds'] = seconds | |
| progress['last'] = time.monotonic() | |
| sess['ff_last'] = time.monotonic() # ليس دليلاً على المصدر إلا في HLS | |
| sess['reattaches'] = 0 | |
| self.set(status='FFmpeg يرسل — تحقق من معاينة YouTube Studio', | |
| output_seconds=round(seconds, 1), last_output_at=time.time(), | |
| stalled=False) | |
| # «استقرار» مرة واحدة لكل جلسة: ثوانٍ من التدفق المقبول تعني أن الحافة لم تعد ترفضنا، | |
| # فلا معنى لبقاء نافذة العقاب ولا لملاحظتها على اللوحة حتى تموت الجلسة (وهو ما كان يحدث). | |
| if not sess.get('stable_notified') and seconds >= float(os.getenv('YOUTUBE_STABLE_SECONDS', '25')): | |
| sess['stable_notified'] = True | |
| cb = getattr(self, 'on_stable', None) | |
| if cb: | |
| try: | |
| cb(user) | |
| except Exception as e: # قرار المشرف لا يجوز أن يقتل بثًا يعمل | |
| self.log(user, '[WATCHDOG] فشل مُبلِّغ الاستقرار: ' + str(e)[:80]) | |
| except (OSError, ValueError): | |
| pass | |
| threading.Thread(target=read_progress, daemon=True).start() | |
| pump = None | |
| if response is not None: | |
| sess['pump_cancel'] = threading.Event() | |
| pump = threading.Thread(target=self._pump_source, | |
| args=(response, p, bridge, stop, sess, read_timeout, retries, sess['pump_cancel']), | |
| daemon=True) | |
| pump.start() | |
| # ---- حشو الفجوات: صورة مجمّدة بدل صمتٍ تقفل بعده حافة يوتيوب النشر ---- | |
| # صمت المصدر 13-30 ث كان يقتل جلسة سليمة ثم نصطدم بإعادة نشر مرفوضة. | |
| # الحشو يُبقي الخطّ الزمني والوصلة حيّين حتى يعود المصدر فيُخاط في نفس | |
| # الاتصال (انحياز الجسر)، بسقف زمني بعده يتصرّف الحارس كصمت مصدر عادي. | |
| fill_state={'started':None,'emitted':0,'last_emit':0.0} | |
| try:fill_after=max(0.5,float(os.getenv('YOUTUBE_FILL_AFTER_SECONDS','2'))) | |
| except (TypeError,ValueError):fill_after=2.0 | |
| try:fill_max=max(0.0,float(os.getenv('YOUTUBE_FILL_MAX_SECONDS','90'))) | |
| except (TypeError,ValueError):fill_max=90.0 | |
| try:fill_step=max(250,int(os.getenv('YOUTUBE_FILL_STEP_MS','2000'))) | |
| except (TypeError,ValueError):fill_step=2000 | |
| def fill_loop(): | |
| try: | |
| while not stop.is_set(): | |
| time.sleep(0.5) | |
| if p is None or p.poll() is not None or p.stdin is None or sess.get('force_stop'): | |
| return | |
| now_f=time.monotonic() | |
| silent_for=now_f-float(sess.get('last') or now_f) | |
| if silent_for<fill_after: | |
| if fill_state['started'] is not None: | |
| filled_for=now_f-fill_state['started'] | |
| self.log(user,'[FILL] عاد المصدر بعد %.0f ث حشو (%d إطارًا مجمّدًا)؛ الاستئناف في نفس اتصال يوتيوب بلا إعادة نشر'%(filled_for,fill_state['emitted'])) | |
| log_event(user,'GAP_FILL_END',{'filled_seconds':round(filled_for,1),'frames':fill_state['emitted'],'silent_audio':bridge.fill_audio_ready()}) | |
| telemetry_event(tele_key,user,'GAP_FILL_END',{'filled_seconds':round(filled_for,1),'frames':fill_state['emitted'],'silent_audio':bridge.fill_audio_ready()}) | |
| fill_state['started']=None;fill_state['emitted']=0 | |
| sess['filling']=False | |
| continue | |
| if not fill_max or bridge._last_keyframe_tag is None: | |
| sess['filling']=False;continue | |
| if fill_state['started'] is None: | |
| fill_state['started']=now_f;fill_state['last_emit']=0.0 | |
| sess['fill_since']=now_f;sess['fill_until']=now_f+fill_max+25.0 | |
| sess['fill_out_start']=max(0.0,float(progress['seconds'])) | |
| self.log(user,'[FILL] صمت المصدر %.0f ث؛ حشو بإطار مجمّد حتى لا يرى يوتيوب فجوة (سقف %.0f ث)'%(silent_for,fill_max)) | |
| log_event(user,'GAP_FILL_START',{'silent_seconds':round(silent_for,1),'cap_seconds':fill_max,'silent_audio':bridge.fill_audio_ready()}) | |
| telemetry_event(tele_key,user,'GAP_FILL_START',{'silent_seconds':round(silent_for,1),'cap_seconds':fill_max,'silent_audio':bridge.fill_audio_ready()}) | |
| if now_f-fill_state['started']>fill_max: | |
| sess['filling']=False;continue | |
| sess['filling']=True;sess['filling_at']=now_f | |
| if now_f-fill_state['last_emit']<fill_step/1000.0*0.95:continue | |
| data=bridge.freeze_span_bytes(fill_step) | |
| if not data:continue | |
| try: | |
| with sess['stdin_lock']: | |
| if time.monotonic()-float(sess.get('last') or 0)<fill_after:continue | |
| # لا كتابة خام محظورة هنا: أنبوب ممتلئ كان يُبقي خيط | |
| # الحشو (وقفل stdin معه) معلّقًا بلا مهلة، فيتجمّد | |
| # الانبعاث ولا يرى الحارس إلا عدّادات مضلّلة. نفس | |
| # مهلة المضخّة: تجديد فقط أثناء تقدّم خرج حقيقي. | |
| fill_limit=self._stall_seconds() | |
| if fill_limit: | |
| self._write_deadline(p.stdin,data, | |
| self._deadline_base(sess)+fill_limit,sess) | |
| else: | |
| view=memoryview(data) | |
| while view: | |
| n=p.stdin.write(view) | |
| if not n:raise BrokenPipeError() | |
| view=view[n:] | |
| except SourceStalled as e: | |
| sess['output_failure']=True | |
| sess['error']=str(e) | |
| sess['filling']=False | |
| return | |
| except (BrokenPipeError,OSError,ValueError): | |
| return | |
| fill_state['emitted']+=1;fill_state['last_emit']=time.monotonic();sess['fill_emit_at']=fill_state['last_emit'] | |
| except Exception: | |
| sess['filling']=False | |
| return | |
| filler=None | |
| if response is not None: | |
| filler=threading.Thread(target=fill_loop,daemon=True);filler.start() | |
| # ---- الحارس: لا تعتمد على poll() ولا على رمز الخروج ---- | |
| stall_limit = self._stall_seconds() | |
| progress_limit = max(30.0, float(os.getenv('YOUTUBE_PROGRESS_STALL_SECONDS', '60'))) | |
| fill_output_grace = self._fill_output_grace() | |
| next_report = 0.0 | |
| while not stop.is_set(): | |
| if self.pending_sticky and self.apply_pending_sticky(user): | |
| break | |
| now = time.monotonic() | |
| # الدليل على أن المصدر لم يصمت هو بايتات تيك توك نفسها. تقدّم FFmpeg وحده لا | |
| # يكفي: في وضع النسخ يفرّغ المخزن الداخلي لدقائق بعد صمت المصدر، فيبدو البث | |
| # حيًا ويوتيوب يبقى على آخر ثانية — وهذا هو التعليق الذي شكا منه المستخدم. | |
| # في HLS يستحيل القياس إلا على FFmpeg (هو من يفتح القائمة)، فنقبل تقدّمه. | |
| _fresh = (sess['ff_last'] if response is None else sess['last']) | |
| fill_output_stale = self._fill_output_stale( | |
| sess, now, fill_output_grace, progress['seconds']) | |
| swap_for = (now - float(sess.get('swap_started') or 0)) if sess.get('swapping') else 0.0 | |
| # Give a newly opened HTTP source time to deliver its first actual bytes. | |
| # Do not alter sess['last']: source age remains honest while we wait. | |
| recent_attach = now < float(sess.get('reattach_not_before') or 0) | |
| tried_reattach = False | |
| source_done = response is not None and (sess.get('ended') or sess.get('error') or | |
| (pump is not None and not pump.is_alive())) | |
| if (source_done and not recent_attach and not sess.get('force_stop') | |
| and not sess.get('writing') and not sess.get('output_failure')): | |
| tried_reattach = True | |
| attached = self._try_reattach(user, p, bridge, sess, stop, refresh, | |
| read_timeout, retries, response, pump) | |
| if attached: | |
| response, pump = attached | |
| continue | |
| # swapping must not disable the stall watchdog for longer than the stall | |
| # limit: refresh()+HTTP open was measured at 40–57s of frozen viewers. | |
| if self._recovering_copy: | |
| ran=now-float(sess.get('spawned_at') or now) | |
| if ran>12 and progress['seconds']<0.5: | |
| self.copy_recovery_fails+=1 | |
| self._recovering_copy=False | |
| self.log(user,'[RECOVERY] COPY produced no output in %.0fs; staying on %s (fail %d)'%( | |
| ran, self.sticky_mode or 'light', self.copy_recovery_fails)) | |
| raise SourceStalled('COPY recovery produced no output; reverting to light') | |
| try:verify_seconds=max(5.0,float(os.getenv('YOUTUBE_COPY_RECOVERY_VERIFY_SECONDS','15'))) | |
| except (TypeError,ValueError):verify_seconds=15.0 | |
| if progress['seconds']>=verify_seconds: | |
| self.log(user,'[RECOVERY] COPY output verified (%.1fs); leaving light sticky'%progress['seconds']) | |
| self.sticky_mode=None | |
| self.sticky_since=0.0 | |
| self._recovering_copy=False | |
| self.copy_recovery_fails=0 | |
| try: | |
| from ingest import hub | |
| hub.release_light(user) | |
| except Exception: | |
| pass | |
| # The journal copy of the mode is the one that used to | |
| # resurrect the transcode on every later FFmpeg. A | |
| # verified probe must clear it there too. | |
| cb=getattr(self,'on_sticky_clear',None) | |
| if cb: | |
| try:cb(user) | |
| except Exception as e:self.log(user,'[RECOVERY] could not clear persisted mode: '+str(e)[:80]) | |
| log_event(user,'COPY_RECOVERY',{'mode':'copy','verified_seconds':round(progress['seconds'],1)}) | |
| telemetry_event(tele_key,user,'COPY_RECOVERY',{'verified_seconds':round(progress['seconds'],1)}) | |
| # Governor: a transcode that cannot keep real time is the | |
| # bottleneck itself (measured: HEVC 960x1920 in matchgop on | |
| # cpu-basic -> CPU saturation -> YouTube starved). End this | |
| # publisher and let the next one probe byte copy instead of | |
| # burning the whole event behind a falling-behind encoder. | |
| if (self.active_mode not in ('','copy') and progress.get('slow_since') | |
| and progress['seconds'] >= 20 and not sess.get('governor_switch') | |
| and not getattr(self,'sticky_operator_locked',False) | |
| and self._copy_supported(self.current_candidate)): | |
| try:slow_for=max(10.0,float(os.getenv('YOUTUBE_TRANSCODE_SLOW_SECONDS','30'))) | |
| except (TypeError,ValueError):slow_for=30.0 | |
| if now-float(progress['slow_since'])>=slow_for: | |
| sess['governor_switch']='copy_probe' | |
| # ترميز يخنق البث أخطر من GOP مضطرب: الحاكم يملك وحده مسح قفل الصيغة | |
| self.format_unstable=False | |
| self.sticky_since=min(float(self.sticky_since or now),now-3600.0) | |
| self.log(user,'[GOVERNOR] الترميز أبطأ من الزمن الحقيقي %.0f ث — إنهاء هذا الناشر وفحص COPY بدل استمرار ترميز خاسر'%slow_for) | |
| log_event(user,'TRANSCODE_TOO_SLOW',{'mode':self.active_mode,'slow_seconds':round(slow_for,1), | |
| 'output_seconds':round(progress['seconds'],1)}) | |
| telemetry_event(tele_key,user,'TRANSCODE_TOO_SLOW',{'mode':self.active_mode, | |
| 'slow_seconds':round(slow_for,1),'output_seconds':round(progress['seconds'],1)}) | |
| break | |
| if (stall_limit and now - _fresh > 3 and not recent_attach and not tried_reattach | |
| and not sess.get('force_stop') | |
| and not sess.get('swapping') | |
| and not sess.get('swap_source') and not sess.get('writing') | |
| and not sess.get('output_failure')): | |
| tried_reattach = True | |
| attached = self._try_reattach(user, p, bridge, sess, stop, refresh, | |
| read_timeout, retries, response, pump) | |
| if attached: | |
| response, pump = attached | |
| continue | |
| if (stall_limit and now - _fresh > stall_limit and | |
| (not recent_attach or sess.get('output_failure') or sess.get('writing')) and | |
| not (sess.get('filling') and now<float(sess.get('fill_until') or 0) | |
| and not fill_output_stale) and | |
| (not sess.get('swapping') or swap_for > stall_limit)): | |
| # A full stdin pipe proves the source delivered a chunk but FFmpeg stopped | |
| # consuming it. Reopening TikTok cannot repair a blocked RTMPS output. | |
| if sess.get('writing') or sess.get('output_failure') or fill_output_stale: | |
| self.last_failure_side = 'output' | |
| sess['output_failure'] = True | |
| elif not tried_reattach: | |
| attached = self._try_reattach(user, p, bridge, sess, stop, refresh, | |
| read_timeout, retries, response, pump) | |
| if attached: | |
| response, pump = attached | |
| continue | |
| # نُفيق الخيط أولًا (قد يكون محبوسًا في write على أنبوب ممتلئ)، ثم نرفع: | |
| # إغلاق stdin يجعل FFmpeg يرى EOF ويتوقف بلطف بدل SIGKILL. | |
| sess['force_stop'] = True | |
| if p.stdin: | |
| try: | |
| p.stdin.close() | |
| except (OSError, ValueError): | |
| pass | |
| if sess.get('output_failure'): | |
| if fill_output_stale: | |
| raise SourceStalled('توقف خرج RTMPS أثناء الحشو: لا تقدّم فعلي منذ %.0f ث ' | |
| 'رغم إطارات/صوت الحشو (صمت المصدر %.0f ث)' | |
| % (now - float(sess.get('ff_last') or now), now - _fresh)) | |
| raise SourceStalled('توقف FFmpeg عن استهلاك المصدر لأن خرج RTMPS عالق منذ %.0f ث' | |
| % (now - _fresh)) | |
| self.last_failure_side = 'source' | |
| raise SourceStalled('انقطع تدفق المصدر منذ %.0f ث (FFmpeg ما زال حيًا، ' | |
| 'ويوتيوب يبقى على آخر ثانية وصلته)' % (now - _fresh)) | |
| rc = p.poll() | |
| if rc is not None: | |
| if not sess['ended'] and not stop.is_set() and rc: | |
| gap = max(0.0, time.monotonic() - sess['last']) if sess.get('src_metered') else 0.0 | |
| suspect = float(os.getenv('YOUTUBE_SOURCE_GAP_SUSPECT_SECONDS', '8')) | |
| blame = ('؛ المصدر كان صامتًا %.0f ث قبل القفل — المرجّح جوع التغذية لا رفض المستقبِل' | |
| % gap) if gap >= suspect else '' | |
| # «لم يظهر سبب مصنّف» لا يعني أن FFmpeg سكت: قد يكون قال شيئًا ليس في القائمة، | |
| # لذا تُطبع آخر رسالة بعد الإخفاء دائمًا إن وُجدت — هي وحدها ما يفرّق بين | |
| # «يوتيوب رفض المفتاح» و«توقيتنا فاسد» و«انتهت الصورة قبل الإرسال». | |
| tail = (' | آخر رسالة FFmpeg: ' + raw_tail[-1]) if raw_tail else '' | |
| self.log(user, 'FFmpeg خرج برمز %s%s%s%s' % ( | |
| rc, ' — ' + '؛ '.join(diagnostics[-3:]) if diagnostics else '؛ لم يظهر سبب مصنّف', blame, tail)) | |
| break | |
| if now - progress['last'] > progress_limit: | |
| raise SourceStalled('لم يتقدّم FFmpeg منذ %.0f ث' % (now - progress['last'])) | |
| if sess['ended'] and not sess['error'] and not recent_attach: | |
| self.log(user, '[MONITOR] أغلق تيك توك الاستجابة؛ الجلسة تُعاد من جديد') | |
| break | |
| if sess['error'] and not recent_attach: | |
| raise SourceStalled(sess['error']) | |
| if response is not None and now >= next_report: | |
| # بمعدل مضبوط: لا logging لكل بايت (كان سيملأ السجل في دقائق). | |
| next_report = now + max(15, int(os.getenv('YOUTUBE_WATCHDOG_REPORT_SECONDS', '60'))) | |
| self.log(user, '[WATCHDOG] المصدر يعمل؛ آخر بايت قبل %.0f ث، خرج %.1f ث' | |
| % (now - sess['last'], max(0, progress['seconds']))) | |
| if pump is not None and not pump.is_alive() and response is not None and not stop.is_set(): | |
| if sess['force_stop'] or ((sess['error'] or sess['ended']) and not recent_attach): | |
| break | |
| if now - tele_last_observe >= 15.0: | |
| elapsed_tele = max(0.001, now - tele_last_observe) | |
| snap_tele = self.snapshot() | |
| current_bytes = float(snap_tele.get('source_bytes') or 0) | |
| tele_sample = { | |
| 'output_seconds': max(0.0, progress['seconds']), | |
| 'source_age_seconds': max(0.0, now - sess['last']), | |
| 'source_bytes_total': current_bytes, | |
| 'mode': self.active_mode or self.send_mode(), | |
| } | |
| if sess.get('src_metered'): | |
| tele_sample['source_bps'] = max( | |
| 0.0, (current_bytes - tele_last_source_bytes) * 8.0 / elapsed_tele) | |
| telemetry_observe(tele_key, tele_sample) | |
| tele_last_source_bytes = current_bytes | |
| tele_last_observe = now | |
| if now - tele_last_checkpoint >= 60.0: | |
| telemetry_checkpoint(tele_key, 'periodic') | |
| tele_last_checkpoint = now | |
| time.sleep(1) | |
| except (OSError, ValueError) as e: | |
| if p is None: | |
| sess['spawn_error'] = 'تعذر بدء FFmpeg/المصدر (%s)' % type(e).__name__ | |
| self.log(user, '[RELAY] فشل بدء الإرسال: %s' % sess['spawn_error']) | |
| log_event(user,'CONNECTION_ERROR',{'scope':self.last_failure_side or 'startup', | |
| 'error_type':type(e).__name__,'stage':'open'});sess['connection_event']=True | |
| telemetry_event(tele_key,user,'SESSION_FAILURE',{ | |
| 'scope':self.last_failure_side or 'startup','stage':'open', | |
| 'error_type':type(e).__name__}) | |
| raise | |
| except SourceStalled as e: | |
| sess['stall_reason'] = str(e) | |
| if not self.last_failure_side: | |
| self.last_failure_side = 'source' | |
| log_event(user,'CONNECTION_ERROR',{'scope':self.last_failure_side, | |
| 'error_type':type(e).__name__,'reason':str(e)[:300]});sess['connection_event']=True | |
| telemetry_event(tele_key,user,'SESSION_FAILURE',{ | |
| 'scope':self.last_failure_side,'stage':'watchdog', | |
| 'error_type':type(e).__name__,'reason':str(e)[:300]}) | |
| tag = 'خرج الإرسال' if self.last_failure_side == 'output' else 'المصدر' | |
| self.log(user, '[WATCHDOG] توقف %s: %s' % (tag, e)) | |
| raise | |
| finally: | |
| done.set() | |
| if p is None: | |
| # لا شيء لنُنهيّه: إما أن FFmpeg لم يولد أو أن الاستجابة رفضت قبل ذلك. | |
| try: | |
| if response: | |
| response.close() | |
| except (OSError, ValueError): | |
| pass | |
| with self.lock: | |
| self.session = None | |
| self.set(status='', stalled=False, note='لم يبدأ FFmpeg أصلًا') | |
| tele_finish('start_failed',{'spawn_error':sess.get('spawn_error') or '', | |
| 'source_bytes':self.data.get('source_bytes',0), | |
| 'mode':self.active_mode or ''}) | |
| raise RelayStartError(sess.get('spawn_error') or 'تعذر بدء جلسة الإرسال') | |
| _why = self._close_reason(sess,p) | |
| if self.last_failure_side=='output' and not sess.get('connection_event'): | |
| log_event(user,'CONNECTION_ERROR',{'scope':'youtube_rtmp','reason':_why[:300], | |
| 'ffmpeg_exit_code':p.poll()});sess['connection_event']=True | |
| if destination.startswith(('rtmp://','rtmps://')) and progress['seconds'] <= 0: | |
| self.log(user, '[RTMP] لم يصل أول إطار؛ مرحلة=%s؛ دخل=%d بايت؛ pipe_busy=%s' % | |
| (self.data.get('rtmp_phase') or 'opening', self.data.get('source_bytes',0), bool(sess.get('writing')))) | |
| for line in self.last_errors[-8:]: | |
| self.log(user, '[RTMP] ' + redact_stderr(line)) | |
| self._teardown(user, p, response, sess, stop, _why) | |
| if error_reader: | |
| error_reader.join(timeout=1) | |
| if pump: | |
| # مهلة قصيرة عمدًا: الخيط قد يكون محبوسًا في next() على مقبس مُغلق؛ لا معنى | |
| # لتعليق الإشراف كله لأجله — FFmpeg أُقرِت والطرف القارئ أُغلق. | |
| pump.join(timeout=float(os.getenv('YOUTUBE_PUMP_JOIN_SECONDS', '2'))) | |
| if filler: | |
| filler.join(timeout=1) | |
| if pump.is_alive(): | |
| sess['orphan'] = True | |
| self.log(user, '[CLEANUP] ⚠ خيط القراءة ما زال يعمل؛ سيُترك كـ daemon ويُعاد الرصد') | |
| with self.lock: | |
| self.session = None | |
| tele_finish(_why or sess.get('error') or 'session_ended',{ | |
| 'ffmpeg_exit_code':p.poll(),'output_seconds':round(max(0.0,progress['seconds']),3), | |
| 'source_bytes':self.data.get('source_bytes',0), | |
| 'last_failure_side':self.last_failure_side or '', | |
| 'mode':self.active_mode or '','orphan':bool(sess.get('orphan'))}) | |
| if sess['orphan']: | |
| raise RuntimeError('بقيت عملية/خيط من جلسة سابقة؛ تُراجع قبل جلسة جديدة') | |
| return progress['seconds'] > 0 | |