""" QuitSure Refund Eligibility Form - Backend FastAPI app that serves the form, checks DB eligibility, and logs to Google Sheets. """ import hashlib import logging import os from contextlib import contextmanager from datetime import date, datetime, timedelta from enum import Enum from typing import Optional import httpx import pymysql from dotenv import load_dotenv from fastapi import FastAPI, Request from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import HTMLResponse, JSONResponse from fastapi.staticfiles import StaticFiles from fastapi.templating import Jinja2Templates from pydantic import BaseModel, EmailStr, Field, validator from slowapi import Limiter from slowapi.errors import RateLimitExceeded from slowapi.util import get_remote_address from starlette.middleware.base import BaseHTTPMiddleware from starlette.responses import Response load_dotenv() # ========== LOGGING ========== logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", ) logger = logging.getLogger("refund-form") # ========== CONFIG ========== DB_CONFIG = { "host": os.environ.get("DB_HOST", ""), "user": os.environ.get("DB_USER", ""), "password": os.environ.get("DB_PASSWORD", ""), "database": os.environ.get("DB_NAME", "QuitSure_Production"), "charset": "utf8mb4", "connect_timeout": 10, "read_timeout": 15, } GOOGLE_SHEET_WEBHOOK_URL = os.environ.get("GOOGLE_SHEET_WEBHOOK_URL", "") GOOGLE_SHEET_SECRET = os.environ.get("GOOGLE_SHEET_SECRET", "") # ========== RATE LIMITING ========== limiter = Limiter(key_func=get_remote_address) # ========== SECURITY HEADERS MIDDLEWARE ========== class SecurityHeadersMiddleware(BaseHTTPMiddleware): async def dispatch(self, request: Request, call_next) -> Response: response = await call_next(request) response.headers["X-Content-Type-Options"] = "nosniff" response.headers["X-Frame-Options"] = "DENY" response.headers["X-XSS-Protection"] = "1; mode=block" response.headers["Referrer-Policy"] = "strict-origin-when-cross-origin" response.headers["Content-Security-Policy"] = ( "default-src 'self'; " "style-src 'self' 'unsafe-inline' https://fonts.googleapis.com; " "font-src 'self' https://fonts.gstatic.com; " "img-src 'self' data:; " "script-src 'self' 'unsafe-inline'; " "connect-src 'self';" ) return response # ========== APP ========== app = FastAPI( title="QuitSure Refund Form", docs_url=None, redoc_url=None, openapi_url=None ) app.state.limiter = limiter app.add_exception_handler(RateLimitExceeded, lambda req, exc: JSONResponse( status_code=429, content={"detail": "Too many requests. Please try again later."}, )) app.add_middleware(SecurityHeadersMiddleware) app.add_middleware( CORSMiddleware, allow_origins=[os.environ.get("ALLOWED_ORIGIN", "*")], allow_methods=["GET", "POST"], allow_headers=["Content-Type"], ) app.mount("/static", StaticFiles(directory="static"), name="static") templates = Jinja2Templates(directory="templates") # ========== DUPLICATE SUBMISSION TRACKING ========== SUBMISSIONS_FILE = os.path.join(os.path.dirname(__file__), "data", "submitted_emails.txt") os.makedirs(os.path.dirname(SUBMISSIONS_FILE), exist_ok=True) def _load_submitted_hashes() -> set: try: with open(SUBMISSIONS_FILE, "r") as f: return {line.strip() for line in f if line.strip()} except FileNotFoundError: return set() def _save_submitted_hash(email_hash: str) -> None: with open(SUBMISSIONS_FILE, "a") as f: f.write(email_hash + "\n") # ========== PYDANTIC MODELS ========== class PlatformEnum(str, Enum): android = "Android" ios = "iOS" web = "Web" class ProgramEnum(str, Enum): original = "original" relaxed = "relaxed" class YesNo(str, Enum): yes = "yes" no = "no" class RefundFormData(BaseModel): email: EmailStr name: str = Field(min_length=2, max_length=200) platform: PlatformEnum program: ProgramEnum purchaseDate: str = Field(max_length=10) completed: YesNo firstSub: YesNo accountActive: YesNo reason: str = Field(default="", max_length=5000) country: str = Field(default="", max_length=100) paypalEmail: str = Field(default="", max_length=200) formEligible: bool formRejectReasons: list = Field(default_factory=list) @validator("purchaseDate") @classmethod def validate_purchase_date(cls, v): try: d = datetime.strptime(v, "%Y-%m-%d").date() except ValueError: raise ValueError("Invalid date format, expected YYYY-MM-DD") if d > date.today(): raise ValueError("Purchase date cannot be in the future") if d < date(2020, 1, 1): raise ValueError("Purchase date is too far in the past") return v @validator("name") @classmethod def validate_name(cls, v): stripped = v.strip() if not any(c.isalpha() for c in stripped): raise ValueError("Name must contain at least one letter") return stripped # ========== DB HELPERS ========== @contextmanager def get_db_connection(): """Context manager for safe DB connections with automatic cleanup.""" conn = None try: conn = pymysql.connect(**DB_CONFIG, cursorclass=pymysql.cursors.DictCursor) yield conn finally: if conn: conn.close() def _json_safe(val): """Convert MySQL types (Decimal, datetime, date) to JSON-serializable values.""" from decimal import Decimal if val is None: return "" if isinstance(val, Decimal): return float(val) if isinstance(val, (datetime, date)): return val.isoformat() return val def sha256_hash(text: str) -> str: """Hash email the same way QuitSure does.""" return hashlib.sha256(text.lower().strip().encode()).hexdigest() def _to_date(val): """Parse a date value from MySQL into a Python date object.""" if val is None: return None if isinstance(val, date) and not isinstance(val, datetime): return val if isinstance(val, datetime): return val.date() if isinstance(val, str): return datetime.strptime(val[:10], "%Y-%m-%d").date() return None FREE_TRIAL_DAYS = {9: 7} # P9 = 7-day trial; all others default to 3 def lookup_user(email_hash: str = None, user_id: int = None) -> dict: """Look up user and return eligibility data. Key by hashed email OR by user_id.""" result = { "user_found": False, "user_id": None, "platform": None, "program": None, "program_id": None, "first_sub_date": None, "days_since_purchase": None, "account_deleted": None, "has_prior_refund": None, "program_completed": None, "days_completed": None, } try: with get_db_connection() as conn: with conn.cursor() as cursor: # ── 1. Find user by hashed email OR by user_id ── if user_id is not None: cursor.execute(""" SELECT /*+ MAX_EXECUTION_TIME(15000) */ iUserID, vDeviceType, bDeleted, bTestUser FROM tbl_Users WHERE iUserID = %s LIMIT 1 """, (user_id,)) else: cursor.execute(""" SELECT /*+ MAX_EXECUTION_TIME(15000) */ iUserID, vDeviceType, bDeleted, bTestUser FROM tbl_Users WHERE vHashedEmail = %s ORDER BY bTestUser ASC, bDeleted ASC, iUserID DESC LIMIT 1 """, (email_hash,)) user = cursor.fetchone() if not user: return result result["user_found"] = True result["user_id"] = user["iUserID"] result["account_deleted"] = bool(user["bDeleted"]) device = (user.get("vDeviceType") or "").lower() if "android" in device: result["platform"] = "Android" elif "ios" in device or "iphone" in device: result["platform"] = "iOS" else: result["platform"] = "Unknown" user_id = user["iUserID"] # ── 2. Get ALL UserPrograms rows ── cursor.execute(""" SELECT /*+ MAX_EXECUTION_TIME(15000) */ iUserProgramID, iProgramId, dFirstSubDate, bPreviouslySubscribed, bSubscribed, bRefund FROM tbl_UserPrograms WHERE iUserID = %s ORDER BY dFirstSubDate DESC """, (user_id,)) all_programs = cursor.fetchall() # ── 2a. Check prior refund across ALL programs ── result["has_prior_refund"] = False for prog in all_programs: b = prog.get("bRefund") if b and str(b) != "0": result["has_prior_refund"] = True result["refund_program"] = prog["iProgramId"] break # Pick the most recent program with a subscription date subs_with_date = [p for p in all_programs if p.get("dFirstSubDate")] sub = subs_with_date[0] if subs_with_date else None if sub: pid = sub["iProgramId"] result["program_id"] = pid result["program"] = "Relaxed" if pid == 9 else "Original" # ── 2b. bPreviouslySubscribed gate ── # If 0, the payment failed (SubHistory may have decPrice>0 anyway) if not sub.get("bPreviouslySubscribed"): result["no_payment_found"] = True result["date_source"] = "bPreviouslySubscribed=0 (payment not completed)" else: # ── 2c. Get ALL SubHistory rows ── user_program_id = sub["iUserProgramID"] cursor.execute(""" SELECT /*+ MAX_EXECUTION_TIME(15000) */ dStartDate, dExpiryDate, decPrice, vCurrency, vProductId, vLabel, iRenewed, bRestore FROM UserSubHistory WHERE iUserProgramID = %s ORDER BY dStartDate ASC """, (user_program_id,)) all_sub_rows = cursor.fetchall() # Classify rows into paid periods vs free paid_periods = [] for row in all_sub_rows: price = row.get("decPrice") product = (row.get("vProductId") or "").lower() label = (row.get("vLabel") or "").lower() is_free_product = ( "ft" in product or "free" in label or "trial" in label ) # decPrice = 0 → free trial / restore if price is not None and float(price) == 0: continue # decPrice IS NULL + free trial product → legacy free if is_free_product and price is None: continue # Everything else is a real subscription period: # decPrice > 0 (paid) or decPrice IS NULL (legacy paid) paid_periods.append(row) if not paid_periods: # bPreviouslySubscribed=1 but no real paid rows result["no_payment_found"] = True result["date_source"] = "No paid rows in SubHistory" else: # ── Repeat subscriber check ── if len(paid_periods) >= 2: result["is_repeat_subscriber"] = True result["total_subscription_periods"] = len(paid_periods) # Use latest paid row for current refund assessment latest = paid_periods[-1] # ── Payment date ── start = _to_date(latest["dStartDate"]) if start: # Android free trial: dStartDate = trial start, not payment if (latest.get("vLabel") or "").lower() == "free-trial": trial_days = FREE_TRIAL_DAYS.get(pid, 3) pay_date = start + timedelta(days=trial_days) result["date_source"] = ( "SubHistory +{}d trial adjustment".format(trial_days) ) else: pay_date = start result["date_source"] = "SubHistory payment" result["first_sub_date"] = pay_date.isoformat() result["days_since_purchase"] = ( date.today() - pay_date ).days if latest.get("decPrice") is not None: result["payment_amount"] = str(latest["decPrice"]) result["payment_currency"] = latest.get("vCurrency") or "" # ── Product ID + duration ── product_id = ( latest.get("vProductId") or latest.get("vLabel") or "" ) result["product_id"] = product_id cursor.execute(""" SELECT /*+ MAX_EXECUTION_TIME(15000) */ vDuration FROM tbl_ProductCodes WHERE bActive = 1 AND (vIosSubId = %s OR vAndroidSubId = %s OR vAndroidProdId = %s OR vWebId = %s OR vRazorId = %s) LIMIT 1 """, (product_id,) * 5) dur_row = cursor.fetchone() duration = ( (dur_row["vDuration"] if dur_row else None) or "mo" ) result["sub_duration"] = duration duration_days = { "wk": 7, "mo": 30, "6mo": 180 }.get(duration, 30) # ── Renewal detection via sub_length ── expiry = _to_date(latest.get("dExpiryDate")) raw_start = _to_date(latest["dStartDate"]) if expiry and raw_start: sub_length = (expiry - raw_start).days result["is_renewal"] = ( sub_length > duration_days * 1.5 ) else: result["is_renewal"] = False # ── 3. Program completion + speed detection ── pid = result["program_id"] if pid and 1 <= pid <= 10: user_tbl = "tbl_User{}Chapters".format(pid) master_tbl = "tbl_{}Chapters".format(pid) required = 42 if pid == 9 else 6 cursor.execute(""" SELECT /*+ MAX_EXECUTION_TIME(15000) */ DISTINCT iDayId FROM {} WHERE bDeleted = 0 ORDER BY iDayId ASC LIMIT %s """.format(master_tbl), (required,)) core_days = [row["iDayId"] for row in cursor.fetchall()] if core_days: placeholders = ",".join(["%s"] * len(core_days)) cursor.execute(""" SELECT /*+ MAX_EXECUTION_TIME(15000) */ COUNT(DISTINCT ch.iDayId) as days_completed, COUNT(*) as total_chapters, SUM(CASE WHEN TIMESTAMPDIFF(SECOND, uc.dDateCreated, uc.dDateUpdated) < 10 THEN 1 ELSE 0 END) as speed_chapters, ROUND(AVG(TIMESTAMPDIFF(SECOND, uc.dDateCreated, uc.dDateUpdated))) as avg_seconds FROM {} uc JOIN {} ch ON uc.iChapterID = ch.iChapterId WHERE uc.iUserID = %s AND uc.bCompleted = 1 AND ch.iDayId IN ({}) """.format(user_tbl, master_tbl, placeholders), (user_id, *core_days)) progress = cursor.fetchone() days_done = progress["days_completed"] if progress else 0 result["program_completed"] = days_done >= required result["days_completed"] = days_done if progress and progress["total_chapters"]: total = progress["total_chapters"] speed = progress["speed_chapters"] or 0 result["speed_chapters"] = speed result["total_chapters"] = total result["avg_seconds_per_chapter"] = progress["avg_seconds"] result["suspected_speed_completion"] = ( total >= 5 and (speed / total) > 0.5 ) # ── 4. Duplicate accounts — only flag PAID ones ── cursor.execute(""" SELECT /*+ MAX_EXECUTION_TIME(15000) */ vUDID FROM tbl_Users WHERE iUserID = %s """, (user_id,)) udid_row = cursor.fetchone() udid = udid_row.get("vUDID", "") if udid_row else "" if (udid and len(udid) >= 14 and udid != "-" and udid != "0000000000000000"): cursor.execute(""" SELECT /*+ MAX_EXECUTION_TIME(15000) */ u.iUserID, upr.bRefund FROM tbl_Users u JOIN tbl_UserPrograms upr ON u.iUserID = upr.iUserID WHERE u.vUDID = %s AND u.iUserID != %s AND u.bTestUser = 0 AND upr.bPreviouslySubscribed = 1 """, (udid, user_id)) paid_dupes = cursor.fetchall() result["other_accounts_same_device"] = len( set(r["iUserID"] for r in paid_dupes) ) for d in paid_dupes: b = d.get("bRefund") if b and str(b) != "0": result["other_account_refunded"] = True break except pymysql.Error as e: logger.error("Database error during user lookup: %s", e) except Exception as e: logger.error("Unexpected error during user lookup: %s", e) return result # ========== ELIGIBILITY ========== REFUND_WINDOW = {"Original": 21, "Relaxed": 60} REQUIRED_DAYS = {"Original": 6, "Relaxed": 42} def calculate_db_eligibility(db_data: dict) -> dict: """Calculate eligibility based on DB data.""" reasons = [] if not db_data["user_found"]: return {"eligible": None, "reasons": ["User not found in database"]} if db_data.get("no_payment_found"): reasons.append("No payment record found — user may be on free trial or B2B provision") if db_data.get("is_repeat_subscriber"): n = db_data.get("total_subscription_periods", 2) reasons.append( f"Repeat subscriber: {n} subscription periods found. " f"Money-back guarantee applies only to the first subscription." ) if db_data.get("account_deleted"): reasons.append("Account is deleted in database") if db_data.get("has_prior_refund"): prog = db_data.get("refund_program") extra = f" (Program {prog})" if prog else "" reasons.append(f"User has a prior refund on record{extra}") if db_data.get("other_account_refunded"): reasons.append( "Another account on the same device already received a refund" ) if db_data["days_since_purchase"] is not None: max_days = REFUND_WINDOW.get(db_data["program"], 21) if db_data["days_since_purchase"] > max_days: reasons.append( f"The refund request must be made within {max_days} days of purchasing the subscription. " f"It has been {db_data['days_since_purchase']} days since purchase." ) if db_data.get("program_completed") is not True: days_done = db_data.get("days_completed", 0) required = REQUIRED_DAYS.get(db_data["program"], 6) reasons.append( f"Program not completed: {days_done}/{required} days done" ) if db_data.get("is_renewal"): duration = db_data.get("sub_duration") or "unknown" reasons.append( f"User is past first billing cycle (subscription: {duration}). " f"Renewals are not eligible for refund." ) if db_data.get("other_accounts_same_device", 0) > 0: count = db_data["other_accounts_same_device"] reasons.append( f"Found {count} other paid account(s) on the same device. " f"Possible duplicate — coach should verify." ) if db_data.get("suspected_speed_completion"): speed = db_data.get("speed_chapters", 0) total = db_data.get("total_chapters", 0) avg = db_data.get("avg_seconds_per_chapter", 0) reasons.append( f"Suspected speed-completion: {speed}/{total} chapters done in <10 sec " f"(avg {avg}s per chapter). Coach should verify." ) return {"eligible": len(reasons) == 0, "reasons": reasons} # ========== GOOGLE SHEET ========== async def write_to_google_sheet( form_data: RefundFormData, db_data: dict, db_eligibility: dict ) -> dict: """Send data to Google Sheet via Apps Script webhook (async).""" if not GOOGLE_SHEET_WEBHOOK_URL: logger.warning("Google Sheet webhook URL not configured, skipping") return {"status": "skipped", "reason": "No webhook URL configured"} payload = { "secret": GOOGLE_SHEET_SECRET, "timestamp": datetime.now().isoformat(), # User-reported fields "email": form_data.email, "name": form_data.name, "platform_reported": form_data.platform.value, "program_reported": form_data.program.value, "purchase_date_reported": form_data.purchaseDate, "completed_reported": form_data.completed.value, "first_sub_reported": form_data.firstSub.value, "account_active_reported": form_data.accountActive.value, "reason": form_data.reason, "country": form_data.country, "paypal_email": form_data.paypalEmail, "form_eligible": form_data.formEligible, "form_reject_reasons": "; ".join(form_data.formRejectReasons), # DB-verified fields "db_user_found": db_data.get("user_found", ""), "db_user_id": db_data.get("user_id", ""), "db_platform": db_data.get("platform", ""), "db_program": db_data.get("program", ""), "db_first_sub_date": db_data.get("first_sub_date", ""), "db_days_since_purchase": db_data.get("days_since_purchase", ""), "db_program_completed": db_data.get("program_completed", ""), "db_days_completed": db_data.get("days_completed", ""), "db_prior_refund": db_data.get("has_prior_refund", ""), "db_account_deleted": db_data.get("account_deleted", ""), "db_no_payment": db_data.get("no_payment_found", False), "db_payment_amount": db_data.get("payment_amount", ""), "db_payment_currency": db_data.get("payment_currency", ""), "db_date_source": db_data.get("date_source", ""), "db_product_id": db_data.get("product_id", ""), "db_sub_duration": db_data.get("sub_duration", ""), "db_is_renewal": db_data.get("is_renewal", False), "db_is_repeat_subscriber": db_data.get("is_repeat_subscriber", False), "db_total_sub_periods": db_data.get("total_subscription_periods", ""), "db_other_account_refunded": db_data.get("other_account_refunded", False), "db_other_accounts_same_device": db_data.get("other_accounts_same_device", 0), "db_speed_chapters": db_data.get("speed_chapters", ""), "db_total_chapters": db_data.get("total_chapters", ""), "db_avg_sec_per_chapter": db_data.get("avg_seconds_per_chapter", ""), "db_suspected_speed": db_data.get("suspected_speed_completion", False), "db_eligible": db_eligibility.get("eligible", ""), "db_reject_reasons": "; ".join(db_eligibility.get("reasons", [])), # Comparison "user_lied": _check_discrepancies(form_data, db_data), } # Convert any Decimal/datetime values to JSON-safe types payload = {k: _json_safe(v) for k, v in payload.items()} try: async with httpx.AsyncClient(follow_redirects=True) as client: resp = await client.post( GOOGLE_SHEET_WEBHOOK_URL, json=payload, timeout=15.0 ) logger.info("Google Sheet write: status=%d", resp.status_code) return {"status": "success", "code": resp.status_code} except httpx.TimeoutException: logger.error("Google Sheet webhook timed out") return {"status": "error", "reason": "Webhook timeout"} except Exception as e: logger.error("Google Sheet webhook error: %s", e) return {"status": "error", "reason": "Webhook failed"} def _check_discrepancies(form_data: RefundFormData, db_data: dict) -> str: """Compare user-reported vs DB values. Return discrepancies.""" discrepancies = [] if not db_data.get("user_found"): return "User not found in DB" # Platform mismatch form_platform = form_data.platform.value.lower() db_platform = (db_data.get("platform") or "").lower() if form_platform != db_platform and db_platform != "unknown": discrepancies.append( f"Platform: said {form_data.platform.value}, DB says {db_data.get('platform')}" ) # Program mismatch form_program = form_data.program.value db_program = (db_data.get("program") or "").lower() if form_program != "unsure" and form_program != db_program: discrepancies.append( f"Program: said {form_program}, DB says {db_data.get('program')}" ) # Completion mismatch if form_data.completed == YesNo.yes and db_data.get("program_completed") is False: discrepancies.append( f"Completion: said completed, DB says {db_data.get('days_completed', 0)} days done" ) # First sub mismatch if form_data.firstSub == YesNo.yes and db_data.get("has_prior_refund"): discrepancies.append("First sub: said yes, DB shows prior refund") if form_data.firstSub == YesNo.yes and db_data.get("is_repeat_subscriber"): discrepancies.append( "First sub: said yes, DB shows {} subscription periods".format( db_data.get("total_subscription_periods", "2+") ) ) # Account active mismatch if form_data.accountActive == YesNo.yes and db_data.get("account_deleted"): discrepancies.append("Account: said active, DB shows deleted") return "; ".join(discrepancies) if discrepancies else "No discrepancies" # ========== AUTO REFUND ELIGIBILITY (proactive daily list) ========== # Offer (live 2026-06-24): high-intent (vSubIntent high/medium-high) EVEN users who # complete all 6 core days of Program 3 within 6x24h of purchase get a full auto-refund. # This endpoint returns ALL users who have ever met the criteria since launch; the # Google Sheet de-dupes against rows already present, so no one is ever missed or doubled. ELIGIBLE_SECRET = os.environ.get("ELIGIBLE_SECRET", GOOGLE_SHEET_SECRET) OFFER_LAUNCH_DATE = "2026-06-24" CORE_CHAPTER_COUNT = 36 # active Days 1-6 chapters for even users (iABTest IN (0,1)) ELIGIBLE_SQL = """ SELECT /*+ MAX_EXECUTION_TIME(15000) */ u.iUserID AS user_id, upf.vName AS name, CASE WHEN LOWER(u.vDeviceType) LIKE '%%android%%' THEN 'Android' WHEN LOWER(u.vDeviceType) LIKE '%%ios%%' OR LOWER(u.vDeviceType) LIKE '%%iphone%%' THEN 'iOS' ELSE 'Other' END AS platform, uc.vSubIntent AS sub_intent, upr.dFirstSubDate AS purchase_date, comp.last_completion AS completed_at, ROUND(TIMESTAMPDIFF(HOUR, upr.dFirstSubDate, comp.last_completion)) AS hrs_to_complete, pay.vSubscriptionId AS payment_ref FROM tbl_Users u JOIN tbl_UserConfig uc ON uc.iUserID = u.iUserID LEFT JOIN tbl_UserProfile upf ON upf.iUserID = u.iUserID JOIN tbl_UserPrograms upr ON upr.iUserID = u.iUserID AND upr.iProgramId = 3 -- completion gate: all core chapters (Days 1-6, even-variant) completed JOIN ( SELECT uc3.iUserID, MAX(uc3.dDateUpdated) AS last_completion, COUNT(DISTINCT uc3.iChapterID) AS done FROM tbl_User3Chapters uc3 JOIN tbl_3Chapters c3 ON c3.iChapterId = uc3.iChapterID AND c3.iDayId BETWEEN 1 AND 6 AND c3.bDeleted = 0 AND c3.bActive = 1 AND c3.iABTest IN (0,1) WHERE uc3.bCompleted = 1 GROUP BY uc3.iUserID HAVING done = %s ) comp ON comp.iUserID = u.iUserID -- paid gate: real payment exists (excludes free-trial-only); pulls amount + payment reference JOIN ( SELECT iUserProgramID, vSubscriptionId, decPrice, vCurrency, ROW_NUMBER() OVER (PARTITION BY iUserProgramID ORDER BY dStartDate DESC) rn FROM UserSubHistory WHERE decPrice > 0 AND (bRestore = 0 OR bRestore IS NULL) ) pay ON pay.iUserProgramID = upr.iUserProgramID AND pay.rn = 1 WHERE u.iProgramID = 3 AND u.bDeleted = 0 AND u.bTestUser = 0 AND u.iUserID %% 2 = 0 -- even users AND uc.vSubIntent IN ('high','medium-high') -- high intent AND upr.bPreviouslySubscribed = 1 -- paid (gate); prior refund is now an info column AND comp.last_completion >= upr.dFirstSubDate -- finished after buying AND comp.last_completion <= upr.dFirstSubDate + INTERVAL 6 DAY -- within 6x24h AND upr.dFirstSubDate >= %s -- on/after offer launch ORDER BY comp.last_completion ASC """ def get_auto_eligible(since_date: str) -> list: """Return all users eligible for the automatic refund since `since_date`. Gate = paid + completed 6 days within 6 days (ELIGIBLE_SQL). Every row is then enriched with the SAME informational checks the refund form computes (via lookup_user), so the sheet's columns always match the form. These checks are informational only — they never remove anyone from the list. """ with get_db_connection() as conn: with conn.cursor() as cursor: cursor.execute(ELIGIBLE_SQL, (CORE_CHAPTER_COUNT, since_date)) rows = cursor.fetchall() out = [] for r in rows: row = {k: _json_safe(v) for k, v in r.items()} info = lookup_user(user_id=r["user_id"]) # same checks as the refund form row.update({ "no_payment_found": info.get("no_payment_found", False), "payment_amount": info.get("payment_amount", ""), "payment_currency": info.get("payment_currency", ""), "product_id": info.get("product_id", ""), "sub_duration": info.get("sub_duration", ""), "is_repeat_subscriber": info.get("is_repeat_subscriber", False), "total_subscription_periods": info.get("total_subscription_periods", ""), "has_prior_refund": info.get("has_prior_refund", False), "refund_program": info.get("refund_program", ""), "other_account_refunded": info.get("other_account_refunded", False), "other_accounts_same_device": info.get("other_accounts_same_device", 0), "account_deleted": info.get("account_deleted", False), "days_since_purchase": info.get("days_since_purchase", ""), "days_completed": info.get("days_completed", ""), "is_renewal": info.get("is_renewal", False), "suspected_speed_completion": info.get("suspected_speed_completion", False), "speed_chapters": info.get("speed_chapters", ""), "total_chapters": info.get("total_chapters", ""), "avg_seconds_per_chapter": info.get("avg_seconds_per_chapter", ""), }) out.append({k: _json_safe(v) for k, v in row.items()}) return out # ========== ROUTES ========== @app.get("/", response_class=HTMLResponse) async def serve_form(request: Request) -> HTMLResponse: """Serve the refund eligibility form.""" return templates.TemplateResponse("index.html", {"request": request}) @app.get("/health") async def health_check() -> dict: """Health check endpoint for load balancers and monitoring.""" db_ok = False try: with get_db_connection() as conn: with conn.cursor() as cursor: cursor.execute("SELECT 1") db_ok = True except Exception: pass return {"status": "healthy" if db_ok else "degraded", "db": db_ok} @app.get("/eligible") @limiter.limit("30/minute") async def auto_eligible(request: Request, secret: str = "", since: str = "") -> JSONResponse: """Return all users eligible for the automatic refund (proactive daily list). Called by the Google Sheet button. De-duplication happens in the Sheet, so this always returns the FULL eligible set since `since` (default = offer launch date). """ if secret != ELIGIBLE_SECRET: return JSONResponse(status_code=401, content={"status": "error", "detail": "Unauthorized"}) since_date = since or OFFER_LAUNCH_DATE try: datetime.strptime(since_date, "%Y-%m-%d") except ValueError: return JSONResponse(status_code=422, content={"status": "error", "detail": "since must be YYYY-MM-DD"}) try: users = get_auto_eligible(since_date) except Exception as e: logger.error("Auto-eligible query failed: %s", e) return JSONResponse(status_code=500, content={"status": "error", "detail": "Query failed"}) return JSONResponse({"status": "success", "count": len(users), "since": since_date, "users": users}) @app.post("/submit-refund") @limiter.limit("5/minute") async def submit_refund(request: Request) -> JSONResponse: """Process refund form submission with DB verification and Sheet logging.""" try: body = await request.json() form_data = RefundFormData(**body) except Exception as e: logger.warning("Invalid form submission: %s", e) return JSONResponse( status_code=422, content={"status": "error", "detail": "Invalid form data"}, ) email_hash = sha256_hash(form_data.email) logger.info("Refund submission received for email hash: %s", email_hash[:12]) # Duplicate submission check if email_hash in _load_submitted_hashes(): return JSONResponse( status_code=409, content={ "status": "duplicate", "detail": ( "A refund request has already been submitted for this email. " "Only the first submission will be considered. " "If you need to update your request, please email hello@quitsure.app." ), }, ) # Look up in DB db_data = lookup_user(email_hash) db_eligibility = calculate_db_eligibility(db_data) # Write to Google Sheet (async, non-blocking) sheet_result = await write_to_google_sheet(form_data, db_data, db_eligibility) # Record this email as submitted _save_submitted_hash(email_hash) return JSONResponse({ "status": "success", "form_eligible": form_data.formEligible, "db_eligible": db_eligibility.get("eligible"), "sheet_status": sheet_result.get("status"), }) # ========== RUN ========== if __name__ == "__main__": import uvicorn uvicorn.run( "app:app", host="0.0.0.0", port=int(os.environ.get("PORT", 8000)), reload=os.environ.get("ENV", "production") != "production", )