| import sys |
| import os |
| import requests |
| from googleapiclient.discovery import build |
| from googleapiclient.http import MediaFileUpload |
| from google.auth.transport.requests import Request |
| import pickle |
|
|
| TOKEN_FILE = '/app/token.pickle' |
| VIDEO_PATH = '/app/video.mp4' |
|
|
| def get_credentials(): |
| creds = None |
| if os.path.exists(TOKEN_FILE): |
| with open(TOKEN_FILE, 'rb') as f: |
| creds = pickle.load(f) |
| if creds and creds.expired and creds.refresh_token: |
| creds.refresh(Request()) |
| with open(TOKEN_FILE, 'wb') as f: |
| pickle.dump(creds, f) |
| return creds |
|
|
| def download_video(url, path): |
| print(f"Downloading video...") |
| session = requests.Session() |
| for attempt in range(3): |
| try: |
| r = session.get(url, stream=True, timeout=60, headers={ |
| 'User-Agent': 'Mozilla/5.0', |
| 'Referer': 'https://www.pexels.com/' |
| }) |
| with open(path, 'wb') as f: |
| for chunk in r.iter_content(chunk_size=65536): |
| if chunk: |
| f.write(chunk) |
| print("Download complete!") |
| return |
| except Exception as e: |
| print(f"Attempt {attempt+1} failed: {e}") |
| if attempt == 2: |
| raise |
|
|
| def upload_video(title, description): |
| creds = get_credentials() |
| youtube = build('youtube', 'v3', credentials=creds) |
| body = { |
| 'snippet': { |
| 'title': title, |
| 'description': description, |
| 'categoryId': '22', |
| 'tags': ['shorts', 'viral', 'trending'] |
| }, |
| 'status': { |
| 'privacyStatus': 'public', |
| 'selfDeclaredMadeForKids': False |
| } |
| } |
| media = MediaFileUpload(VIDEO_PATH, mimetype='video/mp4', resumable=True) |
| request = youtube.videos().insert(part='snippet,status', body=body, media_body=media) |
| response = None |
| while response is None: |
| status, response = request.next_chunk() |
| if status: |
| print(f"Uploading... {int(status.progress() * 100)}%") |
| print(f"Uploaded! Video ID: {response['id']}") |
| return response['id'] |
|
|
| if __name__ == '__main__': |
| video_url = sys.argv[1] |
| title = sys.argv[2] |
| description = sys.argv[3] |
| download_video(video_url, VIDEO_PATH) |
| video_id = upload_video(title, description) |
| print(f"SUCCESS:{video_id}") |
