#!/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 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 BE_ID = "TRNMOCK1" # simulirani BE/LPFR identifikator 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) # Test PIN (pravi Teron traži PIN za BE karticu) PIN_BE = "1234" # NTech SQLite baza (read-only) — čita podatke o firmi NTECH_DB = os.environ.get("NTECH_SQLITE") or str(Path(__file__).parent.parent / "ntech.db") # ── 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(): f = ucitaj_firmu() last = peek_counter("total") last_num = f"{ESIR_ID}-{BE_ID}-{last}" if last >= 1 else "" return { "isPinRequired": False, "auditRequired": False, "sdcDateTime": sada(), "lastInvoiceNumber": last_num, "protocolVersion": "1.0.0", "serialNumber": ESIR_ID, "tin": f["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('"') if uneti == PIN_BE: log(" 🔓 PIN ispravan") return {"status": "OK", "message": "PIN verifikovan"} log(" 🔒 Pogrešan PIN") return {"status": "ERROR", "code": "2100", "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(): f = ucitaj_firmu() return { "serialNumber": BE_ID, "tin": f["tin"], "name": f["name"], "validFrom": "2024-01-01T00:00:00+01:00", "validTo": "2027-01-01T00:00:00+01:00", "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", []) # Brojači total_cnt = get_counter("total") ext, tip_key = counter_ext(invoice_type, transaction_type) type_cnt = get_counter(tip_key) invoice_number = f"{ESIR_ID}-{BE_ID}-{total_cnt}" invoice_counter = f"{type_cnt}/{total_cnt}{ext}" verification_url = f"https://sandbox.suf.purs.gov.rs/v/?vl={invoice_number}" qr_b64 = generate_qr(verification_url) # 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) firma = ucitaj_firmu() # Odgovor koji ide ka NTech-u (ESIR-u) odgovor = { "requestedBy": ESIR_ID, "signedBy": BE_ID, "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", } # Puni podaci za snimanje i generisanje računa full_data = { **odgovor, "requestId": request_id, "invoiceType": invoice_type, "transactionType": transaction_type, "items": items, "payments": inv_req.get("payment", [{"type": "Cash", "amount": total_amount}]), "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": 0, # 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 receipt_text = generate_receipt(full_data, "latin") 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, "latin") 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(): firma = ucitaj_firmu() 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: {firma['name']} | PIB: {firma['tinPlain']}") log(f" 🆔 ESIR ID: {ESIR_ID} | BE ID: {BE_ID}") log(f" 🔑 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()