| """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}') |
|
|
| |
| 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: |
| |
| 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): |
| |
| |
| |
| |
| 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: |
| |
| 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: |
| |
| time.sleep(1 * (2 ** attempt)) |
| continue |
| raise APIError(None,'networkOrInvalidResponse') from None |
| except Exception: |
| |
| 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()<self.expires:return self.access |
| if not self.configured():raise APIError(401,'missingOAuthSecrets') |
| fields={'client_id':os.environ['GOOGLE_CLIENT_ID'].strip(),'client_secret':os.environ['GOOGLE_CLIENT_SECRET'].strip(), |
| 'refresh_token':os.environ['YOUTUBE_REFRESH_TOKEN'].strip(),'grant_type':'refresh_token'} |
| result=self._request('https://oauth2.googleapis.com/token','POST',urlencode(fields).encode(),{'Content-Type':'application/x-www-form-urlencoded'}) |
| if not result.get('access_token'):raise APIError(401,'missingAccessToken') |
| self.access=result['access_token'];self.expires=time.time()+max(30,int(result.get('expires_in',3600))-90) |
| return self.access |
| def call(self,resource,method='GET',params=None,body=None): |
| token=self.token() |
| self.journal.reserve(1 if method=='GET' else 50) |
| url='https://www.googleapis.com/youtube/v3/'+resource |
| if params:url+='?'+urlencode(params) |
| headers={'Authorization':'Bearer '+token,'Content-Type':'application/json'} |
| try:return self._request(url,method,json.dumps(body).encode() if body is not None else None,headers) |
| except APIError as e: |
| if e.status==401:self.access=None |
| raise |
| def items(self,resource,params,max_pages=10): |
| items=[];query=dict(params) |
| for _ in range(max_pages): |
| r=self.call(resource,params=query);items.extend(r.get('items',[])) |
| if not r.get('nextPageToken'):return items |
| query['pageToken']=r['nextPageToken'] |
| raise APIError(None,'reconciliationPageLimit') |
|
|
