✨ add and update utils

Signed-off-by: peter szemraj <[email protected]>

utils.py

@@ -1,12 +1,12 @@
 """
     utils.py - Utility functions for the project.
 """
-
 import logging
+import os
 import re
 import subprocess
 from collections import defaultdict, deque
-from datetime import datetime
+from datetime import datetime, timedelta
 from itertools import combinations, islice
 from pathlib import Path
 from typing import List
@@ -25,6 +25,86 @@ STOPWORDS = set(
 )
 
 
+def remove_stagnant_files(
+    freq: str = "hourly",
+    search_path: str = ".",
+    substring="DocSumm",
+    remove_suffix=".txt",
+):
+    """
+    remove_stagnant_files - Remove files that have not been modified in a certain amount of time.
+
+    :param str freq: frequency of file removal, defaults to "hourly"
+    :param str search_path: location to search for files, defaults to "."
+    :param str substring: substring to search for in file names, defaults to "DocSumm"
+    :param str remove_suffix: suffix of files to remove, defaults to ".txt"
+    :raises ValueError: if freq is not one of "hourly", "daily", or "weekly"
+    """
+    current_time = datetime.now()
+    search_path = Path(search_path)
+
+    if freq == "hourly":
+        time_threshold = current_time - timedelta(hours=1)
+    elif freq == "daily":
+        time_threshold = current_time - timedelta(days=1)
+    elif freq == "weekly":
+        time_threshold = current_time - timedelta(weeks=1)
+    else:
+        raise ValueError(
+            "Invalid frequency. Supported values are 'hourly', 'daily', and 'weekly'."
+        )
+
+    files_to_remove = []
+    potential_files = [
+        f for f in search_path.iterdir() if f.is_file() and f.suffix == remove_suffix
+    ]
+    logging.info(f"Found {len(potential_files)} files.")
+    for candidate in potential_files:
+        if (
+            candidate.is_file()
+            and substring in candidate.name
+            and candidate.stat().st_mtime < time_threshold.timestamp()
+        ):
+            files_to_remove.append(candidate)
+        logging.debug(f"File {candidate} last modified at {candidate.stat().st_mtime}")
+    logging.info(f"Removing {len(files_to_remove)} files.")
+    for file_path in files_to_remove:
+        file_path.unlink()
+    logging.debug(f"Removed files: {files_to_remove}")
+
+
+def compare_model_size(model_name: str, threshold: int = 500) -> bool:
+    """
+    compare_model_size - compare string representations of model size to a threshold
+
+    :param str model_name: the model name to compare
+    :param int threshold: the threshold to compare against in millions, defaults to 500
+    :return: True if the model size is greater than the threshold, False or None otherwise
+    """
+    pattern = r"(\d+)(M|G|k|b)?"  # param regex
+
+    matches = re.findall(pattern, model_name)
+    if not matches:
+        return None
+
+    # Extract the parameter count and unit from the last match
+    parameter_count, unit = matches[-1]
+
+    parameter_count = int(parameter_count)  # Convert to an integer
+
+    # Convert to the standard form (M for million, G for billion, k for thousand)
+    if unit == "G" or unit == "b":
+        parameter_count *= 1000
+    elif unit == "M":
+        pass
+    elif unit == "k":
+        parameter_count /= 1000
+    else:
+        return None  # Unknown
+
+    return parameter_count > threshold
+
+
 def validate_pytorch2(torch_version: str = None):
     torch_version = torch.__version__ if torch_version is None else torch_version
 
@@ -46,7 +126,7 @@ def get_timestamp(detailed=False) -> str:
     )
 
 
-def truncate_word_count(text, max_words=512):
+def truncate_word_count(text, max_words=1024):
     """
     truncate_word_count - a helper function for the gradio module
     Parameters
@@ -141,6 +221,42 @@ def textlist2html(text_batches):
     return text_html_block
 
 
+def extract_batches(html_string, pattern=None, flags=None) -> list:
+    """
+    Extract batches of text from an HTML string.
+
+    Args:
+        html_string (str): The HTML string to extract batches from.
+        pattern (str, optional): The regular expression pattern to use. Defaults to a pattern that matches batches in the format provided.
+        flags (int, optional): The flags to use with the regular expression. Defaults to re.DOTALL.
+
+    Returns:
+        list: A list of dictionaries where each dictionary represents a batch and has 'title' and 'content' keys.
+    """
+    # Set default pattern if none provided
+    if pattern is None:
+        pattern = r'<h2 style="font-size: 22px; color: #555;">(.*?)</h2>\s*<p style="white-space: pre-line;">(.*?)</p>'
+
+    # Set default flags if none provided
+    if flags is None:
+        flags = re.DOTALL
+
+    try:
+        # Find all matches in the string
+        matches = re.findall(pattern, html_string, flags)
+
+        # Convert matches to a list of dictionaries
+        batches = [
+            {"title": title.strip(), "content": content.strip()}
+            for title, content in matches
+        ]
+
+        return batches
+    except re.error as e:
+        logging.error(f"An error occurred while trying to extract batches: {e}")
+        return []
+
+
 def extract_keywords(
     text: str, num_keywords: int = 3, window_size: int = 5, kw_max_len: int = 20
 ) -> List[str]: