notebookvb
This commit is contained in:
Vladimir Buzalka
@@ -0,0 +1,114 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
import sys as _sys
|
||||
_sys.stdout.reconfigure(encoding="utf-8", errors="replace")
|
||||
_sys.stderr.reconfigure(encoding="utf-8", errors="replace")
|
||||
"""
|
||||
01_parse_seznam.py
|
||||
==================
|
||||
Najde a rozparsuje dávky VZP – Seznam registrovaných pojištěnců (III-1.1.2).
|
||||
Formát souboru: Fxxxmmur.nnn, pevná šířka, kódování CP852.
|
||||
|
||||
Záhlaví (typ H, délka 20):
|
||||
HTYP C 1 0 typ věty 'H'
|
||||
HICP N 8 1 IČP lékaře
|
||||
HPUP N 5 9 počet uznávaných pojištěnců
|
||||
HROK N 2 14 poslední dvojčíslí roku
|
||||
HMES N 2 16 měsíc
|
||||
HDEN N 2 18 den
|
||||
|
||||
Registrace (typ I, délka 82):
|
||||
ITYP C 1 0 typ věty 'I'
|
||||
IPOR N 4 1 pořadové číslo
|
||||
IVS N 2 5 věková skupina
|
||||
IPRI C 30 7 příjmení
|
||||
IJME C 24 37 jméno
|
||||
ICIP C 10 61 číslo pojištěnce
|
||||
IDUOD D 8 71 datum uznání registrace (DDMMRRRR)
|
||||
ICPO C 3 79 kód pojišťovny
|
||||
"""
|
||||
|
||||
import re
|
||||
from pathlib import Path
|
||||
from datetime import date, timedelta
|
||||
|
||||
DAVKY_DIR = Path(r"U:\Dropbox\Ordinace\Dokumentace_ke_zpracování\Zúčtovací zprávy\111 VZP Podání")
|
||||
ENCODING = "cp852"
|
||||
|
||||
|
||||
def parse_davku(path: Path) -> dict:
|
||||
"""Vrátí dict s klíči 'hlavicka' a 'pojistenci'."""
|
||||
lines = path.read_bytes().splitlines()
|
||||
hlavicka = None
|
||||
pojistenci = []
|
||||
|
||||
for raw in lines:
|
||||
line = raw.decode(ENCODING, errors="replace")
|
||||
if not line:
|
||||
continue
|
||||
|
||||
if line[0] == "H":
|
||||
hlavicka = {
|
||||
"icp": line[1:9].strip(),
|
||||
"pocet": int(line[9:14].strip() or 0),
|
||||
"rok": 2000 + int(line[14:16]),
|
||||
"mesic": int(line[16:18]),
|
||||
"den": int(line[18:20]),
|
||||
}
|
||||
|
||||
elif line[0] == "I":
|
||||
if len(line) < 82:
|
||||
continue
|
||||
dat_raw = line[71:79] # DDMMRRRR
|
||||
try:
|
||||
datum = date(int(dat_raw[4:8]), int(dat_raw[2:4]), int(dat_raw[0:2]))
|
||||
except ValueError:
|
||||
datum = None
|
||||
|
||||
pojistenci.append({
|
||||
"por": int(line[1:5].strip() or 0),
|
||||
"vs": line[5:7].strip(),
|
||||
"prijmeni": line[7:37].strip(),
|
||||
"jmeno": line[37:61].strip(),
|
||||
"cip": line[61:71].strip(),
|
||||
"datum_od": datum,
|
||||
"pojistovna": line[79:82].strip(),
|
||||
})
|
||||
|
||||
return {"hlavicka": hlavicka, "pojistenci": pojistenci, "soubor": path.name}
|
||||
|
||||
|
||||
def najdi_davky(adresar: Path) -> list[Path]:
|
||||
"""Vrátí seřazený seznam souborů odpovídajících vzoru Fxxx*.nnn."""
|
||||
return sorted(
|
||||
[p for p in adresar.iterdir()
|
||||
if re.search(r"F\d{3}.*\.\d{3}$", p.name, re.IGNORECASE)],
|
||||
key=lambda p: p.name
|
||||
)
|
||||
|
||||
|
||||
def tiskni_davku(d: dict) -> None:
|
||||
h = d["hlavicka"]
|
||||
if h:
|
||||
print(f"\n=== {d['soubor']} ===")
|
||||
print(f" IČP: {h['icp']} | datum: {h['den']:02d}.{h['mesic']:02d}.{h['rok']} | pojištěnců: {h['pocet']}")
|
||||
print(f" {'#':>4} {'Příjmení':<30} {'Jméno':<24} {'ČIP':<10} {'Datum od':>10} Pojiš.")
|
||||
print(f" {'-'*4} {'-'*30} {'-'*24} {'-'*10} {'-'*10} {'-'*6}")
|
||||
else:
|
||||
print(f"\n=== {d['soubor']} === (záhlaví chybí)")
|
||||
|
||||
for p in d["pojistenci"]:
|
||||
datum_str = p["datum_od"].strftime("%d.%m.%Y") if p["datum_od"] else "?"
|
||||
print(f" {p['por']:>4} {p['prijmeni']:<30} {p['jmeno']:<24} {p['cip']:<10} {datum_str:>10} {p['pojistovna']}")
|
||||
|
||||
|
||||
# ── MAIN ──────────────────────────────────────────────────────────────────────
|
||||
|
||||
davky = najdi_davky(DAVKY_DIR)
|
||||
print(f"Nalezeno dávek: {len(davky)}")
|
||||
|
||||
for cesta in davky:
|
||||
data = parse_davku(cesta)
|
||||
tiskni_davku(data)
|
||||
|
||||
print(f"\nCelkem dávek: {len(davky)}")
|
||||
@@ -0,0 +1,165 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
import sys as _sys
|
||||
_sys.stdout.reconfigure(encoding="utf-8", errors="replace")
|
||||
_sys.stderr.reconfigure(encoding="utf-8", errors="replace")
|
||||
"""
|
||||
02_import_do_db.py
|
||||
==================
|
||||
Vytvoří tabulku seznam_pojistencu_davky v medevio DB a naimportuje
|
||||
všechny dávky F111*.nnn ze složky Podání.
|
||||
|
||||
Spuštění:
|
||||
python 02_import_do_db.py # import všech dávek
|
||||
python 02_import_do_db.py --reset # smaže a znovu vytvoří tabulku před importem
|
||||
"""
|
||||
|
||||
import re
|
||||
import sys
|
||||
import argparse
|
||||
from pathlib import Path
|
||||
from datetime import date
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "Knihovny"))
|
||||
from mysql_db import connect_mysql
|
||||
|
||||
DAVKY_DIR = Path(r"U:\Dropbox\Ordinace\Dokumentace_ke_zpracování\Zúčtovací zprávy\111 VZP Podání")
|
||||
ENCODING = "cp852"
|
||||
|
||||
CREATE_SQL = """
|
||||
CREATE TABLE IF NOT EXISTS seznam_pojistencu_davky (
|
||||
id INT NOT NULL AUTO_INCREMENT PRIMARY KEY,
|
||||
soubor VARCHAR(60) NOT NULL COMMENT 'Název souboru dávky',
|
||||
icp VARCHAR(8) NOT NULL COMMENT 'IČP lékaře (HICP)',
|
||||
davka_rok SMALLINT NOT NULL COMMENT 'Rok dávky (HROK)',
|
||||
davka_mesic TINYINT NOT NULL COMMENT 'Měsíc dávky (HMES)',
|
||||
davka_den TINYINT NOT NULL COMMENT 'Den pořízení seznamu (HDEN)',
|
||||
por SMALLINT NOT NULL COMMENT 'Pořadové číslo v dávce (IPOR)',
|
||||
vs VARCHAR(2) NOT NULL COMMENT 'Věková skupina (IVS)',
|
||||
prijmeni VARCHAR(30) NOT NULL COMMENT 'Příjmení pojištěnce (IPRI)',
|
||||
jmeno VARCHAR(24) NOT NULL COMMENT 'Jméno pojištěnce (IJME)',
|
||||
cip VARCHAR(10) NOT NULL COMMENT 'Číslo pojištěnce (ICIP)',
|
||||
datum_od DATE NULL COMMENT 'Datum uznání registrace (IDUOD)',
|
||||
pojistovna VARCHAR(3) NOT NULL COMMENT 'Kód pojišťovny (ICPO)',
|
||||
vytvoreno TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
||||
UNIQUE KEY uq_soubor_por (soubor, por)
|
||||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_czech_ci
|
||||
COMMENT='VZP seznam registrovaných pojištěnců III-1.1.2';
|
||||
"""
|
||||
|
||||
|
||||
def parse_davku(path: Path) -> dict:
|
||||
lines = path.read_bytes().splitlines()
|
||||
hlavicka = None
|
||||
pojistenci = []
|
||||
|
||||
for raw in lines:
|
||||
line = raw.decode(ENCODING, errors="replace")
|
||||
if not line:
|
||||
continue
|
||||
|
||||
if line[0] == "H":
|
||||
hlavicka = {
|
||||
"icp": line[1:9].strip(),
|
||||
"pocet": int(line[9:14].strip() or 0),
|
||||
"rok": 2000 + int(line[14:16]),
|
||||
"mesic": int(line[16:18]),
|
||||
"den": int(line[18:20]),
|
||||
}
|
||||
|
||||
elif line[0] == "I":
|
||||
if len(line) < 82:
|
||||
continue
|
||||
dat_raw = line[71:79] # DDMMRRRR
|
||||
try:
|
||||
datum = date(int(dat_raw[4:8]), int(dat_raw[2:4]), int(dat_raw[0:2]))
|
||||
except ValueError:
|
||||
datum = None
|
||||
|
||||
pojistenci.append({
|
||||
"por": int(line[1:5].strip() or 0),
|
||||
"vs": line[5:7].strip(),
|
||||
"prijmeni": line[7:37].strip(),
|
||||
"jmeno": line[37:61].strip(),
|
||||
"cip": line[61:71].strip(),
|
||||
"datum_od": datum,
|
||||
"pojistovna": line[79:82].strip(),
|
||||
})
|
||||
|
||||
return {"hlavicka": hlavicka, "pojistenci": pojistenci, "soubor": path.name}
|
||||
|
||||
|
||||
def najdi_davky(adresar: Path) -> list[Path]:
|
||||
return sorted(
|
||||
[p for p in adresar.iterdir()
|
||||
if re.search(r"F\d{3}.*\.\d{3}$", p.name, re.IGNORECASE)],
|
||||
key=lambda p: p.name
|
||||
)
|
||||
|
||||
|
||||
def import_davku(cur, davka: dict) -> tuple[int, int]:
|
||||
"""Vrátí (vloženo, přeskočeno duplicit)."""
|
||||
h = davka["hlavicka"]
|
||||
if not h:
|
||||
print(f" SKIP {davka['soubor']} — chybí záhlaví")
|
||||
return 0, 0
|
||||
|
||||
rows = [
|
||||
(davka["soubor"], h["icp"], h["rok"], h["mesic"], h["den"],
|
||||
p["por"], p["vs"], p["prijmeni"], p["jmeno"],
|
||||
p["cip"], p["datum_od"], p["pojistovna"])
|
||||
for p in davka["pojistenci"]
|
||||
]
|
||||
|
||||
# INSERT IGNORE přeskočí duplicity (unikátní klíč soubor+por) bez chyby
|
||||
affected = cur.executemany(
|
||||
"""INSERT IGNORE INTO seznam_pojistencu_davky
|
||||
(soubor, icp, davka_rok, davka_mesic, davka_den,
|
||||
por, vs, prijmeni, jmeno, cip, datum_od, pojistovna)
|
||||
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)""",
|
||||
rows
|
||||
)
|
||||
vlozeno = affected if affected else 0
|
||||
preskoceno = len(rows) - vlozeno
|
||||
return vlozeno, preskoceno
|
||||
|
||||
|
||||
# ── MAIN ──────────────────────────────────────────────────────────────────────
|
||||
|
||||
parser = argparse.ArgumentParser(description="Import VZP dávek do MySQL")
|
||||
parser.add_argument("--reset", action="store_true",
|
||||
help="Smaže a znovu vytvoří tabulku před importem")
|
||||
args = parser.parse_args()
|
||||
|
||||
conn = connect_mysql()
|
||||
cur = conn.cursor()
|
||||
|
||||
if args.reset:
|
||||
print("DROP TABLE seznam_pojistencu_davky ...")
|
||||
cur.execute("DROP TABLE IF EXISTS seznam_pojistencu_davky")
|
||||
|
||||
print("Vytváření tabulky (pokud neexistuje) ...")
|
||||
cur.execute(CREATE_SQL)
|
||||
|
||||
davky = najdi_davky(DAVKY_DIR)
|
||||
print(f"Nalezeno dávek: {len(davky)}\n")
|
||||
|
||||
celkem_vlozeno = celkem_preskoceno = 0
|
||||
|
||||
for cesta in davky:
|
||||
davka = parse_davku(cesta)
|
||||
h = davka["hlavicka"]
|
||||
if h:
|
||||
print(f"{davka['soubor']} ({h['den']:02d}.{h['mesic']:02d}.{h['rok']}, {len(davka['pojistenci'])} záznamů)")
|
||||
else:
|
||||
print(f"{davka['soubor']} (záhlaví chybí)")
|
||||
|
||||
vlozeno, preskoceno = import_davku(cur, davka)
|
||||
celkem_vlozeno += vlozeno
|
||||
celkem_preskoceno += preskoceno
|
||||
print(f" → vloženo: {vlozeno}, duplicit přeskočeno: {preskoceno}")
|
||||
|
||||
cur.close()
|
||||
conn.close()
|
||||
|
||||
print(f"\nHotovo. Celkem vloženo: {celkem_vlozeno}, přeskočeno: {celkem_preskoceno}")
|
||||
@@ -0,0 +1,226 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
import sys as _sys
|
||||
_sys.stdout.reconfigure(encoding="utf-8", errors="replace")
|
||||
_sys.stderr.reconfigure(encoding="utf-8", errors="replace")
|
||||
"""
|
||||
03_porovnani_davek.py
|
||||
=====================
|
||||
Porovná všechny po sobě jdoucí dávky VZP a vytvoří Excel se třemi listy:
|
||||
1. Přehled – souhrnná tabulka přechodů (kolik odešlo / přibylo)
|
||||
2. Odešli – detail všech, kdo v dané dávce chyběli oproti předchozí
|
||||
3. Přibylo – detail všech, kdo v dané dávce přibyly oproti předchozí
|
||||
|
||||
Pro data s více soubory se použijí unikátní CIPy (dávky jsou totožné).
|
||||
"""
|
||||
|
||||
import sys
|
||||
from datetime import date
|
||||
from pathlib import Path
|
||||
from collections import defaultdict
|
||||
|
||||
import openpyxl
|
||||
from openpyxl.styles import Font, PatternFill, Alignment, Border, Side
|
||||
from openpyxl.utils import get_column_letter
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "Knihovny"))
|
||||
from mysql_db import connect_mysql
|
||||
|
||||
OUTPUT = Path(__file__).parent / "porovnani_davek.xlsx"
|
||||
|
||||
# ── Barvy ─────────────────────────────────────────────────────────────────────
|
||||
CLR_HEADER = "1F4E79" # tmavě modrá
|
||||
CLR_SUBHDR = "2E75B6" # střední modrá
|
||||
CLR_ODEŠLI = "FCE4D6" # světle lososová
|
||||
CLR_PŘIBYLO = "E2EFDA" # světle zelená
|
||||
CLR_ZEBRA = "F2F2F2" # šedá pro zebra řádky
|
||||
CLR_SUMMARY = "DEEAF1" # světle modrá pro souhrn
|
||||
|
||||
# ── Styly ─────────────────────────────────────────────────────────────────────
|
||||
def hdr_font(white=True):
|
||||
return Font(bold=True, color="FFFFFF" if white else "000000", size=11)
|
||||
|
||||
def cell_border():
|
||||
thin = Side(style="thin", color="BFBFBF")
|
||||
return Border(left=thin, right=thin, top=thin, bottom=thin)
|
||||
|
||||
def set_header(cell, text, bg=CLR_HEADER, white=True):
|
||||
cell.value = text
|
||||
cell.font = hdr_font(white)
|
||||
cell.fill = PatternFill("solid", fgColor=bg)
|
||||
cell.alignment = Alignment(horizontal="center", vertical="center", wrap_text=True)
|
||||
cell.border = cell_border()
|
||||
|
||||
def set_cell(cell, value, bg=None, bold=False, align="left", num_fmt=None):
|
||||
cell.value = value
|
||||
cell.font = Font(bold=bold, size=10)
|
||||
cell.alignment = Alignment(horizontal=align, vertical="center")
|
||||
cell.border = cell_border()
|
||||
if bg:
|
||||
cell.fill = PatternFill("solid", fgColor=bg)
|
||||
if num_fmt:
|
||||
cell.number_format = num_fmt
|
||||
|
||||
# ── Načtení dat z DB ──────────────────────────────────────────────────────────
|
||||
|
||||
print("Připojuji se k DB ...")
|
||||
conn = connect_mysql()
|
||||
cur = conn.cursor()
|
||||
|
||||
# Unikátní CIPy per datum + jméno (z prvního souboru daného data)
|
||||
cur.execute("""
|
||||
SELECT davka_rok, davka_mesic, davka_den, cip,
|
||||
MIN(prijmeni) AS prijmeni, MIN(jmeno) AS jmeno
|
||||
FROM seznam_pojistencu_davky
|
||||
GROUP BY davka_rok, davka_mesic, davka_den, cip
|
||||
ORDER BY davka_rok, davka_mesic, davka_den
|
||||
""")
|
||||
rows = cur.fetchall()
|
||||
conn.close()
|
||||
|
||||
# Seskupení do dict: datum -> {cip: (prijmeni, jmeno)}
|
||||
davky: dict[date, dict[str, tuple]] = defaultdict(dict)
|
||||
for rok, mes, den, cip, pri, jme in rows:
|
||||
d = date(rok, mes, den)
|
||||
davky[d][cip] = (pri, jme)
|
||||
|
||||
data = sorted(davky.keys())
|
||||
print(f"Nalezeno {len(data)} unikátních datumů dávek.")
|
||||
|
||||
# ── Porovnání ────────────────────────────────────────────────────────────────
|
||||
|
||||
prehled = [] # (dat_od, dat_do, stav_od, odešlo, přibylo, stav_do)
|
||||
odešli = [] # (dat_do, cip, prijmeni, jmeno)
|
||||
přibylo = [] # (dat_do, cip, prijmeni, jmeno)
|
||||
|
||||
for i in range(1, len(data)):
|
||||
d1, d2 = data[i-1], data[i]
|
||||
cip1 = set(davky[d1])
|
||||
cip2 = set(davky[d2])
|
||||
|
||||
vyšli = cip1 - cip2
|
||||
vstou = cip2 - cip1
|
||||
|
||||
prehled.append((d1, d2, len(cip1), len(vyšli), len(vstou), len(cip2)))
|
||||
|
||||
for cip in sorted(vyšli):
|
||||
pri, jme = davky[d1][cip]
|
||||
odešli.append((d2, cip, pri, jme))
|
||||
|
||||
for cip in sorted(vstou):
|
||||
pri, jme = davky[d2][cip]
|
||||
přibylo.append((d2, cip, pri, jme))
|
||||
|
||||
# ── Excel ─────────────────────────────────────────────────────────────────────
|
||||
|
||||
wb = openpyxl.Workbook()
|
||||
|
||||
# ── List 1: Přehled ───────────────────────────────────────────────────────────
|
||||
ws = wb.active
|
||||
ws.title = "Přehled"
|
||||
ws.freeze_panes = "A3"
|
||||
|
||||
# Nadpis
|
||||
ws.merge_cells("A1:F1")
|
||||
t = ws["A1"]
|
||||
t.value = "Porovnání po sobě jdoucích dávek VZP – seznam registrovaných pojištěnců"
|
||||
t.font = Font(bold=True, size=13, color="FFFFFF")
|
||||
t.fill = PatternFill("solid", fgColor=CLR_HEADER)
|
||||
t.alignment = Alignment(horizontal="center", vertical="center")
|
||||
ws.row_dimensions[1].height = 28
|
||||
|
||||
# Záhlaví sloupců
|
||||
headers = ["Datum od", "Datum do", "Stav\nna začátku", "Odešlo", "Přibylo", "Stav\nna konci"]
|
||||
for col, h in enumerate(headers, 1):
|
||||
set_header(ws.cell(2, col), h)
|
||||
ws.row_dimensions[2].height = 32
|
||||
|
||||
# Data
|
||||
for row_i, (d1, d2, stav_od, vyšli, vstou, stav_do) in enumerate(prehled, 3):
|
||||
bg = CLR_ZEBRA if row_i % 2 == 0 else None
|
||||
set_cell(ws.cell(row_i, 1), d1.strftime("%d.%m.%Y"), bg, align="center")
|
||||
set_cell(ws.cell(row_i, 2), d2.strftime("%d.%m.%Y"), bg, align="center")
|
||||
set_cell(ws.cell(row_i, 3), stav_od, bg, align="right")
|
||||
set_cell(ws.cell(row_i, 4), -vyšli, CLR_ODEŠLI if vyšli else bg, bold=bool(vyšli), align="right")
|
||||
set_cell(ws.cell(row_i, 5), vstou, CLR_PŘIBYLO if vstou else bg, bold=bool(vstou), align="right")
|
||||
set_cell(ws.cell(row_i, 6), stav_do, bg, align="right")
|
||||
|
||||
# Souhrnný řádek
|
||||
sr = len(prehled) + 3
|
||||
ws.merge_cells(f"A{sr}:B{sr}")
|
||||
sc = ws.cell(sr, 1)
|
||||
sc.value = "CELKEM pohybů"
|
||||
sc.font = Font(bold=True, size=10)
|
||||
sc.fill = PatternFill("solid", fgColor=CLR_SUMMARY)
|
||||
sc.alignment = Alignment(horizontal="center", vertical="center")
|
||||
sc.border = cell_border()
|
||||
|
||||
celk_odešlo = sum(p[3] for p in prehled)
|
||||
celk_přibylo = sum(p[4] for p in prehled)
|
||||
set_cell(ws.cell(sr, 3), "", CLR_SUMMARY)
|
||||
set_cell(ws.cell(sr, 4), -celk_odešlo, CLR_SUMMARY, bold=True, align="right")
|
||||
set_cell(ws.cell(sr, 5), celk_přibylo, CLR_SUMMARY, bold=True, align="right")
|
||||
set_cell(ws.cell(sr, 6), "", CLR_SUMMARY)
|
||||
|
||||
# Šířky sloupců
|
||||
for col, w in zip(range(1, 7), [14, 14, 14, 10, 10, 12]):
|
||||
ws.column_dimensions[get_column_letter(col)].width = w
|
||||
|
||||
# ── List 2: Odešli ────────────────────────────────────────────────────────────
|
||||
ws2 = wb.create_sheet("Odešli")
|
||||
ws2.freeze_panes = "A3"
|
||||
|
||||
ws2.merge_cells("A1:D1")
|
||||
t2 = ws2["A1"]
|
||||
t2.value = f"Pacienti, kteří odešli – celkem {len(odešli)} pohybů"
|
||||
t2.font = Font(bold=True, size=13, color="FFFFFF")
|
||||
t2.fill = PatternFill("solid", fgColor="C55A11")
|
||||
t2.alignment = Alignment(horizontal="center", vertical="center")
|
||||
ws2.row_dimensions[1].height = 28
|
||||
|
||||
for col, h in enumerate(["Datum dávky", "Číslo pojištěnce", "Příjmení", "Jméno"], 1):
|
||||
set_header(ws2.cell(2, col), h, bg="C55A11")
|
||||
ws2.row_dimensions[2].height = 24
|
||||
|
||||
for row_i, (dat, cip, pri, jme) in enumerate(odešli, 3):
|
||||
bg = CLR_ODEŠLI if row_i % 2 == 0 else None
|
||||
set_cell(ws2.cell(row_i, 1), dat.strftime("%d.%m.%Y"), bg, align="center")
|
||||
set_cell(ws2.cell(row_i, 2), cip, bg, align="center")
|
||||
set_cell(ws2.cell(row_i, 3), pri, bg)
|
||||
set_cell(ws2.cell(row_i, 4), jme, bg)
|
||||
|
||||
for col, w in zip(range(1, 5), [14, 16, 28, 22]):
|
||||
ws2.column_dimensions[get_column_letter(col)].width = w
|
||||
|
||||
# ── List 3: Přibylo ───────────────────────────────────────────────────────────
|
||||
ws3 = wb.create_sheet("Přibylo")
|
||||
ws3.freeze_panes = "A3"
|
||||
|
||||
ws3.merge_cells("A1:D1")
|
||||
t3 = ws3["A1"]
|
||||
t3.value = f"Pacienti, kteří přibylo – celkem {len(přibylo)} pohybů"
|
||||
t3.font = Font(bold=True, size=13, color="FFFFFF")
|
||||
t3.fill = PatternFill("solid", fgColor="375623")
|
||||
t3.alignment = Alignment(horizontal="center", vertical="center")
|
||||
ws3.row_dimensions[1].height = 28
|
||||
|
||||
for col, h in enumerate(["Datum dávky", "Číslo pojištěnce", "Příjmení", "Jméno"], 1):
|
||||
set_header(ws3.cell(2, col), h, bg="375623")
|
||||
ws3.row_dimensions[2].height = 24
|
||||
|
||||
for row_i, (dat, cip, pri, jme) in enumerate(přibylo, 3):
|
||||
bg = CLR_PŘIBYLO if row_i % 2 == 0 else None
|
||||
set_cell(ws3.cell(row_i, 1), dat.strftime("%d.%m.%Y"), bg, align="center")
|
||||
set_cell(ws3.cell(row_i, 2), cip, bg, align="center")
|
||||
set_cell(ws3.cell(row_i, 3), pri, bg)
|
||||
set_cell(ws3.cell(row_i, 4), jme, bg)
|
||||
|
||||
for col, w in zip(range(1, 5), [14, 16, 28, 22]):
|
||||
ws3.column_dimensions[get_column_letter(col)].width = w
|
||||
|
||||
# ── Uložení ───────────────────────────────────────────────────────────────────
|
||||
wb.save(OUTPUT)
|
||||
print(f"\nExcel uložen: {OUTPUT}")
|
||||
print(f" Přehled: {len(prehled)} přechodů")
|
||||
print(f" Odešli: {len(odešli)} pohybů")
|
||||
print(f" Přibylo: {len(přibylo)} pohybů")
|
||||
@@ -0,0 +1,233 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
import sys as _sys
|
||||
_sys.stdout.reconfigure(encoding="utf-8", errors="replace")
|
||||
_sys.stderr.reconfigure(encoding="utf-8", errors="replace")
|
||||
"""
|
||||
04_najdi_zlomy.py
|
||||
=================
|
||||
Pro pacienty z seznam_pojistencu_davky, kteří NEMAJÍ záznam v vzp_registrace_lekari
|
||||
(skript kdojelekar je nezachytil), najde bod zlomu registrace u naší ambulance.
|
||||
|
||||
Algoritmus:
|
||||
1. Dotaz VZP dnes — je pacient stále registrován u nás (ICP=09305001)?
|
||||
ANO → datum_ukonceni z odpovědi = výsledek
|
||||
2. NE → hledáme rokem dozadu (od dnes, −1 rok, −2 roky …)
|
||||
dokud nenajdeme rok kdy BYL registrován → tím ohraničíme interval [lo, hi]
|
||||
3. V intervalu [lo, hi] binární hledání na den přesně
|
||||
(nebo pokud datum_ukonceni z VZP odpovědi není 3000, použijeme ho přímo)
|
||||
|
||||
Výsledek se uloží do tabulky seznam_pojistencu_zlomy a vytiskne na konzoli.
|
||||
"""
|
||||
|
||||
import time
|
||||
import sys
|
||||
from datetime import date, timedelta
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "Knihovny"))
|
||||
from mysql_db import connect_mysql
|
||||
from vzpb2b_client import VZPB2BClient
|
||||
|
||||
# ── Konfigurace ───────────────────────────────────────────────────────────────
|
||||
PFX_PATH = str(Path(__file__).resolve().parents[1] / "Certificates" / "picka.pfx")
|
||||
PFX_PASSWORD = "Vlado7309208104+"
|
||||
ICZ = "09305000"
|
||||
NASA_ICP = "09305001"
|
||||
API_PAUSE = 2 # sekundy mezi VZP dotazy
|
||||
|
||||
CREATE_SQL = """
|
||||
CREATE TABLE IF NOT EXISTS seznam_pojistencu_zlomy (
|
||||
id INT NOT NULL AUTO_INCREMENT PRIMARY KEY,
|
||||
cip VARCHAR(12) NOT NULL,
|
||||
prijmeni VARCHAR(30) NOT NULL,
|
||||
jmeno VARCHAR(24) NOT NULL,
|
||||
posledni_davka DATE NOT NULL COMMENT 'Poslední měsíc kdy byl v dávce',
|
||||
zlom_datum DATE NULL COMMENT 'Poslední den registrace u nás (NULL=stále aktivní)',
|
||||
zlom_zdroj VARCHAR(60) NULL COMMENT 'Jak byl zlom určen',
|
||||
stav VARCHAR(20) NOT NULL COMMENT 'aktivní / ukončen / nenalezen',
|
||||
dotazeno_dne DATE NOT NULL,
|
||||
UNIQUE KEY uq_cip (cip)
|
||||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='Zlomy registrace pro pacienty bez záznamu v kdojelekar';
|
||||
"""
|
||||
|
||||
# ── VZP klient ────────────────────────────────────────────────────────────────
|
||||
vzp = VZPB2BClient("prod", PFX_PATH, PFX_PASSWORD, icz=ICZ)
|
||||
|
||||
|
||||
def je_registrovan(rc: str, k_datu: date) -> tuple[bool, date | None]:
|
||||
"""
|
||||
Vrátí (registrován_u_nas: bool, datum_ukonceni: date|None).
|
||||
datum_ukonceni = None pokud nelze parsovat nebo pacient není u nás.
|
||||
"""
|
||||
xml = vzp.registrace_lekare(rc, k_datu.isoformat(), odbornosti=["001"])
|
||||
time.sleep(API_PAUSE)
|
||||
try:
|
||||
zaznamy = vzp.parse_registrace_lekare(xml)
|
||||
except Exception as e:
|
||||
print(f" [CHYBA parsování] {e}")
|
||||
return False, None
|
||||
|
||||
for z in zaznamy:
|
||||
if z.get("ma_lekare") and z.get("ICP") == NASA_ICP and z.get("kod_odbornosti") == "001":
|
||||
du_str = z.get("datum_ukonceni")
|
||||
try:
|
||||
du = date.fromisoformat(du_str) if du_str else None
|
||||
except ValueError:
|
||||
du = None
|
||||
return True, du
|
||||
|
||||
return False, None
|
||||
|
||||
|
||||
def najdi_zlom(rc: str, posledni_davka: date) -> tuple[date | None, str, str]:
|
||||
"""
|
||||
Vrátí (zlom_datum, zlom_zdroj, stav).
|
||||
zlom_datum = poslední den kdy byl registrován (None = stále aktivní).
|
||||
"""
|
||||
today = date.today()
|
||||
|
||||
# ── Krok 1: dotaz dnes ────────────────────────────────────────────────────
|
||||
print(f" [dnes {today}]", end=" ", flush=True)
|
||||
reg, du = je_registrovan(rc, today)
|
||||
|
||||
if reg:
|
||||
if du and du.year < 3000:
|
||||
print(f"registrován, ukončení {du}")
|
||||
return du, "VZP datum_ukonceni (dnes)", "ukončen"
|
||||
else:
|
||||
print("registrován, bez data ukončení → stále aktivní")
|
||||
return None, "VZP dnes aktivní", "aktivní"
|
||||
|
||||
print("NENÍ registrován")
|
||||
|
||||
# ── Krok 2: hledání po rocích dozadu ─────────────────────────────────────
|
||||
# lo = víme, že tam BYL (posledni_davka)
|
||||
# hi = víme, že tam NENÍ (today)
|
||||
lo: date = posledni_davka
|
||||
hi: date = today
|
||||
|
||||
probe = today.replace(year=today.year - 1)
|
||||
while probe >= posledni_davka:
|
||||
print(f" [rok {probe}]", end=" ", flush=True)
|
||||
reg_p, du_p = je_registrovan(rc, probe)
|
||||
if reg_p:
|
||||
lo = probe
|
||||
print(f"registrován")
|
||||
if du_p and du_p.year < 3000:
|
||||
print(f" → datum_ukonceni z VZP: {du_p}")
|
||||
return du_p, f"VZP datum_ukonceni (dotaz {probe})", "ukončen"
|
||||
break
|
||||
else:
|
||||
hi = probe
|
||||
print("není")
|
||||
try:
|
||||
probe = probe.replace(year=probe.year - 1)
|
||||
except ValueError:
|
||||
break
|
||||
else:
|
||||
# Ani v posledni_davka není registrován — neobvyklé
|
||||
print(f" ! Ani k datu {posledni_davka} není registrován — zkouším přímo")
|
||||
reg_lo, du_lo = je_registrovan(rc, posledni_davka)
|
||||
if not reg_lo:
|
||||
return None, "nenalezen ani k datu poslední dávky", "nenalezen"
|
||||
lo = posledni_davka
|
||||
if du_lo and du_lo.year < 3000:
|
||||
return du_lo, f"VZP datum_ukonceni ({posledni_davka})", "ukončen"
|
||||
|
||||
# ── Krok 3: binární hledání v intervalu [lo, hi] ─────────────────────────
|
||||
print(f" Binární hledání: {lo} … {hi}")
|
||||
iterace = 0
|
||||
while (hi - lo).days > 1:
|
||||
iterace += 1
|
||||
mid = lo + timedelta(days=(hi - lo).days // 2)
|
||||
print(f" [{iterace}. iterace: {mid}]", end=" ", flush=True)
|
||||
reg_m, du_m = je_registrovan(rc, mid)
|
||||
if reg_m:
|
||||
lo = mid
|
||||
print("registrován")
|
||||
if du_m and du_m.year < 3000:
|
||||
print(f" → datum_ukonceni z VZP: {du_m}")
|
||||
return du_m, f"VZP datum_ukonceni (binární {mid})", "ukončen"
|
||||
else:
|
||||
hi = mid
|
||||
print("není")
|
||||
|
||||
print(f" → Zlom: poslední den registrace = {lo}")
|
||||
return lo, f"binární hledání ({iterace} kroků)", "ukončen"
|
||||
|
||||
|
||||
# ── Načtení pacientů ──────────────────────────────────────────────────────────
|
||||
|
||||
conn = connect_mysql()
|
||||
cur = conn.cursor()
|
||||
cur.execute(CREATE_SQL)
|
||||
|
||||
# Unikátní CIP v seznamu (VZP, pojišťovna 111)
|
||||
cur.execute("SELECT DISTINCT cip FROM seznam_pojistencu_davky WHERE pojistovna='111'")
|
||||
vsechny_cip = {r[0] for r in cur.fetchall()}
|
||||
|
||||
# CIP které jsou v registrace_lekari (u nás, odb 001)
|
||||
cur.execute("""
|
||||
SELECT DISTINCT rc FROM vzp_registrace_lekari
|
||||
WHERE kod_odbornosti='001' AND ICP='09305001' AND ma_lekare=1
|
||||
""")
|
||||
zname_cip = {r[0] for r in cur.fetchall()}
|
||||
|
||||
# Nespárované
|
||||
nesparovane_cip = vsechny_cip - zname_cip
|
||||
|
||||
# Doplním jméno a posledni_davka
|
||||
cur.execute("""
|
||||
SELECT cip, MIN(prijmeni), MIN(jmeno),
|
||||
MAX(DATE(CONCAT(davka_rok, '-', LPAD(davka_mesic,2,'0'), '-01')))
|
||||
FROM seznam_pojistencu_davky
|
||||
WHERE pojistovna='111'
|
||||
GROUP BY cip
|
||||
""")
|
||||
info = {r[0]: (r[1], r[2], r[3]) for r in cur.fetchall()}
|
||||
|
||||
pacienti = [
|
||||
(cip, *info[cip])
|
||||
for cip in sorted(nesparovane_cip)
|
||||
if cip in info
|
||||
]
|
||||
|
||||
print(f"Pacientů ke zpracování: {len(pacienti)}\n")
|
||||
print("=" * 70)
|
||||
|
||||
# ── Hlavní smyčka ─────────────────────────────────────────────────────────────
|
||||
|
||||
vysledky = []
|
||||
for cip, prijmeni, jmeno, posledni_davka in pacienti:
|
||||
print(f"\n{prijmeni} {jmeno} (CIP: {cip}, poslední dávka: {posledni_davka})")
|
||||
try:
|
||||
zlom, zdroj, stav = najdi_zlom(cip, posledni_davka)
|
||||
except Exception as e:
|
||||
print(f" CHYBA: {e}")
|
||||
zlom, zdroj, stav = None, f"chyba: {e}", "chyba"
|
||||
|
||||
vysledky.append((cip, prijmeni, jmeno, posledni_davka, zlom, zdroj, stav))
|
||||
|
||||
cur.execute("""
|
||||
INSERT INTO seznam_pojistencu_zlomy
|
||||
(cip, prijmeni, jmeno, posledni_davka, zlom_datum, zlom_zdroj, stav, dotazeno_dne)
|
||||
VALUES (%s, %s, %s, %s, %s, %s, %s, %s)
|
||||
ON DUPLICATE KEY UPDATE
|
||||
posledni_davka=VALUES(posledni_davka),
|
||||
zlom_datum=VALUES(zlom_datum),
|
||||
zlom_zdroj=VALUES(zlom_zdroj),
|
||||
stav=VALUES(stav),
|
||||
dotazeno_dne=VALUES(dotazeno_dne)
|
||||
""", (cip, prijmeni, jmeno, posledni_davka, zlom, zdroj, stav, date.today()))
|
||||
|
||||
cur.close()
|
||||
conn.close()
|
||||
|
||||
# ── Výsledky ──────────────────────────────────────────────────────────────────
|
||||
print("\n" + "=" * 70)
|
||||
print(f"\n{'Příjmení':<25} {'Jméno':<20} {'CIP':<12} {'Poslední dávka':<15} {'Zlom':<12} Stav")
|
||||
print("-" * 95)
|
||||
for cip, pri, jme, pd, zd, zdroj, stav in vysledky:
|
||||
zlom_str = str(zd) if zd else "—"
|
||||
print(f"{pri:<25} {jme:<20} {cip:<12} {str(pd):<15} {zlom_str:<12} {stav}")
|
||||
@@ -0,0 +1,219 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
import sys as _sys
|
||||
_sys.stdout.reconfigure(encoding="utf-8", errors="replace")
|
||||
_sys.stderr.reconfigure(encoding="utf-8", errors="replace")
|
||||
"""
|
||||
06_nasledny_lekar.py
|
||||
====================
|
||||
Pro všechny ukončené pacienty (103) se dotáže VZP ke dni ukonceni+1
|
||||
na jejich nového praktického lékaře (odbornost 001).
|
||||
|
||||
Výsledek je jeden ze tří stavů:
|
||||
nenalezen → pacient u VZP neexistuje (zemřel / přestal být pojištěný)
|
||||
bez_lekare → pacient existuje, ale nemá GP (dosud se nepřehlásil)
|
||||
prehlasil → přehlásil se k novému lékaři (ukládáme ICP, jméno, datum)
|
||||
|
||||
Ukládá do tabulky seznam_pojistencu_nasledny_lekar.
|
||||
Přeskočí pacienty, kteří tam už jsou (resumovatelný běh).
|
||||
"""
|
||||
|
||||
import sys
|
||||
import time
|
||||
import xml.etree.ElementTree as ET
|
||||
from datetime import date, timedelta
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "Knihovny"))
|
||||
from mysql_db import connect_mysql
|
||||
from vzpb2b_client import VZPB2BClient
|
||||
|
||||
PFX_PATH = str(Path(__file__).resolve().parents[1] / "Certificates" / "picka.pfx")
|
||||
PFX_PASSWORD = "Vlado7309208104+"
|
||||
ICZ = "09305000"
|
||||
API_PAUSE = 2
|
||||
|
||||
CREATE_SQL = """
|
||||
CREATE TABLE IF NOT EXISTS seznam_pojistencu_nasledny_lekar (
|
||||
id INT NOT NULL AUTO_INCREMENT PRIMARY KEY,
|
||||
cip VARCHAR(12) NOT NULL,
|
||||
prijmeni VARCHAR(60) NOT NULL,
|
||||
jmeno VARCHAR(40) NOT NULL,
|
||||
datum_ukonceni DATE NOT NULL COMMENT 'Datum ukončení u nás',
|
||||
datum_dotazu DATE NOT NULL COMMENT 'Dotaz k datu ukonceni+1',
|
||||
stav_vzp VARCHAR(20) NOT NULL COMMENT 'nenalezen / bez_lekare / prehlasil',
|
||||
stav_vyrizeni VARCHAR(10) NULL COMMENT 'stavVyrizeniPozadavku z VZP',
|
||||
novy_icp VARCHAR(20) NULL,
|
||||
novy_icz VARCHAR(20) NULL,
|
||||
novy_nazev VARCHAR(200) NULL COMMENT 'nazevSZZ — jméno lékaře',
|
||||
novy_ordinace VARCHAR(200) NULL COMMENT 'nazevICP — název ordinace',
|
||||
datum_prehlaseni DATE NULL COMMENT 'datumRegistrace u nového lékaře',
|
||||
dotazeno_dne DATE NOT NULL,
|
||||
UNIQUE KEY uq_cip (cip)
|
||||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
|
||||
COMMENT='Stav pojištěnce ke dni ukončení registrace u naší ordinace';
|
||||
"""
|
||||
|
||||
NS = "http://xmlns.gemsystem.cz/B2B/RegistracePojistencePZSB2B/1"
|
||||
|
||||
|
||||
def parse_nasledny(xml_text: str) -> dict:
|
||||
"""
|
||||
Vrátí dict s klíči: stav_vzp, stav_vyrizeni, novy_icp, novy_icz,
|
||||
novy_nazev, novy_ordinace, datum_prehlaseni.
|
||||
"""
|
||||
try:
|
||||
root = ET.fromstring(xml_text)
|
||||
except ET.ParseError:
|
||||
return {"stav_vzp": "chyba_xml", "stav_vyrizeni": None,
|
||||
"novy_icp": None, "novy_icz": None, "novy_nazev": None,
|
||||
"novy_ordinace": None, "datum_prehlaseni": None}
|
||||
|
||||
def find(el, tag):
|
||||
e = el.find(f"{{{NS}}}{tag}")
|
||||
return e.text.strip() if e is not None and e.text else None
|
||||
|
||||
stav_vyrizeni = find(root, "stavVyrizeniPozadavku")
|
||||
|
||||
# Hledám odbornost 001 s jiným lékařem
|
||||
odbornosti = root.findall(f".//{{{NS}}}odbornost")
|
||||
for odb in odbornosti:
|
||||
# Vnější element — má ICZ
|
||||
icz = find(odb, "ICZ")
|
||||
icp = find(odb, "ICP")
|
||||
if not icp:
|
||||
continue
|
||||
# Ověřím odbornost (vnořený subelement)
|
||||
sub = odb.find(f"{{{NS}}}odbornost")
|
||||
if sub is None:
|
||||
continue
|
||||
kod = find(sub, "kod")
|
||||
if kod != "001":
|
||||
continue
|
||||
# Nalezen nový GP
|
||||
dr = date.fromisoformat(find(odb, "datumRegistrace")) \
|
||||
if find(odb, "datumRegistrace") else None
|
||||
return {
|
||||
"stav_vzp": "prehlasil",
|
||||
"stav_vyrizeni": stav_vyrizeni,
|
||||
"novy_icp": icp,
|
||||
"novy_icz": icz,
|
||||
"novy_nazev": find(odb, "nazevSZZ"),
|
||||
"novy_ordinace": find(odb, "nazevICP"),
|
||||
"datum_prehlaseni": dr,
|
||||
}
|
||||
|
||||
# Žádná odbornost 001 nenalezena
|
||||
if stav_vyrizeni == "0":
|
||||
stav = "nenalezen"
|
||||
elif stav_vyrizeni == "1":
|
||||
stav = "bez_lekare"
|
||||
else:
|
||||
# Zkusím ještě — pokud jsou nějaké záznamy ale ne 001, je to bez_lekare
|
||||
stav = "nenalezen" if not odbornosti else "bez_lekare"
|
||||
|
||||
return {"stav_vzp": stav, "stav_vyrizeni": stav_vyrizeni,
|
||||
"novy_icp": None, "novy_icz": None, "novy_nazev": None,
|
||||
"novy_ordinace": None, "datum_prehlaseni": None}
|
||||
|
||||
|
||||
# ── Načtení ukončených pacientů ───────────────────────────────────────────────
|
||||
|
||||
conn = connect_mysql()
|
||||
cur = conn.cursor()
|
||||
cur.execute(CREATE_SQL)
|
||||
|
||||
# Ukončení z vzp_registrace_lekari
|
||||
cur.execute("""
|
||||
SELECT r.rc, r.prijmeni, r.jmeno, r.datum_ukonceni
|
||||
FROM vzp_registrace_lekari r
|
||||
INNER JOIN (
|
||||
SELECT rc, MAX(k_datu) mk
|
||||
FROM vzp_registrace_lekari
|
||||
WHERE kod_odbornosti='001' AND ICP='09305001' AND ma_lekare=1
|
||||
GROUP BY rc
|
||||
) l ON r.rc=l.rc AND r.k_datu=l.mk
|
||||
WHERE r.kod_odbornosti='001' AND r.ICP='09305001'
|
||||
AND r.datum_ukonceni < '3000-01-01' AND r.datum_ukonceni < CURDATE()
|
||||
""")
|
||||
pacienti = [(r[0], r[1] or "", r[2] or "", r[3]) for r in cur.fetchall()]
|
||||
|
||||
# Doplním ze zlomů (13 nespárovaných)
|
||||
cur.execute("""
|
||||
SELECT cip, prijmeni, jmeno, zlom_datum
|
||||
FROM seznam_pojistencu_zlomy
|
||||
WHERE stav='ukončen' AND zlom_datum IS NOT NULL AND zlom_datum < CURDATE()
|
||||
""")
|
||||
for r in cur.fetchall():
|
||||
if r[0] not in {p[0] for p in pacienti}:
|
||||
pacienti.append((r[0], r[1], r[2], r[3]))
|
||||
|
||||
# Přeskočím již zpracované
|
||||
cur.execute("SELECT cip FROM seznam_pojistencu_nasledny_lekar")
|
||||
hotovi = {r[0] for r in cur.fetchall()}
|
||||
ke_zpracovani = [(c, p, j, d) for c, p, j, d in pacienti if c not in hotovi]
|
||||
|
||||
print(f"Ukončených celkem: {len(pacienti)}")
|
||||
print(f"Již zpracováno: {len(hotovi)}")
|
||||
print(f"Ke zpracování: {len(ke_zpracovani)}")
|
||||
|
||||
if not ke_zpracovani:
|
||||
print("Vše již zpracováno.")
|
||||
cur.close(); conn.close(); sys.exit(0)
|
||||
|
||||
vzp = VZPB2BClient("prod", PFX_PATH, PFX_PASSWORD, icz=ICZ)
|
||||
|
||||
# ── Hlavní smyčka ─────────────────────────────────────────────────────────────
|
||||
print()
|
||||
sirka = max(len(f"{p} {j}") for _, p, j, _ in ke_zpracovani) + 2
|
||||
|
||||
for i, (cip, pri, jme, du) in enumerate(ke_zpracovani, 1):
|
||||
datum_dotazu = du + timedelta(days=1)
|
||||
jmeno_str = f"{pri} {jme}"
|
||||
print(f"[{i:>3}/{len(ke_zpracovani)}] {jmeno_str:<{sirka}} ({cip}) k {datum_dotazu}", end=" ", flush=True)
|
||||
|
||||
try:
|
||||
xml = vzp.registrace_lekare(cip, datum_dotazu.isoformat(), odbornosti=["001"])
|
||||
time.sleep(API_PAUSE)
|
||||
vysl = parse_nasledny(xml)
|
||||
except Exception as e:
|
||||
print(f"CHYBA: {e}")
|
||||
vysl = {"stav_vzp": "chyba", "stav_vyrizeni": str(e),
|
||||
"novy_icp": None, "novy_icz": None, "novy_nazev": None,
|
||||
"novy_ordinace": None, "datum_prehlaseni": None}
|
||||
|
||||
# Výpis
|
||||
stav = vysl["stav_vzp"]
|
||||
if stav == "prehlasil":
|
||||
print(f"→ přehlásil se: {vysl['novy_nazev']} (ICP {vysl['novy_icp']}, od {vysl['datum_prehlaseni']})")
|
||||
elif stav == "bez_lekare":
|
||||
print("→ bez nového lékaře")
|
||||
elif stav == "nenalezen":
|
||||
print("→ nenalezen (zemřel / nepojištěný)")
|
||||
else:
|
||||
print(f"→ {stav}")
|
||||
|
||||
cur.execute("""
|
||||
INSERT INTO seznam_pojistencu_nasledny_lekar
|
||||
(cip, prijmeni, jmeno, datum_ukonceni, datum_dotazu, stav_vzp,
|
||||
stav_vyrizeni, novy_icp, novy_icz, novy_nazev, novy_ordinace,
|
||||
datum_prehlaseni, dotazeno_dne)
|
||||
VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)
|
||||
ON DUPLICATE KEY UPDATE
|
||||
datum_ukonceni=VALUES(datum_ukonceni),
|
||||
datum_dotazu=VALUES(datum_dotazu),
|
||||
stav_vzp=VALUES(stav_vzp), stav_vyrizeni=VALUES(stav_vyrizeni),
|
||||
novy_icp=VALUES(novy_icp), novy_icz=VALUES(novy_icz),
|
||||
novy_nazev=VALUES(novy_nazev), novy_ordinace=VALUES(novy_ordinace),
|
||||
datum_prehlaseni=VALUES(datum_prehlaseni),
|
||||
dotazeno_dne=VALUES(dotazeno_dne)
|
||||
""", (cip, pri, jme, du, datum_dotazu, stav,
|
||||
vysl["stav_vyrizeni"], vysl["novy_icp"], vysl["novy_icz"],
|
||||
vysl["novy_nazev"], vysl["novy_ordinace"], vysl["datum_prehlaseni"],
|
||||
date.today()))
|
||||
|
||||
cur.close()
|
||||
conn.close()
|
||||
|
||||
# ── Souhrn ────────────────────────────────────────────────────────────────────
|
||||
print(f"\nHotovo. Zpracováno {len(ke_zpracovani)} pacientů.")
|
||||
@@ -0,0 +1,303 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
import sys as _sys
|
||||
_sys.stdout.reconfigure(encoding="utf-8", errors="replace")
|
||||
_sys.stderr.reconfigure(encoding="utf-8", errors="replace")
|
||||
"""
|
||||
05_report_excel.py
|
||||
==================
|
||||
Kompletní Excel report všech VZP pojištěnců z dávek.
|
||||
Kombinuje čtyři zdroje:
|
||||
- seznam_pojistencu_davky → první/poslední dávka, zda v aktuální
|
||||
- vzp_registrace_lekari → datum_ukonceni od VZP (989 pacientů)
|
||||
- seznam_pojistencu_zlomy → bod zlomu pro 13 nespárovaných
|
||||
- seznam_pojistencu_nasledny_lekar → nový lékař / nenalezen po ukončení
|
||||
|
||||
Listy:
|
||||
1. Všichni pojištěnci – kompletní přehled, řazeno příjmení
|
||||
2. Aktivní – stále registrováni
|
||||
3. Ukončení – registrace ukončena + nový lékař / osud
|
||||
4. Nenalezeni – bez dostatečných dat
|
||||
"""
|
||||
|
||||
import sys
|
||||
from datetime import date
|
||||
from pathlib import Path
|
||||
|
||||
import openpyxl
|
||||
from openpyxl.styles import Font, PatternFill, Alignment, Border, Side
|
||||
from openpyxl.utils import get_column_letter
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "Knihovny"))
|
||||
from mysql_db import connect_mysql
|
||||
|
||||
TODAY = date.today()
|
||||
OUTPUT = Path(__file__).parent / "report_pojistenci.xlsx"
|
||||
|
||||
# ── Barvy ─────────────────────────────────────────────────────────────────────
|
||||
C_HDR_DARK = "1F4E79"
|
||||
C_HDR_MED = "2E75B6"
|
||||
C_AKTIVNI = "E2EFDA" # zelená
|
||||
C_UKONCEN = "FCE4D6" # lososová
|
||||
C_NENALEZEN = "FFF2CC" # žlutá
|
||||
C_ZEBRA = "F2F2F2"
|
||||
C_TITLE_AKT = "375623"
|
||||
C_TITLE_UKO = "C55A11"
|
||||
C_TITLE_NEN = "7F6000"
|
||||
|
||||
THIN = Side(style="thin", color="BFBFBF")
|
||||
|
||||
def border():
|
||||
return Border(left=THIN, right=THIN, top=THIN, bottom=THIN)
|
||||
|
||||
def hdr(cell, text, bg=C_HDR_DARK, fg="FFFFFF", size=10):
|
||||
cell.value = text
|
||||
cell.font = Font(bold=True, color=fg, size=size)
|
||||
cell.fill = PatternFill("solid", fgColor=bg)
|
||||
cell.alignment = Alignment(horizontal="center", vertical="center", wrap_text=True)
|
||||
cell.border = border()
|
||||
|
||||
def cell(ws, row, col, value, bg=None, bold=False, align="left", fmt=None, color="000000"):
|
||||
c = ws.cell(row, col, value)
|
||||
c.font = Font(bold=bold, size=10, color=color)
|
||||
c.alignment = Alignment(horizontal=align, vertical="center")
|
||||
c.border = border()
|
||||
if bg:
|
||||
c.fill = PatternFill("solid", fgColor=bg)
|
||||
if fmt:
|
||||
c.number_format = fmt
|
||||
return c
|
||||
|
||||
def title_row(ws, row, text, ncols, bg, fg="FFFFFF", height=26):
|
||||
ws.merge_cells(start_row=row, start_column=1, end_row=row, end_column=ncols)
|
||||
c = ws.cell(row, 1)
|
||||
c.value = text
|
||||
c.font = Font(bold=True, size=13, color=fg)
|
||||
c.fill = PatternFill("solid", fgColor=bg)
|
||||
c.alignment = Alignment(horizontal="center", vertical="center")
|
||||
ws.row_dimensions[row].height = height
|
||||
|
||||
def autofit(ws, widths):
|
||||
for i, w in enumerate(widths, 1):
|
||||
ws.column_dimensions[get_column_letter(i)].width = w
|
||||
|
||||
# ── Načtení dat ───────────────────────────────────────────────────────────────
|
||||
print("Načítám data z DB ...")
|
||||
conn = connect_mysql()
|
||||
cur = conn.cursor()
|
||||
|
||||
# 1. Ze seznam_pojistencu_davky: rozsah přítomnosti + počet dávek
|
||||
cur.execute("""
|
||||
SELECT
|
||||
cip,
|
||||
MIN(prijmeni) AS prijmeni,
|
||||
MIN(jmeno) AS jmeno,
|
||||
MIN(DATE(CONCAT(davka_rok,'-',LPAD(davka_mesic,2,'0'),'-01'))) AS prvni_davka,
|
||||
MAX(DATE(CONCAT(davka_rok,'-',LPAD(davka_mesic,2,'0'),'-01'))) AS posledni_davka,
|
||||
COUNT(DISTINCT CONCAT(davka_rok,davka_mesic)) AS pocet_davek
|
||||
FROM seznam_pojistencu_davky
|
||||
WHERE pojistovna='111'
|
||||
GROUP BY cip
|
||||
""")
|
||||
seznam = {r[0]: {"prijmeni": r[1], "jmeno": r[2],
|
||||
"prvni": r[3], "posledni": r[4], "pocet_davek": r[5]}
|
||||
for r in cur.fetchall()}
|
||||
|
||||
# Nejnovější datum dávky (pro sloupec "v aktuální dávce")
|
||||
cur.execute("""
|
||||
SELECT MAX(DATE(CONCAT(davka_rok,'-',LPAD(davka_mesic,2,'0'),'-01')))
|
||||
FROM seznam_pojistencu_davky WHERE pojistovna='111'
|
||||
""")
|
||||
nejnovejsi_davka = cur.fetchone()[0]
|
||||
|
||||
cur.execute("""
|
||||
SELECT DISTINCT cip FROM seznam_pojistencu_davky
|
||||
WHERE pojistovna='111'
|
||||
AND CONCAT(davka_rok,LPAD(davka_mesic,2,'0')) = (
|
||||
SELECT MAX(CONCAT(davka_rok,LPAD(davka_mesic,2,'0')))
|
||||
FROM seznam_pojistencu_davky WHERE pojistovna='111'
|
||||
)
|
||||
""")
|
||||
v_aktualni = {r[0] for r in cur.fetchall()}
|
||||
|
||||
# 2. Z vzp_registrace_lekari: nejnovější záznam na pacienta (u nás, odb 001)
|
||||
cur.execute("""
|
||||
SELECT r.rc, r.datum_zahajeni, r.datum_ukonceni, r.k_datu
|
||||
FROM vzp_registrace_lekari r
|
||||
INNER JOIN (
|
||||
SELECT rc, MAX(k_datu) AS max_k
|
||||
FROM vzp_registrace_lekari
|
||||
WHERE kod_odbornosti='001' AND ICP='09305001' AND ma_lekare=1
|
||||
GROUP BY rc
|
||||
) latest ON r.rc = latest.rc AND r.k_datu = latest.max_k
|
||||
WHERE r.kod_odbornosti='001' AND r.ICP='09305001' AND r.ma_lekare=1
|
||||
""")
|
||||
registrace = {r[0]: {"zahajeni": r[1], "ukonceni": r[2], "k_datu": r[3]}
|
||||
for r in cur.fetchall()}
|
||||
|
||||
# 3. Ze seznam_pojistencu_zlomy (13 nespárovaných)
|
||||
cur.execute("SELECT cip, zlom_datum, zlom_zdroj, stav FROM seznam_pojistencu_zlomy")
|
||||
zlomy = {r[0]: {"ukonceni": r[1], "zdroj": r[2], "stav": r[3]}
|
||||
for r in cur.fetchall()}
|
||||
|
||||
# 4. Následný lékař po ukončení
|
||||
cur.execute("""
|
||||
SELECT cip, stav_vzp, novy_nazev, novy_ordinace, novy_icp, datum_prehlaseni
|
||||
FROM seznam_pojistencu_nasledny_lekar
|
||||
""")
|
||||
nasledni = {r[0]: {"stav_vzp": r[1], "novy_nazev": r[2],
|
||||
"novy_ordinace": r[3], "novy_icp": r[4],
|
||||
"datum_prehlaseni": r[5]}
|
||||
for r in cur.fetchall()}
|
||||
|
||||
cur.close()
|
||||
conn.close()
|
||||
|
||||
# ── Sestavení řádků ───────────────────────────────────────────────────────────
|
||||
print(f"Pacientů celkem: {len(seznam)}")
|
||||
|
||||
radky = []
|
||||
for cip, s in seznam.items():
|
||||
v_akt = cip in v_aktualni
|
||||
|
||||
if cip in registrace:
|
||||
reg = registrace[cip]
|
||||
du = reg["ukonceni"]
|
||||
zdroj = f"vzp_registrace ({reg['k_datu']})"
|
||||
elif cip in zlomy:
|
||||
z = zlomy[cip]
|
||||
du = z["ukonceni"]
|
||||
zdroj = z["zdroj"]
|
||||
else:
|
||||
du = None
|
||||
zdroj = "—"
|
||||
|
||||
if du is None:
|
||||
stav = "aktivní"
|
||||
elif du.year >= 3000:
|
||||
stav = "aktivní"
|
||||
du = None # nezobrazujeme 3000-01-01
|
||||
elif du >= TODAY:
|
||||
stav = "aktivní"
|
||||
else:
|
||||
stav = "ukončen"
|
||||
|
||||
if zdroj == "—" and not v_akt:
|
||||
stav = "nenalezen"
|
||||
|
||||
# Následný lékař
|
||||
nl = nasledni.get(cip)
|
||||
if nl:
|
||||
po_ukonceni = nl["stav_vzp"] # prehlasil / nenalezen / bez_lekare
|
||||
novy_lekar = nl["novy_nazev"] or nl["novy_ordinace"] or ""
|
||||
if nl["novy_icp"]:
|
||||
novy_lekar += f" (ICP {nl['novy_icp']})"
|
||||
datum_prehl = nl["datum_prehlaseni"]
|
||||
else:
|
||||
po_ukonceni = ""
|
||||
novy_lekar = ""
|
||||
datum_prehl = None
|
||||
|
||||
radky.append({
|
||||
"cip": cip,
|
||||
"prijmeni": s["prijmeni"],
|
||||
"jmeno": s["jmeno"],
|
||||
"prvni": s["prvni"],
|
||||
"posledni": s["posledni"],
|
||||
"pocet_davek": s["pocet_davek"],
|
||||
"v_aktualni": v_akt,
|
||||
"ukonceni": du,
|
||||
"zdroj": zdroj,
|
||||
"stav": stav,
|
||||
"po_ukonceni": po_ukonceni,
|
||||
"novy_lekar": novy_lekar,
|
||||
"datum_prehl": datum_prehl,
|
||||
})
|
||||
|
||||
radky.sort(key=lambda r: (r["prijmeni"], r["jmeno"]))
|
||||
|
||||
aktivni = [r for r in radky if r["stav"] == "aktivní"]
|
||||
ukonceni = sorted([r for r in radky if r["stav"] == "ukončen"],
|
||||
key=lambda r: r["ukonceni"] or date.min)
|
||||
nenalezeni = [r for r in radky if r["stav"] == "nenalezen"]
|
||||
|
||||
print(f" Aktivní: {len(aktivni)}")
|
||||
print(f" Ukončení: {len(ukonceni)}")
|
||||
print(f" Nenalezeni: {len(nenalezeni)}")
|
||||
|
||||
# ── Excel ─────────────────────────────────────────────────────────────────────
|
||||
SLOUPCE_ALL = ["Příjmení", "Jméno", "ČIP", "První\ndávka", "Poslední\ndávka",
|
||||
"Počet\ndávek", "V aktuální\ndávce", "Ukončení\nregistrace",
|
||||
"Stav", "Po ukončení", "Nový lékař / poznámka", "Datum\npřehlášení"]
|
||||
WIDTHS_ALL = [24, 18, 13, 12, 13, 9, 10, 14, 10, 13, 40, 13]
|
||||
|
||||
SLOUPCE = SLOUPCE_ALL # alias pro funkci zapsat_list
|
||||
WIDTHS = WIDTHS_ALL
|
||||
NCOLS = len(SLOUPCE)
|
||||
DATE_FMT = "DD.MM.YYYY"
|
||||
|
||||
def zapsat_list(ws, nadpis, bg_title, seznam_radku, stav_bg):
|
||||
ws.freeze_panes = "A3"
|
||||
title_row(ws, 1, nadpis, NCOLS, bg_title)
|
||||
ws.row_dimensions[2].height = 30
|
||||
for col, h in enumerate(SLOUPCE, 1):
|
||||
hdr(ws.cell(2, col), h)
|
||||
autofit(ws, WIDTHS)
|
||||
|
||||
PO_CLR = {"prehlasil": "375623", "nenalezen": "C55A11",
|
||||
"bez_lekare": "7F6000", "": "000000"}
|
||||
PO_TXT = {"prehlasil": "přehlásil se", "nenalezen": "nenalezen",
|
||||
"bez_lekare": "bez lékaře", "": ""}
|
||||
|
||||
for ri, r in enumerate(seznam_radku, 3):
|
||||
bg = stav_bg if ri % 2 == 0 else None
|
||||
cell(ws, ri, 1, r["prijmeni"], bg)
|
||||
cell(ws, ri, 2, r["jmeno"], bg)
|
||||
cell(ws, ri, 3, r["cip"], bg, align="center")
|
||||
cell(ws, ri, 4, r["prvni"], bg, align="center", fmt=DATE_FMT)
|
||||
cell(ws, ri, 5, r["posledni"], bg, align="center", fmt=DATE_FMT)
|
||||
cell(ws, ri, 6, r["pocet_davek"], bg, align="right")
|
||||
akt_txt = "✓" if r["v_aktualni"] else "–"
|
||||
akt_clr = "375623" if r["v_aktualni"] else "C55A11"
|
||||
cell(ws, ri, 7, akt_txt, bg, bold=True, align="center", color=akt_clr)
|
||||
cell(ws, ri, 8, r["ukonceni"], bg, align="center", fmt=DATE_FMT)
|
||||
cell(ws, ri, 9, r["stav"], bg, align="center", bold=True,
|
||||
color=("375623" if r["stav"]=="aktivní" else
|
||||
"C55A11" if r["stav"]=="ukončen" else "7F6000"))
|
||||
po = r.get("po_ukonceni", "")
|
||||
cell(ws, ri, 10, PO_TXT.get(po, po), bg, align="center", bold=bool(po),
|
||||
color=PO_CLR.get(po, "000000"))
|
||||
cell(ws, ri, 11, r.get("novy_lekar", ""), bg)
|
||||
cell(ws, ri, 12, r.get("datum_prehl"), bg, align="center", fmt=DATE_FMT)
|
||||
|
||||
wb = openpyxl.Workbook()
|
||||
|
||||
# List 1 – Všichni
|
||||
ws1 = wb.active
|
||||
ws1.title = "Všichni pojištěnci"
|
||||
zapsat_list(ws1,
|
||||
f"VZP pojištěnci — kompletní přehled ({TODAY.strftime('%d.%m.%Y')}) | "
|
||||
f"celkem: {len(radky)} | aktivní: {len(aktivni)} | ukončení: {len(ukonceni)}",
|
||||
C_HDR_DARK, radky, C_ZEBRA)
|
||||
|
||||
# List 2 – Aktivní
|
||||
ws2 = wb.create_sheet("Aktivní")
|
||||
zapsat_list(ws2,
|
||||
f"Aktivní pojištěnci ({TODAY.strftime('%d.%m.%Y')}) — celkem: {len(aktivni)}",
|
||||
C_TITLE_AKT, aktivni, C_AKTIVNI)
|
||||
|
||||
# List 3 – Ukončení
|
||||
ws3 = wb.create_sheet("Ukončení")
|
||||
zapsat_list(ws3,
|
||||
f"Ukončená registrace — celkem: {len(ukonceni)}",
|
||||
C_TITLE_UKO, ukonceni, C_UKONCEN)
|
||||
|
||||
# List 4 – Nenalezeni
|
||||
if nenalezeni:
|
||||
ws4 = wb.create_sheet("Nenalezeni")
|
||||
zapsat_list(ws4,
|
||||
f"Bez dostatečných dat — celkem: {len(nenalezeni)}",
|
||||
C_TITLE_NEN, nenalezeni, C_NENALEZEN)
|
||||
|
||||
wb.save(OUTPUT)
|
||||
print(f"\nExcel uložen: {OUTPUT}")
|
||||
Binary file not shown.
Binary file not shown.
Reference in New Issue
Block a user