`sftp_probe.py`

85 lines
#!/usr/bin/env python3
"""Test SFTP connection, list root, sample a few dirs to understand layout."""
import os, stat, sys
import paramiko

env = {}
with open(os.path.expanduser("~/.kravet-sftp.env")) as f:
    for line in f:
        k,_,v = line.strip().partition("=")
        if k: env[k] = v

HOST = env["KRAVET_SFTP_HOST"]
PORT = int(env["KRAVET_SFTP_PORT"])
USER = env["KRAVET_SFTP_USER"]
PASS = env["KRAVET_SFTP_PASS"]

print(f"connecting to {USER}@{HOST}:{PORT} ...", flush=True)
t = paramiko.Transport((HOST, PORT))
try:
    t.connect(username=USER, password=PASS)
except paramiko.ssh_exception.BadAuthenticationType as e:
    # Server may advertise only keyboard-interactive; retry with handler
    print(f"  plain password rejected — trying keyboard-interactive: {e}")
    def handler(title, instr, prompts):
        return [PASS for _ in prompts]
    t.auth_interactive(USER, handler)
sftp = paramiko.SFTPClient.from_transport(t)
print("connected.")
print(f"cwd: {sftp.getcwd() or sftp.normalize('.')}")

def walk_shallow(path="/", depth=0, max_depth=2):
    try:
        entries = sftp.listdir_attr(path)
    except Exception as e:
        print(f"  {'  '*depth}[err] {path}: {e}")
        return
    # Sort: dirs first, then files; cap listing
    entries.sort(key=lambda a: (0 if stat.S_ISDIR(a.st_mode or 0) else 1, a.filename))
    for i, a in enumerate(entries):
        kind = "d" if stat.S_ISDIR(a.st_mode or 0) else "-"
        print(f"  {'  '*depth}{kind} {a.st_size or 0:>12}  {a.filename}")
        if depth < max_depth and stat.S_ISDIR(a.st_mode or 0) and i < 6:
            sub = (path.rstrip("/") + "/" + a.filename)
            walk_shallow(sub, depth+1, max_depth)

print("\nroot listing:")
walk_shallow(".", 0, 2)

# Try the paths the sheet references: /LJ/HIRES/...
# Full root — top-level brands
print("\nTOP-LEVEL brand dirs and file counts / sizes:")
total_files = 0
total_bytes = 0
brand_totals = []
for a in sorted(sftp.listdir_attr("."), key=lambda x: x.filename):
    if stat.S_ISDIR(a.st_mode or 0):
        brand = a.filename
        fc, sz = 0, 0
        for sub in sftp.listdir_attr(brand):
            if stat.S_ISDIR(sub.st_mode or 0):
                for f in sftp.listdir_attr(brand + "/" + sub.filename):
                    if not stat.S_ISDIR(f.st_mode or 0):
                        fc += 1; sz += f.st_size or 0
            else:
                fc += 1; sz += a.st_size or 0
        brand_totals.append((brand, fc, sz))
        total_files += fc; total_bytes += sz
        print(f"  {brand:10s}  {fc:>8,} files  {sz/1024/1024:>10.1f} MB")
    else:
        print(f"  (file) {a.st_size:>10,}  {a.filename}")

print(f"\nTOTAL: {total_files:,} files, {total_bytes/1024/1024/1024:.2f} GB")

print("\nsample /LJ/HIRES (first 5):")
try:
    for i, a in enumerate(sorted(sftp.listdir_attr("/LJ/HIRES"), key=lambda x: x.filename)):
        if i >= 5: break
        print(f"  {a.st_size:>10}  {a.filename}")
    print(f"  /LJ/HIRES total: {len(sftp.listdir('/LJ/HIRES')):,}")
except Exception as e:
    print(f"  error: {e}")

sftp.close(); t.close()
print("done.")