Setting the file. One moment.
Upload · Pinecone Assistant · pinecone-io/skills · Skills Docs
Back to the top of the page scripts/ upload.py
Python · 222 lines · 7 KB
17 Usage:
18 uv run upload.py --assistant NAME --source PATH [--patterns "*.md,*.pdf,*.docx"]
19
20 Environment Variables:
21 PINECONE_API_KEY: Required Pinecone API key
22
23 Output:
24 Progress updates and summary of uploaded files
25 """
26
27 import os
28 import glob
29 from pathlib import Path
30 from typing import List
31 from datetime import datetime, timezone
32 import typer
33 from rich.console import Console
34 from rich.progress import Progress, SpinnerColumn, TextColumn, BarColumn, TaskProgressColumn
35 from rich.table import Table
36 from rich.panel import Panel
37 from pinecone import Pinecone
38
39 app = typer.Typer()
40 console = Console()
41
42 # Default file patterns - DOCUMENTATION ONLY
43 # Assistant supports: DOCX, JSON, Markdown, PDF, Text
44 DEFAULT_PATTERNS = [ "**/*.md" , "**/*.txt" , "**/*.pdf" , "**/*.docx" , "**/*.json" ]
45
46 # Default directories to exclude
47 DEFAULT_EXCLUDES = [ "node_modules" , ".venv" , "venv" , ".git" , "build" , "dist" , "__pycache__" , ".next" , ".cache" ]
48
49
50 def find_files (source_path: str , patterns: List[ str ], excludes: List[ str ]) -> List[Path]:
51 """Find files matching patterns, excluding certain directories."""
52 source = Path(source_path)
53
54 if not source.exists():
55 console.print( f "[red]Error: Path ' { source_path } ' does not exist[/red]" )
56 raise typer.Exit( 1 )
57
58 # If it's a single file, return it
59 if source.is_file():
60 return [source]
61
62 # Otherwise, scan directory
63 files = []
64 for pattern in patterns:
65 matched = glob.glob( str (source / pattern), recursive = True )
66 files.extend([Path(f) for f in matched])
67
68 # Filter out excluded directories
69 filtered_files = []
70 for file_path in files:
71 # Check if any exclude pattern is in the path
72 if not any (excl in str (file_path) for excl in excludes):
73 filtered_files.append(file_path)
74
75 return sorted ( set (filtered_files))
76
77
78 @app.command ()
79 def main (
80 assistant: str = typer.Option( ... , "--assistant" , "-a" , help = "Name of the assistant to upload to" ),
81 source: str = typer.Option( ... , "--source" , "-s" , help = "File or directory path to upload" ),
82 patterns: str = typer.Option(
83 "," .join( DEFAULT_PATTERNS ),
84 "--patterns" ,
85 "-p" ,
86 help = "Comma-separated glob patterns for documentation files (e.g., '*.md,*.pdf')" ,
87 ),
88 exclude: str = typer.Option(
89 "," .join( DEFAULT_EXCLUDES ),
90 "--exclude" ,
91 "-e" ,
92 help = "Comma-separated directories to exclude" ,
93 ),
94 metadata_json: str = typer.Option(
95 "" ,
96 "--metadata" ,
97 "-m" ,
98 help = "Additional metadata as JSON string" ,
99 ),
100 ):
101 """Upload documentation files to a Pinecone Assistant.
102
103 NOTE : Only documentation files (markdown, text, PDF) are supported.
104 Code files are not recommended for Pinecone Assistant.
105 """
106
107 # Check for API key
108 api_key = os.environ.get( "PINECONE_API_KEY" )
109 if not api_key:
110 console.print( "[red]Error: PINECONE_API_KEY environment variable not set[/red]" )
111 console.print( " \n Get your API key from: https://app.pinecone.io/?sessionType=signup" )
112 raise typer.Exit( 1 )
113
114 # Parse patterns and excludes
115 pattern_list = [p.strip() for p in patterns.split( "," )]
116 exclude_list = [e.strip() for e in exclude.split( "," )]
117
118 # Parse additional metadata if provided
119 extra_metadata = {}
120 if metadata_json:
121 import json
122 try :
123 extra_metadata = json.loads(metadata_json)
124 except json.JSONDecodeError:
125 console.print( "[red]Error: Invalid JSON in --metadata parameter[/red]" )
126 raise typer.Exit( 1 )
127
128 try :
129 # Initialize Pinecone client
130 pc = Pinecone( api_key = api_key, source_tag = "pinecone_skills:assistant" )
131
132 # Find files to upload
133 console.print( f " \n [bold]Scanning for documentation files in:[/bold] { source } " )
134 console.print( f "[dim]Patterns: { ', ' .join(pattern_list) } [/dim] \n " )
135
136 files = find_files(source, pattern_list, exclude_list)
137
138 if not files:
139 console.print( "[yellow]No documentation files found matching the specified patterns[/yellow]" )
140 console.print( " \n [dim]Tip: Pinecone Assistant works with .md, .txt, and .pdf files[/dim]" )
141 return
142
143 console.print( f "[green]Found { len (files) } documentation file(s) to upload[/green] \n " )
144
145 # Upload files with progress bar
146 uploaded = 0
147 failed = 0
148 failed_files = []
149
150 with Progress(
151 SpinnerColumn(),
152 TextColumn( "[progress.description] {task.description} " ),
153 BarColumn(),
154 TaskProgressColumn(),
155 console = console,
156 ) as progress:
157 task = progress.add_task( "[cyan]Uploading files..." , total = len (files))
158
159 for file_path in files:
160 try :
161 # Build metadata
162 rel_path = os.path.relpath( str (file_path), source)
163 stat = file_path.stat()
164 metadata = {
165 "source" : "upload_script" ,
166 "file_path" : rel_path,
167 "file_type" : file_path.suffix,
168 "content_type" : "documentation" ,
169 "mtime" : stat.st_mtime,
170 "size" : stat.st_size,
171 "uploaded_at" : datetime.now(timezone.utc).isoformat(),
172 ** extra_metadata,
173 }
174
175 # Upload file
176 pc.assistants.upload_file(
177 assistant_name = assistant,
178 file_path = str (file_path),
179 metadata = metadata,
180 timeout = None ,
181 )
182 uploaded += 1
183 progress.update(task, advance = 1 , description = f "[cyan]Uploaded: { rel_path } " )
184
185 except Exception as e:
186 failed += 1
187 failed_files.append(( str (file_path), str (e)))
188 progress.update(task, advance = 1 )
189
190 # Summary table
191 console.print()
192 summary = Table( show_header = False , box = None )
193 summary.add_column( "Status" , style = "bold" )
194 summary.add_column( "Count" )
195
196 summary.add_row( "[green]✓ Uploaded[/green]" , str (uploaded))
197 if failed > 0 :
198 summary.add_row( "[red]✗ Failed[/red]" , str (failed))
199
200 console.print(Panel(summary, title = "Upload Summary" , border_style = "blue" ))
201
202 # Show failed files if any
203 if failed_files:
204 console.print( " \n [bold red]Failed uploads:[/bold red]" )
205 for file_path, error in failed_files:
206 console.print( f " • { file_path } : [red] { error } [/red]" )
207
208 # Next steps
209 if uploaded > 0 :
210 next_steps = f """[bold]Next steps:[/bold]
211 <<next_after_upload>>
212
213 [dim]Note: Files are being processed and will be available shortly[/dim]"""
214 console.print(Panel(next_steps, title = "What's Next?" , border_style = "green" ))
215
216 except Exception as e:
217 console.print( f "[red]Error: { e } [/red]" )
218 raise typer.Exit( 1 )
219
220
221 if __name__ == "__main__" :
222 app()