T19-T28: distribusi profit harian (distribute.py, /history, tests)

db.py

@@ -95,3 +95,140 @@ def replace_snapshot(rows, scanned_at):
         conn.commit()
     finally:
         conn.close()
+
+
+# ——————————————————————————————————————————————————————————————————————
+# Riwayat distribusi profit-share (V19..V27). Tabel append — tiap transfer
+# & tiap run tercatat permanen (⊥ hapus). Skema sama untuk kedua backend,
+# beda hanya klausa primary key autoincrement.
+# ——————————————————————————————————————————————————————————————————————
+
+_RUN_COLS = (
+    'run_date TEXT NOT NULL, '
+    'balance_start DOUBLE NOT NULL, '
+    'total_voters INTEGER NOT NULL, '
+    'total_staked DOUBLE NOT NULL, '
+    'total_sent DOUBLE NOT NULL, '
+    'status TEXT NOT NULL, '
+    'created_at TEXT NOT NULL'
+)
+_PAY_COLS = (
+    'run_id INTEGER NOT NULL, '
+    'owner VARCHAR(13) NOT NULL, '
+    'amount DOUBLE NOT NULL, '
+    'memo TEXT NOT NULL, '
+    'txid TEXT, '
+    'status TEXT NOT NULL, '
+    'error TEXT, '
+    'created_at TEXT NOT NULL, '
+    'updated_at TEXT NOT NULL'
+)
+
+_RUN_DDL_SQLITE = f'CREATE TABLE IF NOT EXISTS distribute_runs (run_id INTEGER PRIMARY KEY AUTOINCREMENT, {_RUN_COLS})'
+_PAY_DDL_SQLITE = f'CREATE TABLE IF NOT EXISTS distribute_payments (payment_id INTEGER PRIMARY KEY AUTOINCREMENT, {_PAY_COLS})'
+_RUN_DDL_MYSQL = f'CREATE TABLE IF NOT EXISTS distribute_runs (run_id INT AUTO_INCREMENT PRIMARY KEY, {_RUN_COLS})'
+_PAY_DDL_MYSQL = f'CREATE TABLE IF NOT EXISTS distribute_payments (payment_id INT AUTO_INCREMENT PRIMARY KEY, {_PAY_COLS})'
+
+
+def ensure_distribute_schema():
+    """Buat tabel riwayat distribusi bila belum ada (idempoten, kedua backend)."""
+    conn = connect()
+    try:
+        cur = conn.cursor()
+        if DB_BACKEND == 'sqlite':
+            cur.execute(_RUN_DDL_SQLITE)
+            cur.execute(_PAY_DDL_SQLITE)
+        else:
+            cur.execute(_RUN_DDL_MYSQL)
+            cur.execute(_PAY_DDL_MYSQL)
+        conn.commit()
+    finally:
+        conn.close()
+
+
+def record_run(run_date, balance_start, total_voters, total_staked,
+               total_sent, status, created_at):
+    """Insert baris `distribute_runs`, kembalikan run_id."""
+    ensure_distribute_schema()
+    conn = connect()
+    try:
+        cur = conn.cursor()
+        cur.execute(_translate(
+            'INSERT INTO distribute_runs '
+            '(run_date, balance_start, total_voters, total_staked, total_sent, '
+            ' status, created_at) VALUES (%s, %s, %s, %s, %s, %s, %s)'),
+            (run_date, balance_start, total_voters, total_staked, total_sent,
+             status, created_at))
+        conn.commit()
+        return cur.lastrowid
+    finally:
+        conn.close()
+
+
+def record_payment(run_id, owner, amount, memo, status, created_at,
+                   txid=None, error=None):
+    """Insert baris `distribute_payments` → payment_id."""
+    ensure_distribute_schema()
+    conn = connect()
+    try:
+        cur = conn.cursor()
+        cur.execute(_translate(
+            'INSERT INTO distribute_payments '
+            '(run_id, owner, amount, memo, txid, status, error, created_at, '
+            ' updated_at) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)'),
+            (run_id, owner, amount, memo, txid, status, error, created_at,
+             created_at))
+        conn.commit()
+        return cur.lastrowid
+    finally:
+        conn.close()
+
+
+def update_payment_status(payment_id, status, txid=None, error=None, updated_at=None):
+    """Perbarui status satu transfer (V21): pending → sent|failed."""
+    from datetime import datetime
+    if updated_at is None:
+        updated_at = datetime.now().isoformat(timespec='seconds')
+    conn = connect()
+    try:
+        cur = conn.cursor()
+        cur.execute(_translate(
+            'UPDATE distribute_payments SET status = %s, txid = %s, '
+            'error = %s, updated_at = %s WHERE payment_id = %s'),
+            (status, txid, error, updated_at, payment_id))
+        conn.commit()
+    finally:
+        conn.close()
+
+
+def update_run_status(run_id, status, total_sent):
+    """Tutup run: status ok|partial + total terkirim nyata."""
+    conn = connect()
+    try:
+        cur = conn.cursor()
+        cur.execute(_translate(
+            'UPDATE distribute_runs SET status = %s, total_sent = %s '
+            'WHERE run_id = %s'),
+            (status, total_sent, run_id))
+        conn.commit()
+    finally:
+        conn.close()
+
+
+def list_runs():
+    """Ringkasan run distribusi, terbaru dulu."""
+    ensure_distribute_schema()
+    return query(
+        'SELECT run_id, run_date, balance_start, total_voters, total_staked, '
+        '       total_sent, status, created_at '
+        'FROM distribute_runs ORDER BY run_id DESC')
+
+
+def list_payments():
+    """Transfer per-pemilih, terbaru dulu (gabung run untuk tanggal)."""
+    ensure_distribute_schema()
+    return query(
+        'SELECT p.payment_id, p.run_id, p.owner, p.amount, p.memo, p.txid, '
+        '       p.status, p.error, p.created_at, p.updated_at, r.run_date '
+        'FROM distribute_payments p JOIN distribute_runs r '
+        '  ON p.run_id = r.run_id ORDER BY p.payment_id DESC')