"""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 @staticmethod 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)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) {'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)) @staticmethod 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 @staticmethod 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) @staticmethod 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')) @staticmethod 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 [] @staticmethod 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 [] @staticmethod 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] # ------- مهل المصدر (كلها من البيئة؛ القيم الافتراضية مبرَّرة أدناه) ------- @property 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)) @staticmethod 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 @staticmethod 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 @staticmethod 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 @staticmethod 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')))) @staticmethod 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 @staticmethod 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_forfill_max: sess['filling']=False;continue sess['filling']=True;sess['filling_at']=now_f if now_f-fill_state['last_emit']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 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