""" app/tenant/utils.py Shared helpers used across all tenant portal blueprints. """ import logging from datetime import datetime, timezone from functools import wraps from flask import g, flash, redirect, url_for, request, jsonify from flask_login import current_user logger = logging.getLogger(__name__) def log_tenant_action(action, target_type=None, target_id=None, details=None): """ Write a structured log entry for any tenant create/edit/delete action. Not persisted to audit_log (tenant-level actions use application logs). """ logger.info( "TENANT_ACTION action=%s target_type=%s target_id=%s tenant=%s user=%s details=%s", action, target_type, target_id, getattr(g, 'tenant', None) and g.tenant.id, current_user.get_id() if current_user.is_authenticated else None, details, ) def plan_limit_check(resource: str) -> tuple[bool, str]: """ Check whether the current tenant has headroom to add one more of `resource`. resource: 'staff' | 'location' Returns (allowed: bool, error_message: str | None) """ from app.models.salon import Staff, Location tenant = getattr(g, 'tenant', None) if not tenant or not tenant.plan: return True, None plan = tenant.plan if resource == 'staff': if plan.max_staff is None: return True, None count = Staff.query.filter_by( tenant_id=tenant.id, is_active=True ).filter(Staff.deleted_at.is_(None)).count() if count >= plan.max_staff: return False, ( f"Your plan allows a maximum of {plan.max_staff} active staff members. " "Please upgrade your plan to add more." ) elif resource == 'location': if plan.max_locations is None: return True, None count = Location.query.filter_by( tenant_id=tenant.id, is_active=True ).count() if count >= plan.max_locations: return False, ( f"Your plan allows a maximum of {plan.max_locations} locations. " "Please upgrade your plan to add more." ) return True, None def get_active_promotion(tenant_id: int, item_id: int, item_type: str): """ Promotion engine — resolves the single best active promotion for a service or product line item at checkout time. item_type: 'service' | 'product' Priority: 1. Specific promotion targeting this exact item ID 2. 'all_services' / 'all_products' promotion 3. 'all' promotion (covers both services and products) If multiple promotions match at the same priority level, the highest discount_percent wins. Returns None if no active promotion applies. """ from app.models.salon import Promotion now = datetime.now(timezone.utc) active = Promotion.query.filter_by( tenant_id=tenant_id, is_active=True ).filter( Promotion.starts_at <= now, Promotion.ends_at >= now, ).all() if not active: return None # Collect all candidates candidates = [] for promo in active: at = promo.applies_to if at == 'all': candidates.append((0, promo)) elif at == 'all_services' and item_type == 'service': candidates.append((1, promo)) elif at == 'all_products' and item_type == 'product': candidates.append((1, promo)) elif at == item_type: # Specific IDs — check membership target_ids = promo.target_ids_json or [] if item_id in target_ids: candidates.append((2, promo)) if not candidates: return None # Sort: highest specificity first (2 > 1 > 0), then highest discount candidates.sort(key=lambda x: (x[0], x[1].discount_percent), reverse=True) return candidates[0][1] def apply_promotion_to_price(price, promotion): """ Apply a promotion to a unit price. Returns (discounted_price, discount_percent). If promotion is None, returns original price and 0. """ from decimal import Decimal, ROUND_HALF_UP if promotion is None: return price, 0 pct = promotion.discount_percent price = Decimal(str(price)) discount = (price * Decimal(pct) / Decimal(100)).quantize( Decimal('0.01'), rounding=ROUND_HALF_UP ) return float(price - discount), pct def is_api_request(): return request.path.startswith('/api/') or \ request.accept_mimetypes.best == 'application/json'