"""Byte-only FLV bridge for uninterrupted H.264/HEVC stream copy. Legacy HEVC FLV codec-id 12 tags are rewritten to Enhanced RTMP ``hvc1``; H.264 and already-enhanced tags otherwise pass through. Encoded access units, keyframe flags and composition offsets are untouched. Continuous-source DTS is preserved; reconnect applies only one constant offset to keep the timeline open. A replacement AVCDecoderConfigurationRecord (SPS/PPS) or HEVCDecoderConfigurationRecord (VPS/SPS/PPS) is deliberately emitted in the same stream before the following coded frame. FFmpeg's FLV demuxer transports a replacement as ``AV_PKT_DATA_NEW_EXTRADATA`` and its FLV muxer writes another sequence header, so an isolated same-codec update stays in the current publisher. The Sender separately counts real resolution changes; repeated changes can trigger one per-broadcaster switch to a fixed canvas, reusing the same YouTube event. The complete newest record is authoritative; combining old and new parameter sets can leave conflicting parameter-set IDs and is not done. ``reconnect()`` splices a refreshed TikTok HTTP-FLV into the existing FFmpeg stdin: it removes the second file header and startup metadata, suppresses only the redundant startup decoder record, and shifts DTS monotonically. A genuinely new decoder record is never suppressed. References: - Enhanced RTMP v2, ExVideoTagHeader/SequenceStart/CodedFrames (Veovera VSO) - FFmpeg 7.1 ``flvdec.c``/``flvenc.c`` NEW_EXTRADATA handling - GStreamer bugs 797255/797256 (stream headers/caps changes in streamable FLV) """ import os import threading from collections import deque from stream_monitor import log_event, video_dimensions class FormatChanged(Exception): """Compatibility name; decoder-configuration changes no longer raise it.""" def _video_signature(body): """Return ``FourCC + decoder config`` for a video SequenceStart, else None.""" if not body:return None if body[0]&0x80: # Enhanced RTMP if body[0]&0x0F:return None # PacketType 0 == SequenceStart if len(body)<6:raise ValueError('Truncated Enhanced RTMP sequence header') return bytes(body[1:5])+body[5:] if (body[0]&15)==7 and len(body)>1 and body[1]==0: # legacy AVC sequence header if len(body)<6:raise ValueError('Truncated AVC sequence header') return b'avc1'+body[5:] return None def _tag_ts(header): return (header[7] << 24) | (header[4] << 16) | (header[5] << 8) | header[6] def _set_tag_ts(header, ts): ts = max(0, int(ts)) & 0xFFFFFFFF header[4] = (ts >> 16) & 0xFF header[5] = (ts >> 8) & 0xFF header[6] = ts & 0xFF header[7] = (ts >> 24) & 0xFF _AAC_RATES = (96000, 88200, 64000, 48000, 44100, 32000, 24000, 22050, 16000, 12000, 11025, 8000, 7350) # Silent AAC raw frames digitized from an actual encode of digital silence # (zero-spectral CPE/SCE). A raw AAC frame carries no sample rate, so the same # frame decodes as 1024 silent samples at whatever core rate is configured. _AAC_SILENCE = {1: bytes.fromhex('01182007'), 2: bytes.fromhex('211004608c1c')} def _asc_info(asc): """Parse AudioSpecificConfig bits. Returns ``(channels, is_he, out_rate_index)`` or ``None``; the rate index addresses ``_AAC_RATES`` and names the sample rate the decoder outputs: the plain rate for LC, the SBR rate for HE. Handles explicit SBR (object 5/29: header rate is the output rate, the extension rate is the core) and backward-compatible SBR (object 2 core ASC followed by the 0x2B7 sync extension, which is TikTok's layout). """ if not asc or len(asc) < 2: return None bits = ''.join(format(b, '08b') for b in asc) pos = 0 def take(n): nonlocal pos if pos + n > len(bits): raise ValueError('short ASC') value = int(bits[pos:pos + n] or '0', 2) pos += n return value try: obj = take(5) if obj == 31: obj = 32 + take(6) sfi = take(4) if sfi == 15: take(24) chan = take(4) out_idx, he = (sfi if sfi != 15 else None), False if obj in (5, 29): he = True take(5) # extension object (the LC core) if take(1): # sbrPresentFlag take(4) # core rate index (fill cadence is # stamped at the OUTPUT rate instead; see freeze_span_bytes) elif obj in (1, 2, 3, 4): take(1) # frameLengthFlag if take(1): # dependsOnCoreCoder take(14) take(1) # extensionFlag if len(bits) - pos >= 16 and int(bits[pos:pos + 11], 2) == 0x2B7: take(11) take(5) # extensionAudioObjectType (SBR) if take(1): # sbrPresentFlag he = True ext_idx = take(4) out_idx = ext_idx if ext_idx != 15 else None if chan <= 0 or (he and out_idx is None): return None if not he and out_idx is None: return None # explicit 24-bit rate: unsupported return chan, he, out_idx except (ValueError, IndexError): return None class FLVBridge: def __init__(self,detect_format_change=False,allow_raise=None,user='',event_logger=None): """Create a streaming bridge. ``detect_format_change`` and ``allow_raise`` remain accepted for callers from older releases, but are intentionally non-operative: decoder-record replacement is a normal streamable-FLV event and must not tear down the publisher. """ self.buffer=bytearray();self.header_done=False;self.converted_tags=0 self.detect_format_change=bool(detect_format_change) # compatibility/observability only self.allow_raise=allow_raise self.user=str(user or 'system');self.event_logger=event_logger or log_event self.video_cfg=None;self.video_codec=None;self.video_size=None self.format_changes=0;self.codec_changes=0 self.skip_file_header=False self.skip_script=False self.skip_same_cfg=False self.last_out_ts=None self.ts_bias=0 self.ts_bias_pending=False self._lock=threading.RLock() self._video_samples=deque();self._bitrate_kbps=None;self._bitrate_check_ts=None self._last_keyframe_ts=None;self._keyframe_interval=None;self._keyframe_event_ts=None self._last_keyframe_tag=None # آخر وسم إطار مفتاحي مُرسل كاملًا: مادة حشو الفجوات self._await_keyframe=False # بوّابة الاستئناف: لا إطارات مرجعية حتى أول مفتاحي نظيف self._gate_dropped=0 self.audio_cfg=None;self.audio_he=False self._audio_out_idx=None;self._audio_channels=None;self._audio_frame_ms=None self._audio_seq_payload=None;self._audio_byte0=0xAF self._fill_cursor=None;self._fill_declared_lc=False def _event(self,event,details): try:self.event_logger(self.user,event,details) except Exception:pass def _observe_frame(self,ts,size,keyframe): samples=self._video_samples;samples.append((ts,size)) while samples and ts-samples[0][0]>4000:samples.popleft() if len(samples)>=2 and samples[-1][0]-samples[0][0]>=3000 and ( self._bitrate_check_ts is None or ts-self._bitrate_check_ts>=2000): span=max(1,samples[-1][0]-samples[0][0]);kbps=round(sum(x[1] for x in samples)*8/span,1) old=self._bitrate_kbps;self._bitrate_kbps=kbps;self._bitrate_check_ts=ts ratio=abs(kbps-old)/max(old,1) if old is not None else 0 threshold=max(0.1,float(os.getenv('STREAM_BITRATE_CHANGE_RATIO','0.35'))) minimum=max(1.0,float(os.getenv('STREAM_BITRATE_CHANGE_KBPS','50'))) if old is not None and ratio>=threshold and abs(kbps-old)>=minimum: self._event('BITRATE_CHANGE',{'old_kbps':old,'new_kbps':kbps, 'change_percent':round((kbps-old)*100/max(old,1),1),'source_dts_ms':ts}) if keyframe: if self._last_keyframe_ts is not None and ts>self._last_keyframe_ts: interval=ts-self._last_keyframe_ts;old=self._keyframe_interval if old is not None and abs(interval-old)>=max(500,old*0.30) and ( self._keyframe_event_ts is None or ts-self._keyframe_event_ts>=5000): self._event('KEYFRAME_CHANGE',{'old_interval_ms':round(old,1), 'new_interval_ms':interval,'source_dts_ms':ts}) self._keyframe_event_ts=ts self._keyframe_interval=interval if old is None else old*0.7+interval*0.3 self._last_keyframe_ts=ts def reconnect(self): with self._lock: return self._reconnect_locked() def _reconnect_locked(self): """Prepare to splice a new TikTok FLV into the same FFmpeg stdin. ``video_cfg`` stays live so an identical startup record can be omitted while a changed record is forwarded through the already-open publisher. """ self.buffer=bytearray() self.header_done=False self.skip_file_header=True self.skip_script=True self.skip_same_cfg=True self._fill_cursor=None self.ts_bias_pending=self.last_out_ts is not None # الخياطة الجديدة قد تبدأ من منتصف GOP: إطاراتها المرجعية تشير إلى صور لم # تصل يوتيوب/المُرمِّز قط، وتمريرها يرسم رصاصيًا وألوان تشويش عند المشاهدين # حتى أول مفتاحي. البوّابة تسقطها وتبقي الصورة المجمّدة/الأخيرة حتى مفتاحي نظيف. self._await_keyframe=True;self._gate_dropped=0 self._event('RECONNECT',{'scope':'tiktok_http','outbound_rtmp_reused':True, 'last_output_dts_ms':self.last_out_ts}) def freeze_bytes(self,step_ms=2000): with self._lock: return self._freeze_bytes_locked(step_ms) def _freeze_bytes_locked(self,step_ms=2000): """وسم الإطار المفتاحي الأخير مكرّرًا بطابع زمني متقدّم: صورة مجمّدة صحيحة التشكيل تُبقي الخطّ الزمني (ووصلة يوتيوب) حيًّا أثناء صمت المصدر، بدل صمتٍ تقفل بعده حافة يوتيوب النشر (~14-20 ث موثّق). أول وسم حقيقي بعد الحشو يأخذ انحيازًا ثابتًا من آخر زمن مُرسل — نفس خياطة reconnect() تمامًا.""" if self._last_keyframe_tag is None or self.last_out_ts is None:return None new_ts=int(self.last_out_ts)+max(40,int(step_ms)) header=bytearray(self._last_keyframe_tag[:11]) _set_tag_ts(header,new_ts) self.last_out_ts=new_ts self.ts_bias_pending=True return bytes(header)+self._last_keyframe_tag[11:] @staticmethod def _make_tag(kind,ts,payload): header=bytearray(11);header[0]=kind header[1:4]=len(payload).to_bytes(3,'big') _set_tag_ts(header,ts) return bytes(header)+payload+(11+len(payload)).to_bytes(4,'big') def fill_audio_ready(self): """True when a silent-AAC fill can accompany the frozen keyframe.""" return (self.audio_cfg is not None and self._audio_frame_ms is not None and self._audio_channels in _AAC_SILENCE) def _lc_seq_payload(self): """Plain AAC-LC sequence-header payload at the source output rate.""" value=(2<<11)|((self._audio_out_idx or 0)<<7)|((self._audio_channels or 0)<<3) return bytes([self._audio_byte0,0])+value.to_bytes(2,'big') def freeze_span_bytes(self,step_ms=2000): with self._lock: return self._freeze_span_bytes_locked(step_ms) def _freeze_span_bytes_locked(self,step_ms=2000): """Frozen keyframe for one fill step plus silent AAC covering the span. TikTok audio is HE-AAC: a bare AAC-LC silent frame under an HE header makes decoders spam SBR errors and can starve the audio track, which is what lets the YouTube leg die while only video trickles. So a fill that needs audio first re-declares plain LC *at the same output sample rate the HE decoder was producing* (one replacement sequence header, the same in-stream mechanism the video path already relies on), emits silent frames at the exact 1024-sample cadence, and ``feed()`` restores the source's own HE header in front of the first returning real frame. Keeping the rate constant matters: declaring the 24 kHz core rate instead was measured to make the publisher's ``aresample=async`` chain stretch the fill span ~2x and park the returning audio tens of seconds behind the picture. Without usable audio state this degrades to the historical video-only fill. """ video=self.freeze_bytes(step_ms) if video is None:return None if os.getenv('YOUTUBE_FILL_AUDIO','1').strip().lower() in ('0','off','false','no'): return video if not self.fill_audio_ready():return video end=self.last_out_ts silent=_AAC_SILENCE[self._audio_channels] parts=[] if self._fill_cursor is None or self._fill_cursor=11: header=bytearray(self.buffer[offset:offset+11]);size=int.from_bytes(header[1:4],'big');total=11+size+4 if len(self.buffer)-offset>4)&7;packet=body[1] if packet==0: # HEVC configuration record, no composition time in SequenceStart body=bytes([0x80|(frame<<4)])+b'hvc1'+body[5:] elif packet==1: # CodedFrames includes the same signed 24-bit composition offset body=bytes([0x80|(frame<<4)|1])+b'hvc1'+body[2:] elif packet==2: body=bytes([0x80|(frame<<4)|2])+b'hvc1'+body[5:] else:raise ValueError('Unsupported legacy HEVC packet type') self.converted_tags+=1 if kind==9: signature=_video_signature(body) if signature is not None: tag_is_seq=True previous=self.video_cfg;old_size=self.video_size changed=previous is not None and signature!=previous codec=signature[:4].decode('ascii','replace') new_size=video_dimensions(codec,signature[4:]) if changed: self.format_changes+=1 old_codec=previous[:4].decode('ascii','replace') codec_changed=codec!=old_codec if codec_changed: self.codec_changes+=1;self._video_samples.clear() self._bitrate_kbps=None;self._last_keyframe_ts=None;self._keyframe_interval=None details={'codec':codec,'previous_codec':old_codec,'source_dts_ms':raw_ts, 'old_config_bytes':len(previous)-4,'new_config_bytes':len(signature)-4, 'outbound_rtmp_reused':True} self._event('DECODER_CONFIG_CHANGE',details) if codec_changed:self._event('CODEC_CHANGE',details) if old_size and new_size and old_size!=new_size: self._event('RESOLUTION_CHANGE',{'codec':codec, 'old_width':old_size[0],'old_height':old_size[1], 'new_width':new_size[0],'new_height':new_size[1], 'source_dts_ms':raw_ts,'outbound_rtmp_reused':True}) elif previous is None: self._event('STREAM_CONFIG',{'codec':codec,'width':new_size[0] if new_size else None, 'height':new_size[1] if new_size else None,'config_bytes':len(signature)-4}) self.video_cfg=signature;self.video_codec=codec;self.video_size=new_size if changed: # إعداد فكّ جديد من نفس الجلسة: الإطارات التالية قد ترجع إلى صور # مشفّرة بالإعداد القديم — نفس بوّابة المفتاحي حتى IDR نظيف. self._await_keyframe=True;self._gate_dropped=0 # Only the first identical record after reconnect is redundant. A changed # record is emitted now, before any dependent coded frame in this input. if self.skip_same_cfg: self.skip_same_cfg=False if previous is not None and not changed: offset+=total continue if kind==8 and len(body)>=2 and (body[0]>>4)==10: # AAC audio self._audio_byte0=body[0] if body[1]==0: # sequence header: learn the source audio config tag_is_seq=True self.audio_cfg=body[2:] self._audio_seq_payload=bytes(body) parsed=_asc_info(body[2:]) if parsed: self._audio_channels,self.audio_he,self._audio_out_idx=parsed self._audio_frame_ms=1024.0/_AAC_RATES[self._audio_out_idx]*1000.0 else: self._audio_frame_ms=None # المصدر نفسه أعاد الإعلان: أي إعلان LC من حشونا انتهى أمره self._fill_declared_lc=False elif body[1]==1: # real frame if self._fill_declared_lc and self._audio_seq_payload is not None: # أول إطار حقيقي بعد حشو صامت معلَن LC: نستعيد ترويسة # المصدر (HE) قبله مباشرة فيُفكّ كما كان قبل الصمت. audio_prepend=True self._fill_declared_lc=False self._fill_cursor=None if self.ts_bias_pending and tag_is_seq and self.last_out_ts is not None: # ترويسات الإعداد (AVC/AAC SequenceStart) كثيرًا ما تحمل طابع 0 أو # طابعًا لا يمثّل أول عيّنة وسائط. لا تدعها تضع انحياز الخياطة: # لو استهلكته بطابع 0 صار الانحياز ‏last_out+40، ثم قفز أول إطار # حقيقي (DTS بعيد) ملايين الميلي ثانية إلى المستقبل فينهار خطّ # الإخراج الزمني ويُهمل يوتيوب التدفق رغم أن FFmpeg «يتقدّم». # نُلصق الترويسة عند نقطة الخياطة، ويبقى الانحياز لأول وسم وسائط. out_ts=self.last_out_ts+40 else: if self.ts_bias_pending: self.ts_bias=(self.last_out_ts or 0)+40-raw_ts self.ts_bias_pending=False out_ts=raw_ts+self.ts_bias _set_tag_ts(header, out_ts) # Equal DTS across audio/video (or config/frame) is valid and must not be rewritten. # Only reconnect() adds one constant bias; last_out_ts tracks the emitted maximum. self.last_out_ts=out_ts if self.last_out_ts is None else max(self.last_out_ts,out_ts) is_keyframe=False if kind==9 and body: enhanced=bool(body[0]&0x80);packet=(body[0]&15) if enhanced else (body[1] if len(body)>1 else -1) codec_id=body[0]&15 coded=(enhanced and packet in (1,3)) or (not enhanced and codec_id==7 and packet==1) is_keyframe=coded and ((body[0]>>4)&7)==1 if coded:self._observe_frame(out_ts,len(body),is_keyframe) if self._await_keyframe and kind==9 and coded: if is_keyframe: self._await_keyframe=False self._event('KEYFRAME_RESYNC',{'dropped_tags':self._gate_dropped,'capped':False}) self._gate_dropped=0 else: self._gate_dropped+=1 if self._gate_dropped>=300: # مصدر شاذ بلا مفتاحي قادم: لا نجاعة في تجويع الفيديو أكثر من # ~10 ثوانٍ؛ نفتح ونترك أول مفتاحي حقيقي ينظّف الصورة. self._await_keyframe=False self._event('KEYFRAME_RESYNC',{'dropped_tags':self._gate_dropped,'capped':True}) self._gate_dropped=0 else: offset+=total continue header[1:4]=len(body).to_bytes(3,'big') tag_bytes=bytes(header)+body+(11+len(body)).to_bytes(4,'big') if is_keyframe:self._last_keyframe_tag=tag_bytes if audio_prepend: output.extend(self._make_tag(8,out_ts,self._audio_seq_payload)) output.extend(tag_bytes) offset+=total if offset:del self.buffer[:offset] if len(self.buffer)>17*1024*1024:raise ValueError('FLV packet buffer exceeds limit') return bytes(output)