接入本地订单测试表并替换模拟查单
This commit is contained in:
@@ -0,0 +1,181 @@
|
||||
"""Local, editable order test data. This repository can later be replaced by MCP."""
|
||||
|
||||
import json
|
||||
import re
|
||||
import sqlite3
|
||||
from datetime import datetime
|
||||
|
||||
from .config import get_setting
|
||||
|
||||
|
||||
FIELDS = ('source_order_id', 'tenant_id', 'order_no', 'inbound_code', 'shipment_id',
|
||||
'status', 'country_code', 'transport_mode', 'warehouse_code', 'order_role',
|
||||
'parent_order_no', 'package_count', 'gross_weight_kg', 'volume_cbm',
|
||||
'history_json', 'data_note', 'source_updated_at', 'snapshot_at')
|
||||
STATUS_LABELS = {'DRAFT': '草稿', 'SUBMITTED': '已提交', 'APPROVED': '审核通过',
|
||||
'ACCEPTED': '已接单', 'REJECTED': '审核驳回', 'PROCESSING': '履约中',
|
||||
'COMPLETED': '头程订单已完成', 'CANCELLED': '已取消'}
|
||||
FIELD_LABELS = {'inbound':'入仓码', 'shipment':'Shipment ID', 'country':'目的国',
|
||||
'warehouse':'目的仓', 'transport':'运输方式', 'packages':'件数',
|
||||
'weight':'申报毛重', 'volume':'申报体积', 'role':'订单层级',
|
||||
'parent':'父订单号', 'updated':'来源更新时间'}
|
||||
IDENTIFIER = re.compile(r'(?<![A-Za-z0-9_-])[A-Za-z0-9][A-Za-z0-9_-]{5,63}(?![A-Za-z0-9_-])')
|
||||
|
||||
|
||||
class OrderDataUnavailable(Exception):
|
||||
pass
|
||||
|
||||
|
||||
def init_order_table(connection):
|
||||
connection.execute('''CREATE TABLE IF NOT EXISTS order_test_data (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
source_order_id TEXT, tenant_id TEXT NOT NULL DEFAULT 'test',
|
||||
order_no TEXT NOT NULL COLLATE NOCASE,
|
||||
inbound_code TEXT COLLATE NOCASE, shipment_id TEXT COLLATE NOCASE,
|
||||
status TEXT NOT NULL, country_code TEXT, transport_mode TEXT, warehouse_code TEXT,
|
||||
order_role TEXT, parent_order_no TEXT,
|
||||
package_count INTEGER, gross_weight_kg TEXT, volume_cbm TEXT,
|
||||
history_json TEXT NOT NULL DEFAULT '[]', data_note TEXT,
|
||||
source_updated_at TEXT, snapshot_at TEXT,
|
||||
UNIQUE(tenant_id, order_no)
|
||||
)''')
|
||||
connection.execute('CREATE INDEX IF NOT EXISTS idx_order_test_inbound ON order_test_data(inbound_code)')
|
||||
connection.execute('CREATE INDEX IF NOT EXISTS idx_order_test_shipment ON order_test_data(shipment_id)')
|
||||
|
||||
|
||||
def import_records(connection, records, replace=False):
|
||||
"""Import an explicitly supplied snapshot; preserve manual edits unless requested."""
|
||||
init_order_table(connection)
|
||||
inserted = skipped = 0
|
||||
for source in records:
|
||||
if not isinstance(source, dict) or not str(source.get('order_no') or '').strip():
|
||||
raise ValueError('Each record requires an order_no')
|
||||
row = {key: source.get(key) for key in FIELDS}
|
||||
row['tenant_id'] = str(row['tenant_id'] or 'test')
|
||||
row['source_order_id'] = str(row['source_order_id']) if row['source_order_id'] is not None else None
|
||||
row['order_no'] = str(row['order_no']).strip()
|
||||
row['status'] = str(row['status'] or 'DRAFT')
|
||||
history = row['history_json'] or []
|
||||
if isinstance(history, str): history = json.loads(history)
|
||||
if not isinstance(history, list): raise ValueError('history_json must contain a list')
|
||||
row['history_json'] = json.dumps(history, ensure_ascii=False)
|
||||
row['snapshot_at'] = row['snapshot_at'] or datetime.now().isoformat(timespec='seconds')
|
||||
exists = connection.execute('SELECT 1 FROM order_test_data WHERE tenant_id=? AND order_no=?',
|
||||
(row['tenant_id'], row['order_no'])).fetchone()
|
||||
if exists and not replace:
|
||||
skipped += 1
|
||||
continue
|
||||
columns = ','.join(FIELDS)
|
||||
placeholders = ','.join('?' for _ in FIELDS)
|
||||
updates = ','.join(key+'=excluded.'+key for key in FIELDS if key not in ('tenant_id', 'order_no'))
|
||||
connection.execute(f'INSERT INTO order_test_data ({columns}) VALUES ({placeholders}) '
|
||||
f'ON CONFLICT(tenant_id,order_no) DO UPDATE SET {updates}', [row[key] for key in FIELDS])
|
||||
inserted += 1
|
||||
return {'imported': inserted, 'preserved': skipped}
|
||||
|
||||
|
||||
def lookup_order(reference):
|
||||
"""Exact, bounded lookup; scope comes from server configuration, never model SQL."""
|
||||
from .db import connect
|
||||
|
||||
connection = connect()
|
||||
connection.row_factory = sqlite3.Row
|
||||
try:
|
||||
query = 'SELECT * FROM order_test_data WHERE (order_no=? OR inbound_code=? OR shipment_id=?)'
|
||||
values = [reference] * 3
|
||||
tenant = get_setting('ORDER_DATA_TENANT_ID')
|
||||
if tenant:
|
||||
query += ' AND tenant_id=?'
|
||||
values.append(str(tenant))
|
||||
return [dict(row) for row in connection.execute(query+' ORDER BY id LIMIT 2', values)]
|
||||
except sqlite3.Error:
|
||||
raise OrderDataUnavailable from None
|
||||
finally:
|
||||
connection.close()
|
||||
|
||||
|
||||
def extract_reference(message):
|
||||
message = re.sub(r'\[(?:图片|文件)\]/uploads/\S+|https?://\S+', '', message)
|
||||
references = []
|
||||
for token in IDENTIFIER.findall(message):
|
||||
if sum(character.isdigit() for character in token) < 3:
|
||||
continue
|
||||
if re.fullmatch(r'\d{4}[-_]\d{2}[-_]\d{2}|\d+(?:KG|LBS?|CBM|CM|MM)', token, re.I):
|
||||
continue
|
||||
if token.casefold() not in [value.casefold() for value in references]:
|
||||
references.append(token)
|
||||
return references
|
||||
|
||||
|
||||
def build_card(row):
|
||||
fields = []
|
||||
def field(key, value):
|
||||
if value is not None and str(value).strip(): fields.append({'key': key, 'value': str(value)})
|
||||
field('inbound', row['inbound_code'])
|
||||
field('shipment', row['shipment_id'])
|
||||
field('country', row['country_code'])
|
||||
field('warehouse', row['warehouse_code'])
|
||||
field('transport', {'SEA':'海运', 'AIR':'空运', 'RAIL':'铁路', 'TRUCK':'卡航'}.get(row['transport_mode'], row['transport_mode']))
|
||||
field('packages', row['package_count'])
|
||||
field('weight', str(row['gross_weight_kg'])+' kg' if row['gross_weight_kg'] is not None else None)
|
||||
field('volume', str(row['volume_cbm'])+' m³' if row['volume_cbm'] is not None else None)
|
||||
field('role', {'CHILD':'子订单', 'MASTER':'主订单'}.get(row['order_role']))
|
||||
field('parent', row['parent_order_no'])
|
||||
field('updated', row['source_updated_at'])
|
||||
try:
|
||||
history = json.loads(row['history_json'])
|
||||
except (ValueError, TypeError):
|
||||
history = []
|
||||
nodes = [{'time': str(event.get('time') or ''),
|
||||
'text': STATUS_LABELS.get(str(event.get('status')), str(event.get('status') or '状态更新')), 'place': ''}
|
||||
for event in history[-50:] if isinstance(event, dict)] if isinstance(history, list) else []
|
||||
nodes.reverse()
|
||||
status = STATUS_LABELS.get(row['status'], row['status'])
|
||||
summary = f"订单号 {row['order_no']}，业务状态：{status}。"
|
||||
summary += '；'.join(FIELD_LABELS[item['key']]+'：'+item['value'] for item in fields)
|
||||
return {'kind':'order', 'no':row['order_no'], 'status':status, 'fields':fields,
|
||||
'nodes':nodes, 'summary':summary, 'snapshot_at':row['snapshot_at'],
|
||||
'notice':row['data_note'] or '', 'source':'editable_test_table'}
|
||||
|
||||
|
||||
def answer_query(message, conv_id):
|
||||
references = extract_reference(message)
|
||||
if len(references) > 1:
|
||||
return {'t':'请一次提供一个订单号、入仓码或 Shipment ID，避免查错订单。'}
|
||||
reference = references[0] if references else None
|
||||
if not reference and conv_id and re.search(r'这(?:个|票|单)|刚才|状态|重量|几件|体积|目的地|运费|费用', message):
|
||||
from .db import connect
|
||||
connection = connect()
|
||||
try:
|
||||
recent = connection.execute('SELECT role,content FROM messages WHERE conv_id=? ORDER BY id DESC LIMIT 12', (conv_id,)).fetchall()
|
||||
finally:
|
||||
connection.close()
|
||||
for role, content in recent:
|
||||
if role == 'user':
|
||||
earlier_references = extract_reference(content)
|
||||
if earlier_references:
|
||||
reference = earlier_references[0] if len(earlier_references) == 1 else None
|
||||
break
|
||||
if role == 'assistant' and content.startswith('__CARD__'):
|
||||
try:
|
||||
previous = json.loads(content[8:])
|
||||
if previous.get('kind') == 'order':
|
||||
reference = previous['no']
|
||||
break
|
||||
except (ValueError, KeyError, TypeError):
|
||||
pass
|
||||
if not reference:
|
||||
if re.search(r'查.*(?:订单|运单|物流轨迹)|订单状态|shipment\s*id|(?:列出|搜索|所有|全部|多少).*(?:订单|入仓)|订单.*(?:多少|列表)', message, re.I):
|
||||
return {'t':'请提供订单号、入仓码或 Shipment ID，我会查询本地订单测试表中的记录。'}
|
||||
return None
|
||||
try:
|
||||
matches = lookup_order(reference)
|
||||
except OrderDataUnavailable:
|
||||
return {'t':'订单测试表暂时无法读取，请稍后重试；当前无法确认该订单状态。'}
|
||||
if not matches:
|
||||
return {'t':f'订单测试表中未找到“{reference}”。请核对订单号、入仓码或 Shipment ID，或联系人工客服。'}
|
||||
if len(matches) > 1:
|
||||
return {'t':'该标识匹配到多个订单，请提供该订单唯一的订单号或入仓码。'}
|
||||
if not references and re.search(r'运费|费用|多少钱|价格', message):
|
||||
return {'t':'这份订单测试数据没有实际运费字段，无法据此确认该订单费用，请联系人工客服核实。'}
|
||||
return {'card': build_card(matches[0])}
|
||||
Reference in New Issue
Block a user