更新 update_agent.py
This commit is contained in:
@@ -88,9 +88,79 @@ def _git_rev_parse(ref: str, *, cwd: Path) -> str:
|
|||||||
return out
|
return out
|
||||||
|
|
||||||
|
|
||||||
def _git_is_dirty(*, cwd: Path) -> bool:
|
def _git_has_tracked_changes(*, cwd: Path) -> bool:
|
||||||
out = subprocess.check_output(["git", "status", "--porcelain"], cwd=str(cwd), text=True)
|
"""是否存在 tracked 的未提交修改(含暂存区)。"""
|
||||||
return bool(out.strip())
|
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]:
|
def _compose_cmd() -> list[str]:
|
||||||
@@ -151,7 +221,7 @@ def check_updates(*, paths: Paths, branch: str, log_fp=None) -> dict:
|
|||||||
err = ""
|
err = ""
|
||||||
local = ""
|
local = ""
|
||||||
remote = ""
|
remote = ""
|
||||||
dirty = False
|
dirty_info: dict = {}
|
||||||
try:
|
try:
|
||||||
if log_fp:
|
if log_fp:
|
||||||
_run(["git", "fetch", "origin", branch], cwd=paths.repo_dir, log_fp=log_fp, env=env)
|
_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)
|
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)
|
local = _git_rev_parse("HEAD", cwd=paths.repo_dir)
|
||||||
remote = _git_rev_parse(f"origin/{branch}", 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:
|
except Exception as e:
|
||||||
err = f"{type(e).__name__}: {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,
|
"local_commit": local,
|
||||||
"remote_commit": remote,
|
"remote_commit": remote,
|
||||||
"update_available": update_available,
|
"update_available": update_available,
|
||||||
"dirty": dirty,
|
**(dirty_info or {"dirty": False}),
|
||||||
"error": err,
|
"error": err,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user