3 from abc import ABC, abstractmethod
11 from typing import Any, Callable, List, Optional, Set, Tuple
18 """Base class of a thing that chooses pages"""
20 def get_page_list(self) -> List[str]:
22 valid_filename = re.compile("([^_]+)_(\d+)_([^\.]+)\.html")
26 for f in os.listdir(constants.pages_dir)
27 if os.path.isfile(os.path.join(constants.pages_dir, f))
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))
36 os.path.getmtime(os.path.join(constants.pages_dir, page))
38 age = now - last_modified
39 if age > freshness_requirement:
40 print(f'chooser: "{page}" is too old.')
42 filenames.append(page)
46 def choose_next_page(self) -> Any:
50 class weighted_random_chooser(chooser):
51 """Chooser that does it via weighted RNG."""
53 def __init__(self, filter_list: Optional[List[Callable[[str], bool]]]) -> None:
55 self.valid_filename = re.compile("([^_]+)_(\d+)_([^\.]+)\.html")
56 self.pages: Optional[List[str]] = None
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)
63 def dont_choose_page_twice_in_a_row_filter(self, choice: str) -> bool:
64 if choice == self.last_choice:
66 self.last_choice = choice
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()
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
85 random_pick = random.randrange(0, total_weight - 1)
87 for x in range(0, len(weights)):
89 if so_far > random_pick:
91 choice = self.pages[x]
93 # Allow filter list to suppress pages.
94 choice_is_filtered = False
95 for f in self.filter_list:
97 print(f"chooser: {choice} filtered by {f.__name__}")
98 choice_is_filtered = True
100 if choice_is_filtered:
108 class weighted_random_chooser_with_triggers(weighted_random_chooser):
109 """Same as WRC but has trigger events"""
113 trigger_list: Optional[List[trigger.trigger]],
114 filter_list: List[Callable[[str], bool]],
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(())
122 def check_for_triggers(self) -> bool:
124 for t in self.trigger_list:
125 x = t.get_triggered_page_list()
126 if x is not None and len(x) > 0:
128 self.page_queue.add(y)
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()
136 triggered = self.check_for_triggers()
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...")
144 for t in self.page_queue:
145 if priority is None or t[1] > priority:
148 assert(page is not None)
149 assert(priority is not None)
150 self.page_queue.remove((page, priority))
151 return (page, triggered)
153 # Always show the clock in the middle of the night.
155 for page in self.pages:
159 # Fall back on weighted random choice.
161 return (weighted_random_chooser.choose_next_page(self), False)
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 (
171 # or "mynorthwest" in page
172 # or "seattle" in page
173 # or "stranger" in page
174 # or "twitter" in page
177 # x = weighted_random_chooser_with_triggers([], [ filter_news_during_dinnertime ])
178 # print(x.choose_next_page())