FR-030.2 證據分類支援圖片 / PDF 內容辨識 — Implementation Plan

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: 讓 FR-030 證據分類器的 Docker container 能用 Claude Vision 辨識圖片（png/jpg/gif/webp）與 PDF 內容並分類，原分類 / 歸檔流程一個字不動。

Architecture: 方式 A — 把 container_entrypoint.py 的分類管道從「字串進」改成「Anthropic content blocks 進」。新增純函式 build_content_blocks(path, mime, filename)，依副檔名組 text / image / document block；classify_with_claude 改收 blocks；process_file_worker 串接並把超限 / 不支援檔 skip 進既有未處理清單。文字檔走的路徑語意不變（只是字串包進 text block）。

Tech Stack: Python 3.11、anthropic>=0.77（image + document block）、pypdf（新增，只讀 PDF 頁數）、stdlib base64、pytest。

改動檔（全部在 container 目錄內，不碰主專案 app/domain/infra）：

Modify: scripts/evidence/classify/docker/container_entrypoint.py
Modify: scripts/evidence/classify/docker/requirements.txt（加 pypdf）
Create: scripts/evidence/classify/docker/test_container_entrypoint.py（新測試，目前無）

Spec： docs/features/FR-030.2-2605-evidence-classify-image-vision/design.md

§1

File Structure

檔案	職責	本計畫動作
`container_entrypoint.py`	分類 container 主程式	改 3 個函式 + 加 3 個 helper + 改 prompt 措辭
`requirements.txt`	container 依賴	加 `pypdf>=4`
`test_container_entrypoint.py`	純函式單元測試	新建

不動：_state.json / _report-original.json 寫出邏輯（write_outputs）、Drive 歸檔、review UI、主專案任何檔。

§2

Task 0: Pre-flight 驗證（無 code，先做）

plan 寫好到開工常有時間差，先 verify 假設仍成立再動手。

Files: 無（只讀 / 跑命令）

Step 1: 確認改動點函式仍在預期位置 / 簽名

Run:

cd /Users/chouraymond/Projects/Billows/Audit-Manager/compliance-manager-be
grep -n "def extract_text\|def classify_with_claude\|def process_file_worker\|def build_system_block\|DEFAULT_MODEL\|MAX_TEXT_CHARS" scripts/evidence/classify/docker/container_entrypoint.py

Expected: extract_text(~208) / classify_with_claude(~277, 簽名含 filename, content) / process_file_worker(~312) / build_system_block(~233) / DEFAULT_MODEL = "claude-sonnet-4-6" / MAX_TEXT_CHARS = 8000 皆在。若行號漂移以實際為準。

Step 2: 確認 model 為 vision-capable + SDK 支援 document block

Run:

grep -n "anthropic" scripts/evidence/classify/docker/requirements.txt
python3 -c "import anthropic; print(anthropic.__version__)"

Expected: anthropic>=0.77；claude-sonnet-4-6 為多模態 model（支援 image + PDF document block）。若實際傳入的 model（--model 或 DEFAULT_MODEL）非 sonnet-4.x，停下回報 user。

Step 3: 確認測試環境能 import 該 module（無 side effect）

Run:

cd scripts/evidence/classify/docker && python3 -c "import container_entrypoint as c; print('ok', hasattr(c,'extract_text'))"

Expected: ok True，無 traceback、不會啟動 main（確認有 if __name__ == "__main__" guard）。若 import 觸發 main 或缺依賴，先解決再繼續。

Step 4: 確認 write_outputs 只消費 matches/primary（流程不變的結構前提）

Run:

grep -n "skipped_reason\|\[\"matches\"\]\|\.get(\"matches\"\|\[\"primary\"\]\|\.get(\"error\"" scripts/evidence/classify/docker/container_entrypoint.py

Expected: 確認下游只讀 matches / primary / error，未讀其它 per-file key → 新增 skipped_reason 向後相容。

§3

Task 1: 加 `pypdf` 依賴 + 兩個 helper（`_img_media_type` / `_pdf_page_count`）

Files:

Modify: scripts/evidence/classify/docker/requirements.txt
Modify: scripts/evidence/classify/docker/container_entrypoint.py（加 import base64、from pypdf import PdfReader、兩個 helper）
Test: scripts/evidence/classify/docker/test_container_entrypoint.py

Step 1: 加依賴

在 requirements.txt 末加一行：

pypdf>=4

Step 2: 寫 failing test（helper）

建 test_container_entrypoint.py：

import sys
from pathlib import Path

HERE = Path(__file__).resolve().parent
sys.path.insert(0, str(HERE))
import container_entrypoint as c


def test_img_media_type():
    assert c._img_media_type(".png") == "image/png"
    assert c._img_media_type(".jpg") == "image/jpeg"
    assert c._img_media_type(".jpeg") == "image/jpeg"
    assert c._img_media_type(".gif") == "image/gif"
    assert c._img_media_type(".webp") == "image/webp"


def test_pdf_page_count(tmp_path):
    # 用 pypdf 寫一個 2 頁 PDF
    from pypdf import PdfWriter
    w = PdfWriter()
    w.add_blank_page(width=72, height=72)
    w.add_blank_page(width=72, height=72)
    p = tmp_path / "two.pdf"
    with open(p, "wb") as f:
        w.write(f)
    assert c._pdf_page_count(p) == 2

Step 3: 跑 test 確認 fail

Run: cd scripts/evidence/classify/docker && python3 -m pytest test_container_entrypoint.py -v Expected: FAIL — AttributeError: module has no attribute '_img_media_type'（且需先 pip install pypdf）

Step 4: 實作 helper

在 container_entrypoint.py 加 import base64（與其它 stdlib import 並列）、from pypdf import PdfReader（與外部 deps 並列），並在 extract_text 附近加：

_IMG_MEDIA = {
    ".png": "image/png",
    ".jpg": "image/jpeg",
    ".jpeg": "image/jpeg",
    ".gif": "image/gif",
    ".webp": "image/webp",
}


def _img_media_type(ext: str) -> str:
    return _IMG_MEDIA[ext.lower()]


def _pdf_page_count(path: Path) -> int:
    return len(PdfReader(str(path)).pages)

Step 5: 跑 test 確認 pass

Run: cd scripts/evidence/classify/docker && python3 -m pytest test_container_entrypoint.py -v Expected: PASS（2 passed）

Step 6: Commit

git add scripts/evidence/classify/docker/requirements.txt \
        scripts/evidence/classify/docker/container_entrypoint.py \
        scripts/evidence/classify/docker/test_container_entrypoint.py
git commit -m "feat(evidence-classify): 加 pypdf 依賴與 img/pdf helper"

§4

Task 2: 把 `extract_text` 改名為 `_extract_text_legacy`（行為零變更）

Files:

Modify: scripts/evidence/classify/docker/container_entrypoint.py

文字抽取邏輯整段保留（含 MAX_TEXT_CHARS 截斷），只改函式名，並更新呼叫端。build_content_blocks（Task 3）會包它的輸出。

Step 1: 改名 + 更新呼叫端

實際簽名帶 type hint（一行）：def extract_text(path: Path, max_chars: int = MAX_TEXT_CHARS) -> str: → 改名為 def _extract_text_legacy(path: Path, max_chars: int = MAX_TEXT_CHARS) -> str:（函式體一字不改，MAX_TEXT_CHARS 截斷留在內）。把 process_file_worker 內 content = extract_text(local_path)（~line 330）暫時改為 content = _extract_text_legacy(local_path)（Task 4 會再改成走 build_content_blocks）。

Step 2: 跑既有 import smoke

Run: cd scripts/evidence/classify/docker && python3 -c "import importlib, container_entrypoint as c; importlib.reload(c); print(hasattr(c,'_extract_text_legacy'), not hasattr(c,'extract_text'))" Expected: True True

Step 3: 跑測試確認 Task 1 test 仍綠

Run: python3 -m pytest test_container_entrypoint.py -v Expected: PASS

Step 4: Commit

git add scripts/evidence/classify/docker/container_entrypoint.py
git commit -m "refactor(evidence-classify): extract_text 改名 _extract_text_legacy"

§5

Task 3: `build_content_blocks` — 四分支（text / image / pdf / unsupported）

Files:

Modify: scripts/evidence/classify/docker/container_entrypoint.py
Test: scripts/evidence/classify/docker/test_container_entrypoint.py

Step 1: 寫 failing tests（涵蓋全部分支 + 超限）

加到 test_container_entrypoint.py：

import base64


def _make_png(tmp_path, name="a.png", size_bytes=100):
    # 最小合法 PNG header + padding
    png = (b"\x89PNG\r\n\x1a\n" + b"\x00" * max(0, size_bytes - 8))
    p = tmp_path / name
    p.write_bytes(png)
    return p


def test_blocks_text_file(tmp_path):
    p = tmp_path / "note.txt"
    p.write_text("hello evidence")
    blocks, skip = c.build_content_blocks(p, "text/plain", "note.txt")
    assert skip is None
    assert len(blocks) == 1
    assert blocks[0]["type"] == "text"
    assert "note.txt" in blocks[0]["text"]
    assert "hello evidence" in blocks[0]["text"]


def test_blocks_image(tmp_path):
    p = _make_png(tmp_path)
    blocks, skip = c.build_content_blocks(p, "image/png", "a.png")
    assert skip is None
    assert blocks[0]["type"] == "text" and "a.png" in blocks[0]["text"]
    assert blocks[1]["type"] == "image"
    assert blocks[1]["source"]["media_type"] == "image/png"
    assert base64.standard_b64decode(blocks[1]["source"]["data"])  # 可解碼


def test_blocks_image_oversize(tmp_path):
    p = _make_png(tmp_path, name="big.png", size_bytes=5 * 1024 * 1024 + 1)
    blocks, skip = c.build_content_blocks(p, "image/png", "big.png")
    assert blocks == []
    assert "5MB" in skip


def test_blocks_pdf(tmp_path):
    from pypdf import PdfWriter
    w = PdfWriter(); w.add_blank_page(width=72, height=72)
    p = tmp_path / "doc.pdf"
    with open(p, "wb") as f:
        w.write(f)
    blocks, skip = c.build_content_blocks(p, "application/pdf", "doc.pdf")
    assert skip is None
    assert blocks[0]["type"] == "text" and "doc.pdf" in blocks[0]["text"]
    assert blocks[1]["type"] == "document"
    assert blocks[1]["source"]["media_type"] == "application/pdf"


def test_blocks_pdf_too_many_pages(tmp_path, monkeypatch):
    from pypdf import PdfWriter
    w = PdfWriter(); w.add_blank_page(width=72, height=72)
    p = tmp_path / "many.pdf"
    with open(p, "wb") as f:
        w.write(f)
    monkeypatch.setattr(c, "_pdf_page_count", lambda _p: 101)
    blocks, skip = c.build_content_blocks(p, "application/pdf", "many.pdf")
    assert blocks == []
    assert "100" in skip


def test_blocks_unsupported(tmp_path):
    p = tmp_path / "movie.mp4"
    p.write_bytes(b"\x00\x00")
    blocks, skip = c.build_content_blocks(p, "video/mp4", "movie.mp4")
    assert blocks == []
    assert "unsupported" in skip.lower()

Step 2: 跑 test 確認 fail

Run: python3 -m pytest test_container_entrypoint.py -k blocks -v Expected: FAIL — module has no attribute 'build_content_blocks'

Step 3: 實作 build_content_blocks

在 _extract_text_legacy 後加：

_IMAGE_EXTS = {".png", ".jpg", ".jpeg", ".gif", ".webp"}
_MAX_IMAGE_BYTES = 5 * 1024 * 1024
_MAX_PDF_BYTES = 32 * 1024 * 1024
_MAX_PDF_PAGES = 100


def build_content_blocks(path: Path, mime: str, filename: str):
    """回傳 (content_blocks, skip_reason)。skip_reason 非 None → 跳過該檔。"""
    ext = path.suffix.lower()

    if ext in {".docx", ".csv", ".log", ".txt", ".md"}:
        text = _extract_text_legacy(path)
        return [{
            "type": "text",
            "text": f"Filename: {filename}\nContent preview:\n\"\"\"\n{text}\n\"\"\"",
        }], None

    if ext in _IMAGE_EXTS:
        if path.stat().st_size > _MAX_IMAGE_BYTES:
            return [], "image exceeds 5MB API limit"
        b64 = base64.standard_b64encode(path.read_bytes()).decode()
        return [
            {"type": "text", "text": f"Filename: {filename}\n(image evidence)"},
            {"type": "image", "source": {
                "type": "base64", "media_type": _img_media_type(ext), "data": b64}},
        ], None

    if ext == ".pdf":
        if path.stat().st_size > _MAX_PDF_BYTES:
            return [], "pdf exceeds 32MB API limit"
        if _pdf_page_count(path) > _MAX_PDF_PAGES:
            return [], "pdf exceeds 100-page API limit"
        b64 = base64.standard_b64encode(path.read_bytes()).decode()
        return [
            {"type": "text", "text": f"Filename: {filename}\n(PDF evidence)"},
            {"type": "document", "source": {
                "type": "base64", "media_type": "application/pdf", "data": b64}},
        ], None

    return [], f"unsupported file type: {ext}"

Step 4: 跑 test 確認 pass

Run: python3 -m pytest test_container_entrypoint.py -k blocks -v Expected: PASS（6 passed）

Step 5: Commit

git add scripts/evidence/classify/docker/container_entrypoint.py \
        scripts/evidence/classify/docker/test_container_entrypoint.py
git commit -m "feat(evidence-classify): build_content_blocks 支援圖片/PDF + 超限 skip"

§6

Task 4: `classify_with_claude` 改收 content_blocks + `process_file_worker` 串接

Files:

Modify: scripts/evidence/classify/docker/container_entrypoint.py（classify_with_claude ~277、process_file_worker ~312）
Test: scripts/evidence/classify/docker/test_container_entrypoint.py

Step 1: 寫 failing tests（mock Claude client）

加到測試：

class _FakeContent:
    def __init__(self, text): self.text = text

class _FakeResp:
    def __init__(self, text): self.content = [_FakeContent(text)]

class _FakeMessages:
    def __init__(self, outer): self.outer = outer
    def create(self, **kwargs):
        self.outer.last_kwargs = kwargs
        return _FakeResp('{"matches": [{"ao_id": "AC.L1-3.1.1[a]", '
                         '"confidence": 0.9, "reasoning": "x"}], '
                         '"primary": "AC.L1-3.1.1[a]"}')

class _FakeClient:
    def __init__(self): self.messages = _FakeMessages(self); self.last_kwargs = None


def test_classify_passes_blocks_through():
    client = _FakeClient()
    blocks = [{"type": "text", "text": "Filename: a.txt"}]
    out = c.classify_with_claude(client, "claude-sonnet-4-6", "SYS", blocks)
    # content blocks 原樣傳給 messages.create
    assert client.last_kwargs["messages"][0]["content"] is blocks
    # system block 仍帶 prompt caching
    assert client.last_kwargs["system"][0]["cache_control"] == {"type": "ephemeral"}
    assert out["primary"] == "AC.L1-3.1.1[a]"

Step 2: 跑 test 確認 fail

Run: python3 -m pytest test_container_entrypoint.py -k classify -v Expected: FAIL —目前 classify_with_claude 簽名是 (client, model, system_block, filename, content, retries=3)，呼叫缺參數會 TypeError。

Step 3: 改 classify_with_claude 簽名

實際簽名橫跨兩行且帶 type hint，把：

def classify_with_claude(client, model: str, system_block: str,
                         filename: str, content: str, retries: int = 3) -> dict:
    user_message = (
        f"Filename: {filename}\n"
        f"Content preview:\n\"\"\"\n{content}\n\"\"\""
    )
    ...
        messages=[{"role": "user", "content": user_message}],

改為：

def classify_with_claude(client, model: str, system_block: str,
                         content_blocks, retries: int = 3) -> dict:
    ...
        messages=[{"role": "user", "content": content_blocks}],

（移除 user_message 組裝；retry / fence 去除 / json.loads / cache_control system block 全部不動。）

Step 4: 改 process_file_worker 串接（保留 try/finally + error 分支）

在 try: 內把（此時已是 Task 2 改名後的狀態）：

        download_drive_file(drive, file_id, mime, local_path)
        content = _extract_text_legacy(local_path)
        result = classify_with_claude(claude_client, args.model, system_block, name, content)

改為：

        download_drive_file(drive, file_id, mime, local_path)
        blocks, skip_reason = build_content_blocks(local_path, mime, name)
        if skip_reason:
            return {"file_drive_id": file_id, "file_name": name,
                    "matches": [], "primary": None, "skipped_reason": skip_reason}
        result = classify_with_claude(claude_client, args.model, system_block, blocks)

保留外層 try/finally（local_path.unlink() cleanup）與既有 except Exception → {error: str(exc), matches: [], primary: None} 分支不變。注意：return 在 try 內，finally 仍會跑 cleanup（正確）。

Step 5: 跑 test 確認 pass

Run: python3 -m pytest test_container_entrypoint.py -v Expected: PASS（全部）

Step 6: Commit

git add scripts/evidence/classify/docker/container_entrypoint.py \
        scripts/evidence/classify/docker/test_container_entrypoint.py
git commit -m "feat(evidence-classify): classify_with_claude 收 content_blocks + worker 串接 skip"

§7

Task 5: `build_system_block` prompt 措辭放寬

Files:

Modify: scripts/evidence/classify/docker/container_entrypoint.py（build_system_block ~233）
Test: scripts/evidence/classify/docker/test_container_entrypoint.py

Step 1: 寫 failing test

def test_system_block_mentions_image_and_pdf():
    catalog = {"domains": [{"code": "AC", "name": "Access Control", "controls": [
        {"id": "AC.L1-3.1.1", "name": "x", "statement": "s",
         "assessment_objectives": [{"letter": "a", "text": "t"}]}]}]}
    sb = c.build_system_block(catalog)
    assert "image" in sb and "PDF" in sb
    assert "screenshots" in sb or "PDF pages" in sb

Step 2: 跑 test 確認 fail

Run: python3 -m pytest test_container_entrypoint.py -k system_block -v Expected: FAIL — 現 prompt 只提 "filename + content preview"，無 image/PDF。

Step 3: 改 prompt 兩處措辭

把 Task 段：

        "Given an evidence file (filename + content preview), classify which",
        "Assessment Objective(s) the file best satisfies as audit evidence. A single",

改為：

        "Given an evidence file (filename + its content, which may be text, an",
        "image, or a PDF document), classify which Assessment Objective(s) the file",
        "best satisfies as audit evidence. A single",

把 reasoning 規則：

        "- The reasoning must cite specific evidence from the file content.",

改為：

        "- The reasoning must cite specific evidence from the file content "
        "(including what is visible in screenshots / images / PDF pages).",

Step 4: 跑 test 確認 pass

Run: python3 -m pytest test_container_entrypoint.py -v Expected: PASS（全部）

Step 5: Commit

git add scripts/evidence/classify/docker/container_entrypoint.py \
        scripts/evidence/classify/docker/test_container_entrypoint.py
git commit -m "feat(evidence-classify): prompt 放寬讓 Claude 知道可能收到圖/PDF"

§8

Task 6: 全套件測試 + Dockerfile rebuild 驗證

Files: 無新增（驗證 + 文件）

Step 1: 跑全部單元測試

Run: cd scripts/evidence/classify/docker && python3 -m pytest test_container_entrypoint.py -v Expected: 全 PASS（helper 2 + blocks 6 + classify 1 + system_block 1）

Step 2: 確認 Dockerfile 會帶到新依賴

Dockerfile 已 COPY requirements.txt → pip install -r requirements.txt，新增 pypdf 會自動被裝，無需改 Dockerfile。確認測試檔不會被 COPY 進 image（image 只 COPY container_entrypoint.py + jedi_helpers/，測試檔留在 repo 即可）。

Run（可選，需 docker）:

docker build -t cmmc-classifier:fr0302-test -f scripts/evidence/classify/docker/Dockerfile scripts/evidence/classify/docker && echo BUILD_OK

Expected: BUILD_OK（pypdf 裝成功）

Step 3: （手動 / user）端對端 smoke

需真實 Drive + Claude API key，由 user 在備好證據（含 1 張截圖 + 1 份 PDF）的測試專案觸發一次分類，確認：截圖 / PDF 出現在 report 且有 match、超大檔列未處理。此步交 user 驗收，不在自動測試內。

Step 4: 收尾文件（等 user 下令才做）

依 CLAUDE.md「收尾必須等 user 下命令才做」：fix code commit 完停下，給 user 一句話 status + 手測 checklist。changelog（docs/changelog/YYYY-MM-DD-feat-evidence-classify-image-vision.md）/ design.md §11 reconciliation 等收尾文件，等 user 明確說「收尾」再 batch 寫。

§9

注意事項（implementer 必讀）

不切 branch：在 user 當下 branch 工作；發現 branch 不對停下問 user。
git add 顯式檔名：禁用 -am（此 repo working tree 有其它未提交 doc 變更，會被誤掃）。
改 container code 後：container 是獨立 image，主專案 BE 重啟無關；但若 user 已 build 過舊 image，需提醒重新 docker build 才生效。
不動主專案 app/domain/infra：本計畫全部改動侷限在 scripts/evidence/classify/docker/。
流程不變的紅線：write_outputs / _state.json 結構 / Drive 歸檔 / review UI 一律不碰。

File Structure

Task 0: Pre-flight 驗證（無 code，先做）

Task 1: 加 pypdf 依賴 + 兩個 helper（_img_media_type / _pdf_page_count）

Task 2: 把 extract_text 改名為 _extract_text_legacy（行為零變更）

Task 3: build_content_blocks — 四分支（text / image / pdf / unsupported）

Task 4: classify_with_claude 改收 content_blocks + process_file_worker 串接

Task 5: build_system_block prompt 措辭放寬

Task 6: 全套件測試 + Dockerfile rebuild 驗證

注意事項（implementer 必讀）

Task 1: 加 `pypdf` 依賴 + 兩個 helper（`_img_media_type` / `_pdf_page_count`）

Task 2: 把 `extract_text` 改名為 `_extract_text_legacy`（行為零變更）

Task 3: `build_content_blocks` — 四分支（text / image / pdf / unsupported）

Task 4: `classify_with_claude` 改收 content_blocks + `process_file_worker` 串接

Task 5: `build_system_block` prompt 措辭放寬