liqing/holy-python
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

按后台范围同步看板凭证图片

Browse Source
This commit is contained in:
李清
2026-09-02 16:47:52 +08:00
parent e4627e6ba3
commit 2b841b7a15
2 changed files with 33 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files

30
ocr_dashboard_service.py
View File

@@ -164,38 +164,28 @@ def list_evidences(params: dict[str, Any]) -> dict[str, Any]:
try:
response = SESSION.get(
f"{Config.EVIDENCE_ADMIN_BASE_URL}/api/evidences",
params={"batch_key": query["batch_key"], "question_code": query["question_code"]},
params={
"batch_key": query["batch_key"],
"collection_key": query.get("collection_key", ""),
"question_code": query["question_code"],
},
timeout=min(Config.CRAWLER_TIMEOUT_SECONDS, 10),
)
response.raise_for_status()
manual_items = (response.json().get("data") or [])
items = data.setdefault("items", [])
existing_urls = {str(item.get("cos_url") or "") for item in items}
object_indexes = {
str(item.get("cos_object_key")): index for index, item in enumerate(items)
if item.get("cos_object_key")
}
for item in manual_items:
url = str(item.get("cos_url") or "")
object_key = str(item.get("cos_object_key") or "")
if object_key and object_key in object_indexes:
items[object_indexes[object_key]] = item
existing_urls.add(url)
elif url and url not in existing_urls:
items.append(item)
if object_key:
object_indexes[object_key] = len(items) - 1
existing_urls.add(url)
data["items"] = manual_items
evidence_type_order = {
"body_ranking": 0, "first_screen": 1, "reference_source": 2,
"before": 3, "after": 3, "before_after": 3, "viral_note": 4,
}
items.sort(key=lambda item: (
manual_items.sort(key=lambda item: (
evidence_type_order.get(str(item.get("evidence_type") or ""), 99),
int(item.get("sort_order") or 0),
int(item.get("image_id") or item.get("evidence_id") or item.get("id") or 0),
))
data["total"] = len(data.get("items") or [])
data["total"] = len(manual_items)
data["limit"] = limit
data["is_complete"] = True
except (requests.RequestException, ValueError):
pass
return data

24
tests/test_ocr_dashboard_service.py
View File

@@ -1,11 +1,33 @@
import unittest
from unittest.mock import Mock, patch
from ocr_dashboard_service import export_evidences, get_comparison, get_summary, list_batches, list_configured_notes, list_results
from ocr_dashboard_service import export_evidences, get_comparison, get_summary, list_batches, list_configured_notes, list_evidences, list_results
from task_service import TaskError
class OcrDashboardServiceTest(unittest.TestCase):
@patch("ocr_dashboard_service.Config.EVIDENCE_ADMIN_BASE_URL", "http://admin")
@patch("ocr_dashboard_service._get")
@patch("ocr_dashboard_service.SESSION.get")
def test_evidences_use_exact_admin_scope_as_display_source(self, get, upstream_get):
upstream_get.return_value = {"items": [{"cos_object_key": "old.jpg"}], "total": 1}
response = Mock()
response.json.return_value = {"data": [{
"id": 2, "image_id": 3, "evidence_type": "first_screen",
"cos_object_key": "managed.jpg", "sort_order": 1,
}]}
get.return_value = response
result = list_evidences({
"batch_key": "b1", "collection_key": "c1", "question_code": "Q1", "limit": 100,
})
self.assertEqual([item["cos_object_key"] for item in result["items"]], ["managed.jpg"])
self.assertEqual(result["total"], 1)
self.assertEqual(get.call_args.kwargs["params"], {
"batch_key": "b1", "collection_key": "c1", "question_code": "Q1",
})
@patch("ocr_dashboard_service.SESSION.get")
def test_configured_notes_reads_admin_list(self, get):
response = Mock()