287 lines
15 KiB
Python
287 lines
15 KiB
Python
"""Scoped summaries, five-row previews and durable conversational list pages."""
|
||
|
||
from collections import Counter,deque
|
||
import json
|
||
import re
|
||
import sqlite3
|
||
|
||
from .db import connect
|
||
from .order_query import parse_query,filters,STAGE_LABELS
|
||
from .order_tracking import current_stage
|
||
|
||
|
||
def page_command(message):
|
||
text=re.sub(r'[\s，,。.!！?？]','',message)
|
||
match=re.fullmatch(r'(?:请)?(?:订单)?第(\d{1,5})页',text)
|
||
if match:return int(match.group(1))
|
||
if re.fullmatch(r'(?:请|继续)?(?:下一页|下页|继续显示|接着显示(?:剩下的)?订单)',text):return 'next'
|
||
if re.fullmatch(r'(?:请)?(?:上一页|上页)',text):return 'previous'
|
||
return None
|
||
|
||
|
||
def valid_filters(query):
|
||
from .order_data import STATUS_LABELS
|
||
return (isinstance(query,dict) and set(query)=={'status','stage','exception'}
|
||
and (query['status'] is None or query['status'] in STATUS_LABELS)
|
||
and (query['stage'] is None or query['stage'] in STAGE_LABELS)
|
||
and isinstance(query['exception'],bool))
|
||
|
||
|
||
def load_state(conv_id):
|
||
from .customer_service import current_customer_id,current_scope_key
|
||
from .visitor_service import visitor_id
|
||
from .customer_faq import is_acknowledgement
|
||
from .order_followups import ordinal_request,shown_references,is_relative_question
|
||
from .order_data import is_order_followup
|
||
if not conv_id:return None
|
||
connection=connect()
|
||
try:
|
||
row=connection.execute('SELECT order_list_state FROM chat_history WHERE id=? AND visitor_id=? '
|
||
'AND customer_id=? AND customer_scope=?',
|
||
(conv_id,visitor_id(),current_customer_id(),current_scope_key())).fetchone()
|
||
state=json.loads(row[0]) if row and row[0] else None
|
||
if not isinstance(state,dict) or not valid_filters(state.get('query')):return None
|
||
if type(state.get('page')) is not int or not 1<=state['page']<=10000:return None
|
||
if state.get('size') not in (5,20) or type(state.get('message_id')) is not int:return None
|
||
if 'shown' in state:
|
||
shown=state['shown']
|
||
if not isinstance(shown,list) or len(shown)>20:return None
|
||
if any(not isinstance(item,dict) or set(item)!={'inbound_code','order_no'}
|
||
or not isinstance(item['order_no'],str) or not item['order_no']
|
||
or item['inbound_code'] is not None and not isinstance(item['inbound_code'],str) for item in shown):
|
||
return None
|
||
state['selected']=False
|
||
for (text,) in connection.execute("SELECT content FROM messages WHERE conv_id=? AND role='user' AND id>? ORDER BY id",
|
||
(conv_id,state['message_id'])):
|
||
if is_acknowledgement(text):continue
|
||
if is_relative_question(text):continue
|
||
ordinal=ordinal_request(text)
|
||
if ordinal is not None:
|
||
state['selected']=1<=ordinal<=len(shown_references(connection,conv_id,state))
|
||
continue
|
||
if state['selected'] and is_order_followup(text):continue
|
||
if page_command(text) is None and parse_query(text) is None and text.strip() not in ('查看全部','查看全部订单'):
|
||
return None
|
||
return state
|
||
except (ValueError,TypeError):return None
|
||
finally:connection.close()
|
||
|
||
|
||
def save_state(conv_id,state,message_id):
|
||
"""Called only after the reply and customer binding have been persisted."""
|
||
from .customer_service import current_customer_id,current_scope_key
|
||
from .visitor_service import visitor_id
|
||
if not conv_id or type(message_id) is not int:return
|
||
connection=connect()
|
||
try:
|
||
state=dict(state)
|
||
state['message_id']=message_id
|
||
connection.execute('UPDATE chat_history SET order_list_state=? WHERE id=? AND visitor_id=? '
|
||
'AND customer_id=? AND customer_scope=? '
|
||
'AND NOT EXISTS(SELECT 1 FROM messages WHERE conv_id=? AND id>?)',
|
||
(json.dumps(state,ensure_ascii=False),conv_id,visitor_id(),current_customer_id(),current_scope_key(),conv_id,message_id))
|
||
connection.commit()
|
||
finally:connection.close()
|
||
|
||
|
||
def scoped_orders(connection,tenants,status=None):
|
||
"""Use one authorized projection and normalization for lists and overviews."""
|
||
from .order_data import tenant_filter
|
||
scope=tenant_filter(tenant_ids=tenants)
|
||
if scope is None:raise ValueError('Explicit customer scope required')
|
||
where=' WHERE 1=1'+scope[0]
|
||
values=scope[1]
|
||
if status:
|
||
where+=' AND status=?'
|
||
values=values+[status]
|
||
rows=connection.execute('SELECT order_no,inbound_code,status,warehouse_code,tracking_json,exception_reason '
|
||
'FROM order_test_data'+where+' ORDER BY COALESCE(julianday(source_updated_at),julianday(snapshot_at),0) DESC,id DESC',values)
|
||
for row in rows:
|
||
item=dict(row)
|
||
item['exception_reason']=(item['exception_reason'] or '').strip()
|
||
item['stage']=current_stage(item.pop('tracking_json'))
|
||
yield item
|
||
|
||
|
||
def read_overview(tenants):
|
||
"""Summarize the customer's current orders in one database snapshot."""
|
||
connection=connect()
|
||
try:
|
||
connection.row_factory=sqlite3.Row
|
||
connection.execute('BEGIN')
|
||
total=processing=exceptions=0
|
||
stages=Counter()
|
||
abnormal_rows=[]
|
||
processing_rows=[]
|
||
for row in scoped_orders(connection,tenants):
|
||
total+=1
|
||
abnormal=bool(row['exception_reason'])
|
||
if abnormal:
|
||
exceptions+=1
|
||
if len(abnormal_rows)<5:abnormal_rows.append(row)
|
||
if row['status']=='PROCESSING':
|
||
processing+=1
|
||
stages[row['stage']]+=1
|
||
if not abnormal and len(processing_rows)<5:processing_rows.append(row)
|
||
return {'total':total,'processing':processing,'exceptions':exceptions,
|
||
'stages':[{'code':code,'count':stages[code]} for code in STAGE_LABELS if stages[code]],
|
||
'orders':(abnormal_rows+processing_rows)[:5]}
|
||
finally:
|
||
connection.close()
|
||
|
||
|
||
def answer_overview(message,tenants):
|
||
question=re.sub(r'[\s，,。.!！?？]','',message)
|
||
chinese=re.fullmatch(r'(?:请|帮我|请帮我)?(?:看一下|看看|查看|查询|显示)?(?:一下)?'
|
||
r'(?:(?:我|当前账号)的?|当前|现在)?(?:订单|运单)(?:概览|概况|总体情况|整体情况|情况|汇总)(?:呢|吧|吗)?',question)
|
||
english=re.fullmatch(r'(?:please\s+)?(?:(?:show|give)\s+(?:me\s+)?)?(?:my\s+)?(?:order|shipment)s?\s+(?:overview|summary)',message.strip(' .!?'),re.I)
|
||
if not chinese and not english:return None
|
||
if not tenants:
|
||
from .customer_service import order_login_response
|
||
return order_login_response()
|
||
try:
|
||
data=read_overview(tenants)
|
||
except sqlite3.Error:
|
||
return {'t':'订单信息暂时无法读取，请稍后重试。'}
|
||
lines=[f'您当前共有 {data["total"]} 票订单，履约中 {data["processing"]} 票，异常订单 {data["exceptions"]} 票。']
|
||
if data['stages']:
|
||
lines.append('履约中订单的当前实际节点：')
|
||
lines.extend(f'- {STAGE_LABELS[stage["code"]]}：{stage["count"]}票' for stage in data['stages'])
|
||
if data['orders']:
|
||
lines.append('重点订单（异常优先，最多5票）：')
|
||
lines.extend(row_text(order) for order in data['orders'])
|
||
return {'t':'\n'.join(lines),'_customer_data':True}
|
||
|
||
|
||
def matches_filters(row,query):
|
||
return ((not query['status'] or row['status']==query['status'])
|
||
and (not query['stage'] or row['stage']==query['stage'])
|
||
and (not query['exception'] or bool(row['exception_reason'])))
|
||
|
||
|
||
def read_orders(tenants,query,page=1,size=5,comparisons=()):
|
||
"""One scoped snapshot; each order occupies one actual-node bucket."""
|
||
if not valid_filters(query) or any(not valid_filters(part) for part in comparisons):
|
||
raise ValueError('Explicit filters required')
|
||
connection=connect()
|
||
try:
|
||
connection.row_factory=sqlite3.Row
|
||
connection.execute('BEGIN')
|
||
total=with_inbound=0;groups=Counter();selected=[];tail=deque(maxlen=size)
|
||
comparison_groups=[Counter() for _ in comparisons]
|
||
for item in scoped_orders(connection,tenants):
|
||
for index,comparison in enumerate(comparisons):
|
||
if matches_filters(item,comparison):comparison_groups[index][item['stage']]+=1
|
||
if not matches_filters(item,query):continue
|
||
total+=1;with_inbound+=bool((item['inbound_code'] or '').strip());groups[item['stage']]+=1
|
||
tail.append(item)
|
||
if (page-1)*size<total<=page*size:selected.append(item)
|
||
last=max(1,(total+size-1)//size)
|
||
if page>last:
|
||
page=last;selected=list(tail)[-(total-(last-1)*size):] if total else []
|
||
return {'total':total,'with_inbound':with_inbound,'groups':groups,'rows':selected,'page':page,
|
||
'counts':[sum(group.values()) for group in comparison_groups],'comparison_groups':comparison_groups}
|
||
finally:connection.close()
|
||
|
||
|
||
def row_text(row,reason_only=False):
|
||
from .order_data import STATUS_LABELS
|
||
def clean(value):return re.sub(r'[\r\n\t]+',' ',str(value or '未登记'))
|
||
text=f'- 入仓码：{clean(row["inbound_code"])}；订单号：{clean(row["order_no"])}'
|
||
if not reason_only:
|
||
text+=f'；业务状态：{STATUS_LABELS.get(row["status"],row["status"])}；目的仓：{clean(row["warehouse_code"])}；当前节点：{STAGE_LABELS[row["stage"]]}'
|
||
if row['exception_reason']:text+='；异常原因：'+clean(row['exception_reason'])
|
||
return text
|
||
|
||
|
||
def unsupported(listing=False):
|
||
return {'t':f'暂不支持按这个条件{"列出订单" if listing else "统计"}。可以查询订单总数、业务状态、明确标记的异常，或按当前实际物流节点统计；请一次指定一个节点。'}
|
||
|
||
|
||
def summary_text(query,total):
|
||
from .order_data import STATUS_LABELS
|
||
if query['stage']:
|
||
text=f'您当前有 {total} 票当前节点为“{STAGE_LABELS[query["stage"]]}”的订单'
|
||
elif query['status']:
|
||
text=f'您当前有 {total} 票业务状态为“{STATUS_LABELS[query["status"]]}”的订单'
|
||
elif query['exception']:text=f'您当前有 {total} 票明确标记异常的订单'
|
||
else:text=f'您当前共有 {total} 票订单'
|
||
if query['exception'] and (query['stage'] or query['status']):text+='（已明确标记异常）'
|
||
if query['stage'] and query['status']:text+=f'（业务状态为{STATUS_LABELS[query["status"]]}）'
|
||
return text
|
||
|
||
|
||
def stage_distribution(groups):
|
||
return ['按当前实际物流节点分布（每票只计一次）：']+[
|
||
f'- {label}：{groups[code]}票' for code,label in STAGE_LABELS.items() if groups[code]]
|
||
|
||
|
||
def answer_summary(intent,tenants,conv_id=None,reasons=False):
|
||
from .customer_service import order_login_response
|
||
if not tenants:return order_login_response()
|
||
if intent['unsupported']:return unsupported()
|
||
requested=intent.get('summaries') or [intent]
|
||
summaries=[part for part in requested if not part['unsupported']]
|
||
if not summaries:return unsupported()
|
||
query=filters(summaries[-1])
|
||
comparisons=[filters(part) for part in summaries[:-1]]
|
||
try:
|
||
data=read_orders(tenants,query,comparisons=comparisons)
|
||
total=data['total']
|
||
lines=[]
|
||
for part,count,groups in zip(summaries[:-1],data['counts'],data['comparison_groups']):
|
||
lines.append(summary_text(part,count)+'。')
|
||
if part.get('grouped'):lines.extend(stage_distribution(groups))
|
||
lines.append(summary_text(query,total)+f'，其中 {data["with_inbound"]} 票已登记入仓码。')
|
||
if summaries[-1].get('grouped'):
|
||
lines.extend(stage_distribution(data['groups']))
|
||
for part in requested:
|
||
if part['unsupported']:
|
||
question=re.sub(r'\s+',' ',part['question'])[:120]
|
||
lines.append(f'“{question}”的筛选条件尚未明确，请确认具体含义，可按业务状态、异常或当前实际节点查询。')
|
||
if data['rows']:
|
||
lines.append('已记录的异常原因（最多5票）。' if reasons else f'最近更新的{len(data["rows"])}票：')
|
||
lines.extend(row_text(row,reason_only=reasons) for row in data['rows'])
|
||
lines.append('回复“查看全部”展开列表，或发送入仓码查询详情。')
|
||
state={'query':query,'page':1,'size':5}
|
||
previous=load_state(conv_id)
|
||
if previous and previous['query']==query:state={key:previous[key] for key in ('query','page','size')}
|
||
state['shown']=[{'inbound_code':row['inbound_code'],'order_no':row['order_no']} for row in data['rows']]
|
||
return {'t':'\n'.join(lines),'_customer_data':True,'_order_list_state':state}
|
||
except sqlite3.Error:return {'t':'订单信息暂时无法读取，请稍后重试。'}
|
||
|
||
|
||
def answer_list(message,tenants,conv_id=None):
|
||
from .order_data import is_lookup_help,STATUS_LABELS
|
||
from .customer_service import order_login_response
|
||
command=page_command(message)
|
||
expand=message.strip(' ，,。.!！?？') in ('查看全部','查看全部订单')
|
||
initial=(not is_lookup_help(message) and re.search(r'订单|运单|\borders?\b|\bshipments?\b',message,re.I)
|
||
and re.search(r'列出|列表|全部|所有|\blist\b',message,re.I))
|
||
if command is None and not initial and not expand:return None
|
||
if not tenants:return order_login_response()
|
||
try:
|
||
previous=load_state(conv_id)
|
||
if command is not None or expand:
|
||
if not previous:return {'t':'请先在当前对话中发送“列出我的订单”或查询订单数量，再翻页查看。'}
|
||
query=previous['query'];size=20 if expand else previous['size']
|
||
page=1 if expand else int(command) if isinstance(command,int) else max(1,previous['page']+(1 if command=='next' else -1))
|
||
if not 1<=page<=10000:return {'t':'请提供1至10000之间的有效页码。'}
|
||
else:
|
||
intent=parse_query(message,listing=True)
|
||
if intent is None or intent['unsupported']:return unsupported(True)
|
||
query=filters(intent);page=1;size=20
|
||
data=read_orders(tenants,query,page,size);page=data['page'];total=data['total']
|
||
state={'query':query,'page':page,'size':size,
|
||
'shown':[{'inbound_code':row['inbound_code'],'order_no':row['order_no']} for row in data['rows']]}
|
||
if not total:return {'t':'当前账号下没有符合条件的订单。','_customer_data':True,'_order_list_state':state}
|
||
label=STATUS_LABELS[query['status']] if query['status'] else '全部'
|
||
first=f'先列出{size}票；' if page==1 and total>size else ''
|
||
extra=('；当前节点：'+STAGE_LABELS[query['stage']] if query['stage'] else '')+('；明确标记异常' if query['exception'] else '')
|
||
lines=[f'您当前的订单列表（{first}第{page}页；业务状态：{label}{extra}）：',f'共{total}票，当前显示第{(page-1)*size+1}至{(page-1)*size+len(data["rows"])}票。']
|
||
lines.extend(row_text(row) for row in data['rows'])
|
||
lines.append('回复“下一页”查看后续订单。' if page*size<total else '当前已是最后一页。')
|
||
lines.append('发送入仓码可查询该票订单和物流轨迹。')
|
||
return {'t':'\n'.join(lines),'_customer_data':True,'_order_list_state':state}
|
||
except sqlite3.Error:return {'t':'订单信息暂时无法读取，请稍后重试。'}
|