70 lines
2.1 KiB
Python
70 lines
2.1 KiB
Python
import sqlite3
|
|
|
|
|
|
class Database:
|
|
def __init__(self):
|
|
self.db_path = "data/shifts.db"
|
|
self._setup_tables()
|
|
pass
|
|
|
|
def _get_connection(self):
|
|
return sqlite3.connect(self.db_path)
|
|
|
|
def _setup_tables(self):
|
|
# create the shifts table.
|
|
con = self._get_connection()
|
|
cur = con.cursor()
|
|
cur.execute('''
|
|
CREATE TABLE IF NOT EXISTS shifts (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
shift_start DATETIME NOT NULL,
|
|
shift_end DATETIME NOT NULL,
|
|
department TEXT NOT NULL,
|
|
duration TEXT NOT NULL,
|
|
description TEXT,
|
|
fetched_at DATETIME DEFAULT CURRENT_TIMESTAMP,
|
|
|
|
UNIQUE(shift_start, department) -- No duplicates
|
|
);
|
|
''')
|
|
con.commit()
|
|
con.close()
|
|
|
|
|
|
def insert_shifts(self, shifts):
|
|
# start db connection
|
|
con = self._get_connection()
|
|
cur = con.cursor()
|
|
inserted = 0
|
|
|
|
for shift in shifts:
|
|
# parse all information from the shift
|
|
start = shift.get('start_datetime', 'N/A')
|
|
end = shift.get('end_datetime', 'N/A')
|
|
|
|
department = shift.get('department_name', 'N/A')
|
|
duration = shift.get('duration', 'N/A')
|
|
description = shift.get('shift_remark', shift.get('description', 'N/A'))
|
|
|
|
# insert it into the db.
|
|
cur.execute("""
|
|
INSERT OR IGNORE INTO shifts
|
|
(shift_start, shift_end, department, duration, description)
|
|
VALUES (?, ?, ?, ?, ?)
|
|
""", (start, end, department, duration, description))
|
|
|
|
if cur.rowcount > 0:
|
|
inserted += 1
|
|
|
|
con.commit()
|
|
con.close()
|
|
print(f"✅ Inserted {inserted}/{len(shifts)} new shifts")
|
|
|
|
def delete_future_shifts(self):
|
|
con = self._get_connection()
|
|
cur = con.cursor()
|
|
cur.execute("DELETE FROM shifts WHERE shift_start > current_timestamp")
|
|
con.commit()
|
|
con.close()
|
|
print(f"✅ Deleted all future shifts")
|