diff --git a/CHANGELOG.md b/CHANGELOG.md index e4e9622..ae5ad8c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,14 @@ # Changelog +## [0.6.6] - 2026-10-25 + +识别库智能化：上游更新自动建规则库、自动重算历史；修复 AdGuard Home 读到 0 条（debug14） + +- **修复 AdGuard Home 一条应答都读不到**（0.6.5）：新版 AdGuard Home 的查询日志 API 返回 `{"data": [...]}`，0.6.5 按旧版的 `{"entries": [...]}` 解析，结果"连接成功"却永远是 0 条。现在两种格式（以及直接数组）都认。 +- **上游规则库更新后自动建新规则库**（debug14）：blackmatrix7 / Hagezi / UT1 的上游出现新的目录或列表时，自动登记成新应用——类别按目录名猜测（视频、游戏、购物、银行、AI 等 20 多组关键词，猜不出的归「常用网站」；分类列表猜不出则跳过），登记结果存在 `data/lists/rules_auto.yaml`，重启不丢；手工映射（bm7_map.yaml）优先级更高。汇总 / 区域 / 安全类目录（China、Global、Advertising 等）不参与自动登记。不再需要等新版本手工补映射。 +- **「重算历史分类」会反查 DNS 缓存**：纯 IP 的历史行先查 DNS 缓存（嗅探 / MosDNS / AdGuard Home 积累的 IP→域名映射）拿到域名再分类，AdGuard 接入后积累的映射能回溯到历史流量上；DNS 缓存过期后，已经归到具体应用的行保持不动，不会倒退回「国内网络 / 未识别」。 +- **识别库更新后自动重算历史分类**：识别规则保持「自动更新」开启时，识别库有实际变化（规则更新、DNS 映射积累）就自动执行一次历史重算（按内容指纹判断，没变化时只有几条计数查询的开销）；点「立即检查更新」后同样触发。识别规则页的手动按钮保留。 + ## [0.6.5] - 2026-10-25 外部数据源支持自有 DNS 服务器（AdGuard Home）、ASN 识别中国自营内容商、历史分类重算；DPI 协议 17 色去重（debug13） diff --git a/backend/app/__init__.py b/backend/app/__init__.py index 7bbb2ef..4c513f3 100644 --- a/backend/app/__init__.py +++ b/backend/app/__init__.py @@ -1 +1 @@ -__version__ = "0.6.5" +__version__ = "0.6.6" diff --git a/backend/app/classify/lists.py b/backend/app/classify/lists.py index 7831746..4c81e4c 100644 --- a/backend/app/classify/lists.py +++ b/backend/app/classify/lists.py @@ -5,6 +5,9 @@ - Hagezi DNS blocklists（GPL-3.0）：成人、博彩、VPN/代理、社交、广告、各厂商设备遥测等专项分类列表。每天更新。 - UT1（Université Toulouse Capitole，CC BY-SA 4.0）：游戏、购物、新闻、求职、银行等分类列表。每天更新。 +上游出现新的目录 / 列表时自动登记成新应用（debug14）：类别按目录名猜测，登记结果存在 +data/lists/rules_auto.yaml，下次更新自动带上；手工映射仍在 bm7_map.yaml，优先级更高。 + 识别顺序和权重（详见 app/classify/__init__.py）： - blackmatrix7 是应用级：和内置库 / v2fly 一起按域名匹配，匹配到的后缀更长者赢，一样长时内置库和 v2fly 优先。 - Hagezi / UT1 是分类级：只知道是哪类网站，排在所有应用级信号（域名、IP 段、端口、ASN）之后、国内域名列表之前； @@ -16,6 +19,7 @@ import hashlib import ipaddress import os +import re import threading import time import urllib.parse @@ -32,6 +36,47 @@ MAP_FILE = os.path.join(os.path.dirname(__file__), "bm7_map.yaml") BM7_URL = "https://raw.githubusercontent.com/blackmatrix7/ios_rule_script/master/rule/Clash/{0}/{0}.list" HAGEZI_URL = "https://raw.githubusercontent.com/hagezi/dns-blocklists/main/wildcard/{0}-onlydomains.txt" UT1_URL = "https://raw.githubusercontent.com/olbat/ut1-blacklists/master/blacklists/{0}/domains" +# 上游目录清单（自动发现用，每次更新最多查一次） +BM7_LIST_API = "https://api.github.com/repos/blackmatrix7/ios_rule_script/contents/rule/Clash" +HAGEZI_LIST_API = "https://api.github.com/repos/hagezi/dns-blocklists/contents/wildcard" +UT1_LIST_API = "https://api.github.com/repos/olbat/ut1-blacklists/contents/blacklists" +LIST_API = {"bm7": BM7_LIST_API, "hagezi": HAGEZI_LIST_API, "ut1": UT1_LIST_API} +# bm7 的汇总 / 区域 / 安全类目录：不是单个应用，不自动登记 +BM7_SKIP = { + "Global", "Proxy", "Direct", "Lan", "Reject", "AdBlock", "Advertising", "AdvertisingLite", "AntiAd", + "China", "ChinaMedia", "ChinaTest", "ChinaMax", "ChinaMaxNoIP", "ChinaMaxNoMedia", "ChinaNoMedia", + "ChinaIPs", "ChinaIPsBGP", "ChinaDNS", "DomesticIPs", "PublicDNS", "LocalDNS", "PrivateDNS", "DNS", + "GoogleDNS", "ProxyGFW", "GFW", "GreatFire", "AntiPolicy", "Hijacking", "PotentialMalware", "Safety", + "Spam", "TrackersList", "ZoneTelekom", "Classic", "3Type", "Cloud", "GlobalMedia", "IPTVOther", + "Speedtest", "MicrosoftEdge", "AllAI", "Download", "Crypto", "Cryptocurrency", "Developer", "Anime", + "AsianMedia", "Game", "Mail", "GoogleSearch", "Antutu", "Region", "TldCheap", "TldFake", +} +# 目录名 / 列表名 → 类别的猜测规则（debug14：上游新增内容自动归类；猜不出 bm7 归「常用网站」、分类列表跳过） +GUESS = [ + (r"adult|porn|nsfw|xxx", "adult"), (r"gambl|bet|casino|lotter", "gambling"), + (r"spotify|music|radio|podcast|tidal|deezer|netease_music|kugou|kuwo", "music"), + (r"(^|[-_])(live|stream)([-_]|$)|twitch|douyu|huya|yy", "live"), + (r"video|tv|movie|film|cinema|anime|manga|mgtv|youku|iqiyi|bilibili|netflix|hbo|disney|emby|iptv|cctv", "video"), + (r"game|steam|epic|blizzard|riot|xbox|playstation|nintendo|4399|9game|taptap|mihoyo", "game"), + (r"shop|taobao|tmall|jd|jingdong|pinduoduo|suning|vipshop|aliexpress|amazon|ebay|shopee|gome", "shopping"), + (r"bank|icbc|ccb|cmb|abc|psbc|boc|bocom|citic|spdb|cib|ceb|cgb|spdb|unionpay|alipay|tenpay|paypal|pay", "bank"), + (r"mail|smtp|imap|webmail", "mail"), + (r"news|media|press|bbc|cnn|reuters|nytimes", "news"), + (r"social|weibo|twitter|facebook|instagram|tiktok|douyin|reddit|forum|snapchat|linkedin|pin", "social"), + (r"wechat|weixin|whatsapp|telegram|line|slack|dingtalk|feishu|skype|zoom|meeting", "im"), + (r"openai|claude|gemini|copilot|chatgpt|anthropic|perplexity|midjourney|hf|\bai\b", "ai"), + (r"cloud|drive|pan|onedrive|dropbox|aliyundrive|quark|s3|oss", "cloud"), + (r"github|gitlab|gitcode|gitee|docker|npm|python|jetbrains|dev", "dev"), + (r"proxy|vpn|tor|gfw|warp|outline|shadowsocks|v2ray|trojan|clash", "proxy"), + (r"travel|trip|hotel|flight|airline|booking|ctrip|qunar|fliggy|12306|ticket", "travel"), + (r"edu|mooc|university|course|learn|school|xueersi|yuanfudao|zuoyebang", "education"), + (r"health|hospital|medic|pharma|doctor|clinic", "life"), + (r"food|takeaway|waimai|meituan|eleme|restaurant|recipe", "life"), + (r"cdn|static|cache|cloudfront|akamai|cloudflare|fastly|edge", "cdn"), + (r"tracker|warez|pt|download|thunder|xunlei", "download"), + (r"mine|mining|pool|nicehash", "mining"), + (r"remote|ssh|rdp|vnc|teamviewer|todesk|rustdesk|anydesk", "remote"), +] # 分类列表：(列表名, 归到的应用)。顺序就是同一域名在多个列表里时的优先级 HAGEZI = [(f"native.{v}", "gen_telemetry") for v in @@ -138,6 +183,15 @@ def load_map() -> dict: return {"owner": owner, "apps": m.get("apps") or {}} +def guess_category(name: str) -> tuple[str, bool]: + """按目录名 / 列表名猜类别（debug14 自动登记用）。返回 (类别, 是否有把握)；猜不出归「常用网站」""" + low = name.lower() + for pat, cat in GUESS: + if re.search(pat, low): + return cat, True + return "site", False + + class Lists: def __init__(self): self.lock = threading.Lock() @@ -148,22 +202,53 @@ class Lists: self.cat = _HashTable() self.state: dict[str, dict] = {} # 来源 id -> {count, apps, updated, error} self._map = None + self._auto: dict = {} # 自动登记的应用（debug14），存 data/lists/rules_auto.yaml + self._load_auto() @property def map(self) -> dict: if self._map is None: - self._map = load_map() + base = load_map() + for aid, a in self._auto.items(): + base["apps"].setdefault(aid, {"category": a["category"], "name": a["name"], "name_en": a["name_en"]}) + if a.get("source") == "bm7" and a.get("dir"): + base["owner"].setdefault(a["dir"], aid) + self._map = base return self._map def _dir(self, sid: str) -> str: return os.path.join(os.environ.get("DATA_DIR", "data"), "lists", sid) + def _auto_file(self) -> str: + return os.path.join(self._dir("bm7"), "rules_auto.yaml") + + def _load_auto(self): + try: + with open(self._auto_file(), encoding="utf-8") as f: + data = yaml.safe_load(f) or {} + self._auto = data.get("apps") or {} + except (OSError, yaml.YAMLError): + self._auto = {} + + def _save_auto(self): + path = self._auto_file() + os.makedirs(os.path.dirname(path), exist_ok=True) + with open(path, "w", encoding="utf-8") as f: + f.write("# 自动发现并登记的第三方规则（debug14）：上游出现新目录 / 新列表时自动生成。\n" + "# 手工映射在镜像内的 bm7_map.yaml，优先级更高；删除本文件后，下次更新按当时的上游重新生成。\n") + yaml.safe_dump({"apps": self._auto}, f, allow_unicode=True, sort_keys=True) + + def _spec(self, sid: str) -> list[tuple[str, str]]: + """分类列表的 (列表名, 应用)：手工映射在前，自动登记的追加在后面""" + base = HAGEZI if sid == "hagezi" else UT1 + extra = [(a["list"], aid) for aid, a in self._auto.items() if a.get("source") == sid] + return list(base) + extra + def _files(self, sid: str) -> list[tuple[str, str]]: """(本地文件名, 下载地址)""" if sid == "bm7": return [(f"{n}.list", BM7_URL.format(urllib.parse.quote(n))) for n in sorted(self.map["owner"])] - spec, url = (HAGEZI, HAGEZI_URL) if sid == "hagezi" else (UT1, UT1_URL) - return [(f"{n}.txt", url.format(n)) for n, _app in spec] + return [(f"{n}.txt", (HAGEZI_URL if sid == "hagezi" else UT1_URL).format(n)) for n, _app in self._spec(sid)] # ---------- 加载 ---------- @staticmethod @@ -201,9 +286,8 @@ class Lists: return suffix, full, ips, n, len(apps) def _load_cat(self, sid: str) -> list[tuple[str, list[str]]]: - spec = HAGEZI if sid == "hagezi" else UT1 out = [] - for name, app in spec: + for name, app in self._spec(sid): doms = [] for line in self._read(os.path.join(self._dir(sid), f"{name}.txt")): s = line.strip().lower().lstrip("*.").strip(".") @@ -250,6 +334,11 @@ class Lists: def update(self, force: bool = False) -> bool: """各来源满了更新周期（bm7 一周、其余一天）就重新下载；单个文件失败保留旧文件""" changed = False + if _enabled("bm7") and (force or time.time() - self._stamp("bm7") >= MAX_AGE["bm7"]): + try: + self.discover() # debug14：上游新增的目录 / 列表先登记，同一轮就会下载 + except Exception: # noqa: BLE001 发现不了（没外网 / API 限流）不影响本轮更新 + pass for sid in ("bm7", "hagezi", "ut1"): if not _enabled(sid): continue @@ -263,6 +352,59 @@ class Lists: self.load() return changed + def discover(self) -> int: + """上游仓库出现新的规则目录 / 分类列表时自动登记成应用（debug14）。返回新登记个数。""" + added = self._discover_bm7() + self._discover_spec("hagezi") + self._discover_spec("ut1") + if added: + self._save_auto() + self._map = None # 让 map 重新合并，新应用 / 新目录立即生效 + return added + + @staticmethod + def _list_names(url: str, kind: str) -> set[str]: + """GitHub 目录清单：kind=dir 取目录名；kind=file 取去掉 onlydomains.txt 后缀的文件名""" + r = httpx.get(url, timeout=30.0, headers={"User-Agent": "ros-dpi", "Accept": "application/vnd.github+json"}) + r.raise_for_status() + out = set() + for x in r.json(): + name = x.get("name", "") + if x.get("type") == kind and not name.startswith("."): + if kind == "file" and name.endswith("-onlydomains.txt"): + name = name[:-len("-onlydomains.txt")] + out.add(name) + return out + + def _register(self, source: str, names: set[str], strict: bool) -> int: + added = 0 + known = set(self.map["owner"]) + for n in sorted(names): + if n in known or n in BM7_SKIP: + continue + cat, confident = guess_category(n) + if strict and not confident: + continue # 分类列表猜不出类别就跳过：宁缺毋滥，等手工映射 + aid = "gen_" + re.sub(r"[^a-z0-9]+", "_", n.lower()).strip("_") if source != "bm7" \ + else "bm7_" + re.sub(r"[^a-z0-9]+", "_", n.lower()).strip("_") + if aid in self._auto or aid in self.map["apps"] or aid in self.map["owner"].values(): + continue + entry = {"category": cat, "name": n, "name_en": n, "source": source} + if source == "bm7": + entry["dir"] = n + else: + entry["list"] = n + self._auto[aid] = entry + known.add(n) + added += 1 + return added + + def _discover_bm7(self) -> int: + return self._register("bm7", self._list_names(LIST_API["bm7"], "dir"), strict=False) + + def _discover_spec(self, sid: str) -> int: + if sid == "hagezi": + return self._register("hagezi", self._list_names(LIST_API[sid], "file"), strict=True) + return self._register("ut1", self._list_names(LIST_API[sid], "dir"), strict=True) + def _fetch_all(self, sid: str) -> tuple[int, int]: d = self._dir(sid) os.makedirs(d, exist_ok=True) diff --git a/backend/app/integrations.py b/backend/app/integrations.py index d16ef01..c6288bd 100644 --- a/backend/app/integrations.py +++ b/backend/app/integrations.py @@ -259,9 +259,14 @@ class AdguardPoller: self._connect() r = self._client.get("/control/querylog", params={"limit": self.PAGE, "response_status": "all"}) r.raise_for_status() - data = r.json() - # 新版本返回 {"entries": [...]}，旧版本直接返回数组 - return (data.get("entries") if isinstance(data, dict) else data) or [] + return self._entries(r.json()) + + @staticmethod + def _entries(data) -> list: + # 新版本返回 {"data": [...]}，旧版本 {"entries": [...]} 或直接数组 + if isinstance(data, dict): + return data.get("data") or data.get("entries") or [] + return data or [] def poll(self): if not self.enabled: diff --git a/backend/app/main.py b/backend/app/main.py index a8f3f6b..6cd298e 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -46,7 +46,8 @@ def _on_db_ready(): """数据库连接成功后（启动时或安装向导保存后）执行。各步互不依赖：一步出错只记日志，不挡住后面的 （0.6.0 / 0.6.1 的主域名回填出错时，后面的外部数据源配置、自定义规则都没加载，外部数据源显示“未启用”）""" ensure_admin() # 建管理员失败要报出来（安装向导靠这个提示） - for name, fn in (("load_state", engine.load_state), ("apply_settings", _apply_saved)): + for name, fn in (("load_state", engine.load_state), ("apply_settings", _apply_saved), + ("auto_reclass", _auto_reclassify)): try: fn() except Exception as e: # noqa: BLE001 @@ -70,6 +71,19 @@ def _auto_update_rules(): updater.check(reload=engine.reload_classifier) +def _auto_reclassify(): + """识别库更新 / DNS 映射积累后自动重算历史分类（debug14）：指纹没变就不动，变了才重算""" + if not _auto_update_on(): + return + from app.reclass import auto_reclassify + with database.session() as db: + res = auto_reclassify(engine, db) + db.commit() + if res: + print(f"[reclass] 识别规则有更新，已自动重算历史分类：搬移 {res['bytes_moved'] / 1e6:.1f} MB" + f"（{res['seconds']}s）", flush=True) + + def _auto_update_db(fn): """ASN / 补充库的定期下载跟随同一个开关；关掉后只用已下载的数据，手动「立即检查更新」仍然可用""" def run(): @@ -104,6 +118,8 @@ async def lifespan(app: FastAPI): sched.add_job(_safe(integrations.singbox.poll, "singbox"), "interval", seconds=5, max_instances=1, coalesce=True) sched.add_job(_safe(integrations.mosdns.poll, "mosdns"), "interval", seconds=5, max_instances=1, coalesce=True) sched.add_job(_safe(integrations.adguard.poll, "adguard"), "interval", seconds=5, max_instances=1, coalesce=True) + # 识别库更新 / DNS 映射积累后的自动重算：跟在规则更新任务后面跑，指纹没变时开销只有几条计数查询 + sched.add_job(_safe(_auto_reclassify, "auto-reclass"), "interval", hours=6, max_instances=1, coalesce=True) sched.add_job(_safe(_auto_update_rules, "rules-update"), "interval", hours=6, max_instances=1, coalesce=True) # 审计报表：日报每天 08:00 推送昨天的，周报每周一 08:30 推送近 7 天的（在审计 → 报表里开启） sched.add_job(_safe(lambda: audit.push_report("daily"), "report-daily"), "cron", hour=8, minute=0) diff --git a/backend/app/reclass.py b/backend/app/reclass.py index 6319c9d..0df5fde 100644 --- a/backend/app/reclass.py +++ b/backend/app/reclass.py @@ -1,17 +1,27 @@ -"""按当前规则重算历史分类。 +"""历史分类重算：规则 / 识别库升级后，把历史流量搬到新归属。 以 host_domain_day（每天每终端每域名，31 天）为源：对每一行用现在的规则重新分类，分类结果变化的行， 把 traffic_day 里对应 (天, 终端, 旧应用) 的上下行与连接数按字节占比搬到新应用，并把 host_domain_day 的 app 标记同步改掉（保证重复执行幂等）。host_domain_day 只覆盖有域名、或进了当分钟 Top30 IP 的流量， 其余流量保持原分类；traffic_minute / traffic_hour 不动，随保留期自然滚动。 + +纯 IP 的历史行会用 dns_cache 反查域名再分类（嗅探 / MosDNS / AdGuard Home 积累的映射）——DNS 缓存 +会过期，所以已经归到具体应用、又拿不到新域名证据的行保持不动，避免缓存过期让历史分类倒退。 + +识别库更新 / DNS 映射积累后自动重算：fingerprint() 覆盖各层库的规模和 DNS 缓存行数，指纹变了才执行 +（auto_reclassify），识别规则页的手动「重算历史分类」直接调 reclassify()。 """ import ipaddress import time from collections import defaultdict -from sqlalchemy import select +from sqlalchemy import func, select -from app.models import HostDomainDay, TrafficDay +from app.models import DnsCache, HostDomainDay, Setting, TrafficDay + +FP_KEY = "reclass_fp" +# 纯 IP 行没有域名证据时，只有原本就归在这些笼统应用里的才重算（ASN / IP 段升级的直接受益者） +GENERIC_OLD = {"cn_net", "cn_web", "other", "unknown", "https", "quic", "http", "web", "site"} def _is_ip(s: str) -> bool: @@ -22,9 +32,17 @@ def _is_ip(s: str) -> bool: return False +def _dns_domain_map(db) -> dict[str, str]: + """未过期的 IP→域名映射（嗅探 / MosDNS / AdGuard Home 积累的），给纯 IP 的历史行反查域名""" + now = int(time.time()) + rows = db.execute(select(DnsCache.ip, DnsCache.name).where(DnsCache.expires > now)).all() + return dict(rows) + + def reclassify(engine, db) -> dict: t0 = time.time() classifier = engine.classifier + cache = _dns_domain_map(db) moves: dict = defaultdict(lambda: defaultdict(int)) # (ts, host, 旧应用) -> {新应用: 字节} marks: list[dict] = [] rows = db.execute( @@ -34,7 +52,13 @@ def reclassify(engine, db) -> dict: if not b: continue if _is_ip(dom): - new_app = classifier.classify(dom, 6, 0, None, None) + qname = cache.get(dom) + if qname: + new_app = classifier.classify(dom, 6, 0, qname, None) + elif old_app in GENERIC_OLD: + new_app = classifier.classify(dom, 6, 0, None, None) + else: + continue # 已归到具体应用、又没有新的域名证据：保持不动（DNS 缓存过期不应让历史倒退） else: new_app = classifier.classify("0.0.0.0", 6, 443, dom, None) new_app = new_app or "other" @@ -90,3 +114,30 @@ def reclassify(engine, db) -> dict: dst.flows += flows return {"scanned": len(rows), "rows_moved": len(dsts), "bytes_moved": sum(v[0] + v[1] for v in dsts.values()), "flows_moved": sum(v[2] for v in dsts.values()), "seconds": round(time.time() - t0, 1)} + + +def fingerprint(engine, db) -> str: + """识别规则的有效指纹：任何一层库变了（v2fly 提交、第三方库规模、ASN 行数）或 DNS 映射积累变多都不一样""" + from app.asn import asndb + from app.classify.lists import lists as extra_lists + from app.classify.updater import updater + now = int(time.time()) + dns_n = db.scalar(select(func.count()).select_from(DnsCache).where(DnsCache.expires > now)) or 0 + libs = "|".join(f"{r['id']}:{r.get('domains', r.get('rules', 0))}" for r in extra_lists.rows()) + return "|".join([f"dns{dns_n}", f"asn{asndb.rows}", f"v2{updater.state.get('latest') or ''}", + f"apps{len(engine.classifier.apps)}", libs]) + + +def auto_reclassify(engine, db) -> dict | None: + """识别库更新 / DNS 映射积累后自动重算（debug14）：指纹没变就什么都不做，变了才重算并记住新指纹""" + fp = fingerprint(engine, db) + row = db.get(Setting, FP_KEY) + if row is not None and row.value == fp: + return None + res = reclassify(engine, db) + row = db.get(Setting, FP_KEY) + if row is None: + db.add(Setting(key=FP_KEY, value=fp)) + else: + row.value = fp + return res diff --git a/backend/app/routers/admin.py b/backend/app/routers/admin.py index c3a551c..729ced8 100644 --- a/backend/app/routers/admin.py +++ b/backend/app/routers/admin.py @@ -297,6 +297,15 @@ def library_update(request: Request, user: str = Depends(current_user), db: Sess from app.classify.lists import lists lists.update(force=True) oplog.log(db, request, user, "library_update", {"result": updater.state.get("result")}) + try: + from app.reclass import auto_reclassify + res = auto_reclassify(engine, db) + if res: + db.commit() + oplog.log(db, request, user, "reclassify", res) + except Exception as e: # noqa: BLE001 重算失败不影响更新结果 + db.rollback() + print(f"[reclass] 手动更新后自动重算失败：{type(e).__name__}: {e}", flush=True) return _library(db) diff --git a/backend/tests/test_debug13.py b/backend/tests/test_debug13.py index fdd9448..9ae8450 100644 --- a/backend/tests/test_debug13.py +++ b/backend/tests/test_debug13.py @@ -65,6 +65,16 @@ def test_adguard_parse_entry(): assert p._parse_entry({"domain": "nx.example.com", "client": "192.168.9.6"}) is None +def test_adguard_entries_shapes(): + """新版 AdGuard 返回 {"data": [...]}（线上实测），旧版 {"entries": [...]}，更老的直接数组""" + p = integrations.AdguardPoller(_engine()) + assert p._entries({"data": [1, 2], "oldest": "x"}) == [1, 2] + assert p._entries({"entries": [3]}) == [3] + assert p._entries([4]) == [4] + assert p._entries({}) == [] + assert p._entries({"data": None}) == [] + + class _FakeEngine: def __init__(self): self.calls = [] diff --git a/backend/tests/test_debug14.py b/backend/tests/test_debug14.py new file mode 100644 index 0000000..2c489e2 --- /dev/null +++ b/backend/tests/test_debug14.py @@ -0,0 +1,126 @@ +"""debug14：上游规则库自动发现、自动登记成应用、识别库更新后自动重算历史分类。""" +import os +import time + +import pytest + +from app.classify import lists as L +from app.classify.lists import guess_category, lists +from app.engine import Engine +from app.models import DnsCache, HostDomainDay, Setting, TrafficDay +from app.reclass import auto_reclassify, reclassify + +DAY = 1792900000 +HOST = "IP:192.168.8.31" + + +class _Resp: + def __init__(self, payload): + self._p = payload + + def raise_for_status(self): + return None + + def json(self): + return self._p + + +def test_guess_category(): + assert guess_category("MangoTV")[0] == "video" + assert guess_category("ICBCBank")[0] == "bank" + assert guess_category("OpenAI")[0] == "ai" + assert guess_category("IPTVMainland2")[0] == "video" + # 猜不出：bm7 用「常用网站」兜底登记；Hagezi / UT1 分类列表直接跳过 + assert guess_category("RandomCorp") == ("site", False) + + +def test_discover_registers_new_dirs(tmp_path, monkeypatch): + monkeypatch.setenv("DATA_DIR", str(tmp_path)) + fake = [{"name": n, "type": "dir"} for n in ["BiliBili", "MangoTV", "RandomCorp", "Global", "HotMovies"]] + called = [] + monkeypatch.setattr(L.httpx, "get", + lambda url, **kw: called.append(url) or (_Resp(fake) if "ios_rule_script" in url else _Resp([]))) + try: + lists.reset() + lists._map = None + lists._auto = {} + lists._load_auto() + added = lists.discover() + assert added == 3 # BiliBili 已映射、Global 在跳过名单；其余 3 个自动登记 + m = lists.map + assert m["apps"]["bm7_mangotv"]["category"] == "video" + assert m["apps"]["bm7_randomcorp"]["category"] == "site" + assert m["owner"]["MangoTV"] == "bm7_mangotv" and m["owner"]["HotMovies"] == "bm7_hotmovies" + assert os.path.exists(lists._auto_file()) + assert called and "rule/Clash" in called[0] + assert lists.discover() == 0 # 再发现一次：没有新增 + # 自动登记的应用分类器认得出（和手工映射同一条管线） + from app.classify import Classifier + c = Classifier() + assert "bm7_mangotv" in c.apps and c.apps["bm7_mangotv"].category == "video" + finally: + lists.reset() + lists._map = None + lists._auto = {} + + +def test_reclass_uses_dns_cache_for_ip_rows(): + import app.database as database + database.connect_from_saved(force=True) + s = database.SessionLocal() + try: + now = int(time.time()) + s.add(DnsCache(ip="203.0.113.9", name="v.douyin.com", expires=now + 3600)) + s.add(HostDomainDay(ts=DAY, host=HOST, domain="203.0.113.9", app="cn_net", base=None, bytes=500)) + s.add(TrafficDay(ts=DAY, host=HOST, app="cn_net", up=100, down=900, flows=10)) + s.commit() + + class _C: + @staticmethod + def classify(ip, proto, port, qname, cname): + return "douyin" if qname == "v.douyin.com" else "cn_net" + + eng = Engine() + eng.classifier = _C() + res = reclassify(eng, s) + s.commit() + assert res["bytes_moved"] >= 500 + marks = {d.domain: d.app for d in s.query(HostDomainDay).filter(HostDomainDay.host == HOST).all()} + assert marks["203.0.113.9"] == "douyin" + + # 防倒退：DNS 缓存过期后，已归到具体应用的行不再被搬回笼统分类 + s.query(DnsCache).filter(DnsCache.ip == "203.0.113.9").delete() + s.commit() + res2 = reclassify(eng, s) + s.commit() + assert res2["bytes_moved"] == 0 + marks = {d.domain: d.app for d in s.query(HostDomainDay).filter(HostDomainDay.host == HOST).all()} + assert marks["203.0.113.9"] == "douyin" + finally: + s.query(TrafficDay).filter(TrafficDay.host == HOST).delete() + s.query(HostDomainDay).filter(HostDomainDay.host == HOST).delete() + s.query(DnsCache).filter(DnsCache.ip == "203.0.113.9").delete() + s.commit() + s.close() + + +def test_auto_reclassify_runs_on_fingerprint_change(): + import app.database as database + database.connect_from_saved(force=True) + s = database.SessionLocal() + try: + s.query(Setting).filter(Setting.key == "reclass_fp").delete() + s.commit() + eng = Engine() + r1 = auto_reclassify(eng, s) + s.commit() + assert r1 is not None and "scanned" in r1 + assert auto_reclassify(eng, s) is None # 指纹没变：什么都不做 + s.add(DnsCache(ip="198.51.100.1", name="x.example.com", expires=int(time.time()) + 600)) + s.commit() + assert auto_reclassify(eng, s) is not None # DNS 映射变多：指纹变化 → 重算 + finally: + s.query(Setting).filter(Setting.key == "reclass_fp").delete() + s.query(DnsCache).filter(DnsCache.ip == "198.51.100.1").delete() + s.commit() + s.close()