wehub-resource-sync
1437 lines
56 KiB
Python
Executable File
1437 lines
56 KiB
Python
Executable File
# Copyright 2023-2026 llmware
|
|
|
|
# Licensed under the Apache License, Version 2.0 (the "License"); you
|
|
# may not use this file except in compliance with the License. You
|
|
# may obtain a copy of the License at
|
|
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS,
|
|
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
|
|
# implied. See the License for the specific language governing
|
|
# permissions and limitations under the License.
|
|
|
|
"""The library module implements the logic for managing unstructured information (the text).
|
|
|
|
The module implements the two classes Library and LibraryCatalog. Library is responsible for organizing a
|
|
collection of text and is the interface for the Parser and Embedding classes. In addition, the Library object
|
|
is passed to the Query and Prompt objects. The Library class uses the LibraryCatalog for creating, deleting,
|
|
updating, and other tasks pertaining to Libraries via the Library Card.
|
|
"""
|
|
|
|
import shutil
|
|
import os
|
|
import json
|
|
import logging
|
|
|
|
from llmware.configs import LLMWareConfig, LLMWareTableSchema, LLMWareException
|
|
from llmware.util import Utilities
|
|
from llmware.parsers import Parser
|
|
from llmware.models import ModelCatalog
|
|
from llmware.resources import CollectionRetrieval, CollectionWriter, CloudBucketManager
|
|
from llmware.embeddings import EmbeddingHandler
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class Library:
|
|
|
|
"""Implements the interface to manage a collection of unstructured information as a ``Library``, i.e. a
|
|
library is an indexed collection of texts, tables and images extracted from parsed files.
|
|
|
|
Returns
|
|
-------
|
|
library : Library
|
|
A new ``Library`` object.
|
|
"""
|
|
|
|
def __init__(self):
|
|
|
|
# default settings for basic parameters
|
|
self.account_name = None
|
|
self.library_name = None
|
|
|
|
# base file paths in each library
|
|
self.library_main_path = None
|
|
|
|
# each of these paths hang off library_main_path
|
|
self.file_copy_path = None
|
|
self.image_path = None
|
|
self.dataset_path = None
|
|
self.nlp_path = None
|
|
self.output_path = None
|
|
self.tmp_path = None
|
|
self.embedding_path = None
|
|
|
|
# default key structure of block -> re-order for nicer display
|
|
self.default_keys = ["block_ID", "doc_ID", "content_type", "file_type","master_index","master_index2",
|
|
"coords_x", "coords_y", "coords_cx", "coords_cy", "author_or_speaker", "modified_date",
|
|
"created_date", "creator_tool", "added_to_collection", "file_source",
|
|
# changing to 'table_block' and 'text_block'
|
|
"table_block", "external_files", "text_block", "header_text", "text_search",
|
|
"user_tags", "special_field1", "special_field2", "special_field3","graph_status","dialog",
|
|
"embedding_flags"]
|
|
|
|
self.library_block_schema = LLMWareTableSchema().get_block_schema()
|
|
|
|
# default library card elements
|
|
self.default_library_card = ["library_name", "embedding_status", "embedding_model", "embedding_db",
|
|
"embedded_blocks", "embedding_dims", "time_stamp",
|
|
"knowledge_graph", "unique_doc_id", "documents", "blocks", "images", "pages",
|
|
"tables"]
|
|
|
|
self.block_size_target_characters = 400
|
|
|
|
# attributes used in parsing workflow
|
|
self.doc_ID = 0
|
|
self.block_ID = 0
|
|
|
|
# check for llmware path & create if not already set up
|
|
if not os.path.exists(LLMWareConfig.get_llmware_path()):
|
|
|
|
# if not explicitly set up by user, then create folder directory structure
|
|
LLMWareConfig.setup_llmware_workspace()
|
|
|
|
# add checks for /tmp and /accounts in place
|
|
library_path = LLMWareConfig.get_library_path()
|
|
tmp_path = LLMWareConfig.get_tmp_path()
|
|
|
|
if not os.path.exists(library_path):
|
|
os.mkdir(library_path)
|
|
os.chmod(library_path, 0o777)
|
|
|
|
if not os.path.exists(tmp_path):
|
|
os.mkdir(tmp_path)
|
|
os.chmod(tmp_path, 0o777)
|
|
|
|
# explicit constructor to create a new library
|
|
def create_new_library(self, library_name, account_name="llmware"):
|
|
"""Explicit constructor to create a new library with selected name.
|
|
|
|
If a library with the same name already exists, it will load the existing library.
|
|
|
|
Checks if library_name is safe. If not, it will change library_name to a safe name.
|
|
|
|
Parameters
|
|
----------
|
|
library_name : str
|
|
name of the library to create
|
|
|
|
account_name : str, default="llmware"
|
|
name of the account associated with the library
|
|
|
|
Returns
|
|
-------
|
|
library : Library
|
|
A new ``Library`` object representing the newly created or loaded existing library
|
|
"""
|
|
|
|
# note: default behavior - if library with same name already exists, then it loads existing library
|
|
|
|
self.library_name = library_name
|
|
self.account_name = account_name
|
|
|
|
# apply safety check to library_name path
|
|
library_name = Utilities().secure_filename(library_name)
|
|
|
|
library_exists = self.check_if_library_exists(library_name,account_name)
|
|
|
|
if library_exists:
|
|
|
|
# do not create
|
|
logger.info(f"Library - create_new_library - library already exists - returning library - {library_name} - {account_name}")
|
|
|
|
return self.load_library(library_name, account_name)
|
|
|
|
# assign self.library_name to the 'safe' library_name
|
|
self.library_name = library_name
|
|
|
|
# allow 'dynamic' creation of a new account path
|
|
account_path = os.path.join(LLMWareConfig.get_library_path(), account_name)
|
|
if not os.path.exists(account_path):
|
|
os.makedirs(account_path,exist_ok=True)
|
|
|
|
# safety check for name based on db
|
|
safe_name = CollectionRetrieval(library_name,account_name=self.account_name).safe_name(library_name)
|
|
|
|
if safe_name != library_name:
|
|
|
|
logger.warning(f"Library - create_new_library - selected library name is being changed for safety on selected resource - "
|
|
f"{safe_name}")
|
|
|
|
if isinstance(safe_name,str):
|
|
library_name = safe_name
|
|
self.library_name = safe_name
|
|
|
|
else:
|
|
raise LLMWareException(message=f"Library - create_new_library - selected name is not "
|
|
f"valid library name - {library_name}")
|
|
|
|
self.library_main_path = os.path.join(LLMWareConfig.get_library_path(), account_name, library_name)
|
|
|
|
# add new file dir for this collection
|
|
self.file_copy_path = os.path.join(self.library_main_path,"uploads" + os.sep )
|
|
self.image_path = os.path.join(self.library_main_path, "images" + os.sep)
|
|
self.dataset_path = os.path.join(self.library_main_path, "datasets" + os.sep)
|
|
self.nlp_path = os.path.join(self.library_main_path, "nlp" + os.sep)
|
|
self.output_path = os.path.join(self.library_main_path, "output" + os.sep)
|
|
self.tmp_path = os.path.join(self.library_main_path, "tmp" + os.sep)
|
|
self.embedding_path = os.path.join(self.library_main_path, "embedding" + os.sep)
|
|
|
|
library_folder = os.path.exists(self.library_main_path)
|
|
|
|
# this is a new library to create -> build file paths for work products
|
|
if not library_folder:
|
|
os.mkdir(self.library_main_path)
|
|
os.mkdir(self.file_copy_path)
|
|
os.mkdir(self.image_path)
|
|
os.mkdir(self.dataset_path)
|
|
os.mkdir(self.nlp_path)
|
|
os.mkdir(self.output_path)
|
|
os.mkdir(self.tmp_path)
|
|
os.mkdir(self.embedding_path)
|
|
os.chmod(self.dataset_path, 0o777)
|
|
os.chmod(self.nlp_path, 0o777)
|
|
os.chmod(self.output_path, 0o777)
|
|
os.chmod(self.tmp_path, 0o777)
|
|
os.chmod(self.embedding_path, 0o777)
|
|
|
|
new_library_entry = {"library_name": self.library_name,
|
|
|
|
# track embedding status - each embedding tracked as new dict in list
|
|
# --by default, when library created, no embedding in place
|
|
|
|
"embedding": [{"embedding_status": "no", "embedding_model": "none", "embedding_db": "none",
|
|
"embedded_blocks":0, "embedding_dims":0, "time_stamp": "NA"}],
|
|
|
|
# knowledge graph
|
|
# deprecated in 0.5.0 -> will be removed in upcoming releases
|
|
"knowledge_graph": "no",
|
|
|
|
# doc trackers
|
|
"unique_doc_id": 0, "documents": 0, "blocks": 0, "images": 0, "pages": 0, "tables": 0,
|
|
|
|
# options to create and set different accounts
|
|
"account_name": self.account_name
|
|
}
|
|
|
|
# LibraryCatalog will register the new library card
|
|
new_library_card = LibraryCatalog(self).create_new_library_card(new_library_entry)
|
|
|
|
if CollectionWriter(self.library_name,account_name=self.account_name).check_if_table_build_required():
|
|
|
|
CollectionWriter(self.library_name,account_name=self.account_name).create_table(self.library_name,
|
|
self.library_block_schema)
|
|
|
|
# update collection text index in collection after adding documents
|
|
CollectionWriter(self.library_name,account_name=self.account_name).build_text_index()
|
|
|
|
return self
|
|
|
|
def load_library(self, library_name, account_name="llmware"):
|
|
|
|
"""Load an existing library by invoking the library string name.
|
|
|
|
Parameters
|
|
----------
|
|
library_name : str
|
|
Name of the library to load
|
|
|
|
account_name : str, default="llmware"
|
|
Name of the account associated with the library
|
|
|
|
Returns
|
|
-------
|
|
library : Library
|
|
A new ``Library`` object representing the loaded library
|
|
"""
|
|
|
|
# first check that library exists
|
|
library_exists = self.check_if_library_exists(library_name, account_name=account_name)
|
|
|
|
if not library_exists:
|
|
raise LLMWareException(message=f"Library - load_library - library not found -"
|
|
f"library_name = {library_name} with account_name = {account_name}")
|
|
|
|
self.library_name = library_name
|
|
self.account_name = account_name
|
|
self.library_main_path = os.path.join(LLMWareConfig.get_library_path(), account_name, library_name)
|
|
|
|
# add new file dir for this collection
|
|
self.file_copy_path = os.path.join(self.library_main_path, "uploads" + os.sep)
|
|
self.image_path = os.path.join(self.library_main_path, "images" + os.sep)
|
|
self.dataset_path = os.path.join(self.library_main_path, "datasets" + os.sep)
|
|
self.nlp_path = os.path.join(self.library_main_path, "nlp" + os.sep)
|
|
self.output_path = os.path.join(self.library_main_path, "output" + os.sep)
|
|
self.tmp_path = os.path.join(self.library_main_path, "tmp" + os.sep)
|
|
self.embedding_path = os.path.join(self.library_main_path, "embedding" + os.sep)
|
|
|
|
os.makedirs(self.library_main_path, exist_ok=True)
|
|
os.makedirs(self.file_copy_path,exist_ok=True)
|
|
os.makedirs(self.image_path,exist_ok=True)
|
|
os.makedirs(self.dataset_path,exist_ok=True)
|
|
os.makedirs(self.nlp_path,exist_ok=True)
|
|
os.makedirs(self.output_path,exist_ok=True)
|
|
os.makedirs(self.tmp_path,exist_ok=True)
|
|
os.makedirs(self.embedding_path,exist_ok=True)
|
|
|
|
return self
|
|
|
|
def get_library_card(self, library_name=None, account_name="llmware"):
|
|
|
|
"""Retrieves the library card dictionary with key attributes of library.
|
|
|
|
Parameters
|
|
----------
|
|
library_name : str, default=None
|
|
Name of the library to retrieve. If not provided, uses self.library_name
|
|
|
|
account_name : str, default="llmware"
|
|
Name of the account associated to the library. If not provided, uses self.account_name
|
|
|
|
Returns
|
|
-------
|
|
library_card : dict or None
|
|
The library card dictionary containing key atrributes of the library. If not found, returns None
|
|
"""
|
|
|
|
library_card = None
|
|
|
|
if library_name:
|
|
lib_lookup_name = library_name
|
|
acct_lookup_name = account_name
|
|
else:
|
|
lib_lookup_name = self.library_name
|
|
acct_lookup_name = self.account_name
|
|
|
|
if lib_lookup_name and acct_lookup_name:
|
|
library_card= LibraryCatalog().get_library_card(lib_lookup_name, account_name=acct_lookup_name)
|
|
|
|
if not library_card:
|
|
logger.warning(f"Library - get_library_card - library card not found - {library_name} - {account_name}")
|
|
|
|
return library_card
|
|
|
|
def check_if_library_exists(self, library_name, account_name="llmware"):
|
|
|
|
"""Check if library exists by library string name.
|
|
|
|
Parameters
|
|
----------
|
|
library_name : str
|
|
Name of library to check.
|
|
|
|
account_name : str, default="llmware"
|
|
Name of account associated with library.
|
|
|
|
Returns
|
|
-------
|
|
library_card : dict or None
|
|
The library card dict if the library exists. If not found, returns None.
|
|
|
|
"""
|
|
|
|
# first look in library catalog
|
|
library_card = LibraryCatalog().get_library_card(library_name, account_name=account_name)
|
|
|
|
# check file path
|
|
lib_path = os.path.join(LLMWareConfig.get_library_path(), account_name, library_name)
|
|
library_folder = os.path.exists(lib_path)
|
|
|
|
# if all checks consistent
|
|
if library_card and library_folder:
|
|
# library exists and is in good state
|
|
return library_card
|
|
|
|
if not library_card and not library_folder:
|
|
# library does not exist conclusively
|
|
return None
|
|
|
|
# may be error state - some artifacts exist and others do not
|
|
if library_card:
|
|
# view the library_card as the definitive record
|
|
return library_card
|
|
|
|
return library_card
|
|
|
|
def update_embedding_status (self, status_message, embedding_model, embedding_db,
|
|
embedded_blocks=0, embedding_dims=0,time_stamp="NA",delete_record=False):
|
|
|
|
"""Invoked at the end of the embedding job to update the library card and embedding record -- generally,
|
|
this method does not need to be invoked directly.
|
|
|
|
Parameters
|
|
----------
|
|
status_message : str
|
|
Status message for the embedding process. If "delete", the record will be marked for deletion.
|
|
|
|
embedding_model : str
|
|
Name of the embedding model used.
|
|
|
|
embedding_db : str
|
|
Name of the embedding database used.
|
|
|
|
embedded_blocks : int, default=0
|
|
Number of embedded blocks.
|
|
|
|
embedding_dims : int, default=0
|
|
Dimensions of the embedding.
|
|
|
|
time_stamp : str, default="NA"
|
|
Timestamp of the embedding process.
|
|
|
|
delete_record : bool, default=False
|
|
If True, marks the record for deletion.
|
|
|
|
Returns
|
|
-------
|
|
bool
|
|
True if the embedding status was successfully updated.
|
|
"""
|
|
|
|
# special handling for updating "embedding" in update_library_card
|
|
# -- append/insert this new embedding dict to the end of the embedding list
|
|
|
|
if status_message == "delete":
|
|
delete_record = True
|
|
|
|
update_dict = {"embedding": {"embedding_status": status_message,
|
|
"embedding_model": embedding_model,
|
|
"embedding_db": embedding_db,
|
|
"embedding_dims": embedding_dims,
|
|
"embedded_blocks": embedded_blocks,
|
|
"time_stamp": time_stamp}}
|
|
|
|
updater = LibraryCatalog(self).update_library_card(self.library_name, update_dict,
|
|
delete_record=delete_record, account_name=self.account_name)
|
|
|
|
return True
|
|
|
|
def get_embedding_status (self):
|
|
|
|
"""Pulls the embedding record for the current library from the library card.
|
|
|
|
Returns
|
|
-------
|
|
embedding_record : list or None
|
|
The embedding record, which is a list of dictionaries containing embedding status, model, and database.
|
|
If the library card or embedding record is not found, returns None.
|
|
"""
|
|
|
|
library_card = LibraryCatalog(self).get_library_card(self.library_name, account_name=self.account_name)
|
|
|
|
if not library_card:
|
|
|
|
raise LLMWareException(message=f"Library - get_embedding_status - library not found -"
|
|
f"library_name = {self.library_name} with account_name = {self.account_name}")
|
|
|
|
# embedding record will be a list of {"embedding_status" | "embedding_model" | "embedding_db"}
|
|
logger.info(f"Library - get_embedding_status - library_card - {library_card}")
|
|
|
|
if "embedding" in library_card:
|
|
embedding_record = library_card["embedding"]
|
|
else:
|
|
logger.warning(f"Library - get_embedding_status - could not identify embedding record in library card - {library_card}")
|
|
embedding_record = None
|
|
|
|
return embedding_record
|
|
|
|
def get_and_increment_doc_id(self):
|
|
|
|
"""Convenience method in library class - mirrors method in LibraryCatalog - increments, tracks and provides a
|
|
unique doc id for the library.
|
|
|
|
Returns
|
|
-------
|
|
unique_doc_id : int
|
|
The new unique document ID for the library.
|
|
"""
|
|
|
|
unique_doc_id = LibraryCatalog(self).get_and_increment_doc_id(self.library_name)
|
|
return unique_doc_id
|
|
|
|
def set_incremental_docs_blocks_images(self, added_docs=0, added_blocks=0, added_images=0, added_pages=0,
|
|
added_tables=0):
|
|
|
|
"""Updates the library card with incremental counters after completing a parsing job.
|
|
|
|
Parameters
|
|
----------
|
|
added_docs : int, default=0
|
|
Number of documents added.
|
|
|
|
added_blocks : int, default=0
|
|
Number of blocks added.
|
|
|
|
added_images : int, default=0
|
|
Number of images added.
|
|
|
|
added_pages : int, default=0
|
|
Number of pages added.
|
|
|
|
added_tables : int, default=0
|
|
Number of tables added.
|
|
|
|
Returns
|
|
-------
|
|
bool
|
|
True if the incremental counters were successfully updated.
|
|
"""
|
|
|
|
# updates counting parameters at end of parsing
|
|
updater = LibraryCatalog(self).set_incremental_docs_blocks_images(added_docs=added_docs,
|
|
added_blocks=added_blocks,
|
|
added_images=added_images,
|
|
added_pages=added_pages,
|
|
added_tables=added_tables)
|
|
|
|
return True
|
|
|
|
def add_file(self, file_path):
|
|
|
|
"""Ingests, parses, text chunks and indexes a single selected file to a library -
|
|
provide the full path to file.
|
|
|
|
Parameters
|
|
----------
|
|
file_path : str
|
|
The full path to the file to be ingested and indexed.
|
|
|
|
Returns
|
|
-------
|
|
self : Library
|
|
The updated ``Library`` object after adding the file.
|
|
"""
|
|
|
|
# Ensure the input path exists
|
|
os.makedirs(LLMWareConfig.get_input_path(), exist_ok=True)
|
|
|
|
file_name = os.path.basename(file_path)
|
|
target_path = os.path.join(LLMWareConfig.get_input_path(), file_name)
|
|
|
|
shutil.copyfile(file_path,target_path)
|
|
return self.add_files()
|
|
|
|
def add_files (self, input_folder_path=None, encoding="utf-8",chunk_size=400,
|
|
get_images=True,get_tables=True, smart_chunking=1, max_chunk_size=600,
|
|
table_grid=True, get_header_text=True, table_strategy=1, strip_header=False,
|
|
verbose_level=2, copy_files_to_library=True, set_custom_logging=-1,
|
|
use_logging_file=False):
|
|
|
|
"""Main method to integrate documents into a Library - pass a local filepath folder and all files will be
|
|
routed to appropriate parser by file type extension.
|
|
|
|
Parameters
|
|
----------
|
|
input_folder_path : str, default=None
|
|
The path to the folder containing files to be ingested. If not provided, defaults to None.
|
|
|
|
encoding : str, default="utf-8"
|
|
The encoding to use for reading files.
|
|
|
|
chunk_size : int, default=400
|
|
The size of text chunks to create during parsing.
|
|
|
|
get_images : bool, default=True
|
|
Whether to extract images from the documents.
|
|
|
|
get_tables : bool, default=True
|
|
Whether to extract tables from the documents.
|
|
|
|
smart_chunking : int, default=1
|
|
The strategy for smart chunking of text.
|
|
|
|
max_chunk_size : int, default=600
|
|
The maximum size of text chunks.
|
|
|
|
table_grid : bool, default=True
|
|
Whether to use a grid for tables.
|
|
|
|
get_header_text : bool, default=True
|
|
Whether to extract header text from the documents.
|
|
|
|
table_strategy : int, default=1
|
|
The strategy to use for table extraction.
|
|
|
|
strip_header : bool, default=False
|
|
Whether to strip headers from the documents.
|
|
|
|
verbose_level : int, default=2
|
|
The level of verbosity for logging.
|
|
|
|
copy_files_to_library : bool, default=True
|
|
Whether to copy the files to the library.
|
|
|
|
set_custom_logging : int, default=-1, will apply a custom logging level between 0-50 for the
|
|
parsing job.
|
|
|
|
use_logging_file : bool, default=False
|
|
Whether parse should log to stdout (default) or to file (set to True)
|
|
|
|
Returns
|
|
-------
|
|
output_results : dict or None
|
|
A dictionary containing the results of the document integration process, including counts of added documents,
|
|
blocks, images, pages, tables, and rejected files. If the library card could not be identified, returns None.
|
|
"""
|
|
|
|
if not input_folder_path:
|
|
input_folder_path = LLMWareConfig.get_input_path()
|
|
|
|
# get overall counters at start of process
|
|
lib_counters_before = self.get_library_card()
|
|
|
|
parsing_results = Parser(library=self,
|
|
encoding=encoding,
|
|
chunk_size=chunk_size,
|
|
max_chunk_size=max_chunk_size,
|
|
smart_chunking=smart_chunking,
|
|
get_tables=get_tables,
|
|
get_images=get_images,
|
|
get_header_text=get_header_text,
|
|
table_strategy=table_strategy,
|
|
strip_header=strip_header,
|
|
table_grid=table_grid,
|
|
verbose_level=verbose_level,
|
|
copy_files_to_library=copy_files_to_library,
|
|
set_custom_logging=set_custom_logging,
|
|
use_logging_file=use_logging_file).ingest(input_folder_path,dupe_check=True)
|
|
|
|
logger.debug(f"Library - add_files - parsing results - {parsing_results}")
|
|
|
|
# post-processing: get the updated lib_counters
|
|
lib_counters_after = self.get_library_card()
|
|
|
|
# parsing_results = {"processed_files" | "rejected_files" | "duplicate_files"}
|
|
output_results = None
|
|
|
|
if lib_counters_after and lib_counters_before:
|
|
output_results = {"docs_added": lib_counters_after["documents"] - lib_counters_before["documents"],
|
|
"blocks_added": lib_counters_after["blocks"] - lib_counters_before["blocks"],
|
|
"images_added": lib_counters_after["images"] - lib_counters_before["images"],
|
|
"pages_added": lib_counters_after["pages"] - lib_counters_before["pages"],
|
|
"tables_added": lib_counters_after["tables"] - lib_counters_before["tables"],
|
|
"rejected_files": parsing_results["rejected_files"]}
|
|
else:
|
|
logger.error("Library - add_files - unexpected - could not identify the library_card correctly")
|
|
|
|
logger.info(f"Library - add_files - output_results - {output_results}")
|
|
|
|
# update collection text index in collection after adding documents
|
|
# LibraryCollection(self).create_index()
|
|
CollectionWriter(self.library_name,account_name=self.account_name).build_text_index()
|
|
|
|
return output_results
|
|
|
|
def export_library_to_txt_file(self, output_fp=None, output_fn=None, include_text=True, include_tables=True,
|
|
include_images=False):
|
|
|
|
"""Exports library collection of indexed text chunks to a txt file.
|
|
|
|
Parameters
|
|
----------
|
|
output_fp : str, default=None
|
|
The file path where the output file will be saved. If not provided, defaults to None.
|
|
|
|
output_fn : str, default=None
|
|
The name of the output file. If not provided, defaults to None.
|
|
|
|
include_text : bool, default=True
|
|
Whether to include text content in the export.
|
|
|
|
include_tables : bool, default=True
|
|
Whether to include tables in the export.
|
|
|
|
include_images : bool, default=False
|
|
Whether to include images in the export.
|
|
|
|
Returns
|
|
-------
|
|
file_location : str
|
|
The location of the exported txt file.
|
|
"""
|
|
|
|
if not output_fp:
|
|
output_fp = self.output_path
|
|
|
|
if not output_fn:
|
|
output_fn = self.library_name + "_" + str(Utilities().get_current_time_now())
|
|
|
|
filter_list = []
|
|
if include_text: filter_list.append("text")
|
|
if include_tables: filter_list.append("table")
|
|
if include_images: filter_list.append("image")
|
|
|
|
if not filter_list:
|
|
# go with default - text only
|
|
filter_list = ["text"]
|
|
|
|
results = CollectionRetrieval(self.library_name,
|
|
account_name=self.account_name).filter_by_key_value_range("content_type",filter_list)
|
|
|
|
file_location = os.path.join(output_fp, output_fn + ".txt")
|
|
output_file = open(file_location, "w", encoding='utf-8')
|
|
text_field = "text_search"
|
|
for elements in results:
|
|
new_entry = elements[text_field].strip() + "\n"
|
|
output_file.write(new_entry)
|
|
|
|
output_file.close()
|
|
|
|
return file_location
|
|
|
|
def export_library_to_jsonl_file(self, output_fp, output_fn, include_text=True, include_tables=True,
|
|
include_images=False, dict_keys=None):
|
|
|
|
"""Exports collection of text chunks to a jsonl file.
|
|
|
|
Parameters
|
|
----------
|
|
output_fp : str
|
|
The file path where the output file will be saved.
|
|
|
|
output_fn : str
|
|
The name of the output file.
|
|
|
|
include_text : bool, default=True
|
|
Whether to include text content in the export.
|
|
|
|
include_tables : bool, default=True
|
|
Whether to include tables in the export.
|
|
|
|
include_images : bool, default=False
|
|
Whether to include images in the export.
|
|
|
|
dict_keys : list of str, default=None
|
|
The keys to include in the JSONL entries. If not provided, defaults to None.
|
|
|
|
Returns
|
|
-------
|
|
file_location : str
|
|
The location of the exported JSONL file.
|
|
"""
|
|
|
|
if not output_fp:
|
|
output_fp = self.output_path
|
|
|
|
if not output_fn:
|
|
output_fn = self.library_name + "_" + str(Utilities().get_current_time_now())
|
|
|
|
# expects dict_keys to be a list of dictionary keys
|
|
if not dict_keys:
|
|
dict_keys = self.default_keys
|
|
|
|
filter_list = []
|
|
if include_text: filter_list.append("text")
|
|
if include_tables: filter_list.append("table")
|
|
if include_images: filter_list.append("image")
|
|
|
|
if not filter_list:
|
|
# go with default - text only
|
|
filter_list = ["text"]
|
|
|
|
results = CollectionRetrieval(self.library_name,
|
|
account_name=self.account_name).filter_by_key_value_range("content_type", filter_list)
|
|
|
|
file_location = os.path.join(output_fp, output_fn + ".jsonl")
|
|
output_file = open(file_location, "w", encoding='utf-8')
|
|
|
|
for elements in results:
|
|
|
|
# package up each jsonl entry as dict with selected keys to extract
|
|
new_dict_entry = {}
|
|
for keys in dict_keys:
|
|
if keys in elements:
|
|
new_dict_entry.update({keys:elements[keys]})
|
|
|
|
if new_dict_entry:
|
|
jsonl_row = json.dumps(new_dict_entry)
|
|
output_file.write(jsonl_row)
|
|
output_file.write("\n")
|
|
|
|
output_file.close()
|
|
|
|
return file_location
|
|
|
|
def pull_files_from_cloud_bucket (self, aws_access_key=None, aws_secret_key=None, bucket_name=None):
|
|
|
|
"""Pull files from private S3 bucket into local cache for further processing.
|
|
|
|
Parameters
|
|
----------
|
|
aws_access_key : str, default=None
|
|
The AWS access key for connecting to the S3 bucket.
|
|
|
|
aws_secret_key : str, default=None
|
|
The AWS secret key for connecting to the S3 bucket.
|
|
|
|
bucket_name : str, default=None
|
|
The name of the S3 bucket from which to pull files.
|
|
|
|
Returns
|
|
-------
|
|
files_copied : list
|
|
A list of file paths that were copied from the S3 bucket to the local cache.
|
|
"""
|
|
|
|
files_copied = CloudBucketManager().connect_to_user_s3_bucket (aws_access_key, aws_secret_key,
|
|
bucket_name, LLMWareConfig.get_input_path())
|
|
|
|
return files_copied
|
|
|
|
def install_new_embedding (self, embedding_model_name=None, vector_db=None,
|
|
from_hf= False, from_sentence_transformer=False, model=None, tokenizer=None, model_api_key=None,
|
|
vector_db_api_key=None, batch_size=500, max_len=None, use_gpu=True):
|
|
|
|
"""Main method for installing a new embedding on a library.
|
|
|
|
Parameters
|
|
----------
|
|
embedding_model_name : str, default=None
|
|
The name of the embedding model to use.
|
|
|
|
vector_db : str, default=None
|
|
The name of the vector database to use.
|
|
|
|
from_hf : bool, default=False
|
|
Whether the model is from Hugging Face.
|
|
|
|
from_sentence_transformer : bool, default=False
|
|
Whether the model is a Sentence Transformer.
|
|
|
|
model : object, default=None
|
|
The pre-loaded model to use.
|
|
|
|
tokenizer : object, default=None
|
|
The tokenizer associated with the pre-loaded model.
|
|
|
|
model_api_key : str, default=None
|
|
The API key for accessing the model.
|
|
|
|
vector_db_api_key : str, default=None
|
|
The API key for accessing the vector database.
|
|
|
|
batch_size : int, default=500
|
|
The batch size to use for embedding.
|
|
|
|
max_len : int, default=None
|
|
The maximum length for embedding.
|
|
|
|
use_gpu : bool, default=True
|
|
Whether to use GPU for embedding.
|
|
|
|
Returns
|
|
-------
|
|
embeddings : dict or None
|
|
The created embeddings dict, or None if no embeddings could be created.
|
|
"""
|
|
|
|
embeddings = None
|
|
my_model = None
|
|
|
|
# step 1 - load selected model from ModelCatalog - will pass 'loaded' model to the EmbeddingHandler
|
|
|
|
# check if instantiated model and tokenizer -> load as HuggingFace model
|
|
if model:
|
|
if from_hf:
|
|
logger.info("Library - install_new_embedding - loading hf model")
|
|
my_model = ModelCatalog().load_hf_embedding_model(model, tokenizer)
|
|
batch_size = 50
|
|
|
|
if from_sentence_transformer:
|
|
logger.info("library - install_new_embedding - loading sentence transformer model")
|
|
if not embedding_model_name:
|
|
raise LLMWareException(message=f"Library - install_new_embedding - to use "
|
|
f"sentence_transformer model requires providing the model name.")
|
|
|
|
my_model = ModelCatalog().load_sentence_transformer_model(model,embedding_model_name)
|
|
else:
|
|
# if no model explicitly passed, then look up in the model catalog
|
|
if embedding_model_name:
|
|
my_model = ModelCatalog().load_model(selected_model=embedding_model_name, api_key=model_api_key)
|
|
|
|
if not my_model:
|
|
logger.error("Library - install_new_embedding - can not identify a selected model")
|
|
return -1
|
|
|
|
# new - insert - handle no vector_db passed
|
|
if not vector_db:
|
|
vector_db = LLMWareConfig().get_config("vector_db")
|
|
# end - new insert
|
|
|
|
if vector_db not in LLMWareConfig().get_supported_vector_db():
|
|
raise LLMWareException(message=f"Library - install_new_embedding - selected "
|
|
f"vector db is not supported - {vector_db}")
|
|
|
|
if my_model and max_len:
|
|
my_model.max_len = max_len
|
|
|
|
# step 2 - pass loaded embedding model to EmbeddingHandler, which will route to the appropriate resource
|
|
embeddings = EmbeddingHandler(self).create_new_embedding(vector_db, my_model, batch_size=batch_size)
|
|
|
|
if not embeddings:
|
|
logger.warning("Library - install_new_embedding - no embeddings created")
|
|
|
|
return embeddings
|
|
|
|
def delete_library(self, library_name=None, confirm_delete=False, account_name="llmware"):
|
|
|
|
""" Deletes all artifacts of a library
|
|
|
|
Parameters
|
|
----------
|
|
library_name : str, default=None
|
|
The name of the library to delete. If not provided, defaults to None.
|
|
|
|
confirm_delete : bool, default=False
|
|
Confirmation flag to proceed with deletion. Must be set to True to delete the library.
|
|
|
|
Returns
|
|
-------
|
|
success_code : int
|
|
Returns 1 if the deletion was successful, or -1 if an error occurred.
|
|
|
|
"""
|
|
|
|
if library_name:
|
|
self.library_name = library_name
|
|
|
|
# loads the library specific path information if required
|
|
self.load_library(library_name,account_name=account_name)
|
|
|
|
success_code = 1
|
|
|
|
try:
|
|
if confirm_delete:
|
|
|
|
# 1st - remove the blocks - drop the collection in database
|
|
CollectionWriter(self.library_name, account_name=self.account_name).destroy_collection(confirm_destroy=True)
|
|
|
|
# 2nd - Eliminate the local file structure
|
|
file_path = self.library_main_path
|
|
shutil.rmtree(file_path)
|
|
|
|
# 3rd - remove record in LibraryCatalog
|
|
LibraryCatalog(self).delete_library_card(self.library_name)
|
|
|
|
logger.info("Library - delete_library - deleted all library file artifacts + folders")
|
|
|
|
except:
|
|
logger.exception("Library - delete_library - error destroying library")
|
|
success_code = -1
|
|
|
|
return success_code
|
|
|
|
def update_block (self, doc_id, block_id, key, new_value):
|
|
|
|
"""Convenience method to update the record of a specific block - identified by doc_ID and block_ID
|
|
in text collection database.
|
|
|
|
Parameters
|
|
----------
|
|
doc_id : int
|
|
The ID of the document containing the block to update.
|
|
|
|
block_id : int
|
|
The ID of the block to update.
|
|
|
|
key : str
|
|
The key in the block record to update.
|
|
|
|
new_value : str
|
|
The new value to set for the specified key.
|
|
|
|
Returns
|
|
-------
|
|
completed : bool
|
|
True if the block was successfully updated, False otherwise.
|
|
"""
|
|
|
|
completed = (CollectionWriter(self.library_name, account_name=self.account_name).
|
|
update_block(doc_id, block_id,key,new_value,self.default_keys))
|
|
|
|
return completed
|
|
|
|
def add_website (self, url, get_links=True, max_links=5):
|
|
|
|
"""Main method to ingest a website into a library.
|
|
|
|
Parameters
|
|
----------
|
|
url : str
|
|
The URL of the website to ingest.
|
|
|
|
get_links : bool, default=True
|
|
Whether to follow and ingest links found on the website.
|
|
|
|
max_links : int, default=5
|
|
The maximum number of links to follow and ingest.
|
|
|
|
Returns
|
|
-------
|
|
self : Library
|
|
The updated ``Library`` object after ingesting the website.
|
|
"""
|
|
|
|
Parser(library=self).parse_website(url,get_links=get_links,max_links=max_links)
|
|
CollectionWriter(self.library_name, account_name=self.account_name).build_text_index()
|
|
|
|
return self
|
|
|
|
def add_wiki(self, topic_list,target_results=10):
|
|
|
|
"""Main method to add a wikipedia article to a library - enter a list of topics.
|
|
|
|
Parameters
|
|
----------
|
|
topic_list : list of str
|
|
A list of topics to search for on Wikipedia.
|
|
|
|
target_results : int, default=10
|
|
The target number of results to retrieve for each topic.
|
|
|
|
Returns
|
|
-------
|
|
self : Library
|
|
The updated ``Library`` object after adding the Wikipedia articles.
|
|
"""
|
|
|
|
Parser(library=self).parse_wiki(topic_list,target_results=target_results)
|
|
CollectionWriter(self.library_name, account_name=self.account_name).build_text_index()
|
|
|
|
return self
|
|
|
|
def add_dialogs(self, input_folder=None):
|
|
|
|
"""Main method to add an AWS dialog transcript into a library.
|
|
|
|
Parameters
|
|
----------
|
|
input_folder : str, default=None
|
|
The path to the folder containing the dialog transcripts. If not provided, defaults to None.
|
|
|
|
Returns
|
|
-------
|
|
self : Library
|
|
The updated ``Library`` object after adding the dialog transcripts.
|
|
"""
|
|
|
|
if not input_folder:
|
|
input_folder = LLMWareConfig.get_input_path()
|
|
|
|
output = Parser(library=self).parse_dialog(input_folder)
|
|
|
|
return self
|
|
|
|
def add_image(self, input_folder=None):
|
|
|
|
"""Main method to add image and scanned OCR content into a library.
|
|
|
|
Parameters
|
|
----------
|
|
input_folder : str, default=None
|
|
The path to the folder containing the images. If not provided, defaults to None
|
|
|
|
Returns
|
|
-------
|
|
self : Library
|
|
The updated ``Library`` object after adding the image and OCR content.
|
|
"""
|
|
|
|
if not input_folder:
|
|
input_folder = LLMWareConfig.get_input_path()
|
|
|
|
output = Parser(library=self).parse_image(input_folder)
|
|
|
|
return self
|
|
|
|
def add_pdf_by_ocr(self, input_folder=None):
|
|
|
|
"""Alternative method to ingest PDFs that are scanned, or can not be otherwise parsed.
|
|
|
|
Parameters
|
|
----------
|
|
input_folder : str, default=None
|
|
The path to the folder containing the PDFs. If not provided, defaults to None
|
|
|
|
Returns
|
|
-------
|
|
self : Library
|
|
The updated ``Library`` object after adding the PDFs through OCR.
|
|
"""
|
|
|
|
if not input_folder:
|
|
input_folder = LLMWareConfig.get_input_path()
|
|
|
|
output = Parser(library=self).parse_pdf_by_ocr_images(input_folder)
|
|
|
|
return self
|
|
|
|
def add_pdf(self, input_folder=None):
|
|
|
|
"""Convenience method to directly add PDFs only - note, in most cases, 'add_files' is the better option.
|
|
|
|
Parameters
|
|
----------
|
|
input_folder : str, default=None
|
|
The path to the folder containing the PDFs. If not provided, defaults to None
|
|
|
|
Returns
|
|
-------
|
|
self : Library
|
|
The updated ``Library`` object after adding the PDFs.
|
|
"""
|
|
|
|
if not input_folder:
|
|
input_folder = LLMWareConfig.get_input_path()
|
|
|
|
output = Parser(library=self).parse_pdf(input_folder)
|
|
|
|
return self
|
|
|
|
def add_office(self, input_folder=None):
|
|
|
|
"""Convenience method to directly add PDFs only - note, in most cases, 'add_files' is the better option.
|
|
|
|
Parameters
|
|
----------
|
|
input_folder : str, default=None
|
|
The path to the folder containing the Office documents. If not provided, defaults to None.
|
|
|
|
Returns
|
|
-------
|
|
self : Library
|
|
The updated ``Library`` object after adding the Office documents.
|
|
"""
|
|
|
|
if not input_folder:
|
|
input_folder = LLMWareConfig.get_input_path()
|
|
|
|
output = Parser(library=self).parse_office(input_folder)
|
|
|
|
return self
|
|
|
|
def get_all_library_cards(self, account_name='llmware'):
|
|
|
|
"""Get all library cards for all libraries on account.
|
|
|
|
Parameters
|
|
----------
|
|
account_name : str, default='llmware'
|
|
The name of the account for which to retrieve all library cards.
|
|
|
|
Returns
|
|
-------
|
|
library_cards : list of dict
|
|
A list of all library card dictionaries for the specified account.
|
|
"""
|
|
|
|
library_cards = LibraryCatalog(account_name=account_name).all_library_cards()
|
|
return library_cards
|
|
|
|
def delete_installed_embedding(self, embedding_model_name, vector_db, vector_db_api_key=None):
|
|
|
|
"""Deletes an installed embedding on specific combination of vector_db + embedding_model_name.
|
|
|
|
Parameters
|
|
----------
|
|
embedding_model_name : str
|
|
The name of the embedding model to delete.
|
|
|
|
vector_db : str
|
|
The name of the vector database from which to delete the embedding.
|
|
|
|
vector_db_api_key : str, default=None
|
|
The API key for accessing the vector database. If not provided, defaults to None
|
|
|
|
Returns
|
|
-------
|
|
int
|
|
Returns 1 if the embedding was successfully deleted.
|
|
"""
|
|
|
|
# insert safety check - confirm that this is valid combination with installed embedding
|
|
lib_card = LibraryCatalog(self).get_library_card(self.library_name)
|
|
embedding_list = lib_card["embedding"]
|
|
found_match = False
|
|
embedding_dims = 0
|
|
|
|
for entries in embedding_list:
|
|
if entries["embedding_model"] == embedding_model_name and entries["embedding_db"] == vector_db:
|
|
found_match = True
|
|
embedding_dims = entries["embedding_dims"]
|
|
|
|
logger.info(f"Library - delete_installed_embedding - found matching"
|
|
f"embedding record - {entries}")
|
|
break
|
|
|
|
if found_match:
|
|
EmbeddingHandler(self).delete_index(vector_db,embedding_model_name, embedding_dims)
|
|
else:
|
|
# update exception
|
|
raise LLMWareException(message=f"Library - delete_installed_embedding - library not found -"
|
|
f"library_name = {self.library_name} with account_name = {self.account_name}")
|
|
|
|
return 1
|
|
|
|
def run_ocr_on_images(self, add_to_library=False,chunk_size=400,min_size=10, realtime_progress=True):
|
|
|
|
"""Convenience method in Library class to pass Library to Parser to run OCR on all of the images
|
|
found in the Library, and OCR-extracted text from the images directly into the Library as additional
|
|
blocks.
|
|
|
|
Parameters
|
|
----------
|
|
add_to_library : bool, default=False
|
|
Whether to add the OCR-extracted text directly into the Library as additional blocks.
|
|
|
|
chunk_size : int, default=400
|
|
The size of text chunks to create during OCR processing.
|
|
|
|
min_size : int, default=10
|
|
The minimum size of text chunks to consider during OCR processing.
|
|
|
|
realtime_progress : bool, default=True
|
|
Whether to display real-time progress during OCR processing.
|
|
|
|
Returns
|
|
-------
|
|
output : int
|
|
Returns 1 if running the OCR on the images was successful.
|
|
"""
|
|
|
|
output = Parser(library=self).ocr_images_in_library(add_to_library=add_to_library,
|
|
chunk_size=chunk_size,min_size=min_size,
|
|
realtime_progress=realtime_progress)
|
|
|
|
return output
|
|
|
|
def expand_text_result_before(self, block, window_size=400):
|
|
|
|
""" Expands text result before. Duplicate of Query method - added to Library class for direct access and
|
|
convenience in certain use cases."""
|
|
|
|
block_id = block["block_ID"] -1
|
|
doc_id = block["doc_ID"]
|
|
|
|
before_text = ""
|
|
pre_blocks = []
|
|
|
|
while len(before_text) < window_size and block_id >= 0:
|
|
|
|
before_block = self.block_lookup(block_id, doc_id)
|
|
|
|
if before_block:
|
|
before_text += before_block["text"]
|
|
pre_blocks.append(before_block)
|
|
|
|
output = {"expanded_text": before_text, "results": pre_blocks}
|
|
|
|
return output
|
|
|
|
def expand_text_result_after(self, block, window_size=400):
|
|
|
|
""" Expands text result after. Duplicate of Query method - added to Library class for direct access and
|
|
convenience in certain use cases. """
|
|
|
|
block_id = block["block_ID"] + 1
|
|
doc_id = block["doc_ID"]
|
|
|
|
after_text = ""
|
|
post_blocks = []
|
|
|
|
while len(after_text) < window_size:
|
|
after_block = self.block_lookup(block_id, doc_id)
|
|
if not after_block:
|
|
break # Break if no block is found
|
|
|
|
after_text += after_block["text"]
|
|
post_blocks.append(after_block)
|
|
block_id += 1 # Increment block_id for next iteration
|
|
|
|
output = {"expanded_text": after_text, "results": post_blocks}
|
|
return output
|
|
|
|
def block_lookup(self, block_id, doc_id):
|
|
|
|
""" Look up by a specific pair of doc_id and block_id in a library. Duplicate of Query method -
|
|
added to Library class for direct access and convenience in certain use cases."""
|
|
|
|
result = None
|
|
|
|
kv_dict = {"doc_ID": doc_id, "block_ID": block_id}
|
|
|
|
output = CollectionRetrieval(self.library_name, account_name=self.account_name).filter_by_key_dict(kv_dict)
|
|
|
|
if len(output) == 0:
|
|
logger.info(f"Library - block_lookup - block not found: {block_id}")
|
|
result = None
|
|
|
|
return result
|
|
|
|
if len(output) > 1:
|
|
result = output[0]
|
|
|
|
if len(output) == 1:
|
|
result = output[0]
|
|
|
|
# if arrived this point, then positive result has been identified
|
|
result.update({"matches": []})
|
|
result.update({"page_num": result["master_index"]})
|
|
|
|
return result
|
|
|
|
|
|
class LibraryCatalog:
|
|
|
|
"""Implements the management of tracking details for libraries via the library card, which is stored
|
|
in the `library` table of the text collection database. It is used by the ``Library`` class.
|
|
|
|
``LibraryCatalog`` is responsible for managing tracking details. This includes creating,
|
|
reading, updating, and deleting library cards. The library card is stored in the table library
|
|
of the chosen text collection database. In most cases, ``LibraryCatalog`` does not need to be directly
|
|
invoked, instead it is used indirectly through the methods of ``Library``.
|
|
|
|
Parameters
|
|
----------
|
|
library : Library, default=None
|
|
The library with which the ``LibraryCatalog`` interacts.
|
|
|
|
library_path : str or pathlib.Path object, default=None
|
|
The path to the llmware directory. If set, then the default from ``LLMWareconfig`` is used.
|
|
|
|
account_name : str, default='llmware'
|
|
Name of the account.
|
|
|
|
Returns
|
|
-------
|
|
library_catalog : LibraryCatalog
|
|
A new ``LibraryCatalog`` object.
|
|
"""
|
|
|
|
def __init__(self, library=None, library_path=None, account_name="llmware"):
|
|
|
|
self.library = library
|
|
if library:
|
|
self.library_name = library.library_name
|
|
self.account_name = library.account_name
|
|
else:
|
|
self.library_name = None
|
|
self.account_name = account_name
|
|
|
|
self.schema = LLMWareTableSchema().get_library_card_schema()
|
|
|
|
# if table does not exist, then create
|
|
if CollectionWriter("library",account_name=self.account_name).check_if_table_build_required():
|
|
CollectionWriter("library", account_name=self.account_name).create_table("library", self.schema)
|
|
|
|
# check for llmware path & create if not already set up
|
|
if not os.path.exists(LLMWareConfig.get_llmware_path()):
|
|
LLMWareConfig.setup_llmware_workspace()
|
|
|
|
if not library_path:
|
|
self.library_path = LLMWareConfig.get_llmware_path()
|
|
else:
|
|
self.library_path = library_path
|
|
|
|
def get_library_card (self, library_name, account_name="llmware"):
|
|
|
|
""" Gets the selected library card for the selected library_name """
|
|
|
|
# note: will return either library_card {} or None
|
|
|
|
db_record = CollectionRetrieval("library", account_name=account_name).lookup("library_name", library_name)
|
|
|
|
if isinstance(db_record, list):
|
|
if len(db_record) > 0:
|
|
db_record = db_record[0]
|
|
|
|
library_card = db_record
|
|
|
|
return library_card
|
|
|
|
def all_library_cards(self):
|
|
|
|
""" Get all library cards """
|
|
|
|
all_library_cards_cursor = CollectionRetrieval("library",
|
|
account_name=self.account_name).get_whole_collection()
|
|
|
|
all_library_cards = all_library_cards_cursor.pull_all()
|
|
|
|
return all_library_cards
|
|
|
|
def create_new_library_card(self, new_library_card):
|
|
|
|
""" Creates new library card entry """
|
|
|
|
new_lib_name = new_library_card["library_name"]
|
|
|
|
logger.debug(f"LibraryCatalog - create_new_library_card - {new_lib_name} - {new_library_card}")
|
|
|
|
CollectionWriter("library", account_name=self.account_name).write_new_record(new_library_card)
|
|
|
|
# test to pull card here
|
|
# lib_card = self.get_library_card(new_lib_name)
|
|
# end - test get card
|
|
|
|
return 0
|
|
|
|
def update_library_card(self, library_name, update_dict, account_name="llmware", delete_record=False):
|
|
|
|
""" Updates library card entry """
|
|
|
|
lib_card = self.get_library_card(library_name, account_name=account_name)
|
|
|
|
updater = CollectionWriter("library",
|
|
account_name=self.account_name).update_library_card(library_name,
|
|
update_dict,
|
|
lib_card,
|
|
delete_record=delete_record)
|
|
|
|
return 1
|
|
|
|
def delete_library_card(self, library_name=None, account_name="llmware"):
|
|
|
|
""" Deletes library card """
|
|
|
|
if not library_name:
|
|
library_name = self.library_name
|
|
|
|
if account_name != "llmware":
|
|
self.account_name = account_name
|
|
|
|
f = {"library_name": library_name}
|
|
|
|
# self.library_card_collection.delete_one(f)
|
|
CollectionWriter("library", account_name=self.account_name).delete_record_by_key("library_name", library_name)
|
|
|
|
return 1
|
|
|
|
def get_and_increment_doc_id (self, library_name, account_name="llmware"):
|
|
|
|
""" Gets and increments unique doc id counter for library """
|
|
|
|
if account_name != "llmware":
|
|
self.account_name = account_name
|
|
|
|
cw = CollectionWriter("library", account_name=self.account_name)
|
|
unique_doc_id = cw.get_and_increment_doc_id(library_name)
|
|
|
|
return unique_doc_id
|
|
|
|
def set_incremental_docs_blocks_images(self, added_docs=0, added_blocks=0, added_images=0, added_pages=0,
|
|
added_tables=0):
|
|
|
|
""" Updates library card with incremental counters after parsing """
|
|
|
|
# updates counting parameters at end of parsing
|
|
cw = CollectionWriter("library", account_name=self.account_name)
|
|
|
|
cw.set_incremental_docs_blocks_images(self.library_name,added_docs=added_docs,added_blocks=added_blocks,
|
|
added_images=added_images, added_pages=added_pages,
|
|
added_tables=added_tables)
|
|
return 0
|
|
|