84 lines
2.1 KiB
Python
84 lines
2.1 KiB
Python
from .database import DatabaseController
|
|
from models.stats import Stats
|
|
from datetime import datetime
|
|
from utils.general_utils import is_within_x_days
|
|
|
|
|
|
class StatsController(DatabaseController):
|
|
FIELDS = ['id', 'userID', 'productID', 'viewDate']
|
|
TYPE = Stats
|
|
|
|
def __init__(self):
|
|
super().__init__()
|
|
|
|
def create(self, view: Stats):
|
|
params = [
|
|
view.userID,
|
|
view.productID,
|
|
view.viewDate
|
|
]
|
|
query = """
|
|
INSERT INTO Views
|
|
(userID, productID, viewDate)
|
|
VALUES (?, ?, ?)
|
|
"""
|
|
|
|
self.do(query, params)
|
|
|
|
def read(self) -> list[Stats] | None:
|
|
query = "SELECT * FROM Views"
|
|
self.get_many(query, [])
|
|
|
|
def read_product(self, product_id: int = 0) -> list[Stats] | None:
|
|
params = [
|
|
product_id
|
|
]
|
|
query = """
|
|
SELECT * FROM Views
|
|
WHERE productID = ?
|
|
"""
|
|
|
|
return self.get_many(query, params)
|
|
|
|
def read_user(self, user_id: int) -> list[Stats] | None:
|
|
params = [
|
|
user_id
|
|
]
|
|
query = """
|
|
SELECT * FROM Views
|
|
WHERE userID = ?
|
|
"""
|
|
|
|
return self.get_many(query, params)
|
|
|
|
def read_days(self, product_id: int, prev_days: int = 7
|
|
) -> dict[int, list[Stats]] | None:
|
|
""" Returns data from within the given number of days """
|
|
data = self.read_product(product_id)
|
|
|
|
# Filter the values to only be within X prev days
|
|
filtered_data = list()
|
|
if data is not None:
|
|
filtered_data = list(filter(
|
|
lambda d: is_within_x_days(d.viewDate, prev_days),
|
|
data
|
|
))
|
|
|
|
day_views: dict[int, list[Stats]] = dict()
|
|
|
|
for i in range(0, prev_days):
|
|
day_views[i] = list()
|
|
|
|
# Organise data into distinct
|
|
for view in filtered_data:
|
|
diff = datetime.now()-view.viewDate
|
|
day_views[diff.days].append(view)
|
|
|
|
return day_views
|
|
|
|
def update(self):
|
|
print("Doing work")
|
|
|
|
def delete(self):
|
|
print("Doing work")
|