gitea-admin/gh-christianlouis-docuelevate
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
bdb67de5cbe2fb4832065c2958c41eba708ac8db
gh-christianlouis-docuelevate/app/utils/subscription.py
T
copilot-swe-agent[bot] 9d11d741f4 fix(ui): update plan descriptions to reflect per-user pricing 
- Rewrite taglines and feature lists for all four tiers with
  concrete, use-case-driven copy (freelancers, knowledge workers,
  power users — no team or enterprise framing)
- Rename 'Business' display name to 'Power' (plan_id stays
  'business' for backwards DB compatibility)
- Replace 'enterprise' language in pricing page hero with
  'per person, per month' copy
- Swap fa-building icon for fa-bolt on the Power tier
- Fix support level for Power tier to 'Priority' (was 'Dedicated')
- Update docs/SubscriptionTiers.md with new names, table, and
  intended-use-case section
- Add test_business_tier_display_name_is_power assertion

Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>
2026-03-07 20:12:28 +00:00

505 lines
19 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""
Subscription tier definitions and enforcement utilities for DocuElevate SaaS.
All plans are priced per user per month (or per year with ~20 % discount).
Four tiers (prices ex-VAT; German customers +19 % MwSt):
- free $0/mo — 50 lifetime docs, 150 lifetime OCR pages, 1 dest
- starter $2.99/mo — 50/mo, 300 OCR pp/mo, 2 dests, 1 mailbox
- professional $5.99/mo — 150/mo, 750 OCR pp/mo, 5 dests, 3 mailboxes
- power $7.99/mo — 300/mo, 1500 OCR pp/mo, 10 dests, unlimited mailboxes
Limits use 0 to represent "unlimited".
All paid tiers include a 30-day free trial (trial_days field).
--- Cost analysis at maximum usage (Hetzner Option-A infra, Azure Read + GPT-4o mini) ---
Infrastructure: CX32 (app+Redis €7.59) + CX22 (worker €3.79) + BX21 (storage €7.22) ≈ $24/mo
At 100 users infra share ≈ $0.24/user/mo.
Starter : OCR $0.45 + AI $0.012 + infra $0.24 + Stripe $0.34 = $1.04 → 65 % gross margin
Professional: OCR $1.13 + AI $0.035 + infra $0.24 + Stripe $0.42 = $1.82 → 70 % gross margin
Power : OCR $2.25 + AI $0.069 + infra $0.24 + Stripe $0.48 = $3.04 → 62 % gross margin
After ~30 % German corporate tax: Starter 45 %, Professional 49 %, Power 43 %.
At average usage (~40 % of quota) margins improve to 55-65 % after tax.
⚠ If GPT-4o (not mini) is configured, Power AI cost at max rises to ~$1.92/user,
reducing after-tax margin to ~33 %. Recommend GPT-4o mini as default in production.
"""
from __future__ import annotations
import logging
from datetime import date, datetime, timezone
from typing import Any
from sqlalchemy import func
from sqlalchemy.orm import Session
from app.config import settings
logger = logging.getLogger(__name__)
# ---------------------------------------------------------------------------
# Tier catalogue
# ---------------------------------------------------------------------------
TIER_DEFAULTS: dict[str, dict[str, Any]] = {
"free": {
"id": "free",
"name": "Free",
"tagline": "Try DocuElevate free — no credit card needed",
"price_monthly": 0,
"price_yearly": 0,
"trial_days": 0,
"highlight": False,
# Hard caps — 0 = unlimited
"lifetime_file_limit": 50, # total docs ever processed (enforced at upload)
"daily_upload_limit": 0, # no per-day cap (lifetime cap applies instead)
"monthly_upload_limit": 0, # no per-month cap (lifetime cap applies instead)
"max_storage_destinations": 1,
"max_ocr_pages_monthly": 150, # informational; enforced when OCR quota tracking lands
"max_file_size_mb": 5,
"max_mailboxes": 0, # no email ingestion on free tier
"api_access": False,
# Marketing feature list (shown on pricing page)
"features": [
"50 documents — lifetime total",
"150 OCR pages — lifetime total",
"1 storage destination",
"5 MB max file size",
"Basic AI metadata extraction",
"Community support",
],
"cta": "Get started free",
"badge": None,
},
"starter": {
"id": "starter",
"name": "Starter",
# Use case: freelancer sending ~50 invoices, contracts, or scanned receipts a month
"tagline": "Perfect for freelancers and side-project owners",
"price_monthly": 2.99,
"price_yearly": 28.99, # ≈ 80 % of monthly × 12 — save ~19 % (≈ 2½ months free)
"trial_days": 30,
"highlight": False,
"lifetime_file_limit": 0,
"daily_upload_limit": 0, # no daily cap
"monthly_upload_limit": 50,
"max_storage_destinations": 2,
"max_ocr_pages_monthly": 300,
"max_file_size_mb": 25,
"max_mailboxes": 1,
"api_access": True,
"features": [
"50 documents / month — invoices, contracts, receipts",
"2 storage destinations",
"300 OCR pages / month",
"25 MB max file size",
"Full AI metadata extraction",
"1 email ingestion mailbox",
"API access",
"Email support",
],
"cta": "Start free trial",
"badge": None,
},
"professional": {
"id": "professional",
"name": "Professional",
# Use case: consultant or knowledge worker handling ~150 docs/month across multiple platforms
"tagline": "For knowledge workers managing documents daily",
"price_monthly": 5.99,
"price_yearly": 57.99, # ≈ 80 % of monthly × 12 — save ~19 %
"trial_days": 30,
"highlight": True, # shown as "Most popular"
"lifetime_file_limit": 0,
"daily_upload_limit": 0, # no daily cap
"monthly_upload_limit": 150,
"max_storage_destinations": 5,
"max_ocr_pages_monthly": 750,
"max_file_size_mb": 100,
"max_mailboxes": 3,
"api_access": True,
"features": [
"150 documents / month — reports, contracts, invoices",
"5 storage destinations",
"750 OCR pages / month",
"100 MB max file size",
"Advanced AI workflows",
"3 email ingestion mailboxes",
"Email & URL ingestion",
"Webhooks",
"Priority email support",
],
"cta": "Start free trial",
"badge": "Most Popular",
},
"business": {
"id": "business",
"name": "Power",
# Use case: power user — real estate agent, bookkeeper, or researcher processing ~10 docs/day
"tagline": "For power users with high-volume document workflows",
"price_monthly": 7.99,
"price_yearly": 76.99, # ≈ 80 % of monthly × 12 — save ~20 %
"trial_days": 30,
"highlight": False,
"lifetime_file_limit": 0,
"daily_upload_limit": 0, # no daily cap
"monthly_upload_limit": 300,
"max_storage_destinations": 10,
"max_ocr_pages_monthly": 1500,
"max_file_size_mb": 0, # unlimited file size
"max_mailboxes": 0, # unlimited mailboxes
"api_access": True,
"features": [
"300 documents / month — ~10 documents per day",
"10 storage destinations",
"1,500 OCR pages / month",
"Unlimited file size",
"All AI processing steps",
"Unlimited email ingestion mailboxes",
"All ingestion methods",
"Webhooks & full API access",
"Priority support",
],
"cta": "Start free trial",
"badge": "Best Value",
},
}
# Backward-compatible alias
TIERS = TIER_DEFAULTS
# Display order for the pricing page
TIER_ORDER = ["free", "starter", "professional", "business"]
# Default tier assigned to new users
DEFAULT_TIER = "free"
# ---------------------------------------------------------------------------
# DB → dict conversion
# ---------------------------------------------------------------------------
def _plan_to_dict(plan: Any) -> dict[str, Any]:
"""Convert a SubscriptionPlan ORM object to the same dict shape as TIER_DEFAULTS entries."""
import json
features: list[str] = []
if plan.features:
try:
features = json.loads(plan.features)
except (json.JSONDecodeError, TypeError):
features = []
return {
"id": plan.plan_id,
"name": plan.name,
"tagline": plan.tagline or "",
"price_monthly": plan.price_monthly,
"price_yearly": plan.price_yearly,
"trial_days": plan.trial_days,
"highlight": plan.is_highlighted,
"lifetime_file_limit": plan.lifetime_file_limit,
"daily_upload_limit": plan.daily_upload_limit,
"monthly_upload_limit": plan.monthly_upload_limit,
"max_storage_destinations": plan.max_storage_destinations,
"max_ocr_pages_monthly": plan.max_ocr_pages_monthly,
"max_file_size_mb": plan.max_file_size_mb,
"max_mailboxes": plan.max_mailboxes,
"api_access": plan.api_access,
"features": features,
"cta": plan.cta_text or "Get started",
"badge": plan.badge_text,
"overage_percent": plan.overage_percent,
"allow_overage_billing": plan.allow_overage_billing,
}
# ---------------------------------------------------------------------------
# Getters
# ---------------------------------------------------------------------------
def get_tier(tier_id: str, db: Session | None = None) -> dict[str, Any]:
"""Return plan config dict; DB-first when db is provided, falls back to TIER_DEFAULTS."""
if db is not None:
from app.models import SubscriptionPlan
plan = (
db.query(SubscriptionPlan)
.filter(
SubscriptionPlan.plan_id == tier_id,
SubscriptionPlan.is_active.is_(True),
)
.first()
)
if plan is not None:
return _plan_to_dict(plan)
return TIER_DEFAULTS.get(tier_id, TIER_DEFAULTS["free"])
def get_all_tiers(db: Session | None = None) -> list[dict[str, Any]]:
"""Return plans in display order; DB-first when db is provided."""
if db is not None:
from app.models import SubscriptionPlan
plans = (
db.query(SubscriptionPlan)
.filter(SubscriptionPlan.is_active.is_(True))
.order_by(SubscriptionPlan.sort_order)
.all()
)
if plans:
return [_plan_to_dict(p) for p in plans]
return [TIER_DEFAULTS[tid] for tid in TIER_ORDER]
def seed_default_plans(db: Session) -> int:
"""Seed subscription_plans table from TIER_DEFAULTS if the table is empty.
Called at application startup. Returns the number of plans inserted (0 if already seeded).
"""
import json
from app.models import SubscriptionPlan
try:
if db.query(SubscriptionPlan).count() > 0:
return 0
except Exception:
return 0 # table may not exist yet during first migration
inserted = 0
for sort_order, (_, tier) in enumerate(TIER_DEFAULTS.items()):
plan = SubscriptionPlan(
plan_id=tier["id"],
name=tier["name"],
tagline=tier.get("tagline", ""),
price_monthly=tier["price_monthly"],
price_yearly=tier["price_yearly"],
trial_days=tier.get("trial_days", 0),
is_highlighted=tier.get("highlight", False),
badge_text=tier.get("badge"),
cta_text=tier.get("cta", "Get started"),
lifetime_file_limit=tier["lifetime_file_limit"],
daily_upload_limit=tier["daily_upload_limit"],
monthly_upload_limit=tier["monthly_upload_limit"],
max_storage_destinations=tier["max_storage_destinations"],
max_ocr_pages_monthly=tier["max_ocr_pages_monthly"],
max_file_size_mb=tier["max_file_size_mb"],
max_mailboxes=tier.get("max_mailboxes", 0),
api_access=tier.get("api_access", False),
features=json.dumps(tier.get("features", [])),
overage_percent=20,
allow_overage_billing=False,
sort_order=sort_order,
is_active=True,
)
db.add(plan)
inserted += 1
try:
db.commit()
logger.info("Seeded %d default subscription plans", inserted)
except Exception as exc:
db.rollback()
logger.error("Failed to seed subscription plans: %s", exc)
inserted = 0
return inserted
# ---------------------------------------------------------------------------
# Usage queries
# ---------------------------------------------------------------------------
def _today_utc() -> date:
return datetime.now(timezone.utc).date()
def _scalar_count(query: Any) -> int:
"""Execute a count query and return an int, defaulting to 0 for NULL."""
return query.scalar() or 0
def get_lifetime_file_count(db: Session, owner_id: str) -> int:
"""Total files ever processed by this user (not counting duplicates)."""
from app.models import FileRecord
return _scalar_count(
db.query(func.count(FileRecord.id)).filter(FileRecord.owner_id == owner_id, FileRecord.is_duplicate.is_(False))
)
def get_today_file_count(db: Session, owner_id: str) -> int:
"""Files processed by this user today (UTC, not counting duplicates)."""
from app.models import FileRecord
today = _today_utc()
return _scalar_count(
db.query(func.count(FileRecord.id)).filter(
FileRecord.owner_id == owner_id,
FileRecord.is_duplicate.is_(False),
func.date(FileRecord.created_at) == today,
)
)
def get_month_file_count(db: Session, owner_id: str) -> int:
"""Files processed by this user this calendar month (UTC, not counting duplicates)."""
from app.models import FileRecord
today = _today_utc()
return _scalar_count(
db.query(func.count(FileRecord.id)).filter(
FileRecord.owner_id == owner_id,
FileRecord.is_duplicate.is_(False),
func.strftime("%Y-%m", FileRecord.created_at) == today.strftime("%Y-%m"),
)
)
def get_year_file_count(db: Session, owner_id: str, period_start: datetime) -> int:
"""Files processed since the start of the current annual subscription period."""
from app.models import FileRecord
return _scalar_count(
db.query(func.count(FileRecord.id)).filter(
FileRecord.owner_id == owner_id,
FileRecord.is_duplicate.is_(False),
FileRecord.created_at >= period_start,
)
)
def _months_elapsed(period_start: datetime, now: datetime) -> int:
"""Calendar months elapsed since *period_start*, clamped to [1, 12]."""
elapsed = (now.year - period_start.year) * 12 + (now.month - period_start.month) + 1
return max(1, min(elapsed, 12))
# ---------------------------------------------------------------------------
# Limit enforcement
# ---------------------------------------------------------------------------
class QuotaExceeded(Exception):
"""Raised when a user has hit a subscription limit."""
def __init__(self, message: str, limit_type: str, limit_value: int, current_value: int) -> None:
super().__init__(message)
self.limit_type = limit_type
self.limit_value = limit_value
self.current_value = current_value
def check_upload_allowed(db: Session, owner_id: str | None, tier_id: str | None) -> None:
"""Raise :class:`QuotaExceeded` if this user is not allowed to upload another file.
Skipped entirely when *owner_id* or *tier_id* is ``None`` (single-user mode).
Enforcement model
-----------------
* **Announced limit** — the quota shown on the pricing page
(``monthly_upload_limit`` in the plan).
* **Overage buffer** — each plan stores ``overage_percent`` (default 20).
Enforcement = announced × (1 + overage_percent / 100). A 150-doc/month
plan with 20 % buffer is enforced at 180 docs.
* **Overage flag** — if ``UserProfile.allow_overage`` is ``True``, quota
checks are bypassed entirely so usage can be billed retroactively.
(Not yet exposed in the admin UI — baked in for future billing.)
* **Yearly carry-over** — yearly subscribers have cumulative quota:
effective limit = monthly_limit × months_elapsed × overage_factor.
Unused quota from earlier months rolls forward automatically.
* **No daily cap** — ``daily_upload_limit`` is kept for display purposes
only; it is never enforced.
"""
if owner_id is None or tier_id is None:
return
tier = get_tier(tier_id, db)
# Per-plan overage_percent overrides global config default
overage_percent: int = tier.get("overage_percent", settings.subscription_overage_percent)
overage_factor: float = 1.0 + overage_percent / 100.0
from app.models import UserProfile
profile = db.query(UserProfile).filter(UserProfile.user_id == owner_id).first()
allow_overage: bool = bool(profile.allow_overage) if profile else False
billing_cycle: str = (profile.subscription_billing_cycle if profile else None) or "monthly"
period_start: datetime | None = profile.subscription_period_start if profile else None
# 1. Lifetime file cap (free tier) — always enforced regardless of overage flag
lifetime_limit: int = tier["lifetime_file_limit"]
if lifetime_limit > 0:
enforcement_limit = int(lifetime_limit * overage_factor)
count = get_lifetime_file_count(db, owner_id)
if count >= enforcement_limit:
raise QuotaExceeded(
f"Lifetime file limit of {lifetime_limit} reached for the {tier['name']} plan. "
"Please upgrade to continue processing documents.",
limit_type="lifetime",
limit_value=lifetime_limit,
current_value=count,
)
# 2. Monthly cap — bypassed when allow_overage is True (future billing)
if allow_overage:
return
monthly_limit: int = tier["monthly_upload_limit"]
if monthly_limit > 0:
if billing_cycle == "yearly" and period_start is not None:
now = datetime.now(timezone.utc)
months = _months_elapsed(period_start, now)
cumulative_budget = int(monthly_limit * months * overage_factor)
cumulative_used = get_year_file_count(db, owner_id, period_start)
if cumulative_used >= cumulative_budget:
raise QuotaExceeded(
f"Annual document quota for the {tier['name']} plan has been reached. "
"Unused monthly quota carries forward — your limit resets on your annual "
"renewal date, or you can upgrade your plan.",
limit_type="monthly",
limit_value=monthly_limit,
current_value=cumulative_used,
)
else:
count = get_month_file_count(db, owner_id)
enforcement_limit = int(monthly_limit * overage_factor)
if count >= enforcement_limit:
raise QuotaExceeded(
f"Monthly file limit of {monthly_limit} reached for the {tier['name']} plan. "
"Please upgrade your plan for more documents this month.",
limit_type="monthly",
limit_value=monthly_limit,
current_value=count,
)
def get_user_tier_id(db: Session, owner_id: str | None) -> str:
"""Return the subscription tier id for *owner_id*, defaulting to 'free'."""
if owner_id is None:
return DEFAULT_TIER
from app.models import UserProfile
profile = db.query(UserProfile).filter(UserProfile.user_id == owner_id).first()
if profile and profile.subscription_tier:
return profile.subscription_tier
return DEFAULT_TIER
def get_user_usage(db: Session, owner_id: str) -> dict[str, int]:
"""Return file counts for *owner_id*, including carry-over data for yearly plans."""
from app.models import UserProfile
profile = db.query(UserProfile).filter(UserProfile.user_id == owner_id).first()
result: dict[str, int] = {
"lifetime": get_lifetime_file_count(db, owner_id),
"today": get_today_file_count(db, owner_id),
"month": get_month_file_count(db, owner_id),
}
if profile and (profile.subscription_billing_cycle or "monthly") == "yearly" and profile.subscription_period_start:
result["year_to_date"] = get_year_file_count(db, owner_id, profile.subscription_period_start)
return result
Reference in New Issue View Git Blame Copy Permalink