| """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<<zeros)-1+(self.read(zeros) if zeros else 0) |
| def se(self): |
| value=self.ue() |
| return (value+1)//2 if value&1 else -(value//2) |
|
|
|
|
| def _rbsp(nal): |
| out=bytearray();zeros=0 |
| for byte in nal: |
| if zeros>=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) |
|
|
