baiyue-logistics-chat/logistics_chat/human_service.py

"""Human conversation lifecycle; typing state is supplied by the app."""

import time
from datetime import datetime, timedelta, timezone

from .db import connect

SERVICE_TIMEZONE = timezone(timedelta(hours=8))


def human_request(data):
    conv_id = data.get('conv_id')
    if not conv_id:
        return ({'ok': False, 'error': 'no conv'}, 400)
    service_time = datetime.now(SERVICE_TIMEZONE)
    if not 8 <= service_time.hour < 22:
        return {'ok': False, 'closed': True, 'msg': '人工客服工作时间为每天北京时间8:00-22:00，当前为非工作时间，请留言'}
    now = service_time.strftime('%Y-%m-%d %H:%M:%S')
    conn = connect()
    conn.execute('BEGIN IMMEDIATE')
    if not conn.execute('SELECT 1 FROM chat_history WHERE id=?', (conv_id,)).fetchone():
        conn.close()
        return ({'ok':False, 'error':'conversation_not_found'}, 404)
    row = conn.execute("SELECT id FROM human_sessions WHERE conv_id=? AND status!='closed' ORDER BY id DESC LIMIT 1", (conv_id,)).fetchone()
    if row:
        sid = row[0]
    else:
        cur = conn.execute('INSERT INTO human_sessions(conv_id, status, ctime) VALUES(?,?,?)', (conv_id, 'waiting', now))
        sid = cur.lastrowid
    conn.commit()
    conn.close()
    return {'ok': True, 'session_id': sid}


def human_poll(sid, after, typing_map):
    conn = connect()
    msgs = conn.execute('SELECT id, side, content, ctime, read_at, recalled_id FROM human_messages WHERE session_id=? AND id>? ORDER BY id', (sid, after)).fetchall()
    st = conn.execute("SELECT status, COALESCE(note, '') FROM human_sessions WHERE id=?", (sid,)).fetchone()
    conn.close()
    is_typing = False
    ts = typing_map.get(str(sid))
    if ts and time.time() - ts < 5:
        is_typing = True
    return {'status': st[0] if st else 'closed', 'note': st[1] if st else '',
            'typing': is_typing, 'messages': [dict(id=m[0], side=m[1], content=m[2], ctime=m[3],
                read=bool(m[4]), **({'recalled_id': m[5]} if m[5] else {})) for m in msgs]}


def human_send(data):
    sid = data.get('sid')
    side = data.get('side', 'customer')
    content = data.get('content') or ''
    if not sid or not isinstance(content, str) or not content.strip() or side not in ('customer', 'agent'):
        return ({'ok': False}, 400)
    content = content.strip()
    now = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    conn = connect()
    conn.execute('BEGIN IMMEDIATE')
    row = conn.execute('SELECT conv_id, status FROM human_sessions WHERE id=?', (sid,)).fetchone()
    if not row:
        conn.close()
        return ({'ok': False, 'error': 'session_not_found'}, 404)
    if row[1] == 'closed' or (data.get('conv_id') is not None and data['conv_id'] != row[0]):
        conn.close()
        return ({'ok': False, 'error': 'conversation_changed'}, 409)
    conn.execute('INSERT INTO human_messages(session_id, side, content, ctime) VALUES(?,?,?,?)', (sid, side, content, now))
    conn.commit()
    conn.close()
    return {'ok': True}


def human_queue():
    conn = connect()
    rows = conn.execute("SELECT id, conv_id, status, ctime FROM human_sessions WHERE status='waiting' ORDER BY id").fetchall()
    conn.close()
    return [{'id': r[0], 'conv_id': r[1], 'ctime': r[3]} for r in rows]


def human_active():
    conn = connect()
    rows = conn.execute("SELECT id, conv_id FROM human_sessions WHERE status='active' ORDER BY id").fetchall()
    conn.close()
    return [{'id': r[0], 'conv_id': r[1]} for r in rows]


def human_accept(data):
    sid = data.get('sid')
    conn = connect()
    changed = conn.execute("UPDATE human_sessions SET status='active' WHERE id=? AND status IN ('waiting','active') "
                           'AND EXISTS(SELECT 1 FROM chat_history WHERE id=human_sessions.conv_id)', (sid,)).rowcount
    conn.commit()
    conn.close()
    return {'ok': True} if changed else ({'ok':False, 'error':'session_closed_or_missing'}, 409)


def human_close(data):
    sid = data.get('sid')
    conn = connect()
    conn.execute("UPDATE human_sessions SET status='closed' WHERE id=?", (sid,))
    conn.commit()
    conn.close()
    return {'ok': True}


def human_transfer(data, typing_map):
    sid = data.get('sid')
    conn = connect()
    changed = conn.execute("UPDATE human_sessions SET status='waiting' WHERE id=? AND status='active'", (sid,)).rowcount
    if not changed:
        conn.close()
        return ({'ok': False, 'error': '会话当前无法转接'}, 409)
    conn.execute('INSERT INTO human_messages(session_id, side, content, ctime) VALUES(?,?,?,?)',
                 (sid, 'system', '客服已转接，请其他客服接入', datetime.now().strftime('%Y-%m-%d %H:%M:%S')))
    conn.commit()
    conn.close()
    typing_map.pop(str(sid), None)
    return {'ok': True}


def human_note(data):
    note = data.get('note', '')
    if not isinstance(note, str):
        return ({'ok': False}, 400)
    conn = connect()
    changed = conn.execute('UPDATE human_sessions SET note=? WHERE id=?', (note, data.get('sid'))).rowcount
    conn.commit()
    conn.close()
    return {'ok': True} if changed else ({'ok': False}, 404)


def human_recall(data):
    conn = connect()
    mid = data.get('mid')
    # Only the first request can change the original message; retries emit no duplicate event.
    conn.execute('BEGIN IMMEDIATE')
    row = conn.execute('SELECT session_id, side, content FROM human_messages WHERE id=?', (mid,)).fetchone()
    if not row:
        conn.close()
        return ({'ok': False}, 404)
    if row[1] == 'system' and row[2] == '该消息已撤回':
        conn.close()
        return {'ok': True}
    if row[1] != 'agent':
        conn.close()
        return ({'ok': False}, 400)
    conn.execute("UPDATE human_messages SET side='system', content='该消息已撤回' WHERE id=?", (mid,))
    conn.execute('INSERT INTO human_messages(session_id, side, content, ctime, recalled_id) VALUES(?,?,?,?,?)',
                 (row[0], 'system', '该消息已撤回', datetime.now().strftime('%Y-%m-%d %H:%M:%S'), mid))
    conn.commit()
    conn.close()
    return {'ok': True}


def human_typing(data, typing_map):
    sid = data.get('sid')
    typing_map[str(sid)] = time.time()
    return {'ok': True}


def human_stop_typing(data, typing_map):
    sid = data.get('sid')
    typing_map.pop(str(sid), None)
    return {'ok': True}


def human_read(data):
    sid = data.get('sid')
    conn = connect()
    conn.execute("UPDATE human_messages SET read_at=? WHERE session_id=? AND side='customer' AND read_at IS NULL", (datetime.now().strftime('%Y-%m-%d %H:%M:%S'), sid))
    conn.commit()
    conn.close()
    return {'ok': True}