our-claude-skills/custom-skills/90-reference-curator/01-reference-discovery/code/scripts/discover.py

#!/usr/bin/env python3
"""Reference Discovery CLI — manage source discovery and manifests.

Claude performs the actual web search. This script handles:
- Receiving search results and creating manifests
- Deduplicating URLs against the existing repository
- Registering new sources in the database

Usage:
    python discover.py create-manifest --topic "prompt engineering" --output manifest.json < urls.json
    python discover.py dedup --manifest manifest.json [--output deduped.json]
    python discover.py register-source --name "Anthropic Docs" --type official_docs --url "https://docs.anthropic.com" --tier tier1_official --vendor anthropic
    python discover.py list-sources [--vendor anthropic]
"""

import json
import sys
from datetime import datetime
from pathlib import Path

import click
from rich.console import Console
from rich.table import Table

from refcurator.db import db_session
from refcurator.manifest import write_manifest, read_manifest, dedup_manifest_urls
from refcurator.models import Manifest, ManifestURL
from refcurator.utils import normalize_url

console = Console()


@click.group()
def cli():
    """Reference Discovery — manage source manifests."""
    pass


@cli.command("create-manifest")
@click.option("--topic", required=True, help="Discovery topic")
@click.option("--output", required=True, type=click.Path(), help="Output manifest path")
@click.option("--input", "input_file", type=click.Path(exists=True),
              help="Input JSON file with URL entries (or pipe via stdin)")
def create_manifest(topic, output, input_file):
    """Create a manifest from discovered URLs.

    Input format (JSON array):
    [
      {"url": "https://...", "title": "...", "credibility_score": 0.85, "source_type": "official_docs"},
      ...
    ]
    """
    if input_file:
        data = json.loads(Path(input_file).read_text())
    elif not sys.stdin.isatty():
        data = json.load(sys.stdin)
    else:
        console.print("[red]Error:[/red] Provide URLs via --input file or stdin")
        sys.exit(1)

    urls = [ManifestURL(**entry) for entry in data]

    # Normalize and deduplicate
    seen = {}
    for u in urls:
        norm = normalize_url(u.url)
        if norm not in seen or (u.credibility_score or 0) > (seen[norm].credibility_score or 0):
            seen[norm] = u

    unique_urls = list(seen.values())

    manifest = Manifest(
        discovery_date=datetime.now().isoformat(),
        topic=topic,
        total_urls=len(unique_urls),
        urls=unique_urls,
    )

    out_path = Path(output)
    write_manifest(manifest, out_path)
    console.print(f"[green]Created manifest:[/green] {len(unique_urls)} URLs → {output}")


@cli.command()
@click.option("--manifest", required=True, type=click.Path(exists=True), help="Manifest to dedup")
@click.option("--output", type=click.Path(), help="Output path (defaults to overwriting input)")
def dedup(manifest, output):
    """Deduplicate manifest URLs against existing repository."""
    m = read_manifest(Path(manifest))
    original_count = len(m.urls)

    with db_session() as db:
        existing = db.fetch_all("SELECT url FROM documents WHERE url IS NOT NULL")
        existing_urls = {r["url"] for r in existing}

    deduped = dedup_manifest_urls(m, existing_urls)
    removed = original_count - len(deduped.urls)

    out_path = Path(output) if output else Path(manifest)
    write_manifest(deduped, out_path)

    console.print(
        f"[green]Deduped:[/green] {original_count} → {len(deduped.urls)} URLs "
        f"({removed} duplicates removed)"
    )


@cli.command("register-source")
@click.option("--name", required=True, help="Source name")
@click.option("--type", "source_type", required=True,
              type=click.Choice(["official_docs", "engineering_blog", "research_paper",
                                 "github_repo", "community_guide", "pdf_document", "api_reference"]))
@click.option("--url", required=True, help="Base URL")
@click.option("--tier", default="tier3_community",
              type=click.Choice(["tier1_official", "tier2_verified", "tier3_community"]))
@click.option("--vendor", help="Vendor name (e.g., anthropic, openai)")
def register_source(name, source_type, url, tier, vendor):
    """Register a new source in the repository."""
    with db_session() as db:
        # Check for existing source with same base_url
        existing = db.fetch_one(
            "SELECT source_id, source_name FROM sources WHERE base_url = %s",
            (url,),
        )
        if existing:
            console.print(
                f"[yellow]Already registered:[/yellow] source_id={existing['source_id']} "
                f"({existing['source_name']})"
            )
            return

        source_id = db.insert_returning_id(
            """INSERT INTO sources (source_name, source_type, base_url, credibility_tier, vendor)
               VALUES (%s, %s, %s, %s, %s)""",
            (name, source_type, url, tier, vendor),
        )

    console.print(f"[green]Registered:[/green] source_id={source_id} — {name}")
    click.echo(json.dumps({"source_id": source_id, "name": name, "url": url}))


@cli.command("list-sources")
@click.option("--vendor", help="Filter by vendor")
@click.option("--tier", type=click.Choice(["tier1_official", "tier2_verified", "tier3_community"]),
              help="Filter by credibility tier")
def list_sources(vendor, tier):
    """List registered sources."""
    with db_session() as db:
        if vendor:
            rows = db.fetch_all(
                "SELECT * FROM sources WHERE vendor = %s ORDER BY credibility_tier",
                (vendor,),
            )
        elif tier:
            rows = db.fetch_all(
                "SELECT * FROM sources WHERE credibility_tier = %s ORDER BY vendor",
                (tier,),
            )
        else:
            rows = db.fetch_all("SELECT * FROM sources ORDER BY credibility_tier, vendor")

    table = Table(title="Registered Sources")
    table.add_column("ID", style="cyan")
    table.add_column("Name")
    table.add_column("Type")
    table.add_column("Tier")
    table.add_column("Vendor")
    table.add_column("URL")
    for r in rows:
        table.add_row(
            str(r.get("source_id", "")),
            str(r.get("source_name", "")),
            str(r.get("source_type", "")),
            str(r.get("credibility_tier", "")),
            str(r.get("vendor", "")),
            str(r.get("base_url", ""))[:40],
        )
    console.print(table)


if __name__ == "__main__":
    cli()