1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
|
#!/usr/bin/env python3
"""Self-update for the audiobook generator's own git checkout.
The generator ships as a plain ``git clone`` with no tags or releases, so
updating means moving the checkout to the remote's default-branch HEAD —
the same fetch + hard-reset flow the backend checkouts use
(``backends.common.git_update``). Everything stateful lives in untracked,
gitignored paths (``app/envs/``, ``input/``, ``output/``, ``voices/``, the
backend checkouts, logs), which a hard reset never touches.
The one tracked file users are expected to edit — ``app/converter/
config.py`` — gets explicit protection: every locally modified tracked
file is snapshotted before the reset and written back afterwards, so an
update can never silently overwrite user configuration with upstream's
new version. A kept file whose upstream version changed is reported, so
new upstream options can be merged by hand.
Run via the hub's Configure Backends > Update Generator action; the
module is stdlib-only like ``backends.common`` / ``backends.envs``.
"""
from pathlib import Path
from typing import Dict, List, Optional
from backends import common
from backends.envs import TTS_ROOT
def _root(root: Optional[Path]) -> Path:
return Path(root) if root is not None else TTS_ROOT
def is_git_checkout(root: Optional[Path] = None) -> bool:
"""True when ROOT (the install by default) is a git working copy.
``.git`` is a directory for a normal clone and a file for a linked
worktree or submodule checkout — both count; a zip download or a copy
without git history does not.
"""
return (_root(root) / ".git").exists()
def current_commit(root: Optional[Path] = None) -> Optional[str]:
"""The checked-out commit's short SHA, or None when git cannot answer."""
proc = common.run_console_subprocess_quiet(
["git", "-C", str(_root(root)), "rev-parse", "--short", "HEAD"])
if proc is None or proc.returncode != 0:
return None
sha = proc.stdout.decode("ascii", errors="replace").strip()
return sha or None
def modified_tracked_files(root: Optional[Path] = None) -> List[str]:
"""Tracked files with local changes (staged or unstaged), repo-relative.
Untracked files are excluded (``--untracked-files=no``): they are
invisible to ``git reset --hard`` anyway. Deletions are included —
the reset restores a deleted tracked file, so they show up in the
update confirm as state the reset will undo. An empty list also
results when git itself fails (no checkout), so callers must gate on
is_git_checkout for the error case.
"""
proc = common.run_console_subprocess_quiet(
["git", "-C", str(_root(root)), "status", "--porcelain",
"--untracked-files=no"])
if proc is None or proc.returncode != 0:
return []
files: List[str] = []
for line in proc.stdout.decode("utf-8", errors="replace").splitlines():
# Porcelain: "XY<space>path"; staged renames read "old -> new".
if len(line) < 4:
continue
path = line[3:]
if " -> " in path:
path = path.rsplit(" -> ", 1)[1]
path = path.strip().strip('"')
if path and path not in files:
files.append(path)
return files
def _say(message: str, emit=None) -> None:
if emit is None:
print(message)
else:
emit(message)
def _snapshot(rel_paths: List[str], root: Path) -> Dict[str, bytes]:
"""Read every existing file in REL_PATHS (missing ones — deletions —
are skipped: the reset restores those by itself)."""
snapshot: Dict[str, bytes] = {}
for rel in rel_paths:
path = root / rel
try:
if path.is_file():
snapshot[rel] = path.read_bytes()
except OSError:
continue
return snapshot
def _restore(snapshot: Dict[str, bytes], root: Path) -> List[str]:
"""Write the user's file contents back over the reset's result.
A file whose post-reset content already matches the snapshot (the
user's version is what upstream now ships) is left untouched, so only
genuinely skipped upstream changes are reported. Returns the
repo-relative paths that were actually written.
"""
restored: List[str] = []
for rel, data in snapshot.items():
path = root / rel
try:
if path.is_file() and path.read_bytes() == data:
continue
path.parent.mkdir(parents=True, exist_ok=True)
path.write_bytes(data)
except OSError:
continue
restored.append(rel)
return restored
def _rev_parse(root: Path, ref: str) -> Optional[str]:
proc = common.run_console_subprocess_quiet(
["git", "-C", str(root), "rev-parse", "--short", ref])
if proc is None or proc.returncode != 0:
return None
sha = proc.stdout.decode("ascii", errors="replace").strip()
return sha or None
def update_generator(*, emit=None, cancel=None,
root: Optional[Path] = None) -> int:
"""Update the generator's own checkout to the remote's default branch.
Fetch, then — only when the remote moved — hard-reset to
``origin/<default-branch>`` (an up-to-date checkout skips the reset
entirely, so a no-op update can never discard anything). Locally
modified tracked files are snapshotted first and written back after
the reset attempt (even a cancelled or failed one — the tree may be
mid-reset when CANCEL fires), which is what keeps user config intact.
EMIT/CANCEL behave like git_update's. Returns the exit code of the
first failing step (0 when the checkout now matches the remote, or
already did).
"""
base = _root(root)
snapshot = _snapshot(modified_tracked_files(base), base)
_say("[INFO] Fetching the latest generator code...", emit)
fetch_argv = ["git", "-C", str(base), "fetch"]
if emit is not None:
# --progress makes git report percentage updates even though stderr
# is piped (it normally only does so on a terminal), feeding the
# task view — same as git_update.
fetch_argv.append("--progress")
fetch_argv.append("origin")
fetch_rc = common.run_console_subprocess(fetch_argv, emit=emit,
cancel=cancel)
if fetch_rc != 0:
_restore(snapshot, base)
return fetch_rc
branch = common.origin_default_branch(base)
remote = _rev_parse(base, f"origin/{branch}")
if remote is not None and _rev_parse(base, "HEAD") == remote:
_say(f"[INFO] The generator is already up to date ({remote}).", emit)
_restore(snapshot, base)
return 0
_say(f"[INFO] Updating the checkout to origin/{branch} "
f"({remote or 'unknown commit'})...", emit)
reset_rc = common.run_console_subprocess(
["git", "-C", str(base), "reset", "--hard", f"origin/{branch}"],
emit=emit, cancel=cancel)
for rel in _restore(snapshot, base):
_say(f"[OK] Kept your local {rel} (upstream's changes to it were "
"skipped — merge new options by hand if needed).", emit)
return reset_rc
|
