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 (coffees.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             f += ")"
  40         else:
  41             assert b != ""
  42             f = f"WHERE flavor = '{b}'"
  43         return f"""
  44
  45         SELECT count(*), users.name FROM coffees
  46         LEFT JOIN identifiers on coffees.id = identifiers.userid
  47         LEFT JOIN users on identifiers.userid = users.id
  48         {f}
  49         AND (coffees.time BETWEEN
  50             {dtf}
  51             AND {dtt})
  52         GROUP BY identifiers.userid
  53
  54         """
  55
  56
  57 class Db:
  58     def __init__(self, db_path):
  59         self.con = connect(db_path)
  60         self.cur = self.con.cursor()
  61
  62     def __del__(self):
  63         if self.con:
  64             self.con.close()
  65
  66     def get_top_drinks(self):
  67         q = Q("get_drinks")
  68         top = []
  69         for (cnt, dn) in self.cur.execute(q):
  70             top.append((dn, cnt))
  71         top.sort(key=lambda x: (x[1], x[0]), reverse=True)
  72         return tuple(top)
  73
  74     def get_top_mate_drinkers(self):
  75         """Return list of pairs of name, count for Mate drinkers."""
  76         users = {}
  77         que = Q("get_drinkers_of", "Club-Mate 0,5 l")
  78         for (cnt, un) in self.cur.execute(que):
  79             users[un] = cnt * 0.5
  80         que = Q("get_drinkers_of", "Club-Mate 0,33 l")
  81         for (cnt, un) in self.cur.execute(que):
  82             if un in users:
  83                 users[un] += cnt * 0.33
  84             else:
  85                 users[un] = cnt * 0.33
  86         top = []
  87         for (un, cnt) in users.items():
  88             top.append((un, cnt))
  89         top.sort(key=lambda x: (x[1], x[0]), reverse=True)
  90         return tuple(top)
  91
  92     def get_top_tea_drinkers(self):
  93         """Return list of pairs of name, count for tea drinkers."""
  94         q = Q("get_drinkers_of", "tea")
  95         top = []
  96         for (cnt, un) in self.cur.execute(q):
  97             top.append((un, cnt))
  98         top.sort(key=lambda x: (x[1], x[0]), reverse=True)
  99         return tuple(top)
 100
 101     def get_drunk_sum(self, flavor, dtf, dtt):
 102         """Return list of drunken ``flavor`` from ``dtf`` to ``dtt``.
 103
 104         Keyword arguments:
 105         flavor -- The flavor of beverage.
 106         dtf -- Date and time *from*.
 107         dtt -- Date and time *to*.
 108         """
 109         que = Q("get_drinkers_of", flavor.split(";"), (dtf, dtt))
 110         drunk = []
 111         for (cnt, un) in self.cur.execute(que):
 112             drunk.append((un, cnt))
 113         return tuple(drunk)
 114
 115     def get_drunk_list(self, flavor, dtf, dtt):
 116         """Return dict of lists of drunken ``flavor`` from ``dtf`` to ``dtt``.
 117
 118         Keyword arguments:
 119         flavor -- The flavor of beverage.
 120         dtf -- Date and time *from*.
 121         dtt -- Date and time *to*.
 122         """
 123         flavors = flavor.split(";")
 124         drunk = {}
 125         i = 0
 126         for f in flavors:
 127             que = Q("get_drinkers_of", f, (dtf, dtt))
 128             for (cnt, un) in self.cur.execute(que):
 129                 if un not in drunk:
 130                     drunk[un] = [0 for j in range(i)]
 131                 drunk[un].append(cnt)
 132             i += 1
 133         return drunk