admin/PACS
1
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

Refine skip and manual save annotations

Browse Source
This commit is contained in:
Codex
2026-05-27 10:38:03 +08:00
parent 45017fbdfd
commit bbf8466d56
5 changed files with 149 additions and 74 deletions
Download Patch File Download Diff File Expand all files Collapse all files

119
PACS_DICOM处理/数据处理网页端/app.py
View File

@@ -483,8 +483,6 @@ def bool_or_none(value: Any) -> bool | None:
def merge_parts(manual_parts: list[str], ai_parts: list[str], skipped: bool) -> list[str]:
if skipped:
return []
merged: list[str] = []
for part in manual_parts + ai_parts:
if part in BODY_PARTS and part not in merged:
@@ -493,14 +491,12 @@ def merge_parts(manual_parts: list[str], ai_parts: list[str], skipped: bool) ->
def effective_phase(manual_phase: str, ai_phase: str, body_parts: list[str], skipped: bool) -> str:
if skipped or "upper_abdomen" not in body_parts:
if "upper_abdomen" not in body_parts:
return ""
return valid_phase(manual_phase) or valid_phase(ai_phase)
def effective_plain_ct(manual_plain_ct: bool | None, ai_plain_ct: bool | None, fallback: bool, skipped: bool) -> bool:
if skipped:
return False
if manual_plain_ct is not None:
return manual_plain_ct
if ai_plain_ct is not None:
@@ -529,22 +525,35 @@ def normalize_annotation(row: dict[str, Any] | None, default_skipped: bool = Fal
)
return {
"body_parts": body_parts,
"manual_body_parts": [] if skipped else manual_parts,
"ai_body_parts": [] if skipped else ai_parts,
"manual_body_parts": manual_parts,
"ai_body_parts": ai_parts,
"upper_abdomen_phase": effective_phase(manual_phase, ai_phase, body_parts, skipped),
"manual_upper_abdomen_phase": "" if skipped else manual_phase,
"ai_upper_abdomen_phase": "" if skipped else ai_phase,
"manual_upper_abdomen_phase": manual_phase,
"ai_upper_abdomen_phase": ai_phase,
"plain_ct": plain_ct,
"manual_plain_ct": None if skipped else bool_or_none(row.get("manual_plain_ct")),
"ai_plain_ct": None if skipped else bool_or_none(row.get("ai_plain_ct")),
"manual_plain_ct": bool_or_none(row.get("manual_plain_ct")),
"ai_plain_ct": bool_or_none(row.get("ai_plain_ct")),
"skipped": skipped,
"ai_skipped": ai_skipped,
"notes": row.get("notes", ""),
"updated_at": row.get("updated_at", ""),
"ai_model": row.get("ai_model", ""),
"updated_by": row.get("updated_by", ""),
}
def dicom_auto_annotation(meta: dict[str, str]) -> tuple[list[str], str]:
marker = f"{meta.get('BodyPartExamined', '')} {meta.get('SeriesDescription', '')}".upper()
parts: list[str] = []
phase = ""
if "CHEST" in marker and "chest" not in parts:
parts.append("chest")
if "ABDOMEN" in marker and "upper_abdomen" not in parts:
parts.append("upper_abdomen")
phase = "unknown"
return parts, phase
def get_annotations(ct_number: str) -> dict[str, dict[str, Any]]:
try:
ensure_annotation_table()
@@ -553,7 +562,7 @@ def get_annotations(ct_number: str) -> dict[str, dict[str, Any]]:
SELECT
series_instance_uid, body_parts, manual_body_parts, ai_body_parts,
upper_abdomen_phase, manual_upper_abdomen_phase, ai_upper_abdomen_phase,
plain_ct, manual_plain_ct, ai_plain_ct, skipped, ai_skipped, notes, updated_at, ai_model
plain_ct, manual_plain_ct, ai_plain_ct, skipped, ai_skipped, notes, updated_at, ai_model, updated_by
FROM public.pacs_dicom_series_annotations
WHERE ct_number = {sql_literal(ct_number)}
"""
@@ -586,6 +595,7 @@ def scan_study(ct_number: str) -> dict[str, Any]:
series_list = []
file_map = {}
default_skip_rows = []
auto_annotation_rows = []
for uid, items in grouped.items():
items.sort(key=sort_key)
first = items[0][1]
@@ -593,13 +603,18 @@ def scan_study(ct_number: str) -> dict[str, Any]:
file_map[uid] = [path for path, _ in items]
raw_annotation = annotations.get(uid)
annotation = normalize_annotation(raw_annotation)
default_skipped = len(items) < 80 and not annotation.get("body_parts") and not annotation.get("plain_ct")
auto_parts, auto_phase = dicom_auto_annotation(first)
can_apply_auto = bool(auto_parts) and not annotation.get("body_parts")
if can_apply_auto:
annotation["manual_body_parts"] = auto_parts
annotation["body_parts"] = auto_parts
if auto_phase:
annotation["manual_upper_abdomen_phase"] = auto_phase
annotation["upper_abdomen_phase"] = auto_phase
auto_annotation_rows.append((uid, first, auto_parts, auto_phase))
default_skipped = len(items) < 80 and (not raw_annotation or annotation.get("updated_by") == "system")
if default_skipped:
annotation["skipped"] = True
annotation["manual_body_parts"] = []
annotation["ai_body_parts"] = []
annotation["body_parts"] = []
annotation["plain_ct"] = False
if default_skipped:
default_skip_rows.append((uid, first, len(items)))
series_list.append(
@@ -636,6 +651,42 @@ def scan_study(ct_number: str) -> dict[str, Any]:
}
)
for uid, first, auto_parts, auto_phase in auto_annotation_rows:
try:
pg_scalar(
f"""
INSERT INTO public.pacs_dicom_series_annotations (
ct_number, study_instance_uid, series_instance_uid, series_description,
body_parts, manual_body_parts, upper_abdomen_phase, manual_upper_abdomen_phase,
updated_by, updated_at
)
VALUES (
{sql_literal(ct_number)},
{sql_literal(first.get('StudyInstanceUID', ''))},
{sql_literal(uid)},
{sql_literal(first.get('SeriesDescription', '') or '未命名序列')},
{sql_literal(json.dumps(auto_parts, ensure_ascii=False))}::jsonb,
{sql_literal(json.dumps(auto_parts, ensure_ascii=False))}::jsonb,
{sql_literal(auto_phase)},
{sql_literal(auto_phase)},
'system',
now()
)
ON CONFLICT (ct_number, series_instance_uid) DO UPDATE SET
body_parts = EXCLUDED.body_parts,
manual_body_parts = EXCLUDED.manual_body_parts,
upper_abdomen_phase = EXCLUDED.upper_abdomen_phase,
manual_upper_abdomen_phase = EXCLUDED.manual_upper_abdomen_phase,
updated_by = 'system',
updated_at = now()
WHERE pacs_dicom_series_annotations.updated_by = 'system'
AND jsonb_array_length(pacs_dicom_series_annotations.body_parts) = 0
""",
timeout=8,
)
except Exception:
pass
for uid, first, count in default_skip_rows:
try:
pg_scalar(
@@ -662,13 +713,6 @@ def scan_study(ct_number: str) -> dict[str, Any]:
)
ON CONFLICT (ct_number, series_instance_uid) DO UPDATE SET
skipped = true,
body_parts = '[]'::jsonb,
manual_body_parts = '[]'::jsonb,
ai_body_parts = '[]'::jsonb,
upper_abdomen_phase = '',
manual_upper_abdomen_phase = '',
ai_upper_abdomen_phase = '',
plain_ct = false,
notes = CASE
WHEN pacs_dicom_series_annotations.notes = ''
THEN EXCLUDED.notes
@@ -676,8 +720,7 @@ def scan_study(ct_number: str) -> dict[str, Any]:
END,
updated_by = 'system',
updated_at = now()
WHERE jsonb_array_length(pacs_dicom_series_annotations.body_parts) = 0
AND pacs_dicom_series_annotations.plain_ct IS NOT TRUE
WHERE pacs_dicom_series_annotations.updated_by = 'system'
""",
timeout=8,
)
@@ -991,14 +1034,14 @@ def save_annotation_payload(
{sql_literal(series_uid)},
{sql_literal(series_row.get('description', ''))},
{sql_literal(json.dumps(body_parts, ensure_ascii=False))}::jsonb,
{sql_literal(json.dumps([] if skipped else manual_parts, ensure_ascii=False))}::jsonb,
{sql_literal(json.dumps([] if skipped else ai_parts, ensure_ascii=False))}::jsonb,
{sql_literal(json.dumps(manual_parts, ensure_ascii=False))}::jsonb,
{sql_literal(json.dumps(ai_parts, ensure_ascii=False))}::jsonb,
{sql_literal(phase)},
{sql_literal('' if skipped else manual_phase)},
{sql_literal('' if skipped else ai_phase)},
{sql_literal(manual_phase)},
{sql_literal(ai_phase)},
{'true' if plain_ct else 'false'},
{sql_bool_or_null(None if skipped else manual_plain_ct)},
{sql_bool_or_null(None if skipped else ai_plain_ct)},
{sql_bool_or_null(manual_plain_ct)},
{sql_bool_or_null(ai_plain_ct)},
{'true' if skipped else 'false'},
{'true' if ai_skipped else 'false'},
{sql_literal(notes)},
@@ -1031,14 +1074,14 @@ def save_annotation_payload(
return normalize_annotation(
{
"body_parts": body_parts,
"manual_body_parts": [] if skipped else manual_parts,
"ai_body_parts": [] if skipped else ai_parts,
"manual_body_parts": manual_parts,
"ai_body_parts": ai_parts,
"upper_abdomen_phase": phase,
"manual_upper_abdomen_phase": "" if skipped else manual_phase,
"ai_upper_abdomen_phase": "" if skipped else ai_phase,
"manual_upper_abdomen_phase": manual_phase,
"ai_upper_abdomen_phase": ai_phase,
"plain_ct": plain_ct,
"manual_plain_ct": None if skipped else manual_plain_ct,
"ai_plain_ct": None if skipped else ai_plain_ct,
"manual_plain_ct": manual_plain_ct,
"ai_plain_ct": ai_plain_ct,
"skipped": skipped,
"ai_skipped": ai_skipped,
"notes": notes,