habit-tracker/scripts/deploy_sftp.py

"""SFTP로 시놀로지 NAS에 소스를 동기화하는 배포 스크립트.

deploy.env(git 미포함, deploy.env.example 참고)의 접속 정보를 읽어, Dockerfile이 COPY하는
파일/디렉터리(pyproject.toml, alembic.ini, app/, migrations/, scripts/)와 Dockerfile,
docker-compose.yml을 원격 경로로 동기화한다. 로컬 mtime이 원격보다 최신인 파일만 올리고,
원격에만 있는 파일은 건드리지 않는다(단방향 추가/갱신, 삭제 없음). .env는 절대 동기화하지 않는다
(원격 프로덕션 .env를 덮어쓰면 안 되므로).

컨테이너 재시작/재빌드는 이 스크립트가 하지 않는다 — 파일을 올린 뒤 직접 재시작할 것.

사용법: python scripts/deploy_sftp.py [--dry-run]
"""

import sys
from pathlib import Path

import paramiko

sys.stdout.reconfigure(encoding="utf-8")

PROJECT_ROOT = Path(__file__).resolve().parent.parent
DEPLOY_ENV_PATH = PROJECT_ROOT / "deploy.env"

# Dockerfile이 COPY하는 것과 동일한 목록 + 컨테이너 정의 파일
SYNC_TARGETS = ["pyproject.toml", "alembic.ini", "Dockerfile", "docker-compose.yml", "app", "migrations", "scripts"]
SKIP_NAMES = {"__pycache__"}
SKIP_SUFFIXES = {".pyc"}


def load_deploy_env(path: Path) -> dict[str, str]:
    if not path.exists():
        print(f"{path}가 없습니다. deploy.env.example을 복사해 값을 채워주세요.")
        sys.exit(1)
    values = {}
    for line in path.read_text(encoding="utf-8").splitlines():
        line = line.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        key, _, value = line.partition("=")
        values[key.strip()] = value.strip()
    return values


def iter_local_files(target: Path):
    if target.is_file():
        yield target
        return
    for path in target.rglob("*"):
        if path.is_dir():
            continue
        if any(part in SKIP_NAMES for part in path.parts):
            continue
        if path.suffix in SKIP_SUFFIXES:
            continue
        yield path


def ensure_remote_dir(sftp: paramiko.SFTPClient, remote_dir: str) -> None:
    parts = remote_dir.strip("/").split("/")
    current = ""
    for part in parts:
        current += "/" + part
        try:
            sftp.stat(current)
        except FileNotFoundError:
            sftp.mkdir(current)


def remote_mtime(sftp: paramiko.SFTPClient, remote_path: str) -> float | None:
    try:
        return sftp.stat(remote_path).st_mtime
    except FileNotFoundError:
        return None


def main() -> None:
    dry_run = "--dry-run" in sys.argv
    env = load_deploy_env(DEPLOY_ENV_PATH)

    host = env["SFTP_HOST"]
    port = int(env.get("SFTP_PORT", "22"))
    username = env["SFTP_USERNAME"]
    password = env["SFTP_PASSWORD"]
    remote_root = env["SFTP_REMOTE_PATH"].rstrip("/")

    print(f"{host}:{port} ({username}) -> {remote_root} 로 동기화{'(dry-run)' if dry_run else ''}")

    transport = paramiko.Transport((host, port))
    transport.connect(username=username, password=password)
    sftp = paramiko.SFTPClient.from_transport(transport)
    assert sftp is not None

    uploaded, skipped = 0, 0
    try:
        for target_name in SYNC_TARGETS:
            local_target = PROJECT_ROOT / target_name
            if not local_target.exists():
                continue
            for local_file in iter_local_files(local_target):
                rel_path = local_file.relative_to(PROJECT_ROOT).as_posix()
                remote_path = f"{remote_root}/{rel_path}"
                local_mtime = local_file.stat().st_mtime
                existing_mtime = remote_mtime(sftp, remote_path)

                if existing_mtime is not None and existing_mtime >= local_mtime:
                    skipped += 1
                    continue

                print(f"  업로드: {rel_path}")
                if not dry_run:
                    ensure_remote_dir(sftp, str(Path(remote_path).parent.as_posix()))
                    sftp.put(str(local_file), remote_path)
                uploaded += 1
    finally:
        sftp.close()
        transport.close()

    print(f"완료: {uploaded}개 업로드, {skipped}개 변경 없음. 컨테이너 재시작/재빌드는 직접 해주세요.")


if __name__ == "__main__":
    main()