""" Pricing plans, credits, quotas and upgrade nudges (pricing policy of October 2026). Unit: one CREDIT = one document of up to CREDIT_WORDS (15,000) words; a longer document uses one credit per started 15,000 words (a 40,000-word thesis = 3 credits). The number is shown before the run and charged only after a successful run (a failed run costs nothing). The same file can be re-run FREE_RERUNS times free within RERUN_DAYS. guest paste only: 300 words a paste, 3 a day, 2 Smart passes free 5 pastes a day (300 words); 1 trial document up to 1,500 words; 3 detector files a day (trimmed 2026-10-28 so free use costs at most about Rs 15 a day per account) single ₹199 / $3 one credit, valid 6 months pack ₹799 / $12 five credits, valid 6 months scholar ₹999 / $14 a month 10 credits a month, pastes up to 1,500 words, unlimited detector pro ₹2,999 / $39 a month 40 credits a month, pastes up to 5,000 words, priority institution ₹9,999 / $129 a month 150 credits a month (activated by the admin, invoice billing) extra ₹149 / $2 one extra credit for monthly-plan users whose month's credits are used admin everything free and unlimited Journal Finder: free. Journal Formatter: 1 file a month free, included in plans (5 / 20 / 60 a month), ₹99 / $2 per file otherwise (shown on the Pricing page; enforced once the Formatter shares the suite's login). Monthly plans: 30 days from activation, credits do not carry over, renewal before expiry adds a month on top; not renewed by the end date -> switched off automatically (plan_history + email). Prices include GST (INR). Money comes in through `activate_plan()` / `add_credits()` (Admin panel today, Razorpay webhook later). """ import calendar import html import math import os import threading import time import gradio as gr import auth as A import db def _env_i(k, d): return int(os.getenv(k, str(d))) def _env_f(k, d): return float(os.getenv(k, str(d))) FREE_TRIAL_DOCS = _env_i("FREE_TRIAL_DOCS", 1) FREE_DETECT_DAY = _env_i("FREE_DETECT_DAY", 3) # detector file reports a day on the free account CREDIT_WORDS = _env_i("CREDIT_WORDS", 15000) # words per credit CREDIT_MONTHS = _env_i("CREDIT_MONTHS", 6) # bought credits stay valid this many months FREE_RERUNS = _env_i("FREE_RERUNS", 2) # free re-runs of the same file ... RERUN_DAYS = _env_i("RERUN_DAYS", 7) # ... within this many days PAY_EMAIL = os.getenv("PAY_EMAIL", "info@cssoftwaresolutions.tech") PAY_WHATSAPP = os.getenv("PAY_WHATSAPP", "918886306793") PROMO_TEXT = os.getenv("PROMO_TEXT", "") UNLIMITED = 10 ** 9 # docs_month = credits per subscription month; format_month = Journal Formatter files a month (display) PLANS = { "guest": dict(name="Guest", inr=0, usd=0, docs_month=0, docs_day=0, pastes=_env_i("GUEST_PASTES", 3), paste_words=300, passes=2, doc_words=0, format_month=0, priority=False), "free": dict(name="Free", inr=0, usd=0, docs_month=0, docs_day=0, pastes=_env_i("FREE_PASTES", 5), paste_words=300, passes=2, doc_words=_env_i("FREE_DOC_WORDS", 1500), format_month=1, priority=False), "scholar": dict(name="Scholar", inr=_env_i("PRICE_SCHOLAR", 999), usd=_env_f("PRICE_SCHOLAR_USD", 14), docs_month=_env_i("SCHOLAR_CREDITS", 10), docs_day=0, pastes=UNLIMITED, paste_words=1500, passes=10, doc_words=150000, format_month=5, priority=False), "pro": dict(name="Pro", inr=_env_i("PRICE_PRO", 2999), usd=_env_f("PRICE_PRO_USD", 39), docs_month=_env_i("PRO_CREDITS", 40), docs_day=0, pastes=UNLIMITED, paste_words=5000, passes=10, doc_words=150000, format_month=20, priority=True), "institution": dict(name="Institution", inr=_env_i("PRICE_INSTITUTION", 9999), usd=_env_f("PRICE_INSTITUTION_USD", 129), docs_month=_env_i("INSTITUTION_CREDITS", 150), docs_day=0, pastes=UNLIMITED, paste_words=5000, passes=10, doc_words=150000, format_month=60, priority=True), # earlier plan (₹4,999 for 300 documents a month): kept so existing subscriptions run to their end date "monthly": dict(name="Monthly (earlier plan)", inr=4999, usd=59, docs_month=_env_i("MONTHLY_DOCS", 300), docs_day=0, pastes=UNLIMITED, paste_words=5000, passes=10, doc_words=150000, format_month=20, priority=True), "admin": dict(name="Admin", inr=0, usd=0, docs_month=0, docs_day=0, pastes=UNLIMITED, paste_words=20000, passes=10, doc_words=UNLIMITED, format_month=UNLIMITED, priority=True), } for _old in ("student", "agency"): # names from an earlier price list PLANS[_old] = PLANS["scholar"] PAID = ("scholar", "pro", "institution", "monthly", "student", "agency") _PAID_SQL = "('scholar','pro','institution','monthly','student','agency')" SELL = ("scholar", "pro") # monthly plans on the Pricing page (institution: on request) ADMIN_PLANS = ("scholar", "pro", "institution", "monthly") # credit products SINGLE = dict(name="Single credit", inr=_env_i("PRICE_SINGLE", 199), usd=_env_f("PRICE_SINGLE_USD", 3), credits=1) PACK = dict(name="Pack of 5 credits", inr=_env_i("PRICE_PACK", 799), usd=_env_f("PRICE_PACK_USD", 12), credits=5) EXTRA = dict(name="Extra credit", inr=_env_i("PRICE_EXTRA", 149), usd=_env_f("PRICE_EXTRA_USD", 2), credits=1) FORMAT_FILE = dict(name="Journal formatting", inr=_env_i("PRICE_FORMAT", 99), usd=_env_f("PRICE_FORMAT_USD", 2)) EXTRA_DOC = SINGLE # older name INR_NOTE = os.getenv("PRICE_INR_NOTE", "Prices include 18% GST.") DEFAULT_CURRENCY = os.getenv("PRICE_CURRENCY", "INR").strip().upper() _ITEMS = {"single": SINGLE, "pack": PACK, "extra": EXTRA, "format": FORMAT_FILE} def _cur(currency): c = str(currency or DEFAULT_CURRENCY).upper() return "USD" if c.startswith(("USD", "$")) else "INR" def money(amount, cur="INR", per=""): """₹999 / $14 ('' -> Free).""" if not amount: return "Free" return f"₹{amount:,.0f}{per}" if cur == "INR" else f"${amount:,.0f}{per}" def price_of(item, cur="INR"): p = _ITEMS.get(item) or PLANS[item] return money(p["usd"] if cur == "USD" else p["inr"], cur) def both(item): """'₹999 ($14)' with the default currency first.""" c1 = _cur(DEFAULT_CURRENCY) c2 = "USD" if c1 == "INR" else "INR" return f"{price_of(item, c1)} ({price_of(item, c2)})" # Yearly plans: twelve months paid at once, YEARLY_DISCOUNT percent off twelve monthly payments. Activated with # activate_plan(email, plan, 12): the monthly credits work exactly as on the monthly plan. YEARLY_DISCOUNT = _env_f("YEARLY_DISCOUNT", 20) def yearly_amount(plan, cur="INR"): """Price of one year of `plan` (INR rounded to ₹10, USD to $1).""" p = PLANS[plan] v = 12 * (p["usd"] if _cur(cur) == "USD" else p["inr"]) * (1 - YEARLY_DISCOUNT / 100) return round(v) if _cur(cur) == "USD" else int(round(v / 10) * 10) def yearly_price(plan, cur="INR"): return money(yearly_amount(plan, cur), _cur(cur)) def credits_needed(words): """Credits a document of `words` words uses (one per started CREDIT_WORDS words).""" return max(1, math.ceil((words or 0) / CREDIT_WORDS)) def per_day_text(p): return f"{p['docs_month']} credits per month" # ---------------------------------------------------------------- storage def _db(): con = A._db() if db.IS_MYSQL: return con cols = {r[1] for r in con.execute("PRAGMA table_info(users)")} for c, ddl in (("plan", "TEXT DEFAULT 'free'"), ("plan_until", "TEXT"), ("period_start", "TEXT"), ("period_docs", "INTEGER DEFAULT 0"), ("doc_credits", "INTEGER DEFAULT 0"), ("trial_used", "INTEGER DEFAULT 0"), ("plan_started", "TEXT"), ("plan_status", "TEXT DEFAULT 'free'"), ("credits_until", "TEXT")): if c not in cols: try: con.execute(f"ALTER TABLE users ADD COLUMN {c} {ddl}") except Exception as err: # added at the same moment by another thread if "duplicate column" not in str(err).lower(): raise con.execute("CREATE TABLE IF NOT EXISTS payments (id INTEGER PRIMARY KEY AUTOINCREMENT, " "ts TEXT DEFAULT CURRENT_TIMESTAMP, email TEXT, item TEXT, amount TEXT, source TEXT, note TEXT)") con.execute("CREATE TABLE IF NOT EXISTS upgrade_requests (id INTEGER PRIMARY KEY AUTOINCREMENT, " "ts TEXT DEFAULT CURRENT_TIMESTAMP, email TEXT, name TEXT, item TEXT, status TEXT DEFAULT 'open')") con.execute("CREATE TABLE IF NOT EXISTS plan_history (id INTEGER PRIMARY KEY AUTOINCREMENT, " "ts TEXT DEFAULT CURRENT_TIMESTAMP, email TEXT, plan TEXT, started TEXT, until TEXT, " "event TEXT, note TEXT)") con.execute("CREATE TABLE IF NOT EXISTS reruns (email TEXT, fkey TEXT, first_ts TEXT, runs INTEGER DEFAULT 0, " "PRIMARY KEY (email, fkey))") con.execute("CREATE TABLE IF NOT EXISTS daily_usage (email TEXT, day TEXT, tool TEXT, n INTEGER DEFAULT 0, " "PRIMARY KEY (email, day, tool))") con.execute("CREATE TABLE IF NOT EXISTS charges (id INTEGER PRIMARY KEY AUTOINCREMENT, ref TEXT UNIQUE, " "ts TEXT DEFAULT CURRENT_TIMESTAMP, email TEXT, tool TEXT, name TEXT, words INTEGER, seconds REAL, " "gpu_usd REAL, space_usd REAL, llm_usd REAL, cost_usd REAL, amount_inr REAL, amount_usd REAL, " "status TEXT DEFAULT 'pending', output TEXT, paid_ts TEXT, source TEXT, note TEXT)") return con def _now(): return time.time() def _ts(t): return time.strftime("%Y-%m-%d %H:%M:%S", time.gmtime(t)) def add_month(t): """Same day and time next calendar month (31 Jan -> 28/29 Feb).""" g = time.gmtime(t) y, m = (g.tm_year + 1, 1) if g.tm_mon == 12 else (g.tm_year, g.tm_mon + 1) d = min(g.tm_mday, calendar.monthrange(y, m)[1]) return calendar.timegm((y, m, d, g.tm_hour, g.tm_min, g.tm_sec)) def _parse(s): try: return calendar.timegm(time.strptime(s[:19], "%Y-%m-%d %H:%M:%S")) except Exception: return 0 def _ist_date(s): """'2026-10-02 10:00:00' (UTC) -> '02 Nov 2026' in IST.""" t = _parse(s or "") return time.strftime("%d %b %Y", time.gmtime(t + 330 * 60)) if t else "" # ---------------------------------------------------------------- automatic discontinue def expire_due(email=None): """Switch off every paid plan whose month ended without a renewal. Returns the emails switched off.""" now = _ts(_now()) with _db() as con: q = ("SELECT email, plan, plan_started, plan_until FROM users WHERE plan IN " + _PAID_SQL + " " "AND plan_until IS NOT NULL AND plan_until <= ?") rows = con.execute(q + (" AND email=?" if email else ""), (now, email) if email else (now,)).fetchall() for e, plan, started, until in rows: con.execute("UPDATE users SET plan='free', plan_status='expired', period_docs=0 WHERE email=?", (e,)) con.execute("INSERT INTO plan_history (email, plan, started, until, event, note) VALUES (?,?,?,?,?,?)", (e, plan, started, until, "expired", "not renewed by the end of the month")) if rows: A.request_sync() try: import mailer as M for e, plan, _, until in rows: M.send_plan_expired(e, PLANS[plan]["name"], _ist_date(until)) except Exception as err: print(f"[plans] expiry email skipped: {err}", flush=True) print(f"[plans] {len(rows)} plan(s) discontinued (not renewed)", flush=True) return [r[0] for r in rows] def _expiry_loop(): while True: try: expire_due() except Exception as err: print(f"[plans] expiry check failed: {err}", flush=True) time.sleep(int(os.getenv("PLAN_CHECK_SECONDS", "3600"))) threading.Thread(target=_expiry_loop, daemon=True).start() # ---------------------------------------------------------------- state def _month_window(pstart): """Start of the current subscription month (period_start moved forward month by month) -> (start_ts, changed).""" t = _parse(pstart) if pstart else 0 if not t: return _now(), True changed = False while add_month(t) <= _now(): t, changed = add_month(t), True return t, changed def state(user): """Everything the UI and the checks need about a user's plan.""" if A.is_admin(user): return dict(key="admin", **PLANS["admin"], until=None, started=None, used=0, used_today=0, credits=0, credits_until=None, trial_left=0, paid=True, admin=True, status="admin", month_end=None) if not user: return dict(key="guest", **PLANS["guest"], until=None, started=None, used=0, used_today=0, credits=0, credits_until=None, trial_left=0, paid=False, admin=False, status="guest", month_end=None) with _db() as con: row = con.execute("SELECT plan, plan_until, plan_started, doc_credits, trial_used, plan_status, " "period_start, period_docs, credits_until FROM users WHERE email=?", (user["email"],)).fetchone() plan, until, started, credits, trial, status, pstart, pdocs, cuntil = \ row or ("free", None, None, 0, 0, "free", None, 0, None) if credits and cuntil and _parse(cuntil) <= _now(): # bought credits are valid CREDIT_MONTHS months with _db() as con: con.execute("UPDATE users SET doc_credits=0 WHERE email=?", (user["email"],)) con.execute("INSERT INTO plan_history (email, plan, started, until, event, note) VALUES (?,?,?,?,?,?)", (user["email"], "credits", None, cuntil, "credits expired", f"{credits} credit(s)")) credits = 0 if plan in PAID and until and _parse(until) <= _now(): expire_due(user["email"]) # month over, not renewed: switch it off now plan, status = "free", "expired" used, month_end = 0, None if plan in PAID: # documents used in the current subscription month t, changed = _month_window(pstart) used = 0 if changed else (pdocs or 0) if changed: with _db() as con: con.execute("UPDATE users SET period_start=?, period_docs=0 WHERE email=?", (_ts(t), user["email"])) month_end = _ts(min(add_month(t), _parse(until))) if until else _ts(add_month(t)) p = PLANS.get(plan, PLANS["free"]) return dict(key=plan if plan in PLANS else "free", **p, until=until if plan in PAID else None, started=started if plan in PAID else None, used=used, used_today=used, credits=credits or 0, credits_until=cuntil if credits else None, trial_left=max(0, FREE_TRIAL_DOCS - (trial or 0)), month_end=month_end, paid=plan in PAID or (credits or 0) > 0, admin=False, status="active" if plan in PAID else (status or "free")) def docs_available(st): """Credits the user can still spend: monthly allowance left + bought credits (+ free trial documents).""" if st["key"] in PAID: return max(0, st["docs_month"] - st["used"]) + st["credits"] return st["credits"] + st["trial_left"] # ---------------------------------------------------------------- checks (before a run) def _upgrade_msg(reason): return f"{reason} Add money to your wallet on the Pricing tab to continue." def _ist_day(): return time.strftime("%Y-%m-%d", time.gmtime(_now() + 330 * 60)) def _daily(email, tool): with _db() as con: r = con.execute("SELECT n FROM daily_usage WHERE email=? AND day=? AND tool=?", (email, _ist_day(), tool)).fetchone() return r[0] if r else 0 def _daily_add(email, tool): with _db() as con: k = con.execute("UPDATE daily_usage SET n=n+1 WHERE email=? AND day=? AND tool=?", (email, _ist_day(), tool)).rowcount if not k: con.execute("INSERT INTO daily_usage (email, day, tool, n) VALUES (?,?,?,1)", (email, _ist_day(), tool)) def _free_rerun(email, fkey): """True if this file was charged within RERUN_DAYS and still has free re-runs left.""" if not fkey: return False with _db() as con: r = con.execute("SELECT first_ts, runs FROM reruns WHERE email=? AND fkey=?", (email, fkey)).fetchone() return bool(r) and _parse(r[0]) + RERUN_DAYS * 86400 > _now() and (r[1] or 0) < FREE_RERUNS def _is_detector(tool): return "detector" in str(tool or "").lower() def check_doc(user, words=None, tool="", fkey=None): """Before a document run: plan / credits / trial / word cap. Raises A.Upsell with a friendly message.""" st = state(user) if st["admin"]: return e = (user or {}).get("email") if _is_detector(tool): # detector file reports: unlimited on plans, a few a day free if st["key"] in PAID or st["credits"] > 0: return if _daily(e, "detector") >= FREE_DETECT_DAY: if PAY_PER_USE: return "wallet" # beyond the free reports: paid from the wallet raise A.Upsell(f"The free account includes {FREE_DETECT_DAY} detector file reports a day. Paste text " f"for more checks today, or take the Scholar plan ({both('scholar')}/month) for " "unlimited reports.", kind="limit", title="Today's free reports are used") return if e and _free_rerun(e, fkey): return # free re-run of a file charged in the last days need = credits_needed(words) cap = st["doc_words"] if st["key"] in PAID or st["credits"] > 0 else PLANS["free"]["doc_words"] if st["key"] not in PAID and st["credits"] <= 0 and st["trial_left"] > 0: if words and words > cap: if PAY_PER_USE: return "wallet" # longer than the free trial covers: paid from the wallet raise A.Upsell(f"This document has {words:,} words; the free trial covers documents up to {cap:,} words. " f"A single credit ({both('single')}) covers up to {CREDIT_WORDS:,} words.", title="This document is longer than the free trial") return have = docs_available(st) if have < need and PAY_PER_USE: biggest = PLANS["scholar"]["doc_words"] if words and words > biggest: raise A.Upsell(f"This document has {words:,} words; the largest document is {biggest:,} words. " "Split it into chapters.", title="This document is very long") return "wallet" # no credits left: paid from the prepaid wallet (wallet.py) if have < need: if st["key"] in PAID: raise A.Upsell(f"This document needs {need} credit{'s' if need > 1 else ''} ({words or 0:,} words; one " f"credit = {CREDIT_WORDS:,} words) and you have {have} left this month. Add extra credits " f"at {both('extra')} each; your monthly credits start again on " f"{_ist_date(st['month_end'])}.", kind="extra", title="Monthly credits used up") msg = ("Your plan was not renewed, so it has been switched off. Renew it on the 💎 Pricing tab, or buy " f"credits ({both('single')} each, {both('pack')} for 5)." if st["status"] == "expired" else "Your subscription has been ended. Please contact us, or choose an option on the 💎 Pricing tab." if st["status"] == "terminated" else f"This document needs {need} credit{'s' if need > 1 else ''} (one credit = {CREDIT_WORDS:,} words). " f"Buy a single credit for {both('single')}, a pack of 5 for {both('pack')}, or take Scholar: " f"{PLANS['scholar']['docs_month']} credits for {both('scholar')}/month.") raise A.Upsell(msg, title="Your plan has ended" if st["status"] in ("expired", "terminated") else "Credits needed") biggest = st["doc_words"] if st["key"] in PAID else PLANS["scholar"]["doc_words"] # bought credits: any size if words and words > biggest: raise A.Upsell(f"This document has {words:,} words; the largest document is {biggest:,} words. " "Split it into chapters.", title="This document is very long") def doc_cost_text(user, words, tool=""): """'This run uses 2 credits (21,400 words)' for the upload area ('' when not applicable).""" if not words or _is_detector(tool): return "" st = state(user) if st["admin"]: return "" n = credits_needed(words) return f"This document ({words:,} words) uses {n} credit{'s' if n > 1 else ''}; you have {docs_available(st)}." def paste_limit(user): return state(user)["paste_words"] def check_paste(user, text): st = state(user) n = len((text or "").split()) if n > st["paste_words"]: if st["key"] not in PAID: raise A.Upsell(f"Free pastes are up to {st['paste_words']} words (yours: {n:,}). For longer text, put it " "in a Word file and upload it: it is paid from your wallet (pay as you go), and the usual " "price is shown before the run.", kind="wallet", title="Your text is longer than the free limit") raise A.Upsell(f"Your plan allows {st['paste_words']:,} words per paste (yours: {n:,}). Upload a .docx for " "longer text.", kind="limit", title="Your text is a bit long") def daily_paste_limit(user): return state(user)["pastes"] def max_passes(user, requested, document=False): """Smart Humanize passes: the plan's limit; the free trial document gets the full automatic passes.""" st = state(user) cap = st["passes"] if document and (st["credits"] > 0 or (st["key"] == "free" and st["trial_left"] > 0)): cap = 10 return int(min(int(requested), cap)) def doc_words(path): """Quick word count of an uploaded file (for credits and the word cap).""" try: p = str(path).lower() if p.endswith(".docx"): from docx import Document d = Document(path) n = sum(len(par.text.split()) for par in d.paragraphs) for t in d.tables: for r in t.rows: for c in r.cells: n += len(c.text.split()) return n if p.endswith(".pdf"): import pymupdf as fitz with fitz.open(path) as doc: return sum(len(pg.get_text().split()) for pg in doc) with open(path, encoding="utf-8", errors="ignore") as f: return len(f.read().split()) except Exception: return None # ---------------------------------------------------------------- consume (after a successful run) def consume_doc(user, words=None, tool="", fkey=None): """Charge a successful document run: free re-run if allowed; detector files only count on the free account; otherwise credits_needed(words) from the month's allowance, then bought credits, then the free trial.""" st = state(user) if st["admin"] or not user: return e = user["email"] if _is_detector(tool): if st["key"] not in PAID and st["credits"] <= 0: _daily_add(e, "detector") return with _db() as con: if fkey: r = con.execute("SELECT first_ts, runs FROM reruns WHERE email=? AND fkey=?", (e, fkey)).fetchone() if r and _parse(r[0]) + RERUN_DAYS * 86400 > _now() and (r[1] or 0) < FREE_RERUNS: con.execute("UPDATE reruns SET runs=COALESCE(runs,0)+1 WHERE email=? AND fkey=?", (e, fkey)) A.request_sync() return need = credits_needed(words) if st["key"] not in PAID and st["credits"] <= 0 and st["trial_left"] > 0: con.execute("UPDATE users SET trial_used=COALESCE(trial_used,0)+1 WHERE email=?", (e,)) else: from_month = min(need, max(0, st["docs_month"] - st["used"])) if st["key"] in PAID else 0 if from_month: con.execute("UPDATE users SET period_docs=COALESCE(period_docs,0)+? WHERE email=?", (from_month, e)) rest = need - from_month if rest: con.execute("UPDATE users SET doc_credits=MAX(0, COALESCE(doc_credits,0)-?) WHERE email=?", (rest, e)) if fkey: con.execute("INSERT OR REPLACE INTO reruns (email, fkey, first_ts, runs) VALUES (?,?,?,0)", (e, fkey, _ts(_now()))) A.request_sync() # ---------------------------------------------------------------- money in (admin today, Razorpay later) def activate_plan(email, plan, months=1, source="admin", amount="", note=""): """Start or renew a monthly plan. Renewed before the end date: the end date moves on by `months`. Renewed after it ended (or a new plan): a new subscription starts today.""" email = (email or "").strip().lower() months = max(1, int(months or 1)) if plan not in PLANS or plan in ("guest", "admin"): return False, f"Unknown plan {plan!r}." expire_due(email) with _db() as con: row = con.execute("SELECT plan, plan_until, plan_started FROM users WHERE email=?", (email,)).fetchone() if not row: return False, f"No user found with {email}." cur, until, started = row now = _now() if plan == "free": con.execute("UPDATE users SET plan='free', plan_until=NULL, plan_status='free' WHERE email=?", (email,)) con.execute("INSERT INTO plan_history (email, plan, started, until, event, note) VALUES (?,?,?,?,?,?)", (email, cur, started, until, "cancelled", note or source)) msg = f"{email} is now on the free plan." else: active = cur in PAID and until and _parse(until) > now if active: # on-time renewal / upgrade: continue from the end date start_t, keep_started, event = _parse(until), started or _ts(now), "renewed" if cur != plan: event = f"changed from {cur}" else: start_t, keep_started, event = now, _ts(now), "started" end_t = start_t for _ in range(months): end_t = add_month(end_t) if active: # same subscription continues: the monthly count runs on con.execute("UPDATE users SET plan=?, plan_until=?, plan_started=?, plan_status='active' WHERE email=?", (plan, _ts(end_t), keep_started, email)) else: # new subscription: a fresh month of documents con.execute("UPDATE users SET plan=?, plan_until=?, plan_started=?, period_start=?, period_docs=0, " "plan_status='active', trial_used=MAX(COALESCE(trial_used,0), ?) WHERE email=?", (plan, _ts(end_t), keep_started, _ts(start_t), FREE_TRIAL_DOCS, email)) con.execute("INSERT INTO plan_history (email, plan, started, until, event, note) VALUES (?,?,?,?,?,?)", (email, plan, _ts(start_t), _ts(end_t), event, note or source)) msg = (f"{email} is now on {PLANS[plan]['name']} ({event}): paid until {_ist_date(_ts(end_t))}. " f"Subscribed since {_ist_date(keep_started)}.") con.execute("INSERT INTO payments (email, item, amount, source, note) VALUES (?,?,?,?,?)", (email, f"{plan} {months} month(s)", amount, source, note)) con.execute("UPDATE upgrade_requests SET status='done' WHERE email=? AND status='open'", (email,)) A.request_sync() return True, msg def terminate_plan(email, reason="", notify=True, source="admin"): """Admin: end a paid subscription now (refund, misuse, request...). The user goes back to the free plan with status 'terminated'; the event is written to plan_history and, if `notify`, the user gets an email.""" email = (email or "").strip().lower() with _db() as con: row = con.execute("SELECT plan, plan_started, plan_until FROM users WHERE email=?", (email,)).fetchone() if not row: return False, f"No user found with {email}." plan, started, until = row if plan not in PAID: return False, f"{email} has no active paid plan to terminate." now = _ts(_now()) con.execute("UPDATE users SET plan='free', plan_until=?, plan_status='terminated', period_docs=0 " "WHERE email=?", (now, email)) con.execute("INSERT INTO plan_history (email, plan, started, until, event, note) VALUES (?,?,?,?,?,?)", (email, plan, started, now, "terminated", (reason or "").strip() or source)) A.request_sync() if notify: try: import mailer as M M.send_plan_terminated(email, PLANS[plan]["name"], (reason or "").strip()) except Exception as err: print(f"[plans] termination email skipped: {err}", flush=True) return True, (f"{PLANS[plan]['name']} plan of {email} terminated (was paid until {_ist_date(until)})." + (" The user was emailed." if notify else "")) def add_credits(email, n, source="admin", amount="", note=""): """Add bought credits (single, pack or extra); all of the user's bought credits are then valid CREDIT_MONTHS months from today.""" email = (email or "").strip().lower() t = _now() for _ in range(CREDIT_MONTHS): t = add_month(t) with _db() as con: k = con.execute("UPDATE users SET doc_credits=COALESCE(doc_credits,0)+?, credits_until=? WHERE email=?", (int(n), _ts(t), email)).rowcount if not k: return False, f"No user found with {email}." con.execute("INSERT INTO payments (email, item, amount, source, note) VALUES (?,?,?,?,?)", (email, f"{int(n)} credit(s)", amount, source, note)) con.execute("UPDATE upgrade_requests SET status='done' WHERE email=? AND status='open'", (email,)) A.request_sync() return True, f"Added {int(n)} credit(s) to {email}, valid until {_ist_date(_ts(t))}." def request_upgrade(user, item): """A signed-in user clicked Buy: log the request so the admin can follow up (until Razorpay is live).""" if not user: return with _db() as con: con.execute("INSERT INTO upgrade_requests (email, name, item) VALUES (?,?,?)", (user["email"], user.get("name", ""), item)) # ---------------------------------------------------------------- UI pieces def badge(user): st = state(user) if st["admin"]: return 'Admin · free' if st["key"] in PAID: return f'{html.escape(st["name"])}' if user: return 'Pay as you go' return "" def usage_line(user): st = state(user) if st["admin"]: return "Admin — everything is free and unlimited." if not user: return (f"Guest: {st['paste_words']}-word pastes, {st['pastes']} runs a day. Sign up free for a trial " f"document (up to {PLANS['free']['doc_words']:,} words) and {FREE_DETECT_DAY} detector reports a day.") parts = [] if st["key"] in PAID: parts.append(f"{st['name']} plan: {min(st['used'], st['docs_month'])}/{st['docs_month']} credits used this " f"month (new month starts {_ist_date(st['month_end'])}) · paid until {_ist_date(st['until'])}, " "renew by then to keep it") else: import wallet as W parts.append(f"Pay as you go: wallet {W.inr(W.balance(user.get('email')))}") if st["trial_left"] > 0: parts.append(f"{st['trial_left']} free trial document(s) left") if st["credits"]: parts.append(f"{st['credits']} credit(s) available until {_ist_date(st.get('credits_until'))} " f"(1 credit = 1 document up to {CREDIT_WORDS:,} words)") return " · ".join(parts) def nudge(user, document): """Shown under a finished run for guests and free users.""" st = state(user) if st["admin"] or st["key"] in PAID or st["credits"] > 0: return "" offer = "Pay as you go: add money to your wallet and pay only for what each run uses." if not user: msg = (f"Want to check a whole chapter? Sign up free for a trial document (up to " f"{PLANS['free']['doc_words']:,} words): layout kept, highlighted PDF report included.") elif st["trial_left"] > 0: msg = f"Your free trial document is waiting. {offer}" else: msg = f"Your free trial is used. {offer}" return f'
💳 {msg} Open the Pricing tab.
' BILLING = ("Monthly", f"Yearly (save {YEARLY_DISCOUNT:.0f}%)") def pricing_html(currency=None, billing="Monthly"): """The Pricing page: pay per use, credits, and the Scholar / Pro plans billed monthly or yearly.""" cur = _cur(currency) other = "USD" if cur == "INR" else "INR" where = "outside India" if other == "USD" else "in India" yearly = str(billing or "").lower().startswith("year") sc, pr = PLANS["scholar"], PLANS["pro"] common = ["Smart Humanize + AI detector PDF report", "Final check & proofread, done automatically", "Word layout, tables, citations and equations kept", f"{FREE_RERUNS} free re-runs of the same file within {RERUN_DAYS} days"] def card(title, price, per, alt, feats, featured=False, tag_text="Best value"): li = "".join(f"
  • {html.escape(f)}
    • " for f in feats) tag = f'
    {html.escape(tag_text)}
    ' if featured else "" return (f'
    {tag}
    ' f'{html.escape(title)}
    {price}{per}
    ' f'
    {alt}
    ') def plan_card(key, p, feats, tag_text): if yearly: # one payment for twelve months per_month = yearly_amount(key, cur) / 12 return card(p["name"], yearly_price(key, cur), "/year", f"{yearly_price(key, other)}/year {where} · works out at " f"{money(round(per_month), cur)}/month instead of {price_of(key, cur)}", [f"Save {YEARLY_DISCOUNT:.0f}%: 12 months for the price of " f"{12 * (1 - YEARLY_DISCOUNT / 100):.1f}".replace(".0", "")] + feats, featured=True, tag_text=tag_text) return card(p["name"], price_of(key, cur), "/month", f"{price_of(key, other)}/month {where}", feats + [f"Yearly: {yearly_price(key, cur)} a year (save {YEARLY_DISCOUNT:.0f}%)"], featured=True, tag_text=tag_text) cards = [] if PAY_PER_USE: low = money(PRICE_MIN_INR, "INR") if cur == "INR" else f"${PRICE_MIN_INR / PRICE_USD_INR:.2f}" cards.append(card("Pay per use", f"from {low}", "/document", "No plan and no credits needed", ["Run any document without buying anything first", "See your AI scores first; pay only to download the result", "Priced by what your document actually used (its length and the work it needed)", f"The exact price is shown before you pay and includes {PRICE_GST:.0f}% GST"] + common[:3])) cards += [ card("Single credit", price_of("single", cur), "/credit", f"{price_of('single', other)}/credit {where}", [f"1 document up to {CREDIT_WORDS:,} words", f"Valid {CREDIT_MONTHS} months", "Journal formatting of the same file included"] + common), card("Pack of 5", price_of("pack", cur), "/5 credits", f"{price_of('pack', other)}/5 credits {where}", ["5 documents (a full thesis)", f"Valid {CREDIT_MONTHS} months", "Journal formatting of each file " "included"] + common), plan_card("scholar", sc, [f"{sc['docs_month']} credits every month", f"Pasted text up to {sc['paste_words']:,} words, " "unlimited", "Unlimited AI detector reports", f"{sc['format_month']} journal-formatting files a " "month", f"Extra credits {price_of('extra', cur)} each"] + common, "Best value"), plan_card("pro", pr, [f"{pr['docs_month']} credits every month", f"Pasted text up to {pr['paste_words']:,} words, " "unlimited", "Unlimited AI detector reports", f"{pr['format_month']} journal-formatting files a " "month", "Priority queue", f"Extra credits {price_of('extra', cur)} each"] + common, "For heavy users"), ] promo = f'
    🎉 {html.escape(PROMO_TEXT)}
    ' if PROMO_TEXT else "" extra = (f'
    ' f'Free account: pasted text up to {PLANS["free"]["paste_words"]} words ({PLANS["free"]["pastes"]} a ' f'day), {FREE_TRIAL_DOCS} trial document up to {PLANS["free"]["doc_words"]:,} words, {FREE_DETECT_DAY} ' f'detector reports a day, 1 journal-formatting file a month. ' f'Monthly and yearly plans give the same credits every month; a yearly plan is paid once for ' f'twelve months. ' f'Journal Finder: free with a free account. Journal formatting on its own: ' f'{price_of("format", cur)} per file. Institutions and agencies: from ' f'{price_of("institution", cur)}/month for {PLANS["institution"]["docs_month"]} credits, several logins and ' f'a GST invoice, contact us.' + (f' {html.escape(INR_NOTE)}' if INR_NOTE and cur == "INR" else "") + "
    ") return (promo + '
    ' + "".join(cards) + "
    " + extra) BUY_ITEMS = ("Single credit", "Pack of 5 credits", "Scholar plan", "Pro plan", "Scholar plan (yearly)", "Pro plan (yearly)") def price_buttons(currency=None): """Labels of the buy buttons in the chosen currency (same order as BUY_ITEMS).""" cur = _cur(currency) return (f"Buy 1 credit — {price_of('single', cur)}", f"Buy 5 credits — {price_of('pack', cur)}", f"Scholar — {price_of('scholar', cur)}/month", f"Pro — {price_of('pro', cur)}/month", f"Scholar yearly — {yearly_price('scholar', cur)}/year", f"Pro yearly — {yearly_price('pro', cur)}/year") def pay_instructions(user, item): """Until online payment is live: how to pay + the request is logged for the admin.""" if A.is_admin(user): return '
    You are an admin — every feature is free for you.
    ' if not user: return ('
    Please sign in (or create a free account) first, so we can ' 'activate it on your account.
    ') request_upgrade(user, item) links = [] if PAY_WHATSAPP: txt = f"Hi, I want to buy: {item} for {user['email']}".replace(" ", "%20") links.append(f'' f'💬 Pay via WhatsApp / UPI') if PAY_EMAIL: links.append(f'✉️ Email us to pay') contact = "".join(links) or (f'
    Contact us through ' f'cssoftwaresolutions.in.
    ') return (f'
    {html.escape(item)} requested for ' f'{html.escape(user["email"])}. Online card payment is launching shortly; for now pay by UPI or ' f'bank transfer and it is switched on the same day.
    {contact}') # ---------------------------------------------------------------- pay per use # A signed-in customer without credits or a plan can still run a document. The run is priced by what it used: # cost = GPU time of the job x the Space's hourly price (Hugging Face Nvidia T4 small) # + the idle share of the Space (it runs all day; each busy second carries COST_IDLE_FACTOR - 1 idle ones) # + what the language model charged for this job (groqai's account of the job) # price = cost x (1 + PRICE_MARGIN %) + GST (PRICE_GST %) in rupees at PRICE_USD_INR, rounded up # The customer sees the scores at once; the result itself is locked until the charge is paid. The pop-up's button # opens PAY_URL (a payment page; {amount} {ref} {email} {item} are filled in). Until a payment gateway confirms # payments by itself, the admin marks a charge paid (Admin -> Plans & credits), which unlocks the result; a # gateway's webhook only has to call settle_charge(ref, source="razorpay", amount=...). PAY_PER_USE = os.getenv("PAY_PER_USE", "1") != "0" COST_GPU_HOUR_USD = _env_f("COST_GPU_HOUR_USD", 0.40) COST_IDLE_FACTOR = max(1.0, _env_f("COST_IDLE_FACTOR", 5)) PRICE_MARGIN = _env_f("PRICE_MARGIN", 400) # profit, percent of cost (as the wallet) PRICE_GST = _env_f("PRICE_GST", 18) PRICE_USD_INR = _env_f("PRICE_USD_INR", 85) PRICE_MIN_INR = _env_f("PRICE_MIN_INR", 49) # smallest charge (a payment below this is not worth its fee) PAY_URL = os.getenv("PAY_URL", "").strip() def usage_price(seconds, llm_usd): """Price of one job from its usage. Returns the full breakdown (US$) and the amount to pay (inr, usd).""" gpu = max(0.0, float(seconds or 0)) * COST_GPU_HOUR_USD / 3600 space = gpu * (COST_IDLE_FACTOR - 1) llm = max(0.0, float(llm_usd or 0)) cost = gpu + space + llm profit = cost * PRICE_MARGIN / 100 gst = (cost + profit) * PRICE_GST / 100 total = cost + profit + gst inr = max(PRICE_MIN_INR, math.ceil(total * PRICE_USD_INR)) return dict(seconds=round(float(seconds or 0), 1), gpu_usd=gpu, space_usd=space, llm_usd=llm, cost_usd=cost, profit_usd=profit, gst_usd=gst, total_usd=total, inr=int(inr), usd=round(inr / PRICE_USD_INR, 2)) def create_charge(email, tool, name, words, price, output): """Record a job whose result is locked until it is paid. Returns the charge reference (shown to the customer).""" import secrets ref = "CS-" + secrets.token_hex(4).upper() with _db() as con: con.execute("INSERT INTO charges (ref, email, tool, name, words, seconds, gpu_usd, space_usd, llm_usd, " "cost_usd, amount_inr, amount_usd, output) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)", (ref, (email or "").lower(), tool, name, int(words or 0), price["seconds"], price["gpu_usd"], price["space_usd"], price["llm_usd"], price["cost_usd"], price["inr"], price["usd"], str(output or ""))) A.request_sync() return ref _CHARGE = ("ref, ts, email, tool, name, words, seconds, gpu_usd, space_usd, llm_usd, cost_usd, amount_inr, amount_usd, " "status, output, paid_ts, source, note") def _charge_row(r): return dict(zip([c.strip() for c in _CHARGE.split(",")], r)) if r else None def charge_of(ref): with _db() as con: return _charge_row(con.execute(f"SELECT {_CHARGE} FROM charges WHERE ref=?", ((ref or "").strip().upper(),) ).fetchone()) def latest_charge(email, status=None): """The customer's most recent charge (optionally only 'pending' / 'paid').""" q, args = f"SELECT {_CHARGE} FROM charges WHERE email=?", [(email or "").lower()] if status: q += " AND status=?" args.append(status) with _db() as con: return _charge_row(con.execute(q + " ORDER BY id DESC LIMIT 1", args).fetchone()) def settle_charge(ref, source="admin", amount="", note=""): """Mark a charge paid: its result is unlocked for the customer. Called by the admin today, by the payment gateway's webhook later. Returns (ok, message).""" c = charge_of(ref) if not c: return False, f"No charge found with reference {ref!r}." if c["status"] == "paid": return True, f"{c['ref']} was already paid." with _db() as con: con.execute("UPDATE charges SET status='paid', paid_ts=?, source=?, note=? WHERE ref=?", (_ts(_now()), source, note, c["ref"])) con.execute("INSERT INTO payments (email, item, amount, source, note) VALUES (?,?,?,?,?)", (c["email"], f"pay per use {c['ref']} ({c['name']})", amount or f"₹{c['amount_inr']:.0f}", source, note)) A.request_sync() return True, f"{c['ref']} marked paid: {c['email']} can now download {c['name']}." def pay_link(charge, email=""): """Where the pop-up's Pay button goes: PAY_URL when set (payment page), else the WhatsApp / email contact.""" from urllib.parse import quote amount, ref = f"{charge['amount_inr']:.0f}", charge["ref"] item = f"{charge['tool']}: {charge['name']}" if PAY_URL: return (PAY_URL.replace("{amount}", amount).replace("{ref}", quote(ref)) .replace("{email}", quote(email or charge.get("email") or "")).replace("{item}", quote(item))) text = quote(f"Hi, I want to pay ₹{amount} for {ref} ({item}) on account {email or charge.get('email') or ''}") if PAY_WHATSAPP: return f"https://wa.me/{PAY_WHATSAPP}?text={text}" return f"mailto:{PAY_EMAIL}?subject={quote('Payment ' + ref)}&body={text}" CHARGE_COLS = ["When (UTC)", "Reference", "Email", "Document", "Words", "GPU s", "Cost $", "Price ₹", "Status"] def admin_charges(user): """Pay-per-use charges, newest first (pending ones are waiting for payment).""" A._need_admin(user) with _db() as con: rows = con.execute("SELECT ts, ref, email, name, words, seconds, cost_usd, amount_inr, status FROM charges " "ORDER BY status='paid', id DESC LIMIT 200").fetchall() return [[(r[0] or "")[:16], r[1], r[2], r[3], r[4], round(r[5] or 0), round(r[6] or 0, 4), round(r[7] or 0), r[8]] for r in rows] # ---------------------------------------------------------------- admin PAY_COLS = ["When (UTC)", "Email", "Item", "Amount", "Source", "Note"] REQ_COLS = ["When (UTC)", "Email", "Name", "Wants", "Status"] def admin_payments(user): A._need_admin(user) with _db() as con: rows = con.execute("SELECT ts, email, item, amount, source, note FROM payments ORDER BY id DESC LIMIT 200" ).fetchall() return [[(r[0] or "")[:16], *r[1:]] for r in rows] def admin_requests(user): A._need_admin(user) with _db() as con: rows = con.execute("SELECT ts, email, name, item, status FROM upgrade_requests ORDER BY id DESC LIMIT 200" ).fetchall() return [[(r[0] or "")[:16], *r[1:]] for r in rows] SUB_COLS = ["Email", "Plan", "Status", "Subscribed since", "Current month from", "Paid until", "Days left"] HIST_COLS = ["When (UTC)", "Email", "Plan", "From", "Until", "Event", "Note"] def admin_subscriptions(user): """Every paid / ended subscription: plan, start date, end date, status.""" A._need_admin(user) expire_due() now = _now() with _db() as con: rows = con.execute("SELECT email, plan, plan_status, plan_started, period_start, plan_until FROM users " "WHERE plan IN " + _PAID_SQL + " OR plan_status IN ('expired','terminated') " "ORDER BY plan_until DESC").fetchall() out = [] for e, plan, status, started, pstart, until in rows: left = max(0, int((_parse(until) - now) // 86400)) if until and plan in PAID else 0 out.append([e, PLANS.get(plan, PLANS["free"])["name"], (status or "").capitalize(), _ist_date(started), _ist_date(pstart), _ist_date(until), left if plan in PAID else "—"]) return out def admin_history(user): A._need_admin(user) with _db() as con: rows = con.execute("SELECT ts, email, plan, started, until, event, note FROM plan_history " "ORDER BY id DESC LIMIT 300").fetchall() return [[(r[0] or "")[:16], r[1], r[2], _ist_date(r[3]), _ist_date(r[4]), r[5], r[6]] for r in rows] def admin_revenue_tiles(user): """Paying users and estimated MRR for the admin overview.""" A._need_admin(user) now = _ts(_now()) with _db() as con: rows = con.execute("SELECT plan, COUNT(*) FROM users WHERE plan IN " + _PAID_SQL + " " "AND plan_until > ? GROUP BY plan", (now,)).fetchall() ended = con.execute("SELECT COUNT(*) FROM users WHERE plan_status='expired'").fetchone()[0] opens = con.execute("SELECT COUNT(*) FROM upgrade_requests WHERE status='open'").fetchone()[0] paying = sum(n for _, n in rows) mrr = sum(PLANS[p]["inr"] * n for p, n in rows) # list price in ₹ return [("Paying users", f"{paying:,}"), ("Monthly revenue (list price)", f"₹{mrr:,.0f}"), ("Plans ended (not renewed)", f"{ended:,}"), ("Open upgrade requests", f"{opens:,}")]