#!/usr/bin/env python3 """ animily_music 예약 업로드 — auto_shorts와 동일한 큐 방식 upload_queue.json에서 현재 시간에 해당하는 항목만 업로드 crontab: 매시간 실행 0 * * * * cd /home/javamon/project/animily_music && venv/bin/python3 upload_scheduled.py >> logs/upload.log 2>&1 """ import json import logging import os import pickle import sys from datetime import datetime, timezone, timedelta from pathlib import Path from google.auth.transport.requests import Request from googleapiclient.discovery import build from googleapiclient.http import MediaFileUpload logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s") logger = logging.getLogger(__name__) KST = timezone(timedelta(hours=9)) PROJECT_DIR = Path("/home/javamon/project/animily_music") QUEUE_PATH = PROJECT_DIR / "upload_queue.json" TOKEN_PATH = PROJECT_DIR / "token_animily_music.pickle" DESCRIPTION_TEMPLATE = """{title} ━━━━━━━━━━━━━━━━━━━━━━━━ 🎵 무료 BGM | 사용 시 출처 표기 부탁드립니다 아래 내용을 영상 설명란에 복사해 주세요: 🎵 Music by 애니밀리 뮤직 (Animily Music) 🔗 https://www.youtube.com/@animily-music ━━━━━━━━━━━━━━━━━━━━━━━━ ✅ 사용 가능 범위 • YouTube, Instagram, TikTok, 블로그 등 개인 콘텐츠 무료 사용 • 상업적 용도 (광고, 기업 영상 등) 무료 사용 • 사용 시 설명란에 출처 (채널명 + 링크) 표기 필수 ❌ 금지 사항 • 음원 단독 재업로드 및 재배포 • 자신의 음악인 것처럼 등록 또는 판매 • 음원 파일 자체를 다운로드 링크로 공유 ━━━━━━━━━━━━━━━━━━━━━━━━ 🐾 애니밀리 — 반려동물 행동교정 플랫폼 📲 앱 다운로드: 앱스토어/구글플레이에서 '애니밀리' 검색 🌐 https://animily.co.kr/ 🤖 제작: AI 자동 생성 콘텐츠 (ACE-Step 1.5) #무료BGM #저작권프리 #배경음악 #BGM #copyrightfree #royaltyfree #backgroundmusic #animily """ FIXED_COMMENT = """🎵 무료 BGM — 사용 시 출처만 남겨주세요! ✅ 사용 방법: 아래 내용을 영상 설명란에 붙여넣기 ━━━━━━━━━━━━━━━━━━ 🎵 Music by 애니밀리 뮤직 (Animily Music) 🔗 https://www.youtube.com/@animily-music ━━━━━━━━━━━━━━━━━━ YouTube, Instagram, TikTok, 블로그 등 자유롭게 사용 가능합니다. 상업적 용도도 무료! 출처만 표기해 주세요 🙏 🎵 Free BGM — Just credit us! Free for personal and commercial use. No attribution needed ✅""" # 재생목록 캐시 _playlist_cache = {} def get_youtube_client(): with open(TOKEN_PATH, "rb") as f: creds = pickle.load(f) if creds.expired and creds.refresh_token: creds.refresh(Request()) with open(TOKEN_PATH, "wb") as f: pickle.dump(creds, f) logger.info("Token refreshed") return build("youtube", "v3", credentials=creds) def get_or_create_playlist(youtube, category: str) -> str: if category in _playlist_cache: return _playlist_cache[category] # 기존 재생목록 검색 try: pl = youtube.playlists().list(part="snippet", mine=True, maxResults=50).execute() for item in pl.get("items", []): if item["snippet"]["title"] == category: _playlist_cache[category] = item["id"] return item["id"] except Exception as e: logger.error(f"Playlist search error: {e}") # 없으면 생성 try: body = { "snippet": { "title": category, "description": f"ANIMILY {category} - AI 생성 저작권 프리 BGM 모음", }, "status": {"privacyStatus": "public"}, } resp = youtube.playlists().insert(part="snippet,status", body=body).execute() pl_id = resp["id"] _playlist_cache[category] = pl_id logger.info(f"Created playlist: {category} ({pl_id})") return pl_id except Exception as e: logger.error(f"Playlist create error: {e}") return None def upload_one(youtube, item: dict) -> bool: video_path = item["video_path"] if not os.path.exists(video_path): logger.error(f"Video not found: {video_path}") return False title = item["title"][:100] description = DESCRIPTION_TEMPLATE.format(title=title) tags = item.get("youtube_tags", [])[:30] body = { "snippet": { "title": title, "description": description, "tags": tags, "categoryId": "10", "defaultLanguage": "ko", }, "status": { "privacyStatus": "public", "madeForKids": False, "selfDeclaredMadeForKids": False, }, } logger.info(f"Uploading: {title}") media = MediaFileUpload(video_path, mimetype="video/mp4", resumable=True, chunksize=50*1024*1024) request = youtube.videos().insert(part="snippet,status", body=body, media_body=media) response = None while response is None: status, response = request.next_chunk() video_id = response.get("id") logger.info(f"Upload complete: https://youtu.be/{video_id}") # 썸네일 thumb = item.get("thumbnail_path", "") if thumb and os.path.exists(thumb): try: youtube.thumbnails().set( videoId=video_id, media_body=MediaFileUpload(thumb, mimetype="image/png"), ).execute() logger.info("Thumbnail set") except Exception as e: logger.warning(f"Thumbnail failed: {e}") # 재생목록 category = item.get("playlist_category", "") if category: pl_id = get_or_create_playlist(youtube, category) if pl_id: try: youtube.playlistItems().insert( part="snippet", body={"snippet": {"playlistId": pl_id, "resourceId": {"kind": "youtube#video", "videoId": video_id}}}, ).execute() logger.info(f"Added to playlist: {category}") except Exception as e: logger.warning(f"Playlist add failed: {e}") # 고정 댓글 try: youtube.commentThreads().insert( part="snippet", body={"snippet": {"videoId": video_id, "topLevelComment": {"snippet": {"textOriginal": FIXED_COMMENT}}}}, ).execute() logger.info("Comment posted") except Exception as e: logger.warning(f"Comment failed: {e}") # 임시 파일 삭제 try: os.remove(video_path) if thumb and os.path.exists(thumb): os.remove(thumb) logger.info(f"Temp files cleaned") except: pass item["uploaded"] = True item["uploaded_at"] = datetime.now(KST).strftime("%Y-%m-%d %H:%M:%S") item["video_id"] = video_id return True def main(): now = datetime.now(KST) current_hour = now.hour if not QUEUE_PATH.exists(): return queue = json.loads(QUEUE_PATH.read_text(encoding="utf-8")) if not queue: return # 현재 시간에 해당하는 미업로드 항목 targets = [item for item in queue if not item.get("uploaded") and item.get("upload_hour") == current_hour] if not targets: return logger.info(f"=== Upload Scheduled ({current_hour}:00) — {len(targets)}건 ===") youtube = get_youtube_client() for item in targets: try: upload_one(youtube, item) except Exception as e: logger.error(f"Upload error: {e}") item["uploaded"] = False item["error"] = str(e)[:200] # 큐 저장 QUEUE_PATH.write_text(json.dumps(queue, ensure_ascii=False, indent=2), encoding="utf-8") logger.info(f"Queue updated: {len([i for i in queue if i.get('uploaded')])} uploaded total") if __name__ == "__main__": main()