Spaces:
Running
Running
| import os | |
| import sqlite3 | |
| from fastapi import FastAPI, HTTPException | |
| from huggingface_hub import hf_hub_download | |
| app = FastAPI(title="Alien X OSINT API") | |
| # Secret se token uthana | |
| HF_TOKEN = os.environ.get("HF_TOKEN") | |
| def get_db(): | |
| print("⏳ Downloading database...") | |
| db_path = hf_hub_download( | |
| repo_id="Watchhrr/Pakistan-Master-Vault", | |
| filename="Pakistan_Master_Vault.db", | |
| repo_type="dataset", | |
| token=HF_TOKEN | |
| ) | |
| return db_path | |
| DB_FILE = get_db() | |
| def home(): | |
| return {"message": "🚀 Alien X OSINT API is Live!", "usage": "/search?pnum=92xxx"} | |
| def search(pnum: str = None, cnic: str = None): | |
| try: | |
| conn = sqlite3.connect(DB_FILE) | |
| cursor = conn.cursor() | |
| if pnum: | |
| cursor.execute("SELECT * FROM vault WHERE pnum = ?", (pnum,)) | |
| elif cnic: | |
| cursor.execute("SELECT * FROM vault WHERE cnic = ?", (cnic,)) | |
| else: | |
| raise HTTPException(status_code=400, detail="Provide pnum or cnic") | |
| row = cursor.fetchone() | |
| conn.close() | |
| if row: | |
| return { | |
| "status": "success", | |
| "data": { | |
| "pnum": row[0], | |
| "cnic": row[1], | |
| "name": row[2], | |
| "address": row[3], | |
| "source": row[4] | |
| } | |
| } | |
| return {"status": "not_found"} | |
| except Exception as e: | |
| raise HTTPException(status_code=500, detail=str(e)) | |