Load data from a file, compute embeddings, and store in the database. The file can have one of multiple formats: 1. Two columns per row, separated by tabs: URL and JSON 2. One column: JSON only (URL will be extracted from the JSON) 3. CSV file with headers 4. RSS/Atom feed
(file_path: str, site: str, batch_size: int = 100, delete_existing: bool = False, force_recompute: bool = False, database: str | None = None)
| 677 | pass |
| 678 | |
| 679 | async def loadJsonToDB(file_path: str, site: str, batch_size: int = 100, delete_existing: bool = False, force_recompute: bool = False, database: str | None = None): |
| 680 | """ |
| 681 | Load data from a file, compute embeddings, and store in the database. |
| 682 | |
| 683 | The file can have one of multiple formats: |
| 684 | 1. Two columns per row, separated by tabs: URL and JSON |
| 685 | 2. One column: JSON only (URL will be extracted from the JSON) |
| 686 | 3. CSV file with headers |
| 687 | 4. RSS/Atom feed |
| 688 | 5. URL pointing to any of the above types |
| 689 | |
| 690 | Args: |
| 691 | file_path: Path to the input file or URL |
| 692 | site: Site identifier |
| 693 | batch_size: Number of documents to process and upload in each batch |
| 694 | delete_existing: Whether to delete existing entries for this site before loading |
| 695 | force_recompute: Whether to force recomputation of embeddings |
| 696 | database: Specific database endpoint to use (if None, uses preferred endpoint) |
| 697 | """ |
| 698 | # Check if this is a URL |
| 699 | is_url_path = await is_url(file_path) |
| 700 | temp_path = None |
| 701 | |
| 702 | if is_url_path: |
| 703 | temp_path, _ = await save_url_content(file_path) |
| 704 | original_path = file_path # Keep original URL for reference |
| 705 | file_path = temp_path |
| 706 | else: |
| 707 | original_path = file_path # Use original path for non-URLs |
| 708 | |
| 709 | try: |
| 710 | # First, check if the file exists at the given path |
| 711 | if os.path.exists(file_path): |
| 712 | resolved_path = file_path # Use directly if it exists |
| 713 | else: |
| 714 | # If not, try to resolve it from the JSON data folder |
| 715 | resolved_path = os.path.join(CONFIG.nlweb.json_data_folder, file_path) |
| 716 | if not os.path.exists(resolved_path): |
| 717 | # If still not found, check if it's an absolute path |
| 718 | if os.path.isabs(file_path): |
| 719 | resolved_path = file_path |
| 720 | else: |
| 721 | # Last attempt: just use the provided path (might not exist) |
| 722 | resolved_path = file_path |
| 723 | |
| 724 | # Use specified database or fall back to preferred endpoint |
| 725 | endpoint_name = database or CONFIG.write_endpoint |
| 726 | |
| 727 | print(f"Loading data from {original_path} (resolved to {resolved_path}) for site {site} using database endpoint '{endpoint_name}'") |
| 728 | |
| 729 | # Detect file type |
| 730 | file_type, has_embeddings = await detect_file_type(resolved_path) |
| 731 | print(f"Detected file type: {file_type}") |
| 732 | |
| 733 | # If embeddings are detected, switch to loadJsonWithEmbeddingsToDB |
| 734 | if has_embeddings and not force_recompute: |
| 735 | print("File already contains embeddings, switching to direct loading mode...") |
| 736 | return await loadJsonWithEmbeddingsToDB(resolved_path, site, batch_size, delete_existing, endpoint_name) |
no test coverage detected