File size: 2,518 Bytes
29cdbaa | 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 | from fastapi import APIRouter, HTTPException
from pydantic import BaseModel
from typing import Optional
from app.validator import proxy_validator, ValidationResult
router = APIRouter(prefix="/api/v1/validate", tags=["validation"])
class ValidateProxyRequest(BaseModel):
proxy_url: str
ip: Optional[str] = None
class ValidateProxyResponse(BaseModel):
success: bool
latency_ms: Optional[int] = None
anonymity: Optional[str] = None
can_access_google: Optional[bool] = None
country_code: Optional[str] = None
country_name: Optional[str] = None
proxy_type: Optional[str] = None
quality_score: Optional[int] = None
error_message: Optional[str] = None
@router.post("/proxy", response_model=ValidateProxyResponse)
async def validate_proxy(request: ValidateProxyRequest):
if not request.ip:
try:
ip_port = request.proxy_url.replace("http://", "").replace("https://", "").replace("socks4://", "").replace("socks5://", "")
ip = ip_port.split(":")[0]
except Exception:
raise HTTPException(status_code=400, detail="Could not extract IP from proxy_url. Please provide IP explicitly.")
else:
ip = request.ip
try:
result = await proxy_validator.validate_comprehensive(request.proxy_url, ip)
quality_score = None
if result.success:
quality_score = await proxy_validator.calculate_quality_score(
result.latency_ms,
result.anonymity,
result.can_access_google,
result.proxy_type
)
return ValidateProxyResponse(
success=result.success,
latency_ms=result.latency_ms,
anonymity=result.anonymity,
can_access_google=result.can_access_google,
country_code=result.country_code,
country_name=result.country_name,
proxy_type=result.proxy_type,
quality_score=quality_score,
error_message=result.error_message
)
except Exception as e:
raise HTTPException(status_code=500, detail=f"Validation failed: {str(e)}")
@router.post("/proxy/format", response_model=dict)
async def validate_proxy_format(request: ValidateProxyRequest):
is_valid = await proxy_validator.validate_format(request.proxy_url)
return {
"valid": is_valid,
"proxy_url": request.proxy_url
}
|