app.py

import os
import re
import ast
import sys
import json
import time
from collections import deque

from dotenv import load_dotenv
from google import genai
from google.genai import types
from google.genai.errors import APIError

load_dotenv()

from ui_module import *

# ---------------------------------------------------------------------------
# Initialization: 
# 1. Set up the Google API client
# 2. Define the response schema
# ---------------------------------------------------------------------------
GOOGLE_API_KEY = os.environ.get("GOOGLE_API_KEY")
if not GOOGLE_API_KEY:
    sys.exit("API Key not found! Please configure GOOGLE_API_KEY in Settings → Variables and secrets.")
client = genai.Client(api_key=GOOGLE_API_KEY)
LLM_MODEL = os.environ.get("LLM_MODEL")

# ---------------------------------------------------------------------------
# MODULE 5 - Output Schema Definition
# ---------------------------------------------------------------------------
response_schema = types.Schema(
    type=types.Type.OBJECT,
    required=["result", "accuracy", "summary", "issues"],
    properties={

        # → VERDICT card: "Pass" or "Fail"
        "result": types.Schema(
            type=types.Type.STRING,
            enum=["Pass", "Fail"]
        ),

        # → ACCURACY card: 0–100 integer (renders as "96%")
        "accuracy": types.Schema(
            type=types.Type.INTEGER,
        ),

        # → SUMMARY card: short prose explanation
        "summary": types.Schema(
            type=types.Type.STRING,
        ),

        # → ISSUES DETECTED list: each bullet point
        "issues": types.Schema(
            type=types.Type.STRING,
        )
    }
)

# ---------------------------------------------------------------------------
# MODULE 2 — Validation & Flow Management Module
# ---------------------------------------------------------------------------
MIN_DESCRIPTION_CHARS = 20
MAX_DESCRIPTION_CHARS = 3000
MIN_CODE_CHARS        = 10
MAX_CODE_CHARS        = 8000
MAX_CODE_LINES        = 300

FORBIDDEN_PATTERNS = [
    r"ignore (all |previous |above )?instructions",
    r"disregard (all |previous |above )?instructions",
    r"you are now",
    r"act as (a |an )?",
    r"<\s*(script|iframe|object|embed)",
    r"system\s*prompt",
    r"jailbreak",
]

PYTHON_KEYWORDS = {
    "def", "class", "import", "from", "return", "if", "else", "elif",
    "for", "while", "try", "except", "with", "lambda", "yield", "pass",
    "raise", "assert", "in", "not", "and", "or", "True", "False", "None",
    "print", "len", "range", "self",
}

class RateLimiter:
    def __init__(self, max_calls: int = 5, window_seconds: int = 60):
        self.max_calls      = max_calls
        self.window_seconds = window_seconds
        self._timestamps: deque = deque()

    def is_allowed(self) -> tuple[bool, str]:
        now = time.time()
        while self._timestamps and now - self._timestamps[0] > self.window_seconds:
            self._timestamps.popleft()
        if len(self._timestamps) >= self.max_calls:
            wait = int(self.window_seconds - (now - self._timestamps[0])) + 1
            return False, (
                f"⏳ Rate limit reached — {self.max_calls} requests in "
                f"{self.window_seconds}s. Please wait ~{wait}s and try again."
            )
        self._timestamps.append(now)
        return True, ""

_rate_limiter = RateLimiter(max_calls=5, window_seconds=60)

def _check_forbidden(text: str) -> str | None:
    lower = text.lower()
    for pattern in FORBIDDEN_PATTERNS:
        if re.search(pattern, lower):
            return (
                "Input contains disallowed content. "
                "Please remove prompt-injection or HTML patterns and try again."
            )
    return None

def _looks_like_python(code: str) -> tuple[bool, str]:
    tokens = set(re.findall(r"[A-Za-z_]\w*", code))
    if not tokens.intersection(PYTHON_KEYWORDS):
        return False, (
            "🐍 The code doesn't appear to be Python — no recognisable Python "
            "keywords found (e.g. def, class, import, return). "
            "Please submit Python code only."
        )
    try:
        ast.parse(code)
    except SyntaxError as exc:
        line_hint = f" (line {exc.lineno})" if exc.lineno else ""
        return False, (
            f"🐍 Python syntax error{line_hint}: {exc.msg}. "
            "Please fix the syntax error before evaluating."
        )
    return True, ""

def validate_inputs(description: str, code: str) -> list[str]:
    errors: list[str] = []

    if not description or not description.strip():
        errors.append("📋 Requirements description is required.")
    if not code or not code.strip():
        errors.append("🐍 Python code is required.")
    if errors:
        return errors

    desc, code_ = description.strip(), code.strip()

    if len(desc) < MIN_DESCRIPTION_CHARS:
        errors.append(f"📋 Description too short ({len(desc)} chars) — minimum is {MIN_DESCRIPTION_CHARS} characters.")
    if len(code_) < MIN_CODE_CHARS:
        errors.append(f"🐍 Code too short ({len(code_)} chars) — minimum is {MIN_CODE_CHARS} characters.")
    if len(desc) > MAX_DESCRIPTION_CHARS:
        errors.append(f"📋 Description too long ({len(desc):,} chars) — max is {MAX_DESCRIPTION_CHARS:,} characters.")
    if len(code_) > MAX_CODE_CHARS:
        errors.append(f"🐍 Code too long ({len(code_):,} chars) — max is {MAX_CODE_CHARS:,} characters.")
    if len(code_.splitlines()) > MAX_CODE_LINES:
        errors.append(f"🐍 Code has too many lines ({len(code_.splitlines())}) — max is {MAX_CODE_LINES} lines.")

    if err := _check_forbidden(desc):
        errors.append(f"📋 {err}")
    if err := _check_forbidden(code_):
        errors.append(f"🐍 {err}")

    if not errors:
        is_python, py_error = _looks_like_python(code_)
        if not is_python:
            errors.append(py_error)

    if not errors:
        allowed, rate_msg = _rate_limiter.is_allowed()
        if not allowed:
            errors.append(rate_msg)

    return errors

# ---------------------------------------------------------------------------
# Format the evaluation results for display
# ---------------------------------------------------------------------------- 
def format_for_display(raw_output: dict) -> tuple[str, str, str]:
    emoji   = "✅" if raw_output["result"].upper() == "PASS" else ("❌" if raw_output["result"].upper() == "FAIL" else "⚠️")
    verdict = f"{emoji}  {raw_output['result']}"
    acc_str = f"{raw_output['accuracy']}%" if raw_output["accuracy"] >= 0 else "N/A"
    metrics = f"Accuracy: {acc_str}\n\nSummary: {raw_output['summary']}"
    issues_text = raw_output["issues"]
    return verdict, metrics, issues_text

# ---------------------------------------------------------------------------
# Core function
# ----------------------------------------------------------------------------
def validate_and_evaluate(description: str, code: str):
    errors = validate_inputs(description, code)
    if errors:
        return "", "", "", "\n".join(f"{i+1}. {e}" for i, e in enumerate(errors))
    prompt = build_prompt(description, code)
    try:
        raw_output = generate_response(prompt)
    except APIError as exc:
        # 1. If it's a 500 error (and the retries failed), show the friendly timeout message
        if exc.code == 500:
            timeout_msg = (
                "❌ API Connection Timeout: The upstream AI provider is currently overloaded "
                "or unresponsive. Please try again in a few moments."
            )
            return "", "", "", timeout_msg
            
        # 2. If it's a different API error (e.g., 400 Bad Request, 403 Invalid Key)
        return "", "", "", f"❌ Upstream API Error ({exc.code}): {exc}"
        
    except Exception as exc:
        # 3. Catch any local application bugs or JSON parsing failures here
        return "", "", "", f"❌ Internal Application Error: {exc}"

    verdict, metrics, issues = format_for_display(raw_output)
    return verdict, metrics, issues, ""

# ---------------------------------------------------------------------------
# MODULE 3 — Build the prompt for code evaluation
# ---------------------------------------------------------------------------
def build_prompt(description, code):
    prompt = f"""
#ROLE
You are a Python code reviewer. Your goal is to determine if the provided Python code strictly complies with the requirements mentioned by the user.
 
#CONTEXT
Developers may create code that does not entirely meet the specified requirements. You will examine the relationship between a natural-language description and a Python code sample and create a structured evaluation report.
 
#INPUT DATA
Description: {description}
Code to review:
```python
{code}
```
 
#TASK
1. Carefully read the requirements description.
2. Examine each line of the Python code.
3. Determine whether the code fulfils ALL requirements stated in the description.
4. Estimate the accuracy in percentages to assess how accurately the code matches the description.
5. List specific requirements that are either absent or incorrectly applied.
6. Base your evaluation solely on static code analysis and logical reasoning.
 
#CONSTRAINTS
1. Do not execute the code.
2. The output must be in the specified format.
3. The output should be clear and concise.
 
#OUTPUT FORMAT (to be strictly followed without any changes)
- Result: either "Pass" or "Fail" (Pass if the code is functional and matches the description, Fail if it has bugs or doesn't match).
- Accuracy: an integer from 0 to 100 representing how well the code matches the description and is bug-free.
- Summary: a short prose explanation of the result (e.g. "Code correctly handles all described requirements including edge cases.").    
- Issues: The value should be a single string containing a numbered list. For each item, include the severity (Error, Warning, or Info) followed by a dash and description of the issue (e.g. "Missing type hints on function signature"). List each issue on a new line starting with a number. Use newline characters (\n) to separate each line.
...
 
"""
    return prompt


# ---------------------------------------------------------------------------
# MODULE 4 — Generation Module
# Responsible for communicating with Gemini 2.5 Flash and generating
# ---------------------------------------------------------------------------
def generate_response(prompt: str, retries: int = 3, initial_delay: float = 1.5) -> dict:
    delay = initial_delay
    
    for i in range(retries):
        try:
            response = client.models.generate_content(
                model=LLM_MODEL, 
                contents=prompt,
                config=types.GenerateContentConfig(
                    response_mime_type="application/json",
                    response_schema=response_schema, 
                )
            )
            # If successful, parse the JSON and return immediately
            print("✅ Successful response from Gemini API.")
            return json.loads(response.text)
            
        except APIError as exc:
            # Catch transient 500 Internal Server Errors and retry
            if exc.code == 500 and i <= retries - 1:
                print(f"⚠️ Gemini API 500 error. Retrying attempt {i + 1}/{retries} in {delay}s...")
                time.sleep(delay)
                delay *= 2  # Exponential backoff
                continue
            
            # If out of retries, or if it's a 400/403 error, raise it up to validate_and_evaluate
            raise exc
            
        except json.JSONDecodeError as exc:
            # Failsafe: In rare cases, if the API drops a malformed payload, catch the JSON error
            raise RuntimeError(f"API returned invalid JSON format: {exc}")

# Build the Gradio UI and connect it to the validation and evaluation function
app = build_ui(validate_and_evaluate)

# Launch the web application
app.launch(css=CUSTOM_CSS)