references/fetcher-notes.md
# Fetcher Architecture and Anti-Bot Notes
Reference for understanding and extending `scripts/fetch-sources.py`.
<architecture>
## How the Fetcher Works
The fetcher uses `concurrent.futures.ThreadPoolExecutor` to fetch multiple URLs simultaneously, with per-domain rate limiting to avoid triggering anti-bot protections on any single site.
```
manifest.md (pending entries)
│
├─ Thread 1: ref-01 (arxiv.org) ─┐
├─ Thread 2: ref-02 (redhat.com) ─┤ concurrent, different domains
├─ Thread 3: ref-03 (redhat.com) ─┤ Thread 3 waits for Thread 2's
│ │ domain lock to release
└─ Thread 4: ref-04 (thenewstack.io) ─┘
│
▼
sources/ directory (one .md file per successful fetch)
manifest.md (updated with status, file paths, notes)
```
**Domain-aware throttling:** Each domain has a threading lock and a timestamp. Before fetching, a thread acquires the lock for its domain and sleeps if the last request to that domain was less than `--delay` seconds ago. This means requests to *different* domains run in parallel while requests to the *same* domain are serialized with the configured delay.
</architecture>
<known_blocked_domains>
## Known WAF-Blocked and Access-Restricted Domains
Some domains return 200 OK on HEAD requests but serve "Access Denied" body content
on GET, or block non-browser clients at the application layer. These are distinct
from 403 at the proxy layer.
| Domain | Behavior | Workaround |
|--------|----------|------------|
| `docs.redhat.com` | ✅ Accessible — Nuxt.js SPA with SSR; use `#main-content` selector | Use `--stealth` + single-page format (`/html-single/...`) for best results |
| `developers.redhat.com` | 403 at request layer | `--stealth` sometimes helps; otherwise headless browser or manual copy-paste |
| `medium.com` | JS-rendered behind bot check | Headless browser fallback |
| `access.redhat.com` | Usually accessible (200 OK) | N/A |
**docs.redhat.com is accessible.** It serves a Nuxt.js SPA with server-side rendering — the full document content is in the initial HTML. Use the `/html-single/` URL format (single-page) rather than `/html/` (multi-page). The `#main-content` selector in the fetcher targets the correct section. Strip `<nav>`, `<header>`, `<footer>`, and `<aside>` tags to remove boilerplate.
</known_blocked_domains>
<anti_bot>
## Why Sources Return 403
Sites like `developers.redhat.com` and `medium.com` use several layers of bot detection:
1. **IP reputation:** Rapid sequential requests from one IP get flagged. The concurrent fetcher with domain throttling helps, but the IP itself may already be flagged from prior sessions.
2. **User-Agent filtering:** Old or obviously fake UAs get blocked. The `--stealth` flag rotates through 5 current browser UA strings.
3. **Missing browser headers:** Real browsers send `Accept`, `Accept-Language`, `Sec-Fetch-*`, etc. Bot requests typically omit these. The `--stealth` flag adds a full set of browser-mimicry headers.
4. **JavaScript challenge pages:** Cloudflare and similar CDNs serve a JS challenge page that `requests` can't execute. This is the hardest to bypass without a real browser.
5. **Cookie/session requirements:** Some sites require an initial page load to set cookies before the article URL works. `requests` handles cookies within a session but doesn't execute JS-based cookie setters.
</anti_bot>
<proxy_usage>
## Using a Proxy / VPN
When your IP is flagged (403s on sites that normally work, or your regular browsing slows down after a fetch session):
**Option A: SOCKS5 proxy through VPN**
If your VPN exposes a SOCKS5 proxy (most do, e.g., on `127.0.0.1:1080`):
```bash
pip install "requests[socks]"
python3 fetch-sources.py research/subject/ --retry-failed --stealth --proxy socks5://127.0.0.1:1080
```
**Option B: System-wide VPN**
Turn on your VPN before running the fetcher. All traffic routes through the VPN tunnel:
```bash
# VPN already active
python3 fetch-sources.py research/subject/ --retry-failed --stealth
```
**Option C: HTTP proxy**
If you have an HTTP proxy available:
```bash
python3 fetch-sources.py research/subject/ --retry-failed --stealth --proxy http://proxy-host:8080
```
**Split-tunneling tip:** If your VPN supports split tunneling, route only the fetch script through the VPN while keeping your browser on your normal connection. This prevents both from being rate-limited simultaneously.
</proxy_usage>
<browser_fallback>
## Future: Headless Browser Fallback
For sites with aggressive JS-based bot protection (Cloudflare challenges, JS-rendered content), a headless browser fallback could handle the cases `requests` cannot.
**Approach:** A separate script (`fetch-sources-browser.py`) that:
1. Reads the manifest for entries with `status: failed` and `notes: HTTP 403`
2. Launches headless Chromium via `playwright`
3. Navigates to each URL, waits for JS to render, extracts content
4. Saves to the same `sources/` directory and updates the manifest
**Dependencies:**
```bash
pip install playwright
playwright install chromium
```
**Why separate:** Playwright is a heavy dependency (~150 MB for Chromium). Keeping it in a separate script means the main fetcher stays lightweight and the browser fallback is opt-in.
**Trade-offs:**
- Much slower (2-5 seconds per page vs. 0.5-1s with `requests`)
- Higher resource usage (headless Chromium process)
- But handles JS challenges, cookie-gated pages, and client-side-rendered content
- Not needed if proxy + stealth headers resolve the 403s
This is not yet implemented. If proxy + stealth doesn't achieve >85% capture rate, implementing this fallback is the next step.
</browser_fallback>
<pdf_support>
## PDF Extraction
The fetcher detects `application/pdf` responses and extracts text using `pdfplumber` (optional dependency). Each page is separated by a horizontal rule. If `pdfplumber` is not installed, PDFs are logged as failures with an install hint.
```bash
pip install pdfplumber
```
Limitations:
- Scanned/image PDFs won't extract text (would need OCR)
- Complex table layouts may not preserve structure
- Very large PDFs are truncated to 500K characters
</pdf_support>
<content_extraction>
## HTML Content Extraction Strategy
The fetcher uses a three-tier extraction strategy to find the main article content:
1. **Domain-specific selectors:** Known CSS selectors for common domains (Red Hat docs, Medium, arXiv, Microsoft Learn). Checked first because they're most accurate.
2. **Generic article selectors:** Common patterns like `<article>`, `<main>`, `[role="main"]`, `.post-content`, etc. Covers most well-structured sites.
3. **Body fallback:** If neither of the above produces >= 200 chars of useful text, falls back to the full `<body>`.
If the final extracted text is under 200 characters, the fetcher marks it as `low-content` in the manifest notes. This flags pages that returned mostly navigation or JavaScript placeholders.
To add support for a new domain, add an entry to `DOMAIN_SELECTORS` in the script.
</content_extraction>
references/verification-patterns.md
# Verification Patterns
Common claim types and how to check them against sources.
<numeric_claims>
## Numeric Claims (Cost, Performance, Percentages)
These are the highest-risk claims — specific numbers carry authority but are easy to misrepresent.
**Check for:**
- Is the number actually in the source? (not inferred or rounded)
- What conditions does the source attach? ("up to", "under specific workload", "assuming full utilization")
- Is the comparison fair? (apples-to-apples or apples-to-oranges)
- What time horizon? (5-year amortization vs. 1-year)
- What baseline? (on-demand vs. reserved vs. spot pricing)
- Are "up to" figures presented as typical?
**Red flags:**
- Source says "up to 18x" under specific conditions, article says "18x cost advantage"
- Source compares against frontier API, article implies comparison against equivalent infrastructure
- Breakeven calculated against on-demand pricing when enterprises use reserved instances
- Numbers from vendor marketing materials presented as independent analysis
</numeric_claims>
<maturity_claims>
## Maturity and Readiness Claims
Articles often present features as production-ready when sources describe them as preview or experimental.
**Check for:**
- Does the source mark features as GA, Tech Preview, Developer Preview, Alpha, or Beta?
- Is there a support statement? ("not covered by Red Hat support agreements")
- Are there known limitations listed that the article omits?
- Is the feature on a "roadmap" vs. "available now"?
- What version introduced it? Is that version widely deployed?
**Red flags:**
- Official docs say "Technology Preview" but article treats feature as production-ready
- GitHub repo shows alpha version numbers (v0.x) but article presents as mature
- Feature described in a Red Hat "Emerging Technologies" blog (experimental) vs. official product docs (supported)
</maturity_claims>
<architecture_claims>
## Architecture and Implementation Claims
These tend to be the most reliable category — architecture is factual and well-documented.
**Check for:**
- Does the described stack match official documentation?
- Are component names and versions correct?
- Is the dependency chain accurate? (A requires B which requires C)
- Are configuration examples syntactically valid?
**Typical confidence:** High — these are either right or wrong, and official docs are the authority.
</architecture_claims>
<strategic_claims>
## Strategic and Qualitative Claims
Hardest to verify because they express opinions, positioning, or predictions.
**Check for:**
- Is the framing consistent with the source's intent?
- Does the source support the specific argument being made, or just the general topic?
- Is vendor positioning presented as independent analysis?
- Are competitive comparisons fair?
**Red flags:**
- Red Hat blog posts cited as independent validation of Red Hat products
- Vendor whitepapers treated as objective analysis
- Industry trends described as certainties
</strategic_claims>
<source_quality>
## Source Quality Assessment
Not all sources carry equal weight.
**Tier 1 — Highest confidence:**
- Official product documentation (docs.redhat.com, docs.nvidia.com)
- Peer-reviewed papers (arXiv with citations, IEEE, ACM)
- Standards body publications (CNCF, Kubernetes SIGs)
**Tier 2 — Good confidence:**
- Official vendor blogs by named engineers/architects
- Established tech journalism (The New Stack, InfoQ, LWN)
- Community blogs with working examples and YAML
**Tier 3 — Moderate confidence:**
- Vendor marketing materials and whitepapers
- Medium posts and personal blogs (check author credentials)
- Conference talks and slide decks
**Tier 4 — Low confidence:**
- SEO-optimized comparison sites
- Unnamed or AI-generated content
- Paywalled content you can't read
- Sources that are no longer accessible
</source_quality>
<synthesis_patterns>
## Patterns to Watch For in Synthesis
When reviewing findings across all batches, look for these systemic patterns:
- **Cherry-picking**: Consistently selecting the most favorable number from each source
- **Context stripping**: Omitting conditions, caveats, or alternative scenarios
- **Maturity inflation**: Treating preview features as production-ready across the board
- **Circular citation**: Article A cites Article B which cites Article A (or a common vendor source)
- **Advocacy posture**: Every topic area presented in the most favorable light
- **AI-generated style**: Uniform sentence structure, no personal experience, generic transitions
</synthesis_patterns>
scripts/fetch-sources.py
#!/usr/bin/env python3
"""
fetch-sources.py — Batch URL fetcher for research workflows.
Reads a manifest file, fetches each URL marked as 'pending', saves content
to the sources/ directory, and updates the manifest with results.
Usage:
python3 fetch-sources.py <research-dir>
python3 fetch-sources.py <research-dir> --workers 4 --stealth
python3 fetch-sources.py <research-dir> --retry-failed --proxy socks5://127.0.0.1:1080
Where <research-dir> contains:
manifest.md — source manifest with URLs and status
sources/ — directory for fetched content (created if missing)
The manifest uses this format (one entry per source):
| ref-id | url | status | file | notes |
Status values: pending, fetched, failed, skipped
Dependencies:
Required: requests, beautifulsoup4, markdownify
Optional: requests[socks] (for SOCKS5 proxy), pdfplumber (for PDF extraction)
"""
import sys
import re
import os
import io
import time
import random
import argparse
import threading
from pathlib import Path
from urllib.parse import urlparse
from collections import defaultdict
from concurrent.futures import ThreadPoolExecutor, as_completed
try:
import requests
from bs4 import BeautifulSoup
from markdownify import markdownify as md
except ImportError:
print("Missing dependencies. Install with:")
print(" pip install requests beautifulsoup4 markdownify")
sys.exit(1)
TIMEOUT = 30
MAX_CONTENT_BYTES = 500_000
MIN_USEFUL_CHARS = 200
USER_AGENTS = [
"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/125.0.0.0 Safari/537.36",
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/125.0.0.0 Safari/537.36",
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/125.0.0.0 Safari/537.36",
"Mozilla/5.0 (X11; Linux x86_64; rv:128.0) Gecko/20100101 Firefox/128.0",
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10.15; rv:128.0) Gecko/20100101 Firefox/128.0",
]
STEALTH_HEADERS = {
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,*/*;q=0.8",
"Accept-Language": "en-US,en;q=0.9",
"Accept-Encoding": "gzip, deflate, br",
"Sec-Fetch-Dest": "document",
"Sec-Fetch-Mode": "navigate",
"Sec-Fetch-Site": "none",
"Sec-Fetch-User": "?1",
"Upgrade-Insecure-Requests": "1",
"Cache-Control": "max-age=0",
}
ARTICLE_SELECTORS = [
"article",
"main",
'[role="main"]',
".post-content",
".article-content",
".entry-content",
".content-body",
"#content",
".markdown-body",
".prose",
]
DOMAIN_SELECTORS = {
"developers.redhat.com": [".assembly", ".rh-article", '[data-analytics-category="article"]'],
"medium.com": [".meteredContent", "article"],
"arxiv.org": [".ltx_document", "#content"],
"docs.redhat.com": ["#main-content"],
"learn.microsoft.com": [".content", "main"],
"ai-on-openshift.io": [".md-content", "article"],
}
_domain_locks = defaultdict(threading.Lock)
_domain_last_request = defaultdict(float)
def parse_manifest(manifest_path: Path) -> list[dict]:
"""Parse markdown table rows from manifest into list of dicts."""
entries = []
with open(manifest_path) as f:
lines = f.readlines()
in_table = False
headers = []
for line in lines:
line = line.strip()
if not line.startswith("|"):
in_table = False
continue
cells = [c.strip() for c in line.split("|")[1:-1]]
if not in_table:
headers = [h.lower().replace(" ", "_") for h in cells]
in_table = True
continue
if all(c.startswith("-") or c.startswith(":") for c in cells):
continue
entry = {}
for i, header in enumerate(headers):
entry[header] = cells[i] if i < len(cells) else ""
entries.append(entry)
return entries
def write_manifest(manifest_path: Path, entries: list[dict], preamble: str):
"""Write entries back as a markdown table, preserving preamble text."""
if not entries:
return
headers = list(entries[0].keys())
header_line = "| " + " | ".join(headers) + " |"
sep_line = "| " + " | ".join("---" for _ in headers) + " |"
with open(manifest_path, "w") as f:
f.write(preamble)
f.write(header_line + "\n")
f.write(sep_line + "\n")
for entry in entries:
row = "| " + " | ".join(str(entry.get(h, "")) for h in headers) + " |"
f.write(row + "\n")
def extract_preamble(manifest_path: Path) -> str:
"""Extract all text before the first markdown table."""
with open(manifest_path) as f:
lines = f.readlines()
preamble_lines = []
for line in lines:
if line.strip().startswith("|"):
break
preamble_lines.append(line)
return "".join(preamble_lines)
def sanitize_filename(ref_id: str) -> str:
"""Convert a ref ID to a safe filename."""
return re.sub(r"[^a-zA-Z0-9_-]", "-", ref_id).strip("-").lower()
def domain_throttle(url: str, min_delay: float):
"""Enforce per-domain rate limiting across threads."""
domain = urlparse(url).netloc
with _domain_locks[domain]:
elapsed = time.time() - _domain_last_request[domain]
if elapsed < min_delay:
time.sleep(min_delay - elapsed)
_domain_last_request[domain] = time.time()
def build_headers(stealth: bool) -> dict:
"""Build request headers, optionally with stealth browser-mimicry headers."""
headers = {"User-Agent": random.choice(USER_AGENTS)}
if stealth:
headers.update(STEALTH_HEADERS)
return headers
def extract_article_content(soup: BeautifulSoup, url: str) -> str:
"""
Extract the main article content from a parsed HTML page.
Uses domain-specific selectors first, then generic article selectors,
then falls back to <body>.
"""
domain = urlparse(url).netloc.replace("www.", "")
for base_domain, selectors in DOMAIN_SELECTORS.items():
if base_domain in domain:
for sel in selectors:
el = soup.select_one(sel)
if el:
text = md(str(el), heading_style="ATX", strip=["img"])
if len(text.strip()) >= MIN_USEFUL_CHARS:
return text.strip()
for sel in ARTICLE_SELECTORS:
el = soup.select_one(sel)
if el:
text = md(str(el), heading_style="ATX", strip=["img"])
if len(text.strip()) >= MIN_USEFUL_CHARS:
return text.strip()
body = soup.find("body")
if body:
return md(str(body), heading_style="ATX", strip=["img"]).strip()
return md(str(soup), heading_style="ATX", strip=["img"]).strip()
def fetch_pdf(content_bytes: bytes) -> tuple[str, str]:
"""Extract text from PDF bytes. Returns (text, error)."""
try:
import pdfplumber
except ImportError:
return "", "PDF — install pdfplumber for PDF support"
try:
with pdfplumber.open(io.BytesIO(content_bytes)) as pdf:
pages = []
for page in pdf.pages:
text = page.extract_text()
if text:
pages.append(text)
full_text = "\n\n---\n\n".join(pages)
return full_text[:MAX_CONTENT_BYTES], ""
except Exception as e:
return "", f"PDF extraction failed: {str(e)[:100]}"
def fetch_url(url: str, stealth: bool = False, proxies: dict = None) -> tuple[str, str]:
"""
Fetch a URL and return (content_as_markdown, error_or_empty).
Converts HTML to markdown, extracts PDF text, handles plain text and JSON.
"""
try:
headers = build_headers(stealth)
resp = requests.get(
url,
timeout=TIMEOUT,
headers=headers,
allow_redirects=True,
proxies=proxies,
)
resp.raise_for_status()
content_type = resp.headers.get("content-type", "")
if "application/pdf" in content_type:
return fetch_pdf(resp.content)
if "text/html" in content_type or "application/xhtml" in content_type:
html = resp.text[:MAX_CONTENT_BYTES]
soup = BeautifulSoup(html, "html.parser")
for tag in soup(["script", "style", "nav", "footer", "header", "aside"]):
tag.decompose()
text = extract_article_content(soup, url)
if len(text) < MIN_USEFUL_CHARS:
return text, f"low-content ({len(text)} chars)"
return text, ""
if "text/plain" in content_type or "text/markdown" in content_type:
return resp.text[:MAX_CONTENT_BYTES].strip(), ""
if "application/json" in content_type:
return resp.text[:MAX_CONTENT_BYTES].strip(), ""
return "", f"Unsupported content-type: {content_type}"
except requests.exceptions.Timeout:
return "", "Timeout"
except requests.exceptions.ConnectionError:
return "", "Connection error"
except requests.exceptions.HTTPError as e:
return "", f"HTTP {e.response.status_code}"
except Exception as e:
return "", str(e)[:200]
def process_entry(entry: dict, sources_dir: Path, stealth: bool,
proxies: dict, domain_delay: float) -> dict:
"""Fetch a single entry and update its status. Thread-safe."""
url = entry.get("url", "").strip()
ref_id = entry.get("ref_id", entry.get("ref-id", "unknown"))
if not url or url == "-":
entry["status"] = "skipped"
entry["notes"] = "No URL"
return entry
domain_throttle(url, domain_delay)
filename = sanitize_filename(ref_id) + ".md"
filepath = sources_dir / filename
content, error = fetch_url(url, stealth=stealth, proxies=proxies)
if error and not content:
entry["status"] = "failed"
entry["notes"] = error
entry["file"] = "-"
elif content:
source_header = f"# Source: {ref_id}\n\n"
source_header += f"**URL:** {url}\n"
source_header += f"**Fetched:** {time.strftime('%Y-%m-%d %H:%M:%S')}\n\n"
source_header += "---\n\n"
with open(filepath, "w") as f:
f.write(source_header + content)
status = "fetched"
notes = f"{len(content)} chars"
if error:
notes += f" ({error})"
entry["status"] = status
entry["file"] = f"sources/{filename}"
entry["notes"] = notes
else:
entry["status"] = "failed"
entry["notes"] = error or "Empty response"
entry["file"] = "-"
return entry
def main():
parser = argparse.ArgumentParser(
description="Batch-fetch URLs from a research manifest"
)
parser.add_argument("research_dir", help="Path to research directory")
parser.add_argument(
"--retry-failed", action="store_true",
help="Re-attempt previously failed fetches"
)
parser.add_argument(
"--delay", type=float, default=2.0,
help="Minimum seconds between requests to the same domain (default: 2.0)"
)
parser.add_argument(
"--workers", type=int, default=4,
help="Number of concurrent fetch threads (default: 4)"
)
parser.add_argument(
"--stealth", action="store_true",
help="Use browser-mimicry headers (rotating UA, Accept, Sec-Fetch-*)"
)
parser.add_argument(
"--proxy",
help="HTTP or SOCKS5 proxy URL (e.g., socks5://127.0.0.1:1080, http://proxy:8080)"
)
args = parser.parse_args()
research_dir = Path(args.research_dir)
manifest_path = research_dir / "manifest.md"
sources_dir = research_dir / "sources"
if not manifest_path.exists():
print(f"Error: {manifest_path} not found")
sys.exit(1)
sources_dir.mkdir(exist_ok=True)
if args.proxy and "socks" in args.proxy.lower():
try:
import socks # noqa: F401
except ImportError:
print("SOCKS proxy requires PySocks. Install with:")
print(" pip install requests[socks]")
sys.exit(1)
proxies = None
if args.proxy:
proxies = {"http": args.proxy, "https": args.proxy}
print(f"Using proxy: {args.proxy}")
if args.stealth:
print("Stealth mode: rotating UA + browser-mimicry headers")
preamble = extract_preamble(manifest_path)
entries = parse_manifest(manifest_path)
if not entries:
print("No entries found in manifest table.")
sys.exit(1)
to_fetch = []
for entry in entries:
status = entry.get("status", "").lower()
if status == "pending":
to_fetch.append(entry)
elif status == "failed" and args.retry_failed:
to_fetch.append(entry)
total = len(entries)
fetch_count = len(to_fetch)
print(f"Found {total} total entries, {fetch_count} to fetch "
f"(workers={args.workers}, domain_delay={args.delay}s).")
if fetch_count == 0:
print("Nothing to fetch.")
return
fetched = 0
failed = 0
completed = 0
with ThreadPoolExecutor(max_workers=args.workers) as executor:
future_to_entry = {}
for entry in to_fetch:
future = executor.submit(
process_entry, entry, sources_dir,
args.stealth, proxies, args.delay
)
future_to_entry[future] = entry
for future in as_completed(future_to_entry):
entry = future_to_entry[future]
ref_id = entry.get("ref_id", entry.get("ref-id", "?"))
completed += 1
try:
result = future.result()
status = result.get("status", "failed")
notes = result.get("notes", "")
if status == "fetched":
fetched += 1
print(f" [{completed}/{fetch_count}] {ref_id}: OK — {notes}")
elif status == "skipped":
print(f" [{completed}/{fetch_count}] {ref_id}: SKIPPED — {notes}")
else:
failed += 1
print(f" [{completed}/{fetch_count}] {ref_id}: FAILED — {notes}")
except Exception as e:
failed += 1
entry["status"] = "failed"
entry["notes"] = f"Thread error: {str(e)[:100]}"
entry["file"] = "-"
print(f" [{completed}/{fetch_count}] {ref_id}: ERROR — {e}")
write_manifest(manifest_path, entries, preamble)
print(f"\nDone. Fetched: {fetched}, Failed: {failed}, Total: {total}")
print(f"Manifest updated: {manifest_path}")
if __name__ == "__main__":
main()
scripts/fetch-transcript.py
#!/usr/bin/env python3
"""
fetch-transcript.py — YouTube transcript fetcher for research and library workflows.
Fetches transcripts from YouTube videos and saves them as markdown files,
suitable for use in research workspaces or the personal reference library.
Usage:
python3 fetch-transcript.py <youtube-url> <output-path>
python3 fetch-transcript.py <youtube-url> <output-path> --lang en
python3 fetch-transcript.py --batch <urls-file> <output-dir>
Where:
<youtube-url> — A YouTube video URL (any format)
<output-path> — Path for the output .md file, or directory for batch mode
<urls-file> — Text file with one YouTube URL per line (batch mode)
Dependencies:
Required: youtube-transcript-api
Install: pip install youtube-transcript-api
"""
import sys
import re
import os
import time
import argparse
from pathlib import Path
try:
from youtube_transcript_api import YouTubeTranscriptApi
from youtube_transcript_api._errors import (
TranscriptsDisabled,
NoTranscriptFound,
VideoUnavailable,
)
except ImportError:
print("Missing dependency. Install with:")
print(" pip install youtube-transcript-api")
sys.exit(1)
try:
import requests
except ImportError:
requests = None
def extract_video_id(url: str) -> str | None:
"""Extract video ID from various YouTube URL formats."""
patterns = [
r"(?:v=|/v/|youtu\.be/)([a-zA-Z0-9_-]{11})",
r"^([a-zA-Z0-9_-]{11})$",
]
for pattern in patterns:
match = re.search(pattern, url)
if match:
return match.group(1)
return None
def fetch_video_metadata(video_id: str) -> dict:
"""Fetch basic video metadata via oembed (no API key needed)."""
if not requests:
return {"title": f"Video {video_id}", "author": "Unknown"}
try:
url = f"https://www.youtube.com/oembed?url=https://www.youtube.com/watch?v={video_id}&format=json"
resp = requests.get(url, timeout=10)
if resp.status_code == 200:
data = resp.json()
return {
"title": data.get("title", f"Video {video_id}"),
"author": data.get("author_name", "Unknown"),
}
except Exception:
pass
return {"title": f"Video {video_id}", "author": "Unknown"}
def fetch_transcript(video_id: str, lang: str = "en") -> tuple[list | None, str | None]:
"""Fetch transcript for a video. Returns (segments, error)."""
try:
ytt = YouTubeTranscriptApi()
transcript = ytt.fetch(video_id, languages=[lang])
segments = []
for snippet in transcript:
segments.append({
"text": snippet.text,
"start": snippet.start,
"duration": snippet.duration,
})
return segments, None
except TranscriptsDisabled:
return None, "Transcripts are disabled for this video"
except NoTranscriptFound:
try:
ytt = YouTubeTranscriptApi()
transcript = ytt.fetch(video_id)
segments = []
for snippet in transcript:
segments.append({
"text": snippet.text,
"start": snippet.start,
"duration": snippet.duration,
})
return segments, f"No '{lang}' transcript; used auto-detected language"
except Exception as e:
return None, f"No transcript found: {e}"
except VideoUnavailable:
return None, "Video is unavailable"
except Exception as e:
return None, f"Error: {e}"
def format_timestamp(seconds: float) -> str:
"""Convert seconds to HH:MM:SS or MM:SS format."""
h = int(seconds // 3600)
m = int((seconds % 3600) // 60)
s = int(seconds % 60)
if h > 0:
return f"{h}:{m:02d}:{s:02d}"
return f"{m}:{s:02d}"
def segments_to_markdown(segments: list, timestamps: bool = True) -> str:
"""Convert transcript segments to readable markdown text."""
if not timestamps:
return " ".join(seg["text"] for seg in segments)
lines = []
paragraph = []
last_break = 0
for seg in segments:
paragraph.append(seg["text"])
if seg["start"] - last_break >= 60 or seg["text"].rstrip().endswith((".", "?", "!")):
ts = format_timestamp(last_break)
text = " ".join(paragraph)
lines.append(f"**[{ts}]** {text}")
paragraph = []
last_break = seg["start"]
if paragraph:
ts = format_timestamp(last_break)
text = " ".join(paragraph)
lines.append(f"**[{ts}]** {text}")
return "\n\n".join(lines)
def save_transcript(video_id: str, metadata: dict, segments: list,
output_path: Path, note: str = None):
"""Save transcript as a markdown file."""
url = f"https://www.youtube.com/watch?v={video_id}"
title = metadata.get("title", f"Video {video_id}")
author = metadata.get("author", "Unknown")
duration_secs = 0
if segments:
last = segments[-1]
duration_secs = last["start"] + last["duration"]
duration_str = format_timestamp(duration_secs)
header = f"# Transcript: {title}\n\n"
header += f"- **Channel:** {author}\n"
header += f"- **URL:** {url}\n"
header += f"- **Duration:** {duration_str}\n"
header += f"- **Fetched:** {time.strftime('%Y-%m-%d %H:%M:%S')}\n"
header += f"- **Segments:** {len(segments)}\n"
if note:
header += f"- **Note:** {note}\n"
header += "\n---\n\n"
body = segments_to_markdown(segments)
plain = "\n\n---\n\n## Plain Text\n\n"
plain += " ".join(seg["text"] for seg in segments)
output_path.parent.mkdir(parents=True, exist_ok=True)
with open(output_path, "w") as f:
f.write(header + body + plain)
return len(segments)
def generate_filename(metadata: dict, video_id: str) -> str:
"""Generate a filesystem-safe filename from video metadata."""
title = metadata.get("title", video_id)
slug = re.sub(r"[^\w\s-]", "", title.lower())
slug = re.sub(r"[\s_]+", "-", slug).strip("-")
slug = slug[:80]
return f"{slug}.md"
def main():
parser = argparse.ArgumentParser(
description="Fetch YouTube transcripts and save as markdown"
)
parser.add_argument("url", nargs="?", help="YouTube video URL")
parser.add_argument("output", nargs="?", help="Output file path or directory")
parser.add_argument(
"--lang", default="en",
help="Preferred transcript language (default: en)"
)
parser.add_argument(
"--batch", metavar="FILE",
help="Batch mode: text file with one YouTube URL per line"
)
parser.add_argument(
"--no-timestamps", action="store_true",
help="Output plain text without timestamps"
)
args = parser.parse_args()
if args.batch:
batch_file = Path(args.batch)
if not batch_file.exists():
print(f"Error: {batch_file} not found")
sys.exit(1)
output_dir = Path(args.output) if args.output else Path(".")
output_dir.mkdir(parents=True, exist_ok=True)
urls = [line.strip() for line in batch_file.read_text().splitlines() if line.strip() and not line.startswith("#")]
print(f"Batch mode: {len(urls)} URLs")
success = 0
failed = 0
for url in urls:
video_id = extract_video_id(url)
if not video_id:
print(f" SKIP: {url} (can't extract video ID)")
failed += 1
continue
metadata = fetch_video_metadata(video_id)
segments, error = fetch_transcript(video_id, args.lang)
if not segments:
print(f" FAIL: {metadata['title']} — {error}")
failed += 1
continue
filename = generate_filename(metadata, video_id)
output_path = output_dir / filename
count = save_transcript(video_id, metadata, segments, output_path, note=error)
print(f" OK: {metadata['title']} ({count} segments) → {output_path}")
success += 1
print(f"\nDone: {success} fetched, {failed} failed")
else:
if not args.url:
parser.print_help()
sys.exit(1)
video_id = extract_video_id(args.url)
if not video_id:
print(f"Error: Can't extract video ID from '{args.url}'")
sys.exit(1)
print(f"Fetching metadata for {video_id}...")
metadata = fetch_video_metadata(video_id)
print(f" Title: {metadata['title']}")
print(f" Channel: {metadata['author']}")
print(f"Fetching transcript (lang={args.lang})...")
segments, error = fetch_transcript(video_id, args.lang)
if not segments:
print(f"Error: {error}")
sys.exit(1)
if error:
print(f" Note: {error}")
if args.output:
output_path = Path(args.output)
else:
filename = generate_filename(metadata, video_id)
output_path = Path(filename)
if output_path.is_dir():
filename = generate_filename(metadata, video_id)
output_path = output_path / filename
count = save_transcript(video_id, metadata, segments, output_path, note=error)
print(f"Saved {count} segments to {output_path}")
if __name__ == "__main__":
main()
SKILL.md
---
name: research-and-analyze
description: >-
YouTube transcript ingestion via fetch-transcript.py (captions to markdown under
research/), plus systematic research, citation verification, and evidence-based
analysis of articles, docs, and talks. Use when the user pastes youtube.com or
youtu.be URLs, asks to save or pull a video transcript, verify claims against
sources, evaluate a talk or video, run gather/manifest/analyze pipelines, or build
assessments from source material. For transcript-only + library stub routing,
youtube-transcript-library is the narrow entry skill; this skill holds the full
workflows and scripts.
---
<essential_principles>
### 1. Gather First, Analyze Second
Never try to fetch and analyze in the same pass. External fetches timeout, content is too large for inline processing, and conversation resets lose everything. Instead:
- **Gather phase**: Fetch all sources to disk. Produce a manifest of successes/failures.
- **Analyze phase**: Read files from disk in manageable batches. Write findings to disk incrementally.
- **Synthesize phase**: Compile per-source findings into an overall assessment.
### 2. Everything Goes to Disk
Source content, intermediate findings, and final assessment all live on the filesystem. This means:
- Work survives conversation resets
- Content can be re-analyzed without re-fetching
- Findings accumulate across sessions
- Large content doesn't overwhelm context windows
### 3. Work in Batches — Parallelize When Possible
Don't try to analyze 20 sources at once. Group into batches of 3-5 by topic:
- Each batch reads source files from disk and writes its own findings file
- **Batches are independent** — launch them as parallel Task agents (up to 4 at once)
- Each agent gets: the original article, its source files, specific claims to check, and the findings template
- Compare claims against what sources actually say
- Write findings for that batch to disk
- After all batches return, verify all expected output files exist
### 4. Track Provenance
Every claim should trace back to: where the article says it → what the cited source actually says → whether they match. Use the manifest to track which sources have been checked and which remain.
### 5. Workspace Convention
All research artifacts go in a `research/` directory within the project, organized by subject:
```
research/
├── {subject}/
│ ├── manifest.md # What was fetched, status, file mapping
│ ├── sources/ # Raw fetched content
│ │ ├── ref-01.md
│ │ ├── ref-02.md
│ │ └── ...
│ ├── findings/ # Per-batch analysis notes
│ │ ├── batch-01.md
│ │ ├── batch-02.md
│ │ └── ...
│ └── assessment.md # Final synthesized assessment
```
</essential_principles>
<intake>
**HARD STOP — present these options and wait. Do not proceed until the user responds.**
Even if the user has already provided a URL, article, or YouTube link, you do not know:
- Whether they want the full pipeline or a specific phase
- Whether they want the transcript-only path or citation verification
- What subject slug to use for the research directory
Present this menu and wait:
---
What would you like to do?
1. **Start new research** — article or document with citations to verify
2. **YouTube / single-source transcript** — evaluate a talk, video, or single source with no citations
3. **Continue gathering** — I have a manifest with unfetched sources
4. **Analyze sources** — sources are on disk, ready for claim verification
5. **Synthesize findings** — analysis is done, compile the assessment
6. **Full pipeline** — do everything: gather → analyze → synthesize
Also tell me: **what subject slug should I use for the research directory?** (e.g., `miessler-single-da-thesis`, `openai-gpt5-launch`) — this becomes `research/{slug}/`.
---
**Do not take any action until the user responds to the above.**
</intake>
<routing>
| Response | Workflow |
|----------|----------|
| YouTube URL pasted / “transcript” / “captions” / library ingest for a video | **`youtube-transcript-library`** skill first (`fetch-transcript.py` + library checklist); then this skill if they want claims analysis or full pipeline |
| 1, "new", "start", "verify", "article" | `workflows/gather-sources.md` |
| 2, "youtube", "transcript", "video", "talk", "single source" | `workflows/gather-sources.md` — transcript variant (Step 2T) |
| 3, "continue", "fetch", "retry" | `workflows/gather-sources.md` (resume mode) |
| 4, "analyze", "check", "verify claims" | `workflows/analyze-claims.md` |
| 5, "synthesize", "compile", "summarize", "assessment" | `workflows/synthesize-findings.md` |
| 6, "full", "everything", "pipeline", "all" | Run all three workflows in sequence, stopping at each phase-boundary checkpoint |
**After reading the workflow, follow it exactly. Each workflow has a phase-boundary checkpoint at the end — stop there and wait for explicit user confirmation before starting the next phase.**
</routing>
<reference_index>
All domain knowledge in `references/`:
**Patterns:** verification-patterns.md — common claim types and how to check them
**Fetcher:** fetcher-notes.md — architecture, anti-bot strategies, proxy/VPN usage, browser fallback roadmap
</reference_index>
<workflows_index>
| Workflow | Purpose |
|----------|---------|
| gather-sources.md | Extract URLs, batch-fetch to disk, produce manifest |
| analyze-claims.md | Read source files in batches, verify claims, write findings |
| synthesize-findings.md | Compile findings into overall confidence assessment |
</workflows_index>
<templates_index>
| Template | Purpose |
|----------|---------|
| manifest-template.md | Source tracking: URL, status, file path, notes |
| batch-findings-template.md | Per-batch analysis structure |
| assessment-template.md | Final assessment with confidence table |
</templates_index>
<scripts_index>
| Script | Purpose |
|----------|---------|
| fetch-sources.py | Concurrent batch URL fetcher with stealth headers, proxy support, PDF extraction, and domain-aware rate limiting |
| fetch-transcript.py | YouTube transcript fetcher — single video or batch mode, saves timestamped markdown to disk |
| (skill) `../youtube-transcript-library/SKILL.md` | YouTube-only ingest routing + library checklist — invokes this script |
</scripts_index>
templates/assessment-template.md
# Verification Assessment: {{Article Title}}
**Source:** {{article URL}}
**Assessment date:** {{date}}
**Sources checked:** {{N}} of {{total}} cited references
**Sources unreachable:** {{N}}
---
## Summary
{{3-5 sentence overview of what the analysis found. Is the article generally trustworthy? Where does it fall short? What should readers know?}}
---
## Confidence by Topic Area
| Topic area | Confidence | Basis |
| --- | --- | --- |
| {{area}} | **High** / **Medium-High** / **Medium** / **Medium-Low** / **Low** / **Unverifiable** | {{1-line explanation}} |
---
## Key Findings
**1. {{Finding title}}**
{{Evidence and explanation. Reference specific batch findings.}}
**2. {{Finding title}}**
{{Evidence and explanation.}}
**3. {{Finding title}}**
{{Evidence and explanation.}}
---
## What to Trust
{{List of topic areas / claims that are well-supported and can be cited confidently}}
## What to Verify Independently
{{Claims that are directionally correct but need additional context or caveats}}
## What to Discard or Caveat Heavily
{{Claims that are misleading, unsupported, or stripped of essential context}}
---
## Methodology
- **Tool:** research-and-analyze skill (fetch → analyze → synthesize pipeline)
- **Fetcher:** `fetch-sources.py` with `requests` + `beautifulsoup4` + `markdownify`
- **Analysis approach:** Batch comparison of article claims against source content
- **Sources checked:** {{N}} of {{total}} ({{percentage}}%)
- **Unreachable sources:** {{list with reasons}}
- **Limitations:** {{e.g., PDFs not parsed, paywalled content, non-English sources}}
templates/batch-findings-template.md
# Batch {{NN}} Findings: {{Topic Area}}
**Sources analyzed:** {{list of ref IDs}}
**Date:** {{date}}
---
## {{ref_id}}: {{source title or description}}
**Article claims:** {{what the original article says, citing this source}}
**Source actually says:** {{what the source content actually states}}
**Verdict:** {{VERIFIED | VERIFIED WITH CAVEATS | MISLEADING | UNSUPPORTED | UNVERIFIABLE}}
**Details:** {{explanation of match or mismatch, context omitted, numbers compared}}
**Impact:** {{how this affects the credibility of the article's argument in this area}}
---
## {{ref_id}}: {{next source}}
{{repeat pattern}}
---
## Batch Summary
- **Verified:** {{count}}
- **Verified with caveats:** {{count}}
- **Problematic:** {{count}}
- **Unverifiable:** {{count}}
- **Key pattern in this batch:** {{observation}}
templates/manifest-template.md
# Source Manifest
**Subject:** {{article title}}
**URL:** {{article URL}}
**Analysis started:** {{date}}
**Total references:** {{count}}
---
| ref_id | url | status | file | notes |
| --- | --- | --- | --- | --- |
| ref-01 | {{url}} | pending | - | {{what the article claims this supports}} |
| ref-02 | {{url}} | pending | - | {{what the article claims this supports}} |
workflows/analyze-claims.md
# Workflow: Analyze Claims
<required_reading>
**Read these before proceeding:**
1. templates/batch-findings-template.md
2. references/verification-patterns.md
3. The manifest at `research/{subject}/manifest.md` (to know what's available)
</required_reading>
<process>
## Step 1: Understand the Scope
Read the manifest to determine:
- How many sources were fetched successfully
- Which topic areas they cover
- Whether this is a standard article (claims map to external citations) or a transcript/single-source (claims are assertions within the source itself)
**For transcript/single-source analysis:** The manifest will list extracted claims (`C1`, `C2`, ...) rather than cited references. Each claim is evaluated for:
- Internal coherence and self-consistency
- Verifiability against external sources (for factual/architectural claims)
- Directional plausibility (for predictive claims)
- Workspace connections and tensions (for talks that relate to existing research)
If the original article is on disk (`sources/original-article.md`) or the transcript is at `sources/ref-01-transcript.md`, read it to confirm the claims list is complete before batching.
## Step 2: Plan Batches
**Standard path (article with citations):** Group sources into batches of 3-5 files by topic area. Prioritize:
1. **Economic/quantitative claims** — these are most likely to be misleading
2. **Architecture claims** — verifiable against official documentation
3. **Feature maturity claims** — check for GA vs. preview vs. alpha status
4. **Strategic/qualitative claims** — hardest to falsify, check last
**Transcript/single-source path:** Group the extracted claims into thematic batches of 5-8 claims. Suggested groupings:
1. **Factual/verifiable claims** — specific numbers, product states, dates
2. **Architectural claims** — how systems work (check against public repos/docs if available)
3. **Predictive/framework claims** — evaluate coherence and supporting evidence
4. **Workspace connections** — how claims connect to or tension with existing research, essays, library entries
Create a batch plan:
```
Batch 1: Economic claims (refs 1, 61, 62)
Batch 2: Serving architecture (refs 3, 8, 12)
Batch 3: ...
```
## Step 3: Process Batches in Parallel
**Batches are independent — launch them concurrently using Task agents.**
For each batch, launch a `generalPurpose` Task agent with this prompt structure:
```
You are analyzing sources for a research verification exercise.
Read source files from {research_dir}/sources/ and compare them
against the original article at {research_dir}/sources/original-article.md.
Sources to analyze: [list of ref files for this batch]
The article claims: [specific claims mapped to these sources]
For each source:
1. Read the source file
2. Find the specific claims the article maps to it
3. Compare article vs source — note omissions, context stripping, maturity levels
4. Assign verdict: VERIFIED | VERIFIED WITH CAVEATS | MISLEADING | UNSUPPORTED | UNVERIFIABLE
Write findings to: {research_dir}/findings/batch-{nn}-{topic}.md
Use the batch findings template structure.
```
**Parallelization rules:**
- Launch up to 4 batch agents simultaneously in a single message (multiple Task tool calls)
- Each agent reads/writes its own files — no conflicts
- Wait for all agents to return before proceeding
- If more than 4 batches, launch the first 4, wait, then launch the remainder
- After all batches complete, verify all expected findings files exist on disk
**Each agent needs these context elements in its prompt:**
- Path to the original article on disk
- Path to the source files it should read
- The specific claims from the article that map to those sources
- The batch findings template structure (inline or by reference)
- Instructions to check for maturity levels (GA vs Tech Preview vs alpha)
Key questions per source (include in each agent's prompt):
- Does the article accurately represent what the source says?
- Is important context omitted?
- Are numbers quoted correctly, including their conditions and caveats?
- Is the source authoritative for the claim being made?
- What maturity/readiness level does the source describe?
## Step 4: Handle Gaps
For claims whose sources couldn't be fetched:
- Can the claim be cross-checked against other fetched sources?
- Can a web search find the specific number or fact?
- If neither, mark as "unverifiable" with a note on what we tried
## Step 5: Incremental Progress
After each batch:
- Write findings to disk immediately (don't accumulate in memory)
- Update the manifest to note which refs have been analyzed
- Give the user a brief status update
If the conversation needs to reset, all progress is on disk and the next session can pick up where this one left off.
## Step 6: Phase-Boundary Checkpoint — HARD STOP
**Do not proceed to synthesis. Stop here and report to the user.**
Tell the user:
- Total claims analyzed: verified / verified-with-caveats / misleading / unsupported / unverifiable
- Which topic areas had the most issues
- Key pattern emerging across the batches (1-2 sentences)
- Confirm all findings files are on disk: list `research/{subject}/findings/` contents
Then ask:
> **Analysis phase complete.** Findings are at `research/{subject}/findings/`. Ready to synthesize into a final assessment?
> - Yes — proceed to `synthesize-findings.md`
> - No — describe what to revisit
**Wait for explicit confirmation before starting synthesis.**
</process>
<success_criteria>
This workflow is complete when:
- [ ] All sources/claims have been read and evaluated
- [ ] Findings for each batch are written to disk (verified by listing the files)
- [ ] Each finding includes: what is claimed, what the source says, verdict, impact
- [ ] Unverifiable claims are documented with explanation
- [ ] User has explicitly confirmed readiness to proceed to synthesis
</success_criteria>
workflows/gather-sources.md
# Workflow: Gather Sources
<required_reading>
**Read these before proceeding:**
1. templates/manifest-template.md
2. scripts/fetch-sources.py (understand the interface)
3. scripts/fetch-transcript.py (for YouTube/transcript variant)
</required_reading>
<process>
## Step 1: Identify the Subject and Create Workspace
Confirm with the user (this should already be answered from the intake):
- What is the subject slug? (e.g., `miessler-single-da-thesis`)
- Is this the standard article path or the transcript variant (Step 2T)?
**Directory naming rule: use a meaningful subject slug, never a URL fragment, video ID, or technical identifier.**
Create the directory structure:
```bash
mkdir -p research/{subject-slug}/sources
mkdir -p research/{subject-slug}/findings
```
## Step 2: Extract References from the Source Material
*(Standard path — article with citations. For YouTube/transcript, skip to Step 2T.)*
Read the article or document being analyzed. Extract every cited reference into a structured list:
- Reference ID (e.g., `ref-01`, `ref-02`, or the article's own numbering)
- URL (if present)
- What the article claims the reference supports
- Which section of the article cites it
If the article is a URL, fetch it first and save it as `research/{subject}/sources/original-article.md`.
## Step 2T: Transcript Variant — YouTube Video or Single Source
*(Use this path when the source is a YouTube video, a talk, or any single source with no external citations to verify.)*
**Fast path (Cursor / Pi):** When the user only needs captions + library ingest, the **`youtube-transcript-library`** skill (`../youtube-transcript-library/SKILL.md`) is the scoped entry point — same `fetch-transcript.py` command below, explicit anti–`yt-dlp`-only guidance, and the library checklist.
**Fetch the transcript:**
```bash
python3 .cursor/skills/research-and-analyze/scripts/fetch-transcript.py \
"{youtube-url}" \
research/{subject-slug}/sources/ref-01-transcript.md
```
The script fetches metadata (title, channel, duration) and saves a timestamped markdown file with both timestamped paragraphs and a plain-text section.
**Extract claims for analysis.** A talk has no external citations — the "references" are the assertions made in the talk itself. After fetching, read the transcript and build a claims list in the manifest:
- `C1`, `C2`, ... for each substantive claim
- What the speaker asserts
- Whether it is factual/verifiable, directional/predictive, or personal/framework
For talks, claims fall into these categories:
1. **Factual** — specific numbers, dates, product states (can be verified against external sources)
2. **Architectural** — descriptions of how a system works (verify against repo/docs if public)
3. **Predictive** — directional claims about where things are going (evaluate internal coherence and supporting evidence)
4. **Framework** — proprietary models or terminology (evaluate coherence, not external correctness)
5. **Relational** — connections to other people, companies, events (verify against public record)
Build the manifest with one row per claim. Mark each `pending`. The `analyze-claims.md` workflow handles claim-by-claim evaluation.
## Step 3: Build the Manifest
Copy `templates/manifest-template.md` to `research/{subject}/manifest.md`.
Fill in the preamble with:
- Article title and URL
- Date of analysis
- Total number of references
Fill in the table with one row per reference, all marked `pending`.
> **Column name warning:** The fetcher identifies entries by the `ref_id` column.
> Do **not** rename this column to `slug`, `id`, or anything else — the fetcher
> will silently fall back to `unknown` and all output files will collide as
> `unknown.md`. The template is correct; copy it as-is.
## Step 4: Install Dependencies (if needed)
Check if the Python dependencies are available:
```bash
python3 -c "import requests, bs4, markdownify" 2>/dev/null && echo "OK" || echo "MISSING"
```
If missing:
```bash
pip install requests beautifulsoup4 markdownify
```
## Step 5: Run the Fetcher
Execute the fetch script:
```bash
python3 .agents/skills/research-and-analyze/scripts/fetch-sources.py research/{subject}/ --stealth
```
**Available flags:**
- `--stealth` — Use browser-mimicry headers (rotating UA, Accept, Sec-Fetch-*). **Recommended for all runs.**
- `--workers N` — Concurrent fetch threads (default: 4). Fetches from different domains run in parallel; same-domain requests are rate-limited.
- `--delay N` — Minimum seconds between requests to the same domain (default: 2.0). Increase for sites that rate-limit aggressively.
- `--proxy URL` — Route through an HTTP or SOCKS5 proxy (e.g., `socks5://127.0.0.1:1080`). Useful when your IP is being rate-limited. Requires `pip install requests[socks]` for SOCKS proxies.
- `--retry-failed` — Re-attempt previously failed fetches.
**Typical invocations:**
```bash
# Standard run with stealth headers and 4 concurrent workers
python3 .cursor/skills/research-and-analyze/scripts/fetch-sources.py research/{subject}/ --stealth
# Retry failures through a VPN/proxy
python3 .cursor/skills/research-and-analyze/scripts/fetch-sources.py research/{subject}/ --retry-failed --stealth --proxy socks5://127.0.0.1:1080
# Conservative: single-threaded, longer delays
python3 .cursor/skills/research-and-analyze/scripts/fetch-sources.py research/{subject}/ --stealth --workers 1 --delay 3.0
```
## Step 5b: Rename Fetched Files to `.txt`
The fetcher saves content as `.md` files. Rename them to `.txt` before committing:
```bash
cd research/{subject}/sources/
for f in *.md; do mv "$f" "${f%.md}.txt"; done
```
This prevents `relative-link-guard` false positives on web-relative URLs
(e.g. `/en/products`, `some-page.html`) that appear in scraped content but
are not repository-relative paths. Update the manifest `file` column to
reflect the `.txt` extension if the fetcher has already written `.md` paths.
## Step 6: Review the Manifest
Read the updated manifest. Report to the user:
- How many sources were fetched successfully
- How many failed (and why — timeout, 404, PDF, etc.)
- How many were skipped
For failed sources, consider:
- **Timeout**: May work with a retry (`--retry-failed`)
- **404**: URL may have moved — try a web search for the title
- **PDF**: Note as "manual review needed" — user can download and convert separately
- **Connection error**: Site may be down — retry later
## Step 7: Retry and Fill Gaps
If there are failed fetches:
1. Run `--retry-failed` once for timeouts
2. For persistent failures, use WebSearch to find alternative URLs or mirrors
3. If an alternative is found, manually update the manifest URL and re-run
4. For truly unreachable sources, mark as `unreachable` and note in findings
## Step 8: Create Library Entry Stub — REQUIRED BEFORE CHECKPOINT
Before stopping, create a minimal library entry stub so no transcript is left orphaned:
1. Check whether a library entry already exists for this source:
```bash
ls library/ | grep -i "{subject-slug}"
```
2. If none exists, create `library/{subject-slug}.md` with at minimum:
- Metadata block (title, author, URL, type, tags, added date, wing)
- One-line "Why This Matters (personal)" placeholder: `> Stub — full entry pending analysis phase.`
3. Append to `library/log.md`:
```
## [YYYY-MM-DD] ingest | {Title}
- **Entry:** [{subject-slug}.md]({subject-slug}.md)
- **Wing:** {wing}
- **Source:** {type} / research/{subject-slug}/sources/
- **Note:** Transcript fetched; library entry stub created. Full enrichment pending analysis.
```
4. Add a row to `library/catalog.md` (or note that it already exists).
This step is non-optional. An orphaned transcript with no library entry is an incomplete ingest.
## Step 9: Phase-Boundary Checkpoint — HARD STOP
**Do not proceed to analysis. Stop here and report to the user.**
Tell the user:
- Research directory location and structure (show the file tree)
- Number of sources on disk vs. total
- For transcript variant: confirm the transcript was saved and show the segment count
- For standard path: fetch success/failure counts
- Confirm the library entry stub exists and is logged in `library/log.md`
Then ask:
> **Gather phase complete.** Files are on disk at `research/{subject-slug}/`. Library stub created at `library/{subject-slug}.md`. Ready to move to claim analysis?
> - Yes — proceed to `analyze-claims.md`
> - No — describe what to fix
**Wait for explicit confirmation before starting analysis.**
</process>
<success_criteria>
This workflow is complete when:
- [ ] Research directory exists with proper structure (`sources/`, `findings/`, `manifest.md`)
- [ ] Subject slug is meaningful (not a URL fragment or video ID)
- [ ] Manifest has one entry per reference or claim, all marked `pending`
- [ ] Fetch script has been run at least once and output confirmed
- [ ] Failed fetches have been retried or documented
- [ ] Library entry stub exists at `library/{subject-slug}.md`
- [ ] Entry logged in `library/log.md` with wing tag
- [ ] User has explicitly confirmed readiness to proceed to analysis
</success_criteria>
workflows/synthesize-findings.md
# Workflow: Synthesize Findings
<required_reading>
**Read these before proceeding:**
1. templates/assessment-template.md
2. All files in `research/{subject}/findings/`
3. The manifest at `research/{subject}/manifest.md`
</required_reading>
<process>
## Step 1: Read All Findings
Read every batch findings file from `research/{subject}/findings/`. Build a mental model of:
- Which claims were verified and at what confidence level
- Which claims had problems (misleading, stripped of context, unsupported)
- Which claims couldn't be verified and why
- What patterns emerged across the analysis
## Step 2: Categorize by Confidence
Sort all analyzed claims into tiers:
- **High confidence** — verified by authoritative sources (official docs, peer-reviewed papers)
- **Medium confidence** — verified directionally but with caveats (maturity omissions, context stripping)
- **Low confidence** — problematic (misleading numbers, unsupported by cited source)
- **Unverifiable** — source unreachable and no alternative confirmation found
## Step 3: Identify Patterns
Look for systemic patterns across findings:
- Are economic claims consistently overstated?
- Are maturity levels systematically omitted?
- Is the article advocacy, analysis, or documentation?
- Which topic areas are strongest/weakest?
- Are there signs of AI-generated content (consistent style, lack of nuance)?
## Step 4: Write the Assessment
Copy `templates/assessment-template.md` to `research/{subject}/assessment.md`.
Fill in:
- **Summary** — 3-5 sentence overview of what we found
- **Confidence table** — one row per topic area with confidence rating and basis
- **Key findings** — numbered findings with supporting evidence from batch analysis
- **Recommendations** — what a reader should trust, what they should verify independently, what they should discard
- **Methodology** — how many sources checked, what we couldn't reach, tools used
## Step 5: Cross-Reference Back
Review the assessment against the original article one more time:
- Is our assessment fair? Are we being too harsh or too generous?
- Did we miss any major claim areas?
- Are there claims we initially flagged that actually hold up when considering the full picture?
## Step 6: Deliver — Final Checkpoint
Write the assessment to disk first (`research/{subject}/assessment.md`), then present to the user:
- Confirm the file is written: show the path
- Brief verbal summary of key findings (3-5 sentences max — the file has the detail)
- Confidence table highlights: what scored high, what scored low, what's unverifiable
- Any recommended follow-up actions
Then explicitly close the pipeline:
> **Pipeline complete.** Full assessment is at `research/{subject}/assessment.md`.
>
> Suggested next steps (pick any):
> - Add a library entry (`/reference add`)
> - Cross-link into related essays
> - Add a new thread or backlog item based on the findings
> - No action needed — work is on disk for future reference
**Do not automatically proceed to any next step. Wait for the user to decide.**
</process>
<success_criteria>
This workflow is complete when:
- [ ] All batch findings have been read and synthesized
- [ ] Confidence table covers every major topic area
- [ ] Key findings are supported by specific evidence from batch files
- [ ] Assessment file is written to disk and path confirmed to user
- [ ] Patterns and systemic issues are documented
- [ ] User has been offered suggested next steps and given the choice
</success_criteria>