import asyncio import httpx import re from datetime import datetime from urllib.parse import quote_plus from fastapi import FastAPI, Query, HTTPException from fastapi.middleware.cors import CORSMiddleware app = FastAPI(title="WASIF CLOUD Complete API") app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"], ) DEV_INFO = { "developer": "WASIF ALI", "telegram": "@FREEHACKS95", "brand": "WASIF CLOUD" } client = httpx.AsyncClient(timeout=30.0) # ---------------------------------------------------------------------- # 1. AI Hub – using only the specified endpoints # ---------------------------------------------------------------------- MODELS_DB = { "gpt4o": ["https://apis.prexzyvilla.site/ai/gpt4o?q={}"], "gpt4-turbo": ["https://apis.prexzyvilla.site/ai/gpt4?q={}"], "gpt3.5": ["https://apis.prexzyvilla.site/ai/gpt3?q={}"], "gemini-1.5-pro": ["https://gemini.rudyy.workers.dev/chat?message={}"], "wormgpt": ["https://ai-chat.apisimpacientes.workers.dev/chat?model=wormgpt&prompt={}"] } def extract_text_from_json(data): if isinstance(data, str): return data.strip() if data.strip() else None if isinstance(data, dict): for key in ["answer", "out", "response", "result", "text", "content", "message", "data"]: if key in data: val = data[key] extracted = extract_text_from_json(val) if extracted: return extracted for val in data.values(): extracted = extract_text_from_json(val) if extracted: return extracted if isinstance(data, list): for item in data: extracted = extract_text_from_json(item) if extracted: return extracted return None async def fetch_ai(model_key: str, prompt: str): urls = MODELS_DB.get(model_key) if not urls: return {"ok": False, "error": "Model not found"} for url_template in urls: try: url = url_template.format(quote_plus(prompt)) resp = await client.get(url) if resp.status_code == 200: try: data = resp.json() except: text = resp.text.strip() if text: return {"ok": True, "answer": text} continue answer = extract_text_from_json(data) if answer: return {"ok": True, "answer": answer} except Exception: continue return {"ok": False} # ---------------------------------------------------------------------- # 2. Instagram Info (unchanged) # ---------------------------------------------------------------------- async def instagram_info(username: str): url = f"https://i.instagram.com/api/v1/users/web_profile_info/?username={username}" headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/114.0.0.0 Safari/537.36", "x-ig-app-id": "936619743392459", "Accept": "application/json", "Referer": f"https://www.instagram.com/{username}/" } try: resp = await client.get(url, headers=headers) if resp.status_code == 404: return {"error": "not_found"} if not resp.ok: return {"error": "http_error", "code": resp.status_code} data = resp.json() udata = data.get("data", {}).get("user") if not udata: return {"error": "no_data"} media = udata.get("edge_owner_to_timeline_media", {}) edges = media.get("edges", []) recent = [] for edge in edges[:8]: node = edge.get("node", {}) recent.append({ "id": node.get("id"), "code": node.get("shortcode"), "img": node.get("display_url"), "cap": node.get("edge_media_to_caption", {}).get("edges", [{}])[0].get("node", {}).get("text") }) return { "id": udata.get("id"), "username": udata.get("username"), "name": udata.get("full_name"), "bio": udata.get("biography"), "verified": udata.get("is_verified", False), "private": udata.get("is_private", False), "pic": udata.get("profile_pic_url_hd") or udata.get("profile_pic_url"), "followers": udata.get("edge_followed_by", {}).get("count", 0), "following": udata.get("edge_follow", {}).get("count", 0), "posts": media.get("count", 0), "recent": recent } except: return {"error": "request_failed"} # ---------------------------------------------------------------------- # 3. TikTok Dedicated Endpoint (working) # ---------------------------------------------------------------------- async def tiktok_info(video_url: str): apis = [ f"https://tikwm.com/api/?url={quote_plus(video_url)}", f"https://api.tikmate.app/api/lookup?url={quote_plus(video_url)}", f"https://tikdown.org/api/ajaxSearch?url={quote_plus(video_url)}" ] for api_url in apis: try: resp = await client.get(api_url, timeout=15) if resp.status_code != 200: continue data = resp.json() if "tikwm.com" in api_url: if data.get("code") == 0 and data.get("data"): d = data["data"] return { "success": True, "title": d.get("title"), "cover": d.get("cover"), "duration": d.get("duration"), "play_count": d.get("play_count"), "video_no_watermark": d.get("play"), "video_watermark": d.get("wmplay"), "music": d.get("music") } elif "tikmate.app" in api_url: if data.get("success"): return { "success": True, "title": data.get("title"), "cover": data.get("thumbnail"), "duration": data.get("duration"), "video_no_watermark": data.get("video_no_watermark") or data.get("video"), "music": data.get("music") } elif "tikdown.org" in api_url: if data.get("success"): return { "success": True, "title": data.get("title"), "cover": data.get("thumbnail"), "duration": data.get("duration"), "video_no_watermark": data.get("video"), "music": data.get("music") } except: continue return {"error": "failed_to_fetch", "msg": "All TikTok APIs failed"} # ---------------------------------------------------------------------- # 4. SIM Database (without FreshSim) # ---------------------------------------------------------------------- NETWORKS = { '0300': 'Jazz', '0301': 'Jazz', '0302': 'Jazz', '0303': 'Jazz', '0304': 'Jazz', '0305': 'Jazz', '0306': 'Jazz', '0307': 'Jazz', '0308': 'Jazz', '0309': 'Jazz', '0310': 'Zong', '0311': 'Zong', '0312': 'Zong', '0313': 'Zong', '0314': 'Zong', '0315': 'Zong', '0316': 'Zong', '0317': 'Zong', '0318': 'Zong', '0319': 'Zong', '0320': 'Warid', '0321': 'Warid', '0322': 'Warid', '0323': 'Warid', '0324': 'Warid', '0325': 'Warid', '0330': 'Ufone', '0331': 'Ufone', '0332': 'Ufone', '0333': 'Ufone', '0334': 'Ufone', '0335': 'Ufone', '0336': 'Ufone', '0340': 'Telenor', '0341': 'Telenor', '0342': 'Telenor', '0343': 'Telenor', '0344': 'Telenor', '0345': 'Telenor', '0346': 'Telenor', '0347': 'Telenor', '0355': 'Jazz' } def detect_network(phone: str) -> str: if not phone: return 'Unknown' clean = re.sub(r'\D', '', phone) if clean.startswith('0'): with_zero = clean else: with_zero = '0' + clean prefix = with_zero[:4] return NETWORKS.get(prefix, 'Unknown') def clean_number(phone: str) -> str: return re.sub(r'\D', '', phone) def is_valid_cnic(s: str) -> bool: return len(s) == 13 and s.isdigit() def is_valid_record(record: dict) -> bool: mobile = record.get("Mobile", "") name = record.get("Name", "") return bool(mobile and mobile.strip() and '*' not in mobile and name and name.strip() and '*' not in name) def generate_number_formats(number: str): clean = clean_number(number) formats = set() if clean.startswith('0'): formats.add(clean) formats.add(clean[1:]) formats.add('92' + clean[1:]) else: formats.add('0' + clean) formats.add(clean) formats.add('92' + clean) if clean.startswith('92'): formats.add(clean) formats.add('0' + clean[2:]) formats.add(clean[2:]) return list(formats) SIM_APIS = [ { "name": "BlackSim", "url": "https://blacksimdetail.vercel.app/public_apis/simdetailsapi.php", "method": "GET", "params": lambda q: {"number": q}, "parse": lambda d: [{"Mobile": item.get("Mobile") or item.get("number") or item.get("phone", ""), "Name": item.get("Name") or item.get("name") or item.get("owner", ""), "CNIC": item.get("CNIC") or item.get("cnic", ""), "Address": item.get("Address") or item.get("address", "")} for item in d.get("data", [])] if d.get("status") == "success" else [] }, { "name": "PsychoSim", "url": "https://sychosimdatabase.vercel.app/api/lookup", "method": "GET", "params": lambda q: {"query": q}, "parse": lambda d: [{"Mobile": item.get("number") or item.get("phone", ""), "Name": item.get("name") or item.get("owner", ""), "CNIC": item.get("cnic") or item.get("nic", ""), "Address": item.get("address", "")} for item in (d.get("records") or d.get("data") or [])] }, { "name": "Amscript", "url": "https://amscript.xyz/PublicApi/Siminfo.php", "method": "GET", "params": lambda q: {"number": q}, "parse": lambda d: [{"Mobile": item.get("phone", ""), "Name": item.get("full_name", ""), "CNIC": item.get("cnic", ""), "Address": item.get("address", "")} for item in d.get("data", [])] if d.get("success") else [] }, { "name": "WasifAPI", "url": "https://wasif-ali-sim.vercel.app/api/data", "method": "GET", "params": lambda q: {"phone": q}, "parse": lambda d: [{"Mobile": item.get("Mobile", ""), "Name": item.get("Name", ""), "CNIC": item.get("CNIC", ""), "Address": item.get("Address", "")} for item in d.get("records", [])] if d.get("success") else [] } ] async def fetch_from_api(api: dict, query: str): try: params = api["params"](query) url = api["url"] if api["method"] == "GET": resp = await client.get(url, params=params) else: resp = await client.post(url, data=params) if resp.status_code != 200: return [] data = resp.json() records = api["parse"](data) for r in records: r["_source"] = api["name"] return records except: return [] async def search_sim(phone_input: str): clean = clean_number(phone_input) if not clean or len(clean) < 10: raise ValueError("INVALID_INPUT") is_cnic = is_valid_cnic(clean) all_records = [] if is_cnic: for api in SIM_APIS: recs = await fetch_from_api(api, clean) all_records.extend(recs) else: formats = generate_number_formats(clean) for fmt in formats: for api in SIM_APIS: recs = await fetch_from_api(api, fmt) all_records.extend(recs) valid = [r for r in all_records if is_valid_record(r)] unique = {} for rec in valid: mob = rec.get("Mobile", "") norm = clean_number(mob) if norm.startswith('92'): norm = '0' + norm[2:] elif not norm.startswith('0'): norm = '0' + norm norm = norm[:11] if norm not in unique: rec["Network"] = detect_network(mob) del rec["_source"] unique[norm] = rec return list(unique.values()) # ---------------------------------------------------------------------- # API Endpoints # ---------------------------------------------------------------------- @app.get("/") async def home(): return { "service": "WASIF CLOUD Complete API", "endpoints": { "/api/ai/list": "List AI models", "/api/ai/master": "AI chat (gpt4o, gpt4-turbo, gpt3.5, gemini-1.5-pro, wormgpt)", "/api/instagram": "Instagram profile info", "/api/tiktok": "TikTok video info (dedicated)", "/api/sim": "SIM database lookup (phone or CNIC)" }, **DEV_INFO } @app.get("/api/ai/list") async def list_models(): return {"status": True, "available_models": list(MODELS_DB.keys()), **DEV_INFO} @app.get("/api/ai/master") async def ai_master(model: str = Query(...), prompt: str = Query(...)): model_key = model.lower() if model_key not in MODELS_DB: raise HTTPException(status_code=400, detail={"status": False, "msg": f"Model '{model}' not found.", **DEV_INFO}) start = datetime.now() result = await fetch_ai(model_key, prompt) duration = (datetime.now() - start).total_seconds() if result["ok"]: return { "status": True, "model": model, "response": result["answer"], "execution_time": f"{duration:.2f}s", **DEV_INFO } else: raise HTTPException(status_code=503, detail={"status": False, "msg": "All AI services are currently unavailable.", **DEV_INFO}) @app.get("/api/instagram") async def instagram(username: str = Query(...)): data = await instagram_info(username) if "error" in data: status = 404 if data["error"] == "not_found" else 400 raise HTTPException(status_code=status, detail={**data, **DEV_INFO}) return {**data, **DEV_INFO} @app.get("/api/tiktok") async def tiktok(url: str = Query(...)): data = await tiktok_info(url) if "error" in data: raise HTTPException(status_code=400, detail={**data, **DEV_INFO}) return {**data, **DEV_INFO} @app.get("/api/sim") async def sim(phone: str = Query(...)): try: records = await search_sim(phone) if not records: return {"success": True, "count": 0, "message": "No valid records found", "records": [], **DEV_INFO} return {"success": True, "count": len(records), "records": records, **DEV_INFO} except ValueError: raise HTTPException(status_code=400, detail={"error": "Invalid phone number or CNIC format", **DEV_INFO}) except Exception: raise HTTPException(status_code=500, detail={"error": "Internal server error", **DEV_INFO}) # ---------------------------------------------------------------------- @app.on_event("shutdown") async def shutdown(): await client.aclose()