"""Unified Credit System — tool cost registry and credit constants. Every tool has a credit cost. Lighter tools cost 1 credit, heavier server-side conversions cost 2, CPU/ML-intensive tools cost 3, and AI-powered tools cost 5+. Heavy and AI tools use dynamic pricing based on file/request size. Light and medium tools keep a fixed cost per invocation. This module is the single source of truth for all credit-related constants consumed by policy_service, credit_service, quote_service, and the frontend config endpoint. """ from __future__ import annotations import math import os from dataclasses import dataclass # ── Credit allocations per rolling 30-day window ──────────────── FREE_CREDITS_PER_WINDOW = int(os.getenv("FREE_CREDITS_PER_WINDOW", "50")) PRO_CREDITS_PER_WINDOW = int(os.getenv("PRO_CREDITS_PER_WINDOW", "500")) CREDIT_WINDOW_DAYS = int(os.getenv("CREDIT_WINDOW_DAYS", "30")) # ── Guest demo budget (anonymous, pre-registration) ──────────── GUEST_DEMO_BUDGET = int(os.getenv("GUEST_DEMO_BUDGET", "3")) GUEST_DEMO_TTL_HOURS = int(os.getenv("GUEST_DEMO_TTL_HOURS", "24")) # ── API quota (Pro only, per rolling window) ──────────────────── PRO_API_CREDITS_PER_WINDOW = int(os.getenv("PRO_API_CREDITS_PER_WINDOW", "1000")) # ── Cost tiers ────────────────────────────────────────────────── TIER_LIGHT = 1 # Fast, in-memory or trivial server ops TIER_MEDIUM = 2 # Server-side conversion (LibreOffice, Ghostscript, etc.) TIER_HEAVY = 3 # CPU/ML-intensive (OCR, background removal, compression) TIER_AI = 5 # AI-powered tools (LLM API calls) # ── Dynamic pricing model ────────────────────────────────────── # Tools marked as DYNAMIC_PRICING_TIERS use size-based cost instead of # fixed tier cost. The formula is: # cost = base + ceil(file_size_kb / step_kb) * per_step # capped at max_credits. AI tools also add an estimated-tokens surcharge. DYNAMIC_PRICING_TIERS: set[str] = {"heavy", "ai"} @dataclass(frozen=True) class DynamicPricingRule: """Size-based pricing rule for one tool family.""" base: int # minimum credits charged step_kb: int # every `step_kb` KB adds `per_step` credits per_step: int # credits added per step max_credits: int # hard cap on credits per invocation # AI-specific: extra credits per estimated 1 000 input tokens token_step: int = 0 # 0 means no token surcharge per_token_step: int = 0 # Default rules per tier — individual tools can override via # TOOL_DYNAMIC_OVERRIDES below. HEAVY_DEFAULT_RULE = DynamicPricingRule( base=3, step_kb=500, per_step=1, max_credits=10, ) AI_DEFAULT_RULE = DynamicPricingRule( base=5, step_kb=200, per_step=1, max_credits=20, token_step=1000, per_token_step=1, ) # Per-tool overrides (tool slug → rule). TOOL_DYNAMIC_OVERRIDES: dict[str, DynamicPricingRule] = { # Translation is heavier than chat/summarize "translate-pdf": DynamicPricingRule( base=6, step_kb=150, per_step=1, max_credits=25, token_step=1000, per_token_step=1, ), } def _tier_label(cost: int) -> str: """Return the tier family name for a fixed cost value.""" if cost >= TIER_AI: return "ai" if cost >= TIER_HEAVY: return "heavy" if cost >= TIER_MEDIUM: return "medium" return "light" def _get_dynamic_rule(tool: str, tier: str) -> DynamicPricingRule | None: """Return the dynamic pricing rule for *tool*, or None if fixed.""" if tier not in DYNAMIC_PRICING_TIERS: return None if tool in TOOL_DYNAMIC_OVERRIDES: return TOOL_DYNAMIC_OVERRIDES[tool] return AI_DEFAULT_RULE if tier == "ai" else HEAVY_DEFAULT_RULE def is_dynamic_tool(tool: str) -> bool: """Return True if *tool* uses size-based dynamic pricing.""" base = TOOL_CREDIT_COSTS.get(tool, DEFAULT_CREDIT_COST) return _tier_label(base) in DYNAMIC_PRICING_TIERS def calculate_dynamic_cost( tool: str, file_size_kb: float = 0, estimated_tokens: int = 0, ) -> int: """Calculate the credit cost for a dynamic tool given size metrics. For fixed-price tools this returns the static tier cost unchanged. """ base_cost = TOOL_CREDIT_COSTS.get(tool, DEFAULT_CREDIT_COST) tier = _tier_label(base_cost) rule = _get_dynamic_rule(tool, tier) if rule is None: return base_cost size_surcharge = math.ceil(max(0, file_size_kb) / rule.step_kb) * rule.per_step token_surcharge = 0 if rule.token_step and estimated_tokens > 0: token_surcharge = math.ceil(estimated_tokens / rule.token_step) * rule.per_token_step total = rule.base + size_surcharge + token_surcharge return min(total, rule.max_credits) # ── Per-tool credit costs ─────────────────────────────────────── # Keys match the `tool` parameter passed to record_usage_event / routes. TOOL_CREDIT_COSTS: dict[str, int] = { # ─── PDF Core (light operations) ──────────────────────────── "merge-pdf": TIER_LIGHT, "split-pdf": TIER_LIGHT, "rotate-pdf": TIER_LIGHT, "reorder-pdf": TIER_LIGHT, "extract-pages": TIER_LIGHT, "page-numbers": TIER_LIGHT, "watermark-pdf": TIER_LIGHT, "protect-pdf": TIER_LIGHT, "unlock-pdf": TIER_LIGHT, "flatten-pdf": TIER_LIGHT, "repair-pdf": TIER_LIGHT, "pdf-metadata": TIER_LIGHT, "crop-pdf": TIER_LIGHT, "sign-pdf": TIER_LIGHT, "pdf-to-images": TIER_LIGHT, "images-to-pdf": TIER_LIGHT, # ─── Conversion (medium — server-side rendering) ──────────── "pdf-to-word": TIER_MEDIUM, "word-to-pdf": TIER_MEDIUM, "pdf-to-excel": TIER_MEDIUM, "excel-to-pdf": TIER_MEDIUM, "pdf-to-pptx": TIER_MEDIUM, "pptx-to-pdf": TIER_MEDIUM, "html-to-pdf": TIER_MEDIUM, "pdf-editor": TIER_MEDIUM, # ─── Image (light to medium) ──────────────────────────────── "image-converter": TIER_LIGHT, "image-resize": TIER_LIGHT, "image-crop": TIER_LIGHT, "image-rotate-flip": TIER_LIGHT, "image-to-svg": TIER_MEDIUM, # ─── Image / PDF heavy (CPU/ML) ──────────────────────────── "compress-pdf": TIER_HEAVY, "compress-image": TIER_HEAVY, "ocr": TIER_HEAVY, "remove-background": TIER_HEAVY, "remove-watermark-pdf": TIER_HEAVY, # ─── Utility ──────────────────────────────────────────────── "qr-code": TIER_LIGHT, "barcode-generator": TIER_LIGHT, "video-to-gif": TIER_MEDIUM, "word-counter": TIER_LIGHT, "text-cleaner": TIER_LIGHT, # ─── AI-powered ───────────────────────────────────────────── "chat-pdf": TIER_AI, "summarize-pdf": TIER_AI, "translate-pdf": TIER_AI, "extract-tables": TIER_AI, "pdf-flowchart": TIER_AI, # ─── Route-specific aliases ───────────────────────────────────── # Some routes record a tool name that differs from the manifest slug. # Both names must map to the same cost. "barcode": TIER_LIGHT, # manifest: barcode-generator "image-convert": TIER_LIGHT, # manifest: image-converter "ocr-image": TIER_HEAVY, # manifest: ocr "ocr-pdf": TIER_HEAVY, # manifest: ocr "pdf-flowchart-sample": TIER_AI, # manifest: pdf-flowchart "pdf-edit": TIER_MEDIUM, # manifest: pdf-editor "edit-metadata": TIER_LIGHT, # manifest: pdf-metadata "remove-watermark": TIER_HEAVY, # manifest: remove-watermark-pdf "remove-bg": TIER_HEAVY, # manifest: remove-background "video-frames": TIER_MEDIUM, # route alias for video-to-gif "edit-pdf-text": TIER_MEDIUM, # route alias for pdf-editor } # Default cost for any tool not explicitly listed DEFAULT_CREDIT_COST = TIER_LIGHT def get_tool_credit_cost(tool: str) -> int: """Return the *fixed* credit cost for a given tool slug. For dynamic tools this is the base/minimum cost. Use :func:`calculate_dynamic_cost` when file size is known. """ return TOOL_CREDIT_COSTS.get(tool, DEFAULT_CREDIT_COST) def get_credits_for_plan(plan: str) -> int: """Return the total credits per window for a plan.""" return PRO_CREDITS_PER_WINDOW if plan == "pro" else FREE_CREDITS_PER_WINDOW def get_all_tool_costs() -> dict[str, int]: """Return the full cost registry — used by the config API endpoint.""" return dict(TOOL_CREDIT_COSTS) def get_dynamic_tools_info() -> dict[str, dict]: """Return metadata about tools that use dynamic pricing. Used by the config/credit-info endpoint so the frontend can display "price varies by file size" for these tools. """ result: dict[str, dict] = {} seen: set[str] = set() for slug, cost in TOOL_CREDIT_COSTS.items(): tier = _tier_label(cost) rule = _get_dynamic_rule(slug, tier) if rule is None: continue if slug in seen: continue seen.add(slug) result[slug] = { "base": rule.base, "step_kb": rule.step_kb, "per_step": rule.per_step, "max_credits": rule.max_credits, "has_token_surcharge": rule.token_step > 0, } return result