src/scipeerai/modules/field_saturation.py

# src/scipeerai/modules/field_saturation.py
#
# Module 23: Field Saturation Detector
# Detects over-published research topics, measures novelty space,
# identifies redundant contribution patterns, and scores field
# crowding relative to claimed contributions.
#
# Score attribute: saturation_score (0.0 = fresh field, 1.0 = saturated)
# Part of SciPeerAI Phase 6 — v2.3.0

from __future__ import annotations

import re
import math
from collections import Counter
from dataclasses import dataclass


# ── Data classes ──────────────────────────────────────────────────────────────

@dataclass
class SaturationFlag:
    flag_type:   str
    severity:    str
    description: str
    evidence:    str
    suggestion:  str


@dataclass
class SaturationResult:
    topic_keywords:          list
    keyword_density:         float
    redundancy_score:        float
    novelty_claim_score:     float
    contribution_vagueness:  float
    overcrowding_signals:    int
    saturation_score:        float
    risk_level:              str
    summary:                 str
    flags:                   list
    flags_count:             int


# ── Compiled patterns ─────────────────────────────────────────────────────────

_NOVELTY_CLAIMS = re.compile(
    r'\b(novel|innovative|first|pioneer|unique|original|breakthrough|'
    r'state.of.the.art|cutting.edge|unprecedented|revolutionar\w+|'
    r'new approach|new method|new framework|we propose|we present|'
    r'we introduce|we develop)\b',
    re.IGNORECASE,
)

_SATURATION_SIGNALS = re.compile(
    r'\b(many studies|numerous studies|extensive research|'
    r'widely studied|well.studied|well.known|well.established|'
    r'extensively investigated|much attention|growing body|'
    r'large body of|considerable research|substantial literature|'
    r'abundant literature|intensively studied|heavily researched|'
    r'significant amount of work|proliferation of|surge of interest)\b',
    re.IGNORECASE,
)

_VAGUE_CONTRIBUTION = re.compile(
    r'\b(improve\w*|enhanc\w*|better\w*|outperform\w*|superior\w*|'
    r'more efficient|more effective|more accurate|higher performance|'
    r'significant improvement|notable improvement|'
    r'promising results|competitive results)\b',
    re.IGNORECASE,
)

_KEYWORD_STOP = {
    'the', 'a', 'an', 'and', 'or', 'but', 'in', 'on', 'at', 'to',
    'for', 'of', 'with', 'by', 'from', 'is', 'are', 'was', 'were',
    'be', 'been', 'has', 'have', 'had', 'do', 'does', 'did', 'will',
    'would', 'could', 'should', 'may', 'might', 'this', 'that', 'these',
    'those', 'it', 'its', 'we', 'our', 'their', 'as', 'not', 'also',
    'can', 'which', 'such', 'than', 'other', 'into', 'used', 'using',
    'based', 'paper', 'study', 'research', 'method', 'approach', 'work',
    'show', 'shows', 'shown', 'propose', 'present', 'results', 'data',
}

_SENTENCE_SPLIT = re.compile(r'(?<=[.!?])\s+')
_WORD_RE        = re.compile(r'\b[a-zA-Z]{4,}\b')


# ── Engine ────────────────────────────────────────────────────────────────────

class FieldSaturationDetector:
    """
    Field Saturation Detector — Module 23.

    Analyses paper text to detect:
    - Over-saturated research topics with redundant contributions
    - Vague novelty claims unsupported by specific advances
    - Crowding signals indicating field congestion
    - Mismatch between claimed novelty and acknowledged saturation
    """

    def analyze(self, text: str) -> SaturationResult:
        text = (text or "").strip()
        if not text:
            return self._empty_result("No text provided for saturation analysis.")

        keywords             = self._extract_topic_keywords(text)
        keyword_density      = self._compute_keyword_density(text, keywords)
        redundancy_score     = self._compute_redundancy(text)
        novelty_claim_score  = self._compute_novelty_claims(text)
        contribution_vague   = self._compute_contribution_vagueness(text)
        overcrowding_signals = self._count_overcrowding_signals(text)

        saturation_score = self._compute_score(
            keyword_density,
            redundancy_score,
            novelty_claim_score,
            contribution_vague,
            overcrowding_signals,
        )

        risk_level = (
            "critical" if saturation_score >= 0.75 else
            "high"     if saturation_score >= 0.55 else
            "medium"   if saturation_score >= 0.30 else
            "low"
        )

        flags   = self._build_flags(
            keywords, keyword_density, redundancy_score,
            novelty_claim_score, contribution_vague,
            overcrowding_signals, saturation_score,
        )

        summary = self._build_summary(
            keywords, overcrowding_signals,
            redundancy_score, saturation_score, risk_level,
        )

        return SaturationResult(
            topic_keywords         = keywords[:15],
            keyword_density        = round(keyword_density, 4),
            redundancy_score       = round(redundancy_score, 4),
            novelty_claim_score    = round(novelty_claim_score, 4),
            contribution_vagueness = round(contribution_vague, 4),
            overcrowding_signals   = overcrowding_signals,
            saturation_score       = round(saturation_score, 4),
            risk_level             = risk_level,
            summary                = summary,
            flags                  = flags,
            flags_count            = len(flags),
        )

    # ── Extraction ────────────────────────────────────────────────────────────

    def _extract_topic_keywords(self, text: str) -> list:
        words   = _WORD_RE.findall(text.lower())
        cleaned = [w for w in words if w not in _KEYWORD_STOP and len(w) >= 4]
        counter = Counter(cleaned)
        return [w for w, _ in counter.most_common(30) if counter[w] >= 2]

    # ── Analysis ──────────────────────────────────────────────────────────────

    def _compute_keyword_density(self, text: str, keywords: list) -> float:
        if not keywords:
            return 0.0
        words       = _WORD_RE.findall(text.lower())
        total_words = max(len(words), 1)
        kw_set      = set(keywords[:10])
        kw_count    = sum(1 for w in words if w in kw_set)
        raw_density = kw_count / total_words
        return round(min(1.0, raw_density * 8), 4)

    def _compute_redundancy(self, text: str) -> float:
        sentences = _SENTENCE_SPLIT.split(text.strip())
        sentences = [s.strip() for s in sentences if len(s.strip()) > 20]
        if len(sentences) < 4:
            return 0.0

        word_sets = []
        for s in sentences:
            words = set(_WORD_RE.findall(s.lower())) - _KEYWORD_STOP
            if words:
                word_sets.append(words)

        if len(word_sets) < 4:
            return 0.0

        overlap_count = 0
        total_pairs   = 0
        for i in range(len(word_sets)):
            for j in range(i + 1, min(i + 6, len(word_sets))):
                a, b        = word_sets[i], word_sets[j]
                union       = a | b
                intersection = a & b
                if union:
                    jaccard = len(intersection) / len(union)
                    if jaccard > 0.35:
                        overlap_count += 1
                total_pairs += 1

        if total_pairs == 0:
            return 0.0
        return round(min(1.0, overlap_count / total_pairs * 2.5), 4)

    def _compute_novelty_claims(self, text: str) -> float:
        words        = _WORD_RE.findall(text)
        total_words  = max(len(words), 1)
        claim_count  = len(_NOVELTY_CLAIMS.findall(text))
        claim_ratio  = claim_count / total_words

        sat_count    = len(_SATURATION_SIGNALS.findall(text))

        if sat_count > 0 and claim_count > 0:
            mismatch = min(1.0, (claim_count / max(sat_count, 1)) * 0.25)
        else:
            mismatch = 0.0

        base = min(1.0, claim_ratio * 60)
        return round(min(1.0, base * 0.5 + mismatch * 0.5), 4)

    def _compute_contribution_vagueness(self, text: str) -> float:
        words       = _WORD_RE.findall(text)
        total_words = max(len(words), 1)
        vague_count = len(_VAGUE_CONTRIBUTION.findall(text))
        return round(min(1.0, (vague_count / total_words) * 80), 4)

    def _count_overcrowding_signals(self, text: str) -> int:
        return len(_SATURATION_SIGNALS.findall(text))

    # ── Scoring ───────────────────────────────────────────────────────────────

    def _compute_score(
        self,
        keyword_density:     float,
        redundancy_score:    float,
        novelty_claim_score: float,
        contribution_vague:  float,
        overcrowding:        int,
    ) -> float:
        score  = 0.0
        score += keyword_density     * 0.15
        score += redundancy_score    * 0.30
        score += novelty_claim_score * 0.25
        score += contribution_vague  * 0.20
        score += min(overcrowding * 0.05, 0.20)
        return round(min(1.0, score), 4)

    # ── Flags ─────────────────────────────────────────────────────────────────

    def _build_flags(
        self,
        keywords:            list,
        keyword_density:     float,
        redundancy_score:    float,
        novelty_claim_score: float,
        contribution_vague:  float,
        overcrowding:        int,
        saturation_score:    float,
    ) -> list:
        flags = []

        if overcrowding >= 3:
            flags.append(SaturationFlag(
                flag_type   = "field_overcrowding_acknowledged",
                severity    = "high",
                description = (
                    f"Paper acknowledges {overcrowding} field saturation signals, "
                    f"yet still claims novel contribution — a credibility mismatch."
                ),
                evidence    = (
                    f"{overcrowding} saturation phrases detected: "
                    f"'many studies', 'well-studied', 'extensive research', etc. "
                    f"Field is crowded by the paper's own admission."
                ),
                suggestion  = (
                    "Clearly articulate what specific gap this paper fills that "
                    "prior work has not. Vague novelty in saturated fields is a "
                    "major rejection signal for top journals."
                ),
            ))

        if redundancy_score >= 0.45:
            flags.append(SaturationFlag(
                flag_type   = "high_content_redundancy",
                severity    = "medium",
                description = (
                    "High sentence-level content redundancy detected — paper "
                    "repeats concepts across sections without adding new information."
                ),
                evidence    = (
                    f"Jaccard similarity analysis across sentence windows "
                    f"returned redundancy score of {redundancy_score:.2f} "
                    f"(threshold: 0.45). Significant content overlap present."
                ),
                suggestion  = (
                    "Restructure paper to eliminate repetitive content. "
                    "Each section should introduce new information, not restate previous points."
                ),
            ))

        if novelty_claim_score >= 0.45:
            flags.append(SaturationFlag(
                flag_type   = "overclaimed_novelty",
                severity    = "high",
                description = (
                    "Paper makes disproportionately high novelty claims relative "
                    "to the specificity of its actual contributions."
                ),
                evidence    = (
                    f"Novelty claim density score: {novelty_claim_score:.2f}. "
                    f"High frequency of terms like 'novel', 'first', 'innovative', "
                    f"'breakthrough' without commensurate technical specificity."
                ),
                suggestion  = (
                    "Replace broad novelty claims with precise technical statements. "
                    "Quantify what makes the contribution different, not just that it is different."
                ),
            ))

        if contribution_vague >= 0.40:
            flags.append(SaturationFlag(
                flag_type   = "vague_contribution_language",
                severity    = "medium",
                description = (
                    "Contribution language is vague and non-specific — paper "
                    "relies on generic improvement claims rather than measurable advances."
                ),
                evidence    = (
                    f"Vagueness score: {contribution_vague:.2f}. Overuse of "
                    f"'improve', 'enhance', 'better', 'outperform' without "
                    f"quantified baselines or specific metrics."
                ),
                suggestion  = (
                    "State contributions with exact numbers: 'X% improvement on Y benchmark "
                    "over Z baseline' instead of 'improved performance'."
                ),
            ))

        if keyword_density >= 0.55 and len(keywords) >= 5:
            flags.append(SaturationFlag(
                flag_type   = "topic_keyword_saturation",
                severity    = "low",
                description = (
                    "Topic keyword density is high — a small set of terms "
                    "dominates the entire paper, suggesting narrow scope."
                ),
                evidence    = (
                    f"Keyword density index: {keyword_density:.2f}. "
                    f"Top terms: {', '.join(keywords[:6])}. "
                    f"Repeated heavily throughout without broadening context."
                ),
                suggestion  = (
                    "Contextualize the work within broader research themes. "
                    "Narrow keyword scope can signal incremental rather than transformative contribution."
                ),
            ))

        if not flags:
            flags.append(SaturationFlag(
                flag_type   = "field_saturation_acceptable",
                severity    = "low",
                description = "No significant field saturation indicators detected.",
                evidence    = (
                    f"Saturation score: {saturation_score:.2f}. "
                    f"Contribution language is sufficiently specific. "
                    f"Field crowding signals are within acceptable range."
                ),
                suggestion  = "Continue ensuring each section adds incremental specificity.",
            ))

        return flags

    # ── Summary ───────────────────────────────────────────────────────────────

    def _build_summary(
        self,
        keywords:         list,
        overcrowding:     int,
        redundancy_score: float,
        saturation_score: float,
        risk_level:       str,
    ) -> str:
        top_kw = ', '.join(keywords[:5]) if keywords else 'none detected'
        return (
            f"Field saturation analysis complete. "
            f"Top topic keywords: {top_kw}. "
            f"{overcrowding} field crowding signal(s) detected in text. "
            f"Content redundancy score: {redundancy_score:.2f}. "
            f"Overall saturation risk: {risk_level.upper()}."
        )

    # ── Fallback ──────────────────────────────────────────────────────────────

    def _empty_result(self, msg: str) -> SaturationResult:
        return SaturationResult(
            topic_keywords         = [],
            keyword_density        = 0.0,
            redundancy_score       = 0.0,
            novelty_claim_score    = 0.0,
            contribution_vagueness = 0.0,
            overcrowding_signals   = 0,
            saturation_score       = 0.0,
            risk_level             = "low",
            summary                = msg,
            flags                  = [],
            flags_count            = 0,
        )