From 089580341d0a70d84d84648c1733f1fad0e4e5e3 Mon Sep 17 00:00:00 2001 From: DankerMu Date: Sat, 21 Feb 2026 11:03:46 +0800 Subject: [PATCH 1/2] =?UTF-8?q?feat(qa):=20add=20rule-based=20consistency?= =?UTF-8?q?=20checker=20for=20MVP-=CE=B2=20#14?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implements 5 consistency checks: - character_status: dead characters appearing in later scenes - timeline: narrative_day conflicts with chapter order - possession: items owned by multiple characters - plot_thread: resolved threads referenced again - repetition: configurable n-gram detection API: POST /api/qa/check with project_id, ngram_n, ngram_threshold Service: run_consistency_check() with scene indexing and KG analysis Tests: 12 tests covering all check types and API endpoint Co-Authored-By: Claude Opus 4.6 --- backend/app/api/qa.py | 31 ++ backend/app/api/schemas.py | 13 + backend/app/main.py | 2 + backend/app/services/consistency.py | 427 ++++++++++++++++++++++++++++ backend/tests/test_consistency.py | 319 +++++++++++++++++++++ frontend/src/lib/types.ts | 14 + 6 files changed, 806 insertions(+) create mode 100644 backend/app/api/qa.py create mode 100644 backend/app/services/consistency.py create mode 100644 backend/tests/test_consistency.py diff --git a/backend/app/api/qa.py b/backend/app/api/qa.py new file mode 100644 index 0000000..02892d6 --- /dev/null +++ b/backend/app/api/qa.py @@ -0,0 +1,31 @@ +"""Quality Assurance API endpoints.""" + +from fastapi import APIRouter, Depends +from pydantic import BaseModel +from sqlalchemy.ext.asyncio import AsyncSession + +from app.api.schemas import ConsistencyResult +from app.core.database import get_db +from app.services.consistency import run_consistency_check + +router = APIRouter(prefix="/api", tags=["qa"]) + + +class ConsistencyCheckRequest(BaseModel): + project_id: int + ngram_n: int = 4 + ngram_threshold: int = 3 + + +@router.post("/qa/check", response_model=list[ConsistencyResult]) +async def check_consistency( + body: ConsistencyCheckRequest, + db: AsyncSession = Depends(get_db), +): + """Run rule-based consistency checks on a project.""" + return await run_consistency_check( + db, + body.project_id, + ngram_n=body.ngram_n, + ngram_threshold=body.ngram_threshold, + ) diff --git a/backend/app/api/schemas.py b/backend/app/api/schemas.py index 206f9bb..71df570 100644 --- a/backend/app/api/schemas.py +++ b/backend/app/api/schemas.py @@ -198,6 +198,19 @@ class LoreEntryOut(BaseModel): model_config = {"from_attributes": True} +# --- Consistency Check --- + +class ConsistencyResult(BaseModel): + type: str + severity: str + confidence: float + source: str + message: str + evidence: list[str] + evidence_locations: list[str] + suggest_fix: str + + # --- KG --- class KGNodeOut(BaseModel): diff --git a/backend/app/main.py b/backend/app/main.py index 951bde4..9a28eb8 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -9,6 +9,7 @@ from app.api.kg import router as kg_router from app.api.lorebook import router as lorebook_router from app.api.projects import router as projects_router +from app.api.qa import router as qa_router from app.api.summary import router as summary_router @@ -46,6 +47,7 @@ async def lifespan(app: FastAPI): app.include_router(summary_router) app.include_router(export_router) app.include_router(kg_router) +app.include_router(qa_router) @app.get("/health") diff --git a/backend/app/services/consistency.py b/backend/app/services/consistency.py new file mode 100644 index 0000000..c43a82f --- /dev/null +++ b/backend/app/services/consistency.py @@ -0,0 +1,427 @@ +"""Rule-based consistency checker for novel projects.""" + +import json +from collections import defaultdict + +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from app.models.tables import Book, Chapter, KGEdge, KGNode, KGProposal, Scene, SceneTextVersion + + +def _loads(raw: str, default=None): + if default is None: + default = {} + try: + return json.loads(raw) if raw else default + except (json.JSONDecodeError, TypeError): + return default + + +def _extract_ngrams(text: str, n: int) -> list[str]: + """Extract character-level n-grams from text.""" + tokens = text.split() + if len(tokens) < n: + # fall back to character n-grams for dense Chinese text + chars = [c for c in text if c.strip()] + return ["" .join(chars[i : i + n]) for i in range(len(chars) - n + 1)] + return [" ".join(tokens[i : i + n]) for i in range(len(tokens) - n + 1)] + + +async def _build_scene_index(db: AsyncSession, project_id: int) -> list[dict]: + """Return ordered list of {chapter_sort, chapter_id, scene_id, scene_sort, text, location}.""" + # Subquery: latest version per scene + from sqlalchemy import func + + latest = ( + select( + SceneTextVersion.scene_id, + func.max(SceneTextVersion.version).label("max_ver"), + ) + .group_by(SceneTextVersion.scene_id) + .subquery() + ) + + result = await db.execute( + select( + Book.id.label("book_id"), + Book.sort_order.label("book_sort"), + Chapter.id.label("chapter_id"), + Chapter.sort_order.label("chapter_sort"), + Scene.id.label("scene_id"), + Scene.sort_order.label("scene_sort"), + SceneTextVersion.content_md, + ) + .join(Chapter, Chapter.book_id == Book.id) + .join(Scene, Scene.chapter_id == Chapter.id) + .join(latest, latest.c.scene_id == Scene.id) + .join( + SceneTextVersion, + (SceneTextVersion.scene_id == Scene.id) + & (SceneTextVersion.version == latest.c.max_ver), + ) + .where(Book.project_id == project_id) + .order_by(Book.sort_order, Chapter.sort_order, Scene.sort_order) + ) + rows = result.all() + + scenes = [] + for row in rows: + scenes.append( + { + "book_id": row.book_id, + "book_sort": row.book_sort, + "chapter_id": row.chapter_id, + "chapter_sort": row.chapter_sort, + "scene_id": row.scene_id, + "scene_sort": row.scene_sort, + "text": row.content_md or "", + "location": f"chapter:{row.chapter_id}:scene:{row.scene_id}", + } + ) + return scenes + + +# ---------- Check 1: character_status ---------- + +async def _check_character_status( + db: AsyncSession, project_id: int, scenes: list[dict] +) -> list[dict]: + """Detect dead characters appearing in later scene text.""" + result = await db.execute( + select(KGNode).where( + KGNode.project_id == project_id, + KGNode.label == "Character", + ) + ) + characters = result.scalars().all() + + dead_chars: list[tuple[str, str]] = [] # (name, death_location or "") + for char in characters: + props = _loads(char.properties_json, {}) + status = props.get("status", "") or props.get("Status", "") + if str(status).lower() == "dead": + death_loc = props.get("death_location", "") + dead_chars.append((char.name, death_loc)) + + if not dead_chars: + return [] + + conflicts = [] + for name, death_loc in dead_chars: + # Find the earliest scene index where the character "dies" + death_idx: int | None = None + if death_loc: + for i, scene in enumerate(scenes): + if death_loc in scene["location"]: + death_idx = i + break + + # Scan scenes after death for name occurrence + check_from = (death_idx + 1) if death_idx is not None else 0 + reappearances = [ + scene["location"] + for scene in scenes[check_from:] + if name in scene["text"] + ] + + if reappearances: + evidence = [] + if death_loc: + evidence.append(f"Character '{name}' marked dead at {death_loc}") + else: + evidence.append(f"Character '{name}' has status=dead in KG") + evidence.append(f"Reappears in: {reappearances[0]}") + + conflicts.append( + { + "type": "character_status", + "severity": "high", + "confidence": 1.0, + "source": "rule", + "message": f"Dead character '{name}' appears in later scene text.", + "evidence": evidence, + "evidence_locations": ([death_loc] if death_loc else []) + reappearances, + "suggest_fix": ( + f"Remove or justify all references to '{name}'" + " after their death." + ), + } + ) + return conflicts + + +# ---------- Check 2: timeline ---------- + +async def _check_timeline(db: AsyncSession, project_id: int) -> list[dict]: + """Detect events with narrative_day property ordered inconsistently with chapter sort.""" + # Gather events from KGProposals with narrative_day in data_json + result = await db.execute( + select(KGProposal).where(KGProposal.project_id == project_id) + ) + proposals = result.scalars().all() + + # Also look at KGEdge properties for narrative_day + edge_result = await db.execute( + select(KGEdge).where(KGEdge.project_id == project_id) + ) + edges = edge_result.scalars().all() + + # Map chapter_id -> sort_order + ch_result = await db.execute( + select(Chapter.id, Chapter.sort_order) + .join(Book, Book.id == Chapter.book_id) + .where(Book.project_id == project_id) + ) + chapter_sort: dict[int, int] = {row[0]: row[1] for row in ch_result.all()} + + events: list[dict] = [] # {name, narrative_day, chapter_id, chapter_sort, location} + + for p in proposals: + data = _loads(p.data_json, {}) + day = data.get("narrative_day") or data.get("properties", {}).get("narrative_day") + if day is not None: + try: + day = int(day) + except (ValueError, TypeError): + continue + ch_sort = chapter_sort.get(p.chapter_id, 0) + events.append( + { + "name": data.get("name") or data.get("label") or f"proposal:{p.id}", + "narrative_day": day, + "chapter_id": p.chapter_id, + "chapter_sort": ch_sort, + "location": p.evidence_location or f"chapter:{p.chapter_id}", + } + ) + + for e in edges: + props = _loads(e.properties_json, {}) + day = props.get("narrative_day") + if day is not None: + try: + day = int(day) + except (ValueError, TypeError): + continue + events.append( + { + "name": f"edge:{e.id}", + "narrative_day": day, + "chapter_id": None, + "chapter_sort": 0, + "location": f"edge:{e.id}", + } + ) + + conflicts = [] + # Compare every pair: if A has higher narrative_day but lower chapter_sort than B -> conflict + for i in range(len(events)): + for j in range(i + 1, len(events)): + a, b = events[i], events[j] + # a appears textually before b (lower chapter_sort) but has later narrative_day + if a["chapter_sort"] < b["chapter_sort"] and a["narrative_day"] > b["narrative_day"]: + conflicts.append( + { + "type": "timeline", + "severity": "medium", + "confidence": 1.0, + "source": "rule", + "message": ( + f"Timeline conflict: '{a['name']}'" + f" (narrative_day={a['narrative_day']})" + f" appears before '{b['name']}'" + f" (narrative_day={b['narrative_day']})" + " in chapter order." + ), + "evidence": [ + f"'{a['name']}' narrative_day={a['narrative_day']} at {a['location']}", + f"'{b['name']}' narrative_day={b['narrative_day']} at {b['location']}", + ], + "evidence_locations": [a["location"], b["location"]], + "suggest_fix": ( + f"Reorder events so narrative_day={b['narrative_day']} " + f"comes before narrative_day={a['narrative_day']}." + ), + } + ) + return conflicts + + +# ---------- Check 3: possession ---------- + +async def _check_possession(db: AsyncSession, project_id: int) -> list[dict]: + """Detect items owned by multiple characters simultaneously.""" + result = await db.execute( + select(KGEdge).where( + KGEdge.project_id == project_id, + KGEdge.relation.in_(["owns", "possesses"]), + ) + ) + edges = result.scalars().all() + + # Group by target_node_id -> list of source_node_ids + ownership: dict[int, list[int]] = defaultdict(list) + for e in edges: + ownership[e.target_node_id].append(e.source_node_id) + + # Fetch node names for conflicting entries + conflicts = [] + for target_id, owners in ownership.items(): + if len(owners) <= 1: + continue + + target_node = await db.get(KGNode, target_id) + item_name = target_node.name if target_node else str(target_id) + + owner_names = [] + for oid in owners: + node = await db.get(KGNode, oid) + owner_names.append(node.name if node else str(oid)) + + conflicts.append( + { + "type": "possession", + "severity": "medium", + "confidence": 1.0, + "source": "rule", + "message": ( + f"Item '{item_name}' is simultaneously owned by multiple characters: " + + ", ".join(f"'{n}'" for n in owner_names) + + "." + ), + "evidence": [ + f"'{n}' owns '{item_name}'" for n in owner_names + ], + "evidence_locations": [f"kg_node:{target_id}"], + "suggest_fix": ( + f"Clarify which character currently owns '{item_name}', " + "or add a transfer-of-ownership event." + ), + } + ) + return conflicts + + +# ---------- Check 4: plot_thread ---------- + +async def _check_plot_thread(db: AsyncSession, project_id: int, scenes: list[dict]) -> list[dict]: + """Detect resolved plot threads referenced as active in later scenes.""" + result = await db.execute( + select(KGNode).where( + KGNode.project_id == project_id, + KGNode.label.in_(["Event", "PlotThread", "Plot"]), + ) + ) + nodes = result.scalars().all() + + conflicts = [] + for node in nodes: + props = _loads(node.properties_json, {}) + status = str(props.get("status", "") or props.get("Status", "")).lower() + if status != "resolved": + continue + + resolved_loc = props.get("resolved_location", "") + resolved_idx: int | None = None + if resolved_loc: + for i, scene in enumerate(scenes): + if resolved_loc in scene["location"]: + resolved_idx = i + break + + check_from = (resolved_idx + 1) if resolved_idx is not None else 0 + reappearances = [ + scene["location"] + for scene in scenes[check_from:] + if node.name in scene["text"] + ] + + if reappearances: + evidence = [f"Plot thread '{node.name}' marked resolved"] + if resolved_loc: + evidence.append(f"Resolved at {resolved_loc}") + evidence.append(f"Referenced again at {reappearances[0]}") + + conflicts.append( + { + "type": "plot_thread", + "severity": "medium", + "confidence": 1.0, + "source": "rule", + "message": ( + f"Resolved plot thread '{node.name}' is referenced again in later scenes." + ), + "evidence": evidence, + "evidence_locations": ([resolved_loc] if resolved_loc else []) + reappearances, + "suggest_fix": ( + f"Remove or update references to '{node.name}' after it was resolved." + ), + } + ) + return conflicts + + +# ---------- Check 5: repetition ---------- + +def _check_repetition_in_scene( + scene: dict, ngram_n: int, ngram_threshold: int +) -> list[dict]: + """Return conflicts for repeated n-grams within a single scene.""" + text = scene["text"] + if not text.strip(): + return [] + + ngrams = _extract_ngrams(text, ngram_n) + counts: dict[str, int] = defaultdict(int) + for ng in ngrams: + counts[ng] += 1 + + conflicts = [] + for ng, count in counts.items(): + if count >= ngram_threshold: + conflicts.append( + { + "type": "repetition", + "severity": "low", + "confidence": 1.0, + "source": "rule", + "message": ( + f"{ngram_n}-gram '{ng}' appears {count} times " + f"(threshold={ngram_threshold}) in scene." + ), + "evidence": [ + f"Repeated phrase: '{ng}' ({count}x)", + ], + "evidence_locations": [scene["location"]], + "suggest_fix": ( + f"Rephrase repeated text to avoid repetitive use of '{ng}'." + ), + } + ) + return conflicts + + +# ---------- Main entry point ---------- + +async def run_consistency_check( + db: AsyncSession, + project_id: int, + ngram_n: int = 4, + ngram_threshold: int = 3, +) -> list[dict]: + """Run all consistency checks and return a flat list of conflict dicts.""" + scenes = await _build_scene_index(db, project_id) + + results: list[dict] = [] + results.extend(await _check_character_status(db, project_id, scenes)) + results.extend(await _check_timeline(db, project_id)) + results.extend(await _check_possession(db, project_id)) + results.extend(await _check_plot_thread(db, project_id, scenes)) + + for scene in scenes: + results.extend( + _check_repetition_in_scene(scene, ngram_n, ngram_threshold) + ) + + return results diff --git a/backend/tests/test_consistency.py b/backend/tests/test_consistency.py new file mode 100644 index 0000000..b54f435 --- /dev/null +++ b/backend/tests/test_consistency.py @@ -0,0 +1,319 @@ +"""Tests for consistency check service and API.""" + +import json + +import pytest + +from app.models.tables import KGEdge, KGNode +from app.services.consistency import run_consistency_check + + +# ---------- Helpers ---------- + +async def _setup_project(client) -> int: + resp = await client.post("/api/projects", json={"title": "Consistency Test"}) + return resp.json()["id"] + + +async def _setup_book(client, project_id: int) -> int: + resp = await client.post( + "/api/books", json={"project_id": project_id, "title": "Book 1"} + ) + return resp.json()["id"] + + +async def _setup_chapter(client, book_id: int, sort_order: int = 0) -> int: + resp = await client.post( + "/api/chapters", + json={"book_id": book_id, "title": f"Chapter {sort_order}", "sort_order": sort_order}, + ) + return resp.json()["id"] + + +async def _setup_scene_with_text(client, chapter_id: int, text: str) -> int: + resp = await client.post( + "/api/scenes", json={"chapter_id": chapter_id, "title": "Scene"} + ) + sid = resp.json()["id"] + await client.post( + f"/api/scenes/{sid}/versions", + json={"content_md": text, "created_by": "user"}, + ) + return sid + + +def _make_node( + project_id: int, + label: str, + name: str, + properties: dict, +) -> KGNode: + return KGNode( + project_id=project_id, + label=label, + name=name, + properties_json=json.dumps(properties, ensure_ascii=False), + ) + + +def _make_edge( + project_id: int, + source_node_id: int, + target_node_id: int, + relation: str, + properties: dict | None = None, +) -> KGEdge: + return KGEdge( + project_id=project_id, + source_node_id=source_node_id, + target_node_id=target_node_id, + relation=relation, + properties_json=json.dumps(properties or {}, ensure_ascii=False), + ) + + +# ---------- Test: character_status ---------- + +@pytest.mark.asyncio +async def test_character_status_conflict(client, db_session): + """Dead character appearing in later scene text is flagged.""" + pid = await _setup_project(client) + bid = await _setup_book(client, pid) + + ch1 = await _setup_chapter(client, bid, sort_order=0) + await _setup_scene_with_text(client, ch1, "The hero Lin Yuan died in battle.") + + ch2 = await _setup_chapter(client, bid, sort_order=1) + await _setup_scene_with_text(client, ch2, "Lin Yuan arrived at the castle.") + + node = _make_node(pid, "Character", "Lin Yuan", {"status": "dead"}) + db_session.add(node) + await db_session.flush() + + results = await run_consistency_check(db_session, pid) + + char_conflicts = [r for r in results if r["type"] == "character_status"] + assert len(char_conflicts) >= 1 + conflict = char_conflicts[0] + assert conflict["severity"] == "high" + assert conflict["confidence"] == 1.0 + assert conflict["source"] == "rule" + assert "Lin Yuan" in conflict["message"] + assert len(conflict["evidence"]) >= 1 + assert len(conflict["evidence_locations"]) >= 1 + assert conflict["suggest_fix"] + + +@pytest.mark.asyncio +async def test_character_status_no_conflict(client, db_session): + """Dead character not mentioned in any scene text produces no conflict.""" + pid = await _setup_project(client) + bid = await _setup_book(client, pid) + ch1 = await _setup_chapter(client, bid, sort_order=0) + # Scene text does not mention the dead character by name + await _setup_scene_with_text(client, ch1, "A brave warrior perished in battle.") + + node = _make_node(pid, "Character", "Lin Yuan", {"status": "dead"}) + db_session.add(node) + await db_session.flush() + + results = await run_consistency_check(db_session, pid) + char_conflicts = [r for r in results if r["type"] == "character_status"] + assert char_conflicts == [] + + +# ---------- Test: possession ---------- + +@pytest.mark.asyncio +async def test_possession_conflict(client, db_session): + """Same item owned by two characters is flagged.""" + pid = await _setup_project(client) + + char_a = _make_node(pid, "Character", "Alice", {}) + char_b = _make_node(pid, "Character", "Bob", {}) + item = _make_node(pid, "Item", "Magic Sword", {}) + db_session.add_all([char_a, char_b, item]) + await db_session.flush() + + edge_a = _make_edge(pid, char_a.id, item.id, "owns") + edge_b = _make_edge(pid, char_b.id, item.id, "owns") + db_session.add_all([edge_a, edge_b]) + await db_session.flush() + + results = await run_consistency_check(db_session, pid) + poss_conflicts = [r for r in results if r["type"] == "possession"] + assert len(poss_conflicts) >= 1 + c = poss_conflicts[0] + assert c["severity"] == "medium" + assert "Magic Sword" in c["message"] + assert "Alice" in c["message"] or "Bob" in c["message"] + assert c["suggest_fix"] + + +@pytest.mark.asyncio +async def test_possession_no_conflict_single_owner(client, db_session): + """Item with only one owner produces no possession conflict.""" + pid = await _setup_project(client) + + char_a = _make_node(pid, "Character", "Alice", {}) + item = _make_node(pid, "Item", "Shield", {}) + db_session.add_all([char_a, item]) + await db_session.flush() + + edge = _make_edge(pid, char_a.id, item.id, "owns") + db_session.add(edge) + await db_session.flush() + + results = await run_consistency_check(db_session, pid) + poss_conflicts = [r for r in results if r["type"] == "possession"] + assert poss_conflicts == [] + + +# ---------- Test: repetition ---------- + +@pytest.mark.asyncio +async def test_repetition_detected(client, db_session): + """Scene with a repeated n-gram is flagged.""" + pid = await _setup_project(client) + bid = await _setup_book(client, pid) + ch1 = await _setup_chapter(client, bid, sort_order=0) + # Repeat the phrase 4 times so 4-gram appears >= 3 times in char ngrams + repeated = "the quick brown fox " * 5 + await _setup_scene_with_text(client, ch1, repeated.strip()) + + results = await run_consistency_check(db_session, pid, ngram_n=4, ngram_threshold=3) + rep_conflicts = [r for r in results if r["type"] == "repetition"] + assert len(rep_conflicts) >= 1 + c = rep_conflicts[0] + assert c["severity"] == "low" + assert c["confidence"] == 1.0 + assert c["suggest_fix"] + + +@pytest.mark.asyncio +async def test_repetition_configurable_threshold(client, db_session): + """Higher threshold suppresses detection; lower threshold catches more.""" + pid = await _setup_project(client) + bid = await _setup_book(client, pid) + ch1 = await _setup_chapter(client, bid, sort_order=0) + repeated = "alpha beta gamma delta " * 4 + await _setup_scene_with_text(client, ch1, repeated.strip()) + + # High threshold: no conflicts + results_high = await run_consistency_check(db_session, pid, ngram_n=4, ngram_threshold=10) + rep_high = [r for r in results_high if r["type"] == "repetition"] + assert rep_high == [] + + # Low threshold: conflicts detected + results_low = await run_consistency_check(db_session, pid, ngram_n=4, ngram_threshold=2) + rep_low = [r for r in results_low if r["type"] == "repetition"] + assert len(rep_low) >= 1 + + +@pytest.mark.asyncio +async def test_repetition_no_false_positive(client, db_session): + """Unique text produces no repetition conflict.""" + pid = await _setup_project(client) + bid = await _setup_book(client, pid) + ch1 = await _setup_chapter(client, bid, sort_order=0) + await _setup_scene_with_text( + client, ch1, "The sun rose over the distant mountains casting long shadows." + ) + + results = await run_consistency_check(db_session, pid, ngram_n=4, ngram_threshold=3) + rep_conflicts = [r for r in results if r["type"] == "repetition"] + assert rep_conflicts == [] + + +# ---------- Test: clean data = empty results ---------- + +@pytest.mark.asyncio +async def test_clean_project_no_conflicts(client, db_session): + """A project with no KG data and clean text returns no conflicts.""" + pid = await _setup_project(client) + bid = await _setup_book(client, pid) + ch1 = await _setup_chapter(client, bid, sort_order=0) + await _setup_scene_with_text(client, ch1, "Once upon a time in a land far away.") + + results = await run_consistency_check(db_session, pid) + assert results == [] + + +# ---------- Test: API endpoint ---------- + +@pytest.mark.asyncio +async def test_api_endpoint_returns_list(client, db_session): + """POST /api/qa/check returns a list (possibly empty).""" + pid = await _setup_project(client) + + resp = await client.post("/api/qa/check", json={"project_id": pid}) + assert resp.status_code == 200 + data = resp.json() + assert isinstance(data, list) + + +@pytest.mark.asyncio +async def test_api_endpoint_conflict_format(client, db_session): + """API response items match ConsistencyResult schema fields.""" + pid = await _setup_project(client) + bid = await _setup_book(client, pid) + ch1 = await _setup_chapter(client, bid, sort_order=0) + repeated = "the quick brown fox " * 5 + await _setup_scene_with_text(client, ch1, repeated.strip()) + + resp = await client.post( + "/api/qa/check", + json={"project_id": pid, "ngram_n": 4, "ngram_threshold": 3}, + ) + assert resp.status_code == 200 + results = resp.json() + rep = [r for r in results if r["type"] == "repetition"] + assert len(rep) >= 1 + + required_fields = { + "type", "severity", "confidence", "source", + "message", "evidence", "evidence_locations", "suggest_fix", + } + for field in required_fields: + assert field in rep[0], f"Missing field: {field}" + + +@pytest.mark.asyncio +async def test_api_endpoint_custom_ngram_params(client, db_session): + """Configurable ngram_n and ngram_threshold forwarded correctly.""" + pid = await _setup_project(client) + bid = await _setup_book(client, pid) + ch1 = await _setup_chapter(client, bid, sort_order=0) + repeated = "hello world foo bar " * 4 + await _setup_scene_with_text(client, ch1, repeated.strip()) + + # Very high threshold: no repetition flagged + resp = await client.post( + "/api/qa/check", + json={"project_id": pid, "ngram_n": 4, "ngram_threshold": 100}, + ) + assert resp.status_code == 200 + assert all(r["type"] != "repetition" for r in resp.json()) + + +@pytest.mark.asyncio +async def test_api_possession_conflict_via_endpoint(client, db_session): + """Possession conflict detected through the API endpoint.""" + pid = await _setup_project(client) + + char_a = _make_node(pid, "Character", "Eve", {}) + char_b = _make_node(pid, "Character", "Mallory", {}) + item = _make_node(pid, "Item", "Golden Key", {}) + db_session.add_all([char_a, char_b, item]) + await db_session.flush() + + edge_a = _make_edge(pid, char_a.id, item.id, "possesses") + edge_b = _make_edge(pid, char_b.id, item.id, "possesses") + db_session.add_all([edge_a, edge_b]) + await db_session.flush() + + resp = await client.post("/api/qa/check", json={"project_id": pid}) + assert resp.status_code == 200 + poss = [r for r in resp.json() if r["type"] == "possession"] + assert len(poss) >= 1 + assert "Golden Key" in poss[0]["message"] diff --git a/frontend/src/lib/types.ts b/frontend/src/lib/types.ts index f832e79..313c36d 100644 --- a/frontend/src/lib/types.ts +++ b/frontend/src/lib/types.ts @@ -110,6 +110,20 @@ export interface KGProposal { created_at: string } +export type ConflictSeverity = 'high' | 'medium' | 'low' +export type ConflictType = 'character_status' | 'timeline' | 'possession' | 'plot_thread' | 'repetition' + +export interface ConsistencyResult { + type: ConflictType + severity: ConflictSeverity + confidence: number + source: string + message: string + evidence: string[] + evidence_locations: string[] + suggest_fix: string +} + export interface KGNode { id: number project_id: number From e157e3a962f3b28fbb6798657e067fee66925372 Mon Sep 17 00:00:00 2001 From: DankerMu Date: Sat, 21 Feb 2026 11:06:32 +0800 Subject: [PATCH 2/2] fix(qa): fix import formatting in test_consistency.py Remove extra blank line after imports to pass ruff check. Co-Authored-By: Claude Opus 4.6 --- backend/tests/test_consistency.py | 1 - 1 file changed, 1 deletion(-) diff --git a/backend/tests/test_consistency.py b/backend/tests/test_consistency.py index b54f435..2f97fc5 100644 --- a/backend/tests/test_consistency.py +++ b/backend/tests/test_consistency.py @@ -7,7 +7,6 @@ from app.models.tables import KGEdge, KGNode from app.services.consistency import run_consistency_check - # ---------- Helpers ---------- async def _setup_project(client) -> int: