validation_module.py

"""
validation_module.py
--------------------
Clinical validation module for radiology reports.

Performs:
  1. Structural completeness check
  2. Medical terminology validation
  3. Logical consistency analysis (findings → impression)
  4. Confidence score computation
  5. Detailed validation feedback
"""

import logging
import re
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Tuple

logger = logging.getLogger(__name__)


# ---------------------------------------------------------------------------
# Constants — Medical Terminology Dictionaries
# ---------------------------------------------------------------------------

# Positive radiological finding terms (suggest pathology is mentioned)
FINDING_TERMS = {
    # Pulmonary
    "opacity", "consolidation", "infiltrate", "effusion", "pneumothorax",
    "atelectasis", "nodule", "mass", "lesion", "cardiomegaly", "pneumonia",
    "edema", "fibrosis", "emphysema", "pleural", "cavitation", "calcification",
    # Neuro
    "hemorrhage", "infarct", "midline shift", "herniation", "hydrocephalus",
    "atrophy", "gliosis", "white matter", "leukoencephalopathy",
    # Abdominal
    "hepatomegaly", "splenomegaly", "ascites", "obstruction", "perforation",
    "ileus", "dilatation", "gallstone", "calculi", "adenopathy",
    # Musculoskeletal
    "fracture", "dislocation", "lytic", "sclerotic", "arthritis", "erosion",
    "osteoporosis", "osteopenia", "hardware", "implant", "arthroplasty",
    "healed", "healing", "surgical", "fixation", "screw", "prosthesis",
    "joint", "bone", "density", "cortical", "trabecular", "alignment",
    # Vascular
    "stenosis", "occlusion", "aneurysm", "thrombosis", "calcified",
    # Soft tissue
    "swelling", "inflammation", "fluid", "collection", "thickening",
    # Normal descriptors
    "normal", "unremarkable", "no acute", "within normal limits", "clear",
    "intact", "preserved", "well-healed", "satisfactory", "stable",
    "demonstrates", "appears", "shows", "noted", "visible", "evident",
}

# Impression must include at least one of these conclusion terms
IMPRESSION_CONCLUSION_TERMS = {
    "consistent with", "suggestive of", "findings indicate", "impression",
    "likely", "possible", "probable", "no evidence", "cannot exclude",
    "differential", "consider", "normal", "unremarkable", "recommend",
    "represents", "compatible with", "appears to", "evidence of",
    "healed", "healing", "demonstrates", "shows", "indicates",
    "no significant", "well-healed", "satisfactory", "stable",
    "there is", "there are", "findings are",
}

# Recommendation terms
RECOMMENDATION_TERMS = {
    "correlation", "follow-up", "repeat", "ct scan", "mri", "biopsy",
    "consult", "refer", "further", "additional", "clinical", "evaluation",
    "monitoring", "treatment", "management", "urgent", "emergent",
    "recommended", "suggested", "advised", "periodic", "regular",
    "orthopedic", "specialist", "physiotherapy", "physical therapy",
    "bone density", "dexa", "rheumatologist", "scan", "imaging",
}

# Required report sections
REQUIRED_SECTIONS = [
    "examination_details",
    "clinical_information",
    "findings",
    "impression",
    "recommendations",
]

# Minimum word counts per section
MINIMUM_WORDS = {
    "examination_details": 5,
    "clinical_information": 8,
    "findings": 10,
    "impression": 5,
    "recommendations": 5,
}


# ---------------------------------------------------------------------------
# Data Models
# ---------------------------------------------------------------------------


@dataclass
class ValidationResult:
    is_valid: bool = True
    confidence_score: float = 0.50
    section_scores: Dict[str, float] = field(default_factory=dict)
    issues: List[str] = field(default_factory=list)
    warnings: List[str] = field(default_factory=list)
    suggestions: List[str] = field(default_factory=list)

    def to_dict(self) -> dict:
        return {
            "is_valid": self.is_valid,
            "confidence_score": round(self.confidence_score, 3),
            "section_scores": {k: round(v, 3) for k, v in self.section_scores.items()},
            "issues": self.issues,
            "warnings": self.warnings,
            "suggestions": self.suggestions,
        }

    def summary(self) -> str:
        status = "✓ VALID" if self.is_valid else "✗ INVALID"
        lines = [
            f"Validation Status : {status}",
            f"Confidence Score  : {self.confidence_score:.2f}",
        ]
        if self.issues:
            lines.append("Issues:")
            lines.extend(f"  • {i}" for i in self.issues)
        if self.warnings:
            lines.append("Warnings:")
            lines.extend(f"  ⚠ {w}" for w in self.warnings)
        if self.suggestions:
            lines.append("Suggestions:")
            lines.extend(f"  → {s}" for s in self.suggestions)
        return "\n".join(lines)


# ---------------------------------------------------------------------------
# Validator
# ---------------------------------------------------------------------------


class ClinicalValidator:
    """
    Multi-stage clinical validation pipeline for radiology reports.
    """

    WEIGHTS = {
        "completeness": 0.30,
        "content_length": 0.15,
        "terminology": 0.25,
        "logical_consistency": 0.20,
        "recommendation_quality": 0.10,
    }

    def validate(self, report) -> ValidationResult:
        result = ValidationResult()
        scores: Dict[str, float] = {}

        sections = {
            "examination_details": report.examination_details or "",
            "clinical_information": report.clinical_information or "",
            "findings": report.findings or "",
            "impression": report.impression or "",
            "recommendations": report.recommendations or "",
        }

        scores["completeness"] = self._check_completeness(sections, result)
        scores["content_length"] = self._check_content_length(sections, result)
        scores["terminology"] = self._check_terminology(sections, result)

        # ✅ UPDATED LOGIC USED HERE
        scores["logical_consistency"] = self._check_logical_consistency(sections, result)

        scores["recommendation_quality"] = self._check_recommendations(sections, result)

        weighted = sum(scores[k] * self.WEIGHTS[k] for k in self.WEIGHTS)

        result.confidence_score = round(min(max(weighted, 0.30), 0.88), 3)
        result.section_scores = scores

        result.is_valid = (
            result.confidence_score >= 0.55
            and len(result.issues) == 0
        )

        return result

    # ------------------------------------------------------------------

    def _check_completeness(self, sections: Dict[str, str], result: ValidationResult) -> float:
        missing = []
        for sec in REQUIRED_SECTIONS:
            if not sections.get(sec, "").strip():
                missing.append(sec)

        if missing:
            result.issues.append(
                f"Missing required sections: {', '.join(missing)}"
            )
            return 1.0 - (len(missing) / len(REQUIRED_SECTIONS))

        return 1.0

    # ------------------------------------------------------------------

    def _check_content_length(self, sections: Dict[str, str], result: ValidationResult) -> float:
        total, passing = 0, 0
        for sec, min_words in MINIMUM_WORDS.items():
            total += 1
            word_count = len(sections.get(sec, "").split())
            if word_count >= min_words:
                passing += 1
            else:
                result.warnings.append(
                    f"Section '{sec}' is brief ({word_count} words; minimum recommended: {min_words})."
                )

        return passing / total if total > 0 else 0.0

    # ------------------------------------------------------------------

    def _check_terminology(self, sections: Dict[str, str], result: ValidationResult) -> float:
        findings_text = sections.get("findings", "").lower()
        impression_text = sections.get("impression", "").lower()

        combined = findings_text + " " + impression_text

        matched_terms = FINDING_TERMS.intersection(
            set(re.findall(r"\b\w+\b", combined))
        )

        for term in FINDING_TERMS:
            if " " in term and term in combined:
                matched_terms.add(term)

        score = min(len(matched_terms) / 2.0, 1.0)

        if len(matched_terms) == 0:
            result.issues.append(
                "No recognised radiological terminology found in findings/impression."
            )
        elif len(matched_terms) < 2:
            result.warnings.append(
                "Limited radiological terminology detected. Report may lack specificity."
            )
        else:
            result.suggestions.append(
                f"Detected radiological terms: {', '.join(sorted(matched_terms)[:6])}."
            )

        return score

    # ------------------------------------------------------------------
    # ✅ FIXED FUNCTION (MAIN CHANGE)
    # ------------------------------------------------------------------

    def _check_logical_consistency(
        self, sections: Dict[str, str], result: ValidationResult
    ) -> float:

        findings = sections.get("findings", "").lower()
        impression = sections.get("impression", "").lower()

        if not findings or not impression:
            return 0.5

        score = 1.0

        has_conclusion = any(
            term in impression for term in IMPRESSION_CONCLUSION_TERMS
        )
        if not has_conclusion:
            result.warnings.append(
                "Impression does not contain a clear diagnostic conclusion."
            )
            score -= 0.3

        # ✅ NEW CLEAN LOGIC
        ABNORMAL_TERMS = {
            "opacity", "opacities", "ground glass", "consolidation",
            "lesion", "mass", "nodule", "effusion", "infiltrate",
            "atelectasis", "fibrosis", "infection"
        }

        NORMAL_TERMS = {
            "normal", "unremarkable", "no acute", "within normal limits",
            "clear", "intact", "preserved"
        }

        findings_has_pathology = any(t in findings for t in ABNORMAL_TERMS)
        impression_says_normal = any(t in impression for t in NORMAL_TERMS)

        # 🚨 STRONG ERROR DETECTION
        if findings_has_pathology and impression_says_normal:
            result.issues.append(
                "Critical error: Findings indicate abnormal pathology but impression states normal."
            )
            result.is_valid = False
            score -= 0.7

        # ⚠ Mixed statement detection
        if "no" in findings and findings_has_pathology:
            result.warnings.append(
                "Findings contain both abnormal and negative descriptors; verify interpretation."
            )

        return max(score, 0.0)

    # ------------------------------------------------------------------

    def _check_recommendations(self, sections: Dict[str, str], result: ValidationResult) -> float:
        recs = sections.get("recommendations", "").lower()

        if not recs:
            result.warnings.append("Recommendations section is empty.")
            return 0.0

        matched = RECOMMENDATION_TERMS.intersection(
            set(re.findall(r"\b\w+\b", recs))
        )

        for term in RECOMMENDATION_TERMS:
            if " " in term and term in recs:
                matched.add(term)

        score = min(len(matched) / 2.0, 1.0)

        if len(matched) == 0:
            result.warnings.append(
                "Recommendations section does not contain specific clinical guidance."
            )

        return score


# ---------------------------------------------------------------------------
# Convenience Function
# ---------------------------------------------------------------------------


def validate_report(report) -> ValidationResult:
    """Top-level convenience function."""
    validator = ClinicalValidator()
    return validator.validate(report)


# ---------------------------------------------------------------------------
# Smoke Test
# ---------------------------------------------------------------------------

if __name__ == "__main__":
    import sys
    sys.path.insert(0, ".")
    from backend.report_generator import PatientInfo, ImagingInfo, RadiologyReport

    mock_report = RadiologyReport(
        examination_details="Chest X-ray, posteroanterior view.",
        clinical_information="52-year-old male presenting with persistent cough and fever for 3 days.",
        findings=(
            "Patchy consolidation in the right lower lobe with air bronchograms. "
            "No pleural effusion. Cardiac silhouette within normal limits. "
            "Costophrenic angles are sharp bilaterally."
        ),
        impression=(
            "Findings are consistent with right lower lobe pneumonia. "
            "No evidence of pleural complication."
        ),
        recommendations=(
            "Clinical correlation recommended. "
            "Consider repeat chest X-ray in 4 to 6 weeks to confirm resolution. "
            "If symptoms persist, CT chest may be warranted."
        ),
    )

    result = validate_report(mock_report)
    print(result.summary())
    print(f"\nFull JSON: {result.to_dict()}")