Python module to lookup location, organisation and ISP from an IP

locate_ips.py

"""Look up the geographic location of IP addresses found in a text file.
A self-contained Python module needing only the standard library.

The input is expected to contain pipe-separated lines where the IP address
is the last column, e.g.:

    2779 | t           |          |           |       1 | 217.61.227.209

Lines without a parseable IP in the last column are skipped (handles psql
header / separator rows naturally). Private, loopback, link-local and
reserved IPs are reported as such without contacting the external API.

Lookups use ip-api.com's free batch endpoint:

    http://ip-api.com/batch

Limits at time of writing: ~15 batch requests / min, up to 100 IPs per
batch (~1500 IPs / min sustained). HTTPS requires the paid tier.

Usage as a script:
    python locate_ips.py path/to/file.txt

Usage as a module:
    from locate_ips import lookup_ips_in_file
    for ip, location in lookup_ips_in_file("ips.txt"):
        ...
"""
from __future__ import annotations

import argparse
import ipaddress
import json
import sys
import urllib.error
import urllib.request
from collections.abc import Iterable, Iterator
from typing import Any

_API_URL = "http://ip-api.com/batch"
_BATCH_SIZE = 100
_HTTP_TIMEOUT_S = 10


def extract_ips(lines: Iterable[str]) -> Iterator[str]:
    """Yield the IP token from the last pipe-delimited column of each line.

    Lines without a parseable IP in that column are skipped silently.
    """
    for line in lines:
        stripped = line.strip()
        if not stripped or "|" not in stripped:
            continue
        candidate = stripped.rsplit("|", 1)[-1].strip()
        try:
            ipaddress.ip_address(candidate)
        except ValueError:
            continue
        yield candidate


def _lookup_batch(ips: list[str]) -> list[dict[str, Any]]:
    """Resolve a batch of IPs via ip-api.com. Returns one dict per IP, in order."""
    body = json.dumps([{"query": ip} for ip in ips]).encode()
    request = urllib.request.Request(
        _API_URL,
        data=body,
        headers={"Content-Type": "application/json"},
    )
    with urllib.request.urlopen(request, timeout=_HTTP_TIMEOUT_S) as response:
        return json.loads(response.read())


def _format_result(result: dict[str, Any]) -> str:
    parts = [
        result.get("city") or "",
        result.get("regionName") or "",
        result.get("country") or "",
    ]
    location = ", ".join(p for p in parts if p) or "unknown"

    # ip-api.com returns "org" (registered owner of the IP block) and "isp"
    # (the connection provider); they often match. De-duplicate so the
    # suffix is concise.
    extras: list[str] = []
    for field in ("org", "isp"):
        value = result.get(field)
        if value and value not in extras:
            extras.append(value)
    if extras:
        location = f"{location} ({' / '.join(extras)})"
    return location


def lookup_ips_in_file(path: str) -> Iterator[tuple[str, str]]:
    """Yield (ip, location) pairs for each unique IP found in `path`.

    Private / reserved IPs short-circuit with a "private/reserved" tag.
    Public IPs are batched in groups of 100 to ip-api.com.
    """
    with open(path, "r", encoding="utf-8", errors="replace") as handle:
        # dict.fromkeys preserves order while deduplicating.
        unique_ips = list(dict.fromkeys(extract_ips(handle)))

    public: list[str] = []
    for ip in unique_ips:
        addr = ipaddress.ip_address(ip)
        if (
            addr.is_private
            or addr.is_loopback
            or addr.is_reserved
            or addr.is_link_local
            or addr.is_multicast
        ):
            yield ip, "private/reserved"
        else:
            public.append(ip)

    for start in range(0, len(public), _BATCH_SIZE):
        chunk = public[start : start + _BATCH_SIZE]
        try:
            results = _lookup_batch(chunk)
        except (urllib.error.URLError, TimeoutError, json.JSONDecodeError) as exc:
            for ip in chunk:
                yield ip, f"lookup failed: {exc}"
            continue
        for ip, result in zip(chunk, results):
            if result.get("status") == "success":
                yield ip, _format_result(result)
            else:
                yield ip, f"lookup failed: {result.get('message') or 'unknown'}"


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(
        description="Look up locations of IP addresses listed in a text file."
    )
    parser.add_argument(
        "file",
        help="Text file with pipe-delimited lines; IP must be the last column.",
    )
    args = parser.parse_args(argv)
    for ip, location in lookup_ips_in_file(args.file):
        print(f"{ip}\t{location}")
    return 0


if __name__ == "__main__":
    sys.exit(main())