feat(phase7): auto-trigger asset classification

apps/web/src/pages/workspace/ProjectAssets.tsx

@@ -75,6 +75,7 @@ const ProjectAssets: React.FC = () => {
     queryKey: ['assets', libraryId],
     queryFn: () => getAssets(libraryId),
     enabled: !!libraryId,
+    refetchInterval: 3000,
   });
 
   const ingestJobQuery = useQuery({
@@ -102,7 +103,7 @@ const ProjectAssets: React.FC = () => {
     if (status === 'completed') {
       setUploading(false);
       assetsQuery.refetch();
-      message.success('素材导入完成');
+      message.success('素材导入完成，已自动发起分类');
     }
     if (status === 'failed') {
       setUploading(false);
@@ -166,7 +167,8 @@ const ProjectAssets: React.FC = () => {
   }, [assetsQuery.data, classificationFilter]);
 
   const renderClassification = (asset: AssetItem) => {
-    if (classifyingAssetId === asset.id && classificationJobQuery.data?.status !== 'failed') {
+    const autoClassification = String(asset.metadata?.auto_classification || '');
+    if (classifyingAssetId === asset.id || autoClassification === 'queued') {
       return <Tag color="processing">分类中</Tag>;
     }
     const classification = String(asset.metadata?.classification || '');
@@ -178,6 +180,7 @@ const ProjectAssets: React.FC = () => {
       <Space>
         <Tag color="blue">{classification}</Tag>
         <Tag>{Math.round(confidence * 100)}%</Tag>
+        {autoClassification === 'completed' ? <Tag color="green">自动</Tag> : null}
       </Space>
     );
   };
@@ -207,11 +210,12 @@ const ProjectAssets: React.FC = () => {
       render: (_: unknown, record: AssetItem) => {
         const hasClassification = !!record.metadata?.classification;
         const isCurrentClassifying = classifyingAssetId === record.id;
+        const isAutoClassifying = record.metadata?.auto_classification === 'queued';
         return (
           <Button
             icon={<TagsOutlined />}
             loading={isCurrentClassifying && classificationMutation.isPending}
-            disabled={!!classifyingAssetId && classifyingAssetId !== record.id}
+            disabled={isAutoClassifying || (!!classifyingAssetId && classifyingAssetId !== record.id)}
             onClick={() => {
               setClassifyingAssetId(record.id);
               classificationMutation.mutate({
@@ -221,7 +225,7 @@ const ProjectAssets: React.FC = () => {
               });
             }}
           >
-            {isCurrentClassifying ? '分类中...' : hasClassification ? '重新分类' : '发起分类'}
+            {isAutoClassifying ? '自动分类中...' : isCurrentClassifying ? '分类中...' : hasClassification ? '重新分类' : '发起分类'}
           </Button>
         );
       },

apps/worker/tasks.py

@@ -1,13 +1,13 @@
 from datetime import datetime, timezone
 import random
 
-from app.core.config import get_settings
+from app.config import get_settings
 from .celery_app import celery_app
 from packages.adapters.sqlalchemy_impl.session import SessionLocal, build_session_factory
 from packages.adapters.sqlalchemy_impl.ingest_job_repository import SQLAlchemyIngestJobRepository
 from packages.adapters.sqlalchemy_impl.asset_repository import SQLAlchemyAssetRepository
 from packages.adapters.sqlalchemy_impl.classification_job_repository import SQLAlchemyClassificationJobRepository
-from packages.domain import Asset, AssetClassification, ClassificationJobStatus, IngestJobStatus
+from packages.domain import Asset, AssetClassification, ClassificationJob, ClassificationJobStatus, IngestJobStatus
 
 settings = get_settings()
 if SessionLocal is None:
@@ -25,6 +25,7 @@ def ingest_asset(job_id: str) -> dict:
     try:
         ingest_repo = SQLAlchemyIngestJobRepository(session)
         asset_repo = SQLAlchemyAssetRepository(session)
+        classification_repo = SQLAlchemyClassificationJobRepository(session)
 
         job = ingest_repo.get(job_id)
         if job is None:
@@ -55,7 +56,7 @@ def ingest_asset(job_id: str) -> dict:
             name=filename,
             storage_key=storage_key,
             mime_type=mime_type,
-            metadata={"source": "ingest_task"},
+            metadata={"source": "ingest_task", "auto_classification": "queued"},
         )
         asset_repo.create(asset)
 
@@ -64,7 +65,20 @@ def ingest_asset(job_id: str) -> dict:
         job.updated_at = datetime.now(timezone.utc)
         ingest_repo.update(job)
 
-        return {"ok": True, "job_id": job.id, "asset_id": asset.id}
+        classification_job = ClassificationJob.create(
+            workspace_id=job.workspace_id,
+            project_id=job.project_id,
+            asset_id=asset.id,
+        )
+        classification_repo.create(classification_job)
+        celery_app.send_task("worker.classify_asset", args=[classification_job.id])
+
+        return {
+            "ok": True,
+            "job_id": job.id,
+            "asset_id": asset.id,
+            "classification_job_id": classification_job.id,
+        }
     except Exception as error:
         try:
             ingest_repo = SQLAlchemyIngestJobRepository(session)
@@ -125,6 +139,7 @@ def classify_asset(job_id: str) -> dict:
             **asset.metadata,
             "classification": classification,
             "classification_confidence": confidence,
+            "auto_classification": "completed",
         }
         asset_repo.update(asset)