Fisk: fiskalni mock server (myLPFR) sa pokretačem fisk-test.sh

Kartica (bezbednosni element) čita profil firme iz NTech baze (read-only):
PIB, MB, naziv, adresa, poslovna jedinica. Fiksni test PIN. U git ide kod i
locale; generisani računi, QR, logovi i pycache su ignorisani.
This commit is contained in:
2026-06-21 19:06:35 +02:00
parent 774f4915c1
commit 3a4b9f2c1f
8 changed files with 1683 additions and 0 deletions
+9
View File
@@ -29,6 +29,15 @@ web/static/uploads/
*.kate-swp
.*.kate-swp
# Fiskalni test server — kod ide u repo, ali ne i runtime artefakti
Fisk/__pycache__/
Fisk/data/invoices/
Fisk/data/qr/
Fisk/data/receipts/
Fisk/data/*.log
Fisk/data/*.pid
Fisk/data/counter.txt
# IDE podešavanja
.vscode/
.idea/
+48
View File
@@ -0,0 +1,48 @@
# Locale: Serbian Cyrillic
fiscal-invoice=ФИСКАЛНИ РАЧУН
non-fiscal-invoice=ОВО НИЈЕ ФИСКАЛНИ РАЧУН
buyer-id=ИД купца
buyer-cost-center=Опција купца
cashier-id=Касир
pos-invoice-number=ЕСИР број
pos-time=ЕСИР време
ref-doc-number=Реф. број
ref-doc-dt=Реф. време
sdc-invoice-counter=Бројач
sdc-invoice-number=Број рачуна
sdc-time=Време
uid=УИД
items=Артикли
item-name=Назив
item-qty=Кол.
item-price=Цена
item-amount=Износ
gtin=ГТИН
labels=Ознаке
name=Назив
qty=Кол.
quantity=Количина
price=Цена
amount=Износ
unitPrice=Јед. цена
tax-label=ПДВ
tax-name=Порез
tax-rate=Стопа
tax-amount=Порез
tax=Порез
total-tax=Укупан порез
advance-tax=Авансни порез
to-pay=За уплату
paid-in-advance=Плаћено авансно
remaining=Преостало
refund=Рефундација
total-refund=Укупна рефундација
total-payment=Укупно уплаћено
payment=Уплаћено
summary=УКУПАН ПРОМЕТ
invoice-count=Број рачуна
period=Период
report-number=Број извештаја
per-transaction-type=ПРОМЕТ ПО ВРСТИ
customer-signature=Потпис купца
end-of-fiscal-invoice=КРАЈ ФИСКАЛНОГ РАЧУНА
+48
View File
@@ -0,0 +1,48 @@
# Locale: Serbian Latin
fiscal-invoice=FISKALNI RAČUN
non-fiscal-invoice=OVO NIJE FISKALNI RAČUN
buyer-id=ID kupca
buyer-cost-center=Opcija kupca
cashier-id=Kasir
pos-invoice-number=ESIR broj
pos-time=ESIR vreme
ref-doc-number=Ref. broj
ref-doc-dt=Ref. vreme
sdc-invoice-counter=Brojač
sdc-invoice-number=Broj računa
sdc-time=Vreme
uid=UID
items=Artikli
item-name=Naziv
item-qty=Kol.
item-price=Cena
item-amount=Iznos
gtin=GTIN
labels=Oznake
name=Naziv
qty=Kol.
quantity=Količina
price=Cena
amount=Iznos
unitPrice=Jed. cena
tax-label=PDV
tax-name=Porez
tax-rate=Stopa
tax-amount=Porez
tax=Porez
total-tax=Ukupan porez
advance-tax=Avansni porez
to-pay=Za uplatu
paid-in-advance=Plaćeno avansno
remaining=Preostalo
refund=Refundacija
total-refund=Ukupna refundacija
total-payment=Ukupno uplaćeno
payment=Uplaćeno
summary=UKUPAN PROMET
invoice-count=Broj računa
period=Period
report-number=Broj izveštaja
per-transaction-type=PROMET PO VRSTI
customer-signature=Potpis kupca
end-of-fiscal-invoice=KRAJ FISKALNOG RAČUNA
+629
View File
@@ -0,0 +1,629 @@
#!/usr/bin/env python3
"""
Generisanje fiskalnog računa za A4 štampač.
Prati izgled definisan u LPFR VM template-u i locale fajlovima.
"""
from pathlib import Path
from datetime import datetime
DATA_DIR = Path(__file__).parent / "data"
def load_locale(lang="latin"):
"""Učitava lokalizacioni fajl u dict."""
filename = f"locale_{lang}.properties"
path = DATA_DIR / filename
if not path.exists():
return {}
locale = {}
with open(path, encoding="utf-8") as f:
for line in f:
line = line.strip()
if not line or line.startswith("#") or "=" not in line:
continue
key, _, value = line.partition("=")
locale[key.strip()] = value.strip()
return locale
def price(n):
"""Formatira cenu: ###,###.00"""
return f"{n:,.2f}"
def qty(n):
"""Formatira količinu: ###,###.000"""
return f"{n:,.3f}"
def amount(n):
"""Formatira iznos: ###,###.00"""
return f"{n:,.2f}"
def number(n):
"""Formatira broj: ###,###"""
return f"{n:,.0f}"
def dt(iso_string):
"""Konvertuje ISO datetime u format: dd.MM.yyyy. HH:mm:ss"""
try:
d = datetime.fromisoformat(iso_string.replace("+02:00", "").replace("Z", ""))
return d.strftime("%d.%m.%Y. %H:%M:%S")
except Exception:
return iso_string
def center(text, width=48):
"""Centrira tekst unutar širine."""
return text.center(width)
def layout(left, right, width=48):
"""Formatira dve kolone: levo poravnato levo, desno poravnato desno."""
left_str = str(left) if left else ""
right_str = str(right) if right else ""
space = width - len(left_str) - len(right_str)
if space < 1:
space = 1
return left_str + " " * space + right_str
def wrap(text, width=48):
"""Prelamanje teksta (vraća listu linija)."""
if not text:
return []
words = text.split()
lines = []
current = ""
for word in words:
if len(current) + len(word) + 1 <= width:
current = (current + " " + word).strip()
else:
if current:
lines.append(current)
current = word
if current:
lines.append(current)
return lines if lines else [text]
def separator(char="-", width=48):
"""Linija separatora."""
return char * width
def title(text, char="=", width=48):
"""Naslov okružen linijama."""
return separator(char, width) + "\n" + center(text, width) + "\n" + separator(char, width)
# ── Transakcije prevod ──────────────────────────────────────
TRANSACTION_TYPES_CYR = {
"NSX": "ПРОМЕТ - ПРОДАЈА", "NRX": "ПРОМЕТ - РЕФУНДАЦИЈА",
"CSX": "КОПИЈА - ПРОДАЈА", "CRX": "КОПИЈА - РЕФУНДАЦИЈА",
"ASX": "АВАНС - ПРОДАЈА", "ARX": "АВАНС - РЕФУНДАЦИЈА",
"PSX": "ПРЕДРАЧУН - ПРОДАЈА", "PRX": "ПРЕДРАЧУН - РЕФУНДАЦИЈА",
"TSX": "ОБУКА - ПРОДАЈА", "TRX": "ОБУКА - РЕФУНДАЦИЈА",
}
TRANSACTION_TYPES_LAT = {
"NSX": "PROMET - PRODAJA", "NRX": "PROMET - REFUNDACIJA",
"CSX": "KOPIJA - PRODAJA", "CRX": "KOPIJA - REFUNDACIJA",
"ASX": "AVANS - PRODAJA", "ARX": "AVANS - REFUNDACIJA",
"PSX": "PREDRAČUN - PRODAJA", "PRX": "PREDRAČUN - REFUNDACIJA",
"TSX": "OBUKA - PRODAJA", "TRX": "OBUKA - REFUNDACIJA",
}
# ── Glavna funkcija ─────────────────────────────────────────
def generate_receipt(invoice_data, lang="latin"):
"""
Generiše fiskalni račun spreman za A4 štampu.
Prati agent-invoice.vm template 1:1.
"""
m = load_locale(lang)
if not m:
m = load_locale("latin")
tx_types = TRANSACTION_TYPES_LAT if lang == "latin" else TRANSACTION_TYPES_CYR
W = 48
inv = invoice_data
lines = []
# ── PREAMBLE ──
preamble = inv.get("preamble", "")
if preamble:
for pl in wrap(preamble, W):
lines.append(pl)
lines.append("")
# ── NASLOV ──
is_fiscal = inv.get("isFiscal", True)
if is_fiscal:
lines.append(title(m.get("fiscal-invoice", "FISKALNI RAČUN"), "=", W))
else:
lines.append(title(m.get("non-fiscal-invoice", "OVO NIJE FISKALNI RAČUN"), "=", W))
# ── ZAGLAVLJE ──
for field in ["tin", "company", "store", "address", "district"]:
val = inv.get(field, "")
if val:
lines.append(center(str(val), W))
lines.append(separator("-", W))
# Buyer info
if inv.get("buyerId"):
lines.append(layout(m.get("buyer-id", "ID kupca"), inv["buyerId"], W))
if inv.get("buyerCostCenterId"):
lines.append(layout(m.get("buyer-cost-center", "Opcija kupca"), inv["buyerCostCenterId"], W))
if inv.get("cashier"):
lines.append(layout(m.get("cashier-id", "Kasir"), inv["cashier"], W))
if inv.get("posNumber"):
lines.append(layout(m.get("pos-invoice-number", "ESIR broj"), inv["posNumber"], W))
if inv.get("posDateTime"):
lines.append(layout(m.get("pos-time", "ESIR vreme"), dt(inv["posDateTime"]), W))
# Referentni dokument (storno)
if inv.get("referentDocumentNumber"):
lines.append(layout(m.get("ref-doc-number", "Ref. broj"), inv["referentDocumentNumber"], W))
if inv.get("referentDocumentDT"):
lines.append(layout(m.get("ref-doc-dt", "Ref. vreme"), dt(inv["referentDocumentDT"]), W))
# ── TIP TRANSAKCIJE ──
tx_code = inv.get("transactionType", "NSX")
tx_label = tx_types.get(tx_code, tx_code)
lines.append(title(tx_label, "-", W))
# ── ARTIKLI ──
lines.append(center(m.get("items", "Artikli"), W))
lines.append(separator("=", W))
# Zaglavlje tabele (prati agent-invoice.vm layout)
lbl_name = m.get("item-name", "Naziv")
lbl_price = m.get("item-price", "Cena")
lbl_qty = m.get("item-qty", "Kol.")
lbl_amount = m.get("item-amount", "Ukupno")
header = lbl_name + lbl_price.rjust(10) + lbl_qty.rjust(11)
lines.append(layout(header, lbl_amount, W))
for item in inv.get("items", []):
name = item.get("name", "")
gtin = item.get("gtin", "")
labels = item.get("labels", [])
label_str = " ".join(labels) if labels else ""
if gtin:
name_line = f"{gtin} {name} {label_str}".strip()
else:
name_line = f"{name} {label_str}".strip()
for wl in wrap(name_line, W):
lines.append(wl)
ip = float(item.get("unitPrice") or item.get("price", 0))
iq = float(item.get("quantity") or item.get("qty", 0))
ia = float(item.get("amount") or (ip * iq))
sign = "-" if inv.get("transactionType") == "Refund" else ""
row = price(ip).rjust(14) + qty(iq).rjust(9)
lines.append(layout(row, sign + amount(ia), W))
lines.append(separator("-", W))
# ── UKUPNO ──
if inv.get("transactionType") == "Refund":
lines.append(layout(m.get("total-refund", "Ukupna refundacija"), amount(inv.get("totalAmount", 0)), W))
else:
lines.append(layout(m.get("to-pay", "Za uplatu"), amount(inv.get("totalAmount", 0)), W))
# ── PLAĆANJA ──
advance = float(inv.get("advance", 0))
advance_tax = float(inv.get("advanceTax", 0))
is_covered_by_advance = inv.get("coveredByAdvance", False)
if advance:
lines.append(layout(m.get("paid-in-advance", "Uplaćeno avansom"), amount(advance), W))
if advance_tax:
lines.append(layout(m.get("advance-tax", "PDV na avans"), amount(advance_tax), W))
if not is_covered_by_advance:
for p in inv.get("payments", []):
ptype = m.get(p.get("type", ""), p.get("type", "Drugo"))
lines.append(layout(ptype, amount(float(p.get("amount", 0))), W))
if inv.get("invoiceType") == "Proforma":
lines.append(layout(m.get("refund", "Povraćaj"), amount(0), W))
else:
lines.append(layout(m.get("refund", "Povraćaj"), amount(float(inv.get("refund", 0))), W))
if advance:
lines.append(layout(m.get("remaining", "Preostalo"), amount(float(inv.get("remaining", 0))), W))
lines.append(separator("=", W))
# ── NEFISKALNI RAČUN ──
if not is_fiscal:
lines.append(center(m.get("non-fiscal-invoice", "OVO NIJE FISKALNI RAČUN"), W))
lines.append(separator("-", W))
# ── POREZI ──
tax_hdr = (m.get("tax-label", "Oznaka") +
m.get("tax-name", "Ime").rjust(8) +
m.get("tax-rate", "Stopa").rjust(8))
lines.append(layout(tax_hdr, m.get("tax-amount", "Porez"), W))
for tax in inv.get("taxItems", []):
row = (str(tax.get("label", "")) +
str(tax.get("name", "")).rjust(13) +
number(float(tax.get("rate", 0))).rjust(7) + "%")
lines.append(layout(row, amount(float(tax.get("amount", 0))), W))
lines.append(separator("-", W))
lines.append(layout(m.get("total-tax", "Ukupan porez"), amount(float(inv.get("totalTax", 0))), W))
lines.append(separator("=", W))
# ── PFR VREDNOSTI ──
lines.append(layout(m.get("sdc-time", "PFR vreme"), dt(inv.get("sdcDateTime", "")), W))
lines.append(layout(m.get("sdc-invoice-number", "PFR broj računa"), str(inv.get("invoiceNumber", "")), W))
lines.append(layout(m.get("sdc-invoice-counter", "Brojač računa"), str(inv.get("invoiceNumber", "")), W))
lines.append(separator("=", W))
# ── QR KOD ──
lines.append("{{{{QR-KOD}}}}")
lines.append("")
# ── POTPIS KUPCA (Copy + Refund) ──
if inv.get("invoiceType") == "Copy" and inv.get("transactionType") == "Refund":
lines.append("")
lines.append(f"{m.get('customer-signature', 'Potpis kupca')}: ______________________")
lines.append("")
# ── KRAJ ──
if is_fiscal:
lines.append(title(m.get("end-of-fiscal-invoice", "KRAJ FISKALNOG RAČUNA"), "=", W))
else:
lines.append(title(m.get("non-fiscal-invoice", "OVO NIJE FISKALNI RAČUN"), "=", W))
# ── PORUKA ──
msg = inv.get("message", "")
if msg:
lines.append("")
for ml in wrap(msg, W):
lines.append(ml)
return "\n".join(lines)
def generate_receipt_html(invoice_data, lang="latin"):
"""
Generiše HTML verziju fiskalnog računa — za A4 štampu iz browsera.
Prati agent-invoice.vm template 1:1.
"""
m = load_locale(lang)
if not m:
m = load_locale("latin")
tx_types = TRANSACTION_TYPES_LAT if lang == "latin" else TRANSACTION_TYPES_CYR
inv = invoice_data
tx_code = inv.get("transactionType", "NSX")
tx_label = tx_types.get(tx_code, tx_code)
is_fiscal = inv.get("isFiscal", True)
is_refund = inv.get("transactionType") == "Refund"
inv_type = inv.get("invoiceType", "Normal")
sign = "-" if is_refund else ""
# Preamble
preamble_html = ""
preamble = inv.get("preamble", "")
if preamble:
preamble_html = f'<div class="preamble">{preamble}</div><div class="sep"></div>'
# Items
items_rows = ""
for item in inv.get("items", []):
gtin = f' <small>(GTIN: {item["gtin"]})</small>' if item.get("gtin") else ""
labels = " ".join(item.get("labels", []))
ip = float(item.get("unitPrice") or item.get("price", 0))
iq = float(item.get("quantity") or item.get("qty", 0))
ia = float(item.get("amount") or (ip * iq))
items_rows += f"""
<tr>
<td class="l">{item.get('name', '')}{gtin} {labels}</td>
<td class="r">{price(ip)}</td>
<td class="r">{qty(iq)}</td>
<td class="r">{sign}{amount(ia)}</td>
</tr>"""
# Payments
payments_rows = ""
advance = float(inv.get("advance", 0))
advance_tax = float(inv.get("advanceTax", 0))
covered = inv.get("coveredByAdvance", False)
if advance:
payments_rows += f'<tr><td class="l">{m.get("paid-in-advance", "Uplaćeno avansom")}</td><td class="r">{amount(advance)}</td></tr>'
if advance_tax:
payments_rows += f'<tr><td class="l">{m.get("advance-tax", "PDV na avans")}</td><td class="r">{amount(advance_tax)}</td></tr>'
if not covered:
for p in inv.get("payments", []):
ptype = m.get(p.get("type", ""), p.get("type", "Drugo"))
payments_rows += f'<tr><td class="l">{ptype}</td><td class="r">{amount(float(p.get("amount", 0)))}</td></tr>'
if inv_type == "Proforma":
payments_rows += f'<tr><td class="l">{m.get("refund", "Povraćaj")}</td><td class="r">{amount(0)}</td></tr>'
else:
payments_rows += f'<tr><td class="l">{m.get("refund", "Povraćaj")}</td><td class="r">{amount(float(inv.get("refund", 0)))}</td></tr>'
if advance:
payments_rows += f'<tr><td class="l">{m.get("remaining", "Preostalo")}</td><td class="r">{amount(float(inv.get("remaining", 0)))}</td></tr>'
# Non-fiscal notice
non_fiscal_html = ""
if not is_fiscal:
non_fiscal_html = f'<div class="c"><strong>{m.get("non-fiscal-invoice", "OVO NIJE FISKALNI RAČUN")}</strong></div><div class="sep"></div>'
# Tax
tax_rows = ""
for tax in inv.get("taxItems", []):
tax_rows += f"""
<tr>
<td class="l">{tax.get('label', '')}</td>
<td class="l">{tax.get('name', '')}</td>
<td class="r">{number(float(tax.get('rate', 0)))}%</td>
<td class="r">{amount(float(tax.get('amount', 0)))}</td>
</tr>"""
# Customer signature (Copy + Refund)
sig_html = ""
if inv_type == "Copy" and inv.get("transactionType") == "Refund":
sig_html = f'<div style="margin-top:5mm;">{m.get("customer-signature", "Potpis kupca")}: ______________________</div>'
# Message
msg_html = ""
msg = inv.get("message", "")
if msg:
msg_html = f'<div class="sep"></div><div>{msg}</div>'
qr_src = f"data:image/png;base64,{inv.get('qrCode', '')}"
title_text = m.get("fiscal-invoice", "FISKALNI RAČUN") if is_fiscal else m.get("non-fiscal-invoice", "OVO NIJE FISKALNI RAČUN")
end_text = m.get("end-of-fiscal-invoice", "KRAJ FISKALNOG RAČUNA") if is_fiscal else m.get("non-fiscal-invoice", "")
# ── Buyer info (dvokolonski: levo=header, desno=buyer) ──
buyer_col = ""
if inv.get("buyerId"):
buyer_col += f'<div class="buyer-line"><span class="buyer-lbl">{m.get("buyer-id", "ID kupca")}</span><span>{inv["buyerId"]}</span></div>'
if inv.get("buyerCostCenterId"):
buyer_col += f'<div class="buyer-line"><span class="buyer-lbl">{m.get("buyer-cost-center", "Opcija kupca")}</span><span>{inv["buyerCostCenterId"]}</span></div>'
ref_doc = ""
if inv.get("referentDocumentNumber"):
ref_doc += f'<div class="ref-line"><span>{m.get("ref-doc-number", "Ref. broj")}: {inv["referentDocumentNumber"]}</span></div>'
if inv.get("referentDocumentDT"):
ref_doc += f'<div class="ref-line"><span>{m.get("ref-doc-dt", "Ref. vreme")}: {dt(inv["referentDocumentDT"])}</span></div>'
html = f"""<!DOCTYPE html>
<html lang="sr">
<head>
<meta charset="utf-8">
<title>{title_text} | {inv.get('invoiceNumber', '')}</title>
<style>
@page {{ size: A4; margin: 8mm 11mm; }}
body {{ font-family: 'Consolas', 'Lucida Console', 'IBM Plex Mono', 'Roboto Condensed', monospace; font-size: 7pt; width: 188mm; margin: 0 auto; color: #000; }}
.c {{ text-align: center; }}
.r {{ text-align: right; }}
.l {{ text-align: left; }}
.sep {{ border-top: 1px solid #000; margin: 1mm 0; }}
.sep-double {{ border-top: 3px double #000; margin: 1mm 0; }}
table {{ width: 100%; border-collapse: collapse; }}
td {{ padding: 0.5px 1px; vertical-align: top; font-size: 7pt; }}
th {{ padding: 0.5px 1px; font-size: 7pt; font-weight: normal; }}
.hdr {{ font-size: 11pt; font-weight: bold; margin: 0 0 0.5mm 0; }}
.hdr-sub {{ font-size: 11pt; font-weight: bold; }}
.title {{ font-weight: bold; font-size: 12pt; margin: 1.5mm 0; }}
.qr {{ text-align: center; margin: 2mm 0; }}
.qr img {{ width: 25mm; height: 25mm; }}
.preamble {{ font-style: italic; margin: 1mm 0; font-size: 7pt; }}
.row {{ display: flex; justify-content: space-between; }}
.col-left {{ width: 72%; }}
.col-right {{ width: 25%; text-align: right; }}
.buyer-line {{ font-size: 7pt; }}
.buyer-lbl {{ margin-right: 3mm; }}
.ref-line {{ font-size: 7pt; }}
.inv-total {{ font-size: 8pt; font-weight: bold; }}
.end-title {{ font-size: 9pt; font-weight: bold; }}
@media print {{ body {{ -webkit-print-color-adjust: exact; }} }}
</style>
</head>
<body>
{preamble_html}
<div class="c title">{title_text}</div>
<div class="sep-double"></div>
<div class="c hdr">{inv.get('company', '')}</div>
<div class="c hdr-sub">{inv.get('tin', '')}</div>
<div class="c hdr-sub">{inv.get('store', '')}</div>
<div class="c hdr-sub">{inv.get('address', '')}</div>
<div class="c hdr-sub">{inv.get('district', '')}</div>
<div class="sep"></div>
<div class="row">
<div class="col-left">
<div class="ref-line">{m.get('cashier-id', 'Kasir')}: {inv.get('cashier', '')}</div>
<div class="ref-line">{m.get('pos-invoice-number', 'ESIR broj')}: {inv.get('posNumber', '')}</div>
<div class="ref-line">{m.get('pos-time', 'ESIR vreme')}: {dt(inv.get('posDateTime', ''))}</div>
</div>
<div class="col-right">
{buyer_col}
</div>
</div>
{ref_doc}
<div class="sep"></div>
<div class="c"><strong>{tx_label}</strong></div>
<div class="sep"></div>
<table>
<tr><th class="l">{m.get('item-name', 'Naziv')}</th><th class="r">{m.get('item-price', 'Cena')}</th><th class="r">{m.get('item-qty', 'Kol.')}</th><th class="r">{m.get('item-amount', 'Ukupno')}</th></tr>
{items_rows}
</table>
<div class="sep"></div>
<table>
<tr class="inv-total"><td class="l">{m.get('to-pay', 'Za uplatu') if not is_refund else m.get('total-refund', 'Ukupna refundacija')}</td><td class="r">{amount(float(inv.get('totalAmount', 0)))}</td></tr>
{payments_rows}
</table>
<div class="sep-double"></div>
{non_fiscal_html}
<table>
<tr><th class="l">{m.get('tax-label', 'Oznaka')}</th><th class="l">{m.get('tax-name', 'Naziv')}</th><th class="r">{m.get('tax-rate', 'Stopa')}</th><th class="r">{m.get('tax-amount', 'Porez')}</th></tr>
{tax_rows}
</table>
<div class="sep"></div>
<table>
<tr class="inv-total"><td class="l">{m.get('total-tax', 'Ukupan porez')}</td><td class="r">{amount(float(inv.get('totalTax', 0)))}</td></tr>
</table>
<div class="sep-double"></div>
<table>
<tr><td class="l">{m.get('sdc-time', 'PFR vreme')}</td><td class="r">{dt(inv.get('sdcDateTime', ''))}</td></tr>
<tr><td class="l">{m.get('sdc-invoice-number', 'PFR broj računa')}</td><td class="r">{inv.get('invoiceNumber', '')}</td></tr>
<tr><td class="l">{m.get('sdc-invoice-counter', 'Brojač računa')}</td><td class="r">{inv.get('invoiceNumber', '')}</td></tr>
</table>
<div class="sep-double"></div>
<div class="qr"><img src="{qr_src}" alt="QR kod za verifikaciju"></div>
{sig_html}
<div class="c end-title">{end_text}</div>
{msg_html}
</body>
</html>"""
return html
# ═══════════════════════════════════════════════════════════════
# DNEVNI / PERIODIČNI IZVEŠTAJI (standard-report.vm)
# ═══════════════════════════════════════════════════════════════
def generate_report(report_data, lang="latin"):
"""
Generiše dnevni ili periodični izveštaj.
Prati standard-report.vm template 1:1.
report_data = {
"title": "DNEVNI IZVEŠTAJ",
"number": 1,
"dateTime": "2026-06-21T16:00:00.000+02:00",
"tin": "123456789",
"businessName": "Test Company DOO",
"locationName": "Test Location",
"address": "Test Address 1, Beograd",
"district": "Savski Venac",
"uid": "550e8400...",
"startDate": "2026-06-21",
"endDate": "2026-06-21",
"total": {
"invoiceCount": 42,
"payments": [
{"paymentType": "Cash", "amount": 25000.00},
{"paymentType": "Card", "amount": 15000.00},
],
"totalPayments": 40000.00,
"taxItems": [
{"label": "S", "rate": 20.0, "total": 250000.00, "amount": 50000.00},
{"label": "P", "rate": 10.0, "total": 100000.00, "amount": 10000.00},
],
"totalTax": 60000.00,
},
"perTransactionType": [
{
"transactionTypeExt": "NSX",
"invoiceCount": 30,
"payments": [...],
"totalPayments": 30000.00,
"taxItems": [...],
"totalTax": 45000.00,
}
],
}
"""
m = load_locale(lang)
if not m:
m = load_locale("latin")
W = 48
rep = report_data
lines = []
lines.append(separator("=", W))
# Zaglavlje
for field in ["tin", "businessName", "locationName", "address", "district"]:
val = rep.get(field, "")
if val:
lines.append(center(str(val), W))
lines.append(separator("-", W))
# Naslov
title_text = rep.get("title", "IZVEŠTAJ")
lines.append(center(title_text, W))
# Period
start = rep.get("startDate", "")
end = rep.get("endDate", "")
if start and end:
period_text = m.get("period", "PERIOD") + f": {start} - {end}"
lines.append(center(period_text, W))
lines.append(separator("-", W))
# Broj izveštaja, JID, vreme
if rep.get("number"):
lines.append(layout(m.get("report-number", "Broj izveštaja"), str(rep["number"]), W))
lines.append(layout(m.get("uid", "JID"), rep.get("uid", ""), W))
lines.append(layout(m.get("sdc-time", "PFR vreme"), dt(rep.get("dateTime", "")), W))
lines.append(separator("-", W))
lines.append(center(m.get("summary", "UKUPAN PROMET"), W))
lines.append(separator("-", W))
total = rep.get("total", {})
# Broj računa
lines.append(layout(m.get("invoice-count", "Broj računa"), str(total.get("invoiceCount", 0)), W))
# Plaćanja
lines.append(title(m.get("payment", "Uplaćeno"), "=", W))
for p in total.get("payments", []):
ptype = m.get(p.get("paymentType", ""), p.get("paymentType", "Drugo"))
lines.append(layout(ptype, amount(float(p.get("amount", 0))), W))
lines.append(separator("-", W))
lines.append(layout(m.get("total-payment", "Ukupno uplaćeno"), amount(float(total.get("totalPayments", 0))), W))
# Porezi
lines.append(title(m.get("tax", "Porez"), "=", W))
tax_hdr = m.get("tax-rate", "Stopa") + m.get("item-amount", "Osnovica").rjust(20)
lines.append(layout(tax_hdr, m.get("tax-amount", "Porez"), W))
lines.append(separator("-", W))
for t in total.get("taxItems", []):
row = str(t.get("label", "")) + number(float(t.get("rate", 0))).rjust(7) + amount(float(t.get("total", 0))).rjust(17)
lines.append(layout(row, amount(float(t.get("amount", 0))), W))
lines.append(separator("-", W))
lines.append(layout(m.get("total-tax", "Ukupan porez"), amount(float(total.get("totalTax", 0))), W))
# Po tipu transakcije
per_tx = rep.get("perTransactionType", [])
if per_tx:
lines.append(separator("=", W))
lines.append(center(m.get("per-transaction-type", "PROMET PO VRSTI"), W))
for summary in per_tx:
lines.append(separator("=", W))
tx_code = summary.get("transactionTypeExt", "")
tx_label_full = TRANSACTION_TYPES_LAT.get(tx_code, "") if lang == "latin" else TRANSACTION_TYPES_CYR.get(tx_code, "")
if tx_label_full:
lines.append(center(tx_label_full, W))
lines.append(separator("-", W))
lines.append(layout(m.get("invoice-count", "Broj računa"), str(summary.get("invoiceCount", 0)), W))
lines.append(title(m.get("payment", "Uplaćeno"), "=", W))
for p in summary.get("payments", []):
ptype = m.get(p.get("paymentType", ""), p.get("paymentType", "Drugo"))
lines.append(layout(ptype, amount(float(p.get("amount", 0))), W))
lines.append(separator("-", W))
lbl = m.get("total-refund", "Refundacija") if "Refund" in tx_code else m.get("total-payment", "Ukupno")
lines.append(layout(lbl, amount(float(summary.get("totalPayments", 0))), W))
lines.append(title(m.get("tax", "Porez"), "=", W))
lines.append(layout(tax_hdr, m.get("tax-amount", "Porez"), W))
lines.append(separator("-", W))
for t in summary.get("taxItems", []):
row = str(t.get("label", "")) + number(float(t.get("rate", 0))).rjust(7) + amount(float(t.get("total", 0))).rjust(17)
lines.append(layout(row, amount(float(t.get("amount", 0))), W))
lines.append(separator("-", W))
lines.append(layout(m.get("total-tax", "Ukupan porez"), amount(float(summary.get("totalTax", 0))), W))
lines.append(separator("=", W))
return "\n".join(lines)
Executable
+669
View File
@@ -0,0 +1,669 @@
#!/usr/bin/env python3
"""
myLPFR Mock Server Kompletan fiskalni server za testiranje
================================================================
Podržava sve /agent/v3, /api/v3 i /extension/v3 endpoint-e.
Automatski generiše QR kodove, snima račune, vodi log.
Pokreće se preko start.sh, gasi preko stop.sh.
"""
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 qrcode
from receipt import generate_receipt, generate_receipt_html, generate_report, load_locale
# ── Konfiguracija ──────────────────────────────────────────
PORT = 8989
HOST = "0.0.0.0"
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"
# Inicijalizuj foldere
for d in [DATA_DIR, INVOICES_DIR, QR_DIR, RECEIPTS_DIR]:
d.mkdir(parents=True, exist_ok=True)
# Prefix računa (čitaj iz fajla, ili kreni od 1)
COUNTER_FILE = DATA_DIR / "counter.txt"
# ── Bezbednosni element (mock kartica) ─────────────────────
# Fiksni test PIN — pravu karticu otključava korisnik svojim PIN-om,
# ovde je samo test vrednost kojom glumimo otključavanje.
PIN_BE = "1234"
# Putanja do NTech SQLite baze — čita se read-only. Podrazumevano ../ntech.db
# (koren repozitorijuma), može se promeniti preko NTECH_SQLITE.
NTECH_DB = os.environ.get("NTECH_SQLITE") or str(Path(__file__).parent.parent / "ntech.db")
def ucitaj_firmu():
"""Čita podatke o firmi iz NTech baze (read-only) i vraća ih kao dict.
Bezbednosni element 'već zna' identitet poreskog obveznika ovde to glumimo
čitanjem profila firme iz tabele podesavanja. Ako baza ili ključ nedostaje,
vraćamo test vrednosti da server i dalje radi."""
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"
return {
"name": naziv,
"tin": podaci.get("pib") or "123456789",
"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",
}
def get_next_invoice_number():
"""Vraća i inkrementira broj računa."""
if COUNTER_FILE.exists():
num = int(COUNTER_FILE.read_text().strip())
else:
num = 1
COUNTER_FILE.write_text(str(num + 1))
return f"{num:06d}"
def log(msg):
"""Upisuje poruku u log fajl i na stdout."""
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")
def generate_qr(url):
"""Pravi QR kod PNG i vraća base64 string."""
img = qrcode.make(url)
buf = io.BytesIO()
img.save(buf, format="PNG")
return base64.b64encode(buf.getvalue()).decode("utf-8")
# ── Gradi rute iz OpenAPI speca ─────────────────────────────
# Ako postoji myLPFR-api-docs.json, koristi ga.
# Ako ne, koristi hardkodovane rute.
SPEC_FILE = Path(__file__).parent.parent / "myLPFR-api-docs.json"
# Default rute (pale su iz Swagger speca)
DEFAULT_ROUTES = [
# Agent API
("GET", "agent/v3/attention", "attention"),
("GET", "agent/v3/environment-parameters", "environment"),
("POST", "agent/v3/invoices", "invoice"),
("GET", "agent/v3/invoices/:requestId", "invoice_lookup"),
("POST", "agent/v3/open-drawer", "open_drawer"),
("POST", "agent/v3/pin", "verify_pin"),
("POST", "agent/v3/print-text", "print_text"),
("GET", "agent/v3/receipts/:requestId", "receipt"),
("GET", "agent/v3/receipts/:requestId/text","receipt_text"),
("GET", "agent/v3/receipts/:requestId/html","receipt_html"),
("GET", "agent/v3/reports/daily", "daily_report"),
("GET", "agent/v3/reports/daily/text", "daily_report_text"),
("GET", "agent/v3/reports/periodic", "periodic_report"),
("GET", "agent/v3/reports/periodic/text", "periodic_report_text"),
("GET", "agent/v3/status", "status"),
("GET", "agent/v3/subject", "subject"),
# E-SDC API
("GET", "api/v3/attention", "attention"),
("GET", "api/v3/environment-parameters", "environment"),
("POST", "api/v3/invoices", "invoice"),
("GET", "api/v3/invoices/:requestId", "invoice_lookup"),
("POST", "api/v3/pin", "verify_pin"),
("GET", "api/v3/status", "status"),
# Extension API
("GET", "extension/v3/notifications", "notifications"),
("GET", "extension/v3/reports/daily", "daily_report"),
("GET", "extension/v3/reports/periodic", "periodic_report"),
("GET", "extension/v3/status-codes", "status_codes"),
("GET", "extension/v3/subject", "subject"),
]
# ── Response handler-i ──────────────────────────────────────
def sada():
"""Trenutno vreme u ISO formatu sa +02:00."""
tz = timezone(timedelta(hours=2))
return datetime.now(tz).strftime("%Y-%m-%dT%H:%M:%S.000+02:00")
def resp_attention():
return {"sdcDateTime": sada(), "status": "OK"}
def resp_status():
return {
"isPinRequired": True,
"auditRequired": False,
"sdcDateTime": sada(),
"lastInvoiceNumber": get_last_invoice_number(),
"protocolVersion": "1.0.0.0",
"secureElementVersion": "1.0",
"hardwareVersion": "1.0",
"softwareVersion": "0.3.18",
"deviceSerialNumber": "50-0002-NX6LC40XR3TQ",
"make": "MyOffice DOO",
"model": "myLPFR",
"mssc": [],
"gsc": ["1300", "0210"],
"supportedLanguages": ["sr-Cyrl-RS", "sr-Latin-RS"],
"uid": "",
"taxCoreApi": "https://suf-sandbox.purs.gov.rs",
"currentTaxRates": None,
"allTaxRates": [],
}
def resp_environment():
f = ucitaj_firmu()
return {
"tin": f["tin"],
"uid": "550e8400-e29b-41d4-a716-446655440000",
"taxCoreApi": "https://suf-sandbox.purs.gov.rs",
"sufVersion": "3.0",
"supportedLanguages": ["sr-Cyrl-RS", "sr-Latin-RS"],
"taxRates": [{
"validFrom": "2026-01-01",
"groupId": 1,
"taxCategories": [{
"categoryId": 1,
"name": "PDV",
"categoryType": "0",
"orderId": 1,
"taxRates": [
{"rateId": 1, "rate": 20.0, "label": "S"},
{"rateId": 2, "rate": 10.0, "label": "P"},
],
}],
}],
}
def resp_subject():
f = ucitaj_firmu()
return {
"tin": f["tin"],
"mb": f["mb"],
"uid": "550e8400-e29b-41d4-a716-446655440000",
"name": f["name"],
"address": f["address"],
"city": f["city"],
"country": "RS",
"district": f["district"],
"locationName": f["locationName"],
"businessUnitId": f["businessUnitId"],
}
def resp_invoice(request_id, request_body=None):
invoice_number = get_next_invoice_number()
verification_url = f"https://suf-sandbox.purs.gov.rs/verify/{request_id}"
qr_b64 = generate_qr(verification_url)
uid = f"550e8400-e29b-41d4-a716-{invoice_number.zfill(12)}"
invoice_data = {
"uid": uid,
"requestId": request_id,
"signedXml": f"<Invoice><UID>{uid}</UID><Number>{invoice_number}</Number><RequestId>{request_id}</RequestId><SignedAt>{sada()}</SignedAt></Invoice>",
"sdcDateTime": sada(),
"invoiceNumber": invoice_number,
"verificationUrl": verification_url,
"qrCode": qr_b64,
"encryptedInternalData": f"ENC_{uid}",
"signature": f"SIG_{invoice_number}_{request_id[:8]}",
}
# Proširi podatke iz tela zahteva (za štampu)
full_data = dict(invoice_data)
if request_body:
full_data.update(request_body)
full_data["invoiceNumber"] = invoice_number
full_data["sdcDateTime"] = invoice_data["sdcDateTime"]
full_data["qrCode"] = qr_b64
full_data["isFiscal"] = full_data.get("isFiscal", True)
f = ucitaj_firmu()
full_data.setdefault("tin", f["tin"])
full_data.setdefault("company", f["name"])
full_data.setdefault("store", f["locationName"])
full_data.setdefault("address", f["address"])
full_data.setdefault("district", f["district"])
full_data.setdefault("cashier", "Marko Marković")
full_data.setdefault("transactionType", "NSX")
full_data.setdefault("totalAmount", sum(item.get("amount", item.get("unitPrice", 0) * item.get("quantity", 0)) for item in full_data.get("items", [])))
full_data.setdefault("payments", [{"type": "Cash", "amount": full_data["totalAmount"]}])
full_data.setdefault("taxItems", [])
full_data.setdefault("totalTax", sum(t.get("amount", 0) for t in full_data.get("taxItems", [])))
full_data.setdefault("refund", 0)
full_data.setdefault("invoiceType", "Normal")
# Snimi kompletan račun
invoice_path = INVOICES_DIR / f"{invoice_number}_{request_id}.json"
with open(invoice_path, "w", encoding="utf-8") as f:
json.dump(full_data, f, indent=2, ensure_ascii=False)
# Snimi QR kod
qr_path = QR_DIR / f"{invoice_number}_{request_id}.png"
with open(qr_path, "wb") as f:
f.write(base64.b64decode(qr_b64))
# Generiši i snimi tekst računa (latinica)
receipt_text = generate_receipt(full_data, "latin")
receipt_path = RECEIPTS_DIR / f"{invoice_number}_{request_id}.txt"
receipt_path.write_text(receipt_text, encoding="utf-8")
# Generiši HTML račun
receipt_html = generate_receipt_html(full_data, "latin")
html_path = RECEIPTS_DIR / f"{invoice_number}_{request_id}.html"
html_path.write_text(receipt_html, encoding="utf-8")
log(f" 🧾 RAČUN {invoice_number} | requestId={request_id} | QR={qr_path.name} | Račun={receipt_path.name} | HTML={html_path.name}")
return invoice_data
def resp_invoice_lookup(request_id):
"""Pronađi postojeći račun po requestId."""
for f in INVOICES_DIR.glob("*.json"):
try:
data = json.loads(f.read_text(encoding="utf-8"))
if data.get("requestId") == request_id:
log(f" 🔍 Pronađen račun: {f.name}")
return data
except Exception:
continue
return None
def get_last_invoice_number():
"""Poslednji broj računa (bez inkrementiranja)."""
if COUNTER_FILE.exists():
num = int(COUNTER_FILE.read_text().strip()) - 1
return f"{num:06d}" if num >= 1 else ""
return ""
def resp_verify_pin(request_body=None):
"""Glumi otključavanje kartice PIN-om. Prihvata telo kao JSON {"pin": "..."}
ili kao goli string. Poredi sa fiksnim test PIN-om PIN_BE."""
uneti = ""
if isinstance(request_body, dict):
uneti = str(request_body.get("pin", "")).strip()
elif isinstance(request_body, str):
uneti = request_body.strip().strip('"')
if uneti == PIN_BE:
log(" 🔓 PIN ispravan — kartica otključana")
return {"status": "OK", "message": "PIN verifikovan"}
log(" 🔒 Pogrešan PIN")
return {"status": "ERROR", "code": "E003", "message": "Pogrešan PIN"}
def resp_open_drawer():
return {"status": "OK", "message": "Fioka otvorena"}
def resp_print_text():
return {"status": "OK", "message": "Tekst odštampan"}
def resp_receipt(request_id):
"""Vraća sačuvani račun u tekst formatu (za štampu)."""
# Prvo probaj da nađeš po requestId
for f in sorted(RECEIPTS_DIR.glob("*.txt"), reverse=True):
if request_id in f.stem:
return {
"contentType": "text/plain; charset=utf-8",
"receiptText": f.read_text(encoding="utf-8"),
"requestId": request_id,
}
return {
"contentType": "text/plain; charset=utf-8",
"receiptText": "Račun nije pronađen.",
"requestId": request_id,
}
def resp_receipt_html(request_id):
"""Vraća sačuvani račun u HTML formatu (za A4 štampu iz browsera)."""
for f in sorted(RECEIPTS_DIR.glob("*.html"), reverse=True):
if request_id in f.stem:
return f.read_text(encoding="utf-8")
return "<h1>Račun nije pronađen</h1>"
def _build_report(title, start_date=None, end_date=None):
"""Pravi izveštaj iz snimljenih računa."""
_firma = ucitaj_firmu()
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
total_payments_by_type = {}
total_tax_by_label = {}
per_tx_data = {}
invoice_count = 0
grand_total = 0.0
grand_tax = 0.0
for inv in invoices:
invoice_count += 1
# Plaćanja
for p in inv.get("payments", []):
ptype = p.get("type", "Other")
amt = float(p.get("amount", 0))
total_payments_by_type[ptype] = total_payments_by_type.get(ptype, 0.0) + amt
grand_total += amt
# Porezi
for t in inv.get("taxItems", []):
lbl = t.get("label", "")
rate = float(t.get("rate", 0))
amt = float(t.get("amount", 0))
key = f"{lbl}_{rate}"
if key not in total_tax_by_label:
total_tax_by_label[key] = {"label": lbl, "rate": rate, "total": 0.0, "amount": 0.0}
total_tax_by_label[key]["total"] += float(inv.get("totalAmount", 0))
total_tax_by_label[key]["amount"] += amt
grand_tax += amt
# Po tipu transakcije
tx = inv.get("transactionType", "NSX")
if tx not in per_tx_data:
per_tx_data[tx] = {"transactionTypeExt": tx, "invoiceCount": 0, "payments": {}, "taxItems": {}}
per_tx_data[tx]["invoiceCount"] += 1
for p in inv.get("payments", []):
ptype = p.get("type", "Other")
amt = float(p.get("amount", 0))
per_tx_data[tx]["payments"][ptype] = per_tx_data[tx]["payments"].get(ptype, 0.0) + amt
for t in inv.get("taxItems", []):
lbl = t.get("label", "")
rate = float(t.get("rate", 0))
amt = float(t.get("amount", 0))
key = f"{lbl}_{rate}"
if key not in per_tx_data[tx]["taxItems"]:
per_tx_data[tx]["taxItems"][key] = {"label": lbl, "rate": rate, "total": 0.0, "amount": 0.0}
per_tx_data[tx]["taxItems"][key]["total"] += float(inv.get("totalAmount", 0))
per_tx_data[tx]["taxItems"][key]["amount"] += amt
# Formatiraj
payments_list = [{"paymentType": k, "amount": v} for k, v in total_payments_by_type.items()]
tax_list = list(total_tax_by_label.values())
per_tx_list = []
for tx, data in per_tx_data.items():
tx_payments = [{"paymentType": k, "amount": v} for k, v in data["payments"].items()]
tx_taxes = list(data["taxItems"].values())
tx_total_pmts = sum(p["amount"] for p in tx_payments)
tx_total_taxes = sum(t["amount"] for t in tx_taxes)
per_tx_list.append({
"transactionTypeExt": tx,
"invoiceCount": data["invoiceCount"],
"payments": tx_payments,
"totalPayments": tx_total_pmts,
"taxItems": tx_taxes,
"totalTax": tx_total_taxes,
})
report_data = {
"title": title,
"number": 1,
"dateTime": sada(),
"tin": _firma["tin"],
"businessName": _firma["name"],
"locationName": _firma["locationName"],
"address": _firma["address"],
"district": _firma["district"],
"uid": "550e8400-e29b-41d4-a716-000000000001",
"startDate": start_date or datetime.now().strftime("%Y-%m-%d"),
"endDate": end_date or datetime.now().strftime("%Y-%m-%d"),
"total": {
"invoiceCount": invoice_count,
"payments": payments_list,
"totalPayments": grand_total,
"taxItems": tax_list,
"totalTax": grand_tax,
},
"perTransactionType": per_tx_list,
}
return report_data
def resp_daily_report():
today = datetime.now().strftime("%Y-%m-%d")
locale = load_locale("latin")
report_data = _build_report(locale.get("daily-report", "DNEVNI IZVEŠTAJ"), today, today)
# Snimi izveštaj
report_path = RECEIPTS_DIR / f"daily-report-{today}.json"
with open(report_path, "w", encoding="utf-8") as f:
json.dump(report_data, f, indent=2, ensure_ascii=False)
# Generiši tekst izveštaj
report_text = generate_report(report_data, "latin")
text_path = RECEIPTS_DIR / f"daily-report-{today}.txt"
text_path.write_text(report_text, encoding="utf-8")
log(f" 📊 DNEVNI IZVEŠTAJ | računa: {report_data['total']['invoiceCount']} | ukupno: {report_data['total']['totalPayments']:.2f}")
return report_data
def resp_periodic_report():
today = datetime.now().strftime("%Y-%m-%d")
locale = load_locale("latin")
report_data = _build_report(locale.get("periodic-report", "PERIODIČNI IZVEŠTAJ"), "2026-01-01", today)
report_path = RECEIPTS_DIR / f"periodic-report-{today}.json"
with open(report_path, "w", encoding="utf-8") as f:
json.dump(report_data, f, indent=2, ensure_ascii=False)
report_text = generate_report(report_data, "latin")
text_path = RECEIPTS_DIR / f"periodic-report-{today}.txt"
text_path.write_text(report_text, encoding="utf-8")
log(f" 📊 PERIODIČNI IZVEŠTAJ | računa: {report_data['total']['invoiceCount']} | ukupno: {report_data['total']['totalPayments']:.2f}")
return report_data
def resp_notifications():
return [{
"id": "1",
"type": "INFO",
"message": "Sistem funkcioniše ispravno",
"timestamp": sada(),
}]
def resp_status_codes():
return {
"codes": [
{"code": "S001", "description": "Uspešno potpisan račun"},
{"code": "E001", "description": "Greška pri potpisivanju"},
{"code": "E002", "description": "Kartica nije prisutna"},
{"code": "E003", "description": "Pogrešan PIN"},
{"code": "E004", "description": "Nema konekcije ka SUF serveru"},
],
}
# Mapiranje handler-a
HANDLERS = {
"attention": resp_attention,
"status": resp_status,
"environment": resp_environment,
"subject": resp_subject,
"invoice": resp_invoice,
"invoice_lookup": resp_invoice_lookup,
"verify_pin": resp_verify_pin,
"open_drawer": resp_open_drawer,
"print_text": resp_print_text,
"receipt": resp_receipt,
"receipt_text": resp_receipt,
"receipt_html": resp_receipt_html,
"daily_report": resp_daily_report,
"daily_report_text": resp_daily_report,
"periodic_report": resp_periodic_report,
"periodic_report_text": resp_periodic_report,
"notifications": resp_notifications,
"status_codes": resp_status_codes,
}
# ── HTTP Handler ─────────────────────────────────────────────
class FiscalHandler(http.server.BaseHTTPRequestHandler):
"""Glavni handler za sve fiskalne endpoint-e."""
def log_message(self, fmt, *args):
"""Override — koristi naš log umesto default stderr."""
pass # Logujemo ručno u _handle
def do_GET(self):
self._handle("GET")
def do_POST(self):
self._handle("POST")
def do_OPTIONS(self):
"""CORS preflight."""
self.send_response(200)
self.send_header("Access-Control-Allow-Origin", "*")
self.send_header("Access-Control-Allow-Methods", "GET,POST,OPTIONS")
self.send_header("Access-Control-Allow-Headers", "Content-Type, Accept-Language, RequestId")
self.end_headers()
def _handle(self, method):
path = self.path.split("?")[0]
hdrs = {
"Access-Control-Allow-Origin": "*",
"Access-Control-Allow-Methods": "GET,POST,OPTIONS",
"Access-Control-Allow-Headers": "Content-Type, Accept-Language, RequestId",
}
status = 404
body = None
matched_route = None
# Nađi rutu
r_handler_name = None
for r_method, r_pattern, r_handler_name in DEFAULT_ROUTES:
if r_method != method:
continue
# Konvertuj :param u regex
regex = re.sub(r":(\w+)", r"(?P<\1>[^/]+)", r_pattern)
regex = f"^{regex}$"
m = re.match(regex, path.strip("/"))
if m:
matched_route = r_pattern
handler = HANDLERS.get(r_handler_name)
if handler:
request_id = self.headers.get("RequestId", "unknown")
params = m.groupdict()
if r_handler_name == "invoice":
# Pročitaj telo zahteva
content_len = int(self.headers.get("Content-Length", 0))
request_body = None
if content_len > 0:
try:
raw = self.rfile.read(content_len)
request_body = json.loads(raw.decode("utf-8"))
except Exception:
request_body = None
body = handler(request_id, request_body)
elif r_handler_name == "verify_pin":
# Pročitaj telo (PIN) — JSON {"pin": "..."} ili goli string
content_len = int(self.headers.get("Content-Length", 0))
request_body = None
if content_len > 0:
try:
raw = self.rfile.read(content_len).decode("utf-8")
try:
request_body = json.loads(raw)
except Exception:
request_body = raw
except Exception:
request_body = None
body = handler(request_body)
elif r_handler_name == "invoice_lookup":
rid = params.get("requestId", "unknown")
result = handler(rid)
if result:
body = result
else:
status = 404
body = {"error": f"Račun {rid} nije pronađen"}
elif r_handler_name == "receipt":
body = handler(params.get("requestId", "unknown"))
elif r_handler_name == "receipt_text":
result = handler(params.get("requestId", "unknown"))
body = result.get("receiptText", "Račun nije pronađen.")
elif r_handler_name == "receipt_html":
body = handler(params.get("requestId", "unknown"))
else:
body = handler()
status = 200
break
# Log
emoji = "" if status == 200 else ""
client = self.client_address[0]
log(f" {emoji} {method} {path}{matched_route or '404'} | {client}")
# Pošalji odgovor
if r_handler_name in ("daily_report_text", "periodic_report_text") and body:
report_text = generate_report(body, "latin")
resp_bytes = report_text.encode("utf-8")
hdrs["Content-Type"] = "text/plain; charset=utf-8"
elif r_handler_name == "receipt_html" and body:
# HTML odgovor
resp_bytes = body.encode("utf-8") if isinstance(body, str) else body
hdrs["Content-Type"] = "text/html; charset=utf-8"
elif isinstance(body, str) and r_handler_name == "receipt_text":
# Tekst odgovor
resp_bytes = body.encode("utf-8") if isinstance(body, str) else body
hdrs["Content-Type"] = "text/plain; charset=utf-8"
elif body is not None:
resp_bytes = json.dumps(body, indent=2, ensure_ascii=False).encode("utf-8")
hdrs["Content-Type"] = "application/json; charset=utf-8"
else:
resp_bytes = json.dumps({"error": "Not Found", "path": path}, ensure_ascii=False).encode("utf-8")
hdrs["Content-Type"] = "application/json; charset=utf-8"
self.send_response(status)
for k, v in hdrs.items():
self.send_header(k, v)
self.send_header("Content-Length", str(len(resp_bytes)))
self.end_headers()
self.wfile.write(resp_bytes)
# ── Main ─────────────────────────────────────────────────────
def main():
log("╔══════════════════════════════════════════════╗")
log("║ 🧾 myLPFR Mock Server — Fiskalni server ║")
log("║ http://{}:{}/ ║".format(HOST, PORT))
log("{} ruta | QR: AUTO | Snimanje: UKLJUČENO ║".format(len(DEFAULT_ROUTES)))
log("╚══════════════════════════════════════════════╝")
log(f" 📁 Podaci: {DATA_DIR}")
log(f" 🧾 Računi: {INVOICES_DIR}")
log(f" 📱 QR PNG: {QR_DIR}")
log(f" 📝 Log: {LOG_FILE}")
f = ucitaj_firmu()
log(f" 💳 Kartica (BE) — baza: {NTECH_DB}")
log(f" Firma: {f['name']} | PIB: {f['tin']} | MB: {f['mb']}")
log(f" Adresa: {f['address']} | Test PIN: {PIN_BE}")
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()
Executable
+80
View File
@@ -0,0 +1,80 @@
#!/bin/bash
# ── start.sh — Pokreće fiskalni mock server ─────────────────
# ./start.sh — pokreće server u pozadini
# ./start.sh -f — pokreće server u prvom planu (vidiš logove)
# ./start.sh status — proverava da li je server živ
SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)"
PID_FILE="$SCRIPT_DIR/data/server.pid"
PORT=8989
# Proveri da li server već radi
check_running() {
if [ -f "$PID_FILE" ]; then
PID=$(cat "$PID_FILE")
if kill -0 "$PID" 2>/dev/null; then
return 0 # radi
fi
fi
# Proveri i port
if ss -tlnp 2>/dev/null | grep -q ":$PORT "; then
return 0 # radi
fi
return 1 # ne radi
}
stop_server() {
echo "⏹ Zaustavljam server..."
if [ -f "$PID_FILE" ]; then
PID=$(cat "$PID_FILE")
kill "$PID" 2>/dev/null && echo " PID $PID ugašen"
rm -f "$PID_FILE"
fi
fuser -k "${PORT}/tcp" 2>/dev/null
echo "✅ Server zaustavljen"
}
case "${1:-}" in
status)
if check_running; then
echo "🟢 Server je POKRENUT na http://localhost:$PORT"
echo " PID: $(cat "$PID_FILE" 2>/dev/null || echo '?')"
else
echo "🔴 Server NIJE pokrenut"
fi
;;
stop)
stop_server
;;
-f|--foreground)
echo "🔧 Pokrećem server u PRVOM PLANU (Ctrl+C gasi)..."
cd "$SCRIPT_DIR"
python3 -u server.py
;;
*)
if check_running; then
echo "⚠️ Server je VEĆ pokrenut na http://localhost:$PORT"
echo " Koristi './start.sh stop' da ga zaustaviš, ili './start.sh status'"
exit 1
fi
echo "🔧 Pokrećem server u pozadini..."
cd "$SCRIPT_DIR"
mkdir -p data
nohup python3 -u server.py > "$SCRIPT_DIR/data/server_stdout.log" 2>&1 &
PID=$!
echo "$PID" > "$PID_FILE"
sleep 2
if check_running; then
echo "✅ Server pokrenut! PID: $PID"
echo " http://localhost:$PORT"
echo ""
echo " ./start.sh stop — zaustavi server"
echo " ./start.sh status — proveri status"
echo " ./start.sh -f — pokreni u prvom planu (vidiš logove uživo)"
else
echo "❌ Server nije uspeo da se pokrene. Proveri: data/server_stdout.log"
rm -f "$PID_FILE"
exit 1
fi
;;
esac
Executable
+29
View File
@@ -0,0 +1,29 @@
#!/bin/bash
# ── stop.sh — Zaustavlja fiskalni mock server ───────────────
SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)"
PID_FILE="$SCRIPT_DIR/data/server.pid"
PORT=8989
echo "⏹ Zaustavljam fiskalni mock server..."
# Pokušaj preko PID fajla
if [ -f "$PID_FILE" ]; then
PID=$(cat "$PID_FILE")
if kill "$PID" 2>/dev/null; then
echo " PID $PID ugašen."
fi
rm -f "$PID_FILE"
fi
# Osiguraj da port nije zauzet
fuser -k "${PORT}/tcp" 2>/dev/null
sleep 0.5
# Proveri
if ss -tlnp 2>/dev/null | grep -q ":$PORT "; then
echo "⚠️ Port $PORT je još uvek zauzet. Pokušaj: kill -9 \$(fuser ${PORT}/tcp)"
else
echo "✅ Server zaustavljen. Port $PORT je slobodan."
fi
Executable
+171
View File
@@ -0,0 +1,171 @@
#!/bin/bash
# ═══════════════════════════════════════════════════════════════
# fisk-test.sh — Interaktivni fiskalni test server
# ═══════════════════════════════════════════════════════════════
# Pokreće myLPFR Mock Server, pokazuje info,
# i čeka da pritisneš 'q' da ga ugasiš.
set -e
SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)"
FISK_DIR="$SCRIPT_DIR/Fisk"
PID_FILE="$FISK_DIR/data/server.pid"
PORT=8989
HOST="localhost"
BASE_URL="http://$HOST:$PORT"
# ── Boje za output ─────────────────────────────────────────
GREEN='\033[0;32m'
YELLOW='\033[1;33m'
RED='\033[0;31m'
CYAN='\033[0;36m'
BOLD='\033[1m'
NC='\033[0m' # No Color
# ── Cleanup na izlaz ───────────────────────────────────────
cleanup() {
echo ""
echo -e "${YELLOW}⏹ Gasim server...${NC}"
# Preko PID fajla
if [ -f "$PID_FILE" ]; then
PID=$(cat "$PID_FILE")
kill "$PID" 2>/dev/null && echo -e " PID ${BOLD}$PID${NC} ugašen"
rm -f "$PID_FILE"
fi
# Pobrini se da port 8989 bude slobodan
fuser -k "${PORT}/tcp" 2>/dev/null || true
sleep 0.5
if ss -tlnp 2>/dev/null | grep -q ":$PORT "; then
echo -e "${RED}⚠️ Port $PORT je još uvek zauzet.${NC}"
echo " Ručno: kill -9 \$(fuser ${PORT}/tcp)"
else
echo -e "${GREEN}✅ Server zaustavljen. Port $PORT je slobodan.${NC}"
fi
exit 0
}
trap cleanup INT TERM
# ── Provera da li server već radi ──────────────────────────
check_running() {
# Provera preko PID fajla
if [ -f "$PID_FILE" ]; then
PID=$(cat "$PID_FILE")
if kill -0 "$PID" 2>/dev/null; then
return 0
fi
fi
# Provera preko porta
if ss -tlnp 2>/dev/null | grep -q ":$PORT " || \
netstat -tlnp 2>/dev/null | grep -q ":$PORT "; then
return 0
fi
return 1
}
# ── Dohvatanje PID-a servera ───────────────────────────────
get_pid() {
if [ -f "$PID_FILE" ]; then
cat "$PID_FILE"
else
ss -tlnp 2>/dev/null | grep ":$PORT " | sed -E 's/.*pid=([0-9]+).*/\1/' | head -1
fi
}
# ═══════════════════════════════════════════════════════════════
# MAIN
# ═══════════════════════════════════════════════════════════════
echo ""
echo -e "${BOLD}╔══════════════════════════════════════════════════╗${NC}"
echo -e "${BOLD}║ 🧾 myLPFR Fiskalni Test Server ║${NC}"
echo -e "${BOLD}╚══════════════════════════════════════════════════╝${NC}"
echo ""
if check_running; then
PID=$(get_pid)
echo -e "${GREEN}🟢 Server je VEĆ pokrenut${NC}"
echo -e " URL: ${CYAN}$BASE_URL${NC}"
echo -e " PID: ${BOLD}$PID${NC}"
else
echo -e "${YELLOW}🔧 Pokrećem fiskalni server...${NC}"
# Osiguraj data folder
mkdir -p "$FISK_DIR/data"
# Pokreni server u pozadini
cd "$FISK_DIR"
nohup python3 -u server.py > "$FISK_DIR/data/server_stdout.log" 2>&1 &
PID=$!
echo "$PID" > "$PID_FILE"
cd "$SCRIPT_DIR"
# Sačekaj da server bude spreman
echo -n " Čekam server"
for i in $(seq 1 20); do
sleep 0.3
echo -n "."
if check_running; then
echo ""
break
fi
done
if check_running; then
echo ""
echo -e "${GREEN}✅ Server pokrenut!${NC}"
else
echo ""
echo -e "${RED}❌ Server nije uspeo da se pokrene!${NC}"
echo " Proveri log: tail -f $FISK_DIR/data/server_stdout.log"
rm -f "$PID_FILE"
exit 1
fi
fi
# ── Prikaži info o serveru ─────────────────────────────────
echo ""
echo -e "${BOLD}── Podaci o serveru ──────────────────────────────${NC}"
echo -e " 🌐 URL: ${CYAN}$BASE_URL${NC}"
echo -e " 📦 PID: ${BOLD}$(get_pid)${NC}"
echo -e " 📁 Data dir: ${FISK_DIR}/data"
echo -e " 🧾 Računi: ${FISK_DIR}/data/invoices"
echo -e " 📱 QR kodovi: ${FISK_DIR}/data/qr"
echo -e " 📝 Log servera: ${FISK_DIR}/data/server.log"
echo -e " 📤 Stdout log: ${FISK_DIR}/data/server_stdout.log"
echo ""
# Prikaži trenutni brojač ako postoji
COUNTER_FILE="$FISK_DIR/data/counter.txt"
if [ -f "$COUNTER_FILE" ]; then
echo -e " 🔢 Sledeći račun: ${BOLD}$(cat "$COUNTER_FILE")${NC}"
else
echo -e " 🔢 Sledeći račun: ${BOLD}000001${NC}"
fi
# Prikaži poslednjih 5 log linija servera
if [ -f "$FISK_DIR/data/server.log" ] && [ -s "$FISK_DIR/data/server.log" ]; then
echo ""
echo -e "${BOLD}── Poslednje log linije ─────────────────────────${NC}"
tail -5 "$FISK_DIR/data/server.log" | while read -r line; do
echo -e " ${CYAN}$line${NC}"
done
fi
echo ""
echo -e "${BOLD}──────────────────────────────────────────────────${NC}"
echo -e " ${GREEN}Server je spreman za testiranje.${NC}"
echo -e " Pritisni ${BOLD}q${NC} + Enter da zaustaviš server."
echo -e "${BOLD}──────────────────────────────────────────────────${NC}"
echo ""
# ── Čekaj 'q' ─────────────────────────────────────────────
while true; do
read -r -p " ⌨ Unesi 'q' za gašenje: " input
if [ "$input" = "q" ] || [ "$input" = "Q" ]; then
cleanup
fi
done