#Practical project one: Intelligent customer service work order classification system

import pandas as pd
import matplotlib.pyplot as plt
import seaborn as sns

df = pd.read_csv('customer_tickets.csv')
print(f"数据规模: {df.shape[0]} 条, {df.shape[1]} 列")
print(f"核心字段缺失值:\n{df[['title','content','category']].isnull().sum()}")

# 可视化类别分布 —— 查看是否有严重的类别不平衡
category_dist = df['category'].value_counts()
plt.figure(figsize=(8, 4))
sns.barplot(x=category_dist.index, y=category_dist.values)
plt.title("工单类别分布")
plt.xticks(rotation=30)
plt.show()

import re
import string
from zhon.hanzi import punctuation

def clean_text(text):
    """基础中文文本清洗：保留有用内容，去除冗余符号"""
    if pd.isna(text):
        return ""
    text = str(text)
    # 合并多余空格
    text = re.sub(r'\s+', ' ', text).strip()
    # 去除URL、邮箱、11位手机号（按需保留）
    text = re.sub(r'http[s]?://\S+|@\S+|\d{11}', ' ', text)
    # 去除中英文标点
    text = re.sub(f'[{punctuation}{string.punctuation}]', ' ', text)
    return text

# 构建新字段
df['text'] = df['title'].fillna('') + ' ' + df['content'].fillna('')
df['cleaned_text'] = df['text'].apply(clean_text)
# 过滤太短的无效文本（如只有“你好”）
df = df[df['cleaned_text'].str.len() > 5]

# 类别转数字标签
label2id = {cat: i for i, cat in enumerate(df['category'].unique())}
id2label = {v: k for k, v in label2id.items()}
df['label'] = df['category'].map(label2id)

from sklearn.model_selection import train_test_split
from sklearn.utils.class_weight import compute_class_weight

# 先切出70%训练，剩余30%均分为验证集和测试集
train_df, temp_df = train_test_split(
    df, test_size=0.3, stratify=df['label'], random_state=42
)
val_df, test_df = train_test_split(
    temp_df, test_size=0.5, stratify=temp_df['label'], random_state=42
)

# 计算各类别权重
class_weights = compute_class_weight(
    'balanced', classes=list(label2id.values()), y=train_df['label']
)
class_weights = dict(zip(list(label2id.values()), class_weights))
print(f"类别权重: {class_weights}")

from transformers import (
    AutoTokenizer, AutoModelForSequenceClassification,
    TrainingArguments, Trainer
)
from sklearn.metrics import accuracy_score, f1_score
import torch

# 1. 加载中文 RoBERTa
MODEL_NAME = "hfl/chinese-roberta-wwm-ext"
tokenizer = AutoTokenizer.from_pretrained(MODEL_NAME)
model = AutoModelForSequenceClassification.from_pretrained(
    MODEL_NAME, num_labels=len(label2id),
    id2label=id2label, label2id=label2id
)

# 2. 自定义数据集（适配 Trainer）
class TicketDataset(torch.utils.data.Dataset):
    def __init__(self, texts, labels, tokenizer, max_len=256):
        self.texts = texts
        self.labels = labels
        self.tokenizer = tokenizer
        self.max_len = max_len

    def __len__(self):
        return len(self.texts)

    def __getitem__(self, idx):
        text = str(self.texts[idx])
        label = self.labels[idx]
        encoding = self.tokenizer(
            text, truncation=True, padding='max_length',
            max_length=self.max_len, return_tensors='pt'
        )
        return {
            'input_ids': encoding['input_ids'].flatten(),
            'attention_mask': encoding['attention_mask'].flatten(),
            'labels': torch.tensor(label, dtype=torch.long)
        }

# 3. 准备数据集
train_dataset = TicketDataset(
    train_df['cleaned_text'].tolist(), train_df['label'].tolist(), tokenizer
)
val_dataset = TicketDataset(
    val_df['cleaned_text'].tolist(), val_df['label'].tolist(), tokenizer
)

# 4. 定义评估指标（准确率 + 加权 F1）
def compute_metrics(eval_pred):
    predictions, labels = eval_pred
    predictions = predictions.argmax(axis=-1)
    return {
        'accuracy': accuracy_score(labels, predictions),
        'f1': f1_score(labels, predictions, average='weighted')
    }

# 5. 训练参数配置
training_args = TrainingArguments(
    output_dir='./ticket_classifier',
    num_train_epochs=3,
    per_device_train_batch_size=16,
    per_device_eval_batch_size=32,
    learning_rate=2e-5,
    weight_decay=0.01,
    warmup_ratio=0.1,
    evaluation_strategy="epoch",
    save_strategy="epoch",
    load_best_model_at_end=True,
    metric_for_best_model="f1",
    fp16=torch.cuda.is_available(),     # 混合精度加速
    logging_steps=100,
    seed=42,
    report_to=None                      # 不上传 W&B
)

# 6. 启动训练
trainer = Trainer(
    model=model,
    args=training_args,
    train_dataset=train_dataset,
    eval_dataset=val_dataset,
    compute_metrics=compute_metrics,
    class_weight=class_weights         # 传入类别权重
)
trainer.train()
trainer.save_model('./best_ticket_classifier')

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
from transformers import pipeline
import torch
import time
from typing import List, Optional

app = FastAPI(
    title="智能客服工单分类API",
    version="1.0.0",
    description="轻量可落地的中文工单分类服务"
)

# 服务启动时加载模型（不会阻塞请求）
MODEL_PATH = "./best_ticket_classifier"
classifier = pipeline(
    "text-classification",
    model=MODEL_PATH,
    tokenizer=MODEL_PATH,
    device=0 if torch.cuda.is_available() else -1,
    truncation=True,
    max_length=256
)

# 请求与响应数据结构
class SingleTicket(BaseModel):
    title: str = Field(..., description="工单标题")
    content: Optional[str] = Field("", description="工单内容（可选）")

class BatchTickets(BaseModel):
    tickets: List[SingleTicket]

class SingleResponse(BaseModel):
    category: str
    confidence: float
    processing_time: float

class BatchResponse(BaseModel):
    results: List[SingleResponse]
    total_processing_time: float

@app.on_event("startup")
async def startup():
    print("模型加载完成，服务已启动！")

@app.get("/health", tags=["健康检查"])
async def health_check():
    return {"status": "healthy", "timestamp": time.strftime('%Y-%m-%d %H:%M:%S')}

@app.get("/categories", tags=["辅助接口"])
async def get_categories():
    """返回所有支持的类别列表"""
    return {"categories": list(classifier.model.config.id2label.values())}

@app.post("/classify/single", tags=["分类接口"], response_model=SingleResponse)
async def classify_single(ticket: SingleTicket):
    start = time.time()
    try:
        full_text = f"{ticket.title} {ticket.content}".strip()
        result = classifier(full_text)[0]
        return SingleResponse(
            category=result["label"],
            confidence=round(result["score"], 4),
            processing_time=round(time.time() - start, 4)
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"分类失败: {str(e)}")

@app.post("/classify/batch", tags=["分类接口"], response_model=BatchResponse)
async def classify_batch(batch: BatchTickets):
    start = time.time()
    try:
        full_texts = [f"{t.title} {t.content}".strip() for t in batch.tickets]
        raw_results = classifier(full_texts, batch_size=8)
        results = [
            SingleResponse(
                category=r["label"],
                confidence=round(r["score"], 4),
                processing_time=0.0
            ) for r in raw_results
        ]
        return BatchResponse(
            results=results,
            total_processing_time=round(time.time() - start, 4)
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"批量分类失败: {str(e)}")

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

import logging
from starlette.middleware.base import BaseHTTPMiddleware
import time

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("ticket_api")

class LogMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        start = time.time()
        response = await call_next(request)
        duration = time.time() - start
        logger.info(
            f"{request.method} {request.url.path} "
            f"- {response.status_code} - {duration:.4f}s"
        )
        return response

app.add_middleware(LogMiddleware)