app/adi.py

# =====================================================================================================
# Anti-Dump Algorithm (ADI) - FIXED VERSION
# A mathematical framework for evaluating and filtering low-quality, unproductive text inputs.
#
# Copyright 2008 - 2026 S. Volkan Kücükbudak
# NOTE: This file contains the core logic for calculating the ADI. It is not an application itself.
# It serves as a library to be integrated into other tools.
#
# IF YOU USE THIS CODE, PLEASE READ THE LICENSE FILE.
# Do not steal free code. Respecting developers' credits ensures that projects like this remain open-source.
# =====================================================================================================
# https://github.com/VolkanSah/Anti-Dump-Index
# =====================================================================================================
# QUICK USAGE EXAMPLE
# This section demonstrates how to initialize the analyzer and run it on sample texts.
# =====================================================================================================
#
# analyzer = DumpindexAnalyzer()
#
# test_inputs = [
#     "Pls fix my code. Urgent!!!",
#     """I'm trying to implement a login function in Python. 
#     When calling auth.login(), I get a TypeError. 
#     Here's my code:
#     ```python
#     def login(username, password):
#         return auth.login(username)
#     ```
#     I'm using Python 3.8 and the auth library version 2.1."""
# ]
#
# for input_text in test_inputs:
#     result = analyzer.analyze_input(input_text)
#     print("-" * 50)
#     print(f"Analysis for: {input_text[:50]}...")
#     print(f"ADI: {result['adi']}")
#     print(f"Decision: {result['decision']}")
#     print("Recommendations:")
#     for rec in result['recommendations']:
#         print(f"- {rec}")
#     print("\nMetrics:", result['metrics'])
# print("-" * 50)
#
# =====================================================================================================
# END OF EXAMPLE
# =====================================================================================================

from dataclasses import dataclass
from typing import List, Dict, Tuple, Optional
import re
import numpy as np
import json
from pathlib import Path

@dataclass
class InputMetrics:
    noise: float
    effort: float
    context: float
    details: float
    bonus_factors: float
    penalty_factors: float
    repetition_penalty: float = 0.0

class DumpindexAnalyzer:
    def __init__(self, weights: Dict[str, float] = None, enable_logging: bool = False):
        """
        Initialize the ADI Analyzer.
        
        Args:
            weights: Custom weight configuration for your use case
            enable_logging: If True, logs all analyses to adi_logs.jsonl for later optimization
        """
        self.weights = weights or {
            'noise': 1.0,
            'effort': 2.0,
            'context': 1.5,
            'details': 1.5,
            'bonus': 0.5,
            'penalty': 1.0
        }
        
        self.enable_logging = enable_logging
        self.log_file = Path('adi_logs.jsonl')
        
        # Pattern definitions for metric extraction 
        # !!!! Only demo examples! In production you need your own or get data from  vectors!!!!
        self.noise_patterns = {
            'urgency': r'\b(urgent|asap|emergency|!!+|\?\?+)\b',
            'informal': r'\b(pls|plz|thx|omg|wtf)\b',
            'vague': r'\b(something|somehow|maybe|probably)\b'
        }
        
        self.detail_patterns = {
            'code_elements': r'\b(function|class|method|variable|array|object|def|return)\b',
            'technical_terms': r'\b(error|exception|bug|issue|crash|fail|traceback|stack)\b',
            'specifics': r'[a-zA-Z_][a-zA-Z0-9_]*\.[a-zA-Z_][a-zA-Z0-9_]*'
        }
        
        self.context_indicators = {
            'background': r'\b(because|since|as|when|while)\b',
            'environment': r'\b(using|version|environment|platform|system)\b',
            'goal': r'\b(trying to|want to|need to|goal is|attempting to)\b'
        }

    def _has_negation_before(self, text: str, match_pos: int, window_size: int = 50) -> bool:
        """
        Check if a negation word appears within a specified window before the match position.
        This prevents false positives like 'I have no idea when this started' counting as context.
        
        Args:
            text: The full input text
            match_pos: Position of the matched pattern
            window_size: Number of characters to look back (default: 50)
        
        Returns:
            True if negation found, False otherwise
        """
        window_start = max(0, match_pos - window_size)
        window = text[window_start:match_pos].lower()
        return bool(re.search(r'\b(no|not|never|without|dont|don\'t|doesnt|doesn\'t)\b', window))

    def calculate_repetition_penalty(self, text: str) -> float:
        """
        Calculate penalty for keyword stuffing and repetitive patterns.
        This prevents gaming the system by repeating technical terms.
        
        Returns:
            Penalty score (0 to 3, where higher means more repetition)
        """
        words = text.lower().split()
        if len(words) == 0:
            return 0.0
        
        # Calculate unique word ratio
        unique_ratio = len(set(words)) / len(words)
        
        # Detect excessive repetition of the same word
        word_counts = {}
        for word in words:
            if len(word) > 3:  # Ignore short words like 'the', 'and'
                word_counts[word] = word_counts.get(word, 0) + 1
        
        max_repetition = max(word_counts.values()) if word_counts else 1
        repetition_factor = min(max_repetition / len(words), 0.5)
        
        # Combined penalty
        penalty = (1 - unique_ratio) * 2 + repetition_factor * 2
        return min(penalty, 3.0)

    def calculate_noise(self, text: str) -> Tuple[float, Dict]:
        """
        Calculates the noise ratio in the input text by detecting irrelevant or informal words.
        Returns the ratio of noise words to total words, and a dictionary of all matched patterns.
        """
        noise_count = 0
        noise_details = {}
        
        for category, pattern in self.noise_patterns.items():
            matches = re.findall(pattern, text.lower())
            noise_count += len(matches)
            noise_details[category] = matches
        
        total_words = len(text.split())
        return (noise_count / max(total_words, 1), noise_details)

    def calculate_effort(self, text: str) -> float:
        """
        Assesses the effort invested in the input's structure.
        FIXED: Now handles edge cases like very short sentences properly.
        """
        sentences = [s.strip() for s in re.split(r'[.!?]+', text) if s.strip()]
        if not sentences:
            return 0.0
        
        avg_sentence_length = np.mean([len(s.split()) for s in sentences])
        has_formatting = bool(re.search(r'```|\*\*|\n\s*\n', text))
        has_punctuation = bool(re.search(r'[.,;:]', text))
        
        # FIX: Weight sentence count AND length, not just length range
        sentence_quality = (
            (len(sentences) >= 3) * 1.0 +  # Bonus for multiple sentences
            (20 <= avg_sentence_length <= 50) * 2.0 +  # Ideal length range
            (avg_sentence_length >= 5) * 0.5  # Minimum meaningful length
        )
        
        effort_score = min(5.0, (
            sentence_quality +
            has_formatting * 1.5 +
            has_punctuation * 1.5
        ))
        
        return effort_score

    def calculate_context(self, text: str) -> float:
        """
        Measures the presence of background information.
        FIXED: Now checks for negations to avoid false positives.
        """
        context_score = 0.0
        
        for category, pattern in self.context_indicators.items():
            for match in re.finditer(pattern, text.lower()):
                # Only count if NOT preceded by negation
                if not self._has_negation_before(text, match.start()):
                    context_score += 1.0
                    break  # Only count once per category
        
        return min(5.0, context_score)

    def calculate_details(self, text: str) -> Tuple[float, Dict]:
        """
        Quantifies the level of technical depth. This function looks for specific
        technical keywords, code snippets, and structured data that adds value.
        """
        detail_score = 0.0
        detail_findings = {}
        
        for category, pattern in self.detail_patterns.items():
            matches = re.findall(pattern, text.lower())
            score = len(matches) * 0.5
            detail_findings[category] = matches
            detail_score += score
        
        # Cap the score to prevent keyword stuffing from dominating
        return (min(5.0, detail_score), detail_findings)

    def calculate_bonus_factors(self, text: str) -> float:
        """
        Identifies and rewards positive formatting elements like code blocks,
        links, or bulleted lists, which significantly improve clarity.
        """
        bonus_score = 0.0
        
        if re.search(r'```[\s\S]*?```', text):
            bonus_score += 1.0
        if re.search(r'\[.*?\]\(.*?\)', text):
            bonus_score += 0.5
        if re.search(r'\n\s*[-*+]\s', text):
            bonus_score += 0.5
        
        return bonus_score

    def calculate_penalty_factors(self, text: str) -> Tuple[float, Dict]:
        """
        Deducts points for negative characteristics, such as excessive capitalization,
        redundant punctuation, or inputs that are too short to be useful.
        """
        penalties = {}
        
        # Excessive capitalization
        alpha_chars = re.findall(r'[a-zA-Z]', text)
        if alpha_chars:
            caps_ratio = len(re.findall(r'[A-Z]', text)) / len(alpha_chars)
            if caps_ratio > 0.7:
                penalties['excessive_caps'] = caps_ratio
        
        # Excessive punctuation
        excessive_punctuation = len(re.findall(r'[!?]{2,}', text))
        if excessive_punctuation:
            penalties['excessive_punctuation'] = excessive_punctuation
        
        # Too short
        if len(text.split()) < 10:
            penalties['too_short'] = 1.0
        
        penalty_score = sum(penalties.values()) if penalties else 0
        return (min(5.0, penalty_score), penalties)

    def calculate_adi(self, metrics: InputMetrics) -> float:
        """
        Calculates the final Anti-Dump Index (ADI) score using the weighted formula.
        FIXED: Now includes repetition penalty in the denominator to dampen gaming attempts.
        """
        try:
            numerator = (
                self.weights['noise'] * metrics.noise -
                (self.weights['effort'] * metrics.effort +
                 self.weights['bonus'] * metrics.bonus_factors)
            )
            
            # FIX: Add repetition penalty to denominator to reduce impact of keyword stuffing
            denominator = (
                self.weights['context'] * metrics.context +
                self.weights['details'] * metrics.details +
                self.weights['penalty'] * metrics.penalty_factors +
                metrics.repetition_penalty
            )
            
            # Ensure we never divide by zero
            return numerator / max(denominator, 0.1)
            
        except Exception as e:
            print(f"Error calculating ADI: {e}")
            return float('inf')

    def analyze_input(self, text: str, user_context: Optional[Dict] = None) -> Dict:
        """
        Main entry point for the analysis. Orchestrates the entire workflow.
        
        Args:
            text: The input text to analyze
            user_context: Optional dict with 'tier', 'history_avg' for context-aware routing
        
        Returns:
            Dictionary with ADI score, metrics, decision, and recommendations
        """
        # Calculate all metrics
        noise_value, noise_details = self.calculate_noise(text)
        effort_value = self.calculate_effort(text)
        context_value = self.calculate_context(text)
        details_value, detail_findings = self.calculate_details(text)
        bonus_value = self.calculate_bonus_factors(text)
        penalty_value, penalty_details = self.calculate_penalty_factors(text)
        repetition_value = self.calculate_repetition_penalty(text)
        
        metrics = InputMetrics(
            noise=noise_value,
            effort=effort_value,
            context=context_value,
            details=details_value,
            bonus_factors=bonus_value,
            penalty_factors=penalty_value,
            repetition_penalty=repetition_value
        )
        
        adi = self.calculate_adi(metrics)
        
        # Context-aware adjustment (if user tier provided)
        adi_adjusted = adi
        if user_context:
            if user_context.get('tier') == 'enterprise':
                adi_adjusted *= 0.8  # More lenient for paying customers
            if user_context.get('history_avg', 0) < 0:
                adi_adjusted *= 0.9  # Boost for users with good track record
        
        decision = self._make_decision(adi_adjusted)
        recommendations = self._generate_recommendations(
            metrics, noise_details, detail_findings, penalty_details
        )
        
        result = {
            'adi': round(adi, 3),
            'adi_adjusted': round(adi_adjusted, 3) if user_context else None,
            'metrics': {
                'noise': round(noise_value, 3),
                'effort': round(effort_value, 3),
                'context': round(context_value, 3),
                'details': round(details_value, 3),
                'bonus_factors': round(bonus_value, 3),
                'penalty_factors': round(penalty_value, 3),
                'repetition_penalty': round(repetition_value, 3)
            },
            'decision': decision,
            'recommendations': recommendations,
            'details': {
                'noise_findings': noise_details,
                'technical_details': detail_findings,
                'penalties': penalty_details
            }
        }
        
        # Optional logging for later weight optimization
        if self.enable_logging:
            self._log_analysis(text, adi, metrics)
        
        return result

    def _make_decision(self, adi: float) -> str:
        """
        Translates the numerical ADI score into a categorical decision.
        """
        if adi > 1:
            return "REJECT"
        elif 0 <= adi <= 1:
            return "MEDIUM_PRIORITY"
        else:
            return "HIGH_PRIORITY"

    def _generate_recommendations(self, metrics: InputMetrics,
                                 noise_details: Dict,
                                 detail_findings: Dict,
                                 penalty_details: Dict) -> List[str]:
        """
        Generates actionable suggestions to help the user improve their input.
        """
        recommendations = []
        
        if metrics.noise > 0.3:
            recommendations.append("Reduce informal or urgent expressions.")
            
        if metrics.context < 1.0:
            recommendations.append("Provide more context (environment, background, goal).")
            
        if metrics.details < 1.0:
            recommendations.append("Include specific technical details or error messages.")
            
        if metrics.effort < 2.0:
            recommendations.append("Improve the structure of your input with proper sentences.")
        
        if metrics.repetition_penalty > 1.0:
            recommendations.append("Avoid repeating the same keywords excessively.")
            
        if metrics.penalty_factors > 0:
            if 'excessive_caps' in penalty_details:
                recommendations.append("Avoid excessive capitalization.")
            if 'excessive_punctuation' in penalty_details:
                recommendations.append("Reduce excessive punctuation marks.")
            if 'too_short' in penalty_details:
                recommendations.append("Provide a more detailed description (minimum 10 words).")
        
        if not recommendations:
            recommendations.append("Your input quality is excellent. No improvements needed.")
                    
        return recommendations

    def _log_analysis(self, text: str, adi: float, metrics: InputMetrics):
        """
        Log analysis results to file for later weight optimization.
        Format: One JSON object per line (JSONL).
        """
        log_entry = {
            'text_hash': hash(text),
            'text_length': len(text),
            'adi': round(adi, 3),
            'metrics': {
                'noise': round(metrics.noise, 3),
                'effort': round(metrics.effort, 3),
                'context': round(metrics.context, 3),
                'details': round(metrics.details, 3),
                'bonus_factors': round(metrics.bonus_factors, 3),
                'penalty_factors': round(metrics.penalty_factors, 3),
                'repetition_penalty': round(metrics.repetition_penalty, 3)
            }
        }
        
        with open(self.log_file, 'a') as f:
            f.write(json.dumps(log_entry) + '\n')

    def validate_weights(self, test_cases: List[Tuple[str, str]]) -> float:
        """
        Validate current weights against manually labeled test cases.
        
        Args:
            test_cases: List of (input_text, expected_decision) tuples
                       Example: [("Help pls!", "REJECT"), ("Python KeyError...", "HIGH_PRIORITY")]
        
        Returns:
            Accuracy score (0.0 to 1.0)
        """
        if not test_cases:
            raise ValueError("test_cases cannot be empty")
        
        correct = 0
        for text, expected in test_cases:
            result = self.analyze_input(text)
            if result['decision'] == expected:
                correct += 1
        
        accuracy = correct / len(test_cases)
        print(f"Weight Validation: {correct}/{len(test_cases)} correct ({accuracy:.1%})")
        return accuracy


# =====================================================================================================
# USAGE EXAMPLE
# =====================================================================================================
# if __name__ == "__main__":
#    analyzer = DumpindexAnalyzer(enable_logging=False)
#    
#    test_inputs = [
#        "Pls fix my code. Urgent!!!",
#        """I'm trying to implement a login function in Python. 
#        When calling auth.login(), I get a TypeError. 
#        Here's my code:
#        ```python
#        def login(username, password):
# #           return auth.login(username)
#        ```
#        I'm using Python 3.8 and the auth library version 2.1.""",
#        "error error error bug bug crash crash function method class object variable",  # Keyword stuffing test
#    ]
    
#    for input_text in test_inputs:
#        result = analyzer.analyze_input(input_text)
#        print("-" * 70)
#        print(f"Input: {input_text[:60]}...")
#        print(f"ADI: {result['adi']}")
#        print(f"Decision: {result['decision']}")
#        print("Recommendations:")
#        for rec in result['recommendations']:
#            print(f"  - {rec}")
#        print(f"Metrics: {result['metrics']}")
#    print("-" * 70)

# Have fun :) Volkan Sah