""" Use-case functions for Criteria Workspace operations. Ported from AI_Screening_UI criteria_workspace/use_cases/*.py. Each use case is a standalone async function accepting a context dict. No Panel widgets or UI dependencies. """ from __future__ import annotations import logging from .ai_service import CriteriaAIService, DEFAULT_CRITERIA_MODEL from .models import ExclusionCriterion logger = logging.getLogger(__name__) async def execute_generate(context: dict) -> list[ExclusionCriterion]: """ Generate inclusion/exclusion criteria from project context. Context keys: project_id (int): Project ID project_description (str): Project description text research_questions (list[str]): Research questions additional_notes (str, optional): Extra context for generation existing_criteria (list[ExclusionCriterion], optional): For deduplication pico_extraction (dict, optional): Cached PICO extraction to include output_type (str): "Exclusion", "Inclusion", or "Both" (default: "Exclusion") model (str, optional): OpenAI model name api_key (str, optional): OpenAI API key """ project_id = context.get("project_id", 0) desc = context.get("project_description") rqs = context.get("research_questions") notes = context.get("additional_notes") existing = context.get("existing_criteria") output_type = context.get("output_type", "Exclusion") model = context.get("model", DEFAULT_CRITERIA_MODEL) api_key = context.get("api_key") # Append PICO elements to notes if provided pico = context.get("pico_extraction") if pico: pico_lines = ["EXTRACTED PICO ELEMENTS:"] for element, data in pico.items(): if not isinstance(data, dict): continue element_title = element.replace("_", " ").title() explicit = data.get("explicit", []) implicit = data.get("implicit", []) all_items = explicit + implicit if all_items: pico_lines.append(f"- {element_title}: {', '.join(all_items)}") if len(pico_lines) > 1: pico_text = "\n".join(pico_lines) notes = f"{notes}\n\n{pico_text}" if notes else pico_text # Append conflict summary if provided conflict_summary = context.get("conflict_summary") if conflict_summary: notes = f"{notes}\n\n{conflict_summary}" if notes else conflict_summary service = CriteriaAIService(project_id=project_id) logger.info("execute_generate: output_type=%s", output_type) if output_type == "Exclusion": return await service.generate_exclusion_criteria( project_description=desc, research_questions=rqs, additional_notes=notes, existing_criteria=existing, model=model, api_key=api_key, ) elif output_type == "Inclusion": return await service.generate_inclusion_criteria( project_description=desc, research_questions=rqs, additional_notes=notes, existing_criteria=existing, model=model, api_key=api_key, ) else: # "Both" - run sequentially, concatenate exclusions = await service.generate_exclusion_criteria( project_description=desc, research_questions=rqs, additional_notes=notes, existing_criteria=existing, model=model, api_key=api_key, ) inclusions = await service.generate_inclusion_criteria( project_description=desc, research_questions=rqs, additional_notes=notes, existing_criteria=existing, model=model, api_key=api_key, ) return exclusions + inclusions async def execute_extract_pico(context: dict) -> dict: """ Extract PICO elements from project context. Context keys: project_id (int): Project ID project_description (str): Project description text research_questions (list[str]): Research questions existing_criteria (list[dict], optional): Existing criteria dicts model (str, optional): OpenAI model name api_key (str, optional): OpenAI API key Returns: Dict with pico_extraction, gap_flags, accepted_elements """ project_id = context.get("project_id", 0) desc = context.get("project_description") rqs = context.get("research_questions") existing = context.get("existing_criteria") model = context.get("model", DEFAULT_CRITERIA_MODEL) api_key = context.get("api_key") service = CriteriaAIService(project_id=project_id) logger.info("execute_extract_pico: starting extraction") result = await service.extract_pico( project_description=desc, research_questions=rqs, existing_criteria=existing, model=model, api_key=api_key, ) pico_extraction = result.get("pico_extraction", {}) gap_flags = result.get("gap_flags", []) # Build accepted elements (explicit + implicit accepted by default) accepted_elements = { element: { "explicit": list(range(len(data.get("explicit", [])))), "implicit": list(range(len(data.get("implicit", [])))), "ambiguous": [], # Ambiguous not accepted by default } for element, data in pico_extraction.items() if isinstance(data, dict) } return { "pico_extraction": pico_extraction, "gap_flags": gap_flags, "accepted_elements": accepted_elements, } async def execute_refine(context: dict) -> list[dict]: """ Generate criteria refinement suggestions from reconciliation patterns. Context keys: project_id (int): Project ID reconciliation_patterns (dict): Output of pattern analysis current_exclusion_criteria (list[dict]): Current exclusion criteria as dicts current_inclusion_criteria (list[dict]): Current inclusion criteria as dicts project_description (str, optional): Project description model (str, optional): OpenAI model name api_key (str, optional): OpenAI API key Returns: List of suggestion dicts with keys: action, category, text, rationale """ project_id = context.get("project_id", 0) patterns = context.get("reconciliation_patterns", {}) current_excl = context.get("current_exclusion_criteria", []) current_incl = context.get("current_inclusion_criteria", []) desc = context.get("project_description", "") model = context.get("model", DEFAULT_CRITERIA_MODEL) api_key = context.get("api_key") if not patterns: return [] service = CriteriaAIService(project_id=project_id) logger.info("execute_refine: analyzing reconciliation patterns") return await service.refine_criteria( reconciliation_patterns=patterns, current_criteria=current_excl, inclusion_criteria=current_incl, project_description=desc, model=model, api_key=api_key, ) async def execute_reconcile_conflicts(context: dict) -> list[ExclusionCriterion]: """ Analyze human/AI decision conflicts and generate refined criteria. Context keys: project_id (int): Project ID conflict_papers (list[dict]): Papers with conflicting decisions active_criteria (list[ExclusionCriterion], optional): Current active criteria project_description (str, optional): Project description research_questions (list[str], optional): Research questions model (str, optional): OpenAI model name api_key (str, optional): OpenAI API key Returns: List of ExclusionCriterion objects to improve screening accuracy """ project_id = context.get("project_id", 0) papers = context.get("conflict_papers", []) active = context.get("active_criteria") desc = context.get("project_description", "") rqs = context.get("research_questions") model = context.get("model", DEFAULT_CRITERIA_MODEL) api_key = context.get("api_key") if not papers: return [] service = CriteriaAIService(project_id=project_id) logger.info("execute_reconcile_conflicts: papers=%d", len(papers)) return await service.reconcile_conflicts( conflict_papers=papers, active_criteria=active, project_description=desc, research_questions=rqs, model=model, api_key=api_key, )