feat: move approved face image job

pyproject.toml

@@ -19,6 +19,7 @@ dependencies = [
     "pydantic>=2.9.2",
     "python-dotenv>=1.0.1",
     "python-multipart>=0.0.17",
+    "pytz>=2024.2",
     "sqlmodel>=0.0.22",
     "uvicorn>=0.32.0",
 ]

src/core/scheduler.py

@@ -1,3 +1,7 @@
+import datetime
+from typing import List
+
+import pytz
 from apscheduler.schedulers.asyncio import AsyncIOScheduler
 from persica.factory.component import AsyncInitializingComponent
 
@@ -11,3 +15,15 @@ class TimeScheduler(AsyncInitializingComponent):
 
     async def shutdown(self):
         self.scheduler.shutdown()
+
+    def run_job(self, name: str, func, args: List, seconds: int = 5):
+        self.scheduler.add_job(
+            func,
+            trigger="date",
+            args=args,
+            id=name,
+            name=name,
+            run_date=datetime.datetime.now(pytz.timezone("Asia/Shanghai"))
+            + datetime.timedelta(seconds=seconds),
+            replace_existing=True,
+        )

src/frontend/sa-view/face/face-list.html

@@ -183,14 +183,14 @@
 						this.get(selection[0]);
 					},
 					// 批准
-					approveByIdsF: function (ids, is_approved) {
+					approveByIdsF: function (ids, uids, is_approved) {
-						sa.ajax("/face/admin/image/approve_by_ids", {ids: ids, is_approved: is_approved}, function(res) {
+						sa.ajax("/face/admin/image/approve_by_ids", {ids: ids, uids: uids, is_approved: is_approved}, function(res) {
 							const msg = is_approved ? "批准成功" : "撤销批准成功"
 							sa.alert(msg, this.f5);
 						}.bind(this), {});
 					},
 					approve: function (data, is_approved) {
-						this.approveByIdsF([data.id], is_approved);
+						this.approveByIdsF([data.id], [data.user_id], is_approved);
 					},
 					approveByIds: function (is_approved) {
 						var selection = this.$refs['data-table'].selection;
@@ -198,10 +198,12 @@
 							return sa.msg('请选择一条数据')
 						}
 						const ids = [];
+						const uids = [];
 						selection.forEach(function (i) {
 							ids.push(i.id);
+							uids.push(i.user_id);
 						});
-						this.approveByIdsF(ids, is_approved);
+						this.approveByIdsF(ids, uids, is_approved);
 					},
 					// 删除
 					del: function(data) {

src/route/face_image_admin.py

@@ -1,12 +1,17 @@
+import asyncio
+from concurrent.futures import ThreadPoolExecutor
+
 from fastapi_amis_admin.crud import BaseApiOut
 from starlette import status
 from starlette.exceptions import HTTPException
 
-from src.plugin import handler
+from src.core.scheduler import TimeScheduler
+from src.plugin import handler, job
 from src.plugin.plugin import Plugin
 from src.services.face_image.schemas import FaceList, FaceApproveList
 from src.services.face_image.services import FaceImageServices
 from src.services.users.services import UserRoleServices
+from src.utils.move_files import move_files, move_files_by_uid
 
 
 class FaceImageAdminRoutes(Plugin):
@@ -16,9 +21,11 @@ class FaceImageAdminRoutes(Plugin):
         self,
         face_image_services: FaceImageServices,
         user_role_services: UserRoleServices,
+        sche: TimeScheduler,
     ):
         self.face_image_services = face_image_services
         self.user_role_services = user_role_services
+        self.sche = sche
         self.avatar_path = "/face/view/"
 
     @handler.post("/list")
@@ -53,7 +60,7 @@ class FaceImageAdminRoutes(Plugin):
 
     @handler.post("/approve_by_ids")
     async def approve_by_ids(self, data: FaceApproveList):
-        ids, is_approved = data.ids, data.is_approved
+        ids, uids, is_approved = data.ids, data.uids, data.is_approved
         if not ids:
             return BaseApiOut(code=500, msg="指定的人脸 id 不存在")
         try:
@@ -61,6 +68,8 @@ class FaceImageAdminRoutes(Plugin):
                 await self.face_image_services.approve_image(ids[0], is_approved)
             else:
                 await self.face_image_services.approve_images(ids, is_approved)
+            for uid in set(uids):
+                self.add_move_face_image_by_uid_job(uid)
             return BaseApiOut(code=0, msg="请求成功")
         except FileNotFoundError:
             return BaseApiOut(code=500, msg="指定的人脸 id 不存在")
@@ -71,3 +80,35 @@ class FaceImageAdminRoutes(Plugin):
                 status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
                 detail=f"Error Execute SQL：{e}",
             ) from e
+
+    def add_move_face_image_by_uid_job(self, uid: int):
+        self.sche.run_job(
+            f"move_face_image_by_uid_job_{uid}",
+            self.move_face_image_by_uid_job,
+            [uid],
+        )
+
+    async def move_face_image_by_uid_job(self, uid: int):
+        print("Move face image job start uid:", uid)
+        data = await self.face_image_services.get_all(
+            user_id=uid, approved=True, ignore_deleted=True
+        )
+        paths = [i.image for i in data if i.image]
+
+        loop = asyncio.get_event_loop()
+        with ThreadPoolExecutor() as executor:
+            await loop.run_in_executor(executor, move_files_by_uid, [paths, uid])
+        print("Move face image job done uid:", uid)
+
+    @job.cron(name="move_face_image_job", hour=3, minute=0, second=0)
+    async def move_face_image_job(self):
+        print("Move face image job start")
+        data = await self.face_image_services.get_all(
+            approved=True, ignore_deleted=True
+        )
+        paths = [i.image for i in data if i.image]
+
+        loop = asyncio.get_event_loop()
+        with ThreadPoolExecutor() as executor:
+            await loop.run_in_executor(executor, move_files, paths)
+        print("Move face image job done")

src/services/face_image/schemas.py

@@ -63,4 +63,5 @@ class FaceList(BaseModel):
 
 class FaceApproveList(BaseModel):
     ids: List[int]
+    uids: List[int]
     is_approved: bool

src/services/face_image/services.py

@@ -25,10 +25,15 @@ class FaceImageServices(AsyncInitializingComponent):
         return await self.repo.get_all(user_id=user_id, ignore_deleted=ignore_deleted)
 
     async def get_all(
-        self, approved: Optional[bool] = None, ignore_deleted: bool = True
+        self,
+        user_id: Optional[int] = None,
+        approved: Optional[bool] = None,
+        ignore_deleted: bool = True,
     ):
         return await self.repo.get_all(
-            is_approved=approved, ignore_deleted=ignore_deleted
+            user_id=user_id,
+            is_approved=approved,
+            ignore_deleted=ignore_deleted,
         )
 
     async def upload_by_user_id(self, user_id: int, image: str) -> FaceImageModel:

src/utils/_path.py

@@ -12,3 +12,6 @@ AVATAR_DATA_PATH.mkdir(exist_ok=True)
 
 FACE_IMAGE_DATA_PATH = DATA_PATH / "face_image"
 FACE_IMAGE_DATA_PATH.mkdir(exist_ok=True)
+
+FACE_IMAGE_DATABASE_PATH = DATA_PATH / "face_image_database"
+FACE_IMAGE_DATABASE_PATH.mkdir(exist_ok=True)

src/utils/move_files.py

@@ -0,0 +1,36 @@
+import contextlib
+import shutil
+from threading import Lock
+from typing import List, Tuple
+
+from ._path import FACE_IMAGE_DATA_PATH, FACE_IMAGE_DATABASE_PATH
+
+LOCK = Lock()
+
+
+def move_files(paths: List[str]):
+    with LOCK:
+        shutil.rmtree(FACE_IMAGE_DATABASE_PATH, ignore_errors=True)
+        FACE_IMAGE_DATABASE_PATH.mkdir(exist_ok=True)
+
+        for path in paths:
+            p = FACE_IMAGE_DATA_PATH / path
+            d = FACE_IMAGE_DATABASE_PATH / path
+            d.parent.mkdir(exist_ok=True)
+            with contextlib.suppress(Exception):
+                shutil.copy(p, d)
+
+
+def move_files_by_uid(data: Tuple[List[str], int]):
+    paths, uid = data
+    with LOCK:
+        shutil.rmtree(FACE_IMAGE_DATABASE_PATH / str(uid), ignore_errors=True)
+        if not paths:
+            return
+        (FACE_IMAGE_DATABASE_PATH / str(uid)).mkdir(exist_ok=True)
+
+        for path in paths:
+            p = FACE_IMAGE_DATA_PATH / path
+            d = FACE_IMAGE_DATABASE_PATH / path
+            with contextlib.suppress(Exception):
+                shutil.copy(p, d)