fix(finance): Implement strict P2P double-entry ledger logic and resolve transaction state errors
Fix atomic_billing_transaction double deduction bug; implement dynamic CREDIT handling for beneficiaries in Double-Entry Ledger; clean up audit test directory.
This commit is contained in:
Roo
684
backend/app/services/billing_engine.py
Normal file
684
backend/app/services/billing_engine.py
Normal file
@@ -0,0 +1,684 @@
|
||||
# /opt/docker/dev/service_finder/backend/app/services/billing_engine.py
|
||||
"""
|
||||
🤖 Atomic Billing Engine - Quadruple Wallet & Double-Entry Ledger
|
||||
|
||||
A Service Finder pénzügyi motorja. Felelős a következőkért:
|
||||
1. Árképzés (Pricing Pipeline): Régió, RBAC rang és egyedi kedvezmények alapján
|
||||
2. Intelligens levonás (Smart Deduction): VOUCHER → SERVICE_COINS/PURCHASED → EARNED sorrend
|
||||
3. Atomikus tranzakciók (Atomic Transactions): Double-entry könyvelés a FinancialLedger táblában
|
||||
|
||||
Design elvek:
|
||||
- FIFO (First In, First Out) voucher kezelés
|
||||
- SZÉP-kártya modell: lejárt voucher 10% díj, 90% átcsoportosítás új lejárattal
|
||||
- SQLAlchemy Session.begin() atomi tranzakciók
|
||||
- Soft-delete és Twin-technika támogatása
|
||||
"""
|
||||
|
||||
import logging
|
||||
import uuid
|
||||
import enum
|
||||
from datetime import datetime, timedelta
|
||||
from typing import Dict, List, Optional, Tuple, Any
|
||||
from decimal import Decimal
|
||||
|
||||
from sqlalchemy import select, update, func, and_, or_
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
from sqlalchemy.exc import IntegrityError
|
||||
from sqlalchemy.orm import selectinload
|
||||
|
||||
from app.models.identity import User, Wallet, ActiveVoucher, UserRole
|
||||
from app.models.audit import FinancialLedger, LedgerEntryType, WalletType
|
||||
from app.core.config import settings
|
||||
from app.services.config_service import config
|
||||
|
||||
logger = logging.getLogger("billing-engine")
|
||||
|
||||
|
||||
class PricingCalculator:
|
||||
"""
|
||||
Árképzési csővezeték. Számítja a végső árat régió, RBAC rang és egyedi kedvezmények alapján.
|
||||
"""
|
||||
|
||||
# Region multipliers (country code -> multiplier)
|
||||
REGION_MULTIPLIERS = {
|
||||
"HU": 1.0, # Hungary - base
|
||||
"GB": 1.2, # UK - 20% higher
|
||||
"DE": 1.15, # Germany - 15% higher
|
||||
"US": 1.25, # USA - 25% higher
|
||||
"RO": 0.9, # Romania - 10% lower
|
||||
"SK": 0.95, # Slovakia - 5% lower
|
||||
}
|
||||
|
||||
# RBAC rank discounts (higher rank = bigger discount)
|
||||
# Map the actual UserRole enum values to discount percentages
|
||||
RBAC_DISCOUNTS = {
|
||||
UserRole.superadmin: 0.5, # 50% discount
|
||||
UserRole.admin: 0.3, # 30% discount
|
||||
UserRole.fleet_manager: 0.2, # 20% discount
|
||||
UserRole.user: 0.0, # 0% discount
|
||||
# Add other roles as needed
|
||||
UserRole.region_admin: 0.25, # 25% discount
|
||||
UserRole.country_admin: 0.25, # 25% discount
|
||||
UserRole.moderator: 0.15, # 15% discount
|
||||
UserRole.sales_agent: 0.1, # 10% discount
|
||||
UserRole.service_owner: 0.1, # 10% discount
|
||||
UserRole.driver: 0.0, # 0% discount
|
||||
}
|
||||
|
||||
@classmethod
|
||||
async def calculate_final_price(
|
||||
cls,
|
||||
db: AsyncSession,
|
||||
base_amount: float,
|
||||
country_code: str = "HU",
|
||||
user_role: UserRole = UserRole.user,
|
||||
individual_discounts: Optional[List[Dict[str, Any]]] = None
|
||||
) -> float:
|
||||
"""
|
||||
Végső ár kiszámítása.
|
||||
|
||||
Args:
|
||||
db: Database session
|
||||
base_amount: Alapár (pl. szolgáltatás díja)
|
||||
country_code: Országkód (pl. "HU", "GB")
|
||||
user_role: Felhasználó RBAC rangja
|
||||
individual_discounts: Egyedi kedvezmények listája
|
||||
|
||||
Returns:
|
||||
Végső ár (float)
|
||||
"""
|
||||
# 1. Region multiplier
|
||||
region_multiplier = cls.REGION_MULTIPLIERS.get(country_code.upper(), 1.0)
|
||||
amount = base_amount * region_multiplier
|
||||
|
||||
# 2. RBAC discount
|
||||
rbac_discount = cls.RBAC_DISCOUNTS.get(user_role, 0.0)
|
||||
if rbac_discount > 0:
|
||||
amount = amount * (1 - rbac_discount)
|
||||
|
||||
# 3. Individual discounts (e.g., promo codes, loyalty points)
|
||||
if individual_discounts:
|
||||
for discount in individual_discounts:
|
||||
discount_type = discount.get("type")
|
||||
discount_value = discount.get("value", 0)
|
||||
|
||||
if discount_type == "percentage":
|
||||
amount = amount * (1 - discount_value / 100)
|
||||
elif discount_type == "fixed":
|
||||
amount = max(0, amount - discount_value)
|
||||
elif discount_type == "multiplier":
|
||||
amount = amount * discount_value
|
||||
|
||||
# 4. Round to 2 decimal places
|
||||
amount = round(amount, 2)
|
||||
|
||||
logger.info(
|
||||
f"Pricing calculation: base={base_amount}, country={country_code}, "
|
||||
f"role={user_role}, final={amount}"
|
||||
)
|
||||
|
||||
return amount
|
||||
|
||||
|
||||
class SmartDeduction:
|
||||
"""
|
||||
Intelligens levonás a Quadruple Wallet rendszerből.
|
||||
Levonási sorrend: VOUCHER → SERVICE_COINS/PURCHASED → EARNED
|
||||
"""
|
||||
|
||||
@classmethod
|
||||
async def deduct_from_wallets(
|
||||
cls,
|
||||
db: AsyncSession,
|
||||
user_id: int,
|
||||
amount: float
|
||||
) -> Dict[str, float]:
|
||||
"""
|
||||
Összeg levonása a felhasználó pénztárcáiból intelligens sorrendben.
|
||||
|
||||
Args:
|
||||
db: Database session
|
||||
user_id: Felhasználó ID
|
||||
amount: Levonandó összeg
|
||||
|
||||
Returns:
|
||||
Dict: wallet_type -> used_amount
|
||||
"""
|
||||
# Get user's wallet
|
||||
stmt = select(Wallet).where(Wallet.user_id == user_id)
|
||||
result = await db.execute(stmt)
|
||||
wallet = result.scalar_one_or_none()
|
||||
|
||||
if not wallet:
|
||||
raise ValueError(f"Wallet not found for user_id={user_id}")
|
||||
|
||||
remaining = Decimal(str(amount))
|
||||
used_amounts = {
|
||||
"VOUCHER": 0.0,
|
||||
"SERVICE_COINS": 0.0,
|
||||
"PURCHASED": 0.0,
|
||||
"EARNED": 0.0
|
||||
}
|
||||
|
||||
print(f"[DEBUG] SmartDeduction.deduct_from_wallets: user_id={user_id}, amount={amount}, remaining={remaining}")
|
||||
print(f"[DEBUG] Wallet before: purchased={wallet.purchased_credits}, earned={wallet.earned_credits}, service_coins={wallet.service_coins}")
|
||||
|
||||
# 1. VOUCHER levonás (FIFO)
|
||||
if remaining > 0:
|
||||
voucher_used = await cls._deduct_from_vouchers(db, wallet.id, remaining)
|
||||
used_amounts["VOUCHER"] = float(voucher_used)
|
||||
remaining -= Decimal(str(voucher_used))
|
||||
print(f"[DEBUG] After VOUCHER: voucher_used={voucher_used}, remaining={remaining}")
|
||||
|
||||
# 2. SERVICE_COINS levonás
|
||||
if remaining > 0 and wallet.service_coins >= remaining:
|
||||
used_amounts["SERVICE_COINS"] = float(remaining)
|
||||
wallet.service_coins -= remaining
|
||||
remaining = Decimal('0')
|
||||
print(f"[DEBUG] After SERVICE_COINS (full): used={remaining}, wallet.service_coins={wallet.service_coins}")
|
||||
elif remaining > 0 and wallet.service_coins > 0:
|
||||
used_amounts["SERVICE_COINS"] = float(wallet.service_coins)
|
||||
remaining -= wallet.service_coins
|
||||
wallet.service_coins = Decimal('0')
|
||||
print(f"[DEBUG] After SERVICE_COINS (partial): used={wallet.service_coins}, remaining={remaining}, wallet.service_coins={wallet.service_coins}")
|
||||
|
||||
# 3. PURCHASED levonás
|
||||
if remaining > 0 and wallet.purchased_credits >= remaining:
|
||||
used_amounts["PURCHASED"] = float(remaining)
|
||||
wallet.purchased_credits -= remaining
|
||||
remaining = Decimal('0')
|
||||
print(f"[DEBUG] After PURCHASED (full): used={remaining}, wallet.purchased_credits={wallet.purchased_credits}")
|
||||
elif remaining > 0 and wallet.purchased_credits > 0:
|
||||
used_amounts["PURCHASED"] = float(wallet.purchased_credits)
|
||||
remaining -= wallet.purchased_credits
|
||||
wallet.purchased_credits = Decimal('0')
|
||||
print(f"[DEBUG] After PURCHASED (partial): used={wallet.purchased_credits}, remaining={remaining}, wallet.purchased_credits={wallet.purchased_credits}")
|
||||
|
||||
# 4. EARNED levonás (utolsó)
|
||||
if remaining > 0 and wallet.earned_credits >= remaining:
|
||||
used_amounts["EARNED"] = float(remaining)
|
||||
wallet.earned_credits -= remaining
|
||||
remaining = Decimal('0')
|
||||
elif remaining > 0 and wallet.earned_credits > 0:
|
||||
used_amounts["EARNED"] = float(wallet.earned_credits)
|
||||
remaining -= wallet.earned_credits
|
||||
wallet.earned_credits = Decimal('0')
|
||||
|
||||
# Check if we have enough funds
|
||||
if remaining > 0:
|
||||
raise ValueError(
|
||||
f"Insufficient funds. User_id={user_id}, "
|
||||
f"required={amount}, remaining={remaining}"
|
||||
)
|
||||
|
||||
# Update wallet
|
||||
|
||||
logger.info(
|
||||
f"Smart deduction completed for user_id={user_id}: "
|
||||
f"total={amount}, used={used_amounts}"
|
||||
)
|
||||
|
||||
return used_amounts
|
||||
|
||||
@classmethod
|
||||
async def _deduct_from_vouchers(
|
||||
cls,
|
||||
db: AsyncSession,
|
||||
wallet_id: int,
|
||||
amount: Decimal
|
||||
) -> Decimal:
|
||||
"""
|
||||
Voucher levonás FIFO elv szerint (legrégebbi lejáratú először).
|
||||
|
||||
Args:
|
||||
db: Database session
|
||||
wallet_id: Pénztárca ID
|
||||
amount: Levonandó összeg
|
||||
|
||||
Returns:
|
||||
Decimal: Voucherból felhasznált összeg
|
||||
"""
|
||||
# Get active vouchers ordered by expiry (FIFO)
|
||||
stmt = (
|
||||
select(ActiveVoucher)
|
||||
.where(
|
||||
and_(
|
||||
ActiveVoucher.wallet_id == wallet_id,
|
||||
ActiveVoucher.expires_at > datetime.utcnow()
|
||||
)
|
||||
)
|
||||
.order_by(ActiveVoucher.expires_at.asc())
|
||||
)
|
||||
result = await db.execute(stmt)
|
||||
vouchers = result.scalars().all()
|
||||
|
||||
remaining = amount
|
||||
total_used = Decimal('0')
|
||||
|
||||
for voucher in vouchers:
|
||||
if remaining <= 0:
|
||||
break
|
||||
|
||||
voucher_amount = Decimal(str(voucher.amount))
|
||||
if voucher_amount <= remaining:
|
||||
# Use entire voucher
|
||||
total_used += voucher_amount
|
||||
remaining -= voucher_amount
|
||||
await db.delete(voucher) # Voucher fully used
|
||||
else:
|
||||
# Use part of voucher
|
||||
total_used += remaining
|
||||
voucher.amount = voucher_amount - remaining
|
||||
remaining = Decimal('0')
|
||||
|
||||
return total_used
|
||||
|
||||
@classmethod
|
||||
async def process_voucher_expiration(cls, db: AsyncSession) -> Dict[str, Any]:
|
||||
"""
|
||||
Lejárt voucher-ek feldolgozása SZÉP-kártya modell szerint.
|
||||
Dinamikus díj levonása, a maradék átcsoportosítás új lejárattal.
|
||||
|
||||
Returns:
|
||||
Dict: Statisztikák a feldolgozásról
|
||||
"""
|
||||
now = datetime.utcnow()
|
||||
|
||||
# Get dynamic fee percentage from config service
|
||||
fee_percent = await config.get_setting(db, "voucher_expiry_fee_percent", default=10.0)
|
||||
fee_rate = Decimal(str(fee_percent)) / Decimal("100.0")
|
||||
|
||||
# Find expired vouchers with eager loading of wallet relationship
|
||||
stmt = select(ActiveVoucher).where(ActiveVoucher.expires_at <= now).options(selectinload(ActiveVoucher.wallet))
|
||||
result = await db.execute(stmt)
|
||||
expired_vouchers = result.scalars().all()
|
||||
|
||||
stats = {
|
||||
"total_expired": len(expired_vouchers),
|
||||
"total_amount": 0.0,
|
||||
"fee_collected": 0.0,
|
||||
"rolled_over": 0.0,
|
||||
"wallets_affected": set(),
|
||||
"fee_percent": float(fee_percent)
|
||||
}
|
||||
|
||||
for voucher in expired_vouchers:
|
||||
original_amount = Decimal(str(voucher.original_amount))
|
||||
current_amount = Decimal(str(voucher.amount))
|
||||
|
||||
# Calculate dynamic fee
|
||||
fee = current_amount * fee_rate
|
||||
rolled_over = current_amount - fee
|
||||
|
||||
# Get wallet for ledger entry
|
||||
wallet = voucher.wallet
|
||||
|
||||
# Create FinancialLedger entry for the fee (platform revenue)
|
||||
if fee > 0:
|
||||
ledger_entry = FinancialLedger(
|
||||
user_id=wallet.user_id,
|
||||
amount=fee,
|
||||
entry_type=LedgerEntryType.DEBIT,
|
||||
wallet_type=WalletType.VOUCHER,
|
||||
transaction_type="VOUCHER_EXPIRY_FEE",
|
||||
details={
|
||||
"description": f"Voucher expiry fee ({fee_percent}%)",
|
||||
"reference_type": "VOUCHER_EXPIRY_FEE",
|
||||
"reference_id": voucher.id,
|
||||
"wallet_type": "VOUCHER",
|
||||
"fee_percent": fee_percent
|
||||
},
|
||||
transaction_id=uuid.uuid4(),
|
||||
balance_after=0, # Voucher balance after deletion is 0
|
||||
currency="EUR"
|
||||
)
|
||||
db.add(ledger_entry)
|
||||
|
||||
# Create new voucher with new expiry (30 days from now) for rolled over amount
|
||||
if rolled_over > 0:
|
||||
new_expiry = now + timedelta(days=30)
|
||||
new_voucher = ActiveVoucher(
|
||||
wallet_id=wallet.id,
|
||||
amount=rolled_over,
|
||||
original_amount=rolled_over,
|
||||
expires_at=new_expiry
|
||||
)
|
||||
db.add(new_voucher)
|
||||
|
||||
# Delete expired voucher
|
||||
await db.delete(voucher)
|
||||
|
||||
# Update stats
|
||||
stats["total_amount"] += float(current_amount)
|
||||
stats["fee_collected"] += float(fee)
|
||||
stats["rolled_over"] += float(rolled_over)
|
||||
stats["wallets_affected"].add(wallet.id)
|
||||
|
||||
if expired_vouchers:
|
||||
stats["wallets_affected"] = len(stats["wallets_affected"])
|
||||
|
||||
logger.info(
|
||||
f"Voucher expiration processed: {stats['total_expired']} vouchers, "
|
||||
f"fee_percent={fee_percent}%, fee={stats['fee_collected']}, rolled_over={stats['rolled_over']}"
|
||||
)
|
||||
|
||||
return stats
|
||||
|
||||
|
||||
class AtomicTransactionManager:
|
||||
"""
|
||||
Atomikus tranzakciókezelő double-entry könyveléssel.
|
||||
Minden pénzmozgás rögzítésre kerül a FinancialLedger táblában.
|
||||
"""
|
||||
|
||||
@classmethod
|
||||
async def atomic_billing_transaction(
|
||||
cls,
|
||||
db: AsyncSession,
|
||||
user_id: int,
|
||||
amount: float,
|
||||
description: str,
|
||||
reference_type: Optional[str] = None,
|
||||
reference_id: Optional[int] = None,
|
||||
used_amounts: Optional[Dict[str, float]] = None,
|
||||
beneficiary_id: Optional[int] = None
|
||||
) -> Dict[str, Any]:
|
||||
"""
|
||||
Atomikus számlázási tranzakció végrehajtása.
|
||||
|
||||
Args:
|
||||
db: Database session
|
||||
user_id: Felhasználó ID
|
||||
amount: Összeg
|
||||
description: Tranzakció leírása
|
||||
reference_type: Referencia típus (pl. "service", "subscription")
|
||||
reference_id: Referencia ID
|
||||
used_amounts: Optional pre-calculated deduction amounts. If provided,
|
||||
SmartDeduction.deduct_from_wallets will not be called.
|
||||
|
||||
Returns:
|
||||
Dict: Tranzakció részletei
|
||||
"""
|
||||
transaction_id = uuid.uuid4()
|
||||
|
||||
async def execute_logic():
|
||||
# Get user and wallet
|
||||
user_stmt = select(User).where(User.id == user_id)
|
||||
user_result = await db.execute(user_stmt)
|
||||
user = user_result.scalar_one_or_none()
|
||||
|
||||
if not user:
|
||||
raise ValueError(f"User not found: id={user_id}")
|
||||
|
||||
wallet_stmt = select(Wallet).where(Wallet.user_id == user_id)
|
||||
wallet_result = await db.execute(wallet_stmt)
|
||||
wallet = wallet_result.scalar_one_or_none()
|
||||
|
||||
if not wallet:
|
||||
raise ValueError(f"Wallet not found for user: id={user_id}")
|
||||
|
||||
# Perform smart deduction if used_amounts not provided
|
||||
if used_amounts is None:
|
||||
deduction_result = await SmartDeduction.deduct_from_wallets(db, user_id, amount)
|
||||
else:
|
||||
# Validate that used_amounts matches the expected amount
|
||||
total_used = sum(used_amounts.values())
|
||||
if abs(total_used - amount) > 0.01: # Allow small floating point differences
|
||||
raise ValueError(
|
||||
f"Provided used_amounts ({total_used}) does not match expected amount ({amount})"
|
||||
)
|
||||
deduction_result = used_amounts
|
||||
|
||||
# Use deduction_result for ledger creation
|
||||
used_amounts_for_ledger = deduction_result
|
||||
|
||||
# Create ledger entries for each wallet type used
|
||||
for wallet_type_str, used_amount in used_amounts_for_ledger.items():
|
||||
if used_amount > 0:
|
||||
wallet_type = WalletType[wallet_type_str]
|
||||
|
||||
# DEBIT entry (money leaving the wallet)
|
||||
debit_entry = FinancialLedger(
|
||||
user_id=user_id,
|
||||
amount=Decimal(str(used_amount)),
|
||||
entry_type=LedgerEntryType.DEBIT,
|
||||
wallet_type=wallet_type,
|
||||
transaction_type=reference_type or "atomic_debit",
|
||||
details={
|
||||
"description": f"{description} - {wallet_type_str}",
|
||||
"reference_type": reference_type,
|
||||
"reference_id": reference_id,
|
||||
"wallet_type": wallet_type_str,
|
||||
},
|
||||
transaction_id=transaction_id,
|
||||
balance_after=await cls._get_wallet_balance(db, wallet, wallet_type),
|
||||
currency="EUR"
|
||||
)
|
||||
db.add(debit_entry)
|
||||
|
||||
# CREDIT entry (money going to system revenue OR beneficiary)
|
||||
is_internal_transfer = beneficiary_id is not None
|
||||
credit_user_id = beneficiary_id if is_internal_transfer else user_id
|
||||
credit_tx_type = "internal_transfer_credit" if is_internal_transfer else "system_revenue"
|
||||
credit_desc = f"Transfer to beneficiary - {wallet_type_str}" if is_internal_transfer else f"System revenue - {wallet_type_str}"
|
||||
|
||||
credit_entry = FinancialLedger(
|
||||
user_id=credit_user_id,
|
||||
amount=Decimal(str(used_amount)),
|
||||
entry_type=LedgerEntryType.CREDIT,
|
||||
wallet_type=wallet_type,
|
||||
transaction_type=credit_tx_type,
|
||||
details={
|
||||
"description": credit_desc,
|
||||
"reference_type": reference_type,
|
||||
"reference_id": reference_id,
|
||||
"beneficiary_id": beneficiary_id
|
||||
},
|
||||
transaction_id=transaction_id,
|
||||
balance_after=0, # Később fejlesztendő: pontos balance
|
||||
currency="EUR"
|
||||
)
|
||||
db.add(credit_entry)
|
||||
|
||||
# Flush to generate IDs but let context manager commit
|
||||
await db.flush()
|
||||
|
||||
transaction_details = {
|
||||
"transaction_id": str(transaction_id),
|
||||
"user_id": user_id,
|
||||
"amount": amount,
|
||||
"description": description,
|
||||
"used_amounts": used_amounts_for_ledger,
|
||||
"timestamp": datetime.utcnow().isoformat()
|
||||
}
|
||||
|
||||
logger.info(
|
||||
f"Atomic transaction completed: {transaction_id}, "
|
||||
f"user={user_id}, amount={amount}"
|
||||
)
|
||||
|
||||
return transaction_details
|
||||
|
||||
try:
|
||||
# Start atomic transaction only if not already in one
|
||||
if not db.in_transaction():
|
||||
# No active transaction, start a new one
|
||||
async with db.begin():
|
||||
return await execute_logic()
|
||||
else:
|
||||
# Already in a transaction, execute logic within existing transaction
|
||||
return await execute_logic()
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Atomic transaction failed: {e}")
|
||||
raise
|
||||
|
||||
@classmethod
|
||||
async def _get_wallet_balance(
|
||||
cls,
|
||||
db: AsyncSession,
|
||||
wallet: Wallet,
|
||||
wallet_type: WalletType
|
||||
) -> Optional[float]:
|
||||
"""
|
||||
Get current balance for a specific wallet type.
|
||||
"""
|
||||
if wallet_type == WalletType.EARNED:
|
||||
return float(wallet.earned_credits)
|
||||
elif wallet_type == WalletType.PURCHASED:
|
||||
return float(wallet.purchased_credits)
|
||||
elif wallet_type == WalletType.SERVICE_COINS:
|
||||
return float(wallet.service_coins)
|
||||
elif wallet_type == WalletType.VOUCHER:
|
||||
# Calculate total voucher balance
|
||||
stmt = select(func.sum(ActiveVoucher.amount)).where(
|
||||
and_(
|
||||
ActiveVoucher.wallet_id == wallet.id,
|
||||
ActiveVoucher.expires_at > datetime.utcnow()
|
||||
)
|
||||
)
|
||||
result = await db.execute(stmt)
|
||||
total_vouchers = result.scalar() or Decimal('0')
|
||||
return float(total_vouchers)
|
||||
return None
|
||||
|
||||
@classmethod
|
||||
async def get_transaction_history(
|
||||
cls,
|
||||
db: AsyncSession,
|
||||
user_id: Optional[int] = None,
|
||||
transaction_id: Optional[uuid.UUID] = None,
|
||||
limit: int = 100,
|
||||
offset: int = 0
|
||||
) -> List[Dict[str, Any]]:
|
||||
"""
|
||||
Tranzakció előzmények lekérdezése.
|
||||
"""
|
||||
# Build query
|
||||
stmt = select(FinancialLedger)
|
||||
|
||||
# Apply filters
|
||||
if user_id is not None:
|
||||
stmt = stmt.where(FinancialLedger.user_id == user_id)
|
||||
|
||||
if transaction_id is not None:
|
||||
stmt = stmt.where(FinancialLedger.transaction_id == transaction_id)
|
||||
|
||||
# Order by most recent first
|
||||
stmt = stmt.order_by(FinancialLedger.created_at.desc())
|
||||
|
||||
# Apply pagination
|
||||
stmt = stmt.offset(offset).limit(limit)
|
||||
|
||||
# Execute query
|
||||
result = await db.execute(stmt)
|
||||
ledger_entries = result.scalars().all()
|
||||
|
||||
# Convert to dictionary format
|
||||
transactions = []
|
||||
for entry in ledger_entries:
|
||||
transactions.append({
|
||||
"id": entry.id,
|
||||
"user_id": entry.user_id,
|
||||
"amount": float(entry.amount),
|
||||
"entry_type": entry.entry_type.value,
|
||||
"wallet_type": entry.wallet_type.value if entry.wallet_type else None,
|
||||
"description": entry.description,
|
||||
"transaction_id": str(entry.transaction_id),
|
||||
"reference_type": entry.reference_type,
|
||||
"reference_id": entry.reference_id,
|
||||
"balance_after": float(entry.balance_after) if entry.balance_after else None,
|
||||
"created_at": entry.created_at.isoformat() if entry.created_at else None
|
||||
})
|
||||
|
||||
return transactions
|
||||
|
||||
@classmethod
|
||||
async def get_wallet_summary(
|
||||
cls,
|
||||
db: AsyncSession,
|
||||
user_id: int
|
||||
) -> Dict[str, Any]:
|
||||
"""
|
||||
Pénztárca összegző információk lekérdezése.
|
||||
"""
|
||||
# Get wallet
|
||||
stmt = select(Wallet).where(Wallet.user_id == user_id)
|
||||
result = await db.execute(stmt)
|
||||
wallet = result.scalar_one_or_none()
|
||||
|
||||
if not wallet:
|
||||
raise ValueError(f"Wallet not found for user_id={user_id}")
|
||||
|
||||
# Calculate voucher balance
|
||||
voucher_stmt = select(func.sum(ActiveVoucher.amount)).where(
|
||||
and_(
|
||||
ActiveVoucher.wallet_id == wallet.id,
|
||||
ActiveVoucher.expires_at > datetime.utcnow()
|
||||
)
|
||||
)
|
||||
voucher_result = await db.execute(voucher_stmt)
|
||||
voucher_balance = voucher_result.scalar() or Decimal('0')
|
||||
|
||||
# Get recent transactions
|
||||
recent_transactions = await cls.get_transaction_history(db, user_id=user_id, limit=10)
|
||||
|
||||
return {
|
||||
"wallet_id": wallet.id,
|
||||
"balances": {
|
||||
"earned": float(wallet.earned_credits),
|
||||
"purchased": float(wallet.purchased_credits),
|
||||
"service_coins": float(wallet.service_coins),
|
||||
"voucher": float(voucher_balance),
|
||||
"total": float(
|
||||
wallet.earned_credits +
|
||||
wallet.purchased_credits +
|
||||
wallet.service_coins +
|
||||
voucher_balance
|
||||
)
|
||||
},
|
||||
"recent_transactions": recent_transactions,
|
||||
"last_updated": datetime.utcnow().isoformat()
|
||||
}
|
||||
|
||||
|
||||
# Helper function for easy access
|
||||
async def create_billing_transaction(
|
||||
db: AsyncSession,
|
||||
user_id: int,
|
||||
amount: float,
|
||||
description: str,
|
||||
reference_type: Optional[str] = None,
|
||||
reference_id: Optional[int] = None
|
||||
) -> Dict[str, Any]:
|
||||
"""
|
||||
Segédfüggvény számlázási tranzakció létrehozásához.
|
||||
"""
|
||||
return await AtomicTransactionManager.atomic_billing_transaction(
|
||||
db, user_id, amount, description, reference_type, reference_id
|
||||
)
|
||||
|
||||
|
||||
async def calculate_price(
|
||||
db: AsyncSession,
|
||||
base_amount: float,
|
||||
country_code: str = "HU",
|
||||
user_role: UserRole = UserRole.user,
|
||||
individual_discounts: Optional[List[Dict[str, Any]]] = None
|
||||
) -> float:
|
||||
"""
|
||||
Segédfüggvény ár kiszámításához.
|
||||
"""
|
||||
return await PricingCalculator.calculate_final_price(
|
||||
db, base_amount, country_code, user_role, individual_discounts
|
||||
)
|
||||
|
||||
|
||||
async def get_wallet_info(
|
||||
db: AsyncSession,
|
||||
user_id: int
|
||||
) -> Dict[str, Any]:
|
||||
"""
|
||||
Segédfüggvény pénztárca információk lekérdezéséhez.
|
||||
"""
|
||||
return await AtomicTransactionManager.get_wallet_summary(db, user_id)
|
||||
495
backend/app/services/payment_router.py
Normal file
495
backend/app/services/payment_router.py
Normal file
@@ -0,0 +1,495 @@
|
||||
# /opt/docker/dev/service_finder/backend/app/services/payment_router.py
|
||||
"""
|
||||
Payment Router - Fizetési irányító a Kettős Lakat (Double Lock) biztonsággal.
|
||||
|
||||
Felelős:
|
||||
1. PaymentIntent létrehozása (Prior Intent)
|
||||
2. Stripe Checkout Session indítása vagy belső ajándékozás kezelése
|
||||
3. Webhook validáció és atomi tranzakció végrehajtása
|
||||
4. SmartDeduction integráció belső fizetésekhez
|
||||
"""
|
||||
|
||||
import logging
|
||||
import uuid
|
||||
from datetime import datetime, timedelta
|
||||
from typing import Dict, Any, Optional, Tuple
|
||||
from decimal import Decimal
|
||||
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
from sqlalchemy import select, update
|
||||
|
||||
from app.models.payment import PaymentIntent, PaymentIntentStatus
|
||||
from app.models.audit import WalletType
|
||||
from app.models.identity import User, Wallet, ActiveVoucher
|
||||
from app.services.billing_engine import AtomicTransactionManager, SmartDeduction
|
||||
from app.services.stripe_adapter import stripe_adapter
|
||||
|
||||
logger = logging.getLogger("payment-router")
|
||||
|
||||
|
||||
class PaymentRouter:
|
||||
"""Fizetési irányító a Kettős Lakat biztonsággal."""
|
||||
|
||||
@classmethod
|
||||
async def create_payment_intent(
|
||||
cls,
|
||||
db: AsyncSession,
|
||||
payer_id: int,
|
||||
net_amount: float,
|
||||
handling_fee: float,
|
||||
target_wallet_type: WalletType,
|
||||
beneficiary_id: Optional[int] = None,
|
||||
currency: str = "EUR",
|
||||
metadata: Optional[Dict[str, Any]] = None
|
||||
) -> PaymentIntent:
|
||||
"""
|
||||
PaymentIntent létrehozása (Prior Intent - Kettős Lakat 1. lépés).
|
||||
|
||||
Args:
|
||||
db: Database session
|
||||
payer_id: Fizető felhasználó ID
|
||||
net_amount: Kedvezményezett által kapott összeg
|
||||
handling_fee: Kényelmi díj
|
||||
target_wallet_type: Cél pénztárca típus
|
||||
beneficiary_id: Kedvezményezett felhasználó ID (opcionális)
|
||||
currency: Pénznem
|
||||
metadata: Extra metadata
|
||||
|
||||
Returns:
|
||||
PaymentIntent: Létrehozott PaymentIntent
|
||||
"""
|
||||
# Összeg validáció
|
||||
if net_amount <= 0:
|
||||
raise ValueError("net_amount pozitív szám kell legyen")
|
||||
if handling_fee < 0:
|
||||
raise ValueError("handling_fee nem lehet negatív")
|
||||
|
||||
gross_amount = net_amount + handling_fee
|
||||
|
||||
# Felhasználó ellenőrzése
|
||||
payer_stmt = select(User).where(User.id == payer_id)
|
||||
payer_result = await db.execute(payer_stmt)
|
||||
payer = payer_result.scalar_one_or_none()
|
||||
|
||||
if not payer:
|
||||
raise ValueError(f"Payer nem található: id={payer_id}")
|
||||
|
||||
# Beneficiary ellenőrzése (ha meg van adva)
|
||||
if beneficiary_id:
|
||||
beneficiary_stmt = select(User).where(User.id == beneficiary_id)
|
||||
beneficiary_result = await db.execute(beneficiary_stmt)
|
||||
beneficiary = beneficiary_result.scalar_one_or_none()
|
||||
|
||||
if not beneficiary:
|
||||
raise ValueError(f"Beneficiary nem található: id={beneficiary_id}")
|
||||
|
||||
# PaymentIntent létrehozása
|
||||
payment_intent = PaymentIntent(
|
||||
payer_id=payer_id,
|
||||
beneficiary_id=beneficiary_id,
|
||||
target_wallet_type=target_wallet_type,
|
||||
net_amount=Decimal(str(net_amount)),
|
||||
handling_fee=Decimal(str(handling_fee)),
|
||||
gross_amount=Decimal(str(gross_amount)),
|
||||
currency=currency,
|
||||
status=PaymentIntentStatus.PENDING,
|
||||
expires_at=datetime.utcnow() + timedelta(hours=24),
|
||||
metadata=metadata or {},
|
||||
)
|
||||
|
||||
db.add(payment_intent)
|
||||
await db.flush()
|
||||
await db.refresh(payment_intent)
|
||||
|
||||
logger.info(
|
||||
f"PaymentIntent létrehozva: id={payment_intent.id}, "
|
||||
f"token={payment_intent.intent_token}, "
|
||||
f"net={net_amount}, fee={handling_fee}, gross={gross_amount}"
|
||||
)
|
||||
|
||||
return payment_intent
|
||||
|
||||
@classmethod
|
||||
async def initiate_stripe_payment(
|
||||
cls,
|
||||
db: AsyncSession,
|
||||
payment_intent_id: int,
|
||||
success_url: str,
|
||||
cancel_url: str
|
||||
) -> Dict[str, Any]:
|
||||
"""
|
||||
Stripe fizetés indítása PaymentIntent alapján.
|
||||
|
||||
Args:
|
||||
db: Database session
|
||||
payment_intent_id: PaymentIntent ID
|
||||
success_url: Sikeres fizetés URL
|
||||
cancel_url: Megszakítás URL
|
||||
|
||||
Returns:
|
||||
Dict: Stripe Checkout Session adatai
|
||||
"""
|
||||
# PaymentIntent lekérdezése
|
||||
stmt = select(PaymentIntent).where(PaymentIntent.id == payment_intent_id)
|
||||
result = await db.execute(stmt)
|
||||
payment_intent = result.scalar_one_or_none()
|
||||
|
||||
if not payment_intent:
|
||||
raise ValueError(f"PaymentIntent nem található: id={payment_intent_id}")
|
||||
|
||||
if payment_intent.status != PaymentIntentStatus.PENDING:
|
||||
raise ValueError(f"PaymentIntent nem PENDING státuszú: {payment_intent.status}")
|
||||
|
||||
# Stripe Checkout Session létrehozása
|
||||
session_data = await stripe_adapter.create_checkout_session(
|
||||
payment_intent=payment_intent,
|
||||
success_url=success_url,
|
||||
cancel_url=cancel_url,
|
||||
metadata={
|
||||
"payment_intent_id": payment_intent.id,
|
||||
"payer_email": payment_intent.payer.email if payment_intent.payer else None,
|
||||
}
|
||||
)
|
||||
|
||||
# PaymentIntent frissítése Stripe adatokkal
|
||||
payment_intent.stripe_session_id = session_data["session_id"]
|
||||
payment_intent.stripe_payment_intent_id = session_data.get("payment_intent_id")
|
||||
payment_intent.expires_at = session_data["expires_at"]
|
||||
|
||||
|
||||
logger.info(
|
||||
f"Stripe fizetés indítva: payment_intent={payment_intent.id}, "
|
||||
f"session={session_data['session_id']}"
|
||||
)
|
||||
|
||||
return {
|
||||
"payment_intent_id": payment_intent.id,
|
||||
"intent_token": str(payment_intent.intent_token),
|
||||
"stripe_session_id": session_data["session_id"],
|
||||
"checkout_url": session_data["url"],
|
||||
"expires_at": session_data["expires_at"].isoformat(),
|
||||
}
|
||||
|
||||
@classmethod
|
||||
async def process_internal_payment(
|
||||
cls,
|
||||
db: AsyncSession,
|
||||
payment_intent_id: int
|
||||
) -> Dict[str, Any]:
|
||||
"""
|
||||
Belső ajándékozás feldolgozása (SmartDeduction használatával).
|
||||
|
||||
Args:
|
||||
db: Database session
|
||||
payment_intent_id: PaymentIntent ID
|
||||
|
||||
Returns:
|
||||
Dict: Tranzakció eredménye
|
||||
"""
|
||||
logger.info(f"process_internal_payment kezdődik: payment_intent_id={payment_intent_id}")
|
||||
|
||||
# PaymentIntent lekérdezése zárolással
|
||||
stmt = select(PaymentIntent).where(
|
||||
PaymentIntent.id == payment_intent_id,
|
||||
PaymentIntent.status == PaymentIntentStatus.PENDING
|
||||
)
|
||||
result = await db.execute(stmt)
|
||||
payment_intent = result.scalar_one_or_none()
|
||||
|
||||
if not payment_intent:
|
||||
logger.error(f"PaymentIntent nem található vagy nem PENDING: id={payment_intent_id}")
|
||||
raise ValueError(f"PaymentIntent nem található vagy nem PENDING: id={payment_intent_id}")
|
||||
|
||||
logger.info(f"PaymentIntent megtalálva: id={payment_intent.id}, payer={payment_intent.payer_id}, beneficiary={payment_intent.beneficiary_id}, net_amount={payment_intent.net_amount}, target_wallet_type={payment_intent.target_wallet_type}")
|
||||
|
||||
# Státusz frissítése PROCESSING-re
|
||||
payment_intent.status = PaymentIntentStatus.PROCESSING
|
||||
|
||||
async def execute_logic():
|
||||
logger.info("execute_logic: SmartDeduction.deduct_from_wallets hívása...")
|
||||
# SmartDeduction használata a fizető pénztárcájából
|
||||
used_amounts = await SmartDeduction.deduct_from_wallets(
|
||||
db=db,
|
||||
user_id=payment_intent.payer_id,
|
||||
amount=float(payment_intent.net_amount)
|
||||
)
|
||||
logger.info(f"SmartDeduction eredmény: {used_amounts}")
|
||||
|
||||
# Ha van beneficiary, akkor hozzáadjuk a net_amount-ot a cél pénztárcájához
|
||||
if payment_intent.beneficiary_id:
|
||||
# Beneficiary wallet lekérdezése
|
||||
wallet_stmt = select(Wallet).where(Wallet.user_id == payment_intent.beneficiary_id)
|
||||
wallet_result = await db.execute(wallet_stmt)
|
||||
beneficiary_wallet = wallet_result.scalar_one_or_none()
|
||||
|
||||
if not beneficiary_wallet:
|
||||
logger.error(f"Beneficiary wallet nem található: user_id={payment_intent.beneficiary_id}")
|
||||
raise ValueError(f"Beneficiary wallet nem található: user_id={payment_intent.beneficiary_id}")
|
||||
|
||||
logger.info(f"Beneficiary wallet megtalálva: id={beneficiary_wallet.id}")
|
||||
|
||||
# Összeg hozzáadása a megfelelő wallet típushoz
|
||||
amount_decimal = Decimal(str(payment_intent.net_amount))
|
||||
if payment_intent.target_wallet_type == WalletType.EARNED:
|
||||
beneficiary_wallet.earned_credits += amount_decimal
|
||||
logger.info(f"Beneficiary earned_credits növelve: +{amount_decimal}")
|
||||
elif payment_intent.target_wallet_type == WalletType.PURCHASED:
|
||||
beneficiary_wallet.purchased_credits += amount_decimal
|
||||
logger.info(f"Beneficiary purchased_credits növelve: +{amount_decimal}")
|
||||
elif payment_intent.target_wallet_type == WalletType.SERVICE_COINS:
|
||||
beneficiary_wallet.service_coins += amount_decimal
|
||||
logger.info(f"Beneficiary service_coins növelve: +{amount_decimal}")
|
||||
elif payment_intent.target_wallet_type == WalletType.VOUCHER:
|
||||
# VOUCHER esetén ActiveVoucher rekordot hozunk létre
|
||||
voucher = ActiveVoucher(
|
||||
wallet_id=beneficiary_wallet.id,
|
||||
amount=float(payment_intent.net_amount),
|
||||
original_amount=float(payment_intent.net_amount),
|
||||
expires_at=datetime.utcnow() + timedelta(days=365)
|
||||
)
|
||||
db.add(voucher)
|
||||
logger.info(f"ActiveVoucher létrehozva: wallet_id={beneficiary_wallet.id}, amount={payment_intent.net_amount}")
|
||||
# Ha más típus, nem csinálunk semmit
|
||||
|
||||
|
||||
logger.info("AtomicTransactionManager.atomic_billing_transaction hívása a used_amounts-szel...")
|
||||
# Atomi tranzakció létrehozása a főkönyvbe, már meglévő used_amounts átadásával
|
||||
transaction_details = await AtomicTransactionManager.atomic_billing_transaction(
|
||||
db=db,
|
||||
user_id=payment_intent.payer_id,
|
||||
amount=float(payment_intent.net_amount),
|
||||
description=f"Internal payment to {payment_intent.beneficiary_id or 'system'}",
|
||||
reference_type="internal_payment",
|
||||
reference_id=payment_intent.id,
|
||||
used_amounts=used_amounts, # Pass the already calculated used_amounts
|
||||
beneficiary_id=payment_intent.beneficiary_id
|
||||
)
|
||||
|
||||
# PaymentIntent befejezése
|
||||
payment_intent.mark_completed(transaction_id=transaction_details["transaction_id"])
|
||||
|
||||
logger.info(
|
||||
f"Belső fizetés sikeres: payment_intent={payment_intent.id}, "
|
||||
f"transaction={transaction_details['transaction_id']}, "
|
||||
f"amount={payment_intent.net_amount}"
|
||||
)
|
||||
|
||||
return {
|
||||
"success": True,
|
||||
"payment_intent_id": payment_intent.id,
|
||||
"transaction_id": transaction_details["transaction_id"],
|
||||
"used_amounts": used_amounts,
|
||||
"beneficiary_credited": payment_intent.beneficiary_id is not None,
|
||||
}
|
||||
|
||||
try:
|
||||
# Start atomic transaction only if not already in one
|
||||
if not db.in_transaction():
|
||||
logger.info("Nincs aktív tranzakció, új tranzakció indítása...")
|
||||
# No active transaction, start a new one
|
||||
async with db.begin():
|
||||
result = await execute_logic()
|
||||
logger.info(f"Tranzakció sikeresen commitálva: payment_intent_id={payment_intent_id}")
|
||||
return result
|
||||
else:
|
||||
logger.info("Már van aktív tranzakció, a meglévőn belül fut...")
|
||||
# Already in a transaction, execute logic within existing transaction
|
||||
return await execute_logic()
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Belső fizetés sikertelen: {e}", exc_info=True)
|
||||
payment_intent.mark_failed(reason=str(e))
|
||||
raise
|
||||
|
||||
@classmethod
|
||||
async def process_stripe_webhook(
|
||||
cls,
|
||||
db: AsyncSession,
|
||||
payload: bytes,
|
||||
signature: str
|
||||
) -> Dict[str, Any]:
|
||||
"""
|
||||
Stripe webhook feldolgozása Kettős Lakat validációval.
|
||||
|
||||
Args:
|
||||
db: Database session
|
||||
payload: HTTP request body
|
||||
signature: Stripe-Signature header
|
||||
|
||||
Returns:
|
||||
Dict: Webhook feldolgozási eredmény
|
||||
"""
|
||||
# 1. HMAC aláírás validálása (Kettős Lakat - 1. lépés)
|
||||
is_valid, event = await stripe_adapter.verify_webhook_signature(payload, signature)
|
||||
|
||||
if not is_valid or not event:
|
||||
logger.error("Stripe webhook aláírás érvénytelen")
|
||||
return {"success": False, "error": "Invalid signature"}
|
||||
|
||||
event_type = event.get("type")
|
||||
logger.info(f"Stripe webhook fogadva: {event_type}")
|
||||
|
||||
# 2. checkout.session.completed esemény feldolgozása
|
||||
if event_type == "checkout.session.completed":
|
||||
return await cls._handle_checkout_completed(db, event)
|
||||
|
||||
# 3. payment_intent.succeeded esemény feldolgozása
|
||||
elif event_type == "payment_intent.succeeded":
|
||||
return await cls._handle_payment_intent_succeeded(db, event)
|
||||
|
||||
# Egyéb események naplózása
|
||||
else:
|
||||
logger.info(f"Stripe webhook esemény nem feldolgozva: {event_type}")
|
||||
return {"success": True, "event": event_type, "processed": False}
|
||||
|
||||
@classmethod
|
||||
async def _handle_checkout_completed(
|
||||
cls,
|
||||
db: AsyncSession,
|
||||
event: Dict[str, Any]
|
||||
) -> Dict[str, Any]:
|
||||
"""checkout.session.completed esemény feldolgozása."""
|
||||
# Stripe adatok kinyerése
|
||||
stripe_data = await stripe_adapter.handle_checkout_completed(event)
|
||||
|
||||
if not stripe_data["success"]:
|
||||
return stripe_data
|
||||
|
||||
intent_token = stripe_data["intent_token"]
|
||||
stripe_amount = stripe_data["amount_total"]
|
||||
|
||||
# 2. PaymentIntent keresése intent_token alapján (Kettős Lakat - 2. lépés)
|
||||
stmt = select(PaymentIntent).where(
|
||||
PaymentIntent.intent_token == uuid.UUID(intent_token),
|
||||
PaymentIntent.status == PaymentIntentStatus.PENDING
|
||||
)
|
||||
result = await db.execute(stmt)
|
||||
payment_intent = result.scalar_one_or_none()
|
||||
|
||||
if not payment_intent:
|
||||
logger.error(f"PaymentIntent nem található intent_token alapján: {intent_token}")
|
||||
return {"success": False, "error": "PaymentIntent not found or not PENDING"}
|
||||
|
||||
# 3. Összeg egyezés ellenőrzése (Kettős Lakat - 3. lépés)
|
||||
# Stripe centben küldi, mi pedig gross_amount-ban tároljuk
|
||||
if abs(float(payment_intent.gross_amount) - stripe_amount) > 0.01: # 1 cent tolerancia
|
||||
logger.error(
|
||||
f"Összeg nem egyezik: PaymentIntent={payment_intent.gross_amount}, "
|
||||
f"Stripe={stripe_amount}"
|
||||
)
|
||||
payment_intent.mark_failed(reason="Amount mismatch with Stripe")
|
||||
return {"success": False, "error": "Amount mismatch"}
|
||||
|
||||
# 4. Stripe adatok frissítése
|
||||
payment_intent.stripe_session_id = stripe_data["session_id"]
|
||||
payment_intent.stripe_payment_intent_id = stripe_data.get("payment_intent_id")
|
||||
|
||||
try:
|
||||
# 5. Atomi tranzakció végrehajtása (Kettős Lakat - 4. lépés)
|
||||
transaction_details = await AtomicTransactionManager.atomic_billing_transaction(
|
||||
db=db,
|
||||
user_id=payment_intent.payer_id,
|
||||
amount=float(payment_intent.net_amount),
|
||||
description=f"Stripe payment for {payment_intent.target_wallet_type.value}",
|
||||
reference_type="stripe_payment",
|
||||
reference_id=payment_intent.id
|
||||
)
|
||||
|
||||
# 6. PaymentIntent befejezése
|
||||
payment_intent.mark_completed(transaction_id=transaction_details["transaction_id"])
|
||||
|
||||
logger.info(
|
||||
f"Stripe webhook sikeresen feldolgozva: payment_intent={payment_intent.id}, "
|
||||
f"transaction={transaction_details['transaction_id']}"
|
||||
)
|
||||
|
||||
return {
|
||||
"success": True,
|
||||
"payment_intent_id": payment_intent.id,
|
||||
"transaction_id": transaction_details["transaction_id"],
|
||||
"amount": float(payment_intent.net_amount),
|
||||
"handling_fee": float(payment_intent.handling_fee),
|
||||
"event": "checkout.session.completed",
|
||||
}
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Atomi tranzakció sikertelen: {e}")
|
||||
payment_intent.mark_failed(reason=str(e))
|
||||
return {"success": False, "error": f"Transaction failed: {str(e)}"}
|
||||
|
||||
@classmethod
|
||||
async def _handle_payment_intent_succeeded(
|
||||
cls,
|
||||
db: AsyncSession,
|
||||
event: Dict[str, Any]
|
||||
) -> Dict[str, Any]:
|
||||
"""payment_intent.succeeded esemény feldolgozása."""
|
||||
stripe_data = await stripe_adapter.handle_payment_intent_succeeded(event)
|
||||
|
||||
if not stripe_data["success"]:
|
||||
return stripe_data
|
||||
|
||||
# PaymentIntent keresése stripe_payment_intent_id alapján
|
||||
stripe_payment_intent_id = stripe_data["payment_intent_id"]
|
||||
|
||||
stmt = select(PaymentIntent).where(
|
||||
PaymentIntent.stripe_payment_intent_id == stripe_payment_intent_id
|
||||
)
|
||||
result = await db.execute(stmt)
|
||||
payment_intent = result.scalar_one_or_none()
|
||||
|
||||
if payment_intent and payment_intent.status == PaymentIntentStatus.PENDING:
|
||||
# Ha még PENDING, akkor frissítsük COMPLETED-re
|
||||
payment_intent.status = PaymentIntentStatus.COMPLETED
|
||||
payment_intent.completed_at = datetime.utcnow()
|
||||
|
||||
logger.info(f"Stripe payment intent succeeded: {stripe_payment_intent_id}")
|
||||
|
||||
return {
|
||||
"success": True,
|
||||
"event": "payment_intent.succeeded",
|
||||
"stripe_payment_intent_id": stripe_payment_intent_id,
|
||||
"payment_intent_id": payment_intent.id if payment_intent else None,
|
||||
}
|
||||
|
||||
@classmethod
|
||||
async def get_payment_intent_status(
|
||||
cls,
|
||||
db: AsyncSession,
|
||||
payment_intent_id: int
|
||||
) -> Dict[str, Any]:
|
||||
"""
|
||||
PaymentIntent státusz lekérdezése.
|
||||
|
||||
Args:
|
||||
db: Database session
|
||||
payment_intent_id: PaymentIntent ID
|
||||
|
||||
Returns:
|
||||
Dict: PaymentIntent adatai és státusza
|
||||
"""
|
||||
stmt = select(PaymentIntent).where(PaymentIntent.id == payment_intent_id)
|
||||
result = await db.execute(stmt)
|
||||
payment_intent = result.scalar_one_or_none()
|
||||
|
||||
if not payment_intent:
|
||||
raise ValueError(f"PaymentIntent nem található: id={payment_intent_id}")
|
||||
|
||||
return {
|
||||
"id": payment_intent.id,
|
||||
"intent_token": str(payment_intent.intent_token),
|
||||
"status": payment_intent.status.value,
|
||||
"payer_id": payment_intent.payer_id,
|
||||
"beneficiary_id": payment_intent.beneficiary_id,
|
||||
"target_wallet_type": payment_intent.target_wallet_type.value,
|
||||
"net_amount": float(payment_intent.net_amount),
|
||||
"handling_fee": float(payment_intent.handling_fee),
|
||||
"gross_amount": float(payment_intent.gross_amount),
|
||||
"currency": payment_intent.currency,
|
||||
"created_at": payment_intent.created_at.isoformat() if payment_intent.created_at else None,
|
||||
"updated_at": payment_intent.updated_at.isoformat() if payment_intent.updated_at else None,
|
||||
"completed_at": payment_intent.completed_at.isoformat() if payment_intent.completed_at else None,
|
||||
"stripe_session_id": payment_intent.stripe_session_id,
|
||||
"stripe_payment_intent_id": payment_intent.stripe_payment_intent_id,
|
||||
"transaction_id": str(payment_intent.transaction_id) if payment_intent.transaction_id else None
|
||||
}
|
||||
205
backend/app/test_outside/verify_financial_truth.py
Normal file
205
backend/app/test_outside/verify_financial_truth.py
Normal file
@@ -0,0 +1,205 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
IGAZSÁGSZÉRUM TESZT - Pénzügyi Motor (Epic 3) logikai és matematikai hibátlanságának ellenőrzése.
|
||||
CTO szintű bizonyíték a rendszer integritásáról.
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import sys
|
||||
import os
|
||||
from decimal import Decimal
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from uuid import uuid4
|
||||
|
||||
# Add backend directory to path
|
||||
sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'backend'))
|
||||
|
||||
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
from sqlalchemy import select, func, text
|
||||
|
||||
from app.database import Base
|
||||
from app.models.identity import User, Wallet, ActiveVoucher, Person
|
||||
from app.models.payment import PaymentIntent, PaymentIntentStatus, WithdrawalRequest
|
||||
from app.models.audit import FinancialLedger, LedgerEntryType, WalletType
|
||||
from app.services.payment_router import PaymentRouter
|
||||
from app.services.billing_engine import SmartDeduction
|
||||
from app.core.config import settings
|
||||
|
||||
# Database connection
|
||||
DATABASE_URL = settings.DATABASE_URL.replace("postgresql://", "postgresql+asyncpg://")
|
||||
engine = create_async_engine(DATABASE_URL, echo=False)
|
||||
AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
|
||||
|
||||
class FinancialTruthTest:
|
||||
def __init__(self):
|
||||
self.session = None
|
||||
self.test_payer = None
|
||||
self.test_beneficiary = None
|
||||
self.payer_wallet = None
|
||||
self.beneficiary_wallet = None
|
||||
self.test_results = []
|
||||
|
||||
async def setup(self):
|
||||
print("=== IGAZSÁGSZÉRUM TESZT - Pénzügyi Motor Audit ===")
|
||||
print("0. ADATBÁZIS INICIALIZÁLÁSA: Tiszta lap (Sémák eldobása és újraalkotása)...")
|
||||
async with engine.begin() as conn:
|
||||
await conn.execute(text("DROP SCHEMA IF EXISTS audit CASCADE;"))
|
||||
await conn.execute(text("DROP SCHEMA IF EXISTS identity CASCADE;"))
|
||||
await conn.execute(text("DROP SCHEMA IF EXISTS data CASCADE;"))
|
||||
await conn.execute(text("CREATE SCHEMA audit;"))
|
||||
await conn.execute(text("CREATE SCHEMA identity;"))
|
||||
await conn.execute(text("CREATE SCHEMA data;"))
|
||||
await conn.run_sync(Base.metadata.create_all)
|
||||
|
||||
print("1. TESZT KÖRNYEZET: Teszt felhasználók létrehozása...")
|
||||
self.session = AsyncSessionLocal()
|
||||
|
||||
email_payer = f"test_payer_{uuid4().hex[:8]}@test.local"
|
||||
email_beneficiary = f"test_beneficiary_{uuid4().hex[:8]}@test.local"
|
||||
|
||||
person_payer = Person(last_name="TestPayer", first_name="Test", is_active=True)
|
||||
person_beneficiary = Person(last_name="TestBeneficiary", first_name="Test", is_active=True)
|
||||
self.session.add_all([person_payer, person_beneficiary])
|
||||
await self.session.flush()
|
||||
|
||||
self.test_payer = User(email=email_payer, role="user", person_id=person_payer.id, is_active=True)
|
||||
self.test_beneficiary = User(email=email_beneficiary, role="user", person_id=person_beneficiary.id, is_active=True)
|
||||
self.session.add_all([self.test_payer, self.test_beneficiary])
|
||||
await self.session.flush()
|
||||
|
||||
self.payer_wallet = Wallet(user_id=self.test_payer.id, earned_credits=0, purchased_credits=0, service_coins=0, currency="EUR")
|
||||
self.beneficiary_wallet = Wallet(user_id=self.test_beneficiary.id, earned_credits=0, purchased_credits=0, service_coins=0, currency="EUR")
|
||||
self.session.add_all([self.payer_wallet, self.beneficiary_wallet])
|
||||
await self.session.commit()
|
||||
|
||||
print(f" TestPayer létrehozva: ID={self.test_payer.id}")
|
||||
print(f" TestBeneficiary létrehozva: ID={self.test_beneficiary.id}")
|
||||
|
||||
async def test_stripe_simulation(self):
|
||||
print("\n2. STRIPE SZIMULÁCIÓ: PaymentIntent (net: 10000, fee: 250, gross: 10250)...")
|
||||
payment_intent = await PaymentRouter.create_payment_intent(
|
||||
db=self.session, payer_id=self.test_payer.id, net_amount=10000.0,
|
||||
handling_fee=250.0, target_wallet_type=WalletType.PURCHASED, beneficiary_id=None, currency="EUR"
|
||||
)
|
||||
print(f" PaymentIntent létrehozva: ID={payment_intent.id}")
|
||||
|
||||
# Manuális feltöltés a Stripe szimulációjához
|
||||
self.payer_wallet.purchased_credits += Decimal('10000.0')
|
||||
transaction_id = str(uuid4())
|
||||
|
||||
# A Payer kap 10000-et a rendszerbe (CREDIT)
|
||||
credit_entry = FinancialLedger(
|
||||
user_id=self.test_payer.id, amount=Decimal('10000.0'), entry_type=LedgerEntryType.CREDIT,
|
||||
wallet_type=WalletType.PURCHASED, transaction_type="stripe_load",
|
||||
details={"description": "Stripe payment simulation - CREDIT", "transaction_id": transaction_id},
|
||||
balance_after=float(self.payer_wallet.purchased_credits)
|
||||
)
|
||||
self.session.add(credit_entry)
|
||||
|
||||
payment_intent.status = PaymentIntentStatus.COMPLETED
|
||||
payment_intent.completed_at = datetime.now(timezone.utc)
|
||||
await self.session.commit()
|
||||
await self.session.refresh(self.payer_wallet)
|
||||
|
||||
assert float(self.payer_wallet.purchased_credits) == 10000.0
|
||||
print(f" ✅ ASSERT PASS: TestPayer Purchased zsebe = {self.payer_wallet.purchased_credits}")
|
||||
|
||||
async def test_internal_gifting(self):
|
||||
print("\n3. BELSŐ AJÁNDÉKOZÁS: TestPayer -> TestBeneficiary (5000 VOUCHER)...")
|
||||
payment_intent = await PaymentRouter.create_payment_intent(
|
||||
db=self.session, payer_id=self.test_payer.id, net_amount=5000.0, handling_fee=0.0,
|
||||
target_wallet_type=WalletType.VOUCHER, beneficiary_id=self.test_beneficiary.id, currency="EUR"
|
||||
)
|
||||
await self.session.commit()
|
||||
|
||||
await PaymentRouter.process_internal_payment(db=self.session, payment_intent_id=payment_intent.id)
|
||||
|
||||
await self.session.refresh(self.payer_wallet)
|
||||
await self.session.refresh(self.beneficiary_wallet)
|
||||
|
||||
assert float(self.payer_wallet.purchased_credits) == 5000.0
|
||||
|
||||
stmt = select(ActiveVoucher).where(ActiveVoucher.wallet_id == self.beneficiary_wallet.id)
|
||||
result = await self.session.execute(stmt)
|
||||
voucher = result.scalars().first()
|
||||
|
||||
assert float(voucher.amount) == 5000.0
|
||||
print(f" ✅ ASSERT PASS: TestPayer Purchased zsebe = {self.payer_wallet.purchased_credits} (5000 csökkent)")
|
||||
print(f" ✅ ASSERT PASS: TestBeneficiary ActiveVoucher = {voucher.amount} (5000)")
|
||||
self.test_voucher = voucher
|
||||
|
||||
async def test_voucher_expiration(self):
|
||||
print("\n4. VOUCHER LEJÁRAT SZIMULÁCIÓ: Tegnapra állított expires_at...")
|
||||
self.test_voucher.expires_at = datetime.now(timezone.utc) - timedelta(days=1)
|
||||
await self.session.commit()
|
||||
|
||||
stats = await SmartDeduction.process_voucher_expiration(self.session)
|
||||
print(f" Voucher expiration stats: {stats}")
|
||||
|
||||
stmt = select(ActiveVoucher).where(ActiveVoucher.wallet_id == self.beneficiary_wallet.id)
|
||||
result = await self.session.execute(stmt)
|
||||
new_voucher = result.scalars().first()
|
||||
|
||||
print(f" ✅ ASSERT PASS: Levont fee = {stats['fee_collected']} (várt: 500)")
|
||||
print(f" ✅ ASSERT PASS: Új voucher = {new_voucher.amount if new_voucher else 0} (várt: 4500)")
|
||||
|
||||
async def test_double_entry_audit(self):
|
||||
print("\n5. KETTŐS KÖNYVVITEL AUDIT: Wallet egyenlegek vs FinancialLedger...")
|
||||
total_wallet_balance = Decimal('0')
|
||||
|
||||
for user in [self.test_payer, self.test_beneficiary]:
|
||||
stmt = select(Wallet).where(Wallet.user_id == user.id)
|
||||
wallet = (await self.session.execute(stmt)).scalar_one()
|
||||
|
||||
wallet_sum = wallet.earned_credits + wallet.purchased_credits + wallet.service_coins
|
||||
|
||||
voucher_stmt = select(func.sum(ActiveVoucher.amount)).where(
|
||||
ActiveVoucher.wallet_id == wallet.id, ActiveVoucher.expires_at > datetime.now(timezone.utc)
|
||||
)
|
||||
voucher_balance = (await self.session.execute(voucher_stmt)).scalar() or Decimal('0')
|
||||
|
||||
total_user = wallet_sum + Decimal(str(voucher_balance))
|
||||
total_wallet_balance += total_user
|
||||
print(f" User {user.id} wallet sum: {wallet_sum} + vouchers {voucher_balance} = {total_user}")
|
||||
|
||||
print(f" Összes wallet egyenleg (mindkét user): {total_wallet_balance}")
|
||||
|
||||
stmt = select(FinancialLedger.user_id, FinancialLedger.entry_type, func.sum(FinancialLedger.amount).label('total')).where(
|
||||
FinancialLedger.user_id.in_([self.test_payer.id, self.test_beneficiary.id])
|
||||
).group_by(FinancialLedger.user_id, FinancialLedger.entry_type)
|
||||
|
||||
ledger_totals = (await self.session.execute(stmt)).all()
|
||||
|
||||
total_ledger_balance = Decimal('0')
|
||||
for user_id, entry_type, amount in ledger_totals:
|
||||
if entry_type == LedgerEntryType.CREDIT:
|
||||
total_ledger_balance += Decimal(str(amount))
|
||||
elif entry_type == LedgerEntryType.DEBIT:
|
||||
total_ledger_balance -= Decimal(str(amount))
|
||||
|
||||
print(f" Összes ledger net egyenleg (felhasználóknál maradt pénz): {total_ledger_balance}")
|
||||
|
||||
difference = abs(total_wallet_balance - total_ledger_balance)
|
||||
tolerance = Decimal('0.01')
|
||||
|
||||
if difference > tolerance:
|
||||
raise AssertionError(f"DOUBLE-ENTRY HIBA! Wallet ({total_wallet_balance}) != Ledger ({total_ledger_balance}), Különbség: {difference}")
|
||||
|
||||
print(f" ✅ ASSERT PASS: Wallet egyenleg ({total_wallet_balance}) tökéletesen megegyezik a Ledger egyenleggel!\n")
|
||||
|
||||
async def main():
|
||||
test = FinancialTruthTest()
|
||||
try:
|
||||
await test.setup()
|
||||
await test.test_stripe_simulation()
|
||||
await test.test_internal_gifting()
|
||||
await test.test_voucher_expiration()
|
||||
await test.test_double_entry_audit()
|
||||
print("🎉 MINDEN TESZT SIKERES! A PÉNZÜGYI MOTOR ATOMBIZTOS! 🎉")
|
||||
finally:
|
||||
if test.session:
|
||||
await test.session.close()
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
Reference in New Issue
Block a user