Spaces:
Running
Running
| """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:] | |
| 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<end-2*max(40,int(step_ms)): | |
| # مرساة جديدة: بداية هذا المدى (أو مؤشر قديم تركه استئناف بصري فقط) | |
| self._fill_cursor=float(end-max(40,int(step_ms))) | |
| if self.audio_he and not self._fill_declared_lc: | |
| parts.append(self._make_tag(8,int(self._fill_cursor),self._lc_seq_payload())) | |
| self._fill_declared_lc=True | |
| while self._fill_cursor<end-1: | |
| parts.append(self._make_tag(8,int(round(self._fill_cursor)), | |
| bytes([self._audio_byte0,1])+silent)) | |
| self._fill_cursor+=self._audio_frame_ms | |
| return b''.join(parts)+video | |
| def feed(self,chunk): | |
| with self._lock: | |
| return self._feed_locked(chunk) | |
| def _feed_locked(self,chunk): | |
| self.buffer.extend(chunk);output=bytearray() | |
| if not self.header_done: | |
| if len(self.buffer)<9:return b'' | |
| if self.buffer[:3]!=b'FLV':raise ValueError('Source is not FLV') | |
| header_size=int.from_bytes(self.buffer[5:9],'big')+4 | |
| if not 13<=header_size<=1048576:raise ValueError('Invalid FLV header size') | |
| if len(self.buffer)<header_size:return b'' | |
| if not self.skip_file_header: | |
| output.extend(self.buffer[:header_size]) | |
| del self.buffer[:header_size];self.header_done=True | |
| offset=0 | |
| while len(self.buffer)-offset>=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<total:break | |
| body=bytes(self.buffer[offset+11:offset+11+size]);tag_type=header[0] | |
| if tag_type & 0x20:raise ValueError('Filtered/encrypted FLV tag is unsupported') | |
| kind = tag_type & 31;raw_ts=_tag_ts(header);audio_prepend=False;tag_is_seq=False | |
| if kind==18 and self.skip_script: | |
| offset+=total | |
| continue | |
| if kind==9 and body and not (body[0]&0x80) and (body[0]&15)==12: | |
| if len(body)<5:raise ValueError('Truncated HEVC FLV video header') | |
| frame=(body[0]>>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) | |