Merge branch 'feat/access-stats': basic PV/UV access statistics

This commit is contained in:
kairee committed 2026-11-01 15:40:53 +08:00
commit a3e7589ad3
29 files changed
+1898 -29

No files matched your search

+32 -1
View File
@@ -217,6 +217,7 @@ PRAGMA synchronous = NORMAL;
│ ├── scheduler.py # 内置任务调度
│ ├── logging_json.py # 脱敏 JSON 日志
│ ├── classify.py # 标题归一化、时长解析与打分
│ ├── client_ip.py # 来源 IP 提取、路由归一化、机器人判定(纯函数)
│ ├── video_ref.py # BV/av/URL/b23.tv 引用解析(纯函数)
│ ├── timeutil.py # 上海时区与每日时间
│ ├── bilibili/
@@ -238,8 +239,12 @@ PRAGMA synchronous = NORMAL;
│ └── web/ # P2 只读看板
│ ├── server.py # HTTP、路由、鉴权、静态资源
│ ├── queries.py # 看板 SQL 与数据整形
│ ├── visits.py # 访问统计内存缓冲与批量落库
│ └── static/ # 原生 JS/CSS、ECharts、预览页面
│
├── deploy/
│ └── nginx/ # 反代示例:真实来源 IP(realip)与站点配置
│
├── tools/
│ ├── seed_preview_db.py # 复制真实库并回填 7 天预览数据
│ └── import_fans_csv.py # 从 B 站粉丝历史 CSV 回填 up_stat
@@ -329,6 +334,7 @@ PRAGMA synchronous = NORMAL;
- `snapshot` 主键为 `(stat_date, aid)`,保证同日幂等;
- `up_stat` 主键为 `(stat_date, mid)`;
- `visit` 访问统计表主键为 `(stat_date, ip, path)`,`hits` 累加;数据先落内存缓冲、批量 upsert,`PV = SUM(hits)`、`UV = COUNT(DISTINCT ip)`;
- `folder` / `folder_resource` 保留为历史数据与重排序参考,**不再由服务写入**;
- `up_blacklist` 只在发现任务中生效,不影响已入库视频;
- `seen_aid` 永久保留搜索去重标记,候选过期删除后也不重复通知;
@@ -499,6 +505,7 @@ P3 实现说明:
### 9.1 API 原则
- P2 看板**只读**:业务接口只注册 GET,写方法统一返回 `405`;
- 仅显式放行两个 POST 端点:`POST /login`(建立会话)与 `POST /api/visit`(访问统计埋点);二者都不修改业务数据,其余 POST/PUT/PATCH/DELETE 仍统一 `405`;
- 数据管理(入库、忽略、拉黑、移出库、移除黑名单)继续通过 Telegram、CLI 或数据库完成;
- `/api/health` 和 `/api/session` 可匿名;
- 首页、榜单、视频/UP、粉丝页面和接口访客可见;
@@ -525,6 +532,8 @@ P3 实现说明:
| `GET` | `/api/candidates` | 候选池(只读) |
| `GET` | `/api/system` | 最近任务、下次任务、备份、黑名单 |
| `GET` | `/api/about` | 数据说明页当前口径(配置阈值、权重、覆盖度;无凭据) |
| `POST` | `/api/visit` | 访问统计埋点(匿名):归一化路径后写内存缓冲,定期批量落库 |
| `GET` | `/api/stats` | 访问统计(需登录):每日 PV/UV 序列、Top 页面,合并未落库缓冲 |
### 9.2 页面
@@ -538,7 +547,7 @@ P3 实现说明:
| UP 页 | 该 UP 的头像、mid、首RE时间、最近发布时间、全部视频、聚合趋势、单视频对比、总时长与播放·小时、入库时与当前粉丝数。 |
| 粉丝页 | 粉丝总数、日增、7 日净增、累计增长(标注起始日期);支持预设和自选日期,范围超过一年时按每月峰值日期聚合;最近记录默认 14 天并显示“较上一成功日”增量;不叠加视频数据。 |
| 候选池 | 待确认候选、规则命中、剩余保留时间;操作指向 Telegram/CLI。 |
| 系统状态 | 最近扫描、同步、快照、下次运行、备份、鉴权和黑名单。 |
| 系统状态 | 最近扫描、同步、快照、下次运行、备份、鉴权、黑名单与访问统计(近 30 天 PV/UV 与 Top 页面)。 |
### 9.3 预览模式
@@ -550,6 +559,15 @@ P3 实现说明:
- 重新生成预览库前必须停止预览服务,避免 Windows 文件占用;
- 第 9 轮起本地不再保留 `data/app.db` 真实副本;重新生成预览时用 `--source` 指向 `data/backup/` 中的真实库备份(如 `app-local-before-reseed-20261101.db`)。
### 9.4 访问统计与真实来源 IP
- 前端在整页加载与站内路由切换时通过 `navigator.sendBeacon("/api/visit")` 上报路径(`sort`/筛选等同路径重渲染不重复计数);
- 服务端 `reaction_board/web/visits.py` 把上报按 `(日期, IP, 路由)` 聚合到内存缓冲,后台线程每 `web.track.flush_seconds`(默认 60 秒)批量 upsert 到 `visit` 表;`server_close()` 时做最后一次 flush;
- `/api/stats` 合并未落库缓冲,因此系统页基本实时;`keep_days` 到期数据由后台按日清理;
- 真实 IP 由 nginx 用 `realip` 模块按 CDN 网段归一化后写入 `X-Real-IP`;应用(`reaction_board/client_ip.py`)仅在 socket 对端属于 `web.track.trusted_proxies` 时采信请求头,否则回退对端,避免直连请求伪造 `CF-Connecting-IP`/`EO-Connecting-IP`/`X-Forwarded-For`;
- 直连公网 EIP、Cloudflare、EdgeOne 三种来源的 nginx 配置与网段刷新脚本见 `deploy/nginx/`;配置项见 README 的 `web.track` 说明;
- 进程被强杀时最多丢失一个刷新周期的上报(可接受)。
---
## 10. 热度与趋势口径
@@ -632,6 +650,8 @@ uv run python tests/test_moves.py
- Telegram 用户与 chat ID 授权校验;
- 看板查询的筛选、分页、增长冷启动与状态过滤;
- 只读路由、鉴权、静态资源路径穿越和预览标记校验;
- 来源 IP 提取(直连 EIP/Cloudflare/EdgeOne、不可信对端伪造、IPv6、带端口、回退)与路由归一化、机器人判定;
- 访问统计内存缓冲(聚合、跨日、单键上限、flush 幂等、写失败并回、保留期清理)与 `POST /api/visit`、`GET /api/stats` 契约;
- 预览库生成的确定性、单调性、异常样本与真实库隔离;
- 粉丝历史 CSV 解析、同日去重、缺口计数与幂等/替换回填;
- UP 粉丝快照:首次写入不可覆盖、同轮去重、失败不中断、`live_since` 实时/回填标记、`backfill` 的 dry-run 与 limit;
@@ -759,6 +779,17 @@ uv run python tests/test_moves.py
- `/api/ups` 返回 `up_face` 与 `follower_latest`,`/api/ups/{mid}` 返回 `up_face`;
- 自动化测试更新:`test_migrations.py`、`test_database.py`、`test_web_queries.py`、`test_bilibili_client.py`、`test_library.py`、`test_snapshot.py`、`web_fixtures.py`。
### 第 12 轮:访问统计(PV/UV)✅(2026-11-01)
- 迁移 v7:新增 `visit(stat_date, ip, path, hits, first_at, last_at)` 聚合表;
- `reaction_board/client_ip.py`:来源 IP 提取、路由归一化与机器人判定纯函数(三种来源 + 伪造防护);
- `reaction_board/web/visits.py`:访问上报先落内存缓冲,后台每分钟批量 upsert,支持单键上限与保留期清理,关闭时最后一次 flush;
- 前端 `navigator.sendBeacon("/api/visit")` 埋点;Web 只读边界放宽为显式 POST 白名单(`/api/visit` + `/login`);
- `config.json` 新增 `web.track`(`enabled`/`header_order`/`trusted_proxies`/`count_bots`/`flush_seconds`/`keep_days`/`max_hits_per_ip_path_per_day`);
- 系统状态页展示近 30 天 PV/UV 与 Top 页面(`GET /api/stats`,需登录,合并未落库缓冲);
- `deploy/nginx/` 提供直连 EIP / Cloudflare / EdgeOne 的 realip 与站点配置、网段刷新脚本和说明;
- 自动化测试新增 `tests/test_client_ip.py`、`tests/test_visits.py`,并扩展 `test_config.py`、`test_web_queries.py`、`test_web_server.py`、`test_migrations.py` 等。
### P4:可选部署
- 反向代理、TLS、远程访问鉴权;
+30 -3
View File
@@ -22,6 +22,7 @@
| 搜索发现与候选池 | 可用(P3) | 每小时搜索前 2 页,分类打分后写入候选池。 |
| Telegram 通知与入库 | 可用(P3) | 入库 / 忽略 / 拉黑按钮;私聊消息手动入库。 |
| 本地数据看板 | 可用(P2/P9) | 只读总览、视频榜、UP 主榜、内容效率、视频/UP 详情、粉丝曲线、数据说明、候选池和系统状态。 |
| 访问统计(PV/UV) | 可用 | 前端埋点上报,按客户端 IP 与页面去重;先落内存、每分钟批量入库;系统页展示近 30 天。 |
## 核心工作流
@@ -176,10 +177,11 @@ uv run python app.py serve --preview --config data/preview/config.json
- 粉丝页只绘制粉丝曲线,不叠加视频数据;支持 7/30/90/全部与自选日期,范围超过一年时按每月峰值日期聚合;最近记录默认 14 天并显示较上一成功日的增量;
- 前端使用 History API 的干净路径路由(`/videos`、`/ups`、`/video/{aid}`、`/up/{mid}`、`/about`),不使用 `#/` hash;旧 `/rank` 链接仍可用;
- 候选池和系统状态只读展示,操作入口指向 Telegram/CLI;
- 系统状态页展示访问统计(近 30 天每日 PV/UV 与 Top 页面):前端在整页加载与站内路由切换时通过 `navigator.sendBeacon("/api/visit")` 上报,服务端按客户端 IP 与归一化路由聚合,先写内存缓冲、每分钟批量落库(详见 [访问统计与真实来源 IP](#访问统计与真实来源-ip));
- 首页、视频榜、UP 主榜、视频/UP 详情、粉丝页和数据说明页对访客可见;
- 配置 `web.auth_token` 后,“候选”和“系统”页面及其接口需要登录,启用登录页和 HttpOnly Cookie;访客的底部导航只显示公开页面;
- 本地回环地址且未配置 token 时视为无鉴权,全部页面可见;非回环地址且未配置 token 时拒绝启动;
- 所有业务接口只接受 GET,写方法返回 `405`;
- 所有业务接口只接受 GET,写方法返回 `405`;仅显式放行 `POST /api/visit`(访问统计埋点)与 `POST /login`(会话),其余 POST/PUT/PATCH/DELETE 仍统一 `405`;
- 首页/榜单聚合结果缓存 1 小时,粉丝结果缓存 24 小时;快照、发现、入库/忽略/拉黑、候选清理和粉丝 CSV 导入会立即让对应缓存失效。
### 预览数据
@@ -227,6 +229,28 @@ uv run python tools/import_fans_csv.py --csv E:\Downloads\history.csv --replace
- 写入前建议先备份 `data/app.db`,例如 `data/backup/app-pre-fans-YYYYMMDD.db`;
- 容器内运行时用 `--db /data/app.db` 或 `--config /config/config.json`,避免默认去找 `/app/config.json`。
## 访问统计与真实来源 IP
看板提供基础访问统计(PV/UV)。前端在整页加载与站内路由切换时通过
`navigator.sendBeacon("/api/visit")` 上报路径;服务端按 `(日期, 客户端 IP, 归一化路由)`
聚合,**先写入进程内缓冲、每分钟批量落库**到 `visit` 表,`/system` 系统状态页展示近
30 天每日 PV/UV 与 Top 页面。
统计按客户端 IP 去重,因此反代必须还原真实客户端 IP。链路为
`域名 → CDN → 公网 EIP → 云服务器内网 → 宿主机 nginx → 127.0.0.1:8080(容器)`:
容器内看到的 socket 对端是 Docker 网关,真实 IP 只能来自 nginx 注入的请求头。
- nginx 用 `realip` 模块按 CDN 网段把真实 IP 归一化到 `$remote_addr`,再写
`X-Real-IP` 传给容器;直连公网 EIP 时 `$remote_addr` 本身就是客户端;
- 应用只读 `X-Real-IP`,且仅当 socket 对端属于可信代理网段时采信,否则回退对端,
因此直连请求伪造的 `CF-Connecting-IP` / `EO-Connecting-IP` / `X-Forwarded-For` 无效;
- 三种来源(直连 EIP / Cloudflare / EdgeOne)的完整 nginx 配置、网段刷新脚本与
配置项说明见 [`deploy/nginx/README.md`](deploy/nginx/README.md)。
`config.json` 中的 `web.track` 控制统计行为:`enabled` / `header_order` /
`trusted_proxies` / `count_bots` / `flush_seconds` / `keep_days` /
`max_hits_per_ip_path_per_day`;默认过滤常见爬虫 UA、每分钟落库、保留 180 天。
## 快速部署(Docker)
仓库根目录已包含 `Dockerfile`、`compose.example.yml`、`.dockerignore` 和 `config.docker.example.json`。Compose 文件是示例,需要显式指定:
@@ -298,6 +322,7 @@ Copy-Item config.example.json config.json
- `track_fans`:需要每日追踪粉丝数的 UP mid;
- `discover`:搜索关键词、页数与时间窗(`keywords` / `pages` / `window_days`);
- `storage`:数据库与备份目录;
- `web`:绑定地址、端口、访问令牌与 `web.track`(访问统计开关、可信代理网段、上报头顺序、落库周期与保留期);
- `schedule` / `notify` / `classify` / `heat_weights`:任务节奏、通知阈值、分类规则和热度权重。
配置每 60 秒检查一次 mtime,凭据、分类规则等变更无需重启即可在下次任务读取;但 `schedule.discover_minutes` 与 `schedule.snapshot_at` 在启动时固化,修改后需重启 `run` 进程。
@@ -398,6 +423,8 @@ uv run python tests/test_auth.py # Cookie 合并、设备指纹、Set
uv run python tests/test_moves.py # 排序移动脚本:随机排列与头部用例
uv run python tests/test_favsort_http.py # favsort httpx 请求头与匿名回退
uv run python tests/test_config.py # 配置加载、校验、ENV 覆盖与热加载
uv run python tests/test_client_ip.py # 来源 IP 提取、路由归一化与机器人判定(纯函数)
uv run python tests/test_visits.py # 访问统计内存缓冲、批量落库与保留期清理
uv run python tests/test_migrations.py # 迁移、备份、回滚与保留期
uv run python tests/test_database.py # 单写队列、事务与并发
uv run python tests/test_repository.py # 收藏夹/视频状态机与黑名单
@@ -416,8 +443,8 @@ uv run python tests/test_timeutil.py # 上海时区与每日任务时间
uv run python tests/test_analytics.py # 热度、互动率、增长与封面
uv run python tests/test_preview_seed.py # 预览库回填、确定性与安全边界
uv run python tests/test_fans_csv.py # 粉丝 CSV 解析、替换与幂等回填
uv run python tests/test_web_queries.py # 看板查询层与冷启动
uv run python tests/test_web_server.py # 只读路由、鉴权、静态资源与预览模式
uv run python tests/test_web_queries.py # 看板查询层、冷启动与访问统计聚合
uv run python tests/test_web_server.py # 只读路由、鉴权、静态资源、访问统计与预览模式
uv run python tests/test_scheduler.py # 内置调度器到期/失败隔离
uv run python tests/test_provider.py # 凭据变化重建客户端与配置 revision
uv run python tests/test_favsort_run.py # 排序参数校验与混合资源类型
+19 -1
View File
@@ -20,7 +20,25 @@
"host": "0.0.0.0",
"port": 8080,
"auth_token": "replace-with-a-long-random-token",
"public_base_url": ""
"public_base_url": "",
"track": {
"enabled": true,
"header_order": ["X-Real-IP", "X-Forwarded-For"],
"trusted_proxies": [
"127.0.0.0/8",
"::1/128",
"10.0.0.0/8",
"172.16.0.0/12",
"192.168.0.0/16",
"169.254.0.0/16",
"fc00::/7",
"fe80::/10"
],
"count_bots": false,
"flush_seconds": 60,
"keep_days": 180,
"max_hits_per_ip_path_per_day": 0
}
},
"storage": {
"database_path": "/data/app.db",
+19 -1
View File
@@ -30,7 +30,25 @@
"host": "127.0.0.1",
"port": 8080,
"auth_token": "replace-with-a-long-random-token",
"public_base_url": ""
"public_base_url": "",
"track": {
"enabled": true,
"header_order": ["X-Real-IP", "X-Forwarded-For"],
"trusted_proxies": [
"127.0.0.0/8",
"::1/128",
"10.0.0.0/8",
"172.16.0.0/12",
"192.168.0.0/16",
"169.254.0.0/16",
"fc00::/7",
"fe80::/10"
],
"count_bots": false,
"flush_seconds": 60,
"keep_days": 180,
"max_hits_per_ip_path_per_day": 0
}
},
"storage": {
"database_path": "data/app.db",
+105
View File
@@ -0,0 +1,105 @@
# 访问统计与真实来源 IP
看板新增了基础访问统计(PV/UV)。前端在页面访问时通过
`POST /api/visit`(`navigator.sendBeacon`)上报,服务端先写入进程内缓冲、
每分钟批量落库到 `visit` 表(迁移 v7)。`/system` 系统状态页展示近 30 天
PV/UV 与 Top 页面。
统计按客户端 IP 去重,因此**必须在反代层还原真实客户端 IP**。
## 链路与 socket 对端
```text
浏览器 → 域名 → CDN(Cloudflare/EdgeOne) → 公网 EIP → 10.10.8.x(云服务器内网)
→ 宿主机 nginx → proxy_pass http://127.0.0.1:8080 → 容器
```
容器内 socket 对端经端口映射 SNAT 后是 Docker 网关(如 `172.17.0.1`),
不是真实客户端。因此真实 IP 只能来自 nginx 注入的请求头,且有三条来源:
| 场景 | nginx 处理 |
|---|---|
| 域名直接解析到公网 EIP | `$remote_addr` 即客户端,nginx 直接写 `X-Real-IP` |
| 经 Cloudflare | `realip` + `CF-Connecting-IP`(仅当对端属于 CF 网段才采信) |
| 经 EdgeOne | `realip` + `EO-Connecting-IP`(仅当对端属于 EdgeOne 网段才采信) |
应用只读 `X-Real-IP`,且**仅当 socket 对端属于可信代理网段**
(默认回环 + RFC1918 + Docker 网段,可配置)时才采信,否则回退到对端。
这样直连 EIP 的请求即使伪造 `CF-Connecting-IP` / `EO-Connecting-IP` /
`X-Forwarded-For` 也无法污染统计。
> 关键:用 `realip` 模块按 CDN 网段校验,而不是无条件相信 CDN 头。
> 若无条件相信,任何能直连公网 EIP 的人都能伪造来源 IP。
## 文件
| 文件 | 用途 |
|---|---|
| `reaction-board.conf` | 站点 `server {}`,反代到 `127.0.0.1:8080` 并注入真实 IP 头 |
| `realip-cloudflare.conf` | Cloudflare 场景,放在 `http {}` 内 |
| `realip-edgeone.conf` | EdgeOne 场景,放在 `http {}` 内 |
| `update-cloudflare-ips.sh` | 生成 Cloudflare 网段清单 `cloudflare-ips.conf` |
| `update-edgeone-ips.sh` | 由运维维护的 CIDR 清单生成 `edgeone-ips.conf` |
按实际来源**二选一**引入对应的 `realip-*.conf`(放在 `nginx.conf` 的
`http {}` 内),再 `include` 站点配置。
## Cloudflare
```nginx
# nginx.conf 的 http {} 内
include /etc/nginx/realip-cloudflare.conf;
include /etc/nginx/conf.d/reaction-board.conf;
```
```bash
sudo bash deploy/nginx/update-cloudflare-ips.sh
```
Cloudflare 的 IP 段变化很少,建议加 cron 每月刷新一次。
## EdgeOne
EdgeOne 节点网段会变动,需从 EdgeOne 控制台或 API 获取后维护成 CIDR 清单:
```bash
# source 为每行一个 CIDR 的清单
sudo bash deploy/nginx/update-edgeone-ips.sh /path/to/edgeone-cidrs.txt
```
```nginx
# nginx.conf 的 http {} 内
include /etc/nginx/realip-edgeone.conf;
include /etc/nginx/conf.d/reaction-board.conf;
```
若无法自动同步网段,可退化为:在安全组/防火墙只放行 EdgeOne 节点访问源站,
此时可放宽为直接采信 `EO-Connecting-IP`;但**同时保留直连公网 EIP 访问**的
场景下,仍建议使用 `realip` 网段校验。
## 应用侧配置
```jsonc
{
"web": {
"track": {
"enabled": true,
"header_order": ["X-Real-IP", "X-Forwarded-For"],
"trusted_proxies": [
"127.0.0.0/8", "::1/128",
"10.0.0.0/8", "172.16.0.0/12", "192.168.0.0/16", "169.254.0.0/16",
"fc00::/7", "fe80::/10"
],
"count_bots": false,
"flush_seconds": 60,
"keep_days": 180,
"max_hits_per_ip_path_per_day": 0
}
}
}
```
- `flush_seconds`:内存缓冲落库周期;设为 `0` 可关闭后台刷新(仅调试用)。
- `count_bots`:是否统计命中常见爬虫/监控 UA 的请求,默认过滤。
- `max_hits_per_ip_path_per_day`:单个(IP, 页面, 日期)计数上限,`0` 表示不限。
- `keep_days`:保留天数,后台按日清理更早的记录。
+26
View File
@@ -0,0 +1,26 @@
# 站点反代:域名 → CDN → 公网 EIP → 本机 nginx → 127.0.0.1:8080(容器)。
#
# 配合 realip-*.conf(放在 http {} 内)先把真实客户端 IP 归一化到 $remote_addr,
# 再通过 X-Real-IP 传给容器。应用仅在 socket 对端属于可信代理网段时采信该头。
server {
listen 80;
server_name board.example.com; # TODO: 改成你的域名
# 回源 HTTP(CDN 侧终结 TLS)。若要 nginx 直接对公网提供 HTTPS,
# 另建 443 server 并配置证书,下列 proxy_set_header 保持不变。
client_max_body_size 1m;
proxy_http_version 1.1;
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr; # 归一化后的真实客户端 IP
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; # 追加真实对端
proxy_set_header X-Forwarded-Proto $scheme;
location / {
proxy_pass http://127.0.0.1:8080;
}
# POST /api/visit(访问统计埋点)与其余只读接口走同一上游,无需额外配置。
}
+12
View File
@@ -0,0 +1,12 @@
# 放到 nginx.conf 的 http {} 内(或 http 级 include)。
#
# 仅当与 nginx 建立连接的对端属于下列 Cloudflare 网段时,才采信
# CF-Connecting-IP;直连公网 EIP 的请求无法伪造该头。
#
# 网段清单由 update-cloudflare-ips.sh 生成到 /etc/nginx/cloudflare-ips.conf。
# 若暂时未生成,可先注释掉 include(此时 CDN 访客会被记为 CF 节点 IP)。
real_ip_header CF-Connecting-IP;
real_ip_recursive on;
include /etc/nginx/cloudflare-ips.conf;
+16
View File
@@ -0,0 +1,16 @@
# 放到 nginx.conf 的 http {} 内(或 http 级 include)。
#
# 仅当与 nginx 建立连接的对端属于 EdgeOne 节点网段时,才采信
# EO-Connecting-IP;直连公网 EIP 的请求无法伪造该头。
#
# 网段清单 /etc/nginx/edgeone-ips.conf 由 update-edgeone-ips.sh 生成,
# 源数据从 EdgeOne 控制台或 API 导出(会变动,建议定期刷新)。
real_ip_header EO-Connecting-IP;
real_ip_recursive on;
include /etc/nginx/edgeone-ips.conf;
# 备选:用 EdgeOne 的 X-Forwarded-For(首段为真实客户端)配合网段递归提取。
# real_ip_header X-Forwarded-For;
# real_ip_recursive on;
+21
View File
@@ -0,0 +1,21 @@
#!/usr/bin/env bash
# 生成 Cloudflare 节点网段清单,供 realip-cloudflare.conf 使用。
#
# 用法: sudo bash deploy/nginx/update-cloudflare-ips.sh [输出文件]
# 默认输出: /etc/nginx/cloudflare-ips.conf
set -euo pipefail
OUT="${1:-/etc/nginx/cloudflare-ips.conf}"
{
for url in https://www.cloudflare.com/ips-v4 https://www.cloudflare.com/ips-v6; do
curl -fsSL "$url" | sed '/^[[:space:]]*$/d; s/^/set_real_ip_from /; s/$/;/'
done
} > "${OUT}.tmp"
mv "${OUT}.tmp" "$OUT"
echo "已写入 $OUT"
if command -v nginx >/dev/null 2>&1; then
nginx -t && nginx -s reload
fi
+20
View File
@@ -0,0 +1,20 @@
#!/usr/bin/env bash
# 由 EdgeOne 节点 CIDR 清单生成 realip 配置,供 realip-edgeone.conf 使用。
#
# EdgeOne 节点网段会变动,需从 EdgeOne 控制台或 API 导出为「每行一个 CIDR」
# 的清单后传入本脚本(不硬编码易过期的网段)。
#
# 用法: sudo bash deploy/nginx/update-edgeone-ips.sh <cidr-list-file> [输出文件]
# 默认输出: /etc/nginx/edgeone-ips.conf
set -euo pipefail
SOURCE="${1:?用法: update-edgeone-ips.sh <cidr-list-file> [输出文件]}"
OUT="${2:-/etc/nginx/edgeone-ips.conf}"
sed '/^[[:space:]]*#/d; /^[[:space:]]*$/d; s|^|set_real_ip_from |; s|$|;|' "$SOURCE" > "${OUT}.tmp"
mv "${OUT}.tmp" "$OUT"
echo "已写入 $OUT"
if command -v nginx >/dev/null 2>&1; then
nginx -t && nginx -s reload
fi
+8
View File
@@ -883,5 +883,13 @@ P0、P1、P2、P3、第 9 轮与第 10 轮已实施;P2 交付了只读本地
2. 基于粉丝数的"破圈"维度(播放/入库时粉丝数、粉丝分段、UP 中位数)——数据已就绪,榜单未实现;
3. P4(容器化 + 反代)继续保持可选。
第 12 轮(2026-11-01)交付:**基础访问统计(PV/UV)**:
1. 迁移 v7 新增 `visit(stat_date, ip, path, hits, first_at, last_at)` 聚合表;
2. 前端 `sendBeacon` 埋点 `POST /api/visit`;服务端内存缓冲、每分钟批量落库,关闭时 flush;
3. 来源 IP 兼容「域名直连公网 EIP / Cloudflare / EdgeOne」:nginx 用 realip 按 CDN 网段归一化到 `X-Real-IP`,应用仅在 socket 对端可信时采信(见 `deploy/nginx/`);
4. `/system` 系统状态页展示近 30 天 PV/UV 与 Top 页面;Web 只读边界放宽为显式 POST 白名单;
5. 纯函数 `client_ip.py` 与内存缓冲 `web/visits.py` 均有离线单元测试。
> 本地数据:第 9 轮起本地只保留可再生的预览库 `data/preview/app.db`(`tools/seed_preview_db.py` 生成),
> 原 `data/app.db` 真实副本已删除;需要重新生成或恢复时,用 `--source data/backup/app-local-before-reseed-20261101.db`。
+221
View File
@@ -0,0 +1,221 @@
# -*- coding: utf-8 -*-
"""来源 IP 提取与访问统计用纯函数。
服务经``域名 → CDN → 公网 EIP → 内网 → 宿主机 nginx → 容器``链路访问,容器内
socket 对端是 Docker 网关而非真实客户端,因此真实 IP 只能来自 nginx 注入的
请求头。安全前提:
* 只有当 socket 对端(``peer``)属于可信代理网段时才采信请求头;
* nginx 用 ``realip`` 模块按 CDN 网段校验后写入 ``X-Real-IP``(见文档);
* 直接解析到公网 EIP 的请求没有 CDN 头,回退到 ``peer``。
本模块不依赖网络、文件或时钟,便于离线单元测试。
"""
from __future__ import annotations
import ipaddress
import re
from typing import Iterable, Mapping, Sequence
# 默认可信代理网段:回环、RFC1918、链路本地、IPv6 ULA/链路本地。
# Docker 默认桥接网段在 172.16.0.0/12 内,容器内看到的对端即网关地址。
DEFAULT_TRUSTED_PROXIES: tuple[str, ...] = (
"127.0.0.0/8",
"::1/128",
"10.0.0.0/8",
"172.16.0.0/12",
"192.168.0.0/16",
"169.254.0.0/16",
"fc00::/7",
"fe80::/10",
)
# 依次尝试的请求头:nginx 归一化后的 X-Real-IP 优先,X-Forwarded-For 兜底。
DEFAULT_CLIENT_IP_HEADERS: tuple[str, ...] = ("X-Real-IP", "X-Forwarded-For")
# 已知前端路由:未知路径统一归为 /other,避免卡片式路径污染统计。
KNOWN_ROUTES: frozenset[str] = frozenset({
"/home", "/videos", "/ups", "/fans", "/about",
"/candidates", "/system", "/rank", "/login", "/logout",
})
_VIDEO_ROUTE_RE = re.compile(r"^/video/\d+$")
_UP_ROUTE_RE = re.compile(r"^/up/\d+$")
_MAX_ROUTE_LENGTH = 200
# 常见爬虫 / 监控 UA 关键词;命中时视为机器人(是否计数由调用方决定)。
# 这些标记在 UA 中通常作为子串出现(如 Googlebot / bingbot / Baiduspider),
# 因此不使用词边界,避免漏判。
_BOT_PATTERN = re.compile(
r"(?i)("
r"bot|crawler|spider|crawl|slurp|bingpreview|facebookexternalhit|"
r"python-requests|python-urllib|httpx|aiohttp|curl|wget|libwww|"
r"monitor|uptime|healthcheck|headlesschrome|phantomjs|"
r"semrush|ahrefs|dotbot|yandex|baiduspider|sogou|bytespider"
r")"
)
def normalize_ip(value: object) -> str | None:
"""规范化 IP 字符串;非法或空值返回 ``None``。
兼容 ``[::1]:443``、``1.2.3.4:5678`` 带端口写法,并把 IPv4-mapped IPv6
折叠为 IPv4(``::ffff:1.2.3.4`` → ``1.2.3.4``),保证同一客户端去重一致。
"""
if not isinstance(value, str):
return None
text = value.strip()
if not text:
return None
if text.startswith("["):
end = text.find("]")
if end == -1:
return None
text = text[1:end]
ip = _parse_ip(text)
if ip is None:
# host:port(IPv6 已在上面的直接解析里成功,不会走到这里)
host, sep, port = text.rpartition(":")
if not sep or not port.isdigit() or not host:
return None
ip = _parse_ip(host)
if ip is None:
return None
if isinstance(ip, ipaddress.IPv6Address) and ip.ipv4_mapped is not None:
ip = ip.ipv4_mapped
return ip.compressed
def _parse_ip(text: str) -> ipaddress.IPv4Address | ipaddress.IPv6Address | None:
try:
return ipaddress.ip_address(text)
except ValueError:
return None
def parse_networks(values: Iterable[str]) -> tuple[ipaddress.IPv4Network | ipaddress.IPv6Network, ...]:
"""把 CIDR / 裸 IP 字符串解析为网段元组;非法项忽略。"""
networks: list[ipaddress.IPv4Network | ipaddress.IPv6Network] = []
for raw in values:
if not isinstance(raw, str) or not raw.strip():
continue
text = raw.strip()
try:
networks.append(ipaddress.ip_network(text, strict=False))
except ValueError:
continue
return tuple(networks)
def is_trusted_proxy(
peer_ip: str | None,
networks: Sequence[ipaddress.IPv4Network | ipaddress.IPv6Network],
) -> bool:
ip = _parse_ip(peer_ip) if peer_ip else None
if ip is None:
return False
for network in networks:
if ip.version != network.version:
continue
if ip in network:
return True
return False
def _header_value(headers: Mapping[str, object], name: str) -> str | None:
"""大小写不敏感地读取请求头;兼容单值与列表两种取值。"""
plain = headers.get(name)
if plain is None:
lowered = name.lower()
for key, value in headers.items():
if isinstance(key, str) and key.lower() == lowered:
plain = value
break
if plain is None:
return None
if isinstance(plain, (list, tuple)):
return ",".join(str(item) for item in plain)
return str(plain)
def _rightmost_valid_forwarded(value: str) -> str | None:
"""X-Forwarded-For 取最右侧合法 IP。
nginx 的 ``$proxy_add_x_forwarded_for`` 会把真实对端追加到最右侧,客户端
自带的伪造条目在左侧,因此最右侧最可信。
"""
for part in reversed(value.split(",")):
normalized = normalize_ip(part)
if normalized is not None:
return normalized
return None
def extract_client_ip(
headers: Mapping[str, object],
peer_ip: str | None,
*,
trusted_proxies: Sequence[ipaddress.IPv4Network | ipaddress.IPv6Network],
header_order: Sequence[str] = DEFAULT_CLIENT_IP_HEADERS,
) -> str:
"""返回用于统计的客户端 IP。
* 对端不可信(或无法解析)时忽略所有请求头,直接回退到对端;
* 对端可信时按 ``header_order`` 依次取第一个合法值;
* 全部缺失时回退到规范化后的对端;对端也无法解析时返回 ``""``。
"""
peer = normalize_ip(peer_ip)
if peer is None:
return ""
if not is_trusted_proxy(peer, trusted_proxies):
return peer
for name in header_order:
value = _header_value(headers, name)
if value is None:
continue
if name.lower() == "x-forwarded-for":
candidate = _rightmost_valid_forwarded(value)
else:
candidate = normalize_ip(value)
if candidate is not None:
return candidate
return peer
def is_bot(user_agent: object) -> bool:
"""UA 命中已知机器人/监控模式时返回 ``True``;空 UA 不算机器人。"""
if not isinstance(user_agent, str) or not user_agent.strip():
return False
return bool(_BOT_PATTERN.search(user_agent))
def normalize_route(path: object) -> str:
"""把上报路径归一化为有限集合中的路由标签。
``/`` → ``/home``;``/video/123`` → ``/video/:aid``;``/up/123`` →
``/up/:mid``;已知路由原样保留;其余(含空值、超长、带查询串)→ ``/other``。
"""
if not isinstance(path, str):
return "/other"
text = path.strip()
if not text:
return "/other"
# 丢弃查询串与片段,并折叠末尾斜杠
text = text.split("?", 1)[0].split("#", 1)[0]
if not text.startswith("/"):
return "/other"
if len(text) > _MAX_ROUTE_LENGTH:
return "/other"
if text == "/" or text == "":
return "/home"
if text != "/" and text.endswith("/"):
text = text.rstrip("/") or "/home"
if _VIDEO_ROUTE_RE.match(text):
return "/video/:aid"
if _UP_ROUTE_RE.match(text):
return "/up/:mid"
if text in KNOWN_ROUTES:
return text
return "/other"
+84
View File
@@ -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"]
+20
View File
@@ -224,6 +224,26 @@ MIGRATIONS: tuple[Migration, ...] = (
"ALTER TABLE video ADD COLUMN up_face TEXT",
),
),
Migration(
version=7,
name="visit access statistics",
statements=(
# 访问统计按 (日期, IP, 路由) 聚合:PV = SUM(hits),UV = COUNT(DISTINCT ip)。
# 数据先落内存缓冲、批量 upsert,因此 hits 累加而非逐条插入。
"""
CREATE TABLE visit (
stat_date TEXT NOT NULL,
ip TEXT NOT NULL,
path TEXT NOT NULL,
hits INTEGER NOT NULL DEFAULT 0,
first_at INTEGER NOT NULL,
last_at INTEGER NOT NULL,
PRIMARY KEY (stat_date, ip, path)
)
""",
"CREATE INDEX idx_visit_date ON visit(stat_date)",
),
),
)
+85
View File
@@ -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()
+118 -2
View File
@@ -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:
+24
View File
@@ -57,3 +57,27 @@ export async function session() {
const payload = await response.json();
return payload?.data ?? payload;
}
export async function stats(params) {
return api("/api/stats", params);
}
export function visit(path) {
// 访问统计埋点:优先 sendBeacon(不阻塞导航),失败时用 keepalive fetch 兜底。
const body = JSON.stringify({ path });
try {
if (typeof navigator !== "undefined" && typeof navigator.sendBeacon === "function") {
const blob = new Blob([body], { type: "application/json" });
if (navigator.sendBeacon("/api/visit", blob)) return;
}
} catch {
// 忽略,走 fetch 兜底
}
fetch("/api/visit", {
method: "POST",
headers: { "Content-Type": "application/json" },
body,
credentials: "same-origin",
keepalive: true,
}).catch(() => {});
}
+7 -1
View File
@@ -1,4 +1,4 @@
import { session } from "./api.js";
import { session, visit } from "./api.js";
import { disposeAll } from "./charts.js";
import { onRoute, parseLocation, startRouter } from "./router.js";
import { renderCandidates } from "./views/candidates.js";
@@ -153,9 +153,15 @@ async function boot() {
toast("无法连接看板服务", "error");
}
renderNav();
let lastVisitPath = null;
onRoute(({ path, route }) => {
disposeAll();
setTitle(route, path);
// 访问统计:仅在路由路径变化时上报(排序/筛选等同路径重渲染不计 PV)。
if (path !== lastVisitPath) {
lastVisitPath = path;
visit(path);
}
});
startRouter({
root: document.querySelector("#app"),
+72
View File
@@ -189,3 +189,75 @@ export function emptyChart(element, message = "暂无数据") {
if (!element) return;
element.innerHTML = `<div class="chart-empty">${message}</div>`;
}
export function mixedChart(element, { dates, bars, line, barName = "PV", lineName = "UV" }) {
if (!element) return null;
const theme = colors();
const option = {
animationDuration: 300,
backgroundColor: "transparent",
grid: { left: 8, right: 12, top: 34, bottom: 8, containLabel: true },
legend: {
data: [barName, lineName],
top: 0,
textStyle: { color: theme.muted, fontSize: 11 },
},
tooltip: {
trigger: "axis",
axisPointer: { type: "shadow" },
backgroundColor: theme.surface,
borderColor: theme.grid,
textStyle: { color: theme.text, fontSize: 12 },
},
xAxis: {
type: "category",
data: dates,
axisLine: { lineStyle: { color: theme.grid } },
axisTick: { show: false },
axisLabel: {
color: theme.muted,
fontSize: 11,
formatter: (value) => formatDay(value),
hideOverlap: true,
},
},
yAxis: [
{
type: "value",
name: barName,
minInterval: 1,
splitLine: { lineStyle: { color: theme.grid, type: "dashed" } },
axisLabel: { color: theme.muted, fontSize: 11, formatter: (value) => formatAxis(value) },
},
{
type: "value",
name: lineName,
minInterval: 1,
splitLine: { show: false },
axisLabel: { color: theme.muted, fontSize: 11, formatter: (value) => formatAxis(value) },
},
],
series: [
{
name: barName,
type: "bar",
data: bars,
barMaxWidth: 22,
itemStyle: { color: theme.blue, borderRadius: [4, 4, 0, 0] },
},
{
name: lineName,
type: "line",
yAxisIndex: 1,
data: line,
smooth: false,
connectNulls: false,
symbol: "circle",
symbolSize: 5,
lineStyle: { width: 2.2, color: theme.accent },
itemStyle: { color: theme.accent },
},
],
};
return mount(element, option);
}
+51 -2
View File
@@ -1,4 +1,5 @@
import { api } from "../api.js";
import { api, stats } from "../api.js";
import { emptyChart, mixedChart } from "../charts.js";
import { emptyState, sectionHeader } from "../components.js";
import {
escapeHtml, formatCount, formatDateTime, formatNumber,
@@ -30,11 +31,42 @@ function taskRows(tasks) {
}
export async function renderSystem(root) {
const system = await api("/api/system");
const [system, visitStats] = await Promise.all([
api("/api/system"),
stats({ days: 30 }).catch(() => null),
]);
const health = system.snapshot_health || { total: 0, ok: 0, failed: 0 };
const counts = system.counts || {};
const next = system.next_runs || {};
const visitExtra = visitStats
? `<span class="section-note">近 ${visitStats.days} 天 · ${visitStats.pending ? "含未落库缓冲" : "已落库"}</span>`
: "";
const visitPanel = visitStats ? `
<section class="panel chart-panel">
${sectionHeader("访问统计", visitExtra)}
<div class="mini-stats">
<div><span>今日 PV</span><strong>${formatNumber(visitStats.today_pv)}</strong></div>
<div><span>今日 UV</span><strong>${formatNumber(visitStats.today_uv)}</strong></div>
<div><span>${visitStats.days} 天 PV</span><strong>${formatNumber(visitStats.total_pv)}</strong></div>
<div><span>${visitStats.days} 天 UV</span><strong>${formatNumber(visitStats.total_uv)}</strong></div>
</div>
<div class="chart" id="visit-chart"></div>
<p class="section-note">柱状为每日 PV,折线为每日 UV(按客户端 IP 去重)。数据来自前端埋点,先落内存、每分钟批量入库;仅统计页面访问,不含 API 与静态资源请求。</p>
${(visitStats.top_pages || []).length ? `<div class="table-wrap">
<table class="data-table">
<thead><tr><th>页面</th><th>PV</th><th>UV</th></tr></thead>
<tbody>
${visitStats.top_pages.map((page) => `<tr>
<td>${escapeHtml(page.path)}</td>
<td>${formatNumber(page.pv)}</td>
<td>${formatNumber(page.uv)}</td>
</tr>`).join("")}
</tbody>
</table>
</div>` : `<p class="section-note">暂无访问记录。</p>`}
</section>` : "";
root.innerHTML = `<section class="page system-page">
<header class="page-head">
<div>
@@ -51,6 +83,8 @@ export async function renderSystem(root) {
<div><span>已移出库</span><strong>${formatCount(counts.dropped)}</strong></div>
</div>
${visitPanel}
<section class="panel">
${sectionHeader("最近任务")}
${taskRows(system.tasks)}
@@ -119,4 +153,19 @@ export async function renderSystem(root) {
</div>` : emptyState("黑名单为空")}
</section>
</section>`;
const visitChart = root.querySelector("#visit-chart");
if (visitChart) {
if (visitStats && (visitStats.total_pv || visitStats.total_uv)) {
mixedChart(visitChart, {
dates: visitStats.series.dates,
bars: visitStats.series.pv,
line: visitStats.series.uv,
barName: "PV",
lineName: "UV",
});
} else {
emptyChart(visitChart, "暂无访问数据");
}
}
}
+197
View File
@@ -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)
+1 -1
View File
@@ -51,7 +51,7 @@ def test_init_and_status():
path = write_config(tmp)
code, output = run_cli(["init", "--config", str(path)])
assert code == 0
assert '"schema_version": 6' in output
assert '"schema_version": 7' in output
db = Database(Path(tmp) / "app.db")
db.initialize()
+268
View File
@@ -0,0 +1,268 @@
# -*- coding: utf-8 -*-
"""来源 IP 提取与路由/机器人判定纯函数测试(离线)。
运行:uv run python tests/test_client_ip.py
"""
import os
import sys
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
sys.stdout.reconfigure(encoding="utf-8")
from reaction_board.client_ip import ( # noqa: E402
DEFAULT_TRUSTED_PROXIES,
extract_client_ip,
is_bot,
is_trusted_proxy,
normalize_ip,
normalize_route,
parse_networks,
)
TRUSTED = parse_networks(DEFAULT_TRUSTED_PROXIES)
# 容器内看到的 Docker 网关地址,属于可信代理网段。
DOCKER_GATEWAY = "172.17.0.1"
LOOPBACK = "127.0.0.1"
PUBLIC_CLIENT = "203.0.113.7"
def test_normalize_ip():
assert normalize_ip("203.0.113.7") == "203.0.113.7"
assert normalize_ip(" 203.0.113.7 ") == "203.0.113.7"
assert normalize_ip("::ffff:203.0.113.7") == "203.0.113.7"
assert normalize_ip("2001:db8::1") == "2001:db8::1"
assert normalize_ip("2001:0db8:0000:0000:0000:0000:0000:0001") == "2001:db8::1"
# 带端口 / 方括号
assert normalize_ip("203.0.113.7:5678") == "203.0.113.7"
assert normalize_ip("[2001:db8::1]:443") == "2001:db8::1"
# 空值、非法、非字符串
assert normalize_ip("") is None
assert normalize_ip(" ") is None
assert normalize_ip("not-an-ip") is None
assert normalize_ip("203.0.113.7:abc") is None
assert normalize_ip("[2001:db8::1") is None
assert normalize_ip(None) is None
assert normalize_ip(123) is None
assert normalize_ip("999.1.1.1") is None
def test_is_trusted_proxy():
assert is_trusted_proxy(DOCKER_GATEWAY, TRUSTED)
assert is_trusted_proxy(LOOPBACK, TRUSTED)
assert is_trusted_proxy("10.10.8.5", TRUSTED)
assert is_trusted_proxy("::1", TRUSTED)
assert not is_trusted_proxy(PUBLIC_CLIENT, TRUSTED)
assert not is_trusted_proxy("2001:db8::1", TRUSTED)
assert not is_trusted_proxy(None, TRUSTED)
assert not is_trusted_proxy("garbage", TRUSTED)
def test_extract_direct_public_eip():
"""场景一:域名直接解析公网 EIP,无 CDN 头,取对端。"""
ip = extract_client_ip(
{"User-Agent": "Mozilla/5.0"},
"198.51.100.9", # 直连客户端被 nginx 记为 $remote_addr,但 app 看到的是可信对端
trusted_proxies=TRUSTED,
header_order=("X-Real-IP", "X-Forwarded-For"),
)
# 对端是可信 nginx,X-Real-IP 缺失 → 回退对端
assert ip == "198.51.100.9"
def test_extract_direct_uses_x_real_ip_from_nginx():
"""nginx 设置 X-Real-IP,app 采信之。"""
ip = extract_client_ip(
{"X-Real-IP": PUBLIC_CLIENT},
DOCKER_GATEWAY,
trusted_proxies=TRUSTED,
)
assert ip == PUBLIC_CLIENT
def test_extract_cloudflare():
"""场景二:Cloudflare 中转,nginx realip 归一化到 X-Real-IP。"""
ip = extract_client_ip(
{
"X-Real-IP": PUBLIC_CLIENT,
"CF-Connecting-IP": PUBLIC_CLIENT,
"X-Forwarded-For": f"{PUBLIC_CLIENT}, 172.68.1.1",
},
DOCKER_GATEWAY,
trusted_proxies=TRUSTED,
)
assert ip == PUBLIC_CLIENT
def test_extract_cloudflare_via_cf_header_only():
ip = extract_client_ip(
{"CF-Connecting-IP": PUBLIC_CLIENT},
DOCKER_GATEWAY,
trusted_proxies=TRUSTED,
header_order=("CF-Connecting-IP", "X-Real-IP", "X-Forwarded-For"),
)
assert ip == PUBLIC_CLIENT
def test_extract_edgeone():
"""场景三:EdgeOne 中转,X-Real-IP 由 nginx 归一化给出。"""
ip = extract_client_ip(
{
"X-Real-IP": "198.51.100.23",
"EO-Connecting-IP": "198.51.100.23",
"X-Forwarded-For": "198.51.100.23, 43.152.0.9, 43.152.0.10",
},
DOCKER_GATEWAY,
trusted_proxies=TRUSTED,
)
assert ip == "198.51.100.23"
def test_forwarded_for_takes_rightmost():
"""X-Forwarded-For 取最右侧(nginx 追加的真实对端),忽略左侧伪造值。"""
ip = extract_client_ip(
{"X-Forwarded-For": "1.2.3.4, 5.6.7.8, 203.0.113.7"},
DOCKER_GATEWAY,
trusted_proxies=TRUSTED,
header_order=("X-Forwarded-For",),
)
assert ip == "203.0.113.7"
def test_forwarded_for_skips_invalid_rightmost():
ip = extract_client_ip(
{"X-Forwarded-For": "203.0.113.7, not-an-ip, "},
DOCKER_GATEWAY,
trusted_proxies=TRUSTED,
header_order=("X-Forwarded-For",),
)
assert ip == "203.0.113.7"
def test_untrusted_peer_ignores_spoofed_headers():
"""不可信对端(直连容器)伪造 X-Real-IP/XFF 必须被忽略。"""
ip = extract_client_ip(
{
"X-Real-IP": "1.1.1.1",
"X-Forwarded-For": "2.2.2.2, 3.3.3.3",
"CF-Connecting-IP": "4.4.4.4",
},
PUBLIC_CLIENT,
trusted_proxies=TRUSTED,
header_order=("CF-Connecting-IP", "X-Real-IP", "X-Forwarded-For"),
)
assert ip == PUBLIC_CLIENT
def test_extract_fallbacks_and_edges():
# 可信对端但无任何头 → 对端
assert extract_client_ip({}, LOOPBACK, trusted_proxies=TRUSTED) == LOOPBACK
# 头部为空值 → 回退
assert extract_client_ip(
{"X-Real-IP": " "}, DOCKER_GATEWAY, trusted_proxies=TRUSTED
) == DOCKER_GATEWAY
# 头部非法 → 尝试下一个头
assert extract_client_ip(
{"X-Real-IP": "bad", "X-Forwarded-For": PUBLIC_CLIENT},
DOCKER_GATEWAY,
trusted_proxies=TRUSTED,
) == PUBLIC_CLIENT
# 对端不可解析 → 空串
assert extract_client_ip(
{"X-Real-IP": PUBLIC_CLIENT}, "garbage", trusted_proxies=TRUSTED
) == ""
# 对端缺失 → 空串
assert extract_client_ip({"X-Real-IP": PUBLIC_CLIENT}, None, trusted_proxies=TRUSTED) == ""
# 大小写不敏感的请求头查找
assert extract_client_ip(
{"x-real-ip": PUBLIC_CLIENT}, DOCKER_GATEWAY, trusted_proxies=TRUSTED,
) == PUBLIC_CLIENT
# 多值请求头(http.server 可能给 list)
assert extract_client_ip(
{"X-Real-IP": [PUBLIC_CLIENT]}, DOCKER_GATEWAY, trusted_proxies=TRUSTED,
) == PUBLIC_CLIENT
# 空的 header_order → 回退对端
assert extract_client_ip(
{"X-Real-IP": PUBLIC_CLIENT}, DOCKER_GATEWAY,
trusted_proxies=TRUSTED, header_order=(),
) == DOCKER_GATEWAY
def test_extract_ipv6_client():
ip = extract_client_ip(
{"X-Real-IP": "2001:db8::beef"},
LOOPBACK,
trusted_proxies=TRUSTED,
)
assert ip == "2001:db8::beef"
def test_parse_networks_tolerates_junk():
networks = parse_networks(["10.0.0.0/8", "bad", "", "192.168.1.1", 42])
assert len(networks) == 2
assert is_trusted_proxy("10.1.2.3", networks)
assert is_trusted_proxy("192.168.1.1", networks)
assert not is_trusted_proxy("11.0.0.1", networks)
def test_is_bot():
assert is_bot("Mozilla/5.0 (Windows NT 10.0; Win64; x64) Chrome/155.0") is False
assert is_bot("") is False
assert is_bot(None) is False
assert is_bot("Googlebot/2.1 (+http://www.google.com/bot.html)") is True
assert is_bot("Mozilla/5.0 (compatible; bingbot/2.0)") is True
assert is_bot("python-requests/2.31") is True
assert is_bot("curl/8.4.0") is True
assert is_bot("Uptime-Kuma/1.23") is True
assert is_bot("Bytespider") is True
def test_normalize_route():
assert normalize_route("/") == "/home"
assert normalize_route("") == "/other"
assert normalize_route(None) == "/other"
assert normalize_route("/home") == "/home"
assert normalize_route("/videos") == "/videos"
assert normalize_route("/ups") == "/ups"
assert normalize_route("/fans") == "/fans"
assert normalize_route("/about") == "/about"
assert normalize_route("/candidates") == "/candidates"
assert normalize_route("/system") == "/system"
assert normalize_route("/rank") == "/rank"
# 带查询串 / 片段
assert normalize_route("/videos?sort=view&year=2026") == "/videos"
assert normalize_route("/video/114775988111033#top") == "/video/:aid"
# 末尾斜杠
assert normalize_route("/videos/") == "/videos"
# 动态路由折叠
assert normalize_route("/video/123") == "/video/:aid"
assert normalize_route("/up/230020975") == "/up/:mid"
# 非法 / 未知 / 超长
assert normalize_route("videos") == "/other"
assert normalize_route("/video/abc") == "/other"
assert normalize_route("/unknown") == "/other"
assert normalize_route("/" + "x" * 500) == "/other"
assert normalize_route(123) == "/other"
def main():
test_normalize_ip()
test_is_trusted_proxy()
test_extract_direct_public_eip()
test_extract_direct_uses_x_real_ip_from_nginx()
test_extract_cloudflare()
test_extract_cloudflare_via_cf_header_only()
test_extract_edgeone()
test_forwarded_for_takes_rightmost()
test_forwarded_for_skips_invalid_rightmost()
test_untrusted_peer_ignores_spoofed_headers()
test_extract_fallbacks_and_edges()
test_extract_ipv6_client()
test_parse_networks_tolerates_junk()
test_is_bot()
test_normalize_route()
print("OK: 来源 IP 提取与路由/机器人判定测试全部通过")
if __name__ == "__main__":
main()
+70 -1
View File
@@ -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__":
+4 -4
View File
@@ -24,7 +24,7 @@ def test_write_read_transaction():
with tempfile.TemporaryDirectory() as tmp:
db = make_db(tmp)
try:
assert db.schema_version == 6
assert db.schema_version == 7
db.write(lambda c: c.execute("INSERT INTO kv(k, v) VALUES ('a', '1')"))
assert db.read(lambda c: c.execute("SELECT v FROM kv WHERE k='a'").fetchone()[0]) == "1"
@@ -94,8 +94,8 @@ def test_close_and_reinitialize():
db2 = make_db(tmp)
try:
assert db2.schema_version == 6
assert db2.read(lambda c: c.execute("SELECT MAX(version) FROM schema_version").fetchone()[0]) == 6
assert db2.schema_version == 7
assert db2.read(lambda c: c.execute("SELECT MAX(version) FROM schema_version").fetchone()[0]) == 7
finally:
db2.close()
@@ -123,7 +123,7 @@ def test_migration_backup_on_existing_db():
db = Database(path, backup_dir=Path(tmp) / "backup")
try:
applied = db.initialize()
assert applied == [1, 2, 3, 4, 5, 6]
assert applied == [1, 2, 3, 4, 5, 6, 7]
assert len(list((Path(tmp) / "backup").glob("app-pre-v0-*.db"))) == 1
finally:
db.close()
+15 -8
View File
@@ -22,6 +22,7 @@ from reaction_board.migrations import ( # noqa: E402
EXPECTED_TABLES = {
"video", "snapshot", "folder", "folder_resource", "up_blacklist",
"up_stat", "up_profile", "candidate", "seen_aid", "notify", "fav_action", "kv",
"visit",
}
@@ -37,8 +38,8 @@ def test_fresh_and_idempotent():
conn = sqlite3.connect(Path(tmp) / "app.db")
try:
applied = run_migrations(conn)
assert applied == [1, 2, 3, 4, 5, 6]
assert current_version(conn) == 6
assert applied == [1, 2, 3, 4, 5, 6, 7]
assert current_version(conn) == 7
assert EXPECTED_TABLES <= table_names(conn)
assert "schema_version" in table_names(conn)
columns = {
@@ -63,10 +64,16 @@ def test_fresh_and_idempotent():
"mid", "follower_first", "following_first", "first_captured_at",
"follower_latest", "last_refreshed",
} <= profile_columns
visit_columns = {
row[1] for row in conn.execute("PRAGMA table_info(visit)")
}
assert {
"stat_date", "ip", "path", "hits", "first_at", "last_at",
} <= visit_columns
# 再跑一次不应重复应用
assert run_migrations(conn) == []
row = conn.execute("SELECT COUNT(*) FROM schema_version").fetchone()
assert row[0] == 6
assert row[0] == 7
finally:
conn.close()
@@ -81,15 +88,15 @@ def test_backup_before_upgrade():
conn.execute("INSERT INTO kv(k, v) VALUES ('hello', 'world')")
conn.commit()
v7 = Migration(7, "add probe", ("CREATE TABLE probe (id INTEGER PRIMARY KEY)",))
v8 = Migration(8, "add probe", ("CREATE TABLE probe (id INTEGER PRIMARY KEY)",))
applied = run_migrations(
conn,
backup_dir=backup_dir,
now=1_800_000_000,
migrations=MIGRATIONS + (v7,),
migrations=MIGRATIONS + (v8,),
)
assert applied == [7]
backups = list(backup_dir.glob("app-pre-v6-*.db"))
assert applied == [8]
backups = list(backup_dir.glob("app-pre-v7-*.db"))
assert len(backups) == 1
check = sqlite3.connect(backups[0])
try:
@@ -174,7 +181,7 @@ def test_v5_normalizes_unfaved():
)
conn.commit()
applied = run_migrations(conn)
assert applied == [5, 6]
assert applied == [5, 6, 7]
rows = dict(conn.execute("SELECT aid, status FROM video").fetchall())
assert rows == {1: "active", 2: "dropped"}
finally:
+203
View File
@@ -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()
+62 -2
View File
@@ -459,7 +459,7 @@ def test_about_payload():
assert about["snapshot_days"] == 3
assert about["snapshot_first_date"] == DATES[0]
assert about["fans_start_date"] == DATES[0]
assert about["schema_version"] == 6
assert about["schema_version"] == 7
assert about["analytics"] == {
"min_duration_seconds": 600,
"maturity_days": 2,
@@ -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__":
+88 -2
View File
@@ -75,7 +75,7 @@ def test_readonly_routes_and_static():
assert status == 200 and payload["ok"] is True
status, _, payload = running.json("/api/about")
assert status == 200
assert payload["data"]["schema_version"] == 6
assert payload["data"]["schema_version"] == 7
assert payload["data"]["analytics"]["min_duration_seconds"] == 600
status, _, overview = running.json("/api/overview")
@@ -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__":