285 lines
11 KiB
Python
285 lines
11 KiB
Python
"""
|
||
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)
|