diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml new file mode 100644 index 0000000..094dbaa --- /dev/null +++ b/.github/workflows/tests.yml @@ -0,0 +1,28 @@ +name: Tests + +on: + push: + branches: [main] + pull_request: + branches: [main] + +jobs: + unittest: + name: Python ${{ matrix.python }} on ${{ matrix.os }} + runs-on: ${{ matrix.os }} + strategy: + fail-fast: false + matrix: + # Windows is not optional here: the WeChat path-discovery tests exist + # because of Windows-only layouts, and they force IS_WINDOWS=True. + os: [ubuntu-latest, windows-latest, macos-latest] + python: ["3.9", "3.12"] + steps: + - uses: actions/checkout@v4 + + - uses: actions/setup-python@v5 + with: + python-version: ${{ matrix.python }} + + - name: Run unit tests + run: python -m unittest discover -s tests -v diff --git a/cli/paths.py b/cli/paths.py index 4fa2e5e..436a807 100644 --- a/cli/paths.py +++ b/cli/paths.py @@ -18,6 +18,7 @@ import shutil import subprocess import sys +import time from dataclasses import dataclass from pathlib import Path from typing import Optional @@ -427,6 +428,7 @@ def _mac_xwechat_search_paths() -> list[Path]: "xwechat_files", "wechat files", "all_users", + "all users", "backup", "old_backup", "app_data", @@ -442,7 +444,12 @@ def _looks_like_account_dir(p: Path) -> bool: except (PermissionError, OSError): return False name = p.name.lower() - if name in _IGNORED_WECHAT_ACCOUNT_DIRS or name.startswith("all"): + # Exact matches only. This used to also reject anything starting with "all", + # which was meant for `all_users` but silently swallowed real accounts whose + # WeChat alias happens to start with those letters (allen_9f3a, ally_1a2b …). + # Such a user could paste the correct path at every level and still be told + # nothing was found — `all_users`/`all users` are in the set above instead. + if name in _IGNORED_WECHAT_ACCOUNT_DIRS: return False try: return (p / "db_storage").is_dir() @@ -587,6 +594,36 @@ def _wechat_root_config_paths() -> list[Path]: return _dedupe_paths(out) +def _user_supplied_roots() -> list[Path]: + """Paths the user explicitly handed us — normalised, but NOT fanned out into + the hardcoded wrapper variants. + + `_wechat_root_config_paths()` expands each saved path into a fixed list of + plausible layouts (`Tencent/xwechat_files`, `Documents/xwechat_files`, …). + That list cannot cover every folder name people pick, so we keep the bare + user path around too and search under it directly — see + `_descend_for_account_dirs`. + """ + raws: list[str] = [] + env_raw = os.environ.get("MURMUR_WECHAT_ROOT", "").strip() + if env_raw: + raws.extend(env_raw.split(os.pathsep)) + cfg = load_config() + saved = cfg.get("wechat_roots", []) + if isinstance(saved, str): + saved = [saved] + if isinstance(saved, list): + raws.extend(x for x in saved if isinstance(x, str)) + + out: list[Path] = [] + for raw in raws: + raw = raw.strip().strip('"') + if not raw: + continue + out.append(_normalize_user_root(Path(os.path.expandvars(os.path.expanduser(raw))))) + return _dedupe_paths(out) + + def wechat_search_paths() -> list[Path]: """All candidate WeChat data roots Murmur will inspect.""" env_candidates = _wechat_root_env_paths() @@ -898,6 +935,58 @@ def worker(): return result["entries"] +def _descend_for_account_dirs( + root: Path, + *, + max_depth: int = 4, + budget: int = 3000, + deadline_s: float = 8.0, +) -> list[Path]: + """Bounded breadth-first search under a USER-SUPPLIED root for account dirs. + + Only called on paths the user typed, and only after the shallow lookup came + up empty. `_windows_xwechat_variants` enumerates a fixed set of wrapper + names, so someone who keeps WeChat under `D:\\Weixin\\`, `D:\\微信文件\\` + or `D:\\data\\wx\\` pastes a perfectly correct path and is still told + nothing was found — the single most confusing failure this UI can produce. + Walking a few levels down handles any layout without another round of + whack-a-mole on folder names. + + Bounded on depth, directories visited AND wall clock so pasting `D:\\` + degrades into a quick shallow look rather than a whole-drive crawl; that job + belongs to /api/scan-disks. The wall clock matters independently: every + listdir here can burn up to `_safe_listdir`'s own 1.5s timeout, so a budget + counted only in directories could still stall the request for minutes on a + dying disk or a disconnected network drive. Account dirs are not descended + into — WeChat data holds tens of thousands of media files and nothing we + want is below that level. + """ + found: list[Path] = [] + visited = 0 + started = time.monotonic() + frontier: list[tuple[Path, int]] = [(root, 0)] + while frontier and visited < budget: + if time.monotonic() - started > deadline_s: + break + cur, depth = frontier.pop(0) + entries = _safe_listdir(cur) + if entries is None: # TCC-blocked or hung — skip, don't fail the walk + continue + visited += 1 + for entry in entries: + try: + if not entry.is_dir(): + continue + except (PermissionError, OSError): + continue + if _looks_like_account_dir(entry): + found.append(entry) + continue + if depth + 1 <= max_depth and entry.name.lower() not in _SCAN_SKIP_NAMES: + frontier.append((entry, depth + 1)) + return found + + _LAST_TCC_BLOCKED: bool = False # set by discover_wechat_profiles @@ -994,38 +1083,62 @@ def _db_storage_size(d: Path) -> int: picked.append(max(dirs, key=_db_storage_size)) wxid_subs = picked - for sub in wxid_subs: - wxid_full = sub.name - profile_key = str(sub) - if profile_key in seen_profiles: - continue - # Require the wxid dir to actually contain decryptable data. - # Without this guard, a `wxid_*/` directory whose `db_storage/` is - # empty (or missing entirely) gets reported as a valid profile — - # diagnose then says "微信数据 已找到 ✓", refresh.py iterates 0 DBs - # and exits 0, the post-decrypt promote finds no session.db, and - # the user sees "decrypt subprocess returned 0 but no decrypted - # directory found". Filter such empty shells out at discovery. - db_storage = sub / "db_storage" + _add_profiles(wxid_subs, profiles, seen_profiles, plat) + + # The user pasted a path and we still found nothing. Before reporting + # failure — the worst outcome this screen has — search under what they gave + # us. The hardcoded wrapper list can't know they keep WeChat in `D:\Weixin\`. + if not profiles: + for user_root in _user_supplied_roots(): try: - if not db_storage.is_dir(): - continue - if not any(db_storage.rglob("*.db")): + if not user_root.exists(): continue except (PermissionError, OSError): + _LAST_TCC_BLOCKED = True continue - seen_profiles.add(profile_key) - wxid_short = _wechat_account_short(wxid_full) - profiles.append(WeChatProfile( - wxid=wxid_full, - wxid_short=wxid_short, - encrypted_root=db_storage, - cache_root=sub, - platform=plat, - )) + _add_profiles(_descend_for_account_dirs(user_root), profiles, seen_profiles, plat) + return profiles +def _add_profiles( + account_dirs: list[Path], + profiles: list[WeChatProfile], + seen_profiles: set[str], + plat: str, +) -> None: + """Validate candidate account dirs and append the usable ones to `profiles`.""" + for sub in account_dirs: + wxid_full = sub.name + profile_key = str(sub) + if profile_key in seen_profiles: + continue + # Require the wxid dir to actually contain decryptable data. + # Without this guard, a `wxid_*/` directory whose `db_storage/` is + # empty (or missing entirely) gets reported as a valid profile — + # diagnose then says "微信数据 已找到 ✓", refresh.py iterates 0 DBs + # and exits 0, the post-decrypt promote finds no session.db, and + # the user sees "decrypt subprocess returned 0 but no decrypted + # directory found". Filter such empty shells out at discovery. + db_storage = sub / "db_storage" + try: + if not db_storage.is_dir(): + continue + if not any(db_storage.rglob("*.db")): + continue + except (PermissionError, OSError): + continue + seen_profiles.add(profile_key) + wxid_short = _wechat_account_short(wxid_full) + profiles.append(WeChatProfile( + wxid=wxid_full, + wxid_short=wxid_short, + encrypted_root=db_storage, + cache_root=sub, + platform=plat, + )) + + # (db_filename, sentinel_table) pairs we expect under a healthy decrypted dir. # session.db is *required* — without it Murmur cannot list any sessions. The # rest are *advisory*: if any of them is present but its sentinel table is diff --git a/tests/test_wechat_path_discovery.py b/tests/test_wechat_path_discovery.py new file mode 100644 index 0000000..6fc9fe3 --- /dev/null +++ b/tests/test_wechat_path_discovery.py @@ -0,0 +1,141 @@ +"""Regression tests for manually-pasted WeChat data paths. + +Reported from the field (Windows): pasting the folder WeChat's 「文件管理」 +shows still ended in "已保存路径,但里面还没找到 wxid_*/db_storage". Two +independent causes, both covered here: + + 1. `_windows_xwechat_variants` enumerates a fixed set of wrapper folder names, + so any custom layout (`D:\\Weixin\\xwechat_files`, `D:\\微信文件\\...`) + was invisible no matter which level the user pasted. + 2. `_looks_like_account_dir` rejected every directory whose name started with + "all" — intended for `all_users`, but it also swallowed real accounts + (`allen_9f3a`). +""" +import sys +import tempfile +import unittest +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT)) + +from cli import paths # noqa: E402 + + +def make_tree(base: Path, rel: str, account: str = "wxid_a1b2c3d4e5") -> Path: + """Create ///db_storage/session/session.db.""" + session = base / rel / account / "db_storage" / "session" + session.mkdir(parents=True, exist_ok=True) + (session / "session.db").write_bytes(b"SQLite format 3\x00" + b"\x00" * 64) + return base / rel / account + + +class WeChatPathDiscoveryTest(unittest.TestCase): + def setUp(self) -> None: + self._tmp = tempfile.TemporaryDirectory() + self.tmp = Path(self._tmp.name) + cfg = self.tmp / "config.json" + cfg.write_text("{}", encoding="utf-8") + + self.cfg = cfg + self._orig_cfg_path = paths.murmur_config_path + self._orig_is_windows = paths.IS_WINDOWS + self._orig_is_mac = paths.IS_MAC + paths.murmur_config_path = lambda: cfg + # Exercise the Windows-only branches regardless of the host OS. + paths.IS_WINDOWS = True + paths.IS_MAC = False + + def tearDown(self) -> None: + paths.murmur_config_path = self._orig_cfg_path + paths.IS_WINDOWS = self._orig_is_windows + paths.IS_MAC = self._orig_is_mac + self._tmp.cleanup() + + def discover_after_pasting(self, pasted: Path) -> list[str]: + # save_wechat_root() keeps up to 8 previous roots, so each case starts + # from a clean config — otherwise earlier subTests leak in as extra hits. + self.cfg.write_text("{}", encoding="utf-8") + paths.save_wechat_root(str(pasted)) + return [p.wxid for p in paths.discover_wechat_profiles()] + + # ── layout variants: the user pastes WeChat's configured storage folder ── + + def test_finds_account_under_custom_wrapper_folders(self) -> None: + for rel in ( + "xwechat_files", + "Tencent/xwechat_files", + "WeChat/xwechat_files", + "Weixin/xwechat_files", + "wechatData/xwechat_files", + "微信文件/xwechat_files", + "data/wx/xwechat_files", + ): + with self.subTest(layout=rel): + base = Path(tempfile.mkdtemp(dir=self.tmp)) + make_tree(base, rel) + self.assertEqual( + self.discover_after_pasting(base), ["wxid_a1b2c3d4e5"] + ) + + def test_finds_account_at_every_level_the_user_might_paste(self) -> None: + base = Path(tempfile.mkdtemp(dir=self.tmp)) + account = make_tree(base, "xwechat_files") + for label, pasted in ( + ("parent of xwechat_files", base), + ("xwechat_files", account.parent), + ("account dir", account), + ("db_storage", account / "db_storage"), + ("a .db file", account / "db_storage" / "session" / "session.db"), + ): + with self.subTest(pasted=label): + self.assertEqual( + self.discover_after_pasting(pasted), ["wxid_a1b2c3d4e5"] + ) + + # ── account names beginning with "all" ── + + def test_account_named_like_all_users_is_still_found(self) -> None: + for account in ("allen_9f3a", "ally_1a2b", "alice_2b7c"): + with self.subTest(account=account): + base = Path(tempfile.mkdtemp(dir=self.tmp)) + make_tree(base, "xwechat_files", account=account) + self.assertEqual(self.discover_after_pasting(base), [account]) + + def test_all_users_directory_is_still_ignored(self) -> None: + base = Path(tempfile.mkdtemp(dir=self.tmp)) + make_tree(base, "xwechat_files", account="wxid_real1234") + # WeChat ships a shared `all_users` folder next to the real accounts; + # giving it a db_storage must not make it look like an account. + noise = base / "xwechat_files" / "all_users" / "db_storage" + noise.mkdir(parents=True) + (noise / "x.db").write_bytes(b"SQLite format 3\x00") + self.assertEqual(self.discover_after_pasting(base), ["wxid_real1234"]) + + # ── guards on the bounded descent ── + + def test_empty_account_shell_is_not_reported(self) -> None: + # db_storage exists but holds no .db files — decrypt would return 0 rows + # and the user would get "no decrypted directory found" much later. + base = Path(tempfile.mkdtemp(dir=self.tmp)) + (base / "xwechat_files" / "wxid_empty0001" / "db_storage").mkdir(parents=True) + self.assertEqual(self.discover_after_pasting(base), []) + + def test_descent_does_not_run_past_its_depth_budget(self) -> None: + base = Path(tempfile.mkdtemp(dir=self.tmp)) + make_tree(base, "a/b/c/d/e/f/xwechat_files") + self.assertEqual(self.discover_after_pasting(base), []) + + def test_descent_stops_at_the_account_dir(self) -> None: + # Nothing below an account dir should be walked: real ones hold tens of + # thousands of media files. + base = Path(tempfile.mkdtemp(dir=self.tmp)) + account = make_tree(base, "xwechat_files") + buried = account / "msg" / "attach" / "xwechat_files" / "wxid_nested999" + (buried / "db_storage").mkdir(parents=True) + (buried / "db_storage" / "s.db").write_bytes(b"SQLite format 3\x00") + self.assertEqual(self.discover_after_pasting(base), ["wxid_a1b2c3d4e5"]) + + +if __name__ == "__main__": + unittest.main()