From 3c16caac1ad9bf3356cc2c2eba0cb51fcc2d015b Mon Sep 17 00:00:00 2001 From: kaireewu Date: Sun, 1 Nov 2026 15:34:53 +0800 Subject: [PATCH] feat(stats): buffer visits in memory and expose POST /api/visit + /api/stats --- reaction_board/config.py | 84 ++++++++++++++ reaction_board/web/queries.py | 85 ++++++++++++++ reaction_board/web/server.py | 120 +++++++++++++++++++- reaction_board/web/visits.py | 197 +++++++++++++++++++++++++++++++++ tests/test_config.py | 71 +++++++++++- tests/test_visits.py | 203 ++++++++++++++++++++++++++++++++++ tests/test_web_queries.py | 62 ++++++++++- tests/test_web_server.py | 88 ++++++++++++++- 8 files changed, 905 insertions(+), 5 deletions(-) create mode 100644 reaction_board/web/visits.py create mode 100644 tests/test_visits.py diff --git a/reaction_board/config.py b/reaction_board/config.py index eab2cf2..b6455c1 100644 --- a/reaction_board/config.py +++ b/reaction_board/config.py @@ -16,6 +16,12 @@ from dataclasses import dataclass, field from pathlib import Path from typing import Any, Mapping +from .client_ip import ( + DEFAULT_CLIENT_IP_HEADERS, + DEFAULT_TRUSTED_PROXIES, + parse_networks, +) + DEFAULT_CLASSIFY: dict[str, Any] = { "require_title": "咻咻满", "self_mid": 37754047, @@ -57,6 +63,15 @@ DEFAULT_CONFIG: dict[str, Any] = { "port": 8080, "auth_token": "", "public_base_url": "", + "track": { + "enabled": True, + "header_order": list(DEFAULT_CLIENT_IP_HEADERS), + "trusted_proxies": list(DEFAULT_TRUSTED_PROXIES), + "count_bots": False, + "flush_seconds": 60, + "keep_days": 180, + "max_hits_per_ip_path_per_day": 0, + }, }, "storage": { "database_path": "data/app.db", @@ -130,6 +145,18 @@ class WebConfig: port: int auth_token: str public_base_url: str + track: "TrackConfig" + + +@dataclass(frozen=True) +class TrackConfig: + enabled: bool + header_order: tuple[str, ...] + trusted_proxies: tuple[str, ...] + count_bots: bool + flush_seconds: int + keep_days: int + max_hits_per_ip_path_per_day: int @dataclass(frozen=True) @@ -270,6 +297,30 @@ def _as_str_list(value: Any, name: str) -> list[str]: return out +def _as_str_tuple(value: Any, name: str, *, allow_empty: bool) -> tuple[str, ...]: + if not isinstance(value, list): + raise ConfigError(f"{name} 必须是数组") + out: list[str] = [] + for item in value: + if not isinstance(item, str) or not item.strip(): + raise ConfigError(f"{name}[] 必须是非空字符串") + text = item.strip() + if text not in out: + out.append(text) + if not allow_empty and not out: + raise ConfigError(f"{name} 不能为空") + return tuple(out) + + +def _validate_networks(value: Any, name: str) -> tuple[str, ...]: + """校验 CIDR / 裸 IP 列表，任意非法项都视为配置错误。""" + texts = _as_str_tuple(value, name, allow_empty=True) + parsed = parse_networks(texts) + if len(parsed) != len(texts): + raise ConfigError(f"{name} 含非法 IP/CIDR") + return tuple(network.with_prefixlen for network in parsed) + + def _validate_folders(value: Any) -> tuple[FolderConfig, ...]: if not isinstance(value, list): raise ConfigError("folders 必须是数组") @@ -321,6 +372,38 @@ def _build_config(data: Mapping[str, Any], path: Path) -> AppConfig: web_raw = data["web"] if not isinstance(web_raw, Mapping): raise ConfigError("web 必须是对象") + track_raw = web_raw.get("track") + if track_raw is not None and not isinstance(track_raw, Mapping): + raise ConfigError("web.track 必须是对象") + track_raw = track_raw or {} + track = TrackConfig( + enabled=bool(track_raw.get("enabled", True)), + header_order=_as_str_tuple( + track_raw.get("header_order", list(DEFAULT_CLIENT_IP_HEADERS)), + "web.track.header_order", + allow_empty=False, + ), + trusted_proxies=_validate_networks( + track_raw.get("trusted_proxies", list(DEFAULT_TRUSTED_PROXIES)), + "web.track.trusted_proxies", + ), + count_bots=bool(track_raw.get("count_bots", False)), + flush_seconds=_as_int( + track_raw.get("flush_seconds", 60), + "web.track.flush_seconds", + minimum=0, + ), + keep_days=_as_int( + track_raw.get("keep_days", 180), + "web.track.keep_days", + minimum=1, + ), + max_hits_per_ip_path_per_day=_as_int( + track_raw.get("max_hits_per_ip_path_per_day", 0), + "web.track.max_hits_per_ip_path_per_day", + minimum=0, + ), + ) web = WebConfig( host=_as_str(web_raw.get("host", "127.0.0.1"), "web.host"), port=_as_int(web_raw.get("port", 8080), "web.port", minimum=1, maximum=65535), @@ -328,6 +411,7 @@ def _build_config(data: Mapping[str, Any], path: Path) -> AppConfig: public_base_url=_as_str( web_raw.get("public_base_url", "") or "", "web.public_base_url" ), + track=track, ) storage_raw = data["storage"] diff --git a/reaction_board/web/queries.py b/reaction_board/web/queries.py index dd6cfb0..37f8bb9 100644 --- a/reaction_board/web/queries.py +++ b/reaction_board/web/queries.py @@ -1560,6 +1560,91 @@ class DashboardQueries: } return self.db.read(run) + # ------------------------------------------------------------ stats + + def stats( + self, + *, + days: int = 30, + pending: Sequence[Mapping[str, Any]] | None = None, + ) -> dict[str, Any]: + """访问统计：区间内每日 PV/UV 与 Top 页面。 + + ``pending`` 为内存缓冲中尚未落库的上报行；与已落库数据按 + (日期, IP, 路由) 合并去重后统一聚合，因此页面基本是实时可见的。 + """ + def run(conn: sqlite3.Connection) -> dict[str, Any]: + cfg = self.config_store.get() + today = self._today() + start = _date_minus(today, int(days) - 1) + rows: list[Mapping[str, Any]] = [ + dict(row) for row in conn.execute( + "SELECT stat_date, ip, path, hits, first_at, last_at FROM visit " + "WHERE stat_date >= ? AND stat_date <= ?", + (start, today), + ).fetchall() + ] + if pending: + rows.extend(dict(row) for row in pending) + + buckets: dict[tuple[str, str, str], list[int]] = {} + for row in rows: + day = str(row["stat_date"]) + ip = str(row["ip"]) + path = str(row["path"]) + hits = int(row.get("hits") or 0) + first = int(row.get("first_at") or 0) + last = int(row.get("last_at") or 0) + bucket = buckets.get((day, ip, path)) + if bucket is None: + buckets[(day, ip, path)] = [hits, first, last] + else: + bucket[0] += hits + bucket[1] = min(bucket[1], first) if first else bucket[1] + bucket[2] = max(bucket[2], last) + + dates = _date_range(start, today) + daily_pv = {day: 0 for day in dates} + daily_uv: dict[str, set[str]] = {day: set() for day in dates} + page_pv: dict[str, int] = {} + page_uv: dict[str, set[str]] = {} + total_pv = 0 + total_ips: set[str] = set() + for (day, ip, path), bucket in buckets.items(): + if day not in daily_pv: + continue + daily_pv[day] += bucket[0] + daily_uv[day].add(ip) + total_pv += bucket[0] + total_ips.add(ip) + page_pv[path] = page_pv.get(path, 0) + bucket[0] + page_uv.setdefault(path, set()).add(ip) + + top_pages = [ + {"path": path, "pv": hits, "uv": len(page_uv[path])} + for path, hits in sorted( + page_pv.items(), key=lambda kv: (-kv[1], kv[0]) + )[:10] + ] + return { + "enabled": bool(cfg.web.track.enabled), + "days": int(days), + "range": {"start": start, "end": today}, + "today": today, + "today_pv": daily_pv.get(today, 0), + "today_uv": len(daily_uv.get(today, set())), + "total_pv": total_pv, + "total_uv": len(total_ips), + "series": { + "dates": dates, + "pv": [daily_pv[day] for day in dates], + "uv": [len(daily_uv[day]) for day in dates], + }, + "top_pages": top_pages, + "pending": bool(pending), + } + return self.db.read(run) + def candidates(self) -> dict[str, Any]: def run(conn: sqlite3.Connection) -> dict[str, Any]: cfg = self.config_store.get() diff --git a/reaction_board/web/server.py b/reaction_board/web/server.py index 996c40f..0869b08 100644 --- a/reaction_board/web/server.py +++ b/reaction_board/web/server.py @@ -21,15 +21,20 @@ from typing import Any, Callable, Mapping from urllib.parse import parse_qs, quote, unquote, urlsplit from ..analytics import is_loopback +from ..client_ip import extract_client_ip, is_bot, normalize_route, parse_networks from ..config import ConfigStore from ..database import Database from .queries import DashboardQueries +from .visits import VisitCollector STATIC_ROOT = Path(__file__).resolve().parent / "static" COOKIE_NAME = "rb_session" SESSION_TTL = 7 * 86400 -PROTECTED_API = {"/api/candidates", "/api/system"} +PROTECTED_API = {"/api/candidates", "/api/system", "/api/stats"} PROTECTED_PAGES = {"/candidates", "/system"} +# 显式放行的 POST 端点：访问统计埋点（业务写操作仍统一 405）。 +POST_API = {"/api/visit"} +VISIT_BODY_LIMIT = 2048 class ApiError(Exception): @@ -49,6 +54,7 @@ class WebContext: clock: Callable[[], float] = time.time db_path: str | None = None schema_version: int | None = None + visits: VisitCollector | None = None @property def auth_token(self) -> str: @@ -63,6 +69,13 @@ class DashboardHTTPServer(ThreadingHTTPServer): self.context = context super().__init__(address, DashboardHandler) + def server_close(self) -> None: + # 关闭前停止访问统计刷新线程，保证最后一次缓冲落库。 + collector = getattr(self.context, "visits", None) + if collector is not None: + collector.stop() + super().server_close() + class DashboardHandler(BaseHTTPRequestHandler): protocol_version = "HTTP/1.1" @@ -142,6 +155,16 @@ class DashboardHandler(BaseHTTPRequestHandler): self._send_error(HTTPStatus.UNAUTHORIZED, "unauthorized", "需要登录") return if method == "POST": + if path in POST_API: + self._api_post(path, query) + return + self._send_error( + HTTPStatus.METHOD_NOT_ALLOWED, + "method_not_allowed", + "不支持该操作", + ) + return + if path in POST_API: self._send_error( HTTPStatus.METHOD_NOT_ALLOWED, "method_not_allowed", @@ -285,6 +308,61 @@ border-color:#3a2c22;color:#f3e9dc}button{background:#e9b97c;color:#2a1a10}} # ------------------------------------------------------------ API + def _api_post(self, path: str, query: Mapping[str, list[str]]) -> None: + if path == "/api/visit": + self._visit_post(query) + return + raise ApiError(HTTPStatus.METHOD_NOT_ALLOWED, "method_not_allowed", "不支持该操作") + + def _visit_post(self, query: Mapping[str, list[str]]) -> None: + """记录一次页面浏览；前端 sendBeacon 上报，写内存后定期批量落库。""" + track = self.context.config_store.get().web.track + collector = self.context.visits + if not track.enabled or collector is None: + self._send_json(HTTPStatus.OK, {"recorded": False, "reason": "disabled"}) + return + + path_value = _param(query, "path", "") + if not path_value: + body = self._read_small_body() + if body: + path_value = _json_path(body) + + user_agent = self.headers.get("User-Agent") + if not track.count_bots and is_bot(user_agent): + self._send_json(HTTPStatus.OK, {"recorded": False, "reason": "bot"}) + return + + peer = self.client_address[0] if self.client_address else None + ip = extract_client_ip( + self.headers, + peer, + trusted_proxies=parse_networks(track.trusted_proxies), + header_order=track.header_order, + ) + if not ip: + self._send_json(HTTPStatus.OK, {"recorded": False, "reason": "no_ip"}) + return + + route = normalize_route(path_value) + recorded = collector.record(ip, route) + self._send_json(HTTPStatus.OK, { + "recorded": bool(recorded), + "route": route, + "reason": None if recorded else "capped", + }) + + def _read_small_body(self) -> bytes: + try: + length = int(self.headers.get("Content-Length", "0")) + except (TypeError, ValueError): + length = 0 + if length <= 0: + return b"" + if length > VISIT_BODY_LIMIT: + raise ApiError(HTTPStatus.BAD_REQUEST, "bad_request", "请求体过大") + return self.rfile.read(length) + def _api(self, path: str, query: Mapping[str, list[str]], method: str) -> None: queries = self.context.queries if path == "/api/overview": @@ -392,6 +470,16 @@ border-color:#3a2c22;color:#f3e9dc}button{background:#e9b97c;color:#2a1a10}} if path == "/api/about": self._send_json(HTTPStatus.OK, queries.about()) return + if path == "/api/stats": + pending = ( + self.context.visits.pending_rows() + if self.context.visits is not None else [] + ) + self._send_json(HTTPStatus.OK, queries.stats( + days=_int_param(query, "days", 30, 1, 3650), + pending=pending, + )) + return if path == "/api/candidates": self._send_json(HTTPStatus.OK, queries.candidates()) return @@ -510,6 +598,22 @@ def _safe_next(value: str | None) -> str: return text +def _json_path(body: bytes) -> str: + """从 JSON（或表单编码）请求体里取 ``path`` 字段；失败返回空串。""" + text = body.decode("utf-8", errors="replace").strip() + if not text: + return "" + try: + data = json.loads(text) + except (ValueError, TypeError): + data = None + if isinstance(data, Mapping): + value = data.get("path") + return value if isinstance(value, str) else "" + values = parse_qs(text).get("path") + return values[0] if values else "" + + def _param(query: Mapping[str, list[str]], name: str, default: str) -> str: values = query.get(name) return default if not values else str(values[0]) @@ -604,6 +708,15 @@ def create_server( ) if preview and not _preview_marker(db): raise ValueError("目标数据库不是预览库（缺少 preview:seeded=1 标记）") + track = cfg.web.track + visits = VisitCollector( + db, + clock=clock, + flush_seconds=track.flush_seconds, + keep_days=track.keep_days, + max_hits_per_key=track.max_hits_per_ip_path_per_day, + logger=logger, + ) context = WebContext( queries=DashboardQueries(db, config_store, clock=clock), config_store=config_store, @@ -612,8 +725,11 @@ def create_server( clock=clock, db_path=str(db.path), schema_version=db.schema_version, + visits=visits, ) - return DashboardHTTPServer((bind_host, bind_port), context) + server = DashboardHTTPServer((bind_host, bind_port), context) + visits.start() + return server def serve_forever(server: DashboardHTTPServer, *, logger: Any = None) -> None: diff --git a/reaction_board/web/visits.py b/reaction_board/web/visits.py new file mode 100644 index 0000000..a9e4ffc --- /dev/null +++ b/reaction_board/web/visits.py @@ -0,0 +1,197 @@ +# -*- coding: utf-8 -*- +"""访问统计的内存缓冲与批量入库。 + +前端通过 ``POST /api/visit`` 上报页面浏览。为降低 SQLite 压力，上报先写入进程内 +聚合缓冲，再由后台线程按固定周期（默认 60 秒）批量 upsert 到 ``visit`` 表。按 +(日期, IP, 路由) 聚合，同一分钟的重复浏览只累加计数。 + +进程被强杀时最多丢失一个刷新周期的数据，属可接受范围。批量写入经 Database 的 +单写线程串行执行，网络/请求路径不直接触碰数据库。 +""" + +from __future__ import annotations + +import threading +import time +from typing import Any, Callable + +from ..database import Database +from ..timeutil import shanghai_date + +# 缓冲里每个键保存 [hits, first_at, last_at]。 +_UPSERT_SQL = """ + INSERT INTO visit(stat_date, ip, path, hits, first_at, last_at) + VALUES (?, ?, ?, ?, ?, ?) + ON CONFLICT(stat_date, ip, path) DO UPDATE SET + hits = visit.hits + excluded.hits, + first_at = MIN(visit.first_at, excluded.first_at), + last_at = MAX(visit.last_at, excluded.last_at) +""" + + +class VisitCollector: + def __init__( + self, + db: Database, + *, + clock: Callable[[], float] = time.time, + flush_seconds: int = 60, + keep_days: int = 180, + max_hits_per_key: int = 0, + logger: Any = None, + ) -> None: + self._db = db + self._clock = clock + self._flush_seconds = max(0, int(flush_seconds)) + self._keep_days = max(1, int(keep_days)) + self._max_hits = max(0, int(max_hits_per_key)) + self._logger = logger + self._lock = threading.RLock() + self._buckets: dict[tuple[str, str, str], list[int]] = {} + self._last_prune_date: str | None = None + self._stop = threading.Event() + self._thread: threading.Thread | None = None + + # ------------------------------------------------------------ buffering + + def record(self, ip: str, path: str, *, now: float | None = None) -> bool: + """把一次浏览累加到内存缓冲；超过单键上限时返回 ``False``。""" + if not ip or not path: + return False + ts = int(self._clock() if now is None else now) + key = (shanghai_date(ts), ip, path) + with self._lock: + bucket = self._buckets.get(key) + if bucket is None: + bucket = [0, ts, ts] + self._buckets[key] = bucket + if self._max_hits and bucket[0] >= self._max_hits: + return False + bucket[0] += 1 + bucket[1] = min(bucket[1], ts) + bucket[2] = max(bucket[2], ts) + return True + + def pending_rows(self) -> list[dict[str, Any]]: + """返回未落库缓冲的快照，供 /api/stats 合并展示。""" + with self._lock: + return [ + { + "stat_date": day, + "ip": ip, + "path": path, + "hits": bucket[0], + "first_at": bucket[1], + "last_at": bucket[2], + } + for (day, ip, path), bucket in self._buckets.items() + ] + + @property + def pending_keys(self) -> int: + with self._lock: + return len(self._buckets) + + # ------------------------------------------------------------ flush + + def flush(self) -> int: + """把当前缓冲批量写入数据库，返回写入的聚合行数。""" + with self._lock: + if not self._buckets: + return 0 + take = self._buckets + self._buckets = {} + rows = [ + (day, ip, path, bucket[0], bucket[1], bucket[2]) + for (day, ip, path), bucket in take.items() + ] + + def write(conn) -> None: + conn.executemany(_UPSERT_SQL, rows) + + try: + self._db.transaction(write) + except Exception as exc: # noqa: BLE001 写失败不应影响服务 + # 按用户容忍度可丢弃，但内存极小，这里并回缓冲等待下次重试。 + with self._lock: + self._merge_back(take) + self._log("error", "visit.flush_error", error=str(exc)) + return 0 + self._maybe_prune() + self._log( + "info", "visit.flush", + rows=len(rows), hits=sum(row[3] for row in rows), + ) + return len(rows) + + def _merge_back(self, take: dict[tuple[str, str, str], list[int]]) -> None: + for key, bucket in take.items(): + current = self._buckets.get(key) + if current is None: + self._buckets[key] = list(bucket) + else: + current[0] += bucket[0] + current[1] = min(current[1], bucket[1]) + current[2] = max(current[2], bucket[2]) + + def _maybe_prune(self) -> None: + if self._keep_days <= 0: + return + today = shanghai_date(self._clock()) + if today == self._last_prune_date: + return + cutoff = shanghai_date( + self._clock() - self._keep_days * 86400 + ) + + def prune(conn) -> int: + cur = conn.execute( + "DELETE FROM visit WHERE stat_date < ?", (cutoff,) + ) + return int(cur.rowcount or 0) + + try: + removed = self._db.write(prune) + except Exception as exc: # noqa: BLE001 + self._log("error", "visit.prune_error", error=str(exc)) + return + self._last_prune_date = today + if removed: + self._log("info", "visit.prune", removed=removed, cutoff=cutoff) + + # ------------------------------------------------------------ lifecycle + + def start(self) -> None: + if self._flush_seconds <= 0 or self._thread is not None: + return + self._stop.clear() + self._thread = threading.Thread( + target=self._loop, name="visit-flusher", daemon=True + ) + self._thread.start() + + def _loop(self) -> None: + while not self._stop.wait(self._flush_seconds): + try: + self.flush() + except Exception as exc: # noqa: BLE001 后台线程自保 + self._log("error", "visit.flush_loop_error", error=str(exc)) + + def stop(self) -> None: + """停止后台线程并做最后一次 flush。""" + self._stop.set() + thread = self._thread + if thread is not None: + thread.join(timeout=5) + self._thread = None + try: + self.flush() + except Exception as exc: # noqa: BLE001 + self._log("error", "visit.stop_flush_error", error=str(exc)) + + def _log(self, level: str, event: str, **fields: Any) -> None: + if self._logger is None: + return + method = getattr(self._logger, level, None) + if method is not None: + method(event, **fields) \ No newline at end of file diff --git a/tests/test_config.py b/tests/test_config.py index 810d2e9..5066b91 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -172,13 +172,82 @@ def test_analytics_defaults_and_validation(): raise AssertionError(f"非法 analytics 必须拒绝: {bad}") +def test_web_track_defaults_and_validation(): + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "config.json" + write(path, base_config()) + cfg = load_config(path, env={}) + track = cfg.web.track + assert track.enabled is True + assert track.header_order == ("X-Real-IP", "X-Forwarded-For") + assert "172.16.0.0/12" in track.trusted_proxies + assert track.count_bots is False + assert track.flush_seconds == 60 + assert track.keep_days == 180 + assert track.max_hits_per_ip_path_per_day == 0 + + data = base_config() + data["web"] = { + "host": "0.0.0.0", "port": 8080, "auth_token": "t", + "public_base_url": "", + "track": { + "enabled": False, + "header_order": ["X-Forwarded-For", "X-Real-IP", "X-Real-IP"], + "trusted_proxies": ["10.0.0.0/8", "192.168.1.5"], + "count_bots": True, + "flush_seconds": 0, + "keep_days": 30, + "max_hits_per_ip_path_per_day": 500, + }, + } + write(path, data) + cfg = load_config(path, env={}) + assert cfg.web.track.enabled is False + # 去重且保持顺序 + assert cfg.web.track.header_order == ("X-Forwarded-For", "X-Real-IP") + assert cfg.web.track.trusted_proxies == ("10.0.0.0/8", "192.168.1.5/32") + assert cfg.web.track.count_bots is True + assert cfg.web.track.flush_seconds == 0 + assert cfg.web.track.max_hits_per_ip_path_per_day == 500 + + for bad, hint in ( + ({"header_order": []}, "header_order"), + ({"header_order": ["X-Real-IP", 5]}, "header_order"), + ({"trusted_proxies": ["not-a-cidr"]}, "trusted_proxies"), + ({"flush_seconds": -1}, "flush_seconds"), + ({"keep_days": 0}, "keep_days"), + ({"max_hits_per_ip_path_per_day": -1}, "max_hits_per_ip_path_per_day"), + ): + data = base_config() + data["web"] = {"track": bad} + write(path, data) + try: + load_config(path, env={}) + except ConfigError as exc: + assert hint in str(exc), (bad, str(exc)) + else: + raise AssertionError(f"非法 web.track 必须拒绝: {bad}") + + # web.track 不是对象 + data = base_config() + data["web"] = {"track": "nope"} + write(path, data) + try: + load_config(path, env={}) + except ConfigError as exc: + assert "web.track" in str(exc) + else: + raise AssertionError("web.track 必须是对象") + + def main(): test_dual_accounts_and_env_override() test_flat_and_missing_account_rejected() test_validation_require_read_auth_and_hot_reload() test_invalid_values() test_analytics_defaults_and_validation() - print("OK: 双账号配置加载、拒绝 flat 格式、ENV 覆盖、热加载与 analytics 测试全部通过") + test_web_track_defaults_and_validation() + print("OK: 双账号配置加载、拒绝 flat 格式、ENV 覆盖、热加载与 analytics/web.track 测试全部通过") if __name__ == "__main__": diff --git a/tests/test_visits.py b/tests/test_visits.py new file mode 100644 index 0000000..bbbcd73 --- /dev/null +++ b/tests/test_visits.py @@ -0,0 +1,203 @@ +# -*- coding: utf-8 -*- +"""访问统计内存缓冲与批量落库测试（离线）。 + +运行：uv run python tests/test_visits.py +""" + +import os +import sys +import tempfile +from pathlib import Path + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +sys.stdout.reconfigure(encoding="utf-8") + +from reaction_board.database import Database # noqa: E402 +from reaction_board.timeutil import day_bounds # noqa: E402 +from reaction_board.web.visits import VisitCollector # noqa: E402 + +DAY_A = "2026-10-30" +DAY_B = "2026-10-31" + + +def ts(day: str, hour: int = 12) -> int: + return int(day_bounds(day)[0]) + hour * 3600 + + +class _StubDB: + """只让 transaction/write 抛错，用于验证 flush 失败时的并回行为。""" + + def transaction(self, fn): + raise RuntimeError("db down") + + def write(self, fn): + raise RuntimeError("db down") + + +def _open(tmp: str) -> Database: + db = Database(Path(tmp) / "app.db") + db.initialize() + return db + + +def _rows(db: Database): + return db.read(lambda conn: [ + dict(row) for row in conn.execute( + "SELECT stat_date, ip, path, hits, first_at, last_at FROM visit " + "ORDER BY stat_date, ip, path" + ).fetchall() + ]) + + +def test_record_and_aggregate(): + with tempfile.TemporaryDirectory() as tmp: + db = _open(tmp) + try: + collector = VisitCollector(db, flush_seconds=0) + assert collector.record("1.1.1.1", "/home", now=ts(DAY_A)) is True + assert collector.record("1.1.1.1", "/home", now=ts(DAY_A)) is True + assert collector.record("1.1.1.1", "/videos", now=ts(DAY_A)) is True + assert collector.record("2.2.2.2", "/home", now=ts(DAY_A)) is True + # 跨日单独成键 + assert collector.record("1.1.1.1", "/home", now=ts(DAY_B)) is True + + assert collector.pending_keys == 4 + rows = {(r["stat_date"], r["ip"], r["path"]): r + for r in collector.pending_rows()} + home_a = rows[(DAY_A, "1.1.1.1", "/home")] + assert home_a["hits"] == 2 + assert home_a["first_at"] == ts(DAY_A) + assert home_a["last_at"] == ts(DAY_A) + assert rows[(DAY_B, "1.1.1.1", "/home")]["hits"] == 1 + + # 空值不计数 + assert collector.record("", "/home", now=ts(DAY_A)) is False + assert collector.record("1.1.1.1", "", now=ts(DAY_A)) is False + finally: + db.close() + + +def test_flush_writes_and_is_idempotent(): + with tempfile.TemporaryDirectory() as tmp: + db = _open(tmp) + try: + collector = VisitCollector(db, flush_seconds=0) + assert collector.flush() == 0 # 空缓冲 no-op + + collector.record("1.1.1.1", "/home", now=ts(DAY_A)) + collector.record("1.1.1.1", "/home", now=ts(DAY_A)) + assert collector.flush() == 1 + assert collector.pending_keys == 0 + rows = _rows(db) + assert len(rows) == 1 + assert rows[0]["hits"] == 2 + assert rows[0]["ip"] == "1.1.1.1" + assert rows[0]["path"] == "/home" + + # 再次 flush 不重复累加 + assert collector.flush() == 0 + assert _rows(db)[0]["hits"] == 2 + + # 与已有行累加（先在库中放一条 hits=5 的记录） + db.write(lambda conn: conn.execute( + "INSERT INTO visit(stat_date, ip, path, hits, first_at, last_at) " + "VALUES (?, '3.3.3.3', '/home', 5, ?, ?)", + (DAY_A, ts(DAY_A), ts(DAY_A)), + )) + collector.record("3.3.3.3", "/home", now=ts(DAY_A, hour=13)) + assert collector.flush() == 1 + row = next(r for r in _rows(db) if r["ip"] == "3.3.3.3") + assert row["hits"] == 6 + assert row["first_at"] == ts(DAY_A) + assert row["last_at"] == ts(DAY_A, hour=13) + finally: + db.close() + + +def test_flush_failure_merges_back(): + with tempfile.TemporaryDirectory() as tmp: + db = _open(tmp) + try: + collector = VisitCollector(db, flush_seconds=0) + collector.record("1.1.1.1", "/home", now=ts(DAY_A)) + collector.record("1.1.1.1", "/home", now=ts(DAY_A)) + + original = collector._db + collector._db = _StubDB() + try: + assert collector.flush() == 0 + assert collector.pending_keys == 1 + assert collector.pending_rows()[0]["hits"] == 2 + finally: + collector._db = original + + # 恢复后能正常落库且计数不丢、不重复 + assert collector.flush() == 1 + assert _rows(db)[0]["hits"] == 2 + finally: + db.close() + + +def test_max_hits_per_key_cap(): + with tempfile.TemporaryDirectory() as tmp: + db = _open(tmp) + try: + collector = VisitCollector(db, flush_seconds=0, max_hits_per_key=2) + assert collector.record("1.1.1.1", "/home", now=ts(DAY_A)) is True + assert collector.record("1.1.1.1", "/home", now=ts(DAY_A)) is True + assert collector.record("1.1.1.1", "/home", now=ts(DAY_A)) is False + assert collector.pending_rows()[0]["hits"] == 2 + # 不同键各自独立 + assert collector.record("1.1.1.1", "/videos", now=ts(DAY_A)) is True + finally: + db.close() + + +def test_prune_old_rows(): + with tempfile.TemporaryDirectory() as tmp: + db = _open(tmp) + try: + collector = VisitCollector(db, flush_seconds=0, keep_days=7) + # 30 天前的老记录 + old_day = "2026-09-30" + db.write(lambda conn: conn.execute( + "INSERT INTO visit(stat_date, ip, path, hits, first_at, last_at) " + "VALUES (?, '9.9.9.9', '/home', 1, 1, 1)", + (old_day,), + )) + collector.record("1.1.1.1", "/home", now=ts(DAY_A)) + assert collector.flush() == 1 + remaining = {row["stat_date"] for row in _rows(db)} + assert old_day not in remaining + assert DAY_A in remaining + finally: + db.close() + + +def test_stop_flushes_pending(): + with tempfile.TemporaryDirectory() as tmp: + db = _open(tmp) + try: + collector = VisitCollector(db, flush_seconds=0) + collector.start() # flush_seconds=0 → 不启动线程 + assert collector._thread is None + collector.record("1.1.1.1", "/home", now=ts(DAY_A)) + collector.stop() + assert collector.pending_keys == 0 + assert _rows(db)[0]["hits"] == 1 + finally: + db.close() + + +def main(): + test_record_and_aggregate() + test_flush_writes_and_is_idempotent() + test_flush_failure_merges_back() + test_max_hits_per_key_cap() + test_prune_old_rows() + test_stop_flushes_pending() + print("OK: 访问统计内存缓冲与批量落库测试全部通过") + + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/tests/test_web_queries.py b/tests/test_web_queries.py index efaca28..b622224 100644 --- a/tests/test_web_queries.py +++ b/tests/test_web_queries.py @@ -477,6 +477,65 @@ def test_about_payload(): db.close() +def test_stats_query(): + with tempfile.TemporaryDirectory() as tmp: + db, q = setup(tmp) + try: + empty = q.stats(days=7) + assert empty["today"] == "2026-10-31" + assert empty["today_pv"] == 0 and empty["today_uv"] == 0 + assert len(empty["series"]["dates"]) == 7 + assert empty["series"]["dates"][-1] == "2026-10-31" + assert empty["total_pv"] == 0 and empty["total_uv"] == 0 + assert empty["top_pages"] == [] + assert empty["enabled"] is True + assert empty["pending"] is False + + def insert(conn, day, ip, path, hits, first, last): + conn.execute( + "INSERT INTO visit(stat_date, ip, path, hits, first_at, last_at) " + "VALUES (?, ?, ?, ?, ?, ?)", + (day, ip, path, hits, first, last), + ) + + db.write(lambda conn: [ + insert(conn, "2026-10-31", "1.1.1.1", "/home", 3, 1, 2), + insert(conn, "2026-10-31", "2.2.2.2", "/home", 1, 1, 1), + insert(conn, "2026-10-31", "1.1.1.1", "/videos", 2, 1, 1), + insert(conn, "2026-10-30", "1.1.1.1", "/home", 5, 1, 1), + insert(conn, "2026-09-01", "9.9.9.9", "/home", 100, 1, 1), + ]) + stats = q.stats(days=7) + assert stats["today_pv"] == 6 + assert stats["today_uv"] == 2 + assert stats["total_pv"] == 11 # 今天 6 + 昨天 5，区间外不计 + assert stats["total_uv"] == 2 + assert stats["series"]["pv"][-1] == 6 + assert stats["series"]["pv"][-2] == 5 + assert stats["series"]["uv"][-1] == 2 + pages = {page["path"]: page for page in stats["top_pages"]} + assert pages["/home"]["pv"] == 9 and pages["/home"]["uv"] == 2 + assert pages["/videos"]["pv"] == 2 and pages["/videos"]["uv"] == 1 + assert stats["top_pages"][0]["path"] == "/home" + + # 内存缓冲合并：不同键相加，重叠键去重合并 + pending = [ + {"stat_date": "2026-10-31", "ip": "3.3.3.3", + "path": "/about", "hits": 4, "first_at": 1, "last_at": 1}, + {"stat_date": "2026-10-31", "ip": "1.1.1.1", + "path": "/home", "hits": 1, "first_at": 3, "last_at": 3}, + ] + merged = q.stats(days=7, pending=pending) + assert merged["today_pv"] == 11 + assert merged["today_uv"] == 3 + assert merged["pending"] is True + pages = {page["path"]: page for page in merged["top_pages"]} + assert pages["/about"]["pv"] == 4 + assert pages["/home"]["pv"] == 10 + finally: + db.close() + + def main_test(): test_overview() test_video_listing_filters_and_sorting() @@ -487,9 +546,10 @@ def main_test(): test_duration_efficiency_queries() test_year_filter_and_published_sort() test_about_payload() + test_stats_query() test_fans_custom_range_modes() test_fans_candidates_system_and_export() - print("OK: Web 查询层总览/榜单/详情/UP/时长效率/粉丝/候选/系统测试全部通过") + print("OK: Web 查询层总览/榜单/详情/UP/时长效率/粉丝/候选/系统/访问统计测试全部通过") if __name__ == "__main__": diff --git a/tests/test_web_server.py b/tests/test_web_server.py index 2cd1a56..030234a 100644 --- a/tests/test_web_server.py +++ b/tests/test_web_server.py @@ -189,6 +189,87 @@ def test_readonly_routes_and_static(): running.close() +def test_visit_endpoint_and_stats(): + with tempfile.TemporaryDirectory() as tmp: + running = RunningServer(tmp) + try: + ua = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) Chrome/155.0"} + # 上报：可信对端 127.0.0.1 + X-Real-IP → 记录真实 IP 与归一化路由 + status, _, payload = running.json( + "/api/visit", method="POST", + headers={**ua, "Content-Type": "application/json", "X-Real-IP": "203.0.113.7"}, + body=json.dumps({"path": "/video/123"}), + ) + assert status == 200 and payload["data"]["recorded"] is True + assert payload["data"]["route"] == "/video/:aid" + + status, _, payload = running.json( + "/api/visit", method="POST", + headers={**ua, "Content-Type": "application/json", "X-Real-IP": "203.0.113.7"}, + body=json.dumps({"path": "/videos"}), + ) + assert status == 200 and payload["data"]["recorded"] is True + + collector = running.server.context.visits + assert collector.pending_keys == 2 + + # 机器人 UA 被过滤，不计数 + status, _, payload = running.json( + "/api/visit", method="POST", + headers={"User-Agent": "Googlebot/2.1", "Content-Type": "application/json"}, + body=json.dumps({"path": "/home"}), + ) + assert status == 200 and payload["data"]["recorded"] is False + assert payload["data"]["reason"] == "bot" + assert collector.pending_keys == 2 + + # 超大请求体被拒绝 + status, _, _ = running.json( + "/api/visit", method="POST", + headers={**ua, "Content-Type": "application/json"}, + body=json.dumps({"path": "/" + "x" * 4000}), + ) + assert status == 400 + + # GET 埋点端点 → 405；其他业务 POST 仍 405 + status, _, _ = running.request("/api/visit") + assert status == 405 + status, _, _ = running.request("/api/videos/101", method="POST") + assert status == 405 + + # 未 flush 前库里没有行，但 /api/stats 合并内存缓冲可见 + assert running.db.read( + lambda c: c.execute("SELECT COUNT(*) FROM visit").fetchone()[0] + ) == 0 + status, _, payload = running.json("/api/stats?days=7") + assert status == 200 + data = payload["data"] + assert data["pending"] is True + assert data["today_pv"] == 2 + assert data["today_uv"] == 1 + pages = {page["path"]: page for page in data["top_pages"]} + assert pages["/video/:aid"]["pv"] == 1 + assert pages["/videos"]["pv"] == 1 + + # 手动 flush 后落库，且 IP 正确 + assert collector.flush() == 2 + assert running.db.read( + lambda c: c.execute("SELECT COUNT(*) FROM visit").fetchone()[0] + ) == 2 + ips = running.db.read( + lambda c: [row[0] for row in c.execute("SELECT DISTINCT ip FROM visit")] + ) + assert ips == ["203.0.113.7"] + + status, _, payload = running.json("/api/stats?days=7") + data = payload["data"] + assert data["pending"] is False + assert data["today_pv"] == 2 + assert data["range"]["end"] == "2026-10-31" + finally: + running.close() + + def test_token_auth_and_login(): with tempfile.TemporaryDirectory() as tmp: running = RunningServer(tmp, token="test-token") @@ -210,6 +291,8 @@ def test_token_auth_and_login(): assert status == 401 and body["error"]["code"] == "unauthorized" status, _, body = running.json("/api/system") assert status == 401 + status, _, body = running.json("/api/stats") + assert status == 401 status, headers, _ = running.request("/candidates") assert status == 303 and headers["Location"] == "/login?next=/candidates" status, headers, _ = running.request("/system") @@ -246,6 +329,8 @@ def test_token_auth_and_login(): assert status == 200 and body["ok"] is True status, _, body = running.json("/api/system", headers={"Cookie": cookie}) assert status == 200 and body["ok"] is True + status, _, body = running.json("/api/stats", headers={"Cookie": cookie}) + assert status == 200 and body["ok"] is True status, _, body = running.json( "/api/candidates", headers={"Authorization": "Bearer test-token"} ) @@ -299,9 +384,10 @@ def test_preview_and_loopback_guards(): def main_test(): test_readonly_routes_and_static() + test_visit_endpoint_and_stats() test_token_auth_and_login() test_preview_and_loopback_guards() - print("OK: Web 只读路由/鉴权/预览/安全边界测试全部通过") + print("OK: Web 只读路由/鉴权/预览/访问统计/安全边界测试全部通过") if __name__ == "__main__":