cli v0.8.0: 'yakcloud datasource migrate' — 블랙리스트(.migrateignore) 스냅샷 캡처/빈타깃 적용(부트스트랩). pg/mysql/mariadb/mongo/minio, 미리보기+제외검증+apply-skip+reset복원 docker E2E 검증. dev reset named볼륨 삭제 수정. 문서·/yakcloud 갱신

2026-08-27 20:15:18 +09:00
parent a82de865f7
commit a498a402e6
6 changed files with 340 additions and 4 deletions
--- a/.claude/commands/yakcloud.md
+++ b/.claude/commands/yakcloud.md
@ -21,6 +21,8 @@ allowed-tools: Bash, Read, Write, Edit, Glob, Grep, AskUserQuestion
  (사용자가 값을 직접 주면 `YAKCLOUD_URL=… YAKCLOUD_TOKEN=… GITEA_USER=… GITEA_TOKEN=… yakcloud config` 로 저장하되 **값은 화면에 출력 금지**.)
 - **`dev …`**(`dev up|run|down|status|logs|env|reset|doctor`) → 로컬 docker 데이터소스 + `.env.dev`.
  `docker` 데몬이 떠 있어야 함(없으면 `yakcloud dev doctor` 로 진단). `dev run -- <명령>` 은 앱을 배포와 동일한 `<ALIAS>_*` 로 실행.
+- **`datasource migrate …`**(`capture|apply|status`) → 데이터소스 스냅샷(블랙리스트 `db/<source>/.migrateignore`).
+  `capture` = 소스 전체−블랙리스트 덤프(**읽기전용**, 먼저 미리보기로 무엇이 잡히는지 보여줄 것). `apply` = **빈 타깃만** 복원(데이터 있으면 스킵). prod 적용은 콘솔 소스관리자 컨펌 + 백업 후(롤백) — 이 명령으로 prod를 직접 덮지 말 것.
 - **그 외 전부**(`project deploy|info|check|update`, `datasource ls`, `domain`, `scale`, `set`, `env`,
  `source`, `bind`, `unbind`, `config`, `upgrade`) → `yakcloud $ARGUMENTS` 를 실행하고 **출력을 그대로** 보여준 뒤,
  필요할 때만 한 줄로 해석. 배포/삭제성(`deploy`·`source rm`·`unbind`·`domain`·`dev down --volumes`)은 실행 전 무엇을 하는지 한 줄 알리고 진행.
--- a/.claude/skills/yakcloud-deploy/SKILL.md
+++ b/.claude/skills/yakcloud-deploy/SKILL.md
@ -32,6 +32,7 @@ YakCloud 콘솔 API로 **데이터소스 리컨실 + 워크로드 배포 + 바
 - `yakcloud dev up` — `requires[]` 를 로컬 docker 로 기동 + `.env.dev`(프로덕션과 **바이트 동일**한 `<ALIAS>_*`) 생성
 - `yakcloud dev run -- <명령>` — up 보장 후 `.env.dev` 주입해 앱 실행(배포된 파드와 동일 env → 무수정 이식)
 - `yakcloud dev status|logs|down [--volumes]|reset|doctor` — 자격은 시드에서 결정적 파생(`.yakcloud/dev/`, gitignore)
+- `yakcloud datasource migrate capture|apply|status` — 블랙리스트(`db/<source>/.migrateignore`) 스냅샷: 전체−제외 캡처(읽기전용), **빈 타깃만** 적용(부트스트랩). prod는 콘솔 소스관리자 컨펌+백업 후

 **배포환경 설정(앱별) — 매니페스트 수정 + 배포중이면 재빌드 없이 라이브 반영**
 - `yakcloud domain <fqdn> [wl]`  도메인 등록 + 워크로드 할당
--- a/DATA-SOURCES.md
+++ b/DATA-SOURCES.md
@ -48,6 +48,20 @@ yakcloud dev status | logs | down [--volumes] | reset | doctor
 - `.env.dev`·`.yakcloud/`·`docker-compose.yakcloud-dev.yml` 은 커밋 금지(스캐폴드 `.gitignore` 에 포함).
 - env 계약은 프로덕션 백엔드 `_bind_env_for` 와 **바이트 동일**하게 유지(단일 진실원천).

+## 스키마·데이터 마이그레이션(`yakcloud datasource migrate`) — 블랙리스트 스냅샷
+개발 DB는 테스트로 엉망이 되기 쉬우니, **전부 캡처하되 블랙리스트로 뺀다**(denylist). 새 환경을 dev 상태로 채우는 **부트스트랩** 도구.
+```sh
+yakcloud datasource migrate capture [source…] --dry-run   # 무엇이 잡히는지 미리보기(읽기전용)
+yakcloud datasource migrate capture [source…]             # db/<source>/snapshot/ 에 전체−블랙리스트 덤프
+yakcloud datasource migrate apply [source…]               # 빈 타깃만 복원(데이터 있으면 스킵, force 없음)
+yakcloud datasource migrate status
+```
+- **블랙리스트**: `db/<source>/.migrateignore` (글롭, git 추적). 예: `audit_log_*`, `tmp_*`, `cache/*`. 민감/개인정보·쓰레기는 반드시 여기에.
+- **capture=읽기전용**(안전), 전체 native 덤프(pg_dump/mysqldump/mongodump/mc mirror). 지원: postgres·mysql·mariadb·mongodb·minio(redis/solr/rabbitmq/oracle=다음 단계).
+- **apply=빈 타깃만**. 데이터가 있으면 **절대 안 덮음**(스킵). 라이브 prod 지속 동기화가 아님.
+- **db/** 는 커밋·리뷰 대상(`.env.dev`·`.yakcloud/` 와 반대로 gitignore 아님).
+- prod 적용은 콘솔 종류별 소스관리자 **컨펌 + 항상 백업 후(롤백 보장)** — CLI 로 prod 직접 덮지 않음.
+
 ## 앱 코드 예시
 ```python
 # Python — alias=db (postgres)
--- a/bin/yakcloud
+++ b/bin/yakcloud
@ -18,7 +18,7 @@ set -uo pipefail   # -e 미사용: 'test && action' 관용구가 값 없을 때

 REPO="${YAKCLOUD_STARTER_REPO:-https://gitea.yakenator.io/yakenator/yakcloud-starter}"
 BRANCH="${YAKCLOUD_STARTER_BRANCH:-main}"
-VERSION="0.7.0"
+VERSION="0.8.0"
 CONFIG_DIR="${YAKCLOUD_CONFIG_DIR:-$HOME/.config/yakcloud}"
 CONFIG_FILE="$CONFIG_DIR/config"

@ -45,6 +45,12 @@ dev() {
  python3 scripts/yakcloud_dev.py "$@"
 }

+migrate() {
+  [ -f scripts/yakcloud_migrate.py ] || die "프로젝트 폴더가 아닙니다 — 'yakcloud project init' 로 초기화하세요."
+  ensure_pyyaml
+  python3 scripts/yakcloud_migrate.py "$@"
+}
+
 # ── 자격 저장 ── 비밀(배포토큰·Gitea토큰)=OS 키체인(암호화), 그 외(URL·사용자)=config 파일(600) ──
 _KC="none"
 if command -v security >/dev/null 2>&1; then _KC="macos"
@ -326,7 +332,8 @@ cmd_project_update() {
  mkdir -p scripts .gitea/workflows .claude/skills
  cp "$tmp/s/scripts/yakcloud_deploy.py" scripts/ 2>/dev/null || true
  cp "$tmp/s/scripts/yakcloud_ctl.py"    scripts/ 2>/dev/null || true
-  cp "$tmp/s/scripts/yakcloud_dev.py"    scripts/ 2>/dev/null || true
+  cp "$tmp/s/scripts/yakcloud_dev.py"     scripts/ 2>/dev/null || true
+  cp "$tmp/s/scripts/yakcloud_migrate.py" scripts/ 2>/dev/null || true
  cp "$tmp/s/.gitea/workflows/deploy.yml" .gitea/workflows/ 2>/dev/null || true
  cp "$tmp/s/DATA-SOURCES.md" DATA-SOURCES.md 2>/dev/null || true
  rm -rf .claude/skills/yakcloud-deploy
@ -362,6 +369,8 @@ yakcloud $VERSION — YakCloud 프로젝트 CLI
  dev up [source…]             requires[] 를 로컬 docker 로 기동 + .env.dev 생성
  dev run -- <명령>            up 보장 후 .env.dev 로 앱 실행(배포와 동일 <ALIAS>_* 주입)
  dev down [--volumes] | status | logs [src] | env | reset [src] | doctor
+  datasource migrate capture [src…] [--dry-run]   소스 전체−블랙리스트(.migrateignore) 스냅샷(읽기전용)
+  datasource migrate apply [src…]                 스냅샷을 빈 타깃에 적용(데이터 있으면 스킵) | migrate status
 env: YAKCLOUD_URL, YAKCLOUD_TOKEN(배포토큰 yakd_…), (선택) YAKCLOUD_CLUSTER
 EOF
 }
@ -383,7 +392,10 @@ case "${1:-help}" in
  config) cmd_config ;;
  logout) cmd_logout ;;
  dev) shift; dev "$@" ;;
-  datasource|ds) shift; ctl source "$@" ;;
+  migrate) shift; migrate "$@" ;;
+  datasource|ds)
+    shift
+    if [ "${1:-}" = migrate ]; then shift; migrate "$@"; else ctl source "$@"; fi ;;
  domain|scale|set|env|source|bind|unbind) ctl "$@" ;;
  upgrade|self-update) cmd_upgrade ;;
  version|-v|--version) echo "yakcloud $VERSION" ;;
--- a/scripts/yakcloud_dev.py
+++ b/scripts/yakcloud_dev.py
@ -707,7 +707,10 @@ def cmd_reset(a) -> None:
    m = load_manifest()
    if os.path.exists(COMPOSE_FILE):
        if a.source:
-            compose(m, "rm", "-fsv", a.source, check=False)
+            compose(m, "rm", "-fs", a.source, check=False)
+            # named 볼륨은 compose rm -v 로 안 지워짐 → 명시 삭제(진짜 초기화)
+            sh("docker", "volume", "rm", "-f", f"{compose_project(m)}_yakdev-{a.source}-data",
+               check=False, capture=True)
        else:
            compose(m, "down", "-v", check=False)
    print(f"✓ 초기화 완료{'('+a.source+')' if a.source else ''} — 'yakcloud dev up' 로 재기동(초기화 재실행)")
--- a/scripts/yakcloud_migrate.py
+++ b/scripts/yakcloud_migrate.py
@ -0,0 +1,304 @@
+#!/usr/bin/env python3
+"""yakcloud datasource migrate — 데이터소스 스냅샷(블랙리스트) 캡처 + 빈 타깃 적용(부트스트랩).
+
+모델(확정):
+  capture  = 소스 전체 native 덤프 − 블랙리스트(db/<source>/.migrateignore, 글롭) → db/<source>/snapshot/.
+             **읽기전용**(항상 안전). 커밋 전 '무엇이 잡히는지' 미리보기.
+  apply    = 타깃이 **비어 있을 때만** 복원. 데이터가 있으면 **스킵**(절대 안 덮음, force 없음).
+             즉 새 환경을 dev 상태로 채우는 부트스트랩/시딩 도구. (prod 적용은 백엔드 경유 — 별도 승인)
+
+dev 채널: 러너를 DB 클라이언트 이미지 one-shot 으로 compose 네트워크에 join → 소스 컨테이너 직접(자격 미노출).
+자격은 yakcloud_dev 의 결정적 파생(creds_for)을 그대로 사용. 지원 타입: postgresql·mysql·mariadb·mongodb·minio.
+"""
+from __future__ import annotations
+
+import argparse
+import fnmatch
+import json
+import os
+import subprocess
+import sys
+
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+import yakcloud_dev as dev  # noqa: E402  (동일 scripts/ 의 dev 엔진 헬퍼 재사용)
+
+SNAP = "db"                      # 프로젝트 루트 db/<source>/ (git 추적 — .gitignore 아님)
+IGNORE = ".migrateignore"
+SUPPORTED = ("postgresql", "mysql", "mariadb", "mongodb", "minio")
+IMG = {"postgresql": "postgres:16-alpine", "mysql": "mysql:8.4", "mariadb": "mariadb:11.4",
+       "mongodb": "mongo:7", "minio": "minio/mc:latest"}
+
+
+def die(m: str) -> None:
+    sys.exit(f"  ✗ {m}")
+
+
+def _net(m: dict) -> str:
+    return f"{dev.compose_project(m)}_default"
+
+
+def _running(m: dict, source: str) -> bool:
+    j = dev._ps_state(m).get(source) or {}
+    return (j.get("State") or "") == "running"
+
+
+def _srcdir(source: str) -> str:
+    return os.path.join(SNAP, source)
+
+
+def _blacklist(source: str) -> list[str]:
+    p = os.path.join(_srcdir(source), IGNORE)
+    if not os.path.exists(p):
+        return []
+    out = []
+    for ln in open(p):
+        ln = ln.split("#", 1)[0].strip()
+        if ln:
+            out.append(ln)
+    return out
+
+
+def _excluded(name: str, patterns: list[str]) -> bool:
+    return any(fnmatch.fnmatch(name, pat) for pat in patterns)
+
+
+def drun(net: str, image: str, shell: str, mounts=None, capture=True, check=True):
+    """docker run one-shot(sh -c) — 모든 명령을 쉘 경유(리다이렉트/파이프 지원)."""
+    args = ["docker", "run", "--rm", "--network", net]
+    for host, cont in (mounts or []):
+        os.makedirs(host, exist_ok=True)
+        args += ["-v", f"{os.path.abspath(host)}:{cont}"]
+    args += ["--entrypoint", "sh", image, "-c", shell]
+    r = subprocess.run(args, text=True,
+                       stdout=subprocess.PIPE if capture else None,
+                       stderr=subprocess.PIPE if capture else None)
+    if check and r.returncode != 0:
+        die(f"클라이언트 실행 실패({r.returncode}): {(r.stderr or r.stdout or '').strip()[:400]}")
+    return r
+
+
+# ── 타입별: 대상 단위 나열(미리보기·글롭 확장용) ────────────────────────────
+def list_units(m: dict, source: str, stype: str, cr: dict) -> list[str]:
+    net = _net(m)
+    if stype == "postgresql":
+        url = f"postgresql://{cr['user']}:{cr['password']}@{source}:5432/{cr['db']}"
+        q = "SELECT tablename FROM pg_tables WHERE schemaname='public' ORDER BY tablename"
+        r = drun(net, IMG[stype], f"psql '{url}' -tAc \"{q}\"", check=False)
+    elif stype in ("mysql", "mariadb"):
+        q = f"SELECT table_name FROM information_schema.tables WHERE table_schema='{cr['db']}'"
+        r = drun(net, IMG[stype],
+                 f"mysql -h {source} -u{cr['user']} -p{cr['password']} -N -B -e \"{q}\" 2>/dev/null", check=False)
+    elif stype == "mongodb":
+        js = "db.getCollectionNames().filter(c=>!c.startsWith('_yakcloud')).join('\\n')"
+        r = drun(net, IMG[stype],
+                 f"mongosh --quiet --host {source} -u {cr['user']} -p {cr['password']} "
+                 f"--authenticationDatabase {cr['db']} {cr['db']} --eval \"{js}\"", check=False)
+    elif stype == "minio":
+        # minio/mc 이미지엔 awk 가 없음 → --json 으로 받아 파이썬에서 key 파싱.
+        setup = f"mc alias set s http://{source}:9000 {cr['access']} {cr['secret']} >/dev/null 2>&1"
+        r = drun(net, IMG[stype],
+                 f"{setup} && mc ls --recursive --json s/{cr['bucket']} 2>/dev/null", check=False)
+        keys = []
+        for ln in (r.stdout or "").splitlines():
+            ln = ln.strip()
+            if not ln:
+                continue
+            try:
+                j = json.loads(ln)
+                if j.get("key"):
+                    keys.append(j["key"])
+            except Exception:
+                pass
+        return keys
+    else:
+        return []
+    return [x.strip() for x in (r.stdout or "").splitlines() if x.strip() and "Warning" not in x]
+
+
+# ── 미리보기 ────────────────────────────────────────────────────────────
+def preview(m: dict, targets: list[tuple[str, str]]) -> dict:
+    seed = dev.ensure_seed()
+    plan = {}
+    print("캡처 미리보기 (읽기전용) — ✓=포함, ✗=제외(블랙리스트)")
+    for source, stype in targets:
+        cr = dev.creds_for(seed, source, stype)
+        bl = _blacklist(source)
+        units = list_units(m, source, stype, cr)
+        inc = [u for u in units if not _excluded(u, bl)]
+        exc = [u for u in units if _excluded(u, bl)]
+        plan[source] = {"type": stype, "include": inc, "exclude": exc, "blacklist": bl}
+        unit_word = {"minio": "오브젝트"}.get(stype, "테이블/컬렉션")
+        print(f"\n  ● {source} ({stype}) — {unit_word} {len(units)}개 · 포함 {len(inc)} / 제외 {len(exc)}")
+        for u in inc[:20]:
+            print(f"      ✓ {u}")
+        if len(inc) > 20:
+            print(f"      … 외 {len(inc)-20}개")
+        for u in exc:
+            print(f"      ✗ {u}  (블랙리스트)")
+        if not bl:
+            print(f"      · 블랙리스트 없음 → 전부 포함. 민감/개인정보·쓰레기는 "
+                  f"{os.path.join(_srcdir(source), IGNORE)} 에 글롭으로 제외 권장")
+    return plan
+
+
+# ── 캡처(전체 native 덤프 − 블랙리스트) ───────────────────────────────────
+def capture_source(m: dict, source: str, stype: str, cr: dict) -> dict:
+    net, outdir = _net(m), os.path.join(_srcdir(source), "snapshot")
+    os.makedirs(outdir, exist_ok=True)
+    bl = _blacklist(source)
+    units = list_units(m, source, stype, cr)
+    excluded = [u for u in units if _excluded(u, bl)]
+    info = {"type": stype, "excluded": excluded, "blacklist": bl}
+    if stype == "postgresql":
+        url = f"postgresql://{cr['user']}:{cr['password']}@{source}:5432/{cr['db']}"
+        # pg_dump 는 패턴(*)을 지원 → 블랙리스트 글롭을 그대로 넘겨 소유 시퀀스(<tbl>_id_seq 등)까지 제외.
+        ex = " ".join(f"--exclude-table='{g}'" for g in bl)
+        drun(net, IMG[stype], f"pg_dump '{url}' {ex} -f /out/snapshot.sql", [(outdir, "/out")])
+        info["file"] = "snapshot.sql"
+    elif stype in ("mysql", "mariadb"):
+        ig = " ".join(f"--ignore-table={cr['db']}.{e}" for e in excluded)
+        drun(net, IMG[stype],
+             f"mysqldump -h {source} -u{cr['user']} -p{cr['password']} --skip-comments {ig} "
+             f"{cr['db']} > /out/snapshot.sql", [(outdir, "/out")])
+        info["file"] = "snapshot.sql"
+    elif stype == "mongodb":
+        ex = " ".join(f"--excludeCollection={e}" for e in excluded)
+        drun(net, IMG[stype],
+             f"mongodump --host {source} -u {cr['user']} -p {cr['password']} "
+             f"--authenticationDatabase {cr['db']} --db {cr['db']} {ex} --archive=/out/dump.archive",
+             [(outdir, "/out")])
+        info["file"] = "dump.archive"
+    elif stype == "minio":
+        setup = f"mc alias set s http://{source}:9000 {cr['access']} {cr['secret']} >/dev/null"
+        exflag = " ".join(f"--exclude '{e}'" for e in bl)
+        drun(net, IMG[stype],
+             f"{setup} && mc mirror --overwrite {exflag} s/{cr['bucket']} /out/bucket", [(outdir, "/out")])
+        info["file"], info["bucket"] = "bucket/", cr["bucket"]
+    else:
+        die(f"미지원 타입(캡처): {stype}")
+    json.dump(info, open(os.path.join(outdir, "manifest.json"), "w"), ensure_ascii=False, indent=2)
+    return info
+
+
+# ── 타깃 비어있음 검사 + 적용(부트스트랩) ─────────────────────────────────
+def is_empty(m: dict, source: str, stype: str, cr: dict) -> bool:
+    return len(list_units(m, source, stype, cr)) == 0
+
+
+def apply_source(m: dict, source: str, stype: str, cr: dict) -> str:
+    net, outdir = _net(m), os.path.join(_srcdir(source), "snapshot")
+    if not os.path.isdir(outdir):
+        return "스냅샷 없음(먼저 capture)"
+    if not is_empty(m, source, stype, cr):
+        return "타깃에 데이터 있음 → 스킵(덮지 않음)"
+    if stype == "postgresql":
+        url = f"postgresql://{cr['user']}:{cr['password']}@{source}:5432/{cr['db']}"
+        drun(net, IMG[stype], f"psql '{url}' -v ON_ERROR_STOP=1 -f /out/snapshot.sql", [(outdir, "/out")])
+    elif stype in ("mysql", "mariadb"):
+        drun(net, IMG[stype],
+             f"mysql -h {source} -u{cr['user']} -p{cr['password']} {cr['db']} < /out/snapshot.sql",
+             [(outdir, "/out")])
+    elif stype == "mongodb":
+        drun(net, IMG[stype],
+             f"mongorestore --host {source} -u {cr['user']} -p {cr['password']} "
+             f"--authenticationDatabase {cr['db']} --archive=/out/dump.archive", [(outdir, "/out")])
+    elif stype == "minio":
+        setup = f"mc alias set s http://{source}:9000 {cr['access']} {cr['secret']} >/dev/null"
+        drun(net, IMG[stype],
+             f"{setup} && (mc mb -p s/{cr['bucket']} 2>/dev/null || true) && "
+             f"mc mirror --overwrite /out/bucket s/{cr['bucket']}", [(outdir, "/out")])
+    else:
+        die(f"미지원 타입(적용): {stype}")
+    return "적용 완료(부트스트랩)"
+
+
+def select_targets(m: dict, names: list[str], require_running: bool) -> list[tuple[str, str]]:
+    reqs = {r["name"]: r["type"] for r in dev.requires(m)}
+    if names:
+        miss = [n for n in names if n not in reqs]
+        if miss:
+            die(f"requires 에 없음: {', '.join(miss)}")
+        chosen = [(n, reqs[n]) for n in names]
+    else:
+        chosen = list(reqs.items())
+    out = []
+    for s, t in chosen:
+        if t not in SUPPORTED:
+            print(f"  · {s}({t}) — 마이그레이션 미지원(다음 단계: redis/solr/rabbitmq/oracle). 건너뜀")
+            continue
+        if require_running and not _running(m, s):
+            print(f"  · {s} — dev 컨테이너 미기동('yakcloud dev up {s}' 먼저). 건너뜀")
+            continue
+        out.append((s, t))
+    if not out:
+        die("대상 소스가 없습니다.")
+    return out
+
+
+# ── 명령 ────────────────────────────────────────────────────────────────
+def cmd_capture(a) -> None:
+    m = dev.load_manifest()
+    targets = select_targets(m, a.source, require_running=True)
+    preview(m, targets)
+    if a.dry_run:
+        print("\n(미리보기 전용 — 캡처 안 함)")
+        return
+    if not a.yes and sys.stdin.isatty():
+        if input("\n위 대상을 db/<source>/snapshot/ 에 캡처할까요? [y/N] ").strip().lower() not in ("y", "yes"):
+            print("취소.")
+            return
+    seed = dev.ensure_seed()
+    for source, stype in targets:
+        cr = dev.creds_for(seed, source, stype)
+        info = capture_source(m, source, stype, cr)
+        print(f"  ✓ {source} 캡처 → {os.path.join(_srcdir(source), 'snapshot', info['file'])}"
+              f" (제외 {len(info['excluded'])})")
+    print("  · db/ 를 커밋하세요(리뷰 대상). 적용: 'yakcloud datasource migrate apply'(빈 타깃만)")
+
+
+def cmd_apply(a) -> None:
+    m = dev.load_manifest()
+    targets = select_targets(m, a.source, require_running=True)
+    seed = dev.ensure_seed()
+    for source, stype in targets:
+        cr = dev.creds_for(seed, source, stype)
+        print(f"  ▸ {source} ({stype}): {apply_source(m, source, stype, cr)}")
+    print("  · prod 적용은 콘솔 소스관리자 컨펌 + 백업 후(롤백 보장) — 별도 경로")
+
+
+def cmd_status(a) -> None:
+    m = dev.load_manifest()
+    reqs = {r["name"]: r["type"] for r in dev.requires(m)}
+    names = a.source or list(reqs)
+    print("migrate 스냅샷 상태")
+    for s in names:
+        t = reqs.get(s, "?")
+        snap = os.path.join(_srcdir(s), "snapshot", "manifest.json")
+        bl = _blacklist(s)
+        if os.path.exists(snap):
+            info = json.load(open(snap))
+            print(f"  {s} ({t}): 스냅샷 있음(제외 {len(info.get('excluded', []))}) · 블랙리스트 {len(bl)}줄")
+        else:
+            print(f"  {s} ({t}): 스냅샷 없음 · 블랙리스트 {len(bl)}줄")
+
+
+def build_parser() -> argparse.ArgumentParser:
+    p = argparse.ArgumentParser(prog="yakcloud datasource migrate")
+    sub = p.add_subparsers(dest="cmd", required=True)
+    c = sub.add_parser("capture", help="소스 전체 상태 − 블랙리스트 를 db/<source>/snapshot/ 에 캡처(읽기전용)")
+    c.add_argument("source", nargs="*")
+    c.add_argument("--dry-run", action="store_true", help="미리보기만")
+    c.add_argument("-y", "--yes", action="store_true", help="확인 없이")
+    c.set_defaults(fn=cmd_capture)
+    ap = sub.add_parser("apply", help="스냅샷을 타깃에 적용(빈 타깃만, 데이터 있으면 스킵)")
+    ap.add_argument("source", nargs="*")
+    ap.set_defaults(fn=cmd_apply)
+    st = sub.add_parser("status", help="스냅샷·블랙리스트 상태")
+    st.add_argument("source", nargs="*")
+    st.set_defaults(fn=cmd_status)
+    return p
+
+
+if __name__ == "__main__":
+    args = build_parser().parse_args()
+    args.fn(args)