Этот Python-скрипт извлекает все IP-адреса из access-лога nginx или Apache, локально считает запросы по уникальным посетителям, а затем обращается к бесплатному API ip.xakki.pro только для уникальных, глобально маршрутизируемых IP, чтобы вывести топ стран и ASN по суммарному числу запросов — приватные/loopback/CGNAT-адреса учитываются, но никогда не отправляются в API, так как там нечего искать.
Зависимости: нет — только стандартная библиотека Python 3.8+ (urllib, json, csv, ipaddress, argparse). pip install не требуется.
#!/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()
Скрипт распространяется по лицензии MIT (заголовок SPDX-License-Identifier: MIT в начале файла) — копируйте, изменяйте и используйте свободно в своём проекте, указывать авторство не обязательно.
Извлечение IP основано на одном намеренно простом допущении: IP — это первый токен строки до пробела, что соответствует и Common Log Format, и стандартному «combined»-формату nginx/Apache. Кастомный log_format, где первым идёт что-то другое (ID запроса, метка времени), потребует своего регулярного выражения — это рецепт для адаптации, а не универсальный парсер логов. Строки без распознаваемого IP в начале засчитываются как «unparsed» и пропускаются, не прерывая выполнение.
Агрегация выполняется в два прохода не случайно: подсчёт обращений по уникальным IP — это чисто локальная работа над файлом лога (collections.Counter из Python, вообще без сети), и только ВТОРОЙ проход — по одному запросу к API на каждый уникальный маршрутизируемый IP — обращается к сети. Лог с 50 000 запросами от 200 уникальных посетителей стоит 200 обращений, а не 50 000, что в точности повторяет модель лимита API по уникальному IP в сутки, а не идёт ей наперекор.
Итоги по странам/ASN взвешены по числу ЗАПРОСОВ, а не по числу уникальных IP — иначе один IP-скрапер, обратившийся 5000 раз, и страна с 5000 разными разовыми посетителями выглядели бы одинаково при наивном подсчёте по уникальным IP. Уникальный IP, для которого группа «network» в этом прогоне пришла с ограничением по частоте, исключается из итогов (а не засчитывается как страна с нулём запросов), и скрипт печатает, сколько IP было исключено, чтобы итоги никогда не выглядели тихо неполными.
GET /api/v1/ip никогда не отвечает жёстким 429 от самого приложения — группы обогащения (geo, network, security) лимитируются по уникальному целевому IP в сутки (UTC) плюс более жёсткий поминутный бюджет; при превышении группа мягко придерживается: HTTP 200, "status":"partial", имя группы в "limited", и заголовки RateLimit-Limit/RateLimit-Remaining/RateLimit-Reset в ответе. Этот скрипт читает эти сигналы, а не угадывает, и явно фиксирует придержанный результат (никогда как пустое/null значение — это выглядело бы так же, как «у API нет данных по этому IP»). Настоящий 429/503 всё же возможен на границе — это собственный анти-флуд nginx, независимый от лимитов выше — если слать запросы слишком быстро с одной машины; скрипт делает экспоненциальные повторы на 429/5xx и учитывает заголовок Retry-After, если сервер его прислал. У анонимных вызовов небольшой суточный/поминутный бюджет на группу (см. заголовок RateLimit-Limit в любом ответе или /api/v1/tiers); для заметно бо́льшего бюджета зарегистрируйте бесплатный API-токен.
Сам этот скрипт и есть пример работы с API — отдельный блок curl/ответ не нужен, код выше и есть запрос. Полная схема ответа и все параметры — в документации API:
Понравился инструмент? Бесплатный аккаунт нашего API даёт 1000 запросов в день.Смотреть документацию API →