88 lines
2.7 KiB
Python
88 lines
2.7 KiB
Python
import sqlite3
|
|
from datetime import datetime, timedelta
|
|
from typing import List, Tuple
|
|
|
|
AUTO_CHECKOUT_HOURS = 6
|
|
|
|
|
|
class CheckinDB:
|
|
def __init__(self, db_name: str) -> None:
|
|
self.con = sqlite3.connect(db_name)
|
|
self.con.row_factory = sqlite3.Row
|
|
self.try_create_tables()
|
|
|
|
def try_create_tables(self) -> bool:
|
|
try:
|
|
with self.con:
|
|
self.con.execute(
|
|
"CREATE TABLE checkin (ID INTEGER PRIMARY KEY AUTOINCREMENT, NAME VARCHAR UNIQUE, IS_PRESENT BOOL, UPDATED INT)"
|
|
)
|
|
print("Creating new table")
|
|
except sqlite3.OperationalError:
|
|
print("Table already exists in database, nothing to create.")
|
|
return True
|
|
|
|
def update(self, name: str, is_present: bool) -> bool:
|
|
present_int = 1 if is_present else 0
|
|
with self.con:
|
|
self.con.execute(
|
|
f"INSERT INTO checkin (NAME, IS_PRESENT, UPDATED) VALUES ('{name}', '{present_int}', strftime('%s','now')) ON CONFLICT(name) DO UPDATE SET is_present=excluded.is_present"
|
|
)
|
|
return True
|
|
return False
|
|
|
|
def get_full_entries(self) -> List[Tuple]:
|
|
with self.con:
|
|
return [
|
|
(x["name"], x["is_present"], x["updated"])
|
|
for x in self.con.execute(
|
|
"SELECT name, is_present, updated FROM checkin ORDER BY name"
|
|
)
|
|
]
|
|
|
|
def get_checkin_names(self) -> List[str]:
|
|
with self.con:
|
|
return [
|
|
x["name"]
|
|
for x in self.con.execute(
|
|
"SELECT name FROM checkin WHERE is_present ORDER BY name"
|
|
)
|
|
]
|
|
|
|
def get_checkin_count(self) -> int:
|
|
with self.con:
|
|
return self.con.execute(
|
|
"SELECT COUNT(name) FROM checkin WHERE is_present"
|
|
).fetchone()[0]
|
|
|
|
def clear_rows(self) -> None:
|
|
with self.con:
|
|
self.con.execute("DELETE FROM checkin")
|
|
|
|
def get_updated_for_name(self, name: str) -> datetime:
|
|
with self.con:
|
|
return datetime.fromtimestamp(
|
|
float(
|
|
self.con.execute(
|
|
f"SELECT updated FROM checkin WHERE name = '{name}'"
|
|
).fetchone()[0]
|
|
)
|
|
)
|
|
|
|
def auto_checkout(self) -> None:
|
|
filter_time = (
|
|
datetime.now() - timedelta(hours=AUTO_CHECKOUT_HOURS)
|
|
).timestamp()
|
|
with self.con:
|
|
self.con.execute(
|
|
f"UPDATE checkin SET IS_PRESENT = 0 WHERE updated <= {filter_time}"
|
|
)
|
|
|
|
|
|
def main() -> None:
|
|
counter_bot = CheckinDB("checkin.db")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|