Convert Teradata BTEQ scripts to modern cloud SQL

-- daily_sales_rank.bteq
.LOGON $TDHOST/$TDUSER,$TDPWD;
.SET WIDTH 500;
.SET ERROROUT STDOUT;

DELETE FROM sales_rank_stg ALL;
.IF ERRORCODE <> 0 THEN .QUIT 8;

INSERT INTO sales_rank_stg
SELECT  store_id,
        product_id,
        SUM(amount)  AS total_amt
FROM    sales_fact
WHERE   sale_dt BETWEEN DATE '2026-01-01' AND DATE
        '2026-01-31'
GROUP BY 1,2
QUALIFY RANK() OVER (PARTITION BY store_id
                     ORDER BY SUM(amount) DESC) <= 10;
.IF ERRORCODE <> 0 THEN .QUIT 8;

.IF ACTIVITYCOUNT = 0 THEN .GOTO NODATA;
COLLECT STATISTICS ON sales_rank_stg COLUMN(store_id);

.LABEL NODATA
.LOGOFF;
.QUIT 0;

# BTEQ control flow → explicit steps
import snowflake.connector

TOP_N_SALES = """
INSERT INTO sales_rank_stg
SELECT store_id, product_id, total_amt FROM (
    SELECT  store_id, product_id,
            SUM(amount) AS total_amt,
            RANK() OVER (PARTITION BY store_id
                         ORDER BY SUM(amount) DESC) AS rnk
    FROM    sales_fact
    WHERE   sale_dt BETWEEN '2026-01-01' AND '2026-01-31'
    GROUP BY store_id, product_id
) WHERE rnk <= 10;
"""

with snowflake.connector.connect(**cfg) as cx:
    cur = cx.cursor()
    cur.execute("TRUNCATE TABLE sales_rank_stg")   # DELETE ALL
    cur.execute(TOP_N_SALES)                       # QUALIFY rewritten
    if cur.rowcount == 0:                          # ACTIVITYCOUNT
        log.warning("no rows loaded; skipping stats")
    # .IF ERRORCODE handling is exception handling here;
    # COLLECT STATISTICS has no equivalent and is dropped.