import asyncio
import aiohttp
import json
import os
from google.oauth2.credentials import Credentials
from google_auth_oauthlib.flow import InstalledAppFlow
from googleapiclient.discovery import build
from moviepy.editor import VideoFileClip, TextClip, CompositeVideoClip
from tikapi import TikAPI
import platform
# Configuration
NOTEBOOKLM_API_URL = "https://api.notebooklm.google.com/v1/audio" # Hypothetical
YOUTUBE_API_SCOPES = ["https://www.googleapis.com/auth/youtube.upload"]
TIKTOK_API_KEY = "your_tiktok_api_key" # Replace with actual key
YOUTUBE_CLIENT_SECRETS = "client_secrets.json" # OAuth client secrets
OUTPUT_DIR = "output_clips"
# Initialize TikTok API
tikapi = TikAPI(TIKTOK_API_KEY)
async def fetch_notebooklm_audio(youtube_url, output_lang="ko"):
"""Fetch AI podcast audio from NotebookLM."""
async with aiohttp.ClientSession() as session:
payload = {
"youtube_url": youtube_url,
"output_language": output_lang,
"style": "radio_dialogue",
"hosts": 2
}
async with session.post(NOTEBOOKLM_API_URL, json=payload) as response:
if response.status == 200:
data = await response.json()
audio_url = data.get("audio_url")
summary = data.get("summary")
title = data.get("title", "AI Podcast Clip")
return audio_url, summary, title
else:
raise Exception("NotebookLM API error")
async def download_audio(audio_url, output_path):
"""Download audio file."""
async with aiohttp.ClientSession() as session:
async with session.get(audio_url) as response:
if response.status == 200:
with open(output_path, "wb") as f:
f.write(await response.read())
return output_path
else:
raise Exception("Audio download failed")
def create_video_clip(audio_path, output_path, subtitle_text):
"""Create video clip with audio, subtitles, and background."""
# Load audio as video clip
audio_clip = VideoFileClip(audio_path)
# Create subtitle clip
subtitle_clip = TextClip(
subtitle_text,
fontsize=24,
color="white",
bg_color="black",
size=(audio_clip.w, None),
method="caption"
).set_duration(audio_clip.duration)
# Composite video (background + subtitles)
video = CompositeVideoClip([audio_clip.set_audio(audio_clip.audio), subtitle_clip.set_pos(("center", "bottom"))])
video.write_videofile(output_path, codec="libx264", audio_codec="aac")
return output_path
async def post_to_threads(summary, audio_path, video_path):
"""Post to Threads (simplified, API TBD)."""
# Hypothetical Threads API call
async with aiohttp.ClientSession() as session:
payload = {
"text": f"{summary}\n#AIPodcast #KoreanTranslation",
"media": {"audio": audio_path, "video": video_path}
}
async with session.post("https://api.threads.net/v1/post", json=payload) as response:
if response.status == 200:
print("Posted to Threads")
else:
print("Threads posting failed")
async def post_to_tiktok(video_path, caption):
"""Post to TikTok."""
try:
response = tikapi.post_video(
video_file=video_path,
description=f"{caption}\n#AIPodcast #TikTokKorea",
privacy_level="PUBLIC"
)
if response.status_code == 200:
print("Posted to TikTok")
else:
print("TikTok posting failed")
except Exception as e:
print(f"TikTok error: {e}")
def get_youtube_service():
"""Authenticate and return YouTube API service."""
flow = InstalledAppFlow.from_client_secrets_file(YOUTUBE_CLIENT_SECRETS, YOUTUBE_API_SCOPES)
credentials = flow.run_local_server(port=0)
return build("youtube", "v3", credentials=credentials)
async def post_to_youtube(video_path, title, description):
"""Post to YouTube Shorts."""
youtube = get_youtube_service()
request_body = {
"snippet": {
"title": title,
"description": f"{description}\n#AIPodcast #YouTubeShorts",
"categoryId": "22",
"tags": ["AI", "podcast", "Korean"],
"defaultLanguage": "ko"
},
"status": {"privacyStatus": "public"}
}
media_file = video_path
request = youtube.videos().insert(
part="snippet,status",
body=request_body,
media_body=media_file
)
response = request.execute()
print("Posted to YouTube")
return response["id"]
async def main(youtube_url):
"""Main function to process YouTube URL and post to platforms."""
# Create output directory
if not os.path.exists(OUTPUT_DIR):
os.makedirs(OUTPUT_DIR)
# Fetch AI podcast from NotebookLM
audio_url, summary, title = await fetch_notebooklm_audio(youtube_url)
audio_path = os.path.join(OUTPUT_DIR, "podcast_audio.mp3")
video_path = os.path.join(OUTPUT_DIR, "podcast_clip.mp4")
# Download audio
await download_audio(audio_url, audio_path)
# Create video clip with subtitles
create_video_clip(audio_path, video_path, summary[:100]) # Truncate for subtitle
# Preview content
print(f"Title: {title}")
print(f"Summary: {summary}")
print(f"Audio: {audio_path}")
print(f"Video: {video_path}")
confirm = input("Post to platforms? (y/n): ")
if confirm.lower() == "y":
# Post to Threads
await post_to_threads(summary, audio_path, video_path)
# Post to TikTok
await post_to_tiktok(video_path, summary)
# Post to YouTube
await post_to_youtube(video_path, title, summary)
else:
print("Posting cancelled")
if platform.system() == "Emscripten":
asyncio.ensure_future(main("https://www.youtube.com/watch?v=example"))
else:
if __name__ == "__main__":
asyncio.run(main("https://www.youtube.com/watch?v=example"))
