RomanNum3ral
/
cai
mirror of https://github.com/aliasrobotics/cai.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity
cai/tools/logs.py

682 lines
24 KiB
Python
Raw Permalink Blame History

"""
This script is used to create a web-based logs analysis dashboard.
It allows you to visualize the logs in different ways and see the PyPI download statistics.
Usage:
# Show all logs
python tools/web_logs.py <(cat ./logs.txt)
# Show last 10 logs and enable map
python tools/web_logs.py --enable-map <(tail -n 10 ./logs.txt)
Ideas for further improvements:
- Re-generate the log heatmap with only top 20 IPs
- Create a map with the top 20 IPs
- Dive into the logs
"""
import matplotlib
matplotlib.use("Agg")
from flask import Flask, render_template
import pandas as pd
import matplotlib.pyplot as plt
import io
import base64
from datetime import datetime
import os
import folium
import requests
import argparse
from typing import Dict, Optional
import numpy as np
import re
app = Flask(__name__)
# Configuration for enabled visualizations
class Config:
def __init__(self):
self.enable_map = False # Default to disabled
self.enable_daily_logs = True
self.enable_system_dist = True
self.enable_user_activity = True
@classmethod
def from_args(cls, args):
config = cls()
# Handle map options - disable takes precedence
if hasattr(args, "disable_map") and args.disable_map:
config.enable_map = False
elif hasattr(args, "enable_map") and args.enable_map:
config.enable_map = True
if hasattr(args, "disable_daily"):
config.enable_daily_logs = not args.disable_daily
if hasattr(args, "disable_system"):
config.enable_system_dist = not args.disable_system
if hasattr(args, "disable_users"):
config.enable_user_activity = not args.disable_users
return config
# Visualization components
class Visualizations:
def __init__(self, df: pd.DataFrame, config: Config):
self.df = df
self.config = config
def create_daily_logs(self) -> Optional[str]:
if not self.config.enable_daily_logs:
return None
plt.figure(figsize=(12, 6))
daily_counts = self.df.set_index("timestamp").resample("D").size()
daily_counts.index = daily_counts.index.strftime(
"%Y-%m-%d"
) # Format the index to 'yyyy-mm-dd'
# Plot bar chart for daily counts
ax = daily_counts.plot(kind="bar", color="skyblue", label="Daily Count")
# Plot line chart for cumulative counts
cumulative_counts = daily_counts.cumsum()
total_cumulative_count = cumulative_counts.iloc[-1] # Get the total cumulative count
cumulative_counts.plot(
kind="line",
color="orange",
secondary_y=True,
ax=ax,
label=f"Cumulative Count (Total: {total_cumulative_count})",
)
# Add vertical red line on 2025-04-09
if "2025-04-09" in daily_counts.index:
red_line_index = daily_counts.index.get_loc("2025-04-09")
ax.axvline(
x=red_line_index, color="red", linestyle="--", label="Public Release v0.3.11"
)
# Add grey-ish background to all elements prior to the red line
ax.axvspan(0, red_line_index, color="grey", alpha=0.3)
# Add vertical blue line on 2025-05-30
if "2025-05-30" in daily_counts.index:
green_line_index = daily_counts.index.get_loc("2025-05-30")
ax.axvline(
x=green_line_index,
color="green",
linestyle="--",
label='"CAIv0.4.0" and "alias1" releases',
)
# Add vertical yellow line on 2025-04-01
if "2025-04-01" in daily_counts.index:
yellow_line_index = daily_counts.index.get_loc("2025-04-01")
ax.axvline(
x=yellow_line_index,
color="yellow",
linestyle="--",
label="Professional Bug Bounty Test",
)
# Set titles and labels
ax.set_title("Number of Logs by Day")
ax.set_xlabel("Date")
ax.set_ylabel("Number of Logs")
ax.right_ax.set_ylabel("Cumulative Count")
ax.set_xticklabels(daily_counts.index, rotation=45)
# Add legends
ax.legend(loc="upper left")
ax.right_ax.legend(loc="upper right")
plt.tight_layout()
return self._get_plot_base64()
def create_system_distribution(self) -> Optional[str]:
if not self.config.enable_system_dist:
return None
plt.figure(figsize=(10, 6))
system_map = {
"linux": "Linux",
"darwin": "Darwin",
"windows": "Windows",
"microsoft": "Windows",
"wsl": "Windows",
}
self.df["system_grouped"] = self.df["system"].map(system_map).fillna("Other")
system_counts = self.df["system_grouped"].value_counts()
system_counts.plot(kind="bar")
plt.title("Total Number of Logs per System")
plt.xlabel("System")
plt.ylabel("Number of Logs")
plt.tight_layout()
return self._get_plot_base64()
def create_user_activity(self) -> Optional[str]:
if not self.config.enable_user_activity:
return None
plt.figure(figsize=(12, 6))
user_counts = self.df["username"].value_counts().head(50)
total_unique_users = self.df["username"].nunique()
ax = user_counts.plot(kind="bar")
plt.title(f"Top 50 Most Active Users (out of {total_unique_users} different users)")
plt.xlabel("Username")
plt.ylabel("Number of Logs")
plt.xticks(rotation=45)
# Add the actual number on top of each bar
for i, count in enumerate(user_counts):
ax.text(i, count, str(count), ha="center", va="bottom")
plt.tight_layout()
return self._get_plot_base64()
def create_map(self) -> Optional[str]:
if not self.config.enable_map:
return None
m = folium.Map(location=[40, -3], zoom_start=4)
for _, row in self.df.iterrows():
location = get_location(row["ip_address"])
folium.Marker(
location,
popup=f"{row['username']} ({row['ip_address']})<br>{row['timestamp']}",
tooltip=row["username"],
).add_to(m)
return m._repr_html_()
def create_ip_date_heatmap(self) -> Optional[str]:
# Only create if there are valid IPs (not 'disabled')
df = self.df[self.df["ip_address"] != "disabled"].copy()
if df.empty:
return None
# Use only date part for columns now
df["date"] = df["timestamp"].dt.strftime("%Y-%m-%d")
# Pivot: rows=ip, columns=date, values=count
pivot = df.pivot_table(
index="ip_address", columns="date", values="size", aggfunc="count", fill_value=0
)
if pivot.empty:
return None
# Order IPs by total logs (descending)
ip_order = pivot.sum(axis=1).sort_values(ascending=True).index.tolist()
pivot = pivot.loc[ip_order]
# Get human-readable locations for each IP
ip_labels = []
#
# TODO: note API limits
# for ip in pivot.index:
# loc = self._get_ip_location_label(ip)
# ip_labels.append(f"{ip} ({loc})")
#
for ip in pivot.index:
ip_labels.append(ip)
plt.figure(figsize=(max(6, 0.5 * len(pivot.columns)), min(20, 1 + 0.5 * len(pivot.index))))
ax = plt.gca()
im = ax.imshow(pivot.values, aspect="auto", cmap="YlOrRd", origin="lower")
plt.colorbar(im, ax=ax, label="Number of Logs")
ax.set_xticks(range(len(pivot.columns)))
ax.set_xticklabels(pivot.columns, rotation=90, fontsize=8)
ax.set_yticks(range(len(ip_labels)))
ax.set_yticklabels(ip_labels, fontsize=8)
plt.title("Log Heatmap: Number of Logs per IP Address and Date")
plt.xlabel("Date")
plt.ylabel("IP Address (Location)")
plt.tight_layout()
return self._get_plot_base64()
def _get_ip_location_label(self, ip: str) -> str:
# Try to get city/country from ip-api.com
if ip in ("127.0.0.1", "localhost"):
return "Vitoria, Spain"
try:
response = requests.get(f"http://ip-api.com/json/{ip}", timeout=5)
data = response.json()
if response.status_code == 200 and data.get("status") == "success":
city = data.get("city", "")
country = data.get("country", "")
if city and country:
return f"{city}, {country}"
elif country:
return country
except Exception:
pass
# Fallback to lat/lon
try:
lat, lon = get_location(ip)
return f"{lat:.2f},{lon:.2f}"
except Exception:
return "Unknown"
def _get_plot_base64(self) -> str:
buf = io.BytesIO()
plt.savefig(buf, format="png", bbox_inches="tight")
buf.seek(0)
plot_data = base64.b64encode(buf.getvalue()).decode()
plt.close()
return plot_data
def parse_logs(file_path, parse_ips=False):
logs = []
# Regex patterns for the three formats
# 1. Old: ...-cai_20250405_091537_root_linux_6.10.14-linuxkit_81_38_188_36.jsonl
old_pattern = re.compile(
r"cai_(\d{8})_(\d{6})_([^_]+)_([^_]+)_([^_]+)_(\d+)_(\d+)_(\d+)_(\d+)\.jsonl$"
)
# 2. New: uuid_cai_uuid_20250426_054313_root_linux_6.12.13-amd64_177_91_253_204.jsonl
new_pattern = re.compile(
r"([\w-]+)_cai_([\w-]+)_(\d{8})_(\d{6})_([^_]+)_([^_]+)_([^_]+)_([\d]+)_([\d]+)_([\d]+)_([\d]+)\.jsonl$"
)
# 3. Intermediate: logs/sessions/uuid/intermediate_20250422_222021.jsonl
intermediate_pattern = re.compile(r"intermediate_(\d{8})_(\d{6})\.jsonl$")
with open(file_path, "r") as file:
for line in file:
try:
parts = line.strip().split(None, 2)
if len(parts) != 3:
continue
size = parts[2].split()[0]
filename = parts[2].split()[1] if len(parts[2].split()) > 1 else parts[2]
# --- Old and New format ---
if "cai_" in filename:
# Try new format first
m_new = new_pattern.search(filename)
if m_new:
# uuid_cai_uuid_YYYYMMDD_HHMMSS_user_system_version_ip.jsonl
# Groups: 3=date, 4=time, 5=username, 6=system, 7=version, 8-11=ip
date_str = m_new.group(3)
time_str = m_new.group(4)
ts = f"{date_str[:4]}-{date_str[4:6]}-{date_str[6:]} {time_str[:2]}:{time_str[2:4]}:{time_str[4:]}"
username = m_new.group(5)
system = m_new.group(6).lower()
version = m_new.group(7)
if "microsoft" in system or "wsl" in version.lower():
system = "windows"
if parse_ips:
ip_address = ".".join(
[m_new.group(8), m_new.group(9), m_new.group(10), m_new.group(11)]
)
else:
ip_address = "disabled"
logs.append([ts, size, ip_address, system, username])
continue
# Try old format
m_old = old_pattern.search(filename)
if m_old:
# Groups: 1=date, 2=time, 3=username, 4=system, 5=version, 6-9=ip
date_str = m_old.group(1)
time_str = m_old.group(2)
ts = f"{date_str[:4]}-{date_str[4:6]}-{date_str[6:]} {time_str[:2]}:{time_str[2:4]}:{time_str[4:]}"
username = m_old.group(3)
system = m_old.group(4).lower()
version = m_old.group(5)
if "microsoft" in system or "wsl" in version.lower():
system = "windows"
if parse_ips:
ip_address = ".".join(
[m_old.group(6), m_old.group(7), m_old.group(8), m_old.group(9)]
)
else:
ip_address = "disabled"
logs.append([ts, size, ip_address, system, username])
continue
# --- Intermediate format ---
m_inter = intermediate_pattern.search(filename)
if m_inter:
# Only date is relevant
date_str = m_inter.group(1)
time_str = m_inter.group(2)
# Compose a timestamp from the extracted date/time
ts = f"{date_str[:4]}-{date_str[4:6]}-{date_str[6:]} {time_str[:2]}:{time_str[2:4]}:{time_str[4:]}"
logs.append([ts, size, "disabled", "unknown", "unknown"])
continue
# If none matched, skip
continue
except Exception as e:
print(f"Error parsing line: {line.strip()} -> {e}")
continue
return logs
def get_location(ip):
if ip in ("127.0.0.1", "localhost"):
return 42.85, -2.67 # Vitoria
# API 1: ip-api.com
try:
response = requests.get(f"http://ip-api.com/json/{ip}", timeout=5)
data = response.json()
if response.status_code == 200 and data.get("status") == "success":
return data["lat"], data["lon"]
except Exception:
pass
# API 2: ipinfo.io
try:
response = requests.get(f"https://ipinfo.io/{ip}/json", timeout=5)
data = response.json()
if response.status_code == 200 and "loc" in data:
lat, lon = map(float, data["loc"].split(","))
return lat, lon
except Exception:
pass
# API 3: ipwho.is
try:
response = requests.get(f"https://ipwho.is/{ip}", timeout=5)
data = response.json()
if response.status_code == 200 and data.get("success") is True:
return data["latitude"], data["longitude"]
except Exception:
pass
# Fallback
return 42.85, -2.67
def get_overall_stats():
"""Fetch overall download statistics for cai-framework"""
url = "https://pypistats.org/api/packages/cai-framework/overall"
response = requests.get(url)
if response.status_code == 200:
return response.json()
else:
print(f"Error fetching overall stats: {response.status_code}")
return None
def get_system_stats():
"""Fetch system-specific download statistics for cai-framework"""
url = "https://pypistats.org/api/packages/cai-framework/system"
response = requests.get(url)
if response.status_code == 200:
return response.json()
else:
print(f"Error fetching system stats: {response.status_code}")
return None
def create_pypi_plot():
# Get the data
overall_stats = get_overall_stats()
system_stats = get_system_stats()
if not overall_stats or not system_stats:
print("Error: Could not fetch PyPI statistics")
return None, None
# Create a figure with custom layout
plt.figure(figsize=(15, 8))
# Convert data to DataFrames
df_overall = pd.DataFrame(overall_stats["data"])
df_system = pd.DataFrame(system_stats["data"])
# Filter for downloads without mirrors (matches website reporting)
df_overall_no_mirrors = df_overall[df_overall["category"] == "without_mirrors"]
without_mirrors_total = df_overall_no_mirrors["downloads"].sum()
# Process the data
daily_downloads = df_overall_no_mirrors.groupby("date")["downloads"].sum().reset_index()
daily_downloads["date"] = pd.to_datetime(daily_downloads["date"])
# Add cumulative downloads
daily_downloads["cumulative_downloads"] = daily_downloads["downloads"].cumsum()
# Get release date (first date in the dataset)
release_date = daily_downloads["date"].min()
# Calculate system percentages for each day
system_pivot = df_system.pivot(index="date", columns="category", values="downloads")
system_pivot.index = pd.to_datetime(system_pivot.index)
system_pivot = system_pivot.fillna(0)
# Keep track of the total downloads per system for the legend
system_totals = system_pivot.sum()
# Create main plot with two y-axes
ax1 = plt.subplot(111)
ax2 = ax1.twinx() # Create a second y-axis sharing the same x-axis
# Plot total cumulative downloads on the left axis
ax1.plot(
daily_downloads["date"],
daily_downloads["cumulative_downloads"],
linewidth=3,
color="black",
label=f"Total Downloads (without mirrors): {without_mirrors_total:,}",
)
# Define color mapping for systems
color_map = {
"Darwin": "#1E88E5", # Blue
"Linux": "#FB8C00", # Orange
"Windows": "#43A047", # Green
"null": "#E53935", # Red
}
# Plot system distribution on the right axis
bottom = np.zeros(len(system_pivot))
# Ensure specific order of systems
desired_order = ["Darwin", "Linux", "Windows", "null"]
for col in desired_order:
if col in system_pivot.columns:
ax2.bar(
system_pivot.index,
system_pivot[col],
bottom=bottom,
label=col,
color=color_map[col],
alpha=0.5,
width=0.8,
)
bottom += system_pivot[col]
# Add release date annotation
ax1.axvline(x=release_date, color="#E53935", linestyle="--", alpha=0.7)
ax1.annotate(
"Release Date",
xy=(release_date, ax1.get_ylim()[1]),
xytext=(10, 10),
textcoords="offset points",
color="#E53935",
fontsize=10,
bbox=dict(boxstyle="round,pad=0.3", fc="white", ec="#E53935", alpha=0.8),
)
# Set the x-ticks to be at each date in the dataset
ax1.set_xticks(system_pivot.index)
ax1.set_xticklabels(
[date.strftime("%Y-%m-%d") for date in system_pivot.index],
rotation=45,
fontsize=10,
ha="right",
)
# Add padding between x-axis and the date labels
ax1.tick_params(axis="x", which="major", pad=10)
ax1.set_title("CAI Framework Download Statistics", fontsize=14, pad=20)
ax1.set_ylabel("Total Cumulative Downloads", fontsize=14, color="black")
ax2.set_ylabel("Daily Downloads by System", fontsize=14, color="black")
ax1.set_xlabel("Date", fontsize=14)
# Set grid and tick parameters
ax1.grid(True, linestyle="--", alpha=0.7)
ax1.tick_params(axis="y", colors="black")
ax2.tick_params(axis="y", colors="black")
# Add legend with combined information
handles1, labels1 = ax1.get_legend_handles_labels()
handles2, labels2 = [], []
# Add bars to legend in the desired order with correct colors
for col in desired_order:
if col in system_pivot.columns:
# Create a proxy artist with the correct color
proxy = plt.Rectangle((0, 0), 1, 1, fc=color_map[col], alpha=0.5)
handles2.append(proxy)
# Calculate percentage of both system total and overall total
system_percentage = (system_totals[col] / system_totals.sum()) * 100
website_percentage = (system_totals[col] / without_mirrors_total) * 100
labels2.append(f"{col} ({int(system_totals[col]):,} total, {system_percentage:.1f}%)")
# Create legend with updated colors
ax1.legend(
handles1 + handles2,
labels1 + labels2,
title="Operating Systems",
bbox_to_anchor=(1.05, 1),
loc="upper left",
fontsize=12,
title_fontsize=14,
)
plt.tight_layout()
# Create a BytesIO buffer for the image
buf = io.BytesIO()
plt.savefig(buf, format="png", bbox_inches="tight", dpi=300)
plt.close()
# Encode the image to base64 string
buf.seek(0)
image_base64 = base64.b64encode(buf.getvalue()).decode("utf-8")
# Prepare statistics for the template
stats = {
"total_downloads": without_mirrors_total,
"latest_downloads": daily_downloads.iloc[-1]["downloads"]
if not daily_downloads.empty
else 0,
"first_date": daily_downloads["date"].min().strftime("%Y-%m-%d")
if not daily_downloads.empty
else "N/A",
"last_date": daily_downloads["date"].max().strftime("%Y-%m-%d")
if not daily_downloads.empty
else "N/A",
"system_totals": {
col: int(system_totals[col])
for col in system_totals.index
if col in system_pivot.columns
},
"system_percentages": {
col: (system_totals[col] / system_totals.sum()) * 100
for col in system_totals.index
if col in system_pivot.columns
},
}
return f"data:image/png;base64,{image_base64}", stats
@app.route("/")
def index():
# Get log file path from app config
log_file = app.config["LOG_FILE"]
# Parse logs
logs = parse_logs(log_file, parse_ips=True)
if not logs:
return f"No logs were parsed. Please check if the file {log_file} exists and contains valid log entries."
df = pd.DataFrame(logs, columns=["timestamp", "size", "ip_address", "system", "username"])
df["timestamp"] = pd.to_datetime(df["timestamp"])
# Create visualizations
viz = Visualizations(df, app.config["VIZ_CONFIG"])
# Only create enabled visualizations
visualizations = {
"logs_by_day": viz.create_daily_logs(),
"logs_by_system": viz.create_system_distribution(),
"active_users": viz.create_user_activity(),
"ip_date_heatmap": viz.create_ip_date_heatmap(),
"config": app.config["VIZ_CONFIG"],
}
# Only create map if enabled
if app.config["VIZ_CONFIG"].enable_map:
visualizations["map_html"] = viz.create_map()
# Generate PyPI plot
pypi_plot, pypi_stats = create_pypi_plot()
visualizations["pypi_plot"] = pypi_plot
visualizations["pypi_stats"] = pypi_stats
return render_template("logs.html", **visualizations)
@app.route("/pypi-stats")
def pypi_stats():
# Generate PyPI plot
pypi_plot, stats = create_pypi_plot()
return render_template("pypi_stats.html", pypi_plot=pypi_plot, stats=stats)
def parse_args():
parser = argparse.ArgumentParser(description="Web-based log analysis dashboard")
parser.add_argument(
"log_file",
nargs="?",
default="/tmp/logs.txt",
help="Path to the log file (default: /tmp/logs.txt)",
)
# Map control group
map_group = parser.add_mutually_exclusive_group()
map_group.add_argument(
"--enable-map",
action="store_true",
help="Enable the geographic distribution map (default: disabled)",
)
map_group.add_argument(
"--disable-map",
action="store_true",
help="Disable the geographic distribution map (takes precedence)",
)
parser.add_argument("--disable-daily", action="store_true", help="Disable the daily logs chart")
parser.add_argument(
"--disable-system", action="store_true", help="Disable the system distribution chart"
)
parser.add_argument(
"--disable-users", action="store_true", help="Disable the user activity chart"
)
parser.add_argument(
"--port", type=int, default=5001, help="Port to run the server on (default: 5001)"
)
return parser.parse_args()
def main():
args = parse_args()
# Ensure the log file exists
if not os.path.exists(args.log_file):
print(f"Error: {args.log_file} not found!")
exit(1)
# Configure the application
app.config["LOG_FILE"] = args.log_file
app.config["VIZ_CONFIG"] = Config.from_args(args)
print(f"Starting web server on http://localhost:{args.port}")
print(f"Using log file: {args.log_file}")
app.run(host="0.0.0.0", port=args.port, debug=True)
if __name__ == "__main__":
main()
Reference in New Issue View Git Blame Copy Permalink