feat(z01): 기초단가 표 수준 출처 — 출처표(data_master_sources) 를 kind 마다 목록으로(내는 곳 · 받는 자리 · 주기 · 최신 공표일 · 우리 판 · 확인한 날 · outdated) · 우리 판은 출처표 글자가 아니라 실제 파일에서 · 뒤처짐 판정은 서버만 · 판 기준일은 한 함수(effective_date · 파생본 derived_from.effective_date) · 모아 보기 항목에 단위(브레인)

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01BW6Jdsh18WPtYUR6THZJqn

Z01_MasterData/Z01_MasterData_BasePrices.py

@@ -10,6 +10,7 @@
 from __future__ import annotations
 
 import json
+from datetime import date
 from decimal import Decimal
 from functools import lru_cache
 from pathlib import Path
@@ -20,6 +21,9 @@ from Z01_MasterData import Z01_MasterData_Overrides as overrides
 from Z01_MasterData import Z01_MasterData_Tables as tables
 
 KINDS = ("labor", "machine", "material", "oil", "rate")
+SOURCES_DIR = (
+    tables.RESOURCES / "data_master_sources"
+)  # 출처표(데스크탑 서브) — 어디서 받고 최신이 무엇인지
 _KEY = "자료 열쇠 — 바꾸면 덮개·다른 표가 이 줄을 못 찾음"
 _SOURCE = "공표 원문 칸 — 원본 갱신으로만 바뀜"
 _RATE_EDITABLE = (
@@ -53,6 +57,58 @@ def _doc(file_id: str) -> tuple[dict[str, Any], str]:
     return _read(str(path), path.stat().st_mtime_ns), path.name
 
 
+def edition_of(doc: dict[str, Any]) -> str | None:
+    """판 기준일 — 자료마다 자리가 달라 **여기 한 곳**에서 뽑음(`effective_date` · 파생본은 `derived_from.effective_date`)."""
+    derived = doc.get("derived_from")
+    return doc.get("effective_date") or (
+        derived.get("effective_date") if isinstance(derived, dict) else None
+    )
+
+
+def _date(text: Any) -> date | None:
+    try:
+        return date.fromisoformat(str(text))
+    except ValueError:
+        return None
+
+
+def sources(kind: str) -> list[dict[str, Any]]:
+    """표 수준 출처 — kind 하나에 판이 둘인 곳(노임·유가·기계)이 있어 목록.
+
+    `our_edition` 은 출처표 글자가 아니라 **실제 파일**에서 읽음(글자가 굳으면 또 뒤처짐) ·
+    `outdated` 판정은 서버만(최신 공표일 > 우리 판 · 둘 중 하나라도 모르면 None).
+    """
+    found = sorted(SOURCES_DIR.glob("sources_*.json")) if SOURCES_DIR.is_dir() else []
+    if not found:
+        return []
+    listing = _read(str(found[-1]), found[-1].stat().st_mtime_ns)
+    out = []
+    for entry in listing.get("sources") or []:
+        if entry.get("kind") != kind:
+            continue
+        editions = []
+        for name in entry.get("files") or []:
+            path = (tables.ROOT / name).resolve()
+            if path.is_file() and path.is_relative_to(tables.RESOURCES):
+                editions.append(_date(edition_of(_read(str(path), path.stat().st_mtime_ns))))
+        ours = min((e for e in editions if e), default=None)
+        latest = _date(entry.get("latest_published"))
+        out.append(
+            {
+                "source_id": entry.get("source_id"),
+                "name": entry.get("name_ko") or entry.get("source_id"),
+                "publisher": entry.get("publisher") or "",
+                "where": entry.get("where_to_get") or "",
+                "cycle": entry.get("cycle") or "",
+                "latest_published": latest.isoformat() if latest else None,
+                "our_edition": ours.isoformat() if ours else None,
+                "checked_at": entry.get("checked_on") or None,
+                "outdated": None if ours is None or latest is None else latest > ours,
+            }
+        )
+    return out
+
+
 def _records(doc: dict[str, Any]) -> list[dict[str, Any]]:
     return next(
         v["records"] for v in doc["variables"].values() if isinstance(v.get("records"), list)
@@ -70,7 +126,7 @@ def _labor() -> list[dict[str, Any]]:
                 "@id": f"{dataset}/{r['occupation_code']}",
                 "@source": name,
                 "dataset_id": dataset,
-                "effective_date": doc.get("effective_date"),
+                "effective_date": edition_of(doc),
                 **r,
             }
             row.setdefault("daily_wage_krw", None)  # 미공표 직종 — 사람이 넣을 수 있음
@@ -80,7 +136,7 @@ def _labor() -> list[dict[str, Any]]:
 
 def _material() -> list[dict[str, Any]]:
     doc, name = _doc("mat_price_public")
-    date = doc.get("effective_date")
+    date = edition_of(doc)
     return [
         {"@id": str(r["item_code"]), "@source": name, "effective_date": date, **r}
         for r in _records(doc)
@@ -96,7 +152,7 @@ def _oil() -> list[dict[str, Any]]:
             common = {
                 "scope": v.get("scope"),
                 "fuel": variable,
-                "effective_date": doc.get("effective_date"),  # 전국 · 지역 판 기준일이 다름
+                "effective_date": edition_of(doc),  # 전국 · 지역 판 기준일이 다름
                 **head,
             }
             if "records" not in v:
@@ -140,7 +196,7 @@ def _rate() -> list[dict[str, Any]]:
                 "@source": name,
                 "variable": variable,
                 "part": "",
-                "effective_date": doc.get("effective_date"),
+                "effective_date": edition_of(doc),
                 **_flat(v),
             }
         )
@@ -161,7 +217,7 @@ def _rate() -> list[dict[str, Any]]:
                         "@source": name,
                         "variable": variable,
                         "part": part,
-                        "effective_date": doc.get("effective_date"),
+                        "effective_date": edition_of(doc),
                         "base": v.get("base"),
                         **item,
                     }
@@ -176,7 +232,7 @@ def base_rows(kind: str) -> list[dict[str, Any]]:
     """덮개 얹기 전 원본 줄 — 기계는 입력 칸까지(계산 칸은 덮개를 얹은 뒤 셈)."""
     if kind == "machine":
         doc, name = _doc("mach_base")
-        return machine.base_rows(name, doc.get("effective_date"))
+        return machine.base_rows(name, edition_of(doc))
     return _BUILDERS[kind]()
 
 
@@ -196,7 +252,7 @@ def _machine_inputs() -> machine.Inputs:
         wage_date=labor[0]["effective_date"] if labor else None,
         oil={r["fuel"]: Decimal(str(r["value"])) for r in oil},
         oil_dates={r["fuel"]: r.get("date") or r["effective_date"] for r in oil},
-        operating_date=_doc("pum_const")[0].get("effective_date"),
+        operating_date=edition_of(_doc("pum_const")[0]),
     )
 
 
@@ -300,6 +356,7 @@ def table(
         "rows": [public(r) for r in hits[start : start + size]],
         "total": len(hits),
         **spec(kind, columns),
+        "source": sources(kind),
     }
 
 
@@ -335,7 +392,9 @@ def override_items(
             if (kind and each != kind) or (state and item["state"] != state):
                 continue
             item["row_label"] = row_label(each, by_id.get(item["row_id"]))
-            item["column_label"] = column_meta(each, item["column"])["label"]
+            meta = column_meta(each, item["column"])
+            item["column_label"] = meta["label"]
+            item["unit"] = meta["unit"]
             out.append(item)
     out.sort(key=lambda i: overrides.STATE_ORDER.index(i["state"]))
     size = max(1, min(size, tables.MAX_PAGE_SIZE))

resources/tester/test_z01_base_prices.py

@@ -267,6 +267,7 @@ def test_원본이_바뀐_덮개는_덮개_값을_쓰고_세게_드러냄_주인
         "current",
         "value",
         "state",
+        "unit",
     }
     assert first["row_label"] == "작업반장" and (
         first["original"],
@@ -409,3 +410,66 @@ def test_줄은_있는데_칸이_사라진_덮개도_주인_없음(client: TestC
     items = client.get("/api/master-data/overrides", params={"kind": "rate"}).json()["items"]
     assert [(i["row_id"], i["state"]) for i in items] == [(fee["@id"], overrides.ORPHAN)]
     assert "@overrides" not in _row(client, "rate", fee["@id"])  # 없는 칸을 새로 만들지 않음
+
+
+SOURCE_KEYS = {
+    "source_id",
+    "name",
+    "publisher",
+    "where",
+    "cycle",
+    "latest_published",
+    "our_edition",
+    "checked_at",
+    "outdated",
+}
+
+
+def test_표마다_출처와_뒤처짐을_서버가_판정(client: TestClient) -> None:
+    """브레인 — 건설 노임이 한 판 뒤처진 것을 반 년간 아무도 몰랐음 → 표 수준에 출처 · outdated 는 서버 판정."""
+    labor = {s["source_id"]: s for s in _get(client, "labor", size=1)["source"]}
+    const = labor["labor_const"]
+    assert set(const) == SOURCE_KEYS
+    assert (const["our_edition"], const["latest_published"], const["outdated"]) == (
+        "2026-01-01",
+        "2026-09-01",
+        True,
+    )
+    assert "cak.or.kr" in const["where"] and const["publisher"] == "대한건설협회"
+    assert const["checked_at"] == "2026-09-15"
+    assert labor["labor_mfg"]["outdated"] is None  # 최신 공표일 모름
+    machine = {s["source_id"]: s for s in _get(client, "machine", size=1)["source"]}
+    assert (
+        machine["machine_operating"]["our_edition"] == "2026-01-01"
+    )  # derived_from.effective_date
+    assert machine["mach_base"]["outdated"] is False
+    assert [s["source_id"] for s in _get(client, "oil", size=1)["source"]] == [
+        "oil",
+        "oil_regional",
+    ]
+
+
+def test_우리_판은_출처표_글자가_아니라_파일에서_읽음(
+    client: TestClient, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    from Z01_MasterData import Z01_MasterData_BasePrices as base_prices
+
+    real = sorted((ROOT / "resources/data_master_sources").glob("sources_*.json"))[-1]
+    doc = json.loads(real.read_text(encoding="utf-8"))
+    for s in doc["sources"]:
+        if s["source_id"] == "rates":  # 표 글자가 굳어 틀려도
+            s.update(our_edition="1999-01-01", latest_published="2027-01-01", status="current")
+    folder = tmp_path / "sources"
+    folder.mkdir()
+    (folder / "sources_2099-01-01.json").write_text(json.dumps(doc), encoding="utf-8")
+    monkeypatch.setattr(base_prices, "SOURCES_DIR", folder)
+    rate = _get(client, "rate", size=1)["source"]
+    assert [(s["our_edition"], s["outdated"]) for s in rate] == [("2026-04-13", True)]
+    monkeypatch.setattr(base_prices, "SOURCES_DIR", tmp_path / "없음")
+    assert _get(client, "rate", size=1)["source"] == []  # 출처표가 없으면 빈 목록(모양은 그대로)
+
+
+def test_모아_보기_항목에_단위(client: TestClient) -> None:
+    _put(client, "labor", "labor_const/1001", {"daily_wage_krw": 230000})
+    (item,) = client.get("/api/master-data/overrides").json()["items"]
+    assert item["unit"] == "원" and item["column_label"] == "일 노임"