153 lines
6.2 KiB
Python
153 lines
6.2 KiB
Python
"""Feedback, reporting, login, orders and offline tickets."""
|
||
|
||
from datetime import datetime
|
||
|
||
from .db import connect, now_str
|
||
from .tracking import card_to_text
|
||
|
||
|
||
def feedback(data):
|
||
"""收集 👍/👎 反馈，存到 feedback 表"""
|
||
if not isinstance(data, dict):
|
||
return {'ok': False, 'error': '评价格式无效。'}, 400
|
||
raw_vote = data.get('vote')
|
||
votes = {'up': 1, 'down': -1, '1': 1, '-1': -1}
|
||
vote = votes.get(raw_vote) if isinstance(raw_vote, str) else raw_vote
|
||
question, answer = data.get('question', ''), data.get('answer', '')
|
||
if type(vote) is not int or vote not in (-1, 1) or not isinstance(question, str) or not isinstance(answer, str):
|
||
return {'ok': False, 'error': '评价内容无效，请重试。'}, 400
|
||
conn = connect()
|
||
conn.execute('INSERT INTO feedback(question, answer, vote, ctime) VALUES (?,?,?,?)',
|
||
(question[:500], answer[:2000], vote, now_str()))
|
||
conn.commit()
|
||
conn.close()
|
||
return {'ok': True}
|
||
|
||
|
||
def admin_stats():
|
||
conn = connect()
|
||
total_conv = conn.execute('SELECT COUNT(*) FROM chat_history').fetchone()[0]
|
||
total_msg = conn.execute('SELECT COUNT(*) FROM messages').fetchone()[0]
|
||
good = conn.execute("SELECT COUNT(*) FROM feedback WHERE vote IN ('1', 'up')").fetchone()[0]
|
||
bad = conn.execute("SELECT COUNT(*) FROM feedback WHERE vote IN ('-1', 'down')").fetchone()[0]
|
||
today = datetime.now().strftime('%Y-%m-%d')
|
||
today_conv = conn.execute('SELECT COUNT(*) FROM chat_history WHERE date(ctime)=date(?)', (today,)).fetchone()[0]
|
||
rating_avg, rating_count = conn.execute('SELECT COALESCE(AVG(stars), 0), COUNT(*) FROM ratings').fetchone()
|
||
conn.close()
|
||
return {'total_conv': total_conv, 'total_msg': total_msg, 'good': good, 'bad': bad, 'today_conv': today_conv,
|
||
'rating_avg': round(rating_avg, 1), 'rating_count': rating_count}
|
||
|
||
|
||
def my_orders(phone):
|
||
if not phone:
|
||
return {'orders': []}
|
||
conn = connect()
|
||
rows = conn.execute('SELECT track_no, status, created FROM orders WHERE phone=? ORDER BY id DESC', (phone,)).fetchall()
|
||
conn.close()
|
||
return {'orders': [{'track_no': r[0], 'status': r[1], 'created': r[2]} for r in rows]}
|
||
|
||
|
||
def add_ticket(data):
|
||
if not isinstance(data, dict):
|
||
return {'ok': False, 'error': '留言格式无效。'}, 400
|
||
content, phone = data.get('content'), data.get('phone', '')
|
||
if not isinstance(content, str) or not 1 <= len(content.strip()) <= 2000:
|
||
return {'ok': False, 'error': '请输入 1–2000 字的留言。'}, 400
|
||
if not isinstance(phone, str) or len(phone.strip()) > 32:
|
||
return {'ok': False, 'error': '联系电话格式无效。'}, 400
|
||
conn = connect()
|
||
conn.execute('INSERT INTO tickets(content, phone, ctime, status) VALUES(?,?,?,?)',
|
||
(content.strip(), phone.strip(), datetime.now().strftime('%Y-%m-%d %H:%M:%S'), 'pending'))
|
||
conn.commit()
|
||
conn.close()
|
||
return {'ok': True}
|
||
|
||
|
||
def list_tickets():
|
||
conn = connect()
|
||
rows = conn.execute("SELECT id, content, phone, ctime FROM tickets WHERE status='pending' ORDER BY id DESC").fetchall()
|
||
conn.close()
|
||
return [{'id': r[0], 'content': r[1], 'phone': r[2], 'ctime': r[3]} for r in rows]
|
||
|
||
|
||
def admin_login(data):
|
||
u = data.get('username', '')
|
||
p = data.get('password', '')
|
||
conn = connect()
|
||
row = conn.execute('SELECT name FROM agents WHERE username=? AND password=?', (u, p)).fetchone()
|
||
conn.close()
|
||
if row:
|
||
return {'ok': True, 'name': row[0]}
|
||
return ({'ok': False}, 401)
|
||
|
||
|
||
def agents_list():
|
||
conn = connect()
|
||
# rowid also works with the legacy agents table that has an explicit INTEGER id.
|
||
rows = conn.execute('SELECT rowid, username, name FROM agents ORDER BY rowid').fetchall()
|
||
conn.close()
|
||
return [{'id': row[0], 'username': row[1], 'name': row[2]} for row in rows]
|
||
|
||
|
||
def rate_chat(data):
|
||
sid, stars = data.get('sid'), data.get('stars', 0)
|
||
comment = data.get('comment', '')
|
||
if not sid or type(stars) is not int or not 1 <= stars <= 5 or not isinstance(comment, str):
|
||
return ({'ok': False}, 400)
|
||
conn = connect()
|
||
if not conn.execute('SELECT 1 FROM human_sessions WHERE id=?', (sid,)).fetchone():
|
||
conn.close()
|
||
return ({'ok': False}, 404)
|
||
conn.execute('INSERT INTO ratings(sid, stars, comment, ctime) VALUES(?,?,?,?)',
|
||
(sid, stars, comment, datetime.now().strftime('%Y-%m-%d %H:%M:%S')))
|
||
conn.commit()
|
||
conn.close()
|
||
return {'ok': True}
|
||
|
||
|
||
def search_chat(keyword):
|
||
if not keyword:
|
||
return {'results': []}
|
||
conn = connect()
|
||
rows = conn.execute('''SELECT h.title, m.role, m.content, m.ctime FROM messages m
|
||
JOIN chat_history h ON m.conv_id=h.id WHERE m.content LIKE ? ORDER BY m.id DESC LIMIT 50''',
|
||
('%' + keyword + '%',)).fetchall()
|
||
conn.close()
|
||
return [{'title': row[0], 'role': row[1], 'content': card_to_text(row[2]), 'ctime': row[3]} for row in rows]
|
||
|
||
|
||
def export_chat(format='xlsx'):
|
||
conn = connect()
|
||
rows = conn.execute('SELECT h.title, m.role, m.content, m.ctime FROM messages m JOIN chat_history h ON m.conv_id=h.id ORDER BY m.id').fetchall()
|
||
conn.close()
|
||
import io
|
||
if format == 'csv':
|
||
import csv
|
||
buf = io.StringIO()
|
||
w = csv.writer(buf)
|
||
w.writerow(['标题', '角色', '内容', '时间'])
|
||
for r in rows:
|
||
w.writerow([r[0], r[1], r[2] or '', r[3]])
|
||
return buf.getvalue()
|
||
from openpyxl import Workbook
|
||
from openpyxl.styles import Alignment, Font, PatternFill
|
||
|
||
workbook = Workbook()
|
||
sheet = workbook.active
|
||
sheet.title = '聊天记录'
|
||
sheet.append(['对话标题', '角色', '内容', '时间'])
|
||
for cell in sheet[1]:
|
||
cell.font = Font(bold=True, color='FFFFFF')
|
||
cell.fill = PatternFill('solid', fgColor='315FD5')
|
||
cell.alignment = Alignment(horizontal='center')
|
||
roles = {'user': '客户', 'assistant': '客服', 'system': '系统'}
|
||
for title, role, content, ctime in rows:
|
||
sheet.append([title, roles.get(role, role), content or '', ctime])
|
||
for cell in sheet[sheet.max_row]:
|
||
cell.data_type = 's'
|
||
for column, width in zip('ABCD', (20, 8, 80, 20)):
|
||
sheet.column_dimensions[column].width = width
|
||
buf = io.BytesIO()
|
||
workbook.save(buf)
|
||
return buf.getvalue()
|