| import logging |
| import re |
| from pathlib import Path |
|
|
| logger = logging.getLogger(__name__) |
|
|
| MAX_FILE_SIZE = 5 * 1024 * 1024 |
|
|
|
|
| def load_txt(path: str) -> str: |
| return Path(path).read_text(encoding="utf-8", errors="replace") |
|
|
|
|
| def load_md(path: str) -> str: |
| text = Path(path).read_text(encoding="utf-8", errors="replace") |
| |
| text = re.sub(r"^#{1,6}\s+", "", text, flags=re.MULTILINE) |
| text = re.sub(r"\*{1,3}(.+?)\*{1,3}", r"\1", text) |
| text = re.sub(r"`{1,3}[^`]*`{1,3}", "", text, flags=re.DOTALL) |
| text = re.sub(r"!\[.*?\]\(.*?\)", "", text) |
| text = re.sub(r"\[(.+?)\]\(.*?\)", r"\1", text) |
| text = re.sub(r"^\s*[-*+]\s+", "", text, flags=re.MULTILINE) |
| return text |
|
|
|
|
| def load_pdf(path: str) -> str: |
| try: |
| import pdfplumber |
| except ImportError: |
| logger.warning("pdfplumber not installed, cannot load PDF: %s", path) |
| return "" |
|
|
| pages = [] |
| with pdfplumber.open(path) as pdf: |
| for i, page in enumerate(pdf.pages, start=1): |
| page_text = page.extract_text() or "" |
| pages.append(f"--- strona {i} ---\n{page_text}") |
| return "\n".join(pages) |
|
|
|
|
| def load_file(path: str) -> str: |
| p = Path(path) |
| ext = p.suffix.lower() |
| if ext == ".pdf": |
| return load_pdf(path) |
| if ext == ".md": |
| return load_md(path) |
| return load_txt(path) |
|
|
|
|
| def build_corpus(file_paths: list) -> dict[str, list[str]]: |
| corpus: dict[str, list[str]] = {} |
| for fp in file_paths: |
| p = Path(fp) |
| try: |
| size = p.stat().st_size |
| if size > MAX_FILE_SIZE: |
| logger.warning("Plik %s przekracza 5 MB (%d B) - pomijam.", p.name, size) |
| continue |
| text = load_file(str(p)) |
| corpus[p.name] = text.splitlines() |
| except Exception as exc: |
| logger.warning("Nie można wczytać pliku %s: %s", p.name, exc) |
| return corpus |
|
|
