"""記事C 最小実験：バックアップからの復元演習（架空データ限定）— 改訂版 rev2（2026-09-19） 演習は「演習領域」の中だけで動く。演習領域は既定で examples/sample-c/work/。 環境変数 DRILL_WORK_DIR で別の場所を指定できるが、examples/sample-c の直下で名前が work で始まるフォルダに限る。 本番・共有・外部のデータベースには一切接続しない。 安全のための約束（コードで強制している範囲。共通の検査は drill_safety.py）: - 信頼する基点は examples/sample-c だけ。演習領域はその直下の work* フォルダだけ。 - setup は、新しいフォルダか、目印のない空のフォルダか、内容つきの目印が一致するフォルダにしか演習領域を作らない。 無関係なファイルがある既存フォルダには、目印を付けず、何も変更せずに拒否する。 - reset は「--confirm」を付け、かつ内容つきの目印（種別と実パス）が一致する場合しか削除しない。 空の旧形式の目印は無効として扱い、削除の許可にしない。 - 読取り元・書込み先・削除先は、親フォルダだけでなく最終ファイルまで実際の場所を検査する。 演習領域以下のシンボリックリンク（リンク先がまだ無いものを含む）は拒否する。 - backup は世代フォルダ（gen-001, gen-002 ...）を新しく作る。既存の世代は上書きしない。 - restore は復元元の指紋（SHA-256）を先に照合し、復元先が既にあれば（リンクを含め）拒否する。既存の復元物を消さない。 - verify は不一致があれば終了コード 1 を返す。レポートは不一致でも書き出す。 前提は「単独実行・架空データ・通常のローカルファイル」。目印やコードを故意に書き換える利用者や、 別プロセスによる差替えは対象外。 注意: setup／after-backup／break は演習領域内の practice.db を変更する。「元データを上書きしない」のは 復元（restore）についての約束であり、演習の全手順が practice.db を変えないという意味ではない。 使い方（このフォルダで、番号順に実行）: python3 drill.py reset --confirm # 演習領域を作り直す（やり直し用。目印ファイルがある領域だけ削除） python3 drill.py setup # 架空の依頼・履歴データを作る python3 drill.py backup # 新しい世代にバックアップを取り、バックアップ自体から基準の控えを作る python3 drill.py after-backup # バックアップ後の業務変更を起こす（新規依頼・承認） python3 drill.py break # 障害を起こす：誤った掃除スクリプト v1.1 を実行する python3 drill.py freeze # 壊れた状態を別名で保全する python3 drill.py rollback-code # 掃除スクリプトを既知版 v1.0 に戻して再実行 → データは戻らないことを見る python3 drill.py restore [gen] # 指定世代（1つしかなければ省略可）を別の場所へ復元する python3 drill.py verify [gen] # 復元結果を基準と照合する。不一致なら終了コード 1 """ import hashlib import json import os import shutil import sqlite3 import sys from collections import Counter from datetime import datetime, timezone import drill_safety as safety from drill_safety import SafetyError HERE = os.path.dirname(os.path.abspath(__file__)) try: WORK = safety.resolve_work() except SafetyError as _e: print(f"中止: {_e}", flush=True) sys.exit(2) MARKER = safety.marker_path(WORK) DB = os.path.join(WORK, "practice.db") BACKUP_ROOT = os.path.join(WORK, "backup") INCIDENT_DIR = os.path.join(WORK, "incident") RESTORE_ROOT = os.path.join(WORK, "restore") SCHEMA = """ CREATE TABLE requests ( id TEXT PRIMARY KEY, requester_id TEXT NOT NULL, title TEXT NOT NULL, state TEXT NOT NULL, approver_id TEXT ); CREATE TABLE history ( seq INTEGER PRIMARY KEY AUTOINCREMENT, request_id TEXT NOT NULL, action TEXT NOT NULL, actor_id TEXT NOT NULL, at TEXT NOT NULL, note TEXT ); """ REQUEST_COLS = ["id", "requester_id", "title", "state", "approver_id"] HISTORY_COLS = ["request_id", "action", "actor_id", "at", "note"] # seq は複製で保たれるが照合は業務列で行う # ---------------------------------------------------------------- 共通 def now(): return datetime.now(timezone.utc).isoformat(timespec="seconds") def log(msg): print(f"[{now()}] {msg}", flush=True) def fail(msg, code=1): print(f"中止: {msg}", flush=True) sys.exit(code) def _file(path, what, must_exist=None): """演習領域内の通常ファイルとして検査して返す（最終ファイルまで実パス・リンクを確認）。""" try: return safety.check_inside(WORK, path, what, must_exist=must_exist) except SafetyError as e: fail(str(e)) def _dir(path, what): try: return safety.check_dir_inside(WORK, path, what) except SafetyError as e: fail(str(e)) def _require_initialized(): try: safety.require_initialized(WORK) except SafetyError as e: fail(str(e)) def sha256(path): h = hashlib.sha256() with open(path, "rb") as f: for chunk in iter(lambda: f.read(65536), b""): h.update(chunk) return h.hexdigest() def open_ro(path, what="読取り元"): """読取り専用で開く。復元結果や保全ファイルを誤って変更しないため。開く前に場所を検査する。""" path = _file(path, what, must_exist=True) uri = "file:" + path.replace("\\", "/") + "?mode=ro" return sqlite3.connect(uri, uri=True) def snapshot(conn): """件数・内容・関係の照合に使う要約を作る。行は業務列をすべて含め、重複行も数えられるようリストで持つ。""" reqs = conn.execute( "SELECT id, requester_id, title, state, approver_id FROM requests ORDER BY id" ).fetchall() hist = conn.execute( "SELECT request_id, action, actor_id, at, note FROM history ORDER BY seq" ).fetchall() orphan = conn.execute( "SELECT COUNT(*) FROM history h LEFT JOIN requests r ON r.id = h.request_id WHERE r.id IS NULL" ).fetchone()[0] approved_without_history = conn.execute( "SELECT COUNT(*) FROM requests r WHERE r.state='承認済み' AND NOT EXISTS (" " SELECT 1 FROM history h WHERE h.request_id=r.id AND h.action='承認')" ).fetchone()[0] return { "requests_count": len(reqs), "history_count": len(hist), "requests": [dict(zip(REQUEST_COLS, r)) for r in reqs], "history": [dict(zip(HISTORY_COLS, h)) for h in hist], "orphan_history": orphan, "approved_without_history": approved_without_history, } def print_summary(label, snap): print(f"--- {label} ---") print(f" 依頼 {snap['requests_count']} 件 / 履歴 {snap['history_count']} 件" f" / 依頼のない履歴 {snap['orphan_history']} 件 / 承認履歴のない承認済み {snap['approved_without_history']} 件") for r in snap["requests"]: print(f" {r['id']} {r['state']:4s} {r['title']}") def _generations(): if not os.path.isdir(BACKUP_ROOT) or os.path.islink(BACKUP_ROOT): return [] return sorted(d for d in os.listdir(BACKUP_ROOT) if d.startswith("gen-") and os.path.isdir(os.path.join(BACKUP_ROOT, d)) and not os.path.islink(os.path.join(BACKUP_ROOT, d))) def _pick_generation(arg, purpose): gens = _generations() if not gens: fail("バックアップの世代がない。先に backup を実行する") if arg is None: if len(gens) == 1: return gens[0] fail(f"世代が複数ある（{', '.join(gens)}）。{purpose} する世代を引数で指定する（例: 1）") name = arg if arg.startswith("gen-") else f"gen-{int(arg):03d}" if name not in gens: fail(f"世代 {name} は存在しない。ある世代: {', '.join(gens)}") return name # ---------------------------------------------------------------- 各手順 def cmd_reset(args): if "--confirm" not in args: fail("reset は演習領域を削除する。実行するなら --confirm を付ける", code=2) if os.path.lexists(WORK): if os.path.islink(WORK) or not os.path.isdir(WORK): fail(f"演習領域がフォルダでない（リンク等）。削除しない: {WORK}", code=2) if not safety.marker_valid(WORK): fail(f"目印ファイル {MARKER} が無いか、内容がこの演習領域と一致しない。演習領域と確認できないため削除しない", code=2) shutil.rmtree(WORK) os.makedirs(WORK) safety.write_marker(WORK) log(f"演習領域を初期化した: {WORK}") def cmd_setup(args): if os.path.lexists(WORK): if os.path.islink(WORK) or not os.path.isdir(WORK): fail(f"演習領域がフォルダでない（リンク等）: {WORK}") if safety.marker_valid(WORK): pass # 初期化済みの演習領域 elif os.path.lexists(MARKER): fail(f"目印ファイル {MARKER} の内容がこの演習領域と一致しない（空の旧形式を含む）。何も変更しない") elif os.listdir(WORK): fail(f"演習領域にしようとしたフォルダに既にファイルがある。無関係なフォルダの可能性があるため、目印を付けず何も変更しない: {WORK}") else: safety.write_marker(WORK) else: os.makedirs(WORK) safety.write_marker(WORK) if os.path.lexists(DB): fail("practice.db が既にある。やり直すなら先に reset --confirm を実行する") _file(DB, "practice.db", must_exist=False) conn = sqlite3.connect(DB) conn.executescript(SCHEMA) conn.executemany("INSERT INTO requests VALUES (?,?,?,?,?)", [ ("REQ-C01", "u-sato", "会議室プロジェクタ手配", "承認済み", "u-suzuki"), ("REQ-C02", "u-sato", "名刺の追加発注", "提出済み", "u-suzuki"), ("REQ-C03", "u-tanaka", "備品棚の移設", "差戻し", "u-takahashi"), ("REQ-C04", "u-tanaka", "社用車の予約", "承認済み", "u-takahashi"), ("REQ-C05", "u-sato", "来客用駐車場の確保", "下書き", "u-suzuki"), ]) conn.executemany("INSERT INTO history (request_id, action, actor_id, at, note) VALUES (?,?,?,?,?)", [ ("REQ-C01", "提出", "u-sato", "2026-09-08T01:00:00+00:00", None), ("REQ-C01", "承認", "u-suzuki", "2026-09-08T03:00:00+00:00", None), ("REQ-C02", "提出", "u-sato", "2026-09-09T01:00:00+00:00", None), ("REQ-C03", "提出", "u-tanaka", "2026-09-09T02:00:00+00:00", None), ("REQ-C03", "差戻し", "u-takahashi", "2026-09-09T04:00:00+00:00", "移設先の階数が未記入"), ("REQ-C04", "提出", "u-tanaka", "2026-09-10T01:00:00+00:00", None), ("REQ-C04", "承認", "u-takahashi", "2026-09-10T02:00:00+00:00", None), ]) conn.commit() print_summary("初期状態（架空データ）", snapshot(conn)) conn.close() log(f"practice.db を作成した: {DB}") def cmd_backup(args): _require_initialized() if not os.path.lexists(DB): fail("practice.db がない。先に setup を実行する") _file(DB, "practice.db", must_exist=True) _dir(BACKUP_ROOT, "バックアップ置き場") gens = _generations() gen = f"gen-{len(gens) + 1:03d}" gen_dir = os.path.join(BACKUP_ROOT, gen) if os.path.lexists(gen_dir): fail(f"世代フォルダ {gen_dir} が既にある。上書きしない") _dir(gen_dir, "バックアップ世代フォルダ") os.makedirs(gen_dir) backup_db = _file(os.path.join(gen_dir, "practice_backup.db"), "バックアップファイル", must_exist=False) manifest_path = _file(os.path.join(gen_dir, "manifest.json"), "基準の控え（manifest）", must_exist=False) # 演習の前提：この時点で practice.db を更新する処理は動いておらず、接続は閉じている。 src = sqlite3.connect(DB) dst = sqlite3.connect(backup_db) with dst: src.backup(dst) # SQLite のオンラインバックアップ機能でファイル全体を複製する src.close() dst.close() # 基準の控えは、元DBではなく「バックアップファイルそのもの」から作る。 ro = open_ro(backup_db, "バックアップファイル") baseline = snapshot(ro) ro.close() manifest = { "generation": gen, "taken_at": now(), "source": os.path.relpath(DB, WORK), "backup_file": os.path.relpath(backup_db, WORK), "sha256": sha256(backup_db), "baseline_from": "backup_file", "baseline": baseline, } with open(manifest_path, "w", encoding="utf-8") as f: json.dump(manifest, f, ensure_ascii=False, indent=2) log(f"バックアップ世代 {gen} を作成した: {backup_db}") log(f"基準の控え（バックアップ自体から取得）: 依頼 {baseline['requests_count']} 件 / 履歴 {baseline['history_count']} 件 / sha256 {manifest['sha256'][:16]}…") if gens: log(f"注意: 既存の世代 {', '.join(gens)} はそのまま残っている。restore/verify では世代を明示する") def cmd_after_backup(args): _require_initialized() _file(DB, "practice.db", must_exist=True) # 接続前に、リンクでなく演習領域内の通常ファイルであることを確認 conn = sqlite3.connect(DB) with conn: conn.execute("INSERT INTO requests VALUES ('REQ-C06','u-tanaka','会議用弁当の手配','提出済み','u-suzuki')") conn.execute("INSERT INTO history (request_id, action, actor_id, at) VALUES ('REQ-C06','提出','u-tanaka',?)", (now(),)) conn.execute("UPDATE requests SET state='承認済み' WHERE id='REQ-C02' AND state='提出済み'") conn.execute("INSERT INTO history (request_id, action, actor_id, at) VALUES ('REQ-C02','承認','u-suzuki',?)", (now(),)) print_summary("バックアップ後の業務変更を反映した状態", snapshot(conn)) conn.close() log("バックアップ後に、REQ-C06 の提出と REQ-C02 の承認が起きた（架空）") def cmd_break(args): _require_initialized() import cleanup_v1_1 log("誤った掃除スクリプト cleanup_v1_1 を実行する（架空の障害の引き金。演習領域内の practice.db だけが対象）") try: cleanup_v1_1.run(DB) except SafetyError as e: fail(str(e)) conn = open_ro(DB, "practice.db") print_summary("障害後の状態", snapshot(conn)) conn.close() def cmd_freeze(args): _require_initialized() _file(DB, "practice.db", must_exist=True) _dir(INCIDENT_DIR, "保全置き場") os.makedirs(INCIDENT_DIR, exist_ok=True) stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") dest = _file(os.path.join(INCIDENT_DIR, f"practice_broken_{stamp}.db"), "保全先", must_exist=False) shutil.copy2(DB, dest) log(f"壊れた状態を別名で保全した（元ファイルは変更しない）: {dest}") log(f"保全ファイル sha256 {sha256(dest)[:16]}…") def cmd_rollback_code(args): _require_initialized() import cleanup_v1_0 log("掃除スクリプトを既知版 cleanup_v1_0 に戻して再実行する（コードの切戻し）") try: cleanup_v1_0.run(DB) except SafetyError as e: fail(str(e)) conn = open_ro(DB, "practice.db") print_summary("コード切戻し後の状態（データは戻っていないはず）", snapshot(conn)) conn.close() def cmd_restore(args): _require_initialized() gen = _pick_generation(args[0] if args else None, "復元") gen_dir = os.path.join(BACKUP_ROOT, gen) backup_db = os.path.join(gen_dir, "practice_backup.db") manifest_path = os.path.join(gen_dir, "manifest.json") restore_dir = os.path.join(RESTORE_ROOT, gen) restored_db = os.path.join(restore_dir, "practice_restored.db") record_path = os.path.join(restore_dir, "restore_record.json") # 1. 復元先が既にあれば拒否する（既存の復元物を消さない。リンク先の無いリンクも「既にある」として扱う） _dir(RESTORE_ROOT, "復元置き場") _dir(restore_dir, "復元先フォルダ") if os.path.lexists(restored_db): fail(f"復元先 {restored_db} が既にある（リンクを含む）。既存の復元物は消さない。別の世代を使うか、演習領域を reset --confirm で作り直す") _file(record_path, "復元記録", must_exist=False) # 2. 復元元を検査する（復元先に触れる前） if not os.path.lexists(manifest_path) or not os.path.lexists(backup_db): fail(f"世代 {gen} のバックアップまたは控えがない") _file(manifest_path, "基準の控え（manifest）", must_exist=True) _file(backup_db, "復元元（バックアップファイル）", must_exist=True) with open(manifest_path, encoding="utf-8") as f: manifest = json.load(f) actual = sha256(backup_db) if actual != manifest["sha256"]: fail(f"復元元（バックアップ）の指紋が控えと一致しない。復元を中止する。控え={manifest['sha256']} 実際={actual}") log(f"復元元（世代 {gen} のバックアップファイル）の指紋が控えと一致した") try: ro = open_ro(backup_db, "復元元（バックアップファイル）") ro.execute("SELECT COUNT(*) FROM requests").fetchone() ro.close() except sqlite3.Error as e: fail(f"復元元がデータベースとして読めない: {e}") # 3. 別の場所へ複製する（コピー直前に復元先を最終ファイルまで再検査） os.makedirs(restore_dir, exist_ok=True) _file(restored_db, "復元先", must_exist=False) shutil.copy2(backup_db, restored_db) restored_hash = sha256(restored_db) record = { "generation": gen, "restored_at": now(), "restored_file": os.path.relpath(restored_db, WORK), "source_sha256": actual, "restored_sha256": restored_hash, "source_matches_restored": restored_hash == actual, } with open(record_path, "w", encoding="utf-8") as f: json.dump(record, f, ensure_ascii=False, indent=2) log(f"別の場所へ復元した（元の practice.db は上書きしていない）: {restored_db}") log(f"復元後のファイルの指紋 {restored_hash[:16]}…（復元元と{'一致' if record['source_matches_restored'] else '不一致'}）") def _compare_rows(label, base_rows, cur_rows, cols, key_cols): """行の内容を照合する。重複行を消さないよう Counter（個数付きの集合）で比べる。""" problems = [] base_c = Counter(tuple(r[c] for c in cols) for r in base_rows) cur_c = Counter(tuple(r[c] for c in cols) for r in cur_rows) missing = base_c - cur_c # 基準にあって復元結果にない extra = cur_c - base_c # 復元結果にあって基準にない for row, n in missing.items(): problems.append(f"{label}: 基準にあるが復元結果にない ×{n}: {dict(zip(cols, row))}") for row, n in extra.items(): problems.append(f"{label}: 復元結果にあるが基準にない ×{n}: {dict(zip(cols, row))}") return problems def cmd_verify(args): _require_initialized() gen = _pick_generation(args[0] if args else None, "照合") gen_dir = os.path.join(BACKUP_ROOT, gen) manifest_path = os.path.join(gen_dir, "manifest.json") restore_dir = os.path.join(RESTORE_ROOT, gen) restored_db = os.path.join(restore_dir, "practice_restored.db") record_path = os.path.join(restore_dir, "restore_record.json") report_path = os.path.join(restore_dir, "verify_report.md") if not os.path.lexists(restored_db): fail(f"世代 {gen} の復元結果がない。先に restore を実行する") _file(restored_db, "復元結果", must_exist=True) _file(manifest_path, "基準の控え（manifest）", must_exist=True) with open(manifest_path, encoding="utf-8") as f: manifest = json.load(f) base = manifest["baseline"] record = {} if os.path.lexists(record_path): _file(record_path, "復元記録", must_exist=True) record = json.load(open(record_path, encoding="utf-8")) checks = [] # (項目, 基準, 実際, 合否) problems = [] # A. ファイルの指紋（復元後のファイルが復元記録と一致するか） cur_hash = sha256(restored_db) if record: checks.append(("復元後ファイルの指紋", record["restored_sha256"][:16] + "…", cur_hash[:16] + "…", cur_hash == record["restored_sha256"])) # B. データベースとして読めるか try: ro = open_ro(restored_db, "復元結果") cur = snapshot(ro) ro.close() checks.append(("データベースとして読める", "読める", "読める", True)) except sqlite3.Error as e: checks.append(("データベースとして読める", "読める", f"読めない: {e}", False)) cur = None if cur is not None: # C. 件数 checks.append(("依頼の件数", base["requests_count"], cur["requests_count"], base["requests_count"] == cur["requests_count"])) checks.append(("履歴の件数", base["history_count"], cur["history_count"], base["history_count"] == cur["history_count"])) # D. 内容（全業務列） p_req = _compare_rows("依頼", base["requests"], cur["requests"], REQUEST_COLS, ["id"]) p_his = _compare_rows("履歴", base["history"], cur["history"], HISTORY_COLS, []) checks.append(("依頼の内容（ID・依頼者・件名・状態・承認者）", f"{base['requests_count']} 行", f"不一致 {len(p_req)} 件", not p_req)) checks.append(("履歴の内容（依頼ID・操作・操作者・時刻・備考）", f"{base['history_count']} 行", f"不一致 {len(p_his)} 件", not p_his)) problems += p_req + p_his # E. 関係と不変条件 checks.append(("依頼のない履歴（関係の破れ）", 0, cur["orphan_history"], cur["orphan_history"] == 0)) checks.append(("承認履歴のない承認済み（不変条件）", 0, cur["approved_without_history"], cur["approved_without_history"] == 0)) all_ok = all(c[3] for c in checks) lines = ["# 復元確認レポート（自動生成）", ""] lines.append(f"- 生成日時: {now()}") lines.append(f"- 世代: {gen}　基準時点（バックアップ取得時刻）: {manifest['taken_at']}") lines.append(f"- 基準の控えの出所: {manifest.get('baseline_from', '不明')}（バックアップファイル自体から取得）") lines.append(f"- 復元先: {os.path.relpath(restored_db, WORK)}") lines.append("") lines.append("## 1. 基準時点との照合（指紋・可読性・件数・内容・関係）") lines.append("") lines.append("| 項目 | 基準 | 復元結果 | 判定 |") lines.append("|---|---|---|---|") for name, b, c, ok in checks: lines.append(f"| {name} | {b} | {c} | {'一致' if ok else '不一致'} |") if problems: lines.append("") lines.append("不一致の明細:") lines.append("") for p in problems: lines.append(f"- {p}") lines.append("") lines.append("指紋の一致・データベースとして読めること・件数の一致・内容の一致・関係の整合は、それぞれ別の確認である。") lines.append("指紋が一致しても、控えが誤った時点のものなら業務要件への適合は示せない。") lines.append("") # F. 基準時点より後の変更（復元成功とは別の判断） lines.append("## 2. 基準時点より後に起きて、復元では戻らない変更（復元成功とは別の判断）") lines.append("") broken_files = sorted(os.listdir(INCIDENT_DIR)) if (os.path.isdir(INCIDENT_DIR) and not os.path.islink(INCIDENT_DIR)) else [] if cur is not None and broken_files: bro = open_ro(os.path.join(INCIDENT_DIR, broken_files[-1]), "保全ファイル") bs = snapshot(bro) bro.close() lines.append(f"比較対象（保全した障害時のファイル）: {broken_files[-1]}") lines.append("") lines.append("| 対象 | 障害時のファイルにある内容 | 復元結果 | 扱い |") lines.append("|---|---|---|---|") r_ids = {r["id"] for r in cur["requests"]} for r in bs["requests"]: if r["id"] not in r_ids: lines.append(f"| 依頼 {r['id']}（{r['title']}） | 存在（{r['state']}） | 存在しない | 再入力または再提出が必要 |") else: rstate = next(x["state"] for x in cur["requests"] if x["id"] == r["id"]) if rstate != r["state"]: lines.append(f"| 依頼 {r['id']} の状態 | {r['state']} | {rstate} | 基準時点以降の操作を再実施するか判断 |") cur_hist = Counter(tuple(h[c] for c in HISTORY_COLS) for h in cur["history"]) bro_hist = Counter(tuple(h[c] for c in HISTORY_COLS) for h in bs["history"]) for row, n in (bro_hist - cur_hist).items(): h = dict(zip(HISTORY_COLS, row)) lines.append(f"| 履歴 {h['request_id']} {h['action']}（{h['actor_id']}） ×{n} | 存在 | 存在しない | 再操作で再生成するか、記録として扱う |") lines.append("") lines.append("注意: 基準時点より後に作られ、障害で失われた変更は、どちらのファイルにも存在しないため、この照合では検出できない。") lines.append("別系統の記録（通知、申請記録など）があれば回収できる場合がある。この演習ではそれらは用意していない。") else: lines.append("（保全した障害時のファイルがないため、この節は省略）") lines.append("") lines.append("## 3. 判定") lines.append("") lines.append(f"- 基準時点との照合: {'すべて一致' if all_ok else '不一致あり'}") lines.append("- これは自動照合の結果であり、利用者による業務確認でも、利用再開の承認でもない。") lines.append(" 再開の判断は復旧確認表（templates/recovery_checklist_template.md）で、業務確認の完了後に別途行う。") report = "\n".join(lines) _file(report_path, "レポート出力先") # 既存なら通常ファイルであること。リンクなら拒否 with open(report_path, "w", encoding="utf-8") as f: f.write(report + "\n") print(report) log(f"レポートを書き出した: {report_path}") if not all_ok: log("不一致があるため終了コード 1 で終わる") sys.exit(1) log("終了コード 0") COMMANDS = { "reset": cmd_reset, "setup": cmd_setup, "backup": cmd_backup, "after-backup": cmd_after_backup, "break": cmd_break, "freeze": cmd_freeze, "rollback-code": cmd_rollback_code, "restore": cmd_restore, "verify": cmd_verify, } if __name__ == "__main__": if len(sys.argv) < 2 or sys.argv[1] not in COMMANDS: print(__doc__) sys.exit(2) COMMANDS[sys.argv[1]](sys.argv[2:])