Pinecone:Assistant

#!/usr/bin/env python3
# /// script
# dependencies = [
#   "pinecone>=8.0.0",
#   "typer>=0.15.0",
#   "rich>=13.0.0",
# ]
# ///
"""
Chat with a Pinecone Assistant and receive cited responses.

Usage:
    uv run chat.py --assistant NAME --message "Your question" [--stream]

Environment Variables:
    PINECONE_API_KEY: Required Pinecone API key

Output:
    Assistant's response with citations to source documents
"""

import os
import typer
from rich.console import Console
from rich.panel import Panel
from rich.table import Table
from pinecone import Pinecone
from pinecone_plugins.assistant.models.chat import Message

app = typer.Typer()
console = Console()


@app.command()
def main(
    assistant: str = typer.Option(..., "--assistant", "-a", help="Name of the assistant to chat with"),
    message: str = typer.Option(..., "--message", "-m", help="Your question or message"),
):
    """Chat with a Pinecone Assistant and receive answers with source citations."""

    # Check for API key
    api_key = os.environ.get("PINECONE_API_KEY")
    if not api_key:
        console.print("[red]Error: PINECONE_API_KEY environment variable not set[/red]")
        console.print("\nGet your API key from: https://app.pinecone.io/?sessionType=signup")
        raise typer.Exit(1)

    try:
        # Initialize Pinecone client
        pc = Pinecone(api_key=api_key,source_tag="claude_code_plugin:assistant")
        asst = pc.assistant.Assistant(assistant_name=assistant)

        # Create message
        user_msg = Message(role="user", content=message)

        # Display user question
        console.print(Panel(f"[bold cyan]Question:[/bold cyan] {message}", border_style="cyan"))

        # Get response
        with console.status("[bold blue]Thinking...[/bold blue]"):
            response = asst.chat(messages=[user_msg], stream=False)

        answer_content = response.message.content
        citations = response.citations if hasattr(response, 'citations') else []
        usage = response.usage if hasattr(response, 'usage') else None

        # Display assistant's response (same for both modes)
        console.print("\n[bold green]Answer:[/bold green]\n")

        if answer_content:
            console.print(Panel(answer_content, border_style="green", title="Assistant Response"))
        else:
            console.print("[yellow]No response content received[/yellow]")

        # Display citations if available
        if citations and len(citations) > 0:

            console.print("\n[bold yellow]Citations:[/bold yellow]\n")

            citations_table = Table(show_header=True, header_style="bold yellow")
            citations_table.add_column("#", style="dim", width=4)
            citations_table.add_column("File", style="cyan", width=40)
            citations_table.add_column("Pages", style="blue", width=15)
            citations_table.add_column("Position", style="green", width=10)

            citation_num = 0
            for citation in citations:
                # Each citation has a list of references
                if hasattr(citation, 'references') and citation.references:
                    for reference in citation.references:
                        citation_num += 1

                        # Get file name
                        file_name = "Unknown"
                        if hasattr(reference, 'file') and hasattr(reference.file, 'name'):
                            file_name = reference.file.name

                        # Get pages
                        pages = []
                        if hasattr(reference, 'pages') and reference.pages:
                            pages = reference.pages

                        # Format pages
                        if pages:
                            pages_str = ", ".join(str(p) for p in pages)
                        else:
                            pages_str = "N/A"

                        # Get position from citation
                        position = getattr(citation, 'position', 'N/A')

                        citations_table.add_row(
                            str(citation_num),
                            file_name,
                            pages_str,
                            str(position)
                        )

            console.print(citations_table)

            # Optionally show download links
            console.print("\n[dim]Tip: File URLs are temporary signed links valid for ~1 hour[/dim]")

        # Display token usage
        if usage:
            usage_info = f"""[dim]Tokens used:[/dim]
• Prompt: {getattr(usage, 'prompt_tokens', 'N/A')}
• Completion: {getattr(usage, 'completion_tokens', 'N/A')}
• Total: {getattr(usage, 'total_tokens', 'N/A')}"""
            console.print(Panel(usage_info, border_style="dim", title="Usage Stats"))

        # Follow-up suggestion
        console.print(f"\n[dim]Continue the conversation with another message using the same command[/dim]")

    except Exception as e:
        console.print(f"[red]Error: {e}[/red]")
        raise typer.Exit(1)


if __name__ == "__main__":
    app()

#!/usr/bin/env python3
# /// script
# dependencies = [
#   "pinecone>=8.0.0",
#   "typer>=0.15.0",
#   "rich>=13.0.0",
# ]
# ///
"""
Retrieve context snippets from a Pinecone Assistant's knowledge base.

Usage:
    uv run context.py --assistant NAME --query "search text" [--top-k 5] [--json]

Environment Variables:
    PINECONE_API_KEY: Required Pinecone API key

Output:
    Relevant context snippets with file sources, page numbers, and relevance scores
"""

import os
import json as json_module
import typer
from rich.console import Console
from rich.panel import Panel
from rich.table import Table
from rich.text import Text
from pinecone import Pinecone

app = typer.Typer()
console = Console()


@app.command()
def main(
    assistant: str = typer.Option(..., "--assistant", "-a", help="Name of the assistant"),
    query: str = typer.Option(..., "--query", "-q", help="Search query text"),
    top_k: int = typer.Option(5, "--top-k", "-k", help="Number of results to return (max 16)"),
    snippet_size: int = typer.Option(1024, "--snippet-size", "-s", help="Maximum tokens per snippet"),
    json: bool = typer.Option(False, "--json", help="Output in JSON format"),
):
    """Retrieve relevant context snippets from an assistant's knowledge base."""

    # Check for API key
    api_key = os.environ.get("PINECONE_API_KEY")
    if not api_key:
        console.print("[red]Error: PINECONE_API_KEY environment variable not set[/red]")
        console.print("\nGet your API key from: https://app.pinecone.io/?sessionType=signup")
        raise typer.Exit(1)

    try:
        # Initialize Pinecone client
        pc = Pinecone(api_key=api_key, source_tag="claude_code_plugin:assistant")
        asst = pc.assistant.Assistant(assistant_name=assistant)

        # Display query
        if not json:
            console.print(Panel(f"[bold cyan]Query:[/bold cyan] {query}", border_style="cyan"))

        # Retrieve context
        with console.status("[bold blue]Searching knowledge base...[/bold blue]", spinner="dots"):
            response = asst.context(query=query, top_k=top_k, snippet_size=snippet_size)

        # Get snippets from response
        snippets = response.snippets if hasattr(response, 'snippets') else []

        if json:
            # JSON output
            results = []
            for snippet in snippets:
                file_name = "Unknown"
                pages = []
                if hasattr(snippet, 'reference') and snippet.reference:
                    ref = snippet.reference
                    if hasattr(ref, 'file') and hasattr(ref.file, 'name'):
                        file_name = ref.file.name
                    if hasattr(ref, 'pages') and ref.pages:
                        pages = ref.pages

                results.append({
                    "file_name": file_name,
                    "pages": pages,
                    "content": getattr(snippet, 'content', ''),
                    "score": getattr(snippet, 'score', 0.0),
                    "type": getattr(snippet, 'type', 'text'),
                })
            print(json_module.dumps({"snippets": results, "count": len(results)}, indent=2))
        else:
            # Rich formatted output
            if not snippets or len(snippets) == 0:
                console.print("[yellow]No context found for this query[/yellow]")
                return

            console.print(f"\n[bold]Found {len(snippets)} relevant snippet(s):[/bold]\n")

            for idx, snippet in enumerate(snippets, 1):
                # Extract file info from reference
                file_name = "Unknown"
                pages = []
                if hasattr(snippet, 'reference') and snippet.reference:
                    ref = snippet.reference
                    if hasattr(ref, 'file') and hasattr(ref.file, 'name'):
                        file_name = ref.file.name
                    if hasattr(ref, 'pages') and ref.pages:
                        pages = ref.pages

                score = getattr(snippet, 'score', 0.0)
                content = getattr(snippet, 'content', '')

                # Create header
                header = f"#{idx} - {file_name}"
                if pages:
                    pages_str = ", ".join(str(p) for p in pages)
                    header += f" (Page {pages_str})"
                header += f" - Score: {score:.3f}" if isinstance(score, (int, float)) else f" - Score: {score}"

                console.print(Panel(
                    content,
                    title=header,
                    border_style="blue",
                    subtitle=f"[dim]Relevance: {score:.2%}[/dim]" if isinstance(score, (int, float)) else None
                ))
                console.print()

            # Suggest next action
            next_action = f"""[bold]Next steps:[/bold]
\u2022 Ask a question: [cyan]/pinecone:assistant[/cyan] \u2014 "ask {assistant} about [your question]"
\u2022 Upload more files: [cyan]/pinecone:assistant[/cyan] \u2014 "upload files from [path] to {assistant}\""""
            console.print(Panel(next_action, title="What's Next?", border_style="green"))

    except AttributeError as e:
        # Handle case where context method doesn't exist or response structure is different
        console.print(f"[red]Error: Context retrieval failed[/red]")
        console.print(f"[dim]Details: {e}[/dim]")
        console.print("\n[yellow]Note:[/yellow] Context API requires SDK version with assistant.context() support")
        console.print("\n[yellow]Try using chat instead:[/yellow]")
        console.print(f"  /pinecone:assistant — \"ask {assistant} about \\\"{query}\\\"\"")
        raise typer.Exit(1)
    except Exception as e:
        console.print(f"[red]Error: {e}[/red]")
        raise typer.Exit(1)


if __name__ == "__main__":
    app()

#!/usr/bin/env python3
# /// script
# dependencies = [
#   "pinecone>=8.0.0",
#   "typer>=0.15.0",
#   "rich>=13.0.0",
# ]
# ///
"""
Create a new Pinecone Assistant.

Usage:
    uv run create.py --name ASSISTANT_NAME [--instructions TEXT] [--region us|eu] [--timeout SECONDS]

Environment Variables:
    PINECONE_API_KEY: Required Pinecone API key

Output:
    Success message with assistant details including host URL for MCP configuration
"""

import os
import typer
from rich.console import Console
from rich.panel import Panel
from rich.table import Table
from pinecone import Pinecone

app = typer.Typer()
console = Console()


@app.command()
def main(
    name: str = typer.Option(..., "--name", "-n", help="Unique name for the assistant"),
    instructions: str = typer.Option(
        "",
        "--instructions",
        "-i",
        help="Instructions for assistant behavior (max 16KB)",
    ),
    region: str = typer.Option(
        "us",
        "--region",
        "-r",
        help="Deployment region: 'us' or 'eu'",
    ),
    timeout: int = typer.Option(
        30,
        "--timeout",
        "-t",
        help="Seconds to wait for ready status",
    ),
):
    """Create a new Pinecone Assistant for document Q&A with citations."""

    # Validate region
    if region not in ["us", "eu"]:
        console.print("[red]Error: Region must be 'us' or 'eu'[/red]")
        raise typer.Exit(1)

    # Check for API key
    api_key = os.environ.get("PINECONE_API_KEY")
    if not api_key:
        console.print("[red]Error: PINECONE_API_KEY environment variable not set[/red]")
        console.print("\nGet your API key from: https://app.pinecone.io/?sessionType=signup")
        raise typer.Exit(1)

    try:
        # Initialize Pinecone client
        with console.status(f"[bold blue]Creating assistant '{name}'...[/bold blue]"):
            pc = Pinecone(api_key=api_key, source_tag="claude_code_plugin:assistant")

            # Create assistant
            assistant = pc.assistant.create_assistant(
                assistant_name=name,
                instructions=instructions if instructions else None,
                region=region,
                timeout=timeout,
                metadata={"agentic-ide-source":"claude-code-plugin"}
            )

        # Success message
        console.print(f"\n[bold green]✓ Assistant '{name}' created successfully![/bold green]\n")

        # Display assistant details in a table
        table = Table(show_header=False, box=None)
        table.add_column("Property", style="cyan")
        table.add_column("Value", style="white")

        table.add_row("Name", assistant.name)
        table.add_row("Region", region)
        table.add_row("Status", f"[yellow]{assistant.status}[/yellow]")
        table.add_row("Host", getattr(assistant, "host", "N/A"))
        if instructions:
            instructions_preview = instructions[:80] + "..." if len(instructions) > 80 else instructions
            table.add_row("Instructions", instructions_preview)

        console.print(table)

        # MCP configuration info
        host = getattr(assistant, "host", "")
        if host:
            mcp_info = f"""[bold]MCP Endpoint:[/bold]
{host}/mcp/assistants/{name}

[bold]Set environment variable:[/bold]
export PINECONE_ASSISTANT_HOST="{host}"
"""
            console.print(Panel(mcp_info, title="MCP Configuration", border_style="blue"))

        # Next steps
        next_steps = f"""[bold]Next steps:[/bold]
1. Upload files: [cyan]/pinecone:assistant[/cyan] \u2014 upload files from [path] to {name}
2. Chat: [cyan]/pinecone:assistant[/cyan] \u2014 ask {name} about [your question]
3. Get context: [cyan]/pinecone:assistant[/cyan] \u2014 search {name} for context about [topic]"""

        console.print(Panel(next_steps, title="What's Next?", border_style="green"))

    except Exception as e:
        console.print(f"[red]Error creating assistant: {e}[/red]")
        raise typer.Exit(1)


if __name__ == "__main__":
    app()

#!/usr/bin/env python3
# /// script
# dependencies = [
#   "pinecone>=8.0.0",
#   "typer>=0.15.0",
#   "rich>=13.0.0",
# ]
# ///
"""
List all Pinecone Assistants in the account.

Usage:
    uv run list.py [--json] [--files]

Environment Variables:
    PINECONE_API_KEY: Required Pinecone API key

Output:
    Formatted table or JSON list of assistants with name, region, status, and host
    Optionally include files for each assistant with --files flag
"""

import os
import sys
import json
import typer
from rich.console import Console
from rich.table import Table
from rich.panel import Panel
from pinecone import Pinecone

app = typer.Typer()
console = Console()


@app.command()
def main(
    json_output: bool = typer.Option(False, "--json", help="Output in JSON format"),
    files: bool = typer.Option(False, "--files", "-f", help="Include file listing for each assistant"),
):
    """List all Pinecone Assistants in your account."""

    # Check for API key
    api_key = os.environ.get('PINECONE_API_KEY')
    if not api_key:
        console.print("[red]Error: PINECONE_API_KEY environment variable not set[/red]")
        console.print("\nGet your API key from: https://app.pinecone.io/?sessionType=signup")
        raise typer.Exit(1)

    try:
        # Initialize Pinecone client
        pc = Pinecone(api_key=api_key, source_tag="claude_code_plugin:assistant")

        # List assistants
        assistants = pc.assistant.list_assistants()

        if not assistants:
            if json_output:
                print(json.dumps({"assistants": [], "count": 0}))
            else:
                console.print("[yellow]No assistants found.[/yellow]\n")
                console.print("Create your first assistant with:")
                console.print("  [cyan]/pinecone:assistant[/cyan] — \"create a new assistant called [name]\"")
            return

        if json_output:
            # JSON output
            assistants_data = []
            for asst in assistants:
                asst_data = {
                    "name": asst.name,
                    "region": getattr(asst, 'region', 'unknown'),
                    "status": asst.status,
                    "host": getattr(asst, 'host', ''),
                }

                if files:
                    # Get files for this assistant
                    try:
                        assistant_instance = pc.assistant.Assistant(assistant_name=asst.name)
                        file_list = assistant_instance.list_files()
                        asst_data["files"] = [
                            {
                                "name": f.name,
                                "id": f.id,
                                "status": f.status,
                                "metadata": getattr(f, 'metadata', {}),
                            }
                            for f in file_list
                        ]
                        asst_data["file_count"] = len(file_list)
                    except Exception as e:
                        asst_data["files"] = []
                        asst_data["file_count"] = 0
                        asst_data["file_error"] = str(e)

                assistants_data.append(asst_data)

            result = {
                "assistants": assistants_data,
                "count": len(assistants)
            }
            print(json.dumps(result, indent=2))
        else:
            # Rich table output
            console.print(f"\n[bold]Found {len(assistants)} assistant(s):[/bold]\n")

            # Assistants table
            table = Table(show_header=True, header_style="bold cyan")
            table.add_column("Name", style="green", width=30)
            table.add_column("Region", style="blue", width=10)
            table.add_column("Status", style="yellow", width=15)
            if files:
                table.add_column("Files", style="magenta", width=10)
            table.add_column("Host", style="dim", width=40 if files else 50)

            for asst in assistants:
                name = asst.name
                region = getattr(asst, 'region', 'unknown')
                status = asst.status
                host = getattr(asst, 'host', '')

                # Color code status
                if status == 'ready':
                    status_display = f"[green]{status}[/green]"
                elif status == 'indexing':
                    status_display = f"[yellow]{status}[/yellow]"
                else:
                    status_display = status

                if files:
                    # Get file count for this assistant
                    try:
                        assistant_instance = pc.assistant.Assistant(assistant_name=asst.name)
                        file_list = assistant_instance.list_files()
                        file_count = str(len(file_list))
                    except Exception:
                        file_count = "?"

                    table.add_row(name, region, status_display, file_count, host)
                else:
                    table.add_row(name, region, status_display, host)

            console.print(table)
            console.print()

            # If --files flag is set, show detailed file listing for each assistant
            if files:
                console.print("[bold]File Details:[/bold]\n")
                for asst in assistants:
                    try:
                        assistant_instance = pc.assistant.Assistant(assistant_name=asst.name)
                        file_list = assistant_instance.list_files()

                        if file_list:
                            # Create a table for this assistant's files
                            file_table = Table(show_header=True, header_style="bold blue", title=f"[cyan]{asst.name}[/cyan]")
                            file_table.add_column("#", style="dim", width=4)
                            file_table.add_column("File Name", style="green", width=50)
                            file_table.add_column("Status", style="yellow", width=15)
                            file_table.add_column("ID", style="dim", width=30)

                            for idx, file_obj in enumerate(file_list, 1):
                                file_name = file_obj.name
                                file_id = file_obj.id
                                file_status = file_obj.status

                                # Color code file status
                                if file_status == 'available':
                                    file_status_display = f"[green]{file_status}[/green]"
                                elif file_status == 'processing':
                                    file_status_display = f"[yellow]{file_status}[/yellow]"
                                else:
                                    file_status_display = file_status

                                file_table.add_row(str(idx), file_name, file_status_display, file_id)

                            console.print(file_table)
                            console.print()
                        else:
                            console.print(f"[dim]{asst.name}: No files uploaded[/dim]\n")
                    except Exception as e:
                        console.print(f"[red]Error listing files for {asst.name}: {e}[/red]\n")

            # Next steps panel
            next_steps = """[bold]Next steps:[/bold]
\u2022 List with files: [cyan]/pinecone:assistant[/cyan] \u2014 list my assistants with their files
\u2022 Chat: [cyan]/pinecone:assistant[/cyan] \u2014 ask [name] about [your question]
\u2022 Upload: [cyan]/pinecone:assistant[/cyan] \u2014 upload files from [path] to [name]
\u2022 Context: [cyan]/pinecone:assistant[/cyan] \u2014 search [name] for context about [topic]"""

            console.print(Panel(next_steps, title="Available Commands", border_style="blue"))

    except Exception as e:
        console.print(f"[red]Error listing assistants: {e}[/red]")
        raise typer.Exit(1)


if __name__ == "__main__":
    app()

#!/usr/bin/env python3
# /// script
# dependencies = [
#   "pinecone>=8.0.0",
#   "typer>=0.15.0",
#   "rich>=13.0.0",
# ]
# ///
"""
Sync local files to a Pinecone Assistant, only uploading new or changed files.

Usage:
    uv run sync.py --assistant NAME --source PATH [--delete-missing] [--dry-run]

Environment Variables:
    PINECONE_API_KEY: Required Pinecone API key

Output:
    Shows files to add, update, and optionally delete, with confirmation prompt
"""

import os
import hashlib
from pathlib import Path
from datetime import datetime, timezone
import typer
from rich.console import Console
from rich.panel import Panel
from rich.table import Table
from rich.progress import Progress, SpinnerColumn, TextColumn
from pinecone import Pinecone

app = typer.Typer()
console = Console()

# Supported file extensions
SUPPORTED_EXTENSIONS = {'.md', '.txt', '.pdf', '.docx', '.json'}

# Directories to exclude
EXCLUDE_DIRS = {'node_modules', '.venv', '.git', 'build', 'dist', '__pycache__', '.pytest_cache'}


def should_exclude_path(path: Path, source_root: Path) -> bool:
    """Check if path should be excluded based on directory patterns."""
    try:
        rel_path = path.relative_to(source_root)
        for part in rel_path.parts:
            if part in EXCLUDE_DIRS or part.startswith('.'):
                return True
    except ValueError:
        return True
    return False


def find_files(source_path: Path) -> list[Path]:
    """Find all supported files in source directory, excluding common build/dependency dirs."""
    files = []

    if source_path.is_file():
        if source_path.suffix.lower() in SUPPORTED_EXTENSIONS:
            return [source_path]
        else:
            return []

    for file_path in source_path.rglob('*'):
        if file_path.is_file():
            if file_path.suffix.lower() in SUPPORTED_EXTENSIONS:
                if not should_exclude_path(file_path, source_path):
                    files.append(file_path)

    return sorted(files)


def get_file_info(file_path: Path):
    """Get file modification time and size."""
    stat = file_path.stat()
    return {
        'mtime': stat.st_mtime,
        'size': stat.st_size,
    }


def file_changed(local_info: dict, remote_metadata: dict) -> bool:
    """Check if local file differs from remote using mtime and size."""
    remote_mtime = remote_metadata.get('mtime')
    remote_size = remote_metadata.get('size')

    if remote_mtime is None or remote_size is None:
        # No stored metadata, assume changed
        return True

    return (local_info['mtime'] != float(remote_mtime) or
            local_info['size'] != int(remote_size))


@app.command()
def main(
    assistant: str = typer.Option(..., "--assistant", "-a", help="Name of the assistant"),
    source: str = typer.Option(..., "--source", "-s", help="Local file or directory path"),
    delete_missing: bool = typer.Option(False, "--delete-missing", help="Delete files from assistant that don't exist locally"),
    dry_run: bool = typer.Option(False, "--dry-run", help="Show what would change without making changes"),
    yes: bool = typer.Option(False, "--yes", "-y", help="Skip confirmation prompt"),
):
    """Sync local files to Pinecone Assistant, only uploading new or changed files."""

    # Check for API key
    api_key = os.environ.get("PINECONE_API_KEY")
    if not api_key:
        console.print("[red]Error: PINECONE_API_KEY environment variable not set[/red]")
        console.print("\nGet your API key from: https://app.pinecone.io/?sessionType=signup")
        raise typer.Exit(1)

    # Validate source path
    source_path = Path(source).resolve()
    if not source_path.exists():
        console.print(f"[red]Error: Source path does not exist: {source}[/red]")
        raise typer.Exit(1)

    try:
        # Initialize Pinecone client
        pc = Pinecone(api_key=api_key, source_tag="claude_code_plugin:assistant")
        asst = pc.assistant.Assistant(assistant_name=assistant)

        console.print(Panel(
            f"[bold cyan]Assistant:[/bold cyan] {assistant}\n"
            f"[bold cyan]Source:[/bold cyan] {source_path}",
            title="Sync Configuration",
            border_style="cyan"
        ))

        # Step 1: Get current files in assistant
        with console.status("[bold blue]Fetching assistant files...[/bold blue]", spinner="dots"):
            remote_files = asst.list_files()

        # Build map of file_path -> file object
        remote_file_map = {}
        for f in remote_files:
            metadata = getattr(f, 'metadata', {}) or {}
            file_path = metadata.get('file_path', f.name)
            remote_file_map[file_path] = {
                'file_obj': f,
                'metadata': metadata
            }

        console.print(f"[dim]Found {len(remote_files)} file(s) in assistant[/dim]\n")

        # Step 2: Find local files
        with console.status("[bold blue]Scanning local files...[/bold blue]", spinner="dots"):
            local_files = find_files(source_path)

        if not local_files:
            console.print("[yellow]No supported files found in source path[/yellow]")
            console.print(f"Supported extensions: {', '.join(sorted(SUPPORTED_EXTENSIONS))}")
            raise typer.Exit(0)

        console.print(f"[dim]Found {len(local_files)} local file(s)[/dim]\n")

        # Step 3: Determine what needs syncing
        to_upload = []  # New files
        to_update = []  # Changed files (delete + re-upload)
        to_delete = []  # Files in assistant but not local
        unchanged = []  # Files that match

        # Track which remote files we've seen
        seen_remote_paths = set()

        for local_file in local_files:
            # Get relative path from source root
            if source_path.is_file():
                rel_path = local_file.name
            else:
                rel_path = str(local_file.relative_to(source_path))

            local_info = get_file_info(local_file)

            if rel_path in remote_file_map:
                # File exists remotely, check if changed
                seen_remote_paths.add(rel_path)
                remote_info = remote_file_map[rel_path]

                if file_changed(local_info, remote_info['metadata']):
                    to_update.append({
                        'local_path': local_file,
                        'rel_path': rel_path,
                        'remote_file_id': remote_info['file_obj'].id,
                        'local_info': local_info
                    })
                else:
                    unchanged.append(rel_path)
            else:
                # New file
                to_upload.append({
                    'local_path': local_file,
                    'rel_path': rel_path,
                    'local_info': local_info
                })

        # Find files to delete (in remote but not local)
        if delete_missing:
            for rel_path, remote_info in remote_file_map.items():
                if rel_path not in seen_remote_paths:
                    to_delete.append({
                        'rel_path': rel_path,
                        'remote_file_id': remote_info['file_obj'].id
                    })

        # Step 4: Show summary
        console.print("[bold]Sync Summary:[/bold]\n")

        summary_table = Table(show_header=True, header_style="bold cyan")
        summary_table.add_column("Action", style="yellow", width=15)
        summary_table.add_column("Count", style="green", width=10)

        summary_table.add_row("New files", str(len(to_upload)))
        summary_table.add_row("Updated files", str(len(to_update)))
        if delete_missing:
            summary_table.add_row("Deleted files", str(len(to_delete)))
        summary_table.add_row("Unchanged", str(len(unchanged)))

        console.print(summary_table)
        console.print()

        # Show details if there are changes
        if to_upload:
            console.print("[bold green]Files to upload:[/bold green]")
            for item in to_upload[:10]:  # Show first 10
                console.print(f"  + {item['rel_path']}")
            if len(to_upload) > 10:
                console.print(f"  ... and {len(to_upload) - 10} more")
            console.print()

        if to_update:
            console.print("[bold yellow]Files to update:[/bold yellow]")
            for item in to_update[:10]:
                console.print(f"  ~ {item['rel_path']}")
            if len(to_update) > 10:
                console.print(f"  ... and {len(to_update) - 10} more")
            console.print()

        if to_delete:
            console.print("[bold red]Files to delete:[/bold red]")
            for item in to_delete[:10]:
                console.print(f"  - {item['rel_path']}")
            if len(to_delete) > 10:
                console.print(f"  ... and {len(to_delete) - 10} more")
            console.print()

        # If no changes, exit early
        if not (to_upload or to_update or to_delete):
            console.print("[green]✓ All files are up to date![/green]")
            return

        # Dry run mode
        if dry_run:
            console.print("[yellow]Dry run mode: No changes made[/yellow]")
            return

        # Confirmation prompt
        if not yes:
            proceed = typer.confirm("\nProceed with sync?")
            if not proceed:
                console.print("[yellow]Sync cancelled[/yellow]")
                return

        console.print()

        # Step 5: Execute sync
        uploaded_count = 0
        updated_count = 0
        deleted_count = 0

        with Progress(
            SpinnerColumn(),
            TextColumn("[progress.description]{task.description}"),
            console=console
        ) as progress:

            # Upload new files
            if to_upload:
                task = progress.add_task(f"Uploading {len(to_upload)} new file(s)...", total=len(to_upload))
                for item in to_upload:
                    try:
                        asst.upload_file(
                            file_path=str(item['local_path']),
                            metadata={
                                'file_path': item['rel_path'],
                                'mtime': item['local_info']['mtime'],
                                'size': item['local_info']['size'],
                                'uploaded_at': datetime.now(timezone.utc).isoformat(),
                                'source': 'sync_script',
                            },
                            timeout=None
                        )
                        uploaded_count += 1
                        progress.advance(task)
                    except Exception as e:
                        console.print(f"[red]Failed to upload {item['rel_path']}: {e}[/red]")

            # Update changed files (delete old + upload new)
            if to_update:
                task = progress.add_task(f"Updating {len(to_update)} file(s)...", total=len(to_update) * 2)
                for item in to_update:
                    try:
                        # Delete old version
                        asst.delete_file(file_id=item['remote_file_id'])
                        progress.advance(task)

                        # Upload new version
                        asst.upload_file(
                            file_path=str(item['local_path']),
                            metadata={
                                'file_path': item['rel_path'],
                                'mtime': item['local_info']['mtime'],
                                'size': item['local_info']['size'],
                                'uploaded_at': datetime.now(timezone.utc).isoformat(),
                                'source': 'sync_script',
                            },
                            timeout=None
                        )
                        updated_count += 1
                        progress.advance(task)
                    except Exception as e:
                        console.print(f"[red]Failed to update {item['rel_path']}: {e}[/red]")

            # Delete missing files
            if to_delete:
                task = progress.add_task(f"Deleting {len(to_delete)} file(s)...", total=len(to_delete))
                for item in to_delete:
                    try:
                        asst.delete_file(file_id=item['remote_file_id'])
                        deleted_count += 1
                        progress.advance(task)
                    except Exception as e:
                        console.print(f"[red]Failed to delete {item['rel_path']}: {e}[/red]")

        # Final summary
        console.print()
        console.print(Panel(
            f"[green]✓ Sync complete![/green]\n\n"
            f"Uploaded: {uploaded_count}\n"
            f"Updated: {updated_count}\n"
            + (f"Deleted: {deleted_count}\n" if delete_missing else "") +
            f"Unchanged: {len(unchanged)}",
            title="Results",
            border_style="green"
        ))

    except Exception as e:
        console.print(f"[red]Error: {e}[/red]")
        raise typer.Exit(1)


if __name__ == "__main__":
    app()

#!/usr/bin/env python3
# /// script
# dependencies = [
#   "pinecone>=8.0.0",
#   "typer>=0.15.0",
#   "rich>=13.0.0",
# ]
# ///
"""
Upload files or repository contents to a Pinecone Assistant.

IMPORTANT: Only uploads DOCUMENTATION and DATA files.
Supported: DOCX (.docx), JSON (.json), Markdown (.md), PDF (.pdf), Text (.txt)
Code files are NOT supported by Pinecone Assistant.

Usage:
    uv run upload.py --assistant NAME --source PATH [--patterns "*.md,*.pdf,*.docx"]

Environment Variables:
    PINECONE_API_KEY: Required Pinecone API key

Output:
    Progress updates and summary of uploaded files
"""

import os
import glob
from pathlib import Path
from typing import List
from datetime import datetime, timezone
import typer
from rich.console import Console
from rich.progress import Progress, SpinnerColumn, TextColumn, BarColumn, TaskProgressColumn
from rich.table import Table
from rich.panel import Panel
from pinecone import Pinecone

app = typer.Typer()
console = Console()

# Default file patterns - DOCUMENTATION ONLY
# Assistant supports: DOCX, JSON, Markdown, PDF, Text
DEFAULT_PATTERNS = ["**/*.md", "**/*.txt", "**/*.pdf", "**/*.docx", "**/*.json"]

# Default directories to exclude
DEFAULT_EXCLUDES = ["node_modules", ".venv", "venv", ".git", "build", "dist", "__pycache__", ".next", ".cache"]


def find_files(source_path: str, patterns: List[str], excludes: List[str]) -> List[Path]:
    """Find files matching patterns, excluding certain directories."""
    source = Path(source_path)

    if not source.exists():
        console.print(f"[red]Error: Path '{source_path}' does not exist[/red]")
        raise typer.Exit(1)

    # If it's a single file, return it
    if source.is_file():
        return [source]

    # Otherwise, scan directory
    files = []
    for pattern in patterns:
        matched = glob.glob(str(source / pattern), recursive=True)
        files.extend([Path(f) for f in matched])

    # Filter out excluded directories
    filtered_files = []
    for file_path in files:
        # Check if any exclude pattern is in the path
        if not any(excl in str(file_path) for excl in excludes):
            filtered_files.append(file_path)

    return sorted(set(filtered_files))


@app.command()
def main(
    assistant: str = typer.Option(..., "--assistant", "-a", help="Name of the assistant to upload to"),
    source: str = typer.Option(..., "--source", "-s", help="File or directory path to upload"),
    patterns: str = typer.Option(
        ",".join(DEFAULT_PATTERNS),
        "--patterns",
        "-p",
        help="Comma-separated glob patterns for documentation files (e.g., '*.md,*.pdf')",
    ),
    exclude: str = typer.Option(
        ",".join(DEFAULT_EXCLUDES),
        "--exclude",
        "-e",
        help="Comma-separated directories to exclude",
    ),
    metadata_json: str = typer.Option(
        "",
        "--metadata",
        "-m",
        help="Additional metadata as JSON string",
    ),
):
    """Upload documentation files to a Pinecone Assistant.

    NOTE: Only documentation files (markdown, text, PDF) are supported.
    Code files are not recommended for Pinecone Assistant.
    """

    # Check for API key
    api_key = os.environ.get("PINECONE_API_KEY")
    if not api_key:
        console.print("[red]Error: PINECONE_API_KEY environment variable not set[/red]")
        console.print("\nGet your API key from: https://app.pinecone.io/?sessionType=signup")
        raise typer.Exit(1)

    # Parse patterns and excludes
    pattern_list = [p.strip() for p in patterns.split(",")]
    exclude_list = [e.strip() for e in exclude.split(",")]

    # Parse additional metadata if provided
    extra_metadata = {}
    if metadata_json:
        import json
        try:
            extra_metadata = json.loads(metadata_json)
        except json.JSONDecodeError:
            console.print("[red]Error: Invalid JSON in --metadata parameter[/red]")
            raise typer.Exit(1)

    try:
        # Initialize Pinecone client
        pc = Pinecone(api_key=api_key, source_tag="claude_code_plugin:assistant")
        asst = pc.assistant.Assistant(assistant_name=assistant)

        # Find files to upload
        console.print(f"\n[bold]Scanning for documentation files in:[/bold] {source}")
        console.print(f"[dim]Patterns: {', '.join(pattern_list)}[/dim]\n")

        files = find_files(source, pattern_list, exclude_list)

        if not files:
            console.print("[yellow]No documentation files found matching the specified patterns[/yellow]")
            console.print("\n[dim]Tip: Pinecone Assistant works with .md, .txt, and .pdf files[/dim]")
            return

        console.print(f"[green]Found {len(files)} documentation file(s) to upload[/green]\n")

        # Upload files with progress bar
        uploaded = 0
        failed = 0
        failed_files = []

        with Progress(
            SpinnerColumn(),
            TextColumn("[progress.description]{task.description}"),
            BarColumn(),
            TaskProgressColumn(),
            console=console,
        ) as progress:
            task = progress.add_task("[cyan]Uploading files...", total=len(files))

            for file_path in files:
                try:
                    # Build metadata
                    rel_path = os.path.relpath(str(file_path), source)
                    stat = file_path.stat()
                    metadata = {
                        "source": "upload_script",
                        "file_path": rel_path,
                        "file_type": file_path.suffix,
                        "content_type": "documentation",
                        "mtime": stat.st_mtime,
                        "size": stat.st_size,
                        "uploaded_at": datetime.now(timezone.utc).isoformat(),
                        **extra_metadata,
                    }

                    # Upload file
                    asst.upload_file(
                        file_path=str(file_path),
                        metadata=metadata,
                        timeout=None,
                    )
                    uploaded += 1
                    progress.update(task, advance=1, description=f"[cyan]Uploaded: {rel_path}")

                except Exception as e:
                    failed += 1
                    failed_files.append((str(file_path), str(e)))
                    progress.update(task, advance=1)

        # Summary table
        console.print()
        summary = Table(show_header=False, box=None)
        summary.add_column("Status", style="bold")
        summary.add_column("Count")

        summary.add_row("[green]✓ Uploaded[/green]", str(uploaded))
        if failed > 0:
            summary.add_row("[red]✗ Failed[/red]", str(failed))

        console.print(Panel(summary, title="Upload Summary", border_style="blue"))

        # Show failed files if any
        if failed_files:
            console.print("\n[bold red]Failed uploads:[/bold red]")
            for file_path, error in failed_files:
                console.print(f"  • {file_path}: [red]{error}[/red]")

        # Next steps
        if uploaded > 0:
            next_steps = f"""[bold]Next steps:[/bold]
• Chat: [cyan]/pinecone:assistant[/cyan] — "ask {assistant} about [your question]"
• Context: [cyan]/pinecone:assistant[/cyan] — "search {assistant} for context about [topic]"

[dim]Note: Files are being processed and will be available shortly[/dim]"""
            console.print(Panel(next_steps, title="What's Next?", border_style="green"))

    except Exception as e:
        console.print(f"[red]Error: {e}[/red]")
        raise typer.Exit(1)


if __name__ == "__main__":
    app()