Automating FTP Deployments in Python with Smart Change Detection & Retry

"""
ftp_deploy.py — upload only changed files, retry on drop, clear caches.

Usage:
    python ftp_deploy.py
"""
import ftplib
import io
import os
import socket
import time
import urllib.request
import urllib.error

# ── config ────────────────────────────────────────────────────────────────
FTP_HOST   = "example.com"
FTP_USER   = "your_ftp_user"
FTP_PASS   = "your_ftp_password"
REMOTE_ROOT = "public_html"          # directory that contains artisan / index.php
LOCAL_ROOT  = r"C:\Projects\myapp"  # local project root
SITE_URL    = "https://example.com"

FILES = [
    "bootstrap/app.php",
    "app/Http/Controllers/HomeController.php",
    "resources/views/home.blade.php",
    "public/css/app.css",
]

RETRIES    = 4
TIMEOUT    = 60
TIME_SLACK = 120   # seconds of clock skew tolerance


# ── connection wrapper with auto-reconnect ────────────────────────────────
class FTPClient:
    def __init__(self):
        self.ftp = None
        self._connect()

    def _connect(self):
        if self.ftp:
            try: self.ftp.close()
            except Exception: pass
        self.ftp = ftplib.FTP(FTP_HOST, timeout=TIMEOUT)
        self.ftp.login(FTP_USER, FTP_PASS)
        self.ftp.voidcmd("TYPE I")

    def retry(self, fn, label: str):
        for attempt in range(1, RETRIES + 1):
            try:
                return fn(self.ftp)
            except (EOFError, ConnectionError, TimeoutError,
                    socket.error, ftplib.error_temp) as exc:
                print(f"  ! {label} dropped ({exc!r}) — reconnecting ({attempt}/{RETRIES})")
                time.sleep(2 * attempt)
                try: self._connect()
                except Exception as ce: print(f"  ! reconnect failed: {ce}")
        raise RuntimeError(f"gave up on {label} after {RETRIES} attempts")


# ── remote helpers ────────────────────────────────────────────────────────
def remote_stat(client: FTPClient, rpath: str):
    """Return (size, mtime_epoch) or (None, None) if the file does not exist."""
    def do(ftp):
        try:
            ftp.voidcmd("TYPE I")
            size = ftp.size(rpath)
        except ftplib.error_perm:
            return (None, None)
        mtime = None
        try:
            import calendar
            ts = ftp.voidcmd(f"MDTM {rpath}")   # '213 YYYYMMDDHHMMSS'
            mtime = calendar.timegm(time.strptime(ts[4:18].strip(), "%Y%m%d%H%M%S"))
        except Exception:
            pass
        return (size, mtime)
    return client.retry(do, f"STAT {rpath}")


def ensure_dirs(client: FTPClient, rpath: str) -> None:
    """Create every directory component of rpath that does not yet exist."""
    parts = rpath.split("/")[:-1]
    cur = ""
    for part in parts:
        cur = f"{cur}/{part}" if cur else part
        def mk(ftp, d=cur):
            try: ftp.mkd(d)
            except ftplib.error_perm: pass   # already exists — fine
        client.retry(mk, f"MKD {cur}")


def upload_bytes(client: FTPClient, data: bytes, rpath: str) -> None:
    client.retry(lambda ftp: ftp.storbinary("STOR " + rpath, io.BytesIO(data)), f"STOR {rpath}")


# ── smart upload ──────────────────────────────────────────────────────────
def upload_if_changed(client: FTPClient, relpath: str) -> bool:
    lpath = os.path.join(LOCAL_ROOT, relpath.replace("/", os.sep))
    if not os.path.exists(lpath):
        print(f"  MISSING locally — skipped: {relpath}")
        return False

    with open(lpath, "rb") as f:
        data = f.read()

    rpath = f"{REMOTE_ROOT}/{relpath}"
    rsize, rmtime = remote_stat(client, rpath)
    lmtime = os.path.getmtime(lpath)

    if rsize == len(data) and rmtime is not None and rmtime >= lmtime - TIME_SLACK:
        print(f"  SAME  {relpath}")
        return False

    ensure_dirs(client, rpath)
    upload_bytes(client, data, rpath)
    print(f"  UP    {relpath}  ({len(data):,} bytes)")
    return True


# ── post-deploy cache clear ───────────────────────────────────────────────
RUNNER = """\
<?php
// One-time runner — self-deletes after execution.
if (($_GET['t'] ?? '') !== '__TOKEN__') { http_response_code(403); exit; }
require __DIR__ . '/../vendor/autoload.php';
$app    = require __DIR__ . '/../bootstrap/app.php';
$kernel = $app->make(Illuminate\\Contracts\\Console\\Kernel::class);
foreach (['config:clear', 'route:clear', 'view:clear'] as $cmd) {
    $kernel->call($cmd);
}
echo json_encode(['ok' => true]);
@unlink(__FILE__);
"""

def clear_caches(client: FTPClient) -> None:
    import secrets
    token = secrets.token_urlsafe(16)
    fname = f"_run_{secrets.token_hex(4)}.php"
    rpath = f"{REMOTE_ROOT}/public/{fname}"

    upload_bytes(client, RUNNER.replace("__TOKEN__", token).encode(), rpath)
    url = f"{SITE_URL}/{fname}?t={token}"

    try:
        with urllib.request.urlopen(url, timeout=60) as r:
            print(f"  Cache clear response: {r.read().decode()[:120]}")
    except urllib.error.HTTPError as e:
        print(f"  HTTP {e.code} calling runner")
    except Exception as e:
        print(f"  Could not reach runner: {e}")

    try:
        client.retry(lambda ftp: ftp.delete(rpath), f"DELE {rpath}")
    except Exception:
        pass   # runner self-deleted — fine


# ── main ──────────────────────────────────────────────────────────────────
def main():
    client  = FTPClient()
    changed = sum(upload_if_changed(client, f) for f in FILES)
    print(f"\n{changed} file(s) uploaded.")

    if changed:
        clear_caches(client)

    try: client.ftp.quit()
    except Exception: pass


if __name__ == "__main__":
    main()