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/.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/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). 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..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 { @@ -110,6 +169,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,39 +202,37 @@ 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. +## 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 60% rename from scripts/extract-metadata/extract.py rename to scripts/extract-metadata/claude_extract.py index d6b023c..245664b 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" @@ -134,6 +122,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 +160,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 +238,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 @@ -334,196 +329,37 @@ 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", - } - - if include_transcript: - ydl_opts.update( - { - "writeautomaticsub": True, - "writesubtitles": True, - "subtitleslangs": ["en"], - "subtitlesformat": "json3", - } - ) - - 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: - try: - 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 - - 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) # --- 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. @@ -532,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("}") @@ -570,12 +398,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: @@ -608,17 +433,12 @@ 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"), "city": location.get("city"), "state": location.get("state"), "latitude": location.get("latitude"), @@ -648,33 +468,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) @@ -683,51 +492,67 @@ def process_single_url( return entry -def process_urls_batch( - urls: list[str], +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, 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 = [] - for url, yt_data in youtube_data.items(): + # 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"], description=yt_data["description"], @@ -739,7 +564,7 @@ def process_urls_batch( requests.append( { - "custom_id": url, + "custom_id": video_id, "params": { "model": model, "max_tokens": 4096, @@ -758,7 +583,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( @@ -781,52 +606,119 @@ 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): - url = entry.custom_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) - 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}") - - # 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}") + video_id = entry.custom_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) + _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 _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 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", @@ -842,57 +734,36 @@ 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) - 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) + # 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) + + 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) + print(f"Loaded {len(youtube_data_list)} entries from {args.input}.", file=sys.stderr) + # Initialize Anthropic client try: client = anthropic.Anthropic() @@ -903,76 +774,29 @@ def main(): ) sys.exit(1) - include_transcript = not args.no_transcript - # Load existing entries if appending existing_entries = [] 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) - 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) - - # Process URLs + existing_entries, existing_urls = _load_existing_output(Path(args.output)) + 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) + 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) - try: - entry = process_single_url(url, client, args.model, include_transcript) - 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) - - 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) + _process_sequential(youtube_data_list, client, args.model, results, errors) - # 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} URL(s).", 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 new file mode 100644 index 0000000..e304c53 --- /dev/null +++ b/scripts/extract-metadata/fetch_youtube.py @@ -0,0 +1,456 @@ +#!/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, + 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, + 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 cookies_from_browser: + ydl_opts["cookiesfrombrowser"] = (cookies_from_browser,) + + 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 _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. + + 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 not sub_url: + 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_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 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).strip() + if clean: + lines.append(clean) + + 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: + """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 _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 _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.", + ) + 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.", + ) + parser.add_argument( + "--delay", + "-d", + type=float, + 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) + + urls = _collect_urls(args) + if not urls: + print("Error: No URLs to process.", file=sys.stderr) + sys.exit(1) + + # Load existing entries if appending + existing_entries = [] + if args.append and args.output: + existing_entries, existing_urls = _load_existing_output(Path(args.output)) + 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) + + results = list(existing_entries) + output_path = Path(args.output) if args.output else None + if output_path: + output_path.parent.mkdir(parents=True, exist_ok=True) + + 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, + ) + + _print_summary(output_path, args.output, results, len(existing_entries), errors) + + +if __name__ == "__main__": + main() 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..41d7f72 --- /dev/null +++ b/scripts/list-channel/list_channel.py @@ -0,0 +1,290 @@ +#!/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) + + +_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. + + 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://", "https://")): + return _strip_channel_path_suffix(channel) + "/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 _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": True, + "ignoreerrors": True, + } + + # 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 _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, + 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 + + duration = entry.get("duration") or 0 + if duration < min_duration: + return None + + upload_date = entry.get("upload_date", "") + 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: + 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": upload_date, + } + + +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 (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) + + 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 [] + + videos = [] + for entry in info.get("entries") or []: + 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: + 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 _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.", + 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 and convert to YYYYMMDD for yt-dlp comparison + 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) + + videos = fetch_channel_videos( + channel_url=channel_url, + max_results=args.max_results, + after_date=after_yyyymmdd, + before_date=before_yyyymmdd, + 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..aca66cd 100644 --- a/scripts/seed-videos.sh +++ b/scripts/seed-videos.sh @@ -100,18 +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 (requires a street address) if [[ "$LAT" == "null" || "$LNG" == "null" ]]; then - ADDRESS_PARTS="" - [[ -n "$LOC_NAME" ]] && ADDRESS_PARTS="$LOC_NAME" - [[ -n "$LOC_CITY" ]] && ADDRESS_PARTS="${ADDRESS_PARTS:+$ADDRESS_PARTS, }$LOC_CITY" - [[ -n "$LOC_STATE" ]] && ADDRESS_PARTS="${ADDRESS_PARTS:+$ADDRESS_PARTS, }$LOC_STATE" + if [[ -n "$LOC_STREET" ]]; then + ADDRESS_PARTS="$LOC_STREET" + [[ -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") @@ -119,8 +119,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 @@ -150,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 // []') @@ -164,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" \ 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__/**