mirror of
https://github.com/migatu/astrololo.git
synced 2026-07-14 21:38:37 +00:00
16d35c16dc
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>
144 lines
5.5 KiB
Python
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)
|