chooser.py

   1 #!/usr/bin/env python3
   2
   3 from abc import ABC, abstractmethod
   4 import datetime
   5 import glob
   6 import os
   7 import random
   8 import re
   9 import sys
  10 import time
  11 from typing import Any, Callable, List, Optional, Set, Tuple
  12
  13 import constants
  14 import trigger
  15
  16
  17 class chooser(ABC):
  18     """Base class of a thing that chooses pages"""
  19
  20     def get_page_list(self) -> List[str]:
  21         now = time.time()
  22         valid_filename = re.compile("([^_]+)_(\d+)_([^\.]+)\.html")
  23         filenames = []
  24         pages = [
  25             f
  26             for f in os.listdir(constants.pages_dir)
  27             if os.path.isfile(os.path.join(constants.pages_dir, f))
  28         ]
  29         for page in pages:
  30             result = re.match(valid_filename, page)
  31             if result is not None:
  32                 print(f'chooser: candidate page: "{page}"')
  33                 if result.group(3) != "none":
  34                     freshness_requirement = int(result.group(3))
  35                     last_modified = int(
  36                         os.path.getmtime(os.path.join(constants.pages_dir, page))
  37                     )
  38                     age = now - last_modified
  39                     if age > freshness_requirement:
  40                         print(f'chooser: "{page}" is too old.')
  41                         continue
  42                 filenames.append(page)
  43         return filenames
  44
  45     @abstractmethod
  46     def choose_next_page(self) -> Any:
  47         pass
  48
  49
  50 class weighted_random_chooser(chooser):
  51     """Chooser that does it via weighted RNG."""
  52
  53     def __init__(self, filter_list: Optional[List[Callable[[str], bool]]]) -> None:
  54         self.last_choice = ""
  55         self.valid_filename = re.compile("([^_]+)_(\d+)_([^\.]+)\.html")
  56         self.pages: Optional[List[str]] = None
  57         self.count = 0
  58         self.filter_list: List[Callable[[str], bool]] = []
  59         if filter_list is not None:
  60             self.filter_list.extend(filter_list)
  61         self.filter_list.append(self.dont_choose_page_twice_in_a_row_filter)
  62
  63     def dont_choose_page_twice_in_a_row_filter(self, choice: str) -> bool:
  64         if choice == self.last_choice:
  65             return False
  66         self.last_choice = choice
  67         return True
  68
  69     def choose_next_page(self) -> Any:
  70         if self.pages is None or self.count % 100 == 0:
  71             self.pages = self.get_page_list()
  72
  73         total_weight = 0
  74         weights = []
  75         for page in self.pages:
  76             result = re.match(self.valid_filename, page)
  77             if result is not None:
  78                 weight = int(result.group(2))
  79                 weights.append(weight)
  80                 total_weight += weight
  81         if total_weight <= 0:
  82             raise Exception
  83
  84         while True:
  85             random_pick = random.randrange(0, total_weight - 1)
  86             so_far = 0
  87             for x in range(0, len(weights)):
  88                 so_far += weights[x]
  89                 if so_far > random_pick:
  90                     break
  91             choice = self.pages[x]
  92
  93             # Allow filter list to suppress pages.
  94             choice_is_filtered = False
  95             for f in self.filter_list:
  96                 if not f(choice):
  97                     print(f"chooser: {choice} filtered by {f.__name__}")
  98                     choice_is_filtered = True
  99                     break
 100             if choice_is_filtered:
 101                 continue
 102
 103             # We're good...
 104             self.count += 1
 105             return choice
 106
 107
 108 class weighted_random_chooser_with_triggers(weighted_random_chooser):
 109     """Same as WRC but has trigger events"""
 110
 111     def __init__(
 112         self,
 113         trigger_list: Optional[List[trigger.trigger]],
 114         filter_list: List[Callable[[str], bool]],
 115     ) -> None:
 116         weighted_random_chooser.__init__(self, filter_list)
 117         self.trigger_list: List[trigger.trigger] = []
 118         if trigger_list is not None:
 119             self.trigger_list.extend(trigger_list)
 120         self.page_queue: Set[Tuple[str, int]] = set(())
 121
 122     def check_for_triggers(self) -> bool:
 123         triggered = False
 124         for t in self.trigger_list:
 125             x = t.get_triggered_page_list()
 126             if x is not None and len(x) > 0:
 127                 for y in x:
 128                     self.page_queue.add(y)
 129                     triggered = True
 130         return triggered
 131
 132     def choose_next_page(self) -> Tuple[str, bool]:
 133         if self.pages is None or self.count % 100 == 0:
 134             self.pages = self.get_page_list()
 135
 136         triggered = self.check_for_triggers()
 137
 138         # First try to satisfy from the page queue.
 139         now = datetime.datetime.now()
 140         if len(self.page_queue) > 0:
 141             print("chooser: Pulling page from queue...")
 142             page = None
 143             priority = None
 144             for t in self.page_queue:
 145                 if priority is None or t[1] > priority:
 146                     page = t[0]
 147                     priority = t[1]
 148             assert(page is not None)
 149             assert(priority is not None)
 150             self.page_queue.remove((page, priority))
 151             return (page, triggered)
 152
 153         # Always show the clock in the middle of the night.
 154         elif now.hour < 7:
 155             for page in self.pages:
 156                 if "clock" in page:
 157                     return (page, False)
 158
 159         # Fall back on weighted random choice.
 160         else:
 161             return (weighted_random_chooser.choose_next_page(self), False)
 162
 163
 164 # Test
 165 # def filter_news_during_dinnertime(page):
 166 #    now = datetime.datetime.now()
 167 #    is_dinnertime = now.hour >= 17 and now.hour <= 20
 168 #    return not is_dinnertime or not (
 169 #        "cnn" in page
 170 #        or "news" in page
 171 #        or "mynorthwest" in page
 172 #        or "seattle" in page
 173 #        or "stranger" in page
 174 #        or "twitter" in page
 175 #        or "wsj" in page
 176 #    )
 177 # x = weighted_random_chooser_with_triggers([], [ filter_news_during_dinnertime ])
 178 # print(x.choose_next_page())