#!/usr/bin/env python3 """렌더 완료 웹진 숏츠를 YouTube에 올리고 성공한 로컬 영상만 삭제한다.""" from __future__ import annotations import argparse import json import logging import os import shutil import sys import time from pathlib import Path import requests BASE_DIR=Path(__file__).resolve().parents[1] if str(BASE_DIR) not in sys.path:sys.path.insert(0,str(BASE_DIR)) from db import get_conn LOCK_NAME='multi_webzine_shorts_upload_worker' # blog-manager 운영 서버는 현재 공인 HTTPS 인증서가 적용되지 않아 # https 기본값을 사용하면 CERTIFICATE_VERIFY_FAILED로 업로드가 중단된다. # 기존 Cafe24 관리자/API와 동일한 HTTP 엔드포인트를 기본값으로 사용하되, # 공인 인증서 적용 후 환경변수로 HTTPS 주소를 지정할 수 있게 유지한다. TOKEN_URL=os.getenv('MULTI_SHORTS_YOUTUBE_TOKEN_URL','http://blog-manager.hongheemarketing.com/client-api/multi_shorts_youtube_token.php') API_KEY=os.getenv('V1_SHORTS_API_KEY','honghee-shorts-secret-2026') TEMP_ROOT=Path(os.getenv('MULTI_SHORTS_TEMP_DIR',str(BASE_DIR/'temp'/'multi_shorts'))) LOGGER=logging.getLogger('multi_shorts_upload_worker') def claim_delete(conn,job_id=0): conn.begin() try: sql="""SELECT p.*,j.realtor_id,j.temp_file_path,j.article_id FROM multi_shorts_platform_posts p INNER JOIN multi_shorts_jobs j ON j.id=p.shorts_job_id WHERE p.platform='youtube' AND p.upload_status IN ('delete_pending','delete_retry_wait') AND COALESCE(p.remote_media_id,'')<>''""" params=[] if job_id:sql+=' AND j.id=%s';params.append(job_id) sql+=' ORDER BY p.id LIMIT 1 FOR UPDATE' with conn.cursor() as cur: cur.execute(sql,params);row=cur.fetchone() if not row:conn.rollback();return None cur.execute("UPDATE multi_shorts_platform_posts SET upload_status='deleting',upload_error=NULL,updated_at=NOW() WHERE id=%s",(row['id'],)) conn.commit();return row except Exception:conn.rollback();raise def delete_youtube_video(video_id:str,access_token:str): response=requests.delete('https://www.googleapis.com/youtube/v3/videos',params={'id':video_id},headers={'Authorization':'Bearer '+access_token},timeout=60) if response.status_code in (204,404):return 'deleted' # 삭제 권한(scope)이 부족한 계정은 공개 노출부터 즉시 중단한다. # 업로드 권한으로 상태 변경이 가능하면 private 전환 후 웹진 연결을 해제한다. if response.status_code in (401,403): private_response=requests.put( 'https://www.googleapis.com/youtube/v3/videos',params={'part':'status'}, headers={'Authorization':'Bearer '+access_token,'Content-Type':'application/json; charset=UTF-8'}, data=json.dumps({'id':video_id,'status':{'privacyStatus':'private'}},ensure_ascii=False).encode('utf-8'),timeout=60, ) if private_response.status_code==200:return 'private' raise RuntimeError(f'YouTube 삭제 및 비공개 전환 실패 DELETE {response.status_code} / PRIVATE {private_response.status_code}: {private_response.text[:800]}') raise RuntimeError(f'YouTube 영상 삭제 실패 HTTP {response.status_code}: {response.text[:800]}') def finish_delete(conn,row,remote_action='deleted'): with conn.cursor() as cur: cur.execute("""UPDATE multi_shorts_platform_posts SET upload_status='cancelled',remote_media_id=NULL, remote_url=NULL,upload_error=NULL,uploaded_at=NULL,updated_at=NOW() WHERE id=%s""",(row['id'],)) cur.execute("""UPDATE multi_shorts_jobs SET queue_status='cancelled',finished_at=NOW(),temp_file_path=NULL, error_message='관리자 요청으로 YouTube 영상 삭제',updated_at=NOW() WHERE id=%s""",(row['shorts_job_id'],)) cur.execute("""INSERT INTO multi_shorts_job_events (shorts_job_id,event_type,from_status,to_status,message,event_data,created_at) VALUES (%s,'youtube_deleted','uploaded','cancelled','YouTube 영상 삭제 및 웹진 연결 해제',%s,NOW())""", (row['shorts_job_id'],json.dumps({'video_id':str(row.get('remote_media_id') or ''),'remote_action':remote_action},ensure_ascii=False))) conn.commit() def fail_delete(conn,row,error): conn.rollback() retry=int(row.get('retry_count') or 0)+1;status='delete_failed' if retry>=3 else 'delete_retry_wait' with conn.cursor() as cur: cur.execute("UPDATE multi_shorts_platform_posts SET upload_status=%s,retry_count=%s,upload_error=%s,updated_at=NOW() WHERE id=%s",(status,retry,str(error)[:60000],row['id'])) conn.commit() def claim(conn,job_id=0): conn.begin() try: sql="""SELECT p.*,j.realtor_id,j.temp_file_path,j.shorts_title,j.hashtags,j.article_id FROM multi_shorts_platform_posts p INNER JOIN multi_shorts_jobs j ON j.id=p.shorts_job_id WHERE p.platform='youtube' AND p.upload_status IN ('pending','retry_wait') AND j.queue_status='rendered'""" params=[] if job_id:sql+=' AND j.id=%s';params.append(job_id) sql+=' ORDER BY p.id LIMIT 1 FOR UPDATE' with conn.cursor() as cur: cur.execute(sql,params);row=cur.fetchone() if not row:conn.rollback();return None cur.execute("UPDATE multi_shorts_platform_posts SET upload_status='uploading',upload_error=NULL,updated_at=NOW() WHERE id=%s",(row['id'],)) conn.commit();return row except Exception:conn.rollback();raise def token(realtor_id): response=requests.post(TOKEN_URL,json={'realtor_id':int(realtor_id)},headers={'X-API-KEY':API_KEY},timeout=40) data=response.json() if response.content else {} if response.status_code!=200 or not data.get('ok') or not data.get('access_token'): raise RuntimeError(str(data.get('error') or f'token HTTP {response.status_code}')) return str(data['access_token']) def upload(video:Path,access_token:str,title:str,description:str,privacy:str): metadata={'snippet':{'title':title[:100],'description':description[:5000],'categoryId':'19'}, 'status':{'privacyStatus':privacy if privacy in ('public','unlisted','private') else 'public','selfDeclaredMadeForKids':False}} init=requests.post('https://www.googleapis.com/upload/youtube/v3/videos',params={'uploadType':'resumable','part':'snippet,status'}, headers={'Authorization':'Bearer '+access_token,'Content-Type':'application/json; charset=UTF-8', 'X-Upload-Content-Type':'video/mp4','X-Upload-Content-Length':str(video.stat().st_size)}, data=json.dumps(metadata,ensure_ascii=False).encode('utf-8'),timeout=60) if init.status_code not in (200,201) or not init.headers.get('Location'): raise RuntimeError(f'YouTube 업로드 세션 생성 실패 HTTP {init.status_code}: {init.text[:500]}') with video.open('rb') as stream: done=requests.put(init.headers['Location'],data=stream,headers={'Content-Type':'video/mp4','Content-Length':str(video.stat().st_size)},timeout=1800) payload=done.json() if done.content else {} video_id=str(payload.get('id') or '') if done.status_code not in (200,201) or not video_id: raise RuntimeError(f'YouTube 영상 업로드 실패 HTTP {done.status_code}: {done.text[:800]}') return video_id,'https://www.youtube.com/shorts/'+video_id def find_thumbnail(row): path=TEMP_ROOT/f"job_{row['shorts_job_id']}"/'youtube_thumbnail.jpg' return path if path.is_file() else None def set_youtube_thumbnail(video_id:str,access_token:str,thumbnail:Path): last_error='' for attempt in range(1,4): with thumbnail.open('rb') as stream: response=requests.post( 'https://www.googleapis.com/upload/youtube/v3/thumbnails/set', params={'videoId':video_id,'uploadType':'media'}, headers={'Authorization':'Bearer '+access_token,'Content-Type':'image/jpeg'}, data=stream,timeout=120, ) if response.status_code==200:return last_error=f'HTTP {response.status_code}: {response.text[:800]}' if attempt<3:time.sleep(attempt*2) raise RuntimeError('YouTube 인트로 썸네일 등록 실패 '+last_error) def finish(conn,row,video_id,url,thumbnail_ok=True,thumbnail_error=''): with conn.cursor() as cur: cur.execute("""UPDATE multi_shorts_platform_posts SET upload_status='uploaded',remote_media_id=%s, remote_url=%s,upload_error=NULL,uploaded_at=NOW(),updated_at=NOW() WHERE id=%s""",(video_id,url,row['id'])) cur.execute("UPDATE multi_shorts_jobs SET queue_status='uploaded',finished_at=NOW(),error_message=NULL,updated_at=NOW() WHERE id=%s",(row['shorts_job_id'],)) cur.execute("INSERT INTO multi_shorts_job_events (shorts_job_id,event_type,from_status,to_status,message,event_data,created_at) VALUES (%s,'youtube_uploaded','rendered','uploaded','YouTube 업로드 완료',%s,NOW())",(row['shorts_job_id'],json.dumps({'video_id':video_id,'url':url,'intro_thumbnail':bool(thumbnail_ok)},ensure_ascii=False))) if not thumbnail_ok: cur.execute("INSERT INTO multi_shorts_job_events (shorts_job_id,event_type,from_status,to_status,message,event_data,created_at) VALUES (%s,'youtube_thumbnail_failed','uploaded','uploaded','인트로 썸네일 등록 실패(영상 업로드는 완료)',%s,NOW())",(row['shorts_job_id'],json.dumps({'video_id':video_id,'error':str(thumbnail_error)[:2000]},ensure_ascii=False))) conn.commit() def fail(conn,row,error): conn.rollback() retry=int(row.get('retry_count') or 0)+1;status='failed' if retry>=3 else 'retry_wait' with conn.cursor() as cur: cur.execute("UPDATE multi_shorts_platform_posts SET upload_status=%s,retry_count=%s,upload_error=%s,updated_at=NOW() WHERE id=%s",(status,retry,str(error)[:60000],row['id'])) conn.commit() def cleanup(conn,row,video): try: if video.is_file():video.unlink() job_dir=TEMP_ROOT/f"job_{row['shorts_job_id']}" if job_dir.is_dir():shutil.rmtree(job_dir) with conn.cursor() as cur:cur.execute("UPDATE multi_shorts_jobs SET temp_file_path=NULL,updated_at=NOW() WHERE id=%s",(row['shorts_job_id'],)) conn.commit();print(f"[MULTI SHORTS LOCAL DELETE] job_id={row['shorts_job_id']} status=deleted") except Exception as exc: conn.rollback();LOGGER.warning('업로드 완료 후 임시파일 삭제 실패: %s',exc) def main(): parser=argparse.ArgumentParser();parser.add_argument('--job-id',type=int,default=0);parser.add_argument('--dry-run',action='store_true');args=parser.parse_args() logging.basicConfig(level=logging.INFO,format='%(asctime)s [%(levelname)s] %(message)s');conn=get_conn();row=None;delete_row=None;locked=False try: with conn.cursor() as cur:cur.execute('SELECT GET_LOCK(%s,0) acquired',(LOCK_NAME,));locked=int((cur.fetchone() or {}).get('acquired') or 0)==1 if not locked:print('[MULTI SHORTS UPLOAD SKIP] another_worker_running');return delete_row=claim_delete(conn,args.job_id) if delete_row: if args.dry_run: with conn.cursor() as cur:cur.execute("UPDATE multi_shorts_platform_posts SET upload_status='delete_pending',updated_at=NOW() WHERE id=%s",(delete_row['id'],)) conn.commit();print(json.dumps({'ok':True,'action':'delete','job_id':delete_row['shorts_job_id'],'video_id':delete_row['remote_media_id']},ensure_ascii=False));return access=token(delete_row['realtor_id']);remote_action=delete_youtube_video(str(delete_row['remote_media_id']),access);finish_delete(conn,delete_row,remote_action) print(f"[MULTI SHORTS YOUTUBE DELETE DONE] job_id={delete_row['shorts_job_id']} action={remote_action}");return row=claim(conn,args.job_id) if not row:print('[MULTI SHORTS UPLOAD DONE] no_rendered_job');return video=Path(str(row.get('temp_file_path') or '')) if not video.is_file():raise RuntimeError('렌더 완료 영상 파일을 찾을 수 없습니다: '+str(video)) if args.dry_run: with conn.cursor() as cur:cur.execute("UPDATE multi_shorts_platform_posts SET upload_status='pending',updated_at=NOW() WHERE id=%s",(row['id'],)) conn.commit();print(json.dumps({'ok':True,'job_id':row['shorts_job_id'],'video':str(video)},ensure_ascii=False));return access=token(row['realtor_id']);title=str(row.get('shorts_title') or '부동산 매물 숏츠');description=(str(row.get('hashtags') or '')+'\n\n#부동산 #매물 #Shorts').strip() thumbnail=find_thumbnail(row) video_id,url=upload(video,access,title,description,str(row.get('privacy_status') or 'public')) thumbnail_ok=False;thumbnail_error='' if thumbnail: try: set_youtube_thumbnail(video_id,access,thumbnail);thumbnail_ok=True except Exception as exc: thumbnail_error=str(exc);LOGGER.warning('%s',thumbnail_error) else: thumbnail_error='운영용 인트로 썸네일 파일을 찾을 수 없습니다.';LOGGER.warning('%s',thumbnail_error) # 영상 업로드가 성공한 뒤 썸네일만 실패한 경우 재실행으로 영상이 중복되지 않게 # 업로드는 완료 처리하고 경고 이벤트를 별도로 기록한다. finish(conn,row,video_id,url,thumbnail_ok,thumbnail_error);cleanup(conn,row,video) print(f"[MULTI SHORTS UPLOAD DONE] job_id={row['shorts_job_id']} video_id={video_id} thumbnail={'set' if thumbnail_ok else 'warning'} url={url}") except Exception as exc: if delete_row: fail_delete(conn,delete_row,exc);LOGGER.exception('YouTube 영상 삭제 실패: %s',exc) elif row: fail(conn,row,exc);LOGGER.exception('YouTube 업로드 실패: %s',exc) else:LOGGER.exception('YouTube 작업 실패: %s',exc) raise SystemExit(1) finally: if locked: try: with conn.cursor() as cur:cur.execute('SELECT RELEASE_LOCK(%s)',(LOCK_NAME,)) except Exception:pass conn.close() if __name__=='__main__':main()