05/07 Phase 3: initial codes

app/tenant/utils.py

@@ -0,0 +1,152 @@
+"""
+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'