From cd9d69dddbe78d7145c090dc876ba22d71dab62f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=EC=9C=A0=EC=B0=BD=EC=9A=B1?= Date: Fri, 12 Jun 2026 17:48:26 +0900 Subject: [PATCH] feat: knowledge entry update/deactivate/reactivate endpoints with audit events --- src/rights_filter/server/http_app.py | 9 +++ src/rights_filter/server/sqlite_store.py | 63 +++++++++++++++++++++ tests/rights_filter/server/test_http_app.py | 48 ++++++++++++++++ 3 files changed, 120 insertions(+) diff --git a/src/rights_filter/server/http_app.py b/src/rights_filter/server/http_app.py index c7616d3..9a435ef 100644 --- a/src/rights_filter/server/http_app.py +++ b/src/rights_filter/server/http_app.py @@ -152,6 +152,12 @@ def build_server( elif path.startswith("/api/knowledge/") and path.endswith("/exclude-watchlist"): entry_id = unquote(path.split("/")[3]) self._json(store.exclude_watchlist_entry(entry_id, str(body.get("reason", "")))) + elif path.startswith("/api/knowledge/") and path.endswith("/deactivate"): + entry_id = unquote(path.split("/")[3]) + self._json(store.deactivate_knowledge_entry(entry_id, str(body.get("reason", "")))) + elif path.startswith("/api/knowledge/") and path.endswith("/reactivate"): + entry_id = unquote(path.split("/")[3]) + self._json(store.reactivate_knowledge_entry(entry_id, str(body.get("reason", "")))) elif path == "/api/providers/emergency-disable": self._json(store.emergency_disable_external_providers()) else: @@ -177,6 +183,9 @@ def build_server( raise KeyError(provider_id) enabled = not current["enabled"] self._json(store.set_provider_enabled(provider_id, bool(enabled))) + elif path.startswith("/api/knowledge/"): + entry_id = unquote(path.split("/")[3]) + self._json(store.update_knowledge_entry(entry_id, body)) else: self._json({"error": "not found"}, HTTPStatus.NOT_FOUND) except KeyError: diff --git a/src/rights_filter/server/sqlite_store.py b/src/rights_filter/server/sqlite_store.py index a9b14e8..e8612bc 100644 --- a/src/rights_filter/server/sqlite_store.py +++ b/src/rights_filter/server/sqlite_store.py @@ -1073,6 +1073,69 @@ class CopyrighterStore: ) return self.bootstrap() + def update_knowledge_entry(self, entry_id: str, payload: dict[str, Any]) -> dict[str, Any]: + entry = self._get("knowledge_entries", entry_id) + updates: dict[str, Any] = {} + if "aliases" in payload: + updates["aliases"] = _text_list(payload.get("aliases")) + if "keywords" in payload: + updates["keywords"] = _text_list(payload.get("keywords")) + if "memo" in payload: + updates["memo"] = str(payload.get("memo", "")).strip() + if not updates: + raise ValueError("aliases, keywords, memo 중 수정할 값이 필요합니다") + + before = {key: entry.get(key) for key in updates} + entry.update(updates) + self._put("knowledge_entries", entry_id, entry) + self.add_audit_event( + "rights.ops", + "Knowledge entry updated", + str(entry.get("name", entry_id)), + f"{json.dumps(before, ensure_ascii=False)} -> {json.dumps(updates, ensure_ascii=False)}", + ) + return self.bootstrap() + + def deactivate_knowledge_entry(self, entry_id: str, reason: str = "") -> dict[str, Any]: + entry = self._get("knowledge_entries", entry_id) + if entry.get("entryStatus", "confirmed") != "confirmed": + raise ValueError("확정 DB 항목만 비활성화할 수 있습니다") + if not entry.get("active", False): + raise ValueError("이미 비활성 상태입니다") + entry["active"] = False + entry["deactivatedAt"] = _now_label() + entry["deactivatedBy"] = "rights.ops" + entry["deactivatedReason"] = reason.strip() + self._put("knowledge_entries", entry_id, entry) + self.add_audit_event( + "rights.ops", + "Knowledge entry deactivated", + str(entry.get("name", entry_id)), + reason.strip() or "운영자 비활성화", + ) + return self.bootstrap() + + def reactivate_knowledge_entry(self, entry_id: str, reason: str) -> dict[str, Any]: + if not reason.strip(): + raise ValueError("재활성에는 사유 메모가 필요합니다") + entry = self._get("knowledge_entries", entry_id) + if entry.get("entryStatus", "confirmed") != "confirmed": + raise ValueError("확정 DB 항목만 재활성화할 수 있습니다") + if entry.get("active", False): + raise ValueError("이미 활성 상태입니다") + entry["active"] = True + entry["reactivatedAt"] = _now_label() + entry["reactivatedBy"] = "rights.ops" + entry["reactivatedReason"] = reason.strip() + self._put("knowledge_entries", entry_id, entry) + self.add_audit_event( + "rights.ops", + "Knowledge entry reactivated", + str(entry.get("name", entry_id)), + reason.strip(), + ) + return self.bootstrap() + def _create_or_update_watchlist_entry( self, submission_id: str, diff --git a/tests/rights_filter/server/test_http_app.py b/tests/rights_filter/server/test_http_app.py index 81abd51..c39c924 100644 --- a/tests/rights_filter/server/test_http_app.py +++ b/tests/rights_filter/server/test_http_app.py @@ -1160,3 +1160,51 @@ def test_http_server_serves_bundled_font_with_mime_and_immutable_cache(tmp_path: assert font_headers["Content-Type"] == "font/woff2" assert font_headers["Cache-Control"] == "public, max-age=31536000, immutable" assert "Cache-Control" not in html_headers + + +def test_knowledge_entry_update_deactivate_reactivate_lifecycle(tmp_path: Path): + static_dir, image_store, store = _fixtures(tmp_path) + server = build_server(host="127.0.0.1", port=0, store=store, image_store=image_store, static_dir=static_dir) + _start(server) + base = f"http://127.0.0.1:{server.server_port}" + + try: + _json( + base + "/api/knowledge/manual", + method="POST", + body={"name": "윈터", "type": "public_figure", "aliases": ["에스파 윈터"], "keywords": ["aespa"], "memo": "초상권 주의"}, + ) + entry = next(item for item in store._all("knowledge_entries") if item["name"] == "윈터") + entry_id = entry["id"] + + updated = _json( + base + f"/api/knowledge/{entry_id}", + method="PATCH", + body={"aliases": "윈터, winter", "keywords": ["aespa", "윈터 화보"], "memo": "수정된 메모"}, + ) + updated_entry = next(item for item in updated["knowledgeEntries"] if item["id"] == entry_id) + assert updated_entry["aliases"] == ["윈터", "winter"] + assert updated_entry["keywords"] == ["aespa", "윈터 화보"] + assert updated_entry["memo"] == "수정된 메모" + + deactivated = _json(base + f"/api/knowledge/{entry_id}/deactivate", method="POST", body={"reason": "중복 항목"}) + deactivated_entry = next(item for item in deactivated["knowledgeEntries"] if item["id"] == entry_id) + assert deactivated_entry["active"] is False + + with pytest.raises(Exception): + _json(base + f"/api/knowledge/{entry_id}/reactivate", method="POST", body={"reason": ""}) + + reactivated = _json(base + f"/api/knowledge/{entry_id}/reactivate", method="POST", body={"reason": "검토 후 재사용"}) + reactivated_entry = next(item for item in reactivated["knowledgeEntries"] if item["id"] == entry_id) + assert reactivated_entry["active"] is True + + with pytest.raises(Exception): + _json(base + f"/api/knowledge/{entry_id}", method="PATCH", body={}) + + events = _json(base + "/api/audit-events") + event_names = [event["event"] for event in events] + assert "Knowledge entry updated" in event_names + assert "Knowledge entry deactivated" in event_names + assert "Knowledge entry reactivated" in event_names + finally: + server.shutdown()