scripts/update.py
Rama main · réplica de lectura del repositorio original.
#!/usr/bin/env python
"""Safely update a local DeepTutor git checkout.
The updater is intentionally conservative:
1. Fetch the remote for the current branch.
2. Show the local-vs-remote gap and ask for confirmation.
3. Fast-forward pull only when the branch can be updated safely.
"""
from __future__ import annotations
import argparse
from dataclasses import dataclass
from pathlib import Path
import subprocess
import sys
from typing import Sequence
PROJECT_ROOT = Path(__file__).resolve().parent.parent
class UpdateError(Exception):
"""Raised for user-actionable update failures."""
@dataclass(frozen=True)
class GitResult:
args: tuple[str, ...]
returncode: int
stdout: str
stderr: str
@dataclass(frozen=True)
class BranchTarget:
local_branch: str
remote: str
remote_branch: str
remote_ref: str
upstream: str | None
source: str
@property
def display_remote_ref(self) -> str:
return f"{self.remote}/{self.remote_branch}"
@dataclass(frozen=True)
class BranchGap:
local_sha: str
local_subject: str
remote_sha: str
remote_subject: str
ahead: int
behind: int
incoming_commits: list[str]
outgoing_commits: list[str]
diff_stat: str
dirty_entries: list[str]
@property
def is_up_to_date(self) -> bool:
return self.ahead == 0 and self.behind == 0
@property
def is_fast_forwardable(self) -> bool:
return self.ahead == 0 and self.behind > 0
@property
def is_diverged(self) -> bool:
return self.ahead > 0 and self.behind > 0
class Git:
def __init__(self, repo_root: Path) -> None:
self.repo_root = repo_root
def run(
self,
args: Sequence[str],
*,
check: bool = True,
timeout: int | None = 60,
) -> GitResult:
cmd = ["git", *args]
completed = subprocess.run(
cmd,
cwd=self.repo_root,
capture_output=True,
encoding="utf-8",
errors="replace",
text=True,
timeout=timeout,
)
result = GitResult(
args=tuple(args),
returncode=completed.returncode,
stdout=completed.stdout.strip(),
stderr=completed.stderr.strip(),
)
if check and result.returncode != 0:
message = result.stderr or result.stdout or f"git {' '.join(args)} failed"
raise UpdateError(message)
return result
def _print_section(title: str) -> None:
print()
print(f"== {title} ==")
def _print_list(title: str, items: list[str], *, empty: str) -> None:
print(f"{title}:")
if not items:
print(f" {empty}")
return
for item in items:
print(f" {item}")
def ensure_git_checkout(git: Git) -> None:
inside = git.run(["rev-parse", "--is-inside-work-tree"]).stdout
if inside != "true":
raise UpdateError("This directory is not inside a git checkout.")
def current_branch(git: Git) -> str:
branch = git.run(["branch", "--show-current"]).stdout
if not branch:
raise UpdateError(
"Detached HEAD detected. Please switch to a branch before running the updater."
)
return branch
def git_config(git: Git, key: str) -> str | None:
result = git.run(["config", "--get", key], check=False)
if result.returncode != 0:
return None
return result.stdout or None
def available_remotes(git: Git) -> list[str]:
remotes = git.run(["remote"], check=False).stdout.splitlines()
return [remote.strip() for remote in remotes if remote.strip()]
def normalize_merge_ref(merge_ref: str | None, fallback_branch: str) -> str:
if not merge_ref:
return fallback_branch
prefix = "refs/heads/"
if merge_ref.startswith(prefix):
return merge_ref[len(prefix) :]
return merge_ref
def resolve_branch_target(git: Git, branch: str) -> BranchTarget:
remote = git_config(git, f"branch.{branch}.remote")
merge_ref = git_config(git, f"branch.{branch}.merge")
if remote and remote != ".":
remote_branch = normalize_merge_ref(merge_ref, branch)
return BranchTarget(
local_branch=branch,
remote=remote,
remote_branch=remote_branch,
remote_ref=f"refs/remotes/{remote}/{remote_branch}",
upstream=f"{remote}/{remote_branch}",
source="configured upstream",
)
remotes = available_remotes(git)
if not remotes:
raise UpdateError("No git remote is configured for this checkout.")
selected_remote = "origin" if "origin" in remotes else remotes[0]
return BranchTarget(
local_branch=branch,
remote=selected_remote,
remote_branch=branch,
remote_ref=f"refs/remotes/{selected_remote}/{branch}",
upstream=None,
source=f"default remote '{selected_remote}' with matching branch name",
)
def fetch_remote(git: Git, target: BranchTarget) -> None:
print(f"Fetching latest refs from {target.remote} ...")
git.run(["fetch", "--prune", target.remote], timeout=None)
def verify_remote_branch(git: Git, target: BranchTarget) -> None:
result = git.run(
["rev-parse", "--verify", "--quiet", f"{target.remote_ref}^{{commit}}"],
check=False,
)
if result.returncode == 0:
return
raise UpdateError(
"Remote branch not found after fetch: "
f"{target.display_remote_ref}. Set the branch upstream or push it first."
)
def short_commit(git: Git, ref: str) -> tuple[str, str]:
sha = git.run(["rev-parse", "--short", ref]).stdout
subject = git.run(["log", "-1", "--format=%s", ref]).stdout
return sha, subject
def log_lines(git: Git, revision_range: str, limit: int = 8) -> list[str]:
result = git.run(
["log", "--oneline", "--decorate", f"--max-count={limit}", revision_range],
check=False,
)
if result.returncode != 0 or not result.stdout:
return []
return result.stdout.splitlines()
def tracked_dirty_entries(git: Git) -> list[str]:
git.run(["update-index", "-q", "--refresh"], check=False)
result = git.run(["status", "--porcelain=v1", "--untracked-files=no"], check=False)
if result.returncode != 0 or not result.stdout:
return []
return result.stdout.splitlines()
def analyze_gap(git: Git, target: BranchTarget) -> BranchGap:
local_sha, local_subject = short_commit(git, "HEAD")
remote_sha, remote_subject = short_commit(git, target.remote_ref)
counts = git.run(["rev-list", "--left-right", "--count", f"HEAD...{target.remote_ref}"])
ahead_str, behind_str = counts.stdout.split()
ahead = int(ahead_str)
behind = int(behind_str)
diff_stat = ""
if behind:
diff_stat = git.run(
["diff", "--stat", "--compact-summary", f"HEAD..{target.remote_ref}"]
).stdout
return BranchGap(
local_sha=local_sha,
local_subject=local_subject,
remote_sha=remote_sha,
remote_subject=remote_subject,
ahead=ahead,
behind=behind,
incoming_commits=log_lines(git, f"HEAD..{target.remote_ref}"),
outgoing_commits=log_lines(git, f"{target.remote_ref}..HEAD"),
diff_stat=diff_stat,
dirty_entries=tracked_dirty_entries(git),
)
def print_gap(target: BranchTarget, gap: BranchGap) -> None:
_print_section("Detected branch")
print(f"Local branch: {target.local_branch}")
print(f"Remote branch: {target.display_remote_ref}")
print(f"Selection: {target.source}")
if target.upstream:
print(f"Upstream: {target.upstream}")
else:
print("Upstream: not configured")
_print_section("Local vs remote")
print(f"Local HEAD: {gap.local_sha} {gap.local_subject}")
print(f"Remote HEAD: {gap.remote_sha} {gap.remote_subject}")
print(f"Gap: local is {gap.ahead} commit(s) ahead, {gap.behind} commit(s) behind")
_print_list(
"Incoming commits from remote",
gap.incoming_commits,
empty="none",
)
_print_list(
"Local commits not on remote",
gap.outgoing_commits,
empty="none",
)
if gap.diff_stat:
print("Incoming file summary:")
for line in gap.diff_stat.splitlines():
print(f" {line}")
if gap.dirty_entries:
_print_list(
"Tracked local changes",
gap.dirty_entries,
empty="none",
)
def confirm_update(assume_yes: bool, target: BranchTarget, gap: BranchGap) -> bool:
if assume_yes:
return True
if not sys.stdin.isatty():
print()
print("Confirmation required, but stdin is not interactive. Re-run with --yes to update.")
return False
print()
answer = input(
"Confirm this branch mapping and update "
f"{target.local_branch} from {target.display_remote_ref}? [y/N] "
).strip()
return answer.lower() in {"y", "yes"}
def ensure_safe_to_update(gap: BranchGap) -> None:
if gap.dirty_entries:
raise UpdateError(
"Tracked local changes are present. Commit or stash them before updating."
)
if gap.is_diverged:
raise UpdateError(
"Local and remote branches have diverged. Resolve manually, then rerun the updater."
)
if gap.ahead > 0 and gap.behind == 0:
raise UpdateError(
"Local branch has commits not present on the remote, and there are no remote "
"commits to pull."
)
if not gap.is_fast_forwardable:
raise UpdateError("This branch cannot be updated with a safe fast-forward pull.")
def dependency_hints(changed_files: list[str]) -> list[str]:
hints: list[str] = []
if any(path == "pyproject.toml" or path.startswith("requirements/") for path in changed_files):
hints.append("Backend dependencies changed: consider running python -m pip install -e .")
if any(
path in {"web/package.json", "web/package-lock.json", "web/pnpm-lock.yaml"}
or path == "web/yarn.lock"
for path in changed_files
):
hints.append("Frontend dependencies changed: consider running cd web && npm install")
return hints
def pull_updates(git: Git, target: BranchTarget) -> list[str]:
old_sha = git.run(["rev-parse", "HEAD"]).stdout
_print_section("Updating")
git.run(["pull", "--ff-only", target.remote, target.remote_branch], timeout=None)
new_sha = git.run(["rev-parse", "HEAD"]).stdout
if old_sha == new_sha:
return []
changed = git.run(["diff", "--name-only", f"{old_sha}..{new_sha}"]).stdout
return [line for line in changed.splitlines() if line.strip()]
def run_update(repo_root: Path, *, assume_yes: bool) -> int:
git = Git(repo_root)
ensure_git_checkout(git)
branch = current_branch(git)
target = resolve_branch_target(git, branch)
fetch_remote(git, target)
verify_remote_branch(git, target)
gap = analyze_gap(git, target)
print_gap(target, gap)
if gap.dirty_entries:
raise UpdateError(
"Tracked local changes are present. Commit or stash them before updating."
)
if gap.is_diverged:
raise UpdateError(
"Local and remote branches have diverged. Resolve manually, then rerun the updater."
)
if gap.is_up_to_date:
print()
print("Already up to date. No update is needed.")
return 0
if gap.ahead > 0 and gap.behind == 0:
print()
print("No remote commits to pull. Local branch is ahead of the remote.")
return 0
if not confirm_update(assume_yes, target, gap):
print("Update cancelled.")
return 0
ensure_safe_to_update(gap)
changed_files = pull_updates(git, target)
print()
print(f"Updated {target.local_branch} from {target.display_remote_ref}.")
if changed_files:
print(f"Changed files: {len(changed_files)}")
for hint in dependency_hints(changed_files):
print(f"Next step: {hint}")
print("Restart DeepTutor if it is currently running.")
return 0
def parse_args(argv: Sequence[str] | None = None) -> argparse.Namespace:
parser = argparse.ArgumentParser(
description="Fetch, review, and fast-forward update a local DeepTutor checkout."
)
parser.add_argument(
"--yes",
"-y",
action="store_true",
help="Skip the interactive confirmation after printing the branch comparison.",
)
parser.add_argument(
"--repo",
type=Path,
default=PROJECT_ROOT,
help="Path to the git checkout to update. Defaults to this DeepTutor repository.",
)
return parser.parse_args(argv)
def main(argv: Sequence[str] | None = None) -> int:
args = parse_args(argv)
try:
return run_update(args.repo.resolve(), assume_yes=args.yes)
except KeyboardInterrupt:
print()
print("Update cancelled.")
return 130
except (OSError, subprocess.SubprocessError, UpdateError) as exc:
print()
print(f"Update failed: {exc}", file=sys.stderr)
return 1
if __name__ == "__main__":
raise SystemExit(main())