728 lines
26 KiB
Python
728 lines
26 KiB
Python
"""
|
|
IP Geo Report Generator
|
|
- Batch processing for 480M rows
|
|
- LRU memory cache with disk spillover
|
|
- Resume strategy via checkpoint
|
|
- Daily visit count per country report
|
|
- Uses ipinfo bundle_location_lite.mmdb + API fallback
|
|
- Bot/attack detection and filtering
|
|
"""
|
|
|
|
import os
|
|
import re
|
|
import csv
|
|
import json
|
|
import time
|
|
import pickle
|
|
import shelve
|
|
import logging
|
|
import ipaddress
|
|
from datetime import datetime, date
|
|
from collections import defaultdict, OrderedDict
|
|
from typing import Optional, Dict, Tuple, Set
|
|
from dataclasses import dataclass, field
|
|
|
|
# ─── Config ───────────────────────────────────────────────────────────────────
|
|
|
|
DB_CONFIG = {
|
|
"host": "localhost",
|
|
"user": "user",
|
|
"password": "password",
|
|
"database": "database",
|
|
"charset": "utf8mb4",
|
|
}
|
|
|
|
TABLE_NAME = "wp_statpress" # your table name
|
|
IP_COLUMN = "ip" # IP address column
|
|
DATE_COLUMN = "date" # date column (DATE or DATETIME)
|
|
UA_COLUMN = "agent" # user agent column (set None if not available)
|
|
PATH_COLUMN = "urlrequested"
|
|
BATCH_SIZE = 50_000 # rows per batch
|
|
LRU_MAX_SIZE = 100_000 # max IPs in memory cache
|
|
DISK_CACHE_PATH = "./cache/ip_geo" # shelve disk cache path
|
|
CHECKPOINT_FILE = "./cache/checkpoint.json"
|
|
REPORT_OUTPUT = "./report" # output directory
|
|
GEOIP_DB_PATH = "./ipinfo_lite.mmdb" # ipinfo free DB path
|
|
IPINFO_TOKEN = "6f2ffc9ac24285" # ipinfo.io API token (free tier)
|
|
|
|
# Disk cache settings
|
|
DISK_CACHE_TTL = 86400 * 30 # 30 days in seconds
|
|
DISK_FLUSH_EVERY = 10_000 # flush disk cache every N lookups
|
|
|
|
# Bot detection thresholds
|
|
BOT_IP_REQ_THRESHOLD = 5000 # flag IP if it appears more than this in a batch
|
|
BOT_RATE_WINDOW_SEC = 60 # time window for rate detection
|
|
|
|
# ─── Logging ──────────────────────────────────────────────────────────────────
|
|
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format="%(asctime)s [%(levelname)s] %(message)s",
|
|
handlers=[
|
|
logging.FileHandler("ip_geo_report.log"),
|
|
logging.StreamHandler()
|
|
]
|
|
)
|
|
log = logging.getLogger(__name__)
|
|
|
|
# ─── Bot Detector ─────────────────────────────────────────────────────────────
|
|
|
|
class BotDetector:
|
|
"""
|
|
Detects and filters bots/attacks based on:
|
|
1. IP address (private, reserved, known bot ranges)
|
|
2. User agent string (known bot patterns)
|
|
3. Request path (scanner/attack patterns)
|
|
4. IP request frequency (too many hits = bot)
|
|
"""
|
|
|
|
# Known bot / crawler user agent patterns
|
|
BOT_UA_PATTERNS = re.compile(
|
|
r"(?i)("
|
|
r"bot|crawler|spider|scraper|scan|curl|wget|python|java|"
|
|
r"go-http|ruby|perl|php|axios|libwww|httpclient|okhttp|"
|
|
r"zgrab|masscan|nmap|nikto|sqlmap|dirbuster|nuclei|"
|
|
r"semrush|ahrefs|mj12bot|dotbot|rogerbot|bingbot|"
|
|
r"googlebot|yandexbot|baiduspider|duckduckbot|"
|
|
r"facebookexternalhit|twitterbot|linkedinbot|"
|
|
r"archive\.org|ia_archiver|wayback|"
|
|
r"headlesschrome|phantomjs|selenium|puppeteer|playwright"
|
|
r")"
|
|
)
|
|
|
|
# Suspicious request path patterns (scanners / attacks)
|
|
ATTACK_PATH_PATTERNS = re.compile(
|
|
r"(?i)("
|
|
r"wp-login|xmlrpc|\.env|\.git|\.svn|\.htaccess|"
|
|
r"phpmyadmin|adminer|manager/html|solr/admin|"
|
|
r"actuator|/etc/passwd|/proc/self|"
|
|
r"select\s+.+from|union\s+select|waitfor\s+delay|"
|
|
r"pg_sleep|exec\(|eval\(|base64_decode|"
|
|
r"\.php\?.*=http|\.php\?.*=//|"
|
|
r"\.\./|%2e%2e|%252e|" # path traversal
|
|
r"<script|javascript:|onerror=|onload=|" # XSS
|
|
r"\/shell|\/backdoor|\/cmd|c99|r57|webshell" # webshells
|
|
r")"
|
|
)
|
|
|
|
# Private / reserved IP ranges (RFC 1918, loopback, etc.)
|
|
PRIVATE_RANGES = [
|
|
ipaddress.ip_network("10.0.0.0/8"),
|
|
ipaddress.ip_network("172.16.0.0/12"),
|
|
ipaddress.ip_network("192.168.0.0/16"),
|
|
ipaddress.ip_network("127.0.0.0/8"),
|
|
ipaddress.ip_network("169.254.0.0/16"), # link-local
|
|
ipaddress.ip_network("::1/128"), # IPv6 loopback
|
|
ipaddress.ip_network("fc00::/7"), # IPv6 private
|
|
]
|
|
|
|
def __init__(self):
|
|
self.ip_freq: Dict[str, int] = defaultdict(int) # IP frequency in current batch
|
|
self.skipped_ips: Set[str] = set() # known bot IPs this run
|
|
self.stats = {
|
|
"total": 0,
|
|
"skipped_ua": 0,
|
|
"skipped_path": 0,
|
|
"skipped_ip": 0,
|
|
"skipped_freq": 0,
|
|
"skipped_private": 0,
|
|
}
|
|
|
|
def is_private_ip(self, ip: str) -> bool:
|
|
"""Check if IP is private/reserved."""
|
|
try:
|
|
addr = ipaddress.ip_address(ip)
|
|
return any(addr in net for net in self.PRIVATE_RANGES)
|
|
except ValueError:
|
|
return True # invalid IP = skip
|
|
|
|
def is_bot_ua(self, ua: str) -> bool:
|
|
"""Check if user agent matches known bot patterns."""
|
|
if not ua or ua.strip() == "-":
|
|
return True # missing UA = likely bot
|
|
return bool(self.BOT_UA_PATTERNS.search(ua))
|
|
|
|
def is_attack_path(self, path: str) -> bool:
|
|
"""Check if request path matches attack/scanner patterns."""
|
|
if not path:
|
|
return False
|
|
return bool(self.ATTACK_PATH_PATTERNS.search(path))
|
|
|
|
def track_ip_frequency(self, ip: str) -> bool:
|
|
"""Track IP hit count. Returns True if IP exceeds threshold."""
|
|
self.ip_freq[ip] += 1
|
|
if self.ip_freq[ip] > BOT_IP_REQ_THRESHOLD:
|
|
self.skipped_ips.add(ip)
|
|
return True
|
|
return False
|
|
|
|
def reset_frequency(self):
|
|
"""Reset per-batch frequency counter."""
|
|
self.ip_freq.clear()
|
|
|
|
def is_bot(self, ip: str, ua: str = None, path: str = None) -> Tuple[bool, str]:
|
|
"""
|
|
Main bot detection method.
|
|
Returns (is_bot: bool, reason: str)
|
|
"""
|
|
self.stats["total"] += 1
|
|
|
|
# Already flagged as bot IP
|
|
if ip in self.skipped_ips:
|
|
self.stats["skipped_ip"] += 1
|
|
return True, "known_bot_ip"
|
|
|
|
# Private / invalid IP
|
|
if self.is_private_ip(ip):
|
|
self.stats["skipped_private"] += 1
|
|
return True, "private_ip"
|
|
|
|
# IP frequency check
|
|
if self.track_ip_frequency(ip):
|
|
self.stats["skipped_freq"] += 1
|
|
log.info(f"Bot detected by frequency: {ip} ({self.ip_freq[ip]} hits)")
|
|
return True, "high_frequency"
|
|
|
|
# User agent check
|
|
if ua is not None and self.is_bot_ua(ua):
|
|
self.stats["skipped_ua"] += 1
|
|
return True, "bot_ua"
|
|
|
|
# Attack path check
|
|
if path is not None and self.is_attack_path(path):
|
|
self.stats["skipped_path"] += 1
|
|
return True, "attack_path"
|
|
|
|
return False, ""
|
|
|
|
def report(self) -> dict:
|
|
total = self.stats["total"]
|
|
skipped = sum(v for k, v in self.stats.items() if k != "total")
|
|
return {
|
|
**self.stats,
|
|
"skipped_total": skipped,
|
|
"skip_rate": f"{skipped/total*100:.1f}%" if total else "0%",
|
|
"bot_ips_found": len(self.skipped_ips),
|
|
}
|
|
|
|
|
|
# ─── LRU Memory Cache ─────────────────────────────────────────────────────────
|
|
|
|
class LRUCache:
|
|
"""LRU cache. Returns evicted key so caller can spill to disk."""
|
|
|
|
def __init__(self, max_size: int = 100_000):
|
|
self.max_size = max_size
|
|
self.cache = OrderedDict()
|
|
self.hits = 0
|
|
self.misses = 0
|
|
|
|
def get(self, key: str) -> Optional[str]:
|
|
if key in self.cache:
|
|
self.cache.move_to_end(key)
|
|
self.hits += 1
|
|
return self.cache[key]
|
|
self.misses += 1
|
|
return None
|
|
|
|
def put(self, key: str, value: str) -> Optional[str]:
|
|
if key in self.cache:
|
|
self.cache.move_to_end(key)
|
|
self.cache[key] = value
|
|
if len(self.cache) > self.max_size:
|
|
evicted_key, _ = self.cache.popitem(last=False)
|
|
return evicted_key
|
|
return None
|
|
|
|
def stats(self) -> dict:
|
|
total = self.hits + self.misses
|
|
return {
|
|
"size": len(self.cache),
|
|
"hits": self.hits,
|
|
"misses": self.misses,
|
|
"hit_rate": f"{self.hits/total*100:.1f}%" if total else "0%"
|
|
}
|
|
|
|
|
|
# ─── Disk Cache ───────────────────────────────────────────────────────────────
|
|
|
|
class DiskCache:
|
|
"""Persistent shelve cache with TTL expiry."""
|
|
|
|
def __init__(self, path: str, ttl: int = DISK_CACHE_TTL):
|
|
os.makedirs(os.path.dirname(path) or ".", exist_ok=True)
|
|
self.ttl = ttl
|
|
self.db = shelve.open(path, writeback=False)
|
|
self.hits = 0
|
|
self.misses = 0
|
|
log.info(f"Disk cache opened: {path} ({len(self.db)} entries)")
|
|
|
|
def get(self, key: str) -> Optional[str]:
|
|
entry = self.db.get(key)
|
|
if entry:
|
|
value, ts = entry
|
|
if time.time() - ts < self.ttl:
|
|
self.hits += 1
|
|
return value
|
|
del self.db[key]
|
|
self.misses += 1
|
|
return None
|
|
|
|
def put(self, key: str, value: str):
|
|
self.db[key] = (value, time.time())
|
|
|
|
def flush(self):
|
|
self.db.sync()
|
|
|
|
def close(self):
|
|
self.db.sync()
|
|
self.db.close()
|
|
|
|
def stats(self) -> dict:
|
|
total = self.hits + self.misses
|
|
return {
|
|
"size": len(self.db),
|
|
"hits": self.hits,
|
|
"misses": self.misses,
|
|
"hit_rate": f"{self.hits/total*100:.1f}%" if total else "0%"
|
|
}
|
|
|
|
|
|
# ─── Two-Level Cache ──────────────────────────────────────────────────────────
|
|
|
|
class TwoLevelCache:
|
|
"""L1: LRU memory | L2: Disk shelve (handles LRU evictions)"""
|
|
|
|
def __init__(self, lru_size: int, disk_path: str):
|
|
self.lru = LRUCache(max_size=lru_size)
|
|
self.disk = DiskCache(disk_path)
|
|
self.lookup_cnt = 0
|
|
|
|
def get(self, ip: str) -> Optional[str]:
|
|
val = self.lru.get(ip)
|
|
if val is not None:
|
|
return val
|
|
val = self.disk.get(ip)
|
|
if val is not None:
|
|
self.lru.put(ip, val)
|
|
return val
|
|
return None
|
|
|
|
def put(self, ip: str, country: str):
|
|
evicted_key = self.lru.put(ip, country)
|
|
if evicted_key:
|
|
self.disk.put(evicted_key, country)
|
|
self.lookup_cnt += 1
|
|
if self.lookup_cnt % DISK_FLUSH_EVERY == 0:
|
|
self.disk.flush()
|
|
log.info(
|
|
f"Cache → Memory: {self.lru.stats()} | "
|
|
f"Disk: {self.disk.stats()}"
|
|
)
|
|
|
|
def close(self):
|
|
self.disk.flush()
|
|
self.disk.close()
|
|
|
|
def stats(self) -> dict:
|
|
return {"memory": self.lru.stats(), "disk": self.disk.stats()}
|
|
|
|
|
|
# ─── GeoIP Lookup ─────────────────────────────────────────────────────────────
|
|
|
|
class GeoIPLookup:
|
|
"""
|
|
1. ipinfo bundle_location_lite.mmdb (offline, no limit)
|
|
Fields: country, country_code, continent, asn, as_name, as_domain
|
|
2. ipinfo API fallback (50k/month free tier)
|
|
"""
|
|
|
|
def __init__(self, db_path: str = None, api_token: str = None):
|
|
self.reader = None
|
|
self.client = None
|
|
self.api_calls = 0
|
|
sources = []
|
|
|
|
if db_path and os.path.exists(db_path):
|
|
try:
|
|
import maxminddb
|
|
self.reader = maxminddb.open_database(db_path)
|
|
db_type = self.reader.metadata().database_type
|
|
sources.append("mmdb")
|
|
log.info(f"mmdb loaded: {db_path} (type: {db_type})")
|
|
test = self.reader.get("8.8.8.8")
|
|
if test:
|
|
log.info(f"mmdb fields: {list(test.keys())}")
|
|
except ImportError:
|
|
log.error("pip install maxminddb")
|
|
except Exception as e:
|
|
log.error(f"mmdb open failed: {e}")
|
|
else:
|
|
log.warning(f"mmdb not found: {db_path}")
|
|
|
|
if api_token:
|
|
try:
|
|
import ipinfo
|
|
self.client = ipinfo.getHandler(api_token)
|
|
sources.append("api")
|
|
log.info("ipinfo API loaded as fallback")
|
|
log.warning("API active — watch 50k/month free limit!")
|
|
except ImportError:
|
|
log.error("pip install ipinfo")
|
|
|
|
if not sources:
|
|
log.warning("No GeoIP source — all IPs = Unknown")
|
|
else:
|
|
log.info(f"Lookup order: {' → '.join(sources)}")
|
|
|
|
def _extract_country(self, result: dict) -> Optional[str]:
|
|
"""
|
|
ipinfo bundle_location_lite structure:
|
|
{'country': 'United States', 'country_code': 'US', ...}
|
|
"""
|
|
if not result:
|
|
return None
|
|
return result.get("country") or None
|
|
|
|
def _lookup_mmdb(self, ip: str) -> Optional[str]:
|
|
try:
|
|
return self._extract_country(self.reader.get(ip))
|
|
except Exception:
|
|
return None
|
|
|
|
def _lookup_api(self, ip: str) -> Optional[str]:
|
|
try:
|
|
details = self.client.getDetails(ip)
|
|
self.api_calls += 1
|
|
if self.api_calls % 100 == 0:
|
|
log.info(f"API calls: {self.api_calls:,}")
|
|
return (
|
|
getattr(details, "country_name", None) or
|
|
getattr(details, "country", None)
|
|
)
|
|
except Exception as e:
|
|
log.warning(f"API failed for {ip}: {e}")
|
|
return None
|
|
|
|
def lookup(self, ip: str) -> str:
|
|
if self.reader:
|
|
country = self._lookup_mmdb(ip)
|
|
if country:
|
|
return country
|
|
if self.client:
|
|
country = self._lookup_api(ip)
|
|
if country:
|
|
return country
|
|
return "Unknown"
|
|
|
|
def close(self):
|
|
if self.reader:
|
|
self.reader.close()
|
|
if self.api_calls > 0:
|
|
log.info(f"Total API fallback calls: {self.api_calls:,}")
|
|
|
|
|
|
# ─── Checkpoint ───────────────────────────────────────────────────────────────
|
|
|
|
@dataclass
|
|
class Checkpoint:
|
|
last_id: int = 0
|
|
rows_processed: int = 0
|
|
batches_done: int = 0
|
|
started_at: str = field(default_factory=lambda: datetime.now().isoformat())
|
|
updated_at: str = field(default_factory=lambda: datetime.now().isoformat())
|
|
|
|
def save(self, path: str):
|
|
self.updated_at = datetime.now().isoformat()
|
|
with open(path, "w") as f:
|
|
json.dump(self.__dict__, f, indent=2)
|
|
|
|
@classmethod
|
|
def load(cls, path: str) -> "Checkpoint":
|
|
if os.path.exists(path):
|
|
with open(path) as f:
|
|
data = json.load(f)
|
|
log.info(f"Resuming from checkpoint: {data}")
|
|
return cls(**data)
|
|
log.info("No checkpoint — starting fresh")
|
|
return cls()
|
|
|
|
|
|
# ─── Report Aggregator ────────────────────────────────────────────────────────
|
|
|
|
class ReportAggregator:
|
|
"""Aggregates visit counts by country and date. Survives crashes via partial saves."""
|
|
|
|
def __init__(self, output_dir: str):
|
|
os.makedirs(output_dir, exist_ok=True)
|
|
self.output_dir = output_dir
|
|
self.partial_file = os.path.join(output_dir, "partial_data.pkl")
|
|
self.data: Dict[str, Dict[str, int]] = defaultdict(lambda: defaultdict(int))
|
|
self._load_partial()
|
|
|
|
def _load_partial(self):
|
|
if os.path.exists(self.partial_file):
|
|
with open(self.partial_file, "rb") as f:
|
|
saved = pickle.load(f)
|
|
for country, dates in saved.items():
|
|
for d, cnt in dates.items():
|
|
self.data[country][d] += cnt
|
|
log.info(f"Loaded partial data: {len(self.data)} countries")
|
|
|
|
def add(self, country: str, visit_date):
|
|
d = (
|
|
visit_date.strftime("%Y-%m-%d")
|
|
if hasattr(visit_date, "strftime")
|
|
else str(visit_date)[:10]
|
|
)
|
|
self.data[country][d] += 1
|
|
|
|
def save_partial(self):
|
|
with open(self.partial_file, "wb") as f:
|
|
pickle.dump(dict(self.data), f)
|
|
|
|
def export_csv(self) -> str:
|
|
path = os.path.join(self.output_dir, f"report_{date.today()}.csv")
|
|
with open(path, "w", newline="", encoding="utf-8") as f:
|
|
writer = csv.writer(f)
|
|
writer.writerow(["Country", "Date", "Visit Count"])
|
|
for country in sorted(self.data):
|
|
for d in sorted(self.data[country]):
|
|
writer.writerow([country, d, self.data[country][d]])
|
|
log.info(f"CSV saved: {path}")
|
|
return path
|
|
|
|
def export_summary(self) -> str:
|
|
path = os.path.join(self.output_dir, f"summary_{date.today()}.csv")
|
|
totals = {c: sum(d.values()) for c, d in self.data.items()}
|
|
with open(path, "w", newline="", encoding="utf-8") as f:
|
|
writer = csv.writer(f)
|
|
writer.writerow(["Country", "Total Visits"])
|
|
for country, total in sorted(totals.items(), key=lambda x: -x[1]):
|
|
writer.writerow([country, total])
|
|
log.info(f"Summary saved: {path}")
|
|
return path
|
|
|
|
def print_top(self, n: int = 20):
|
|
totals = {c: sum(d.values()) for c, d in self.data.items()}
|
|
top = sorted(totals.items(), key=lambda x: -x[1])[:n]
|
|
print(f"\n{'─'*45}")
|
|
print(f"{'Country':<30} {'Total Visits':>12}")
|
|
print(f"{'─'*45}")
|
|
for country, count in top:
|
|
print(f"{country:<30} {count:>12,}")
|
|
print(f"{'─'*45}\n")
|
|
|
|
|
|
# ─── Main Processor ───────────────────────────────────────────────────────────
|
|
|
|
class IPGeoProcessor:
|
|
|
|
def __init__(self):
|
|
os.makedirs("./cache", exist_ok=True)
|
|
os.makedirs(REPORT_OUTPUT, exist_ok=True)
|
|
|
|
self.cache = TwoLevelCache(LRU_MAX_SIZE, DISK_CACHE_PATH)
|
|
self.geoip = GeoIPLookup(
|
|
db_path = GEOIP_DB_PATH,
|
|
api_token = IPINFO_TOKEN or None
|
|
)
|
|
self.bot = BotDetector()
|
|
self.checkpoint = Checkpoint.load(CHECKPOINT_FILE)
|
|
self.report = ReportAggregator(REPORT_OUTPUT)
|
|
self.conn = None
|
|
self.cursor = None
|
|
|
|
def connect(self):
|
|
import mysql.connector
|
|
self.conn = mysql.connector.connect(**DB_CONFIG)
|
|
self.cursor = self.conn.cursor(buffered=False)
|
|
log.info("MySQL connected")
|
|
|
|
def disconnect(self):
|
|
if self.cursor:
|
|
self.cursor.close()
|
|
if self.conn:
|
|
self.conn.close()
|
|
log.info("MySQL disconnected")
|
|
|
|
def get_total_rows(self) -> int:
|
|
self.cursor.execute(f"SELECT COUNT(*) FROM {TABLE_NAME}")
|
|
return self.cursor.fetchone()[0]
|
|
|
|
def build_select(self) -> str:
|
|
"""Build SELECT query based on available columns."""
|
|
cols = ["id", IP_COLUMN, DATE_COLUMN]
|
|
if UA_COLUMN:
|
|
cols.append(UA_COLUMN)
|
|
if PATH_COLUMN:
|
|
cols.append(PATH_COLUMN)
|
|
return ", ".join(cols)
|
|
|
|
def resolve_ip(self, ip: str) -> str:
|
|
if not ip:
|
|
return "Unknown"
|
|
cached = self.cache.get(ip)
|
|
if cached is not None:
|
|
return cached
|
|
country = self.geoip.lookup(ip)
|
|
self.cache.put(ip, country)
|
|
return country
|
|
|
|
def process_batch(self, last_id: int) -> Tuple[int, int, int]:
|
|
"""
|
|
Returns (last_id_processed, rows_in_batch, skipped_count)
|
|
"""
|
|
cols = self.build_select()
|
|
query = f"""
|
|
SELECT {cols}
|
|
FROM {TABLE_NAME}
|
|
WHERE id > {last_id}
|
|
ORDER BY id ASC
|
|
LIMIT {BATCH_SIZE}
|
|
"""
|
|
self.cursor.execute(query)
|
|
rows = self.cursor.fetchall()
|
|
skipped = 0
|
|
|
|
if not rows:
|
|
return last_id, 0, 0
|
|
|
|
# Reset per-batch frequency tracking
|
|
self.bot.reset_frequency()
|
|
|
|
for row in rows:
|
|
# Unpack columns dynamically
|
|
idx = 0
|
|
row_id = row[idx]; idx += 1
|
|
ip = str(row[idx]).strip() if row[idx] else ""; idx += 1
|
|
visit_date = row[idx]; idx += 1
|
|
ua = str(row[idx]).strip() if UA_COLUMN and idx < len(row) else None
|
|
if ua: idx += 1
|
|
path = str(row[idx]).strip() if PATH_COLUMN and idx < len(row) else None
|
|
|
|
last_id = row_id
|
|
|
|
# Bot / attack detection
|
|
is_bot, reason = self.bot.is_bot(ip, ua, path)
|
|
if is_bot:
|
|
skipped += 1
|
|
continue
|
|
|
|
# Resolve IP → country
|
|
country = self.resolve_ip(ip)
|
|
self.report.add(country, visit_date)
|
|
|
|
return last_id, len(rows), skipped
|
|
|
|
def run(self):
|
|
log.info("=" * 60)
|
|
log.info("IP Geo Report Generator starting")
|
|
log.info("=" * 60)
|
|
log.info(f"Bot detection: UA={'on' if UA_COLUMN else 'off'} | "
|
|
f"Path={'on' if PATH_COLUMN else 'off'} | "
|
|
f"Frequency threshold={BOT_IP_REQ_THRESHOLD}")
|
|
|
|
self.connect()
|
|
|
|
total = self.get_total_rows()
|
|
start_id = self.checkpoint.last_id
|
|
processed = self.checkpoint.rows_processed
|
|
batches = self.checkpoint.batches_done
|
|
start_time = time.time()
|
|
|
|
log.info(f"Total rows : {total:,}")
|
|
log.info(f"Resuming at : ID {start_id:,} ({processed:,} already done)")
|
|
|
|
try:
|
|
while True:
|
|
last_id, count, skipped = self.process_batch(start_id)
|
|
|
|
if count == 0:
|
|
log.info("All rows processed!")
|
|
break
|
|
|
|
processed += count
|
|
batches += 1
|
|
start_id = last_id
|
|
elapsed = time.time() - start_time
|
|
rate = processed / elapsed if elapsed > 0 else 0
|
|
remaining = (total - processed) / rate if rate > 0 else 0
|
|
|
|
log.info(
|
|
f"Batch {batches:>6,} | "
|
|
f"{processed:>12,}/{total:,} "
|
|
f"({processed/total*100:.1f}%) | "
|
|
f"Skipped(bot): {skipped:,} | "
|
|
f"{rate:>8,.0f} rows/s | "
|
|
f"ETA {remaining/3600:.1f}h"
|
|
)
|
|
|
|
# Save checkpoint every 10 batches
|
|
if batches % 10 == 0:
|
|
self.checkpoint.last_id = start_id
|
|
self.checkpoint.rows_processed = processed
|
|
self.checkpoint.batches_done = batches
|
|
self.checkpoint.save(CHECKPOINT_FILE)
|
|
self.report.save_partial()
|
|
log.info("Checkpoint saved")
|
|
|
|
except KeyboardInterrupt:
|
|
log.warning("Interrupted! Saving checkpoint...")
|
|
self.checkpoint.last_id = start_id
|
|
self.checkpoint.rows_processed = processed
|
|
self.checkpoint.batches_done = batches
|
|
self.checkpoint.save(CHECKPOINT_FILE)
|
|
self.report.save_partial()
|
|
log.info("Re-run to resume.")
|
|
|
|
except Exception as e:
|
|
log.error(f"Error: {e}", exc_info=True)
|
|
self.checkpoint.save(CHECKPOINT_FILE)
|
|
self.report.save_partial()
|
|
raise
|
|
|
|
finally:
|
|
self.disconnect()
|
|
self.cache.close()
|
|
self.geoip.close()
|
|
|
|
# Print bot detection summary
|
|
bot_report = self.bot.report()
|
|
log.info(f"Bot detection summary: {bot_report}")
|
|
|
|
# Export reports
|
|
csv_path = self.report.export_csv()
|
|
summary_path = self.report.export_summary()
|
|
self.report.print_top(20)
|
|
|
|
# Print bot stats
|
|
bot_report = self.bot.report()
|
|
print(f"\n{'─'*45}")
|
|
print("Bot Detection Summary")
|
|
print(f"{'─'*45}")
|
|
print(f"Total rows checked : {bot_report['total']:>12,}")
|
|
print(f"Skipped (total) : {bot_report['skipped_total']:>12,} ({bot_report['skip_rate']})")
|
|
print(f" Private IP : {bot_report['skipped_private']:>12,}")
|
|
print(f" High frequency : {bot_report['skipped_freq']:>12,}")
|
|
print(f" Bot user agent : {bot_report['skipped_ua']:>12,}")
|
|
print(f" Attack path : {bot_report['skipped_path']:>12,}")
|
|
print(f"Unique bot IPs : {bot_report['bot_ips_found']:>12,}")
|
|
print(f"{'─'*45}\n")
|
|
|
|
# Cleanup checkpoint
|
|
if os.path.exists(CHECKPOINT_FILE):
|
|
os.remove(CHECKPOINT_FILE)
|
|
|
|
elapsed = time.time() - start_time
|
|
log.info(f"Completed in {elapsed/3600:.2f}h")
|
|
log.info(f"Reports: {csv_path}, {summary_path}")
|
|
log.info(f"Cache: {self.cache.stats()}")
|
|
|
|
|
|
# ─── Entry Point ──────────────────────────────────────────────────────────────
|
|
|
|
if __name__ == "__main__":
|
|
processor = IPGeoProcessor()
|
|
processor.run()
|