import os import re import logging from typing import List, Dict, Tuple import chromadb from chromadb.utils import embedding_functions from config import EMBEDDING_MODEL, DATABASE_DIR logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) class KodeksProcessor: def __init__(self): logger.info(f"Inicjalizacja klienta bazy danych w katalogu: {DATABASE_DIR}") if not os.path.exists(DATABASE_DIR): os.makedirs(DATABASE_DIR) logger.info(f"Utworzono katalog {DATABASE_DIR}") self.client = chromadb.PersistentClient(path=DATABASE_DIR) logger.info("Klient bazy danych zainicjalizowany") try: self.collection = self.client.get_or_create_collection( name="kodeksy", embedding_function=embedding_functions.SentenceTransformerEmbeddingFunction( model_name=EMBEDDING_MODEL ) ) logger.info("Kolekcja 'kodeksy' pobrana lub utworzona") except Exception as e: logger.error(f"Błąd podczas pobierania lub tworzenia kolekcji: {e}") raise def extract_metadata(self, text: str) -> Dict: metadata = {} 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) if nazwa_match: metadata['data_ustawy'] = nazwa_match.group(1).strip() metadata['nazwa'] = nazwa_match.group(2).strip() 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 ] logger.debug(f"Wyodrębnione metadane: {metadata}") return metadata def split_header_and_content(self, text: str) -> Tuple[str, str]: parts = text.split("USTAWA", 1) if len(parts) > 1: return parts[0], "USTAWA" + parts[1] return "", text def process_article(self, article_text: str) -> Dict: 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) if not paragraphs: return { "article_num": article_num, "content": article_text.strip(), "has_paragraphs": False } return { "article_num": article_num, "paragraphs": paragraphs, "has_paragraphs": True } def split_into_chunks(self, text: str, metadata: Dict) -> List[Dict]: chunks = [] articles = re.split(r'(Art\.\s*\d+[a-z]?)', text) 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 "" processed_article = self.process_article(article_title + " " + article_content) chunk_metadata = { **metadata, "article": processed_article["article_num"] } if processed_article["has_paragraphs"]: for par_num, par_content in processed_article["paragraphs"]: chunk = { "text": f"{article_title} §{par_num}. {par_content.strip()}", "metadata": {**chunk_metadata, "paragraph": par_num} } chunks.append(chunk) logger.debug(f"Utworzono chunk: {chunk['text'][:100]}...") else: chunk = { "text": processed_article["content"], "metadata": chunk_metadata } chunks.append(chunk) logger.debug(f"Utworzono chunk: {chunk['text'][:100]}...") logger.debug(f"Podzielono tekst na {len(chunks)} chunków.") return chunks def process_file(self, filepath: str) -> None: logger.info(f"Rozpoczęcie przetwarzania pliku: {filepath}") try: with open(filepath, 'r', encoding='utf-8') as file: content = file.read() logger.info(f"Odczytano zawartość pliku: {filepath}") 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) logger.info(f"Podzielono plik na {len(chunks)} chunków") if chunks: self.collection.add( documents=[chunk["text"] for chunk in chunks], metadatas=[chunk["metadata"] for chunk in chunks], ids=[f"{metadata['filename']}_{chunk['metadata']['article']}_{i}" for i, chunk in enumerate(chunks)] ) logger.info(f"Dodano {len(chunks)} chunków do kolekcji z pliku {metadata['filename']}") else: logger.warning(f"Brak chunków do dodania z pliku: {filepath}") except Exception as e: logger.error(f"Błąd podczas przetwarzania pliku {filepath}: {e}") def process_all_files(self, directory: str) -> None: logger.info(f"Rozpoczęcie przetwarzania wszystkich plików w katalogu: {directory}") if not os.path.exists(directory): logger.error(f"Katalog {directory} nie istnieje!") return try: files = [f for f in os.listdir(directory) if f.endswith('.txt')] logger.info(f"Znaleziono {len(files)} plików .txt") for filename in files: filepath = os.path.join(directory, filename) logger.info(f"Przetwarzanie pliku: {filepath}") self.process_file(filepath) logger.info("Zakończono przetwarzanie plików.") except Exception as e: logger.error(f"Błąd podczas przetwarzania plików: {e}") def verify_data_loading(self): count = self.collection.count() logger.info(f"Całkowita liczba dokumentów w kolekcji: {count}") if count == 0: logger.warning("Nie załadowano żadnych dokumentów do bazy danych.") def test_search(self): test_queries = ["kodeks karny", "art. 1", "przestępstwo"] for query in test_queries: results = self.search(query) logger.info(f"Zapytanie testowe '{query}' zwróciło {len(results['documents'][0])} wyników") def search(self, query: str, n_results: int = 3) -> Dict: logger.info(f"Wyszukiwanie w bazie danych dla zapytania: {query}") try: results = self.collection.query( query_texts=[query], n_results=n_results ) logger.info(f"Znaleziono {len(results['documents'][0])} wyników dla zapytania: {query}") return results except Exception as e: logger.error(f"Błąd podczas wyszukiwania: {e}") return {"documents": [[]], "metadatas": [[]], "distances": [[]]} def list_all_documents(self) -> None: try: all_docs = self.collection.get(include=['metadatas']) if all_docs['metadatas']: for metadata in all_docs['metadatas']: logger.info(f"Dokument: {metadata}") else: logger.info("Brak dokumentów w bazie.") except Exception as e: logger.error(f"Błąd podczas listowania dokumentów: {e}") if __name__ == "__main__": processor = KodeksProcessor() processor.process_all_files("data/kodeksy") processor.verify_data_loading() processor.test_search() processor.list_all_documents()