优化日志管理，支持最近30天的日志加载和日期范围验证功能

logging_helper.py

@@ -5,10 +5,11 @@ a module of logs saving and backuping
 import os
 import datasets as ds
 from apscheduler.schedulers.background import BackgroundScheduler
-from tqdm import tqdm
 from utils import beijing, md5, json_to_str
 from huggingface_hub import HfApi
 import pandas as pd
+import datetime
+from zoneinfo import ZoneInfo
 import glob
 
 hf = HfApi()
@@ -22,6 +23,7 @@ class LoggingHelper:
         repo_id: str,
         local_dir: str = "data/logs",
         synchronize_interval: int = 60,
+        cache_days: int = 30,
     ):
         """
         :param repo_id: the repo_id of the dataset in huggingface
@@ -29,6 +31,7 @@ class LoggingHelper:
         :param synchronize_interval: the interval of synchronizing between local and huggingface
 
         """
+        self.cache_days = cache_days
         self.local_dir = local_dir
         self.repo_id = repo_id
         self.synchronize_interval = synchronize_interval
@@ -39,11 +42,11 @@ class LoggingHelper:
         self.today = beijing().date()
         ds.disable_progress_bar()
         self.dataframe: pd.DataFrame
-        # 缓存相关变量
-        self.cached_df: pd.DataFrame | None = None
-        self.loaded_files: set[str] = set()
-        self.cache_needs_refresh = False
+        self.dataframe_refresh_needed = True
+        # 首先下载所有数据
         self.pull()
+        # 加载最近30天的日志数据到内存
+        self.load_logs()
         self.start_synchronize()
 
     def addlog(self, log: dict):
@@ -57,6 +60,7 @@ class LoggingHelper:
             self.buffer[remotepath] = ds.Dataset.from_dict({})
             self.buffer[remotepath] = self.buffer[remotepath].add_item(log)  # type: ignore
         self.need_push[remotepath] = True
+        self.dataframe_refresh_needed = True
         print("[addlog] Added a log to buffer")
 
     def remotedir(self):
@@ -66,36 +70,6 @@ class LoggingHelper:
         day = now.day.__str__()
         return "/".join([year, month, day])
 
-    def pull(self):
-        try:
-            self.download()
-            remotedir = self.remotedir()
-            print(f"[pull] today dir: {remotedir}")
-            filenames = hf.list_repo_files(
-                repo_id=self.repo_id,
-                repo_type=self.repo_type,
-            )
-            files_to_load = [
-                filename
-                for filename in filenames
-                if filename not in self.buffer
-                and filename.startswith(remotedir)
-                and filename.endswith(".json")
-            ]
-            print(f"[pull] total {len(files_to_load)} to load")
-            for filename in tqdm(files_to_load):
-                print()
-                path = os.sep.join([self.local_dir, filename])
-                with open(path, "r") as f:
-                    data = f.read()
-                if len(data) != 0:
-                    self.buffer[filename] = ds.Dataset.from_json(path)  # type: ignore
-                    self.need_push[filename] = False
-            return True
-        except Exception as e:
-            print(f"[pull] {type(e)}: {e}")
-            return False
-
     def push_yesterday(self) -> bool:
         try:
             year = self.today.year.__str__()
@@ -106,9 +80,6 @@ class LoggingHelper:
             for filename in self.buffer.keys():
                 if not filename.startswith(remotedir):
                     continue
-                if not self.need_push[filename]:
-                    del self.buffer[filename]
-                    del self.need_push[filename]
                 files_to_push.append(filename)
             if len(files_to_push) == 0:
                 return True
@@ -168,25 +139,116 @@ class LoggingHelper:
                 self.need_push[filename] = False
             print(f"[push] Log files pushed to {res}")
             print("[push] Done")
-            # 标记缓存需要刷新
-            self.cache_needs_refresh = True
             return True
         except Exception as e:
             print(f"[push] {type(e)}: {e}")
             return False
 
-    def download(self):
-        print("[download] Starting downloading")
+    def pull(self):
+        print("[pull] Starting downloading")
         try:
             res = hf.snapshot_download(
                 repo_id=self.repo_id,
                 repo_type="dataset",
                 local_dir=self.local_dir,
             )
-            print(f"[download] Downloaded to {res}")
+            print(f"[pull] Downloaded to {res}")
+        except Exception as e:
+            print(f"[pull] {type(e)}: {e}")
+        print("[pull] Done")
+
+    def get_pathes_between(
+        self, from_date: datetime.date, to_date: datetime.date
+    ) -> list[str]:
+        """
+        获取指定日期范围内的路径列表
+
+        :param from_date: 开始日期（格式：YYYY-MM-DD 或 datetime.date），含该日期
+        :param to_date: 结束日期（格式：YYYY-MM-DD 或 datetime.date），含该日期
+        :return: 日期范围内的路径列表，格式为 ["YYYY/MM/DD", ...]
+        """
+        pathes = []
+        current_date = from_date
+        while current_date <= to_date:
+            pathes.append(current_date.strftime("%Y/%m/%d"))
+            current_date += datetime.timedelta(days=1)
+        return pathes
+
+    def load_logs(self):
+        """
+        在启动时加载最近30天的日志数据到内存buffer
+        """
+        print("[load_logs] Starting to load recent 30 days logs")
+
+        try:
+            today = beijing().date()
+            start_date = today - datetime.timedelta(days=self.cache_days)
+            print(f"Loading logs from {start_date} to {today}")
+            # 生成最近30天的日期范围
+            pathes = self.get_pathes_between(start_date, today)
+            total_files_loaded = 0
+
+            # 遍历每一天的日志
+            for path in pathes:
+                date_path = os.sep.join([self.local_dir, path])
+                print(f"[load_logs] Processing directory: {date_path}")
+                # 检查该日期的目录是否存在
+                if not os.path.exists(date_path):
+                    print(f"[load_logs] Directory not found: {date_path}")
+                    continue
+                # 加载该目录下的所有JSON文件
+                json_files = glob.glob(os.path.join(date_path, "*.json"))
+
+                for json_file in json_files:
+                    # 构造相对路径作为buffer的key
+                    relative_path = os.path.relpath(json_file, self.local_dir).replace(
+                        os.sep, "/"
+                    )
+                    try:
+                        # 检查文件是否为空
+                        if os.path.getsize(json_file) == 0:
+                            print(f"[load_logs] Skipping empty file: {relative_path}")
+                            continue
+
+                        # 加载JSON数据到Dataset
+                        dataset = ds.Dataset.from_json(json_file)
+                        if isinstance(dataset, ds.Dataset):
+                            self.buffer[relative_path] = dataset
+                            self.need_push[relative_path] = False
+                            total_files_loaded += 1
+                    except Exception as e:
+                        print(f"[load_logs] Error loading {relative_path}: {e}")
+                        continue
+            print(f"[load_logs] Successfully loaded {total_files_loaded} log files")
+            print(f"[load_logs] Total datasets in buffer: {len(self.buffer)}")
         except Exception as e:
-            print(f"[download] {type(e)}: {e}")
-        print("[download] Done")
+            print(f"[load_logs] Error: {type(e)}: {e}")
+
+    def cleanup_old_logs(self):
+        """清理buffer中超过30天的日志数据"""
+        try:
+            print("[cleanup_old_logs] Starting cleanup of old logs")
+            cache_dir_to_remove = (
+                self.today - datetime.timedelta(days=(self.cache_days + 1))
+            ).strftime("%Y/%m/%d")
+            print(
+                f"[cleanup_old_logs] Removing logs in {cache_dir_to_remove} from buffer"
+            )
+            removed_count = 0
+            for filepath in list(self.buffer.keys()):
+                if filepath.startswith(cache_dir_to_remove):
+                    del self.buffer[filepath]
+                    del self.need_push[filepath]
+                    removed_count += 1
+
+            print(f"[cleanup_old_logs] Cleaned up {removed_count} old log files")
+            print(
+                f"[cleanup_old_logs] Remaining datasets in buffer: {len(self.buffer)}"
+            )
+            print("[cleanup_old_logs] Done")
+
+        except Exception as e:
+            print(f"[cleanup_old_logs] Error: {type(e)}: {e}")
 
     def start_synchronize(self):
         self.scheduler.add_job(
@@ -194,130 +256,64 @@ class LoggingHelper:
             "interval",
             seconds=self.synchronize_interval,
         )
+        # 添加每日清理任务，在每天凌晨2点执行
+        self.scheduler.add_job(
+            self.cleanup_old_logs,
+            "cron",
+            hour=2,
+            minute=0,
+        )
         self.scheduler.start()
 
-    def _load_all_logs(self, from_date=None, to_date=None) -> pd.DataFrame:
+    def refresh_dataframe(self) -> pd.DataFrame:
         """
-        加载日志文件并返回合并后的DataFrame
-
-        使用直接路径构造方式高效地检索特定日期范围内的文件
-
-        :param from_date: 开始日期（格式：YYYY-MM-DD或datetime.date），默认为None
-        :param to_date: 结束日期（格式：YYYY-MM-DD或datetime.date），默认为None
+        加载最近30天的日志文件并返回合并后的DataFrame
         """
-        import datetime
-
-        print("[_load_all_logs] Starting to load logs")
-        print(f"[_load_all_logs] Date range: {from_date} to {to_date}")
-
-        filepathes = []
-
-        # 确定日期范围
-        if from_date is None and to_date is None:
-            # 如果没有指定范围，扫描所有目录
-            files = glob.glob("**/*.json", root_dir=self.local_dir, recursive=True)
-            filepathes = [os.path.join(self.local_dir, file) for file in files]
-        else:
-            # 将日期参数转换为 datetime.date 对象
-            start_date = from_date
-            end_date = to_date
-
-            if isinstance(start_date, str):
-                start_date = datetime.datetime.strptime(start_date, "%Y-%m-%d").date()
-            if isinstance(end_date, str):
-                end_date = datetime.datetime.strptime(end_date, "%Y-%m-%d").date()
-
-            # 如果只指定了一个日期，设置默认值
-            if start_date is None:
-                start_date = end_date
-            if end_date is None:
-                end_date = start_date
-
-            # 确保日期不为 None 的类型检查
-            if start_date is not None and end_date is not None:
-                # 直接构造日期范围内的目录路径，避免 glob 遍历
-                current_date = start_date
-                date_dirs = []
-                while current_date <= end_date:
-                    year = str(current_date.year)
-                    month = str(current_date.month)
-                    day = str(current_date.day)
-                    date_dir = os.path.join(self.local_dir, year, month, day)
-                    date_dirs.append((date_dir, year, month, day))
-                    current_date += datetime.timedelta(days=1)
-
-                print(
-                    f"[_load_all_logs] Constructed {len(date_dirs)} date directories"
-                )
-
-                # 从指定日期目录中查找 JSON 文件
-                for date_dir, year, month, day in date_dirs:
-                    if os.path.isdir(date_dir):
-                        json_files = glob.glob("*.json", root_dir=date_dir)
-                        for json_file in json_files:
-                            filepathes.append(os.path.join(date_dir, json_file))
-
-        print(f"[_load_all_logs] Found {len(filepathes)} files in date range")
-
-        # 加载所有日志文件
-        datasets = []
-        for path in tqdm(filepathes):
-            path = str(path)
-            try:
-                datasets.append(ds.Dataset.from_json(path))
-            except Exception as e:
-                print(f"[_load_all_logs] Error loading {path}: {e}")
-                continue
-
-        # 合并数据集并排序
-        df = pd.DataFrame()
-        if datasets:
-            dataset: ds.Dataset = ds.concatenate_datasets(datasets)
-            df = dataset.to_pandas()
-            assert isinstance(df, pd.DataFrame)
-            df = df.sort_values(by="timestamp", ascending=False)
-
-        print(f"[_load_all_logs] Loaded {len(df)} logs")
-        self.loaded_files = set([os.path.relpath(p, self.local_dir) for p in filepathes])
-        return df
+        datasets = list(self.buffer.values())
+        merged_dataset = ds.concatenate_datasets(datasets)
+        self.dataframe = merged_dataset.to_pandas()  # type: ignore
+        print(f"[refresh_dataframe] Loaded {len(self.dataframe)} logs")  # type: ignore
+        self.dataframe_refresh_needed = False
+        return self.dataframe  # type: ignore
 
     def refresh(self, from_date=None, to_date=None) -> list[dict]:
         """
-        获取刷新后的日志列表，支持日期范围过滤
+        获取刷新后的日志列表，从内存buffer中合并Dataset，支持日期范围过滤
 
-        :param from_date: 开始日期（格式：YYYY-MM-DD或datetime.date），默认为None
-        :param to_date: 结束日期（格式：YYYY-MM-DD或datetime.date），默认为None
-        :return: 日志字典列表
-        """
-        import datetime
+        基于timestamp字段进行日期过滤。时间戳格式为 ISO 8601 格式（如 "2025-09-08T16:01:07.526954+08:00"）
 
-        self.push()
+        :param from_date: 开始日期（格式：YYYY-MM-DD 或 datetime.date），含该日期的所有日志
+        :param to_date: 结束日期（格式：YYYY-MM-DD 或 datetime.date），含该日期的所有日志
+        :return: 按时间戳降序排列的日志字典列表
+        """
+        if self.dataframe_refresh_needed:
+            self.refresh_dataframe()
 
+        df = self.dataframe
         # 将字符串日期转换为 datetime.date 对象
+        tz = ZoneInfo("Asia/Shanghai")
         if isinstance(from_date, str):
-            from_date = datetime.datetime.strptime(from_date, "%Y-%m-%d").date()
+            from_date = (
+                datetime.datetime.strptime(from_date, "%Y-%m-%d")
+                .astimezone(tz)
+                .isoformat()
+            )
+            from_date = str(from_date)
         if isinstance(to_date, str):
-            to_date = datetime.datetime.strptime(to_date, "%Y-%m-%d").date()
-
-        # 如果没有指定日期范围，使用缓存机制
-        if from_date is None and to_date is None:
-            # 如果缓存需要刷新或者缓存为空，重新加载所有日志
-            if self.cache_needs_refresh or self.cached_df is None:
-                print("[refresh] Cache miss, reloading all logs")
-                self.cached_df = self._load_all_logs()
-                self.cache_needs_refresh = False
-            else:
-                print("[refresh] Using cached data")
-
-            # 返回缓存的DataFrame
-            if self.cached_df is None or self.cached_df.empty:
-                return []
-
-            return self.cached_df.to_dict(orient="records")
-        else:
-            # 如果指定了日期范围，直接加载不使用缓存
-            print("[refresh] Date range specified, loading without cache")
-            df = self._load_all_logs(from_date=from_date, to_date=to_date)
-            if df is None or df.empty:
-                return []
-            return df.to_dict(orient="records")
+            to_date = datetime.datetime.strptime(to_date, "%Y-%m-%d").astimezone(tz)
+            to_date += datetime.timedelta(days=1)  # 包含结束日期全天
+            to_date = to_date.isoformat()
+        print(f"[refresh] Filtering logs from {from_date} to {to_date}")
+        # 按timestamp范围过滤（包含边界日期的全天数据）
+        if from_date is not None or to_date is not None:
+            # 创建日期范围过滤条件
+            filter_condition = pd.Series([True] * len(df), index=df.index)
+            if from_date is not None:
+                filter_condition = filter_condition & (df["timestamp"] >= from_date)
+            if to_date is not None:
+                filter_condition = filter_condition & (df["timestamp"] < to_date)
+            df = df[filter_condition]
+        # 按timestamp降序排序（最新日志在前）
+        df = df.sort_values(by="timestamp", ascending=False)
+        print(f"[refresh] Returning {len(df)} logs")
+        return df.to_dict(orient="records")

main.py

@@ -75,15 +75,15 @@ async def root(
     首页端点，支持日期范围查询
 
     查询参数：
-    - from_date: 开始日期（格式：YYYY-MM-DD），不指定时默认加载最近7天
+    - from_date: 开始日期（格式：YYYY-MM-DD），不指定时默认加载最近30天
     - to_date: 结束日期（格式：YYYY-MM-DD），不指定时默认为今天
     """
-    # 如果没有指定日期范围，默认加载最近7天的日志
+    # 如果没有指定日期范围，默认加载最近30天的日志
     if from_date is None and to_date is None:
         today = beijing().date()
-        from_date = str(today - datetime.timedelta(days=6))  # 最近7天（包括今天）
+        from_date = str(today - datetime.timedelta(days=29))  # 最近30天（包括今天）
         to_date = str(today)
-        print(f"[root] No date range specified, using last 7 days: {from_date} to {to_date}")
+        print(f"[root] No date range specified, using last 30 days: {from_date} to {to_date}")
 
     data = logger.refresh(from_date=from_date, to_date=to_date)
     return templates.TemplateResponse(

utils.py

@@ -61,3 +61,43 @@ def md5(text: list[str | bytes] | str | bytes | None = None) -> str:
 
 def json_to_str(obj: dict | list) -> str:
     return json.dumps(obj, separators=(",", ":"))
+
+
+def validate_date_format(date_str: str, format_str: str = "%Y-%m-%d") -> bool:
+    """
+    验证日期字符串的格式是否正确
+
+    :param date_str: 要验证的日期字符串
+    :param format_str: 期望的日期格式（默认：YYYY-MM-DD）
+    :return: 如果格式正确返回 True，否则返回 False
+    """
+    if not date_str:
+        return True  # 空值被认为是有效的（可选参数）
+
+    try:
+        from datetime import datetime as dt
+        dt.strptime(date_str, format_str)
+        return True
+    except ValueError:
+        return False
+
+
+def parse_date_range(from_date: str | None, to_date: str | None) -> tuple[str | None, str | None] | tuple[str, str]:
+    """
+    解析和验证日期范围
+
+    :param from_date: 开始日期（格式：YYYY-MM-DD）
+    :param to_date: 结束日期（格式：YYYY-MM-DD）
+    :return: 验证后的日期范围元组 (from_date, to_date)
+    :raises ValueError: 如果日期格式不正确或范围无效
+    """
+    if from_date and not validate_date_format(from_date):
+        raise ValueError(f"Invalid from_date format: {from_date}")
+
+    if to_date and not validate_date_format(to_date):
+        raise ValueError(f"Invalid to_date format: {to_date}")
+
+    if from_date and to_date and from_date > to_date:
+        raise ValueError(f"from_date ({from_date}) cannot be after to_date ({to_date})")
+
+    return from_date, to_date