审计(debug8) - 访问日志:按“访问”记录终端访问了什么,进行中的访问每分钟写入并更新,可筛选、导出 CSV,保留 30 天 - 终端详情新增访问日志、上网时间线 - 审计规则(大类 / 分类 / 应用 / 域名 / 流量 / 新设备,可限星期、时段、终端),5 个模板 - 通知推送:企业微信、钉钉、飞书、Bark、Telegram、Webhook;发送成功才标记已推送 - 日报、周报、操作日志 debug9 - 只看主路由表的默认路由,排除 DHCP 服务接口:策略路由表指向旁路由时 LAN 网桥被当成 WAN,上下行叠加成几乎一样 - 多条 WAN 合并成一张卡,全网状态始终一排 - 应用统计改成维盟式 DPI 协议三级下钻:大类 → 小类 → 应用,各列“数值 | 本级占比”,可排序、可停止刷新 - sing-box:按规则集 / 策略组名识别应用;实时连接显示策略组和出口节点;新增“代理分流” - 访问目标显示国家 / 地区及分布 - 管理员与普通用户:普通用户只读,用户管理,删除后令牌立即失效 - 手机上饼图图例移到下方,访问日志表不再横向滑动 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
49 lines
1.5 KiB
Python
49 lines
1.5 KiB
Python
"""时间分桶与统计区间(今日 / 本周 / 本月)。周从周一开始,按配置时区计算。"""
|
|
import time
|
|
from datetime import datetime, timedelta
|
|
|
|
from app.config import get_settings
|
|
|
|
|
|
def tz():
|
|
return get_settings().zone
|
|
|
|
|
|
def minute_bucket(ts: float) -> int:
|
|
return int(ts) // 60 * 60
|
|
|
|
|
|
def hour_bucket(ts: float) -> int:
|
|
# 小时按 UTC 对齐;整小时偏移的时区(如 Asia/Shanghai)与本地整点一致
|
|
return int(ts) // 3600 * 3600
|
|
|
|
|
|
def day_bucket(ts: float) -> int:
|
|
z = get_settings().zone
|
|
d = datetime.fromtimestamp(ts, z)
|
|
return int(d.replace(hour=0, minute=0, second=0, microsecond=0).timestamp())
|
|
|
|
|
|
def range_bounds(rng: str, now: float | None = None) -> tuple[int, int]:
|
|
"""返回 [start, end) 的 epoch 秒。rng: today | yesterday | week | month | last7 | last30。"""
|
|
now = now or time.time()
|
|
z = get_settings().zone
|
|
d = datetime.fromtimestamp(now, z).replace(hour=0, minute=0, second=0, microsecond=0)
|
|
end = int(now) + 1
|
|
if rng == "today":
|
|
start = d
|
|
elif rng == "yesterday":
|
|
start = d - timedelta(days=1)
|
|
end = int(d.timestamp())
|
|
elif rng == "week":
|
|
start = d - timedelta(days=d.weekday())
|
|
elif rng == "month":
|
|
start = d.replace(day=1)
|
|
elif rng == "last7":
|
|
start = d - timedelta(days=6)
|
|
elif rng == "last30":
|
|
start = d - timedelta(days=29)
|
|
else:
|
|
raise ValueError(f"unknown range {rng}")
|
|
return int(start.timestamp()), end
|