178 lines
5.6 KiB
Python
178 lines
5.6 KiB
Python
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
|
|
|
|
import logging
|
|
|
|
logging.basicConfig(level=logging.INFO)
|
|
|
|
|
|
# ==================== 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')
|
|
async def get_ip(request: Request):
|
|
forwarded = request.headers.get('X-Forwarded-For')
|
|
user_agent = request.headers.get('User-Agent')
|
|
|
|
if forwarded:
|
|
# Берем ПЕРВЫЙ IP из списка (это реальный клиент)
|
|
client_ip = forwarded.split(',')[0].strip()
|
|
else:
|
|
client_ip = request.client.host
|
|
|
|
async with AsyncClient(headers={'User-Agent': user_agent}) as client:
|
|
response = await client.get(f'https://ipinfo.io/{client_ip}')
|
|
|
|
html = response.text
|
|
logging.info(f'html: {html}')
|
|
ipinfo = IPInfo(html)
|
|
|
|
return ipinfo.dict() |