feat 历史记录删除支持删除源文件

app/api/endpoints/history.py

@@ -62,20 +62,22 @@ def transfer_history(title: str = None,
 
 @router.delete("/transfer", summary="删除转移历史记录", response_model=schemas.Response)
 def delete_transfer_history(history_in: schemas.TransferHistory,
-                            delete_file: bool = False,
+                            deletesrc: bool = False,
+                            deletedest: bool = False,
                             db: Session = Depends(get_db),
                             _: schemas.TokenPayload = Depends(verify_token)) -> Any:
     """
     删除转移历史记录
     """
-    # 触发删除事件
-    if delete_file:
-        history = TransferHistory.get(db, history_in.id)
-        if not history:
-            return schemas.Response(success=False, msg="记录不存在")
-        # 册除文件
-        if history.dest:
-            TransferChain(db).delete_files(Path(history.dest))
+    history = TransferHistory.get(db, history_in.id)
+    if not history:
+        return schemas.Response(success=False, msg="记录不存在")
+    # 册除媒体库文件
+    if deletedest and history.dest:
+        TransferChain(db).delete_files(Path(history.dest))
+    # 删除源文件
+    if deletesrc and history.src:
+        TransferChain(db).delete_files(Path(history.src))
     # 删除记录
     TransferHistory.delete(db, history_in.id)
     return schemas.Response(success=True)

app/db/models/__init__.py

@@ -1,4 +1,4 @@
-from typing import Any
+from typing import Any, Self, List
 
 from sqlalchemy.orm import as_declarative, declared_attr, Session
 
@@ -16,13 +16,13 @@ class Base:
             db.rollback()
             raise err
 
-    def create(self, db: Session):
+    def create(self, db: Session) -> Self:
         db.add(self)
         self.commit(db)
         return self
 
     @classmethod
-    def get(cls, db: Session, rid: int):
+    def get(cls, db: Session, rid: int) -> Self:
         return db.query(cls).filter(cls.id == rid).first()
 
     def update(self, db: Session, payload: dict):
@@ -42,7 +42,7 @@ class Base:
         Base.commit(db)
 
     @classmethod
-    def list(cls, db: Session):
+    def list(cls, db: Session) -> List[Self]:
         return db.query(cls).all()
 
     def to_dict(self):