🎬 My Movie Collection

""" self.src = src self.dst = Path(dst) if dst else Path(os.path.dirname(sys.argv[0])) / 'index.html' self.movie_list = [] self.threads = [] self.api_key = api_key or os.environ.get('OMDB_API_KEY') if not self.api_key: sys.stderr.write( 'error: no OMDb API key found.\n' 'Get a free key at https://www.omdbapi.com/apikey.aspx and set it via:\n' ' export OMDB_API_KEY=your_key_here\n' ) sys.exit(1) self.read_prev_output() self.html_table = None # ------------------------------------------------------------------ # Internal helpers # ------------------------------------------------------------------ @staticmethod def _omdb_type(arg_kind: str): """Return the OMDb `type` filter value for a movie_list kind tag, or None if there's no good match (in which case OMDb's own fuzzy search decides without a type constraint).""" return OMDB_TYPE_MAP.get(arg_kind.lower()) def _omdb_request(self, params: dict): """GET against the OMDb API with the API key attached. Returns the parsed JSON dict. Raises requests.RequestException on network/HTTP failure (the caller is responsible for retrying).""" query = dict(params) query['apikey'] = self.api_key response = requests.get(OMDB_BASE_URL, params=query, timeout=15) response.raise_for_status() return response.json() @staticmethod def _is_daily_limit_error(data: dict) -> bool: return ( data.get('Response') == 'False' and 'request limit' in (data.get('Error') or '').lower() ) def _fallback_row(self, arg: dict, index: int) -> str: """Build a row from cached HTML (even if it has N/A fields) or, if nothing cached exists, from the raw movie_list data. Used when the daily OMDb quota has been exhausted so the run can still finish.""" match = [ row for row in self.prev_html if f'' in row ] if match: match_str = match[0].replace('*', arg['status']).replace('DONE', arg['status']) if '', 1)[1] return f'\n {after_index}' return match_str return ( f'\n ' f'' f'' f'' f'' f'' f'' f'' ) # ------------------------------------------------------------------ # Worker thread # ------------------------------------------------------------------ def _worker(self, arg, index): """Fetch OMDb data for one movie and write it into self.html_table.""" if DAILY_LIMIT_REACHED.is_set(): self.html_table[index] = self._fallback_row(arg, index) return omdb_type = self._omdb_type(arg['kind']) movie_detail = None first_run = True while True: if DAILY_LIMIT_REACHED.is_set(): break if not first_run: time.sleep(10) else: first_run = False try: # ---- attempt 1: direct title+year lookup ---- # OMDb's `t=` endpoint does fuzzy title matching server-side # and returns full details (rating, genre, etc.) in one call. params = {'t': arg['title'], 'plot': 'short'} if arg['year']: params['y'] = arg['year'] if omdb_type: params['type'] = omdb_type data = self._omdb_request(params) if self._is_daily_limit_error(data): DAILY_LIMIT_REACHED.set() sys.stderr.write('OMDb daily request limit reached — stopping further lookups.\n') break # ---- attempt 2: drop the year, in case it's slightly off ---- if data.get('Response') == 'False' and arg['year']: params.pop('y', None) data = self._omdb_request(params) if self._is_daily_limit_error(data): DAILY_LIMIT_REACHED.set() break # ---- attempt 3: fuzzy search, take the top hit, fetch by ID ---- if data.get('Response') == 'False': search_params = {'s': arg['title']} if omdb_type: search_params['type'] = omdb_type search_data = self._omdb_request(search_params) if self._is_daily_limit_error(search_data): DAILY_LIMIT_REACHED.set() break results = search_data.get('Search') or [] if results: data = self._omdb_request({'i': results[0]['imdbID'], 'plot': 'short'}) if self._is_daily_limit_error(data): DAILY_LIMIT_REACHED.set() break if data.get('Response') == 'True': movie_detail = data break except requests.RequestException as exc: sys.stderr.write(f'network error ({arg["title"]}): {exc}, retrying…\n') # ---- build the table row ---- if movie_detail: title = movie_detail.get('Title') or arg['title'] year = movie_detail.get('Year') or arg['year'] rating = movie_detail.get('imdbRating') or 'N/A' genres = movie_detail.get('Genre') or 'N/A' imdb_id = movie_detail.get('imdbID') or '' if imdb_id: html_title_td = ( f'' f'{title}' ) else: html_title_td = title elif DAILY_LIMIT_REACHED.is_set(): self.html_table[index] = self._fallback_row(arg, index) return else: # Nothing found — show the raw data from the list file title = arg['title'] year = arg['year'] rating = 'N/A' genres = 'N/A' html_title_td = title self.html_table[index] = ( f'\n ' f'' f'' f'' f'' f'' f'' f'' ) # ------------------------------------------------------------------ # Public interface # ------------------------------------------------------------------ def gen(self): """ Generate an HTML list based on input, using a threaded worker """ if not self.src: self.src = Path(os.path.dirname(sys.argv[0])) / 'movie_list' else: self.src = Path(self.src) if not self.src.exists(): sys.stderr.write(f'error: input does not exist - {self.src}\n') return False self.movie_list = {} seen_titles = set() with open(self.src, 'r', encoding='utf-8') as fp_handle: mlist_raw = fp_handle.read() idx = 0 for raw_line in mlist_raw.splitlines(): if not raw_line: continue title = raw_line[0:next( (i for i, ch in enumerate(raw_line) if ch in {'<', '('}), None ) - 1] if title in seen_titles: continue seen_titles.add(title) self.movie_list[idx] = { 'title': title, 'kind': raw_line[raw_line.find('<')+1:raw_line.rfind('>')+1].strip('<>') or 'movie', 'year': raw_line[raw_line.find('(')+1:raw_line.find(')')], 'status': raw_line[raw_line.find('[')+1:raw_line.find(']')], } idx += 1 self.html_table = [None] * len(self.movie_list) pbar = progressbar.ProgressBar(max_value=len(self.movie_list)) for idx, movie in self.movie_list.items(): # Re-use the cached row if available and complete (no N/A) match = [ row for row in self.prev_html if f'' in row and 'N/A' not in row ] if match: match_str = match[0] # Sync status field match_str = match_str.replace('*', movie['status']).replace('DONE', movie['status']) # Sync the row index number if '', 1)[1] if '' in match_str else match_str self.html_table[idx] = ( f'\n ' f'{after_index}' ) else: self.html_table[idx] = match_str pbar.increment() elif DAILY_LIMIT_REACHED.is_set(): # Quota already blown earlier in this run — don't bother # spawning a thread that will just no-op. self.html_table[idx] = self._fallback_row(movie, idx) pbar.increment() else: thread = threading.Thread(target=self._worker, args=(movie, idx)) self.threads.append(thread) max_threads = 10 while self.threads: threads_alive = self.get_alive_threads() threads_to_start = [t for t in self.threads if t not in threads_alive] for i in range(min(max_threads, len(threads_to_start))): threads_to_start[i].start() pbar.increment() time.sleep(2) # gentle rate-limiting between thread launches time.sleep(2) self.delete_finished_threads() return True def delete_finished_threads(self): for idx, thread in enumerate(self.threads): if not thread.is_alive() and thread._started.is_set(): thread.join() self.threads[idx] = None self.threads = [t for t in self.threads if t is not None] def get_alive_threads(self): return [t for t in self.threads if t.is_alive() or t._started.is_set()] def write(self, dst=None): """ Write the HTML list to index.html """ out_path = dst if dst else self.dst timestamp = time.strftime('%Y-%m-%d %H:%M:%S', time.gmtime()) final_html = self.html.split('')[0] + '' final_html += ''.join(row for row in self.html_table if row is not None) final_html += f'''

#	Title	Year	Rating	Genre	Status
' in match_str: after_index = match_str.split('
{index + 1}
{index + 1}	{arg["title"]}	{arg["year"]}	N/A	N/A	{arg["status"]}
{index + 1}	{html_title_td}	{year}	{rating}	{genres}	{arg["status"]}
' in match_str: after_index = match_str.split('
{idx + 1}