Julian Carlson
40 lines
1.4 KiB
Python
40 lines
1.4 KiB
Python
#!/usr/bin/env python3
|
|
"""Parallel ingestion of a directory."""
|
|
from concurrent.futures import ThreadPoolExecutor, as_completed
|
|
from pathlib import Path
|
|
import sys
|
|
sys.path.insert(0, '/home/julianocarlson/kb')
|
|
|
|
from app.ingest import ingest_file
|
|
from app.parsers import PARSERS
|
|
|
|
def main(root_path, max_workers=4):
|
|
root = Path(root_path)
|
|
files = [p for p in root.rglob("*")
|
|
if p.is_file() and p.suffix.lower() in PARSERS]
|
|
|
|
print(f"Found {len(files)} files to process")
|
|
|
|
counts = {"indexed": 0, "unchanged": 0, "error": 0, "skipped": 0, "empty": 0}
|
|
|
|
with ThreadPoolExecutor(max_workers=max_workers) as executor:
|
|
futures = {executor.submit(ingest_file, f): f for f in files}
|
|
for i, future in enumerate(as_completed(futures), 1):
|
|
try:
|
|
result = future.result()
|
|
status = result.get("status", "unknown")
|
|
counts[status] = counts.get(status, 0) + 1
|
|
print(f"[{i}/{len(files)}] {status}: {Path(result['path']).name}")
|
|
except Exception as e:
|
|
counts["error"] += 1
|
|
print(f"[{i}/{len(files)}] EXCEPTION: {e}")
|
|
|
|
print("\nSummary:")
|
|
for k, v in counts.items():
|
|
print(f" {k}: {v}")
|
|
|
|
if __name__ == "__main__":
|
|
path = sys.argv[1] if len(sys.argv) > 1 else "/mnt/nas/transcripts"
|
|
workers = int(sys.argv[2]) if len(sys.argv) > 2 else 4
|
|
main(path, workers)
|