"""CountryLedger: offline country reports using IP2Location CSV databases."""
import argparse
from bisect import bisect_right
from collections import Counter
import csv
import html
import ipaddress
import json
from pathlib import Path
import re
import sys
class CountryDatabase:
def __init__(self, path, version=4):
self.starts, self.ends, self.countries = [], [], []
limit = 2 ** (32 if version == 4 else 128)
self.version = version
with Path(path).open(encoding='utf-8-sig', newline='') as source:
for number, row in enumerate(csv.reader(source), 1):
if not row:
continue
if number == 1 and row[0].lower() == 'ip_from':
continue
try:
start, end = int(row[0]), int(row[1])
code, name = row[2], row[3]
if not 0 <= start <= end < limit:
raise ValueError()
if self.ends and start <= self.ends[-1]:
raise ValueError()
if not re.fullmatch(r'[A-Z]{2}|-', code):
raise ValueError()
except (ValueError, IndexError):
raise ValueError(f'Invalid, overlapping or unsorted database range at row {number}') from None
self.starts.append(start)
self.ends.append(end)
self.countries.append((code, name))
if not self.starts:
raise ValueError('Database contains no ranges')
def lookup(self, address):
value = int(address)
index = bisect_right(self.starts, value) - 1
if index >= 0 and value <= self.ends[index]:
return self.countries[index]
return None
COMBINED = re.compile(r'^(\S+)\s+\S+\s+\S+\s+\[[^\]]+\]\s+"[^"\r\n]*"\s+(\d{3})\s+')
def records(path, format):
with Path(path).open(encoding='utf-8-sig', errors='replace', newline='') as source:
if format == 'csv':
reader = csv.DictReader(source)
if not {'ip', 'status'} <= set(reader.fieldnames or []):
raise ValueError('Traffic CSV requires ip and status columns')
for row in reader:
yield row.get('ip'), row.get('status')
else:
for line in source:
match = COMBINED.match(line)
yield match.groups() if match else (None, None)
def summarize(path, databases, format='combined', threshold=10):
if threshold < 2:
raise ValueError('Country-label threshold must be at least 2 requests')
countries, statuses = Counter(), Counter()
skipped = unknown = nonpublic = total = 0
for raw_ip, raw_status in records(path, format):
try:
address = ipaddress.ip_address(raw_ip)
status = int(raw_status)
if not 100 <= status <= 599:
raise ValueError()
except (ValueError, TypeError):
skipped += 1
continue
if address.version == 6 and address.ipv4_mapped:
address = address.ipv4_mapped
total += 1
statuses[f'{status // 100}xx'] += 1
if not address.is_global:
nonpublic += 1
continue
database = databases.get(address.version)
country = database.lookup(address) if database else None
if country is None or country[0] == '-':
unknown += 1
else:
countries[country] += 1
visible = [{'code': code, 'country': name, 'requests': count}
for (code, name), count in sorted(countries.items(), key=lambda item: (-item[1], item[0]))
if count >= threshold]
hidden = sum(count for count in countries.values() if count < threshold)
return {'schema_version': 1, 'requests': total, 'skipped_rows': skipped,
'country_label_threshold': threshold, 'countries': visible,
'requests_in_hidden_countries': hidden, 'unmapped_requests': unknown,
'nonpublic_requests': nonpublic, 'status_classes': dict(sorted(statuses.items())),
'privacy_note': 'Raw IPs, URLs, timestamps and user agents are not exported. Counts are requests, not people. Small-country labels are hidden; this is not an anonymity guarantee.',
'geolocation_note': 'IP geolocation estimates network location, not a person\u2019s precise location. VPNs and proxies can affect results.',
'attribution': 'Country lookup uses IP2Location data. https://www.ip2location.com/'}
def render(summary):
escape = html.escape
rows = ''.join(f'
{escape(c["country"])} {escape(c["code"])}
'
f'
'
f'
{c["requests"]:,}
' for c in summary['countries'])
if not rows:
rows = '
No country meets the display threshold.
'
status = ' · '.join(f'{escape(key)}: {count:,}' for key, count in summary['status_classes'].items())
return f'''
CountryLedger · Traffic without the trail
CountryLedger / Offline traffic report
Where requests come from. Without keeping the trail.
Country-level patterns from local logs, powered by IP2Location.