Hode

Sleeping

Hode / relay.py

Replace app entirely with Reoo88/Tktktk (TikTok->YouTube relay manager)

35c2217 about 2 months ago

13.1 kB

"""One FFmpeg push per live session. Multiple senders run concurrently, one per YouTube event."""
import os, time, threading, subprocess
from collections import deque
from curl_cffi import requests
from flv_bridge import FLVBridge, FormatChanged

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}
        # Timestamps of source resolution changes seen by this sender (anti-flap window).
        self.flap=deque()
        # 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
    def set(self,**kw):
        with self.lock:self.data.update(kw)
    def snapshot(self):
        with self.lock:return dict(self.data)
    def terminate(self):
        with self.lock:p=self.process
        if p and p.poll() is None:
            try:p.terminate()
            except OSError:pass
    def send_mode(self):
        """copy      -> forward the source untouched (original quality, no extra CPU)
           fixed1080 -> re-encode onto a constant 1080x1920 canvas
           fixed720  -> re-encode onto a constant 720x1280 canvas (lighter on CPU)
        """
        mode=os.getenv('YOUTUBE_SEND_MODE','copy').strip().lower()
        if mode not in ('copy','fixed1080','fixed720'):
            raise ValueError('YOUTUBE_SEND_MODE يجب أن يكون copy أو fixed1080 أو fixed720')
        # A source already proven to flap stays on the constant canvas for this whole event.
        if self.sticky_mode:return self.sticky_mode
        # copy mode only: if the phone keeps flipping resolution, stop flapping the
        # YouTube session and absorb the changes on a constant canvas instead.
        if mode=='copy':
            flap=self.flap
            now=time.monotonic()
            while flap and now-flap[0]>600:flap.popleft()
            if len(flap)>=int(os.getenv('YOUTUBE_FLAP_LIMIT','4')):
                mode=os.getenv('YOUTUBE_FLAP_FALLBACK_MODE','fixed720').strip().lower()
                if mode not in ('fixed1080','fixed720'):mode='fixed720'
                self.sticky_mode=mode
        return mode
    def note_format_change(self):
        self.flap.append(time.monotonic())
        if len(self.flap)>=int(os.getenv('YOUTUBE_FLAP_LIMIT','4')):
            self.log('system','تغيّرت أبعاد المصدر عدة مرات خلال 10 دقائق؛ التحول تلقائيًا إلى كانفس ثابت (%s) لمنع تكرار الانقطاع'%os.getenv('YOUTUBE_FLAP_FALLBACK_MODE','fixed720'))
    @staticmethod
    def canvas(width,height):
        return ('scale=w=%d:h=%d:force_original_aspect_ratio=decrease:flags=lanczos,'
                'pad=%d:%d:(ow-iw)/2:(oh-ih)/2:color=black,setsar=1,format=yuv420p'%(width,height,width,height))
    def command(self,destination):
        # Advertise Enhanced RTMP for network output; local test files need no protocol option.
        mode=self.send_mode()
        network=destination.startswith(('rtmp://','rtmps://'))
        enhanced=['-rtmp_enhanced_codecs','hvc1'] if network and mode=='copy' and os.getenv('YOUTUBE_VIDEO_MODE','original').strip().lower()=='original' else []
        if mode=='copy':
            video=['-c:v','copy']
        else:
            width,height,bitrate=(1080,1920,'6000k') if mode=='fixed1080' else (720,1280,'4000k')
            bufsize=str(int(int(bitrate[:-1])*1.5))+'k'
            # A constant canvas is the only thing that guarantees YouTube always receives
            # frames whose real size matches the signalled one, whatever the phone does.
            video=['-vf',self.canvas(width,height),'-c:v','libx264','-preset',os.getenv('YOUTUBE_X264_PRESET','veryfast'),
                   '-tune','zerolatency','-profile:v','high','-crf',os.getenv('YOUTUBE_CRF','22'),
                   '-maxrate',bitrate,'-bufsize',bufsize,'-g','60','-keyint_min','30','-sc_threshold','0',
                   '-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 deliberately delays the YouTube feed behind the live TikTok source
            # and keeps that much data buffered. Two effects, both free of CPU cost: a
            # source hiccup is covered by the buffer instead of starving YouTube's ingest
            # (which it reports as videoIngestionStarved = viewer buffering), and packets
            # are retained across a reconnect instead of being dropped, which is what made
            # segments go missing when the streamer restarted. Cost is latency only.
            ts=int(os.getenv('YOUTUBE_FIFO_TIMESHIFT','10'))
            # 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',
                '-drop_pkts_on_overflow','1','-attempt_recovery','1','-restart_with_keyframe','1',
                '-recovery_wait_time',os.getenv('YOUTUBE_RECOVERY_WAIT','1'),
                *(('-timeshift',str(ts),'-queue_size',str(q)) if ts>0 else ())]
        else:
            output=['-flvflags','no_duration_filesize','-f','flv']
        return [self.ffmpeg,'-hide_banner','-loglevel','error','-nostats','-progress','pipe:1',
            '-fflags','+genpts','-i','pipe:0','-map','0:v:0','-map','0:a:0?',
            *video,'-c:a','aac','-b:a','128k','-ar','48000',
            '-af','aresample=async=1:first_pts=0','-avoid_negative_ts','make_zero',
            *enhanced,'-rw_timeout','20000000',*output,destination]
    def send(self,user,candidate,destination,stop):
        # YouTube's ingest is created with cdn.resolution/frameRate = 'variable', i.e. it
        # auto-detects the incoming resolution, so a source resolution change does not
        # require a new session. Tearing the session down is what produced the visible
        # gaps and the reconnect storm. The guard stays available behind an env var in
        # case YouTube's own healthStatus ever reports videoResolutionUnsupported.
        guard=os.getenv('YOUTUBE_FORMAT_GUARD','off').lower() in ('1','on','true','yes')
        bridge=FLVBridge(detect_format_change=(self.send_mode()=='copy' and guard))
        response=None;p=None;error_reader=None;done=threading.Event()
        # Set when we deliberately tear FFmpeg down because the source resolution
        # changed; suppresses the misleading "FFmpeg exited with code N" log.
        format_changed=False
        progress={'last':time.monotonic(),'seconds':-1}
        self.set(status='الاتصال بيوتيوب',source=candidate['label'],output_seconds=0,last_output_at=None)
        try:
            response=requests.get(candidate['url'],headers={'Referer':'https://www.tiktok.com/',**self.headers_for(user)},impersonate='chrome',stream=True,timeout=20)
            response.raise_for_status()
            p=subprocess.Popen(self.command(destination),stdin=subprocess.PIPE,stdout=subprocess.PIPE,stderr=subprocess.PIPE,bufsize=0)
            with self.lock:self.process=p
            diagnostics=[]
            def read_errors():
                # Classify selected FFmpeg errors; never expose raw stderr or stream URLs.
                patterns=[('Connection refused','رفض اتصال خادم الاستقبال'),('Connection timed out','انتهت مهلة اتصال خادم الاستقبال'),
                    ('Connection reset by peer','خادم الاستقبال قطع الاتصال أثناء الإرسال'),('Broken pipe','انقطع أنبوب الإرسال إلى خادم الاستقبال'),
                    ('End of file','خادم الاستقبال أغلق الاتصال'),('Immediate exit requested','طُلب إنهاء FFmpeg قبل اكتمال الإرسال'),
                    ('Packet mismatch','عدم تطابق حزم الفيديو — تغيّر مقاس المصدر داخل الجلسة'),
                    ('Input/output error','خطأ إدخال/إخراج في الإرسال'),('Server error','خادم الاستقبال أعاد خطأ'),
                    ('Invalid data found','بيانات المصدر غير صالحة للقراءة'),('Video codec','عدم توافق ترميز الفيديو مع الحاوية'),
                    ('Invalid argument','أحد إعدادات FFmpeg غير مقبول'),('Non-monotonic','توقيت الفيديو غير متزايد'),
                    ('Unknown encoder','مرمز غير متوفر'),('Option not found','خيار FFmpeg غير مدعوم')]
                try:
                    for line in iter(p.stderr.readline,b''):
                        text=line.decode('utf-8','replace')
                        for needle,label in patterns:
                            if needle.lower() in text.lower() and label not in diagnostics:diagnostics.append(label)
                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 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()
                                self.set(status='FFmpeg يرسل — تحقق من معاينة YouTube Studio',output_seconds=round(seconds,1),last_output_at=time.time())
                except (OSError,ValueError):pass
            def watchdog():
                while not done.wait(1):
                    if stop.is_set() or time.monotonic()-progress['last']>60:
                        if p.poll() is None:
                            try:p.kill()
                            except OSError:pass
                        return
            threading.Thread(target=read_progress,daemon=True).start()
            threading.Thread(target=watchdog,daemon=True).start()
            for chunk in response.iter_content():
                if stop.is_set() or p.poll() is not None:break
                if chunk:
                    # Unbuffered writes may be partial; send every byte.
                    view=memoryview(bridge.feed(chunk))
                    while view and not stop.is_set():
                        n=p.stdin.write(view)
                        if not n:raise BrokenPipeError()
                        view=view[n:]
        except FormatChanged:
            format_changed=True
            raise
        finally:
            if response:response.close()
            if p:
                try:p.stdin.close()
                except (OSError,ValueError):pass
                try:p.wait(timeout=10)
                except subprocess.TimeoutExpired:p.kill();p.wait()
                done.set()
                if error_reader:error_reader.join(timeout=1)
                if p.returncode and not stop.is_set() and not format_changed:
                    self.log(user,'FFmpeg خرج برمز '+str(p.returncode)+(' — '+'؛ '.join(diagnostics[-3:]) if diagnostics else '؛ لم يظهر سبب مصنّف'))
                with self.lock:self.process=None
        return progress['seconds']>0

	"""One FFmpeg push per live session. Multiple senders run concurrently, one per YouTube event."""
	import os, time, threading, subprocess
	from collections import deque
	from curl_cffi import requests
	from flv_bridge import FLVBridge, FormatChanged

	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}
	# Timestamps of source resolution changes seen by this sender (anti-flap window).
	self.flap=deque()
	# 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
	def set(self,**kw):
	with self.lock:self.data.update(kw)
	def snapshot(self):
	with self.lock:return dict(self.data)
	def terminate(self):
	with self.lock:p=self.process
	if p and p.poll() is None:
	try:p.terminate()
	except OSError:pass
	def send_mode(self):
	"""copy -> forward the source untouched (original quality, no extra CPU)
	fixed1080 -> re-encode onto a constant 1080x1920 canvas
	fixed720 -> re-encode onto a constant 720x1280 canvas (lighter on CPU)
	"""
	mode=os.getenv('YOUTUBE_SEND_MODE','copy').strip().lower()
	if mode not in ('copy','fixed1080','fixed720'):
	raise ValueError('YOUTUBE_SEND_MODE يجب أن يكون copy أو fixed1080 أو fixed720')
	# A source already proven to flap stays on the constant canvas for this whole event.
	if self.sticky_mode:return self.sticky_mode
	# copy mode only: if the phone keeps flipping resolution, stop flapping the
	# YouTube session and absorb the changes on a constant canvas instead.
	if mode=='copy':
	flap=self.flap
	now=time.monotonic()
	while flap and now-flap[0]>600:flap.popleft()
	if len(flap)>=int(os.getenv('YOUTUBE_FLAP_LIMIT','4')):
	mode=os.getenv('YOUTUBE_FLAP_FALLBACK_MODE','fixed720').strip().lower()
	if mode not in ('fixed1080','fixed720'):mode='fixed720'
	self.sticky_mode=mode
	return mode
	def note_format_change(self):
	self.flap.append(time.monotonic())
	if len(self.flap)>=int(os.getenv('YOUTUBE_FLAP_LIMIT','4')):
	self.log('system','تغيّرت أبعاد المصدر عدة مرات خلال 10 دقائق؛ التحول تلقائيًا إلى كانفس ثابت (%s) لمنع تكرار الانقطاع'%os.getenv('YOUTUBE_FLAP_FALLBACK_MODE','fixed720'))
	@staticmethod
	def canvas(width,height):
	return ('scale=w=%d:h=%d:force_original_aspect_ratio=decrease:flags=lanczos,'
	'pad=%d:%d:(ow-iw)/2:(oh-ih)/2:color=black,setsar=1,format=yuv420p'%(width,height,width,height))
	def command(self,destination):
	# Advertise Enhanced RTMP for network output; local test files need no protocol option.
	mode=self.send_mode()
	network=destination.startswith(('rtmp://','rtmps://'))
	enhanced=['-rtmp_enhanced_codecs','hvc1'] if network and mode=='copy' and os.getenv('YOUTUBE_VIDEO_MODE','original').strip().lower()=='original' else []
	if mode=='copy':
	video=['-c:v','copy']
	else:
	width,height,bitrate=(1080,1920,'6000k') if mode=='fixed1080' else (720,1280,'4000k')
	bufsize=str(int(int(bitrate[:-1])*1.5))+'k'
	# A constant canvas is the only thing that guarantees YouTube always receives
	# frames whose real size matches the signalled one, whatever the phone does.
	video=['-vf',self.canvas(width,height),'-c:v','libx264','-preset',os.getenv('YOUTUBE_X264_PRESET','veryfast'),
	'-tune','zerolatency','-profile:v','high','-crf',os.getenv('YOUTUBE_CRF','22'),
	'-maxrate',bitrate,'-bufsize',bufsize,'-g','60','-keyint_min','30','-sc_threshold','0',
	'-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 deliberately delays the YouTube feed behind the live TikTok source
	# and keeps that much data buffered. Two effects, both free of CPU cost: a
	# source hiccup is covered by the buffer instead of starving YouTube's ingest
	# (which it reports as videoIngestionStarved = viewer buffering), and packets
	# are retained across a reconnect instead of being dropped, which is what made
	# segments go missing when the streamer restarted. Cost is latency only.
	ts=int(os.getenv('YOUTUBE_FIFO_TIMESHIFT','10'))
	# 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',
	'-drop_pkts_on_overflow','1','-attempt_recovery','1','-restart_with_keyframe','1',
	'-recovery_wait_time',os.getenv('YOUTUBE_RECOVERY_WAIT','1'),
	*(('-timeshift',str(ts),'-queue_size',str(q)) if ts>0 else ())]
	else:
	output=['-flvflags','no_duration_filesize','-f','flv']
	return [self.ffmpeg,'-hide_banner','-loglevel','error','-nostats','-progress','pipe:1',
	'-fflags','+genpts','-i','pipe:0','-map','0:v:0','-map','0:a:0?',
	*video,'-c:a','aac','-b:a','128k','-ar','48000',
	'-af','aresample=async=1:first_pts=0','-avoid_negative_ts','make_zero',
	enhanced,'-rw_timeout','20000000',output,destination]
	def send(self,user,candidate,destination,stop):
	# YouTube's ingest is created with cdn.resolution/frameRate = 'variable', i.e. it
	# auto-detects the incoming resolution, so a source resolution change does not
	# require a new session. Tearing the session down is what produced the visible
	# gaps and the reconnect storm. The guard stays available behind an env var in
	# case YouTube's own healthStatus ever reports videoResolutionUnsupported.
	guard=os.getenv('YOUTUBE_FORMAT_GUARD','off').lower() in ('1','on','true','yes')
	bridge=FLVBridge(detect_format_change=(self.send_mode()=='copy' and guard))
	response=None;p=None;error_reader=None;done=threading.Event()
	# Set when we deliberately tear FFmpeg down because the source resolution
	# changed; suppresses the misleading "FFmpeg exited with code N" log.
	format_changed=False
	progress={'last':time.monotonic(),'seconds':-1}
	self.set(status='الاتصال بيوتيوب',source=candidate['label'],output_seconds=0,last_output_at=None)
	try:
	response=requests.get(candidate['url'],headers={'Referer':'https://www.tiktok.com/',**self.headers_for(user)},impersonate='chrome',stream=True,timeout=20)
	response.raise_for_status()
	p=subprocess.Popen(self.command(destination),stdin=subprocess.PIPE,stdout=subprocess.PIPE,stderr=subprocess.PIPE,bufsize=0)
	with self.lock:self.process=p
	diagnostics=[]
	def read_errors():
	# Classify selected FFmpeg errors; never expose raw stderr or stream URLs.
	patterns=[('Connection refused','رفض اتصال خادم الاستقبال'),('Connection timed out','انتهت مهلة اتصال خادم الاستقبال'),
	('Connection reset by peer','خادم الاستقبال قطع الاتصال أثناء الإرسال'),('Broken pipe','انقطع أنبوب الإرسال إلى خادم الاستقبال'),
	('End of file','خادم الاستقبال أغلق الاتصال'),('Immediate exit requested','طُلب إنهاء FFmpeg قبل اكتمال الإرسال'),
	('Packet mismatch','عدم تطابق حزم الفيديو — تغيّر مقاس المصدر داخل الجلسة'),
	('Input/output error','خطأ إدخال/إخراج في الإرسال'),('Server error','خادم الاستقبال أعاد خطأ'),
	('Invalid data found','بيانات المصدر غير صالحة للقراءة'),('Video codec','عدم توافق ترميز الفيديو مع الحاوية'),
	('Invalid argument','أحد إعدادات FFmpeg غير مقبول'),('Non-monotonic','توقيت الفيديو غير متزايد'),
	('Unknown encoder','مرمز غير متوفر'),('Option not found','خيار FFmpeg غير مدعوم')]
	try:
	for line in iter(p.stderr.readline,b''):
	text=line.decode('utf-8','replace')
	for needle,label in patterns:
	if needle.lower() in text.lower() and label not in diagnostics:diagnostics.append(label)
	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 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()
	self.set(status='FFmpeg يرسل — تحقق من معاينة YouTube Studio',output_seconds=round(seconds,1),last_output_at=time.time())
	except (OSError,ValueError):pass
	def watchdog():
	while not done.wait(1):
	if stop.is_set() or time.monotonic()-progress['last']>60:
	if p.poll() is None:
	try:p.kill()
	except OSError:pass
	return
	threading.Thread(target=read_progress,daemon=True).start()
	threading.Thread(target=watchdog,daemon=True).start()
	for chunk in response.iter_content():
	if stop.is_set() or p.poll() is not None:break
	if chunk:
	# Unbuffered writes may be partial; send every byte.
	view=memoryview(bridge.feed(chunk))
	while view and not stop.is_set():
	n=p.stdin.write(view)
	if not n:raise BrokenPipeError()
	view=view[n:]
	except FormatChanged:
	format_changed=True
	raise
	finally:
	if response:response.close()
	if p:
	try:p.stdin.close()
	except (OSError,ValueError):pass
	try:p.wait(timeout=10)
	except subprocess.TimeoutExpired:p.kill();p.wait()
	done.set()
	if error_reader:error_reader.join(timeout=1)
	if p.returncode and not stop.is_set() and not format_changed:
	self.log(user,'FFmpeg خرج برمز '+str(p.returncode)+(' — '+'؛ '.join(diagnostics[-3:]) if diagnostics else '؛ لم يظهر سبب مصنّف'))
	with self.lock:self.process=None
	return progress['seconds']>0