backend/routers/ip.py

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
from async_lru import alru_cache
from functools import lru_cache

import asyncio
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)
    
    @lru_cache(maxsize=1000)
    def _parse(self, html: str) -> Optional[IPInfoModel]:
        """Парсит HTML и возвращает модель"""
        soup = BeautifulSoup(html, 'html.parser')
        scripts = soup.find_all('script', type='application/ld+json')

        if not scripts:
            return None

        # Ищем первый скрипт с типом DataFeed
        data = None
        for script in scripts:
            if not script.string:
                continue
            
            try:
                parsed = json.loads(script.string.strip())
                if parsed.get('@type') == 'DataFeed':
                    data = parsed
                    break
            except json.JSONDecodeError:
                continue

        if data is None:
            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'])


@alru_cache(maxsize=1000, ttl=86400)
async def get_ip_info(ip: str, user_agent: str):
    """Асинхронная функция для получения информации об IP с кешированием."""
    async with AsyncClient(headers={'User-Agent': user_agent}) as client:
        response = await client.get(f'https://ipinfo.io/{ip}')
        return response.text


@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

    html = await get_ip_info(client_ip, user_agent)

    result = await asyncio.to_thread(lambda: IPInfo(html).dict())
    logging.info(result)

    return result