File size: 4,635 Bytes
950d6aa 00c4bbe 950d6aa 00c4bbe 950d6aa 00c4bbe 950d6aa 00c4bbe 950d6aa |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 |
from fastapi import FastAPI, HTTPException, Depends
from pydantic import BaseModel
from itertools import islice
from typing import Dict, Any, Optional
from fastapi.middleware.cors import CORSMiddleware
from xnxx_api import search_filters
from xnxx_api.xnxx_api import Client as xnxx_client
# Constants
CREATOR = "EyePatch"
API_VERSION = "1.3.0"
# ----- Pydantic Models -----
class SuccessResponse(BaseModel):
creator: str = CREATOR
status: str = "success"
api_version: str = API_VERSION
data: Dict[str, Any]
class ErrorResponse(BaseModel):
status: str = "error"
creator: str = CREATOR
api_version: str = API_VERSION
error_code: int
message: str
class ItemPayload(BaseModel):
name: str
description: Optional[str] = None
price: float
tags: list[str] = []
# ----- FastAPI Setup -----
app = FastAPI(title="Awesome API", version=API_VERSION)
# CORS Configuration
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_methods=["*"],
allow_headers=["*"],
)
# ----- Helper Functions -----
async def get_api_version():
return API_VERSION
# ----- API Endpoints -----
@app.get("/", response_model=SuccessResponse)
async def root(version: str = Depends(get_api_version)):
"""Service health check endpoint"""
return SuccessResponse(data={
"message": "Service operational",
})
@app.get("/prono/xnxxsearch", response_model=SuccessResponse)
async def xnxx_search(
query: str,
quality: Optional[str] = None,
upload_time: Optional[str] = None,
length: Optional[str] = None,
mode: Optional[str] = None,
# page: Optional[int] = None,
results: Optional[int] = 20
):
data_dict = {
"quality": {
"720p": search_filters.SearchingQuality.X_720p,
"1080p": search_filters.SearchingQuality.X_1080p_plus
},
"upload_time": {
"year": search_filters.UploadTime.year,
"month": search_filters.UploadTime.month
},
"length": {
"0-10min": search_filters.Length.X_0_10min,
"10min+": search_filters.Length.X_10min_plus,
"10-20min": search_filters.Length.X_10_20min,
"20min+": search_filters.Length.X_20min_plus
},
"mode": {
"default": search_filters.Mode.default,
"hits": search_filters.Mode.hits,
"random": search_filters.Mode.random
}
}
try:
# Prepare search parameters
search_kwargs = {
"query": query
}
if length is not None:
search_kwargs["length"] = data_dict["length"][length]
if quality is not None:
search_kwargs["searching_quality"] = data_dict["quality"][quality]
if upload_time is not None:
search_kwargs["upload_time"] = data_dict["upload_time"][upload_time]
if mode is not None:
search_kwargs["mode"] = data_dict["mode"][mode]
# if page is not None:
# search_kwargs["limit"] = page
# Perform the search with only the provided parameters
search = xnxx_client().search(**search_kwargs)
res = search.videos
if results is not None:
response = islice(res, results)
results_list = []
for x in response:
results_list.append({
"title": x.title,
"url": x.url,
"author": x.author,
"length": x.length,
"highest_quality": x.highest_quality,
"publish_date": x.publish_date,
"views": x.views,
"thumb": x.thumbnail_url[0] if isinstance(x.thumbnail_url, list) and len(x.thumbnail_url) > 0 else None
})
return SuccessResponse(
status="True",
data={
"results": results_list
}
)
except Exception as e:
return SuccessResponse(
status="False",
data={"error": f"Error: {e}"}
)
@app.get("/protected", responses={
200: {"model": SuccessResponse},
403: {"model": ErrorResponse}
})
async def protected_route(secret_key: Optional[str] = None):
"""Endpoint demonstrating error responses"""
if not secret_key or secret_key != "supersecret123":
return ErrorResponse(
error_code=403,
message="Invalid or missing secret key"
)
return SuccessResponse(data={
"secret_data": "🔐 You've unlocked premium content!",
"access_level": "VIP"
})
|