#!/usr/bin/env python3
from pathlib import Path
from zoneinfo import ZoneInfo
import datetime,fcntl,hashlib,json,os,time,urllib.parse,urllib.request
ROOT=Path('/data/video-pipeline/HaTramAudio/project/028-Nguoi-Duoc-Goi-Ten-Cuoi-Cung'); RUN='run-20260722T063856Z-87753fd5'; CANON=json.loads((ROOT/'script/project-manifest.json').read_text())['canon_sha256']
LOG=ROOT/'log'; WORK=ROOT/f'work/schedule/{RUN}'; CFG=Path.home()/'.config/ha-tram-audio/postiz.env'; INFO=ROOT/'output/info.txt'; VIDEO=ROOT/'output/final-upload.mp4'
def guard():
 d=json.loads((ROOT/'.ownership-lock.json').read_text())
 if d.get('project_id')!=ROOT.name or d.get('run_id')!=RUN or d.get('owner')!='Levy': raise RuntimeError('ownership guard mismatch')
guard()
manifest=json.loads((ROOT/'script/project-manifest.json').read_text())
if manifest.get('steps',{}).get('schedule')!='pending': raise RuntimeError('schedule gate is not pending')
for dep in manifest.get('dependencies',{}).get('schedule',[]):
 if manifest.get('steps',{}).get(dep)!='completed': raise RuntimeError('schedule dependency incomplete: '+dep)
lease=json.loads((ROOT/'script/gate-leases/schedule.json').read_text())
expires=datetime.datetime.fromisoformat(lease['expires_at'].replace('Z','+00:00'))
if lease.get('status')!='active' or lease.get('run_id')!=RUN or lease.get('owner')!='Levy' or expires<=datetime.datetime.now(datetime.timezone.utc): raise RuntimeError('active schedule lease required')
execution_lock=(ROOT/'script/gate-leases/.schedule-execution.lock').open('a+')
try: fcntl.flock(execution_lock,fcntl.LOCK_EX|fcntl.LOCK_NB)
except BlockingIOError: raise RuntimeError('another schedule process is already active')

def now(): return datetime.datetime.now(datetime.timezone.utc).isoformat().replace('+00:00','Z')
def sha(p):
 h=hashlib.sha256()
 with p.open('rb') as f:
  for b in iter(lambda:f.read(8*1024*1024),b''):h.update(b)
 return h.hexdigest()
def atomic(p,d):
 guard()
 p.parent.mkdir(parents=True,exist_ok=True);t=p.with_name('.'+p.name+'.tmp');t.write_text(json.dumps(d,ensure_ascii=False,indent=2)+'\n');os.replace(t,p)
def load_env():
 d=dict(os.environ)
 for raw in CFG.read_text().splitlines():
  s=raw.strip()
  if s and not s.startswith('#') and '=' in s:
   k,v=s.split('=',1);d.setdefault(k.strip(),v.strip().strip("\"'"))
 for k in ['POSTIZ_BASE_URL','POSTIZ_API_KEY','POSTIZ_FACEBOOK_INTEGRATION_ID','POSTIZ_YOUTUBE_INTEGRATION_ID','POSTIZ_TIMEZONE','POSTIZ_SLOTS']:
  if not d.get(k):raise RuntimeError('missing Postiz config')
 return d
E=load_env();BASE=E['POSTIZ_BASE_URL'].rstrip('/');KEY=E['POSTIZ_API_KEY'];FB=E['POSTIZ_FACEBOOK_INTEGRATION_ID'];YT=E['POSTIZ_YOUTUBE_INTEGRATION_ID'];TARGET={FB,YT};TZ=ZoneInfo(E['POSTIZ_TIMEZONE']);SLOTS=[tuple(map(int,x.strip().split(':'))) for x in E['POSTIZ_SLOTS'].split(',')]
def get(path):
 q=urllib.request.Request(BASE+path,headers={'Authorization':KEY,'Accept':'application/json'})
 with urllib.request.urlopen(q,timeout=60) as r:return json.load(r)
def post(path,payload):
 q=urllib.request.Request(BASE+path,data=json.dumps(payload,ensure_ascii=False).encode(),headers={'Authorization':KEY,'Content-Type':'application/json','Accept':'application/json'},method='POST')
 with urllib.request.urlopen(q,timeout=120) as r:return r.status,json.load(r)
def items(v):
 if isinstance(v,list):return v
 if isinstance(v,dict):
  for k in ('posts','integrations','data'):
   if isinstance(v.get(k),list):return v[k]
 return []
def iid(p):
 x=p.get('integration');return x.get('id') if isinstance(x,dict) else p.get('integrationId')
def dt(v):
 x=datetime.datetime.fromisoformat(v.replace('Z','+00:00'))
 if x.tzinfo is None:x=x.replace(tzinfo=datetime.timezone.utc)
 return x.astimezone(datetime.timezone.utc).replace(microsecond=0)
def calendar(start,end):
 q=urllib.parse.urlencode({'startDate':start.astimezone(datetime.timezone.utc).isoformat().replace('+00:00','Z'),'endDate':end.astimezone(datetime.timezone.utc).isoformat().replace('+00:00','Z')})
 return items(get('/posts?'+q))
def choose_slot():
 utcnow=datetime.datetime.now(datetime.timezone.utc);end=utcnow+datetime.timedelta(days=31);posts=calendar(utcnow-datetime.timedelta(minutes=5),end);occ={FB:set(),YT:set()};ignored=0
 for p in posts:
  i=iid(p);state=str(p.get('state','')).upper();raw=p.get('publishDate') or p.get('date')
  if i not in TARGET:ignored+=1;continue
  if state in {'QUEUE','PUBLISHED'} and raw:occ[i].add(dt(raw))
 localnow=utcnow.astimezone(TZ)
 for n in range(32):
  day=localnow.date()+datetime.timedelta(days=n)
  for h,m in sorted(SLOTS):
   local=datetime.datetime.combine(day,datetime.time(h,m),TZ);u=local.astimezone(datetime.timezone.utc).replace(microsecond=0)
   if local>localnow and u not in occ[FB] and u not in occ[YT]:return local,u,{'facebook_occupied_count':len(occ[FB]),'youtube_occupied_count':len(occ[YT]),'other_posts_ignored':ignored}
 raise RuntimeError('no common slot')
def matches(slot):
 found=[]
 for p in calendar(slot-datetime.timedelta(minutes=10),slot+datetime.timedelta(minutes=10)):
  i=iid(p);raw=p.get('publishDate') or p.get('date')
  if i in TARGET and raw and dt(raw)==slot:
   found.append({'post_id':p.get('id') or p.get('postId'),'integration_id':i,'state':str(p.get('state','')).upper(),'publishDate':raw})
 return found
def finalize(intent,found):
 queues=[x for x in found if x['state']=='QUEUE'];by={x['integration_id']:x for x in queues}
 if set(by)!=TARGET or len(queues)!=2:return False
 accepted=intent.get('accepted_post_ids')
 if accepted and any(by[i]['post_id']!=accepted.get(i) for i in TARGET):return False
 upload=json.loads((LOG/'upload.json').read_text());ready=json.loads((LOG/'publish-ready.json').read_text());receipt={'schema_version':1,'project_id':ROOT.name,'gate':'schedule','status':'completed','verified':True,'run_id':RUN,'scheduled':True,'published':False,'source_canon_sha256':CANON,'source_poster_sha256':upload['source_poster_sha256'],'source_layout_sha256':upload['source_layout_sha256'],'slot_local':intent['slot_local'],'slot_utc':intent['slot_utc'],'artifact_path':str(VIDEO),'artifact_sha256':upload['artifact_sha256'],'video_media':upload['video_media'],'thumbnail_media':upload['thumbnail_media'],'accepted_posts':[{'postId':by[i]['post_id'],'integration':i} for i in (FB,YT)],'calendar_verification':[by[FB],by[YT]],'idempotency_key':intent['idempotency_key'],'completed_at':now()};atomic(LOG/'schedule.json',receipt);intent.update({'status':'scheduled_verified','verified':True,'post_ids':{i:by[i]['post_id'] for i in (FB,YT)},'terminal_at':now()});atomic(LOG/'schedule-intent.json',intent);print(json.dumps({'status':'completed','verified':True,'slot_local':receipt['slot_local'],'posts':receipt['calendar_verification']},ensure_ascii=False));return True
# Validate upstream and integrations before any schedule write.
upload=json.loads((LOG/'upload.json').read_text());ready=json.loads((LOG/'publish-ready.json').read_text());meta=json.loads((LOG/'metadata.json').read_text())
if upload.get('status')!='completed' or upload.get('verified') is not True or ready.get('readiness')!='ready' or ready.get('verified') is not True:raise RuntimeError('upload/publish-ready authority invalid')
if sha(VIDEO)!=upload.get('artifact_sha256') or sha(INFO)!=meta.get('artifact_sha256'):raise RuntimeError('schedule artifact hash mismatch')
ints={x.get('id'):x for x in items(get('/integrations'))}
if not set(ints)>=TARGET or any(ints[i].get('disabled') is True for i in TARGET):raise RuntimeError('target integrations missing or disabled')
if ints[FB].get('identifier')!='facebook' or ints[YT].get('identifier')!='youtube' or any(ints[i].get('name')!='Hạ Trâm Audio' for i in TARGET):raise RuntimeError('target integration identity mismatch')
intent_path=LOG/'schedule-intent.json'
if intent_path.exists():
 intent=json.loads(intent_path.read_text())
 if intent.get('project_id')!=ROOT.name or intent.get('run_id')!=RUN or intent.get('source_canon_sha256')!=CANON:raise RuntimeError('stale/foreign schedule intent')
 if intent.get('video_media_id')!=upload['video_media']['id'] or intent.get('thumbnail_media_id')!=upload['thumbnail_media']['id']:raise RuntimeError('schedule intent media binding mismatch')
 slot=dt(intent['slot_utc'])
 for attempt in range(1,11):
  found=matches(slot)
  if finalize(intent,found):raise SystemExit(0)
  time.sleep(min(30,attempt*3))
 raise RuntimeError('existing schedule intent not reconciled; refusing duplicate POST')
local,slot,scan=choose_slot();title,sep,description=INFO.read_text().strip().partition('\n\n')
if not sep or not 2<=len(title)<=100 or not description.strip():raise RuntimeError('metadata title/description invalid')
video=upload['video_media'];thumb=upload['thumbnail_media'];payload={'type':'schedule','date':slot.isoformat().replace('+00:00','Z'),'shortLink':False,'tags':[],'posts':[{'integration':{'id':FB},'value':[{'content':description.strip(),'image':[video]}],'settings':{'__type':'facebook'}},{'integration':{'id':YT},'value':[{'content':description.strip(),'image':[video]}],'settings':{'__type':'youtube','title':title,'type':'public','selfDeclaredMadeForKids':'no','thumbnail':thumb,'tags':[]}}]};serialized=json.dumps(payload,ensure_ascii=False,sort_keys=True);idem=hashlib.sha256((ROOT.name+RUN+CANON+slot.isoformat()+video['id']+thumb['id']+sha(VIDEO)).encode()).hexdigest();intent={'schema_version':1,'project_id':ROOT.name,'run_id':RUN,'status':'creating','verified':False,'idempotency_key':idem,'source_canon_sha256':CANON,'slot_local':local.isoformat(),'slot_utc':payload['date'],'payload_sha256':hashlib.sha256(serialized.encode()).hexdigest(),'video_media_id':video['id'],'thumbnail_media_id':thumb['id'],'calendar_precheck':scan,'created_at':now()};atomic(intent_path,intent);code,response=post('/posts',payload);atomic(WORK/'posts-response.json',{'http_status':code,'response':response,'received_at':now()})
accepted_rows=items(response);accepted={str(x.get('integration')):x.get('postId') for x in accepted_rows if isinstance(x,dict)}
if set(accepted)!=TARGET or any(not accepted[i] for i in TARGET):raise RuntimeError('schedule POST response did not return exact two target post IDs; intent preserved for reconciliation')
intent.update({'http_status':code,'status':'accepted','accepted_post_ids':accepted,'accepted_at':now()});atomic(intent_path,intent)
for attempt in range(1,11):
 found=matches(slot)
 if finalize(intent,found):raise SystemExit(0)
 time.sleep(min(30,attempt*3))
intent.update({'status':'ambiguous','verified':False,'updated_at':now()});atomic(intent_path,intent);raise RuntimeError('calendar did not verify both QUEUE records; refusing duplicate POST')