181 lines
4.4 KiB
Python
181 lines
4.4 KiB
Python
import os
|
||
from pathlib import Path
|
||
from blake3 import blake3
|
||
import pymysql
|
||
from datetime import date
|
||
|
||
# =========================
|
||
# VÝVOJOVÝ PŘEPÍNAČ
|
||
# =========================
|
||
|
||
RESET_DB = True # !!! POZOR: smaže kapitace a pojištěnce !!!
|
||
|
||
|
||
# =========================
|
||
# KONFIGURACE
|
||
# =========================
|
||
|
||
BASE_PATH = Path(
|
||
r"U:\Dropbox\Ordinace\Dokumentace_ke_zpracování\Výpis pojištěnců"
|
||
)
|
||
|
||
DB_CONFIG = {
|
||
"host": "192.168.1.76",
|
||
"port": 3307,
|
||
"user": "root",
|
||
"password": "Vlado9674+",
|
||
"database": "ordinace",
|
||
"charset": "utf8mb4",
|
||
"autocommit": False,
|
||
}
|
||
|
||
# =========================
|
||
# DB POMOCNÉ FUNKCE
|
||
# =========================
|
||
def reset_kapitace_tables(conn):
|
||
print("!!! RESET_DB=True – mažu data kapitace !!!")
|
||
|
||
with conn.cursor() as cur:
|
||
# kvůli FK
|
||
cur.execute("SET FOREIGN_KEY_CHECKS = 0")
|
||
|
||
cur.execute("TRUNCATE TABLE zp_kapitace_pojistenec")
|
||
cur.execute("TRUNCATE TABLE zp_kapitace_header")
|
||
|
||
cur.execute("SET FOREIGN_KEY_CHECKS = 1")
|
||
|
||
conn.commit()
|
||
print("✔ Kapitace resetována")
|
||
|
||
def get_conn():
|
||
return pymysql.connect(**DB_CONFIG)
|
||
|
||
def blake_exists(cur, blake):
|
||
cur.execute(
|
||
"SELECT id FROM zp_kapitace_header WHERE file_blake3 = %s",
|
||
(blake,),
|
||
)
|
||
return cur.fetchone()
|
||
|
||
# =========================
|
||
# PARSERY
|
||
# =========================
|
||
|
||
def parse_header(line: str):
|
||
return {
|
||
"icp_lekar": line[1:9].strip(),
|
||
"pocet_pojistencu": int(line[9:14]),
|
||
"rok": 2000 + int(line[14:16]),
|
||
"mesic": int(line[16:18]),
|
||
"den": int(line[18:20]),
|
||
}
|
||
|
||
def parse_pojistenec(line: str):
|
||
return {
|
||
"poradi_radku": int(line[1:5]),
|
||
"vekova_skupina": int(line[5:7]),
|
||
"prijmeni": line[7:37].strip(),
|
||
"jmeno": line[37:61].strip(),
|
||
"cislo_pojistence": line[61:71].strip(),
|
||
"kapitace_od": date(
|
||
int(line[75:79]),
|
||
int(line[73:75]),
|
||
int(line[71:73]),
|
||
),
|
||
"zp_kod": line[79:82].strip(),
|
||
}
|
||
|
||
# =========================
|
||
# HLAVNÍ IMPORT
|
||
# =========================
|
||
|
||
def import_file(path: Path):
|
||
raw = path.read_bytes()
|
||
blake = blake3(raw).hexdigest()
|
||
|
||
text = raw.decode("cp852")
|
||
lines = text.splitlines()
|
||
|
||
with get_conn() as conn:
|
||
cur = conn.cursor()
|
||
|
||
if blake_exists(cur, blake):
|
||
print(f"SKIP {path.name} (už existuje)")
|
||
return
|
||
|
||
# ---- HEADER ----
|
||
h = parse_header(lines[0])
|
||
snapshot_date = date(h["rok"], h["mesic"], h["den"])
|
||
|
||
cur.execute(
|
||
"""
|
||
INSERT INTO zp_kapitace_header
|
||
(source_file, zp_kod, icp_lekar, pocet_pojistencu,
|
||
rok, mesic, den, snapshot_date,
|
||
file_blake3, file_content, file_size, file_lines)
|
||
VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)
|
||
""",
|
||
(
|
||
path.name,
|
||
path.name[1:4], # xxx ze jména Fxxx...
|
||
h["icp_lekar"],
|
||
h["pocet_pojistencu"],
|
||
h["rok"],
|
||
h["mesic"],
|
||
h["den"],
|
||
snapshot_date,
|
||
blake,
|
||
text,
|
||
len(raw),
|
||
len(lines),
|
||
),
|
||
)
|
||
|
||
header_id = cur.lastrowid
|
||
|
||
# ---- POJISTENCI ----
|
||
for line in lines[1:]:
|
||
if not line.startswith("I"):
|
||
continue
|
||
|
||
p = parse_pojistenec(line)
|
||
|
||
cur.execute(
|
||
"""
|
||
INSERT INTO zp_kapitace_pojistenec
|
||
(header_id, poradi_radku, vekova_skupina,
|
||
prijmeni, jmeno, cislo_pojistence,
|
||
kapitace_od, zp_kod)
|
||
VALUES (%s,%s,%s,%s,%s,%s,%s,%s)
|
||
""",
|
||
(
|
||
header_id,
|
||
p["poradi_radku"],
|
||
p["vekova_skupina"],
|
||
p["prijmeni"],
|
||
p["jmeno"],
|
||
p["cislo_pojistence"],
|
||
p["kapitace_od"],
|
||
p["zp_kod"],
|
||
),
|
||
)
|
||
|
||
conn.commit()
|
||
print(f"IMPORTED {path.name}")
|
||
|
||
# =========================
|
||
# RUN
|
||
# =========================
|
||
|
||
def main():
|
||
with get_conn() as conn:
|
||
if RESET_DB:
|
||
reset_kapitace_tables(conn)
|
||
|
||
for f in sorted(BASE_PATH.glob("F*.???")):
|
||
import_file(f)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|