ip.py

#! /usr/bin/env python3

"""
This module is used for checking IP GeoLocations of users.
Install Pandas and Requests beforehand via: `pip install pandas requests`
Also remember to enter your API key in line 26.

Sample Command:
$ sudo journalctl -u xray --since "5 min ago" | grep "admin" | cut -d " " -f 6,7,8 | python3 ip.py -

GitHub: https://github.com/Mahyar24/V2Conf
Mahyar@Mahyar24.com, Mon 28 Nov 2022
"""


import sys
from functools import cache
from typing import Final

import pandas as pd
import requests
# from rich_dataframe import prettify


# Enter your own API key.
API_KEY: Final[str] = "*****"


@cache
def get_geolocation(ip: str) -> pd.Series:
    global API_KEY
    link = f"https://api.ipgeolocation.io/ipgeo?apiKey={API_KEY}&ip={ip}"
    data = requests.get(link).json()
    return pd.Series(data).loc[["country_name", "city", "latitude", "longitude", "isp"]]


def make_df(input_data, time_interval: int = 5) -> pd.DataFrame:
    return (
        pd.read_csv(input_data, delimiter=" ", header=None)
        .rename({0: "date", 1: "time", 2: "ip"}, axis=1)
        .assign(
            **{
                "date_time": lambda df_: pd.to_datetime(
                    df_.pop("date") + " " + df_.pop("time")
                ),
                "ip": lambda df_: df_.pop("ip")
                .str.removeprefix("tcp:")
                .str.removeprefix("udp:")
                .str.split(":")
                .str[0],
            }
        )
        .groupby(pd.Grouper(key="date_time", freq=f"{time_interval} min"))["ip"]
        .agg(lambda ser_: ser_.mode().min())
        .loc[lambda ser_: ser_.shift() != ser_]
        .dropna()
        .pipe(
            lambda ser_: pd.concat(
                [
                    ser_.reset_index(),
                    pd.concat(
                        ser_.map(get_geolocation).reset_index(drop=True).to_list(),
                        axis=1,
                    ).T,
                ],
                axis=1,
            )
        )
        .set_index("date_time")
    )


if __name__ == "__main__":
    print(make_df(sys.stdin))