Gan, Jimmy
111 lines
3.8 KiB
Python
111 lines
3.8 KiB
Python
import asyncio
|
|
import html
|
|
import logging
|
|
import aiosqlite
|
|
from googleapiclient.discovery import build
|
|
from googleapiclient.errors import HttpError
|
|
from auth import get_credentials
|
|
from db import DB_PATH
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
async def fetch_subscriptions():
|
|
"""Fetch latest videos from subscribed channels using YouTube Data API v3."""
|
|
try:
|
|
creds = get_credentials()
|
|
except FileNotFoundError as e:
|
|
log.error(f"OAuth setup incomplete: {e}")
|
|
return 0
|
|
except Exception as e:
|
|
log.error(f"OAuth credential error: {e}")
|
|
return 0
|
|
|
|
youtube = build("youtube", "v3", credentials=creds)
|
|
|
|
# Step 1: Get subscribed channel IDs
|
|
channel_ids = []
|
|
try:
|
|
request = youtube.subscriptions().list(
|
|
part="snippet",
|
|
mine=True,
|
|
maxResults=50,
|
|
order="alphabetical"
|
|
)
|
|
while request:
|
|
response = request.execute()
|
|
for item in response.get("items", []):
|
|
cid = item["snippet"]["resourceId"]["channelId"]
|
|
if cid not in channel_ids:
|
|
channel_ids.append(cid)
|
|
request = youtube.subscriptions().list_next(request, response)
|
|
except HttpError as e:
|
|
log.error(f"YouTube API error fetching subscriptions: {e}")
|
|
return 0
|
|
|
|
log.info(f"Found {len(channel_ids)} subscribed channels")
|
|
|
|
if not channel_ids:
|
|
log.warning("No subscribed channels found. Make sure the authenticated account has subscriptions.")
|
|
return 0
|
|
|
|
# Step 2: For each channel, get latest videos
|
|
new_count = 0
|
|
async with aiosqlite.connect(DB_PATH) as db:
|
|
for cid in channel_ids[:20]:
|
|
try:
|
|
req = youtube.search().list(
|
|
part="snippet",
|
|
channelId=cid,
|
|
maxResults=10,
|
|
order="date",
|
|
type="video"
|
|
)
|
|
resp = req.execute()
|
|
for item in resp.get("items", []):
|
|
vid = item["id"]["videoId"]
|
|
snippet = item["snippet"]
|
|
|
|
cursor = await db.execute(
|
|
"SELECT 1 FROM videos WHERE video_id = ?", (vid,)
|
|
)
|
|
if await cursor.fetchone():
|
|
continue
|
|
|
|
pub_time = snippet.get("publishTime", "")
|
|
await db.execute("""
|
|
INSERT INTO videos
|
|
(video_id, title, channel_name, channel_id, url, thumbnail_url, publish_date)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?)
|
|
""", (
|
|
vid,
|
|
html.unescape(snippet.get("title", "Untitled")),
|
|
html.unescape(snippet.get("channelTitle", "Unknown")),
|
|
snippet.get("channelId", cid),
|
|
f"https://www.youtube.com/watch?v={vid}",
|
|
snippet.get("thumbnails", {}).get("high", {}).get("url", ""),
|
|
pub_time[:10] if pub_time else None
|
|
))
|
|
new_count += 1
|
|
log.debug(f"New video: {snippet.get('title', 'Untitled')[:50]}")
|
|
except HttpError as e:
|
|
log.warning(f"Channel {cid} fetch error: {e}")
|
|
continue
|
|
|
|
await db.commit()
|
|
|
|
log.info(f"Ingestion completed. Found {new_count} new videos.")
|
|
return new_count
|
|
|
|
|
|
if __name__ == "__main__":
|
|
import sys
|
|
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
|
|
|
|
async def main():
|
|
from db import init_db
|
|
await init_db()
|
|
count = await fetch_subscriptions()
|
|
print(f"\nDone. {count} new videos ingested.")
|
|
|
|
asyncio.run(main())
|