Update database.py

database.py

@@ -11,19 +11,31 @@ logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(
 
 class KodeksProcessor:
     def __init__(self):
-        logging.info("Inicjalizacja klienta bazy danych...")
+        logging.info(f"Inicjalizacja klienta bazy danych w katalogu: {DATABASE_DIR}")
+        if not os.path.exists(DATABASE_DIR):
+            os.makedirs(DATABASE_DIR)
+            logging.info(f"Utworzono katalog {DATABASE_DIR}")
+        
         self.client = chromadb.PersistentClient(path=DATABASE_DIR)
+        logging.info("Klient bazy danych zainicjalizowany")
+        
         try:
             self.collection = self.client.get_collection("kodeksy")
-            logging.info("Pobrano istniejącą kolekcję 'kodeksy'.")
-        except:
-            self.collection = self.client.create_collection(
-                name="kodeksy",
-                embedding_function=embedding_functions.SentenceTransformerEmbeddingFunction(
-                    model_name=EMBEDDING_MODEL
+            logging.info("Pobrano istniejącą kolekcję 'kodeksy'")
+        except Exception as e:
+            logging.error(f"Błąd podczas pobierania kolekcji: {e}")
+            logging.info("Próba utworzenia nowej kolekcji 'kodeksy'")
+            try:
+                self.collection = self.client.create_collection(
+                    name="kodeksy",
+                    embedding_function=embedding_functions.SentenceTransformerEmbeddingFunction(
+                        model_name=EMBEDDING_MODEL
+                    )
                 )
-            )
-            logging.info("Utworzono nową kolekcję 'kodeksy'.")
+                logging.info("Utworzono nową kolekcję 'kodeksy'")
+            except Exception as e:
+                logging.error(f"Błąd podczas tworzenia kolekcji: {e}")
+                raise
 
     def extract_metadata(self, text: str) -> Dict:
         metadata = {}
@@ -37,6 +49,19 @@ class KodeksProcessor:
             metadata['data_ustawy'] = nazwa_match.group(1).strip()
             metadata['nazwa'] = nazwa_match.group(2).strip()
 
+        # Dodanie przetwarzania historii zmian
+        zmiany = re.findall(r'(\d{4}-\d{2}-\d{2})\s+(zm\.\s+DZ\.U\.(\d{4})\.(\d+)\.(\d+)\s+art\.\s+(\d+)(?:\s+§\s+(\d+))?)', text)
+        if zmiany:
+            metadata['historia_zmian'] = [
+                {
+                    'data': data,
+                    'dz_u': f"Dz.U.{rok}.{numer}.{pozycja}",
+                    'artykul': artykul,
+                    'paragraf': paragraf if paragraf else None
+                }
+                for data, _, rok, numer, pozycja, artykul, paragraf in zmiany
+            ]
+
         logging.info("Wydobyto metadane: %s", metadata)
         return metadata
 
@@ -47,7 +72,7 @@ class KodeksProcessor:
         return "", text
 
     def process_article(self, article_text: str) -> Dict:
-        art_num_match = re.match(r'Art\.\s*(\d+)', article_text)
+        art_num_match = re.match(r'Art\.\s*(\d+[a-z]?)', article_text)
         article_num = art_num_match.group(1) if art_num_match else ""
 
         paragraphs = re.findall(r'§\s*(\d+)\.\s*(.*?)(?=§\s*\d+|Art\.\s*\d+|$)', article_text, re.DOTALL)
@@ -67,9 +92,9 @@ class KodeksProcessor:
 
     def split_into_chunks(self, text: str, metadata: Dict) -> List[Dict]:
         chunks = []
-        articles = re.split(r'(Art\.\s*\d+)', text)  # Podział na artykuły
+        articles = re.split(r'(Art\.\s*\d+[a-z]?)', text)
 
-        for i in range(1, len(articles), 2):  # Przechodzimy przez artykuły
+        for i in range(1, len(articles), 2):
             article_title = articles[i].strip()
             article_content = articles[i + 1].strip() if i + 1 < len(articles) else ""
 
@@ -98,8 +123,12 @@ class KodeksProcessor:
     def process_file(self, filepath: str) -> None:
         logging.info("Przetwarzanie pliku: %s", filepath)
 
-        with open(filepath, 'r', encoding='utf-8') as file:
-            content = file.read()
+        try:
+            with open(filepath, 'r', encoding='utf-8') as file:
+                content = file.read()
+        except Exception as e:
+            logging.error(f"Błąd podczas odczytu pliku {filepath}: {e}")
+            return
 
         header, main_content = self.split_header_and_content(content)
         metadata = self.extract_metadata(main_content)
@@ -107,16 +136,19 @@ class KodeksProcessor:
 
         chunks = self.split_into_chunks(main_content, metadata)
 
-        if chunks:  # Sprawdzenie, czy są jakieś chunk'i do dodania
-            for i, chunk in enumerate(chunks):
-                self.collection.add(
-                    documents=[chunk["text"]],
-                    metadatas=[chunk["metadata"]],
-                    ids=[f"{metadata['filename']}_{chunk['metadata']['article']}_{i}"]
-                )
-                logging.info("Dodano chunk: %s", chunk["text"])  # Logowanie dodawanych chunków
+        if chunks:
+            try:
+                for i, chunk in enumerate(chunks):
+                    self.collection.add(
+                        documents=[chunk["text"]],
+                        metadatas=[chunk["metadata"]],
+                        ids=[f"{metadata['filename']}_{chunk['metadata']['article']}_{i}"]
+                    )
+                    logging.info(f"Dodano chunk: {chunk['text'][:100]}...")  # Logowanie pierwszych 100 znaków chunka
+            except Exception as e:
+                logging.error(f"Błąd podczas dodawania chunków do kolekcji: {e}")
         else:
-            logging.warning("Brak chunków do dodania z pliku: %s", filepath)  # Logowanie braku chunków
+            logging.warning(f"Brak chunków do dodania z pliku: {filepath}")
 
         logging.info("Dodano %d chunków z pliku %s", len(chunks), metadata['filename'])
 
@@ -125,23 +157,53 @@ class KodeksProcessor:
         for filename in os.listdir(directory):
             if filename.endswith('.txt'):
                 filepath = os.path.join(directory, filename)
-                logging.info("Przetwarzanie pliku: %s", filepath)  # Logowanie przetwarzania pliku
                 self.process_file(filepath)
         logging.info("Zakończono przetwarzanie plików.")
 
-    def search(self, query: str, n_results: int = 3) -> Dict:
+    def search(self, query: str, n_results: int = 3, filters: Dict = None) -> Dict:
         logging.info("Wyszukiwanie w bazie danych dla zapytania: %s", query)
-        results = self.collection.query(
-            query_texts=[query],
-            n_results=n_results
-        )
-        logging.info("Znaleziono %d wyników dla zapytania: %s", len(results['documents'][0]), query)
-        return results
+        try:
+            results = self.collection.query(
+                query_texts=[query],
+                n_results=n_results,
+                where=filters
+            )
+            logging.info("Znaleziono %d wyników dla zapytania: %s", len(results['documents'][0]), query)
+            return results
+        except Exception as e:
+            logging.error(f"Błąd podczas wyszukiwania: {e}")
+            return {"documents": [[]], "metadatas": [[]], "distances": [[]]}
 
     def list_all_documents(self) -> None:
-        all_docs = self.collection.query(query_texts=[""], n_results=1000)  # Pobierz wszystkie dokumenty
-        if all_docs['documents']:
-            for doc in all_docs['documents'][0]:
-                logging.info("Dokument: %s", doc)
-        else:
-            logging.info("Brak dokumentów w bazie.")
+        try:
+            all_docs = self.collection.get(include=['metadatas'])
+            if all_docs['metadatas']:
+                for metadata in all_docs['metadatas']:
+                    logging.info("Dokument: %s", metadata)
+            else:
+                logging.info("Brak dokumentów w bazie.")
+        except Exception as e:
+            logging.error(f"Błąd podczas listowania dokumentów: {e}")
+
+    def update_document(self, id: str, new_text: str, new_metadata: Dict) -> None:
+        try:
+            self.collection.update(
+                ids=[id],
+                documents=[new_text],
+                metadatas=[new_metadata]
+            )
+            logging.info(f"Zaktualizowano dokument o id: {id}")
+        except Exception as e:
+            logging.error(f"Błąd podczas aktualizacji dokumentu {id}: {e}")
+
+    def delete_document(self, id: str) -> None:
+        try:
+            self.collection.delete(ids=[id])
+            logging.info(f"Usunięto dokument o id: {id}")
+        except Exception as e:
+            logging.error(f"Błąd podczas usuwania dokumentu {id}: {e}")
+
+if __name__ == "__main__":
+    processor = KodeksProcessor()
+    processor.process_all_files("data/kodeksy")
+    processor.list_all_documents()