notebookvb
This commit is contained in:
Binary file not shown.
@@ -1,135 +0,0 @@
|
|||||||
"""
|
|
||||||
Stahování seznamu registrovaných pojištěnců ČPZP.
|
|
||||||
|
|
||||||
Použij po 01_prihlaseni.py (ten uloží cpzp_cookies.json).
|
|
||||||
|
|
||||||
Co dělá:
|
|
||||||
- Načte cookies z cpzp_cookies.json
|
|
||||||
- Otevře prohlížeč jednou, projde všechny zadané měsíce
|
|
||||||
- Pro každý měsíc vyplní formulář, klikne Hledat, stáhne soubor
|
|
||||||
- Přeskočí měsíce kde soubor v cílovém adresáři už existuje
|
|
||||||
- Uloží jako: YYYY-MM-DD f205MMRR.123
|
|
||||||
|
|
||||||
NASTAVENÍ:
|
|
||||||
OD_MESIC / OD_ROK — první měsíc rozsahu
|
|
||||||
DO_MESIC / DO_ROK — poslední měsíc rozsahu (včetně)
|
|
||||||
"""
|
|
||||||
|
|
||||||
import glob
|
|
||||||
import json
|
|
||||||
import os
|
|
||||||
import sys
|
|
||||||
import time
|
|
||||||
from datetime import date
|
|
||||||
|
|
||||||
from playwright.sync_api import sync_playwright
|
|
||||||
|
|
||||||
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", ".."))
|
|
||||||
from Knihovny.najdi_dropbox import get_dropbox_root
|
|
||||||
|
|
||||||
OD_MESIC = 12
|
|
||||||
OD_ROK = 2024
|
|
||||||
DO_MESIC = 3
|
|
||||||
DO_ROK = 2026
|
|
||||||
|
|
||||||
BASE_URL = "https://portal.cpzp.cz"
|
|
||||||
COOKIES_FILE = os.path.join(os.path.dirname(__file__), "..", "..", "StahováníZpráv", "205 ČPZP", "cpzp_cookies.json")
|
|
||||||
DEST_DIR = os.path.join(
|
|
||||||
get_dropbox_root(),
|
|
||||||
"Ordinace", "Dokumentace_ke_zpracování", "Zúčtovací zprávy", "205 ČPZP",
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
def mesice_v_rozsahu(od_m, od_r, do_m, do_r):
|
|
||||||
"""Generuje (mesic, rok) od od_m/od_r do do_m/do_r včetně."""
|
|
||||||
m, r = od_m, od_r
|
|
||||||
while (r, m) <= (do_r, do_m):
|
|
||||||
yield m, r
|
|
||||||
m += 1
|
|
||||||
if m > 12:
|
|
||||||
m = 1
|
|
||||||
r += 1
|
|
||||||
|
|
||||||
|
|
||||||
def uz_stazeno(mesic: int, rok: int) -> bool:
|
|
||||||
"""Vrátí True pokud soubor pro daný měsíc/rok už existuje v DEST_DIR."""
|
|
||||||
mm = f"{mesic:02d}"
|
|
||||||
rr = str(rok)[-2:]
|
|
||||||
pattern = os.path.join(DEST_DIR, f"* f205{mm}{rr}.*")
|
|
||||||
return bool(glob.glob(pattern))
|
|
||||||
|
|
||||||
|
|
||||||
def stahni_mesic(page, mesic: int, rok: int) -> bool:
|
|
||||||
"""Stáhne soubor pro jeden měsíc. Vrátí True pokud staženo."""
|
|
||||||
today = date.today().strftime("%Y-%m-%d")
|
|
||||||
|
|
||||||
if uz_stazeno(mesic, rok):
|
|
||||||
print(f" [{mesic:02d}/{rok}] přeskočeno — soubor už existuje")
|
|
||||||
return False
|
|
||||||
|
|
||||||
# Vyplň formulář
|
|
||||||
inputs = page.query_selector_all("input[type=text]")
|
|
||||||
if len(inputs) < 2:
|
|
||||||
print(f" [{mesic:02d}/{rok}] CHYBA — inputy nenalezeny")
|
|
||||||
return False
|
|
||||||
|
|
||||||
inputs[0].fill(str(mesic))
|
|
||||||
inputs[1].fill(str(rok))
|
|
||||||
|
|
||||||
page.get_by_text("Hledat", exact=True).click()
|
|
||||||
page.wait_for_load_state("networkidle")
|
|
||||||
|
|
||||||
dl_selector = "a:has-text('Seznam registrovaných pojištěnců')"
|
|
||||||
if not page.query_selector(dl_selector):
|
|
||||||
print(f" [{mesic:02d}/{rok}] CHYBA — download odkaz nenalezen")
|
|
||||||
return False
|
|
||||||
|
|
||||||
with page.expect_download() as dl_info:
|
|
||||||
page.click(dl_selector)
|
|
||||||
download = dl_info.value
|
|
||||||
|
|
||||||
original_name = download.suggested_filename
|
|
||||||
dest_path = os.path.join(DEST_DIR, f"{today} {original_name}")
|
|
||||||
download.save_as(dest_path)
|
|
||||||
print(f" [{mesic:02d}/{rok}] OK — {os.path.basename(dest_path)}")
|
|
||||||
return True
|
|
||||||
|
|
||||||
|
|
||||||
def hlavni() -> None:
|
|
||||||
if not os.path.exists(COOKIES_FILE):
|
|
||||||
raise SystemExit(f"Soubor s cookies nenalezen: {COOKIES_FILE}\nNejdřív spusť 01_prihlaseni.py")
|
|
||||||
|
|
||||||
with open(COOKIES_FILE, encoding="utf-8") as f:
|
|
||||||
cookies = json.load(f)
|
|
||||||
|
|
||||||
os.makedirs(DEST_DIR, exist_ok=True)
|
|
||||||
|
|
||||||
mesice = list(mesice_v_rozsahu(OD_MESIC, OD_ROK, DO_MESIC, DO_ROK))
|
|
||||||
print(f"Celkem měsíců: {len(mesice)} ({OD_MESIC:02d}/{OD_ROK} – {DO_MESIC:02d}/{DO_ROK})")
|
|
||||||
|
|
||||||
with sync_playwright() as p:
|
|
||||||
browser = p.chromium.launch(headless=False)
|
|
||||||
context = browser.new_context()
|
|
||||||
context.add_cookies(cookies)
|
|
||||||
page = context.new_page()
|
|
||||||
|
|
||||||
print("Otevírám stránku klientely...")
|
|
||||||
page.goto(f"{BASE_URL}/app/prohlizeni-klientely/")
|
|
||||||
page.wait_for_load_state("networkidle")
|
|
||||||
|
|
||||||
if "frmPrihlasCert" in page.content():
|
|
||||||
raise SystemExit("Cookies expirovala — nejdřív spusť 01_prihlaseni.py")
|
|
||||||
|
|
||||||
stazeno = 0
|
|
||||||
for mesic, rok in mesice:
|
|
||||||
if stahni_mesic(page, mesic, rok):
|
|
||||||
stazeno += 1
|
|
||||||
time.sleep(2)
|
|
||||||
|
|
||||||
browser.close()
|
|
||||||
|
|
||||||
print(f"\nHotovo: {stazeno} staženo, {len(mesice) - stazeno} přeskočeno.")
|
|
||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
|
||||||
hlavni()
|
|
||||||
@@ -0,0 +1,258 @@
|
|||||||
|
"""
|
||||||
|
Stahování seznamu registrovaných pojištěnců ČPZP — samostatný skript.
|
||||||
|
|
||||||
|
Co dělá:
|
||||||
|
- Přihlásí se certifikátem na portál ČPZP (nevyžaduje 01_prihlaseni.py)
|
||||||
|
- Zjistí poslední stažený měsíc (ze stavového souboru 05_stav.json nebo z existujících souborů)
|
||||||
|
- Zkusí stáhnout právě 1 následující měsíc
|
||||||
|
- Při úspěchu (staženo nebo soubor už existoval) uloží nový stav
|
||||||
|
- Uloží jako: YYYY-MM-DD f205MMRR.123
|
||||||
|
|
||||||
|
Stavový soubor: 05_stav.json vedle tohoto skriptu.
|
||||||
|
{"mesic": 1, "rok": 2026} — poslední úspěšně zpracovaný měsíc
|
||||||
|
"""
|
||||||
|
|
||||||
|
import glob
|
||||||
|
import json
|
||||||
|
import os
|
||||||
|
import re
|
||||||
|
import sys
|
||||||
|
import time
|
||||||
|
from datetime import date
|
||||||
|
|
||||||
|
import requests
|
||||||
|
from bs4 import BeautifulSoup
|
||||||
|
from cryptography.hazmat.primitives import hashes, serialization
|
||||||
|
from cryptography.hazmat.primitives.serialization import pkcs7, pkcs12
|
||||||
|
from playwright.sync_api import sync_playwright
|
||||||
|
|
||||||
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", ".."))
|
||||||
|
from Knihovny.najdi_dropbox import get_dropbox_root
|
||||||
|
|
||||||
|
PFX_PATH = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", "Certificates", "MBQualifiedCert.pfx"))
|
||||||
|
PFX_PASSWORD = b"Vlado7309208104++"
|
||||||
|
|
||||||
|
BASE_URL = "https://portal.cpzp.cz"
|
||||||
|
DEST_DIR = os.path.join(
|
||||||
|
get_dropbox_root(),
|
||||||
|
"Ordinace", "Dokumentace_ke_zpracování", "Zúčtovací zprávy", "SeznamyPojištěnců",
|
||||||
|
)
|
||||||
|
STATE_FILE = os.path.join(os.path.dirname(__file__), "05_stav.json")
|
||||||
|
|
||||||
|
|
||||||
|
def _parse_js_str(js_expr: str) -> str:
|
||||||
|
"""Převede JS string expression ('a' + String.fromCharCode(13,10) + 'b') na Python string."""
|
||||||
|
parts = re.findall(r"'([^']*)'|String\.fromCharCode\(([\d,\s]+)\)", js_expr)
|
||||||
|
result = []
|
||||||
|
for str_part, chars_part in parts:
|
||||||
|
if str_part is not None and (str_part or not chars_part):
|
||||||
|
result.append(str_part)
|
||||||
|
elif chars_part:
|
||||||
|
result.append("".join(chr(int(c.strip())) for c in chars_part.split(",")))
|
||||||
|
return "".join(result)
|
||||||
|
|
||||||
|
|
||||||
|
def prihlaseni() -> list[dict]:
|
||||||
|
"""Přihlásí se certifikátem a vrátí cookies ve formátu pro Playwright."""
|
||||||
|
login_url = f"{BASE_URL}/app/login/"
|
||||||
|
post_url = f"{BASE_URL}/app/"
|
||||||
|
|
||||||
|
session = requests.Session()
|
||||||
|
session.headers.update({
|
||||||
|
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
|
||||||
|
"Origin": BASE_URL,
|
||||||
|
"Referer": login_url,
|
||||||
|
})
|
||||||
|
|
||||||
|
r = session.get(login_url)
|
||||||
|
r.raise_for_status()
|
||||||
|
|
||||||
|
soup = BeautifulSoup(r.content, "html.parser", from_encoding="utf-8")
|
||||||
|
form = soup.find("form", {"name": "frmPrihlasCert"})
|
||||||
|
if not form:
|
||||||
|
raise RuntimeError("Formulář frmPrihlasCert nenalezen — portál nedostupný nebo změnil strukturu")
|
||||||
|
csrf_cert = form.find("input", {"name": "csrfCert"})["value"]
|
||||||
|
|
||||||
|
html = r.content.decode("utf-8")
|
||||||
|
match = re.search(r"certificateLoginKey\s*:\s*(.+?)(?=,\s*\n|\n\s*maxHeight)", html, re.DOTALL)
|
||||||
|
if not match:
|
||||||
|
raise RuntimeError("certificateLoginKey nenalezen v HTML")
|
||||||
|
challenge = _parse_js_str(match.group(1))
|
||||||
|
print(f"Challenge: {challenge[:60]}...")
|
||||||
|
|
||||||
|
with open(PFX_PATH, "rb") as f:
|
||||||
|
private_key, cert, _ = pkcs12.load_key_and_certificates(f.read(), PFX_PASSWORD)
|
||||||
|
|
||||||
|
pem_podpis = (
|
||||||
|
pkcs7.PKCS7SignatureBuilder()
|
||||||
|
.set_data(challenge.encode("utf-8"))
|
||||||
|
.add_signer(cert, private_key, hashes.SHA256())
|
||||||
|
.sign(serialization.Encoding.PEM, [pkcs7.PKCS7Options.DetachedSignature])
|
||||||
|
)
|
||||||
|
|
||||||
|
r = session.post(post_url, data={
|
||||||
|
"csrfCert": csrf_cert,
|
||||||
|
"sign": pem_podpis.decode("ascii").strip(),
|
||||||
|
}, headers={
|
||||||
|
"Content-Type": "application/x-www-form-urlencoded",
|
||||||
|
"Referer": login_url,
|
||||||
|
})
|
||||||
|
r.raise_for_status()
|
||||||
|
|
||||||
|
if "frmPrihlasCert" in r.text:
|
||||||
|
raise RuntimeError("Přihlášení selhalo — server vrátil login stránku znovu")
|
||||||
|
|
||||||
|
print("Přihlášení úspěšné!")
|
||||||
|
|
||||||
|
return [
|
||||||
|
{
|
||||||
|
"name": c.name,
|
||||||
|
"value": c.value,
|
||||||
|
"domain": c.domain if c.domain.startswith(".") else "." + c.domain,
|
||||||
|
"path": c.path or "/",
|
||||||
|
"expires": int(c.expires) if c.expires else -1,
|
||||||
|
"secure": bool(c.secure),
|
||||||
|
"httpOnly": False,
|
||||||
|
"sameSite": "Lax",
|
||||||
|
}
|
||||||
|
for c in session.cookies
|
||||||
|
]
|
||||||
|
|
||||||
|
|
||||||
|
def dalsi_mesic(mesic: int, rok: int) -> tuple[int, int]:
|
||||||
|
mesic += 1
|
||||||
|
if mesic > 12:
|
||||||
|
mesic = 1
|
||||||
|
rok += 1
|
||||||
|
return mesic, rok
|
||||||
|
|
||||||
|
|
||||||
|
def uz_stazeno(mesic: int, rok: int) -> bool:
|
||||||
|
"""Vrátí True pokud soubor pro daný měsíc/rok už existuje v DEST_DIR."""
|
||||||
|
mm = f"{mesic:02d}"
|
||||||
|
rr = str(rok)[-2:]
|
||||||
|
pattern = os.path.join(DEST_DIR, f"* f205{mm}{rr}.*")
|
||||||
|
return bool(glob.glob(pattern))
|
||||||
|
|
||||||
|
|
||||||
|
def zjisti_posledni_ze_souboru() -> tuple[int, int] | None:
|
||||||
|
"""Prohledá DEST_DIR a zjistí nejnovější stažený měsíc/rok z názvů souborů (f205MMRR)."""
|
||||||
|
pattern = os.path.join(DEST_DIR, "* f205????.* ")
|
||||||
|
# Hledáme soubory tvaru '* f205MMRR.*'
|
||||||
|
soubory = glob.glob(os.path.join(DEST_DIR, "* f205????.* "))
|
||||||
|
if not soubory:
|
||||||
|
# zkus bez mezery na konci
|
||||||
|
soubory = glob.glob(os.path.join(DEST_DIR, "* f205????.*"))
|
||||||
|
kandidati = []
|
||||||
|
for s in soubory:
|
||||||
|
nazev = os.path.basename(s)
|
||||||
|
m = re.search(r"f205(\d{2})(\d{2})\.", nazev)
|
||||||
|
if m:
|
||||||
|
mm = int(m.group(1))
|
||||||
|
rr = int(m.group(2))
|
||||||
|
rok = 2000 + rr
|
||||||
|
if 1 <= mm <= 12:
|
||||||
|
kandidati.append((rok, mm))
|
||||||
|
if not kandidati:
|
||||||
|
return None
|
||||||
|
posledni_rok, posledni_mesic = max(kandidati)
|
||||||
|
return posledni_mesic, posledni_rok
|
||||||
|
|
||||||
|
|
||||||
|
def nacti_stav() -> tuple[int, int] | None:
|
||||||
|
"""Načte poslední zpracovaný měsíc/rok. Vrátí None pokud není žádná informace."""
|
||||||
|
if os.path.exists(STATE_FILE):
|
||||||
|
with open(STATE_FILE, encoding="utf-8") as f:
|
||||||
|
data = json.load(f)
|
||||||
|
print(f"Stav ze souboru: poslední zpracovaný {data['mesic']:02d}/{data['rok']}")
|
||||||
|
return data["mesic"], data["rok"]
|
||||||
|
ze_souboru = zjisti_posledni_ze_souboru()
|
||||||
|
if ze_souboru:
|
||||||
|
m, r = ze_souboru
|
||||||
|
print(f"Stav zjištěn ze souborů: poslední stažený {m:02d}/{r}")
|
||||||
|
return m, r
|
||||||
|
return None
|
||||||
|
|
||||||
|
|
||||||
|
def uloz_stav(mesic: int, rok: int) -> None:
|
||||||
|
with open(STATE_FILE, "w", encoding="utf-8") as f:
|
||||||
|
json.dump({"mesic": mesic, "rok": rok}, f, ensure_ascii=False)
|
||||||
|
|
||||||
|
|
||||||
|
def stahni_mesic(page, mesic: int, rok: int) -> bool:
|
||||||
|
"""Stáhne soubor pro jeden měsíc. Vrátí True pokud staženo."""
|
||||||
|
today = date.today().strftime("%Y-%m-%d")
|
||||||
|
|
||||||
|
if uz_stazeno(mesic, rok):
|
||||||
|
print(f" [{mesic:02d}/{rok}] přeskočeno — soubor už existuje")
|
||||||
|
return True # považujeme za úspěch pro účely uložení stavu
|
||||||
|
|
||||||
|
# Vyplň formulář
|
||||||
|
inputs = page.query_selector_all("input[type=text]")
|
||||||
|
if len(inputs) < 2:
|
||||||
|
print(f" [{mesic:02d}/{rok}] CHYBA — inputy nenalezeny")
|
||||||
|
return False
|
||||||
|
|
||||||
|
inputs[0].fill(str(mesic))
|
||||||
|
inputs[1].fill(str(rok))
|
||||||
|
|
||||||
|
page.get_by_text("Hledat", exact=True).click()
|
||||||
|
page.wait_for_load_state("networkidle")
|
||||||
|
|
||||||
|
dl_selector = "a:has-text('Seznam registrovaných pojištěnců')"
|
||||||
|
if not page.query_selector(dl_selector):
|
||||||
|
print(f" [{mesic:02d}/{rok}] CHYBA — download odkaz nenalezen (soubor zřejmě ještě není k dispozici)")
|
||||||
|
return False
|
||||||
|
|
||||||
|
with page.expect_download() as dl_info:
|
||||||
|
page.click(dl_selector)
|
||||||
|
download = dl_info.value
|
||||||
|
|
||||||
|
original_name = download.suggested_filename
|
||||||
|
dest_path = os.path.join(DEST_DIR, f"{today} {original_name}")
|
||||||
|
download.save_as(dest_path)
|
||||||
|
print(f" [{mesic:02d}/{rok}] OK — {os.path.basename(dest_path)}")
|
||||||
|
return True
|
||||||
|
|
||||||
|
|
||||||
|
def hlavni() -> None:
|
||||||
|
os.makedirs(DEST_DIR, exist_ok=True)
|
||||||
|
|
||||||
|
cookies = prihlaseni()
|
||||||
|
|
||||||
|
posledni = nacti_stav()
|
||||||
|
if posledni is None:
|
||||||
|
raise SystemExit(
|
||||||
|
"Nelze zjistit poslední stažený měsíc.\n"
|
||||||
|
f"Vytvoř ručně soubor {STATE_FILE} ve tvaru:\n"
|
||||||
|
' {"mesic": 12, "rok": 2025}'
|
||||||
|
)
|
||||||
|
|
||||||
|
mesic, rok = dalsi_mesic(*posledni)
|
||||||
|
print(f"Zkouším stáhnout: {mesic:02d}/{rok}")
|
||||||
|
|
||||||
|
with sync_playwright() as p:
|
||||||
|
browser = p.chromium.launch(headless=False)
|
||||||
|
context = browser.new_context()
|
||||||
|
context.add_cookies(cookies)
|
||||||
|
page = context.new_page()
|
||||||
|
|
||||||
|
print("Otevírám stránku klientely...")
|
||||||
|
page.goto(f"{BASE_URL}/app/prohlizeni-klientely/")
|
||||||
|
page.wait_for_load_state("networkidle")
|
||||||
|
|
||||||
|
if "frmPrihlasCert" in page.content():
|
||||||
|
raise SystemExit("Portál vyžaduje přihlášení i po předání cookies — zkontroluj certifikát nebo session")
|
||||||
|
|
||||||
|
uspech = stahni_mesic(page, mesic, rok)
|
||||||
|
browser.close()
|
||||||
|
|
||||||
|
if uspech:
|
||||||
|
uloz_stav(mesic, rok)
|
||||||
|
print(f"\nHotovo — stav uložen: {mesic:02d}/{rok}")
|
||||||
|
else:
|
||||||
|
print(f"\nNestaženo — stav nebyl aktualizován, příště se zkusí znovu {mesic:02d}/{rok}")
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
hlavni()
|
||||||
|
Before Width: | Height: | Size: 48 KiB After Width: | Height: | Size: 48 KiB |
|
Before Width: | Height: | Size: 48 KiB After Width: | Height: | Size: 48 KiB |
@@ -0,0 +1,237 @@
|
|||||||
|
"""
|
||||||
|
Podávání žádostí o výpis registrovaných pojištěnců ZPŠ — čistý Python, bez prohlížeče.
|
||||||
|
|
||||||
|
Co dělá:
|
||||||
|
- Přihlásí se certifikátem na portál ZPŠ (requests + cryptography)
|
||||||
|
- Sestaví XML žádosti, podepíše certifikátem (PKCS7/SHA-256)
|
||||||
|
- Odešle POST na JSON API portálu
|
||||||
|
- Zjistí poslední podaný měsíc ze stavového souboru stav.json
|
||||||
|
- Podá žádost pro 1 následující měsíc (poslední den daného měsíce)
|
||||||
|
- Při úspěchu uloží nový stav a referenční číslo do logu
|
||||||
|
|
||||||
|
Stavový soubor: stav.json vedle tohoto skriptu.
|
||||||
|
{"mesic": 2, "rok": 2025} — poslední úspěšně podaný měsíc
|
||||||
|
|
||||||
|
Log podání: log_podani.json — seznam { datum, ref_cislo, podano_kdy }
|
||||||
|
"""
|
||||||
|
|
||||||
|
import calendar
|
||||||
|
import json
|
||||||
|
import os
|
||||||
|
import re
|
||||||
|
import sys
|
||||||
|
from datetime import date, datetime
|
||||||
|
|
||||||
|
import requests
|
||||||
|
from cryptography.hazmat.primitives import hashes, serialization
|
||||||
|
from cryptography.hazmat.primitives.serialization import pkcs7, pkcs12
|
||||||
|
|
||||||
|
PFX_PATH = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", "Certificates", "MBQualifiedCert.pfx"))
|
||||||
|
PFX_PASSWORD = b"Vlado7309208104++"
|
||||||
|
|
||||||
|
BASE_URL = "https://portal.zpskoda.cz"
|
||||||
|
SUBMIT_URL = f"{BASE_URL}/json-api/formular-schranky/29-vypis-registrov-pojistencu/ulozit-formular"
|
||||||
|
ICZ = "25520"
|
||||||
|
|
||||||
|
STATE_FILE = os.path.join(os.path.dirname(__file__), "stav.json")
|
||||||
|
LOG_FILE = os.path.join(os.path.dirname(__file__), "log_podani.json")
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# Přihlášení
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
def prihlaseni() -> requests.Session:
|
||||||
|
"""Přihlásí se certifikátem, vrátí autentizovanou session."""
|
||||||
|
challenge_url = f"{BASE_URL}/json-api/prihlaseni/prihlasovaci-zprava"
|
||||||
|
certlogin_url = f"{BASE_URL}/json-api/prihlaseni/prihlaseni-certifikatem"
|
||||||
|
|
||||||
|
session = requests.Session()
|
||||||
|
session.headers.update({
|
||||||
|
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
|
||||||
|
"X-Requested-With": "XMLHttpRequest",
|
||||||
|
"Origin": BASE_URL,
|
||||||
|
"Referer": BASE_URL + "/",
|
||||||
|
})
|
||||||
|
|
||||||
|
r = session.get(f"{BASE_URL}/app/prihlaseni")
|
||||||
|
r.raise_for_status()
|
||||||
|
session.cookies.set("pzp_sign", "CERT", domain="portal.zpskoda.cz", path="/")
|
||||||
|
|
||||||
|
r = session.post(challenge_url, json={"login_sign": "CERT"},
|
||||||
|
headers={"Content-Type": "application/json; charset=UTF-8"})
|
||||||
|
r.raise_for_status()
|
||||||
|
zprava = r.json()["data"]["zprava"]
|
||||||
|
print(f"Challenge: {zprava[:60]}...")
|
||||||
|
|
||||||
|
with open(PFX_PATH, "rb") as f:
|
||||||
|
private_key, cert, _ = pkcs12.load_key_and_certificates(f.read(), PFX_PASSWORD)
|
||||||
|
|
||||||
|
podpis = (
|
||||||
|
pkcs7.PKCS7SignatureBuilder()
|
||||||
|
.set_data(zprava.encode("utf-8"))
|
||||||
|
.add_signer(cert, private_key, hashes.SHA256())
|
||||||
|
.sign(serialization.Encoding.PEM, [pkcs7.PKCS7Options.DetachedSignature])
|
||||||
|
.decode("ascii").strip()
|
||||||
|
)
|
||||||
|
|
||||||
|
r = session.post(certlogin_url, json={"zprava": zprava, "podpis": podpis},
|
||||||
|
headers={"Content-Type": "application/json; charset=UTF-8"})
|
||||||
|
r.raise_for_status()
|
||||||
|
data = r.json()["data"]
|
||||||
|
|
||||||
|
if not data.get("prihlasen"):
|
||||||
|
raise RuntimeError(f"Přihlášení selhalo: {r.json().get('errMsg', '')}")
|
||||||
|
|
||||||
|
print("Přihlášení úspěšné!")
|
||||||
|
return session
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# Sestavení XML a podpis
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
def build_xml(datum: date) -> str:
|
||||||
|
datum_str = datum.strftime("%d.%m.%Y")
|
||||||
|
return (
|
||||||
|
f'<SchrankaZadost NazevSchranky="VypisPojKap" NazevFiltru="ZZ_VYP_REG">\r\n'
|
||||||
|
f'<PolozkaFiltru Nazev="icz">{ICZ}</PolozkaFiltru>\r\n'
|
||||||
|
f'<PolozkaFiltru Nazev="datum">{datum_str}</PolozkaFiltru>\r\n'
|
||||||
|
f'<PolozkaFiltru Nazev="razeni">jmeno</PolozkaFiltru>\r\n'
|
||||||
|
f'<PolozkaFiltru Nazev="typ">soubor</PolozkaFiltru>\r\n'
|
||||||
|
f'</SchrankaZadost>'
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def sign_xml(xml: str) -> str:
|
||||||
|
"""Podepíše XML certifikátem, vrátí PKCS7 PEM s \\r\\n (stejný formát jako NMSigner)."""
|
||||||
|
with open(PFX_PATH, "rb") as f:
|
||||||
|
private_key, cert, _ = pkcs12.load_key_and_certificates(f.read(), PFX_PASSWORD)
|
||||||
|
|
||||||
|
pem = (
|
||||||
|
pkcs7.PKCS7SignatureBuilder()
|
||||||
|
.set_data(xml.encode("utf-8"))
|
||||||
|
.add_signer(cert, private_key, hashes.SHA256())
|
||||||
|
.sign(serialization.Encoding.PEM, [pkcs7.PKCS7Options.DetachedSignature, pkcs7.PKCS7Options.NoCerts])
|
||||||
|
.decode("ascii")
|
||||||
|
)
|
||||||
|
# NMSigner normalizuje konce řádků na \r\n
|
||||||
|
return pem.replace("\r\n", "\n").replace("\n", "\r\n")
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# Odeslání žádosti
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
def odeslat_zadost(session: requests.Session, datum: date) -> str | None:
|
||||||
|
"""Odešle podepsanou žádost na portál. Vrátí referenční číslo nebo None."""
|
||||||
|
xml = build_xml(datum)
|
||||||
|
podpis = sign_xml(xml)
|
||||||
|
|
||||||
|
payload = {"schrXml": xml, "schrSign": podpis, "schrFiles": []}
|
||||||
|
|
||||||
|
r = session.post(SUBMIT_URL, json=payload, headers={
|
||||||
|
"Content-Type": "application/json; charset=UTF-8",
|
||||||
|
"X-Requested-With": "XMLHttpRequest",
|
||||||
|
"Referer": BASE_URL + "/",
|
||||||
|
})
|
||||||
|
r.raise_for_status()
|
||||||
|
|
||||||
|
try:
|
||||||
|
resp = r.json()
|
||||||
|
except Exception:
|
||||||
|
print(f" Odpověď není JSON: {r.text[:300]}")
|
||||||
|
return None
|
||||||
|
|
||||||
|
# Hledáme referenční číslo v odpovědi
|
||||||
|
resp_str = json.dumps(resp, ensure_ascii=False)
|
||||||
|
m = re.search(r'\b(17\d{7}|18\d{7})\b', resp_str)
|
||||||
|
ref = m.group(1) if m else None
|
||||||
|
|
||||||
|
if resp.get("errMsg") or resp.get("error"):
|
||||||
|
print(f" Chyba od serveru: {resp.get('errMsg') or resp.get('error')}")
|
||||||
|
return None
|
||||||
|
|
||||||
|
if ref:
|
||||||
|
print(f" OK — ref. číslo: {ref}")
|
||||||
|
else:
|
||||||
|
print(f" Odpověď (bez ref. čísla): {resp_str[:300]}")
|
||||||
|
|
||||||
|
return ref or ("OK" if r.ok else None)
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# Stav a log
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
def dalsi_mesic(mesic: int, rok: int) -> tuple[int, int]:
|
||||||
|
mesic += 1
|
||||||
|
if mesic > 12:
|
||||||
|
mesic = 1
|
||||||
|
rok += 1
|
||||||
|
return mesic, rok
|
||||||
|
|
||||||
|
|
||||||
|
def posledni_den(mesic: int, rok: int) -> date:
|
||||||
|
_, last = calendar.monthrange(rok, mesic)
|
||||||
|
return date(rok, mesic, last)
|
||||||
|
|
||||||
|
|
||||||
|
def nacti_stav() -> tuple[int, int] | None:
|
||||||
|
if os.path.exists(STATE_FILE):
|
||||||
|
with open(STATE_FILE, encoding="utf-8") as f:
|
||||||
|
data = json.load(f)
|
||||||
|
print(f"Stav: poslední podaný {data['mesic']:02d}/{data['rok']}")
|
||||||
|
return data["mesic"], data["rok"]
|
||||||
|
return None
|
||||||
|
|
||||||
|
|
||||||
|
def uloz_stav(mesic: int, rok: int) -> None:
|
||||||
|
with open(STATE_FILE, "w", encoding="utf-8") as f:
|
||||||
|
json.dump({"mesic": mesic, "rok": rok}, f, ensure_ascii=False)
|
||||||
|
|
||||||
|
|
||||||
|
def uloz_log(mesic: int, rok: int, ref_cislo: str) -> None:
|
||||||
|
log = []
|
||||||
|
if os.path.exists(LOG_FILE):
|
||||||
|
with open(LOG_FILE, encoding="utf-8") as f:
|
||||||
|
log = json.load(f)
|
||||||
|
log.append({
|
||||||
|
"datum": posledni_den(mesic, rok).strftime("%d.%m.%Y"),
|
||||||
|
"ref_cislo": ref_cislo,
|
||||||
|
"podano_kdy": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
|
||||||
|
})
|
||||||
|
with open(LOG_FILE, "w", encoding="utf-8") as f:
|
||||||
|
json.dump(log, f, indent=2, ensure_ascii=False)
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# Hlavní funkce
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
def hlavni() -> None:
|
||||||
|
posledni = nacti_stav()
|
||||||
|
if posledni is None:
|
||||||
|
raise SystemExit(
|
||||||
|
"Nelze zjistit poslední podaný měsíc.\n"
|
||||||
|
f"Vytvoř ručně soubor {STATE_FILE} ve tvaru:\n"
|
||||||
|
' {"mesic": 2, "rok": 2025}'
|
||||||
|
)
|
||||||
|
|
||||||
|
mesic, rok = dalsi_mesic(*posledni)
|
||||||
|
datum = posledni_den(mesic, rok)
|
||||||
|
print(f"Podávám žádost pro: {datum.strftime('%d.%m.%Y')}")
|
||||||
|
|
||||||
|
session = prihlaseni()
|
||||||
|
ref = odeslat_zadost(session, datum)
|
||||||
|
|
||||||
|
if ref:
|
||||||
|
uloz_stav(mesic, rok)
|
||||||
|
uloz_log(mesic, rok, ref)
|
||||||
|
print(f"\nHotovo — stav uložen: {mesic:02d}/{rok}, ref: {ref}")
|
||||||
|
else:
|
||||||
|
print(f"\nPodání selhalo — stav nebyl aktualizován, příště se zkusí znovu {datum.strftime('%d.%m.%Y')}")
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
hlavni()
|
||||||
@@ -0,0 +1,72 @@
|
|||||||
|
[
|
||||||
|
{
|
||||||
|
"datum": "31.03.2025",
|
||||||
|
"ref_cislo": "178201286",
|
||||||
|
"podano_kdy": "2026-05-12 21:57:24"
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"datum": "30.04.2025",
|
||||||
|
"ref_cislo": "178201295",
|
||||||
|
"podano_kdy": "2026-05-12 21:58:18"
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"datum": "31.05.2025",
|
||||||
|
"ref_cislo": "178201296",
|
||||||
|
"podano_kdy": "2026-05-12 21:58:24"
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"datum": "30.06.2025",
|
||||||
|
"ref_cislo": "178201297",
|
||||||
|
"podano_kdy": "2026-05-12 21:58:31"
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"datum": "31.07.2025",
|
||||||
|
"ref_cislo": "178201298",
|
||||||
|
"podano_kdy": "2026-05-12 21:58:38"
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"datum": "31.08.2025",
|
||||||
|
"ref_cislo": "178201299",
|
||||||
|
"podano_kdy": "2026-05-12 21:58:45"
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"datum": "30.09.2025",
|
||||||
|
"ref_cislo": "178201300",
|
||||||
|
"podano_kdy": "2026-05-12 21:58:50"
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"datum": "31.10.2025",
|
||||||
|
"ref_cislo": "178201301",
|
||||||
|
"podano_kdy": "2026-05-12 21:58:56"
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"datum": "30.11.2025",
|
||||||
|
"ref_cislo": "178201302",
|
||||||
|
"podano_kdy": "2026-05-12 21:59:02"
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"datum": "31.12.2025",
|
||||||
|
"ref_cislo": "178201303",
|
||||||
|
"podano_kdy": "2026-05-12 21:59:07"
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"datum": "31.01.2026",
|
||||||
|
"ref_cislo": "178201310",
|
||||||
|
"podano_kdy": "2026-05-12 21:59:13"
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"datum": "28.02.2026",
|
||||||
|
"ref_cislo": "178201313",
|
||||||
|
"podano_kdy": "2026-05-12 21:59:19"
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"datum": "31.03.2026",
|
||||||
|
"ref_cislo": "178201314",
|
||||||
|
"podano_kdy": "2026-05-12 21:59:24"
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"datum": "30.04.2026",
|
||||||
|
"ref_cislo": "178201321",
|
||||||
|
"podano_kdy": "2026-05-12 21:59:31"
|
||||||
|
}
|
||||||
|
]
|
||||||
@@ -0,0 +1 @@
|
|||||||
|
{"mesic": 4, "rok": 2026}
|
||||||
@@ -1,7 +1,7 @@
|
|||||||
[
|
[
|
||||||
{
|
{
|
||||||
"name": "PHPSESSID",
|
"name": "PHPSESSID",
|
||||||
"value": "jue2dfk7t4k34du7ngg4j706q1",
|
"value": "f9tijs1603id6cj4msbrfm0hv5",
|
||||||
"domain": ".portal.cpzp.cz",
|
"domain": ".portal.cpzp.cz",
|
||||||
"path": "/",
|
"path": "/",
|
||||||
"expires": -1,
|
"expires": -1,
|
||||||
|
|||||||
Reference in New Issue
Block a user