Update database.py

database.py

@@ -1,15 +1,21 @@
 import os
 import re
+import logging
 from typing import List, Dict, Tuple
 import chromadb
 from chromadb.utils import embedding_functions
-from config import CHUNK_SIZE, CHUNK_OVERLAP, DATABASE_DIR, EMBEDDING_MODEL
+from config import EMBEDDING_MODEL, DATABASE_DIR
+
+# Konfiguracja logowania
+logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
 
 class KodeksProcessor:
     def __init__(self):
+        logging.info("Inicjalizacja klienta bazy danych...")
         self.client = chromadb.PersistentClient(path=DATABASE_DIR)
         try:
             self.collection = self.client.get_collection("kodeksy")
+            logging.info("Pobrano istniejącą kolekcję 'kodeksy'.")
         except:
             self.collection = self.client.create_collection(
                 name="kodeksy",
@@ -17,19 +23,21 @@ class KodeksProcessor:
                     model_name=EMBEDDING_MODEL
                 )
             )
+            logging.info("Utworzono nową kolekcję 'kodeksy'.")
 
     def extract_metadata(self, text: str) -> Dict:
         metadata = {}
-        dz_u_match = re.search(r'Dz\.U\.(\d{4})\.(\d+)\.(\d+)', text)
+        dz_u_match = re.search(r'Dz\\.U\\.(\\d{4})\\.(\\d+)\\.(\\d+)', text)
         if dz_u_match:
             metadata['dz_u'] = f"Dz.U.{dz_u_match.group(1)}.{dz_u_match.group(2)}.{dz_u_match.group(3)}"
             metadata['rok'] = dz_u_match.group(1)
-        
-        nazwa_match = re.search(r'USTAWA\s+z dnia(.*?)\n(.*?)\n', text)
+
+        nazwa_match = re.search(r'USTAWA\\s+z dnia(.*?)\\n(.*?)\\n', text)
         if nazwa_match:
             metadata['data_ustawy'] = nazwa_match.group(1).strip()
             metadata['nazwa'] = nazwa_match.group(2).strip()
-        
+
+        logging.info("Wydobyto metadane: %s", metadata)
         return metadata
 
     def split_header_and_content(self, text: str) -> Tuple[str, str]:
@@ -39,18 +47,18 @@ 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+)', article_text)
         article_num = art_num_match.group(1) if art_num_match else ""
-        
-        paragraphs = re.findall(r'§\s*(\d+)[.\s]+(.*?)(?=§\s*\d+|$)', article_text, re.DOTALL)
-        
+
+        paragraphs = re.findall(r'§\\s*(\\d+)[.\\s]+(.*?)(?=§\\s*\\d+|$)', article_text, re.DOTALL)
+
         if not paragraphs:
             return {
                 "article_num": article_num,
                 "content": article_text.strip(),
                 "has_paragraphs": False
             }
-        
+
         return {
             "article_num": article_num,
             "paragraphs": paragraphs,
@@ -59,29 +67,29 @@ class KodeksProcessor:
 
     def split_into_chunks(self, text: str, metadata: Dict) -> List[Dict]:
         chunks = []
-        chapters = re.split(r'(Rozdział \d+\n\n[^\\n]+)\n', text)
+        chapters = re.split(r'(Rozdział \\d+\\n\\n[^\\n]+)\\n', text)
         current_chapter = ""
-        
+
         for i, section in enumerate(chapters):
             if section.startswith('Rozdział'):
                 current_chapter = section.strip()
                 continue
-            
-            articles = re.split(r'(Art\.\s*\d+.*?)(?=Art\.\s*\d+|$)', section)
-            
+
+            articles = re.split(r'(Art\\.\\s*\\d+.*?)(?=Art\\.\\s*\\d+|$)', section)
+
             for article in articles:
                 if not article.strip():
                     continue
-                
+
                 if article.startswith('Art.'):
                     processed_article = self.process_article(article)
-                    
+
                     chunk_metadata = {
                         **metadata,
                         "chapter": current_chapter,
                         "article": processed_article["article_num"]
                     }
-                    
+
                     if processed_article["has_paragraphs"]:
                         for par_num, par_content in processed_article["paragraphs"]:
                             chunks.append({
@@ -93,39 +101,43 @@ class KodeksProcessor:
                             "text": processed_article["content"],
                             "metadata": chunk_metadata
                         })
-        
+
+        logging.info("Podzielono tekst na %d chunków.", len(chunks))
         return chunks
 
     def process_file(self, filepath: str) -> None:
-        print(f"Przetwarzanie pliku: {filepath}")
-        
+        logging.info("Przetwarzanie pliku: %s", filepath)
+
         with open(filepath, 'r', encoding='utf-8') as file:
             content = file.read()
-        
+
         header, main_content = self.split_header_and_content(content)
         metadata = self.extract_metadata(main_content)
         metadata['filename'] = os.path.basename(filepath)
-        
+
         chunks = self.split_into_chunks(main_content, metadata)
-        
+
         for i, chunk in enumerate(chunks):
             self.collection.add(
                 documents=[chunk["text"]],
                 metadatas=[chunk["metadata"]],
                 ids=[f"{metadata['filename']}_{chunk['metadata']['article']}_{i}"]
             )
-        
-        print(f"Dodano {len(chunks)} chunków z pliku {metadata['filename']}")
+
+        logging.info("Dodano %d chunków z pliku %s", len(chunks), metadata['filename'])
 
     def process_all_files(self, directory: str) -> None:
+        logging.info("Rozpoczęcie przetwarzania wszystkich plików w katalogu: %s", directory)
         for filename in os.listdir(directory):
             if filename.endswith('.txt'):
                 filepath = os.path.join(directory, filename)
                 self.process_file(filepath)
 
     def search(self, query: str, n_results: int = 3) -> 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