81 lines
3.0 KiB
Python
81 lines
3.0 KiB
Python
"""Extracted tracking rules with unchanged business behavior."""
|
||
|
||
import json
|
||
import random
|
||
import re
|
||
from datetime import datetime, timedelta
|
||
|
||
CARD_PREFIX = "__CARD__"
|
||
|
||
def card_to_text(content):
|
||
"""把存库的卡片标记转成人话，避免原始 JSON 进入模型上下文"""
|
||
if not content.startswith(CARD_PREFIX):
|
||
return content
|
||
try:
|
||
card = json.loads(content[len(CARD_PREFIX):])
|
||
if card.get('kind') == 'order':
|
||
return card.get('summary', '')
|
||
return f"（历史演示轨迹，非真实物流信息：单号 {card.get('no', '')}，演示状态：{card.get('status', '')}）"
|
||
except Exception:
|
||
return "（已为该运单查询过轨迹）"
|
||
|
||
def is_tracking_no(text):
|
||
"""判断一句话是不是"整句运单号"
|
||
|
||
规则（逐条收紧，避免误判）：
|
||
1. 去掉空格和横杠后，长度 8~20 位，且只能由字母数字组成
|
||
2. 至少包含 3 个数字（英文句子通常只有 1 个数字）
|
||
3. 字母数量不能多于数字（"i want to go 1t how money" 这类会被排除）
|
||
纯数字单号（如 15252125）全部满足上述条件。
|
||
"""
|
||
# 日期格式（如 2026-10-23、2026/10/23）不算运单号
|
||
if re.fullmatch(r"\d{4}[-/.]\d{1,2}[-/.]\d{1,2}", text.strip()):
|
||
return False
|
||
t = re.sub(r"[\s\-]", "", text)
|
||
if not (8 <= len(t) <= 20):
|
||
return False
|
||
if not re.fullmatch(r"[A-Za-z0-9]+", t):
|
||
return False
|
||
digits = sum(c.isdigit() for c in t)
|
||
letters = sum(c.isalpha() for c in t)
|
||
if digits < 3:
|
||
return False
|
||
if letters > digits:
|
||
return False
|
||
return True
|
||
|
||
def extract_tracking_no(text):
|
||
"""从一句话里提取运单号：先看整句是不是，再从句子里找最长的字母数字串"""
|
||
t = text.strip()
|
||
if is_tracking_no(t):
|
||
return t
|
||
# 找句子里最长的 8~20 位字母数字连续串
|
||
candidates = re.findall(r"[A-Za-z0-9]{8,20}", t)
|
||
best = None
|
||
for c in candidates:
|
||
if is_tracking_no(c):
|
||
if best is None or len(c) > len(best):
|
||
best = c
|
||
return best
|
||
|
||
def build_track(no):
|
||
"""Legacy demo helper for old imports; chat queries use order_test_data."""
|
||
rnd = random.Random(no)
|
||
steps = [
|
||
("已揽收", "义乌仓"),
|
||
("出口清关放行", "义乌/宁波口岸"),
|
||
("已发车", "中欧班列/卡航 · 阿拉山口"),
|
||
("到达欧洲中转仓", "波兰 · 华沙"),
|
||
("欧盟清关放行", "德国 · 杜伊斯堡"),
|
||
("派送中", "收件人所在城市"),
|
||
]
|
||
gaps = [rnd.randint(6, 40) for _ in steps] # 相邻节点间隔（小时）
|
||
t = datetime.now() - timedelta(hours=sum(gaps))
|
||
nodes = []
|
||
for (text, place), gap in zip(steps, gaps):
|
||
t = t + timedelta(hours=gap)
|
||
nodes.append({"time": t.strftime("%Y-%m-%d %H:%M"),
|
||
"text": text, "place": place})
|
||
nodes.reverse() # 最新的排最上面
|
||
return {"no": no, "status": nodes[0]["text"], "nodes": nodes}
|