#!/usr/bin/env python3
"""
Day 189 · 03 — 趋势统计与静态看板
输入：01/02 生成的库（scans / scan_assets / events / assets）
输出：
1) 终端 ASCII 趋势（▁▂▃▄▅▆▇█）
2) 单文件 HTML 看板（内联 CSS + SVG，零依赖、可直接扔到任意静态服务器）
口径必须固定，否则趋势会"说谎"：
- 某天的开放端口数 = 当天**最后一次**扫描的观测数（不是当天所有扫描之和）
- 新增资产 = first_seen 落在当天的资产数
- 变更数 = 当天 events 表条数（appeared 另算，避免重复计数）
用法：
python3 03-trend-dashboard.py --db /tmp/day189.db --out /tmp/day189.html
python3 03-trend-dashboard.py --self-test
"""
from __future__ import annotations
import argparse
import html
import random
import sqlite3
import sys
import tempfile
from datetime import datetime, timedelta, timezone
from pathlib import Path
SCHEMA = """
CREATE TABLE IF NOT EXISTS scans (
id INTEGER PRIMARY KEY AUTOINCREMENT,
started_at TEXT NOT NULL,
label TEXT
);
CREATE TABLE IF NOT EXISTS assets (
id INTEGER PRIMARY KEY AUTOINCREMENT,
ip TEXT NOT NULL,
port INTEGER NOT NULL,
proto TEXT NOT NULL DEFAULT 'tcp',
service TEXT,
banner TEXT,
first_seen TEXT NOT NULL,
last_seen TEXT NOT NULL,
state TEXT NOT NULL DEFAULT 'active',
miss_count INTEGER NOT NULL DEFAULT 0,
UNIQUE (ip, port, proto)
);
CREATE TABLE IF NOT EXISTS scan_assets (
scan_id INTEGER NOT NULL,
asset_id INTEGER NOT NULL,
seen_at TEXT NOT NULL,
PRIMARY KEY (scan_id, asset_id)
);
CREATE TABLE IF NOT EXISTS events (
id INTEGER PRIMARY KEY AUTOINCREMENT,
scan_id INTEGER NOT NULL,
seen_at TEXT NOT NULL,
event TEXT NOT NULL,
ip TEXT NOT NULL,
port INTEGER NOT NULL,
proto TEXT NOT NULL,
old_value TEXT,
new_value TEXT
);
CREATE INDEX IF NOT EXISTS idx_scans_started ON scans(started_at);
CREATE INDEX IF NOT EXISTS idx_events_seen ON events(seen_at);
"""
def connect(path: str | Path) -> sqlite3.Connection:
conn = sqlite3.connect(path)
conn.row_factory = sqlite3.Row
conn.executescript(SCHEMA)
return conn
def collect_daily(conn: sqlite3.Connection) -> list[dict]:
"""按天聚合。只取每天最后一次扫描，避免"扫得越勤数字越大"的假趋势。"""
days = conn.execute(
"""SELECT date(started_at) AS d, MAX(id) AS last_scan
FROM scans GROUP BY date(started_at) ORDER BY d"""
).fetchall()
rows = []
for day in days:
d, scan_id = day["d"], day["last_scan"]
open_ports = conn.execute(
"SELECT COUNT(*) FROM scan_assets WHERE scan_id=?", (scan_id,)
).fetchone()[0]
new_assets = conn.execute(
"SELECT COUNT(*) FROM assets WHERE date(first_seen)=?", (d,)
).fetchone()[0]
ev = dict(conn.execute(
"""SELECT event, COUNT(*) AS n FROM events
WHERE date(seen_at)=? GROUP BY event""", (d,)
).fetchall())
rows.append({
"date": d,
"open_ports": open_ports,
"new_assets": new_assets,
"appeared": ev.get("appeared", 0),
"gone": ev.get("gone", 0),
"changed": ev.get("changed", 0),
"missing": ev.get("missing", 0),
})
return rows
BLOCKS = "▁▂▃▄▅▆▇█"
def sparkline(values: list[int]) -> str:
if not values:
return ""
lo, hi = min(values), max(values)
if hi == lo:
return BLOCKS[3] * len(values)
return "".join(BLOCKS[round((v - lo) / (hi - lo) * (len(BLOCKS) - 1))] for v in values)
def print_ascii(rows: list[dict]) -> None:
if not rows:
print("没有数据")
return
print(f"开放端口趋势 {rows[0]['date']} → {rows[-1]['date']}")
print(" " + sparkline([r["open_ports"] for r in rows]))
print(f" min={min(r['open_ports'] for r in rows)} "
f"max={max(r['open_ports'] for r in rows)} "
f"latest={rows[-1]['open_ports']}")
print()
print(f"{'日期':<12}{'端口':>6}{'新增':>6}{'下线':>6}{'变更':>6}")
for r in rows[-10:]:
print(f"{r['date']:<12}{r['open_ports']:>6}{r['new_assets']:>6}"
f"{r['gone']:>6}{r['changed']:>6}")
def _svg(rows: list[dict]) -> str:
w, h, pad = 900, 260, 44
if len(rows) < 2:
return "
数据点不足，无法绘制趋势
"
vals = [r["open_ports"] for r in rows]
lo, hi = min(vals), max(vals)
span = max(1, hi - lo)
step = (w - 2 * pad) / (len(rows) - 1)
def x(i: int) -> float:
return pad + i * step
def y(v: int) -> float:
return h - pad - (v - lo) / span * (h - 2 * pad)
points = " ".join(f"{x(i):.1f},{y(v):.1f}" for i, v in enumerate(vals))
area = f"{pad},{h - pad} " + points + f" {w - pad},{h - pad}"
grid = []
for g in range(5):
gv = lo + span * g / 4
gy = y(gv)
grid.append(
f''
f'{gv:.0f}'
)
labels = (
f'{rows[0]["date"]}'
f''
f'{rows[-1]["date"]}'
)
return (
f'"
)
def render_html(rows: list[dict], conn: sqlite3.Connection, db_path: str) -> str:
latest = rows[-1] if rows else {"open_ports": 0, "date": "-", "new_assets": 0}
total_assets = conn.execute("SELECT COUNT(*) FROM assets").fetchone()[0]
active = conn.execute("SELECT COUNT(*) FROM assets WHERE state='active'").fetchone()[0]
recent = conn.execute(
"""SELECT seen_at, event, ip, port, old_value, new_value
FROM events ORDER BY id DESC LIMIT 15"""
).fetchall()
cards = [
("当前资产", total_assets, "库中全部自然键"),
("活跃端口", active, "state = active"),
("最新快照端口", latest["open_ports"], latest["date"]),
("数据天数", len(rows), "覆盖的日历天"),
]
card_html = "".join(
f'
{html.escape(k)}
'
f'
{v}
{html.escape(s)}
'
for k, v, s in cards
)
trs = "".join(
f"
{html.escape(r['seen_at'])}
"
f"
{html.escape(r['event'])}
"
f"
{html.escape(r['ip'])}:{r['port']}
"
f"
{html.escape(r['old_value'] or '-')} → {html.escape(r['new_value'] or '-')}