198 lines
7.7 KiB
Python
198 lines
7.7 KiB
Python
import os
|
||
import sys
|
||
import argparse
|
||
|
||
from udpak_semistruktur.config import valider_yaml
|
||
from udpak_semistruktur.logger import opsaet_logging, hent_logger
|
||
from udpak_semistruktur import ddl
|
||
|
||
from udpak_semistruktur.extract.reader import læs_filer
|
||
from udpak_semistruktur.extract.extractor import generer_datafil
|
||
from udpak_semistruktur.transform.clean import rens, tag_strip, fjern_linjeskift, upper_lower, filename
|
||
from udpak_semistruktur.transform.reshape import flatten, join, where, id_felt, sammensat_noegle
|
||
from udpak_semistruktur.transform.convert import konverter
|
||
from udpak_semistruktur.transform.hash import beregn_hash
|
||
from udpak_semistruktur.load.file_writer import generer_filer_med_overskrifter, skriv_fil_med_retry
|
||
from udpak_semistruktur.load.db_writer import get_ase_connection_windows, insert_rows_ase
|
||
from udpak_semistruktur.db import læs_json_fil
|
||
from udpak_semistruktur.utils import generer_filnavn, load_env_file
|
||
|
||
logger = hent_logger(__name__)
|
||
|
||
def _byg_argument_parser() -> argparse.ArgumentParser:
|
||
"""Bygger og returnerer CLI argument-parseren."""
|
||
|
||
parser = argparse.ArgumentParser(
|
||
description="Udtræk og transformation af semistrukturerede data (JSON/XML)."
|
||
)
|
||
parser.add_argument("--config", required=True, help="Sti til YAML-konfigurationsfil")
|
||
|
||
# Tilføj DDL-flags via ddl-modulet
|
||
ddl.add_cli_args(parser)
|
||
return parser
|
||
|
||
def _skriv_fejl_fil(
|
||
tmp_data: dict,
|
||
fejl_ext: str,
|
||
base_navn: str,
|
||
kolonner: list,
|
||
separator: str,
|
||
encoding: str,
|
||
) -> str | None:
|
||
"""Skriver fejlede rækker til en fejl-fil hvis der er nogen."""
|
||
if not fejl_ext or not tmp_data.get("fejlede_rækker"):
|
||
return
|
||
|
||
fejl_sti = f"{base_navn}{fejl_ext}"
|
||
|
||
def skriv_fejl():
|
||
with open(fejl_sti, "a", encoding=encoding) as f:
|
||
for række in tmp_data["fejlede_rækker"]:
|
||
linje = separator.join(
|
||
str(række.get(k["navn"], "")) for k in kolonner
|
||
)
|
||
f.write(linje + "\n")
|
||
|
||
skriv_fil_med_retry(skriv_fejl, fejl_sti)
|
||
logger.warning(
|
||
f"Fejl-fil: {fejl_sti} skrevet ({len(tmp_data['fejlede_rækker'])} fejlede rækker)"
|
||
)
|
||
|
||
return fejl_sti
|
||
|
||
def _kør_udtræk(config: dict, global_config: dict) -> None:
|
||
"""Kører den normale udtræks- og transformationspipeline."""
|
||
|
||
fejl_filer_skrevet = []
|
||
|
||
input_fil = global_config.get("input_fil")
|
||
input_fil_liste = global_config.get("input_fil_liste")
|
||
|
||
# Opret DB-forbindelse hvis der er tabel-output
|
||
har_tabel_output = any(
|
||
cfg.get("type") == "tabel"
|
||
for cfg in config.get("output_filer", [])
|
||
)
|
||
conn = None
|
||
if har_tabel_output:
|
||
bruger, password, env, host, port = læs_json_fil(global_config)
|
||
conn = get_ase_connection_windows(
|
||
bruger, password, host, port,
|
||
global_config["database"]
|
||
)
|
||
|
||
for record in læs_filer(global_config, input_fil, input_fil_liste):
|
||
|
||
samlet_antal_rækker = 0
|
||
|
||
for cfg in config.get("output_filer", []):
|
||
# 1) Udtræk
|
||
tmp_data = generer_datafil(record, cfg, global_config)
|
||
|
||
# Hent separator og encoding én gang for begge output-typer
|
||
separator = global_config["separator"]
|
||
encoding = global_config["encoding"]
|
||
|
||
# 2) Transform-pipeline
|
||
tmp_data = join(tmp_data, cfg)
|
||
tmp_data = flatten(tmp_data, cfg)
|
||
tmp_data = rens(tmp_data, cfg, global_config)
|
||
tmp_data = tag_strip(tmp_data, cfg, global_config)
|
||
tmp_data = fjern_linjeskift(tmp_data, cfg, global_config)
|
||
tmp_data = where(tmp_data, cfg, global_config)
|
||
tmp_data = konverter(tmp_data, cfg, global_config)
|
||
tmp_data = upper_lower(tmp_data, cfg, global_config)
|
||
tmp_data = id_felt(tmp_data, cfg)
|
||
tmp_data = sammensat_noegle(tmp_data, cfg, global_config)
|
||
tmp_data = beregn_hash(tmp_data, cfg, global_config)
|
||
tmp_data = filename(tmp_data, cfg, global_config)
|
||
|
||
samlet_antal_rækker += len(tmp_data["rækker"])
|
||
|
||
# 3) Load – afhænger af cfg["type"]
|
||
if cfg.get("type") == "fil":
|
||
fil_navn = generer_filnavn(cfg["fil_navn"], global_config)
|
||
output_sti = global_config["output_path"] + fil_navn
|
||
|
||
overskrifter = cfg.get("overskrifter", True)
|
||
generer_filer_med_overskrifter(overskrifter, output_sti, cfg["kolonner"], global_config)
|
||
|
||
# Bemærk: skriv() lukker over loop-variabler – kaldes straks af skriv_fil_med_retry
|
||
def skriv():
|
||
with open(output_sti, "a", encoding=encoding) as f:
|
||
for række in tmp_data["rækker"]:
|
||
linje = separator.join(
|
||
str(række.get(k["navn"], "")) for k in cfg["kolonner"]
|
||
)
|
||
f.write(linje + "\n")
|
||
|
||
skriv_fil_med_retry(skriv, output_sti)
|
||
logger.info(f"Fil: {output_sti} skrevet ({len(tmp_data['rækker'])} rækker)")
|
||
skrevet = _skriv_fejl_fil(tmp_data, global_config.get("fejl_fil_ext"),
|
||
output_sti, cfg["kolonner"], separator, encoding)
|
||
if skrevet:
|
||
fejl_filer_skrevet.append(skrevet)
|
||
|
||
elif cfg.get("type") == "tabel":
|
||
kolonner = [k["navn"] for k in cfg["kolonner"]]
|
||
indsatte, fejlede = insert_rows_ase(conn, cfg["tabel_navn"], kolonner, tmp_data["rækker"])
|
||
logger.info(f"DB: {indsatte} rækker indsat i {cfg['tabel_navn']}")
|
||
|
||
if fejlede:
|
||
logger.warning(f"DB: {len(fejlede)} rækker fejlede i {cfg['tabel_navn']}")
|
||
|
||
fejl_base = os.path.join(global_config["output_path"],
|
||
cfg["tabel_navn"].replace(".", "_"))
|
||
skrevet = _skriv_fejl_fil(tmp_data, global_config.get("fejl_fil_ext"),
|
||
fejl_base, cfg["kolonner"], separator, encoding)
|
||
if skrevet:
|
||
fejl_filer_skrevet.append(skrevet)
|
||
|
||
# Tjek om alle output-filer gav 0 rækker
|
||
if samlet_antal_rækker == 0:
|
||
logger.warning("0 rækker genereret i alle output-filer for dette input.")
|
||
if global_config.get("stop_ved_0_output"):
|
||
logger.error("Stopper kørsel pga. stop_ved_0_output = true.")
|
||
sys.exit(1)
|
||
|
||
if conn is not None:
|
||
conn.close()
|
||
logger.debug("DB-forbindelse lukket.")
|
||
|
||
if fejl_filer_skrevet and global_config.get("stop_ved_fejl", False):
|
||
logger.error(
|
||
f"stop_ved_fejl: {len(fejl_filer_skrevet)} fejl-fil(er) blev oprettet:\n"
|
||
+ "\n".join(f" - {f}" for f in fejl_filer_skrevet)
|
||
)
|
||
sys.exit(2)
|
||
|
||
def main():
|
||
"""Hovedfunktion der eksekveres ved kørsel af scriptet."""
|
||
|
||
# Indlæs lokal .env under udvikling – springes over i kompileret .exe
|
||
if not getattr(sys, "frozen", False):
|
||
load_env_file("testenv.env")
|
||
|
||
parser = _byg_argument_parser()
|
||
args = parser.parse_args()
|
||
|
||
# Valider og indlæs YAML-konfiguration
|
||
config = valider_yaml(args.config)
|
||
global_config = config["config"]
|
||
|
||
# Opsæt logging baseret på argumenter
|
||
opsaet_logging(
|
||
log_fil=global_config["logfil"],
|
||
niveau=global_config.get("log_niveau", "info"),
|
||
log_output=global_config.get("log_output", "begge"),
|
||
)
|
||
|
||
# Eksekver DDL-flowet
|
||
|
||
if ddl.is_enabled(args):
|
||
ddl.run_ddl_mode(args, config, global_config)
|
||
else:
|
||
_kør_udtræk(config, global_config)
|
||
|
||
if __name__ == "__main__":
|
||
main() |