sdlc-hub/scripts/cutover.py

497 lines
23 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

#!/usr/bin/env python3
"""Переключение живой установки хаба jira-hub -> sdlc-hub (одноразовый скрипт, stdlib, Python >= 3.9).
Без --apply только печатает проверки и шаги, ничего не меняя. С --apply выполняет шаги 2-7 спеки sdlc-rename
по порядку и останавливается на первой ошибке.
"""
from __future__ import annotations
import argparse
import json
import os
import shutil
import socket
import subprocess
import sys
import time
from dataclasses import dataclass, field
from pathlib import Path
from typing import Callable, List, Tuple
OLD_LABEL = "ru.aepif.jira-hub"
NEW_LABEL = "ru.aepif.sdlc-hub"
SUFFIX = ".pre-sdlc"
def _home() -> Path:
return Path(os.environ.get("HOME") or os.path.expanduser("~"))
@dataclass
class Config:
bridge: Path
data_root: Path
launch_agents: Path
logs: Path
remote: str = "" # пусто -> выводится из origin старого checkout заменой /aep/jira-hub на /aep/sdlc-hub
projects: List[Path] = field(default_factory=list)
port: int = 8790
projects_root: Path = None # где искать потребителей автоматически: <projects_root>/*/.mcp.json
@classmethod
def from_env(cls) -> "Config":
e = os.environ
h = _home()
projects = e.get("CUTOVER_PROJECTS")
return cls(
bridge=Path(e.get("CUTOVER_BRIDGE") or h / "projects" / "jira-bridge"),
data_root=Path(e.get("CUTOVER_DATA_ROOT") or h / ".local" / "share"),
launch_agents=Path(e.get("CUTOVER_LAUNCH_AGENTS") or h / "Library" / "LaunchAgents"),
logs=Path(e.get("CUTOVER_LOGS") or h / "Library" / "Logs"),
remote=e.get("CUTOVER_REMOTE") or "",
projects=[Path(p) for p in projects.split(":") if p] if projects is not None
else [h / "projects" / "awp-ca", h / "projects" / "stocks_drrk"],
port=int(e.get("CUTOVER_PORT") or 8790),
projects_root=Path(e.get("CUTOVER_PROJECTS_ROOT") or h / "projects"),
)
@property
def old_checkout(self) -> Path:
return self.bridge / "jira-hub"
@property
def new_checkout(self) -> Path:
return self.bridge / "sdlc-hub"
def _git_env() -> dict:
return dict(os.environ, GIT_OPTIONAL_LOCKS="0", GIT_TERMINAL_PROMPT="0")
def _git(repo: Path, *args: str) -> subprocess.CompletedProcess:
return subprocess.run(["git", "-C", str(repo), *args], capture_output=True, text=True, env=_git_env(),
stdin=subprocess.DEVNULL)
OLD_REPO = "/aep/jira-hub"
NEW_REPO = "/aep/sdlc-hub"
def derive_remote(origin: str) -> str:
"""Адрес нового репозитория из адреса origin: тот же протокол и суффикс, имя другое."""
if OLD_REPO not in origin:
raise ValueError("origin %s не содержит %s" % (origin, OLD_REPO))
return origin.replace(OLD_REPO, NEW_REPO)
def target_remote(cfg: Config) -> Tuple[str, str]:
"""(адрес, проблема): CUTOVER_REMOTE, иначе производный от origin старого checkout."""
if cfg.remote:
return cfg.remote, ""
r = _git(cfg.old_checkout, "remote", "get-url", "origin")
if r.returncode != 0:
return "", "не удалось прочитать origin в %s: %s" % (cfg.old_checkout, r.stderr.strip())
try:
return derive_remote(r.stdout.strip()), ""
except ValueError as e:
return "", "%s: задайте CUTOVER_REMOTE" % e
def _checkout_dirty(old: Path, porcelain: str) -> bool:
"""Чисто, если нет изменений; неотслеживаемая сборка src/jira_hub/ (без tracked-файлов) не считается."""
lines = [ln for ln in porcelain.splitlines() if ln.strip()]
if "?? src/jira_hub/" in lines and not _git(old, "ls-files", "src/jira_hub").stdout.strip():
lines.remove("?? src/jira_hub/")
return bool(lines)
def preflight(cfg: Config) -> List[str]:
"""Проверки шага 1. Только чтение."""
problems: List[str] = []
for tool in ("uv", "make", "claude", "launchctl", "node", "npm"):
if not shutil.which(tool):
problems.append("не найден инструмент %s в PATH (нужен для шагов переключения)" % tool)
old = cfg.old_checkout
remote = cfg.remote
if not old.is_dir():
problems.append("старый checkout %s не найден" % old)
else:
st = _git(old, "status", "--porcelain")
if st.returncode != 0:
problems.append("git status в %s не выполнился: %s" % (old, st.stderr.strip()))
elif _checkout_dirty(old, st.stdout):
problems.append("checkout %s не чист (есть незакоммиченные изменения)" % old)
br = _git(old, "rev-parse", "--abbrev-ref", "HEAD").stdout.strip()
if br != "main":
problems.append("checkout %s не на ветке main (сейчас: %s)" % (old, br or "?"))
head = _git(old, "rev-parse", "HEAD").stdout.strip()
# ls-remote вместо fetch: проверка остаётся строго чтением (fetch пишет FETCH_HEAD)
lr = _git(old, "ls-remote", "origin", "refs/heads/main")
remote_head = lr.stdout.split()[0] if lr.returncode == 0 and lr.stdout.split() else ""
if not remote_head:
problems.append("не удалось получить origin/main из %s" % old)
elif remote_head != head:
problems.append("HEAD не совпадает с origin/main: сделайте git pull --ff-only в %s" % old)
if not remote:
remote, prob = target_remote(cfg)
if prob:
problems.append(prob)
if remote:
lr = subprocess.run(["git", "ls-remote", remote], capture_output=True, text=True, env=_git_env(),
stdin=subprocess.DEVNULL)
if lr.returncode != 0:
problems.append("репозиторий %s недоступен: переименуйте aep/jira-hub -> aep/sdlc-hub в Gitea (%s)"
% (remote, lr.stderr.strip().splitlines()[-1] if lr.stderr.strip() else "нет ответа"))
for p in (cfg.data_root / "sdlc-hub", cfg.new_checkout):
if p.exists():
problems.append("каталог %s уже существует" % p)
if not (cfg.data_root / "jira-hub").exists():
problems.append("каталог данных %s не найден" % (cfg.data_root / "jira-hub"))
return problems
def _pgrep_mcp() -> List[str]:
"""Процессы MCP старого имени (сессии Claude). Нет pgrep - пропуск. Только чтение."""
if not shutil.which("pgrep"):
return []
r = subprocess.run(["pgrep", "-fl", "jira-hub mcp"], capture_output=True, text=True, stdin=subprocess.DEVNULL)
return [ln for ln in r.stdout.splitlines() if ln.strip()]
def warnings(cfg: Config) -> List[str]:
"""Предупреждения (не блокируют): открытые сессии Claude со старым MCP заново создадут каталог данных."""
procs = _pgrep_mcp()
if not procs:
return []
return ["работают процессы `jira-hub mcp` (закройте сессии Claude или после шага 3 они заново создадут "
"%s):" % (cfg.data_root / "jira-hub")] + [" " + p for p in procs]
def consumer_projects(cfg: Config) -> List[Path]:
"""Список проектов-потребителей: заданные плюс найденные <projects_root>/*/.mcp.json с `jira-hub`."""
out: List[Path] = list(cfg.projects)
root = cfg.projects_root
if root is not None and root.is_dir():
for mcp in sorted(root.glob("*/.mcp.json")):
proj = mcp.parent
if proj in out or proj == cfg.bridge or proj.resolve() == cfg.old_checkout.resolve():
continue
try:
if "jira-hub" in mcp.read_text(encoding="utf-8"):
out.append(proj)
except (OSError, UnicodeDecodeError):
continue
return out
def rewrite_mcp_json(path: Path, checkout: Path, unrewritten: List[str] = None) -> bool:
"""Переписывает записи jira-hub/jira-hub-wiki. Запись нестандартной формы (command не uv или нет маркера
mcp/jira-hub в args) не меняется, её описание добавляется в `unrewritten`."""
data = json.loads(path.read_text(encoding="utf-8"))
servers = data.get("mcpServers")
if not isinstance(servers, dict):
return False
names = {"jira-hub": ("sdlc-jira", "jira"), "jira-hub-wiki": ("sdlc-wiki", "wiki")}
changed = False
new: dict = {}
for key, val in servers.items():
if key not in names:
new[key] = val
continue
new_key, module = names[key]
args = list(val.get("args", []))
marker = next((m for m in ("mcp", "jira-hub") if m in args), None)
if os.path.basename(str(val.get("command", ""))) != "uv" or marker is None:
new[key] = val
if unrewritten is not None:
unrewritten.append("%s: запись %s нестандартной формы, не переписана" % (path, key))
continue
tail = args[args.index(marker) + 1:]
if tail and tail[0] in ("jira", "wiki"):
tail = tail[1:]
val = dict(val)
val["args"] = ["run", "--project", str(checkout), "sdlc-hub", "mcp", module] + tail
new[new_key] = val
changed = True
if changed:
data["mcpServers"] = new
path.write_text(json.dumps(data, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
return changed
_REPLACEMENTS = [
("mcp__jira-hub-wiki__", "mcp__sdlc-wiki__"),
("mcp__jira-hub__", "mcp__sdlc-jira__"),
("aep/jira-hub", "aep/sdlc-hub"),
('"jira-hub-wiki"', '"sdlc-wiki"'),
('"jira-hub"', '"sdlc-jira"'),
("MCP `jira-hub`", "MCP `sdlc-jira`"),
("MCP jira-hub", "MCP sdlc-jira"),
]
def rewrite_text(path: Path) -> List[str]:
"""Замены имён серверов и инструментов; возвращает строки с остатками `jira-hub`."""
try:
text = path.read_text(encoding="utf-8")
except (UnicodeDecodeError, OSError):
return []
out = text
for a, b in _REPLACEMENTS:
out = out.replace(a, b)
if out != text:
path.write_text(out, encoding="utf-8")
return [ln.strip() for ln in out.splitlines() if "jira-hub" in ln]
def backup(path: Path) -> Path:
dst = path.with_name(path.name + SUFFIX)
shutil.copy2(str(path), str(dst))
return dst
def plist_text(checkout: Path, log: Path, uv: str) -> str:
args = "".join("<string>%s</string>" % a for a in [uv, "run", "--project", str(checkout), "sdlc-hub", "serve"])
return (
'<?xml version="1.0" encoding="UTF-8"?>\n'
'<!DOCTYPE plist PUBLIC "-//Apple//DTD PLIST 1.0//EN" "http://www.apple.com/DTDs/PropertyList-1.0.dtd">\n'
'<plist version="1.0"><dict>\n'
" <key>Label</key><string>%s</string>\n"
" <key>ProgramArguments</key><array>%s</array>\n"
" <key>EnvironmentVariables</key><dict>\n"
" <key>PATH</key><string>/opt/homebrew/bin:/usr/local/bin:/usr/bin:/bin</string>\n"
" <key>PYTHONUNBUFFERED</key><string>1</string>\n"
" </dict>\n"
" <key>RunAtLoad</key><true/>\n"
" <key>KeepAlive</key><true/>\n"
" <key>StandardOutPath</key><string>%s</string>\n"
" <key>StandardErrorPath</key><string>%s</string>\n"
"</dict></plist>\n" % (NEW_LABEL, args, log, log)
)
def _run(*cmd: str, cwd: Path = None, check: bool = True) -> subprocess.CompletedProcess:
return subprocess.run(list(cmd), cwd=str(cwd) if cwd else None, check=check)
def _port_open(port: int) -> bool:
try:
socket.create_connection(("127.0.0.1", port), timeout=1).close()
return True
except OSError:
return False
_sleep = time.sleep
_clock = time.monotonic
def _wait_port_closed(port: int, timeout: float = 10.0, interval: float = 0.5) -> bool:
deadline = _clock() + timeout
while _port_open(port):
if _clock() >= deadline:
return False
_sleep(interval)
return True
def _domain() -> str:
return "gui/%d" % os.getuid()
def _project_files(project: Path) -> List[Path]:
files: List[Path] = []
claude = project / ".claude"
files += sorted(claude.glob("settings*.json")) if claude.is_dir() else []
if (project / "CLAUDE.md").is_file():
files.append(project / "CLAUDE.md")
skills = claude / "skills"
if skills.is_dir():
files += [p for p in sorted(skills.rglob("*")) if p.is_file() and not p.name.endswith(SUFFIX)]
return files
def steps(cfg: Config, leftovers: List[str] = None, backups: List[Path] = None) -> List[Tuple[str, Callable[[], None]]]:
leftovers = leftovers if leftovers is not None else []
backups = backups if backups is not None else []
new_co = cfg.new_checkout
old_plist = cfg.launch_agents / (OLD_LABEL + ".plist")
kept_plist = old_plist.with_name(old_plist.name + SUFFIX)
new_plist = cfg.launch_agents / (NEW_LABEL + ".plist")
log = cfg.logs / "sdlc-hub.log"
uv = shutil.which("uv") or "/opt/homebrew/bin/uv"
remote_shown = cfg.remote or "производный от origin"
def stop() -> None:
_run("launchctl", "bootout", "%s/%s" % (_domain(), OLD_LABEL), check=False) # не загружен - не ошибка
if not _wait_port_closed(cfg.port):
raise RuntimeError("после остановки %s порт %d всё ещё занят: хаб не остановлен" % (OLD_LABEL, cfg.port))
if old_plist.exists():
shutil.move(str(old_plist), str(kept_plist))
def move_data() -> None:
shutil.move(str(cfg.data_root / "jira-hub"), str(cfg.data_root / "sdlc-hub"))
def move_checkout() -> None:
remote = cfg.remote
if not remote:
remote, prob = target_remote(cfg)
if prob:
raise RuntimeError(prob)
shutil.move(str(cfg.old_checkout), str(new_co))
old_src = new_co / "src" / "jira_hub"
if old_src.is_dir() and not _git(new_co, "ls-files", "src/jira_hub").stdout.strip():
shutil.rmtree(str(old_src)) # неотслеживаемая прежняя сборка web/app
wts = sorted(p for p in (new_co / ".wt").glob("*") if p.is_dir()) if (new_co / ".wt").is_dir() else []
_run("git", "-C", str(new_co), "worktree", "repair", *[str(p) for p in wts])
_run("git", "-C", str(new_co), "remote", "set-url", "origin", remote)
shutil.rmtree(str(new_co / ".venv"), ignore_errors=True)
_run("uv", "sync", cwd=new_co)
_run("make", "web", cwd=new_co)
def start() -> None:
cfg.launch_agents.mkdir(parents=True, exist_ok=True)
cfg.logs.mkdir(parents=True, exist_ok=True)
new_plist.write_text(plist_text(new_co, log, uv), encoding="utf-8")
_run("launchctl", "bootstrap", _domain(), str(new_plist))
deadline = _clock() + 30
while _clock() < deadline:
if _port_open(cfg.port):
return
_sleep(1)
raise RuntimeError("хаб не ответил на порту %d за 30 с (лог %s); остановить: launchctl bootout %s/%s"
% (cfg.port, log, _domain(), NEW_LABEL))
def consumers() -> None:
for project in consumer_projects(cfg):
mcp = project / ".mcp.json"
if mcp.is_file() and "jira-hub" in mcp.read_text(encoding="utf-8"):
backups.append(backup(mcp))
rewrite_mcp_json(mcp, new_co, leftovers)
leftovers.extend("%s: %s" % (mcp, ln.strip())
for ln in mcp.read_text(encoding="utf-8").splitlines() if "jira-hub" in ln)
for f in _project_files(project):
try:
if "jira-hub" not in f.read_text(encoding="utf-8"):
continue
except (UnicodeDecodeError, OSError):
continue
backups.append(backup(f))
leftovers.extend("%s: %s" % (f, ln) for ln in rewrite_text(f))
def user_scope() -> None:
_run("claude", "mcp", "add", "-s", "user", "sdlc-jira", "--",
"uv", "run", "--project", str(new_co), "sdlc-hub", "mcp", "jira")
r = _run("claude", "mcp", "remove", "-s", "user", "jira-hub", check=False)
if r.returncode != 0:
print(" ПРЕДУПРЕЖДЕНИЕ: claude mcp remove -s user jira-hub не выполнился (код %d): удалите запись руками"
% r.returncode)
return [
("2. остановить launchd %s, дождаться закрытия порта, переименовать %s в %s"
% (OLD_LABEL, old_plist.name, kept_plist.name), stop),
("3. перенести данные %s -> %s" % (cfg.data_root / "jira-hub", cfg.data_root / "sdlc-hub"), move_data),
("4. перенести checkout %s -> %s, убрать неотслеживаемый src/jira_hub, worktree repair, origin -> %s, "
"uv sync, make web" % (cfg.old_checkout, new_co, remote_shown), move_checkout),
("5. поставить и запустить %s (лог %s), дождаться порта %d" % (NEW_LABEL, log, cfg.port), start),
("6. править проекты %s (.mcp.json, .claude/settings*.json, CLAUDE.md, .claude/skills; копии *%s)"
% (", ".join(str(p) for p in consumer_projects(cfg)), SUFFIX), consumers),
("7. user scope Claude: add sdlc-jira, затем remove jira-hub", user_scope),
]
def main(argv: List[str]) -> int:
ap = argparse.ArgumentParser(description="Переключение живого хаба jira-hub -> sdlc-hub")
ap.add_argument("--apply", action="store_true", help="выполнить (по умолчанию сухой прогон)")
args = ap.parse_args(argv)
cfg = Config.from_env()
problems = preflight(cfg)
print("1. проверки:")
if problems:
for p in problems:
print(" ПРОБЛЕМА: " + p)
else:
print(" ок")
for w in warnings(cfg):
print(" ПРЕДУПРЕЖДЕНИЕ: " + w if not w.startswith(" ") else w)
leftovers: List[str] = []
backups: List[Path] = []
plan = steps(cfg, leftovers, backups)
if not args.apply:
print("Сухой прогон, ничего не изменено. Шаги:")
for desc, _ in plan:
print(" " + desc)
return 1 if problems else 0
if problems:
print("Остановлено: исправьте проблемы шага 1, ничего не изменено.")
return 1
done: List[str] = []
for desc, fn in plan:
print("-> " + desc)
try:
fn()
except Exception as e: # noqa: BLE001
print("остановлено на шаге %s: %s" % (desc.split(".")[0], e))
print("Выполнено: " + ("; ".join(done) if done else "ничего"))
print("Промежуточное состояние (по выполненным шагам; упавший шаг мог выполниться частично, проверьте):")
for line in _state_lines(done, desc):
print(" " + line)
old_plist = cfg.launch_agents / (OLD_LABEL + ".plist")
print("Откат: остановить ru.aepif.sdlc-hub (launchctl bootout %s/%s), перенести каталоги обратно "
"(данные %s -> %s, checkout %s -> %s, origin вернуть на aep/jira-hub), вернуть plist "
"(mv %s %s) и launchctl bootstrap %s %s; раздел «Переключение с jira-hub» в README." % (
_domain(), NEW_LABEL, cfg.data_root / "sdlc-hub", cfg.data_root / "jira-hub",
cfg.new_checkout, cfg.old_checkout, str(old_plist) + SUFFIX, old_plist, _domain(), old_plist))
_print_tail(cfg, done, backups, leftovers)
return 1
done.append(desc)
print(" готово")
print("\nОсталось владельцу:")
print(" - /mcp reconnect в открытых сессиях Claude")
print(" - codegraph init в %s и в каждом worktree (.wt/*)" % cfg.new_checkout)
print(" - uv sync в каждом worktree (.wt/*): их .venv содержат старые абсолютные пути")
print(" - после запуска проверить лог реле (%s) и agent_status; при «Operation not permitted» выдать python "
"из нового .venv (%s) полный доступ к диску заново" % (cfg.logs / "sdlc-hub.log", cfg.new_checkout / ".venv"))
print(" - скрипт лучше запускать из обычного терминала вне checkout (checkout переезжает во время работы)")
_print_tail(cfg, done, backups, leftovers)
return 0
def _state_lines(done: List[str], failed: str = "") -> List[str]:
n = {d.split(".")[0] for d in done}
out = []
if "2" in n:
out.append("хаб ru.aepif.jira-hub остановлен, его plist переименован в *%s" % SUFFIX)
if "3" in n:
out.append("каталог данных перенесён в sdlc-hub")
if "4" in n:
out.append("checkout перенесён в sdlc-hub, origin переключён")
if "5" in n:
out.append("ru.aepif.sdlc-hub установлен и запущен")
if "6" in n:
out.append("файлы проектов изменены (копии *%s)" % SUFFIX)
if failed:
if failed.startswith("2."):
out.append("упавший шаг 2: хаб ru.aepif.jira-hub мог быть выгружен; его plist мог быть переименован в *%s"
% SUFFIX)
else:
out.append("упавший шаг %s мог выполниться частично, проверьте" % failed.split(".")[0])
return out or ["ничего не изменено"]
def _print_tail(cfg: Config, done: List[str], backups: List[Path], leftovers: List[str]) -> None:
stale = cfg.data_root / "jira-hub"
if any(d.startswith("3.") for d in done) and stale.exists():
print(" - ПРЕДУПРЕЖДЕНИЕ: %s появился заново (сессия Claude со старым MCP): удалите его перед откатом mv" % stale)
print(" - проверить правки и удалить копии *%s:" % SUFFIX)
for b in backups:
print(" " + str(b))
if leftovers:
print(" - проверить руками (остались упоминания jira-hub или запись не переписана):")
for ln in leftovers:
print(" " + ln)
if __name__ == "__main__":
sys.exit(main(sys.argv[1:]))