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

Add PACS image mount cache refresh settings

Browse Source
This commit is contained in:
Codex
2026-05-29 03:18:50 +08:00
parent dbf52d147a
commit 54bae18a21
9 changed files with 694 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

385
DICOM_and_UPP配准/app.py
View File

@@ -2,12 +2,15 @@
from __future__ import annotations
import base64
import datetime as dt
import io
import json
import os
import re
import secrets
import shutil
import subprocess
import threading
import time
from collections import Counter, defaultdict
from pathlib import Path
@@ -54,7 +57,12 @@ WEB_USER = os.getenv("REGISTRATION_WEB_USER", "admin")
WEB_PASSWORD = os.getenv("REGISTRATION_WEB_PASSWORD", "123456")
PACS_VIEWER_URL = os.getenv("PACS_VIEWER_URL", "http://127.0.0.1:8107")
RELATION_VIEWER_URL = os.getenv("RELATION_VIEWER_URL", "http://127.0.0.1:8108")
PROCESSED_ROOT = Path(os.getenv("PACS_PROCESSED_ROOT", "/home/wkmgc/Desktop/Data_Disk_1/PACS数据/DICOM数据/已处理_DICOM数据"))
PACS_IMAGE_DB_ROOT = Path(os.getenv("PACS_IMAGE_DB_ROOT", "/home/wkmgc/Desktop/PACS影像数据库_192.168.3.3"))
PROCESSED_ROOT = Path(os.getenv("PACS_PROCESSED_ROOT", str(PACS_IMAGE_DB_ROOT / "PACS数据" / "DICOM数据" / "已处理_DICOM数据")))
REGISTRATION_CACHE_ROOT = Path(os.getenv("REGISTRATION_CACHE_ROOT", str(PACS_IMAGE_DB_ROOT / "PACS数据" / "DICOM与UPP配准数据")))
RUNTIME_DIR = Path(os.getenv("REGISTRATION_RUNTIME_DIR", str(APP_DIR / "runtime")))
SETTINGS_FILE = RUNTIME_DIR / "settings.json"
CACHE_STATUS_FILE = RUNTIME_DIR / "cache_refresh_status.json"
IDENT_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$")
WINDOWS = {
@@ -116,6 +124,30 @@ app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static")
TOKENS: dict[str, str] = {}
STUDY_CACHE: dict[str, dict[str, Any]] = {}
STACK_CACHE: dict[str, tuple[float, dict[str, Any]]] = {}
CACHE_REFRESH_LOCK = threading.Lock()
CACHE_REFRESH_THREAD: threading.Thread | None = None
SCHEDULER_THREAD: threading.Thread | None = None
DEFAULT_REFRESH_SETTINGS = {
"auto_refresh_enabled": True,
"refresh_time": "03:00",
"cache_root": str(REGISTRATION_CACHE_ROOT),
}
DEFAULT_CACHE_STATUS: dict[str, Any] = {
"running": False,
"stage": "idle",
"message": "尚未刷新",
"progress": 0,
"total_cases": 0,
"processed_cases": 0,
"copied_files": 0,
"skipped_files": 0,
"missing_files": 0,
"errors": [],
"last_started_at": "",
"last_finished_at": "",
"last_scheduled_date": "",
}
CACHE_REFRESH_STATUS: dict[str, Any] = {**DEFAULT_CACHE_STATUS}
class LoginPayload(BaseModel):
@@ -137,6 +169,11 @@ class RegistrationPayload(BaseModel):
notes: str = ""
class RefreshSettingsPayload(BaseModel):
auto_refresh_enabled: bool = True
refresh_time: str = "03:00"
def pg_env() -> dict[str, str]:
env = os.environ.copy()
env.update({"PGHOST": PGHOST, "PGPORT": PGPORT, "PGUSER": PGUSER, "PGDATABASE": PGDATABASE})
@@ -181,6 +218,134 @@ def normalize_ct(value: Any) -> str:
return re.sub(r"\s+", "", str(value or "")).upper()
def now_text() -> str:
return dt.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
def ensure_runtime_dir() -> None:
RUNTIME_DIR.mkdir(parents=True, exist_ok=True)
def valid_refresh_time(value: str) -> str:
value = str(value or "").strip()
if re.fullmatch(r"(?:[01]\d|2[0-3]):[0-5]\d", value):
return value
return "03:00"
def read_refresh_settings() -> dict[str, Any]:
settings = {**DEFAULT_REFRESH_SETTINGS}
try:
if SETTINGS_FILE.exists():
loaded = json.loads(SETTINGS_FILE.read_text(encoding="utf-8"))
if isinstance(loaded, dict):
settings.update(loaded)
except Exception:
pass
settings["auto_refresh_enabled"] = bool(settings.get("auto_refresh_enabled", True))
settings["refresh_time"] = valid_refresh_time(str(settings.get("refresh_time") or "03:00"))
settings["cache_root"] = str(REGISTRATION_CACHE_ROOT)
return settings
def write_refresh_settings(settings: dict[str, Any]) -> dict[str, Any]:
payload = {
"auto_refresh_enabled": bool(settings.get("auto_refresh_enabled", True)),
"refresh_time": valid_refresh_time(str(settings.get("refresh_time") or "03:00")),
"cache_root": str(REGISTRATION_CACHE_ROOT),
}
ensure_runtime_dir()
SETTINGS_FILE.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
return payload
def read_cache_status_file() -> dict[str, Any]:
status = {**DEFAULT_CACHE_STATUS}
try:
if CACHE_STATUS_FILE.exists():
loaded = json.loads(CACHE_STATUS_FILE.read_text(encoding="utf-8"))
if isinstance(loaded, dict):
status.update(loaded)
except Exception:
pass
return status
def write_cache_status(status: dict[str, Any]) -> None:
ensure_runtime_dir()
CACHE_STATUS_FILE.write_text(json.dumps(status, ensure_ascii=False, indent=2), encoding="utf-8")
def update_cache_status(**updates: Any) -> dict[str, Any]:
global CACHE_REFRESH_STATUS
with CACHE_REFRESH_LOCK:
CACHE_REFRESH_STATUS.update(updates)
status = {**CACHE_REFRESH_STATUS}
try:
write_cache_status(status)
except Exception:
pass
return status
def get_cache_status_snapshot() -> dict[str, Any]:
with CACHE_REFRESH_LOCK:
status = {**CACHE_REFRESH_STATUS}
settings = read_refresh_settings()
try:
refresh_hour, refresh_minute = [int(part) for part in settings["refresh_time"].split(":", 1)]
candidate = dt.datetime.now().replace(hour=refresh_hour, minute=refresh_minute, second=0, microsecond=0)
if candidate <= dt.datetime.now():
candidate += dt.timedelta(days=1)
status["next_scheduled_at"] = candidate.strftime("%Y-%m-%d %H:%M:%S") if settings["auto_refresh_enabled"] else ""
except Exception:
status["next_scheduled_at"] = ""
status["cache_root"] = str(REGISTRATION_CACHE_ROOT)
return status
def safe_path_token(value: Any) -> str:
text_value = re.sub(r"\s+", "_", str(value or "unknown").strip())
text_value = re.sub(r'[\\\\/:*?"<>|]+', "_", text_value)
return text_value[:120] or "unknown"
def mapped_path(value: Any) -> Path:
raw = str(value or "").strip()
if not raw:
return Path("")
path = Path(raw)
if path.exists():
return path
mappings = [
(
"/home/wkmgc/Desktop/PACS影像数据库_192.168.3.3",
PACS_IMAGE_DB_ROOT,
),
(
"/home/wkmgc/Desktop/PACS数据处理/PACS_DICOM处理/已处理_DICOM数据",
PACS_IMAGE_DB_ROOT / "PACS数据" / "DICOM数据" / "已处理_DICOM数据",
),
(
"/home/wkmgc/Desktop/Data_Disk_1/PACS数据/DICOM数据/已处理_DICOM数据",
PACS_IMAGE_DB_ROOT / "PACS数据" / "DICOM数据" / "已处理_DICOM数据",
),
(
"/home/wkmgc/Desktop/PACS数据处理/UPP_STL处理/已处理STL数据",
PACS_IMAGE_DB_ROOT / "PACS数据" / "重建STL数据" / "已处理STL数据",
),
(
"/home/wkmgc/Desktop/Data_Disk_1/PACS数据/重建STL数据/已处理STL数据",
PACS_IMAGE_DB_ROOT / "PACS数据" / "重建STL数据" / "已处理STL数据",
),
]
for source, target in mappings:
if raw == source or raw.startswith(source + "/"):
relative = raw.removeprefix(source).lstrip("/")
return target / relative
return path
def parse_json_list(value: Any) -> list[Any]:
if isinstance(value, list):
return value
@@ -328,10 +493,17 @@ def require_auth(authorization: str | None = Header(default=None), access_token:
@app.on_event("startup")
def startup() -> None:
global CACHE_REFRESH_STATUS
CACHE_REFRESH_STATUS = read_cache_status_file()
if CACHE_REFRESH_STATUS.get("running"):
CACHE_REFRESH_STATUS.update({"running": False, "stage": "interrupted", "message": "上次刷新未正常结束，可手动重新刷新"})
write_refresh_settings(read_refresh_settings())
write_cache_status(CACHE_REFRESH_STATUS)
try:
ensure_registration_table()
except Exception:
pass
start_cache_scheduler()
@app.get("/")
@@ -394,6 +566,28 @@ def status(_: str = Depends(require_auth)) -> dict[str, Any]:
}
@app.get("/api/settings")
def settings(_: str = Depends(require_auth)) -> dict[str, Any]:
return {"refresh": read_refresh_settings(), "cache": get_cache_status_snapshot()}
@app.post("/api/settings")
def update_settings(payload: RefreshSettingsPayload, _: str = Depends(require_auth)) -> dict[str, Any]:
settings = write_refresh_settings(payload.dict())
return {"ok": True, "refresh": settings, "cache": get_cache_status_snapshot()}
@app.get("/api/cache/status")
def cache_status(_: str = Depends(require_auth)) -> dict[str, Any]:
return get_cache_status_snapshot()
@app.post("/api/cache/refresh")
def manual_cache_refresh(user: str = Depends(require_auth)) -> dict[str, Any]:
started = start_background_cache_refresh(reason="manual", user=user)
return {"started": started, "cache": get_cache_status_snapshot()}
def annotation_labels_sql() -> str:
return f"""
SELECT
@@ -573,7 +767,7 @@ def get_study_record(ct_number: str) -> dict[str, Any]:
def resolve_study_root(study: dict[str, Any]) -> Path:
root = Path(study.get("processed_path") or "")
root = mapped_path(study.get("processed_path") or "")
if root.exists():
return root
target_folder = str(study.get("target_folder_name") or "")
@@ -879,7 +1073,7 @@ def stl_rows_for_ct(ct_number: str, algorithm_model: str = "") -> list[dict[str,
for asset_index, row in enumerate(rows):
files = parse_json_list(row.get("files"))
if not files and row.get("processed_stl_dir"):
base = Path(str(row["processed_stl_dir"]))
base = mapped_path(row["processed_stl_dir"])
if base.exists():
files = [
{
@@ -896,17 +1090,18 @@ def stl_rows_for_ct(ct_number: str, algorithm_model: str = "") -> list[dict[str,
path = file_info.get("processed_file_path") or file_info.get("source_file_path")
if not path:
continue
resolved_path = mapped_path(path)
output.append(
{
"id": index,
"asset_id": asset_index,
"algorithm_model": row.get("algorithm_model") or "未指定模型",
"file_name": file_info.get("file_name") or Path(path).name,
"segment_name": file_info.get("segment_name") or Path(path).stem,
"family": file_info.get("family") or file_info.get("segment_name") or Path(path).stem,
"file_name": file_info.get("file_name") or resolved_path.name,
"segment_name": file_info.get("segment_name") or resolved_path.stem,
"family": file_info.get("family") or file_info.get("segment_name") or resolved_path.stem,
"category": file_info.get("category") or "未分类",
"size_bytes": int(file_info.get("size_bytes") or 0),
"path": path,
"path": str(resolved_path),
}
)
index += 1
@@ -936,6 +1131,182 @@ def stl_file(ct_number: str, file_id: int, algorithm_model: str = "", _: str = D
return FileResponse(path, media_type="model/stl", filename=path.name)
def copy_if_changed(source: Path, target: Path) -> bool:
if not source.exists() or not source.is_file():
raise FileNotFoundError(str(source))
target.parent.mkdir(parents=True, exist_ok=True)
if target.exists() and target.stat().st_size == source.stat().st_size:
return False
tmp = target.with_suffix(target.suffix + ".tmp")
shutil.copy2(source, tmp)
tmp.replace(target)
return True
def cache_one_case(row: dict[str, Any], cache_root: Path) -> dict[str, int]:
ct_number = normalize_ct(row.get("ct_number") or row.get("ct_key"))
algorithm_model = row.get("algorithm_model") or "未指定模型"
case_dir = cache_root / safe_path_token(ct_number) / safe_path_token(algorithm_model)
stl_dir = case_dir / "stl"
case_dir.mkdir(parents=True, exist_ok=True)
stl_dir.mkdir(parents=True, exist_ok=True)
(case_dir / "case.json").write_text(json.dumps(row, ensure_ascii=False, indent=2, default=str), encoding="utf-8")
counters = {"copied_files": 0, "skipped_files": 0, "missing_files": 0}
# DICOM pixels and per-slice headers stay on-demand; refresh caches only fast index data and STL materials.
series_payload: dict[str, Any] = {
"ct_number": ct_number,
"series_count": row.get("series_count"),
"dicom_file_count": row.get("dicom_file_count"),
"processed_path": str(mapped_path(row.get("processed_path") or "")),
"cached_at": now_text(),
}
(case_dir / "series.json").write_text(json.dumps(series_payload, ensure_ascii=False, indent=2, default=str), encoding="utf-8")
manifest = []
for file_info in stl_rows_for_ct(ct_number, algorithm_model):
source = Path(str(file_info.get("path") or ""))
target_name = f"{int(file_info.get('id') or 0):03d}_{safe_path_token(source.name or file_info.get('file_name'))}"
target = stl_dir / target_name
cached = {**file_info, "source_path": str(source), "cached_path": str(target), "cached_at": now_text()}
try:
if copy_if_changed(source, target):
counters["copied_files"] += 1
cached["cache_status"] = "copied"
else:
counters["skipped_files"] += 1
cached["cache_status"] = "unchanged"
except FileNotFoundError:
counters["missing_files"] += 1
cached["cache_status"] = "missing"
except Exception as exc:
counters["missing_files"] += 1
cached["cache_status"] = f"error: {exc}"
manifest.append(cached)
(case_dir / "stl_manifest.json").write_text(json.dumps(manifest, ensure_ascii=False, indent=2, default=str), encoding="utf-8")
return counters
def relation_rows_for_cache() -> list[dict[str, Any]]:
return pg_json_rows(
f"""
SELECT *
FROM ({relation_select()}) relation
ORDER BY
registration_status = 'registered',
COALESCE(study_date, '') DESC,
COALESCE(study_time, '') DESC,
ct_key,
algorithm_model
""",
timeout=60,
)
def run_cache_refresh(reason: str = "manual", user: str = "system") -> None:
cache_root = REGISTRATION_CACHE_ROOT
update_cache_status(
running=True,
stage="starting",
message=f"{reason} 刷新启动",
progress=1,
total_cases=0,
processed_cases=0,
copied_files=0,
skipped_files=0,
missing_files=0,
errors=[],
last_started_at=now_text(),
)
try:
cache_root.mkdir(parents=True, exist_ok=True)
STUDY_CACHE.clear()
STACK_CACHE.clear()
rows = relation_rows_for_cache()
total = len(rows)
update_cache_status(stage="caching", message="正在缓存 DICOM 序列元数据与 STL 材料", total_cases=total, progress=4)
copied = skipped = missing = 0
errors: list[str] = []
for index, row in enumerate(rows, start=1):
label = f"{row.get('ct_number') or row.get('ct_key')} · {row.get('algorithm_model') or '未指定模型'}"
try:
counters = cache_one_case(row, cache_root)
copied += counters["copied_files"]
skipped += counters["skipped_files"]
missing += counters["missing_files"]
except Exception as exc:
errors.append(f"{label}: {exc}")
update_cache_status(
stage="caching",
message=f"正在缓存 {label}",
processed_cases=index,
copied_files=copied,
skipped_files=skipped,
missing_files=missing,
errors=errors[-20:],
progress=4 + int(index / max(total, 1) * 94),
)
update_cache_status(
running=False,
stage="finished",
message=f"刷新完成：{total} 个 CT/模型，复制 {copied} 个 STL，复用 {skipped} 个 STL",
progress=100,
total_cases=total,
processed_cases=total,
copied_files=copied,
skipped_files=skipped,
missing_files=missing,
errors=errors[-20:],
last_finished_at=now_text(),
)
except Exception as exc:
update_cache_status(
running=False,
stage="error",
message=f"刷新失败：{exc}",
progress=0,
errors=[str(exc)],
last_finished_at=now_text(),
)
def start_background_cache_refresh(reason: str = "manual", user: str = "system") -> bool:
global CACHE_REFRESH_THREAD
with CACHE_REFRESH_LOCK:
if CACHE_REFRESH_THREAD and CACHE_REFRESH_THREAD.is_alive():
return False
if CACHE_REFRESH_STATUS.get("running"):
return False
CACHE_REFRESH_THREAD = threading.Thread(target=run_cache_refresh, kwargs={"reason": reason, "user": user}, daemon=True)
CACHE_REFRESH_THREAD.start()
return True
def cache_scheduler_loop() -> None:
while True:
try:
settings = read_refresh_settings()
if settings.get("auto_refresh_enabled"):
refresh_time = settings.get("refresh_time") or "03:00"
today = dt.date.today().isoformat()
current = dt.datetime.now().strftime("%H:%M")
status = get_cache_status_snapshot()
if current == refresh_time and status.get("last_scheduled_date") != today:
update_cache_status(last_scheduled_date=today)
start_background_cache_refresh(reason="scheduled", user="system")
except Exception:
pass
time.sleep(30)
def start_cache_scheduler() -> None:
global SCHEDULER_THREAD
if SCHEDULER_THREAD and SCHEDULER_THREAD.is_alive():
return
SCHEDULER_THREAD = threading.Thread(target=cache_scheduler_loop, daemon=True)
SCHEDULER_THREAD.start()
@app.get("/api/registrations/{ct_number}")
def get_registration(ct_number: str, algorithm_model: str = "未指定模型", _: str = Depends(require_auth)) -> dict[str, Any]:
ensure_registration_table()