276 lines
9.3 KiB
Python
276 lines
9.3 KiB
Python
import os
|
||
from collections import namedtuple
|
||
|
||
hlav = namedtuple("Hlavicka", ["typ", "char", "dtyp", "dico",
|
||
"dpob", "drok", "dmes", "dcid", "dpoc", "dbody",
|
||
"dfin", "ddpp", "dvdr1", "dvdr2", "ddtyp"])
|
||
av=namedtuple("Aveta",["typ","hcid","hstr","hpoc","hpor","hcpo","htpp","hico","hvar","hodb","hrod","hzdg","hkon","hicz",
|
||
"hcdz","hrez","hccel","hcbod","dtyp"])
|
||
vv=namedtuple("Vveta",["typ","vdat","vkod","vpoc","vodb","vdia","vbod","vtyp"])
|
||
zv=namedtuple("Zveta",["typ","hcid","hstr","hpoc","hpor","hico","hvar","hodb","hrod","hzdg","hrez","hccel","htyp"])
|
||
lv=namedtuple("Lveta",["typ","ldat","ltpr","lzvl","lkod","lmno","lfin","lvdg","ltyp"])
|
||
def lveta(retezec):
|
||
TYP= retezec[0:1]
|
||
LDAT= retezec[1:9]
|
||
LTPR= retezec[9:10]
|
||
LZVL= retezec[10:11]
|
||
LKOD= retezec[11:18]
|
||
LMNO=retezec[18:29]
|
||
LFIN= retezec[29:39]
|
||
LVDG= retezec[39:44]
|
||
VTYP= retezec[44:45]
|
||
print(TYP, LDAT, LTPR, LZVL,LKOD,LMNO,LFIN,LVDG,VTYP)
|
||
Lveta = lv(TYP, LDAT, LTPR, LZVL,LKOD,LMNO,LFIN,LVDG,VTYP)
|
||
print(Lveta)
|
||
|
||
def vveta(retezec):
|
||
TYP= retezec[0:1]
|
||
VDAT= retezec[1:9]
|
||
VKOD= retezec[9:14]
|
||
VPOC= retezec[14:15]
|
||
VODB=retezec[15:18]
|
||
VDIA= retezec[18:23]
|
||
VBOD= retezec[23:30]
|
||
VTYP= retezec[30:31]
|
||
print(TYP, VDAT, VKOD, VPOC,VODB,VDIA,VBOD,VTYP)
|
||
Vveta = vv(TYP, VDAT, VKOD, VPOC,VODB,VDIA,VBOD,VTYP)
|
||
print(Vveta)
|
||
|
||
def zveta(retezec):
|
||
TYP= retezec[0:1]
|
||
HCID= retezec[1:8]
|
||
HSTR= retezec[8:9]
|
||
HPOC= retezec[9:10]
|
||
HPOR= retezec[10:13]
|
||
HICO= retezec[13:21]
|
||
HVAR= retezec[21:27]
|
||
HODB= retezec[27:30]
|
||
HROD= retezec[30:40]
|
||
HZDG= retezec[40:45]
|
||
HREZ= retezec[45:55]
|
||
HCCEL= retezec[55:66]
|
||
DTYP= retezec[66:67]
|
||
print(TYP, HCID, HSTR, HPOC,HPOR,HICO,HVAR,HODB,HROD,HZDG,HREZ,HCCEL,DTYP )
|
||
Zveta = zv(TYP, HCID, HSTR, HPOC,HPOR,HICO,HVAR,HODB,HROD,HZDG,HREZ,HCCEL,DTYP )
|
||
print(Zveta)
|
||
|
||
def prvniradek(retezec):
|
||
TYP = retezec[0:1] # Typ věty „D“ – úvodní věta dávky
|
||
CHAR = retezec[1:2] # Charakter dávky (viz dále ) – určuje stav předkládaných dokladů v dávce
|
||
DTYP = retezec[2:4] # typ dávky, číselný kód
|
||
DICO = retezec[4:12] # IPZ
|
||
DPOB = retezec[12:16] # kod oblastního pracoviště pojišťovny
|
||
DROK = retezec[16:20] # rok
|
||
DMES = retezec[20:22] # měsíc
|
||
DCID = retezec[22:28] # Číslo dávky – jednoznačné číslo dávky v rámci PZS a roku
|
||
DPOC = retezec[28:31] # počet dokladů v dávce, slouží k ověření kompletnosti
|
||
DBODY = retezec[31:42] # nepovinné, počet bodů
|
||
DFIN = float(retezec[42:60]) # nepovinné, celková částka za doklady
|
||
DDPP = retezec[60:61] # druh pojistného plnění
|
||
DVDR1 = retezec[61:74]
|
||
DVDR2 = retezec[74:87]
|
||
DDTYP = retezec[87:88]
|
||
print(TYP, CHAR, DTYP, DICO, DPOB, DROK, DMES, DCID, DPOC, DBODY, DFIN, DDPP, DVDR1,DVDR2,DDTYP)
|
||
hlavicka=hlav(TYP, CHAR, DTYP, DICO, DPOB, DROK, DMES, DCID, DPOC, DBODY, DFIN, DDPP, DVDR1,DVDR2,DDTYP)
|
||
print(hlavicka.dvdr1)
|
||
|
||
def aveta(retezec):
|
||
TYP= retezec[0:1]
|
||
HCID= retezec[1:8]
|
||
HSTR= retezec[8:9]
|
||
HPOC= retezec[9:10]
|
||
HPOR= retezec[10:13]
|
||
HCPO= retezec[13:16]
|
||
HTPP= retezec[16:17]
|
||
HICO= retezec[17:25]
|
||
HVAR= retezec[25:31]
|
||
HODB= retezec[31:34]
|
||
HROD= retezec[34:44]
|
||
HZDG= retezec[44:49]
|
||
HKON= retezec[49:50]
|
||
HICZ= retezec[50:58]
|
||
HCDZ= retezec[58:65]
|
||
HREZ= retezec[65:75]
|
||
HCCEL= retezec[75:85]
|
||
HCBOD= retezec[85:92]
|
||
DTYP= retezec[92:93]
|
||
print(TYP, HCID, HSTR, HPOC,HPOR,HCPO,HTPP,HICO,HVAR,HODB,HROD,HZDG,HKON,HICZ,HCDZ,HREZ,HCCEL,HCBOD,DTYP )
|
||
Aveta = av(TYP, HCID, HSTR, HPOC,HPOR,HCPO,HTPP,HICO,HVAR,HODB,HROD,HZDG,HKON,HICZ,HCDZ,HREZ,HCCEL,HCBOD,DTYP)
|
||
print(Aveta)
|
||
|
||
|
||
|
||
konec=0
|
||
# with open(r"u:\Dropbox\!!!Days\Downloads Z230\Dávky\DataKDavka (6).111","r",encoding='cp852') as f:
|
||
# for importradku in f.readlines():
|
||
# if importradku[0]=="D":
|
||
# print(importradku)
|
||
# prvniradek(importradku)
|
||
# if importradku[0]=="A":
|
||
# konec=0
|
||
# print(importradku)
|
||
# aveta(importradku)
|
||
# if importradku[0] == "V":
|
||
# print(importradku)
|
||
# vveta(importradku)
|
||
# if importradku[0]=="Z":
|
||
# print(importradku)
|
||
# zveta(importradku)
|
||
# if importradku[0] == "L":
|
||
# konec = 1
|
||
# print(importradku)
|
||
# lveta(importradku)
|
||
# if konec==2:
|
||
# exit(0)
|
||
|
||
|
||
#010 Projdi vsechny davky
|
||
|
||
root_davky_dir=r"u:\Dropbox\!!!Days\Downloads Z230\Dávky\Data"
|
||
#010.0 os.walk
|
||
# for root, folders, files in os.walk(root_davky_dir):
|
||
# for file in files:
|
||
# print(file)
|
||
|
||
#010.1 pathlib
|
||
from pathlib import Path
|
||
|
||
# # all txt files in current dir
|
||
# for p in Path(".").glob("*.txt"):
|
||
# print(p)
|
||
|
||
from dataclasses import dataclass
|
||
from decimal import Decimal, InvalidOperation
|
||
import re
|
||
|
||
# ---- header spec (0-based) ----
|
||
# name, type, length, start, description (optional)
|
||
HEADER_SPEC = [
|
||
("TYP", "C", 1, 0, "Typ věty (má být 'D')"),
|
||
("CHAR", "C", 1, 1, "Charakter dávky (např. 'P')"),
|
||
("DTYP", "C", 2, 2, "Typ dávky (např. '98')"),
|
||
("DICO", "C", 8, 4, "IČZ PZS"),
|
||
("DPOB", "C", 4, 12, "Územní pracoviště"),
|
||
("DROK", "N", 4, 16, "Rok uzavření dokladů"),
|
||
("DMES", "N", 2, 20, "Měsíc uzavření dokladů"),
|
||
("DCID", "N", 6, 22, "Číslo dávky"),
|
||
("DPOC", "N", 3, 28, "Počet dokladů"),
|
||
("DBODY", "N", 11, 31, "Počet bodů v dávce (volitelné)"),
|
||
("DFIN", "$", 18, 42, "Celkem Kč (18.2, volitelné)"),
|
||
("DDPP", "C", 1, 60, "Druh pojistného vztahu"),
|
||
("DVDR1", "C", 13, 61, "Verze DR 1"),
|
||
("DVDR2", "C", 13, 74, "Verze DR 2"),
|
||
("DDTYP", "C", 1, 87, "Doplněk typu věty"),
|
||
]
|
||
|
||
def _slice0(s: str, start: int, length: int) -> str:
|
||
"""0-based safe slice."""
|
||
if s is None: return ""
|
||
return s[start:start+length] if start < len(s) else ""
|
||
|
||
def _parse_char(s: str):
|
||
s = (s or "").strip()
|
||
return s or None
|
||
|
||
def _parse_int(s: str):
|
||
if s is None: return None
|
||
digits = re.sub(r"\D", "", s)
|
||
return int(digits) if digits else None
|
||
|
||
def _parse_money(s: str):
|
||
"""Parse money field '18.2': spaces allowed, comma or dot; returns Decimal or None."""
|
||
if s is None: return None
|
||
s = s.replace("\u00A0", " ").strip().replace(" ", "").replace(",", ".")
|
||
if not s: return None
|
||
try:
|
||
# keep precision with Decimal
|
||
val = Decimal(s)
|
||
# optional: round to 2 decimal places
|
||
return val.quantize(Decimal("0.01"))
|
||
except (InvalidOperation, ValueError):
|
||
return None
|
||
|
||
def parse_davka_header_line(line: str, strict=True) -> dict:
|
||
"""
|
||
Parse header (first line) of a dávka according to HEADER_SPEC (0-based).
|
||
Returns dict with typed values. Raises if not a 'DP98' header (strict=True).
|
||
"""
|
||
if not line:
|
||
raise ValueError("Empty header line.")
|
||
if strict and not line.startswith("DP98"):
|
||
raise ValueError(f"Invalid header prefix {line[:4]!r}; expected 'DP98'.")
|
||
|
||
out = {"raw": line}
|
||
# field parsing per spec
|
||
for name, typ, length, start, *_ in HEADER_SPEC:
|
||
raw_val = _slice0(line, start, length)
|
||
if typ == "C":
|
||
out[name] = _parse_char(raw_val)
|
||
elif typ == "N":
|
||
out[name] = _parse_int(raw_val)
|
||
elif typ == "$":
|
||
out[name] = _parse_money(raw_val)
|
||
else:
|
||
out[name] = raw_val # fallback
|
||
|
||
# basic validations (soft)
|
||
if strict:
|
||
if out.get("TYP") != "D":
|
||
raise ValueError(f"TYP must be 'D', got {out.get('TYP')!r}")
|
||
if out.get("DTYP") != "98":
|
||
raise ValueError(f"DTYP must be '98', got {out.get('DTYP')!r}")
|
||
|
||
# Optional month sanity check (accepts '01'..'12' and '00')
|
||
m = out.get("DMES")
|
||
if m is not None and m not in range(0, 13): # 0..12
|
||
# keep value but you could raise or log here
|
||
pass
|
||
|
||
return out
|
||
|
||
|
||
|
||
|
||
|
||
# recursive
|
||
exts=("111","201","205","207","209","211")
|
||
keep={}
|
||
for soubor in Path(root_davky_dir).rglob("KDAVKA.*"):
|
||
if not soubor.is_file():
|
||
print(f"{soubor} neni platny soubor")
|
||
continue
|
||
if soubor.suffix.lstrip(".") not in exts:
|
||
print(f"{soubor} nema platnou koncovku")
|
||
continue
|
||
try:
|
||
with soubor.open("r",encoding="cp1250",errors="strict",newline="") as f:
|
||
if f.read(4)=="DP98": #takovy soubor chceme brat v potaz
|
||
keep[str(soubor)]={"File":soubor}
|
||
except:
|
||
continue
|
||
print(keep)
|
||
|
||
|
||
#020 Read headers of davky in keep
|
||
for soubor in keep: #
|
||
|
||
|
||
# def get_kdavka(root_dir, exts=("111","201","205","207","209","211")):
|
||
# root = Path(root_dir)
|
||
# wanted = {e.upper() for e in exts}
|
||
# keep = []
|
||
# for p in root.rglob("KDAVKA.*"):
|
||
# if not p.is_file():
|
||
# continue
|
||
# if p.suffix.lstrip(".").upper() not in wanted:
|
||
# continue
|
||
# try:
|
||
# with p.open("r", encoding="cp1250", errors="strict", newline="") as f:
|
||
# if f.read(4) == "DP98": # read just 4 chars, cp1250 is single-byte
|
||
# keep.append(p)
|
||
# except UnicodeDecodeError:
|
||
# # bad file — skip
|
||
# continue
|
||
# keep.sort()
|
||
# print(f"{len(keep)} davky (DP98) found under {root}")
|
||
# return keep |