"""Persistent, bounded diagnostics for byte-copy live streams. Events are operational metadata only: no media, URLs, stream keys, cookies, or encoded payloads are written. The JSON array lives under DATA_DIR/logs, appends one compact object per event, and is periodically compacted with atomic replace. """ import json import os import re import threading from datetime import datetime, timezone from pathlib import Path class _Bits: def __init__(self,data): self.data=data;self.bit=0 def read(self,n): if n<0 or self.bit+n>len(self.data)*8:raise ValueError('truncated bitstream') value=0 for _ in range(n): value=(value<<1)|((self.data[self.bit>>3]>>(7-(self.bit&7)))&1);self.bit+=1 return value def ue(self): zeros=0 while self.read(1)==0: zeros+=1 if zeros>31:raise ValueError('invalid Exp-Golomb value') return (1<=2 and byte==3: zeros=0;continue out.append(byte) zeros=zeros+1 if byte==0 else 0 return bytes(out) def _skip_h264_scaling(bits,size): last=next_scale=8 for j in range(size): if next_scale: next_scale=(last+bits.se()+256)%256 last=next_scale or last def _h264_sps_size(nal): data=_rbsp(nal[1:]);b=_Bits(data) profile=b.read(8);b.read(8);b.read(8);b.ue() chroma=1;separate=0 if profile in (44,83,86,100,110,118,122,128,138,139,134,135,144,244): chroma=b.ue() if chroma==3:separate=b.read(1) b.ue();b.ue();b.read(1) if b.read(1): for i in range(8 if chroma!=3 else 12): if b.read(1):_skip_h264_scaling(b,16 if i<6 else 64) b.ue();poc=b.ue() if poc==0:b.ue() elif poc==1: b.read(1);b.se();b.se() for _ in range(b.ue()):b.se() b.ue();b.read(1) width_mbs=b.ue()+1;height_maps=b.ue()+1 frame_only=b.read(1) if not frame_only:b.read(1) b.read(1) crop=(0,0,0,0) if b.read(1):crop=tuple(b.ue() for _ in range(4)) chroma_array=0 if separate else chroma sub_w=(1,2,2,1)[chroma_array];sub_h=(1,2,1,1)[chroma_array] unit_x=1 if chroma_array==0 else sub_w unit_y=(2-frame_only)*(1 if chroma_array==0 else sub_h) width=width_mbs*16-(crop[0]+crop[1])*unit_x height=(2-frame_only)*height_maps*16-(crop[2]+crop[3])*unit_y return (width,height) if width>0 and height>0 else None def _hevc_profile_tier_level(bits,max_sub_layers): bits.read(96) profiles=[];levels=[] for _ in range(max_sub_layers):profiles.append(bits.read(1));levels.append(bits.read(1)) if max_sub_layers: for _ in range(max_sub_layers,8):bits.read(2) for profile,level in zip(profiles,levels): if profile:bits.read(88) if level:bits.read(8) def _hevc_sps_size(nal): data=_rbsp(nal[2:]);b=_Bits(data) b.read(4);max_sub=b.read(3);b.read(1) _hevc_profile_tier_level(b,max_sub) b.ue();chroma=b.ue() separate=b.read(1) if chroma==3 else 0 width=b.ue();height=b.ue();crop=(0,0,0,0) if b.read(1):crop=tuple(b.ue() for _ in range(4)) chroma_array=0 if separate else chroma sub_w=(1,2,2,1)[chroma_array];sub_h=(1,2,1,1)[chroma_array] width-=(crop[0]+crop[1])*sub_w;height-=(crop[2]+crop[3])*sub_h return (width,height) if width>0 and height>0 else None def video_dimensions(codec,config): """Read coded width/height from avcC or hvcC; malformed records return None.""" try: if codec=='avc1': if len(config)<7:return None count=config[5]&31;offset=6 for _ in range(count): size=int.from_bytes(config[offset:offset+2],'big');offset+=2 nal=config[offset:offset+size];offset+=size if nal and nal[0]&31==7:return _h264_sps_size(nal) return None if codec=='hvc1': if len(config)<23:return None offset=23 for _ in range(config[22]): nal_type=config[offset]&63;count=int.from_bytes(config[offset+1:offset+3],'big');offset+=3 for _ in range(count): size=int.from_bytes(config[offset:offset+2],'big');offset+=2 nal=config[offset:offset+size];offset+=size if nal_type==33:return _hevc_sps_size(nal) except (IndexError,ValueError): return None return None class StreamMonitor: def __init__(self,path=None): root=os.getenv('DATA_DIR') or ('/data/tiktok-relay' if Path('/data').is_dir() else './storage') base=Path(root) self.path=Path(path or os.getenv('STREAM_EVENTS_PATH') or base/'logs/stream_events.json') try:self.maximum=max(100,min(int(os.getenv('STREAM_EVENT_MAX','10000')),100000)) except (TypeError,ValueError):self.maximum=10000 self.lock=threading.RLock();self.events=[];self.sequence=0;self.rtmp_opens={} self._load() def _load(self): try: data=json.loads(self.path.read_text()) if isinstance(data,list): self.events=[x for x in data[-self.maximum:] if isinstance(x,dict)] self.sequence=max((int(x.get('sequence') or 0) for x in self.events),default=0) except (OSError,ValueError,TypeError): pass @staticmethod def _safe(value,depth=0): if depth>4:return '…' if isinstance(value,dict): return {str(k)[:80]:StreamMonitor._safe(v,depth+1) for k,v in value.items() if not any(x in str(k).lower() for x in ('url','token','cookie','key','auth'))} if isinstance(value,(list,tuple)):return [StreamMonitor._safe(x,depth+1) for x in value[:50]] if isinstance(value,str): value=re.sub(r'(?i)\b(?:rtmps?|https?)://\S+','«رابط-مخفي»',value) value=re.sub(r'(?i)(sessionid|token|stream[_ -]?key|authorization)\s*[=:]\s*[^\s;&]+', r'\1=«مخفي»',value) return value[:500] if isinstance(value,(int,float,bool)) or value is None:return value return StreamMonitor._safe(str(value),depth+1) def _rewrite(self): self.path.parent.mkdir(parents=True,exist_ok=True) tmp=self.path.with_name(self.path.name+'.tmp') tmp.write_text(json.dumps(self.events,ensure_ascii=False,separators=(',',':'))) try:os.chmod(tmp,0o600) except OSError:pass os.replace(tmp,self.path) def _append(self,item): encoded=json.dumps(item,ensure_ascii=False,separators=(',',':')).encode() try: with self.path.open('r+b') as fh: fh.seek(-1,os.SEEK_END) if fh.read(1)!=b']':raise ValueError('invalid JSON tail') fh.seek(-1,os.SEEK_END) fh.write((b',' if len(self.events)>1 else b'')+encoded+b']');fh.truncate() try:os.chmod(self.path,0o600) except OSError:pass except (OSError,ValueError): self._rewrite() def log(self,broadcaster,event,details=None): try: with self.lock: self.sequence+=1 item={'timestamp':datetime.now(timezone.utc).isoformat(timespec='milliseconds').replace('+00:00','Z'), 'sequence':self.sequence,'broadcaster':str(broadcaster or 'system')[:100], 'event':str(event or 'UNKNOWN')[:80],'details':self._safe(details or {})} self.events.append(item) if len(self.events)>self.maximum: self.events=self.events[-max(100,int(self.maximum*0.9)):];self._rewrite() elif not self.path.exists():self._rewrite() else:self._append(item) return item except (OSError,TypeError,ValueError): return None def read(self,limit=200,broadcaster=None): try:limit=max(1,min(int(limit),2000)) except (TypeError,ValueError):limit=200 with self.lock: rows=self.events if broadcaster:rows=[x for x in rows if x.get('broadcaster')==broadcaster] return [dict(x) for x in rows[-limit:]] def note_rtmp_open(self,broadcaster,details=None): user=str(broadcaster or 'system') with self.lock: count=self.rtmp_opens.get(user,0)+1;self.rtmp_opens[user]=count if count>1: merged={'scope':'youtube_rtmp','session_number':count};merged.update(details or {}) return self.log(user,'RECONNECT',merged) return None monitor=StreamMonitor() def log_event(broadcaster,event,details=None):return monitor.log(broadcaster,event,details) def read_events(limit=200,broadcaster=None):return monitor.read(limit,broadcaster) def note_rtmp_open(broadcaster,details=None):return monitor.note_rtmp_open(broadcaster,details)