# Python 전처리 파이프라인 예시
import re
from konlpy.tag import Mecab

class CSTextPreprocessor:
def __init__(self):
self.mecab = Mecab()

def clean_text(self, text: str) -> str:
# HTML 태그 제거
text = re.sub(r'<[^>]+>', '', text)
# URL 마스킹
text = re.sub(r'https?://\S+', '[URL]', text)
# 전화번호 마스킹
text = re.sub(r'01[0-9]-?\d{3,4}-?\d{4}', '[PHONE]', text)
# 이메일 마스킹
text = re.sub(r'\S+@\S+\.\S+', '[EMAIL]', text)
# 반복 문자 정규화 (3회 이상 → 2회)
text = re.sub(r'(.)\1{2,}', r'\1\1', text)
return text.strip()

def extract_metadata(self, raw_data: dict) -> dict:
return {
'channel': raw_data.get('channel', 'unknown'),
'customer_tier': raw_data.get('customer_tier', 'normal'),
'has_attachment': bool(raw_data.get('attachments')),
'text_length': len(raw_data.get('text', '')),
'hour_of_day': raw_data.get('timestamp').hour
}

def preprocess(self, raw_data: dict) -> dict:
cleaned = self.clean_text(raw_data['text'])
metadata = self.extract_metadata(raw_data)
return {
'cleaned_text': cleaned,
'metadata': metadata,
'original_id': raw_data['id']
}

from transformers import AutoTokenizer, AutoModelForSequenceClassification
from transformers import TrainingArguments, Trainer
import torch

# 카테고리 정의
CATEGORIES = [
"배송_물류", "결제_환불", "기술_버그", 
"계정_보안", "상품_문의", "일반_문의", "기타"
]

class CSClassifier:
def __init__(self, model_name="klue/roberta-base"):
self.tokenizer = AutoTokenizer.from_pretrained(model_name)
self.model = AutoModelForSequenceClassification.from_pretrained(
model_name, 
num_labels=len(CATEGORIES)
)
self.categories = CATEGORIES

def predict(self, text: str) -> dict:
inputs = self.tokenizer(
text, 
return_tensors="pt", 
max_length=512, 
truncation=True,
padding=True
)

with torch.no_grad():
outputs = self.model(**inputs)
probs = torch.softmax(outputs.logits, dim=-1)

pred_idx = probs.argmax().item()
confidence = probs[0][pred_idx].item()

return {
"category": self.categories[pred_idx],
"confidence": confidence,
"all_probs": {
cat: probs[0][i].item() 
for i, cat in enumerate(self.categories)
}
}

def predict_with_fallback(self, text: str, threshold=0.85) -> dict:
result = self.predict(text)

if result["confidence"] < threshold:
# LLM 폴백 호출
result["needs_llm_review"] = True
else:
result["needs_llm_review"] = False

return result

CLASSIFICATION_PROMPT = """
당신은 고객 문의를 분류하는 전문 AI입니다.
다음 고객 문의를 분석하여 JSON 형식으로 응답하세요.

[분류 카테고리]
- 배송_물류: 배송 지연, 배송지 변경, 반품 수거 등
- 결제_환불: 결제 오류, 환불 요청, 취소 요청 등
- 기술_버그: 앱/웹 오류, 기능 이상, 로딩 문제 등
- 계정_보안: 로그인 불가, 비밀번호 찾기, 계정 도용 등
- 상품_문의: 상품 정보, 재고, 사이즈 등
- 일반_문의: 영업시간, 이용방법 등
- 기타: 위 카테고리에 해당하지 않는 경우

[고객 문의]
{customer_message}

[응답 형식]
{
"category": "카테고리명",
"priority": "긴급|높음|보통|낮음",
"sentiment": "매우부정|부정|중립|긍정",
"intent": "문의|불만|요청|취소|기타",
"summary": "문의 내용 한 줄 요약",
"confidence": 0.0~1.0,
"reasoning": "분류 근거 설명"
}
"""

# FastAPI 기반 AI 분류 서버 예시
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import asyncio

app = FastAPI()

class TicketWebhook(BaseModel):
ticket_id: str
subject: str
body: str
customer_email: str
channel: str
created_at: str

class ClassificationResult(BaseModel):
ticket_id: str
category: str
priority: str
sentiment: str
assigned_agent_id: str
confidence: float

# 전역 컴포넌트 초기화
preprocessor = CSTextPreprocessor()
classifier = CSClassifier()
router = SmartRouter(agents=load_agents_from_db())

@app.post("/webhook/new-ticket", response_model=ClassificationResult)
async def process_new_ticket(ticket: TicketWebhook):
try:
# 1. 전처리
processed = preprocessor.preprocess({
"text": f"{ticket.subject}\n{ticket.body}",
"channel": ticket.channel,
"id": ticket.ticket_id
})

# 2. AI 분류
classification = classifier.predict_with_fallback(
processed["cleaned_text"]
)

# 3. LLM 폴백 (필요시)
if classification.get("needs_llm_review"):
classification = await llm_classify(processed["cleaned_text"])

# 4. 담당자 배정
assignment = router.assign_ticket({
"id": ticket.ticket_id,
"category": classification["category"],
"priority": classification["priority"],
"customer_tier": get_customer_tier(ticket.customer_email)
})

# 5. CS 플랫폼에 결과 업데이트 (Zendesk API 등)
await update_ticket_in_platform(
ticket_id=ticket.ticket_id,
classification=classification,
assignment=assignment
)

return ClassificationResult(
ticket_id=ticket.ticket_id,
category=classification["category"],
priority=classification["priority"],
sentiment=classification["sentiment"],
assigned_agent_id=assignment.get("assigned_to", "queue"),
confidence=classification["confidence"]
)

except Exception as e:
raise HTTPException(status_code=500, detail=str(e))

@app.get("/health")
async def health_check():
return {"status": "ok", "model_loaded": True}