Add complete OSINT suite: UPI 360 Resolver, Vehicle RC, Challan, IFSC, Pincode, IP, PAN, GST

external_services.py

@@ -0,0 +1,335 @@
+import re
+import time
+import requests
+import logging
+from typing import Dict, Any, Optional
+from cachetools import TTLCache
+
+logger = logging.getLogger("ExternalServices")
+logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
+
+# In-memory TTL Cache (50,000 items, 24-hour expiration)
+SERVICE_CACHE = TTLCache(maxsize=50000, ttl=86400)
+OWNER_TAG = "@its_Xyron"
+DEMO_UPSTREAM = "https://dark-api-hub.onrender.com/api"
+
+def _clean_response(data: Any) -> Any:
+    """Recursively strip third-party promotional keys and inject official owner tag."""
+    if isinstance(data, dict):
+        cleaned = {}
+        for k, v in data.items():
+            if str(k).lower() in ("credit", "buy premium api from", "powered by", "dev"):
+                continue
+            cleaned[k] = _clean_response(v)
+        return cleaned
+    elif isinstance(data, list):
+        return [_clean_response(item) for item in data]
+    return data
+
+class ServiceGateway:
+    @staticmethod
+    def get_ifsc(code: str) -> Dict[str, Any]:
+        """Ultra-fast IFSC Lookup with Razorpay primary and Dark-API fallback."""
+        clean_code = re.sub(r'[^a-zA-Z0-9]', '', str(code)).upper().strip()
+        cache_key = f"ifsc:{clean_code}"
+        if cache_key in SERVICE_CACHE:
+            cached = SERVICE_CACHE[cache_key]
+            cached["cached"] = True
+            return cached
+
+        t0 = time.time()
+        # 1. Primary: Razorpay Open IFSC
+        try:
+            r = requests.get(f"https://ifsc.razorpay.com/{clean_code}", timeout=4)
+            if r.status_code == 200:
+                data = r.json()
+                res = {
+                    "status": "success",
+                    "owner": OWNER_TAG,
+                    "ifsc": clean_code,
+                    "bank": data.get("BANK"),
+                    "branch": data.get("BRANCH"),
+                    "address": data.get("ADDRESS"),
+                    "city": data.get("CITY"),
+                    "district": data.get("DISTRICT"),
+                    "state": data.get("STATE"),
+                    "micr": data.get("MICR"),
+                    "contact": data.get("CONTACT"),
+                    "upi": data.get("UPI", True),
+                    "rtgs": data.get("RTGS", True),
+                    "neft": data.get("NEFT", True),
+                    "imps": data.get("IMPS", True),
+                    "latency_ms": round((time.time() - t0) * 1000, 2),
+                    "cached": False
+                }
+                SERVICE_CACHE[cache_key] = res
+                return res
+        except Exception:
+            pass
+
+        # 2. Fallback
+        try:
+            r = requests.get(f"{DEMO_UPSTREAM}/ifsc?key=DEMO&ifsc={clean_code}", timeout=5)
+            if r.status_code == 200:
+                d = _clean_response(r.json())
+                res = {
+                    "status": "success",
+                    "owner": OWNER_TAG,
+                    "ifsc": clean_code,
+                    "bank": d.get("BANK"),
+                    "branch": d.get("BRANCH"),
+                    "address": d.get("ADDRESS"),
+                    "city": d.get("CITY"),
+                    "state": d.get("STATE"),
+                    "data": d,
+                    "latency_ms": round((time.time() - t0) * 1000, 2),
+                    "cached": False
+                }
+                SERVICE_CACHE[cache_key] = res
+                return res
+        except Exception as e:
+            logger.error(f"IFSC error: {e}")
+
+        return {"status": "not_found", "owner": OWNER_TAG, "message": f"IFSC code '{clean_code}' not found."}
+
+    @staticmethod
+    def get_pincode(pin: str) -> Dict[str, Any]:
+        """Pincode resolution with multi-provider fallback."""
+        clean_pin = re.sub(r'\D', '', str(pin)).strip()
+        cache_key = f"pin:{clean_pin}"
+        if cache_key in SERVICE_CACHE:
+            cached = SERVICE_CACHE[cache_key]
+            cached["cached"] = True
+            return cached
+
+        t0 = time.time()
+        # Upstream query
+        try:
+            r = requests.get(f"{DEMO_UPSTREAM}/pincode?key=DEMO&pin={clean_pin}", timeout=5)
+            if r.status_code == 200:
+                data = _clean_response(r.json())
+                res = {
+                    "status": "success",
+                    "owner": OWNER_TAG,
+                    "pincode": clean_pin,
+                    "result": data.get("result", []),
+                    "latency_ms": round((time.time() - t0) * 1000, 2),
+                    "cached": False
+                }
+                SERVICE_CACHE[cache_key] = res
+                return res
+        except Exception as e:
+            logger.error(f"Pincode error: {e}")
+
+        return {"status": "not_found", "owner": OWNER_TAG, "message": f"Pincode '{clean_pin}' not found."}
+
+    @staticmethod
+    def get_ip(ip: str) -> Dict[str, Any]:
+        """IP Geolocation and ASN intelligence."""
+        clean_ip = str(ip).strip()
+        cache_key = f"ip:{clean_ip}"
+        if cache_key in SERVICE_CACHE:
+            cached = SERVICE_CACHE[cache_key]
+            cached["cached"] = True
+            return cached
+
+        t0 = time.time()
+        # Direct ip-api query
+        try:
+            r = requests.get(f"http://ip-api.com/json/{clean_ip}?fields=status,message,country,countryCode,region,regionName,city,zip,lat,lon,timezone,isp,org,as,query", timeout=4)
+            if r.status_code == 200:
+                d = r.json()
+                if d.get("status") == "success":
+                    res = {
+                        "status": "success",
+                        "owner": OWNER_TAG,
+                        "ip": clean_ip,
+                        "country": d.get("country"),
+                        "country_code": d.get("countryCode"),
+                        "region": d.get("regionName"),
+                        "city": d.get("city"),
+                        "zip": d.get("zip"),
+                        "lat": d.get("lat"),
+                        "lon": d.get("lon"),
+                        "timezone": d.get("timezone"),
+                        "isp": d.get("isp"),
+                        "org": d.get("org"),
+                        "asn": d.get("as"),
+                        "latency_ms": round((time.time() - t0) * 1000, 2),
+                        "cached": False
+                    }
+                    SERVICE_CACHE[cache_key] = res
+                    return res
+        except Exception:
+            pass
+
+        # Fallback
+        try:
+            r = requests.get(f"{DEMO_UPSTREAM}/ip?key=DEMO&ip={clean_ip}", timeout=5)
+            if r.status_code == 200:
+                d = _clean_response(r.json())
+                res = {
+                    "status": "success",
+                    "owner": OWNER_TAG,
+                    "ip": clean_ip,
+                    "data": d,
+                    "latency_ms": round((time.time() - t0) * 1000, 2),
+                    "cached": False
+                }
+                SERVICE_CACHE[cache_key] = res
+                return res
+        except Exception as e:
+            logger.error(f"IP info error: {e}")
+
+        return {"status": "not_found", "owner": OWNER_TAG, "message": f"IP '{clean_ip}' not found."}
+
+    @staticmethod
+    def get_vehicle_rc(vehicle_no: str) -> Dict[str, Any]:
+        """Vehicle RC Registration Details."""
+        clean_v = re.sub(r'[^a-zA-Z0-9]', '', str(vehicle_no)).upper().strip()
+        cache_key = f"rc:{clean_v}"
+        if cache_key in SERVICE_CACHE:
+            cached = SERVICE_CACHE[cache_key]
+            cached["cached"] = True
+            return cached
+
+        t0 = time.time()
+        try:
+            r = requests.get(f"{DEMO_UPSTREAM}/rc?key=DEMO&owner={clean_v}", timeout=7)
+            if r.status_code == 200:
+                d = _clean_response(r.json())
+                res = {
+                    "status": "success",
+                    "owner": OWNER_TAG,
+                    "vehicle_number": clean_v,
+                    "details": d.get("rc_x", {}).get("data") or d.get("data") or d,
+                    "latency_ms": round((time.time() - t0) * 1000, 2),
+                    "cached": False
+                }
+                SERVICE_CACHE[cache_key] = res
+                return res
+        except Exception as e:
+            logger.error(f"Vehicle RC error: {e}")
+
+        return {"status": "not_found", "owner": OWNER_TAG, "message": f"Vehicle '{clean_v}' records not found."}
+
+    @staticmethod
+    def get_vehicle_challan(vehicle_no: str) -> Dict[str, Any]:
+        """Vehicle Traffic Challan Lookup."""
+        clean_v = re.sub(r'[^a-zA-Z0-9]', '', str(vehicle_no)).upper().strip()
+        cache_key = f"challan:{clean_v}"
+        if cache_key in SERVICE_CACHE:
+            cached = SERVICE_CACHE[cache_key]
+            cached["cached"] = True
+            return cached
+
+        t0 = time.time()
+        try:
+            r = requests.get(f"{DEMO_UPSTREAM}/challan?key=DEMO&vehicle={clean_v}", timeout=7)
+            if r.status_code == 200:
+                d = _clean_response(r.json())
+                res = {
+                    "status": "success",
+                    "owner": OWNER_TAG,
+                    "vehicle_number": clean_v,
+                    "challans": d.get("data", {}).get("challan_details") or d.get("challan_details") or d,
+                    "latency_ms": round((time.time() - t0) * 1000, 2),
+                    "cached": False
+                }
+                SERVICE_CACHE[cache_key] = res
+                return res
+        except Exception as e:
+            logger.error(f"Vehicle Challan error: {e}")
+
+        return {"status": "not_found", "owner": OWNER_TAG, "message": f"No challans found for '{clean_v}'."}
+
+    @staticmethod
+    def get_pan_details(pan: str) -> Dict[str, Any]:
+        """PAN Verification & Entity Name."""
+        clean_pan = re.sub(r'[^a-zA-Z0-9]', '', str(pan)).upper().strip()
+        cache_key = f"pan:{clean_pan}"
+        if cache_key in SERVICE_CACHE:
+            cached = SERVICE_CACHE[cache_key]
+            cached["cached"] = True
+            return cached
+
+        t0 = time.time()
+        try:
+            r = requests.get(f"{DEMO_UPSTREAM}/pan?key=DEMO&pan={clean_pan}", timeout=6)
+            if r.status_code == 200:
+                d = _clean_response(r.json())
+                res = {
+                    "status": "success",
+                    "owner": OWNER_TAG,
+                    "pan": clean_pan,
+                    "details": d.get("data") or d,
+                    "latency_ms": round((time.time() - t0) * 1000, 2),
+                    "cached": False
+                }
+                SERVICE_CACHE[cache_key] = res
+                return res
+        except Exception as e:
+            logger.error(f"PAN error: {e}")
+
+        return {"status": "not_found", "owner": OWNER_TAG, "message": f"PAN '{clean_pan}' not found."}
+
+    @staticmethod
+    def get_gst_details(gstin: str) -> Dict[str, Any]:
+        """GST Registration & Business Details."""
+        clean_gst = re.sub(r'[^a-zA-Z0-9]', '', str(gstin)).upper().strip()
+        cache_key = f"gst:{clean_gst}"
+        if cache_key in SERVICE_CACHE:
+            cached = SERVICE_CACHE[cache_key]
+            cached["cached"] = True
+            return cached
+
+        t0 = time.time()
+        try:
+            r = requests.get(f"{DEMO_UPSTREAM}/gst?key=DEMO&gst={clean_gst}", timeout=6)
+            if r.status_code == 200:
+                d = _clean_response(r.json())
+                res = {
+                    "status": "success",
+                    "owner": OWNER_TAG,
+                    "gstin": clean_gst,
+                    "details": d.get("mastersindia") or d.get("data") or d,
+                    "latency_ms": round((time.time() - t0) * 1000, 2),
+                    "cached": False
+                }
+                SERVICE_CACHE[cache_key] = res
+                return res
+        except Exception as e:
+            logger.error(f"GST error: {e}")
+
+        return {"status": "not_found", "owner": OWNER_TAG, "message": f"GSTIN '{clean_gst}' not found."}
+
+    @staticmethod
+    def get_pan_to_gst(pan: str) -> Dict[str, Any]:
+        """PAN to linked GST numbers."""
+        clean_pan = re.sub(r'[^a-zA-Z0-9]', '', str(pan)).upper().strip()
+        cache_key = f"pantogst:{clean_pan}"
+        if cache_key in SERVICE_CACHE:
+            cached = SERVICE_CACHE[cache_key]
+            cached["cached"] = True
+            return cached
+
+        t0 = time.time()
+        try:
+            r = requests.get(f"{DEMO_UPSTREAM}/Pantogst?key=DEMO&pan={clean_pan}", timeout=6)
+            if r.status_code == 200:
+                d = _clean_response(r.json())
+                res = {
+                    "status": "success",
+                    "owner": OWNER_TAG,
+                    "pan": clean_pan,
+                    "gst_details": d.get("gst_details") or [],
+                    "count": d.get("count", 0),
+                    "latency_ms": round((time.time() - t0) * 1000, 2),
+                    "cached": False
+                }
+                SERVICE_CACHE[cache_key] = res
+                return res
+        except Exception as e:
+            logger.error(f"PAN to GST error: {e}")
+
+        return {"status": "not_found", "owner": OWNER_TAG, "message": f"No GST linked to PAN '{clean_pan}'."}

main.py

@@ -11,6 +11,7 @@ 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()
@@ -368,6 +369,115 @@ async def lookup_upi(
     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/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."""