Spaces:
Running
Running
| """Media ingest helpers. Video path is RTMPS; OAuth is optional metadata only. | |
| Never log the stream key or the full RTMP URL. | |
| """ | |
| import os, threading, time | |
| from datetime import datetime | |
| from zoneinfo import ZoneInfo | |
| APP_VERSION = 'relay-v2.45-probed-canvas' | |
| def env_rtmp_destination(): | |
| """Build rtmps://…/key from secrets. Empty string if not configured.""" | |
| url = (os.getenv('YOUTUBE_RTMP_URL') or os.getenv('YOUTUBE_RTMPS_URL') or '').strip() | |
| key = (os.getenv('YOUTUBE_STREAM_KEY') or os.getenv('YOUTUBE_RTMP_KEY') or '').strip() | |
| if url.startswith(('rtmp://', 'rtmps://')) and key: | |
| if url.rstrip('/').endswith('/' + key): | |
| return url | |
| return url.rstrip('/') + '/' + key | |
| if url.startswith(('rtmp://', 'rtmps://')): | |
| return url | |
| if key: | |
| return 'rtmps://a.rtmps.youtube.com:443/live2/' + key | |
| return '' | |
| def rtmp_configured(): | |
| return bool(env_rtmp_destination()) | |
| def max_concurrent_relays(): | |
| """Cap on simultaneous FFmpeg publishers. Not a quality guarantee. | |
| Default 2: cpu-basic has carried 1 comfortably and 2 at high quota; 3+ is | |
| not a pass. Operators may raise MAX_CONCURRENT_RELAYS; health reports the | |
| value actually in force. Ceiling 8 is only a fork-bomb guard. | |
| """ | |
| raw = os.getenv('MAX_CONCURRENT_RELAYS') | |
| if raw is None or str(raw).strip() == '': | |
| raw = os.getenv('YOUTUBE_MAX_EVENTS') or '0' | |
| try: | |
| n = int(str(raw).strip()) | |
| except (TypeError, ValueError): | |
| n = 0 | |
| # 0 = every LIVE account may copy (one FFmpeg per user). No hidden ceiling. | |
| return n if n > 0 else 0 | |
| def light_max(): | |
| """Operator opt-in ceiling for legacy x264 fallback; zero disables video re-encoding. | |
| The default is deliberately zero: AVC and Enhanced-RTMP HEVC are copied, including decoder | |
| configuration replacement. Historical CBR/x264 modes remain available only by explicitly | |
| raising this limit. | |
| """ | |
| try: | |
| n = int(str(os.getenv('YOUTUBE_LIGHT_MAX', '0')).strip() or '0') | |
| except (TypeError, ValueError): | |
| n = 0 | |
| return max(0, min(n, 8)) | |
| def format_session_title(broadcaster, when=None): | |
| """Baghdad clock of this TikTok LIVE session — not process start time.""" | |
| when = when or datetime.now(ZoneInfo('Asia/Baghdad')) | |
| values = { | |
| 'broadcaster': broadcaster or '', | |
| 'name': broadcaster or '', | |
| 'nickname': broadcaster or '', | |
| 'date': when.strftime('%Y-%m-%d'), | |
| 'time': when.strftime('%H:%M'), | |
| 'title': broadcaster or '', | |
| 'username': '', | |
| 'hashtags': '', | |
| 'room_id': '', | |
| 'live_title': '', | |
| } | |
| tpl = (os.getenv('YOUTUBE_TITLE_TEMPLATE') or '{broadcaster} | {date} {time}').strip() | |
| try: | |
| title = tpl.format(**values).strip()[:100] | |
| except (KeyError, ValueError, IndexError, AttributeError): | |
| title = '%s | %s %s' % (values['broadcaster'], values['date'], values['time']) | |
| return title, values['date'], values['time'] | |
| class RelayHub: | |
| """At most one FFmpeg relay per broadcaster. Owners: 'oauth' or 'direct'.""" | |
| def __init__(self): | |
| self.lock = threading.RLock() | |
| self.slots = {} | |
| self.restarts = {} | |
| # Atomic per-broadcaster reservations prevent simultaneous format fallbacks from | |
| # exceeding YOUTUBE_LIGHT_MAX while their existing FFmpeg children are still exiting. | |
| self.light_reservations = set() | |
| def _ffmpeg_alive(cur): | |
| snd = (cur or {}).get('sender') | |
| try: | |
| return bool(snd and snd.alive()) | |
| except Exception: | |
| return False | |
| def _supervisor_alive(cur): | |
| if not cur: | |
| return False | |
| th = cur.get('thread') | |
| if th and th.is_alive(): | |
| return True | |
| return RelayHub._ffmpeg_alive(cur) | |
| def occupy(self, user, owner, sender=None, thread=None, room_id='', title='', dest=''): | |
| with self.lock: | |
| cur = self.slots.get(user) | |
| if self._supervisor_alive(cur): | |
| return False | |
| if cur: | |
| self.restarts[user] = int(self.restarts.get(user) or 0) + 1 | |
| self.slots[user] = { | |
| 'owner': owner, 'sender': sender, 'thread': thread, | |
| 'room_id': room_id, 'title': title, 'started': time.time(), | |
| 'stop': None, | |
| 'dest': dest or ('env' if owner == 'direct' else 'oauth'), | |
| } | |
| return True | |
| def update(self, user, **kw): | |
| with self.lock: | |
| if user in self.slots: | |
| self.slots[user].update(kw) | |
| def release(self, user, owner=None): | |
| with self.lock: | |
| cur = self.slots.get(user) | |
| if not cur: | |
| self.light_reservations.discard(user) | |
| return | |
| if owner and cur.get('owner') != owner: | |
| return | |
| self.slots.pop(user, None) | |
| self.light_reservations.discard(user) | |
| def busy(self, user): | |
| with self.lock: | |
| return self._supervisor_alive(self.slots.get(user)) | |
| def owner(self, user): | |
| with self.lock: | |
| return (self.slots.get(user) or {}).get('owner') | |
| def snapshot(self, user=None): | |
| with self.lock: | |
| if user is not None: | |
| return dict(self.slots.get(user) or {}) | |
| return {u: dict(v) for u, v in self.slots.items()} | |
| def active_count(self): | |
| """FFmpeg processes actually pushing — not idle supervisor threads.""" | |
| with self.lock: | |
| return sum(1 for v in self.slots.values() if self._ffmpeg_alive(v)) | |
| def _light_count_locked(self, except_user=None): | |
| """Count active encoders plus reserved slots, without counting either twice.""" | |
| active=set();n=0 | |
| for u,v in self.slots.items(): | |
| if except_user and u==except_user:continue | |
| if u in self.light_reservations: | |
| active.add(u);n+=1 | |
| continue | |
| if not self._ffmpeg_alive(v):continue | |
| snd=v.get('sender');mode='' | |
| try: | |
| mode=str(getattr(snd,'sticky_mode',None) or '') | |
| if not mode and snd: | |
| snap=snd.snapshot() or {};mode=str(snap.get('send_mode') or '') | |
| except Exception:pass | |
| if mode in ('matchgop','fixed720','fixed1080'): | |
| active.add(u);n+=1 | |
| n+=sum(1 for u in self.light_reservations if u!=except_user and u not in active) | |
| return n | |
| def light_count(self, except_user=None): | |
| """Publishers and atomic reservations using matchgop/fixed720/fixed1080.""" | |
| with self.lock:return self._light_count_locked(except_user) | |
| def light_available(self, except_user=None): | |
| return self.light_count(except_user) < light_max() | |
| def reserve_light(self,user): | |
| """Atomically claim one per-user encoder slot before changing a live publisher.""" | |
| user=str(user or '') | |
| if not user:return False | |
| with self.lock: | |
| if user in self.light_reservations:return True | |
| limit=light_max() | |
| if limit<=0 or self._light_count_locked(except_user=user)>=limit:return False | |
| self.light_reservations.add(user) | |
| return True | |
| def release_light(self,user): | |
| with self.lock:self.light_reservations.discard(str(user or '')) | |
| def restart_count(self, user): | |
| with self.lock: | |
| return int(self.restarts.get(user) or 0) | |
| def at_capacity(self, except_user=None): | |
| with self.lock: | |
| n = sum(1 for u, v in self.slots.items() | |
| if u != except_user and self._ffmpeg_alive(v)) | |
| cap = max_concurrent_relays() | |
| if cap <= 0: | |
| return False | |
| return n >= cap | |
| def env_key_busy(self, except_user=None): | |
| """A single env stream key can only accept one publisher (any owner).""" | |
| with self.lock: | |
| for u, v in self.slots.items(): | |
| if except_user and u == except_user: | |
| continue | |
| if v.get('dest') == 'env' and self._supervisor_alive(v): | |
| return True | |
| return False | |
| hub = RelayHub() | |