ptt-crawler/scraptt/spiders/utils/parsers/posts/ip/location.py

import re
import requests
from typing import Tuple
from ......models import redis_cli
from requests.adapters import HTTPAdapter, Retry
from scrapy.http.response.html import HtmlResponse


def fetch_ip(ip: str) -> Tuple[str, str]:
    session = requests.Session()
    retries = Retry(total=5, backoff_factor=1, status_forcelist=[502, 503, 504])
    session.mount("http://", HTTPAdapter(max_retries=retries))

    response = requests.get(f"http://www.geoplugin.net/json.gp?ip={ip}")
    result = response.json()
    city = result["geoplugin_city"]
    country = result["geoplugin_countryName"]
    return city, country


def fetch_ip_with_cache(ip) -> Tuple[str, str]:
    ip_result = redis_cli.hgetall(ip)

    if ip_result:
        city = ip_result["city"]
        country = ip_result["country"]
        return city, country

    city, country = fetch_ip(ip)

    ttl = 10 * 60
    redis_cli.hset(ip, mapping={"city": city, "country": country})
    redis_cli.expire(name=ip, time=ttl)

    return city, country


def get_ip_loc(ip: str, ip_cache: bool) -> Tuple[str, str]:
    """The get_ip_loc function get info from an ip.

    Args:
        ip (str): the ip from a ptt post.
        ip_cache (bool): a boolean that is used to determine whether
                         to enable redis service to cache ip.
    Returns:
        a tuple, containing city and country
    """
    if ip_cache:
        return fetch_ip_with_cache(ip)

    return fetch_ip(ip)