4f586f4b57
- faktury_report.py: nový list ED_PODANI (ED_BOOKOFSUBMISSIONS) s přehledem podání pojišťovnám - faktury_report.py: nový list ED_PODANI_DATA s dekódovaným obsahem dávek (KDAVKA, REQUEST XML, odpovědi pojišťoven) - Opraveno kódování: KDAVKA=cp1250, REQUEST detekce BOM (utf-16/utf-8), SERVERRESPONSE/PROTOCOL=iso-8859-2 - Hyperlinky ED_PODANI ↔ ED_PODANI_DATA a Faktura → FAK - FakturaceADavky.md: dokumentace ED_* tabulek, portálů pojišťoven, formátů REQUEST XML - Průzkumné skripty: find_edavky_table, explore_hpn, explore_ed_bookofsubmissions, parse_trace_edavky aj. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
491 lines
15 KiB
Python
491 lines
15 KiB
Python
import fdb
|
||
import openpyxl
|
||
from openpyxl.styles import Font, PatternFill, Alignment
|
||
from openpyxl.utils import get_column_letter
|
||
from datetime import datetime
|
||
import os
|
||
import sys
|
||
|
||
# --- Připojení ---
|
||
conn = fdb.connect(
|
||
dsn=r'localhost:c:\medicus 3\data\medicus.fdb',
|
||
user='SYSDBA', password='masterkey', charset='win1250'
|
||
)
|
||
cur = conn.cursor()
|
||
|
||
# --- Výstupní soubor ---
|
||
output_dir = r'u:\Dropbox\!!!Days\Downloads Z230'
|
||
now = datetime.now()
|
||
filename = now.strftime('%Y-%m-%d_%H-%M-%S') + '_faktury.xlsx'
|
||
output_path = os.path.join(output_dir, filename)
|
||
|
||
# --- Smazání předchozích verzí ---
|
||
for f in os.listdir(output_dir):
|
||
if f.endswith('_faktury.xlsx'):
|
||
os.remove(os.path.join(output_dir, f))
|
||
|
||
wb = openpyxl.Workbook()
|
||
|
||
# =====================
|
||
# Pomocné funkce
|
||
# =====================
|
||
|
||
HEADER_FILL = PatternFill('solid', fgColor='2F5496')
|
||
HEADER_FONT = Font(bold=True, color='FFFFFF')
|
||
LINK_FONT = Font(color='0563C1', underline='single')
|
||
ZEBRA_FILL = PatternFill('solid', fgColor='DCE6F1')
|
||
|
||
def style_header(ws):
|
||
for cell in ws[1]:
|
||
cell.fill = HEADER_FILL
|
||
cell.font = HEADER_FONT
|
||
cell.alignment = Alignment(horizontal='center')
|
||
|
||
def autofit(ws):
|
||
for col in ws.columns:
|
||
max_len = max((len(str(cell.value)) if cell.value is not None else 0) for cell in col)
|
||
ws.column_dimensions[get_column_letter(col[0].column)].width = min(max_len + 2, 40)
|
||
|
||
def fmt(val):
|
||
if val is None:
|
||
return ''
|
||
return val
|
||
|
||
# =====================
|
||
# List 1 – FAK
|
||
# =====================
|
||
|
||
ws1 = wb.active
|
||
ws1.title = 'FAK'
|
||
|
||
cur.execute('''
|
||
SELECT
|
||
ID, CISFAK, POJ, DATUMOD, DATUMDO, DATVYS, DATODE,
|
||
VYKONY, KAPITACE, ZALOHA, CENA, ZAPLACENO, ZUM, HOSPAUSAL,
|
||
PROPLACENO, SPLAT, DRUH, TYP, ROK, OBDOB,
|
||
NAZFAK, POZFAK, OBDFAK,
|
||
ICO, BANKA, UCET,
|
||
ODJMENO, ODULICE, ODMISTO, ODPSC,
|
||
PLNAZEV, PLULICE, PLMISTO, PLPSC,
|
||
ICZ, ICZ1, IDICZ, PORCISLO, DRUHPOJ, POZNAMKA
|
||
FROM FAK
|
||
ORDER BY ID DESC
|
||
''')
|
||
fak_cols = [d[0] for d in cur.description]
|
||
fak_rows = cur.fetchall()
|
||
|
||
ws1.append(fak_cols)
|
||
for i, row in enumerate(fak_rows, start=2):
|
||
ws1.append([fmt(v) for v in row])
|
||
if i % 2 == 0:
|
||
for cell in ws1[i]:
|
||
cell.fill = ZEBRA_FILL
|
||
|
||
style_header(ws1)
|
||
ws1.freeze_panes = 'A2'
|
||
autofit(ws1)
|
||
|
||
# =====================
|
||
# List 2 – FAKDET
|
||
# =====================
|
||
|
||
ws2 = wb.create_sheet('FAKDET')
|
||
|
||
cur.execute('''
|
||
SELECT
|
||
fd.ID, fd.IDFAK,
|
||
f.CISFAK, f.POJ, f.DATUMOD, f.DATUMDO, f.ROK,
|
||
fd.ICP, fd.ODB, fd.IDUZI,
|
||
u.PRIJMENI, u.JMENO,
|
||
fd.CENAVYK, fd.CENALEC, fd.CENAKAP
|
||
FROM FAKDET fd
|
||
JOIN FAK f ON f.ID = fd.IDFAK
|
||
LEFT JOIN UZIVATEL u ON u.IDUZI = fd.IDUZI
|
||
ORDER BY fd.IDFAK DESC, fd.ID DESC
|
||
''')
|
||
det_cols = [d[0] for d in cur.description]
|
||
det_rows = cur.fetchall()
|
||
|
||
ws2.append(det_cols)
|
||
for i, row in enumerate(det_rows, start=2):
|
||
ws2.append([fmt(v) for v in row])
|
||
if i % 2 == 0:
|
||
for cell in ws2[i]:
|
||
cell.fill = ZEBRA_FILL
|
||
|
||
style_header(ws2)
|
||
ws2.freeze_panes = 'A2'
|
||
autofit(ws2)
|
||
|
||
# =====================
|
||
# List 3 – PORTAL (krátké sloupce)
|
||
# =====================
|
||
|
||
ws3 = wb.create_sheet('PORTAL')
|
||
|
||
cur.execute('''
|
||
SELECT ID, IDFAK, ODESLANO, CHYBA, STAV, ID_PODANI, IDPODANI, IDCERT,
|
||
DAVKA_ROK, DAVKA_DISK, DAVKA_IDICZ, DAVKA_DATUMOD, DAVKA_DATUMDO,
|
||
DAVKA_CASTKA, BB_DAVKA, BB_FAKTURA
|
||
FROM PORTAL
|
||
ORDER BY ID DESC
|
||
''')
|
||
portal_cols = [d[0] for d in cur.description]
|
||
portal_rows = cur.fetchall()
|
||
|
||
ws3.append(portal_cols)
|
||
for i, row in enumerate(portal_rows, start=2):
|
||
ws3.append([fmt(v) for v in row])
|
||
if i % 2 == 0:
|
||
for cell in ws3[i]:
|
||
cell.fill = ZEBRA_FILL
|
||
|
||
style_header(ws3)
|
||
ws3.freeze_panes = 'A2'
|
||
autofit(ws3)
|
||
|
||
# =====================
|
||
# List 4 – PORTAL_DATA (BLOBy)
|
||
# =====================
|
||
|
||
ws4 = wb.create_sheet('PORTAL_DATA')
|
||
|
||
cur.execute('''
|
||
SELECT ID, ID_PODANI, DAVKA_ROK, DAVKA_DISK, DAVKA_DATUMOD, DAVKA_DATUMDO,
|
||
KDAVKA, FDAVKA, DATA
|
||
FROM PORTAL
|
||
ORDER BY ID DESC
|
||
''')
|
||
pdata_rows = cur.fetchall()
|
||
|
||
pdata_cols = ['ID', 'ID_PODANI', 'DAVKA_ROK', 'DAVKA_DISK', 'DAVKA_DATUMOD', 'DAVKA_DATUMDO',
|
||
'KDAVKA', 'FDAVKA', 'DATA']
|
||
ws4.append(pdata_cols)
|
||
|
||
WRAP = Alignment(wrap_text=True, vertical='top')
|
||
|
||
# Indexy BLOB sloupců: KDAVKA=6, FDAVKA=7, DATA=8
|
||
# DATA je XML v win1250, KDAVKA/FDAVKA jsou latin2
|
||
BLOB_ENCODINGS = {6: 'cp852', 7: 'cp852', 8: 'cp1250'}
|
||
|
||
def decode_blob(v, enc):
|
||
if hasattr(v, 'read'):
|
||
raw = v.read()
|
||
else:
|
||
raw = v
|
||
if not raw:
|
||
return ''
|
||
if isinstance(raw, str):
|
||
# fdb dekódoval jako win1250 – vrátíme zpět na bytes, pak dekódujeme správně
|
||
raw = raw.encode('cp1250', errors='replace')
|
||
return raw.decode(enc, errors='replace')
|
||
|
||
for i, row in enumerate(pdata_rows, start=2):
|
||
out = []
|
||
for col_idx, v in enumerate(row):
|
||
if col_idx in BLOB_ENCODINGS:
|
||
enc = BLOB_ENCODINGS[col_idx]
|
||
out.append(decode_blob(v, enc))
|
||
else:
|
||
out.append(fmt(v))
|
||
ws4.append(out)
|
||
if i % 2 == 0:
|
||
for cell in ws4[i]:
|
||
cell.fill = ZEBRA_FILL
|
||
for cell in ws4[i]:
|
||
cell.alignment = WRAP
|
||
ws4.row_dimensions[i].height = 80
|
||
|
||
style_header(ws4)
|
||
ws4.freeze_panes = 'A2'
|
||
|
||
# Šířky sloupců PORTAL_DATA
|
||
for col, width in zip(['A','B','C','D','E','F','G','H','I'], [6, 26, 8, 6, 12, 12, 80, 20, 60]):
|
||
ws4.column_dimensions[col].width = width
|
||
|
||
# =====================
|
||
# Hyperlinky PORTAL ↔ PORTAL_DATA
|
||
# =====================
|
||
|
||
# Mapa: portal_id → řádek v každém listu
|
||
portal_id_to_ws3_row = {}
|
||
for i, row in enumerate(portal_rows, start=2):
|
||
portal_id_to_ws3_row[row[0]] = i # row[0] = ID
|
||
|
||
portal_id_to_ws4_row = {}
|
||
for i, row in enumerate(pdata_rows, start=2):
|
||
portal_id_to_ws4_row[row[0]] = i # row[0] = ID
|
||
|
||
# PORTAL → PORTAL_DATA (sloupec A = ID)
|
||
for i, row in enumerate(portal_rows, start=2):
|
||
pid = row[0]
|
||
cell = ws3.cell(row=i, column=1)
|
||
if pid in portal_id_to_ws4_row:
|
||
cell.hyperlink = f'#PORTAL_DATA!A{portal_id_to_ws4_row[pid]}'
|
||
cell.font = LINK_FONT
|
||
|
||
# PORTAL_DATA → PORTAL (sloupec A = ID)
|
||
for i, row in enumerate(pdata_rows, start=2):
|
||
pid = row[0]
|
||
cell = ws4.cell(row=i, column=1)
|
||
if pid in portal_id_to_ws3_row:
|
||
cell.hyperlink = f'#PORTAL!A{portal_id_to_ws3_row[pid]}'
|
||
cell.font = LINK_FONT
|
||
cell.alignment = WRAP
|
||
|
||
# =====================
|
||
# Hyperlinky FAK → FAKDET
|
||
# =====================
|
||
|
||
# Mapa: IDFAK → první řádek na listu FAKDET (řádek 1 = záhlaví, data od 2)
|
||
idfak_to_row = {}
|
||
for i, row in enumerate(det_rows, start=2):
|
||
idfak = row[1] # IDFAK je druhý sloupec
|
||
if idfak not in idfak_to_row:
|
||
idfak_to_row[idfak] = i
|
||
|
||
# Přidat sloupec "→FAKDET" jako první sloupec na listu FAK
|
||
ws1.insert_cols(1)
|
||
ws1.cell(row=1, column=1, value='FAKDET').fill = HEADER_FILL
|
||
ws1.cell(row=1, column=1).font = HEADER_FONT
|
||
ws1.cell(row=1, column=1).alignment = Alignment(horizontal='center')
|
||
ws1.column_dimensions['A'].width = 9
|
||
|
||
for i, row in enumerate(fak_rows, start=2):
|
||
fak_id = row[0] # ID je první sloupec z DB
|
||
cell = ws1.cell(row=i, column=1)
|
||
if fak_id in idfak_to_row:
|
||
target_row = idfak_to_row[fak_id]
|
||
cell.value = '>> det'
|
||
cell.hyperlink = f'#FAKDET!A{target_row}'
|
||
cell.font = LINK_FONT
|
||
else:
|
||
cell.value = ''
|
||
|
||
# =====================
|
||
# List 5 – ED_PODANI (ED_BOOKOFSUBMISSIONS)
|
||
# =====================
|
||
|
||
ws5 = wb.create_sheet('ED_PODANI')
|
||
|
||
REQUESTTYPE_MAP = {0: 'Registrační', 1: 'Výkonová'}
|
||
HICCODE_MAP = {
|
||
'111': 'VZP', '201': 'VoZP', '205': 'ČPZP',
|
||
'207': 'OZP', '209': 'ZPŠ', '211': 'ZPMV',
|
||
'212': '212', '213': '213', '217': '217', '228': '228', '333': '333',
|
||
}
|
||
|
||
cur.execute('''
|
||
SELECT
|
||
ID, CREATED, SENTDATE, HICCODE, REQUESTTYPE,
|
||
SUBMISSIONID, INVOICENUMBER,
|
||
PERIODFROM, PERIODTO, TOTALSUM,
|
||
STATE, CREATOR, HCPPERSONNAME, HCPCODE, UNIQUEID
|
||
FROM ED_BOOKOFSUBMISSIONS
|
||
ORDER BY ID DESC
|
||
''')
|
||
ed_cols_raw = [d[0] for d in cur.description]
|
||
ed_rows = cur.fetchall()
|
||
|
||
# Lidštější záhlaví
|
||
ed_headers = [
|
||
'ID', 'Vytvořeno', 'Odesláno', 'ZP', 'Typ podání',
|
||
'Podací č.', 'Faktura',
|
||
'Období od', 'Období do', 'Částka (Kč)',
|
||
'Stav', 'Autor', 'Lékař', 'IČZ', 'UUID',
|
||
]
|
||
ws5.append(ed_headers)
|
||
|
||
# Mapa CISFAK → řádek na listu FAK (pro hyperlink)
|
||
cisfak_to_fak_row = {}
|
||
for i, row in enumerate(fak_rows, start=2):
|
||
cisfak = row[1] # CISFAK je druhý sloupec z DB (index 1)
|
||
if cisfak and cisfak not in cisfak_to_fak_row:
|
||
cisfak_to_fak_row[cisfak] = i
|
||
|
||
for i, row in enumerate(ed_rows, start=2):
|
||
(eid, created, sentdate, hiccode, reqtype,
|
||
submid, invoicenum,
|
||
perfrom, perto, totalsum,
|
||
state, creator, hcpperson, hcpcode, uniqueid) = row
|
||
|
||
out = [
|
||
eid,
|
||
fmt(created),
|
||
fmt(sentdate),
|
||
f"{hiccode} {HICCODE_MAP.get(str(hiccode), '')}" if hiccode else '',
|
||
REQUESTTYPE_MAP.get(reqtype, str(reqtype)) if reqtype is not None else '',
|
||
fmt(submid),
|
||
fmt(invoicenum),
|
||
fmt(perfrom),
|
||
fmt(perto),
|
||
float(totalsum) if totalsum is not None else '',
|
||
fmt(state),
|
||
fmt(creator),
|
||
fmt(hcpperson),
|
||
fmt(hcpcode),
|
||
fmt(uniqueid),
|
||
]
|
||
ws5.append(out)
|
||
if i % 2 == 0:
|
||
for cell in ws5[i]:
|
||
cell.fill = ZEBRA_FILL
|
||
|
||
style_header(ws5)
|
||
ws5.freeze_panes = 'A2'
|
||
autofit(ws5)
|
||
|
||
# Hyperlink: Faktura (sloupec 7) → list FAK
|
||
for i, row in enumerate(ed_rows, start=2):
|
||
invoicenum = row[6] # INVOICENUMBER
|
||
if invoicenum and invoicenum in cisfak_to_fak_row:
|
||
cell = ws5.cell(row=i, column=7)
|
||
# +1 kvůli vloženému sloupci FAKDET na listu FAK
|
||
cell.hyperlink = f'#FAK!B{cisfak_to_fak_row[invoicenum]}'
|
||
cell.font = LINK_FONT
|
||
|
||
# =====================
|
||
# List 6 – ED_PODANI_DATA (BLOBy z ED_BOOKOFSUBMISSIONS)
|
||
# =====================
|
||
|
||
ws6 = wb.create_sheet('ED_PODANI_DATA')
|
||
|
||
cur.execute('''
|
||
SELECT ID, HICCODE, SENTDATE, SUBMISSIONID, INVOICENUMBER,
|
||
KDAVKACONTENT, REQUEST, SERVERRESPONSE, PROTOCOL
|
||
FROM ED_BOOKOFSUBMISSIONS
|
||
ORDER BY ID DESC
|
||
''')
|
||
ed_data_rows = cur.fetchall()
|
||
|
||
ed_data_headers = ['ID', 'ZP', 'Odesláno', 'Podací č.', 'Faktura',
|
||
'KDAVKA', 'REQUEST (XML)', 'SERVERRESPONSE', 'PROTOCOL']
|
||
ws6.append(ed_data_headers)
|
||
|
||
def decode_ed_blob(v, enc):
|
||
"""Dekóduj BLOB z ED_BOOKOFSUBMISSIONS."""
|
||
if v is None:
|
||
return ''
|
||
if hasattr(v, 'read'):
|
||
raw = v.read()
|
||
else:
|
||
raw = v
|
||
if not raw:
|
||
return ''
|
||
if isinstance(raw, str):
|
||
# fdb vrátil jako win1250 string – zrekonstruuj bytes
|
||
raw = raw.encode('cp1250', errors='replace')
|
||
try:
|
||
text = raw.decode(enc, errors='replace')
|
||
except Exception:
|
||
return repr(raw[:200])
|
||
# Odstraň prázdné řádky a sjednoť odřádkování
|
||
lines = [l for l in text.splitlines() if l.strip()]
|
||
return '\n'.join(lines)
|
||
|
||
for i, row in enumerate(ed_data_rows, start=2):
|
||
eid, hiccode, sentdate, submid, invoicenum, kdavka, request, serverresp, protocol = row
|
||
|
||
# KDAVKACONTENT – fdb vrací str přes win1250 spojení, použij přímo
|
||
if kdavka is None:
|
||
kdavka_txt = ''
|
||
else:
|
||
raw = kdavka.read() if hasattr(kdavka, 'read') else kdavka
|
||
if isinstance(raw, bytes):
|
||
raw = raw.encode('latin-1', errors='replace').decode('cp852', errors='replace')
|
||
lines = [l for l in raw.splitlines() if l.strip()]
|
||
kdavka_txt = '\n'.join(lines)
|
||
|
||
# REQUEST – XML; může být uložen jako UTF-16 binary nebo jako ASCII/UTF-8 string
|
||
request_txt = ''
|
||
if request is not None:
|
||
raw = request.read() if hasattr(request, 'read') else request
|
||
if isinstance(raw, str):
|
||
raw_b = raw.encode('latin-1', errors='replace')
|
||
else:
|
||
raw_b = raw
|
||
if raw_b:
|
||
# Detekce podle BOM – jedině tehdy jde o skutečné UTF-16 binární data
|
||
if raw_b[:2] in (b'\xff\xfe', b'\xfe\xff'):
|
||
request_txt = raw_b.decode('utf-16', errors='replace')
|
||
elif raw_b[:1] == b'<':
|
||
# ASCII/UTF-8 XML – fdb ho vrátil jako string, použij přímo
|
||
request_txt = raw_b.decode('utf-8', errors='replace')
|
||
else:
|
||
request_txt = raw_b.decode('cp1250', errors='replace')
|
||
lines = [l for l in request_txt.splitlines() if l.strip()]
|
||
request_txt = '\n'.join(lines)
|
||
|
||
# SERVERRESPONSE a PROTOCOL – latin-1 re-encoding, pak iso-8859-2
|
||
def decode_latin_blob(v, enc):
|
||
if v is None:
|
||
return ''
|
||
raw = v.read() if hasattr(v, 'read') else v
|
||
if not raw:
|
||
return ''
|
||
if isinstance(raw, str):
|
||
raw = raw.encode('latin-1', errors='replace')
|
||
text = raw.decode(enc, errors='replace')
|
||
lines = [l for l in text.splitlines() if l.strip()]
|
||
return '\n'.join(lines)
|
||
|
||
serverresp_txt = decode_latin_blob(serverresp, 'iso-8859-2')
|
||
protocol_txt = decode_latin_blob(protocol, 'iso-8859-2')
|
||
|
||
out = [
|
||
eid,
|
||
f"{hiccode} {HICCODE_MAP.get(str(hiccode), '')}" if hiccode else '',
|
||
fmt(sentdate),
|
||
fmt(submid),
|
||
fmt(invoicenum),
|
||
kdavka_txt,
|
||
request_txt,
|
||
serverresp_txt,
|
||
protocol_txt,
|
||
]
|
||
ws6.append(out)
|
||
if i % 2 == 0:
|
||
for cell in ws6[i]:
|
||
cell.fill = ZEBRA_FILL
|
||
for cell in ws6[i]:
|
||
cell.alignment = WRAP
|
||
ws6.row_dimensions[i].height = 80
|
||
|
||
style_header(ws6)
|
||
ws6.freeze_panes = 'A2'
|
||
|
||
# Šířky sloupců ED_PODANI_DATA
|
||
for col, width in zip(['A','B','C','D','E','F','G','H','I'],
|
||
[6, 10, 12, 16, 14, 80, 60, 60, 40]):
|
||
ws6.column_dimensions[col].width = width
|
||
|
||
# Hyperlinky ED_PODANI ↔ ED_PODANI_DATA (přes ID)
|
||
ed_id_to_ws5_row = {row[0]: i for i, row in enumerate(ed_rows, start=2)}
|
||
ed_id_to_ws6_row = {row[0]: i for i, row in enumerate(ed_data_rows, start=2)}
|
||
|
||
# ED_PODANI sloupec 1 (ID) → ED_PODANI_DATA
|
||
for i, row in enumerate(ed_rows, start=2):
|
||
eid = row[0]
|
||
if eid in ed_id_to_ws6_row:
|
||
cell = ws5.cell(row=i, column=1)
|
||
cell.hyperlink = f'#ED_PODANI_DATA!A{ed_id_to_ws6_row[eid]}'
|
||
cell.font = LINK_FONT
|
||
|
||
# ED_PODANI_DATA sloupec 1 (ID) → ED_PODANI
|
||
for i, row in enumerate(ed_data_rows, start=2):
|
||
eid = row[0]
|
||
if eid in ed_id_to_ws5_row:
|
||
cell = ws6.cell(row=i, column=1)
|
||
cell.hyperlink = f'#ED_PODANI!A{ed_id_to_ws5_row[eid]}'
|
||
cell.font = LINK_FONT
|
||
cell.alignment = WRAP
|
||
|
||
# =====================
|
||
# Uložení
|
||
# =====================
|
||
|
||
conn.close()
|
||
wb.save(output_path)
|
||
sys.stdout.buffer.write(f'Ulozeno: {output_path}\n'.encode('utf-8'))
|
||
sys.stdout.buffer.write(f'FAK: {len(fak_rows)} radku, FAKDET: {len(det_rows)} radku, PORTAL: {len(portal_rows)} radku, ED_PODANI: {len(ed_rows)} radku\n'.encode('utf-8'))
|