feat: 添加历史记录相关端口

app/api/__init__.py

@@ -28,6 +28,7 @@ from .scripts import router as scripts_router
 from .plan import router as plan_router
 from .queue import router as queue_router
 from .dispatch import router as dispatch_router
+from .history import router as history_router
 from .setting import router as setting_router
 
 __all__ = [
@@ -36,5 +37,6 @@ __all__ = [
     "plan_router",
     "queue_router",
     "dispatch_router",
+    "history_router",
     "setting_router",
 ]

app/api/history.py

@@ -0,0 +1,84 @@
+#   AUTO_MAA:A MAA Multi Account Management and Automation Tool
+#   Copyright © 2024-2025 DLmaster361
+#   Copyright © 2025 MoeSnowyFox
+
+#   This file is part of AUTO_MAA.
+
+#   AUTO_MAA is free software: you can redistribute it and/or modify
+#   it under the terms of the GNU General Public License as published
+#   by the Free Software Foundation, either version 3 of the License,
+#   or (at your option) any later version.
+
+#   AUTO_MAA is distributed in the hope that it will be useful,
+#   but WITHOUT ANY WARRANTY; without even the implied warranty
+#   of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See
+#   the GNU General Public License for more details.
+
+#   You should have received a copy of the GNU General Public License
+#   along with AUTO_MAA. If not, see <https://www.gnu.org/licenses/>.
+
+#   Contact: DLmaster_361@163.com
+
+
+import datetime
+from pathlib import Path
+from fastapi import APIRouter, Body
+
+from app.core import Config
+from app.models.schema import *
+
+router = APIRouter(prefix="/api/history", tags=["历史记录"])
+
+
+@router.post(
+    "/search",
+    summary="搜索历史记录总览信息",
+    response_model=HistorySearchOut,
+    status_code=200,
+)
+async def search_history(history: HistorySearchIn) -> HistorySearchOut:
+
+    try:
+        data = await Config.search_history(
+            history.mode,
+            datetime.datetime.strptime(history.start_date, "%Y-%m-%d"),
+            datetime.datetime.strptime(history.end_date, "%Y-%m-%d"),
+        )
+        for date, users in data.items():
+            for user, records in users.items():
+                record = await Config.merge_statistic_info(records)
+                record["index"] = [HistoryIndexItem(**_) for _ in record["index"]]
+                record = HistoryData(**record)
+                data[date][user] = record
+    except Exception as e:
+        return HistorySearchOut(
+            code=500,
+            status="error",
+            message=f"{type(e).__name__}: {str(e)}",
+            data={},
+        )
+    return HistorySearchOut(data=data)
+
+
+@router.post(
+    "/data",
+    summary="从指定文件内获取历史记录数据",
+    response_model=HistoryDataGetOut,
+    status_code=200,
+)
+async def get_history_data(history: HistoryDataGetIn = Body(...)) -> HistoryDataGetOut:
+
+    try:
+        path = Path(history.jsonPath)
+        data = await Config.merge_statistic_info([path])
+        data.pop("index", None)
+        data["log_content"] = path.with_suffix(".log").read_text(encoding="utf-8")
+        data = HistoryData(**data)
+    except Exception as e:
+        return HistoryDataGetOut(
+            code=500,
+            status="error",
+            message=f"{type(e).__name__}: {str(e)}",
+            data=HistoryData(**{}),
+        )
+    return HistoryDataGetOut(data=data)

app/core/config.py

@@ -1552,7 +1552,7 @@ class AppConfig(GlobalConfig):
 
         logger.success(f"通用日志统计完成，日志路径：{log_path.with_suffix('.log')}")
 
-    def merge_statistic_info(self, statistic_path_list: List[Path]) -> dict:
+    async def merge_statistic_info(self, statistic_path_list: List[Path]) -> dict:
         """
         合并指定数据统计信息文件
 
@@ -1617,11 +1617,13 @@ class AppConfig(GlobalConfig):
                             single_data[key]
                         )
 
-                    data["index"][actual_date] = [
+                    data["index"][actual_date] = {
-                        actual_date.strftime("%d日 %H:%M:%S"),
+                        "date": actual_date.strftime("%d日 %H:%M:%S"),
-                        ("完成" if single_data[key] == "Success!" else "异常"),
+                        "status": (
-                        json_file,
+                            "完成" if single_data[key] == "Success!" else "异常"
-                    ]
+                        ),
+                        "jsonFile": str(json_file),
+                    }
 
         data["index"] = [data["index"][_] for _ in sorted(data["index"])]
 
@@ -1631,7 +1633,7 @@ class AppConfig(GlobalConfig):
 
         return {k: v for k, v in data.items() if v}
 
-    def search_history(
+    async def search_history(
         self, mode: str, start_date: datetime, end_date: datetime
     ) -> dict:
         """
@@ -1694,7 +1696,7 @@ class AppConfig(GlobalConfig):
             for k, v in sorted(history_dict.items(), key=lambda x: x[0], reverse=True)
         }
 
-    def clean_old_history(self):
+    async def clean_old_history(self):
         """删除超过用户设定天数的历史记录文件（基于目录日期）"""
 
         if self.get("Function", "HistoryRetentionTime") == 0:

app/models/schema.py

@@ -407,6 +407,30 @@ class MaaPlanConfig(BaseModel):
     Sunday: Optional[MaaPlanConfig_Item] = Field(None, description="周日")
 
 
+class HistoryIndexItem(BaseModel):
+    date: str = Field(..., description="日期")
+    status: str = Field(..., description="状态")
+    jsonFile: str = Field(..., description="对应JSON文件")
+
+
+class HistoryData(BaseModel):
+    index: Optional[List[HistoryIndexItem]] = Field(
+        None, description="历史记录索引列表"
+    )
+    recruit_statistics: Optional[Dict[str, int]] = Field(
+        None, description="公招统计数据, key为星级, value为对应的公招数量"
+    )
+    drop_statistics: Optional[Dict[str, Dict[str, int]]] = Field(
+        None, description="掉落统计数据, 格式为 { '关卡号': { '掉落物': 数量 } }"
+    )
+    error_info: Optional[Dict[str, str]] = Field(
+        None, description="报错信息, key为时间戳, value为错误描述"
+    )
+    log_content: Optional[str] = Field(
+        None, description="日志内容, 仅在提取单条历史记录数据时返回"
+    )
+
+
 class ScriptCreateIn(BaseModel):
     type: Literal["MAA", "General"] = Field(
         ..., description="脚本类型: MAA脚本, 通用脚本"
@@ -638,6 +662,29 @@ class TaskMessage(BaseModel):
     data: Dict[str, Any] = Field(..., description="消息数据，具体内容根据type类型而定")
 
 
+class HistorySearchIn(BaseModel):
+    mode: Literal["按日合并", "按周合并", "按年月并"] = Field(
+        ..., description="合并模式"
+    )
+    start_date: str = Field(..., description="开始日期, 格式YYYY-MM-DD")
+    end_date: str = Field(..., description="结束日期, 格式YYYY-MM-DD")
+
+
+class HistorySearchOut(OutBase):
+    data: Dict[str, Dict[str, HistoryData]] = Field(
+        ...,
+        description="历史记录索引数据字典, 格式为 { '日期': { '用户名': [历史记录信息] } }",
+    )
+
+
+class HistoryDataGetIn(BaseModel):
+    jsonPath: str = Field(..., description="需要提取数据的历史记录JSON文件")
+
+
+class HistoryDataGetOut(OutBase):
+    data: HistoryData = Field(..., description="历史记录数据")
+
+
 class SettingGetOut(OutBase):
     data: GlobalConfig = Field(..., description="全局设置数据")
 

main.py

@@ -59,6 +59,7 @@ def main():
 
             await Config.init_config()
             await Config.get_stage()
+            await Config.clean_old_history()
             main_timer = asyncio.create_task(MainTimer.second_task())
 
             yield
@@ -79,6 +80,7 @@ def main():
             plan_router,
             queue_router,
             dispatch_router,
+            history_router,
             setting_router,
         )
 
@@ -102,6 +104,7 @@ def main():
         app.include_router(plan_router)
         app.include_router(queue_router)
         app.include_router(dispatch_router)
+        app.include_router(history_router)
         app.include_router(setting_router)
 
         uvicorn.run(app, host="0.0.0.0", port=8000)