""" NETOPIA Payments helper for Flet/Flask apps ------------------------------------------- This module wraps the official NETOPIA Python SDK (API v2) and exposes: • an easy `start_card_payment(...)` helper you can call from your app • IPN verification and order status helpers • an optional Flask Blueprint with REST endpoints you can plug into your existing backend (or the small Flask service that often accompanies Flet apps) Requirements: pip install netopia-sdk flask Environment variables expected (see README at bottom of file too): NETOPIA_API_KEY – API key from admin.netopia-payments.com NETOPIA_POS_SIGNATURE – POS signature NETOPIA_PUBLIC_KEY – RSA public key (PEM string) NETOPIA_PRIVATE_KEY – RSA private key (PEM string, optional) NETOPIA_REDIRECT_URL – your site URL where the buyer returns after payment NETOPIA_NOTIFY_URL – public URL that receives IPN callbacks NETOPIA_CANCEL_URL – optional cancel URL NETOPIA_POS_SIGNATURE_SET – optional, CSV list of allowed POS signatures NETOPIA_IS_LIVE – 'true' to use production; otherwise sandbox Usage from Flet (example): from helpers.netopia import start_card_payment url_payload = start_card_payment( order_id="TG-100045", amount=159.90, currency="RON", description="Comandă #TG-100045", customer={ "email": "client@example.com", "phone": "0712345678", "first_name": "Ion", "last_name": "Popescu", "city": "București", "country": "642", # 642 = Romania "address": "Str. Exemplu 10", "zip": "010101", "county": "B-IF", "language": "ro", }, products=[ {"name": "Mix nuci 500g", "code": "MX500", "category": "fructe-uscate", "price": 79.95, "vat": 0}, {"name": "Caju 500g", "code": "CJ500", "category": "fructe-uscate", "price": 79.95, "vat": 0}, ], installments=1, ) # url_payload typically contains the redirect URL – open it in a webview or browser Notes: • You must expose your IPN endpoint publicly (HTTPS) and configure it in Netopia admin. • Always trust order status updates coming from IPN, not only the browser redirect. """ from __future__ import annotations import os from datetime import datetime, timezone from dataclasses import dataclass from typing import List, Optional, Dict, Any # NETOPIA SDK imports from netopia_sdk.config import Config from netopia_sdk.client import PaymentClient from netopia_sdk.payment import PaymentService from netopia_sdk.requests.models import ( StartPaymentRequest, ConfigData, PaymentData, PaymentOptions, Instrument, OrderData, BillingData, ProductsData, ShippingData, ) # --------------------------- # Configuration & wiring # --------------------------- @dataclass class NetopiaSettings: api_key: str pos_signature: str public_key_str: str notify_url: str redirect_url: str is_live: bool pos_signature_set: List[str] cancel_url: str | None = None private_key_str: str | None = None @classmethod def from_env(cls) -> "NetopiaSettings": print("API_KEY? ", os.getenv("NETOPIA_API_KEY") is not None) is_live_str = os.getenv("NETOPIA_IS_LIVE", "false").strip().lower() is_live = is_live_str in ("1", "true", "yes", "on") pos_sig = os.environ.get("NETOPIA_POS_SIGNATURE", "").strip() pos_sig_set_env = os.getenv("NETOPIA_POS_SIGNATURE_SET", pos_sig) pos_sig_set = [s.strip() for s in pos_sig_set_env.split(",") if s.strip()] return cls( api_key=os.environ.get("NETOPIA_API_KEY", "").strip(), pos_signature=pos_sig, public_key_str=os.environ.get("NETOPIA_PUBLIC_KEY", "").strip(), private_key_str=os.environ.get("NETOPIA_PRIVATE_KEY", "").strip(), notify_url=os.environ.get("NETOPIA_NOTIFY_URL", "").strip(), redirect_url=os.environ.get("NETOPIA_REDIRECT_URL", "").strip(), is_live=is_live, pos_signature_set=pos_sig_set, cancel_url=os.environ.get("NETOPIA_CANCEL_URL", None), ) def _build_payment_service(settings: Optional[NetopiaSettings] = None) -> PaymentService: """Create a PaymentService from settings/env.""" settings = settings or NetopiaSettings.from_env() if not settings.api_key: raise RuntimeError("NETOPIA_API_KEY is missing") if not settings.pos_signature: raise RuntimeError("NETOPIA_POS_SIGNATURE is missing") if not settings.public_key_str: raise RuntimeError("NETOPIA_PUBLIC_KEY is missing (PEM)") if not settings.notify_url: raise RuntimeError("NETOPIA_NOTIFY_URL is missing") if not settings.redirect_url: raise RuntimeError("NETOPIA_REDIRECT_URL is missing") config = Config( api_key=settings.api_key, pos_signature=settings.pos_signature, is_live=settings.is_live, notify_url=settings.notify_url, redirect_url=settings.redirect_url, public_key_str=settings.public_key_str, #private_key_str=settings.private_key_str, pos_signature_set=settings.pos_signature_set, ) client = PaymentClient(config) return PaymentService(client) # --------------------------- # Helpers: BillingData from minimal fields (SDK-supported) # --------------------------- def _derive_billing_from_customer(customer: Dict[str, str]) -> BillingData: """Construct BillingData using only fields supported by the SDK example. Accepts a single free-form address but does not send it (SDK BillingData in v2 typically supports: email, phone, firstName, lastName, city, country). We try to infer city from the first token before a comma if city is missing. """ city = (customer.get("city") or "").strip() if not city: addr = (customer.get("address") or "").strip() if "," in addr: candidate = addr.split(",", 1)[0].strip() if 1 <= len(candidate) <= 64: city = candidate return BillingData( email=customer.get("email", ""), phone=customer.get("phone", ""), firstName=customer.get("first_name", ""), lastName=customer.get("last_name", ""), city=city, country=int(customer.get("country", "642") or 642), countryName=customer.get("countryName", "Romania"), state=customer.get("state", customer.get("county", "")), postalCode=customer.get("zip", ""), details=(customer.get("address") or "").strip(), ) def _derive_shipping_from_customer(customer: Dict[str, str]) -> ShippingData: city = (customer.get("ship_city") or customer.get("city") or "").strip() if not city: addr = (customer.get("ship_address") or customer.get("address") or "").strip() if "," in addr: cand = addr.split(",", 1)[0].strip() if 1 <= len(cand) <= 64: city = cand return ShippingData( email=customer.get("ship_email", customer.get("email", "")), phone=customer.get("ship_phone", customer.get("phone", "")), firstName=customer.get("ship_first_name", customer.get("first_name", "")), lastName=customer.get("ship_last_name", customer.get("last_name", "")), city=city, country=int(customer.get("ship_country", customer.get("country", "642")) or 642), countryName=customer.get("ship_countryName", customer.get("countryName", "Romania")), state=customer.get("ship_state", customer.get("state", customer.get("county", ""))), postalCode=customer.get("ship_zip", customer.get("zip", "")), details=(customer.get("ship_address") or customer.get("address") or "").strip(), ) # --------------------------- # High‑level helpers (call these from your app) # --------------------------- def start_card_payment( *, order_id: str, amount: float, currency: str, description: str, customer: Dict[str, str], products: List[Dict[str, Any]], installments: int = 1, settings: Optional[NetopiaSettings] = None, ) -> Dict[str, Any]: """Create a redirect‑based card payment and return the SDK response. The response typically includes the URL you must redirect the buyer to. You should persist the order locally before calling this. """ svc = _build_payment_service(settings) billing = _derive_billing_from_customer(customer) shipping = _derive_shipping_from_customer(customer) prods: List[ProductsData] = [] for p in products: prods.append( ProductsData( name=str(p["name"]), code=str(p.get("code", p["name"]))[:32], category=str(p.get("category", "")), price=float(p.get("price", 0.0)), vat=int(p.get("vat", 0)), ) ) cfg = NetopiaSettings.from_env() if settings is None else settings now_iso = datetime.now(timezone.utc).strftime('%Y-%m-%dT%H:%M:%SZ') req = StartPaymentRequest( config=ConfigData( emailTemplate=customer.get("email_template", "default"), emailSubject=customer.get("email_subject", "Order Confirmation"), cancelUrl=cfg.cancel_url or cfg.redirect_url, notifyUrl=cfg.notify_url, redirectUrl=cfg.redirect_url, language=customer.get("language", "ro"), ), payment=PaymentData( options=PaymentOptions(installments=int(installments), bonus=0), instrument=None, data={}, ), order=OrderData( ntpID=None, posSignature=None, dateTime=now_iso, orderID=str(order_id), amount=float(amount), currency=str(currency), description=str(description), billing=billing, shipping=shipping, products=prods, installments={"selected": int(installments) if installments else 0, "available": []}, data={}, ), ) return svc.start_payment(req) def verify_ipn(raw_body: bytes, settings: Optional[NetopiaSettings] = None) -> Dict[str, Any]: """Verify an IPN payload coming from NETOPIA. Returns the decoded data. Raise an exception if verification fails. """ svc = _build_payment_service(settings) return svc.verify_ipn(raw_body) def get_status(*, ntp_id: Optional[str] = None, order_id: Optional[str] = None, settings: Optional[NetopiaSettings] = None) -> Dict[str, Any]: """Query order status by ntpID and/or orderID.""" svc = _build_payment_service(settings) return svc.get_status(ntpID=ntp_id, orderID=order_id)