78 lines
2.2 KiB
Python
78 lines
2.2 KiB
Python
#!/usr/bin/env python3
|
|
import re
|
|
import ipaddress
|
|
from datetime import datetime, timezone
|
|
import os
|
|
from local import * # Make sure this defines ACCESS_LOG, LOG_DIR, etc.
|
|
|
|
# ==== CONFIGURATION ====
|
|
MAX_LOG_LINES = 50000 # adjust as needed
|
|
LOG_DIR = os.path.join(os.path.dirname(__file__), "logs")
|
|
os.makedirs(LOG_DIR, exist_ok=True)
|
|
|
|
INTERNAL_NETWORKS = [
|
|
ipaddress.ip_network("10.0.0.0/8"),
|
|
ipaddress.ip_network("192.168.0.0/16"),
|
|
ipaddress.ip_network("172.16.0.0/12"),
|
|
]
|
|
|
|
log_line_re = re.compile(
|
|
r'(?P<ip>\S+) - - \[(?P<time>[^\]]+)\] "(?P<method>\S+) (?P<path>\S+) \S+"'
|
|
)
|
|
|
|
def is_external(ip):
|
|
ip_addr = ipaddress.ip_address(ip)
|
|
return not any(ip_addr in net for net in INTERNAL_NETWORKS)
|
|
|
|
def parse_nginx_line(line):
|
|
match = log_line_re.match(line)
|
|
if not match:
|
|
return None
|
|
data = match.groupdict()
|
|
if not is_external(data["ip"]):
|
|
return None
|
|
dt = datetime.strptime(data["time"], "%d/%b/%Y:%H:%M:%S %z")
|
|
dt_utc = dt.astimezone(timezone.utc)
|
|
iso_time = dt_utc.strftime("%Y-%m-%dT%H:%M:%SZ")
|
|
return f'{iso_time} {data["ip"]} {data["method"]} {data["path"]}'
|
|
|
|
def get_current_logfile():
|
|
"""Find or create the latest log file with line limit."""
|
|
today = datetime.now().strftime("%Y-%m-%d")
|
|
base_name = os.path.join(LOG_DIR, f"filtered_{today}")
|
|
|
|
index = 1
|
|
while True:
|
|
log_file = f"{base_name}_{index}.log"
|
|
if not os.path.exists(log_file):
|
|
return log_file
|
|
# Check line count
|
|
with open(log_file, "r") as f:
|
|
line_count = sum(1 for _ in f)
|
|
if line_count < MAX_LOG_LINES:
|
|
return log_file
|
|
index += 1
|
|
|
|
def process_log():
|
|
output_file = get_current_logfile()
|
|
buffer = []
|
|
|
|
with open(ACCESS_LOG, "r") as f:
|
|
for line in f:
|
|
parsed = parse_nginx_line(line)
|
|
if parsed:
|
|
buffer.append(parsed)
|
|
if buffer:
|
|
with open(output_file, "a") as out:
|
|
out.write("\n".join(buffer) + "\n")
|
|
def flush_access_log():
|
|
"""Safely truncate the access log after processing."""
|
|
with open(ACCESS_LOG, "w"):
|
|
pass # Opening with 'w' truncates file
|
|
|
|
def main():
|
|
process_log()
|
|
#flush_access_log()
|
|
|
|
if __name__ == "__main__":
|
|
main()
|