import os import random import time import json from datetime import datetime, timedelta from flask import Flask, render_template, jsonify, request from faker import Faker app = Flask(__name__) fake = Faker('zh_CN') # --- Configuration --- PORT = 7860 UPLOAD_FOLDER = 'uploads' os.makedirs(UPLOAD_FOLDER, exist_ok=True) # --- In-Memory Data Store --- tickets = [] agents = [ {"id": 1, "name": "王强", "role": "L1 Support", "status": "online", "load": 0, "avatar": "https://api.dicebear.com/7.x/avataaars/svg?seed=王强"}, {"id": 2, "name": "李娜", "role": "L2 Support", "status": "busy", "load": 4, "avatar": "https://api.dicebear.com/7.x/avataaars/svg?seed=李娜"}, {"id": 3, "name": "张伟", "role": "Tech Lead", "status": "online", "load": 1, "avatar": "https://api.dicebear.com/7.x/avataaars/svg?seed=张伟"}, {"id": 4, "name": "刘洋", "role": "Billing Specialist", "status": "offline", "load": 0, "avatar": "https://api.dicebear.com/7.x/avataaars/svg?seed=刘洋"}, {"id": 5, "name": "陈敏", "role": "L1 Support", "status": "online", "load": 2, "avatar": "https://api.dicebear.com/7.x/avataaars/svg?seed=陈敏"}, ] # --- Business Logic & Simulation --- KEYWORDS_URGENT = ["投诉", "退款", "崩溃", "无法登录", "数据丢失", "紧急", "报错", "严重"] KEYWORDS_BILLING = ["发票", "扣款", "续费", "价格", "支付", "账单"] KEYWORDS_TECH = ["API", "Bug", "连接", "延迟", "配置", "代码", "服务器", "数据库"] def calculate_sentiment(text): """ Simulate sentiment analysis. Returns score -1.0 (Negative) to 1.0 (Positive). """ negative_words = ["失望", "垃圾", "慢", "差", "无法", "错误", "失败", "投诉", "骗子", "恶心", "烦", "糟糕"] positive_words = ["谢谢", "棒", "好", "解决", "顺利", "喜欢", "感谢", "赞", "优秀", "完美"] score = 0.0 for w in negative_words: if w in text: score -= 0.3 for w in positive_words: if w in text: score += 0.3 # Random noise score += random.uniform(-0.1, 0.1) return max(min(score, 1.0), -1.0) def auto_triage(ticket): """ AI Triage Logic: Assign Priority and Category based on content. """ text = ticket['content'] + " " + ticket['subject'] # Category Detection if any(k in text for k in KEYWORDS_BILLING): ticket['category'] = 'Billing' elif any(k in text for k in KEYWORDS_TECH): ticket['category'] = 'Technical' else: ticket['category'] = 'General' # Priority Detection if any(k in text for k in KEYWORDS_URGENT): ticket['priority'] = 'High' ticket['tags'].append('Risk') elif ticket['category'] == 'Billing': ticket['priority'] = 'Medium' else: ticket['priority'] = 'Low' # Sentiment ticket['sentiment_score'] = calculate_sentiment(text) return ticket def generate_ticket(): """Generate a realistic support ticket.""" created_at = fake.date_time_between(start_date='-2d', end_date='now') # Generate somewhat context-aware content prob = random.random() if prob < 0.2: subject = f"紧急：{fake.sentence(nb_words=3)}" content = f"我的系统彻底崩溃了，无法登录！报错代码500。{fake.text()}" elif prob < 0.4: subject = f"关于发票的问题" content = f"请问上个月的扣款为什么多出了50元？需要解释。{fake.text()}" elif prob < 0.7: subject = f"功能咨询：{fake.word()}" content = f"我想知道如何配置API密钥，文档看不懂。{fake.text()}" else: subject = fake.sentence(nb_words=4) content = fake.text() ticket = { "id": fake.uuid4()[:8].upper(), "subject": subject, "content": content, "customer_email": fake.email(), "customer_name": fake.name(), "status": "New", "created_at": created_at.strftime("%Y-%m-%d %H:%M:%S"), "tags": [], "history": [], "attachments": [] } return auto_triage(ticket) # Initial Population for _ in range(20): tickets.append(generate_ticket()) # Sort by date tickets.sort(key=lambda x: x['created_at'], reverse=True) # --- Routes --- @app.route('/') def index(): return render_template('index.html') @app.route('/api/stats') def get_stats(): total = len(tickets) open_tickets = len([t for t in tickets if t['status'] != 'Resolved']) high_priority = len([t for t in tickets if t['priority'] == 'High' and t['status'] != 'Resolved']) # Calculate avg sentiment of last 50 tickets recent = tickets[:50] avg_sentiment = sum(t['sentiment_score'] for t in recent) / len(recent) if recent else 0 # Agent load agent_stats = agents return jsonify({ "total_tickets": total, "open_tickets": open_tickets, "high_priority_risk": high_priority, "customer_happiness": round((avg_sentiment + 1) * 50, 1), # Scale -1..1 to 0..100 "agents": agent_stats }) @app.route('/api/tickets') def get_tickets(): # Return recent 50 tickets return jsonify(tickets[:50]) @app.route('/api/simulate', methods=['POST']) def simulate_incoming(): """Simulate receiving 1-3 new tickets""" count = random.randint(1, 3) new_tickets = [] for _ in range(count): t = generate_ticket() # Set to very recent t['created_at'] = datetime.now().strftime("%Y-%m-%d %H:%M:%S") tickets.insert(0, t) new_tickets.append(t) # Auto-assign to least loaded online agent online_agents = [a for a in agents if a['status'] == 'online'] if online_agents: for t in new_tickets: online_agents.sort(key=lambda x: x['load']) target = online_agents[0] target['load'] += 1 t['assigned_to'] = target['name'] t['status'] = 'Open' return jsonify({"added": count, "tickets": new_tickets}) @app.route('/api/upload', methods=['POST']) def upload_file(): if 'file' not in request.files: return jsonify({'error': 'No file part'}), 400 file = request.files['file'] if file.filename == '': return jsonify({'error': 'No selected file'}), 400 if file: filename = file.filename # In a real app, use secure_filename and unique IDs filepath = os.path.join(UPLOAD_FOLDER, filename) file.save(filepath) return jsonify({'success': True, 'filename': filename, 'url': f'/uploads/{filename}'}) @app.route('/api/resolve', methods=['POST']) def resolve_ticket(): """Simulate resolving tickets to keep list clean""" data = request.json ticket_id = data.get('id') for t in tickets: if t['id'] == ticket_id: t['status'] = 'Resolved' # Reduce load of assigned agent if 'assigned_to' in t: for a in agents: if a['name'] == t['assigned_to'] and a['load'] > 0: a['load'] -= 1 return jsonify({"success": True}) return jsonify({"success": False}), 404 @app.errorhandler(404) def page_not_found(e): return render_template('index.html'), 404 # SPA fallback mostly @app.errorhandler(500) def internal_server_error(e): return jsonify(error=str(e)), 500 if __name__ == '__main__': app.run(host='0.0.0.0', port=PORT, debug=True)