AI Automation with Python

import os
from dataclasses import dataclass
from enum import StrEnum
from typing import Literal

from openai import OpenAI
from pydantic import BaseModel, Field


MODEL = os.getenv("OPENAI_MODEL", "gpt-5.6")

client = OpenAI()


class TicketAnalysis(BaseModel):
    summary: str = Field(
        description="A concise description of the customer issue."
    )

    category: Literal[
        "access",
        "billing",
        "performance",
        "security",
        "other",
    ]

    priority: Literal[
        "low",
        "medium",
        "high",
        "critical",
    ]

    sentiment: Literal[
        "positive",
        "neutral",
        "frustrated",
        "angry",
    ]

    recommended_team: Literal[
        "identity-support",
        "billing-support",
        "application-support",
        "security-operations",
        "general-support",
    ]

    requires_human_review: bool
    suggested_next_action: str


class WorkflowStatus(StrEnum):
    RECEIVED = "received"
    PROCESSING = "processing"
    CLASSIFIED = "classified"
    VALIDATED = "validated"
    HUMAN_REVIEW = "human_review"
    ROUTED = "routed"
    COMPLETED = "completed"
    FAILED = "failed"


@dataclass
class WorkflowResult:
    status: WorkflowStatus
    analysis: TicketAnalysis | None = None
    message: str | None = None


TEAM_BY_CATEGORY = {
    "access": "identity-support",
    "billing": "billing-support",
    "performance": "application-support",
    "security": "security-operations",
    "other": "general-support",
}


def analyse_ticket(ticket_text: str) -> TicketAnalysis:
    cleaned_text = ticket_text.strip()

    if not cleaned_text:
        raise ValueError("ticket_text cannot be empty")

    response = client.responses.parse(
        model=MODEL,
        input=[
            {
                "role": "system",
                "content": (
                    "You analyse incoming software support tickets. "
                    "Classify only from the allowed schema values. "
                    "Use critical priority only for severe business impact, "
                    "security incidents, or complete service unavailability. "
                    "Set requires_human_review to true for security issues, "
                    "critical priority, unclear requests, or sensitive cases. "
                    "Do not invent facts that are not present in the ticket."
                ),
            },
            {
                "role": "user",
                "content": cleaned_text,
            },
        ],
        text_format=TicketAnalysis,
    )

    result = response.output_parsed

    if result is None:
        raise RuntimeError(
            "The model did not return a structured ticket analysis."
        )

    return result


def validate_business_rules(
    analysis: TicketAnalysis,
) -> TicketAnalysis:
    expected_team = TEAM_BY_CATEGORY[analysis.category]

    if analysis.recommended_team != expected_team:
        analysis.recommended_team = expected_team

    if analysis.category == "security":
        analysis.priority = "critical"
        analysis.requires_human_review = True

    if analysis.priority == "critical":
        analysis.requires_human_review = True

    return analysis


def requires_escalation(
    analysis: TicketAnalysis,
) -> bool:
    return (
        analysis.priority == "critical"
        or analysis.category == "security"
        or analysis.requires_human_review
    )


def route_ticket(
    analysis: TicketAnalysis,
) -> WorkflowResult:
    if requires_escalation(analysis):
        return WorkflowResult(
            status=WorkflowStatus.HUMAN_REVIEW,
            analysis=analysis,
            message=(
                "Ticket requires human review before "
                "creating or updating an external record."
            ),
        )

    print(
        f"Routing ticket to {analysis.recommended_team}"
    )

    return WorkflowResult(
        status=WorkflowStatus.COMPLETED,
        analysis=analysis,
        message="Ticket routed successfully.",
    )


def process_ticket(
    ticket_text: str,
) -> WorkflowResult:
    try:
        analysis = analyse_ticket(ticket_text)
        validated = validate_business_rules(analysis)
        return route_ticket(validated)

    except Exception as exc:
        return WorkflowResult(
            status=WorkflowStatus.FAILED,
            message=str(exc),
        )


if __name__ == "__main__":
    ticket = """
    Since this morning, nobody in our finance department
    can access the dashboard after resetting their passwords.
    The browser shows an authentication error.
    This is blocking our month-end reporting.
    """

    result = process_ticket(ticket)

    print(f"Status: {result.status}")

    if result.analysis:
        print(
            result.analysis.model_dump_json(indent=2)
        )

    if result.message:
        print(result.message)