import json from typing import Optional, Dict, Any, Tuple from bs4 import BeautifulSoup from pydantic import BaseModel, Field from fastapi import APIRouter, Request from httpx import AsyncClient # ==================== Pydantic модели ==================== class LocationModel(BaseModel): """Геоданные""" city: Optional[str] = None region: Optional[str] = None country: Optional[str] = None postal_code: Optional[str] = None latitude: Optional[float] = None longitude: Optional[float] = None timezone: Optional[str] = None class ASNModel(BaseModel): """ASN данные""" asn: Optional[str] = None name: Optional[str] = None route: Optional[str] = None class CompanyModel(BaseModel): """Данные компании""" name: Optional[str] = None abuse_contact: Optional[str] = None class SecurityModel(BaseModel): """Данные безопасности""" vpn: bool = False proxy: bool = False tor: bool = False hosting: bool = False class IPInfoModel(BaseModel): """Главная модель IP информации""" ip: str url: Optional[str] = None location: LocationModel = Field(default_factory=LocationModel) asn: ASNModel = Field(default_factory=ASNModel) company: CompanyModel = Field(default_factory=CompanyModel) security: SecurityModel = Field(default_factory=SecurityModel) # ==================== Парсер ==================== class IPInfo: """Парсер IP информации из JSON-LD""" def __init__(self, html: str): self._data = self._parse(html) def _parse(self, html: str) -> Optional[IPInfoModel]: """Парсит HTML и возвращает модель""" soup = BeautifulSoup(html, 'html.parser') script = soup.find('script', type='application/ld+json') if not script or not script.string: return None try: data: Dict = json.loads(script.string.strip()) except json.JSONDecodeError: return None if data.get('@type') != 'DataFeed': return None # Базовые поля result = IPInfoModel( ip=data.get('name', ''), url=data.get('url') ) # Парсим локацию location = data.get('contentLocation', {}) address = location.get('address', {}) geo = location.get('geo', {}) result.location = LocationModel( city=address.get('addressLocality'), region=address.get('addressRegion'), country=address.get('addressCountry'), postal_code=address.get('postalCode'), latitude=geo.get('latitude'), longitude=geo.get('longitude'), timezone=self._find_value(data, 'Timezone') ) # Парсим ASN result.asn = ASNModel( asn=self._find_value(data, 'ASN'), name=self._find_value(data, 'AS Name'), route=self._find_value(data, 'AS Route') ) # Парсим компанию result.company = CompanyModel( name=self._find_value(data, 'Company Name'), abuse_contact=self._find_value(data, 'Abuse Contact') ) # Парсим безопасность result.security = SecurityModel( vpn=self._find_bool(data, 'VPN'), proxy=self._find_bool(data, 'Proxy'), tor=self._find_bool(data, 'Tor'), hosting=self._find_bool(data, 'Hosting') ) return result def _find_value(self, data: Dict, name: str) -> Optional[str]: """Поиск значения в variableMeasured""" for item in data.get('variableMeasured', []): if item.get('name') == name: return item.get('value') return None def _find_bool(self, data: Dict, name: str) -> bool: """Поиск булевого значения""" value = self._find_value(data, name) return value == 'Yes' def dict(self) -> Optional[Dict[str, Any]]: """Возвращает словарь с данными""" if self._data: return self._data.model_dump(exclude_none=True) return None @property def is_valid(self) -> bool: """Проверяет, удалось ли распарсить данные""" return self._data is not None @property def coordinates(self) -> Optional[Tuple[float, float]]: """Возвращает координаты как tuple""" if self._data and self._data.location.latitude and self._data.location.longitude: return (self._data.location.latitude, self._data.location.longitude) return None # ==================== Роутер ==================== router = APIRouter(prefix='api/v1', tags=['ip']) @router.get('/ip', response_model=IPInfoModel) async def get_ip(request: Request): ip = request.headers.get('X-Forwarded-For') async with AsyncClient() as client: response = await client.get(f'https://ipinfo.io/{ip}') html = response.text ipinfo = IPInfo(html) return ipinfo._data