Files
astrololo/services/data/app/providers/excel_provider.py
T
gitea 16d35c16dc Szkielet aplikacji trójwarstwowej (prezentacja / logika / dane)
Trzy niezależne usługi FastAPI komunikujące się przez HTTP/JSON, każda
zna tylko adres warstwy bezpośrednio pod nią:

- presentation (:8000) — strona WWW + formularz
- logic (:8001) — reguły biznesowe, pośrednik
- data (:8002) — wyszukiwanie danych za interfejsem DataProvider

Warstwa danych: czytanie setek plików .xlsx z wykrywaniem nagłówka i
mapowaniem układu kolumn na schemat kanoniczny, z 4-poziomowym cache
(schemat L1, Parquet L2, wyniki zapytań L3, odwrócony indeks L4) i
unieważnianiem po odcisku pliku. Gotowa ścieżka migracji do SQL
(ingest/to_sql.py + SqlDataProvider, przełączane przez DATA_PROVIDER).

Zawiera docker-compose, Makefile, generator danych przykładowych.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-26 17:35:09 +02:00

144 lines
5.5 KiB
Python

"""ExcelDataProvider — wyszukiwanie w setkach plików .xlsx z 4-poziomowym cache.
Ścieżka zapytania (od najszybszej):
1) QueryCache (in-memory) -> gotowy wynik
2) InvertedIndex (SQLite) -> które pliki w ogóle otwierać (zamiast skanu setek)
3) FrameCache (Parquet) -> wczytanie pliku bez parsowania .xlsx
4) SchemaCache (SQLite) -> bez ponownego wykrywania nagłówka/układu kolumn
...dopiero gdy wszystko spudłuje, czytamy .xlsx i wypełniamy cache.
Cała ta złożoność jest UKRYTA za interfejsem DataProvider.
"""
from __future__ import annotations
import time
from pathlib import Path
import pandas as pd
from app.cache.fingerprint import fingerprint
from app.cache.frame_cache import FrameCache
from app.cache.index import InvertedIndex
from app.cache.query_cache import QueryCache
from app.cache.schema_cache import SchemaCache
from app.config import Settings
from app.excel.header_detect import detect_header_row
from app.excel.layout import build_column_mapping
from app.models import HealthInfo, SearchQuery, SearchResult
from app.providers.base import DataProvider
class ExcelDataProvider(DataProvider):
name = "excel"
def __init__(self, settings: Settings) -> None:
self.s = settings
self.schema = SchemaCache(settings.cache_dir)
self.frames = FrameCache(settings.cache_dir)
self.index = InvertedIndex(settings.cache_dir)
self.queries = QueryCache(settings.query_cache_size, settings.query_cache_ttl)
# ---- ładowanie pojedynczego arkusza z pełnym cache ----
def _load_frame(self, path: str, sheet: str | int = 0) -> pd.DataFrame:
fp = fingerprint(path)
sheet_key = str(sheet)
cached = self.frames.get(fp, sheet_key) # poziom 2: Parquet
if cached is not None:
return cached
raw = pd.read_excel(path, sheet_name=sheet, header=None, dtype=object)
meta = self.schema.get(fp, sheet_key) # poziom 1: schemat
if meta is None:
header_row = detect_header_row(raw, self.s.header_scan_rows)
header_cells = [str(c) for c in raw.iloc[header_row].tolist()]
mapping = build_column_mapping(header_cells)
self.schema.put(fp, sheet_key, header_row, mapping)
else:
header_row, mapping = meta
header_cells = [str(c) for c in raw.iloc[header_row].tolist()]
data = raw.iloc[header_row + 1 :].copy()
data.columns = header_cells
data = data.dropna(how="all")
inverse = {orig: canon for canon, orig in mapping.items()}
data = data.rename(columns=inverse).reset_index(drop=True)
self.frames.put(fp, sheet_key, data) # zapisz Parquet na przyszłość
return data
# ---- budowa odwróconego indeksu (warmup / po zmianie pliku) ----
def _ensure_indexed(self, path: str) -> None:
fp = fingerprint(path)
if self.index.file_fingerprint(path) == fp:
return # aktualny
frame = self._load_frame(path)
rows: list[tuple[str, str, str]] = []
for key in self.s.indexed_keys:
if key in frame.columns:
for v in frame[key].dropna().astype(str).unique():
rows.append((key, v, "0"))
self.index.reindex_file(path, fp, rows)
def warmup(self) -> None:
for path in self._excel_files():
self._ensure_indexed(path)
def _excel_files(self) -> list[str]:
base = Path(self.s.excel_dir)
return [str(p) for p in sorted(base.glob("**/*.xlsx")) if not p.name.startswith("~$")]
# ---- publiczne API ----
def search(self, query: SearchQuery) -> SearchResult:
t0 = time.perf_counter()
cache_key = f"{query.key}|{query.value}|{query.exact}|{query.limit}|{query.fields}"
hit = self.queries.get(cache_key) # poziom 3: wynik zapytania
if hit is not None:
hit = hit.model_copy(update={"cache": "hit", "elapsed_ms": _ms(t0)})
return hit
candidates = self.index.lookup(query.key, query.value, query.exact)
if not candidates:
# brak w indeksie (np. klucz nieindeksowany) -> przeszukaj wszystkie pliki
candidates = [(p, "0") for p in self._excel_files()]
rows: list[dict] = []
for path, _sheet in candidates:
frame = self._load_frame(path)
if query.key not in frame.columns:
continue
col = frame[query.key].astype(str)
if query.exact:
mask = col.str.lower() == query.value.lower()
else:
mask = col.str.lower().str.contains(query.value.lower(), na=False)
matched = frame[mask]
if query.fields:
keep = [c for c in query.fields if c in matched.columns]
matched = matched[keep]
rows.extend(matched.to_dict(orient="records"))
if len(rows) >= query.limit:
break
result = SearchResult(
rows=rows[: query.limit],
total=len(rows),
elapsed_ms=_ms(t0),
cache="miss",
provider=self.name,
)
self.queries.put(cache_key, result)
return result
def health(self) -> HealthInfo:
return HealthInfo(
provider=self.name,
indexed_files=self.index.count_files(),
details={"excel_dir": str(self.s.excel_dir), "files_on_disk": len(self._excel_files())},
)
def _ms(t0: float) -> float:
return round((time.perf_counter() - t0) * 1000, 2)