refactor!: 删除 Web/CLI 应用层与内置 LLM 抽取，出库单功能下线（2b）
入口只剩 MCP 服务器；编排由宿主 Agent 负责，本项目只留工具与确定性内核。
删除：
- src/web/（Flask 路由、SSE 轮询、自研聊天前端、模板与静态资源，约 3400 行）
- src/main.py、src/pipeline.py、src/pipeline_core.py（CLI 全流程入口与两套重复管道）
- core/extraction/{extractor,llm_extractor}.py（内置 LLM 抽取、llm_stream.log、file_events.log、并行抽取）
- infra/documents/consumable.py 与 infra/documents/pdf.py：出库单功能下线（Word 模板 4fbb4c3 起已不在仓库，
该路径此前一直静默失败），PDF 渲染由 core/extraction/render.py 取代
- mcp__arb__doc_consumable 工具；exceptions.ExtractionError；config 的 consumable_storage / get_llm_config /
load_session_config / SESSION_CONFIG_KEYS（Web 专属，已无引用）
- 针对被删代码的 tests/test_extractor.py、tests/test_llm_extractor.py 与 scripts/ 下的 LLM 调试脚本
- 依赖：llama-index、llama-index-llms-openai-like、python-dotenv（连同其 50+ 传递依赖）
- 提交进来的调试截图 src/images/、根目录 images/ 与 .coverage（并补 .gitignore）
保留与新增：
- 材料入口改为独立进程 src/upload（手机扫码/PC 上传 → cases/<id>/inbox/）
- runner.py 自带 save_invoice_groups，不再依赖 pipeline_core
- scripts/mcp_smoke.py：stdio 自检（拉起自身、列工具、调 case_list、验证 instructions 下发）
- make mcp / mcp-smoke / upload 取代 make run；check 的 mypy 目标从 src/main.py 扩到 src
验证：mcp_smoke 通过（17 个工具、instructions 1449 字节、case_list 正常）；申购单实机生成此前已验证。
门禁：ruff check / format --check(43) / mypy src(36 文件) / deptry / pytest 107 passed。
注：仍用 --no-verify 提交并手动跑等价检查（原因见 2f08bbd 提交说明）。
This commit is contained in:
wandering
@@ -1,332 +0,0 @@
|
||||
"""发票提取编排模块单元测试
|
||||
|
||||
覆盖范围：
|
||||
- extract_invoices：空目录、提取失败、正常流程、分类结果、申请单分离
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
|
||||
from src import exceptions
|
||||
from src.core.extraction import extract_invoices
|
||||
|
||||
# 字段键名（与源码中的字符串字面量保持一致）
|
||||
K_INVOICE_TYPE = "invoice_type"
|
||||
K_INVOICE_NUMBER = "invoice_number"
|
||||
K_TOTAL_AMOUNT = "total_amount"
|
||||
K_ITEM_NAME = "item_name"
|
||||
K_PERSON_NAME = "person_name"
|
||||
K_CARD_DATE = "card_date"
|
||||
K_CARD_NO = "card_no"
|
||||
K_CARD_AMOUNT = "card_amount"
|
||||
K_MATCHED_INVOICES = "_matched_invoices"
|
||||
K_RELATIVE_INVOICE_COUNT = "relative_invoice_count"
|
||||
K_INVOICE_DETAIL = "invoice_detail"
|
||||
K_REMARK = "remark"
|
||||
|
||||
# 发票类型
|
||||
INVOICE_TYPE_TRAIN = "train"
|
||||
INVOICE_TYPE_HOTEL = "hotel"
|
||||
INVOICE_TYPE_GENERAL = "general"
|
||||
INVOICE_TYPE_PAYMENT = "payment"
|
||||
DOCUMENT_TYPE_APPLICATION = "application"
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Fixture helpers
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
|
||||
def _make_invoice(number: str, amount: float, inv_type: str = INVOICE_TYPE_GENERAL) -> dict[str, Any]:
|
||||
inv: dict[str, Any] = {
|
||||
K_INVOICE_NUMBER: number,
|
||||
K_INVOICE_TYPE: inv_type,
|
||||
K_TOTAL_AMOUNT: str(amount),
|
||||
}
|
||||
if inv_type == INVOICE_TYPE_TRAIN:
|
||||
inv[K_PERSON_NAME] = f"person{number}"
|
||||
elif inv_type == INVOICE_TYPE_GENERAL:
|
||||
inv[K_ITEM_NAME] = f"item{number}"
|
||||
return inv
|
||||
|
||||
|
||||
def _make_application() -> dict[str, Any]:
|
||||
return {
|
||||
K_INVOICE_TYPE: DOCUMENT_TYPE_APPLICATION,
|
||||
"applicant": "张三",
|
||||
}
|
||||
|
||||
|
||||
def _make_card(amount: float) -> dict[str, Any]:
|
||||
return {
|
||||
K_INVOICE_TYPE: INVOICE_TYPE_PAYMENT,
|
||||
K_CARD_DATE: "2026-01-01",
|
||||
K_CARD_NO: "6228480000000000",
|
||||
K_CARD_AMOUNT: f"{amount:.2f}",
|
||||
}
|
||||
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# extract_invoices
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestExtractInvoices:
|
||||
"""extract_invoices 编排函数"""
|
||||
|
||||
def test_empty_directory(self, tmp_path: Path, monkeypatch):
|
||||
monkeypatch.setattr("src.core.extraction.extractor._find_all_files", lambda d: [])
|
||||
|
||||
records, apps, groups = extract_invoices(str(tmp_path))
|
||||
assert records == []
|
||||
assert apps == []
|
||||
assert groups == {"travel": [], "general": [], "application": [], "attachments": []}
|
||||
|
||||
def test_extraction_fails(self, tmp_path: Path, monkeypatch):
|
||||
pdf = tmp_path / "broken.pdf"
|
||||
pdf.touch()
|
||||
|
||||
monkeypatch.setattr("src.core.extraction.extractor._find_all_files", lambda d: [pdf])
|
||||
monkeypatch.setattr("src.core.extraction.extractor._extract_document", lambda p, c, s: (None, "parse error"))
|
||||
|
||||
with pytest.raises(exceptions.ExtractionError) as exc_info:
|
||||
extract_invoices(str(tmp_path))
|
||||
|
||||
assert "broken.pdf" in exc_info.value.failed_files
|
||||
assert exc_info.value.details["broken.pdf"] == "parse error"
|
||||
|
||||
def test_normal_flow_general_invoices(self, tmp_path: Path, monkeypatch):
|
||||
pdf1 = tmp_path / "inv1.pdf"
|
||||
pdf2 = tmp_path / "inv2.pdf"
|
||||
pdf1.touch()
|
||||
pdf2.touch()
|
||||
|
||||
inv1 = _make_invoice("INV001", 300.0, INVOICE_TYPE_GENERAL)
|
||||
inv2 = _make_invoice("INV002", 200.0, INVOICE_TYPE_GENERAL)
|
||||
|
||||
monkeypatch.setattr("src.core.extraction.extractor._find_all_files", lambda d: [pdf1, pdf2])
|
||||
|
||||
path_to_result = {pdf1: inv1, pdf2: inv2}
|
||||
|
||||
def fake_extract(path, cache_dir, source_dir):
|
||||
return (path_to_result[path], None)
|
||||
|
||||
monkeypatch.setattr("src.core.extraction.extractor._extract_document", fake_extract)
|
||||
|
||||
def fake_match(invoices, cards):
|
||||
return [
|
||||
{
|
||||
K_CARD_DATE: "2026-01-01",
|
||||
K_CARD_NO: "6228480000000000",
|
||||
K_CARD_AMOUNT: "500.00",
|
||||
K_RELATIVE_INVOICE_COUNT: "2",
|
||||
K_INVOICE_DETAIL: "",
|
||||
K_REMARK: "",
|
||||
K_MATCHED_INVOICES: [inv1, inv2],
|
||||
}
|
||||
]
|
||||
|
||||
monkeypatch.setattr("src.core.extraction.extractor.match_invoices_to_cards", fake_match)
|
||||
|
||||
records, apps, groups = extract_invoices(str(tmp_path))
|
||||
|
||||
assert len(records) == 1
|
||||
assert records[0][K_RELATIVE_INVOICE_COUNT] == "2"
|
||||
assert len(groups["travel"]) == 0
|
||||
assert len(groups["general"]) == 2
|
||||
|
||||
def test_normal_flow_mixed_invoices(self, tmp_path: Path, monkeypatch):
|
||||
pdf1 = tmp_path / "train.pdf"
|
||||
pdf2 = tmp_path / "hotel.pdf"
|
||||
pdf3 = tmp_path / "general.pdf"
|
||||
pdf1.touch()
|
||||
pdf2.touch()
|
||||
pdf3.touch()
|
||||
|
||||
inv_train = _make_invoice("TRAIN001", 500.0, INVOICE_TYPE_TRAIN)
|
||||
inv_hotel = _make_invoice("HOTEL001", 800.0, INVOICE_TYPE_HOTEL)
|
||||
inv_general = _make_invoice("GEN001", 150.0, INVOICE_TYPE_GENERAL)
|
||||
|
||||
monkeypatch.setattr(
|
||||
"src.core.extraction.extractor._find_all_files",
|
||||
lambda d: [pdf1, pdf2, pdf3],
|
||||
)
|
||||
|
||||
invoices_list = [inv_train, inv_hotel, inv_general]
|
||||
path_to_result = dict(zip([pdf1, pdf2, pdf3], invoices_list, strict=True))
|
||||
|
||||
def fake_extract(path, cache_dir, source_dir):
|
||||
return (path_to_result[path], None)
|
||||
|
||||
monkeypatch.setattr("src.core.extraction.extractor._extract_document", fake_extract)
|
||||
|
||||
def fake_match(invoices, cards):
|
||||
return [
|
||||
{
|
||||
K_CARD_DATE: "2026-01-01",
|
||||
K_CARD_NO: "6228480000000000",
|
||||
K_CARD_AMOUNT: "500.00",
|
||||
K_RELATIVE_INVOICE_COUNT: "1",
|
||||
K_INVOICE_DETAIL: "",
|
||||
K_REMARK: "",
|
||||
K_MATCHED_INVOICES: [inv_train],
|
||||
},
|
||||
{
|
||||
K_CARD_DATE: "2026-01-02",
|
||||
K_CARD_NO: "6228480000000000",
|
||||
K_CARD_AMOUNT: "800.00",
|
||||
K_RELATIVE_INVOICE_COUNT: "1",
|
||||
K_INVOICE_DETAIL: "",
|
||||
K_REMARK: "",
|
||||
K_MATCHED_INVOICES: [inv_hotel],
|
||||
},
|
||||
{
|
||||
K_CARD_DATE: "",
|
||||
K_CARD_NO: "",
|
||||
K_CARD_AMOUNT: "",
|
||||
K_RELATIVE_INVOICE_COUNT: "1",
|
||||
K_INVOICE_DETAIL: "",
|
||||
K_REMARK: "unmatched",
|
||||
K_MATCHED_INVOICES: [inv_general],
|
||||
},
|
||||
]
|
||||
|
||||
monkeypatch.setattr("src.core.extraction.extractor.match_invoices_to_cards", fake_match)
|
||||
|
||||
records, apps, groups = extract_invoices(str(tmp_path))
|
||||
|
||||
assert len(records) == 3
|
||||
assert len(groups["travel"]) == 2
|
||||
assert len(groups["general"]) == 1
|
||||
|
||||
travel_numbers = {inv[K_INVOICE_NUMBER] for inv in groups["travel"]}
|
||||
assert "TRAIN001" in travel_numbers
|
||||
assert "HOTEL001" in travel_numbers
|
||||
assert groups["general"][0][K_INVOICE_NUMBER] == "GEN001"
|
||||
|
||||
def test_application_documents_separated(self, tmp_path: Path, monkeypatch):
|
||||
pdf1 = tmp_path / "invoice.pdf"
|
||||
pdf2 = tmp_path / "application.pdf"
|
||||
pdf1.touch()
|
||||
pdf2.touch()
|
||||
|
||||
inv = _make_invoice("INV001", 300.0, INVOICE_TYPE_GENERAL)
|
||||
app = _make_application()
|
||||
|
||||
monkeypatch.setattr(
|
||||
"src.core.extraction.extractor._find_all_files",
|
||||
lambda d: [pdf1, pdf2],
|
||||
)
|
||||
|
||||
path_to_result = {pdf1: inv, pdf2: app}
|
||||
|
||||
def fake_extract(path, cache_dir, source_dir):
|
||||
return (path_to_result[path], None)
|
||||
|
||||
monkeypatch.setattr("src.core.extraction.extractor._extract_document", fake_extract)
|
||||
|
||||
def fake_match(invoices, cards):
|
||||
return [
|
||||
{
|
||||
K_CARD_DATE: "2026-01-01",
|
||||
K_CARD_NO: "6228480000000000",
|
||||
K_CARD_AMOUNT: "300.00",
|
||||
K_RELATIVE_INVOICE_COUNT: "1",
|
||||
K_INVOICE_DETAIL: "",
|
||||
K_REMARK: "",
|
||||
K_MATCHED_INVOICES: [inv],
|
||||
}
|
||||
]
|
||||
|
||||
monkeypatch.setattr("src.core.extraction.extractor.match_invoices_to_cards", fake_match)
|
||||
|
||||
records, apps, groups = extract_invoices(str(tmp_path))
|
||||
|
||||
assert len(apps) == 1
|
||||
assert apps[0]["applicant"] == "张三"
|
||||
assert len(groups["application"]) == 1
|
||||
|
||||
def test_payment_records_included(self, tmp_path: Path, monkeypatch):
|
||||
pdf1 = tmp_path / "invoice.pdf"
|
||||
pdf2 = tmp_path / "card.png"
|
||||
pdf1.touch()
|
||||
pdf2.touch()
|
||||
|
||||
inv = _make_invoice("INV001", 300.0, INVOICE_TYPE_GENERAL)
|
||||
card = _make_card(300.0)
|
||||
|
||||
monkeypatch.setattr(
|
||||
"src.core.extraction.extractor._find_all_files",
|
||||
lambda d: [pdf1, pdf2],
|
||||
)
|
||||
|
||||
path_to_result = {pdf1: inv, pdf2: card}
|
||||
|
||||
def fake_extract(path, cache_dir, source_dir):
|
||||
return (path_to_result[path], None)
|
||||
|
||||
monkeypatch.setattr("src.core.extraction.extractor._extract_document", fake_extract)
|
||||
|
||||
def fake_match(invoices, cards):
|
||||
return [
|
||||
{
|
||||
K_CARD_DATE: "2026-01-01",
|
||||
K_CARD_NO: "6228480000000000",
|
||||
K_CARD_AMOUNT: "300.00",
|
||||
K_RELATIVE_INVOICE_COUNT: "1",
|
||||
K_INVOICE_DETAIL: "",
|
||||
K_REMARK: "",
|
||||
K_MATCHED_INVOICES: [inv],
|
||||
}
|
||||
]
|
||||
|
||||
monkeypatch.setattr("src.core.extraction.extractor.match_invoices_to_cards", fake_match)
|
||||
|
||||
records, apps, groups = extract_invoices(str(tmp_path))
|
||||
|
||||
assert len(records) == 1
|
||||
assert len(groups["general"]) == 1
|
||||
|
||||
def test_partial_extraction_failure(self, tmp_path: Path, monkeypatch):
|
||||
pdf1 = tmp_path / "good.pdf"
|
||||
pdf2 = tmp_path / "bad.pdf"
|
||||
pdf1.touch()
|
||||
pdf2.touch()
|
||||
|
||||
inv = _make_invoice("INV001", 300.0, INVOICE_TYPE_GENERAL)
|
||||
|
||||
monkeypatch.setattr(
|
||||
"src.core.extraction.extractor._find_all_files",
|
||||
lambda d: [pdf1, pdf2],
|
||||
)
|
||||
|
||||
path_to_result = {pdf1: inv, pdf2: None}
|
||||
|
||||
def fake_extract(path, cache_dir, source_dir):
|
||||
result = path_to_result[path]
|
||||
return (result, "parse error") if result is None else (result, None)
|
||||
|
||||
monkeypatch.setattr("src.core.extraction.extractor._extract_document", fake_extract)
|
||||
|
||||
def fake_match(invoices, cards):
|
||||
return [
|
||||
{
|
||||
K_CARD_DATE: "2026-01-01",
|
||||
K_CARD_NO: "6228480000000000",
|
||||
K_CARD_AMOUNT: "300.00",
|
||||
K_RELATIVE_INVOICE_COUNT: "1",
|
||||
K_INVOICE_DETAIL: "",
|
||||
K_REMARK: "",
|
||||
K_MATCHED_INVOICES: [inv],
|
||||
}
|
||||
]
|
||||
|
||||
monkeypatch.setattr("src.core.extraction.extractor.match_invoices_to_cards", fake_match)
|
||||
|
||||
records, apps, groups = extract_invoices(str(tmp_path))
|
||||
|
||||
assert len(records) == 1
|
||||
assert len(groups["general"]) == 1
|
||||
@@ -1,202 +0,0 @@
|
||||
"""LLM 信息提取模块单元测试
|
||||
|
||||
覆盖范围：
|
||||
- `parse_json_response`：纯 JSON、Markdown 包裹、带前缀、解析失败
|
||||
- _image_to_base64：图片转 base64
|
||||
- extract_document：成功提取、LLM 失败
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import base64
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from src.core.extraction import (
|
||||
_image_to_base64,
|
||||
extract_document,
|
||||
parse_json_response,
|
||||
)
|
||||
|
||||
# 字段键名（与源码中的字符串字面量保持一致）
|
||||
K_INVOICE_NUMBER = "invoice_number"
|
||||
K_TOTAL_AMOUNT = "total_amount"
|
||||
K_CARD_DATE = "card_date"
|
||||
K_CARD_NO = "card_no"
|
||||
K_CARD_AMOUNT = "card_amount"
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# parse_json_response
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestParseJsonResponse:
|
||||
"""JSON 响应解析"""
|
||||
|
||||
def test_pure_json(self):
|
||||
raw = json.dumps({K_INVOICE_NUMBER: "123456", K_TOTAL_AMOUNT: "100.00"})
|
||||
result = parse_json_response(raw)
|
||||
assert result[K_INVOICE_NUMBER] == "123456"
|
||||
assert result[K_TOTAL_AMOUNT] == "100.00"
|
||||
|
||||
def test_markdown_json_block(self):
|
||||
raw = f'```json\n{{"{K_INVOICE_NUMBER}": "789"}}\n```'
|
||||
result = parse_json_response(raw)
|
||||
assert result[K_INVOICE_NUMBER] == "789"
|
||||
|
||||
def test_markdown_block_without_lang(self):
|
||||
raw = '```\n{"key": "value"}\n```'
|
||||
result = parse_json_response(raw)
|
||||
assert result["key"] == "value"
|
||||
|
||||
def test_json_prefix(self):
|
||||
raw = f'json\n{{"{K_INVOICE_NUMBER}": "001"}}'
|
||||
result = parse_json_response(raw)
|
||||
assert result[K_INVOICE_NUMBER] == "001"
|
||||
|
||||
def test_json_prefix_with_whitespace(self):
|
||||
raw = ' json \n{"a": 1}'
|
||||
result = parse_json_response(raw)
|
||||
assert result["a"] == 1
|
||||
|
||||
def test_nested_json(self):
|
||||
raw = json.dumps({"outer": {"inner": [1, 2, 3]}})
|
||||
result = parse_json_response(raw)
|
||||
assert result["outer"]["inner"] == [1, 2, 3]
|
||||
|
||||
def test_whitespace_around_json(self):
|
||||
raw = ' \n {"x": 42} \n '
|
||||
result = parse_json_response(raw)
|
||||
assert result["x"] == 42
|
||||
|
||||
def test_invalid_json_raises(self):
|
||||
with pytest.raises(ValueError):
|
||||
parse_json_response("not json at all")
|
||||
|
||||
def test_empty_string_raises(self):
|
||||
with pytest.raises(ValueError):
|
||||
parse_json_response("")
|
||||
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# _image_to_base64
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestImageToBase64:
|
||||
"""图片转 base64"""
|
||||
|
||||
def test_png_to_base64(self, tmp_path: Path):
|
||||
content = b"\x89PNG\r\n\x1a\nfake_png_data"
|
||||
img_path = tmp_path / "test.png"
|
||||
img_path.write_bytes(content)
|
||||
|
||||
result = _image_to_base64(img_path)
|
||||
assert isinstance(result, str)
|
||||
assert base64.b64decode(result) == content
|
||||
|
||||
def test_jpg_to_base64(self, tmp_path: Path):
|
||||
content = b"\xff\xd8\xff\xe0fake_jpg_data"
|
||||
img_path = tmp_path / "test.jpg"
|
||||
img_path.write_bytes(content)
|
||||
|
||||
result = _image_to_base64(img_path)
|
||||
assert base64.b64decode(result) == content
|
||||
|
||||
def test_file_not_found_raises(self, tmp_path: Path):
|
||||
img_path = tmp_path / "nonexistent.png"
|
||||
with pytest.raises(FileNotFoundError):
|
||||
_image_to_base64(img_path)
|
||||
|
||||
def test_returns_utf8_string(self, tmp_path: Path):
|
||||
content = b"test_image_content"
|
||||
img_path = tmp_path / "test.png"
|
||||
img_path.write_bytes(content)
|
||||
|
||||
result = _image_to_base64(img_path)
|
||||
assert type(result) is str
|
||||
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# extract_document (mock LLM)
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestExtractDocument:
|
||||
"""图片提取：通过 mock _llm_query_multimodal 避免真实 LLM 调用"""
|
||||
|
||||
def _mock_multimodal(self, monkeypatch, response_text: str):
|
||||
def fake_query(system_prompt, text, image_b64, max_tokens=4096):
|
||||
return response_text
|
||||
|
||||
monkeypatch.setattr("src.core.extraction.llm_extractor._llm_query_multimodal", fake_query)
|
||||
|
||||
def test_success(self, tmp_path: Path, monkeypatch):
|
||||
img_path = tmp_path / "card.png"
|
||||
img_path.write_bytes(b"fake_image")
|
||||
|
||||
mock_result = {
|
||||
K_CARD_DATE: "2026-01-10",
|
||||
K_CARD_NO: "6228480000000000",
|
||||
K_CARD_AMOUNT: "500.00",
|
||||
}
|
||||
self._mock_multimodal(monkeypatch, json.dumps(mock_result))
|
||||
|
||||
result = extract_document(img_path)
|
||||
assert result[K_CARD_DATE] == "2026-01-10"
|
||||
assert result[K_CARD_NO] == "6228480000000000"
|
||||
assert result[K_CARD_AMOUNT] == "500.00"
|
||||
|
||||
def test_llm_failure_propagates(self, tmp_path: Path, monkeypatch):
|
||||
img_path = tmp_path / "card.png"
|
||||
img_path.write_bytes(b"fake_image")
|
||||
|
||||
def fake_query(system_prompt, text, image_b64, max_tokens=4096):
|
||||
raise RuntimeError("模型不可用")
|
||||
|
||||
monkeypatch.setattr("src.core.extraction.llm_extractor._llm_query_multimodal", fake_query)
|
||||
with pytest.raises(RuntimeError, match="模型不可用"):
|
||||
extract_document(img_path)
|
||||
|
||||
def test_file_not_found_propagates(self, tmp_path: Path, monkeypatch):
|
||||
img_path = tmp_path / "missing.png"
|
||||
self._mock_multimodal(monkeypatch, "{}")
|
||||
with pytest.raises(FileNotFoundError):
|
||||
extract_document(img_path)
|
||||
|
||||
def test_markdown_wrapped_json(self, tmp_path: Path, monkeypatch):
|
||||
img_path = tmp_path / "card.png"
|
||||
img_path.write_bytes(b"fake_image")
|
||||
|
||||
mock_result = {
|
||||
K_CARD_DATE: "2026-02-01",
|
||||
K_CARD_NO: "6228481111111111",
|
||||
K_CARD_AMOUNT: "300.00",
|
||||
}
|
||||
wrapped = f"```json\n{json.dumps(mock_result)}\n```"
|
||||
self._mock_multimodal(monkeypatch, wrapped)
|
||||
|
||||
result = extract_document(img_path)
|
||||
assert result[K_CARD_DATE] == "2026-02-01"
|
||||
|
||||
def test_image_encoded_as_base64(self, tmp_path: Path, monkeypatch):
|
||||
img_path = tmp_path / "card.png"
|
||||
expected_content = b"test_image_data"
|
||||
img_path.write_bytes(expected_content)
|
||||
|
||||
received_b64 = None
|
||||
|
||||
def capture_b64(system_prompt, text, image_b64s, max_tokens=4096):
|
||||
nonlocal received_b64
|
||||
received_b64 = image_b64s
|
||||
return json.dumps({K_CARD_DATE: "2026-01-01", K_CARD_NO: "0000", K_CARD_AMOUNT: "100"})
|
||||
|
||||
monkeypatch.setattr("src.core.extraction.llm_extractor._llm_query_multimodal", capture_b64)
|
||||
extract_document(img_path)
|
||||
|
||||
assert received_b64 is not None
|
||||
assert isinstance(received_b64, list)
|
||||
assert len(received_b64) >= 1
|
||||
assert base64.b64decode(received_b64[0]) == expected_content
|
||||
Reference in New Issue
Block a user