286 lines
16 KiB
Python
286 lines
16 KiB
Python
"""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
|
||
from .order_tracking import build_tracking
|
||
from .visitor_service import customer_tenant_ids
|
||
from .customer_service import order_login_response
|
||
|
||
|
||
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', 'tracking_json', 'data_note', 'source_updated_at', 'snapshot_at')
|
||
STATUS_LABELS = {'DRAFT': '草稿', 'SUBMITTED': '已提交', 'APPROVED': '审核通过',
|
||
'ACCEPTED': '已接单', 'REJECTED': '审核驳回', 'PROCESSING': '履约中',
|
||
'COMPLETED': '头程订单已完成', 'CANCELLED': '已取消'}
|
||
FIELD_LABELS = {'inbound':'入仓码', 'number':'订单号', '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_-])')
|
||
ORDER_DETAIL = r'状态|进度|轨迹|到哪|开船|到港|清关|提柜|还柜|签收|预计|什么时候|何时|多久|重量|毛重|体积|件数|箱数|目的地|目的国|目的仓|仓库|父订单|运费|费用|价格|几件|几箱|公斤|千克|多少钱'
|
||
|
||
|
||
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 '[]', tracking_json TEXT NOT NULL DEFAULT '[]', data_note TEXT,
|
||
source_updated_at TEXT, snapshot_at TEXT,
|
||
UNIQUE(tenant_id, order_no)
|
||
)''')
|
||
columns = {row[1] for row in connection.execute('PRAGMA table_info(order_test_data)')}
|
||
if 'tracking_json' not in columns:
|
||
connection.execute("ALTER TABLE order_test_data ADD COLUMN tracking_json TEXT NOT NULL DEFAULT '[]'")
|
||
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)
|
||
tracking = row['tracking_json'] or []
|
||
if isinstance(tracking, str): tracking = json.loads(tracking)
|
||
if not isinstance(tracking, list): raise ValueError('tracking_json must contain a list')
|
||
row['tracking_json'] = json.dumps(tracking, 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 import_tracking_records(connection, records, replace=False):
|
||
"""Update tracking alone; never replace hand-edited order fields or add unknown orders."""
|
||
init_order_table(connection)
|
||
imported = preserved = missing = 0
|
||
for record in records:
|
||
if not isinstance(record, dict) or not record.get('tenant_id') or not record.get('order_no'):
|
||
raise ValueError('Tracking record requires tenant_id and order_no')
|
||
tracking = record.get('tracking_json')
|
||
if isinstance(tracking, str): tracking = json.loads(tracking)
|
||
if not isinstance(tracking, list): raise ValueError('tracking_json must contain a list')
|
||
row = connection.execute('SELECT id,source_order_id,tracking_json FROM order_test_data '
|
||
'WHERE tenant_id=? AND order_no=?',
|
||
(str(record['tenant_id']), record['order_no'])).fetchone()
|
||
if not row:
|
||
missing += 1
|
||
continue
|
||
if not record.get('source_order_id') or str(record['source_order_id']) != str(row[1]):
|
||
raise ValueError('Tracking source_order_id does not match local order')
|
||
if not replace and row[2] not in (None, '', '[]'):
|
||
preserved += 1
|
||
continue
|
||
connection.execute('UPDATE order_test_data SET tracking_json=? WHERE id=?',
|
||
(json.dumps(tracking, ensure_ascii=False), row[0]))
|
||
imported += 1
|
||
return {'imported': imported, 'preserved': preserved, 'missing': missing}
|
||
|
||
|
||
def lookup_order(reference, tenant_id=None, tenant_ids=None):
|
||
"""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_ids is not None:
|
||
if not isinstance(tenant_ids,(tuple,list,set)) or not 0 < len(tenant_ids) <= 256:
|
||
return []
|
||
allowed = {value for value in tenant_ids if isinstance(value,str) and 0 < len(value.strip()) <= 128}
|
||
if tenant_id is not None:
|
||
allowed.intersection_update({str(tenant_id)})
|
||
if tenant:
|
||
allowed.intersection_update({str(tenant)})
|
||
if not allowed:
|
||
return []
|
||
query += ' AND tenant_id IN (' + ','.join('?' for _ in allowed) + ')'
|
||
values.extend(sorted(allowed))
|
||
elif tenant_id is not None:
|
||
if not str(tenant_id).strip():
|
||
return []
|
||
if tenant and str(tenant_id) != str(tenant):
|
||
return []
|
||
tenant = str(tenant_id)
|
||
if tenant and tenant_ids is None:
|
||
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)
|
||
message = re.sub(r'(?<![A-Za-z0-9_-])\d{4}-\d{1,2}-\d{1,2}'
|
||
r'(?:[T ]\d{1,2}:\d{2}(?::\d{2}(?:\.\d+)?)?(?:Z|[+-]\d{2}:?\d{2})?)?'
|
||
r'(?![A-Za-z0-9_-])', ' ', message)
|
||
message = re.sub(r'(?<![A-Za-z0-9_.-])\d+(?:\.\d+)?\s*[*xX×]\s*\d+(?:\.\d+)?\s*[*xX×]\s*\d+(?:\.\d+)?'
|
||
r'(?:\s*(?:cm|mm|m|厘米|毫米|米))?(?![A-Za-z0-9_.-])', ' ', message, flags=re.I)
|
||
message = re.sub(r'(?<![A-Za-z0-9_-])\d+(?:\.\d+)?\s*'
|
||
r'(?:kg|lbs?|cbm|cm|mm|公斤|千克|克|吨|立方米|方|m[³3]|箱|件)'
|
||
r'(?![A-Za-z0-9_])', ' ', message, flags=re.I)
|
||
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'])
|
||
if row['inbound_code']:
|
||
field('number', row['order_no'])
|
||
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'])
|
||
inbound = row['inbound_code']
|
||
summary = f"入仓码 {inbound}，业务状态：{status}。" if inbound else f"订单号 {row['order_no']}，业务状态：{status}。"
|
||
summary += '；'.join(FIELD_LABELS[item['key']]+'：'+item['value'] for item in fields if item['key'] != 'inbound')
|
||
card = {'kind':'order', 'no':row['order_no'], 'status':status, 'fields':fields,
|
||
'nodes':nodes, 'summary':summary}
|
||
tracking = build_tracking(row.get('tracking_json'))
|
||
if tracking:
|
||
card.update(node_kind='logistics', nodes=tracking['nodes'],
|
||
estimated_nodes=tracking['estimated_nodes'], tracking_status=tracking['status'])
|
||
facts = tracking['nodes'][:10] + tracking['estimated_nodes'][:5]
|
||
card['summary'] += '。物流状态：' + (tracking['status'] or '暂无实际节点') + '。'
|
||
card['summary'] += '；'.join(f"{node['time']} {node['text']} {node['place']}" for node in facts)
|
||
return card
|
||
|
||
|
||
def is_order_followup(message):
|
||
"""Reuse an order only for an explicit reference or a short detail question."""
|
||
if not re.search(ORDER_DETAIL, message):
|
||
return False
|
||
if re.search(r'(?:这|那|该)(?:个|票|单|笔)|该订单|刚才|上一(?:票|单|个订单)', message):
|
||
return True
|
||
question = re.sub(r'[\s，,。.!！?？、：:]', '', message)
|
||
details = (
|
||
r'(?:(?:(?:物流|运输)?(?:状态|进度|轨迹)|重量|毛重|体积|件数|箱数|目的地|目的国|目的仓|仓库|父订单|运费|费用|价格)'
|
||
r'(?:(?:是|有|为)?(?:多少|什么|怎么样|怎样|如何|哪个))?|(?:有)?(?:几件|几箱|多少公斤|多少千克|多少钱))'
|
||
r'(?:呢|吗|呀|啊)?')
|
||
tracking = (
|
||
r'(?:(?:到哪(?:里)?|开船|到港|清关|提柜|还柜|签收)(?:了)?(?:呢|吗|没|了吗|没有)?|'
|
||
r'(?:预计)?(?:什么时候|何时|多久)(?:能|会|可以)?(?:到|到达|到港|到仓|到货|送到)(?:呢|呀|啊)?|还要多久)')
|
||
return bool(re.fullmatch(
|
||
r'(?:那|那么|还有|请问|请|查一下|看一下|现在|目前)?(?:它的|订单的)?'
|
||
r'(?:'+details+'|'+tracking+')', question))
|
||
|
||
|
||
def answer_query(message, conv_id):
|
||
references = extract_reference(message)
|
||
tenants = customer_tenant_ids()
|
||
if references and not tenants:
|
||
return order_login_response()
|
||
if len(references) > 1:
|
||
return {'t':'请一次提供一个入仓码，避免查错订单。'}
|
||
reference = references[0] if references else None
|
||
if not reference and conv_id and is_order_followup(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 not is_order_followup(content):
|
||
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|'
|
||
r'(?:列出|搜索|所有|全部|多少|我的).*(?:订单|运单|入仓)|(?:订单|运单).*(?:多少|列表|查询|在哪|到哪|进度)|'
|
||
r'^\s*(?:查单|查询|查一下|帮我查一下)[。？！?!\s]*$|'
|
||
r'\b(?:my|find|lookup|track)\s+(?:orders?|shipments?)\b|\b(?:order|shipment)\s+(?:status|tracking|lookup)\b|'
|
||
r'\b(?:track|lookup|find|search)\b.*\binbound\b|\binbound\s+(?:code|number)\s+(?:lookup|tracking|status)\b', message, re.I):
|
||
return order_login_response() if not tenants else {'t':'请提供入仓码查询订单和物流轨迹。'}
|
||
return None
|
||
if not tenants:
|
||
return order_login_response()
|
||
try:
|
||
matches = lookup_order(reference,tenant_id=tenants[0]) if len(tenants) == 1 else lookup_order(reference,tenant_ids=tenants)
|
||
except OrderDataUnavailable:
|
||
return {'t':'订单信息暂时无法读取，请稍后重试。'}
|
||
if not matches:
|
||
return {'t':f'未找到“{reference}”对应的订单。请核对入仓码，或联系人工客服。'}
|
||
if len(matches) > 1:
|
||
return {'t':'该标识匹配到多个订单，请提供完整入仓码或联系人工客服确认。'}
|
||
if re.search(r'运费|费用|多少钱|价格', message):
|
||
return {'t':'该订单未记录实际运费，请联系人工客服核实。'}
|
||
return {'card': build_card(matches[0])}
|