192 lines
7.2 KiB
Python
192 lines
7.2 KiB
Python
import os
|
|
import re
|
|
import json
|
|
import sqlite3
|
|
from pathlib import Path
|
|
from datetime import datetime, timezone, timedelta
|
|
from urllib.parse import urlparse
|
|
from typing import Dict, Any, Optional
|
|
|
|
try:
|
|
from googleapiclient.discovery import build
|
|
except ImportError:
|
|
build = None
|
|
|
|
class YouTubeCollector:
|
|
def __init__(self, api_keys: list[str] = None, cache_db: str = ".cache/youtube_metrics.sqlite"):
|
|
self.api_keys = [k.strip() for k in (api_keys or []) if k.strip()]
|
|
self.cache_db = Path(cache_db)
|
|
self.cache_db.parent.mkdir(parents=True, exist_ok=True)
|
|
self._init_db()
|
|
self.client = None
|
|
if self.api_keys and build:
|
|
try:
|
|
self.client = build("youtube", "v3", developerKey=self.api_keys[0], cache_discovery=False)
|
|
except Exception as e:
|
|
print(f"[Warning] Failed to initialize YouTube API client: {e}")
|
|
|
|
def _init_db(self):
|
|
with sqlite3.connect(self.cache_db) as conn:
|
|
conn.execute("""
|
|
CREATE TABLE IF NOT EXISTS channel_metrics (
|
|
channel_id TEXT PRIMARY KEY,
|
|
title TEXT,
|
|
subscribers INTEGER,
|
|
lifetime_views INTEGER,
|
|
video_count INTEGER,
|
|
recent_avg_views INTEGER,
|
|
momentum_30d REAL,
|
|
raw_json TEXT,
|
|
updated_at TEXT
|
|
)
|
|
""")
|
|
|
|
def _get_cached(self, channel_id: str) -> Optional[Dict[str, Any]]:
|
|
with sqlite3.connect(self.cache_db) as conn:
|
|
conn.row_factory = sqlite3.Row
|
|
row = conn.execute("SELECT * FROM channel_metrics WHERE channel_id = ?", (channel_id,)).fetchone()
|
|
if row:
|
|
return dict(row)
|
|
return None
|
|
|
|
def _save_cache(self, metrics: Dict[str, Any]):
|
|
with sqlite3.connect(self.cache_db) as conn:
|
|
conn.execute("""
|
|
INSERT OR REPLACE INTO channel_metrics (
|
|
channel_id, title, subscribers, lifetime_views, video_count,
|
|
recent_avg_views, momentum_30d, raw_json, updated_at
|
|
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
""", (
|
|
metrics["channel_id"],
|
|
metrics["title"],
|
|
metrics["subscribers"],
|
|
metrics["lifetime_views"],
|
|
metrics["video_count"],
|
|
metrics["recent_avg_views"],
|
|
metrics["momentum_30d"],
|
|
json.dumps(metrics),
|
|
datetime.now(timezone.utc).isoformat()
|
|
))
|
|
|
|
def extract_channel_id(self, url: str) -> str:
|
|
"""Extract UC... channel ID from standard YouTube URL."""
|
|
if not url:
|
|
return ""
|
|
match = re.search(r"channel/(UC[a-zA-Z0-9_-]{22})", url)
|
|
if match:
|
|
return match.group(1)
|
|
return url.strip().rstrip("/").split("/")[-1]
|
|
|
|
def get_channel_metrics(self, youtube_url: str, fetch_recent_videos: bool = False) -> Dict[str, Any]:
|
|
channel_id = self.extract_channel_id(youtube_url)
|
|
empty = {
|
|
"channel_id": channel_id,
|
|
"title": "",
|
|
"subscribers": 0,
|
|
"lifetime_views": 0,
|
|
"video_count": 0,
|
|
"recent_avg_views": 0,
|
|
"momentum_30d": 0.0,
|
|
"status": "unresolved"
|
|
}
|
|
if not channel_id:
|
|
empty["status"] = "empty_url"
|
|
return empty
|
|
|
|
# 1. Check local cache first (saves API quota)
|
|
cached = self._get_cached(channel_id)
|
|
if cached:
|
|
cached["status"] = "cached"
|
|
return cached
|
|
|
|
# 2. If client is not available (zero API keys), return baseline
|
|
if not self.client:
|
|
empty["status"] = "no_api_key"
|
|
return empty
|
|
|
|
try:
|
|
# 3. Call channels.list for statistics and contentDetails
|
|
resp = self.client.channels().list(
|
|
part="snippet,statistics,contentDetails",
|
|
id=channel_id
|
|
).execute()
|
|
|
|
items = resp.get("items") or []
|
|
if not items:
|
|
empty["status"] = "not_found"
|
|
return empty
|
|
|
|
item = items[0]
|
|
snippet = item.get("snippet", {})
|
|
stats = item.get("statistics", {})
|
|
content = item.get("contentDetails", {})
|
|
|
|
metrics = {
|
|
"channel_id": channel_id,
|
|
"title": snippet.get("title", ""),
|
|
"subscribers": int(stats.get("subscriberCount") or 0),
|
|
"lifetime_views": int(stats.get("viewCount") or 0),
|
|
"video_count": int(stats.get("videoCount") or 0),
|
|
"recent_avg_views": 0,
|
|
"momentum_30d": 0.0,
|
|
"status": "live_api"
|
|
}
|
|
|
|
# 4. Optional: fetch recent video metrics for average views & momentum
|
|
uploads_id = content.get("relatedPlaylists", {}).get("uploads")
|
|
if fetch_recent_videos and uploads_id:
|
|
recent_views, momentum = self._calc_recent_metrics(uploads_id, metrics["lifetime_views"])
|
|
metrics["recent_avg_views"] = recent_views
|
|
metrics["momentum_30d"] = momentum
|
|
|
|
self._save_cache(metrics)
|
|
return metrics
|
|
|
|
except Exception as e:
|
|
empty["status"] = f"error: {str(e)[:100]}"
|
|
return empty
|
|
|
|
def _calc_recent_metrics(self, uploads_playlist_id: str, lifetime_views: int) -> tuple[int, float]:
|
|
"""Fetch up to 10 latest uploads to compute average views and 30d momentum."""
|
|
try:
|
|
playlist_resp = self.client.playlistItems().list(
|
|
part="contentDetails",
|
|
playlistId=uploads_playlist_id,
|
|
maxResults=10
|
|
).execute()
|
|
|
|
video_ids = [i["contentDetails"]["videoId"] for i in playlist_resp.get("items", [])]
|
|
if not video_ids:
|
|
return 0, 0.0
|
|
|
|
video_resp = self.client.videos().list(
|
|
part="statistics,snippet",
|
|
id=",".join(video_ids)
|
|
).execute()
|
|
|
|
now = datetime.now(timezone.utc)
|
|
thirty_days_ago = now - timedelta(days=30)
|
|
|
|
views = []
|
|
views_30d = 0
|
|
|
|
for v in video_resp.get("items", []):
|
|
v_views = int(v.get("statistics", {}).get("viewCount") or 0)
|
|
views.append(v_views)
|
|
pub_date_str = v.get("snippet", {}).get("publishedAt", "")
|
|
if pub_date_str:
|
|
pub_date = datetime.fromisoformat(pub_date_str.replace("Z", "+00:00"))
|
|
if pub_date >= thirty_days_ago:
|
|
views_30d += v_views
|
|
|
|
avg_views = int(sum(views) / len(views)) if views else 0
|
|
momentum = round((views_30d / lifetime_views) * 100, 3) if lifetime_views > 0 else 0.0
|
|
return avg_views, momentum
|
|
except Exception:
|
|
return 0, 0.0
|
|
|
|
if __name__ == "__main__":
|
|
collector = YouTubeCollector()
|
|
test_res = collector.get_channel_metrics("https://www.youtube.com/channel/UCBOxSirnH1-_-xKtqX051lg")
|
|
print("Test run result:")
|
|
print(test_res) |