"""Google API transport and durable non-secret journal. No credentials in files/logs.""" import copy, json, os, re, ssl, threading, time, uuid from datetime import datetime from zoneinfo import ZoneInfo from urllib.request import Request, urlopen from urllib.error import HTTPError, URLError from urllib.parse import urlencode class APIError(RuntimeError): def __init__(self, status, reason): self.status=status self.reason=re.sub(r'[^a-zA-Z0-9_.-]','',str(reason))[:100] or 'unknown' super().__init__(f'YouTube API: {status or "network"} / {self.reason}') # V8 defaults: name/title/hashtags are pulled from the TikTok account itself. DEFAULT_SETTINGS={'name':'','privacy':'public','title':'{title}', 'description':'بث {name} المباشر\n👤 المذيع: {name} — @{username}\n🔗 https://www.tiktok.com/@{username}\n\n📅 {date} — ⏰ {time} بتوقيت بغداد\n{hashtags}'} OLD_DEFAULT_TITLE='{name} | بث مباشر | {date} — {time}' class Journal: """Schema 7: multiple concurrent events + one reusable ingestion stream per account.""" def __init__(self,path): self.path=path;self.lock=threading.RLock() try:self.path.parent.mkdir(parents=True,exist_ok=True) except OSError:pass try:self.data=json.loads(path.read_text(encoding='utf-8')) except FileNotFoundError: self.data={'schema':7,'instance':uuid.uuid4().hex,'enabled':False,'channel':None, 'streams':{},'events':{},'history':[],'budget':{},'settings':dict(DEFAULT_SETTINGS)} self.change(lambda d:None) except (ValueError,OSError) as e:raise RuntimeError('تعذر قراءة سجل أحداث يوتيوب؛ لا تحذفه. يلزم إصلاح التخزين لمنع التكرار') from None if not isinstance(self.data,dict) or self.data.get('schema') not in (6,7): raise RuntimeError('سجل يوتيوب غير صالح؛ توقف آمن لمنع تكرار الأحداث') if self.data['schema']==6: self.data=self.migrate6(self.data) self.change(lambda d:None) @staticmethod def migrate6(old): """V6 had a single current event and a single reusable stream. V7 keeps one event per live session and one reusable ingestion stream per TikTok account. The old stream is kept for the account that was using it (legacy marker still accepted).""" events={};streams={} cur=old.get('current') if isinstance(cur,dict) and cur.get('session_key') and cur.get('username'): events[cur['session_key']]=cur if old.get('stream_id') or old.get('stream_pending'): streams[cur['username']]={'id':old.get('stream_id'),'pending':bool(old.get('stream_pending')),'legacy':True} settings=old.get('settings') if isinstance(old.get('settings'),dict) else {} clean={**DEFAULT_SETTINGS,**settings} if clean.get('title')==OLD_DEFAULT_TITLE: # Untouched V6/V7 default title -> switch to V8 per-account default. clean['title']=DEFAULT_SETTINGS['title'];clean['name']='' return {'schema':7,'instance':old.get('instance') or uuid.uuid4().hex,'enabled':bool(old.get('enabled')), 'channel':old.get('channel'),'streams':streams,'events':events, 'history':old.get('history') if isinstance(old.get('history'),list) else [], 'budget':old.get('budget') if isinstance(old.get('budget'),dict) else {}, 'settings':clean} def read(self): with self.lock:return copy.deepcopy(self.data) def change(self,fn): with self.lock: new=copy.deepcopy(self.data);fn(new) tmp=self.path.with_suffix('.tmp') tmp.write_text(json.dumps(new,ensure_ascii=False),encoding='utf-8') tmp.replace(self.path) self.data=new def reserve(self,cost): day=datetime.now(ZoneInfo('America/Los_Angeles')).strftime('%Y-%m-%d') def mutate(d): budget=d.get('budget',{}) if budget.get('day')!=day:budget={'day':day,'estimated_units':0} if budget['estimated_units']+cost>int(os.getenv('YOUTUBE_DAILY_BUDGET','8000')): raise APIError(429,'localDailyBudgetReached') budget['estimated_units']+=cost;d['budget']=budget self.change(mutate) class GoogleAPI: def __init__(self,journal): self.journal=journal;self.access=None;self.expires=0;self.lock=threading.Lock() def configured(self): return all(os.getenv(k,'').strip() for k in ('GOOGLE_CLIENT_ID','GOOGLE_CLIENT_SECRET','YOUTUBE_REFRESH_TOKEN')) def _request(self,url,method='GET',data=None,headers=None): # FIX C: retry transient TLS/network failures instead of single urlopen(timeout=25) # Any ephemeral SSLError/URLError previously bubbled as a generic "internal error" # and left a sticky note for 40 minutes. Now we retry with exponential backoff # and only then surface networkOrInvalidResponse. max_attempts = 3 for attempt in range(max_attempts): req=Request(url,data=data,headers=headers or {},method=method) try: with urlopen(req,timeout=25) as response: raw=response.read();return json.loads(raw) if raw else {} except HTTPError as e: # Retry on transient 5xx / 429 (quota) before failing. if e.code in (429, 500, 502, 503, 504) and attempt < max_attempts - 1: time.sleep(1.2 * (attempt + 1)) continue try: err=json.loads(e.read()).get('error',{}) if isinstance(err,str):reason=err else:reason=(err.get('errors') or [{}])[0].get('reason') or err.get('status') or 'httpError' except Exception:reason='httpError' raise APIError(e.code,reason) from None except (URLError, TimeoutError, OSError, ValueError, ssl.SSLError, ssl.SSLWantReadError, ssl.SSLWantWriteError, ssl.SSLEOFError) as e: if attempt < max_attempts - 1: # exponential backoff: 1s, 2s time.sleep(1 * (2 ** attempt)) continue raise APIError(None,'networkOrInvalidResponse') from None except Exception: # Unexpected parse error etc. Don't retry silently; treat as network. if attempt < max_attempts - 1: time.sleep(1 * (2 ** attempt)) continue raise APIError(None,'networkOrInvalidResponse') from None raise APIError(None,'networkOrInvalidResponse') def token(self): with self.lock: if self.access and time.time()