From de6d269fb4a00844961364df25c21ec3810ab096 Mon Sep 17 00:00:00 2001 From: yuyx <237899745@qq.com> Date: Mon, 15 Dec 2025 15:09:34 +0800 Subject: [PATCH] =?UTF-8?q?=E6=9B=B4=E6=96=B0=20update=5Fagent.py?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tools/update_agent.py | 82 +++++++++++++++++++++++++++++++++++++++---- 1 file changed, 76 insertions(+), 6 deletions(-) diff --git a/tools/update_agent.py b/tools/update_agent.py index 9fff7dc..f7396ad 100644 --- a/tools/update_agent.py +++ b/tools/update_agent.py @@ -88,9 +88,79 @@ def _git_rev_parse(ref: str, *, cwd: Path) -> str: return out -def _git_is_dirty(*, cwd: Path) -> bool: - out = subprocess.check_output(["git", "status", "--porcelain"], cwd=str(cwd), text=True) - return bool(out.strip()) +def _git_has_tracked_changes(*, cwd: Path) -> bool: + """是否存在 tracked 的未提交修改(含暂存区)。""" + for cmd in (["git", "diff", "--quiet"], ["git", "diff", "--cached", "--quiet"]): + proc = subprocess.run(cmd, cwd=str(cwd)) + if proc.returncode == 1: + return True + if proc.returncode != 0: + raise RuntimeError(f"{' '.join(cmd)} failed with code {proc.returncode}") + return False + + +def _normalize_prefixes(prefixes: Tuple[str, ...]) -> Tuple[str, ...]: + normalized = [] + for p in prefixes: + text = str(p or "").strip() + if not text: + continue + if not text.endswith("/"): + text += "/" + normalized.append(text) + return tuple(normalized) + + +def _git_has_untracked_changes(*, cwd: Path, ignore_prefixes: Tuple[str, ...]) -> Tuple[bool, int, list[str]]: + """检查 untracked 文件(尊重 .gitignore),并忽略指定前缀目录。""" + ignore_prefixes = _normalize_prefixes(ignore_prefixes) + out = subprocess.check_output(["git", "ls-files", "--others", "--exclude-standard"], cwd=str(cwd), text=True) + paths = [line.strip() for line in out.splitlines() if line.strip()] + + filtered = [] + for p in paths: + if ignore_prefixes and any(p.startswith(prefix) for prefix in ignore_prefixes): + continue + filtered.append(p) + + samples = filtered[:20] + return (len(filtered) > 0), len(filtered), samples + + +def _git_is_dirty(*, cwd: Path, ignore_untracked_prefixes: Tuple[str, ...] = ("data/",)) -> dict: + """ + 判断工作区是否“脏”: + - tracked 变更(含暂存区)一律算脏 + - untracked 文件默认忽略 data/(运行时数据目录,避免后台长期提示) + """ + tracked_dirty = False + untracked_dirty = False + untracked_count = 0 + untracked_samples: list[str] = [] + try: + tracked_dirty = _git_has_tracked_changes(cwd=cwd) + except Exception: + # 若 diff 检测异常,回退到保守策略:认为脏 + tracked_dirty = True + + try: + untracked_dirty, untracked_count, untracked_samples = _git_has_untracked_changes( + cwd=cwd, ignore_prefixes=ignore_untracked_prefixes + ) + except Exception: + # 若 untracked 检测异常,回退到不影响更新:不计入 dirty + untracked_dirty = False + untracked_count = 0 + untracked_samples = [] + + return { + "dirty": bool(tracked_dirty or untracked_dirty), + "dirty_tracked": bool(tracked_dirty), + "dirty_untracked": bool(untracked_dirty), + "dirty_ignore_untracked_prefixes": list(_normalize_prefixes(ignore_untracked_prefixes)), + "untracked_count": int(untracked_count), + "untracked_samples": list(untracked_samples), + } def _compose_cmd() -> list[str]: @@ -151,7 +221,7 @@ def check_updates(*, paths: Paths, branch: str, log_fp=None) -> dict: err = "" local = "" remote = "" - dirty = False + dirty_info: dict = {} try: if log_fp: _run(["git", "fetch", "origin", branch], cwd=paths.repo_dir, log_fp=log_fp, env=env) @@ -159,7 +229,7 @@ def check_updates(*, paths: Paths, branch: str, log_fp=None) -> dict: subprocess.run(["git", "fetch", "origin", branch], cwd=str(paths.repo_dir), env={**os.environ, **env}, check=True) local = _git_rev_parse("HEAD", cwd=paths.repo_dir) remote = _git_rev_parse(f"origin/{branch}", cwd=paths.repo_dir) - dirty = _git_is_dirty(cwd=paths.repo_dir) + dirty_info = _git_is_dirty(cwd=paths.repo_dir, ignore_untracked_prefixes=("data/",)) except Exception as e: err = f"{type(e).__name__}: {e}" @@ -170,7 +240,7 @@ def check_updates(*, paths: Paths, branch: str, log_fp=None) -> dict: "local_commit": local, "remote_commit": remote, "update_available": update_available, - "dirty": dirty, + **(dirty_info or {"dirty": False}), "error": err, }