Rewrite as uv-based Flask web app with Dockerfile
- Flask dashboard with Plotly charts (stores, monthly, hourly, weekday, heatmap) - Spotify Wrapped-style summary with year dropdown and inflation toggle - CSV upload flow (ephemeral, no storage) - Homepage with upload and saved data options - Import script for loading CSVs into SQLite - Dockerized with gunicorn for production Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
84
import_csv.py
Normal file
84
import_csv.py
Normal file
@@ -0,0 +1,84 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Import a DoorDash CSV export into doordash.db."""
|
||||
|
||||
import csv
|
||||
import sqlite3
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
DB_PATH = Path(__file__).parent / "doordash.db"
|
||||
|
||||
CREATE_TABLE = """CREATE TABLE IF NOT EXISTS "doordash"(
|
||||
"ITEM" TEXT, "CATEGORY" TEXT, "STORE_NAME" TEXT, "UNIT_PRICE" TEXT,
|
||||
"QUANTITY" TEXT, "SUBTOTAL" TEXT, "CREATED_AT" TEXT, "DELIVERY_TIME" TEXT,
|
||||
"DELIVERY_ADDRESS" TEXT
|
||||
);"""
|
||||
|
||||
EXPECTED_HEADER = [
|
||||
"ITEM", "CATEGORY", "STORE_NAME", "UNIT_PRICE", "QUANTITY",
|
||||
"SUBTOTAL", "CREATED_AT", "DELIVERY_TIME", "DELIVERY_ADDRESS",
|
||||
]
|
||||
|
||||
|
||||
def import_csv(csv_path: str):
|
||||
path = Path(csv_path)
|
||||
if not path.exists():
|
||||
print(f"Error: {csv_path} not found")
|
||||
sys.exit(1)
|
||||
|
||||
conn = sqlite3.connect(DB_PATH)
|
||||
c = conn.cursor()
|
||||
c.execute(CREATE_TABLE)
|
||||
|
||||
# Get existing row count for reporting
|
||||
before_count = c.execute("SELECT COUNT(*) FROM doordash").fetchone()[0]
|
||||
|
||||
# Check for duplicates using CREATED_AT + ITEM + STORE_NAME
|
||||
existing = set()
|
||||
for row in c.execute("SELECT ITEM, STORE_NAME, CREATED_AT FROM doordash"):
|
||||
existing.add((row[0], row[1], row[2]))
|
||||
|
||||
imported = 0
|
||||
skipped = 0
|
||||
|
||||
with open(path, newline="") as f:
|
||||
reader = csv.reader(f)
|
||||
header = next(reader)
|
||||
|
||||
# Validate header
|
||||
header_upper = [h.strip().upper() for h in header]
|
||||
if header_upper != EXPECTED_HEADER:
|
||||
print(f"Error: unexpected CSV header.\nExpected: {EXPECTED_HEADER}\nGot: {header_upper}")
|
||||
sys.exit(1)
|
||||
|
||||
for row in reader:
|
||||
if len(row) != 9:
|
||||
print(f"Warning: skipping row with {len(row)} columns: {row[:3]}...")
|
||||
skipped += 1
|
||||
continue
|
||||
|
||||
key = (row[0], row[2], row[6]) # ITEM, STORE_NAME, CREATED_AT
|
||||
if key in existing:
|
||||
skipped += 1
|
||||
continue
|
||||
|
||||
c.execute("INSERT INTO doordash VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)", row)
|
||||
existing.add(key)
|
||||
imported += 1
|
||||
|
||||
conn.commit()
|
||||
after_count = c.execute("SELECT COUNT(*) FROM doordash").fetchone()[0]
|
||||
conn.close()
|
||||
|
||||
print(f"Done! Imported {imported} rows, skipped {skipped} duplicates.")
|
||||
print(f"Total rows: {before_count} -> {after_count}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
if len(sys.argv) < 2:
|
||||
print(f"Usage: {sys.argv[0]} <csv_file> [csv_file2 ...]")
|
||||
sys.exit(1)
|
||||
|
||||
for csv_file in sys.argv[1:]:
|
||||
print(f"\nImporting {csv_file}...")
|
||||
import_csv(csv_file)
|
||||
Reference in New Issue
Block a user