Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
28 changes: 28 additions & 0 deletions .github/workflows/tests.yml
Original file line number Diff line number Diff line change
@@ -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
165 changes: 139 additions & 26 deletions cli/paths.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
import shutil
import subprocess
import sys
import time
from dataclasses import dataclass
from pathlib import Path
from typing import Optional
Expand Down Expand Up @@ -427,6 +428,7 @@ def _mac_xwechat_search_paths() -> list[Path]:
"xwechat_files",
"wechat files",
"all_users",
"all users",
"backup",
"old_backup",
"app_data",
Expand All @@ -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()
Expand Down Expand Up @@ -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()
Expand Down Expand Up @@ -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


Expand Down Expand Up @@ -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
Expand Down
141 changes: 141 additions & 0 deletions tests/test_wechat_path_discovery.py
Original file line number Diff line number Diff line change
@@ -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 <base>/<rel>/<account>/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()
Loading