administrator/insurance
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

notebookVB

Browse Source
This commit is contained in:
Vladimir Buzalka
2026-01-26 19:46:49 +01:00
parent b20654b607
commit 73e5eab4a4
4 changed files with 476 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

180
05 Kontrola pojištěnců/10 Čti OZP.py Normal file
View File

@@ -0,0 +1,180 @@
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()

47
05 Kontrola pojištěnců/20 Kapdetais.py Normal file
View File

@@ -0,0 +1,47 @@
from knihovny.medicus_db import MedicusDB # nebo odkud třídu importuješ
def main():
poj = input("Zadej kód pojišťovny (např. 207): ").strip()
db = MedicusDB(
host="192.168.1.4",
db_path=r"z:\medicus 3\data\medicus.fdb"
)
sql = """
SELECT
fak.id,
fak.cisfak,
fak.poj,
fak.kapdetail
FROM fak
WHERE fak.poj = ?
AND fak.kapdetail IS NOT NULL
AND fak.kapdetail <> ''
ORDER BY fak.cisfak DESC
ROWS 1
"""
rows = db.query_dict(sql, (poj,))
if not rows:
print(f"❌ Nenalezena žádná faktura s kapdetails pro pojišťovnu {poj}")
return
r = rows[0]
print("=" * 80)
print(f"FAKTURA ID : {r['id']}")
print(f"ČÍSLO FAKTURY: {r['cisfak']}")
print(f"POJIŠŤOVNA : {r['poj']}")
print("-" * 80)
print("KAPDETAILS:")
print(r["kapdetail"])
print("=" * 80)
db.close()
if __name__ == "__main__":
main()

234
05 Kontrola pojištěnců/30 Porovnani kapitace.py Normal file
View File

@@ -0,0 +1,234 @@
import re
from pathlib import Path
from typing import List, Dict, Set
import pymysql
from knihovny.medicus_db import MedicusDB
# ============================================================
# KONFIGURACE
# ============================================================
# --- Firebird / Medicus ---
FIREBIRD_HOST = "192.168.1.4"
FIREBIRD_DB = r"Z:\medicus 3\data\medicus.fdb"
FIREBIRD_USER = "SYSDBA"
FIREBIRD_PASS = "masterkey"
FIREBIRD_CHARSET = "WIN1250"
# --- MySQL / Kapitace ---
MYSQL_CFG = {
"host": "192.168.1.76",
"port": 3307,
"user": "root",
"password": "Vlado9674+",
"database": "ordinace",
"charset": "utf8mb4",
"autocommit": True,
}
MYSQL_TABLE_HEADER = "zp_kapitace_header"
MYSQL_TABLE_PAC = "zp_kapitace_pojistenec"
# ============================================================
# HELPERY
# ============================================================
_rc_re = re.compile(r"\D+")
def normalize_rc(value: str) -> str:
"""Rodné číslo / číslo pojištěnce → jen číslice."""
if not value:
return ""
return _rc_re.sub("", str(value))
def parse_kapdetail_idpacs(kapdetail: str) -> List[int]:
"""
KAPDETAIL:
<něco>=SOUHRN;idpac;vek|castka;idpac;vek|castka;...
Vrací seznam idpac.
"""
if not kapdetail:
return []
s = kapdetail.strip()
if "=" in s:
s = s.split("=", 1)[1]
parts = [p.strip() for p in s.split(";") if p.strip()]
if not parts:
return []
# první položka je souhrn (není idpac)
if not parts[0].isdigit():
parts = parts[1:]
idpacs: List[int] = []
# idpac je vždy na sudém indexu (0,2,4,...)
for i in range(0, len(parts), 2):
if parts[i].isdigit():
idpacs.append(int(parts[i]))
return idpacs
def chunk_list(lst: List[int], size: int = 500) -> List[List[int]]:
return [lst[i:i + size] for i in range(0, len(lst), size)]
# ============================================================
# FIREBIRD – DATA Z MEDICUSU
# ============================================================
def get_latest_fak_for_poj(db: MedicusDB, poj: str) -> Dict:
sql = """
SELECT
fak.id,
fak.cisfak,
fak.poj,
fak.kapdetail
FROM fak
WHERE fak.poj = ?
AND fak.kapdetail IS NOT NULL
AND fak.kapdetail <> ''
ORDER BY fak.cisfak DESC
ROWS 1
"""
rows = db.query_dict(sql, (poj,))
return rows[0] if rows else {}
def get_rodcis_for_idpacs(db: MedicusDB, idpacs: List[int]) -> Dict[int, str]:
"""
Vrátí mapu: idpac → rodcis
"""
result: Dict[int, str] = {}
if not idpacs:
return result
for batch in chunk_list(idpacs):
placeholders = ",".join("?" for _ in batch)
sql = f"""
SELECT
kar.idpac,
kar.rodcis
FROM kar
WHERE kar.idpac IN ({placeholders})
AND kar.rodcis IS NOT NULL
AND kar.rodcis <> ''
AND (kar.vyrazen IS NULL OR kar.vyrazen <> 'A')
"""
rows = db.query(sql, tuple(batch))
for idpac, rodcis in rows:
result[int(idpac)] = str(rodcis)
return result
# ============================================================
# MYSQL – KAPITACE ZP
# ============================================================
def get_mysql_kapitace_rc(zp_kod: str) -> Set[str]:
sql = f"""
SELECT p.cislo_pojistence
FROM {MYSQL_TABLE_PAC} p
JOIN {MYSQL_TABLE_HEADER} h ON h.id = p.header_id
WHERE h.zp_kod = %s
AND h.snapshot_date = (
SELECT MAX(h2.snapshot_date)
FROM {MYSQL_TABLE_HEADER} h2
WHERE h2.zp_kod = %s
)
"""
with pymysql.connect(**MYSQL_CFG) as conn:
with conn.cursor() as cur:
cur.execute(sql, (zp_kod, zp_kod))
rows = cur.fetchall()
return {
normalize_rc(r[0])
for r in rows
if normalize_rc(r[0])
}
# ============================================================
# MAIN
# ============================================================
def main():
poj = input("Zadej kód pojišťovny (např. 207): ").strip()
# --- Medicus ---
mdb = MedicusDB(
host=FIREBIRD_HOST,
db_path=FIREBIRD_DB,
user=FIREBIRD_USER,
password=FIREBIRD_PASS,
charset=FIREBIRD_CHARSET,
)
fak = get_latest_fak_for_poj(mdb, poj)
if not fak:
print(f"❌ Nenalezena žádná kapitační faktura pro pojišťovnu {poj}")
mdb.close()
return
print("\nNalezena faktura:")
print(f" ID : {fak['id']}")
print(f" CISFAK : {fak['cisfak']}")
print(f" POJ : {fak['poj']}")
idpacs = parse_kapdetail_idpacs(fak["kapdetail"])
print(f"\nKAPDETAIL → idpac celkem: {len(idpacs)}")
idpac_to_rc = get_rodcis_for_idpacs(mdb, idpacs)
mdb.close()
medicus_rc = {
normalize_rc(rc)
for rc in idpac_to_rc.values()
if normalize_rc(rc)
}
print(f"Medicus (KAR.RODCIS): {len(medicus_rc)}")
# --- MySQL ---
mysql_rc = get_mysql_kapitace_rc(poj)
print(f"MySQL kapitace ZP: {len(mysql_rc)}")
# --- Porovnání ---
both = medicus_rc & mysql_rc
only_medicus = medicus_rc - mysql_rc
only_mysql = mysql_rc - medicus_rc
print("\n" + "=" * 80)
print(f"✅ V obou (OK): {len(both)}")
print(f"❌ Jen v Medicusu: {len(only_medicus)}")
print(f"❌ Jen v MySQL (kapitace): {len(only_mysql)}")
print("=" * 80)
if only_medicus:
print("\n❌ Jen v Medicusu (fakturováno, ale není v kapitaci ZP):")
for rc in sorted(only_medicus):
print(" ", rc)
if only_mysql:
print("\n❌ Jen v MySQL (v kapitaci ZP, ale není ve faktuře Medicusu):")
for rc in sorted(only_mysql):
print(" ", rc)
if __name__ == "__main__":
main()

15
knihovny/medicus_db.py
View File

@@ -13,6 +13,21 @@ class MedicusDB:
)
self.cur = self.conn.cursor()
def get_fak_kapitace(self, as_dict=False):
sql = """
SELECT
fak.id,
fak.cisfak,
fak.poj,
fak.kapdetail
FROM fak
WHERE fak.kapdetail IS NOT NULL
AND fak.kapdetail <> ''
"""
if as_dict:
return self.query_dict(sql)
return self.query(sql)
def query(self, sql, params=None):
self.cur.execute(sql, params or ())
return self.cur.fetchall()