63 lines
2.2 KiB
Python
63 lines
2.2 KiB
Python
import os
|
|
import sqlite3
|
|
from datetime import datetime, timedelta
|
|
|
|
def _resolve_sqlite_path(url: str) -> str:
|
|
if not url.startswith('sqlite:///'):
|
|
raise ValueError('Only sqlite URLs are supported')
|
|
return url.replace('sqlite:///', '')
|
|
|
|
def ensure_tables(conn: sqlite3.Connection) -> None:
|
|
cur = conn.cursor()
|
|
cur.execute(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS daily_revenue (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
date DATE NOT NULL UNIQUE,
|
|
amount REAL NOT NULL,
|
|
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
|
|
is_final INTEGER NOT NULL DEFAULT 0,
|
|
updated_at DATETIME DEFAULT CURRENT_TIMESTAMP,
|
|
source TEXT,
|
|
note TEXT
|
|
)
|
|
"""
|
|
)
|
|
cur.execute(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS audit_log (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
date DATE NOT NULL,
|
|
old_amount REAL,
|
|
new_amount REAL,
|
|
reason TEXT,
|
|
actor TEXT,
|
|
type TEXT,
|
|
created_at DATETIME DEFAULT CURRENT_TIMESTAMP
|
|
)
|
|
"""
|
|
)
|
|
conn.commit()
|
|
|
|
def set_yesterday(amount: float) -> None:
|
|
url = os.getenv('DATABASE_URL', 'sqlite:///data.db')
|
|
path = _resolve_sqlite_path(url)
|
|
os.makedirs(os.path.dirname(path) or '.', exist_ok=True)
|
|
conn = sqlite3.connect(path)
|
|
ensure_tables(conn)
|
|
cur = conn.cursor()
|
|
yday = (datetime.now().date() - timedelta(days=1)).isoformat()
|
|
cur.execute('SELECT amount FROM daily_revenue WHERE date=?', (yday,))
|
|
row = cur.fetchone()
|
|
old = row[0] if row else None
|
|
if row:
|
|
cur.execute('UPDATE daily_revenue SET amount=?, is_final=1, source=? , updated_at=CURRENT_TIMESTAMP WHERE date=?', (amount, 'correct', yday))
|
|
else:
|
|
cur.execute('INSERT INTO daily_revenue(date, amount, is_final, source) VALUES(?, ?, 1, ?)', (yday, amount, 'correct'))
|
|
cur.execute('INSERT INTO audit_log(date, old_amount, new_amount, reason, actor, type) VALUES(?, ?, ?, ?, ?, ?)', (yday, old, amount, 'manual_correct', 'admin', 'correct'))
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
if __name__ == '__main__':
|
|
set_yesterday(3629.76)
|