import time import os import re import logging from contextlib import asynccontextmanager from typing import Optional, Dict, Any, List logger = logging.getLogger("LookupApp") from fastapi import FastAPI, Query, HTTPException, Request, Header, Depends from fastapi.middleware.cors import CORSMiddleware from fastapi.middleware.gzip import GZipMiddleware from fastapi.responses import HTMLResponse, FileResponse, JSONResponse from fastapi.staticfiles import StaticFiles from pydantic import BaseModel, Field from engine import LookupEngine from external_services import ServiceGateway from auth import KeyManager, ADMIN_MASTER_KEY, INTERNAL_BOT_KEY start_time = time.time() _engine: Optional[LookupEngine] = None def get_engine() -> LookupEngine: global _engine if _engine is None: _engine = LookupEngine() _engine.warm_up() return _engine @asynccontextmanager async def lifespan(app: FastAPI): get_engine() # Auto-launch Telegram Bot if TELEGRAM_BOT_TOKEN is provided in Space Secrets bot_token = os.environ.get("TELEGRAM_BOT_TOKEN", "").strip() if bot_token: import threading from bot import create_bot def _bot_worker(): try: b = create_bot(bot_token) b_info = b.get_me() print(f"[+] Telegram Bot Daemon online as @{b_info.username}", flush=True) b.infinity_polling(timeout=20, long_polling_timeout=15) except Exception as e: print(f"[!] Telegram Bot Daemon error: {e}", flush=True) t = threading.Thread(target=_bot_worker, daemon=True) t.start() print("[+] Background Telegram Bot thread dispatched.", flush=True) yield app = FastAPI( title="HyperLookup API & SaaS Engine", description="Enterprise-grade ultra-fast lookup API for 2.5B+ Indian Records (Phone & Aadhaar) and 65M+ Telegram Master Database with API Key Auth & Quota Controls.", version="2.2.0", lifespan=lifespan, docs_url="/docs", redoc_url="/redoc" ) # Enable GZip compression app.add_middleware(GZipMiddleware, minimum_size=1000) # Enable CORS for all frontends, bots, and external apps app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # Mount static directory if exists static_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), "static") if os.path.exists(static_dir): app.mount("/static", StaticFiles(directory=static_dir), name="static") # --------------------------------------------------------- # Request Models # --------------------------------------------------------- class CreateKeyRequest(BaseModel): name: str = Field(..., example="Client Alpha") days_valid: Optional[int] = Field(30, description="Validity period in days. Pass 0 or null for lifetime.") rate_limit_per_min: Optional[int] = Field(60, description="Max allowed requests per minute.") total_credits: Optional[int] = Field(1000, description="Total number of search queries allowed. Pass -1 for unlimited.") # --------------------------------------------------------- # Authentication Dependency # --------------------------------------------------------- def get_client_ip(request: Request) -> str: """Extracts client IP considering Cloudflare, proxies, and standard headers.""" cf_ip = request.headers.get("cf-connecting-ip") if cf_ip: return cf_ip.strip() forwarded = request.headers.get("x-forwarded-for") if forwarded: return forwarded.split(",")[0].strip() real_ip = request.headers.get("x-real-ip") if real_ip: return real_ip.strip() return request.client.host if request.client else "127.0.0.1" def verify_api_key( request: Request, api_key_query: Optional[str] = Query(None, alias="api_key"), key_query: Optional[str] = Query(None, alias="key"), x_api_key: Optional[str] = Header(None, alias="X-API-Key"), authorization: Optional[str] = Header(None) ) -> Dict[str, Any]: """Extracts and verifies API key or applies IP-based daily free search with anti-DDoS.""" token = None if x_api_key: token = x_api_key elif authorization and authorization.lower().startswith("bearer "): token = authorization.split(" ", 1)[1] elif api_key_query: token = api_key_query elif key_query: token = key_query client_ip = get_client_ip(request) is_valid, code, msg, meta = KeyManager.validate_request(token, client_ip) if not is_valid: raise HTTPException(status_code=code, detail=msg) return meta def verify_admin_key( x_admin_key: Optional[str] = Header(None, alias="X-Admin-Key"), admin_key_query: Optional[str] = Query(None, alias="admin_key") ): """Requires the master admin key to access administrative operations.""" key = x_admin_key or admin_key_query if key != ADMIN_MASTER_KEY: raise HTTPException(status_code=401, detail="Unauthorized. Master Admin Key required.") return key # --------------------------------------------------------- # Public Endpoints # --------------------------------------------------------- @app.get("/", response_class=HTMLResponse) async def serve_home(): index_file = os.path.join(static_dir, "index.html") if os.path.exists(index_file): return FileResponse(index_file) return HTMLResponse("

HyperLookup Engine API is Online

API documentation available at /docs.

") @app.get("/api/health") async def health_check(): """Liveness probe for monitoring services.""" return {"status": "ok", "timestamp": time.time()} @app.get("/api/public/quota") async def check_public_quota(request: Request): """Returns current daily free search quota for the caller IP.""" client_ip = get_client_ip(request) return KeyManager.get_public_quota(client_ip) @app.get("/api/stats") async def get_system_stats(): """Returns engine health, cache status, and database metrics.""" eng = get_engine() uptime_sec = round(time.time() - start_time, 2) return { "status": "healthy", "uptime_seconds": uptime_sec, "engine_warmed": eng.is_warmed, "cache_entries": len(eng.cache), "database_info": { "total_records_indexed": "478,800,000+ Master DB Records", "active_services": 11, "engine": "DuckDB Vectorized Core", "cluster_state": "Operational" } } # --------------------------------------------------------- # Telegram Search Endpoints (Isolated & Ultra-Fast) # --------------------------------------------------------- @app.get("/api/telegram/lookup") @app.get("/api/tg/lookup") @app.get("/tg/lookup") @app.get("/api/tg") @app.get("/tg") @app.get("/telegram") async def search_telegram( user_id: Optional[str] = Query(None, description="Telegram User ID (e.g. 74252719)"), id: Optional[str] = Query(None, description="Alias for user_id"), q: Optional[str] = Query(None, description="Alias for user_id"), auth_meta: Dict[str, Any] = Depends(verify_api_key) ): target = user_id or id or q if not target: raise HTTPException(status_code=400, detail="Missing required query parameter: 'user_id'") res = get_engine().query_by_telegram(target) if isinstance(res, dict): res["tier"] = auth_meta.get("tier", "free") res["remaining_credits"] = auth_meta.get("remaining", -1) return res @app.get("/api/telegram/{user_id}") @app.get("/api/tg/{user_id}") @app.get("/tg/{user_id}") async def lookup_telegram_path(user_id: str, auth_meta: Dict[str, Any] = Depends(verify_api_key)): res = get_engine().query_by_telegram(user_id) if isinstance(res, dict): res["tier"] = auth_meta.get("tier", "free") res["remaining_credits"] = auth_meta.get("remaining", -1) return res # --------------------------------------------------------- # Individual Master Database Search Endpoints # --------------------------------------------------------- @app.get("/api/truecaller/{phone_number}") async def lookup_truecaller(phone_number: str, auth_meta: Dict[str, Any] = Depends(verify_api_key)): """Search Truecaller 285M Master Database by 10-digit mobile.""" res = get_engine().query_truecaller(phone_number) if isinstance(res, dict): res["tier"] = auth_meta.get("tier", "free") res["remaining_credits"] = auth_meta.get("remaining", -1) return res @app.get("/api/airtel/{phone_number}") async def lookup_airtel(phone_number: str, auth_meta: Dict[str, Any] = Depends(verify_api_key)): """Search Airtel India 73M Master Database by 10-digit mobile.""" res = get_engine().query_airtel(phone_number) if isinstance(res, dict): res["tier"] = auth_meta.get("tier", "free") res["remaining_credits"] = auth_meta.get("remaining", -1) return res @app.get("/api/hdfc/{phone_number}") async def lookup_hdfc(phone_number: str, auth_meta: Dict[str, Any] = Depends(verify_api_key)): """Search HDFC Bank Master Database (CIBIL, Highmark, Loans) by 10-digit mobile.""" res = get_engine().query_hdfc(phone_number) if isinstance(res, dict): res["tier"] = auth_meta.get("tier", "free") res["remaining_credits"] = auth_meta.get("remaining", -1) return res @app.get("/api/bsnl/{phone_number}") async def lookup_bsnl(phone_number: str, auth_meta: Dict[str, Any] = Depends(verify_api_key)): """Search BSNL 19M Master Database by 10-digit mobile.""" res = get_engine().query_bsnl(phone_number) if isinstance(res, dict): res["tier"] = auth_meta.get("tier", "free") res["remaining_credits"] = auth_meta.get("remaining", -1) return res @app.get("/api/paytm/{phone_number}") async def lookup_paytm(phone_number: str, auth_meta: Dict[str, Any] = Depends(verify_api_key)): """Search Paytm India Master Database by 10-digit mobile.""" res = get_engine().query_paytm(phone_number) if isinstance(res, dict): res["tier"] = auth_meta.get("tier", "free") res["remaining_credits"] = auth_meta.get("remaining", -1) return res @app.get("/api/flipkart/{phone_number}") async def lookup_flipkart(phone_number: str, auth_meta: Dict[str, Any] = Depends(verify_api_key)): """Search Flipkart India Master Database by 10-digit mobile.""" res = get_engine().query_flipkart(phone_number) if isinstance(res, dict): res["tier"] = auth_meta.get("tier", "free") res["remaining_credits"] = auth_meta.get("remaining", -1) return res @app.get("/api/instagram/{query}") async def lookup_instagram(query: str, auth_meta: Dict[str, Any] = Depends(verify_api_key)): """Search Instagram 16.7M Master Database by Phone or Username.""" res = get_engine().query_instagram(query) if isinstance(res, dict): res["tier"] = auth_meta.get("tier", "free") res["remaining_credits"] = auth_meta.get("remaining", -1) return res @app.get("/api/twitter/{query}") async def lookup_twitter(query: str, auth_meta: Dict[str, Any] = Depends(verify_api_key)): """Search Twitter / X 5.48M Master Database by User ID or Username.""" res = get_engine().query_twitter(query) if isinstance(res, dict): res["tier"] = auth_meta.get("tier", "free") res["remaining_credits"] = auth_meta.get("remaining", -1) return res @app.get("/api/sbi/{query}") @app.get("/api/sbi") async def lookup_sbi( query: Optional[str] = None, q: Optional[str] = Query(None, description="Search query (phone, aadhaar, pan, name, email)"), phone: Optional[str] = Query(None, description="10-digit mobile number"), auth_meta: Dict[str, Any] = Depends(verify_api_key) ): """Search SBI POSP & Indian Banking Master Database by Phone, Aadhaar, PAN, Name, or Email.""" target = query or q or phone if not target: raise HTTPException(status_code=400, detail="Missing required parameter: provide 'query', 'q', or 'phone'.") res = get_engine().query_sbi(target) if isinstance(res, dict): res["tier"] = auth_meta.get("tier", "free") res["remaining_credits"] = auth_meta.get("remaining", -1) return res @app.get("/api/ind/{query}") @app.get("/api/ind") async def lookup_ind( query: Optional[str] = None, q: Optional[str] = Query(None, description="Search query (phone, email, aadhaar, pan, name)"), phone: Optional[str] = Query(None, description="10-digit mobile number"), email: Optional[str] = Query(None, description="Email address"), auth_meta: Dict[str, Any] = Depends(verify_api_key) ): """Search IND Education, Admissions & KYC Master Database by Phone, Email, Aadhaar, PAN, or Name.""" target = query or q or phone or email if not target: raise HTTPException(status_code=400, detail="Missing required parameter: provide 'query', 'q', 'phone', or 'email'.") res = get_engine().query_ind(target) if isinstance(res, dict): res["tier"] = auth_meta.get("tier", "free") res["remaining_credits"] = auth_meta.get("remaining", -1) return res @app.get("/api/upi/{query}") @app.get("/api/upi") async def lookup_upi( query: Optional[str] = None, q: Optional[str] = Query(None, description="UPI ID / VPA (e.g. 9826287975@paytm) or 10-digit mobile number"), vpa: Optional[str] = Query(None, description="UPI VPA address"), phone: Optional[str] = Query(None, description="10-digit Indian Mobile"), mode: Optional[str] = Query("full", description="Lookup mode: 'full' (360° KYC Dossier), 'upi_to_num' (VPA to Phone), or 'num_to_upi' (Phone to active UPI IDs)"), auth_meta: Dict[str, Any] = Depends(verify_api_key) ): """ UPI & VPA 360° OSINT & KYC Resolver: - Mode 'full': Resolves UPI ID / Mobile to complete 360° cross-database KYC profile, active UPI handles, bank linkages, Aadhaar/PAN, and full address. - Mode 'upi_to_num': Extracts registered phone number, PSP application, acquiring bank & IFSC. - Mode 'num_to_upi': Generates and scores active UPI IDs across PhonePe, Google Pay, Paytm, BHIM, etc. """ target = query or q or vpa or phone if not target: raise HTTPException(status_code=400, detail="Missing required parameter: provide 'query', 'q', 'vpa', or 'phone'.") res = get_engine().query_upi(target, mode=mode) if isinstance(res, dict): res["tier"] = auth_meta.get("tier", "free") res["remaining_credits"] = auth_meta.get("remaining", -1) return res # --------------------------------------------------------- # Banking, Business, Vehicle & Network Intelligence APIs # --------------------------------------------------------- @app.get("/api/ifsc/{code}") @app.get("/api/ifsc") async def lookup_ifsc(code: Optional[str] = None, ifsc: Optional[str] = Query(None), q: Optional[str] = Query(None), auth_meta: Dict[str, Any] = Depends(verify_api_key)): target = code or ifsc or q if not target: raise HTTPException(status_code=400, detail="Missing required parameter: provide 'code', 'ifsc', or 'q'.") res = ServiceGateway.get_ifsc(target) if isinstance(res, dict): res["tier"] = auth_meta.get("tier", "free") res["remaining_credits"] = auth_meta.get("remaining", -1) return res @app.get("/api/pincode/{pin}") @app.get("/api/pincode") async def lookup_pincode(pin: Optional[str] = None, pincode: Optional[str] = Query(None), q: Optional[str] = Query(None), auth_meta: Dict[str, Any] = Depends(verify_api_key)): target = pin or pincode or q if not target: raise HTTPException(status_code=400, detail="Missing required parameter: provide 'pin', 'pincode', or 'q'.") res = ServiceGateway.get_pincode(target) if isinstance(res, dict): res["tier"] = auth_meta.get("tier", "free") res["remaining_credits"] = auth_meta.get("remaining", -1) return res @app.get("/api/ip/{ip_addr}") @app.get("/api/ip") async def lookup_ip(ip_addr: Optional[str] = None, ip: Optional[str] = Query(None), q: Optional[str] = Query(None), auth_meta: Dict[str, Any] = Depends(verify_api_key)): target = ip_addr or ip or q if not target: raise HTTPException(status_code=400, detail="Missing required parameter: provide 'ip_addr', 'ip', or 'q'.") res = ServiceGateway.get_ip(target) if isinstance(res, dict): res["tier"] = auth_meta.get("tier", "free") res["remaining_credits"] = auth_meta.get("remaining", -1) return res @app.get("/api/vehicle/{vehicle_no}") @app.get("/api/rc/{vehicle_no}") @app.get("/api/rc") @app.get("/api/vehicle") async def lookup_vehicle(vehicle_no: Optional[str] = None, vehicle: Optional[str] = Query(None), rc: Optional[str] = Query(None), owner: Optional[str] = Query(None), q: Optional[str] = Query(None), auth_meta: Dict[str, Any] = Depends(verify_api_key)): target = vehicle_no or vehicle or rc or owner or q if not target: raise HTTPException(status_code=400, detail="Missing required parameter: provide vehicle registration number.") res = ServiceGateway.get_vehicle_rc(target) if isinstance(res, dict): res["tier"] = auth_meta.get("tier", "free") res["remaining_credits"] = auth_meta.get("remaining", -1) return res @app.get("/api/challan/{vehicle_no}") @app.get("/api/challan") async def lookup_challan(vehicle_no: Optional[str] = None, vehicle: Optional[str] = Query(None), q: Optional[str] = Query(None), auth_meta: Dict[str, Any] = Depends(verify_api_key)): target = vehicle_no or vehicle or q if not target: raise HTTPException(status_code=400, detail="Missing required parameter: provide vehicle registration number.") res = ServiceGateway.get_vehicle_challan(target) if isinstance(res, dict): res["tier"] = auth_meta.get("tier", "free") res["remaining_credits"] = auth_meta.get("remaining", -1) return res @app.get("/api/pan/{pan_no}") @app.get("/api/pan") async def lookup_pan(pan_no: Optional[str] = None, pan: Optional[str] = Query(None), q: Optional[str] = Query(None), auth_meta: Dict[str, Any] = Depends(verify_api_key)): target = pan_no or pan or q if not target: raise HTTPException(status_code=400, detail="Missing required parameter: provide 'pan_no', 'pan', or 'q'.") res = ServiceGateway.get_pan_details(target) if isinstance(res, dict): res["tier"] = auth_meta.get("tier", "free") res["remaining_credits"] = auth_meta.get("remaining", -1) return res @app.get("/api/gst/{gstin}") @app.get("/api/gst") async def lookup_gst(gstin: Optional[str] = None, gst: Optional[str] = Query(None), q: Optional[str] = Query(None), auth_meta: Dict[str, Any] = Depends(verify_api_key)): target = gstin or gst or q if not target: raise HTTPException(status_code=400, detail="Missing required parameter: provide 'gstin', 'gst', or 'q'.") res = ServiceGateway.get_gst_details(target) if isinstance(res, dict): res["tier"] = auth_meta.get("tier", "free") res["remaining_credits"] = auth_meta.get("remaining", -1) return res @app.get("/api/pantogst/{pan_no}") @app.get("/api/pantogst") async def lookup_pantogst(pan_no: Optional[str] = None, pan: Optional[str] = Query(None), q: Optional[str] = Query(None), auth_meta: Dict[str, Any] = Depends(verify_api_key)): target = pan_no or pan or q if not target: raise HTTPException(status_code=400, detail="Missing required parameter: provide 'pan_no', 'pan', or 'q'.") res = ServiceGateway.get_pan_to_gst(target) if isinstance(res, dict): res["tier"] = auth_meta.get("tier", "free") res["remaining_credits"] = auth_meta.get("remaining", -1) return res @app.get("/api/vehtonum/{vehicle_no}") @app.get("/api/vehtonum") @app.get("/api/veh-to-num") async def lookup_veh_to_num(vehicle_no: Optional[str] = None, vehicle: Optional[str] = Query(None), q: Optional[str] = Query(None), auth_meta: Dict[str, Any] = Depends(verify_api_key)): target = vehicle_no or vehicle or q if not target: raise HTTPException(status_code=400, detail="Missing required parameter: provide vehicle registration number.") res = ServiceGateway.get_veh_to_num(target) if isinstance(res, dict): res["tier"] = auth_meta.get("tier", "free") res["remaining_credits"] = auth_meta.get("remaining", -1) return res @app.get("/api/ration/{query}") @app.get("/api/ration") @app.get("/api/adharration/{query}") @app.get("/api/adharration") async def lookup_ration(query: Optional[str] = None, num: Optional[str] = Query(None), q: Optional[str] = Query(None), auth_meta: Dict[str, Any] = Depends(verify_api_key)): target = query or num or q if not target: raise HTTPException(status_code=400, detail="Missing required parameter: provide 12-digit Aadhaar UID or Ration Card number.") res = ServiceGateway.get_ration(target) if isinstance(res, dict): res["tier"] = auth_meta.get("tier", "free") res["remaining_credits"] = auth_meta.get("remaining", -1) return res @app.get("/api/bgmi/{uid}") @app.get("/api/bgmi") async def lookup_bgmi(uid: Optional[str] = None, q: Optional[str] = Query(None), auth_meta: Dict[str, Any] = Depends(verify_api_key)): target = uid or q if not target: raise HTTPException(status_code=400, detail="Missing required parameter: provide BGMI numeric UID.") res = ServiceGateway.get_bgmi(target) if isinstance(res, dict): res["tier"] = auth_meta.get("tier", "free") res["remaining_credits"] = auth_meta.get("remaining", -1) return res @app.get("/api/github/{query}") @app.get("/api/github") async def lookup_github(query: Optional[str] = None, q: Optional[str] = Query(None), repo: Optional[str] = Query(None), auth_meta: Dict[str, Any] = Depends(verify_api_key)): target = query or repo or q if not target: raise HTTPException(status_code=400, detail="Missing required parameter: provide GitHub username or repo.") res = ServiceGateway.get_github(target) if isinstance(res, dict): res["tier"] = auth_meta.get("tier", "free") res["remaining_credits"] = auth_meta.get("remaining", -1) return res @app.get("/api/youtube/{query}") @app.get("/api/youtube") async def lookup_youtube(query: Optional[str] = None, q: Optional[str] = Query(None), video: Optional[str] = Query(None), auth_meta: Dict[str, Any] = Depends(verify_api_key)): target = query or video or q if not target: raise HTTPException(status_code=400, detail="Missing required parameter: provide YouTube URL or Video ID.") res = ServiceGateway.get_youtube_info(target) if isinstance(res, dict): res["tier"] = auth_meta.get("tier", "free") res["remaining_credits"] = auth_meta.get("remaining", -1) return res @app.get("/api/gas/{query}") @app.get("/api/gas") @app.get("/api/lpg/{query}") @app.get("/api/lpg") @app.get("/api/indane/{query}") @app.get("/api/indane") @app.get("/api/bharatgas/{query}") @app.get("/api/bharatgas") @app.get("/api/hpgas/{query}") @app.get("/api/hpgas") async def lookup_lpg_gas( request: Request, query: Optional[str] = None, num: Optional[str] = Query(None, description="10-digit mobile number or 17-digit LPG ID"), phone: Optional[str] = Query(None), lpg_id: Optional[str] = Query(None), consumer_no: Optional[str] = Query(None), provider: Optional[str] = Query(None, description="Gas provider: 'indane', 'bharat', or 'hp'"), q: Optional[str] = Query(None), auth_meta: Dict[str, Any] = Depends(verify_api_key) ): """ Direct LPG Gas Consumer Intelligence: Resolves 10-digit mobile number or 17-digit LPG ID across all 3 Indian OMCs: Indane Gas (IOCL), Bharat Gas (BPCL), and HP Gas (HPCL). """ target = query or num or phone or lpg_id or consumer_no or q if not target: raise HTTPException(status_code=400, detail="Missing required parameter: provide 10-digit mobile number or 17-digit LPG ID.") auto_prov = provider if (provider and isinstance(provider, str)) else None if not auto_prov: path_lower = request.url.path.lower() if "bharat" in path_lower: auto_prov = "bharat" elif "hp" in path_lower: auto_prov = "hp" elif "indane" in path_lower: auto_prov = "indane" clean_digits = re.sub(r'\D', '', str(target)).strip() user_kyc = None if len(clean_digits) == 10: try: eng = get_engine() phone_res = eng.query_by_phone(clean_digits) if isinstance(phone_res, dict): records = phone_res.get("records", []) if records and isinstance(records, list) and len(records) > 0: rec = records[0] user_kyc = { "name": rec.get("name") or rec.get("fullName"), "fathersName": rec.get("fathersName") or rec.get("father_name"), "phoneNumber": rec.get("phoneNumber") or clean_digits, "aadharNumber": rec.get("aadharNumber") or rec.get("aadhaar"), "address": rec.get("address"), "pincode": rec.get("pincode") or rec.get("pinCode"), "district": rec.get("district"), "state": rec.get("state") } if not user_kyc or not user_kyc.get("name"): comb_res = eng.query_combined(clean_digits) if isinstance(comb_res, dict): summary = comb_res.get("consolidated_summary", {}) p_name = summary.get("primary_name") if not p_name and summary.get("names_found"): p_name = summary["names_found"][0] locs = summary.get("locations_identified", []) loc = locs[0] if locs else "" if p_name: user_kyc = { "name": p_name, "fathersName": None, "phoneNumber": clean_digits, "aadharNumber": None, "address": loc, "pincode": None } except Exception as e: logger.error(f"Gas consumer KYC resolution: {e}") res = ServiceGateway.get_lpg(target, provider=auto_prov, user_kyc=user_kyc) if isinstance(res, dict): res["tier"] = auth_meta.get("tier", "free") res["remaining_credits"] = auth_meta.get("remaining", -1) return res @app.get("/api/phone/{phone_number}") async def lookup_by_phone(phone_number: str, auth_meta: Dict[str, Any] = Depends(verify_api_key)): """Search Telecom 2.5B Database by phone number.""" res = get_engine().query_by_phone(phone_number) if isinstance(res, dict): res["tier"] = auth_meta.get("tier", "free") res["remaining_credits"] = auth_meta.get("remaining", -1) return res @app.get("/api/aadhar/{aadhar_number}") async def lookup_by_aadhar(aadhar_number: str, auth_meta: Dict[str, Any] = Depends(verify_api_key)): """Search Aadhaar Database by 12-digit number.""" res = get_engine().query_by_aadhar(aadhar_number) if isinstance(res, dict): res["tier"] = auth_meta.get("tier", "free") res["remaining_credits"] = auth_meta.get("remaining", -1) return res @app.get("/api/family/{query}") @app.get("/api/family") @app.get("/api/aadhar-family/{query}") @app.get("/api/aadhar_family/{query}") @app.get("/api/aadhar/family/{query}") async def lookup_aadhar_family( query: Optional[str] = None, q: Optional[str] = Query(None, description="Aadhaar UID or Phone Number"), aadhar: Optional[str] = Query(None, description="12-digit Aadhaar UID"), phone: Optional[str] = Query(None, description="10-digit Phone Number"), auth_meta: Dict[str, Any] = Depends(verify_api_key) ): """ Search Aadhaar Family & Household Linkage Database by 12-digit Aadhaar UID or 10-digit Mobile. Returns structured family hierarchy, co-residents, and shared household connections. """ target = query or q or aadhar or phone if not target: raise HTTPException(status_code=400, detail="Missing required parameter: provide 'query', 'q', 'aadhar', or 'phone'.") res = get_engine().query_by_aadhar_family(target) if isinstance(res, dict): res["tier"] = auth_meta.get("tier", "free") res["remaining_credits"] = auth_meta.get("remaining", -1) return res # --------------------------------------------------------- # Unified Combined Search Endpoints (All Databases in Parallel) # --------------------------------------------------------- @app.get("/api/combined/{phone_number}") @app.get("/api/lookup/all") @app.get("/api/lookup/combined") async def combined_lookup( phone_number: Optional[str] = None, phone: Optional[str] = Query(None, description="10-digit Indian Mobile"), q: Optional[str] = Query(None, description="Search query"), auth_meta: Dict[str, Any] = Depends(verify_api_key) ): """ Unified Combined Search: Queries Truecaller, Airtel, HDFC, BSNL, Paytm, Flipkart, Instagram, and Telecom DB in parallel and returns both consolidated profile intelligence and individual database records. """ target = phone_number or phone or q if not target: raise HTTPException(status_code=400, detail="Missing parameter: provide 'phone_number', 'phone', or 'q'.") res = get_engine().query_combined(target) if isinstance(res, dict): res["tier"] = auth_meta.get("tier", "free") res["remaining_credits"] = auth_meta.get("remaining", -1) return res # --------------------------------------------------------- # 🔥 360° Deep Intelligence Dossier Endpoints # --------------------------------------------------------- @app.get("/api/deep/{query}") @app.get("/api/deep") @app.get("/api/dossier/{query}") @app.get("/api/dossier") async def lookup_deep_dossier( query: Optional[str] = None, q: Optional[str] = Query(None, description="Any Indian identifier: Phone, Aadhaar, PAN, UPI, Vehicle, Telegram ID, Email, Name"), target_param: Optional[str] = Query(None, alias="target"), auth_meta: Dict[str, Any] = Depends(verify_api_key) ): """ 🔥 360° Deep Master Intelligence Dossier: Cross-queries all 16 Indian Master Databases in parallel and synthesizes a comprehensive, cross-correlated intelligence profile with KYC consensus score. """ target = query or q or target_param if not target: raise HTTPException(status_code=400, detail="Missing parameter: provide 'query', 'q', or 'target'.") res = get_engine().query_deep_dossier(target) if isinstance(res, dict): res["tier"] = auth_meta.get("tier", "free") res["remaining_credits"] = auth_meta.get("remaining", -1) return res @app.get("/api/muthoot/{query}") @app.get("/api/muthoot") async def lookup_muthoot( query: Optional[str] = None, q: Optional[str] = Query(None), phone: Optional[str] = Query(None), auth_meta: Dict[str, Any] = Depends(verify_api_key) ): """Search Muthoot Finance Gold Loans & Banking KYC Master Database.""" target = query or q or phone if not target: raise HTTPException(status_code=400, detail="Missing parameter: provide 'query', 'q', or 'phone'.") res = get_engine().query_muthoot(target) if isinstance(res, dict): res["tier"] = auth_meta.get("tier", "free") res["remaining_credits"] = auth_meta.get("remaining", -1) return res @app.get("/api/olx/{query}") @app.get("/api/olx") async def lookup_olx( query: Optional[str] = None, q: Optional[str] = Query(None), phone: Optional[str] = Query(None), auth_meta: Dict[str, Any] = Depends(verify_api_key) ): """Search OLX India Auto Dealers & Classifieds Merchants Master Database.""" target = query or q or phone if not target: raise HTTPException(status_code=400, detail="Missing parameter: provide 'query', 'q', or 'phone'.") res = get_engine().query_olx(target) if isinstance(res, dict): res["tier"] = auth_meta.get("tier", "free") res["remaining_credits"] = auth_meta.get("remaining", -1) return res @app.get("/api/icici/{query}") @app.get("/api/icici") async def lookup_icici( query: Optional[str] = None, q: Optional[str] = Query(None), model: Optional[str] = Query(None), auth_meta: Dict[str, Any] = Depends(verify_api_key) ): """Search ICICI Lombard Private Car & Vehicle Insurance Master Database.""" target = query or q or model if not target: raise HTTPException(status_code=400, detail="Missing parameter: provide 'query', 'q', or 'model'.") res = get_engine().query_icici(target) if isinstance(res, dict): res["tier"] = auth_meta.get("tier", "free") res["remaining_credits"] = auth_meta.get("remaining", -1) return res @app.get("/api/xyron/{query}") @app.get("/api/xyron") async def lookup_xyron_master( query: Optional[str] = None, q: Optional[str] = Query(None), phone: Optional[str] = Query(None), auth_meta: Dict[str, Any] = Depends(verify_api_key) ): """Search Unified Xyron Intel Master (Combined Banking, Insurance, Muthoot, OLX).""" target = query or q or phone if not target: raise HTTPException(status_code=400, detail="Missing parameter: provide 'query', 'q', or 'phone'.") res = get_engine().query_xyron_master(target) if isinstance(res, dict): res["tier"] = auth_meta.get("tier", "free") res["remaining_credits"] = auth_meta.get("remaining", -1) return res @app.get("/api/sbi/{query}") @app.get("/api/sbi") async def lookup_sbi( query: Optional[str] = None, q: Optional[str] = Query(None), phone: Optional[str] = Query(None), auth_meta: Dict[str, Any] = Depends(verify_api_key) ): """Search SBI POSP & Indian Banking KYC Master Database.""" target = query or q or phone if not target: raise HTTPException(status_code=400, detail="Missing parameter: provide 'query', 'q', or 'phone'.") res = get_engine().query_sbi(target) if isinstance(res, dict): res["tier"] = auth_meta.get("tier", "free") res["remaining_credits"] = auth_meta.get("remaining", -1) return res @app.get("/api/ind/{query}") @app.get("/api/ind") async def lookup_ind( query: Optional[str] = None, q: Optional[str] = Query(None), phone: Optional[str] = Query(None), auth_meta: Dict[str, Any] = Depends(verify_api_key) ): """Search IND Education Admissions & Student KYC Master Database.""" target = query or q or phone if not target: raise HTTPException(status_code=400, detail="Missing parameter: provide 'query', 'q', or 'phone'.") res = get_engine().query_ind(target) if isinstance(res, dict): res["tier"] = auth_meta.get("tier", "free") res["remaining_credits"] = auth_meta.get("remaining", -1) return res @app.get("/api/lookup") async def unified_lookup( q: Optional[str] = Query(None, description="Search query (phone, aadhar, telegram user_id, username, deep)"), phone: Optional[str] = Query(None, description="Direct 10-digit phone number"), aadhar: Optional[str] = Query(None, description="Direct 12-digit aadhar number"), family: Optional[str] = Query(None, description="Direct Aadhaar family query"), user_id: Optional[str] = Query(None, description="Direct Telegram User ID"), sbi: Optional[str] = Query(None, description="Direct SBI POSP query"), ind: Optional[str] = Query(None, description="Direct IND Admissions & KYC query"), upi: Optional[str] = Query(None, description="Direct UPI ID or VPA query"), gas: Optional[str] = Query(None, description="Direct LPG Gas cylinder or consumer query"), deep: Optional[str] = Query(None, description="Direct 360° Deep Dossier query"), mode: Optional[str] = Query(None, description="Search mode: 'deep' or 'fast'"), auth_meta: Dict[str, Any] = Depends(verify_api_key) ): """ Smart Unified Lookup: Automatically detects input type. - Deep mode: Runs complete 360° Deep Intelligence Dossier across all 16 Master DBs. - 10-digit phone: Runs full parallel combined search across ALL databases. - 12-digit: Searches Aadhaar database. - 16/17/18-digit: Searches LPG Gas database. - Telegram ID: Searches Telegram 65M Master DB. - UPI ID / VPA: Resolves phone, PSP app, bank, and 360° KYC dossier. - Text/Username: Searches Instagram/Twitter. - Family: Discovers family linkages across Aadhaar & Telecom. - Gas: Resolves complete LPG cylinder details across Indane, Bharat Gas, HP Gas. - SBI: Searches SBI POSP & Indian Banking database. - IND: Searches IND Education, Admissions & KYC database. """ eng = get_engine() if deep or mode == "deep": target = deep or q or phone or aadhar or upi or user_id res = eng.query_deep_dossier(target) elif gas: res = ServiceGateway.get_lpg(gas) elif upi: res = eng.query_upi_full_info(upi) elif ind: res = eng.query_ind(ind) elif sbi: res = eng.query_sbi(sbi) elif family: res = eng.query_by_aadhar_family(family) elif user_id: res = eng.query_by_telegram(user_id) elif aadhar: res = eng.query_by_aadhar(aadhar) elif phone: res = eng.query_combined(phone) elif q: q_str = str(q).strip() if "@" in q_str: res = eng.query_upi_full_info(q_str) else: clean = "".join(c for c in q_str if c.isdigit()) if len(clean) in (16, 17, 18): res = ServiceGateway.get_lpg(clean) elif len(clean) == 12: res = eng.query_by_aadhar(clean) elif len(clean) == 10: res = eng.query_combined(clean) elif re.match(r"^[A-Za-z]{5}[0-9]{4}[A-Za-z]$", q_str): res = ServiceGateway.get_pan_details(q_str.upper()) elif 5 <= len(clean) <= 15: tg_res = eng.query_by_telegram(clean) if tg_res.get("status") == "success": res = tg_res else: res = eng.query_combined(clean) else: res = eng.query_instagram(q_str) else: raise HTTPException(status_code=400, detail="Missing parameter: provide 'q', 'phone', 'aadhar', 'upi', or 'user_id'.") if isinstance(res, dict): res["tier"] = auth_meta.get("tier", "free") res["remaining_credits"] = auth_meta.get("remaining", -1) return res @app.get("/api/key/status") async def check_key_status(api_key: str = Query(..., description="API Key to inspect")): status = KeyManager.get_key_status(api_key) if not status: raise HTTPException(status_code=404, detail="API Key not found.") return status # --------------------------------------------------------- # Admin Key Management Endpoints (Master Admin Only) # --------------------------------------------------------- @app.post("/api/admin/keys/create") async def create_api_key(req: CreateKeyRequest, _admin: str = Depends(verify_admin_key)): """Generate a new client API key with custom expiry, credits, and rate limit.""" new_key = KeyManager.create_key( name=req.name, days_valid=req.days_valid, rate_limit_per_min=req.rate_limit_per_min or 60, total_credits=req.total_credits if req.total_credits is not None else 1000 ) return { "success": True, "message": "API key generated successfully.", "key_data": new_key } @app.get("/api/admin/keys") async def list_all_keys(_admin: str = Depends(verify_admin_key)): """List all issued API keys with remaining credits and usage stats.""" return { "success": True, "keys": KeyManager.list_keys() } @app.delete("/api/admin/keys/{target_key}") async def delete_api_key(target_key: str, _admin: str = Depends(verify_admin_key)): """Permanently delete or revoke an API key.""" deleted = KeyManager.delete_key(target_key) if not deleted: raise HTTPException(status_code=404, detail="Key not found.") return {"success": True, "message": "API key deleted successfully."} if __name__ == "__main__": import uvicorn uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)