webshop/app/payment.py

import hashlib
import hmac
import json
import logging
import os
from typing import NoReturn
from pathlib import Path

import httpx
from fastapi import HTTPException
from fastapi.encoders import jsonable_encoder

from pydantic import ValidationError
from app.schema import CartSession, NowPaymentsInvoice, ShippingCostEntry, ShippingRate

logger = logging.getLogger(__name__)

def prepare_stripe_data(cart: CartSession) -> list[dict]:
    """Returns list of items ready for Stripe to be processed.
    Each item is in the form of {price_data: {...}} or {price: <price_id>, quantity: <amount>}
    
    When using product IDs, we need to use price_data to create prices inline:
    list_items = [
        {
            'price_data': {
                'currency': 'usd',
                'product': 'prod_XYZ',
                'unit_amount': 2000  # amount in cents
            },
            'quantity': 1
        }
    ]
    
    When using price IDs:
    list_items = [
        {
            'price': 'price_XYZ',
            'quantity': 1
        }
    ]
    """
    list_items = []

    for item in cart.items.values():
        # Create inline price data using the product ID
        item_data = {
            "price_data": {
                "currency": cart.meta.currency.lower(),
                "product": item.product_id,
                "unit_amount": int(item.price * 100 / item.quantity),  # Convert to cents
            },
            "quantity": item.quantity,
        }

        list_items.append(item_data)

    return list_items


def is_country_in_europe(country: str) -> bool | NoReturn:
    """
    Check if given country is part of a list of countries in Europe.
    """

    dir_path = Path(__file__).resolve().parent
    europe_db_filepath = Path(f"{dir_path}/europe.json")

    try:
        with open(europe_db_filepath, encoding='utf-8') as file:
            europe_db = json.load(file)

            if country in europe_db['countries']:
                return True

            return False

    except FileNotFoundError as error:
        logger.error(f"is-country-in-europe => {error}")
        

def calculate_shipping_cost(country: str, weight: int, currency: str):
    """Returns appropriate shipping price based on given Country and product's Weight."""

    dir_path = Path(__file__).resolve().parent
    shipping_cost_db_filepath = Path(f"{dir_path}/shipping-cost.json")

    try:
        with open(shipping_cost_db_filepath, encoding='utf-8') as f:
            shipping_cost_db = json.load(f)

            area = None
            if country == "Switzerland":
                area = country
                
            elif is_country_in_europe(country):
                area = "Europe"
                
            else:
                area = "International"
            

            shipping_rate = None
            for entry in shipping_cost_db['areas'][area]:

                try:
                    entry = ShippingCostEntry(**entry)

                    if (
                            entry.max is None and weight >= entry.min or
                            entry.min <= weight <= entry.max
                    ):

                        shipping_rate = ShippingRate(price=entry.price, area=area)

                
                except ValidationError as e:
                    logger.error("pydantic validation errors =>")
                    for e in e.errors():
                        logger.error(f"error => {e}")
                

            return shipping_rate


    except FileNotFoundError as error:
        logger.error(f"calculate-shipping-cost => {error}")

        

def prepare_shipping_rate(country: str, weight: int, currency: str):
    """
    """

    shipping_rate = calculate_shipping_cost(country, weight, currency)

    # Stripe needs shipping price set in cents as an integer
    shipping_rate.price = int(shipping_rate.price * 100)

    return {
        "shipping_rate_data": {
            "type": "fixed_amount",
            "fixed_amount": {"amount": shipping_rate.price, "currency": currency},
            "display_name": shipping_rate.area,
        }
    }


async def generate_payment_link(data: NowPaymentsInvoice) -> tuple[str, str] | NoReturn:
    """Generate payment link for NOW Payments crypto payment service.
    See <https://documenter.getpostman.com/view/7907941/2s93JusNJt#f5e4e645-dce2-4b06-b2ca-2a29aaa5e845>.
    """
    environment = os.getenv("ENV")

    API_URL = "https://api.nowpayments.io/v1/"
    NOW_API_KEY = os.getenv("NOWPAYMENTS_API_KEY")

    if environment != "production":
        API_URL = "https://api-sandbox.nowpayments.io/v1/"
        NOW_API_KEY = os.getenv("NOWPAYMENTS_SANDBOX_API_KEY")

    ENDPOINT = "invoice"

    headers = {"x-api-key": NOW_API_KEY}
    async with httpx.AsyncClient(headers=headers) as client:
        resp = await client.post(f"{API_URL}{ENDPOINT}", json=jsonable_encoder(data))

        try:
            resp.raise_for_status()

            if resp.status_code == 200:
                data = resp.json()
                return data["invoice_url"], data["id"]

        except httpx.HTTPStatusError as exc:
            logger.error(f"Error {exc.response.status_code}: {exc.response.text}")
            raise HTTPException(
                status_code=400, detail="NOWPayments link cannot be generated."
            )


def ipn_signature_check(
    np_secret_key: str, np_x_signature: str, payload: dict[str]
) -> bool:
    """Signature verification function as part of the NOW Paymnents's
    Instant Payment Notifications (IPN) webhook workflow.
    See <https://documenter.getpostman.com/view/7907941/2s93JusNJt>.
    """
    sorted_msg = json.dumps(payload, separators=(",", ":"), sort_keys=True)

    digest = hmac.new(
        str(np_secret_key).encode(), f"{sorted_msg}".encode(), hashlib.sha512
    )

    signature = digest.hexdigest()

    if signature == np_x_signature:
        return True
    else:
        return False