I Turned Nginx Access Logs into an IP and ASN Traffic Report

Raw Nginx logs told me which endpoints were busy, but not whether a traffic spike came from customers, a cloud provider, or one unexpected network. I built a small enrichment job that turns unique public IPs into a country and ASN report without putting geolocation calls in the request path.

The IP Geolocation Lookup does the lookup, and the rest is a local Python pipeline with parsing, deduplication, caching, and deliberately boring CSV output.

Start with offline enrichment

Calling a geolocation API from Nginx middleware adds latency and creates a new failure mode for every request. Access logs already contain what the report needs, so the job can run after rotation and fail independently from the application.

Use a log format you control. This example expects the conventional combined format, where the client IP is the first field:

203.0.113.42 - - [24/Jul/2026:13:04:11 +0000] "GET /api/search HTTP/1.1" 200 821 "-" "Mozilla/5.0"

The parser only extracts valid public addresses. Private, loopback, and documentation ranges should not be sent to an external lookup service.

import ipaddress
import re
from collections import Counter
from pathlib import Path

LOG_LINE = re.compile(
    r'^(?P<ip>\S+) \S+ \S+ \[[^]]+\] "(?P<method>\S+) '
    r'(?P<path>\S+) [^"]+" (?P<status>\d{3}) (?P<size>\S+)'
)


def read_public_ip_counts(log_path: str) -> Counter[str]:
    counts: Counter[str] = Counter()

    for line in Path(log_path).read_text(errors="replace").splitlines():
        match = LOG_LINE.match(line)
        if not match:
            continue

        try:
            address = ipaddress.ip_address(match.group("ip"))
        except ValueError:
            continue

        if not address.is_global:
            continue

        counts[address.compressed] += 1

    return counts

If Nginx sits behind a reverse proxy, verify that the first field contains the trusted client address rather than the proxy address. Do not blindly log the first value in X-Forwarded-For; only accept forwarding headers from proxies you operate.

Look up each unique IP once

The synchronous Actor endpoint fits a batch job because each call returns its dataset item directly. The function below applies a timeout and distinguishes an empty result from an HTTP failure.

import os
import requests

APIFY_TOKEN = os.environ.get("APIFY_TOKEN", "YOUR_APIFY_TOKEN")
ACTOR_ID = "weeknds~ip-geolocation-lookup"
LOOKUP_URL = (
    f"https://api.apify.com/v2/acts/{ACTOR_ID}/"
    "run-sync-get-dataset-items"
)


def geolocate(ip_address: str) -> dict:
    response = requests.post(
        LOOKUP_URL,
        params={"token": APIFY_TOKEN},
        json={"ip": ip_address},
        timeout=60,
    )
    response.raise_for_status()

    items = response.json()
    if not isinstance(items, list) or not items:
        return {"ip": ip_address, "lookup_status": "not_found"}

    item = items[0]
    return {
        "ip": ip_address,
        "country_code": item.get("countryCode"),
        "country": item.get("country"),
        "region": item.get("region"),
        "city": item.get("city"),
        "timezone": item.get("timezone"),
        "isp": item.get("isp"),
        "organization": item.get("org"),
        "asn": item.get("as"),
        "lookup_status": "ok",
    }

The token belongs in an environment variable or secret manager. Keeping YOUR_APIFY_TOKEN as the fallback makes accidental production execution fail clearly rather than hiding a real credential in source.

Add a durable lookup cache

A daily log may contain millions of requests but only thousands of unique addresses. The same addresses often return on following days, so a SQLite cache cuts both runtime and API calls.

IP ownership and routing change, which means cache entries need an expiry. Seven days is reasonable for a traffic report; fraud controls or account security may need fresher data.

import json
import sqlite3
from datetime import datetime, timedelta, timezone

CACHE_TTL = timedelta(days=7)


def open_cache(path: str = "ip-enrichment.sqlite3") -> sqlite3.Connection:
    connection = sqlite3.connect(path)
    connection.execute("""
        CREATE TABLE IF NOT EXISTS ip_cache (
            ip TEXT PRIMARY KEY,
            looked_up_at TEXT NOT NULL,
            payload TEXT NOT NULL
        )
    """)
    return connection


def cached_geolocate(connection: sqlite3.Connection, ip_address: str) -> dict:
    row = connection.execute(
        "SELECT looked_up_at, payload FROM ip_cache WHERE ip = ?",
        (ip_address,),
    ).fetchone()

    if row:
        looked_up_at = datetime.fromisoformat(row[0])
        if datetime.now(timezone.utc) - looked_up_at < CACHE_TTL:
            return json.loads(row[1])

    result = geolocate(ip_address)
    now = datetime.now(timezone.utc).isoformat()
    connection.execute(
        """
        INSERT INTO ip_cache (ip, looked_up_at, payload)
        VALUES (?, ?, ?)
        ON CONFLICT(ip) DO UPDATE SET
            looked_up_at = excluded.looked_up_at,
            payload = excluded.payload
        """,
        (ip_address, now, json.dumps(result, sort_keys=True)),
    )
    connection.commit()
    return result

I cache not_found results too. Otherwise a malformed or unsupported address gets retried in every run. A shorter TTL for failures is a useful refinement if temporary upstream gaps are common.

Build a network-level report

Exact city counts look impressive, but IP geolocation is not GPS. Country and network ownership are usually more appropriate for aggregate operations work. I keep the city field for investigation and summarize normal reports by country and ASN.

import csv
from collections import defaultdict


def build_traffic_report(log_path: str, output_path: str) -> None:
    ip_counts = read_public_ip_counts(log_path)
    totals = defaultdict(lambda: {"requests": 0, "unique_ips": 0})

    with open_cache() as connection:
        for ip_address, request_count in ip_counts.items():
            geo = cached_geolocate(connection, ip_address)
            country = geo.get("country_code") or "UNKNOWN"
            asn = geo.get("asn") or "UNKNOWN"
            key = (country, asn)
            totals[key]["requests"] += request_count
            totals[key]["unique_ips"] += 1

    rows = sorted(
        (
            {
                "country": country,
                "asn": asn,
                "requests": values["requests"],
                "unique_ips": values["unique_ips"],
            }
            for (country, asn), values in totals.items()
        ),
        key=lambda row: row["requests"],
        reverse=True,
    )

    with open(output_path, "w", newline="", encoding="utf-8") as handle:
        writer = csv.DictWriter(
            handle,
            fieldnames=("country", "asn", "requests", "unique_ips"),
        )
        writer.writeheader()
        writer.writerows(rows)

This output works with a spreadsheet, a scheduled email, or an ingestion job for Grafana. It also stays inspectable when somebody asks why a network was flagged.

Flag changes against your own baseline

A fixed list of suspicious countries ages badly and can block legitimate users. For operational review, compare each day's distribution with a rolling baseline and flag large changes rather than assigning guilt to a location.

def traffic_share(rows: list[dict], key: str, value: str) -> float:
    total = sum(int(row["requests"]) for row in rows)
    if total == 0:
        return 0.0
    matched = sum(
        int(row["requests"])
        for row in rows
        if row.get(key) == value
    )
    return matched / total


def is_share_spike(
    current_rows: list[dict],
    baseline_rows: list[dict],
    key: str,
    value: str,
    minimum_requests: int = 500,
) -> bool:
    current_requests = sum(
        int(row["requests"])
        for row in current_rows
        if row.get(key) == value
    )
    current = traffic_share(current_rows, key, value)
    baseline = traffic_share(baseline_rows, key, value)

    return current_requests >= minimum_requests and current >= max(0.05, baseline * 3)

A spike should open an investigation, not automatically add a firewall rule. Check paths, status codes, user agents, request rates, and authentication outcomes before taking action. Large cloud ASNs host both normal services and abusive automation.

Cost and data handling

The IP Geolocation Lookup is priced at $0.001 per run, so enriching 10,000 previously unseen IPs costs $10 in Actor charges. Apify platform usage may also apply. With a seven-day cache, repeat visitors do not create another lookup on every report run.

IP addresses can be personal data depending on jurisdiction and context. Set a retention period, restrict access to the raw cache, and aggregate before sending reports broadly. If the report only needs country and ASN totals, delete or rotate the raw address mapping after the operational window closes.

Run the job on rotated logs so each file is stable, write the CSV to a temporary filename, and rename it only after every lookup succeeds. That keeps dashboards from reading a half-written report and gives retries a clean input.