HELPDESK.AI
Next-Generation IT Ticket Inference Engine
System Online • v1.0.0
Powered by FastAPI & Hugging Face Transformers
"""
@app.get("/health", response_model=HealthResponse)
async def health_check():
return HealthResponse(
status="ok",
classifier_loaded=classifier_service._loaded,
ner_loaded=ner_service._loaded,
)
@app.get("/ready", response_model=ReadinessResponse)
async def readiness_check():
require_supabase = os.environ.get("REQUIRE_SUPABASE", "false").lower() == "true"
allow_degraded = os.environ.get("ALLOW_DEGRADED_STARTUP", "0") == "1"
checks = {
"api": True,
"classifier_loaded": classifier_service._loaded,
"ner_loaded": ner_service._loaded,
"duplicate_index_loaded": duplicate_service.is_available(),
"rag_loaded": rag_service.is_available(),
}
if require_supabase:
checks["supabase_configured"] = supabase is not None
# In degraded mode, duplicate and RAG services are optional
if allow_degraded:
required_checks = {k: v for k, v in checks.items() if k not in ["duplicate_index_loaded", "rag_loaded"]}
all_required_pass = all(required_checks.values())
if all_required_pass:
return ReadinessResponse(status="ready", checks=checks)
else:
# Strict mode: all checks must pass
if all(checks.values()):
return ReadinessResponse(status="ready", checks=checks)
return JSONResponse(
status_code=503,
content=jsonable_encoder(ReadinessResponse(status="not_ready", checks=checks)),
)
class TroubleshootRequest(BaseModel):
text: str
category: str
history: list[dict] = []
class TroubleshootResponse(BaseModel):
step_text: str
options: list[str]
is_final: bool
@app.post("/ai/troubleshoot", response_model=TroubleshootResponse)
async def troubleshoot(request: TroubleshootRequest):
"""Get dynamic troubleshooting steps from Gemini."""
if not gemini_service or not gemini_service._initialized:
return TroubleshootResponse(
step_text="AI Troubleshooting is currently unavailable.",
options=["Continue to tracking"],
is_final=True
)
result = gemini_service.get_troubleshooting_step(
request.text,
request.history,
request.category
)
return TroubleshootResponse(**result)
class BugReportAnalysisRequest(BaseModel):
bug_title: str
description: str
steps_to_reproduce: str = ""
console_errors: list[str] = []
class BugReportAnalysisResponse(BaseModel):
probable_cause: str
@app.post("/ai/analyze_bug", response_model=BugReportAnalysisResponse)
async def analyze_bug(request: BugReportAnalysisRequest):
"""Analyze a bug report using Gemini to generate a Probable Cause."""
if not gemini_service or not gemini_service._initialized:
return BugReportAnalysisResponse(
probable_cause="AI Diagnostics are currently unavailable."
)
cause = gemini_service.analyze_bug_report(
request.bug_title,
request.description,
request.steps_to_reproduce,
request.console_errors
)
return BugReportAnalysisResponse(probable_cause=cause)
# ---------------------------------------------------------------------------
# Admin Correction Logging endpoint
# ---------------------------------------------------------------------------
CORRECTIONS_LOG_PATH = Path(__file__).parent / "data" / "corrections_log.json"
@app.post("/ai/log_correction")
async def log_correction(raw_request: Request):
"""Log an admin correction when the AI prediction differs from the human decision."""
try:
body = await raw_request.json()
except Exception as e:
print(f"[CORRECTION ERROR] Could not parse request body: {e}")
return {"status": "error", "message": "Invalid JSON body"}
print(f"[CORRECTION RECEIVED] Payload keys: {list(body.keys())}")
ticket_id = str(body.get("ticket_id", "unknown"))
original_text = str(body.get("original_text", ""))
ocr_text = str(body.get("ocr_text", ""))
confidence = float(body.get("confidence") or 0.0)
original_prediction = body.get("original_prediction") or {}
corrected_prediction = body.get("corrected_prediction") or {}
# Only log if something actually changed
changed_fields = [
field for field in ["category", "subcategory", "priority", "assigned_team"]
if original_prediction.get(field) != corrected_prediction.get(field)
]
if not changed_fields:
return {"status": "no_change", "message": "Prediction matches correction, nothing logged."}
entry = {
"ticket_id": ticket_id,
"original_text": original_text,
"ocr_text": ocr_text,
"original_prediction": original_prediction,
"corrected_prediction": corrected_prediction,
"changed_fields": changed_fields,
"confidence": confidence,
"timestamp": datetime.datetime.utcnow().isoformat() + "Z"
}
try:
if CORRECTIONS_LOG_PATH.exists() and CORRECTIONS_LOG_PATH.stat().st_size > 2:
with open(CORRECTIONS_LOG_PATH, "r", encoding="utf-8") as f:
logs = json.load(f)
else:
logs = []
logs.append(entry)
with open(CORRECTIONS_LOG_PATH, "w", encoding="utf-8") as f:
json.dump(logs, f, indent=2)
print(f"[CORRECTION SAVED] Ticket ID: {ticket_id} | Changed: {changed_fields}")
return {"status": "saved", "changed_fields": changed_fields}
except Exception as e:
print(f"[CORRECTION ERROR] Could not save: {e}")
return {"status": "error", "message": str(e)}
# ---------------------------------------------------------------------------
# Ticket operations (Now via Supabase)
# ---------------------------------------------------------------------------
@app.get("/tickets")
async def get_tickets(company_id: str | None = None):
"""Fetch persistent tickets from Supabase."""
if not supabase:
raise HTTPException(status_code=500, detail="Database connection not initialized")
query = supabase.table("tickets").select("*").order("created_at", desc=True)
if company_id:
query = query.eq("company_id", company_id)
res = query.execute()
return res.data
@app.get("/tickets/search")
async def search_tickets(q: str | None = None, company_id: str | None = None, limit: int = 50, offset: int = 0):
"""Search tickets using tenant-safe full-text search."""
if not supabase:
raise HTTPException(status_code=500, detail="Database connection not initialized")
if not q:
raise HTTPException(status_code=400, detail="Search query is required")
if not company_id:
raise HTTPException(status_code=400, detail="company_id is required for tenant-safe search")
try:
result = supabase.rpc(
"search_tickets",
{
"query_text": q,
"company_id": company_id,
"limit_rows": limit,
"offset_rows": offset,
},
).execute()
return result.data or []
except Exception as e:
raise HTTPException(status_code=500, detail=f"Search failed: {e}")
@app.post("/tickets/save")
async def save_ticket(request_body: TicketSaveRequest):
"""
OFFICIAL PERSISTENCE: Saves the analyzed ticket to Supabase.
This is called AFTER the user confirms the analysis results.
"""
if not supabase:
raise HTTPException(status_code=500, detail="Supabase connection not initialized.")
logger = logging.getLogger(__name__)
try:
final_data = request_body.dict()
# Resolve tenant linkage from user profile with authorization validation.
profile = {}
if request_body.user_id:
try:
profile_res = (
supabase.table("profiles")
.select("company_id, company")
.eq("id", request_body.user_id)
.single()
.execute()
)
profile = profile_res.data or {}
if not profile:
raise HTTPException(status_code=404, detail="User profile not found")
# SELF-HEALING: If company_id is null in database but company name exists, resolve it!
if not profile.get("company_id") and profile.get("company"):
try:
comp_name = profile.get("company").strip()
comp_res = (
supabase.table("companies")
.select("id")
.ilike("name", comp_name)
.execute()
)
if comp_res.data:
resolved_company_id = comp_res.data[0]["id"]
# Backfill the profile table in real-time
supabase.table("profiles").update({"company_id": resolved_company_id}).eq("id", request_body.user_id).execute()
profile["company_id"] = resolved_company_id
logger.info(f"[SELF-HEALING] Backfilled company_id={resolved_company_id} for user={request_body.user_id}")
except Exception as healing_err:
logger.warning(f"[SELF-HEALING WARNING] Failed to backfill company_id: {healing_err}")
except HTTPException:
raise
except Exception as profile_error:
user_hash = hashlib.sha256(str(request_body.user_id).encode()).hexdigest()[:8]
logger.error(f"Tenant resolution error for user {user_hash}: {profile_error}")
raise HTTPException(status_code=503, detail="Failed to resolve tenant linkage") from profile_error
# Validate tenant consistency and authorization.
profile_company_id = profile.get("company_id")
if final_data.get("company_id"):
# User provided company_id: verify it matches their profile.
if profile_company_id and final_data["company_id"] != profile_company_id:
user_hash = hashlib.sha256(str(request_body.user_id).encode()).hexdigest()[:8]
logger.warning(f"Tenant mismatch: user {user_hash} attempted {final_data['company_id']}, assigned to {profile_company_id}")
raise HTTPException(status_code=403, detail="User not authorized for this tenant")
elif profile_company_id:
# Backfill company_id from profile.
final_data["company_id"] = profile_company_id
elif request_body.user_id:
# User has no tenant assignment.
raise HTTPException(status_code=400, detail="User has no tenant assignment")
# Backfill company name if missing.
if not final_data.get("company") and profile.get("company"):
final_data["company"] = profile["company"]
priority = final_data.get("priority")
if not final_data.get("sla_response_due_at"):
final_data["sla_response_due_at"] = calculate_sla_response_at(priority).isoformat().replace("+00:00", "Z")
if not final_data.get("sla_breach_at"):
final_data["sla_breach_at"] = calculate_sla_breach_at(priority).isoformat().replace("+00:00", "Z")
final_data["sla_status"] = final_data.get("sla_status") or classify_sla_status(final_data.get("sla_breach_at"))
final_data["escalation_level"] = int(final_data.get("escalation_level") or 0)
import hashlib
user_hash = hashlib.sha256(str(request_body.user_id).encode()).hexdigest()[:8]
logger.info(f"Tenant linkage: user_hash={user_hash}, company_id={final_data.get('company_id')}")
duplicate_text = (request_body.description or "").strip() or (request_body.subject or "").strip()
duplicate_threshold = get_duplicate_threshold(final_data.get("company_id"), 0.85)
duplicate_result = {
"is_duplicate": False,
"duplicate_ticket_id": None,
"parent_ticket_id": None,
"is_potential_duplicate": False,
"similarity": 0.0,
}
if duplicate_text:
duplicate_result = detect_semantic_duplicate(
duplicate_text,
company_id=final_data.get("company_id"),
threshold=duplicate_threshold,
)
final_data["description_vector"] = duplicate_service.generate_embedding(duplicate_text)
else:
final_data["description_vector"] = None
final_data["is_potential_duplicate"] = duplicate_result.get("is_potential_duplicate", False)
final_data["parent_ticket_id"] = duplicate_result.get("parent_ticket_id")
# --- Sanitize payload to only include valid Supabase DB columns ---
# Extra AI telemetry and non-existent schema fields are merged into the metadata JSONB column
# to avoid 400/500 errors from unknown column names in the insert call.
VALID_TICKET_COLUMNS = {
"user_id", "subject", "description", "category", "subcategory",
"priority", "assigned_team", "status", "auto_resolve", "is_duplicate",
"confidence", "image_url", "company", "company_id", "sla_breach_at", "metadata",
}
# Merge any extra telemetry and SLA/duplicate fields into metadata before filtering
existing_metadata = final_data.get("metadata") or {}
extra_keys = (
"entities", "solution_steps", "ocr_text", "needs_review", "routing_confidence",
"is_potential_duplicate", "parent_ticket_id", "sla_response_due_at", "sla_status", "escalation_level"
)
for extra_key in extra_keys:
if extra_key in final_data and final_data[extra_key] not in (None, "", [], {}):
existing_metadata[extra_key] = final_data[extra_key]
final_data["metadata"] = existing_metadata
# Strip keys not accepted by the DB schema
insert_data = {k: v for k, v in final_data.items() if k in VALID_TICKET_COLUMNS}
res = supabase.table("tickets").insert(insert_data).execute()
if not res.data:
raise Exception("Failed to insert ticket into database.")
ticket_id = res.data[0]["id"]
duplicate_indexed = True
duplicate_index_warning = None
if duplicate_text:
try:
duplicate_service.add_ticket(str(ticket_id), duplicate_text)
except Exception as index_error:
duplicate_indexed = False
duplicate_index_warning = "Duplicate index update failed."
print(f"[WARNING] {duplicate_index_warning} ticket_id={ticket_id} error={index_error}")
else:
duplicate_indexed = False
duplicate_index_warning = "Duplicate index update skipped: no description or subject text was provided."
print(f"[WARNING] {duplicate_index_warning}")
# Add initial system diagnostic message
msg = "Our Neural Engine has successfully triaged your issue and routed it to the designated team."
if final_data["auto_resolve"]:
msg = "AI Auto-Resolution active: A verified solution has been identified. Please review the attached resolution steps."
supabase.table("ticket_messages").insert({
"ticket_id": ticket_id,
"sender_id": "00000000-0000-0000-0000-000000000000", # System ID
"sender_name": "AI Assistant",
"sender_role": "admin",
"message": msg
}).execute()
response = {
"status": "success",
"ticket_id": ticket_id,
"duplicate_indexed": duplicate_indexed,
"is_potential_duplicate": final_data["is_potential_duplicate"],
"parent_ticket_id": final_data["parent_ticket_id"],
}
if duplicate_index_warning:
response["duplicate_index_warning"] = duplicate_index_warning
return response
except Exception as e:
traceback.print_exc()
raise HTTPException(status_code=500, detail=str(e))
@app.get("/tickets/{ticket_id}")
async def get_ticket_by_id(ticket_id: str):
"""Fetch single persistent ticket."""
if not supabase:
raise HTTPException(status_code=500, detail="Database connection not initialized")
res = supabase.table("tickets").select("*").eq("id", ticket_id).single().execute()
if not res.data:
raise HTTPException(status_code=404, detail="Ticket not found")
return res.data
@app.post("/tickets", response_model=TicketRecord)
async def create_ticket(ticket: TicketRecord):
"""Save a new ticket into the system."""
# Check for duplicates before adding
existing = next((t for t in TICKETS_DB if t.ticket_id == ticket.ticket_id), None)
if existing:
return existing
TICKETS_DB.append(ticket)
print(f"[DB] Ticket #{ticket.ticket_id} created for user {ticket.owner_id}")
return ticket
@app.patch("/tickets/{ticket_id}", response_model=TicketRecord)
async def update_ticket(ticket_id: str, updates: dict):
"""Partially update a ticket's fields (e.g., status, viewed_at)."""
for i, ticket in enumerate(TICKETS_DB):
if str(ticket.ticket_id) == str(ticket_id):
# Convert to dict, update, then back to model
ticket_dict = ticket.dict()
ticket_dict.update(updates)
updated_ticket = TicketRecord(**ticket_dict)
TICKETS_DB[i] = updated_ticket
return updated_ticket
raise HTTPException(status_code=404, detail="Ticket not found")
# ---------------------------------------------------------------------------
# Main AI Analyzer endpoint
# ---------------------------------------------------------------------------
@app.post("/ai/analyze_ticket", response_model=TicketResponse)
@limiter.limit("10/minute")
async def analyze_ticket(request_body: TicketRequest, request: Request):
"""
Main endpoint for analyzing a new ticket using the cascade of local AI models.
"""
text = request_body.text
settings = get_system_settings(request_body.company_id)
confidence_threshold = settings["ai_confidence_threshold"]
duplicate_sensitivity = settings["duplicate_sensitivity"]
enable_auto_resolve = settings["enable_auto_resolve"]
# Grab client metadata
client_ip = request.client.host if request.client else "unknown"
user_agent = request.headers.get("user-agent", "unknown")
origin_host = request.headers.get("origin", "unknown")
env_metadata = {
"ip": client_ip,
"user_agent": user_agent,
"origin": origin_host
}
# --- Layer 1: Local OCR (CPU, no API required) ---
local_ocr_text = ""
if request_body.image_base64 and ocr_service:
print("[AI] Extracting text via local OCR...")
local_ocr_text = ocr_service.extract_text(request_body.image_base64)
if local_ocr_text:
text = f"{text} {local_ocr_text}".strip()
print(f"[AI] OCR added {len(local_ocr_text)} chars to context.")
# Initalize Timeline
return await analyze_only(request_body)
@app.post("/ai/analyze")
async def analyze_only(request_body: TicketRequest):
"""
PERFORMANCE UPGRADE: AI Analysis phase only.
Does NOT persist to DB. This allows the user to review the analysis
and duplicate check before committing to a ticket creation.
"""
text = request_body.text
print(f"[AI] Starting Analysis (READ-ONLY) for: {text[:50]}...")
settings = get_system_settings(request_body.company)
confidence_threshold = settings["ai_confidence_threshold"]
duplicate_sensitivity = settings["duplicate_sensitivity"]
enable_auto_resolve = settings["enable_auto_resolve"]
# --- Vague Input Guard ---
# If the text is extremely short or a generic term, skip AI classification and
# return a safe low-priority "General Inquiry" to prevent hallucinated critical categories.
import re as _re
VAGUE_KEYWORDS = {
"demo", "test", "hi", "hello", "check", "try", "ping", "ok", "okay",
"issue", "problem", "error", "bug", "help", "hey", "asdf", "xyz",
"foo", "bar", "nothing", "something", "stuff",
}
_stripped = text.strip().lower()
_word_count = len(_stripped.split())
_is_vague = (len(_stripped) < 15) or (_word_count == 1 and _stripped in VAGUE_KEYWORDS)
if _is_vague:
import datetime as _dt, uuid as _uuid
_sla_breach = calculate_sla_breach_at("Low")
print(f"[AI] Vague input detected: '{text}'. Returning safe General Inquiry classification.")
return TicketResponse(
ticket_id=str(_uuid.uuid4()),
summary=f"General inquiry: {text}",
category="General",
subcategory="General Inquiry",
priority="Low",
auto_resolve=False,
assigned_team="IT Support",
entities=[],
duplicate_ticket=DuplicateInfo(is_duplicate=False),
confidence=0.1,
needs_review=True,
reasoning="Input was too brief for accurate classification. Please provide more context.",
decision_factors=["Input is too short or generic for AI classification."],
image_description="",
ocr_text="",
highlights=[],
timeline={"received": _dt.datetime.utcnow().isoformat() + "Z"},
env_metadata={},
is_potential_duplicate=False,
parent_ticket_id=None,
sla_breach_at=_sla_breach.isoformat().replace("+00:00", "Z"),
)
# --- Context & Environment ---
import datetime
def get_now_ist():
return datetime.datetime.utcnow().isoformat() + "Z"
env_metadata = {
"timestamp": get_now_ist(),
"model_version": "3.0.0-PRO",
"api_endpoint": "/ai/analyze"
}
timeline = {"received": get_now_ist()}
# --- Vision Logic (OCR Awareness) ---
gemini_analysis = {
"ocr_text": request_body.image_text or "",
"image_description": ""
}
if request_body.image_base64 and not gemini_analysis["ocr_text"]:
try:
print("[AI] Detecting visual context via Gemini...")
vision_result = gemini_service.analyze_image(request_body.image_base64, text)
gemini_analysis.update(vision_result)
except Exception as e:
print(f"[VISION ERROR] {e}")
summary = text[:100] + ("…" if len(text) > 100 else "")
# --- Classification ---
try:
classification_v3_res = classifier_v3.predict(text)
if "error" in classification_v3_res:
# Fallback to V1
classification = classifier_service.predict(text)
else:
# Parse V3 output
cat = classification_v3_res.get("Category", {}).get("prediction", "Unknown")
sub = classification_v3_res.get("Subcategory", {}).get("prediction", "Unknown")
pri = classification_v3_res.get("priority", {}).get("prediction", "Medium")
conf = classification_v3_res.get("Category", {}).get("confidence", 0.0)
from backend.services.classifier_service import TEAM_MAP, AUTO_RESOLVE_SUBS
assigned_team = TEAM_MAP.get(cat, "General Support")
auto_resolve = sub in AUTO_RESOLVE_SUBS
classification = {
"category": cat,
"subcategory": sub,
"priority": pri,
"auto_resolve": auto_resolve,
"assigned_team": assigned_team,
"confidence": float(conf)
}
except Exception as e:
traceback.print_exc()
classification = {
"category": "Unknown", "subcategory": "Unknown", "priority": "Medium",
"auto_resolve": False, "assigned_team": "General Support", "confidence": 0.0,
}
timeline["ai_analyzed"] = get_now_ist()
timeline["triaged"] = get_now_ist()
# --- NER ---
try:
entities = ner_service.extract_entities(text)
except Exception:
entities = []
timeline["metadata_harvested"] = get_now_ist()
# --- Duplicate detection ---
duplicate_threshold = get_duplicate_threshold(request_body.company_id, duplicate_sensitivity)
try:
dup_result = detect_semantic_duplicate(
text,
company_id=request_body.company_id,
threshold=duplicate_threshold,
)
except Exception:
dup_result = {
"is_duplicate": False,
"duplicate_ticket_id": None,
"parent_ticket_id": None,
"is_potential_duplicate": False,
"similarity": 0.0,
}
# --- RAG Knowledge Base Check ---
rag_match = None
try:
rag_match = rag_service.search_knowledge_base(text, threshold=0.85)
if rag_match:
classification["auto_resolve"] = True
classification["assigned_team"] = "Auto-Resolve AI"
classification["confidence"] = max(classification["confidence"], float(rag_match["similarity"]))
print(f"[RAG SUCCESS] Found solution for: '{rag_match['title']}'")
except Exception as e:
print(f"[RAG ERROR] {e}")
# --- Reasoning ---
decision_factors = []
if classification["confidence"] > confidence_threshold:
decision_factors.append(f"High confidence match for '{classification['subcategory']}'")
if entities:
decision_factors.append(f"Detected entities: {', '.join([e['text'] for e in entities[:2]])}")
if dup_result["is_duplicate"]:
decision_factors.append(f"Found similar incident ({int(dup_result['similarity']*100)}%)")
if rag_match:
decision_factors.append(f"Found solution article: '{rag_match['title']}'")
reasoning = f"Categorized as '{classification['category']}' - {classification['subcategory']}."
if (
enable_auto_resolve
and classification["confidence"] >= confidence_threshold
and classification["auto_resolve"]
):
classification["auto_resolve"] = True
else:
classification["auto_resolve"] = False
if classification["auto_resolve"]:
reasoning += " Flagged for AI auto-resolution via Knowledge Base." if rag_match else " Flagged for auto-resolution."
timeline["routed"] = get_now_ist()
# --- Gemini Summary ---
if gemini_service and gemini_service._initialized:
summary = gemini_service.get_summary(text)
# Convert priority to the SLA resolution target timestamp for preview.
sla_breach_dt = calculate_sla_breach_at(classification["priority"])
return TicketResponse(
ticket_id=str(uuid.uuid4()), # Temporary ID
summary=summary,
category=classification["category"],
subcategory=classification["subcategory"],
priority=classification["priority"],
auto_resolve=classification["auto_resolve"],
assigned_team=classification["assigned_team"],
entities=[EntityInfo(**e) for e in entities],
duplicate_ticket=DuplicateInfo(**dup_result),
confidence=classification["confidence"],
needs_review=classification["confidence"] < confidence_threshold,
reasoning=reasoning,
decision_factors=decision_factors,
image_description=gemini_analysis["image_description"],
ocr_text=gemini_analysis["ocr_text"],
highlights=entities, # Use entities as highlights for now
timeline=timeline,
env_metadata=env_metadata,
is_potential_duplicate=dup_result.get("is_potential_duplicate", False),
parent_ticket_id=dup_result.get("parent_ticket_id"),
sla_breach_at=sla_breach_dt.isoformat().replace("+00:00", "Z")
)
@app.post("/ai/analyze_stream")
async def analyze_stream(request_body: TicketRequest):
"""
REAL-TIME SSE ENDPOINT: Streams the AI progress to the frontend dynamically.
"""
import datetime
def get_now_ist():
return datetime.datetime.utcnow().isoformat() + "Z"
async def event_generator():
text = request_body.text
env_metadata = {
"timestamp": get_now_ist(),
"model_version": "3.0.0-PRO",
"api_endpoint": "/ai/analyze_stream"
}
timeline = {"received": get_now_ist()}
settings = get_system_settings(request_body.company_id)
confidence_threshold = settings["ai_confidence_threshold"]
duplicate_sensitivity = settings["duplicate_sensitivity"]
enable_auto_resolve = settings["enable_auto_resolve"]
# 1. Reading
yield f"data: {json.dumps({'step': 'Reading your message', 'status': 'in_progress'})}\n\n"
await asyncio.sleep(0.5)
gemini_analysis = {"ocr_text": request_body.image_text or "", "image_description": ""}
if request_body.image_base64 and not gemini_analysis["ocr_text"]:
try:
vision_result = gemini_service.analyze_image(request_body.image_base64, text)
gemini_analysis.update(vision_result)
except Exception as e:
pass
summary = text[:100] + ("…" if len(text) > 100 else "")
# 2. NER
yield f"data: {json.dumps({'step': 'Extracting technical entities', 'status': 'in_progress'})}\n\n"
await asyncio.sleep(0.2)
try:
entities = ner_service.extract_entities(text)
except Exception:
entities = []
timeline["metadata_harvested"] = get_now_ist()
# 3. Classification
yield f"data: {json.dumps({'step': 'Detecting category and priority', 'status': 'in_progress'})}\n\n"
await asyncio.sleep(0.2)
try:
classification_v3_res = classifier_v3.predict(text)
if "error" in classification_v3_res:
classification = classifier_service.predict(text)
else:
cat = classification_v3_res.get("Category", {}).get("prediction", "Unknown")
sub = classification_v3_res.get("Subcategory", {}).get("prediction", "Unknown")
pri = classification_v3_res.get("priority", {}).get("prediction", "Medium")
conf = classification_v3_res.get("Category", {}).get("confidence", 0.0)
from backend.services.classifier_service import TEAM_MAP, AUTO_RESOLVE_SUBS
assigned_team = TEAM_MAP.get(cat, "General Support")
auto_resolve = sub in AUTO_RESOLVE_SUBS
classification = {
"category": cat,
"subcategory": sub,
"priority": pri,
"auto_resolve": auto_resolve,
"assigned_team": assigned_team,
"confidence": float(conf)
}
except Exception as e:
classification = {
"category": "Unknown", "subcategory": "Unknown", "priority": "Medium",
"auto_resolve": False, "assigned_team": "General Support", "confidence": 0.0,
}
timeline["ai_analyzed"] = get_now_ist()
timeline["triaged"] = get_now_ist()
# 4. Duplicates
yield f"data: {json.dumps({'step': 'Checking duplicate issues', 'status': 'in_progress'})}\n\n"
await asyncio.sleep(0.2)
try:
duplicate_threshold = get_duplicate_threshold(request_body.company_id, duplicate_sensitivity)
dup_result = detect_semantic_duplicate(
text,
company_id=request_body.company_id,
threshold=duplicate_threshold,
)
except Exception:
dup_result = {
"is_duplicate": False,
"duplicate_ticket_id": None,
"parent_ticket_id": None,
"is_potential_duplicate": False,
"similarity": 0.0,
}
# 5. RAG / Solutions
yield f"data: {json.dumps({'step': 'Finding possible solutions', 'status': 'in_progress'})}\n\n"
await asyncio.sleep(0.2)
rag_match = None
try:
rag_match = rag_service.search_knowledge_base(text, threshold=0.85)
if rag_match:
classification["auto_resolve"] = True
classification["assigned_team"] = "Auto-Resolve AI"
classification["confidence"] = max(classification["confidence"], float(rag_match["similarity"]))
except Exception as e:
pass
decision_factors = []
if classification["confidence"] > confidence_threshold:
decision_factors.append(f"High confidence match for '{classification['subcategory']}'")
if entities:
decision_factors.append(f"Detected entities: {', '.join([e['text'] for e in entities[:2]])}")
if dup_result["is_duplicate"]:
decision_factors.append(f"Found similar incident ({int(dup_result['similarity']*100)}%)")
if rag_match:
decision_factors.append(f"Found solution article: '{rag_match['title']}'")
if not enable_auto_resolve:
classification["auto_resolve"] = False
reasoning = f"Categorized as '{classification['category']}' - {classification['subcategory']}."
if classification["auto_resolve"]:
reasoning += " Flagged for AI auto-resolution via Knowledge Base." if rag_match else " Flagged for auto-resolution."
timeline["routed"] = get_now_ist()
if gemini_service and gemini_service._initialized:
summary = gemini_service.get_summary(text)
sla_breach_dt = calculate_sla_breach_at(classification["priority"])
ticket_response_dict = {
"ticket_id": str(uuid.uuid4()),
"summary": summary,
"category": classification["category"],
"subcategory": classification["subcategory"],
"priority": classification["priority"],
"auto_resolve": classification["auto_resolve"],
"assigned_team": classification["assigned_team"],
"entities": [e for e in entities],
"duplicate_ticket": dup_result,
"confidence": classification["confidence"],
"needs_review": classification["confidence"] < confidence_threshold,
"reasoning": reasoning,
"decision_factors": decision_factors,
"image_description": gemini_analysis["image_description"],
"ocr_text": gemini_analysis["ocr_text"],
"highlights": entities,
"timeline": timeline,
"env_metadata": env_metadata,
"is_potential_duplicate": dup_result.get("is_potential_duplicate", False),
"parent_ticket_id": dup_result.get("parent_ticket_id"),
"sla_breach_at": sla_breach_dt.isoformat().replace("+00:00", "Z")
}
# 6. Final Result
yield f"data: {json.dumps({'step': 'done', 'result': jsonable_encoder(ticket_response_dict)})}\n\n"
return StreamingResponse(event_generator(), media_type="text/event-stream")
@app.post("/ai/analyze_ticket/legacy")
async def legacy_analyze_and_save(request_body: TicketRequest):
"""
BACKWARD COMPATIBILITY: Strictly performs analysis only.
Does NOT persist to DB to avoid foreign key violations.
"""
return await analyze_only(request_body)
@app.post("/ai/analyze-v2")
async def analyze_ticket_v2(request: TicketRequest):
text = request.text
try:
prediction = classifier_v2.predict(text)
return {
"status": "success",
"category": prediction["category"]["prediction"],
"subcategory": prediction["sub_category"]["prediction"],
"priority": prediction["priority"]["prediction"],
"auto_resolve": prediction["auto_resolve"]["prediction"].lower() == "true",
"assigned_team": prediction["assigned_team"]["prediction"],
"confidence": prediction["category"]["confidence"]
}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))