add netopia payment process
This commit is contained in:
284
UI_V2/helpers/netopia.py
Normal file
284
UI_V2/helpers/netopia.py
Normal file
@@ -0,0 +1,284 @@
|
||||
"""
|
||||
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)
|
||||
Reference in New Issue
Block a user