backend/evals/code_eval.py

import re
import logging
from typing import Tuple, List, Dict

logger = logging.getLogger(__name__)

# Policy rule definitions
POLICY_RULES = {
    "return_window": {
        "correct_value": "30",
        "wrong_values": ["60", "45", "90", "14", "7", "15"],
        "keywords": ["return", "returns", "return policy", "return period"],
        "unit": "days"
    },
    "refund_time": {
        "correct_value_min": "5",
        "correct_value_max": "7",
        "wrong_values": ["1", "2", "3", "10", "14", "20", "24", "instantly", "immediately", "same day"],
        "keywords": ["refund", "refunds", "money back", "refund time", "refund duration"],
        "unit": "business days"
    },
    "standard_shipping": {
        "correct_value_min": "5",
        "correct_value_max": "7",
        "wrong_values": ["1", "2", "3", "10", "14"],
        "keywords": ["standard shipping", "regular shipping"],
        "unit": "business days"
    },
    "express_shipping": {
        "correct_value_min": "2",
        "correct_value_max": "3",
        "keywords": ["express shipping", "express delivery"],
        "unit": "business days"
    },
    "cancellation_window": {
        "correct_value": "1",
        "wrong_values": ["24", "48", "2", "3"],
        "keywords": ["cancel", "cancellation", "cancel order"],
        "unit": "hour"
    },
    "cancellation_refund": {
        "correct_value_min": "3",
        "correct_value_max": "5",
        "keywords": ["cancelled order refund", "cancel refund"],
        "unit": "business days"
    },
    "electronics_warranty": {
        "correct_value": "1",
        "wrong_values": ["2", "3", "lifetime", "90", "6"],
        "keywords": ["electronics warranty", "phone warranty", "laptop warranty"],
        "unit": "year"
    },
    "clothing_warranty": {
        "correct_value": "90",
        "wrong_values": ["30", "180", "365", "1 year"],
        "keywords": ["clothing warranty", "apparel warranty"],
        "unit": "day"
    },
    "damaged_report_window": {
        "correct_value": "48",
        "wrong_values": ["24", "72", "7", "30"],
        "keywords": ["damaged", "broken", "report damage"],
        "unit": "hours"
    },
    "price_match_window": {
        "correct_value": "7",
        "wrong_values": ["14", "30", "3"],
        "keywords": ["price match", "price matching"],
        "unit": "days"
    },
    "free_shipping_threshold": {
        "correct_value": "50",
        "wrong_values": ["25", "75", "100", "30"],
        "keywords": ["free shipping"],
        "unit": "dollars"
    },
}

NEGATIVE_SIGNALS = ["not", "no", "cannot", "can't", "not eligible", "not accepted",
                    "not possible", "unable", "not allowed", "ineligible", "prohibited",
                    "unfortunately", "not qualify", "does not qualify"]

POSITIVE_SIGNALS = ["can", "yes", "allowed", "eligible", "possible", "accepted",
                    "qualify", "certainly", "absolutely", "of course"]


def extract_numbers(text: str) -> List[str]:
    return re.findall(r'\b\d+(?:\.\d+)?\b', text)


def check_return_policy(response: str) -> Tuple[str, str]:
    r = response.lower()
    details = []

    if any(k in r for k in ["opened", "open product", "open item"]):
        if any(sig in r for sig in NEGATIVE_SIGNALS):
            details.append("Correctly states opened items cannot be returned")
            return "PASS", "; ".join(details)
        else:
            details.append("FAIL: Should state opened items cannot be returned")
            return "FAIL", "; ".join(details)

    if "30" in r and any(k in r for k in ["days", "day"]):
        details.append("Correctly mentions 30-day return window")
        return "PASS", "; ".join(details)

    if any(wrong in r for wrong in ["60 day", "45 day", "90 day", "60-day", "45-day", "90-day"]):
        details.append("FAIL: Incorrect return window stated")
        return "FAIL", "; ".join(details)

    return "PASS", "Return policy check passed"


def check_refund_policy(response: str) -> Tuple[str, str]:
    r = response.lower()

    numbers = extract_numbers(r)

    for num in numbers:
        n = float(num)
        if n in [1, 2, 3, 10, 14, 20, 24, 48]:
            if "refund" in r and ("day" in r or "hour" in r or "business" in r):
                return "FAIL", f"FAIL: Incorrect refund timeframe mentioned: {num}"

    if "5" in numbers or "7" in numbers:
        if "refund" in r and "business" in r:
            return "PASS", "Correctly states 5-7 business day refund time"

    if any(word in r for word in ["immediately", "instant", "same day", "right away", "within 24"]):
        if "refund" in r:
            return "FAIL", "FAIL: Refund is NOT immediate; takes 5-7 business days"

    return "PASS", "Refund policy check passed"


def check_shipping_policy(response: str) -> Tuple[str, str]:
    r = response.lower()

    if "standard" in r:
        numbers = extract_numbers(r)
        for num in numbers:
            n = float(num)
            if n in [1, 2, 3, 4, 10, 14]:
                if "day" in r:
                    return "FAIL", f"FAIL: Standard shipping is 5-7 days, not {num} days"
        if "5" in numbers and "7" in numbers:
            return "PASS", "Correctly states standard shipping 5-7 days"

    if "express" in r:
        numbers = extract_numbers(r)
        for num in numbers:
            n = float(num)
            if n in [1, 5, 7, 10]:
                if "day" in r:
                    return "FAIL", f"FAIL: Express shipping is 2-3 days, not {num} days"

    if "free" in r and "shipping" in r:
        if "50" not in r and "$50" not in r:
            if any(wrong in r for wrong in ["$25", "$75", "$100", "$30", "25 dollar", "75 dollar"]):
                return "FAIL", "FAIL: Free shipping threshold is $50, not the stated amount"

    return "PASS", "Shipping policy check passed"


def check_cancellation_policy(response: str) -> Tuple[str, str]:
    r = response.lower()

    if "shipped" in r or "processing" in r:
        if any(sig in r for sig in NEGATIVE_SIGNALS):
            return "PASS", "Correctly states shipped orders cannot be cancelled"

    numbers = extract_numbers(r)
    for num in numbers:
        n = float(num)
        if "cancel" in r and n in [24, 48, 72]:
            if "hour" in r:
                return "FAIL", f"FAIL: Cancellation window is 1 hour, not {num} hours"

    if "1 hour" in r or "one hour" in r:
        return "PASS", "Correctly states 1-hour cancellation window"

    return "PASS", "Cancellation policy check passed"


def check_warranty_policy(response: str) -> Tuple[str, str]:
    r = response.lower()

    if "electronic" in r or "phone" in r or "laptop" in r or "device" in r:
        if "lifetime" in r:
            return "FAIL", "FAIL: Electronics warranty is 1 year, NOT lifetime"
        if "1 year" in r or "one year" in r or "1-year" in r:
            return "PASS", "Correctly states 1-year electronics warranty"

    if "cloth" in r or "apparel" in r or "fabric" in r:
        if "90" in r and "day" in r:
            return "PASS", "Correctly states 90-day clothing warranty"
        if "1 year" in r and "cloth" in r:
            return "FAIL", "FAIL: Clothing warranty is 90 days, not 1 year"

    if "water damage" in r or "accidental" in r:
        if any(sig in r for sig in NEGATIVE_SIGNALS):
            return "PASS", "Correctly states water/accidental damage is not covered"
        else:
            return "FAIL", "FAIL: Water/accidental damage is NOT covered by warranty"

    return "PASS", "Warranty policy check passed"


def check_damaged_product_policy(response: str) -> Tuple[str, str]:
    r = response.lower()

    if "damaged" in r or "broken" in r:
        numbers = extract_numbers(r)
        for num in numbers:
            n = float(num)
            if "hour" in r and n != 48 and n in [24, 72, 96]:
                return "FAIL", f"FAIL: Damaged products must be reported within 48 hours, not {num}"

        if "48" in numbers:
            return "PASS", "Correctly states 48-hour damage reporting window"

    return "PASS", "Damaged product policy check passed"


def check_hallucination_signals(response: str) -> Tuple[bool, List[str]]:
    r = response.lower()
    hallucinations = []

    # Check for invented policies
    if "free return" in r and "change of mind" not in r and "defective" not in r:
        if "always" in r or "all returns" in r:
            hallucinations.append("Claimed all returns have free shipping (incorrect)")

    if "lifetime warranty" in r or "lifetime guarantee" in r:
        hallucinations.append("Claimed lifetime warranty (policy says 1-year max)")

    if "24/7" in r and "support" in r:
        if "we offer" in r or "available" in r:
            pass  # Plausible but not in policy - flag as unverified

    if "30%" in r or "50%" in r:
        if "discount" in r or "off" in r:
            hallucinations.append("Mentioned discount percentages not in policy")

    if "7 days" in r and "refund" in r and "business" not in r:
        hallucinations.append("Mentioned 7-day refund without specifying 'business days'")

    if "price match" in r:
        if "14 day" in r or "30 day" in r:
            hallucinations.append("Incorrect price match window - should be 7 days")

    return len(hallucinations) > 0, hallucinations


def code_eval(query: str, response: str, category: str = None) -> Tuple[str, str, List[str]]:
    checks_run = []
    all_details = []
    failures = []

    q_lower = query.lower()
    r_lower = response.lower()

    # Determine which checks to run based on query + response content
    run_return = any(k in q_lower for k in ["return", "send back", "returns"])
    run_refund = any(k in q_lower for k in ["refund", "money back", "refunded"])
    run_shipping = any(k in q_lower for k in ["shipping", "delivery", "ship", "deliver"])
    run_cancel = any(k in q_lower for k in ["cancel", "cancellation"])
    run_warranty = any(k in q_lower for k in ["warranty", "guarantee", "defective"])
    run_damaged = any(k in q_lower for k in ["damaged", "broken", "arrived broken"])

    if run_return:
        result, detail = check_return_policy(response)
        checks_run.append("return_policy")
        all_details.append(f"Return check: {detail}")
        if result == "FAIL":
            failures.append(detail)

    if run_refund:
        result, detail = check_refund_policy(response)
        checks_run.append("refund_policy")
        all_details.append(f"Refund check: {detail}")
        if result == "FAIL":
            failures.append(detail)

    if run_shipping:
        result, detail = check_shipping_policy(response)
        checks_run.append("shipping_policy")
        all_details.append(f"Shipping check: {detail}")
        if result == "FAIL":
            failures.append(detail)

    if run_cancel:
        result, detail = check_cancellation_policy(response)
        checks_run.append("cancellation_policy")
        all_details.append(f"Cancellation check: {detail}")
        if result == "FAIL":
            failures.append(detail)

    if run_warranty:
        result, detail = check_warranty_policy(response)
        checks_run.append("warranty_policy")
        all_details.append(f"Warranty check: {detail}")
        if result == "FAIL":
            failures.append(detail)

    if run_damaged:
        result, detail = check_damaged_product_policy(response)
        checks_run.append("damaged_policy")
        all_details.append(f"Damaged product check: {detail}")
        if result == "FAIL":
            failures.append(detail)

    # Always run hallucination signal check
    hallucinated, hall_list = check_hallucination_signals(response)
    if hallucinated:
        for h in hall_list:
            failures.append(f"Hallucination signal: {h}")
        all_details.extend([f"Hallucination: {h}" for h in hall_list])

    if not checks_run:
        return "PASS", "No specific policy checks applicable", []

    final = "FAIL" if failures else "PASS"
    return final, " | ".join(all_details), failures