This Python script extracts every IP address from an nginx or Apache access log, counts requests per unique visitor locally, then calls the free ip.xakki.pro API only for the unique, globally-routable IPs to print the top countries and ASNs by total request volume — private/loopback/CGNAT addresses are counted but never sent to the API, since there is nothing to look up there.
placeholder
#!/usr/bin/env python3
# SPDX-License-Identifier: MIT
"""log-aggregate.py -- extract IPs from an nginx/Apache access log and
aggregate hits by country and ASN via the ip.xakki.pro API.
Reads a combined/common-format access log (the IP is the first
whitespace-separated token on each line, e.g. `1.2.3.4 - - [date] "GET / ..."`),
counts requests per unique IP locally with `re`, then enriches only the
UNIQUE, globally-routable IPs (private/loopback/link-local/CGNAT addresses
are counted but never sent to the API -- there is nothing for an API to look
up there) via GET /api/v1/ip?ip=<addr>&include=network, and prints the top-N
countries and ASNs by total request volume (not by unique-IP count).
Zero dependencies -- stdlib only (re, ipaddress, collections, urllib, json,
argparse). Python 3.8+.
Usage:
python3 log-aggregate.py access.log
python3 log-aggregate.py access.log --top 20 --delay 1.0
Why aggregate by unique IP before calling the API: the API's enrichment
groups are metered PER UNIQUE IP PER DAY, not per request -- a log with
50,000 requests from 200 unique visitors costs 200 lookups, not 50,000,
and a repeat lookup of the same IP later the same day is free. This script
mirrors that model rather than fighting it.
"""
import argparse
import ipaddress
import json
import re
import sys
import time
import urllib.error
import urllib.request
from collections import Counter
DEFAULT_BASE_URL = "https://ip.xakki.pro"
USER_AGENT = "myip-log-aggregate-recipe/1.0 (+https://ip.xakki.pro/tools/scripts/log-aggregate.html)"
# First token on the line, up to the first whitespace -- matches the IP
# field of both Common Log Format and Combined Log Format (nginx/Apache
# default `log_format combined`). Deliberately simple: this is a recipe, not
# a general-purpose log parser -- a custom log_format needs its own regex.
IP_TOKEN_RE = re.compile(r"^(\S+)")
def extract_ip(line):
"""Returns the validated IP string at the start of line, or None."""
m = IP_TOKEN_RE.match(line)
if not m:
return None
token = m.group(1)
try:
ipaddress.ip_address(token)
return token
except ValueError:
return None
def fetch_network(base_url, ip_value, max_retries=5, base_backoff=0.5):
"""GET /api/v1/ip?ip=...&include=network with exponential backoff on
429/5xx and transport errors, honoring Retry-After when present.
Returns the parsed JSON body, or None on unrecoverable failure."""
url = "{}/api/v1/ip?ip={}&include=network".format(base_url, ip_value)
req = urllib.request.Request(url, headers={"User-Agent": USER_AGENT, "Accept": "application/json"})
attempt = 0
while True:
try:
with urllib.request.urlopen(req, timeout=15) as resp:
return json.loads(resp.read().decode("utf-8"))
except urllib.error.HTTPError as e:
if not (e.code == 429 or e.code >= 500) or attempt >= max_retries:
print(" ! {} -> HTTP {} (giving up)".format(ip_value, e.code), file=sys.stderr)
return None
wait = float(e.headers.get("Retry-After", 0)) or (base_backoff * (2 ** attempt))
except (urllib.error.URLError, TimeoutError, ConnectionError) as e:
if attempt >= max_retries:
print(" ! {} -> {} (giving up)".format(ip_value, e), file=sys.stderr)
return None
wait = base_backoff * (2 ** attempt)
attempt += 1
time.sleep(wait)
def main():
parser = argparse.ArgumentParser(description="Aggregate an access log's requests by country/ASN.")
parser.add_argument("log_file", help="Path to the access log.")
parser.add_argument("--top", type=int, default=10, help="How many countries/ASNs to print (default: 10).")
parser.add_argument("--base-url", default=DEFAULT_BASE_URL, help="API base URL (default: production).")
parser.add_argument("--delay", type=float, default=0.5, help="Seconds between lookups (default: 0.5).")
args = parser.parse_args()
hits_per_ip = Counter()
unparsed_lines = 0
with open(args.log_file, encoding="utf-8", errors="replace") as f:
for line in f:
ip_value = extract_ip(line)
if ip_value is None:
unparsed_lines += 1
continue
hits_per_ip[ip_value] += 1
routable_ips = []
private_hits = 0
for ip_value, count in hits_per_ip.items():
addr = ipaddress.ip_address(ip_value)
if addr.is_private or addr.is_loopback or addr.is_link_local or addr.is_multicast or addr.is_unspecified:
private_hits += count
else:
routable_ips.append(ip_value)
print("parsed {} unique IPs ({} requests total, {} lines unparsed, {} requests from "
"private/loopback/link-local addresses skipped)".format(
len(hits_per_ip), sum(hits_per_ip.values()), unparsed_lines, private_hits), file=sys.stderr)
hits_per_country = Counter()
hits_per_asn = Counter()
withheld_ips = 0
for i, ip_value in enumerate(routable_ips):
data = fetch_network(args.base_url, ip_value)
if data is None:
continue
if "network" in (data.get("limited") or []):
withheld_ips += 1
continue
network = data.get("network") or {}
geo = data.get("geo") or {}
country = geo.get("countryCode") or "??"
asn_label = "AS{} {}".format(network.get("asn"), network.get("asName") or "") if network.get("asn") else "unknown"
n = hits_per_ip[ip_value]
hits_per_country[country] += n
hits_per_asn[asn_label] += n
if i < len(routable_ips) - 1:
time.sleep(args.delay)
if withheld_ips:
print("note: {} unique IPs were rate-limited (network group withheld this run) and are "
"excluded from the totals below, not counted as zero".format(withheld_ips), file=sys.stderr)
print("\nTop {} countries by request volume:".format(args.top))
for country, n in hits_per_country.most_common(args.top):
print(" {:6d} {}".format(n, country))
print("\nTop {} ASNs by request volume:".format(args.top))
for asn_label, n in hits_per_asn.most_common(args.top):
print(" {:6d} {}".format(n, asn_label))
if __name__ == "__main__":
main()
This script is MIT-licensed (SPDX-License-Identifier: MIT header at the top of the file) — copy, modify and reuse it freely in your own project, no attribution required.
IP extraction uses one deliberately simple assumption: the IP is the first whitespace-separated token on each line, which matches both Common Log Format and nginx/Apache's default "combined" format. A custom log_format that puts something else first (a request ID, a timestamp) needs its own regex — this is a recipe to adapt, not a universal log parser. Lines that don't start with a parseable IP are counted as "unparsed" and skipped, never crashing the run.
Aggregation happens in two passes for a reason: counting hits per unique IP is pure local work over the log file (Python's collections.Counter, no network at all), and only the SECOND pass — one API call per unique routable IP — touches the network. A log with 50,000 requests from 200 unique visitors costs 200 lookups, not 50,000, and mirrors exactly how the API's own per-unique-IP/day metering already works rather than fighting it.
The country/ASN totals are weighted by REQUEST count, not by unique-IP count — a single scraper IP that hit the site 5,000 times and a country with 5,000 distinct one-time visitors would otherwise look identical in a naive unique-IP tally. A unique IP whose "network" group comes back rate-limited this run is excluded from the totals (not counted as a zero-hit country), and the script prints how many were excluded so the totals are never silently incomplete.
GET /api/v1/ip never returns a hard 429 from the app itself — its enrichment groups (geo, network, security) are metered per unique target IP per UTC day, plus a tighter per-minute budget, and going over either soft-withholds the group instead: HTTP 200, "status":"partial", the group's name in "limited", and RateLimit-Limit/RateLimit-Remaining/RateLimit-Reset headers on the response. This script reads those signals rather than guessing, and records a withheld result explicitly (never as a blank/null, which would look identical to "the API has no data for this IP"). A real 429/503 can still happen at the edge — nginx's own flood guard, independent of the metering above — if you call the endpoint too fast from one machine; the script backs off exponentially on any 429/5xx and honors a Retry-After header when the server sends one. Anonymous callers get a modest daily/per-minute budget per group (see the RateLimit-Limit header on any response, or /api/v1/tiers); register a free API token for a much larger one if you plan to run this against more than a handful of IPs.
This script itself is the API example — there is no separate curl/response block to show, because the code above already IS the request. See the full response schema and every parameter in the API docs:
Like this tool? Our API gives free accounts 1,000 requests/day.See the API docs →