warren/wp-statpress-analyze
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

init commit

Browse Source
This commit is contained in:
warren
2026-09-30 21:05:53 +08:00
commit f7410dcfe4
6 changed files with 2219 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
analyze.py
+549
View File
@@ -0,0 +1,549 @@
"""
Report Analyzer
Reads the CSV output from ip_geo_report.py and generates:
1. Monthly total visit count with MoM change
2. Monthly unique country count with MoM change
3. Top countries per month
4. Export to CSV + print to console
"""
import os
import csv
import argparse
from datetime import datetime
from collections import defaultdict
from typing import Dict, List, Tuple, Optional
try:
import matplotlib.pyplot as plt
import matplotlib.ticker as mticker
from matplotlib.gridspec import GridSpec
HAS_MATPLOTLIB = True
except ImportError:
HAS_MATPLOTLIB = False
# ─── Config ───────────────────────────────────────────────────────────────────
REPORT_DIR = "./report"
OUTPUT_DIR = "./analysis"
TOP_N_COUNTRIES = 10 # top N countries to show per month
# ─── Data Loader ──────────────────────────────────────────────────────────────
def load_report(path: str) -> List[dict]:
"""
Load report CSV generated by ip_geo_report.py
Expected columns: Country, Date, Visit Count
"""
rows = []
with open(path, newline="", encoding="utf-8") as f:
reader = csv.DictReader(f)
for row in reader:
try:
rows.append({
"country": row["Country"].strip(),
"date": row["Date"].strip(),
"count": int(row["Visit Count"]),
"year_month": parse_year_month(row["Date"]),
})
except (KeyError, ValueError):
continue
print(f"Loaded {len(rows):,} rows from {path}")
return rows
def parse_year_month(date_str: str) -> str:
"""
Parse date string to YYYY-MM regardless of format:
- 2023-08-01 → 2023-08
- 20230801 → 2023-08
- 2023/08/01 → 2023-08
- 2023-08-01 00:00:00 → 2023-08
"""
date_str = date_str.strip().split(" ")[0] # strip time part
date_str = date_str.replace("/", "-") # normalize slashes
if "-" in date_str:
# YYYY-MM-DD
return date_str[:7]
elif len(date_str) == 8:
# YYYYMMDD
return f"{date_str[:4]}-{date_str[4:6]}"
else:
# fallback — try common formats
for fmt in ("%Y%m%d", "%Y-%m-%d", "%d/%m/%Y", "%m/%d/%Y"):
try:
return datetime.strptime(date_str, fmt).strftime("%Y-%m")
except ValueError:
continue
return date_str[:7] # last resort
"""Find the most recent report_*.csv in report dir."""
files = [
f for f in os.listdir(report_dir)
if f.startswith("report_") and f.endswith(".csv")
]
if not files:
return None
files.sort(reverse=True)
return os.path.join(report_dir, files[0])
# ─── Aggregation ──────────────────────────────────────────────────────────────
def aggregate_monthly(rows: List[dict]) -> Dict[str, dict]:
"""
Aggregate by year_month:
{
"2025-01": {
"total_visits": 12345,
"countries": {"United States": 5000, "Taiwan": 3000, ...},
"unique_countries": 42
}
}
"""
monthly: Dict[str, dict] = defaultdict(lambda: {
"total_visits": 0,
"countries": defaultdict(int)
})
for row in rows:
ym = row["year_month"]
monthly[ym]["total_visits"] += row["count"]
monthly[ym]["countries"][row["country"]] += row["count"]
# Compute unique country count per month
result = {}
for ym in sorted(monthly.keys()):
data = monthly[ym]
result[ym] = {
"total_visits": data["total_visits"],
"countries": dict(data["countries"]),
"unique_countries": len(data["countries"]),
}
return result
# ─── MoM Calculation ──────────────────────────────────────────────────────────
def calc_mom(current: int, previous: int) -> Tuple[float, str]:
"""
Returns (pct_change, formatted_string)
e.g. (12.5, '+12.5%') or (-3.2, '-3.2%') or (0.0, 'N/A')
"""
if previous == 0:
return 0.0, "N/A"
pct = (current - previous) / previous * 100
sign = "+" if pct >= 0 else ""
return pct, f"{sign}{pct:.1f}%"
def build_monthly_stats(monthly: Dict[str, dict]) -> List[dict]:
"""
Build a flat list of monthly stats with MoM columns.
"""
months = sorted(monthly.keys())
records = []
for i, ym in enumerate(months):
prev_ym = months[i - 1] if i > 0 else None
curr_data = monthly[ym]
prev_data = monthly[prev_ym] if prev_ym else None
curr_visits = curr_data["total_visits"]
curr_countries = curr_data["unique_countries"]
prev_visits = prev_data["total_visits"] if prev_data else 0
prev_countries = prev_data["unique_countries"] if prev_data else 0
visits_mom_pct, visits_mom_str = calc_mom(curr_visits, prev_visits)
countries_mom_pct, countries_mom_str = calc_mom(curr_countries, prev_countries)
# Top N countries this month
top = sorted(
curr_data["countries"].items(),
key=lambda x: -x[1]
)[:TOP_N_COUNTRIES]
records.append({
"year_month": ym,
"total_visits": curr_visits,
"visits_prev_month": prev_visits,
"visits_mom": visits_mom_str,
"visits_mom_pct": visits_mom_pct,
"unique_countries": curr_countries,
"countries_prev_month": prev_countries,
"countries_mom": countries_mom_str,
"countries_mom_pct": countries_mom_pct,
"top_countries": top,
})
return records
# ─── Export ───────────────────────────────────────────────────────────────────
def export_monthly_summary(records: List[dict], output_dir: str) -> str:
"""Export monthly summary CSV."""
os.makedirs(output_dir, exist_ok=True)
path = os.path.join(output_dir, f"monthly_summary_{datetime.today().strftime('%Y%m%d')}.csv")
with open(path, "w", newline="", encoding="utf-8") as f:
writer = csv.writer(f)
writer.writerow([
"Year-Month",
"Total Visits",
"Prev Month Visits",
"Visits MoM",
"Unique Countries",
"Prev Month Countries",
"Countries MoM",
"Top 1 Country",
"Top 1 Visits",
"Top 2 Country",
"Top 2 Visits",
"Top 3 Country",
"Top 3 Visits",
])
for r in records:
top = r["top_countries"]
def tc(n, field):
return top[n][field] if len(top) > n else ""
writer.writerow([
r["year_month"],
r["total_visits"],
r["visits_prev_month"],
r["visits_mom"],
r["unique_countries"],
r["countries_prev_month"],
r["countries_mom"],
tc(0, 0), tc(0, 1),
tc(1, 0), tc(1, 1),
tc(2, 0), tc(2, 1),
])
print(f"Monthly summary saved: {path}")
return path
def export_top_countries(records: List[dict], output_dir: str) -> str:
"""Export per-month top countries CSV."""
path = os.path.join(output_dir, f"top_countries_{datetime.today().strftime('%Y%m%d')}.csv")
with open(path, "w", newline="", encoding="utf-8") as f:
writer = csv.writer(f)
writer.writerow(["Year-Month", "Rank", "Country", "Visits", "% of Month"])
for r in records:
total = r["total_visits"]
for rank, (country, visits) in enumerate(r["top_countries"], 1):
pct = f"{visits/total*100:.1f}%" if total else "0%"
writer.writerow([r["year_month"], rank, country, visits, pct])
print(f"Top countries saved: {path}")
return path
# ─── Console Report ───────────────────────────────────────────────────────────
def _mom_arrow(pct: float) -> str:
if pct > 0: return "↑"
if pct < 0: return "↓"
return "→"
def _bar(value: int, max_value: int, width: int = 20) -> str:
if max_value == 0:
return ""
filled = int(value / max_value * width)
return "█" * filled + "░" * (width - filled)
def print_console_report(records: List[dict]):
"""Print a formatted report to the terminal."""
max_visits = max(r["total_visits"] for r in records) if records else 1
max_countries = max(r["unique_countries"] for r in records) if records else 1
# ── Monthly visits ────────────────────────────────────────────────────────
print()
print("═" * 72)
print(" 月份訪問量統計 Monthly Visit Count")
print("═" * 72)
print(f" {'月份':<10} {'訪問量':>12} {'MoM':>10} {'趨勢'}")
print(" " + "─" * 68)
for r in records:
arrow = _mom_arrow(r["visits_mom_pct"])
bar = _bar(r["total_visits"], max_visits)
print(
f" {r['year_month']:<10} "
f"{r['total_visits']:>12,} "
f"{r['visits_mom']:>10} "
f"{arrow} {bar}"
)
print()
# ── Monthly unique countries ───────────────────────────────────────────────
print("═" * 72)
print(" 月份不重複國家數 Monthly Unique Country Count")
print("═" * 72)
print(f" {'月份':<10} {'國家數':>10} {'MoM':>10} {'趨勢'}")
print(" " + "─" * 68)
for r in records:
arrow = _mom_arrow(r["countries_mom_pct"])
bar = _bar(r["unique_countries"], max_countries)
print(
f" {r['year_month']:<10} "
f"{r['unique_countries']:>10,} "
f"{r['countries_mom']:>10} "
f"{arrow} {bar}"
)
print()
# ── Top countries per month ────────────────────────────────────────────────
print("═" * 72)
print(f" 各月 Top {TOP_N_COUNTRIES} 國家 Top Countries per Month")
print("═" * 72)
for r in records:
total = r["total_visits"]
print(f"\n {r['year_month']} (共 {total:,} 次訪問)")
print(f" {'排名':>4} {'國家':<28} {'訪問量':>10} {'佔比':>6}")
print(" " + "─" * 56)
for rank, (country, visits) in enumerate(r["top_countries"], 1):
pct = f"{visits/total*100:.1f}%" if total else "0%"
print(f" {rank:>4}. {country:<28} {visits:>10,} {pct:>6}")
print()
# ── MoM Summary ───────────────────────────────────────────────────────────
if len(records) >= 2:
latest = records[-1]
prev = records[-2]
print("═" * 72)
print(" 最新月份 MoM 摘要 Latest Month MoM Summary")
print("═" * 72)
print(
f" {prev['year_month']} → {latest['year_month']}\n"
f" 訪問量：{prev['total_visits']:,} → {latest['total_visits']:,} "
f"({latest['visits_mom']})\n"
f" 國家數：{prev['unique_countries']} → {latest['unique_countries']} "
f"({latest['countries_mom']})"
)
print()
# ─── Visualization ────────────────────────────────────────────────────────────
def generate_charts(records: List[dict], output_dir: str) -> str:
"""
Generate a 4-panel chart:
1. Monthly total visits (bar + MoM line)
2. Monthly unique countries (bar + MoM line)
3. MoM % change comparison (line chart)
4. Top 5 countries stacked bar (latest 6 months)
"""
if not HAS_MATPLOTLIB:
print("matplotlib not installed — skipping charts.")
print("Install with: pip install matplotlib")
return ""
months = [r["year_month"] for r in records]
visits = [r["total_visits"] for r in records]
ctries = [r["unique_countries"] for r in records]
v_mom = [r["visits_mom_pct"] for r in records]
c_mom = [r["countries_mom_pct"] for r in records]
# replace 0.0 (N/A first month) with nan for cleaner line
# fill_between requires float nan, not None
v_mom_plot = [v if i > 0 else float("nan") for i, v in enumerate(v_mom)]
c_mom_plot = [v if i > 0 else float("nan") for i, v in enumerate(c_mom)]
fig = plt.figure(figsize=(16, 12))
fig.suptitle("IP Geo Report — Monthly Analysis", fontsize=16, fontweight="bold", y=0.98)
gs = GridSpec(2, 2, figure=fig, hspace=0.45, wspace=0.35)
colors = {
"visits": "#4C6EF5",
"countries": "#20C997",
"mom_pos": "#51CF66",
"mom_neg": "#FF6B6B",
"neutral": "#ADB5BD",
}
x = range(len(months))
# ── Panel 1: Monthly visits bar ───────────────────────────────────────────
ax1 = fig.add_subplot(gs[0, 0])
bars = ax1.bar(x, visits, color=colors["visits"], alpha=0.85, zorder=2)
ax1.set_title("Monthly Total Visits", fontweight="bold")
ax1.set_xticks(list(x))
ax1.set_xticklabels(months, rotation=45, ha="right", fontsize=8)
ax1.yaxis.set_major_formatter(mticker.FuncFormatter(lambda v, _: f"{v/1000:.0f}K" if v >= 1000 else str(int(v))))
ax1.set_ylabel("Visits")
ax1.grid(axis="y", linestyle="--", alpha=0.4, zorder=1)
ax1.spines[["top","right"]].set_visible(False)
# MoM % on secondary axis
ax1b = ax1.twinx()
ax1b.plot(list(x), v_mom_plot, color="#FF6B6B", marker="o", linewidth=1.8,
markersize=5, linestyle="--", label="MoM %", zorder=3)
ax1b.axhline(0, color="#FF6B6B", linewidth=0.6, linestyle=":")
ax1b.set_ylabel("MoM %", color="#FF6B6B", fontsize=9)
ax1b.tick_params(axis="y", labelcolor="#FF6B6B")
ax1b.yaxis.set_major_formatter(mticker.FuncFormatter(lambda v, _: f"{v:+.0f}%"))
# value labels on bars
for bar, val in zip(bars, visits):
ax1.text(bar.get_x() + bar.get_width()/2, bar.get_height() + max(visits)*0.01,
f"{val:,}", ha="center", va="bottom", fontsize=7, color="#333")
# ── Panel 2: Monthly unique countries bar ─────────────────────────────────
ax2 = fig.add_subplot(gs[0, 1])
bars2 = ax2.bar(x, ctries, color=colors["countries"], alpha=0.85, zorder=2)
ax2.set_title("Monthly Unique Countries", fontweight="bold")
ax2.set_xticks(list(x))
ax2.set_xticklabels(months, rotation=45, ha="right", fontsize=8)
ax2.set_ylabel("Countries")
ax2.grid(axis="y", linestyle="--", alpha=0.4, zorder=1)
ax2.spines[["top","right"]].set_visible(False)
ax2b = ax2.twinx()
ax2b.plot(list(x), c_mom_plot, color="#FF922B", marker="s", linewidth=1.8,
markersize=5, linestyle="--", label="MoM %", zorder=3)
ax2b.axhline(0, color="#FF922B", linewidth=0.6, linestyle=":")
ax2b.set_ylabel("MoM %", color="#FF922B", fontsize=9)
ax2b.tick_params(axis="y", labelcolor="#FF922B")
ax2b.yaxis.set_major_formatter(mticker.FuncFormatter(lambda v, _: f"{v:+.0f}%"))
for bar, val in zip(bars2, ctries):
ax2.text(bar.get_x() + bar.get_width()/2, bar.get_height() + max(ctries)*0.02,
str(val), ha="center", va="bottom", fontsize=7, color="#333")
# ── Panel 3: MoM % comparison line chart ─────────────────────────────────
ax3 = fig.add_subplot(gs[1, 0])
ax3.plot(list(x), v_mom_plot, color=colors["visits"], marker="o", linewidth=2,
markersize=6, label="Visits MoM %")
ax3.plot(list(x), c_mom_plot, color=colors["countries"], marker="s", linewidth=2,
markersize=6, label="Countries MoM %", linestyle="--")
ax3.axhline(0, color=colors["neutral"], linewidth=1, linestyle=":")
# shade positive / negative regions
import math
x_num = list(x)
ax3.fill_between(x_num, v_mom_plot, 0,
where=[not math.isnan(v) and v > 0 for v in v_mom_plot],
alpha=0.12, color=colors["mom_pos"], interpolate=True)
ax3.fill_between(x_num, v_mom_plot, 0,
where=[not math.isnan(v) and v < 0 for v in v_mom_plot],
alpha=0.12, color=colors["mom_neg"], interpolate=True)
ax3.set_title("MoM % Change Comparison", fontweight="bold")
ax3.set_xticks(list(x))
ax3.set_xticklabels(months, rotation=45, ha="right", fontsize=8)
ax3.set_ylabel("MoM %")
ax3.yaxis.set_major_formatter(mticker.FuncFormatter(lambda v, _: f"{v:+.0f}%"))
ax3.legend(fontsize=8)
ax3.grid(linestyle="--", alpha=0.4)
ax3.spines[["top","right"]].set_visible(False)
# ── Panel 4: Top 5 countries stacked bar (latest 6 months) ───────────────
ax4 = fig.add_subplot(gs[1, 1])
last = records[-6:] if len(records) >= 6 else records
top_months = [r["year_month"] for r in last]
# Collect global top 5 countries across shown months
country_totals: Dict[str, int] = defaultdict(int)
for r in last:
for country, cnt in r["top_countries"]:
country_totals[country] += cnt
top5 = [c for c, _ in sorted(country_totals.items(), key=lambda x: -x[1])[:5]]
palette = ["#4C6EF5","#20C997","#FF922B","#CC5DE8","#FF6B6B"]
bottoms = [0] * len(last)
for i, country in enumerate(top5):
vals = []
for r in last:
cmap = dict(r["top_countries"])
vals.append(cmap.get(country, 0))
ax4.bar(range(len(last)), vals, bottom=bottoms,
label=country, color=palette[i % len(palette)], alpha=0.88)
bottoms = [b + v for b, v in zip(bottoms, vals)]
ax4.set_title(f"Top 5 Countries — Last {len(last)} Months", fontweight="bold")
ax4.set_xticks(range(len(last)))
ax4.set_xticklabels(top_months, rotation=45, ha="right", fontsize=8)
ax4.yaxis.set_major_formatter(mticker.FuncFormatter(lambda v, _: f"{v/1000:.0f}K" if v >= 1000 else str(int(v))))
ax4.set_ylabel("Visits")
ax4.legend(fontsize=7, loc="upper left")
ax4.grid(axis="y", linestyle="--", alpha=0.4, zorder=1)
ax4.spines[["top","right"]].set_visible(False)
# ── Save ─────────────────────────────────────────────────────────────────
os.makedirs(output_dir, exist_ok=True)
path = os.path.join(output_dir, f"analysis_chart_{datetime.today().strftime('%Y%m%d')}.png")
fig.savefig(path, dpi=150, bbox_inches="tight")
plt.close(fig)
print(f"Chart saved: {path}")
return path
# ─── Entry Point ──────────────────────────────────────────────────────────────
def main():
global TOP_N_COUNTRIES
parser = argparse.ArgumentParser(description="IP Geo Report Analyzer")
parser.add_argument(
"--input", "-i",
help="Path to report CSV (default: latest in ./report/)",
default=None
)
parser.add_argument(
"--output", "-o",
help="Output directory (default: ./analysis/)",
default=OUTPUT_DIR
)
parser.add_argument(
"--top", "-n",
help=f"Top N countries per month (default: {TOP_N_COUNTRIES})",
type=int,
default=TOP_N_COUNTRIES
)
args = parser.parse_args()
# Find input file
input_path = args.input
if not input_path:
input_path = find_latest_report(REPORT_DIR)
if not input_path:
print(f"No report_*.csv found in {REPORT_DIR}")
print("Run ip_geo_report.py first, or specify --input path")
return
print(f"Analyzing: {input_path}")
# Load → aggregate → build stats
rows = load_report(input_path)
monthly = aggregate_monthly(rows)
records = build_monthly_stats(monthly)
if not records:
print("No data found in report.")
return
# Print to console
print_console_report(records)
# Export CSVs
export_monthly_summary(records, args.output)
export_top_countries(records, args.output)
# Generate chart
generate_charts(records, args.output)
print(f"\nAnalysis complete. Files saved to: {args.output}/")
if __name__ == "__main__":
main()