251 lines
11 KiB
Python
251 lines
11 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
import base64
|
|
import hashlib
|
|
import os
|
|
import secrets
|
|
import sqlite3
|
|
import sys
|
|
from datetime import date, datetime, timedelta, timezone
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parent.parent
|
|
IST = timezone(timedelta(hours=5, minutes=30))
|
|
|
|
PASSWORD = "staff123"
|
|
|
|
SHOPS = [("KZK", "Kazhakkoottam"), ("SKM", "Sreekaryam"), ("ULR", "Ulloor"), ("PTM", "Pattom")]
|
|
|
|
USERS = [
|
|
("kzk.staff", ["KZK", "SKM"], True),
|
|
("skm.staff", ["SKM"], True),
|
|
("ptm.staff", ["PTM"], False),
|
|
]
|
|
|
|
SCHEMES = {
|
|
0: ("Bhagyathara", "BT"),
|
|
1: ("Sthree Sakthi", "SS"),
|
|
2: ("Dhanalekshmi", "DL"),
|
|
3: ("Karunya Plus", "KN"),
|
|
4: ("Suvarna Keralam", "SK"),
|
|
5: ("Karunya", "KR"),
|
|
6: ("Samrudhi", "SM"),
|
|
}
|
|
|
|
PRIZES = {
|
|
"2851": (5, 2000),
|
|
"1360": (6, 1000),
|
|
"0987": (7, 500),
|
|
"6184": (7, 500),
|
|
"0376": (8, 200),
|
|
"1903": (8, 200),
|
|
"3821": (8, 200),
|
|
"2564": (9, 100),
|
|
"0119": (9, 100),
|
|
"4821": (4, 5000),
|
|
"7342": (6, 1000),
|
|
"5590": (9, 100),
|
|
}
|
|
|
|
KZK_ALL = [
|
|
("0003", 5), ("0018", 2), ("0057", 2), ("0166", 1), ("0231", 6), ("0376", 2), ("0418", 5), ("0745", 2),
|
|
("0887", 10), ("0888", 1), ("0987", 4), ("1122", 3), ("1207", 4), ("1360", 1), ("1573", 3), ("1892", 2),
|
|
("2290", 2), ("2417", 1), ("2564", 3), ("2689", 1), ("3076", 5), ("3345", 10), ("3422", 2), ("3472", 1),
|
|
("3484", 4), ("3584", 3), ("3611", 3), ("3704", 4), ("4129", 1), ("4410", 1), ("4455", 6), ("4568", 6),
|
|
("4852", 2), ("5043", 4), ("5104", 1), ("5105", 5), ("5106", 3), ("5127", 3), ("5936", 10), ("6358", 2),
|
|
("6602", 6), ("6861", 2), ("7021", 1), ("7216", 10), ("7788", 2), ("7790", 12), ("8147", 3), ("8301", 2),
|
|
("8890", 1), ("9064", 1), ("9077", 3), ("9155", 1),
|
|
]
|
|
|
|
|
|
def split_kzk() -> tuple[dict[str, int], dict[str, int]]:
|
|
photo: dict[str, int] = {}
|
|
text: dict[str, int] = {}
|
|
left = 64
|
|
for ending, qty in KZK_ALL:
|
|
take = min(left, qty)
|
|
if take:
|
|
photo[ending] = take
|
|
left -= take
|
|
if qty - take:
|
|
text[ending] = qty - take
|
|
return photo, text
|
|
|
|
|
|
def yesterday_orders() -> list[dict]:
|
|
photo, text = split_kzk()
|
|
return [
|
|
{"shop": "KZK", "time": "12:41", "photo": True, "lines": photo},
|
|
{"shop": "KZK", "time": "11:52", "photo": False, "lines": text, "edited": "13:05"},
|
|
{"shop": "SKM", "time": "12:18", "photo": False, "lines": {
|
|
"0887": 1, "0166": 1, "3611": 1, "4568": 2, "6861": 1, "0018": 1, "0057": 2, "2851": 1, "6184": 1,
|
|
"0119": 1, "9155": 20, "5124": 5, "0003": 3, "7731": 2, "4402": 4, "1288": 3, "3399": 2, "8020": 3}},
|
|
{"shop": "SKM", "time": "10:47", "photo": False, "lines": {
|
|
"1001": 10, "1102": 12, "2203": 8, "3304": 14, "4405": 6, "5506": 10, "6607": 9, "7708": 7,
|
|
"8809": 10}},
|
|
{"shop": "ULR", "time": "11:30", "photo": True, "lines": {
|
|
"1360": 1, "1903": 1, "1010": 20, "2020": 15, "3030": 12, "4040": 18, "5050": 14, "6060": 10,
|
|
"7070": 12, "8080": 7}},
|
|
{"shop": "PTM", "time": "10:05", "photo": False, "lines": {
|
|
"3821": 1, "1111": 15, "2222": 12, "3333": 10, "4444": 8, "5555": 10, "6666": 9, "7777": 5}},
|
|
]
|
|
|
|
|
|
def today_orders() -> list[dict]:
|
|
return [
|
|
{"shop": "KZK", "time": "09:20", "photo": False, "lines": {"0042": 10, "7311": 5, "2208": 5}},
|
|
{"shop": "KZK", "time": "11:10", "photo": False, "lines": {"0987": 2, "4455": 4, "9064": 3, "1573": 6}},
|
|
{"shop": "SKM", "time": "10:15", "photo": False, "lines": {"1102": 8, "3304": 6, "9155": 10, "5124": 2}},
|
|
{"shop": "ULR", "time": "12:05", "photo": True, "lines": {"5521": 8, "3190": 2}},
|
|
{"shop": "PTM", "time": "09:45", "photo": False, "lines": {"1111": 10, "4444": 5, "6666": 4}},
|
|
]
|
|
|
|
|
|
def hash_password(password: str) -> str:
|
|
salt = secrets.token_bytes(16)
|
|
key = hashlib.scrypt(password.encode("utf-8"), salt=salt, n=16384, r=8, p=1, dklen=64, maxmem=64 * 1024 * 1024)
|
|
return "$".join(["scrypt", "16384", "8", "1", base64.b64encode(salt).decode(), base64.b64encode(key).decode()])
|
|
|
|
|
|
def db_path_from_env() -> Path:
|
|
value = os.environ.get("DB_PATH")
|
|
env_file = ROOT / ".env"
|
|
if value is None and env_file.exists():
|
|
for raw in env_file.read_text(encoding="utf-8").splitlines():
|
|
line = raw.strip()
|
|
if line.startswith("DB_PATH="):
|
|
value = line.split("=", 1)[1].strip().strip("'\"")
|
|
return (ROOT / (value or "./data/srs.db")).resolve()
|
|
|
|
|
|
def utc_stamp(day: date, hhmm: str) -> str:
|
|
h, m = (int(x) for x in hhmm.split(":"))
|
|
at = datetime(day.year, day.month, day.day, h, m, tzinfo=IST).astimezone(timezone.utc)
|
|
return at.strftime("%Y-%m-%dT%H:%M:%SZ")
|
|
|
|
|
|
def raw_text(code: str, lines: dict[str, int]) -> str:
|
|
return "\n".join([code] + [f"{e} - {q}" for e, q in sorted(lines.items())])
|
|
|
|
|
|
def main() -> int:
|
|
ap = argparse.ArgumentParser(description="Seed the local SRS database with example data.")
|
|
ap.add_argument("--db", help="database file (default: DB_PATH from .env, or ./data/srs.db)")
|
|
ap.add_argument("--date", help="treat this IST day as today (YYYY-MM-DD)")
|
|
args = ap.parse_args()
|
|
|
|
path = Path(args.db).resolve() if args.db else db_path_from_env()
|
|
if not path.exists():
|
|
print(f"No database at {path}. Run `npm run migrate` first.")
|
|
return 1
|
|
today = date.fromisoformat(args.date) if args.date else datetime.now(IST).date()
|
|
yesterday = today - timedelta(days=1)
|
|
now_ist = datetime.now(IST)
|
|
|
|
db = sqlite3.connect(path, timeout=10)
|
|
db.execute("PRAGMA foreign_keys = ON")
|
|
tables = {r[0] for r in db.execute("SELECT name FROM sqlite_master WHERE type = 'table'")}
|
|
if not {"schema_migrations", "users", "shops", "orders", "order_lines", "draws", "prizes"} <= tables:
|
|
print(f"{path} has no SRS schema. Run `npm run migrate` first.")
|
|
return 1
|
|
|
|
summary: list[str] = []
|
|
try:
|
|
with db:
|
|
# Shops
|
|
shop_ids: dict[str, int] = {}
|
|
for code, name in SHOPS:
|
|
row = db.execute("SELECT id FROM shops WHERE code = ?", (code,)).fetchone()
|
|
if row:
|
|
shop_ids[code] = row[0]
|
|
else:
|
|
shop_ids[code] = db.execute(
|
|
"INSERT INTO shops (name, code) VALUES (?, ?)", (name, code)).lastrowid
|
|
summary.append(f"shop {code} {name}")
|
|
|
|
user_ids: dict[str, int] = {}
|
|
for username, codes, active in USERS:
|
|
row = db.execute(
|
|
"SELECT id FROM users WHERE username = ? COLLATE NOCASE AND deleted_at IS NULL",
|
|
(username,)).fetchone()
|
|
if row:
|
|
user_ids[username] = row[0]
|
|
continue
|
|
uid = db.execute(
|
|
"INSERT INTO users (username, password_hash, phone, is_admin, active) VALUES (?, ?, NULL, 0, ?)",
|
|
(username, hash_password(PASSWORD), 1 if active else 0)).lastrowid
|
|
for code in codes:
|
|
db.execute("INSERT INTO user_shops (user_id, shop_id) VALUES (?, ?)", (uid, shop_ids[code]))
|
|
user_ids[username] = uid
|
|
summary.append(f"user {username} / {PASSWORD}" + ("" if active else " (inactive)"))
|
|
|
|
admin = db.execute(
|
|
"SELECT id FROM users WHERE is_admin = 1 AND deleted_at IS NULL ORDER BY id LIMIT 1").fetchone()
|
|
if not admin:
|
|
raise SystemExit("No admin account. Create one with `npm run user create` first.")
|
|
author = {"KZK": user_ids["kzk.staff"], "SKM": user_ids["skm.staff"], "ULR": admin[0], "PTM": admin[0]}
|
|
|
|
# Orders
|
|
seeded_shops = tuple(shop_ids.values())
|
|
marks = ",".join("?" * len(seeded_shops))
|
|
|
|
def add_orders(day: date, orders: list[dict]) -> None:
|
|
iso = day.isoformat()
|
|
exists = db.execute(
|
|
f"SELECT COUNT(*) FROM orders WHERE draw_date = ? AND shop_id IN ({marks})",
|
|
(iso, *seeded_shops)).fetchone()[0]
|
|
if exists:
|
|
summary.append(f"{iso}: skipped orders, these shops already have {exists}")
|
|
return
|
|
tickets = 0
|
|
for o in orders:
|
|
t = o["time"]
|
|
if day == now_ist.date():
|
|
t = min(t, now_ist.strftime("%H:%M"))
|
|
oid = db.execute(
|
|
"INSERT INTO orders (shop_id, draw_date, source, created_by_user_id, raw_text, created_at,"
|
|
" edited_at) VALUES (?, ?, ?, ?, ?, ?, ?)",
|
|
(shop_ids[o["shop"]], iso, "whatsapp_photo" if o["photo"] else "whatsapp_text",
|
|
author[o["shop"]], raw_text(o["shop"], o["lines"]), utc_stamp(day, t),
|
|
utc_stamp(day, o["edited"]) if o.get("edited") else None)).lastrowid
|
|
db.executemany(
|
|
"INSERT INTO order_lines (order_id, ending, qty) VALUES (?, ?, ?)",
|
|
[(oid, e, q) for e, q in o["lines"].items()])
|
|
tickets += sum(o["lines"].values())
|
|
summary.append(f"{iso}: {len(orders)} orders, {tickets} tickets")
|
|
|
|
add_orders(yesterday, yesterday_orders())
|
|
add_orders(today, today_orders())
|
|
|
|
y = yesterday.isoformat()
|
|
if db.execute("SELECT 1 FROM draws WHERE draw_date = ?", (y,)).fetchone():
|
|
summary.append(f"{y}: a draw already exists, kept it (no demo draw added)")
|
|
else:
|
|
scheme, prefix = SCHEMES[yesterday.weekday()]
|
|
code = f"{prefix}-{900 + yesterday.toordinal() % 100}"
|
|
draw_id = db.execute(
|
|
"INSERT INTO draws (draw_date, scheme, draw_code) VALUES (?, ?, ?)", (y, scheme, code)).lastrowid
|
|
db.executemany(
|
|
"INSERT INTO prizes (draw_id, tier, amount, ending) VALUES (?, ?, ?, ?)",
|
|
[(draw_id, t, a, e) for e, (t, a) in PRIZES.items()])
|
|
if "result_sends" in tables:
|
|
db.execute(
|
|
"INSERT OR IGNORE INTO result_sends (draw_id, user_id) SELECT ?, id FROM users", (draw_id,))
|
|
summary.append(f"{y}: draw {scheme} {code} with {len(PRIZES)} prizes")
|
|
except sqlite3.OperationalError as e:
|
|
print(f"Could not write to {path}: {e}")
|
|
return 1
|
|
finally:
|
|
db.close()
|
|
|
|
print(f"Seeded {path}")
|
|
print(f"Today is {today.isoformat()} (IST)")
|
|
for line in summary or ["nothing to add: everything is already there"]:
|
|
print(f" {line}")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|