File size: 4,383 Bytes
ea2f1f1 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 | """
CONSTABLE – SQLite database helpers.
Tables:
employees – id (TEXT PK), name (TEXT), registered_at (TEXT)
attendance – id (INTEGER PK), employee_id (TEXT FK), timestamp (TEXT), date (TEXT)
"""
import sqlite3
import os
from datetime import datetime, date
DB_DIR = os.path.join(os.path.dirname(__file__))
DB_PATH = os.path.join(DB_DIR, "constable.db")
def get_connection():
conn = sqlite3.connect(DB_PATH, check_same_thread=False)
conn.row_factory = sqlite3.Row
return conn
def init_db():
"""Create tables if they don't exist."""
os.makedirs(DB_DIR, exist_ok=True)
conn = get_connection()
cur = conn.cursor()
cur.executescript("""
CREATE TABLE IF NOT EXISTS employees (
id TEXT PRIMARY KEY,
name TEXT NOT NULL,
registered_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS attendance (
id INTEGER PRIMARY KEY AUTOINCREMENT,
employee_id TEXT NOT NULL,
timestamp TEXT NOT NULL,
date TEXT NOT NULL,
FOREIGN KEY (employee_id) REFERENCES employees(id)
);
""")
conn.commit()
conn.close()
# ---------------------------------------------------------------------------
# Employee helpers
# ---------------------------------------------------------------------------
def add_employee(employee_id: str, name: str) -> bool:
"""Insert or replace an employee record. Returns True on success."""
conn = get_connection()
try:
conn.execute(
"INSERT OR REPLACE INTO employees (id, name, registered_at) VALUES (?, ?, ?)",
(employee_id, name, datetime.now().isoformat(timespec="seconds")),
)
conn.commit()
return True
except Exception as e:
print(f"[DB] add_employee error: {e}")
return False
finally:
conn.close()
def get_employee(employee_id: str):
"""Return employee row or None."""
conn = get_connection()
try:
row = conn.execute(
"SELECT * FROM employees WHERE id = ?", (employee_id,)
).fetchone()
return dict(row) if row else None
finally:
conn.close()
def get_all_employees():
conn = get_connection()
try:
rows = conn.execute("SELECT * FROM employees ORDER BY registered_at DESC").fetchall()
return [dict(r) for r in rows]
finally:
conn.close()
def delete_employee(employee_id: str) -> bool:
"""Delete an employee and their attendance records. Returns True on success."""
conn = get_connection()
try:
conn.execute("DELETE FROM attendance WHERE employee_id = ?", (employee_id,))
conn.execute("DELETE FROM employees WHERE id = ?", (employee_id,))
conn.commit()
return True
except Exception as e:
print(f"[DB] delete_employee error: {e}")
return False
finally:
conn.close()
# ---------------------------------------------------------------------------
# Attendance helpers
# ---------------------------------------------------------------------------
def mark_attendance(employee_id: str) -> dict:
"""
Log attendance for today.
Returns {'status': 'success'|'already_marked', 'timestamp': ...}
"""
today = date.today().isoformat()
conn = get_connection()
try:
existing = conn.execute(
"SELECT id FROM attendance WHERE employee_id = ? AND date = ?",
(employee_id, today),
).fetchone()
if existing:
return {"status": "already_marked"}
ts = datetime.now().strftime("%I:%M %p")
conn.execute(
"INSERT INTO attendance (employee_id, timestamp, date) VALUES (?, ?, ?)",
(employee_id, ts, today),
)
conn.commit()
return {"status": "success", "timestamp": ts}
finally:
conn.close()
def get_today_attendance():
today = date.today().isoformat()
conn = get_connection()
try:
rows = conn.execute(
"""SELECT a.timestamp, e.id, e.name
FROM attendance a
JOIN employees e ON a.employee_id = e.id
WHERE a.date = ?
ORDER BY a.id DESC""",
(today,),
).fetchall()
return [dict(r) for r in rows]
finally:
conn.close()
|