mirror of
https://github.com/migatu/astrololo.git
synced 2026-07-14 21:38:37 +00:00
gitea
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>
53 lines
2.1 KiB
Python
53 lines
2.1 KiB
Python
"""Mapowanie układu kolumn na schemat kanoniczny.
|
|
|
|
Setki plików mogą mieć te same dane pod różnymi nagłówkami i w różnej kolejności
|
|
kolumn ("Imię", "Name", "NAZWA" -> kanoniczne 'name'). Ta warstwa tłumaczy
|
|
faktyczny układ kolumn pliku na wspólny słownik pól, dzięki czemu reszta systemu
|
|
(i przyszła baza SQL) operuje na jednej, stabilnej nazwie pola.
|
|
|
|
Aliasowanie jest świadomie wydzielone i konfigurowalne — to jedyne miejsce do
|
|
edycji, gdy pojawi się nowy wariant nagłówka.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
|
|
# kanoniczne_pole -> zbiór aliasów (po normalizacji)
|
|
CANONICAL_ALIASES: dict[str, set[str]] = {
|
|
"id": {"id", "identyfikator", "nr", "no", "number"},
|
|
"name": {"name", "imie", "nazwa", "nazwisko", "title", "tytul"},
|
|
"symbol": {"symbol", "znak", "sign", "glyph"},
|
|
"date": {"date", "data", "datetime", "timestamp"},
|
|
"value": {"value", "wartosc", "val", "amount", "kwota"},
|
|
"category": {"category", "kategoria", "type", "typ", "group", "grupa"},
|
|
}
|
|
|
|
|
|
def _normalize(col: str) -> str:
|
|
s = str(col).strip().lower()
|
|
s = re.sub(r"[ąàá]", "a", s)
|
|
s = s.replace("ł", "l").replace("ż", "z").replace("ź", "z").replace("ć", "c")
|
|
s = s.replace("ę", "e").replace("ó", "o").replace("ś", "s").replace("ń", "n")
|
|
s = re.sub(r"[^a-z0-9]+", "", s)
|
|
return s
|
|
|
|
|
|
def build_column_mapping(header_cells: list[str]) -> dict[str, str]:
|
|
"""Zwraca mapę pole_kanoniczne -> faktyczna_nazwa_kolumny dla danego pliku.
|
|
|
|
Kolumny nierozpoznane są zachowywane pod swoją (znormalizowaną) nazwą, więc
|
|
nic nie ginie — po prostu nie mają aliasu kanonicznego.
|
|
"""
|
|
reverse: dict[str, str] = {}
|
|
for canonical, aliases in CANONICAL_ALIASES.items():
|
|
for alias in aliases:
|
|
reverse[alias] = canonical
|
|
|
|
mapping: dict[str, str] = {}
|
|
for actual in header_cells:
|
|
norm = _normalize(actual)
|
|
canonical = reverse.get(norm, norm or "col")
|
|
# pierwsze trafienie wygrywa (stabilność przy duplikatach)
|
|
mapping.setdefault(canonical, actual)
|
|
return mapping
|