kussssh/IPO-Analyzer
0
1"""2FastAPI surface for the DRHP-first IPO analyzer.3"""4from __future__ import annotations5 6import asyncio7import concurrent.futures8import json9import os10import re11import shutil12import threading13import traceback14import urllib.error15import urllib.parse16import urllib.request17import uuid18from datetime import datetime, timezone19from pathlib import Path20from typing import Any, Dict21 22from fastapi import FastAPI, File, HTTPException, UploadFile23from fastapi.middleware.cors import CORSMiddleware24from fastapi.responses import FileResponse25from fastapi.staticfiles import StaticFiles26 27from backend.company_pipeline import (28 ANALYSIS_CACHE_VERSION,29 MODULE_ORDER,30 analyze_company_document,31 hydrate_company_documents,32 prewarm_company,33)34from backend.config import (35 BACKGROUND_PRECOMPUTE_LIMIT,36 BACKGROUND_PREP_WORKERS,37 CACHE_DB_PATH,38 CHROMA_BASE_DIR,39 ENABLE_RECENT_COMPANY_PRECOMPUTE,40 HF_CACHE_ROOT,41 OUTPUT_DIR,42 RUNTIME_ROOT,43 UPLOAD_DIR,44)45from backend.database import (46 get_company,47 get_analysis,48 get_company_row,49 get_document_record,50 init_db,51 list_cached_analyses,52 list_companies,53 purge_invalid_catalog_rows,54 upsert_companies,55 get_company_sector,56 save_company_sector,57 get_connection,58 reset_analysis_state,59 update_document_record,60)61from backend.embeddings import log_embedding_runtime_config62from backend.pipeline import run_full_analysis63from backend.sebi import get_latest_filing_highlights, refresh_drhp_company_catalog64 65# Ensure database tables exist automatically on startup66init_db()67log_embedding_runtime_config()68 69# Pre-warm the CrossEncoder reranker in the background so the first analysis70# request doesn't pay the 2–5s model-load cold start cost.71def _prewarm_reranker() -> None:72 try:73 from backend.retriever import load_reranker74 load_reranker()75 print("✅ Reranker pre-warmed and ready.")76 except Exception as exc:77 print(f"⚠️ Reranker pre-warm failed (will load on demand): {exc}")78 79threading.Thread(target=_prewarm_reranker, daemon=True, name="reranker-prewarm").start()80 81# Pre-warm the embedding model in the background.82# For local sentence-transformers: downloads+caches the 90MB model at boot83# so the first analysis doesn't trigger a cold-start download mid-session.84def _prewarm_embeddings() -> None:85 try:86 from backend.embeddings import create_embeddings_client87 client = create_embeddings_client()88 # Run a tiny test embed so the model is fully loaded into memory89 client.embed_query("warmup")90 print("✅ Embedding model pre-warmed and ready.")91 except Exception as exc:92 print(f"⚠️ Embedding pre-warm failed (will load on demand): {exc}")93 94threading.Thread(target=_prewarm_embeddings, daemon=True, name="embeddings-prewarm").start()95 96FRONTEND_DIR = Path(__file__).resolve().parent.parent / "frontend"97 98app = FastAPI(title="IPO Analyzer API", version="3.0.0")99app.add_middleware(100 CORSMiddleware,101 allow_origins=["*"],102 allow_credentials=True,103 allow_methods=["*"],104 allow_headers=["*"],105)106 107if FRONTEND_DIR.exists():108 app.mount("/static", StaticFiles(directory=str(FRONTEND_DIR)), name="static")109 110# Static file serving is removed for /results so that the dynamic API routes111# can accurately fetch historical analysis data from the SQLite database.112 113 114jobs: Dict[str, Dict[str, Any]] = {}115jobs_lock = threading.Lock()116active_analysis_jobs: Dict[tuple[int, str], str] = {}117background_precompute_executor = concurrent.futures.ThreadPoolExecutor(118 max_workers=max(1, BACKGROUND_PREP_WORKERS),119 thread_name_prefix="ipo-precompute",120)121background_precompute_jobs: set[tuple[int, str]] = set()122background_precompute_lock = threading.Lock()123catalog_refresh_state: Dict[str, Any] = {124 "status": "idle",125 "message": "Catalog not refreshed yet",126 "last_count": 0,127 "last_error": None,128 "started_at": None,129 "finished_at": None,130}131filing_highlights_cache: Dict[str, Any] = {132 "data": {"drhp": [], "rhp": [], "prospectus": []},133 "fetched_at": None,134}135MODULE_TITLES = {136 "business": "Business Quality",137 "financials": "Revenue & Profitability",138 "growth_quality": "Growth vs Profitability",139 "valuation": "Valuation vs Peers",140 "promoter_ofs": "Promoter & OFS",141 "use_of_proceeds": "Use of Proceeds",142 "risks": "Risk Factors",143 "institutional": "Anchor & Institutional",144}145 146DOCUMENT_STAGE_MESSAGES = {147 "cataloged": "Document located in catalog",148 "resolving_pdf": "Resolving PDF link",149 "downloading": "Downloading document PDF",150 "re_downloading": "Detected corrupted cached PDF, re-downloading",151 "parsing": "Parsing document pages",152 "chunking": "Chunking extracted content",153 "saving_chunks": "Saving document chunks",154 "indexing": "Building retrieval store",155 "ready": "Document indexed and ready",156 "missing": "Document not available",157 "error": "Document processing failed",158}159STALE_ANALYSIS_MINUTES = 15160 161 162def _safe_count_dir(path: Path) -> int | None:163 try:164 if not path.exists():165 return 0166 return sum(1 for _ in path.iterdir())167 except Exception:168 return None169 170 171def _safe_size_bytes(path: Path) -> int | None:172 try:173 if not path.exists():174 return 0175 if path.is_file():176 return path.stat().st_size177 total = 0178 for child in path.rglob("*"):179 if child.is_file():180 total += child.stat().st_size181 return total182 except Exception:183 return None184 185 186def _clear_runtime_directory(path: Path) -> int:187 if not path.exists():188 return 0189 removed = 0190 for child in path.iterdir():191 if child.is_dir():192 shutil.rmtree(child, ignore_errors=True)193 removed += 1194 else:195 child.unlink(missing_ok=True)196 removed += 1197 return removed198 199 200def _runtime_storage_diagnostics() -> Dict[str, Any]:201 return {202 "runtime_root": str(RUNTIME_ROOT),203 "runtime_root_exists": RUNTIME_ROOT.exists(),204 "data_mount_exists": Path("/data").exists(),205 "db_path": str(CACHE_DB_PATH),206 "db_exists": CACHE_DB_PATH.exists(),207 "db_size_bytes": _safe_size_bytes(CACHE_DB_PATH),208 "uploads_path": str(UPLOAD_DIR),209 "uploads_exists": UPLOAD_DIR.exists(),210 "uploads_entries": _safe_count_dir(UPLOAD_DIR),211 "uploads_size_bytes": _safe_size_bytes(UPLOAD_DIR),212 "outputs_path": str(OUTPUT_DIR),213 "outputs_exists": OUTPUT_DIR.exists(),214 "outputs_entries": _safe_count_dir(OUTPUT_DIR),215 "outputs_size_bytes": _safe_size_bytes(OUTPUT_DIR),216 "chroma_path": str(CHROMA_BASE_DIR),217 "chroma_exists": CHROMA_BASE_DIR.exists(),218 "chroma_entries": _safe_count_dir(CHROMA_BASE_DIR),219 "chroma_size_bytes": _safe_size_bytes(CHROMA_BASE_DIR),220 "hf_cache_path": str(HF_CACHE_ROOT),221 "hf_cache_exists": HF_CACHE_ROOT.exists(),222 "hf_cache_entries": _safe_count_dir(HF_CACHE_ROOT),223 "hf_cache_size_bytes": _safe_size_bytes(HF_CACHE_ROOT),224 "env": {225 "IPO_DATA_DIR": os.getenv("IPO_DATA_DIR"),226 "HF_HOME": os.getenv("HF_HOME"),227 "TRANSFORMERS_CACHE": os.getenv("TRANSFORMERS_CACHE"),228 "SENTENCE_TRANSFORMERS_HOME": os.getenv("SENTENCE_TRANSFORMERS_HOME"),229 "XDG_CACHE_HOME": os.getenv("XDG_CACHE_HOME"),230 },231 }232 233 234import jwt235import bcrypt236from fastapi.security import OAuth2PasswordBearer237from fastapi import Depends238from pydantic import BaseModel239import random240 241JWT_SECRET = os.getenv("JWT_SECRET", "super_secret_ipo_key_123")242oauth2_scheme = OAuth2PasswordBearer(tokenUrl="auth/login")243 244RESEND_API_KEY = os.getenv("RESEND_API_KEY", "")245FROM_EMAIL = os.getenv("FROM_EMAIL", "hello@prospektlab.com")246AIRTABLE_API_TOKEN = os.getenv("AIRTABLE_API_TOKEN", "").strip()247AIRTABLE_BASE_ID = os.getenv("AIRTABLE_BASE_ID", "").strip()248AIRTABLE_TABLE_NAME = os.getenv("AIRTABLE_TABLE_NAME", "").strip()249APP_URL = os.getenv("APP_URL", "https://prospektlab.com")250 251 252class SignupRequest(BaseModel):253 email: str254 password: str255 full_name: str256 phone: str = ""257 otp_code: str = ""258 referral_code: str = ""259 notify_filings: bool = True260 261class LoginRequest(BaseModel):262 email: str263 password: str264 265class OTPRequest(BaseModel):266 email: str267 268class CompleteProfileRequest(BaseModel):269 phone: str270 271class TelemetryRequest(BaseModel):272 event_type: str273 metadata: Dict[str, Any]274 275 276def get_current_user(token: str = Depends(oauth2_scheme)):277 try:278 payload = jwt.decode(token, JWT_SECRET, algorithms=["HS256"])279 return payload.get("user_id")280 except:281 return None282 283 284def _send_email_via_resend(to_email: str, subject: str, html_body: str) -> bool:285 """Send email via Resend API. Returns True on success."""286 if not RESEND_API_KEY:287 print(f"[Resend] No API key — would send to {to_email}: {subject}")288 return False289 try:290 payload = json.dumps({291 "from": FROM_EMAIL,292 "to": [to_email],293 "subject": subject,294 "html": html_body,295 }).encode()296 req = urllib.request.Request(297 "https://api.resend.com/emails",298 data=payload,299 headers={300 "Authorization": f"Bearer {RESEND_API_KEY}", 301 "Content-Type": "application/json",302 "User-Agent": "ProspektLab-Backend/1.0 Mozilla/5.0"303 },304 method="POST"305 )306 with urllib.request.urlopen(req, timeout=10) as r:307 return r.status in (200, 201)308 except urllib.error.HTTPError as e:309 error_body = e.read().decode('utf-8')310 print(f"[Resend] HTTP Error {e.code}: {e.reason}")311 print(f"[Resend] Response Body: {error_body}")312 return False313 except Exception as e:314 print(f"[Resend] Error: {e}")315 return False316 317 318def _sync_to_airtable(user: dict):319 """Push signup data to Airtable CRM (fire-and-forget)."""320 if not all([AIRTABLE_API_TOKEN, AIRTABLE_BASE_ID, AIRTABLE_TABLE_NAME]):321 return322 try:323 payload = json.dumps({324 "records": [325 {326 "fields": {327 "User ID": user.get("id"),328 "Email": user.get("email"),329 "Full Name": user.get("full_name") or "",330 "Phone": user.get("phone") or "",331 "Auth Provider": user.get("auth_provider") or "local",332 "Is Subscribed": bool(user.get("is_subscribed")),333 "Analyses Count": user.get("analyses_count") or 0,334 "Subscribed At": user.get("subscribed_at") or "",335 "Referral Code": user.get("referral_code") or "",336 "Referred By": str(user.get("referred_by")) if user.get("referred_by") else "",337 "Signup Date": user.get("created_at") or datetime.now(timezone.utc).strftime("%Y-%m-%d"),338 }339 }340 ]341 }).encode()342 req = urllib.request.Request(343 f"https://api.airtable.com/v0/{AIRTABLE_BASE_ID}/{urllib.parse.quote(AIRTABLE_TABLE_NAME)}",344 data=payload,345 headers={346 "Authorization": f"Bearer {AIRTABLE_API_TOKEN}",347 "Content-Type": "application/json",348 "User-Agent": "ProspektLab-Backend/1.0"349 },350 method="POST"351 )352 with urllib.request.urlopen(req, timeout=8):353 pass354 except urllib.error.HTTPError as e:355 error_body = e.read().decode('utf-8')356 print(f"[Airtable] HTTP Error {e.code}: {e.reason}")357 print(f"[Airtable] Response Body: {error_body}")358 except Exception as e:359 print(f"[Airtable] Sync error: {e}")360 361 362@app.post("/auth/send-otp")363async def send_otp(req: OTPRequest):364 """Generate and send a 6-digit OTP to the given email for signup verification."""365 from backend.database import store_otp, cleanup_expired_otps366 from backend.disposable_domains import is_disposable_email367 368 if is_disposable_email(req.email):369 raise HTTPException(status_code=400, detail="Disposable or temporary email addresses are not allowed. Please use a real email.")370 371 cleanup_expired_otps()372 otp_code = str(random.randint(100000, 999999))373 from datetime import timedelta374 expires_at = (datetime.now(timezone.utc) + timedelta(minutes=10)).isoformat()375 store_otp(req.email, otp_code, expires_at)376 377 html = f"""378 <div style="font-family:Arial,sans-serif;max-width:480px;margin:0 auto;padding:24px;background:#f9fafb;border-radius:12px">379 <h2 style="color:#1a1a2e;margin-bottom:8px">Verify your email</h2>380 <p style="color:#6b7280">Use the code below to complete your ProspektLab signup. It expires in 10 minutes.</p>381 <div style="font-size:40px;font-weight:900;letter-spacing:8px;color:#0062ff;text-align:center;padding:24px 0">{otp_code}</div>382 <p style="color:#9ca3af;font-size:12px">If you didn't request this, you can safely ignore this email.</p>383 </div>384 """385 sent = _send_email_via_resend(req.email, "Your ProspektLab verification code", html)386 if not sent and RESEND_API_KEY:387 raise HTTPException(status_code=500, detail="Failed to send OTP email. Please try again.")388 return {"success": True, "message": "OTP sent to your email."}389 390 391@app.post("/auth/signup")392async def signup(req: SignupRequest):393 from backend.database import create_user, verify_otp394 from backend.disposable_domains import is_disposable_email395 396 if is_disposable_email(req.email):397 raise HTTPException(status_code=400, detail="Disposable email addresses are not allowed.")398 399 # Verify OTP (required for email/password signups)400 if not req.otp_code:401 raise HTTPException(status_code=400, detail="OTP verification required. Please verify your email first.")402 if not verify_otp(req.email, req.otp_code):403 raise HTTPException(status_code=400, detail="Invalid or expired OTP. Please request a new one.")404 405 salt = bcrypt.gensalt()406 hash_pw = bcrypt.hashpw(req.password.encode('utf-8'), salt).decode('utf-8')407 user_id = create_user(408 req.email, hash_pw, req.full_name,409 phone=req.phone or None,410 referral_code=req.referral_code or None411 )412 if not user_id:413 raise HTTPException(status_code=400, detail="Email already registered")414 415 from backend.database import record_referral, get_user_by_email416 new_user = get_user_by_email(req.email)417 418 # Record referral if applicable419 if req.referral_code and new_user and new_user.get("referred_by"):420 record_referral(new_user["referred_by"], user_id)421 422 if new_user:423 threading.Thread(target=_sync_to_airtable, args=(new_user,), daemon=True).start()424 425 # Create welcome notification426 try:427 from backend.database import create_notification428 create_notification(429 user_id,430 "Welcome to ProspektLab! 🎉",431 f"Hi {req.full_name or 'there'}! Your account is ready. You have {FREE_ANALYSIS_LIMIT} free IPO analyses to get started. Explore the latest SEBI filings and unlock deep intelligence.",432 notif_type="system"433 )434 except Exception:435 pass436 437 token = jwt.encode({"user_id": user_id, "email": req.email}, JWT_SECRET, algorithm="HS256")438 return {"token": token, "user_id": user_id, "email": req.email, "full_name": req.full_name, "needs_phone": not bool(req.phone)}439 440 441@app.post("/auth/login")442async def login(req: LoginRequest):443 from backend.database import get_user_by_email444 user = get_user_by_email(req.email)445 if not user:446 raise HTTPException(status_code=401, detail="Invalid credentials")447 try:448 is_valid = bcrypt.checkpw(req.password.encode('utf-8'), user["password_hash"].encode('utf-8'))449 except Exception:450 is_valid = False451 if not is_valid:452 raise HTTPException(status_code=401, detail="Invalid credentials")453 token = jwt.encode({"user_id": user["id"], "email": user["email"]}, JWT_SECRET, algorithm="HS256")454 return {455 "token": token,456 "user_id": user["id"],457 "email": user["email"],458 "full_name": user["full_name"],459 "needs_phone": not bool(user.get("phone")),460 "is_subscribed": bool(user.get("is_subscribed")),461 }462 463 464class GoogleAuthRequest(BaseModel):465 credential: str466 referral_code: str = ""467 468 469@app.post("/auth/google")470async def google_auth(req: GoogleAuthRequest):471 """Verify a Google ID token from GIS popup and return a session JWT."""472 from backend.database import create_user, get_user_by_email473 474 try:475 url = f"https://oauth2.googleapis.com/tokeninfo?id_token={urllib.parse.quote(req.credential)}"476 with urllib.request.urlopen(url, timeout=10) as resp:477 info = json.loads(resp.read())478 except urllib.error.HTTPError:479 raise HTTPException(status_code=401, detail="Invalid Google token")480 except Exception as e:481 raise HTTPException(status_code=500, detail=f"Google verification failed: {e}")482 483 email = info.get("email")484 name = info.get("name") or info.get("given_name", "")485 if not email or not info.get("email_verified"):486 raise HTTPException(status_code=401, detail="Google account email not verified")487 488 existing = get_user_by_email(email)489 is_new = False490 if existing:491 user_id = existing["id"]492 full_name = existing["full_name"] or name493 has_phone = bool(existing.get("phone"))494 is_subscribed = bool(existing.get("is_subscribed"))495 else:496 is_new = True497 user_id = create_user(email, None, name, auth_provider="google", referral_code=req.referral_code or None)498 if not user_id:499 raise HTTPException(status_code=500, detail="Failed to create user")500 full_name = name501 has_phone = False502 is_subscribed = False503 from backend.database import record_referral, get_user_by_email as gube504 new_user = gube(email)505 506 if req.referral_code and new_user and new_user.get("referred_by"):507 record_referral(new_user["referred_by"], user_id)508 509 if new_user:510 threading.Thread(target=_sync_to_airtable, args=(new_user,), daemon=True).start()511 512 token = jwt.encode({"user_id": user_id, "email": email}, JWT_SECRET, algorithm="HS256")513 return {514 "token": token,515 "user_id": user_id,516 "email": email,517 "full_name": full_name,518 "needs_phone": not has_phone,519 "is_subscribed": is_subscribed,520 }521 522 523@app.post("/auth/complete-profile")524async def complete_profile(req: CompleteProfileRequest, user_id: int = Depends(get_current_user)):525 """Save phone number for a user (used after Google OAuth)."""526 if not user_id:527 raise HTTPException(status_code=401, detail="Not authenticated")528 from backend.database import update_user_phone, get_user_by_id529 update_user_phone(user_id, req.phone)530 user = get_user_by_id(user_id)531 threading.Thread(532 target=_sync_to_airtable,533 args=(user,),534 daemon=True535 ).start()536 return {"success": True}537 538 539@app.get("/referral/stats")540async def referral_stats(user_id: int = Depends(get_current_user)):541 if not user_id:542 raise HTTPException(status_code=401, detail="Not authenticated")543 from backend.database import get_referral_stats544 stats = get_referral_stats(user_id)545 stats["referral_url"] = f"{APP_URL}/signup?ref={stats.get('referral_code', '')}"546 return stats547 548 549@app.post("/telemetry")550async def record_telemetry(req: TelemetryRequest, user_id: int = Depends(get_current_user)):551 if not user_id:552 # Don't write anonymous/invalid-token telemetry — it creates NULL FK rows553 raise HTTPException(status_code=401, detail="Not authenticated")554 try:555 from backend.database import log_telemetry556 log_telemetry(user_id, req.event_type, req.metadata)557 print(f"[Telemetry] user={user_id} event={req.event_type}")558 except Exception as exc:559 # Log the real error to stdout so it appears in HF/production logs560 print(f"[Telemetry] ERROR writing to DB for user={user_id} event={req.event_type}: {exc}")561 # Don't raise — client should never see a 500 for telemetry562 return {"status": "ok"}563 564@app.get("/admin/insights")565async def fetch_insights():566 try:567 from backend.database import get_telemetry_insights568 return {"insights": get_telemetry_insights()}569 except Exception as exc:570 print(f"[Admin/Insights] ERROR: {exc}")571 return {"insights": [], "error": str(exc)}572 573 574# ── Notification endpoints ──────────────────────────────────────────────────────575@app.get("/notifications")576async def list_notifications(user_id: int = Depends(get_current_user)):577 if not user_id:578 raise HTTPException(status_code=401, detail="Not authenticated")579 from backend.database import get_user_notifications, get_unread_count580 notifications = get_user_notifications(user_id)581 unread = get_unread_count(user_id)582 return {"notifications": notifications, "unread_count": unread}583 584 585@app.post("/notifications/{notification_id}/read")586async def read_notification(notification_id: int, user_id: int = Depends(get_current_user)):587 if not user_id:588 raise HTTPException(status_code=401, detail="Not authenticated")589 from backend.database import mark_notification_read590 mark_notification_read(notification_id, user_id)591 return {"success": True}592 593 594@app.post("/notifications/read-all")595async def read_all_notifications(user_id: int = Depends(get_current_user)):596 if not user_id:597 raise HTTPException(status_code=401, detail="Not authenticated")598 from backend.database import mark_all_notifications_read599 mark_all_notifications_read(user_id)600 return {"success": True}601 602 603 604 605import hmac606import hashlib607from fastapi import Request as FastAPIRequest608 609# ── Razorpay Configuration ─────────────────────────────────────────────────────610PAYMENT_ENV_FILES = (611 Path(__file__).resolve().parent.parent / ".env",612 Path(__file__).resolve().parent.parent / "frontend-next" / ".env.local",613)614 615 616def _read_env_file_value(path: Path, *keys: str) -> str:617 if not path.exists():618 return ""619 620 try:621 lines = path.read_text(encoding="utf-8", errors="ignore").splitlines()622 except Exception:623 return ""624 625 for raw_line in lines:626 line = raw_line.strip()627 if not line or line.startswith("#"):628 continue629 if line.lower().startswith("export "):630 line = line[7:].lstrip()631 632 match = re.match(r"^([A-Za-z_][A-Za-z0-9_]*)\s*[:=]\s*(.*)$", line)633 if not match:634 continue635 636 key, value = match.groups()637 if key not in keys:638 continue639 640 cleaned = value.strip()641 if cleaned and cleaned[0] == cleaned[-1] and cleaned[0] in {"'", '"'}:642 cleaned = cleaned[1:-1]643 return cleaned.strip()644 645 return ""646 647 648def _get_runtime_config_value(*keys: str) -> str:649 # Prefer file values so fixes to local .env files win over stale process env.650 for path in PAYMENT_ENV_FILES:651 value = _read_env_file_value(path, *keys)652 if value:653 return value654 655 for key in keys:656 value = os.getenv(key, "").strip()657 if value:658 return value659 660 return ""661 662 663def _get_razorpay_credentials() -> tuple[str, str]:664 key_id = _get_runtime_config_value("RAZORPAY_KEY_ID", "NEXT_PUBLIC_RAZORPAY_KEY_ID")665 key_secret = _get_runtime_config_value("RAZORPAY_KEY_SECRET")666 return key_id, key_secret667 668 669def _get_plan_amount_paise() -> int:670 raw_value = _get_runtime_config_value("RAZORPAY_PRO_MONTHLY_PRICE_PAISE") or "29900"671 try:672 return max(100, int(raw_value))673 except ValueError:674 return 29900675 676 677PLAN_AMOUNT_PAISE = _get_plan_amount_paise()678 679FREE_ANALYSIS_LIMIT = 3 # analyses allowed on free tier680 681 682# ── User Status ───────────────────────────────────────────────────────────────683@app.get("/user/status")684async def get_user_status(user_id: int = Depends(get_current_user)):685 """Return subscription status + usage count for the logged-in user."""686 if not user_id:687 raise HTTPException(status_code=401, detail="Not authenticated")688 from backend.database import get_user_by_id689 user = get_user_by_id(user_id)690 if not user:691 raise HTTPException(status_code=404, detail="User not found")692 return {693 "is_subscribed": bool(user.get("is_subscribed", 0)),694 "analyses_count": user.get("analyses_count", 0),695 "free_limit": FREE_ANALYSIS_LIMIT,696 "analyses_remaining": max(0, FREE_ANALYSIS_LIMIT - user.get("analyses_count", 0)) if not user.get("is_subscribed") else None,697 }698 699 700# ── Track Analysis (paywall gate) ─────────────────────────────────────────────701@app.post("/payment/track-analysis")702async def track_analysis(user_id: int = Depends(get_current_user)):703 """704 Called BEFORE starting an analysis.705 Returns {allowed: true} or {allowed: false, show_paywall: true}.706 """707 if not user_id:708 raise HTTPException(status_code=401, detail="Not authenticated")709 from backend.database import get_user_by_id, increment_user_analysis_count, qualify_referral, check_and_grant_reward710 user = get_user_by_id(user_id)711 if not user:712 raise HTTPException(status_code=404, detail="User not found")713 714 count = user.get("analyses_count", 0)715 if not user.get("is_subscribed") and count >= FREE_ANALYSIS_LIMIT:716 return {"allowed": False, "show_paywall": True, "analyses_count": count}717 718 new_count = increment_user_analysis_count(user_id)719 720 # Trigger referral qualification on 1st analysis721 if new_count == 1:722 referrer_id = qualify_referral(user_id)723 if referrer_id:724 check_and_grant_reward(referrer_id)725 726 if user.get("is_subscribed"):727 return {"allowed": True, "is_subscribed": True}728 729 return {"allowed": True, "analyses_count": new_count, "analyses_remaining": FREE_ANALYSIS_LIMIT - new_count}730 731 732# ── Create Razorpay Order ───────────────────────────────────────────────────────733class CreateOrderRequest(BaseModel):734 plan: str = "Pro"735 736@app.post("/payment/create-order")737async def create_razorpay_order(req: CreateOrderRequest, user_id: int = Depends(get_current_user)):738 """Create a Razorpay order. Returns key_id, order_id, amount for the frontend checkout."""739 if not user_id:740 raise HTTPException(status_code=401, detail="Not authenticated")741 742 razorpay_key_id, razorpay_key_secret = _get_razorpay_credentials()743 plan_amount_paise = _get_plan_amount_paise()744 745 if not razorpay_key_id or not razorpay_key_secret:746 # Dev mode — return mock order so frontend can test the modal747 return {748 "key_id": "rzp_test_mock",749 "order_id": "order_mock_123",750 "amount": plan_amount_paise,751 "currency": "INR",752 "mock": True,753 }754 755 try:756 import razorpay757 client = razorpay.Client(auth=(razorpay_key_id, razorpay_key_secret))758 759 order_data = {760 "amount": plan_amount_paise,761 "currency": "INR",762 "receipt": f"user_{user_id}",763 "notes": {"user_id": str(user_id), "plan": req.plan},764 }765 766 order = client.order.create(data=order_data)767 768 except Exception as e:769 print(f"[Razorpay] create-order error: {e}")770 error_str = str(e)771 if "Authentication failed" in error_str or "unauthorized" in error_str.lower():772 raise HTTPException(773 status_code=401,774 detail="Razorpay Authentication Failed: Your test keys in .env are invalid or revoked. Please regenerate them."775 )776 raise HTTPException(status_code=500, detail=f"Razorpay order creation failed: {e}")777 778 return {779 "key_id": razorpay_key_id,780 "order_id": order["id"],781 "amount": order["amount"],782 "currency": order["currency"],783 }784 785 786# ── Verify Payment (client-side callback) ──────────────────────────────────────787class VerifyPaymentRequest(BaseModel):788 razorpay_order_id: str789 razorpay_payment_id: str790 razorpay_signature: str791 792@app.post("/payment/verify")793async def verify_razorpay_payment(req: VerifyPaymentRequest, user_id: int = Depends(get_current_user)):794 """795 Verifies the HMAC-SHA256 signature from the Razorpay checkout callback.796 If valid, marks the user as subscribed.797 """798 if not user_id:799 raise HTTPException(status_code=401, detail="Not authenticated")800 801 razorpay_key_id, razorpay_key_secret = _get_razorpay_credentials()802 803 if not razorpay_key_secret or razorpay_key_secret == "...":804 # Dev mode — accept mock verification805 if req.razorpay_order_id == "order_mock_123":806 from backend.database import upgrade_user_subscription807 upgrade_user_subscription(user_id)808 return {"success": True, "message": "Mock subscription activated"}809 raise HTTPException(status_code=400, detail="Mock payment: invalid order id")810 811 import razorpay812 client = razorpay.Client(auth=(razorpay_key_id, razorpay_key_secret))813 814 try:815 client.utility.verify_payment_signature({816 'razorpay_order_id': req.razorpay_order_id,817 'razorpay_payment_id': req.razorpay_payment_id,818 'razorpay_signature': req.razorpay_signature819 })820 except razorpay.errors.SignatureVerificationError:821 raise HTTPException(status_code=400, detail="Payment verification failed — invalid signature")822 823 from backend.database import upgrade_user_subscription824 upgrade_user_subscription(user_id)825 print(f" ✅ User {user_id} subscribed via Razorpay payment {req.razorpay_payment_id}")826 return {"success": True, "message": "Subscription activated", "is_subscribed": True}827 828 829# ── Razorpay Webhook (server-to-server event, secondary mechanism) ─────────────830@app.post("/payment/webhook")831async def razorpay_webhook(request: FastAPIRequest):832 """Process Razorpay webhook events for payment.captured / subscription.charged."""833 payload = await request.body()834 sig_header = request.headers.get("x-razorpay-signature", "")835 _, razorpay_key_secret = _get_razorpay_credentials()836 837 if razorpay_key_secret and sig_header:838 expected = hmac.new(839 razorpay_key_secret.encode(),840 payload,841 hashlib.sha256,842 ).hexdigest()843 if not hmac.compare_digest(expected, sig_header):844 raise HTTPException(status_code=400, detail="Invalid webhook signature")845 846 try:847 event = json.loads(payload)848 except Exception:849 raise HTTPException(status_code=400, detail="Invalid JSON payload")850 851 event_type = event.get("event", "")852 if event_type in ("payment.captured", "subscription.charged"):853 notes = (854 event.get("payload", {})855 .get("payment", {})856 .get("entity", {})857 .get("notes", {})858 )859 user_id_str = notes.get("user_id")860 if user_id_str:861 from backend.database import upgrade_user_subscription862 upgrade_user_subscription(int(user_id_str))863 print(f" ✅ Razorpay webhook: User {user_id_str} upgraded")864 865 return {"received": True}866 867 868@app.get("/admin/set-subscription")869async def admin_set_subscription(email: str, status: int, secret_key: str):870 """Hidden endpoint to toggle premium status without accessing HF Spaces terminal"""871 if secret_key != "admin123":872 return {"error": "Unauthorized"}873 874 try:875 import sqlite3876 from backend.config import CACHE_DB_PATH877 conn = sqlite3.connect(CACHE_DB_PATH)878 cursor = conn.cursor()879 cursor.execute("UPDATE users SET is_subscribed = ? WHERE email = ?", (status, email))880 conn.commit()881 updated = cursor.rowcount882 conn.close()883 884 if updated > 0:885 return {"success": True, "message": f"Updated {email} subscription to {status}"}886 else:887 return {"error": f"User {email} not found"}888 except Exception as e:889 return {"error": str(e)}890 891 892@app.get("/admin/clear-users")893async def admin_clear_users(secret_key: str):894 """Hidden endpoint to completely wipe the users table in production."""895 if secret_key != "admin123":896 return {"error": "Unauthorized"}897 898 try:899 import sqlite3900 from backend.config import CACHE_DB_PATH901 conn = sqlite3.connect(CACHE_DB_PATH)902 cursor = conn.cursor()903 904 # Delete all users905 cursor.execute("DELETE FROM users")906 deleted_count = cursor.rowcount907 908 # Reset the auto-increment ID counter for the users table909 cursor.execute("DELETE FROM sqlite_sequence WHERE name='users'")910 911 conn.commit()912 conn.close()913 914 return {"success": True, "message": f"Successfully deleted {deleted_count} users from the production database."}915 except Exception as e:916 return {"error": str(e)}917 918 919def _prewarm_reranker() -> None:920 """Load the CrossEncoder reranker at startup so it's warm for the first analysis.921 922 On HF Spaces the model is downloaded from HuggingFace Hub the first time it's923 needed. Running this at boot (in a background thread) means the first user924 request finds the model already cached in RAM instead of waiting 5–30 seconds.925 """926 try:927 from backend.retriever import load_reranker928 load_reranker()929 print("[Startup] ✅ Reranker pre-warmed and ready.")930 except Exception as exc:931 print(f"[Startup] ⚠️ Reranker pre-warm failed (non-fatal): {exc}")932 933 934@app.on_event("startup")935async def on_startup() -> None:936 init_db()937 diagnostics = _runtime_storage_diagnostics()938 print("[Runtime] Storage diagnostics:")939 print(json.dumps(diagnostics, indent=2))940 # Run the initial, heavy 3-tab SEBI catalog scrape in the background941 # so uvicorn starts the server instantly.942 start_catalog_refresh(background=True, trigger="startup")943 # Pre-warm the CrossEncoder reranker in the background.944 asyncio.create_task(asyncio.to_thread(_prewarm_reranker))945 # Pre-warm filing highlights cache so the first /companies call is instant.946 asyncio.create_task(asyncio.to_thread(get_cached_filing_highlights))947 948@app.get("/")949async def root():950 index_path = FRONTEND_DIR / "index.html"951 if index_path.exists():952 return FileResponse(str(index_path))953 return {"status": "ok"}954 955 956@app.get("/debug/runtime-storage")957async def debug_runtime_storage():958 return _runtime_storage_diagnostics()959 960 961@app.get("/companies")962async def get_companies():963 """Returns companies from DB immediately (never blocks on SEBI scrape)."""964 _expire_stuck_catalog_refresh()965 companies = list_companies()966 if not companies and catalog_refresh_state["status"] == "idle":967 start_catalog_refresh(background=True, trigger="auto")968 return {969 "companies": companies,970 "catalog_refresh": dict(catalog_refresh_state),971 # Always use cached highlights — never block on a live SEBI request here.972 "highlights": filing_highlights_cache["data"],973 }974 975 976@app.post("/companies/refresh")977async def refresh_companies():978 """979 Returns existing DB companies IMMEDIATELY and starts a background SEBI refresh.980 The refresh typically takes 30-120 s; poll GET /companies/refresh/status981 every 2 s until status == 'complete' to get the freshly-scraped list.982 """983 existing = list_companies()984 # Fire background refresh (no-op if one is already running)985 start_catalog_refresh(background=True, trigger="user_refresh")986 return {987 "count": len(existing),988 "companies": existing,989 "catalog_refresh": dict(catalog_refresh_state),990 "highlights": filing_highlights_cache["data"],991 "message": "Refresh started in background. Poll /companies/refresh/status for completion.",992 }993 994 995@app.get("/companies/refresh/status")996async def refresh_status():997 """998 Poll this endpoint (~every 2 s) after calling POST /companies/refresh.999 Returns the current scrape state and, when complete, the fresh company list.1000 """1001 _expire_stuck_catalog_refresh()1002 state = dict(catalog_refresh_state)1003 if state.get("status") == "complete":1004 return {1005 **state,1006 "companies": list_companies(),1007 "highlights": filing_highlights_cache["data"],1008 }1009 return state1010 1011 1012@app.get("/sebi/ipo-tracker")1013async def get_ipo_tracker():1014 """Alias for /companies to support the new frontend API structure."""1015 return await get_companies()1016 1017 1018@app.get("/sebi/sector")1019async def get_sector(name: str):1020 cached = get_company_sector(name)1021 if cached:1022 return {"sector": cached}1023 1024 try:1025 from langchain_core.prompts import ChatPromptTemplate1026 from backend.llm import create_chat_llm1027 1028 llm = create_chat_llm(temperature=0)1029 1030 prompt = ChatPromptTemplate.from_messages([1031 ("system", "You are an expert financial analyst. Categorize the given company into ONE primary business sector from this list: Technology, Pharmaceuticals, Fintech, NBFC, Manufacturing, Aerospace, FMCG, Energy, Real Estate, Retail, Logistics, Agriculture, Construction, Textile, Automobile, Media, Education, Telecom, Chemical, Healthcare. Reply ONLY with the sector name."),1032 ("human", "Company Name: {company_name}")1033 ])1034 1035 msg = prompt | llm1036 result = msg.invoke({"company_name": name})1037 sector = str(result.content).strip()1038 save_company_sector(name, sector)1039 return {"sector": sector}1040 except Exception as e:1041 print(f"Sector resolution failed for {name}: {e}")1042 return {"sector": "General"}1043 1044 1045@app.delete("/companies/{company_id}/analysis/{doc_type}/cache")1046async def clear_analysis_cache(company_id: int, doc_type: str):1047 """Delete cached analysis for a company so the next run re-generates fresh results."""1048 from backend.database import delete_analysis1049 if doc_type not in {"drhp", "rhp", "prospectus", "all"}:1050 raise HTTPException(status_code=400, detail="Invalid doc_type")1051 deleted = delete_analysis(company_id, None if doc_type == "all" else doc_type)1052 return {"deleted_rows": deleted, "message": f"Cleared {doc_type.upper()} cache for company {company_id}"}1053 1054 1055@app.post("/admin/reset-analysis-cache")1056async def reset_analysis_cache():1057 with jobs_lock:1058 jobs.clear()1059 active_analysis_jobs.clear()1060 with background_precompute_lock:1061 background_precompute_jobs.clear()1062 1063 counts = reset_analysis_state()1064 outputs_removed = _clear_runtime_directory(OUTPUT_DIR)1065 chroma_removed = _clear_runtime_directory(CHROMA_BASE_DIR)1066 uploads_removed = _clear_runtime_directory(UPLOAD_DIR)1067 1068 if ENABLE_RECENT_COMPANY_PRECOMPUTE:1069 queue_recent_company_precompute()1070 1071 return {1072 **counts,1073 "outputs_removed": outputs_removed,1074 "chroma_entries_removed": chroma_removed,1075 "uploads_removed": uploads_removed,1076 "message": "Analysis cache reset complete",1077 }1078 1079 1080@app.post("/companies/{company_id}/hydrate")1081async def hydrate_company(company_id: int):1082 company = hydrate_company_documents(company_id, discover_matches=False, prepare_documents=False)1083 threading.Thread(1084 target=_background_discover_company_matches,1085 args=(company_id,),1086 daemon=True,1087 ).start()1088 return {"company": company}1089 1090 1091@app.get("/companies/{company_id}/analysis")1092async def get_company_analysis(company_id: int, doc_type: str = "drhp"):1093 if doc_type not in {"drhp", "rhp", "prospectus"}:1094 raise HTTPException(status_code=400, detail="Invalid document type")1095 1096 company = get_company(company_id)1097 if not company:1098 raise HTTPException(status_code=404, detail="Company not found")1099 1100 payload = _build_company_analysis_status(company_id, doc_type)1101 if payload.get("source") == "cache":1102 payload["source"] = "server_cache"1103 if payload.get("status") == "not_started":1104 queue_company_precompute(company_id, doc_type)1105 payload["company_id"] = company_id1106 payload["company"] = company1107 return payload1108 1109 1110@app.get("/admin/document-prep/{company_id}/{doc_type}")1111async def get_document_preparation_debug(company_id: int, doc_type: str):1112 if doc_type not in {"drhp", "rhp", "prospectus"}:1113 raise HTTPException(status_code=400, detail="Invalid document type")1114 1115 company = get_company(company_id)1116 if not company:1117 raise HTTPException(status_code=404, detail="Company not found")1118 1119 document = get_document_record(company_id, doc_type) or {}1120 metadata = document.get("metadata_json")1121 if isinstance(metadata, str):1122 try:1123 metadata = json.loads(metadata or "{}")1124 except Exception:1125 metadata = {}1126 elif not isinstance(metadata, dict):1127 metadata = {}1128 1129 with background_precompute_lock:1130 queue_depth = len(background_precompute_jobs)1131 queued = (company_id, doc_type) in background_precompute_jobs1132 1133 return {1134 "company_id": company_id,1135 "company": company,1136 "doc_type": doc_type,1137 "document_status": document.get("status"),1138 "pdf_sha256": document.get("pdf_sha256"),1139 "index_ready": bool(metadata.get("index_ready")),1140 "retriever_source": metadata.get("retriever_source"),1141 "retriever_cache_key": metadata.get("retriever_cache_key"),1142 "prepared_via": metadata.get("prepared_via"),1143 "chunk_count": metadata.get("chunk_count"),1144 "avg_chunk_chars": metadata.get("avg_chunk_chars"),1145 "financial_chunk_count": metadata.get("financial_chunk_count"),1146 "narrative_chunk_count": metadata.get("narrative_chunk_count"),1147 "timings": metadata.get("stage_timings_ms", {}),1148 "embed_duration_ms": metadata.get("embed_duration_ms"),1149 "chroma_write_duration_ms": metadata.get("chroma_write_duration_ms"),1150 "queue_depth": queue_depth,1151 "queued_for_background_precompute": queued,1152 "metadata": metadata,1153 }1154 1155 1156@app.get("/companies/{company_id}/documents")1157async def get_company_documents(company_id: int):1158 if not get_company(company_id):1159 raise HTTPException(status_code=404, detail="Company not found")1160 payload = {}1161 for doc_type in ("drhp", "rhp", "prospectus"):1162 payload[doc_type] = get_document_record(company_id, doc_type)1163 return {"company_id": company_id, "documents": payload}1164 1165 1166def _get_live_job_for_company(company_id: int, doc_type: str) -> Dict[str, Any] | None:1167 with jobs_lock:1168 active_job_id = active_analysis_jobs.get((company_id, doc_type))1169 if active_job_id:1170 active_job = jobs.get(active_job_id)1171 if active_job:1172 return active_job1173 for job in reversed(list(jobs.values())):1174 if job.get("company_id") == company_id and job.get("doc_type") == doc_type:1175 return job1176 return None1177 1178 1179def _run_background_precompute(company_id: int, doc_type: str) -> None:1180 try:1181 if not get_company(company_id):1182 return1183 asyncio.run(prewarm_company(company_id, doc_type))1184 if _get_current_cached_analysis(company_id, doc_type):1185 return1186 with jobs_lock:1187 active_job_id = active_analysis_jobs.get((company_id, doc_type))1188 active_job = jobs.get(active_job_id) if active_job_id else None1189 if active_job and active_job.get("status") == "processing":1190 return1191 asyncio.run(analyze_company_document(company_id, doc_type))1192 except Exception as exc:1193 print(1194 f"[Background] Precompute skipped for company {company_id} {doc_type.upper()}: {exc}"1195 )1196 finally:1197 with background_precompute_lock:1198 background_precompute_jobs.discard((company_id, doc_type))1199 1200