File size: 9,087 Bytes
dc5fd17 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 | """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)
|