scope_router.py

import re
from enum import Enum
from typing import Optional

from schemas import CodeXRequest


class ScopeRoute(str, Enum):
    CODE = "code"
    GREETING = "greeting"
    NON_CODE = "non_code"
    IMAGE = "image"
    RESTRICTED = "restricted"
    UNKNOWN = "unknown"


class ScopeDecision:
    def __init__(
        self,
        route: ScopeRoute,
        message: str,
        should_continue_to_codex: bool,
    ):
        self.route = route
        self.message = message
        self.should_continue_to_codex = should_continue_to_codex


GREETING_PATTERNS = [
    r"\bhi\b",
    r"\bhello\b",
    r"\bhey\b",
    r"\bgood morning\b",
    r"\bgood afternoon\b",
    r"\bgood evening\b",
    r"\bhow are you\b",
    r"\bwho are you\b",
    r"\bwhat can you do\b",
]

CODE_PATTERNS = [
    r"\bcode\b",
    r"\bfunction\b",
    r"\bclass\b",
    r"\bmethod\b",
    r"\bvariable\b",
    r"\bbug\b",
    r"\berror\b",
    r"\bexception\b",
    r"\bfix\b",
    r"\brefactor\b",
    r"\breview\b",
    r"\boptimize\b",
    r"\bdebug\b",
    r"\bpython\b",
    r"\bjava\b",
    r"\bjavascript\b",
    r"\btypescript\b",
    r"\bflutter\b",
    r"\bdart\b",
    r"\breact\b",
    r"\bnode\b",
    r"\bapi\b",
    r"\bsql\b",
    r"\bhtml\b",
    r"\bcss\b",
    r"\bjson\b",
    r"\bregex\b",
    r"\balgorithm\b",
    r"\bcompile\b",
    r"\bruntime\b",
    r"\bsyntax\b",
    r"\bprogram\b",
    r"\bscript\b",
    r"\bsoftware\b",
    r"\bbackend\b",
    r"\bfrontend\b",
    r"\bendpoint\b",
    r"\bdatabase\b",
    r"\bquery\b",
    r"\btraceback\b",
    r"\bdocstring\b",
    r"\btype hints?\b",
    r"\bvalidation\b",
    r"\berror handling\b",
    r"\blogging\b",
    r"\bcomments\b",
    r"\bexception handling\b",
]

CODE_MODIFICATION_PATTERNS = [
    r"\badd\b",
    r"\bupdate\b",
    r"\bmodify\b",
    r"\bchange\b",
    r"\brename\b",
    r"\bconvert\b",
    r"\bextend\b",
    r"\binclude\b",
    r"\buse\b",
    r"\breplace\b",
    r"\bremove\b",
    r"\bdelete\b",
    r"\bdocstring\b",
    r"\btype hints?\b",
    r"\bvalidation\b",
    r"\berror handling\b",
    r"\blogging\b",
    r"\bcomments\b",
    r"\bexception handling\b",
    r"\bmake it\b",
    r"\bturn this into\b",
    r"\badd support for\b",
    r"\bnow add\b",
    r"\bnow update\b",
    r"\bnow change\b",
]

IMAGE_PATTERNS = [
    r"\bgenerate an image\b",
    r"\bcreate an image\b",
    r"\bdraw\b",
    r"\bposter\b",
    r"\blogo\b",
    r"\bbanner\b",
    r"\bphoto\b",
    r"\bpicture\b",
    r"\bimage\b",
    r"\bdescribe this image\b",
    r"\bwhat is in this image\b",
    r"\bedit this image\b",
]

NON_CODE_PATTERNS = [
    r"\bweather\b",
    r"\bpresident\b",
    r"\bprime minister\b",
    r"\bnews\b",
    r"\bgold price\b",
    r"\bstock\b",
    r"\brecipe\b",
    r"\btravel\b",
    r"\bhotel\b",
    r"\brestaurant\b",
    r"\bmedical\b",
    r"\blegal\b",
    r"\bmovie\b",
    r"\bsong\b",
    r"\btranslate\b",
    r"\bessay\b",
    r"\bpoem\b",
]

RESTRICTED_PATTERNS = [
    r"\bmalware\b",
    r"\bransomware\b",
    r"\bkeylogger\b",
    r"\bsteal passwords\b",
    r"\bphishing\b",
    r"\btoken grabber\b",
    r"\bremote access trojan\b",
    r"\brat\b",
    r"\bcredential stuffing\b",
    r"\bexploit\b",
    r"\bbypass authentication\b",
    r"\bddos\b",
    r"\bbruteforce\b",
    r"\bsql injection\b",
    r"\bxss exploit\b",
    r"\bvirus\b",
    r"\bspyware\b",
]


def normalize_text(text: Optional[str]) -> str:
    if not text:
        return ""
    text = str(text).strip().lower()
    text = re.sub(r"\s+", " ", text)
    return text


def contains_pattern(text: str, patterns: list[str]) -> bool:
    return any(re.search(pattern, text) for pattern in patterns)


def looks_like_code_block(text: str) -> bool:
    if not text:
        return False

    code_signals = [
        "def ",
        "class ",
        "return ",
        "import ",
        "from ",
        "{",
        "}",
        ";",
        "=>",
        "public ",
        "private ",
        "protected ",
        "const ",
        "let ",
        "var ",
        "function ",
        "if (",
        "for (",
        "while (",
        "</",
        "try:",
        "except:",
    ]

    lowered = text.lower()
    return any(signal.lower() in lowered for signal in code_signals)


def is_follow_up_code_request(request: CodeXRequest) -> bool:
    message = normalize_text(request.message)

    has_session = bool(request.session_id and str(request.session_id).strip())
    has_previous_context = bool(request.previous_context and str(request.previous_context).strip())

    if not (has_session or has_previous_context):
        return False

    if contains_pattern(message, CODE_MODIFICATION_PATTERNS):
        return True

    short_follow_up_phrases = [
        "now add",
        "now update",
        "now change",
        "now remove",
        "now use",
        "add this",
        "update this",
        "change this",
        "fix this",
        "refactor this",
        "review this",
    ]

    return any(phrase in message for phrase in short_follow_up_phrases)


def is_code_related_request(request: CodeXRequest) -> bool:
    message = normalize_text(request.message)
    code = request.code or ""
    error_message = normalize_text(request.error_message)

    if request.code and request.code.strip():
        return True

    if error_message:
        return True

    if looks_like_code_block(request.message):
        return True

    if looks_like_code_block(code):
        return True

    if contains_pattern(message, CODE_PATTERNS):
        return True

    if is_follow_up_code_request(request):
        return True

    return False


def is_greeting_request(request: CodeXRequest) -> bool:
    message = normalize_text(request.message)

    if not message:
        return False

    if contains_pattern(message, GREETING_PATTERNS):
        if not is_code_related_request(request):
            return True

    return False


def is_image_request(request: CodeXRequest) -> bool:
    message = normalize_text(request.message)
    return contains_pattern(message, IMAGE_PATTERNS)


def is_non_code_request(request: CodeXRequest) -> bool:
    message = normalize_text(request.message)

    if is_code_related_request(request):
        return False

    if contains_pattern(message, NON_CODE_PATTERNS):
        return True

    return False


def is_restricted_request(request: CodeXRequest) -> bool:
    message = normalize_text(request.message)
    code = normalize_text(request.code)
    error_message = normalize_text(request.error_message)

    combined = " ".join(part for part in [message, code, error_message] if part)
    return contains_pattern(combined, RESTRICTED_PATTERNS)


def decide_scope(request: CodeXRequest) -> ScopeDecision:
    if is_restricted_request(request):
        return ScopeDecision(
            route=ScopeRoute.RESTRICTED,
            message="Code X cannot help with harmful, abusive, or clearly unsafe coding requests.",
            should_continue_to_codex=False,
        )

    if is_image_request(request):
        return ScopeDecision(
            route=ScopeRoute.IMAGE,
            message="Code X is focused on coding tasks. Image generation, image editing, and picture analysis should be handled by the image workflow instead.",
            should_continue_to_codex=False,
        )

    if is_greeting_request(request):
        return ScopeDecision(
            route=ScopeRoute.GREETING,
            message="Hello. Code X is ready for coding tasks. Send code, an error, or a software request and I will help.",
            should_continue_to_codex=False,
        )

    if is_non_code_request(request):
        return ScopeDecision(
            route=ScopeRoute.NON_CODE,
            message="Code X is focused on software and coding tasks. This request looks outside Code X scope, so it should go to the normal chat assistant.",
            should_continue_to_codex=False,
        )

    if is_code_related_request(request):
        return ScopeDecision(
            route=ScopeRoute.CODE,
            message="Code-related request detected.",
            should_continue_to_codex=True,
        )

    return ScopeDecision(
        route=ScopeRoute.UNKNOWN,
        message="This request is not clearly a coding task. Send code, an error, or a software development request for Code X.",
        should_continue_to_codex=False,
    )