建議 model:opus/effort:medium — 動的是生產中的任務證據表,還要加唯一索引(既有重複列會擋住),冪等寫錯會產生重複證據。
本卡屬 FR-107(母卡 CM-1843)第 .4 棒(子需求卡 CM-1847),子任務 T-4.1:讓系統的任務證據表認得「這筆證據是 AI 分類掛進來的」,並做出真正把檔掛進任務的程式。
系統裡每個稽核任務底下掛的證據,記在 job_evidences 這張表。它現在的「來源」欄位只認兩種值:SYSTEM_UPLOAD(使用者自己傳的)、DRIVE_SYNC(從 Google Drive 同步來的)。
這張卡加第三種 AI_CLASSIFIED(AI 分類後歸檔進來的),並加一個 classification_run_id 欄位記「是哪一次分類掛進來的」(報表要靠它算「歸檔後的正解」)。
同時加一個防重複索引:同一個任務、同一份檔,只能有一筆有效的證據列。這樣連按兩次歸檔、或同一份檔的兩個檢查點剛好對到同一個任務,都只會留一筆。
然後把 T-2.2 留空的「掛證據」轉接頭真的實作出來。
⚠️ 加唯一索引之前要先查現有資料有沒有重複列(Google Drive 同步的早期版本可能留下重複)。查到有就在卡片回報,不要自動合併(決策者 2026-09-16 裁示 5:先查不合併,查出有再裁)。
主專案
scripts/sql/2026-MM-DD-fr107-job-evidences-ai-classified.sql ← 新增(🔴 主線 migration)
infra/flow_engine/models/job_evidence.py
:13 job_evidences 表
:40 file_id int FK→upload_files.id
:55 source 欄位註解(現在只寫 SYSTEM_UPLOAD / DRIVE_SYNC)← 加 AI_CLASSIFIED
:59 drive_file_id UNIQUE
:76 is_deleted / :79 deleted_at
← 加 classification_run_id int NULL
對應 domain entity
core/plugins/evidence_classification.py
JobEvidenceTaskSinkAdapter.attach ← T-2.2 留的 NotImplementedError,本卡實作
JobEvidenceTaskSinkAdapter.find_same_content ← 同上
任務證據列表 API ← 回傳多帶 source/classification_run_id
🔴 照抄這個 pattern(不要自己另寫一套寫證據的邏輯)
app/cloud_integration/service/handlers/import_drive_file_handler.py:210-256
JobEvidenceEntity(source="DRIVE_SYNC", ...) + domain add + 冪等去重
改動 說明
source 新值 AI_CLASSIFIED 欄位是 varchar(20) 無 CHECK 約束,
改 model 註解與 enum 即可
新欄 classification_run_id int NULL 追溯是哪次分類掛進來的;
FR-031 報表據此算「歸檔後正解」
(軟參照,不建 FK)
partial UNIQUE (job_execution_id, file_id) 🔴 冪等鍵
WHERE is_deleted = false 同一批連按兩次歸檔、或同檔兩個檢查點
剛好對到同一任務,都只留一筆
⚠️ 加索引前先查 DEV 有無既存重複列:
SELECT job_execution_id, file_id, count(*)
FROM compliance.job_evidences
WHERE is_deleted = false
GROUP BY 1,2 HAVING count(*) > 1;
有 → 卡片回報筆數與內容,停下等決策者裁;🔴 不要自動合併、不要自己刪
def attach(self, *, job_execution_id: int, file_id: int, file_uid: str,
classification_run_id: int, actor_user_id: int,
content_hash: str | None, description: str | None) -> AttachResult:
# 寫 job_evidences(source="AI_CLASSIFIED")
# 🔴 命中 (job_execution_id, file_id) 既有「未刪」列時回 existed=True,不新寫
# 照 import_drive_file_handler.py:210-256 的 pattern
# 🔴 不複製實體檔——file_id 指向同一筆 upload_files
def find_same_content(self, round_id: int, content_hash: str) -> list[tuple[int, str]]:
# D9 提示:回 [(job_execution_id, evidence_uid)],只提示不阻擋
INSERT public.schema_migrations。只套 DEV。attach 照既有 pattern 寫——import_drive_file_handler.py:210-256 已經有「建 entity + domain add + 冪等去重」的完整範例。先查再寫,不要另創一套。差別只在 source 值與多帶 classification_run_id。existed=True),索引是最後防線(避免併發)。只靠索引會拋 IntegrityError,只靠查詢擋不住併發。source 與 classification_run_id,前端 T-4.3 要用它顯示「AI 分類」標籤。