Spaces:
Sleeping
Sleeping
File size: 2,065 Bytes
5e3be8f 76cbe64 5e3be8f 52de180 5e3be8f 76cbe64 5e3be8f 76cbe64 5e3be8f 76cbe64 5e3be8f 76cbe64 52de180 76cbe64 52de180 5e3be8f | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 | #!/usr/bin/env python3
"""Fetch comments for a Reddit or YouTube URL and print them as JSON.
Usage:
python fetch_comments.py <url>
Credentials are read from the environment (see .env.example); a .env file is
loaded automatically if present.
"""
import json
import logging
import os
import sys
from datetime import datetime, timezone
from urllib.parse import urlparse
from dotenv import load_dotenv
from chorus.api.reddit import RedditClient
from chorus.api.youtube import YouTubeClient
from chorus.logging_config import configure_logging
log = logging.getLogger("chorus.fetch_comments")
def client_for(uri: str):
"""Return the right API client for the given URL based on its host."""
host = (urlparse(uri).hostname or "").lower()
log.debug("Selecting client for host %r", host)
if host.endswith("reddit.com") or host.endswith("redd.it"):
return RedditClient(
client_id=os.environ["REDDIT_CLIENT_ID"],
client_secret=os.environ["REDDIT_CLIENT_SECRET"],
user_agent=os.environ["REDDIT_USER_AGENT"],
)
if host.endswith("youtube.com") or host.endswith("youtu.be"):
return YouTubeClient(api_key=os.environ["YOUTUBE_API_KEY"])
raise ValueError(f"Unsupported URL host: {host!r}")
def main(argv):
if len(argv) != 2:
print(f"Usage: {argv[0]} <url>", file=sys.stderr)
return 1
configure_logging()
load_dotenv()
uri = argv[1]
log.info("Fetching comments for %s", uri)
try:
client = client_for(uri)
metadata = client.get_metadata_by_uri(uri)
comments = client.get_comments_by_uri(uri)
except Exception:
log.exception("Failed to fetch comments for %s", uri)
raise
log.info("Fetched %d comment(s) for %s", len(comments), uri)
output = {
**metadata,
"created_at": datetime.now(timezone.utc).isoformat(),
"comments": comments
}
print(json.dumps(output, indent=2, ensure_ascii=False))
return 0
if __name__ == "__main__":
sys.exit(main(sys.argv))
|