Spaces:
Sleeping
Sleeping
| 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 | |
| # ---------------------------------------------------------------------- | |
| 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 | |
| } | |
| async def list_models(): | |
| return {"status": True, "available_models": list(MODELS_DB.keys()), **DEV_INFO} | |
| 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}) | |
| 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} | |
| 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} | |
| 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}) | |
| # ---------------------------------------------------------------------- | |
| async def shutdown(): | |
| await client.aclose() |