Spaces:
Runtime error
Runtime error
Refactor VideoTemplateEditor component: remove AI editing features and related state management, streamline template loading, and update styles for improved layout. Enhance CSS for responsive design and visual consistency across various components, including dropdowns, sync stats, and quiz elements. Remove deprecated premium badge styles and adjust layout for chat components.
66de9c8 | """Upgrade flow: plan status with meters, nudges, and the seams where the payment provider plugs in. | |
| Stripe goes in create_checkout_url, create_portal_url and the webhook route; everything around it | |
| (meters, nudges, subscription bookkeeping, 402 payloads) is provider agnostic. | |
| """ | |
| from datetime import UTC, datetime | |
| from sqlalchemy import func, select | |
| from sqlalchemy.orm import Session | |
| from app.config import settings | |
| from app.models import Artifact, CalendarItem, Document, Notebook, Source, Subscription, User, VoiceProfile, Workspace | |
| from app.services.plans import PLANS, Plan, effective_plan, next_plan, plan_dict | |
| from app.services.usage import month_usage | |
| from app.services.video_quota import sync_video_quota, video_usage | |
| CYCLES = {"monthly", "annual"} | |
| LOW_FUEL = 0.8 # meters at or above this share of the allowance start nudging | |
| METERS = { | |
| "audio_minutes": ("Audio overviews", "min"), | |
| "videos": ("Videos", ""), | |
| "launch_kits": ("Launch Kits", ""), | |
| } | |
| def _meters(plan: Plan, used: dict, sources: int) -> list[dict]: | |
| rows = [] | |
| meters = dict(METERS) | |
| if plan.audio_overviews >= 0: # a count of audio overviews (Free), not minutes | |
| meters = {("audio_overviews" if k == "audio_minutes" else k): (("Audio overviews", "") if k == "audio_minutes" else v) | |
| for k, v in METERS.items()} | |
| for key, (label, unit) in meters.items(): | |
| limit = getattr(plan, key) | |
| rows.append({"key": key, "label": label, "unit": unit, "used": used[key], "limit": limit, | |
| "pct": 0 if limit < 0 else round(min(1.0, used[key] / max(limit, 1)), 3)}) | |
| rows.append({"key": "sources", "label": "Sources", "unit": "", "used": sources, "limit": plan.sources, | |
| "pct": 0 if plan.sources < 0 else round(min(1.0, sources / max(plan.sources, 1)), 3)}) | |
| return rows | |
| def _more_meters(db: Session, workspace: Workspace, plan: Plan, used: dict) -> list[dict]: | |
| """The rest of the plan's limits, shown when the sidebar card is opened: indexed posts, reports and infographics.""" | |
| posts = db.scalar(select(func.count()).select_from(Document) | |
| .where(Document.workspace_id == workspace.id, Document.path.is_not(None))) or 0 | |
| rows = [("indexed_posts", "Posts indexed", posts, plan.indexed_posts), | |
| ("reports", "Reports", used["reports"], plan.reports), | |
| ("infographics", "Infographics", used["infographics"], plan.infographics)] | |
| return [{"key": k, "label": label, "unit": "", "used": u, "limit": limit, | |
| "pct": 0 if limit < 0 else round(min(1.0, u / max(limit, 1)), 3)} for k, label, u, limit in rows] | |
| def _count(db: Session, model, workspace_id, *where) -> int: | |
| return db.scalar(select(func.count()).select_from(model).where(model.workspace_id == workspace_id, *where)) or 0 | |
| def _nudge(nid: str, tone: str, title: str, body: str, cta: str, to: str | None = None, priority: int = 50) -> dict: | |
| """tone: limit (out of or low on fuel), upgrade (locked perk), action (do more with what you have). | |
| to is an app route, or None when the CTA opens the upgrade popup.""" | |
| return {"id": nid, "tone": tone, "title": title, "body": body, "cta": cta, "to": to, "priority": priority} | |
| def build_nudges(db: Session, workspace: Workspace, plan: Plan, meters: list[dict], can_upgrade: bool) -> list[dict]: | |
| ws = workspace.id | |
| out: list[dict] = [] | |
| month = datetime.now(UTC).strftime("%Y-%m") | |
| if can_upgrade: | |
| for m in meters: | |
| if m["limit"] < 0 or m["key"] == "sources": | |
| continue | |
| label = m["label"].lower() | |
| amount = f"{m['limit']:g} min of {label}" if m["unit"] else f"{m['limit']:g} {label}" | |
| when = "" if plan.lifetime else " this month" | |
| if m["pct"] >= 1: | |
| out.append(_nudge(f"empty-{m['key']}-{month}", "limit", f"Out of {label}{when}", | |
| f"You have used all {amount}. Upgrade to keep launching.", "Refuel now", | |
| priority=100)) | |
| elif m["pct"] >= LOW_FUEL: | |
| left = f"{m['limit'] - m['used']:g}" + (" min" if m["unit"] else "") | |
| out.append(_nudge(f"low-{m['key']}-{month}", "limit", f"Running low on {label}", | |
| f"Only {left} left{when}. Upgrade before you stall mid launch.", | |
| "See plans", priority=90)) | |
| if not plan.voice_cloning: | |
| out.append(_nudge("perk-voice", "upgrade", "Narrate in your own voice", | |
| "Writer unlocks voice cloning, so every audio overview sounds like you.", | |
| "Unlock voice", priority=40)) | |
| sources = _count(db, Source, ws) | |
| if sources == 0: | |
| out.append(_nudge("act-source", "action", "Connect your archive", | |
| "Paste your Substack or blog URL and Notestack reads every post.", "Connect a source", | |
| "/app/sources", priority=95)) | |
| return sorted(out, key=lambda n: -n["priority"]) | |
| if _count(db, Notebook, ws) == 0: | |
| out.append(_nudge("act-notebook", "action", "Open your first notebook", | |
| "Ask your archive anything and get answers with citations to your own posts.", | |
| "Create a notebook", "/app/notebooks", priority=80)) | |
| kits = next(m for m in meters if m["key"] == "launch_kits") | |
| if kits["used"] == 0: | |
| left = "Unlimited" if kits["limit"] < 0 else f"{kits['limit']}" | |
| out.append(_nudge(f"act-kit-{month}", "action", "Launch your latest post", | |
| f"{left} Launch Kits this month, none used yet. One click turns a post into threads, " | |
| "LinkedIn posts and quote cards.", "Build a Launch Kit", | |
| "/app/launch-kit", priority=70)) | |
| elif kits["limit"] > 0 and kits["used"] < kits["limit"]: | |
| left = kits["limit"] - kits["used"] | |
| out.append(_nudge(f"act-kit-left-{month}-{kits['used']}", "action", | |
| f"{left} Launch Kit{'s' if left != 1 else ''} left this month", | |
| "They reset on the 1st. Pick an older post and give it a second orbit.", | |
| "Use one now", "/app/launch-kit", priority=45)) | |
| if _count(db, Artifact, ws, Artifact.type == "audio_overview") == 0: | |
| out.append(_nudge("act-audio", "action", "Hear your archive", | |
| "Make a two host audio overview of any notebook in about a minute.", "Make audio", | |
| "/app/studio", priority=55)) | |
| if not db.scalar(select(VoiceProfile.id).where(VoiceProfile.workspace_id == ws)): | |
| out.append(_nudge("act-voice", "action", "Teach Notestack your voice", | |
| "A voice profile keeps every draft sounding like you wrote it.", "Build voice profile", | |
| "/app/voice", priority=50)) | |
| if _count(db, CalendarItem, ws) == 0: | |
| out.append(_nudge("act-launchpad", "action", "Fill your launch window", | |
| "Schedule this week's posts on the Launchpad and let them go out on their own.", | |
| "Open Launchpad", "/app/launchpad", priority=35)) | |
| out.append(_nudge(f"act-resurface-{month}", "action", "Old posts, new orbit", | |
| "Resurfacing finds evergreen posts worth sharing again this month.", "Find evergreen posts", | |
| "/app/launchpad#ideas", priority=20)) | |
| return sorted(out, key=lambda n: -n["priority"]) | |
| def billing_status(db: Session, workspace: Workspace) -> dict: | |
| plan = effective_plan(db, workspace) | |
| used = month_usage(db, workspace.id, plan.lifetime) | |
| videos = video_usage(db, workspace) | |
| used["videos"] = videos["used"] | |
| sources = _count(db, Source, workspace.id) | |
| meters = _meters(plan, used, sources) | |
| upgrade = next_plan(plan) | |
| can_upgrade = settings.billing_enabled and upgrade is not None | |
| sub = db.query(Subscription).filter_by(workspace_id=workspace.id).one_or_none() | |
| return { | |
| "billing_enabled": settings.billing_enabled, | |
| "plan": plan_dict(plan), | |
| "plans": [plan_dict(p) for p in PLANS.values()], | |
| "next_plan": upgrade.id if upgrade else None, | |
| "can_upgrade": can_upgrade, | |
| "has_billing_account": bool(sub and sub.provider_customer_id), | |
| "period_end": sub.current_period_end.isoformat() if sub and sub.current_period_end else None, | |
| "meters": meters, | |
| "more_meters": _more_meters(db, workspace, plan, used), | |
| "since": used["since"], | |
| "videos_resets_at": videos["resets_at"], | |
| "nudges": build_nudges(db, workspace, plan, meters, can_upgrade), | |
| } | |
| def apply_subscription(db: Session, workspace_id, *, plan: str, status: str, provider: str = "stripe", | |
| customer_id: str | None = None, subscription_id: str | None = None, | |
| current_period_end: datetime | None = None) -> Subscription: | |
| """Upsert the workspace subscription. Call this from the webhook for checkout.session.completed, | |
| customer.subscription.updated and customer.subscription.deleted (status 'canceled' drops to Free).""" | |
| if plan not in PLANS: | |
| raise ValueError(f"Unknown plan {plan}") | |
| sub = db.query(Subscription).filter_by(workspace_id=workspace_id).one_or_none() | |
| workspace = db.get(Workspace, workspace_id) | |
| before = effective_plan(db, workspace).indexed_posts if workspace else None | |
| if not sub: | |
| sub = Subscription(workspace_id=workspace_id) | |
| db.add(sub) | |
| sub.plan, sub.status, sub.provider = plan, status, provider | |
| sub.provider_customer_id = customer_id or sub.provider_customer_id | |
| sub.provider_subscription_id = subscription_id or sub.provider_subscription_id | |
| sub.current_period_end = current_period_end | |
| db.commit() | |
| if workspace: | |
| sync_video_quota(db, workspace) | |
| if workspace and effective_plan(db, workspace).indexed_posts != before: | |
| resync_sources(db, workspace_id) | |
| return sub | |
| def resync_sources(db: Session, workspace_id) -> None: | |
| """The post limit changed: re-sync every feed so locked posts unlock on upgrade (or re-lock on downgrade).""" | |
| from app.pipeline.ingest import IMPORTS_FEED | |
| from app.services.jobs import create_job | |
| for source in db.scalars(select(Source).where(Source.workspace_id == workspace_id, | |
| Source.feed_url != IMPORTS_FEED)): | |
| source.sync_status = "pending" | |
| create_job(db, workspace_id, "ingest", {"source_id": str(source.id)}) | |
| db.commit() | |
| def return_urls() -> tuple[str, str]: | |
| """(success_url, cancel_url) for the hosted checkout page.""" | |
| base = settings.frontend_url.rstrip("/") | |
| return f"{base}/app?checkout=success", f"{base}/app?checkout=cancel" | |
| def create_checkout_url(db: Session, user: User, workspace: Workspace, plan: Plan, cycle: str) -> str: | |
| """Hosted Stripe Checkout for plan on the monthly or annual cycle, or an in place plan change (which may | |
| return Stripe's invoice page when the bank wants 3D Secure). See services/stripe_billing.py.""" | |
| from app.services import stripe_billing | |
| success, cancel = return_urls() | |
| return stripe_billing.start_checkout(db, user, workspace, plan, cycle, success, cancel) | |
| def create_portal_url(db: Session, workspace: Workspace) -> str: | |
| """Stripe customer portal: change card, switch or cancel the plan, download invoices.""" | |
| from app.services import stripe_billing | |
| return stripe_billing.portal_url(db, workspace, f"{settings.frontend_url.rstrip('/')}/app/settings") | |