VesiscanClinicalAndroid/tools/labdb_upload.py

"""
labdb 세션 폴더 일괄 업로드 스크립트.

앱이 만든 표준 세션 폴더 (measurement_<folder>.json 포함) 를 labdb REST API
로 업로드. app/src/main/java/com/medithings/vesiscan/services/labdb/
LabdbUploader.kt 의 buildPayload 로직을 그대로 Python 으로 옮긴 것.

사용법:
    export LABDB_API_KEY=xbk_live_xxxxxxxxxxxxxxxx
    python labdb_upload.py "C:/Users/장동우/Desktop/newdata"

옵션:
    --api-key KEY    apiKey 를 인자로 직접 전달 (env 없이)
    --dry-run        전송하지 않고 payload 만 검증
    --force          이미 labdb_upload.json 있어도 재업로드
    --data-type XXX  meta.data_type 없을 때 fallback (default: 001)
"""

from __future__ import annotations

import argparse
import hashlib
import json
import os
import re
import sys
from datetime import datetime, timezone, timedelta
from pathlib import Path

import urllib.request
import urllib.error


BASE_URL = "https://labdb.medithings.net/api/v1"
DEFAULT_DATA_TYPE = "001"


def sanitize_test_id(raw: str) -> str:
    """^[A-Za-z0-9_-]{1,40}$ 강제. 40자 초과 시 앞 31자 + '_' + 8자 SHA-1."""
    cleaned = re.sub(r"[^A-Za-z0-9_-]", "_", raw)
    if len(cleaned) <= 40:
        return cleaned
    h = hashlib.sha1(cleaned.encode()).hexdigest()[:8]
    return cleaned[:31] + "_" + h


def iso_now_kst() -> str:
    kst = timezone(timedelta(hours=9))
    return datetime.now(kst).strftime("%Y-%m-%dT%H:%M:%S.") + \
        f"{datetime.now(kst).microsecond // 1000:03d}+09:00"


def iso_from_ms(ms: int) -> str:
    kst = timezone(timedelta(hours=9))
    dt = datetime.fromtimestamp(ms / 1000, tz=kst)
    return dt.strftime("%Y-%m-%dT%H:%M:%S.") + f"{dt.microsecond // 1000:03d}+09:00"


def cycle_to_record(cycle: dict, meta: dict) -> dict:
    row_index = cycle.get("cycle", 0)
    ts_ms = cycle.get("timestamp_ms", 0)
    piezo = cycle.get("piezo") or {}
    imu_arr = cycle.get("imu") or []

    latest = imu_arr[-1] if imu_arr else None
    sensor = {}
    if latest:
        sensor["imu"] = {k: latest.get(k, 0.0) for k in ("ax", "ay", "az", "gx", "gy", "gz")}
    sensor["imu_samples"] = imu_arr
    sensor["imu_sample_count"] = len(imu_arr)

    channels = []
    for ch in range(6):
        data = piezo.get(f"CH{ch}")
        if data is None:
            continue
        peak = 0
        peak_idx = 0
        for i, v in enumerate(data):
            if v > peak:
                peak = v
                peak_idx = i
        channels.append({"ch": ch, "peak": peak, "peakIdx": peak_idx, "data": data})

    record = {
        "rowIndex": row_index,
        "datetime": iso_from_ms(ts_ms),
        "commandType": "MTB",
        "sensor": sensor,
        "channels": channels,
    }
    for k in ("align_phase", "align_score", "align_hint", "align_icon",
              "align_detected", "align_urine_lens"):
        if k in cycle:
            record[k] = cycle[k]
    return record


def build_payload(folder_name: str, measurement: dict, default_data_type: str) -> dict:
    meta = measurement.get("meta") or {}
    cycles = measurement.get("cycles") or []
    test_id = sanitize_test_id(folder_name)
    data_type = (meta.get("data_type") or "").strip() or default_data_type

    params = {
        "posture": meta.get("posture", ""),
        "step": meta.get("step", ""),
        "is_alignment": meta.get("is_alignment", False),
        "examiner": meta.get("examiner", ""),
        "app_version": meta.get("app_version_name", ""),
        "captured_cycles": len(cycles),
    }
    for k in ("subject_type", "subject", "firmware_version", "hardware_version",
              "serial_number"):
        v = meta.get(k)
        if isinstance(v, str) and v.strip():
            params[k] = v
    for k in ("true_volume_ml", "abdomen_thickness_mm"):
        if k in meta:
            params[k] = meta[k]

    payload = {
        "testId": test_id,
        "dataType": data_type,
        "sessionName": folder_name,
        "memo": meta.get("notes", ""),
        "params": params,
        "recordCount": len(cycles),
        "records": [cycle_to_record(c, meta) for c in cycles],
    }
    ended_at = meta.get("ended_at")
    if isinstance(ended_at, int) and ended_at > 0:
        payload["savedAt"] = iso_from_ms(ended_at)
    return payload


def post_upload(payload: dict, api_key: str) -> dict:
    url = f"{BASE_URL}/upload/json"
    data = json.dumps(payload, ensure_ascii=False).encode("utf-8")
    req = urllib.request.Request(
        url,
        data=data,
        method="POST",
        headers={
            "Content-Type": "application/json; charset=utf-8",
            "X-API-Key": api_key,
        },
    )
    try:
        with urllib.request.urlopen(req, timeout=60) as resp:
            body = resp.read().decode("utf-8")
            return json.loads(body)
    except urllib.error.HTTPError as e:
        body = e.read().decode("utf-8", errors="replace")
        try:
            err = json.loads(body)
        except Exception:
            err = {"raw": body}
        raise RuntimeError(f"HTTP {e.code}: {err}") from e


def find_measurement_json(folder: Path) -> Path | None:
    named = folder / f"measurement_{folder.name}.json"
    if named.exists():
        return named
    legacy = folder / "measurement.json"
    if legacy.exists():
        return legacy
    for p in folder.glob("measurement_*.json"):
        return p
    return None


def upload_folder(folder: Path, api_key: str, default_data_type: str,
                  dry_run: bool, force: bool) -> tuple[bool, str]:
    marker = folder / "labdb_upload.json"
    if marker.exists() and not force:
        return True, "already uploaded (skip)"

    mfile = find_measurement_json(folder)
    if mfile is None:
        return False, "measurement json not found"

    try:
        measurement = json.loads(mfile.read_text(encoding="utf-8"))
    except Exception as e:
        return False, f"read/parse fail: {e}"

    payload = build_payload(folder.name, measurement, default_data_type)

    if dry_run:
        return True, (f"payload OK: testId={payload['testId']} "
                      f"dataType={payload['dataType']} records={payload['recordCount']}")

    try:
        result = post_upload(payload, api_key)
    except Exception as e:
        err_body = {
            "error": type(e).__name__,
            "message": str(e),
            "failedAt": iso_now_kst(),
        }
        (folder / "labdb_upload_error.json").write_text(
            json.dumps(err_body, ensure_ascii=False, indent=2), encoding="utf-8")
        return False, f"upload fail: {e}"

    marker_body = {
        "ok": result.get("ok", True),
        "sessionId": result.get("sessionId", ""),
        "created": result.get("created", False),
        "totalRecords": result.get("totalRecords", payload["recordCount"]),
        "inserted": result.get("inserted", 0),
        "duplicates": result.get("duplicates", 0),
        "errors": result.get("errors", []),
        "uploadedAt": iso_now_kst(),
    }
    marker.write_text(json.dumps(marker_body, ensure_ascii=False, indent=2), encoding="utf-8")
    (folder / "labdb_upload_error.json").unlink(missing_ok=True)
    return True, (f"uploaded: {marker_body['inserted']}/{marker_body['totalRecords']} "
                  f"(dup={marker_body['duplicates']})")


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("root", help="세션 폴더들이 있는 최상위 디렉토리")
    ap.add_argument("--api-key", default=os.environ.get("LABDB_API_KEY"),
                    help="labdb apiKey (env LABDB_API_KEY 도 가능)")
    ap.add_argument("--dry-run", action="store_true")
    ap.add_argument("--force", action="store_true",
                    help="labdb_upload.json 있어도 재전송")
    ap.add_argument("--data-type", default=DEFAULT_DATA_TYPE)
    args = ap.parse_args()

    root = Path(args.root)
    if not root.is_dir():
        print(f"ERROR: {root} is not a directory", file=sys.stderr)
        return 2

    if not args.dry_run and not args.api_key:
        print("ERROR: LABDB_API_KEY 미설정. env 또는 --api-key 필요", file=sys.stderr)
        return 2

    folders = sorted(p for p in root.iterdir() if p.is_dir())
    if not folders:
        print(f"no folders under {root}")
        return 0

    ok_count = 0
    skip_count = 0
    fail_count = 0
    for f in folders:
        ok, msg = upload_folder(f, args.api_key or "", args.data_type,
                                args.dry_run, args.force)
        marker = "OK  " if ok else "FAIL"
        print(f"[{marker}] {f.name}: {msg}")
        if ok:
            if "skip" in msg:
                skip_count += 1
            else:
                ok_count += 1
        else:
            fail_count += 1

    print(f"\nsummary: {ok_count} uploaded, {skip_count} skipped, {fail_count} failed "
          f"(total {len(folders)})")
    return 0 if fail_count == 0 else 1


if __name__ == "__main__":
    sys.exit(main())