From 8adf23d913edcdd4233db3ae07350e1a740c0477 Mon Sep 17 00:00:00 2001 From: Kelley Glenn Date: Sun, 22 Feb 2026 00:28:39 -0800 Subject: [PATCH 1/9] feat: add channel scraper and geocoding enhancement MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Add scripts/list-channel/ — YouTube channel video URL lister using yt-dlp with date/duration filtering (excludes Shorts). Output is compatible with extract.py --file. - Add streetAddress field to LLM extraction prompt, JSON schema, and output in extract.py. Claude provides street addresses for well-known government buildings, improving downstream geocoding accuracy. - Fix geocode response parsing bug in seed-videos.sh — was reading .latitude/.longitude but the API returns .coordinates.latitude/ .coordinates.longitude, causing geocoding to silently fail. - Prefer streetAddress over location name for geocoding queries in seed-videos.sh when available. - Update docs/llm-extraction-prompt.md and extract-metadata README with streetAddress documentation. Co-Authored-By: Claude Opus 4.6 --- docs/llm-extraction-prompt.md | 7 + scripts/extract-metadata/README.md | 3 +- scripts/extract-metadata/extract.py | 8 + scripts/list-channel/README.md | 116 ++++++++++++ scripts/list-channel/list_channel.py | 255 ++++++++++++++++++++++++++ scripts/list-channel/requirements.txt | 1 + scripts/seed-videos.sh | 13 +- 7 files changed, 398 insertions(+), 5 deletions(-) create mode 100644 scripts/list-channel/README.md create mode 100644 scripts/list-channel/list_channel.py create mode 100644 scripts/list-channel/requirements.txt diff --git a/docs/llm-extraction-prompt.md b/docs/llm-extraction-prompt.md index 0702da8..09ec598 100644 --- a/docs/llm-extraction-prompt.md +++ b/docs/llm-extraction-prompt.md @@ -120,6 +120,9 @@ Extract location information where the encounter occurred. **Fields to extract:** - **name**: Location name such as a landmark (e.g., "Springfield City Hall") or street address. See special instructions below. +- **streetAddress**: The street address of the named location (e.g., "800 E Monroe St"). + If you know the physical street address from your training data, provide it. If you + are not confident, set to null. Do NOT fabricate addresses. - **city**: City name - **state**: State abbreviation (e.g., "CA", "TX") - **latitude** and **longitude**: Set these to null UNLESS they are explicitly stated in @@ -154,6 +157,7 @@ The JSON structure: "videoDate": "YYYY-MM-DD or null", "location": { "name": "location name or null", + "streetAddress": "street address or null", "city": "city name or null", "state": "XX or null", "latitude": 0.0 or null, @@ -247,6 +251,7 @@ The model's response contains XML thinking tags followed by the final JSON: "videoDate": null, "location": { "name": "City Hall", + "streetAddress": "800 E Monroe St", "city": "Springfield", "state": "IL", "latitude": null, @@ -274,6 +279,7 @@ The service parses the last balanced `{...}` block from the response. | `videoDate` | `string \| null` | Yes | Date the incident occurred (ISO 8601 `YYYY-MM-DD`). `null` if not determinable. | | `location` | `object \| null` | Yes | Where the incident took place. `null` if not determinable. | | `location.name` | `string` | Yes* | Specific place name, prioritized: street address > specific landmark > general landmark. | +| `location.streetAddress` | `string \| null` | Yes* | Street address of the named location (e.g., "800 E Monroe St"). `null` if unknown or not confident. | | `location.city` | `string \| null` | Yes* | City name. `null` if not determinable. | | `location.state` | `string \| null` | Yes* | US state abbreviation (e.g., "CA", "TX"). `null` if not determinable. | | `location.latitude` | `number \| null` | Yes* | Latitude. Always `null` unless explicitly stated in description text. | @@ -334,3 +340,4 @@ Since the prompt instructs Claude to set latitude/longitude to `null` unless exp - **Web-app**: Calls the location-service geocode endpoint (`GET /locations/geocode?address=...`) using the extracted name/city/state - **Python CLI**: Uses the location-service geocode endpoint or a local geocoding library +- **`streetAddress` improves geocoding precision**: When available, `streetAddress` should be preferred over `name` for geocoding queries. Street addresses like "800 E Monroe St, Springfield, IL" produce more accurate geocode results than landmark names like "City Hall, Springfield, IL". diff --git a/scripts/extract-metadata/README.md b/scripts/extract-metadata/README.md index 307748e..6dc224e 100644 --- a/scripts/extract-metadata/README.md +++ b/scripts/extract-metadata/README.md @@ -110,6 +110,7 @@ Each entry in the output JSON array follows this schema: "videoDate": "2024-03-15", "location": { "name": "City Hall", + "streetAddress": "200 N Spring St", "city": "Los Angeles", "state": "CA", "latitude": null, @@ -142,7 +143,7 @@ Each entry in the output JSON array follows this schema: | `amendments` | Constitutional amendments relevant to the video (e.g., `FIRST`, `FOURTH`) | | `participants` | Types of participants (e.g., `POLICE`, `CITIZEN`, `GOVERNMENT`) | | `videoDate` | Date of the incident (ISO 8601), or `null` if not determinable | -| `location` | Location object with name, city, state, latitude, longitude; or `null` | +| `location` | Location object with name, streetAddress, city, state, latitude, longitude; or `null` | | `confidence` | Confidence scores (0.0-1.0) for each extracted field | See [docs/llm-extraction-prompt.md](../../docs/llm-extraction-prompt.md) for the full extraction prompt specification, valid enum values, and extraction rules. diff --git a/scripts/extract-metadata/extract.py b/scripts/extract-metadata/extract.py index d6b023c..c2b5629 100644 --- a/scripts/extract-metadata/extract.py +++ b/scripts/extract-metadata/extract.py @@ -134,6 +134,9 @@ **Fields to extract:** - **name**: Location name such as a landmark (e.g., "Springfield City Hall") or street \ address. See special instructions below. +- **streetAddress**: The street address of the named location (e.g., "800 E Monroe St"). \ +If you know the physical street address from your training data, provide it. If you \ +are not confident, set to null. Do NOT fabricate addresses. - **city**: City name - **state**: State abbreviation (e.g., "CA", "TX") - **latitude** and **longitude**: Set these to null UNLESS they are explicitly stated in \ @@ -169,6 +172,7 @@ "videoDate": "YYYY-MM-DD or null", "location": { "name": "location name or null", + "streetAddress": "street address or null", "city": "city name or null", "state": "XX or null", "latitude": 0.0 or null, @@ -246,6 +250,9 @@ candidate against these criteria. - Extract the city (if any) - Extract the state (if any) +- If the location is a well-known government building, courthouse, police station, or \ +other prominent landmark, provide the street address if you know it confidently. \ +If unsure, set streetAddress to null. - For latitude/longitude: Set to null unless you explicitly found coordinates in Step 1 - If no location information was found, note that location should be null - Assess what your confidence score should be based on the specificity of location information @@ -619,6 +626,7 @@ def build_output_entry(url: str, youtube_data: dict, claude_metadata: dict) -> d # Ensure all expected location fields exist location = { "name": location.get("name"), + "streetAddress": location.get("streetAddress"), "city": location.get("city"), "state": location.get("state"), "latitude": location.get("latitude"), diff --git a/scripts/list-channel/README.md b/scripts/list-channel/README.md new file mode 100644 index 0000000..5e9937a --- /dev/null +++ b/scripts/list-channel/README.md @@ -0,0 +1,116 @@ +# YouTube Channel Video Lister + +A Python CLI tool that lists video URLs from a YouTube channel, with optional date and duration filtering. Output is compatible with `extract.py --file` for the metadata extraction pipeline. + +## Prerequisites + +- Python 3.10+ + +## Installation + +```bash +cd scripts/list-channel +pip install -r requirements.txt +``` + +Or with a virtual environment: + +```bash +python -m venv .venv +source .venv/bin/activate # On Windows Git Bash: source .venv/Scripts/activate +pip install -r requirements.txt +``` + +## Usage + +### Basic Usage + +List all non-Shorts videos from a channel (prints to stdout): + +```bash +python list_channel.py "@ChannelName" +``` + +### Limit Results + +```bash +python list_channel.py "@ChannelName" -n 10 +``` + +### Filter by Date + +```bash +python list_channel.py "@ChannelName" --after 2024-01-01 +python list_channel.py "@ChannelName" --after 2024-01-01 --before 2025-01-01 +``` + +### Save to File + +```bash +python list_channel.py "@ChannelName" -o urls.txt +``` + +### Channel Identifier Formats + +The tool accepts multiple formats for specifying a channel: + +```bash +python list_channel.py "@AuditTheAudit" +python list_channel.py "UCwobzUc3z-0PrFpoRxNszXQ" +python list_channel.py "https://www.youtube.com/@AuditTheAudit" +``` + +### Pipeline with extract.py + +List channel URLs, then extract metadata: + +```bash +python list_channel.py "@ChannelName" -n 20 -o urls.txt +cd ../extract-metadata +python extract.py --file ../list-channel/urls.txt --output ../../seed-data/videos.json +``` + +## Output Format + +``` +# Channel: @ChannelName +# Fetched: 2026-02-22 +# Count: 47 +https://www.youtube.com/watch?v=abc123 +https://www.youtube.com/watch?v=def456 +``` + +Lines starting with `#` are comments. `extract.py --file` ignores comment lines and blank lines. + +## Shorts Filtering + +By default, videos shorter than 61 seconds are excluded to filter out YouTube Shorts. Adjust with `--min-duration`: + +```bash +# Include all videos (no duration filter) +python list_channel.py "@ChannelName" --min-duration 0 + +# Only videos longer than 5 minutes +python list_channel.py "@ChannelName" --min-duration 300 +``` + +## CLI Reference + +``` +usage: list_channel.py [-h] [-n MAX_RESULTS] [--after AFTER] [--before BEFORE] + [--min-duration MIN_DURATION] [-o OUTPUT] + channel + +positional arguments: + channel Channel URL, @handle, or UCxxxx channel ID. + +options: + -h, --help show this help message and exit + -n, --max-results MAX_RESULTS + Maximum number of videos to return (default: no limit). + --after AFTER Only include videos published on/after this date (YYYY-MM-DD). + --before BEFORE Only include videos published on/before this date (YYYY-MM-DD). + --min-duration MIN_DURATION + Minimum video duration in seconds (default: 61, filters Shorts). + -o, --output OUTPUT Output file path (default: stdout). +``` diff --git a/scripts/list-channel/list_channel.py b/scripts/list-channel/list_channel.py new file mode 100644 index 0000000..b17e5aa --- /dev/null +++ b/scripts/list-channel/list_channel.py @@ -0,0 +1,255 @@ +#!/usr/bin/env python3 +""" +YouTube channel video URL lister for AccountabilityAtlas. + +Uses yt-dlp to fetch video URLs from a YouTube channel, filtering by date +and duration (to exclude Shorts). Output is compatible with extract.py --file. + +Usage: + python list_channel.py "@ChannelName" + python list_channel.py "@ChannelName" -n 10 + python list_channel.py "@ChannelName" --after 2024-01-01 --before 2025-01-01 + python list_channel.py "https://www.youtube.com/@ChannelName" -o urls.txt +""" + +import argparse +import sys +from datetime import datetime + +try: + import yt_dlp +except ImportError: + print( + "Error: 'yt-dlp' package is not installed. " + "Run: pip install -r requirements.txt", + file=sys.stderr, + ) + sys.exit(1) + + +def normalize_channel_url(channel: str) -> str: + """Normalize a channel identifier to a full YouTube URL with /videos suffix. + + Accepts: + - @handle (e.g., "@AuditTheAudit") + - UCxxxx channel ID (e.g., "UCwobzUc3z-0PrFpoRxNszXQ") + - Full URL (e.g., "https://www.youtube.com/@AuditTheAudit") + + Returns: + Full YouTube URL ending with /videos. + """ + channel = channel.strip() + + # Already a full URL + if channel.startswith("http://") or channel.startswith("https://"): + # Strip trailing slashes and path suffixes like /videos, /shorts, /streams + url = channel.rstrip("/") + for suffix in ("/videos", "/shorts", "/streams", "/playlists", "/community"): + if url.endswith(suffix): + url = url[: -len(suffix)] + break + return url + "/videos" + + # @handle + if channel.startswith("@"): + return f"https://www.youtube.com/{channel}/videos" + + # UCxxxx channel ID + if channel.startswith("UC") and len(channel) == 24: + return f"https://www.youtube.com/channel/{channel}/videos" + + # Assume it's a handle without the @ + return f"https://www.youtube.com/@{channel}/videos" + + +def fetch_channel_videos( + channel_url: str, + max_results: int | None = None, + after_date: str | None = None, + before_date: str | None = None, + min_duration: int = 61, +) -> list[dict]: + """Fetch video metadata from a YouTube channel using yt-dlp. + + Args: + channel_url: Normalized YouTube channel URL (ending in /videos). + max_results: Maximum number of videos to return after filtering. + after_date: Only include videos published on/after this date (YYYY-MM-DD). + before_date: Only include videos published on/before this date (YYYY-MM-DD). + min_duration: Minimum duration in seconds (default 61, filters Shorts). + + Returns: + List of dicts with keys: url, title, duration, upload_date. + """ + ydl_opts: dict = { + "quiet": True, + "no_warnings": True, + "skip_download": True, + "extract_flat": False, + "ignoreerrors": True, + } + + # Use yt-dlp daterange for server-side date filtering + if after_date or before_date: + ydl_opts["daterange"] = yt_dlp.utils.DateRange( + start=after_date or "19700101", + end=before_date or "99991231", + ) + + # Over-fetch to account for Shorts that will be filtered out + if max_results is not None: + ydl_opts["playlistend"] = max_results * 3 + + print(f"Fetching videos from: {channel_url}", file=sys.stderr) + + with yt_dlp.YoutubeDL(ydl_opts) as ydl: + info = ydl.extract_info(channel_url, download=False) + + if not info: + print("Error: Could not extract channel information.", file=sys.stderr) + return [] + + entries = info.get("entries") or [] + + videos = [] + for entry in entries: + if entry is None: + continue + + duration = entry.get("duration") or 0 + if duration < min_duration: + continue + + upload_date = entry.get("upload_date", "") + video_url = entry.get("webpage_url") or entry.get("url", "") + + if not video_url: + continue + + # Ensure it's a proper watch URL, not a channel/playlist URL + video_id = entry.get("id", "") + if video_id and "watch?v=" not in video_url: + video_url = f"https://www.youtube.com/watch?v={video_id}" + + videos.append( + { + "url": video_url, + "title": entry.get("title", ""), + "duration": duration, + "upload_date": upload_date, + } + ) + + if max_results is not None and len(videos) >= max_results: + break + + return videos + + +def format_output(videos: list[dict], channel_name: str) -> str: + """Format video URLs as one-per-line output with header comments. + + Args: + videos: List of video dicts from fetch_channel_videos(). + channel_name: Channel name for the header comment. + + Returns: + Formatted string with # comments and URLs. + """ + lines = [ + f"# Channel: {channel_name}", + f"# Fetched: {datetime.now().strftime('%Y-%m-%d')}", + f"# Count: {len(videos)}", + ] + + for video in videos: + lines.append(video["url"]) + + return "\n".join(lines) + "\n" + + +def main(): + parser = argparse.ArgumentParser( + description="List YouTube video URLs from a channel for AccountabilityAtlas.", + epilog="Output is compatible with extract.py --file (# comment lines are ignored).", + ) + parser.add_argument( + "channel", + help="Channel URL, @handle, or UCxxxx channel ID.", + ) + parser.add_argument( + "-n", + "--max-results", + type=int, + default=None, + help="Maximum number of videos to return (default: no limit).", + ) + parser.add_argument( + "--after", + type=str, + default=None, + help="Only include videos published on/after this date (YYYY-MM-DD).", + ) + parser.add_argument( + "--before", + type=str, + default=None, + help="Only include videos published on/before this date (YYYY-MM-DD).", + ) + parser.add_argument( + "--min-duration", + type=int, + default=61, + help="Minimum video duration in seconds (default: 61, filters Shorts).", + ) + parser.add_argument( + "-o", + "--output", + type=str, + default=None, + help="Output file path (default: stdout).", + ) + + args = parser.parse_args() + + # Validate date formats + for date_arg, name in [(args.after, "--after"), (args.before, "--before")]: + if date_arg is not None: + try: + datetime.strptime(date_arg, "%Y-%m-%d") + except ValueError: + parser.error(f"{name} must be in YYYY-MM-DD format, got: {date_arg}") + + channel_url = normalize_channel_url(args.channel) + + videos = fetch_channel_videos( + channel_url=channel_url, + max_results=args.max_results, + after_date=args.after, + before_date=args.before, + min_duration=args.min_duration, + ) + + if not videos: + print("No videos found matching the criteria.", file=sys.stderr) + sys.exit(0) + + # Try to extract channel name from the first video or use the input + channel_name = args.channel + output = format_output(videos, channel_name) + + if args.output: + with open(args.output, "w", encoding="utf-8") as f: + f.write(output) + print( + f"Wrote {len(videos)} URLs to {args.output}", + file=sys.stderr, + ) + else: + print(output, end="") + + print(f"Found {len(videos)} videos.", file=sys.stderr) + + +if __name__ == "__main__": + main() diff --git a/scripts/list-channel/requirements.txt b/scripts/list-channel/requirements.txt new file mode 100644 index 0000000..90b1fef --- /dev/null +++ b/scripts/list-channel/requirements.txt @@ -0,0 +1 @@ +yt-dlp>=2024.1.0 diff --git a/scripts/seed-videos.sh b/scripts/seed-videos.sh index dbc1e3b..c115f1a 100644 --- a/scripts/seed-videos.sh +++ b/scripts/seed-videos.sh @@ -100,13 +100,18 @@ for i in $(seq 0 $((TOTAL - 1))); do LAT=$(echo "$ENTRY" | jq '.location.latitude') LNG=$(echo "$ENTRY" | jq '.location.longitude') LOC_NAME=$(echo "$ENTRY" | jq -r '.location.name // empty') + LOC_STREET=$(echo "$ENTRY" | jq -r '.location.streetAddress // empty') LOC_CITY=$(echo "$ENTRY" | jq -r '.location.city // empty') LOC_STATE=$(echo "$ENTRY" | jq -r '.location.state // empty') - # If no lat/lng, try geocoding from name/city/state + # If no lat/lng, try geocoding from streetAddress/name/city/state if [[ "$LAT" == "null" || "$LNG" == "null" ]]; then ADDRESS_PARTS="" - [[ -n "$LOC_NAME" ]] && ADDRESS_PARTS="$LOC_NAME" + if [[ -n "$LOC_STREET" ]]; then + ADDRESS_PARTS="$LOC_STREET" + else + [[ -n "$LOC_NAME" ]] && ADDRESS_PARTS="$LOC_NAME" + fi [[ -n "$LOC_CITY" ]] && ADDRESS_PARTS="${ADDRESS_PARTS:+$ADDRESS_PARTS, }$LOC_CITY" [[ -n "$LOC_STATE" ]] && ADDRESS_PARTS="${ADDRESS_PARTS:+$ADDRESS_PARTS, }$LOC_STATE" @@ -119,8 +124,8 @@ for i in $(seq 0 $((TOTAL - 1))); do GEO_BODY=$(echo "$GEOCODE_RESPONSE" | sed '$d') if [[ "$GEO_CODE" == "200" ]]; then - LAT=$(echo "$GEO_BODY" | jq '.latitude') - LNG=$(echo "$GEO_BODY" | jq '.longitude') + LAT=$(echo "$GEO_BODY" | jq '.coordinates.latitude') + LNG=$(echo "$GEO_BODY" | jq '.coordinates.longitude') fi fi fi From 1073613871f434b96e268f4eb7c20aa1f3f33117 Mon Sep 17 00:00:00 2001 From: Kelley Glenn Date: Sun, 22 Feb 2026 00:30:26 -0800 Subject: [PATCH 2/9] docs: add list-channel script to README Co-Authored-By: Claude Opus 4.6 --- README.md | 2 ++ 1 file changed, 2 insertions(+) diff --git a/README.md b/README.md index adae961..10a0eb0 100644 --- a/README.md +++ b/README.md @@ -17,6 +17,7 @@ AccountabilityAtlas/ │ ├── seed-videos.sh # Seed videos from JSON into the running stack │ ├── aws/ # AWS start, stop, and deploy scripts │ ├── extract-metadata/ # Python CLI for AI-powered video metadata extraction +│ ├── list-channel/ # Python CLI to list video URLs from a YouTube channel │ ├── lib/ # Shared script utilities │ └── integration/ # Cross-service integration tests ├── seed-data/ # Generated seed data (JSON, not committed) @@ -100,6 +101,7 @@ See [docs/07-InfrastructureArchitecture.md](docs/07-InfrastructureArchitecture.m The project includes tools for AI-powered extraction of video metadata (amendments, participants, location, date) using Claude: +- **`scripts/list-channel/`** — Python CLI that lists video URLs from a YouTube channel using yt-dlp, with date and duration filtering (excludes Shorts). Output is compatible with `extract.py --file`. See [scripts/list-channel/README.md](scripts/list-channel/README.md). - **`scripts/extract-metadata/`** — Python CLI that uses yt-dlp and the Anthropic SDK to extract metadata from YouTube videos, including optional transcript analysis. See [scripts/extract-metadata/README.md](scripts/extract-metadata/README.md). - **`scripts/seed-videos.sh`** — Seeds videos from a JSON file into the running stack via the API. Reads output from the extract CLI. - **`/api/v1/videos/extract`** — Video-service REST endpoint for real-time extraction (title + description only, no transcript). From 027742aea10f1a0ae34024ef5ebbf10f4f81118a Mon Sep 17 00:00:00 2001 From: Kelley Glenn Date: Sun, 22 Feb 2026 00:32:06 -0800 Subject: [PATCH 3/9] refactor: reduce cognitive complexity in fetch_channel_videos Extract _build_ydl_opts() and _parse_entry() helpers to bring cognitive complexity from 22 to within Sonar's 15 threshold. Co-Authored-By: Claude Opus 4.6 --- scripts/list-channel/list_channel.py | 110 +++++++++++++++------------ 1 file changed, 60 insertions(+), 50 deletions(-) diff --git a/scripts/list-channel/list_channel.py b/scripts/list-channel/list_channel.py index b17e5aa..b22fae8 100644 --- a/scripts/list-channel/list_channel.py +++ b/scripts/list-channel/list_channel.py @@ -62,6 +62,59 @@ def normalize_channel_url(channel: str) -> str: return f"https://www.youtube.com/@{channel}/videos" +def _build_ydl_opts( + max_results: int | None, + after_date: str | None, + before_date: str | None, +) -> dict: + """Build yt-dlp options for channel video extraction.""" + opts: dict = { + "quiet": True, + "no_warnings": True, + "skip_download": True, + "extract_flat": False, + "ignoreerrors": True, + } + + if after_date or before_date: + opts["daterange"] = yt_dlp.utils.DateRange( + start=after_date or "19700101", + end=before_date or "99991231", + ) + + # Over-fetch to account for Shorts that will be filtered out + if max_results is not None: + opts["playlistend"] = max_results * 3 + + return opts + + +def _parse_entry(entry: dict | None, min_duration: int) -> dict | None: + """Parse a single yt-dlp entry into a video dict, or None if filtered out.""" + if entry is None: + return None + + duration = entry.get("duration") or 0 + if duration < min_duration: + return None + + video_url = entry.get("webpage_url") or entry.get("url", "") + if not video_url: + return None + + # Ensure it's a proper watch URL, not a channel/playlist URL + video_id = entry.get("id", "") + if video_id and "watch?v=" not in video_url: + video_url = f"https://www.youtube.com/watch?v={video_id}" + + return { + "url": video_url, + "title": entry.get("title", ""), + "duration": duration, + "upload_date": entry.get("upload_date", ""), + } + + def fetch_channel_videos( channel_url: str, max_results: int | None = None, @@ -81,24 +134,7 @@ def fetch_channel_videos( Returns: List of dicts with keys: url, title, duration, upload_date. """ - ydl_opts: dict = { - "quiet": True, - "no_warnings": True, - "skip_download": True, - "extract_flat": False, - "ignoreerrors": True, - } - - # Use yt-dlp daterange for server-side date filtering - if after_date or before_date: - ydl_opts["daterange"] = yt_dlp.utils.DateRange( - start=after_date or "19700101", - end=before_date or "99991231", - ) - - # Over-fetch to account for Shorts that will be filtered out - if max_results is not None: - ydl_opts["playlistend"] = max_results * 3 + ydl_opts = _build_ydl_opts(max_results, after_date, before_date) print(f"Fetching videos from: {channel_url}", file=sys.stderr) @@ -109,39 +145,13 @@ def fetch_channel_videos( print("Error: Could not extract channel information.", file=sys.stderr) return [] - entries = info.get("entries") or [] - videos = [] - for entry in entries: - if entry is None: - continue - - duration = entry.get("duration") or 0 - if duration < min_duration: - continue - - upload_date = entry.get("upload_date", "") - video_url = entry.get("webpage_url") or entry.get("url", "") - - if not video_url: - continue - - # Ensure it's a proper watch URL, not a channel/playlist URL - video_id = entry.get("id", "") - if video_id and "watch?v=" not in video_url: - video_url = f"https://www.youtube.com/watch?v={video_id}" - - videos.append( - { - "url": video_url, - "title": entry.get("title", ""), - "duration": duration, - "upload_date": upload_date, - } - ) - - if max_results is not None and len(videos) >= max_results: - break + for entry in info.get("entries") or []: + video = _parse_entry(entry, min_duration) + if video is not None: + videos.append(video) + if max_results is not None and len(videos) >= max_results: + break return videos From f9558f369e89c54ec38cb5e218968f2b9f835931 Mon Sep 17 00:00:00 2001 From: Kelley Glenn Date: Sun, 22 Feb 2026 00:41:09 -0800 Subject: [PATCH 4/9] ci: add SonarCloud analysis for Python and shell scripts Co-Authored-By: Claude Opus 4.6 --- .github/workflows/sonar.yaml | 23 +++++++++++++++++++++++ sonar-project.properties | 6 ++++++ 2 files changed, 29 insertions(+) create mode 100644 .github/workflows/sonar.yaml create mode 100644 sonar-project.properties diff --git a/.github/workflows/sonar.yaml b/.github/workflows/sonar.yaml new file mode 100644 index 0000000..9c1dbf1 --- /dev/null +++ b/.github/workflows/sonar.yaml @@ -0,0 +1,23 @@ +name: SonarCloud + +on: + push: + branches: [master] + pull_request: + branches: [master] + +permissions: + contents: read + +jobs: + sonar: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + with: + fetch-depth: 0 + + - name: SonarCloud analysis + uses: SonarSource/sonarqube-scan-action@689fb39b34b9aa95ebc5f8f119343ddd51542402 # v4 + env: + SONAR_TOKEN: ${{ secrets.SONAR_TOKEN }} diff --git a/sonar-project.properties b/sonar-project.properties new file mode 100644 index 0000000..c210c16 --- /dev/null +++ b/sonar-project.properties @@ -0,0 +1,6 @@ +sonar.projectKey=kelleyglenn_AccountabilityAtlas +sonar.projectName=AccountabilityAtlas +sonar.organization=kelleyglenn + +sonar.sources=scripts +sonar.exclusions=**/requirements.txt,**/*.md,**/__pycache__/** From 8d4381b2a1dc13bcc4f9c9c861a8c77c8df1fbd8 Mon Sep 17 00:00:00 2001 From: Kelley Glenn Date: Sun, 22 Feb 2026 10:54:53 -0800 Subject: [PATCH 5/9] fix: improve YouTube rate limiting resilience and batch API compatibility - Add yt-dlp sleep options (sleep_requests, sleep_interval, sleep_subtitles) to avoid triggering YouTube rate limits on subtitle fetches - Add retry with backoff for subtitle URL fetch on HTTP 429 - Fix batch API custom_id to use video ID instead of full URL (must match ^[a-zA-Z0-9_-]{1,64}) - Switch list_channel to extract_flat mode for faster channel listing - Move date filtering from yt-dlp DateRange to Python (fixes YYYY-MM-DD format error and avoids per-video page fetches) - Add seed-data/*.txt to .gitignore Co-Authored-By: Claude Opus 4.6 --- .gitignore | 1 + scripts/extract-metadata/extract.py | 40 ++++++++++++++++----- scripts/list-channel/list_channel.py | 53 ++++++++++++++++------------ 3 files changed, 63 insertions(+), 31 deletions(-) diff --git a/.gitignore b/.gitignore index ab5c428..454434f 100644 --- a/.gitignore +++ b/.gitignore @@ -49,6 +49,7 @@ infra/*.tfvars # Seed data (generated, keep local) seed-data/*.json +seed-data/*.txt # AWS deployment config scripts/aws/config.env diff --git a/scripts/extract-metadata/extract.py b/scripts/extract-metadata/extract.py index c2b5629..1da4a93 100644 --- a/scripts/extract-metadata/extract.py +++ b/scripts/extract-metadata/extract.py @@ -360,6 +360,8 @@ def fetch_youtube_metadata(url: str, include_transcript: bool = True) -> dict: "no_warnings": True, "skip_download": True, "format": "best", + "sleep_requests": 0.75, + "sleep_interval": 2, } if include_transcript: @@ -369,6 +371,7 @@ def fetch_youtube_metadata(url: str, include_transcript: bool = True) -> dict: "writesubtitles": True, "subtitleslangs": ["en"], "subtitlesformat": "json3", + "sleep_subtitles": 5, } ) @@ -413,14 +416,27 @@ def _extract_transcript(info: dict) -> str | None: # If yt-dlp provided a URL but no inline data, we need to fetch it sub_url = en_sub.get("url") if sub_url: - try: - import urllib.request + import urllib.request - with urllib.request.urlopen(sub_url, timeout=15) as resp: - raw = resp.read().decode("utf-8", errors="replace") - return _parse_subtitle_data(raw, en_sub.get("ext", "")) - except Exception: - return None + for attempt in range(3): + try: + with urllib.request.urlopen(sub_url, timeout=15) as resp: + raw = resp.read().decode("utf-8", errors="replace") + return _parse_subtitle_data(raw, en_sub.get("ext", "")) + except urllib.error.HTTPError as e: + if e.code == 429 and attempt < 2: + wait = 5 * (attempt + 1) + print( + f" Subtitle fetch rate-limited, retrying in {wait}s...", + file=sys.stderr, + ) + time.sleep(wait) + else: + print(f" Subtitle fetch failed: {e}", file=sys.stderr) + return None + except Exception as e: + print(f" Subtitle fetch failed: {e}", file=sys.stderr) + return None return None @@ -735,7 +751,12 @@ def process_urls_batch( print(f"\nSubmitting batch of {len(youtube_data)} requests...", file=sys.stderr) requests = [] + # custom_id must be [a-zA-Z0-9_-]{1,64} — use video ID, map back to URL + id_to_url = {} for url, yt_data in youtube_data.items(): + video_id = url.split("watch?v=")[-1].split("&")[0] if "watch?v=" in url else url + id_to_url[video_id] = url + user_message = build_batch_user_message( title=yt_data["title"], description=yt_data["description"], @@ -747,7 +768,7 @@ def process_urls_batch( requests.append( { - "custom_id": url, + "custom_id": video_id, "params": { "model": model, "max_tokens": 4096, @@ -794,7 +815,8 @@ def process_urls_batch( errors = [] for entry in client.messages.batches.results(batch.id): - url = entry.custom_id + video_id = entry.custom_id + url = id_to_url.get(video_id, video_id) if entry.result.type == "succeeded": try: raw_text = entry.result.message.content[0].text.strip() diff --git a/scripts/list-channel/list_channel.py b/scripts/list-channel/list_channel.py index b22fae8..731fb17 100644 --- a/scripts/list-channel/list_channel.py +++ b/scripts/list-channel/list_channel.py @@ -62,34 +62,29 @@ def normalize_channel_url(channel: str) -> str: return f"https://www.youtube.com/@{channel}/videos" -def _build_ydl_opts( - max_results: int | None, - after_date: str | None, - before_date: str | None, -) -> dict: +def _build_ydl_opts(max_results: int | None) -> dict: """Build yt-dlp options for channel video extraction.""" opts: dict = { "quiet": True, "no_warnings": True, "skip_download": True, - "extract_flat": False, + "extract_flat": True, "ignoreerrors": True, } - if after_date or before_date: - opts["daterange"] = yt_dlp.utils.DateRange( - start=after_date or "19700101", - end=before_date or "99991231", - ) - - # Over-fetch to account for Shorts that will be filtered out + # Over-fetch to account for Shorts/date filtering that will remove entries if max_results is not None: opts["playlistend"] = max_results * 3 return opts -def _parse_entry(entry: dict | None, min_duration: int) -> dict | None: +def _parse_entry( + entry: dict | None, + min_duration: int, + after_date: str | None, + before_date: str | None, +) -> dict | None: """Parse a single yt-dlp entry into a video dict, or None if filtered out.""" if entry is None: return None @@ -98,6 +93,14 @@ def _parse_entry(entry: dict | None, min_duration: int) -> dict | None: if duration < min_duration: return None + # Date filtering — upload_date is YYYYMMDD from yt-dlp + upload_date = entry.get("upload_date", "") + if upload_date and (after_date or before_date): + if after_date and upload_date < after_date: + return None + if before_date and upload_date > before_date: + return None + video_url = entry.get("webpage_url") or entry.get("url", "") if not video_url: return None @@ -111,7 +114,7 @@ def _parse_entry(entry: dict | None, min_duration: int) -> dict | None: "url": video_url, "title": entry.get("title", ""), "duration": duration, - "upload_date": entry.get("upload_date", ""), + "upload_date": upload_date, } @@ -127,14 +130,14 @@ def fetch_channel_videos( Args: channel_url: Normalized YouTube channel URL (ending in /videos). max_results: Maximum number of videos to return after filtering. - after_date: Only include videos published on/after this date (YYYY-MM-DD). - before_date: Only include videos published on/before this date (YYYY-MM-DD). + after_date: Only include videos published on/after this date (YYYYMMDD). + before_date: Only include videos published on/before this date (YYYYMMDD). min_duration: Minimum duration in seconds (default 61, filters Shorts). Returns: List of dicts with keys: url, title, duration, upload_date. """ - ydl_opts = _build_ydl_opts(max_results, after_date, before_date) + ydl_opts = _build_ydl_opts(max_results) print(f"Fetching videos from: {channel_url}", file=sys.stderr) @@ -147,7 +150,7 @@ def fetch_channel_videos( videos = [] for entry in info.get("entries") or []: - video = _parse_entry(entry, min_duration) + video = _parse_entry(entry, min_duration, after_date, before_date) if video is not None: videos.append(video) if max_results is not None and len(videos) >= max_results: @@ -222,21 +225,27 @@ def main(): args = parser.parse_args() - # Validate date formats + # Validate date formats and convert to YYYYMMDD for yt-dlp comparison + after_yyyymmdd = None + before_yyyymmdd = None for date_arg, name in [(args.after, "--after"), (args.before, "--before")]: if date_arg is not None: try: datetime.strptime(date_arg, "%Y-%m-%d") except ValueError: parser.error(f"{name} must be in YYYY-MM-DD format, got: {date_arg}") + if args.after: + after_yyyymmdd = args.after.replace("-", "") + if args.before: + before_yyyymmdd = args.before.replace("-", "") channel_url = normalize_channel_url(args.channel) videos = fetch_channel_videos( channel_url=channel_url, max_results=args.max_results, - after_date=args.after, - before_date=args.before, + after_date=after_yyyymmdd, + before_date=before_yyyymmdd, min_duration=args.min_duration, ) From 43410ff7e29409920cbd5b6feca1b460f8bf175e Mon Sep 17 00:00:00 2001 From: Kelley Glenn Date: Sun, 22 Feb 2026 13:05:59 -0800 Subject: [PATCH 6/9] refactor: split extract.py into fetch_youtube.py and claude_extract.py Separates the monolithic extraction script into two independent pipeline stages: YouTube metadata fetching (yt-dlp) and Claude LLM extraction. This allows re-running extraction without re-fetching from YouTube. Both scripts support --append for resumable batch processing. Co-Authored-By: Claude Opus 4.6 --- scripts/extract-metadata/README.md | 185 +++++--- .../{extract.py => claude_extract.py} | 425 ++++-------------- scripts/extract-metadata/fetch_youtube.py | 362 +++++++++++++++ 3 files changed, 577 insertions(+), 395 deletions(-) rename scripts/extract-metadata/{extract.py => claude_extract.py} (66%) create mode 100644 scripts/extract-metadata/fetch_youtube.py diff --git a/scripts/extract-metadata/README.md b/scripts/extract-metadata/README.md index 6dc224e..86cc82b 100644 --- a/scripts/extract-metadata/README.md +++ b/scripts/extract-metadata/README.md @@ -1,11 +1,16 @@ -# Video Metadata Extraction CLI +# Video Metadata Extraction Pipeline -A Python CLI tool that extracts structured metadata from YouTube videos for AccountabilityAtlas seed data. It uses `yt-dlp` to fetch video metadata and auto-generated transcripts, then calls Claude to extract amendments, participants, dates, and locations. +A two-script Python pipeline that extracts structured metadata from YouTube videos for AccountabilityAtlas seed data. + +1. **`fetch_youtube.py`** — fetches video metadata and transcripts via yt-dlp, outputs intermediate JSON +2. **`claude_extract.py`** — reads intermediate JSON, calls Claude to extract amendments/participants/dates/locations, outputs seed-data format JSON + +Splitting the pipeline lets each phase run independently. If the Claude prompt changes or extraction fails, you don't need to re-fetch from YouTube. ## Prerequisites - Python 3.10+ -- `ANTHROPIC_API_KEY` environment variable set with a valid Anthropic API key +- `ANTHROPIC_API_KEY` environment variable set (only needed for `claude_extract.py`) ## Installation @@ -22,80 +27,134 @@ source .venv/bin/activate # On Windows Git Bash: source .venv/Scripts/activate pip install -r requirements.txt ``` -## Usage - -### Single URL +## Pipeline Usage -Prints JSON to stdout: +### Full pipeline: URL list → YouTube data → seed data ```bash -python extract.py "https://www.youtube.com/watch?v=VIDEO_ID" -``` +# 1. (Optional) Generate URL list from a YouTube channel +python ../list-channel/list_channel.py CHANNEL_ID > urls.txt -### Bulk Processing +# 2. Fetch YouTube metadata + transcripts +python fetch_youtube.py --file urls.txt --output youtube-data.json -Process a file of URLs (one per line) and write results to a JSON file: +# 3. Extract structured metadata via Claude +python claude_extract.py --input youtube-data.json --output seed-data/videos.json +``` + +### Single URL (quick test) ```bash -python extract.py --file urls.txt --output seed-data/videos.json +# Fetch metadata to stdout +python fetch_youtube.py "https://www.youtube.com/watch?v=VIDEO_ID" + +# Or pipe directly to claude_extract.py +python fetch_youtube.py "https://www.youtube.com/watch?v=VIDEO_ID" --output single.json +python claude_extract.py --input single.json ``` -### Append to Existing File +## fetch_youtube.py -Add new entries to an existing JSON array file without overwriting: +Fetches video metadata and auto-generated transcripts from YouTube using yt-dlp. + +### Usage ```bash -python extract.py --file more-urls.txt --output seed-data/videos.json --append -``` +# Single URL (prints JSON to stdout) +python fetch_youtube.py "https://www.youtube.com/watch?v=VIDEO_ID" -### Skip Transcript +# Bulk from file +python fetch_youtube.py --file urls.txt --output youtube-data.json -Faster extraction using only title and description (lower confidence scores): +# Skip transcripts (faster) +python fetch_youtube.py --file urls.txt --output youtube-data.json --no-transcript -```bash -python extract.py --no-transcript "https://www.youtube.com/watch?v=VIDEO_ID" +# Resume interrupted batch (skips already-fetched URLs) +python fetch_youtube.py --file urls.txt --output youtube-data.json --append ``` -### Batch Processing +### CLI Reference -Use the [Message Batches API](https://docs.anthropic.com/en/docs/build-with-claude/batch-processing) for **50% cost savings** on bulk processing. Requests are submitted as a single batch and processed asynchronously: +``` +usage: fetch_youtube.py [-h] [--file FILE] [--output OUTPUT] [--no-transcript] + [--append] + [url] -```bash -python extract.py --file urls.txt --output seed-data/videos.json --batch +positional arguments: + url Single YouTube URL to process. + +options: + -h, --help show this help message and exit + --file FILE, -f FILE Path to a text file with one YouTube URL per line. + --output OUTPUT, -o OUTPUT + Output file path for JSON results. + --no-transcript Skip transcript fetch (faster, but less data for extraction). + --append, -a Append to existing output file, skipping URLs already present. ``` -Batch processing may take minutes to hours depending on queue depth. The CLI polls for completion and prints progress updates. The output format is identical to sequential mode. +### Intermediate JSON Format -The `--batch` flag can be combined with other options: +Output is a JSON array where each element has: -```bash -python extract.py --file urls.txt --output videos.json --batch --append --no-transcript +```json +{ + "url": "https://www.youtube.com/watch?v=VIDEO_ID", + "title": "Video Title", + "description": "Full YouTube description", + "channel": "Channel Name", + "thumbnail": "https://i.ytimg.com/.../maxresdefault.jpg", + "duration": 1234, + "published": "20240315", + "transcript": "Full transcript text or null" +} ``` -### Custom Model +## claude_extract.py -Override the default Claude model: +Reads intermediate JSON from `fetch_youtube.py` and calls Claude to extract structured metadata in the seed-data format. + +### Usage ```bash -python extract.py --model claude-sonnet-4-20250514 "https://www.youtube.com/watch?v=VIDEO_ID" +# Sequential processing +python claude_extract.py --input youtube-data.json --output seed-data/videos.json + +# Batch API (50% cost savings, async processing) +python claude_extract.py --input youtube-data.json --output videos.json --batch + +# Resume interrupted extraction (skips already-processed URLs) +python claude_extract.py --input youtube-data.json --output videos.json --append + +# Custom model +python claude_extract.py --input youtube-data.json --output videos.json --model claude-sonnet-4-20250514 + +# Combine flags +python claude_extract.py --input youtube-data.json --output videos.json --batch --append ``` -## URL File Format +Batch processing uses the [Message Batches API](https://docs.anthropic.com/en/docs/build-with-claude/batch-processing) and may take minutes to hours depending on queue depth. The CLI polls for completion and prints progress updates. -One URL per line. Blank lines and lines starting with `#` are ignored: +### CLI Reference ``` -# First Amendment audit videos -https://www.youtube.com/watch?v=abc123 -https://www.youtube.com/watch?v=def456 +usage: claude_extract.py [-h] --input INPUT [--output OUTPUT] [--model MODEL] + [--batch] [--append] -# Police encounter videos -https://www.youtube.com/watch?v=ghi789 +options: + -h, --help show this help message and exit + --input INPUT, -i INPUT + Input JSON file from fetch_youtube.py. + --output OUTPUT, -o OUTPUT + Output file path for JSON results. + --model MODEL, -m MODEL + Claude model to use (default: claude-haiku-4-5-20251001). + --batch, -b Use the Message Batches API for 50% cost savings. + --append, -a Append to existing output file, skipping URLs already present. ``` -## Output Format +### Seed-Data Output Format -Each entry in the output JSON array follows this schema: +Each entry in the output JSON array: ```json { @@ -148,34 +207,32 @@ Each entry in the output JSON array follows this schema: See [docs/llm-extraction-prompt.md](../../docs/llm-extraction-prompt.md) for the full extraction prompt specification, valid enum values, and extraction rules. +## URL File Format + +One URL per line. Blank lines and lines starting with `#` are ignored: + +``` +# First Amendment audit videos +https://www.youtube.com/watch?v=abc123 +https://www.youtube.com/watch?v=def456 + +# Police encounter videos +https://www.youtube.com/watch?v=ghi789 +``` + ## How It Works +### fetch_youtube.py + 1. **Fetch metadata**: Uses the `yt-dlp` Python library to extract video title, description, publication date, channel, thumbnail, and duration without downloading the video. 2. **Fetch transcript**: Optionally retrieves auto-generated English subtitles and parses them into plain text. -3. **Call Claude**: Sends the extraction prompt with XML-tagged video data, following the shared extraction prompt spec from [`docs/llm-extraction-prompt.md`](../../docs/llm-extraction-prompt.md). In sequential mode, this is a user-only prompt (no system prompt) identical to the Java video-service. In batch mode (`--batch`), the shared instructions are sent as a system message with `cache_control` for prompt caching, and only the per-video data is in the user message. Claude responds with XML thinking tags (multi-step analysis) followed by the final JSON object. -4. **Parse response**: Extracts the last balanced JSON object from the response (skipping the XML thinking tags), matching the Java service's parsing logic. -5. **Combine results**: Merges YouTube metadata with Claude's extracted fields into the seed-data format. - -If a transcript is unavailable, the tool falls back to extracting from title and description only, which typically produces lower confidence scores. +3. **Output**: Writes intermediate JSON with all YouTube data for downstream processing. -## CLI Reference +### claude_extract.py -``` -usage: extract.py [-h] [--file FILE] [--output OUTPUT] [--model MODEL] - [--no-transcript] [--append] [--batch] - [url] - -positional arguments: - url Single YouTube URL to process. +1. **Read input**: Loads intermediate JSON from `fetch_youtube.py`. +2. **Call Claude**: Sends the extraction prompt with XML-tagged video data, following the shared extraction prompt spec from [`docs/llm-extraction-prompt.md`](../../docs/llm-extraction-prompt.md). In sequential mode, this is a user-only prompt (no system prompt) identical to the Java video-service. In batch mode (`--batch`), the shared instructions are sent as a system message with `cache_control` for prompt caching, and only the per-video data is in the user message. Claude responds with XML thinking tags (multi-step analysis) followed by the final JSON object. +3. **Parse response**: Extracts the last balanced JSON object from the response (skipping the XML thinking tags), matching the Java service's parsing logic. +4. **Combine results**: Merges YouTube metadata with Claude's extracted fields into the seed-data format. -options: - -h, --help show this help message and exit - --file FILE, -f FILE Path to a text file with one YouTube URL per line. - --output OUTPUT, -o OUTPUT - Output file path for JSON results. - --model MODEL, -m MODEL - Claude model to use (default: claude-haiku-4-5-20251001). - --no-transcript Skip transcript fetch (faster, less accurate). - --append, -a Append to existing output file instead of overwriting. - --batch, -b Use Message Batches API for 50% cost savings (requires --file). -``` +If a transcript is unavailable, the tool falls back to extracting from title and description only, which typically produces lower confidence scores. diff --git a/scripts/extract-metadata/extract.py b/scripts/extract-metadata/claude_extract.py similarity index 66% rename from scripts/extract-metadata/extract.py rename to scripts/extract-metadata/claude_extract.py index 1da4a93..75f60c2 100644 --- a/scripts/extract-metadata/extract.py +++ b/scripts/extract-metadata/claude_extract.py @@ -1,22 +1,20 @@ #!/usr/bin/env python3 """ -Video metadata extraction CLI for AccountabilityAtlas. +Claude LLM metadata extraction for AccountabilityAtlas. -Uses yt-dlp to fetch YouTube metadata and auto-generated transcripts, -then calls Claude to extract structured metadata matching the seed-data format. +Reads intermediate JSON from fetch_youtube.py and uses Claude to extract +structured metadata (amendments, participants, dates, locations) in +the seed-data format. Usage: - python extract.py - python extract.py --file urls.txt --output seed-data/videos.json - python extract.py --file urls.txt --output videos.json --append - python extract.py --file urls.txt --output videos.json --batch - python extract.py --no-transcript - python extract.py --model claude-sonnet-4-20250514 + python claude_extract.py --input youtube-data.json --output seed-data/videos.json + python claude_extract.py --input youtube-data.json --output videos.json --batch + python claude_extract.py --input youtube-data.json --output videos.json --append + python claude_extract.py --input youtube-data.json --output videos.json --model claude-sonnet-4-20250514 """ import argparse import json -import re import sys import time from pathlib import Path @@ -31,16 +29,6 @@ ) sys.exit(1) -try: - import yt_dlp -except ImportError: - print( - "Error: 'yt-dlp' package is not installed. " - "Run: pip install -r requirements.txt", - file=sys.stderr, - ) - sys.exit(1) - DEFAULT_MODEL = "claude-haiku-4-5-20251001" @@ -341,206 +329,18 @@ def _fill_template( ) -# --- YouTube fetching --- - - -def fetch_youtube_metadata(url: str, include_transcript: bool = True) -> dict: - """Fetch video metadata and optionally transcript from YouTube using yt-dlp. - - Args: - url: YouTube video URL. - include_transcript: Whether to attempt fetching auto-generated subtitles. - - Returns: - Dictionary with keys: url, title, description, channel, thumbnail, - duration, published (str or None), transcript (str or None). - """ - ydl_opts = { - "quiet": True, - "no_warnings": True, - "skip_download": True, - "format": "best", - "sleep_requests": 0.75, - "sleep_interval": 2, - } - - if include_transcript: - ydl_opts.update( - { - "writeautomaticsub": True, - "writesubtitles": True, - "subtitleslangs": ["en"], - "subtitlesformat": "json3", - "sleep_subtitles": 5, - } - ) - - with yt_dlp.YoutubeDL(ydl_opts) as ydl: - info = ydl.extract_info(url, download=False) - - transcript = None - if include_transcript: - transcript = _extract_transcript(info) - - thumbnail = _pick_best_thumbnail(info) - - return { - "url": info.get("webpage_url") or url, - "title": info.get("title", ""), - "description": info.get("description", ""), - "channel": info.get("channel") or info.get("uploader", ""), - "thumbnail": thumbnail, - "duration": info.get("duration"), - "published": info.get("upload_date"), - "transcript": transcript, - } - - -def _extract_transcript(info: dict) -> str | None: - """Extract transcript text from yt-dlp subtitle data. - - yt-dlp stores fetched subtitles under 'requested_subtitles' when available. - The json3 format contains an 'events' list with 'segs' (segments) containing 'utf8' text. - Falls back to vtt/srv format text parsing if json3 is unavailable. - """ - subs = info.get("requested_subtitles") or {} - en_sub = subs.get("en") - if not en_sub: - return None - - # If yt-dlp returned the subtitle data directly (json3 format) - sub_data = en_sub.get("data") - if sub_data: - return _parse_subtitle_data(sub_data, en_sub.get("ext", "")) - - # If yt-dlp provided a URL but no inline data, we need to fetch it - sub_url = en_sub.get("url") - if sub_url: - import urllib.request - - for attempt in range(3): - try: - with urllib.request.urlopen(sub_url, timeout=15) as resp: - raw = resp.read().decode("utf-8", errors="replace") - return _parse_subtitle_data(raw, en_sub.get("ext", "")) - except urllib.error.HTTPError as e: - if e.code == 429 and attempt < 2: - wait = 5 * (attempt + 1) - print( - f" Subtitle fetch rate-limited, retrying in {wait}s...", - file=sys.stderr, - ) - time.sleep(wait) - else: - print(f" Subtitle fetch failed: {e}", file=sys.stderr) - return None - except Exception as e: - print(f" Subtitle fetch failed: {e}", file=sys.stderr) - return None - - return None - - -def _parse_subtitle_data(data: str, ext: str) -> str | None: - """Parse subtitle data from various formats into plain text.""" - if ext == "json3": - try: - parsed = json.loads(data) - segments = [] - for event in parsed.get("events", []): - for seg in event.get("segs", []): - text = seg.get("utf8", "").strip() - if text and text != "\n": - segments.append(text) - if segments: - return " ".join(segments) - except (json.JSONDecodeError, KeyError): - pass - - # Fallback: strip timing lines from vtt/srv formats - lines = [] - for line in data.splitlines(): - line = line.strip() - # Skip empty lines, timing lines, WEBVTT headers, and numeric cue IDs - if not line: - continue - if "-->" in line: - continue - if line.startswith("WEBVTT") or line.startswith("Kind:") or line.startswith("Language:"): - continue - if line.isdigit(): - continue - # Remove HTML-style tags (e.g. , , <00:01:02.345>) - clean = re.sub(r"<[^>]+>", "", line) - if clean.strip(): - lines.append(clean.strip()) - - if lines: - # Deduplicate consecutive identical lines (common in vtt) - deduped = [lines[0]] - for ln in lines[1:]: - if ln != deduped[-1]: - deduped.append(ln) - return " ".join(deduped) - - return None - - -def _pick_best_thumbnail(info: dict) -> str | None: - """Select the best thumbnail URL from yt-dlp info. - - Prefers maxresdefault, then high-quality thumbnails, then whatever is available. - """ - thumbnails = info.get("thumbnails") or [] - thumbnail = info.get("thumbnail") - - if not thumbnails: - return thumbnail - - # Prefer known high-quality YouTube thumbnail names - for t in thumbnails: - url = t.get("url", "") - if "maxresdefault" in url: - return url - - for t in thumbnails: - url = t.get("url", "") - if "hqdefault" in url or "sddefault" in url: - return url - - # Fall back to highest resolution available - best = max( - (t for t in thumbnails if t.get("width")), - key=lambda t: (t.get("width", 0) * t.get("height", 0)), - default=None, - ) - if best: - return best.get("url") - - return thumbnail - - # --- Message building --- def build_user_message(title: str, description: str, published: str | None, transcript: str | None) -> str: - """Build the user message for Claude following the shared prompt spec. - - Uses USER_PROMPT_TEMPLATE with the same structure as the Java video-service. - When a transcript is available, it is inserted as an additional XML-tagged section. - """ + """Build the user message for Claude following the shared prompt spec.""" return _fill_template(USER_PROMPT_TEMPLATE, title, description, published, transcript) def build_batch_user_message( title: str, description: str, published: str | None, transcript: str | None ) -> str: - """Build the per-video user message for batch mode. - - Uses BATCH_USER_TEMPLATE which contains only the video data XML tags - and a brief instruction. The shared extraction instructions are in the - system message (BATCH_SYSTEM_PROMPT) for prompt caching optimization. - """ + """Build the per-video user message for batch mode.""" return _fill_template(BATCH_USER_TEMPLATE, title, description, published, transcript) @@ -593,12 +393,9 @@ def extract_metadata_with_claude( ) -> dict: """Call Claude to extract structured metadata from video information. - Uses the same user-only prompt as the Java video-service, with an - additional transcript section when available. See docs/llm-extraction-prompt.md. - Args: client: Anthropic client instance. - youtube_data: Dictionary from fetch_youtube_metadata(). + youtube_data: Dictionary from fetch_youtube.py intermediate JSON. model: Claude model ID to use. Returns: @@ -631,15 +428,9 @@ def extract_metadata_with_claude( def build_output_entry(url: str, youtube_data: dict, claude_metadata: dict) -> dict: - """Combine YouTube metadata and Claude extraction into the seed-data format. - - Returns a dictionary matching the expected output schema with youtubeUrl, - title, description, channelName, thumbnailUrl, durationSeconds, and all - Claude-extracted fields. - """ + """Combine YouTube metadata and Claude extraction into the seed-data format.""" location = claude_metadata.get("location") if location is not None: - # Ensure all expected location fields exist location = { "name": location.get("name"), "streetAddress": location.get("streetAddress"), @@ -672,33 +463,22 @@ def build_output_entry(url: str, youtube_data: dict, claude_metadata: dict) -> d } -def process_single_url( - url: str, +def process_single( + youtube_data: dict, client: anthropic.Anthropic, model: str, - include_transcript: bool, ) -> dict: - """Process a single YouTube URL end-to-end. + """Process a single video entry through Claude extraction. Args: - url: YouTube video URL. + youtube_data: Dictionary from fetch_youtube.py intermediate JSON. client: Anthropic client instance. model: Claude model ID. - include_transcript: Whether to fetch transcript. Returns: Output entry dictionary in seed-data format. """ - print(f"Fetching metadata for: {url}", file=sys.stderr) - youtube_data = fetch_youtube_metadata(url, include_transcript=include_transcript) - - has_transcript = youtube_data.get("transcript") is not None - if include_transcript and not has_transcript: - print( - " Warning: No transcript available. Extracting from title+description only.", - file=sys.stderr, - ) - + url = youtube_data.get("url", "") print(f" Calling Claude ({model})...", file=sys.stderr) claude_metadata = extract_metadata_with_claude(client, youtube_data, model=model) @@ -707,55 +487,34 @@ def process_single_url( return entry -def process_urls_batch( - urls: list[str], +def process_batch( + youtube_data_list: list[dict], client: anthropic.Anthropic, model: str, - include_transcript: bool, ) -> tuple[list[dict], list[str]]: - """Process multiple URLs using the Message Batches API for 50% cost savings. + """Process multiple videos using the Message Batches API for 50% cost savings. Submits all requests as a single batch and polls for completion. The prompt is split into a shared system message (with cache_control) and per-video user messages to maximize prompt caching hits. Args: - urls: List of YouTube video URLs. + youtube_data_list: List of dictionaries from fetch_youtube.py intermediate JSON. client: Anthropic client instance. model: Claude model ID. - include_transcript: Whether to fetch transcripts. Returns: Tuple of (results list, errors list). """ - # Phase 1: Fetch YouTube metadata for all URLs - youtube_data = {} - for i, url in enumerate(urls, 1): - print(f"\n[{i}/{len(urls)}] Fetching metadata for: {url}", file=sys.stderr) - try: - youtube_data[url] = fetch_youtube_metadata(url, include_transcript=include_transcript) - has_transcript = youtube_data[url].get("transcript") is not None - if include_transcript and not has_transcript: - print( - " Warning: No transcript available. Will extract from title+description only.", - file=sys.stderr, - ) - except Exception as e: - print(f" Error fetching metadata: {e}", file=sys.stderr) - # Skip this URL entirely — can't submit to batch without metadata - - if not youtube_data: - return [], [f"Failed to fetch metadata for all {len(urls)} URLs"] - - # Phase 2: Build and submit the batch - print(f"\nSubmitting batch of {len(youtube_data)} requests...", file=sys.stderr) + print(f"\nSubmitting batch of {len(youtube_data_list)} requests...", file=sys.stderr) requests = [] - # custom_id must be [a-zA-Z0-9_-]{1,64} — use video ID, map back to URL - id_to_url = {} - for url, yt_data in youtube_data.items(): - video_id = url.split("watch?v=")[-1].split("&")[0] if "watch?v=" in url else url - id_to_url[video_id] = url + # custom_id must be [a-zA-Z0-9_-]{1,64} — use video ID, map back to index + id_to_index = {} + for idx, yt_data in enumerate(youtube_data_list): + url = yt_data.get("url", "") + video_id = url.split("watch?v=")[-1].split("&")[0] if "watch?v=" in url else f"idx-{idx}" + id_to_index[video_id] = idx user_message = build_batch_user_message( title=yt_data["title"], @@ -787,7 +546,7 @@ def process_urls_batch( batch = client.messages.batches.create(requests=requests) print(f"Batch created: {batch.id}", file=sys.stderr) - # Phase 3: Poll for completion + # Poll for completion while batch.processing_status != "ended": counts = batch.request_counts print( @@ -810,20 +569,26 @@ def process_urls_batch( file=sys.stderr, ) - # Phase 4: Retrieve and process results + # Retrieve and process results results = [] errors = [] for entry in client.messages.batches.results(batch.id): video_id = entry.custom_id - url = id_to_url.get(video_id, video_id) + idx = id_to_index.get(video_id) + if idx is None: + errors.append(f"Unknown custom_id in batch response: {video_id}") + continue + yt_data = youtube_data_list[idx] + url = yt_data.get("url", video_id) + if entry.result.type == "succeeded": try: raw_text = entry.result.message.content[0].text.strip() json_str = _extract_json(raw_text) claude_metadata = json.loads(json_str) - results.append(build_output_entry(url, youtube_data[url], claude_metadata)) - print(f" Processed: {youtube_data[url].get('title', url)}", file=sys.stderr) + results.append(build_output_entry(url, yt_data, claude_metadata)) + print(f" Processed: {yt_data.get('title', url)}", file=sys.stderr) except (json.JSONDecodeError, IndexError, KeyError) as e: errors.append(f"Failed to parse response for {url}: {e}") elif entry.result.type == "errored": @@ -834,29 +599,20 @@ def process_urls_batch( elif entry.result.type == "canceled": errors.append(f"Request canceled for {url}") - # Also count URLs that failed metadata fetch - for url in urls: - if url not in youtube_data: - errors.append(f"Failed to fetch YouTube metadata for {url}") - return results, errors def main(): parser = argparse.ArgumentParser( - description="Extract structured metadata from YouTube videos for AccountabilityAtlas.", + description="Extract structured metadata from YouTube data using Claude for AccountabilityAtlas.", epilog="Requires ANTHROPIC_API_KEY environment variable to be set.", ) parser.add_argument( - "url", - nargs="?", - help="Single YouTube URL to process.", - ) - parser.add_argument( - "--file", - "-f", + "--input", + "-i", type=str, - help="Path to a text file with one YouTube URL per line.", + required=True, + help="Input JSON file from fetch_youtube.py.", ) parser.add_argument( "--output", @@ -872,57 +628,49 @@ def main(): help=f"Claude model to use (default: {DEFAULT_MODEL}).", ) parser.add_argument( - "--no-transcript", + "--batch", + "-b", action="store_true", - help="Skip transcript fetch (faster, but less accurate extraction).", + help="Use the Message Batches API for 50%% cost savings.", ) parser.add_argument( "--append", "-a", action="store_true", - help="Append to existing output file instead of overwriting.", - ) - parser.add_argument( - "--batch", - "-b", - action="store_true", - help=( - "Use the Message Batches API for bulk processing (requires --file). " - "Submits all requests as a single batch for 50%% cost savings." - ), + help="Append to existing output file, skipping URLs already present.", ) args = parser.parse_args() # Validate arguments - if not args.url and not args.file: - parser.error("Provide either a URL argument or --file with a file of URLs.") - if args.url and args.file: - parser.error("Provide either a URL argument or --file, not both.") if args.append and not args.output: parser.error("--append requires --output.") - if args.batch and not args.file: - parser.error("--batch requires --file.") - # Collect URLs - urls = [] - if args.url: - urls.append(args.url.strip()) - else: - file_path = Path(args.file) - if not file_path.exists(): - print(f"Error: File not found: {args.file}", file=sys.stderr) + # Load input data + input_path = Path(args.input) + if not input_path.exists(): + print(f"Error: Input file not found: {args.input}", file=sys.stderr) + sys.exit(1) + + try: + with open(input_path, "r", encoding="utf-8") as f: + youtube_data_list = json.load(f) + if not isinstance(youtube_data_list, list): + print( + f"Error: Input file {args.input} does not contain a JSON array.", + file=sys.stderr, + ) sys.exit(1) - with open(file_path, "r", encoding="utf-8") as f: - for line in f: - line = line.strip() - if line and not line.startswith("#"): - urls.append(line) - - if not urls: - print("Error: No URLs to process.", file=sys.stderr) + except json.JSONDecodeError as e: + print(f"Error: Failed to parse input file {args.input}: {e}", file=sys.stderr) + sys.exit(1) + + if not youtube_data_list: + print("Error: Input file contains no entries.", file=sys.stderr) sys.exit(1) + print(f"Loaded {len(youtube_data_list)} entries from {args.input}.", file=sys.stderr) + # Initialize Anthropic client try: client = anthropic.Anthropic() @@ -933,10 +681,9 @@ def main(): ) sys.exit(1) - include_transcript = not args.no_transcript - # Load existing entries if appending existing_entries = [] + existing_urls = set() if args.append and args.output: output_path = Path(args.output) if output_path.exists(): @@ -949,6 +696,7 @@ def main(): file=sys.stderr, ) sys.exit(1) + existing_urls = {entry.get("youtubeUrl") for entry in existing_entries} print( f"Loaded {len(existing_entries)} existing entries from {args.output}.", file=sys.stderr, @@ -960,21 +708,36 @@ def main(): ) sys.exit(1) - # Process URLs + # Filter out already-processed entries when appending + if existing_urls: + original_count = len(youtube_data_list) + youtube_data_list = [ + d for d in youtube_data_list if d.get("url") not in existing_urls + ] + skipped = original_count - len(youtube_data_list) + if skipped: + print(f"Skipping {skipped} already-extracted URL(s).", file=sys.stderr) + + if not youtube_data_list and existing_entries: + print("All entries already extracted. Nothing to do.", file=sys.stderr) + sys.exit(0) + + # Process entries results = list(existing_entries) errors = [] if args.batch: - batch_results, batch_errors = process_urls_batch( - urls, client, args.model, include_transcript + batch_results, batch_errors = process_batch( + youtube_data_list, client, args.model ) results.extend(batch_results) errors.extend(batch_errors) else: - for i, url in enumerate(urls, 1): - print(f"\n[{i}/{len(urls)}]", file=sys.stderr) + for i, yt_data in enumerate(youtube_data_list, 1): + url = yt_data.get("url", "unknown") + print(f"\n[{i}/{len(youtube_data_list)}] Processing: {url}", file=sys.stderr) try: - entry = process_single_url(url, client, args.model, include_transcript) + entry = process_single(yt_data, client, args.model) results.append(entry) except Exception as e: error_msg = f"Failed to process {url}: {e}" @@ -1002,7 +765,7 @@ def main(): print(f" - {err}", file=sys.stderr) sys.exit(1 if new_count == 0 else 0) else: - print(f"\nSuccessfully processed {new_count} URL(s).", file=sys.stderr) + print(f"\nSuccessfully processed {new_count} entry(ies).", file=sys.stderr) if __name__ == "__main__": diff --git a/scripts/extract-metadata/fetch_youtube.py b/scripts/extract-metadata/fetch_youtube.py new file mode 100644 index 0000000..d98427b --- /dev/null +++ b/scripts/extract-metadata/fetch_youtube.py @@ -0,0 +1,362 @@ +#!/usr/bin/env python3 +""" +YouTube metadata + transcript fetcher for AccountabilityAtlas. + +Uses yt-dlp to fetch video metadata and auto-generated transcripts, +writing intermediate JSON for downstream processing by claude_extract.py. + +Usage: + python fetch_youtube.py + python fetch_youtube.py --file urls.txt --output youtube-data.json + python fetch_youtube.py --file urls.txt --output youtube-data.json --no-transcript + python fetch_youtube.py --file urls.txt --output youtube-data.json --append +""" + +import argparse +import json +import re +import sys +import time +from pathlib import Path + +try: + import yt_dlp +except ImportError: + print( + "Error: 'yt-dlp' package is not installed. " + "Run: pip install -r requirements.txt", + file=sys.stderr, + ) + sys.exit(1) + + +# --- YouTube fetching --- + + +def fetch_youtube_metadata(url: str, include_transcript: bool = True) -> dict: + """Fetch video metadata and optionally transcript from YouTube using yt-dlp. + + Args: + url: YouTube video URL. + include_transcript: Whether to attempt fetching auto-generated subtitles. + + Returns: + Dictionary with keys: url, title, description, channel, thumbnail, + duration, published (str or None), transcript (str or None). + """ + ydl_opts = { + "quiet": True, + "no_warnings": True, + "skip_download": True, + "format": "best", + "sleep_requests": 0.75, + "sleep_interval": 2, + } + + if include_transcript: + ydl_opts.update( + { + "writeautomaticsub": True, + "writesubtitles": True, + "subtitleslangs": ["en"], + "subtitlesformat": "json3", + "sleep_subtitles": 5, + } + ) + + with yt_dlp.YoutubeDL(ydl_opts) as ydl: + info = ydl.extract_info(url, download=False) + + transcript = None + if include_transcript: + transcript = _extract_transcript(info) + + thumbnail = _pick_best_thumbnail(info) + + return { + "url": info.get("webpage_url") or url, + "title": info.get("title", ""), + "description": info.get("description", ""), + "channel": info.get("channel") or info.get("uploader", ""), + "thumbnail": thumbnail, + "duration": info.get("duration"), + "published": info.get("upload_date"), + "transcript": transcript, + } + + +def _extract_transcript(info: dict) -> str | None: + """Extract transcript text from yt-dlp subtitle data. + + yt-dlp stores fetched subtitles under 'requested_subtitles' when available. + The json3 format contains an 'events' list with 'segs' (segments) containing 'utf8' text. + Falls back to vtt/srv format text parsing if json3 is unavailable. + """ + subs = info.get("requested_subtitles") or {} + en_sub = subs.get("en") + if not en_sub: + return None + + # If yt-dlp returned the subtitle data directly (json3 format) + sub_data = en_sub.get("data") + if sub_data: + return _parse_subtitle_data(sub_data, en_sub.get("ext", "")) + + # If yt-dlp provided a URL but no inline data, we need to fetch it + sub_url = en_sub.get("url") + if sub_url: + import urllib.request + + for attempt in range(3): + try: + with urllib.request.urlopen(sub_url, timeout=15) as resp: + raw = resp.read().decode("utf-8", errors="replace") + return _parse_subtitle_data(raw, en_sub.get("ext", "")) + except urllib.error.HTTPError as e: + if e.code == 429 and attempt < 2: + wait = 5 * (attempt + 1) + print( + f" Subtitle fetch rate-limited, retrying in {wait}s...", + file=sys.stderr, + ) + time.sleep(wait) + else: + print(f" Subtitle fetch failed: {e}", file=sys.stderr) + return None + except Exception as e: + print(f" Subtitle fetch failed: {e}", file=sys.stderr) + return None + + return None + + +def _parse_subtitle_data(data: str, ext: str) -> str | None: + """Parse subtitle data from various formats into plain text.""" + if ext == "json3": + try: + parsed = json.loads(data) + segments = [] + for event in parsed.get("events", []): + for seg in event.get("segs", []): + text = seg.get("utf8", "").strip() + if text and text != "\n": + segments.append(text) + if segments: + return " ".join(segments) + except (json.JSONDecodeError, KeyError): + pass + + # Fallback: strip timing lines from vtt/srv formats + lines = [] + for line in data.splitlines(): + line = line.strip() + # Skip empty lines, timing lines, WEBVTT headers, and numeric cue IDs + if not line: + continue + if "-->" in line: + continue + if line.startswith("WEBVTT") or line.startswith("Kind:") or line.startswith("Language:"): + continue + if line.isdigit(): + continue + # Remove HTML-style tags (e.g. , , <00:01:02.345>) + clean = re.sub(r"<[^>]+>", "", line) + if clean.strip(): + lines.append(clean.strip()) + + if lines: + # Deduplicate consecutive identical lines (common in vtt) + deduped = [lines[0]] + for ln in lines[1:]: + if ln != deduped[-1]: + deduped.append(ln) + return " ".join(deduped) + + return None + + +def _pick_best_thumbnail(info: dict) -> str | None: + """Select the best thumbnail URL from yt-dlp info. + + Prefers maxresdefault, then high-quality thumbnails, then whatever is available. + """ + thumbnails = info.get("thumbnails") or [] + thumbnail = info.get("thumbnail") + + if not thumbnails: + return thumbnail + + # Prefer known high-quality YouTube thumbnail names + for t in thumbnails: + url = t.get("url", "") + if "maxresdefault" in url: + return url + + for t in thumbnails: + url = t.get("url", "") + if "hqdefault" in url or "sddefault" in url: + return url + + # Fall back to highest resolution available + best = max( + (t for t in thumbnails if t.get("width")), + key=lambda t: (t.get("width", 0) * t.get("height", 0)), + default=None, + ) + if best: + return best.get("url") + + return thumbnail + + +def main(): + parser = argparse.ArgumentParser( + description="Fetch YouTube video metadata and transcripts for AccountabilityAtlas.", + ) + parser.add_argument( + "url", + nargs="?", + help="Single YouTube URL to process.", + ) + parser.add_argument( + "--file", + "-f", + type=str, + help="Path to a text file with one YouTube URL per line.", + ) + parser.add_argument( + "--output", + "-o", + type=str, + help="Output file path for JSON results. If not specified, prints to stdout.", + ) + parser.add_argument( + "--no-transcript", + action="store_true", + help="Skip transcript fetch (faster, but less data for extraction).", + ) + parser.add_argument( + "--append", + "-a", + action="store_true", + help="Append to existing output file, skipping URLs already present.", + ) + + args = parser.parse_args() + + # Validate arguments + if not args.url and not args.file: + parser.error("Provide either a URL argument or --file with a file of URLs.") + if args.url and args.file: + parser.error("Provide either a URL argument or --file, not both.") + if args.append and not args.output: + parser.error("--append requires --output.") + + # Collect URLs + urls = [] + if args.url: + urls.append(args.url.strip()) + else: + file_path = Path(args.file) + if not file_path.exists(): + print(f"Error: File not found: {args.file}", file=sys.stderr) + sys.exit(1) + with open(file_path, "r", encoding="utf-8") as f: + for line in f: + line = line.strip() + if line and not line.startswith("#"): + urls.append(line) + + if not urls: + print("Error: No URLs to process.", file=sys.stderr) + sys.exit(1) + + include_transcript = not args.no_transcript + + # Load existing entries if appending + existing_entries = [] + existing_urls = set() + if args.append and args.output: + output_path = Path(args.output) + if output_path.exists(): + try: + with open(output_path, "r", encoding="utf-8") as f: + existing_entries = json.load(f) + if not isinstance(existing_entries, list): + print( + f"Error: Existing file {args.output} does not contain a JSON array.", + file=sys.stderr, + ) + sys.exit(1) + existing_urls = {entry.get("url") for entry in existing_entries} + print( + f"Loaded {len(existing_entries)} existing entries from {args.output}.", + file=sys.stderr, + ) + except json.JSONDecodeError as e: + print( + f"Error: Failed to parse existing file {args.output}: {e}", + file=sys.stderr, + ) + sys.exit(1) + + # Filter out already-fetched URLs when appending + if existing_urls: + original_count = len(urls) + urls = [u for u in urls if u not in existing_urls] + skipped = original_count - len(urls) + if skipped: + print(f"Skipping {skipped} already-fetched URL(s).", file=sys.stderr) + + if not urls and existing_entries: + print("All URLs already fetched. Nothing to do.", file=sys.stderr) + sys.exit(0) + + # Fetch metadata for each URL + results = list(existing_entries) + errors = [] + + for i, url in enumerate(urls, 1): + print(f"\n[{i}/{len(urls)}] Fetching metadata for: {url}", file=sys.stderr) + try: + data = fetch_youtube_metadata(url, include_transcript=include_transcript) + results.append(data) + + has_transcript = data.get("transcript") is not None + if include_transcript and not has_transcript: + print( + " Warning: No transcript available for this video.", + file=sys.stderr, + ) + print(f" Done: {data.get('title', 'Unknown')}", file=sys.stderr) + except Exception as e: + error_msg = f"Failed to fetch {url}: {e}" + print(f" Error: {error_msg}", file=sys.stderr) + errors.append(error_msg) + + # Output results + output_json = json.dumps(results, indent=2, ensure_ascii=False) + + if args.output: + output_path = Path(args.output) + output_path.parent.mkdir(parents=True, exist_ok=True) + with open(output_path, "w", encoding="utf-8") as f: + f.write(output_json) + f.write("\n") + print(f"\nWrote {len(results)} entries to {args.output}.", file=sys.stderr) + else: + print(output_json) + + # Summary + new_count = len(results) - len(existing_entries) + if errors: + print(f"\nCompleted with {len(errors)} error(s):", file=sys.stderr) + for err in errors: + print(f" - {err}", file=sys.stderr) + sys.exit(1 if new_count == 0 else 0) + else: + print(f"\nSuccessfully fetched {new_count} URL(s).", file=sys.stderr) + + +if __name__ == "__main__": + main() From 36f24b4fcbfa9a708de8d6f168de8eb92414ce22 Mon Sep 17 00:00:00 2001 From: Kelley Glenn Date: Sun, 22 Feb 2026 14:10:02 -0800 Subject: [PATCH 7/9] fix: harden seed-videos.sh for Windows and missing locations MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Replace python3 URL encoding with jq (@uri filter) — python3 is a non-functional WindowsApps stub on Windows, causing curl to receive unencoded URLs and exit with code 3 under set -euo pipefail - Require street address for geocoding fallback — generic name/city/state queries produce unreliable results - Skip video creation when no location available (locationId is required) - Handle location 409 by extracting existingLocationId from response - Always include locationId in video creation request body Co-Authored-By: Claude Opus 4.6 --- scripts/seed-videos.sh | 35 ++++++++++++++++++++++------------- 1 file changed, 22 insertions(+), 13 deletions(-) diff --git a/scripts/seed-videos.sh b/scripts/seed-videos.sh index c115f1a..aca66cd 100644 --- a/scripts/seed-videos.sh +++ b/scripts/seed-videos.sh @@ -104,19 +104,14 @@ for i in $(seq 0 $((TOTAL - 1))); do LOC_CITY=$(echo "$ENTRY" | jq -r '.location.city // empty') LOC_STATE=$(echo "$ENTRY" | jq -r '.location.state // empty') - # If no lat/lng, try geocoding from streetAddress/name/city/state + # If no lat/lng, try geocoding from streetAddress (requires a street address) if [[ "$LAT" == "null" || "$LNG" == "null" ]]; then - ADDRESS_PARTS="" if [[ -n "$LOC_STREET" ]]; then ADDRESS_PARTS="$LOC_STREET" - else - [[ -n "$LOC_NAME" ]] && ADDRESS_PARTS="$LOC_NAME" - fi - [[ -n "$LOC_CITY" ]] && ADDRESS_PARTS="${ADDRESS_PARTS:+$ADDRESS_PARTS, }$LOC_CITY" - [[ -n "$LOC_STATE" ]] && ADDRESS_PARTS="${ADDRESS_PARTS:+$ADDRESS_PARTS, }$LOC_STATE" + [[ -n "$LOC_CITY" ]] && ADDRESS_PARTS="$ADDRESS_PARTS, $LOC_CITY" + [[ -n "$LOC_STATE" ]] && ADDRESS_PARTS="$ADDRESS_PARTS, $LOC_STATE" - if [[ -n "$ADDRESS_PARTS" ]]; then - ENCODED_ADDRESS=$(echo "$ADDRESS_PARTS" | python3 -c "import sys, urllib.parse; print(urllib.parse.quote(sys.stdin.read().strip()))" 2>/dev/null || echo "$ADDRESS_PARTS") + ENCODED_ADDRESS=$(printf '%s' "$ADDRESS_PARTS" | jq -sRr '@uri') GEOCODE_RESPONSE=$(curl -s -w "\n%{http_code}" \ "$API_URL/locations/geocode?address=$ENCODED_ADDRESS" \ -H "$AUTH_HEADER") @@ -155,10 +150,24 @@ for i in $(seq 0 $((TOTAL - 1))); do if [[ "$LOC_CODE" == "201" ]]; then LOCATION_ID=$(echo "$LOC_BODY_RESP" | jq -r '.id') + elif [[ "$LOC_CODE" == "409" ]]; then + # Location already exists — use the existing one + LOCATION_ID=$(echo "$LOC_BODY_RESP" | jq -r '.existingLocationId // empty') + if [[ -z "$LOCATION_ID" ]]; then + warn " Location 409 but no existingLocationId in response" + fi fi fi fi + # Skip video if no location could be created (locationId is required by the API) + if [[ -z "$LOCATION_ID" ]]; then + echo -e "${RED}failed (no location)${NC}" + warn " Could not create or geocode location — locationId is required" + FAILED=$((FAILED + 1)) + continue + fi + # Build video creation request AMENDMENTS=$(echo "$ENTRY" | jq -c '.amendments // []') PARTICIPANTS=$(echo "$ENTRY" | jq -c '.participants // []') @@ -169,14 +178,14 @@ for i in $(seq 0 $((TOTAL - 1))); do --argjson amendments "$AMENDMENTS" \ --argjson participants "$PARTICIPANTS" \ --arg videoDate "${VIDEO_DATE:-}" \ - --arg locationId "${LOCATION_ID:-}" \ + --arg locationId "$LOCATION_ID" \ '{ youtubeUrl: $youtubeUrl, amendments: $amendments, - participants: $participants + participants: $participants, + locationId: $locationId } - + (if $videoDate != "" then { videoDate: $videoDate } else {} end) - + (if $locationId != "" then { locationId: $locationId } else {} end)') + + (if $videoDate != "" then { videoDate: $videoDate } else {} end)') # Create video VIDEO_RESPONSE=$(curl -s -w "\n%{http_code}" -X POST "$API_URL/videos" \ From ed8e026fef5a280d2ebd6ded76c67101f228c72a Mon Sep 17 00:00:00 2001 From: Kelley Glenn Date: Sun, 22 Feb 2026 18:39:23 -0800 Subject: [PATCH 8/9] refactor: reduce cognitive complexity and add incremental writes Extract helper functions across all three Python scripts to reduce SonarCloud cognitive complexity scores. Add incremental file writing to fetch_youtube.py so progress survives interruptions, and add --delay flag to throttle requests and avoid YouTube rate limiting. Co-Authored-By: Claude Opus 4.6 --- scripts/extract-metadata/claude_extract.py | 136 ++++++----- scripts/extract-metadata/fetch_youtube.py | 265 ++++++++++++--------- scripts/list-channel/list_channel.py | 68 ++++-- 3 files changed, 276 insertions(+), 193 deletions(-) diff --git a/scripts/extract-metadata/claude_extract.py b/scripts/extract-metadata/claude_extract.py index 75f60c2..86f227d 100644 --- a/scripts/extract-metadata/claude_extract.py +++ b/scripts/extract-metadata/claude_extract.py @@ -347,6 +347,19 @@ def build_batch_user_message( # --- JSON extraction --- +def _strip_code_fences(text: str) -> str: + """Strip markdown code fences (```...```) from text.""" + if not text.startswith("```"): + return text + first_newline = text.find("\n") + if first_newline < 0: + return text + last_fence = text.rfind("```") + if last_fence <= first_newline: + return text + return text[first_newline + 1 : last_fence].strip() + + def _extract_json(text: str) -> str: """Extract the last top-level JSON object from the response text. @@ -355,15 +368,7 @@ def _extract_json(text: str) -> str: This finds the last balanced {...} block in the response, matching the Java service's extractJson logic. """ - trimmed = text.strip() - - # Handle code fences if present - if trimmed.startswith("```"): - first_newline = trimmed.index("\n") if "\n" in trimmed else -1 - if first_newline >= 0: - last_fence = trimmed.rfind("```") - if last_fence > first_newline: - trimmed = trimmed[first_newline + 1 : last_fence].strip() + trimmed = _strip_code_fences(text.strip()) # Find the last '}' and walk back to find its matching '{' last_brace = trimmed.rfind("}") @@ -487,6 +492,38 @@ def process_single( return entry +def _process_batch_entry( + entry, + yt_data: dict, + url: str, + results: list[dict], + errors: list[str], +) -> None: + """Process a single result from the Message Batches API response.""" + result_type = entry.result.type + + if result_type != "succeeded": + _ERROR_MESSAGES = { + "errored": lambda: f"API error for {url}: " + + getattr(entry.result.error, "message", str(entry.result.error)), + "expired": lambda: f"Request expired for {url}", + "canceled": lambda: f"Request canceled for {url}", + } + msg_fn = _ERROR_MESSAGES.get(result_type) + if msg_fn: + errors.append(msg_fn()) + return + + try: + raw_text = entry.result.message.content[0].text.strip() + json_str = _extract_json(raw_text) + claude_metadata = json.loads(json_str) + results.append(build_output_entry(url, yt_data, claude_metadata)) + print(f" Processed: {yt_data.get('title', url)}", file=sys.stderr) + except (json.JSONDecodeError, IndexError, KeyError) as e: + errors.append(f"Failed to parse response for {url}: {e}") + + def process_batch( youtube_data_list: list[dict], client: anthropic.Anthropic, @@ -581,27 +618,38 @@ def process_batch( continue yt_data = youtube_data_list[idx] url = yt_data.get("url", video_id) - - if entry.result.type == "succeeded": - try: - raw_text = entry.result.message.content[0].text.strip() - json_str = _extract_json(raw_text) - claude_metadata = json.loads(json_str) - results.append(build_output_entry(url, yt_data, claude_metadata)) - print(f" Processed: {yt_data.get('title', url)}", file=sys.stderr) - except (json.JSONDecodeError, IndexError, KeyError) as e: - errors.append(f"Failed to parse response for {url}: {e}") - elif entry.result.type == "errored": - error_msg = getattr(entry.result.error, "message", str(entry.result.error)) - errors.append(f"API error for {url}: {error_msg}") - elif entry.result.type == "expired": - errors.append(f"Request expired for {url}") - elif entry.result.type == "canceled": - errors.append(f"Request canceled for {url}") + _process_batch_entry(entry, yt_data, url, results, errors) return results, errors +def _load_json_array(path: Path, label: str) -> list: + """Load and validate a JSON array from a file, exiting on error.""" + try: + with open(path, "r", encoding="utf-8") as f: + data = json.load(f) + except json.JSONDecodeError as e: + print(f"Error: Failed to parse {label} {path}: {e}", file=sys.stderr) + sys.exit(1) + + if not isinstance(data, list): + print(f"Error: {label} {path} does not contain a JSON array.", file=sys.stderr) + sys.exit(1) + + return data + + +def _load_existing_output(output_path: Path) -> tuple[list, set]: + """Load existing entries from output file for append mode.""" + if not output_path.exists(): + return [], set() + + entries = _load_json_array(output_path, "Existing file") + urls = {entry.get("youtubeUrl") for entry in entries} + print(f"Loaded {len(entries)} existing entries from {output_path}.", file=sys.stderr) + return entries, urls + + def main(): parser = argparse.ArgumentParser( description="Extract structured metadata from YouTube data using Claude for AccountabilityAtlas.", @@ -652,18 +700,7 @@ def main(): print(f"Error: Input file not found: {args.input}", file=sys.stderr) sys.exit(1) - try: - with open(input_path, "r", encoding="utf-8") as f: - youtube_data_list = json.load(f) - if not isinstance(youtube_data_list, list): - print( - f"Error: Input file {args.input} does not contain a JSON array.", - file=sys.stderr, - ) - sys.exit(1) - except json.JSONDecodeError as e: - print(f"Error: Failed to parse input file {args.input}: {e}", file=sys.stderr) - sys.exit(1) + youtube_data_list = _load_json_array(input_path, "Input file") if not youtube_data_list: print("Error: Input file contains no entries.", file=sys.stderr) @@ -685,28 +722,7 @@ def main(): existing_entries = [] existing_urls = set() if args.append and args.output: - output_path = Path(args.output) - if output_path.exists(): - try: - with open(output_path, "r", encoding="utf-8") as f: - existing_entries = json.load(f) - if not isinstance(existing_entries, list): - print( - f"Error: Existing file {args.output} does not contain a JSON array.", - file=sys.stderr, - ) - sys.exit(1) - existing_urls = {entry.get("youtubeUrl") for entry in existing_entries} - print( - f"Loaded {len(existing_entries)} existing entries from {args.output}.", - file=sys.stderr, - ) - except json.JSONDecodeError as e: - print( - f"Error: Failed to parse existing file {args.output}: {e}", - file=sys.stderr, - ) - sys.exit(1) + existing_entries, existing_urls = _load_existing_output(Path(args.output)) # Filter out already-processed entries when appending if existing_urls: diff --git a/scripts/extract-metadata/fetch_youtube.py b/scripts/extract-metadata/fetch_youtube.py index d98427b..58af68a 100644 --- a/scripts/extract-metadata/fetch_youtube.py +++ b/scripts/extract-metadata/fetch_youtube.py @@ -85,6 +85,32 @@ def fetch_youtube_metadata(url: str, include_transcript: bool = True) -> dict: } +def _fetch_subtitle_from_url(url: str) -> str | None: + """Fetch subtitle data from a URL with retry on rate limiting.""" + import urllib.request + + for attempt in range(3): + try: + with urllib.request.urlopen(url, timeout=15) as resp: + return resp.read().decode("utf-8", errors="replace") + except urllib.error.HTTPError as e: + if e.code == 429 and attempt < 2: + wait = 5 * (attempt + 1) + print( + f" Subtitle fetch rate-limited, retrying in {wait}s...", + file=sys.stderr, + ) + time.sleep(wait) + continue + print(f" Subtitle fetch failed: {e}", file=sys.stderr) + return None + except Exception as e: + print(f" Subtitle fetch failed: {e}", file=sys.stderr) + return None + + return None + + def _extract_transcript(info: dict) -> str | None: """Extract transcript text from yt-dlp subtitle data. @@ -104,75 +130,74 @@ def _extract_transcript(info: dict) -> str | None: # If yt-dlp provided a URL but no inline data, we need to fetch it sub_url = en_sub.get("url") - if sub_url: - import urllib.request - - for attempt in range(3): - try: - with urllib.request.urlopen(sub_url, timeout=15) as resp: - raw = resp.read().decode("utf-8", errors="replace") - return _parse_subtitle_data(raw, en_sub.get("ext", "")) - except urllib.error.HTTPError as e: - if e.code == 429 and attempt < 2: - wait = 5 * (attempt + 1) - print( - f" Subtitle fetch rate-limited, retrying in {wait}s...", - file=sys.stderr, - ) - time.sleep(wait) - else: - print(f" Subtitle fetch failed: {e}", file=sys.stderr) - return None - except Exception as e: - print(f" Subtitle fetch failed: {e}", file=sys.stderr) - return None + if not sub_url: + return None - return None + raw = _fetch_subtitle_from_url(sub_url) + if raw is None: + return None + return _parse_subtitle_data(raw, en_sub.get("ext", "")) -def _parse_subtitle_data(data: str, ext: str) -> str | None: - """Parse subtitle data from various formats into plain text.""" - if ext == "json3": - try: - parsed = json.loads(data) - segments = [] - for event in parsed.get("events", []): - for seg in event.get("segs", []): - text = seg.get("utf8", "").strip() - if text and text != "\n": - segments.append(text) - if segments: - return " ".join(segments) - except (json.JSONDecodeError, KeyError): - pass - - # Fallback: strip timing lines from vtt/srv formats +def _parse_json3_subtitles(data: str) -> str | None: + """Parse json3 format subtitle data into plain text.""" + try: + parsed = json.loads(data) + except (json.JSONDecodeError, KeyError): + return None + + segments = [] + for event in parsed.get("events", []): + for seg in event.get("segs", []): + text = seg.get("utf8", "").strip() + if text and text != "\n": + segments.append(text) + + return " ".join(segments) if segments else None + + +def _is_vtt_metadata_line(line: str) -> bool: + """Check if a line is a VTT/SRV metadata line that should be skipped.""" + if not line: + return True + if "-->" in line: + return True + if line.startswith(("WEBVTT", "Kind:", "Language:")): + return True + return line.isdigit() + + +def _parse_vtt_subtitles(data: str) -> str | None: + """Parse VTT/SRV format subtitle data into plain text.""" lines = [] - for line in data.splitlines(): - line = line.strip() - # Skip empty lines, timing lines, WEBVTT headers, and numeric cue IDs - if not line: - continue - if "-->" in line: - continue - if line.startswith("WEBVTT") or line.startswith("Kind:") or line.startswith("Language:"): - continue - if line.isdigit(): + for raw_line in data.splitlines(): + line = raw_line.strip() + if _is_vtt_metadata_line(line): continue # Remove HTML-style tags (e.g. , , <00:01:02.345>) - clean = re.sub(r"<[^>]+>", "", line) - if clean.strip(): - lines.append(clean.strip()) - - if lines: - # Deduplicate consecutive identical lines (common in vtt) - deduped = [lines[0]] - for ln in lines[1:]: - if ln != deduped[-1]: - deduped.append(ln) - return " ".join(deduped) + clean = re.sub(r"<[^>]+>", "", line).strip() + if clean: + lines.append(clean) - return None + if not lines: + return None + + # Deduplicate consecutive identical lines (common in vtt) + deduped = [lines[0]] + for ln in lines[1:]: + if ln != deduped[-1]: + deduped.append(ln) + return " ".join(deduped) + + +def _parse_subtitle_data(data: str, ext: str) -> str | None: + """Parse subtitle data from various formats into plain text.""" + if ext == "json3": + result = _parse_json3_subtitles(data) + if result: + return result + + return _parse_vtt_subtitles(data) def _pick_best_thumbnail(info: dict) -> str | None: @@ -209,6 +234,53 @@ def _pick_best_thumbnail(info: dict) -> str | None: return thumbnail +def _collect_urls(args) -> list[str]: + """Collect URLs from command-line arguments or file.""" + if args.url: + return [args.url.strip()] + + file_path = Path(args.file) + if not file_path.exists(): + print(f"Error: File not found: {args.file}", file=sys.stderr) + sys.exit(1) + + urls = [] + with open(file_path, "r", encoding="utf-8") as f: + for line in f: + line = line.strip() + if line and not line.startswith("#"): + urls.append(line) + return urls + + +def _write_json_output(path: Path, data: list) -> None: + """Write a JSON array to a file.""" + with open(path, "w", encoding="utf-8") as f: + json.dump(data, f, indent=2, ensure_ascii=False) + f.write("\n") + + +def _load_existing_output(output_path: Path) -> tuple[list, set]: + """Load existing entries from output file for append mode.""" + if not output_path.exists(): + return [], set() + + try: + with open(output_path, "r", encoding="utf-8") as f: + entries = json.load(f) + except json.JSONDecodeError as e: + print(f"Error: Failed to parse existing file {output_path}: {e}", file=sys.stderr) + sys.exit(1) + + if not isinstance(entries, list): + print(f"Error: Existing file {output_path} does not contain a JSON array.", file=sys.stderr) + sys.exit(1) + + urls = {entry.get("url") for entry in entries} + print(f"Loaded {len(entries)} existing entries from {output_path}.", file=sys.stderr) + return entries, urls + + def main(): parser = argparse.ArgumentParser( description="Fetch YouTube video metadata and transcripts for AccountabilityAtlas.", @@ -241,6 +313,13 @@ def main(): action="store_true", help="Append to existing output file, skipping URLs already present.", ) + parser.add_argument( + "--delay", + "-d", + type=float, + default=0, + help="Seconds to wait between videos (default: 0). Use 5-10 to avoid rate limiting.", + ) args = parser.parse_args() @@ -253,19 +332,7 @@ def main(): parser.error("--append requires --output.") # Collect URLs - urls = [] - if args.url: - urls.append(args.url.strip()) - else: - file_path = Path(args.file) - if not file_path.exists(): - print(f"Error: File not found: {args.file}", file=sys.stderr) - sys.exit(1) - with open(file_path, "r", encoding="utf-8") as f: - for line in f: - line = line.strip() - if line and not line.startswith("#"): - urls.append(line) + urls = _collect_urls(args) if not urls: print("Error: No URLs to process.", file=sys.stderr) @@ -277,28 +344,7 @@ def main(): existing_entries = [] existing_urls = set() if args.append and args.output: - output_path = Path(args.output) - if output_path.exists(): - try: - with open(output_path, "r", encoding="utf-8") as f: - existing_entries = json.load(f) - if not isinstance(existing_entries, list): - print( - f"Error: Existing file {args.output} does not contain a JSON array.", - file=sys.stderr, - ) - sys.exit(1) - existing_urls = {entry.get("url") for entry in existing_entries} - print( - f"Loaded {len(existing_entries)} existing entries from {args.output}.", - file=sys.stderr, - ) - except json.JSONDecodeError as e: - print( - f"Error: Failed to parse existing file {args.output}: {e}", - file=sys.stderr, - ) - sys.exit(1) + existing_entries, existing_urls = _load_existing_output(Path(args.output)) # Filter out already-fetched URLs when appending if existing_urls: @@ -315,6 +361,10 @@ def main(): # Fetch metadata for each URL results = list(existing_entries) errors = [] + output_path = Path(args.output) if args.output else None + + if output_path: + output_path.parent.mkdir(parents=True, exist_ok=True) for i, url in enumerate(urls, 1): print(f"\n[{i}/{len(urls)}] Fetching metadata for: {url}", file=sys.stderr) @@ -329,23 +379,24 @@ def main(): file=sys.stderr, ) print(f" Done: {data.get('title', 'Unknown')}", file=sys.stderr) + + # Write incrementally so progress survives interruptions + if output_path: + _write_json_output(output_path, results) + + # Delay between videos to avoid rate limiting + if args.delay > 0 and i < len(urls): + time.sleep(args.delay) except Exception as e: error_msg = f"Failed to fetch {url}: {e}" print(f" Error: {error_msg}", file=sys.stderr) errors.append(error_msg) - # Output results - output_json = json.dumps(results, indent=2, ensure_ascii=False) - - if args.output: - output_path = Path(args.output) - output_path.parent.mkdir(parents=True, exist_ok=True) - with open(output_path, "w", encoding="utf-8") as f: - f.write(output_json) - f.write("\n") + # Final output (stdout mode, or summary for file mode) + if output_path: print(f"\nWrote {len(results)} entries to {args.output}.", file=sys.stderr) else: - print(output_json) + print(json.dumps(results, indent=2, ensure_ascii=False)) # Summary new_count = len(results) - len(existing_entries) diff --git a/scripts/list-channel/list_channel.py b/scripts/list-channel/list_channel.py index 731fb17..41d7f72 100644 --- a/scripts/list-channel/list_channel.py +++ b/scripts/list-channel/list_channel.py @@ -27,6 +27,18 @@ sys.exit(1) +_CHANNEL_PATH_SUFFIXES = ("/videos", "/shorts", "/streams", "/playlists", "/community") + + +def _strip_channel_path_suffix(url: str) -> str: + """Strip known YouTube channel path suffixes from a URL.""" + url = url.rstrip("/") + for suffix in _CHANNEL_PATH_SUFFIXES: + if url.endswith(suffix): + return url[: -len(suffix)] + return url + + def normalize_channel_url(channel: str) -> str: """Normalize a channel identifier to a full YouTube URL with /videos suffix. @@ -41,14 +53,8 @@ def normalize_channel_url(channel: str) -> str: channel = channel.strip() # Already a full URL - if channel.startswith("http://") or channel.startswith("https://"): - # Strip trailing slashes and path suffixes like /videos, /shorts, /streams - url = channel.rstrip("/") - for suffix in ("/videos", "/shorts", "/streams", "/playlists", "/community"): - if url.endswith(suffix): - url = url[: -len(suffix)] - break - return url + "/videos" + if channel.startswith(("http://", "https://")): + return _strip_channel_path_suffix(channel) + "/videos" # @handle if channel.startswith("@"): @@ -79,6 +85,19 @@ def _build_ydl_opts(max_results: int | None) -> dict: return opts +def _is_outside_date_range( + upload_date: str, after_date: str | None, before_date: str | None +) -> bool: + """Check if an upload date falls outside the specified range.""" + if not upload_date: + return False + if after_date and upload_date < after_date: + return True + if before_date and upload_date > before_date: + return True + return False + + def _parse_entry( entry: dict | None, min_duration: int, @@ -93,13 +112,9 @@ def _parse_entry( if duration < min_duration: return None - # Date filtering — upload_date is YYYYMMDD from yt-dlp upload_date = entry.get("upload_date", "") - if upload_date and (after_date or before_date): - if after_date and upload_date < after_date: - return None - if before_date and upload_date > before_date: - return None + if _is_outside_date_range(upload_date, after_date, before_date): + return None video_url = entry.get("webpage_url") or entry.get("url", "") if not video_url: @@ -181,6 +196,17 @@ def format_output(videos: list[dict], channel_name: str) -> str: return "\n".join(lines) + "\n" +def _parse_date_arg(parser, value: str | None, name: str) -> str | None: + """Validate a YYYY-MM-DD date argument and convert to YYYYMMDD.""" + if value is None: + return None + try: + datetime.strptime(value, "%Y-%m-%d") + except ValueError: + parser.error(f"{name} must be in YYYY-MM-DD format, got: {value}") + return value.replace("-", "") + + def main(): parser = argparse.ArgumentParser( description="List YouTube video URLs from a channel for AccountabilityAtlas.", @@ -226,18 +252,8 @@ def main(): args = parser.parse_args() # Validate date formats and convert to YYYYMMDD for yt-dlp comparison - after_yyyymmdd = None - before_yyyymmdd = None - for date_arg, name in [(args.after, "--after"), (args.before, "--before")]: - if date_arg is not None: - try: - datetime.strptime(date_arg, "%Y-%m-%d") - except ValueError: - parser.error(f"{name} must be in YYYY-MM-DD format, got: {date_arg}") - if args.after: - after_yyyymmdd = args.after.replace("-", "") - if args.before: - before_yyyymmdd = args.before.replace("-", "") + after_yyyymmdd = _parse_date_arg(parser, args.after, "--after") + before_yyyymmdd = _parse_date_arg(parser, args.before, "--before") channel_url = normalize_channel_url(args.channel) From 7933384898d4d0835ca90bae41bced3736197684 Mon Sep 17 00:00:00 2001 From: Kelley Glenn Date: Sun, 22 Feb 2026 19:19:03 -0800 Subject: [PATCH 9/9] feat: add --cookies-from-browser flag and reduce main() complexity Add --cookies-from-browser option to fetch_youtube.py for authenticated YouTube sessions (~6x higher rate limits). Extract helper functions from main() in both fetch_youtube.py and claude_extract.py to bring cognitive complexity under SonarCloud's threshold of 15. Co-Authored-By: Claude Opus 4.6 --- scripts/extract-metadata/claude_extract.py | 111 +++++++------ scripts/extract-metadata/fetch_youtube.py | 173 +++++++++++++-------- 2 files changed, 171 insertions(+), 113 deletions(-) diff --git a/scripts/extract-metadata/claude_extract.py b/scripts/extract-metadata/claude_extract.py index 86f227d..245664b 100644 --- a/scripts/extract-metadata/claude_extract.py +++ b/scripts/extract-metadata/claude_extract.py @@ -650,6 +650,64 @@ def _load_existing_output(output_path: Path) -> tuple[list, set]: return entries, urls +def _filter_existing_urls(data_list: list[dict], existing_urls: set) -> list[dict]: + """Remove already-processed entries and report skipped count.""" + if not existing_urls: + return data_list + + filtered = [d for d in data_list if d.get("url") not in existing_urls] + skipped = len(data_list) - len(filtered) + if skipped: + print(f"Skipping {skipped} already-extracted URL(s).", file=sys.stderr) + return filtered + + +def _process_sequential( + youtube_data_list: list[dict], + client: anthropic.Anthropic, + model: str, + results: list[dict], + errors: list[str], +) -> None: + """Process videos one at a time through Claude extraction.""" + for i, yt_data in enumerate(youtube_data_list, 1): + url = yt_data.get("url", "unknown") + print(f"\n[{i}/{len(youtube_data_list)}] Processing: {url}", file=sys.stderr) + try: + entry = process_single(yt_data, client, model) + results.append(entry) + except Exception as e: + error_msg = f"Failed to process {url}: {e}" + print(f" Error: {error_msg}", file=sys.stderr) + errors.append(error_msg) + + +def _write_output(output_arg: str | None, results: list[dict]) -> None: + """Write results to file or stdout.""" + output_json = json.dumps(results, indent=2, ensure_ascii=False) + + if output_arg: + output_path = Path(output_arg) + output_path.parent.mkdir(parents=True, exist_ok=True) + with open(output_path, "w", encoding="utf-8") as f: + f.write(output_json) + f.write("\n") + print(f"\nWrote {len(results)} entries to {output_arg}.", file=sys.stderr) + else: + print(output_json) + + +def _print_summary(new_count: int, errors: list[str]) -> None: + """Print summary and exit with appropriate code.""" + if errors: + print(f"\nCompleted with {len(errors)} error(s):", file=sys.stderr) + for err in errors: + print(f" - {err}", file=sys.stderr) + sys.exit(1 if new_count == 0 else 0) + else: + print(f"\nSuccessfully processed {new_count} entry(ies).", file=sys.stderr) + + def main(): parser = argparse.ArgumentParser( description="Extract structured metadata from YouTube data using Claude for AccountabilityAtlas.", @@ -690,7 +748,6 @@ def main(): args = parser.parse_args() - # Validate arguments if args.append and not args.output: parser.error("--append requires --output.") @@ -701,7 +758,6 @@ def main(): sys.exit(1) youtube_data_list = _load_json_array(input_path, "Input file") - if not youtube_data_list: print("Error: Input file contains no entries.", file=sys.stderr) sys.exit(1) @@ -720,19 +776,9 @@ def main(): # Load existing entries if appending existing_entries = [] - existing_urls = set() if args.append and args.output: existing_entries, existing_urls = _load_existing_output(Path(args.output)) - - # Filter out already-processed entries when appending - if existing_urls: - original_count = len(youtube_data_list) - youtube_data_list = [ - d for d in youtube_data_list if d.get("url") not in existing_urls - ] - skipped = original_count - len(youtube_data_list) - if skipped: - print(f"Skipping {skipped} already-extracted URL(s).", file=sys.stderr) + youtube_data_list = _filter_existing_urls(youtube_data_list, existing_urls) if not youtube_data_list and existing_entries: print("All entries already extracted. Nothing to do.", file=sys.stderr) @@ -743,45 +789,14 @@ def main(): errors = [] if args.batch: - batch_results, batch_errors = process_batch( - youtube_data_list, client, args.model - ) + batch_results, batch_errors = process_batch(youtube_data_list, client, args.model) results.extend(batch_results) errors.extend(batch_errors) else: - for i, yt_data in enumerate(youtube_data_list, 1): - url = yt_data.get("url", "unknown") - print(f"\n[{i}/{len(youtube_data_list)}] Processing: {url}", file=sys.stderr) - try: - entry = process_single(yt_data, client, args.model) - results.append(entry) - except Exception as e: - error_msg = f"Failed to process {url}: {e}" - print(f" Error: {error_msg}", file=sys.stderr) - errors.append(error_msg) - - # Output results - output_json = json.dumps(results, indent=2, ensure_ascii=False) + _process_sequential(youtube_data_list, client, args.model, results, errors) - if args.output: - output_path = Path(args.output) - output_path.parent.mkdir(parents=True, exist_ok=True) - with open(output_path, "w", encoding="utf-8") as f: - f.write(output_json) - f.write("\n") - print(f"\nWrote {len(results)} entries to {args.output}.", file=sys.stderr) - else: - print(output_json) - - # Summary - new_count = len(results) - len(existing_entries) - if errors: - print(f"\nCompleted with {len(errors)} error(s):", file=sys.stderr) - for err in errors: - print(f" - {err}", file=sys.stderr) - sys.exit(1 if new_count == 0 else 0) - else: - print(f"\nSuccessfully processed {new_count} entry(ies).", file=sys.stderr) + _write_output(args.output, results) + _print_summary(len(results) - len(existing_entries), errors) if __name__ == "__main__": diff --git a/scripts/extract-metadata/fetch_youtube.py b/scripts/extract-metadata/fetch_youtube.py index 58af68a..e304c53 100644 --- a/scripts/extract-metadata/fetch_youtube.py +++ b/scripts/extract-metadata/fetch_youtube.py @@ -33,12 +33,17 @@ # --- YouTube fetching --- -def fetch_youtube_metadata(url: str, include_transcript: bool = True) -> dict: +def fetch_youtube_metadata( + url: str, + include_transcript: bool = True, + cookies_from_browser: str | None = None, +) -> dict: """Fetch video metadata and optionally transcript from YouTube using yt-dlp. Args: url: YouTube video URL. include_transcript: Whether to attempt fetching auto-generated subtitles. + cookies_from_browser: Browser name to read cookies from (e.g., "firefox"). Returns: Dictionary with keys: url, title, description, channel, thumbnail, @@ -53,6 +58,9 @@ def fetch_youtube_metadata(url: str, include_transcript: bool = True) -> dict: "sleep_interval": 2, } + if cookies_from_browser: + ydl_opts["cookiesfrombrowser"] = (cookies_from_browser,) + if include_transcript: ydl_opts.update( { @@ -281,6 +289,87 @@ def _load_existing_output(output_path: Path) -> tuple[list, set]: return entries, urls +def _validate_args(parser, args) -> None: + """Validate CLI argument combinations.""" + if not args.url and not args.file: + parser.error("Provide either a URL argument or --file with a file of URLs.") + if args.url and args.file: + parser.error("Provide either a URL argument or --file, not both.") + if args.append and not args.output: + parser.error("--append requires --output.") + + +def _filter_existing_urls(urls: list[str], existing_urls: set) -> list[str]: + """Remove already-fetched URLs and report skipped count.""" + if not existing_urls: + return urls + + filtered = [u for u in urls if u not in existing_urls] + skipped = len(urls) - len(filtered) + if skipped: + print(f"Skipping {skipped} already-fetched URL(s).", file=sys.stderr) + return filtered + + +def _fetch_all( + urls: list[str], + include_transcript: bool, + cookies_from_browser: str | None, + output_path: Path | None, + delay: float, + results: list[dict], +) -> list[str]: + """Fetch metadata for all URLs, writing incrementally. Returns errors list.""" + errors = [] + + for i, url in enumerate(urls, 1): + print(f"\n[{i}/{len(urls)}] Fetching metadata for: {url}", file=sys.stderr) + try: + data = fetch_youtube_metadata( + url, include_transcript=include_transcript, + cookies_from_browser=cookies_from_browser, + ) + results.append(data) + + if include_transcript and data.get("transcript") is None: + print(" Warning: No transcript available for this video.", file=sys.stderr) + print(f" Done: {data.get('title', 'Unknown')}", file=sys.stderr) + + if output_path: + _write_json_output(output_path, results) + if delay > 0 and i < len(urls): + time.sleep(delay) + except Exception as e: + error_msg = f"Failed to fetch {url}: {e}" + print(f" Error: {error_msg}", file=sys.stderr) + errors.append(error_msg) + + return errors + + +def _print_summary( + output_path: Path | None, + output_arg: str | None, + results: list[dict], + existing_count: int, + errors: list[str], +) -> None: + """Print final output and summary.""" + if output_path: + print(f"\nWrote {len(results)} entries to {output_arg}.", file=sys.stderr) + else: + print(json.dumps(results, indent=2, ensure_ascii=False)) + + new_count = len(results) - existing_count + if errors: + print(f"\nCompleted with {len(errors)} error(s):", file=sys.stderr) + for err in errors: + print(f" - {err}", file=sys.stderr) + sys.exit(1 if new_count == 0 else 0) + else: + print(f"\nSuccessfully fetched {new_count} URL(s).", file=sys.stderr) + + def main(): parser = argparse.ArgumentParser( description="Fetch YouTube video metadata and transcripts for AccountabilityAtlas.", @@ -320,93 +409,47 @@ def main(): default=0, help="Seconds to wait between videos (default: 0). Use 5-10 to avoid rate limiting.", ) + parser.add_argument( + "--cookies-from-browser", + type=str, + default=None, + metavar="BROWSER", + help="Browser to read YouTube cookies from (e.g., firefox, chrome). Raises rate limits ~6x.", + ) args = parser.parse_args() + _validate_args(parser, args) - # Validate arguments - if not args.url and not args.file: - parser.error("Provide either a URL argument or --file with a file of URLs.") - if args.url and args.file: - parser.error("Provide either a URL argument or --file, not both.") - if args.append and not args.output: - parser.error("--append requires --output.") - - # Collect URLs urls = _collect_urls(args) - if not urls: print("Error: No URLs to process.", file=sys.stderr) sys.exit(1) - include_transcript = not args.no_transcript - # Load existing entries if appending existing_entries = [] - existing_urls = set() if args.append and args.output: existing_entries, existing_urls = _load_existing_output(Path(args.output)) - - # Filter out already-fetched URLs when appending - if existing_urls: - original_count = len(urls) - urls = [u for u in urls if u not in existing_urls] - skipped = original_count - len(urls) - if skipped: - print(f"Skipping {skipped} already-fetched URL(s).", file=sys.stderr) + urls = _filter_existing_urls(urls, existing_urls) if not urls and existing_entries: print("All URLs already fetched. Nothing to do.", file=sys.stderr) sys.exit(0) - # Fetch metadata for each URL results = list(existing_entries) - errors = [] output_path = Path(args.output) if args.output else None - if output_path: output_path.parent.mkdir(parents=True, exist_ok=True) - for i, url in enumerate(urls, 1): - print(f"\n[{i}/{len(urls)}] Fetching metadata for: {url}", file=sys.stderr) - try: - data = fetch_youtube_metadata(url, include_transcript=include_transcript) - results.append(data) - - has_transcript = data.get("transcript") is not None - if include_transcript and not has_transcript: - print( - " Warning: No transcript available for this video.", - file=sys.stderr, - ) - print(f" Done: {data.get('title', 'Unknown')}", file=sys.stderr) - - # Write incrementally so progress survives interruptions - if output_path: - _write_json_output(output_path, results) - - # Delay between videos to avoid rate limiting - if args.delay > 0 and i < len(urls): - time.sleep(args.delay) - except Exception as e: - error_msg = f"Failed to fetch {url}: {e}" - print(f" Error: {error_msg}", file=sys.stderr) - errors.append(error_msg) - - # Final output (stdout mode, or summary for file mode) - if output_path: - print(f"\nWrote {len(results)} entries to {args.output}.", file=sys.stderr) - else: - print(json.dumps(results, indent=2, ensure_ascii=False)) + errors = _fetch_all( + urls, + include_transcript=not args.no_transcript, + cookies_from_browser=args.cookies_from_browser, + output_path=output_path, + delay=args.delay, + results=results, + ) - # Summary - new_count = len(results) - len(existing_entries) - if errors: - print(f"\nCompleted with {len(errors)} error(s):", file=sys.stderr) - for err in errors: - print(f" - {err}", file=sys.stderr) - sys.exit(1 if new_count == 0 else 0) - else: - print(f"\nSuccessfully fetched {new_count} URL(s).", file=sys.stderr) + _print_summary(output_path, args.output, results, len(existing_entries), errors) if __name__ == "__main__":