From df05b5fdb7e230ef853f267dc7bbf556e6e09f16 Mon Sep 17 00:00:00 2001 From: bing1100 Date: Wed, 22 Jul 2026 18:29:43 +0000 Subject: [PATCH] fixes for fulltext review --- backend/api/citations/router.py | 45 ++++-- .../api/jobs/pipelines/screening_executor.py | 16 +- backend/api/screen/router.py | 138 ++++-------------- backend/api/services/cit_db_service.py | 72 +++------ .../services/screening_eligibility_service.py | 108 ++++++++++++++ .../screening_eligibility_service_test.py | 114 +++++++++++++++ 6 files changed, 308 insertions(+), 185 deletions(-) create mode 100644 backend/api/services/screening_eligibility_service.py create mode 100644 backend/tests/screening_eligibility_service_test.py diff --git a/backend/api/citations/router.py b/backend/api/citations/router.py index 19bda5eb..436f45b5 100644 --- a/backend/api/citations/router.py +++ b/backend/api/citations/router.py @@ -677,19 +677,23 @@ async def list_citation_ids( table_name = (screening or {}).get('table_name') or 'citations' - # Ensure decision columns are never stale before filtering. - # Validation strategy: UI filters by human_l1_decision / human_l2_decision. try: - cp = (sr or {}).get('criteria_parsed') or ( + from ..services.screening_eligibility_service import screening_eligibility_service + + criteria = (sr or {}).get('criteria_parsed') or ( sr or {} ).get('criteria') or {} - await run_in_threadpool(cits_dp_service.backfill_human_decisions, cp, table_name) - except Exception: - # best-effort; listing should still work even if backfill fails - pass - - try: - ids = await run_in_threadpool(cits_dp_service.list_citation_ids, filter_step, table_name) + stage = 'l1' + if str(filter_step or '').strip().lower() == 'l1': + stage = 'l2' + elif str(filter_step or '').strip().lower() == 'l2': + stage = 'extract' + ids = await run_in_threadpool( + screening_eligibility_service.list_eligible_ids, + criteria=criteria, + table_name=table_name, + target_stage=stage, + ) except RuntimeError as rexc: raise HTTPException( status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail=str(rexc), @@ -699,7 +703,6 @@ async def list_citation_ids( # treat it as "no citations" instead of poisoning the shared connection. if _is_undefined_table_error(e): return {'citation_ids': []} - except Exception as e: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Failed to query screening DB: {e}", @@ -1222,9 +1225,13 @@ async def get_citation_export_schema( try: return await run_in_threadpool(citation_export_service.build_schema, sr, table_name) except ExportValidationError as exc: - raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=str(exc)) + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, detail=str(exc), + ) except RuntimeError as exc: - raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail=str(exc)) + raise HTTPException( + status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail=str(exc), + ) @router.post('/{sr_id}/export-citations') @@ -1240,11 +1247,17 @@ async def export_citations_selective( citation_export_service.export_csv, table_name, sr, payload, ) except ExportValidationError as exc: - raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=str(exc)) + raise HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=str(exc), + ) except RuntimeError as exc: - raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail=str(exc)) + raise HTTPException( + status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail=str(exc), + ) return Response( content=csv_bytes, media_type='text/csv; charset=utf-8', - headers={'Content-Disposition': f'attachment; filename="{_safe_export_filename(sr, sr_id)}"'}, + headers={ + 'Content-Disposition': f'attachment; filename="{_safe_export_filename(sr, sr_id)}"', + }, ) diff --git a/backend/api/jobs/pipelines/screening_executor.py b/backend/api/jobs/pipelines/screening_executor.py index fed1ac2d..42bab348 100644 --- a/backend/api/jobs/pipelines/screening_executor.py +++ b/backend/api/jobs/pipelines/screening_executor.py @@ -53,15 +53,13 @@ def _eligible_ids(*, sr_id: str, table_name: str, step: str) -> list[int]: - l2/extract additionally require an uploaded PDF/fulltext (fulltext_url) """ - # Apply filter semantics - filter_step = '' - if step == 'l2': - filter_step = 'l1' - elif step == 'extract': - filter_step = 'l2' - - ids = cits_dp_service.list_citation_ids( - filter_step if filter_step else None, table_name, + from ...services.screening_eligibility_service import screening_eligibility_service + from ...services.sr_db_service import srdb_service + + sr = srdb_service.get_systematic_review(sr_id) or {} + criteria = sr.get('criteria_parsed') or sr.get('criteria') or {} + ids = screening_eligibility_service.list_eligible_ids( + criteria=criteria, table_name=table_name, target_stage=step, ) # PDF gating for l2/extract diff --git a/backend/api/screen/router.py b/backend/api/screen/router.py index 2b905ac4..3d86019a 100644 --- a/backend/api/screen/router.py +++ b/backend/api/screen/router.py @@ -28,6 +28,8 @@ from ..services.cit_db_service import cits_dp_service from ..services.cit_db_service import snake_case from ..services.cit_db_service import snake_case_column +from ..services.screening_eligibility_service import compute_screening_decisions +from ..services.screening_eligibility_service import screening_eligibility_service from ..services.sr_db_service import srdb_service from ..services.storage import storage_service from .agentic_utils import AgentResponseError @@ -277,12 +279,15 @@ async def get_filtered_citation_ids( thr = 0.9 criteria.append({'criterion_key': ck, 'label': q, 'threshold': thr}) - # Scope ids for step (L2 scope depends on human L1 include) - filter_step = '' - if step_norm == 'l2': - filter_step = 'l1' + # Resolve scope through the authoritative eligibility boundary. For L2 this + # repairs stale derived L1 decisions before filtering. try: - ids = await run_in_threadpool(cits_dp_service.list_citation_ids, filter_step, table_name) + ids = await run_in_threadpool( + screening_eligibility_service.list_eligible_ids, + criteria=cp, + table_name=table_name, + target_stage=step_norm, + ) except Exception as e: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, @@ -2207,12 +2212,11 @@ async def get_screening_metrics( thr = 0.9 criteria.append({'criterion_key': ck, 'label': q, 'threshold': thr}) - # Pull all citation ids for this step (L2 list is filtered by human_l1_decision include) - filter_step = '' - if step_norm == 'l2': - filter_step = 'l1' try: - ids = await run_in_threadpool(cits_dp_service.list_citation_ids, filter_step, table_name) + ids = await run_in_threadpool( + screening_eligibility_service.list_eligible_ids, + criteria=cp, table_name=table_name, target_stage=step_norm, + ) except Exception as e: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, @@ -2761,11 +2765,11 @@ async def get_screening_calibration( criteria.append({'criterion_key': ck, 'label': q}) # Determine SR scope ids for step (same as metrics) - filter_step = '' - if step_norm == 'l2': - filter_step = 'l1' try: - ids = await run_in_threadpool(cits_dp_service.list_citation_ids, filter_step, table_name) + ids = await run_in_threadpool( + screening_eligibility_service.list_eligible_ids, + criteria=cp, table_name=table_name, target_stage=step_norm, + ) except Exception as e: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, @@ -3063,11 +3067,11 @@ async def get_live_confidence_histogram( criteria.append({'criterion_key': ck, 'label': q}) # Scope ids for step - filter_step = '' - if step_norm == 'l2': - filter_step = 'l1' try: - ids = await run_in_threadpool(cits_dp_service.list_citation_ids, filter_step, table_name) + ids = await run_in_threadpool( + screening_eligibility_service.list_eligible_ids, + criteria=cp, table_name=table_name, target_stage=step_norm, + ) except Exception as e: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, @@ -3206,11 +3210,11 @@ async def get_calibration_samples( ) # Determine SR scope ids for step - filter_step = '' - if step_norm == 'l2': - filter_step = 'l1' try: - ids = await run_in_threadpool(cits_dp_service.list_citation_ids, filter_step, table_name) + ids = await run_in_threadpool( + screening_eligibility_service.list_eligible_ids, + criteria=cp, table_name=table_name, target_stage=step_norm, + ) except Exception as e: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, @@ -3441,95 +3445,9 @@ def _get_row_fresh() -> dict[str, Any]: try: fresh = await run_in_threadpool(_get_row_fresh) - def _compute_human_decision(step: str) -> str: - """Compute derived decisions used for filtering (pass to next stage). - - Priority rules: - 1. Human-set values have highest priority (include/exclude) - 2. If no human answer exists, fall back to AI/LLM answer - 3. If neither human nor AI answer exists, treat as exclude - - Scope: - - human_l1_decision is derived ONLY from L1 questions. - - human_l2_decision represents "passed to L2/extract" and must consider - BOTH L1 + L2 criteria questions. - """ - cp = sr.get('criteria_parsed') or {} - - # L1: only L1 questions - if step == 'l1': - qs = (cp.get('l1') or {}).get('questions') or [] - # L2: union of L1 + L2 questions - elif step == 'l2': - l1_qs = (cp.get('l1') or {}).get('questions') or [] - l2_qs = (cp.get('l2') or {}).get('questions') or [] - qs = list(l1_qs) + list(l2_qs) - else: - qs = (cp.get(step) or {}).get('questions') or [] - - if not isinstance(qs, list) or not qs: - return 'undecided' - - for q in qs: - core = snake_case(q, max_len=56) if snake_case else '' - hcol = f"human_{core}" if core else 'human_col' - llm_col = f"llm_{core}" if core else 'llm_col' - - # Priority 1: Check human answer - hval = fresh.get(hcol) - selected = None - if hval is not None: - try: - hobj = json.loads(hval) if isinstance( - hval, str, - ) else hval - selected = (hobj or {}).get('selected') - except Exception: - selected = None - - # Treat empty/whitespace as unanswered - if selected is not None and isinstance(selected, str) and selected.strip() == '': - selected = None - - # Priority 2: Fall back to AI/LLM answer if no human answer - if selected is None: - llm_val = fresh.get(llm_col) - if llm_val is not None: - try: - llm_obj = json.loads(llm_val) if isinstance( - llm_val, str, - ) else llm_val - selected = (llm_obj or {}).get('selected') - except Exception: - selected = None - # Check critical answer override if available - if selected is not None and isinstance(llm_val, (dict, str)): - try: - obj = json.loads(llm_val) if isinstance( - llm_val, str, - ) else llm_val - critical = (obj or {}).get('critical') - if isinstance(critical, dict) and critical.get('selected'): - selected = critical['selected'] - except Exception: - pass - - # Treat empty/whitespace as unanswered - if selected is not None and isinstance(selected, str) and selected.strip() == '': - selected = None - - # Priority 3: Neither exists -> exclude - if selected is None: - return 'exclude' - - if 'exclude' in str(selected).lower(): - return 'exclude' - - return 'include' - # Always set both human decisions on any update, so the list filters never go stale. - h1 = _compute_human_decision('l1') - h2 = _compute_human_decision('l2') + criteria = sr.get('criteria_parsed') or sr.get('criteria') or {} + h1, h2 = compute_screening_decisions(fresh, criteria) await run_in_threadpool(cits_dp_service.update_text_column, citation_id, 'human_l1_decision', h1, table_name) await run_in_threadpool(cits_dp_service.update_text_column, citation_id, 'human_l2_decision', h2, table_name) except Exception: diff --git a/backend/api/services/cit_db_service.py b/backend/api/services/cit_db_service.py index 638d633f..f3a6ecba 100644 --- a/backend/api/services/cit_db_service.py +++ b/backend/api/services/cit_db_service.py @@ -1422,7 +1422,9 @@ def fetch_export_rows( } requested = list(dict.fromkeys(['id', *columns])) if any(column not in existing for column in requested): - raise ValueError('Export contains a column that is not present in the citation table') + raise ValueError( + 'Export contains a column that is not present in the citation table', + ) predicates: dict[str, tuple[str, tuple[Any, ...]]] = { 'all': ('', ()), @@ -1466,6 +1468,10 @@ def backfill_human_decisions(self, criteria_parsed: dict[str, Any], table_name: table_name = _validate_ident(table_name, kind='table_name') self._require_psycopg2() + # Local import avoids a module cycle: the eligibility service owns the + # domain rule while this repository owns persistence. + from .screening_eligibility_service import compute_screening_decisions + cp = criteria_parsed or {} l1_qs = (cp.get('l1') or {}).get( 'questions', @@ -1530,49 +1536,6 @@ def _llm_col(q: str) -> str: cur.execute(f'SELECT {sql_cols} FROM "{table_name}" ORDER BY id') rows = cur.fetchall() or [] - def _parse_jsonb(v: Any) -> dict[str, Any]: - if v is None: - return {} - if isinstance(v, dict): - return v - if isinstance(v, str): - s = v.strip() - if s.startswith('{'): - try: - obj = json.loads(s) - return obj if isinstance(obj, dict) else {} - except Exception: - return {} - return {} - - def _selected_from_row(row: dict[str, Any], q: str) -> str | None: - human_col = _human_col(q) - llm_col = _llm_col(q) - - for col in [human_col, llm_col]: - if col not in existing_cols: - continue - obj = _parse_jsonb(row.get(col)) - selected = obj.get('selected') - if isinstance(selected, str): - selected = selected.strip() - if selected: - return str(selected) - return None - - def _compute(step_qs: list[str], row: dict[str, Any]) -> str: - if not step_qs: - return 'undecided' - for q in step_qs: - # Eligibility precedence: - # human answer > llm answer > no answer (exclude) - selected = _selected_from_row(row, q) - if selected is None: - return 'exclude' - if 'exclude' in str(selected).lower(): - return 'exclude' - return 'include' - updates: list[tuple[str, str, int]] = [] for r in rows: if not r: @@ -1582,8 +1545,7 @@ def _compute(step_qs: list[str], row: dict[str, Any]) -> str: rid_i = int(rid) except Exception: continue - d1 = _compute(l1_qs, r) - d2 = _compute(l2_union_qs, r) + d1, d2 = compute_screening_decisions(r, cp) updates.append((d1, d2, rid_i)) if not updates: @@ -1776,9 +1738,14 @@ def attach_fulltext_atomic( OR column_name LIKE 'human_param_%')""", (table_name,), ) - clear_columns = [str(item[0]) for item in (cur.fetchall() or [])] + clear_columns = [ + str(item[0]) + for item in (cur.fetchall() or []) + ] if clear_columns: - assignments = ', '.join(f'"{name}"=NULL' for name in clear_columns) + assignments = ', '.join( + f'"{name}"=NULL' for name in clear_columns + ) cur.execute( f'UPDATE "{table_name}" SET {assignments} WHERE id=%s', (int(citation_id),), @@ -1879,7 +1846,10 @@ def update_pdf_linkage_outcome( pdf_link_status=%s, pdf_link_reason=%s, pdf_link_source=%s, pdf_link_url=%s, pdf_link_error=%s, pdf_link_last_checked_at=now() WHERE id=%s''', - (status, reason, source, url, (error or '')[:2000] or None, int(citation_id)), + ( + status, reason, source, url, (error or '') + [:2000] or None, int(citation_id), + ), ) rows = cur.rowcount conn.commit() @@ -1989,7 +1959,9 @@ def create_table_and_insert_sync( col_defs.append('"pdf_link_reason" TEXT') col_defs.append('"pdf_link_source" TEXT') col_defs.append('"pdf_link_url" TEXT') - col_defs.append('"pdf_link_last_checked_at" TIMESTAMP WITH TIME ZONE') + col_defs.append( + '"pdf_link_last_checked_at" TIMESTAMP WITH TIME ZONE', + ) col_defs.append('"pdf_link_error" TEXT') col_defs.append('"pdf_link_doi_source" TEXT') diff --git a/backend/api/services/screening_eligibility_service.py b/backend/api/services/screening_eligibility_service.py new file mode 100644 index 00000000..9241baa4 --- /dev/null +++ b/backend/api/services/screening_eligibility_service.py @@ -0,0 +1,108 @@ +"""Authoritative screening-stage decision and eligibility rules. + +Answer JSONB columns are the source of truth. The ``human_l*_decision`` +columns are derived caches used for efficient filtering and are repaired before +they are used to determine eligibility for a later screening stage. +""" +from __future__ import annotations + +import json +from typing import Any + +from .cit_db_service import cits_dp_service +from .cit_db_service import snake_case + + +def _answer_object(value: Any) -> dict[str, Any]: + if isinstance(value, dict): + return value + if isinstance(value, str) and value.strip().startswith('{'): + try: + parsed = json.loads(value) + return parsed if isinstance(parsed, dict) else {} + except Exception: + return {} + return {} + + +def selected_answer(row: dict[str, Any], question: str) -> str | None: + """Return the effective answer using human-over-AI precedence.""" + core = snake_case(question, max_len=56) + if not core: + return None + for column in (f'human_{core}', f'llm_{core}'): + selected = _answer_object(row.get(column)).get('selected') + if isinstance(selected, str): + selected = selected.strip() + if selected is not None and selected != '': + return str(selected) + return None + + +def compute_stage_decision( + row: dict[str, Any], questions: list[str], +) -> str: + """Compute include/exclude for one stage from current criterion answers. + + Missing criteria produce ``undecided``. A missing answer is conservatively + excluded, matching the established CAN-SR progression behavior. + """ + valid_questions = [ + q for q in questions if isinstance(q, str) and q.strip() + ] + if not valid_questions: + return 'undecided' + for question in valid_questions: + selected = selected_answer(row, question) + if selected is None or 'exclude' in selected.lower(): + return 'exclude' + return 'include' + + +def compute_screening_decisions( + row: dict[str, Any], criteria: dict[str, Any] | None, +) -> tuple[str, str]: + """Return the derived L1 and cumulative L2 decisions for a citation.""" + criteria = criteria if isinstance(criteria, dict) else {} + l1 = criteria.get('l1') if isinstance(criteria.get('l1'), dict) else {} + l2 = criteria.get('l2') if isinstance(criteria.get('l2'), dict) else {} + l1_questions = l1.get('questions') if isinstance( + l1.get('questions'), list, + ) else [] + l2_questions = l2.get('questions') if isinstance( + l2.get('questions'), list, + ) else [] + return ( + compute_stage_decision(row, l1_questions), + compute_stage_decision(row, list(l1_questions) + list(l2_questions)), + ) + + +class ScreeningEligibilityService: + """Resolve stage scope only after repairing derived decision caches.""" + + def __init__(self, repository: Any = None) -> None: + self.repository = repository or cits_dp_service + + def list_eligible_ids( + self, + *, + criteria: dict[str, Any] | None, + table_name: str, + target_stage: str, + ) -> list[int]: + stage = str(target_stage or '').strip().lower() + if stage not in {'l1', 'l2', 'extract'}: + raise ValueError(f'Unsupported screening stage: {target_stage!r}') + + if stage == 'l1': + return self.repository.list_citation_ids(None, table_name) + + # Later stages depend on derived decisions. Repair first and fail loudly + # if repair fails; an empty list would be a misleading valid result. + self.repository.backfill_human_decisions(criteria or {}, table_name) + source_stage = 'l1' if stage == 'l2' else 'l2' + return self.repository.list_citation_ids(source_stage, table_name) + + +screening_eligibility_service = ScreeningEligibilityService() diff --git a/backend/tests/screening_eligibility_service_test.py b/backend/tests/screening_eligibility_service_test.py new file mode 100644 index 00000000..a54a7384 --- /dev/null +++ b/backend/tests/screening_eligibility_service_test.py @@ -0,0 +1,114 @@ +from __future__ import annotations + +import sys +import types +import unittest +from unittest.mock import Mock + +if 'psycopg' not in sys.modules: + psycopg = types.ModuleType('psycopg') + psycopg.connect = Mock() + sys.modules['psycopg'] = psycopg + +from api.services.screening_eligibility_service import ScreeningEligibilityService +from api.services.screening_eligibility_service import compute_screening_decisions + + +CRITERIA = { + 'l1': {'questions': ['Relevant population?']}, + 'l2': {'questions': ['Eligible study design?']}, +} + + +class ScreeningDecisionTests(unittest.TestCase): + def test_human_answer_overrides_ai_answer(self) -> None: + row = { + 'human_relevant_population': {'selected': 'Include'}, + 'llm_relevant_population': {'selected': 'Exclude'}, + 'human_eligible_study_design': {'selected': 'Include'}, + } + + self.assertEqual( + compute_screening_decisions(row, CRITERIA), + ('include', 'include'), + ) + + def test_ai_answer_is_used_when_human_answer_is_missing(self) -> None: + row = { + 'llm_relevant_population': '{"selected": "Include"}', + 'llm_eligible_study_design': {'selected': 'Include'}, + } + + self.assertEqual( + compute_screening_decisions(row, CRITERIA), + ('include', 'include'), + ) + + def test_l2_decision_is_cumulative_and_missing_answers_exclude(self) -> None: + row = {'human_relevant_population': {'selected': 'Include'}} + + self.assertEqual( + compute_screening_decisions(row, CRITERIA), + ('include', 'exclude'), + ) + + +class ScreeningEligibilityServiceTests(unittest.TestCase): + def test_l2_repairs_decisions_before_filtering(self) -> None: + repository = Mock() + repository.list_citation_ids.return_value = [4, 9] + service = ScreeningEligibilityService(repository) + + result = service.list_eligible_ids( + criteria=CRITERIA, + table_name='screening_table', + target_stage='l2', + ) + + self.assertEqual(result, [4, 9]) + self.assertEqual( + repository.method_calls, + [ + unittest.mock.call.backfill_human_decisions( + CRITERIA, 'screening_table', + ), + unittest.mock.call.list_citation_ids('l1', 'screening_table'), + ], + ) + + def test_l1_does_not_run_unnecessary_repair(self) -> None: + repository = Mock() + repository.list_citation_ids.return_value = [1, 2] + service = ScreeningEligibilityService(repository) + + result = service.list_eligible_ids( + criteria=CRITERIA, + table_name='screening_table', + target_stage='l1', + ) + + self.assertEqual(result, [1, 2]) + repository.backfill_human_decisions.assert_not_called() + repository.list_citation_ids.assert_called_once_with( + None, 'screening_table', + ) + + def test_repair_failure_is_not_hidden_as_an_empty_scope(self) -> None: + repository = Mock() + repository.backfill_human_decisions.side_effect = RuntimeError( + 'repair failed', + ) + service = ScreeningEligibilityService(repository) + + with self.assertRaisesRegex(RuntimeError, 'repair failed'): + service.list_eligible_ids( + criteria=CRITERIA, + table_name='screening_table', + target_stage='l2', + ) + + repository.list_citation_ids.assert_not_called() + + +if __name__ == '__main__': + unittest.main()