feat: knowledge entry update/deactivate/reactivate endpoints with audit events

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:

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,

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()