Su YijunandClaude Opus 5.5
28135807bd
- 应用分类参照 Panabit 改为 应用大类 → 小类 → 应用：10 个大类、41 个小类、277 个应用； CNAME 指向 CDN 时归到 CDN 厂商；ICMP / GRE / ESP 按协议识别；已发布的应用和分类 ID 全部保留 - 图表重做：流量构成双层环图 + 排行表、按应用大类堆叠的趋势（可切换上下行）、 带名次和占比的 Top 排行、桑基图中间列改为应用大类；配色按大类固定并通过色盲校验 - 终端命名：按 MAC 或按 IP，新增命名管理，可提前给未上线设备命名，旧备注名自动迁移 - DNS 嗅探卡片说明作用并显示应答来源；补充内网自建 DNS（AdGuard/Pi-hole）的说明 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
76 lines
2.3 KiB
Python
76 lines
2.3 KiB
Python
"""终端命名：用户按 MAC 或 IP 给终端起名。
|
|
|
|
- 按 MAC：设备换了 IP 名字也跟着走，适合大多数设备。
|
|
- 按 IP：适合固定 IP、但 MAC 会变的设备（手机的随机 MAC），或拿不到 MAC 的终端。
|
|
"""
|
|
import ipaddress
|
|
import re
|
|
import time
|
|
|
|
from sqlalchemy import select
|
|
|
|
from app.models import Host, HostName
|
|
|
|
_MAC_RE = re.compile(r"^[0-9A-F]{12}$")
|
|
|
|
|
|
def norm_mac(s: str) -> str | None:
|
|
"""AA-BB-CC-DD-EE-FF / aabb.ccdd.eeff / aabbccddeeff → AA:BB:CC:DD:EE:FF"""
|
|
h = re.sub(r"[:\-.\s]", "", s or "").upper()
|
|
if not _MAC_RE.match(h):
|
|
return None
|
|
return ":".join(h[i:i + 2] for i in range(0, 12, 2))
|
|
|
|
|
|
def norm_ip(s: str) -> str | None:
|
|
try:
|
|
return str(ipaddress.ip_address((s or "").strip()))
|
|
except ValueError:
|
|
return None
|
|
|
|
|
|
def normalize(kind: str, value: str) -> str | None:
|
|
return norm_mac(value) if kind == "mac" else norm_ip(value) if kind == "ip" else None
|
|
|
|
|
|
def load(db) -> dict[tuple[str, str], str]:
|
|
return {(n.kind, n.value): n.name for n in db.scalars(select(HostName))}
|
|
|
|
|
|
def resolve(names: dict, mac: str | None, ip: str | None) -> tuple[str | None, str | None]:
|
|
"""返回 (名字, 命中方式 mac|ip)。"""
|
|
if mac and (n := names.get(("mac", mac))):
|
|
return n, "mac"
|
|
if ip and (n := names.get(("ip", ip))):
|
|
return n, "ip"
|
|
return None, None
|
|
|
|
|
|
def put(db, kind: str, value: str, name: str | None):
|
|
"""name 为空表示删除这条命名。value 需已规范化。"""
|
|
row = db.get(HostName, (kind, value))
|
|
name = (name or "").strip()
|
|
if not name:
|
|
if row is not None:
|
|
db.delete(row)
|
|
return
|
|
if row is None:
|
|
db.add(HostName(kind=kind, value=value, name=name, updated=int(time.time())))
|
|
else:
|
|
row.name, row.updated = name, int(time.time())
|
|
|
|
|
|
def migrate_aliases(db):
|
|
"""把旧版 hosts.alias 搬到 host_names（有 MAC 按 MAC，否则按 IP），只做一次。"""
|
|
hosts = db.scalars(select(Host).where(Host.alias.is_not(None))).all()
|
|
if not hosts:
|
|
return
|
|
names = load(db)
|
|
for h in hosts:
|
|
kind, value = ("mac", h.mac) if h.mac else ("ip", h.ip)
|
|
if value and (kind, value) not in names and h.alias.strip():
|
|
put(db, kind, value, h.alias)
|
|
names[(kind, value)] = h.alias
|
|
h.alias = None
|
|
db.commit()
|