3 from abc import ABC, abstractmethod
11 from typing import Callable, List
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)
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) -> str:
50 class weighted_random_chooser(chooser):
51 """Chooser that does it via weighted RNG."""
53 def __init__(self, filter_list: List[Callable[[str], bool]]) -> None:
55 self.valid_filename = re.compile("([^_]+)_(\d+)_([^\.]+)\.html")
58 self.filter_list = filter_list
59 if filter_list is None:
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) -> str:
70 if self.pages == 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)
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: List[trigger.trigger],
114 filter_list: List[Callable[[str], bool]],
116 weighted_random_chooser.__init__(self, filter_list)
117 self.trigger_list = trigger_list
118 if trigger_list is None:
119 self.trigger_list = []
120 self.page_queue = 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 != None and len(x) > 0:
128 self.page_queue.add(y)
132 def choose_next_page(self) -> str:
133 if self.pages == 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 if len(self.page_queue) > 0:
140 print("chooser: Pulling page from queue...")
143 for t in self.page_queue:
144 if priority == None or t[1] > priority:
147 self.page_queue.remove((page, priority))
148 return page, triggered
150 # Fall back on weighted random choice.
152 return weighted_random_chooser.choose_next_page(self), False
156 # def filter_news_during_dinnertime(page):
157 # now = datetime.datetime.now()
158 # is_dinnertime = now.hour >= 17 and now.hour <= 20
159 # return not is_dinnertime or not (
162 # or "mynorthwest" in page
163 # or "seattle" in page
164 # or "stranger" in page
165 # or "twitter" in page
168 # x = weighted_random_chooser_with_triggers([], [ filter_news_during_dinnertime ])
169 # print(x.choose_next_page())