Dasko
fd762a6527
- dugme Naplati otvara modal (primljeno/kusur, način plaćanja, prikaz računa) - primljeni iznos se prosleđuje u fiskalni zahtev radi tačnog Gotovina/Povraćaj - nova ruta i handler za prikaz/štampu fiskalnog računa prodaje - fiskalni tab se otvara preko imenovanog prozora (bez blokiranja popup-a) - Fisk mock: refund se sada računa iz stvarnih plaćanja, ne fiksno 0
683 lines
26 KiB
Python
Executable File
683 lines
26 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
"""
|
||
Teron L-PFR Mock Server — glumi Teron fiskalni server za testiranje NTech-a.
|
||
Endpoint-i i format odgovora usklađeni sa Teron API dokumentacijom.
|
||
Port: 4566 (Teron standard)
|
||
"""
|
||
|
||
import json
|
||
import http.server
|
||
import re
|
||
import sys
|
||
import base64
|
||
import io
|
||
import os
|
||
import sqlite3
|
||
import time
|
||
from datetime import datetime, timezone, timedelta
|
||
from pathlib import Path
|
||
|
||
import socket
|
||
import urllib.parse
|
||
|
||
import qrcode
|
||
from receipt import generate_receipt, generate_receipt_html, generate_report, load_locale
|
||
|
||
# ── Konfiguracija ──────────────────────────────────────────
|
||
PORT = 4566 # Teron standard port
|
||
HOST = "0.0.0.0"
|
||
ESIR_ID = "NTECH001" # naš 8-char ESIR identifikator
|
||
|
||
# Kartica emulator (NTech goroutine)
|
||
BE_HOST = os.environ.get("BE_HOST", "127.0.0.1")
|
||
BE_PORT = int(os.environ.get("BE_PORT", "4567"))
|
||
|
||
DATA_DIR = Path(__file__).parent / "data"
|
||
INVOICES_DIR = DATA_DIR / "invoices"
|
||
QR_DIR = DATA_DIR / "qr"
|
||
RECEIPTS_DIR = DATA_DIR / "receipts"
|
||
LOG_FILE = DATA_DIR / "server.log"
|
||
COUNTER_DIR = DATA_DIR / "counters"
|
||
|
||
for d in [DATA_DIR, INVOICES_DIR, QR_DIR, RECEIPTS_DIR, COUNTER_DIR]:
|
||
d.mkdir(parents=True, exist_ok=True)
|
||
|
||
# NTech SQLite baza (read-only) — fallback kad kartica emulator nije dostupan
|
||
NTECH_DB = os.environ.get("NTECH_SQLITE") or str(Path(__file__).parent.parent / "ntech.db")
|
||
|
||
def _ucitaj_verify_host():
|
||
"""Čita verify_host iz env, pa iz NTech SQLite baze."""
|
||
if v := os.environ.get("VERIFY_HOST", ""):
|
||
return v
|
||
try:
|
||
con = sqlite3.connect(f"file:{NTECH_DB}?mode=ro", uri=True)
|
||
try:
|
||
cur = con.execute("SELECT vrednost FROM podesavanja WHERE kljuc='verify_host'")
|
||
row = cur.fetchone()
|
||
return row[0] if row and row[0] else ""
|
||
finally:
|
||
con.close()
|
||
except Exception:
|
||
return ""
|
||
|
||
# Host za verifikacioni link na QR kodu (npr. "ntech.moja-firma.rs:3000").
|
||
# Ako je prazno, koristi se sandbox.suf.purs.gov.rs.
|
||
VERIFY_HOST = _ucitaj_verify_host()
|
||
|
||
def _ucitaj_fiskalni_pismo():
|
||
"""Čita fiskalni_pismo iz env var FISKALNI_PISMO ili iz NTech SQLite baze.
|
||
Vrednosti: 'latin' (podrazumevano) ili 'cyrillic'."""
|
||
if v := os.environ.get("FISKALNI_PISMO", ""):
|
||
return v if v in ("latin", "cyrillic") else "latin"
|
||
try:
|
||
con = sqlite3.connect(f"file:{NTECH_DB}?mode=ro", uri=True)
|
||
try:
|
||
cur = con.execute("SELECT vrednost FROM podesavanja WHERE kljuc='fiskalni_pismo'")
|
||
row = cur.fetchone()
|
||
v = row[0] if row and row[0] else "latin"
|
||
return v if v in ("latin", "cyrillic") else "latin"
|
||
finally:
|
||
con.close()
|
||
except Exception:
|
||
return "latin"
|
||
|
||
# Pismo fiskalnog računa: 'latin' ili 'cyrillic'
|
||
FISKALNI_PISMO = _ucitaj_fiskalni_pismo()
|
||
|
||
|
||
def be_command(cmd: dict) -> dict:
|
||
"""Šalje JSON komandu kartica emulatoru (NTech TCP :4567) i vraća odgovor."""
|
||
try:
|
||
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
|
||
s.settimeout(3)
|
||
s.connect((BE_HOST, BE_PORT))
|
||
s.sendall((json.dumps(cmd) + "\n").encode("utf-8"))
|
||
buf = b""
|
||
while True:
|
||
chunk = s.recv(4096)
|
||
if not chunk:
|
||
break
|
||
buf += chunk
|
||
if b"\n" in buf:
|
||
break
|
||
return json.loads(buf.decode("utf-8").strip())
|
||
except Exception as e:
|
||
log(f" ⚠️ be_command({cmd.get('command')}) greška: {e}")
|
||
return {"status": "error", "message": str(e)}
|
||
|
||
|
||
def build_vl(full_data):
|
||
"""Gradi base64-kodirani payload za vl parametar verifikacionog URL-a."""
|
||
payload = {
|
||
"n": full_data.get("invoiceNumber", ""),
|
||
"ic": full_data.get("invoiceCounter", ""),
|
||
"t": full_data.get("sdcDateTime", ""),
|
||
"a": full_data.get("totalAmount", 0),
|
||
"c": full_data.get("tin", ""),
|
||
"co": full_data.get("company", ""),
|
||
"lo": full_data.get("store", ""),
|
||
"ad": full_data.get("address", ""),
|
||
"g": full_data.get("city", ""),
|
||
"di": full_data.get("district", ""),
|
||
"it": full_data.get("invoiceType", "Normal"),
|
||
"tr": full_data.get("transactionType", "Sale"),
|
||
"tx": full_data.get("taxItems", []),
|
||
"pm": full_data.get("payments", []),
|
||
"ca": full_data.get("cashier", ""),
|
||
"bi": full_data.get("buyerId", ""),
|
||
"items": full_data.get("items", []),
|
||
}
|
||
j = json.dumps(payload, ensure_ascii=False, separators=(",", ":"))
|
||
return base64.b64encode(j.encode("utf-8")).decode("ascii")
|
||
|
||
# ── Firma ───────────────────────────────────────────────────
|
||
|
||
def ucitaj_firmu():
|
||
"""Čita podatke o firmi iz NTech baze."""
|
||
podaci = {}
|
||
try:
|
||
con = sqlite3.connect(f"file:{NTECH_DB}?mode=ro", uri=True)
|
||
try:
|
||
cur = con.execute(
|
||
"SELECT kljuc, vrednost FROM podesavanja WHERE kljuc IN "
|
||
"('naziv_firme','pib','maticni_broj','adresa','telefon',"
|
||
"'poslovna_jedinica_naziv','poslovna_jedinica_oznaka','opstina','grad')"
|
||
)
|
||
podaci = {k: v for k, v in cur.fetchall()}
|
||
finally:
|
||
con.close()
|
||
except Exception as e:
|
||
log(f" ⚠️ Ne mogu da pročitam firmu iz baze ({NTECH_DB}): {e}")
|
||
|
||
naziv = podaci.get("naziv_firme") or "Test Company DOO"
|
||
pib = podaci.get("pib") or "123456789"
|
||
return {
|
||
"name": naziv,
|
||
"tin": f"RS{pib}", # Teron koristi RS prefiks
|
||
"tinPlain": pib,
|
||
"mb": podaci.get("maticni_broj") or "12345678",
|
||
"address": podaci.get("adresa") or "Test Address 1",
|
||
"telefon": podaci.get("telefon") or "",
|
||
"locationName": podaci.get("poslovna_jedinica_naziv") or naziv,
|
||
"businessUnitId": podaci.get("poslovna_jedinica_oznaka") or "BU-001",
|
||
"district": podaci.get("opstina") or "Savski Venac",
|
||
"city": podaci.get("grad") or "Beograd",
|
||
}
|
||
|
||
# ── Brojači ─────────────────────────────────────────────────
|
||
|
||
def get_counter(tip="total"):
|
||
"""Čita i inkrementira brojač za dati tip (total, pp, pr, ap, ar, kp, op, itd.)."""
|
||
f = COUNTER_DIR / f"{tip}.txt"
|
||
num = int(f.read_text().strip()) if f.exists() else 1
|
||
f.write_text(str(num + 1))
|
||
return num
|
||
|
||
def peek_counter(tip="total"):
|
||
"""Čita brojač bez inkrementiranja."""
|
||
f = COUNTER_DIR / f"{tip}.txt"
|
||
num = int(f.read_text().strip()) if f.exists() else 1
|
||
return max(1, num - 1)
|
||
|
||
def counter_ext(invoice_type, transaction_type):
|
||
"""Vraća sufiks tipa transakcije (ПП, ПР, АП...) i ključ brojača."""
|
||
t = (str(invoice_type).lower(), str(transaction_type).lower())
|
||
mapping = {
|
||
("normal", "sale"): ("ПП", "pp"),
|
||
("normal", "refund"): ("ПР", "pr"),
|
||
("advance", "sale"): ("АП", "ap"),
|
||
("advance", "refund"): ("АР", "ar"),
|
||
("copy", "sale"): ("КП", "kp"),
|
||
("copy", "refund"): ("КР", "kr"),
|
||
("training", "sale"): ("ОП", "op"),
|
||
("training", "refund"): ("ОР", "or"),
|
||
}
|
||
return mapping.get(t, ("НН", "other"))
|
||
|
||
# ── Logging ─────────────────────────────────────────────────
|
||
|
||
def log(msg):
|
||
now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||
line = f"[{now}] {msg}"
|
||
print(line, flush=True)
|
||
with open(LOG_FILE, "a", encoding="utf-8") as f:
|
||
f.write(line + "\n")
|
||
|
||
# ── QR ──────────────────────────────────────────────────────
|
||
|
||
def generate_qr(url):
|
||
img = qrcode.make(url)
|
||
buf = io.BytesIO()
|
||
img.save(buf, format="PNG")
|
||
return base64.b64encode(buf.getvalue()).decode("utf-8")
|
||
|
||
# ── Vreme ───────────────────────────────────────────────────
|
||
|
||
def sada():
|
||
tz = timezone(timedelta(hours=2))
|
||
return datetime.now(tz).strftime("%Y-%m-%dT%H:%M:%S.000+02:00")
|
||
|
||
# ── PDV obračun ─────────────────────────────────────────────
|
||
|
||
# Teron koristi: Ж (20%), Ђ (10%), Е (posebna), А (0% neobveznici),
|
||
# Г (oslobođen), З (0% bez prava odbitka)
|
||
# + starije oznake generičkog L-PFR-a za kompatibilnost
|
||
TAX_RATES = {
|
||
# Teron oznake
|
||
"Ж": 20.0, # opšta stopa 20%
|
||
"Ђ": 10.0, # snižena stopa 10%
|
||
"Е": 10.0, # posebna snižena stopa
|
||
"А": 0.0, # neobveznici PDV-a
|
||
"Г": 0.0, # oslobođen bez prava na odbitak
|
||
"З": 0.0, # nije predmet oporezivanja
|
||
# Generičke oznake (kompatibilnost)
|
||
"Б": 20.0, "B": 20.0,
|
||
"В": 0.0, "V": 0.0,
|
||
"Д": 0.0, "D": 0.0,
|
||
"A": 0.0, "G": 0.0, "E": 10.0,
|
||
}
|
||
|
||
def izracunaj_pdv(items):
|
||
"""Grupiše stavke po poreskoj oznaci i izračunava PDV iz bruto iznosa.
|
||
Formula: pdv = bruto * stopa / (100 + stopa)"""
|
||
grupe = {}
|
||
for item in items:
|
||
total = float(item.get("totalAmount", 0))
|
||
for label in item.get("labels", []):
|
||
rate = TAX_RATES.get(label, 0.0)
|
||
if label not in grupe:
|
||
grupe[label] = {"label": label, "rate": rate, "amount": 0.0}
|
||
if rate > 0:
|
||
grupe[label]["amount"] += total * rate / (100 + rate)
|
||
return [
|
||
{
|
||
"label": d["label"],
|
||
"categoryName": "PDV",
|
||
"categoryType": 0,
|
||
"rate": d["rate"],
|
||
"amount": round(d["amount"], 4),
|
||
}
|
||
for d in grupe.values()
|
||
]
|
||
|
||
# ── Response handleri ───────────────────────────────────────
|
||
|
||
def resp_attention():
|
||
return {"sdcDateTime": sada(), "status": "OK"}
|
||
|
||
def resp_status():
|
||
st = be_command({"command": "status"})
|
||
cert = be_command({"command": "certificate"})
|
||
total = st.get("total_counter", 0)
|
||
jid = cert.get("jid", "UNKNOWN")
|
||
tin = cert.get("tin", "RS000000000")
|
||
last_num = f"{ESIR_ID}-{jid}-{total}" if total >= 1 else ""
|
||
return {
|
||
"isPinRequired": st.get("pin_required", False),
|
||
"auditRequired": False,
|
||
"sdcDateTime": sada(),
|
||
"lastInvoiceNumber": last_num,
|
||
"protocolVersion": "1.0.0",
|
||
"serialNumber": ESIR_ID,
|
||
"tin": tin,
|
||
}
|
||
|
||
def resp_verify_pin(request_body=None):
|
||
uneti = ""
|
||
if isinstance(request_body, dict):
|
||
uneti = str(request_body.get("pin", "")).strip()
|
||
elif isinstance(request_body, str):
|
||
uneti = request_body.strip().strip('"')
|
||
resp = be_command({"command": "verify_pin", "pin": uneti})
|
||
if resp.get("status") == "ok":
|
||
log(" 🔓 PIN ispravan")
|
||
return {"status": "OK", "message": "PIN verifikovan"}
|
||
log(" 🔒 Pogrešan PIN")
|
||
return {"status": "ERROR", "code": resp.get("code", "2100"), "message": resp.get("message", "Pogrešan PIN")}
|
||
|
||
def resp_settings_get():
|
||
return {
|
||
"printerType": "Thermal",
|
||
"printerInterface": "None",
|
||
"lpfrEnabled": False,
|
||
"vpfrEnabled": False,
|
||
"authorizeLocalClients": False,
|
||
"authorizeRemoteClients": False,
|
||
"apiKey": "mock-api-key-0000",
|
||
"webserverAddress": f"http://127.0.0.1:{PORT}/",
|
||
}
|
||
|
||
def resp_settings_post():
|
||
return {"status": "OK", "message": "Podešavanja sačuvana"}
|
||
|
||
def resp_certificate():
|
||
c = be_command({"command": "certificate"})
|
||
return {
|
||
"serialNumber": c.get("jid", ESIR_ID),
|
||
"tin": c.get("tin", "RS000000000"),
|
||
"name": c.get("name", ""),
|
||
"validFrom": c.get("valid_from", "2024-01-01T00:00:00+01:00"),
|
||
"validTo": c.get("valid_to", "2027-01-01T00:00:00+01:00"),
|
||
"issuer": c.get("issuer", "Poreska uprava RS"),
|
||
}
|
||
|
||
def _build_invoice_response(req, request_id):
|
||
"""Gradi Teron odgovor za fiskalni račun."""
|
||
# Teron zahtev dolazi unutar invoiceRequest omotača
|
||
inv_req = req.get("invoiceRequest", req)
|
||
|
||
invoice_type = inv_req.get("invoiceType", "Normal")
|
||
transaction_type = inv_req.get("transactionType", "Sale")
|
||
items = inv_req.get("items", [])
|
||
|
||
# PDV i ukupan iznos
|
||
tax_items = izracunaj_pdv(items)
|
||
total_amount = round(sum(float(i.get("totalAmount", 0)) for i in items), 2)
|
||
total_tax = round(sum(t["amount"] for t in tax_items), 2)
|
||
|
||
# Kartica emulator: podatke firme i potpis/brojače
|
||
cert = be_command({"command": "certificate"})
|
||
sign = be_command({
|
||
"command": "sign",
|
||
"invoice_type": invoice_type,
|
||
"transaction_type": transaction_type,
|
||
"total_amount": total_amount,
|
||
})
|
||
|
||
if sign.get("status") == "blocked":
|
||
raise RuntimeError(f"Kartica blokirana: {sign.get('message')}")
|
||
|
||
jid = cert.get("jid", ESIR_ID)
|
||
total_cnt = sign.get("counter", 1)
|
||
type_cnt = sign.get("type_counter", 1)
|
||
ext = sign.get("counter_extension", "ПП")
|
||
|
||
invoice_number = f"{ESIR_ID}-{jid}-{total_cnt}"
|
||
invoice_counter = f"{type_cnt}/{total_cnt}{ext}"
|
||
|
||
# firma podaci sa kartice
|
||
firma = {
|
||
"tinPlain": cert.get("tin_plain", "000000000"),
|
||
"tin": cert.get("tin", "RS000000000"),
|
||
"name": cert.get("name", "Test Company DOO"),
|
||
"locationName": cert.get("location_name", cert.get("name", "Test Company DOO")),
|
||
"address": cert.get("address", "Test Adresa 1"),
|
||
"city": cert.get("city", "Beograd"),
|
||
"district": cert.get("district", "Savski Venac"),
|
||
}
|
||
|
||
# Verifikacioni URL i QR kod
|
||
if VERIFY_HOST:
|
||
vl_payload = {
|
||
"n": invoice_number,
|
||
"ic": invoice_counter,
|
||
"t": sada(),
|
||
"a": total_amount,
|
||
"c": firma["tinPlain"],
|
||
"co": firma["name"],
|
||
"lo": firma["locationName"],
|
||
"ad": firma["address"],
|
||
"g": firma["city"],
|
||
"di": firma["district"],
|
||
"it": invoice_type,
|
||
"tr": transaction_type,
|
||
"tx": tax_items,
|
||
"pm": inv_req.get("payment", [{"type": "Cash", "amount": total_amount}]),
|
||
"ca": inv_req.get("cashier", "Kasir"),
|
||
"bi": inv_req.get("buyerId", ""),
|
||
"items": items,
|
||
}
|
||
vl = base64.b64encode(
|
||
json.dumps(vl_payload, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
|
||
).decode("ascii")
|
||
scheme = "https" if VERIFY_HOST.startswith("https://") else "http"
|
||
host = VERIFY_HOST.removeprefix("https://").removeprefix("http://")
|
||
verification_url = f"{scheme}://{host}/v/?vl={urllib.parse.quote(vl, safe='')}"
|
||
else:
|
||
verification_url = f"https://sandbox.suf.purs.gov.rs/v/?vl={invoice_number}"
|
||
qr_b64 = generate_qr(verification_url)
|
||
|
||
# Odgovor koji ide ka NTech-u (ESIR-u)
|
||
odgovor = {
|
||
"requestedBy": ESIR_ID,
|
||
"signedBy": jid,
|
||
"sdcDateTime": sada(),
|
||
"invoiceCounter": invoice_counter,
|
||
"invoiceCounterExtension": ext,
|
||
"invoiceNumber": invoice_number,
|
||
"verificationUrl": verification_url,
|
||
"verificationQRCode": qr_b64,
|
||
"taxItems": tax_items,
|
||
"totalAmount": total_amount,
|
||
"totalTax": total_tax,
|
||
"messages": "Success",
|
||
}
|
||
|
||
# povraćaj — ako je primljeno više nego što je duženo (npr. kupac dao krupniju
|
||
# novčanicu), razlika se ispisuje na računu; kod refundacije nema povraćaja
|
||
payments = inv_req.get("payment", [{"type": "Cash", "amount": total_amount}])
|
||
total_paid = round(sum(float(p.get("amount", 0)) for p in payments), 2)
|
||
refund = round(total_paid - total_amount, 2) if transaction_type != "Refund" else 0
|
||
if refund < 0:
|
||
refund = 0
|
||
|
||
# Puni podaci za snimanje i generisanje računa
|
||
full_data = {
|
||
**odgovor,
|
||
"requestId": request_id,
|
||
"invoiceType": invoice_type,
|
||
"transactionType": transaction_type,
|
||
"items": items,
|
||
"payments": payments,
|
||
"cashier": inv_req.get("cashier", "Kasir"),
|
||
"buyerId": inv_req.get("buyerId", ""),
|
||
"referentDocumentNumber": inv_req.get("referentDocumentNumber", ""),
|
||
"isFiscal": invoice_type not in ("Copy", "Training", "Proforma"),
|
||
"tin": firma["tinPlain"],
|
||
"company": firma["name"],
|
||
"store": firma["locationName"],
|
||
"address": firma["address"],
|
||
"district": firma["district"],
|
||
"refund": refund,
|
||
# za avansni konačni
|
||
"advancePaid": req.get("advancePaid", 0),
|
||
"advanceTax": req.get("advanceTax", 0),
|
||
}
|
||
|
||
# Snimi JSON
|
||
inv_path = INVOICES_DIR / f"{total_cnt:06d}_{request_id}.json"
|
||
with open(inv_path, "w", encoding="utf-8") as fh:
|
||
json.dump(full_data, fh, indent=2, ensure_ascii=False)
|
||
|
||
# Snimi QR PNG
|
||
qr_path = QR_DIR / f"{total_cnt:06d}_{request_id}.png"
|
||
with open(qr_path, "wb") as fh:
|
||
fh.write(base64.b64decode(qr_b64))
|
||
|
||
# Generiši tekst i HTML račun (pismo određuje podešavanje fiskalni_pismo)
|
||
receipt_text = generate_receipt(full_data, FISKALNI_PISMO)
|
||
txt_path = RECEIPTS_DIR / f"{total_cnt:06d}_{request_id}.txt"
|
||
txt_path.write_text(receipt_text, encoding="utf-8")
|
||
|
||
html_txt = generate_receipt_html(full_data, FISKALNI_PISMO)
|
||
html_path = RECEIPTS_DIR / f"{total_cnt:06d}_{request_id}.html"
|
||
html_path.write_text(html_txt, encoding="utf-8")
|
||
|
||
# Dodaj journal (tekst računa) u odgovor
|
||
odgovor["journal"] = receipt_text
|
||
|
||
log(f" 🧾 {invoice_number} | {ext} | {total_amount:.2f} din | PDV {total_tax:.2f}")
|
||
return odgovor
|
||
|
||
def resp_invoice(request_id, request_body=None):
|
||
if not request_body:
|
||
return {"error": "Telo zahteva je obavezno"}, 400
|
||
return _build_invoice_response(request_body, request_id)
|
||
|
||
def resp_invoice_final(request_id, request_body=None):
|
||
"""Konačni račun koji zatvara avanse (/api/invoices/final)."""
|
||
if not request_body:
|
||
return {"error": "Telo zahteva je obavezno"}, 400
|
||
return _build_invoice_response(request_body, request_id)
|
||
|
||
def resp_invoice_last():
|
||
"""Vraća poslednji sačuvani račun."""
|
||
files = sorted(INVOICES_DIR.glob("*.json"), reverse=True)
|
||
if not files:
|
||
return {"error": "Nema računa"}, 404
|
||
try:
|
||
return json.loads(files[0].read_text(encoding="utf-8"))
|
||
except Exception:
|
||
return {"error": "Greška pri čitanju računa"}, 500
|
||
|
||
def resp_invoice_by_request(request_id):
|
||
for f in INVOICES_DIR.glob("*.json"):
|
||
try:
|
||
data = json.loads(f.read_text(encoding="utf-8"))
|
||
if data.get("requestId") == request_id:
|
||
return data
|
||
except Exception:
|
||
continue
|
||
return {"error": f"Račun {request_id} nije pronađen"}, 404
|
||
|
||
def resp_invoice_by_number(invoice_number):
|
||
for f in INVOICES_DIR.glob("*.json"):
|
||
try:
|
||
data = json.loads(f.read_text(encoding="utf-8"))
|
||
if data.get("invoiceNumber") == invoice_number:
|
||
return data
|
||
except Exception:
|
||
continue
|
||
return {"error": f"Račun {invoice_number} nije pronađen"}, 404
|
||
|
||
def resp_invoice_search(request_body=None):
|
||
"""Osnovna pretraga — vraća CSV."""
|
||
invoices = []
|
||
for f in sorted(INVOICES_DIR.glob("*.json")):
|
||
try:
|
||
data = json.loads(f.read_text(encoding="utf-8"))
|
||
invoices.append(data)
|
||
except Exception:
|
||
continue
|
||
lines = [
|
||
f"{d['invoiceNumber']},{d.get('invoiceType','Normal')},{d.get('transactionType','Sale')},{d.get('sdcDateTime','')},{d.get('totalAmount',0)}"
|
||
for d in invoices
|
||
]
|
||
return "\n".join(lines)
|
||
|
||
# ── Rute ────────────────────────────────────────────────────
|
||
|
||
ROUTES = [
|
||
("GET", "api/attention", "attention"),
|
||
("GET", "api/status", "status"),
|
||
("POST", "api/pin", "verify_pin"),
|
||
("GET", "api/settings", "settings_get"),
|
||
("POST", "api/settings", "settings_post"),
|
||
("GET", "api/certificate", "certificate"),
|
||
("POST", "api/invoices/final", "invoice_final"),
|
||
("GET", "api/invoices/last", "invoice_last"),
|
||
("GET", "api/invoices/request/:requestId", "invoice_by_request"),
|
||
("GET", "api/invoices/:invoiceNumber", "invoice_by_number"),
|
||
("POST", "api/invoices/search", "invoice_search"),
|
||
("POST", "api/invoices", "invoice"),
|
||
]
|
||
|
||
HANDLERS = {
|
||
"attention": resp_attention,
|
||
"status": resp_status,
|
||
"verify_pin": resp_verify_pin,
|
||
"settings_get": resp_settings_get,
|
||
"settings_post": resp_settings_post,
|
||
"certificate": resp_certificate,
|
||
"invoice": resp_invoice,
|
||
"invoice_final": resp_invoice_final,
|
||
"invoice_last": resp_invoice_last,
|
||
"invoice_by_request": resp_invoice_by_request,
|
||
"invoice_by_number": resp_invoice_by_number,
|
||
"invoice_search": resp_invoice_search,
|
||
}
|
||
|
||
# ── HTTP Handler ─────────────────────────────────────────────
|
||
|
||
class FiscalHandler(http.server.BaseHTTPRequestHandler):
|
||
|
||
def log_message(self, fmt, *args):
|
||
pass # koristimo naš log
|
||
|
||
def do_GET(self): self._handle("GET")
|
||
def do_POST(self): self._handle("POST")
|
||
def do_DELETE(self): self._handle("DELETE")
|
||
|
||
def do_OPTIONS(self):
|
||
self.send_response(200)
|
||
self._cors()
|
||
self.end_headers()
|
||
|
||
def _cors(self):
|
||
self.send_header("Access-Control-Allow-Origin", "*")
|
||
self.send_header("Access-Control-Allow-Methods", "GET,POST,DELETE,OPTIONS")
|
||
self.send_header("Access-Control-Allow-Headers", "Content-Type, Authorization")
|
||
|
||
def _read_body(self):
|
||
length = int(self.headers.get("Content-Length", 0))
|
||
if length == 0:
|
||
return None
|
||
raw = self.rfile.read(length).decode("utf-8")
|
||
try:
|
||
return json.loads(raw)
|
||
except Exception:
|
||
return raw
|
||
|
||
def _handle(self, method):
|
||
path = self.path.split("?")[0].strip("/")
|
||
status = 404
|
||
body = {"error": "Not Found", "path": path}
|
||
content_type = "application/json; charset=utf-8"
|
||
matched = None
|
||
|
||
for r_method, r_pattern, r_name in ROUTES:
|
||
if r_method != method:
|
||
continue
|
||
regex = "^" + re.sub(r":(\w+)", r"(?P<\1>[^/]+)", r_pattern) + "$"
|
||
m = re.match(regex, path)
|
||
if not m:
|
||
continue
|
||
|
||
matched = r_name
|
||
handler = HANDLERS.get(r_name)
|
||
params = m.groupdict()
|
||
request_id = self.headers.get("RequestId", f"mock-{int(time.time())}")
|
||
|
||
# Poziv handlera
|
||
if r_name in ("invoice", "invoice_final"):
|
||
result = handler(request_id, self._read_body())
|
||
elif r_name == "verify_pin":
|
||
result = handler(self._read_body())
|
||
elif r_name == "invoice_by_request":
|
||
result = handler(params.get("requestId", ""))
|
||
elif r_name == "invoice_by_number":
|
||
result = handler(params.get("invoiceNumber", ""))
|
||
elif r_name == "invoice_search":
|
||
result = handler(self._read_body())
|
||
elif r_name in ("settings_post",):
|
||
self._read_body()
|
||
result = handler()
|
||
else:
|
||
result = handler()
|
||
|
||
# Razdvoji (body, status) ako handler vratio tuple
|
||
if isinstance(result, tuple):
|
||
body, status = result
|
||
else:
|
||
body = result
|
||
status = 200
|
||
break
|
||
|
||
# Serializacija
|
||
if r_name == "invoice_search" and isinstance(body, str):
|
||
resp_bytes = body.encode("utf-8")
|
||
content_type = "text/csv; charset=utf-8"
|
||
elif isinstance(body, (dict, list)):
|
||
resp_bytes = json.dumps(body, indent=2, ensure_ascii=False).encode("utf-8")
|
||
else:
|
||
resp_bytes = str(body).encode("utf-8")
|
||
|
||
emoji = "✅" if status == 200 else "❌"
|
||
log(f" {emoji} {method} /{path} → {matched or '404'} [{status}]")
|
||
|
||
self.send_response(status)
|
||
self._cors()
|
||
self.send_header("Content-Type", content_type)
|
||
self.send_header("Content-Length", str(len(resp_bytes)))
|
||
self.end_headers()
|
||
self.wfile.write(resp_bytes)
|
||
|
||
# ── Main ─────────────────────────────────────────────────────
|
||
|
||
def main():
|
||
cert = be_command({"command": "certificate"})
|
||
jid = cert.get("jid", "?")
|
||
tin = cert.get("tin_plain", "?")
|
||
name = cert.get("name", "?")
|
||
log("╔══════════════════════════════════════════════════╗")
|
||
log("║ 🧾 Teron L-PFR Mock Server ║")
|
||
log(f"║ http://{HOST}:{PORT}/ ║")
|
||
log("╚══════════════════════════════════════════════════╝")
|
||
log(f" 📁 Podaci: {DATA_DIR}")
|
||
log(f" 🧾 Računi: {INVOICES_DIR}")
|
||
log(f" 📱 QR PNG: {QR_DIR}")
|
||
log(f" 📝 Log: {LOG_FILE}")
|
||
log(f" 🏢 Firma: {name} | PIB: {tin}")
|
||
log(f" 🆔 ESIR ID: {ESIR_ID} | BE JID: {jid} (kartica: {BE_HOST}:{BE_PORT})")
|
||
log(" ▶ Server pokrenut. Ctrl+C za gašenje.")
|
||
|
||
server = http.server.HTTPServer((HOST, PORT), FiscalHandler)
|
||
try:
|
||
server.serve_forever()
|
||
except KeyboardInterrupt:
|
||
log(" ⏹ Server zaustavljen.")
|
||
server.server_close()
|
||
|
||
if __name__ == "__main__":
|
||
main()
|