Automate SFTP and FTP file drops with Python.

folder = "/outbound"

# A file counts as loaded by name and modified time, so a partner that
# reuses a file name still gets each new delivery loaded.
loaded = {
    (row["file_name"], row["file_mtime"])
    for row in qbash.integrations.postgres.execute_query(
        query="SELECT file_name, file_mtime FROM partner_file_loads",
    )["result"]
}
entries = qbash.integrations.sftp.list_files(path=folder)["entries"]
new_files = [
    e for e in entries
    if e["type"] == "file" and e["name"].endswith(".csv")
    and (e["name"], e["mtime"]) not in loaded
]

def load_file(name):
    path = f"{folder}/{name}"
    # Counting first also indexes the file, so each SFTP read seeks to its chunk.
    total = qbash.integrations.sftp.get_row_count(path=path)["row_count"]
    for start in range(0, total, 1000):
        rows = qbash.integrations.sftp.read_chunk(
            path=path, startRow=start, rowCount=1000,
        )["rows"]
        qbash.integrations.postgres.execute_query(
            query=(
                "INSERT INTO partner_orders (order_id, sku, quantity) "
                "SELECT order_id, sku, quantity FROM jsonb_to_recordset($1::jsonb) "
                "AS r(order_id text, sku text, quantity int) "
                "ON CONFLICT (order_id) DO NOTHING"
            ),
            query_params=[rows],
        )
    return total

loaded_now, failed = [], []
for entry in new_files:
    name = entry["name"]
    try:
        row_count = load_file(name)
    except Exception as error:
        failed.append({"file": name, "error": str(error)})
        continue
    qbash.integrations.postgres.execute_query(
        query=(
            "INSERT INTO partner_file_loads (file_name, file_mtime, row_count) "
            "VALUES ($1, $2, $3)"
        ),
        query_params=[name, entry["mtime"], row_count],
    )
    loaded_now.append(name)

if loaded_now or failed:
    qbash.integrations.slack.create_message(
        channel="#partner-data",
        text=f"Loaded {len(loaded_now)} new partner files from {folder}. {len(failed)} failed.",
    )
qbash.output({"files_loaded": loaded_now, "files_failed": failed})