sync: develop의 검수 대기 적재 고침을 data에 올린다

backend/supabase/scripts/hosted_data_plane.py

@@ -121,11 +121,42 @@ def classify_blob(path: str, size_bytes: int) -> str:
     return "postgres_ok"
 
 
+APPLY_CANDIDATE_LABELS = frozenset(
+    {"apply_candidate_pending_geocoded", "apply_candidate_pending_review"}
+)
+
+
 def preview_hash(payload: Mapping[str, Any]) -> str:
-    canonical = json.dumps(payload, sort_keys=True, separators=(",", ":")).encode("utf-8")
+    canonical = json.dumps(payload, sort_keys=True, separators=(",", ":"), default=str).encode(
+        "utf-8"
+    )
     return hashlib.sha256(canonical).hexdigest()
 
 
+def _preview_body(preview: Mapping[str, Any]) -> dict[str, Any]:
+    return {key: value for key, value in preview.items() if key != "previewSha256"}
+
+
+def apply_payload_sha256(
+    rows: Iterable[Mapping[str, Any]],
+    hosted_youtube_ids: Iterable[str],
+) -> str:
+    """Hash only the rows that preview would insert, in a stable order."""
+    payloads = [
+        pending_insert_payload(row)
+        for row in rows
+        if classify_evaluation_row(row, hosted_youtube_ids) in APPLY_CANDIDATE_LABELS
+    ]
+    payloads.sort(
+        key=lambda item: (
+            str(item.get("youtube_link") or ""),
+            str(item.get("trace_id") or ""),
+            str(item.get("origin_name") or ""),
+        )
+    )
+    return preview_hash({"payloads": payloads})
+
+
 def build_apply_preview(
     *,
     local_restaurant_ids: Iterable[str],
@@ -136,12 +167,13 @@ def build_apply_preview(
     docker_class = classify_local_docker_restaurants(
         local_restaurant_ids, hosted_restaurant_ids
     )
+    rows_list = list(evaluation_rows)
     candidates: list[str] = []
     classes: dict[str, int] = {}
-    for row in evaluation_rows:
+    for row in rows_list:
         label = classify_evaluation_row(row, hosted_youtube_ids)
         classes[label] = classes.get(label, 0) + 1
-        if label in {"apply_candidate_pending_geocoded", "apply_candidate_pending_review"}:
+        if label in APPLY_CANDIDATE_LABELS:
             video_id = row_youtube_id(row)
             if video_id:
                 candidates.append(video_id)
@@ -154,6 +186,7 @@ def build_apply_preview(
         "evaluationClasses": classes,
         "applyCandidateVideoIds": candidates,
         "applyCandidateCount": len(candidates),
+        "applyPayloadSha256": apply_payload_sha256(rows_list, hosted_youtube_ids),
         "insertStatus": "pending",
         "overwriteApprovedForbidden": True,
     }
@@ -181,7 +214,12 @@ def assert_apply_authorized(
         if preview.get("dockerRestaurantApply"):
             _deny("forbidden_local_docker_apply")
     expected = preview.get("previewSha256")
-    if not isinstance(expected, str) or expected != presented_preview_sha256:
+    recomputed = preview_hash(_preview_body(preview))
+    if (
+        not isinstance(expected, str)
+        or expected != recomputed
+        or expected != presented_preview_sha256
+    ):
         _deny("preview_hash_mismatch")
     if environment.get(APPROVAL_ENV) != "1":
         _deny("approval_missing")
@@ -215,6 +253,9 @@ def pending_insert_payload(row: Mapping[str, Any]) -> dict[str, Any]:
         "is_missing": bool(row.get("is_missing")),
         "is_not_selected": bool(row.get("is_notSelected") or row.get("is_not_selected")),
         "youtube_meta": row.get("youtube_meta"),
+        "evaluation_results": row.get("evaluation_results")
+        if isinstance(row.get("evaluation_results"), Mapping)
+        else None,
         "source_type": row.get("source_type"),
         "review_count": 0,
     }
@@ -238,7 +279,7 @@ def _json_request(
     extra_headers: Mapping[str, str] | None = None,
 ) -> tuple[int, Any]:
     from urllib.request import Request, urlopen
-    from urllib.error import HTTPError
+    from urllib.error import HTTPError, URLError
 
     headers = {
         "apikey": key,
@@ -255,15 +296,22 @@ def _json_request(
     try:
         with urlopen(request, timeout=30) as response:
             raw = response.read()
-            parsed = json.loads(raw.decode("utf-8")) if raw else None
+            try:
+                parsed = json.loads(raw.decode("utf-8")) if raw else None
+            except json.JSONDecodeError:
+                _deny("hosted_response_invalid")
             return int(response.status), parsed
+    except HostedDataPlaneError:
+        raise
     except HTTPError as exc:
         raw = exc.read()
         try:
             parsed = json.loads(raw.decode("utf-8")) if raw else None
-        except ValueError:
+        except json.JSONDecodeError:
             parsed = None
         return int(exc.code), parsed
+    except (URLError, TimeoutError, OSError):
+        _deny("hosted_request_failed")
 
 
 def fetch_hosted_restaurant_snapshot(
@@ -307,7 +355,10 @@ def load_evaluation_rows(path: str) -> list[dict[str, Any]]:
     for line in source.read_text(encoding="utf-8").splitlines():
         if not line.strip():
             continue
-        parsed = json.loads(line)
+        try:
+            parsed = json.loads(line)
+        except json.JSONDecodeError:
+            _deny("evaluation_row_invalid")
         if isinstance(parsed, dict):
             rows.append(parsed)
     return rows
@@ -335,6 +386,14 @@ def apply_pending_candidates(
         _deny("candidate_count_mismatch")
     if preview.get("dockerRestaurantApply"):
         _deny("forbidden_local_docker_apply")
+    rows_list = list(evaluation_rows)
+    hosted_for_rows = [
+        video_id
+        for row in rows_list
+        if (video_id := row_youtube_id(row)) and video_id not in allowed
+    ]
+    if apply_payload_sha256(rows_list, hosted_for_rows) != preview.get("applyPayloadSha256"):
+        _deny("apply_payload_mismatch")
     inserted: list[str] = []
     skipped: list[str] = []
     unresolved: list[str] = []
@@ -345,10 +404,12 @@ def apply_pending_candidates(
     _APPLIED, _PRESENT, _UNRESOLVED = 3, 2, 1
     outcome: dict[str, int] = {}
     requester = fetch or _json_request
-    for row in evaluation_rows:
+    for row in rows_list:
         video_id = row_youtube_id(row)
         if video_id is None or video_id not in allowed:
             continue
+        if classify_evaluation_row(row, hosted_for_rows) not in APPLY_CANDIDATE_LABELS:
+            continue
         payload = pending_insert_payload(row)
         if payload["status"] != "pending":
             _deny("approved_status_forbidden")

backend/supabase/scripts/test_hosted_data_plane.py

@@ -4,6 +4,7 @@
 from __future__ import annotations
 
 import unittest
+from unittest.mock import patch
 
 from backend.supabase.scripts.hosted_data_plane import (
     APPROVAL_ENV,
@@ -18,6 +19,8 @@
     classify_local_docker_restaurants,
     pending_insert_payload,
     r2_public_object_url,
+    load_evaluation_rows,
+    _json_request,
 )
 
 
@@ -489,6 +492,121 @@ def test_r2_url_uses_issued_r2_dev_not_invented_dns(self) -> None:
         with self.assertRaises(HostedDataPlaneError):
             r2_public_object_url("NOPE", "x")
 
+    def test_changing_candidate_ids_without_rehash_is_rejected(self) -> None:
+        preview = build_apply_preview(
+            local_restaurant_ids=[],
+            hosted_restaurant_ids=[],
+            hosted_youtube_ids=[],
+            evaluation_rows=[
+                {
+                    "youtube_link": "https://www.youtube.com/watch?v=newvideo111",
+                    "trace_id": "trace-new",
+                    "geocoding_success": True,
+                    "lat": 1,
+                    "lng": 2,
+                    "origin_name": "신규집",
+                }
+            ],
+        )
+        preview["applyCandidateVideoIds"] = ["othervideo1"]
+        with self.assertRaises(HostedDataPlaneError) as raised:
+            assert_apply_authorized(
+                preview,
+                environment={APPROVAL_ENV: "1"},
+                presented_preview_sha256=preview["previewSha256"],
+            )
+        self.assertEqual(str(raised.exception), "preview_hash_mismatch")
+
+    def test_skip_row_sharing_a_candidate_video_is_not_posted(self) -> None:
+        candidate = {
+            "youtube_link": "https://www.youtube.com/watch?v=newvideo111",
+            "trace_id": "trace-new",
+            "geocoding_success": True,
+            "lat": 1,
+            "lng": 2,
+            "origin_name": "신규집",
+            "evaluation_results": {"visit_authenticity": {"name": "신규집", "eval_value": 1}},
+        }
+        skipped = {
+            "youtube_link": "https://www.youtube.com/watch?v=newvideo111",
+            "trace_id": "trace-skip",
+            "is_missing": True,
+            "origin_name": "빠진집",
+        }
+        preview = build_apply_preview(
+            local_restaurant_ids=[],
+            hosted_restaurant_ids=[],
+            hosted_youtube_ids=[],
+            evaluation_rows=[candidate, skipped],
+        )
+        calls: list[dict] = []
+
+        def fake_fetch(url, *, key, method="GET", payload=None, extra_headers=None):
+            calls.append(payload)
+            return 201, None
+
+        apply_pending_candidates(
+            preview=preview,
+            evaluation_rows=[candidate, skipped],
+            url=HOSTED_URL,
+            service_role_key="service-role",
+            environment={APPROVAL_ENV: "1"},
+            presented_preview_sha256=preview["previewSha256"],
+            fetch=fake_fetch,
+        )
+        self.assertEqual(len(calls), 1)
+        self.assertEqual(calls[0]["origin_name"], "신규집")
+        self.assertEqual(calls[0]["status"], "pending")
+        self.assertEqual(calls[0]["evaluation_results"]["visit_authenticity"]["eval_value"], 1)
+
+    def test_changed_row_content_is_rejected_before_post(self) -> None:
+        row = {
+            "youtube_link": "https://www.youtube.com/watch?v=newvideo111",
+            "trace_id": "trace-new",
+            "geocoding_success": True,
+            "lat": 1,
+            "lng": 2,
+            "origin_name": "신규집",
+        }
+        preview = build_apply_preview(
+            local_restaurant_ids=[],
+            hosted_restaurant_ids=[],
+            hosted_youtube_ids=[],
+            evaluation_rows=[row],
+        )
+        changed = dict(row)
+        changed["origin_name"] = "다른집"
+        with self.assertRaises(HostedDataPlaneError) as raised:
+            apply_pending_candidates(
+                preview=preview,
+                evaluation_rows=[changed],
+                url=HOSTED_URL,
+                service_role_key="service-role",
+                environment={APPROVAL_ENV: "1"},
+                presented_preview_sha256=preview["previewSha256"],
+                fetch=lambda *args, **kwargs: (201, None),
+            )
+        self.assertEqual(str(raised.exception), "apply_payload_mismatch")
+
+    def test_invalid_evaluation_json_fails_closed(self) -> None:
+        import tempfile
+        from pathlib import Path
+
+        with tempfile.TemporaryDirectory() as tmp:
+            path = Path(tmp) / "rows.jsonl"
+            path.write_text("{not json}\n", encoding="utf-8")
+            with self.assertRaises(HostedDataPlaneError) as raised:
+                load_evaluation_rows(str(path))
+        self.assertEqual(str(raised.exception), "evaluation_row_invalid")
+
+    def test_network_failure_is_a_hosted_data_plane_error(self) -> None:
+        from urllib.error import URLError
+
+        with patch("urllib.request.urlopen", side_effect=URLError("down")):
+            with self.assertRaises(HostedDataPlaneError) as raised:
+                _json_request(HOSTED_URL + "/rest/v1/restaurants", key="service-role")
+        self.assertEqual(str(raised.exception), "hosted_request_failed")
+
 
 if __name__ == "__main__":
     unittest.main()