coffee_getter/db.py

   1 from sqlite3 import connect
   2
   3
   4 def Q(q, b="", t=["now", "-7 days"]):
   5     """Return db query identified by `q`.
   6
   7     :param q: Identifier of query.
   8     :param b: Optionally, specify the beverage(s).
   9     :param t: Optionally, specify the time range.
  10     """
  11     assert isinstance(t, tuple) or isinstance(t, list)
  12     assert len(t) == 2
  13     if t[0] == "now":
  14         dtf = f"datetime('now', 'localtime', '{t[1]}')"
  15         dtt = "datetime('now', 'localtime')"
  16     elif t[1] == "now":
  17         dtf = f"datetime('now', 'localtime', '{t[0]}')"
  18         dtt = "datetime('now', 'localtime')"
  19     else:
  20         dtf = f"datetime('{t[0]}', 'localtime')"
  21         dtt = f"datetime('{t[1]}', 'localtime')"
  22     if q == "get_drinks":
  23         return f"""
  24
  25         SELECT count(*), flavor
  26         FROM coffees
  27         WHERE time BETWEEN
  28             {dtf}
  29             AND {dtt}
  30         GROUP BY flavor
  31
  32         """
  33     elif q == "get_drinkers_of":
  34         if isinstance(b, tuple) or isinstance(b, list):
  35             assert len(b) > 0
  36             f = f"WHERE flavor = '{b[0]}'"
  37             for i in b[1:]:
  38                 f += f" OR flavor = '{i}'"
  39         else:
  40             assert b != ""
  41             f = f"WHERE flavor = '{b}'"
  42         return f"""
  43
  44         SELECT count(*), users.name FROM coffees
  45         LEFT JOIN identifiers on coffees.id = identifiers.userid
  46         LEFT JOIN users on identifiers.userid = users.id
  47         {f}
  48         AND coffees.time BETWEEN
  49             {dtf}
  50             AND {dtt}
  51         GROUP BY identifiers.userid
  52
  53         """
  54
  55
  56 class Db:
  57     def __init__(self, db_path):
  58         self.con = connect(db_path)
  59         self.cur = self.con.cursor()
  60
  61     def __del__(self):
  62         if self.con:
  63             self.con.close()
  64
  65     def get_top_drinks(self):
  66         q = Q("get_drinks")
  67         top = []
  68         for (cnt, dn) in self.cur.execute(q):
  69             top.append((dn, cnt))
  70         top.sort(key=lambda x: (x[1], x[0]), reverse=True)
  71         return tuple(top)
  72
  73     def get_top_mate_drinkers(self):
  74         """Return list of pairs of name, count for Mate drinkers."""
  75         users = {}
  76         que = Q("get_drinkers_of", "Club-Mate 0,5 l")
  77         for (cnt, un) in self.cur.execute(que):
  78             users[un] = cnt * 0.5
  79         que = Q("get_drinkers_of", "Club-Mate 0,33 l")
  80         for (cnt, un) in self.cur.execute(que):
  81             if un in users:
  82                 users[un] += cnt * 0.33
  83             else:
  84                 users[un] = cnt * 0.33
  85         top = []
  86         for (un, cnt) in users.items():
  87             top.append((un, cnt))
  88         top.sort(key=lambda x: (x[1], x[0]), reverse=True)
  89         return tuple(top)
  90
  91     def get_top_tea_drinkers(self):
  92         """Return list of pairs of name, count for tea drinkers."""
  93         q = Q("get_drinkers_of", "tea")
  94         top = []
  95         for (cnt, un) in self.cur.execute(q):
  96             top.append((un, cnt))
  97         top.sort(key=lambda x: (x[1], x[0]), reverse=True)
  98         return tuple(top)
  99
 100     def get_drunk_sum(self, *args, **kwargs):
 101         """Return list of drunken ``flavor`` from ``dtf`` to ``dtt``.
 102
 103         Keyword arguments:
 104         flavor -- The flavor of beverage.
 105         dtf -- Date and time *from*.
 106         dtt -- Date and time *to*.
 107         """
 108         assert ((len(args) == 3 and len(kwargs) == 0)
 109                 or (len(args) == 0 and len(kwargs) == 3))
 110         if args:
 111             flavor = args[0]
 112             dtf = args[1]
 113             dtt = args[2]
 114         elif kwargs:
 115             flavor = kwargs["flavor"]
 116             dtf = kwargs["dtf"]
 117             dtt = kwargs["dtt"]
 118         que = Q("get_drinkers_of", flavor.split(";"), (dtf, dtt))
 119         drunk = []
 120         for (cnt, un) in self.cur.execute(que):
 121             drunk.append((un, cnt))
 122         return tuple(drunk)
 123
 124     def get_drunk_list(self, *args, **kwargs):
 125         """Return dict of lists of drunken ``flavor`` from ``dtf`` to ``dtt``.
 126
 127         Keyword arguments:
 128         flavor -- The flavor of beverage.
 129         dtf -- Date and time *from*.
 130         dtt -- Date and time *to*.
 131         """
 132         assert ((len(args) == 3 and len(kwargs) == 0)
 133                 or (len(args) == 0 and len(kwargs) == 3))
 134         if args:
 135             flavor = args[0]
 136             dtf = args[1]
 137             dtt = args[2]
 138         elif kwargs:
 139             flavor = kwargs["flavor"]
 140             dtf = kwargs["dtf"]
 141             dtt = kwargs["dtt"]
 142         flavors = flavor.split(";")
 143         drunk = {}
 144         i = 0
 145         for f in flavors:
 146             que = Q("get_drinkers_of", f, (dtf, dtt))
 147             for (cnt, un) in self.cur.execute(que):
 148                 if un not in drunk:
 149                     drunk[un] = [0 for j in range(i)]
 150                 drunk[un].append(cnt)
 151             i += 1
 152         return drunk