Files
baiyue-logistics-chat/am.py
T

1513 lines
63 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
from flask import (Flask, render_template, request, jsonify, Response,
stream_with_context, send_from_directory)
import os
import math
import shutil
import requests
import json
import time
import random
import re
import sqlite3
import logging
from datetime import datetime, timedelta
# 日志写到 logs/ 文件夹
LOG_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "logs")
os.makedirs(LOG_DIR, exist_ok=True)
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(message)s",
handlers=[
logging.FileHandler(os.path.join(LOG_DIR, "app.log"), encoding="utf-8"),
logging.StreamHandler()
]
)
try:
import rates_extra as RX # 报价表扩展数据(卡派按方、海外仓、附加费、岛屿邮编)
except Exception: # 缺失时不影响主流程
RX = None
app = Flask(__name__)
# Key 不写进代码,从系统环境变量读取(setx DEEPSEEK_API_KEY "sk-...")
DEEPSEEK_API_KEY = os.environ.get("DEEPSEEK_API_KEY", "").strip()
URL = "https://api.deepseek.com/v1/chat/completions"
MODEL_NAME = "deepseek-chat"
# ===== 简单限流:每个 IP 每分钟最多 60 次(真人够用,挡脚本狂刷)=====
from collections import defaultdict
RATE_LIMIT_PER_MIN = 60
_hits = defaultdict(list)
@app.before_request
def rate_limit():
p = request.path
if p.startswith("/static") or p.startswith("/api/human/"):
return None
ip = request.remote_addr or "unknown"
now = time.time()
arr = _hits[ip]
while arr and now - arr[0] > 60:
arr.pop(0)
if len(arr) >= RATE_LIMIT_PER_MIN:
return jsonify({"e": "请求太频繁,请稍后几秒再试", "d": 1}), 429
arr.append(now)
return None
@app.after_request
def no_cache_html(resp):
"""页面不走浏览器缓存,改完代码刷新就能看到;图片等静态资源不受影响。"""
if resp.mimetype in ("text/html", "text/event-stream"):
resp.headers["Cache-Control"] = "no-store, no-cache, must-revalidate"
resp.headers["Pragma"] = "no-cache"
resp.headers["Expires"] = "0"
return resp
knowledge_list = [
{
"keys":["公司","柏悦","柏悦物流","介绍","哪家","哪家货代","你们是哪家"],
"text":"浙江柏悦物流位于义乌,主营中欧跨境物流专线,依托义乌、宁波口岸优势,服务中小跨境卖家。主要渠道:欧洲海运快线、义乌-中欧铁路、中欧卡航、欧洲空运(空派)、英国空运,均支持欧洲、英国双清包税与VAT递延。义乌、泉州、广州、深圳均可交仓。"
},
{
"keys":["海运","船运","海派","海运快线","樱花快线"],
"text":"欧洲海运快线(双清/递延)单价:21KG+ 12.3元/kg,101KG+ 10.3元/kg,500KG+ 10.2元/kg,2000KG+ 10.0元/kg。时效:开船后38-45天提取。可走樱花快线(+0.3元/kg),递延价减1.7元/kg。主派送为UPS/DHL/DPD/CWE统配。"
},
{
"keys":["铁路","中欧班列","班列","义乌铁路","火车"],
"text":"义乌-中欧铁路专线单价:21KG+ 15.5元/kg,100KG+ 和 1000KG+ 均为13.5元/kg。时效:发车后36-40天提取。VAT递延减1元/kg。只限送义乌仓,义乌装柜经阿拉山口/霍尔果斯出境。"
},
{
"keys":["卡航","卡车","卡派"],
"text":"中欧卡航单价:21KG+ 19.4元/kg,101KG+ 和 500KG+ 17.4元/kg,2000KG+ 及 4000KG+ 17.3元/kg。时效:发车后欧盟16-20天提取。递延减1.5元/kg,普卡时效18-22天。纯纺织品另加5元/kg。"
},
{
"keys":["空运","空派","航空","欧洲空运"],
"text":"欧洲空运专线(空派)单价:21KG+ 39元/kg,45KG+ 38元/kg,71KG+ 37元/kg,101KG+ 36元/kg。时效:10-13天左右提取;快线8-13天,直飞快线12-15天。一票多件单件不足12KG按12KG计费。"
},
{
"keys":["英国海运","英国铁路","中英卡航","英国专线"],
"text":"英国专线(DPD)双清价格:英国海运21KG+ 12.8元/kg、101KG+/301KG+ 10.9元/kg,时效约40天提取;英国铁路21KG+ 18.0元/kg、101KG+ 16.2元/kg,发车后25-30天提取;中英卡航21KG+ 21.4元/kg、101KG+ 19.4元/kg,发车后20-22天提取。VAT自税/递延价更低(海运10.0/8.1、铁路15/13.2、卡航18.4/16.4)。北极航线英国海运25-28天,价格单询。卡航单件不超30kg、尺寸≤98×68×59cm。"
},
{
"keys":["英国空运","英国","UK","伦敦"],
"text":"英国空运专线(双清)单价:22KG+ 48元/kg,51KG+ 44元/kg,101KG+ 42元/kg,时效8-10个自然日提取。VAT自税/递延为44/40/38元/kg,带电货47.5/42.5/41.5元/kg。拒收北爱尔兰、苏格兰高地等偏远地址。"
},
{
"keys":["报价","价格","费用","计费","怎么收费","计费重量","多少钱","运费","单价"],
"text":"运费计算方式:阶梯单价 × 计费重量。计费重量 = 实重与材积重取大值,材积 = 长×宽×高(cm)÷6000,即1CBM=167KG。各渠道最低起收21KG(英国空运22KG)。单件计费重不足13KG的渠道(铁路、卡航)按5元/kg加收超轻费;空运单件不足12KG按12KG计费。偏远地址加7元/kg,单票最低250元。价格不含税,税点以开出的发票为准。"
},
{
"keys":["时效","多久","要几天","多长时间"],
"text":"各渠道时效:欧洲海运快线开船后38-45天提取;中欧铁路发车后36-40天提取;中欧卡航发车后16-20天提取;欧洲空运10-13天提取;英国空运8-10个自然日提取。海关查验、罢工、节假日等不可抗力导致的时间不计入。"
},
{
"keys":["揽收","上门取件","上门收货","交仓","交货"],
"text":"义乌、深圳(龙华观澜、宝安福园)、广州(石井)、泉州均可交仓,义乌仓10:00-21:00收货。欧洲空运交仓义乌+1元/kg,泉州、广州+0.5元/kg;海运、铁路、卡航泉州交货+0.5元/kg。"
},
{
"keys":["入仓","进仓","仓库收货","仓库地址","仓库在哪","义乌仓","交仓地址","送仓"],
"text":"义乌仓:义乌市稠江街道梦娜袜业德贸路东三门进,李东文 19558263692(10:00-21:00)。深圳龙华:龙华区观澜街道君子布社区张一村路黄背坑路1号B栋,张长印 18688937838。深圳宝安:福园二路九号海亨通物流B区C09-C14。广州:恒河沙仓储石井仓211,潘志红 13789206500。送货需打印交接清单随货交仓。"
},
{
"keys":["理赔","货损","丢件","赔付","破损","压坏","索赔"],
"text":"遗失赔付:未提取上网遗失按申报价值赔偿,最高不超过40元/kg,不退运费;已提取遗失按服务商标准,最高100USD/票。开查时间:交仓后7天未提取可申请开查,开查30天无果走索赔。签收后亚马逊未上架需2周内提出调查,包裹异常需1个月内提出。外包装无损的内件丢失不支持理赔,易碎品损坏不支持理赔。可自购保险,保费为货值千分之三(电池千分之五)。"
},
{
"keys":["超轻费","计费重量不足","最低计费"],
"text":"单件计费重量不得低于13KG(空运12KG),可整票按平均13KG算,不足部分按5元/kg加收超轻费。例:单箱只有12KG,加收5×1=5元超轻费。欧洲热门仓卡派直送要求单件平均计费重12KG,不足按2元/kg加收派送费。"
},
{
"keys":["偏远","偏远费","偏远地址"],
"text":"偏远地址加收7元/kg(不含燃油),单票偏远费最低消费250元/票。需要自行到UPS/DPD官网查询目的地址是否偏远,我司出货时不另行通知,以账单为准,偏远费6个月内有效。"
},
{
"keys":["超重","超长","超大","尺寸限制"],
"text":"欧洲空运最长边≤99cm、次长边≤75cm、实重<24KG。英国DPD单件≤30KG,尺寸100×70×60cm以内,超重或超长各加收330元/件。UPS单边最长≤99cm、次边≤75cm、单件<24KG,超出加收240元+100元。热门仓卡派直送:单件超40KG加收30元/件。"
},
{
"keys":["敏感","带电","电池","液体","粉末","禁品","不能寄","违禁品","拒收","拒接"],
"text":"可接带标准箱100瓦以下非危电池/内电产品(C卡航、海运快线均可),资料齐全加1元/kg,资料不齐加2元/kg;手机、平板电脑加5元/kg。拒接:纯液体、纯粉末、药品、食品、易燃易爆、仿牌侵权、木制品(空运)、陶瓷手表等反倾销品、电子烟、管制刀具、枪支。电池第一个托收400元/件,第二个起100元/件,600元封顶。"
},
{
"keys":["清关","递延","VAT","双清","包税","DDP"],
"text":"欧洲、英国渠道均支持双清包税与VAT递延(只付关税,增值税销售后缴纳)。递延需提供法人护照/身份证、欧盟VAT和EORI号及回执、营业执照、4份模板文件原件扫描件PDF。罗马尼亚等欧盟国家有二次清关风险,二次关税由客户自行缴纳。"
},
{
"keys":["报关","单独报关","报关费"],
"text":"单独报关费用:空运350元/票,海运200元/票,铁路250元/票,卡航200元/票,续页费50元/页。一达通报关500元/票。单票品名最多5个,第6个起加收30元/个。报价不含税,税点以开出的发票为准。"
},
{
"keys":["亚马逊","FBA","海外仓","仓库配送","卡派"],
"text":"支持FBA头程送亚马逊仓,热门仓卡派直送(德国WRO5/HAJ1/DTM1/DTM2、法国CDG7/LIL8、意大利TRN3、西班牙BCN8、罗马尼亚EMAG等)。卡派直送优势:超重附加费少、不限尺寸重量、上架快、基本无丢件。法国、意大利、西班牙其他亚马逊仓不能超UPS标准。"
},
{
"keys":["人工客服","人工","客服","转人工","在线吗","上班","工作时间","营业时间","几点","周六","周日","星期六","星期日","星期天","礼拜","周末","双休","节假日","放假","休息"],
"text":"人工客服在线时间:周一至周五 9:00—18:00。周六、周日及法定节假日人工客服不上班。非工作时间可以先留言或提交需求,业务员会在下一个工作日回复您。"
},
{
"keys":["电话","联系方式","怎么联系","联系你们","微信"],
"text":"可以联系您的专属业务员,或在人工客服在线时间(周一至周五 9:00—18:00)通过在线客服咨询。建议提前准备好运单号、货物信息(品名、重量、尺寸、目的国),方便快速处理。"
},
{
"keys":["退件","退回","退货","退运","改地址"],
"text":"海运/铁路为散货拼柜,已装柜不接受退件和扣件。空运国外扣件费200欧/票,收件人改派地址最低200元/票。退回海外仓:德国境内15欧/件,欧盟境内25欧/件,操作费100元/票,5个工作日免仓储,超期1欧/kg/天。重派德国10欧/件、欧盟20欧/件。"
}
]
DB = "history.db"
MAX_CONTEXT = 4 # 每次带给模型的历史消息条数
def now_str():
return datetime.now().strftime("%Y-%m-%d %H:%M")
def init_db():
conn = sqlite3.connect(DB)
cur = conn.cursor()
cur.execute('''
CREATE TABLE IF NOT EXISTS chat_history (
id INTEGER PRIMARY KEY AUTOINCREMENT,
title TEXT NOT NULL,
ctime TEXT NOT NULL
)
''')
# 老库没有 utime 列:补上,用来按"最近活跃"排序
cols = [r[1] for r in cur.execute("PRAGMA table_info(chat_history)").fetchall()]
if "utime" not in cols:
cur.execute("ALTER TABLE chat_history ADD COLUMN utime TEXT")
cur.execute("UPDATE chat_history SET utime = ctime WHERE utime IS NULL")
# 消息表:一条会话对应多条消息
cur.execute('''
CREATE TABLE IF NOT EXISTS messages (
id INTEGER PRIMARY KEY AUTOINCREMENT,
conv_id INTEGER NOT NULL,
role TEXT NOT NULL,
content TEXT NOT NULL,
ctime TEXT NOT NULL
)
''')
cur.execute("CREATE INDEX IF NOT EXISTS idx_msg_conv ON messages(conv_id, id)")
# 👍/👎 反馈表
cur.execute('''
CREATE TABLE IF NOT EXISTS feedback (
id INTEGER PRIMARY KEY AUTOINCREMENT,
question TEXT,
answer TEXT,
vote TEXT,
ctime TEXT
)
''')
# 人工客服会话表
cur.execute('''
CREATE TABLE IF NOT EXISTS human_sessions (
id INTEGER PRIMARY KEY AUTOINCREMENT,
conv_id INTEGER NOT NULL,
status TEXT NOT NULL DEFAULT 'waiting',
ctime TEXT NOT NULL
)
''')
cur.execute('''
CREATE TABLE IF NOT EXISTS human_messages (
id INTEGER PRIMARY KEY AUTOINCREMENT,
session_id INTEGER NOT NULL,
side TEXT NOT NULL,
content TEXT NOT NULL,
ctime TEXT NOT NULL
)
''')
# 补 read_at 列
hm_cols = [r[1] for r in cur.execute("PRAGMA table_info(human_messages)").fetchall()]
if "read_at" not in hm_cols:
cur.execute("ALTER TABLE human_messages ADD COLUMN read_at TEXT")
# 老数据只有标题,补一条 user 消息,点开还能接着聊
old = cur.execute(
"SELECT id, title FROM chat_history "
"WHERE id NOT IN (SELECT DISTINCT conv_id FROM messages)"
).fetchall()
for cid, title in old:
cur.execute(
"INSERT INTO messages(conv_id, role, content, ctime) VALUES (?,?,?,?)",
(cid, "user", title, now_str())
)
conn.commit()
conn.close()
init_db()
def backup_db():
"""启动时把数据库备份到 backups/,文件名带日期,只保留最近 7 天"""
try:
bdir = os.path.join(os.path.dirname(os.path.abspath(__file__)), "backups")
os.makedirs(bdir, exist_ok=True)
today = datetime.now().strftime("%Y%m%d")
dst = os.path.join(bdir, f"history_{today}.db")
if not os.path.exists(dst):
shutil.copy2(DB, dst)
# 清理 7 天前的旧备份
cutoff = datetime.now() - timedelta(days=7)
for f in os.listdir(bdir):
m = re.match(r"history_(\d{8})\.db$", f)
if not m:
continue
try:
if datetime.strptime(m.group(1), "%Y%m%d") < cutoff:
os.remove(os.path.join(bdir, f))
except Exception:
pass
except Exception as e:
print(f"[数据库备份失败] {e}")
def cleanup_uploads():
"""删除 uploads/ 里 7 天前的图片"""
try:
udir = os.path.join(os.path.dirname(os.path.abspath(__file__)), "uploads")
if not os.path.isdir(udir):
return
cutoff = time.time() - 7 * 86400
for f in os.listdir(udir):
fp = os.path.join(udir, f)
if os.path.isfile(fp) and os.path.getmtime(fp) < cutoff:
os.remove(fp)
except Exception as e:
print(f"[清理上传文件失败] {e}")
def cleanup_old_conversations():
"""自动清理 30 天前的旧对话,防止数据库无限膨胀"""
try:
conn = sqlite3.connect(DB)
c = conn.cursor()
cutoff = (datetime.now() - timedelta(days=30)).strftime("%Y-%m-%d %H:%M:%S")
rows = c.execute("SELECT id FROM chat_history WHERE utime < ?", (cutoff,)).fetchall()
ids = [r[0] for r in rows]
if not ids:
return
for cid in ids:
c.execute("DELETE FROM messages WHERE conv_id=?", (cid,))
c.execute(f"DELETE FROM chat_history WHERE id IN ({','.join('?'*len(ids))})", ids)
conn.commit()
conn.close()
print(f"[自动清理] 已删除 {len(ids)} 条 30 天前的旧对话")
except Exception as e:
print(f"[自动清理失败] {e}")
backup_db()
cleanup_uploads()
# 自动清理已关闭,不删旧对话
def make_title(content):
"""生成会话标题:优先用文字内容,只有附件时显示附件数量"""
# 去掉所有 "[附件] xxx" 行
lines = [l for l in content.split("\n") if not l.strip().startswith("[附件]")]
text = " ".join(lines).strip()
if text and text != "(发送了附件)" and text != "请帮我看看这个附件":
return text[:24]
# 只有附件:显示附件数量,不显示文件名
n = len([l for l in content.split("\n") if l.strip().startswith("[附件]")])
if n == 1:
return "📎 已上传附件"
if n > 1:
return f"📎 {n}个附件"
return content[:24]
def save_message(conv_id, role, content):
"""把一条消息写进会话;用户消息做幂等处理,重试不会重复入库。"""
if not conv_id or not content:
return
conn = sqlite3.connect(DB)
cur = conn.cursor()
if role == "user":
last = cur.execute(
"SELECT role, content FROM messages WHERE conv_id=? ORDER BY id DESC LIMIT 1",
(conv_id,)
).fetchone()
if last and last[0] == "user" and last[1] == content:
conn.close()
return "duplicate"
cur.execute(
"INSERT INTO messages(conv_id, role, content, ctime) VALUES (?,?,?,?)",
(conv_id, role, content, now_str())
)
if role == "user":
row = cur.execute("SELECT title FROM chat_history WHERE id=?", (conv_id,)).fetchone()
if row and (not row[0] or row[0] == "新对话"):
cur.execute("UPDATE chat_history SET title=? WHERE id=?",
(make_title(content), conv_id))
cur.execute("UPDATE chat_history SET utime=? WHERE id=?", (now_str(), conv_id))
conn.commit()
conn.close()
def last_user_question(conv_id):
"""上一条用户提问(必须在当前这句入库之前调用)"""
if not conv_id:
return None
conn = sqlite3.connect(DB)
row = conn.execute(
"SELECT content FROM messages WHERE conv_id=? AND role='user' "
"ORDER BY id DESC LIMIT 1",
(conv_id,)
).fetchone()
conn.close()
return row[0] if row else None
def card_to_text(content):
"""把存库的卡片标记转成人话,避免原始 JSON 进入模型上下文"""
if not content.startswith(CARD_PREFIX):
return content
try:
card = json.loads(content[len(CARD_PREFIX):])
return f"(已为运单 {card.get('no', '')} 查询过轨迹,当前状态:{card.get('status', '')})"
except Exception:
return "(已为该运单查询过轨迹)"
def load_context(conv_id, msg):
"""取最近 MAX_CONTEXT 条消息,拼成模型要的 messages(不含 system)。"""
if not conv_id:
return [{"role": "user", "content": msg}]
conn = sqlite3.connect(DB)
rows = conn.execute(
"SELECT role, content FROM messages WHERE conv_id=? ORDER BY id DESC LIMIT ?",
(conv_id, MAX_CONTEXT)
).fetchall()
conn.close()
rows.reverse()
history = [{"role": r, "content": card_to_text(c)} for r, c in rows]
# 本次提问已经落库的话,不要再追加一遍
if not history or history[-1]["role"] != "user" or history[-1]["content"] != msg:
history.append({"role": "user", "content": msg})
return history
def search_knowledge(question):
"""按关键词命中打分:命中的关键词越长越具体,得分越高"""
q = question.lower()
best, best_score = None, 0
for item in knowledge_list:
for k in item["keys"]:
if k.lower() in q:
score = len(k) # 长关键词更具体,优先
if score > best_score:
best, best_score = item["text"], score
return best
def sse(payload):
"""统一的事件格式:data: {"t":正文} / {"e":错误} / {"card":卡片} / {"d":1结束}
JSON 编码可以避免正文里的换行把 SSE 帧撑断。"""
return "data: " + json.dumps(payload, ensure_ascii=False) + "\n\n"
CARD_PREFIX = "__CARD__" # 存库时标记"这条是轨迹卡片"
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):
"""生成固定的模拟轨迹:同一个单号每次查都一样"""
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}
# ===== 报价数据(来源:炜易达 VIP 10月15日 9点执行报价表)=====
# tiers: [(最低计费重量, 单价 RMB/kg), ...],按从高到低匹配
RATES = {
"sea": {
"name": "欧洲海运快线(双清/递延)",
"tiers": [(2000, 10.0), (500, 10.2), (101, 10.3), (21, 12.3)],
"divisor": 6000,
"min_kg": 21,
"time": "开船后 38-45 天提取",
"min_pcs_kg": None,
"remote": 7,
"note": "德国 UPS/DHL/DPD/CWE 统配;欧洲递延 -1.7元/kg;樱花快线 +0.3元/kg;因查验/罢工/节假日顺延",
},
"rail": {
"name": "义乌-中欧铁路专线",
"tiers": [(1000, 13.5), (100, 13.5), (21, 15.5)],
"divisor": 6000,
"min_kg": 21,
"time": "发车后 36-40 天提取",
"min_pcs_kg": 13,
"light_fee": 5,
"remote": 7,
"note": "义乌装柜;VAT 递延 -1元/kg;单件计费重不足 13kg 按超轻费加收",
},
"truck": {
"name": "中欧卡航",
"tiers": [(4000, 17.3), (2000, 17.3), (500, 17.4), (101, 17.4), (21, 19.4)],
"divisor": 6000,
"min_kg": 21,
"time": "发车后欧盟 16-20 天提取",
"min_pcs_kg": 13,
"light_fee": 5,
"remote": 7,
"note": "递延 -1.5元/kg;纯纺织品 +5元/kg;普卡时效 18-22 天",
},
"air": {
"name": "欧洲空运专线(空派)",
"tiers": [(101, 36), (71, 37), (45, 38), (21, 39)],
"divisor": 6000,
"min_kg": 21,
"time": "10-13 天左右提取",
"min_pcs_kg": 12,
"light_fee": None,
"remote": 7,
"note": "一票多件单件不足 12kg 按 12kg 计费;快线 8-13 天;直飞快线 12-15 天",
},
"sea_uk": {
"name": "英国海运(双清)",
"tiers": [(301, 10.9), (101, 10.9), (21, 12.8)],
"divisor": 6000,
"min_kg": 21,
"time": "参考时效 40 天提取",
"remote": None,
"note": "VAT自税/递延 10.0/8.1 元/kg;按KG卡派 12.2/10.3;限时达 13.5/11.6;北极航线 25-28 天(义乌装)",
},
"rail_uk": {
"name": "英国铁路(双清)",
"tiers": [(301, 16.2), (101, 16.2), (21, 18.0)],
"divisor": 6000,
"min_kg": 21,
"time": "发车后 25-30 天提取",
"remote": None,
"note": "VAT自税/递延 15/13.2 元/kg;限时达 19.2/17.4(38天);按KG卡派 17.3/15.5",
},
"truck_uk": {
"name": "中英卡航(双清)",
"tiers": [(301, 19.4), (101, 19.4), (21, 21.4)],
"divisor": 6000,
"min_kg": 21,
"time": "发车后 20-22 天提取",
"remote": None,
"note": "VAT自税/递延 18.4/16.4 元/kg;卡派 20.9/18.9;单件不超30kg,尺寸≤98×68×59cm",
},
"air_uk": {
"name": "英国空运专线(双清)",
"tiers": [(101, 42), (51, 44), (22, 48)],
"divisor": 6000,
"min_kg": 22,
"time": "8-10 个自然日提取",
"min_pcs_kg": 12,
"light_fee": None,
"remote": None,
"note": "不包税 VAT 自税/递延 44/40/38;带电 47.5/42.5/41.5;拒收北爱尔兰/苏格兰高地偏远地址",
},
}
def parse_dims(t):
"""解析 长x宽x高,例如 120*100*100 / 100x80x60 / 120×100×100
用"最长连续数字"匹配三组,要求每组前后都不与其它数字相连,
这样 "100x80x60 200kg" 里的 60 不会被后面的 200 吞成 6020。
"""
for m in re.finditer(r"(\d+)[*xX×](\d+)[*xX×](\d+)", t):
# 逐段向后延长,保证拿到的是完整数字而不是半截
dims = []
ok = True
pos = m.start()
for g in (1, 2, 3):
end = m.end(g)
# 若后面还紧跟数字,说明这一组没取全,放弃这个候选
if end < len(t) and t[end].isdigit():
ok = False
break
# 若是第一组,前面不能紧挨数字
if g == 1 and m.start(1) > 0 and t[m.start(1) - 1].isdigit():
ok = False
break
dims.append(int(m.group(g)))
if ok:
return [float(x) for x in dims]
return None
def parse_shipping_intent(text):
"""从话里抠出:重量(kg)、渠道、长宽高(cm)"""
t = text.replace(" ", "")
intent = {"weight": None, "channel": None, "dims": None}
# 尺寸:在原始文本上解析(空格有助于区分尺寸和后面的重量)
intent["dims"] = parse_dims(text)
if not intent["dims"]:
# 支持“长100宽80高60”
dl = re.search(r"长(\d+(?:\.\d+)?)宽(\d+(?:\.\d+)?)高(\d+(?:\.\d+)?)", t)
if dl:
intent["dims"] = [float(x) for x in dl.groups()]
# 重量:带单位的数字。去掉空格后在原始文本上匹配,
# 但不能把尺寸里的数字(如 120*100*100 的第三个数)当成重量
m = re.search(r"(\d+(?:\.\d+)?)\s*(吨|kg|KG|公斤|千克|tons?|kilo(?:grams?)?|lbs?)", text, re.I)
if m:
before = text[:m.start()].rstrip()
adjacent_sep = bool(before) and before[-1] in "*xX×"
if not adjacent_sep:
n = float(m.group(1))
unit_w = m.group(2).lower()
if unit_w in ("吨", "ton", "tons"):
intent["weight"] = n * 1000
elif unit_w.startswith("lb"):
intent["weight"] = round(n * 0.4536, 2) # 磅转公斤
else:
intent["weight"] = n
intent["weight_raw"] = m.group(0).replace(" ", "")
# 渠道:空运 / 铁路 / 卡航 / 海运(按优先级匹配)
if re.search(r"空运|空派|航空", t) or re.search(r"\bair\b|air ?freight|air ?cargo", text, re.I):
intent["channel"] = "air_uk" if re.search(r"英国|UK|伦敦|英格兰", text, re.I) else "air"
elif re.search(r"铁路|班列|火车", t) or re.search(r"\brail\b|train", text, re.I):
intent["channel"] = "rail"
elif re.search(r"卡航|卡车|卡派", t) or re.search(r"\btruck\b|road", text, re.I):
intent["channel"] = "truck"
elif re.search(r"海运|海派|船运|快船|海运快线|樱花快线", t) or re.search(r"\bsea\b|ocean|shipping", text, re.I):
intent["channel"] = "sea"
return intent
def detect_dim_unit(text):
"""识别尺寸单位:cm(默认)/ m / mm"""
low = text.lower()
if re.search(r"毫米|\d\s*mm", low):
return "mm"
if re.search(r"米", low) and not re.search(r"厘米", low):
return "m"
if re.search(r"\d\s*cm|\d\s*厘米", low):
return "cm"
if re.search(r"\d\s*m", low): # 注意:cm 里也含 m,所以放在 cm 之后判断
return "m"
return "cm"
def dims_to_cm(dims, unit):
"""统一换算成厘米,体积重公式按 cm 计算"""
if unit == "m":
return [d * 100 for d in dims]
if unit == "mm":
return [d / 10 for d in dims]
return list(dims)
def pick_rate(weight_kg, tiers):
"""从阶梯价里选档:tiers 为 [(最低重量, 单价), ...],从高到低匹配"""
for min_w, price in tiers:
if weight_kg >= min_w:
return price, min_w
return tiers[-1][1], tiers[-1][0]
CBM_CHANNELS = ("卡航", "普卡", "铁路", "海运")
def calc_warehouse_card(cbm, channel, warehouse_key=None, country=None):
"""海外仓卡派:按方(CBM)计费,1CBM+ 和 0.5CBM+ 两档
country 传入国家代码(DE/FR/PL/CZ/ES)时,优先选择包含该后缀的仓点组。
"""
if not RX or not RX.WAREHOUSE_CARD:
return None
keys = list(RX.WAREHOUSE_CARD)
key = warehouse_key
if not key and country:
for k in keys:
if ("-" + country) in k.upper():
key = k
break
if not key:
key = keys[0]
row = RX.WAREHOUSE_CARD.get(key) or RX.WAREHOUSE_CARD[keys[0]]
p = row.get(channel) or row["海运"]
price = p["cbm1"] if cbm >= 1 else p["cbm05"]
if price is None:
return None
tier = "1CBM+" if cbm >= 1 else "0.5CBM+"
stores = [s for s in key.splitlines() if s.strip()]
if country:
stores = [s for s in stores if ("-" + country) in s.upper()] or stores
else:
stores = stores[:1]
note = "报价表中的仓点分组:" + "、".join(stores)
return {
"warehouse": "\n".join(stores), "stores": stores,
"channel": channel, "cbm": cbm, "note": note,
"price_cbm": price, "tier": tier,
"total": round(cbm * price, 2),
"formula": f"{cbm:g} CBM × {price:,.0f} 元/CBM = {cbm * price:,.0f} 元",
}
def detect_surcharge(text):
"""识别产品类型,返回报价表中的附加费条目"""
if not RX:
return None
best = None
for product, fee in RX.PRODUCT_SURCHARGES:
# 用产品名里的前几个字做匹配(如“手机”“电池”“玩具”)
for seg in re.split(r"[,,、\s()()]", product):
seg = seg.strip()
if len(seg) >= 2 and seg in text:
if best is None or len(seg) > len(best[2]):
best = (product, fee, seg)
if best:
return best[0], best[1]
return None
def check_island(postcode):
"""按邮编判断是否属于 DPD 岛屿偏远地区(附加费 35 欧/件)"""
if not RX:
return None
pc = re.sub(r"[^0-9]", "", str(postcode))
if not pc:
return None
for country, info in RX.DPD_ISLAND_POSTCODES.items():
seg = info["text"]
# 直接匹配的 4~5 位邮编
for m in re.finditer(r"\d{4,5}", seg):
if pc.startswith(m.group(0)):
return country, info["code"], m.group(0)
# 区间形式 07000–07999
for m in re.finditer(r"(\d{4,5})\s*[–\-]\s*(\d{4,5})", seg):
lo, hi = int(m.group(1)), int(m.group(2))
if len(m.group(1)) == len(pc) and lo <= int(pc) <= hi:
return country, info["code"], m.group(0)
return None
def surcharge_reply(product, fee, weight_kg=None):
"""附加费说明(能算就给出金额)"""
txt = f"【产品附加费】{product}:{fee}"
if weight_kg and "/KG" in fee.upper():
m = re.search(r"\+?(\d+(?:\.\d+)?)", fee)
if m:
extra = float(m.group(1)) * weight_kg
txt += f"\n 按 {weight_kg:g} kg 计:约 {extra:,.0f} 元(以最终计费重为准)"
return txt
def calc_freight(weight_kg, channel, dims=None, unit="cm"):
"""按报价表规则计算运费(数字全部由代码算)
报价表规则:
- 阶梯单价:按总重量落入的档位取价
- 材积:长×宽×高(cm) ÷ 6000,与实重取大值作为计费重量
- 单件最低计费重 / 超轻费:见 RATES 里的 min_pcs_kg、light_fee
- 偏远地址:+7元/kg(不含油),最低 250 元/票
"""
r = RATES[channel]
cm = dims_to_cm(dims, unit) if dims else None
divisor = r.get("divisor", 6000)
vol = round(cm[0] * cm[1] * cm[2] / divisor, 2) if cm else None
real = weight_kg or 0
charge = round(max(real, vol) if vol else real, 2) # 实重与体积重取大
min_kg = r.get("min_kg") or 0
charge = max(charge, min_kg) # 报价表起收档
price, tier = pick_rate(charge, r["tiers"])
total = round(charge * price, 2)
unit_note = ""
if dims and unit != "cm":
unit_note = "(原尺寸 %g×%g×%g %s,已换算为 %g×%g×%g cm)" % (*dims, unit, *cm)
return {
"channel": channel, "name": r["name"],
"real": real, "vol": vol, "charge": charge,
"dims": dims, "dims_cm": cm, "unit": unit, "unit_note": unit_note,
"dim_rule": ("长×宽×高÷%d = %g×%g×%g÷%d = %g kg" % (divisor, *cm, divisor, vol)) if cm else None,
"price": price, "tier": tier,
"formula": f"{charge:g} kg × {price} 元/kg = {total:,.0f} 元",
"total": total,
"time": r["time"],
"min_pcs_kg": r.get("min_pcs_kg"),
"light_fee": r.get("light_fee"),
"remote": r.get("remote"),
"note": r.get("note"),
}
def freight_reply(intent):
"""根据已知信息拼出可读的计算说明(给模型当参考,也直接展示)"""
w = intent["weight"] or 0
dims = intent["dims"]
unit = intent.get("dim_unit") or "cm"
lines = []
channels = [intent["channel"]] if intent["channel"] else ["sea", "air"]
for ch in channels:
c = calc_freight(w, ch, dims, unit)
parts = [f"实重 {w:g} kg" if w else "实重:未提供"]
if c["vol"]:
parts.append(f"体积重 {c['vol']:g} kg")
body = (
f"【{c['name']}】按时效 {c['time']},计费方式:阶梯单价 × 计费重量\n"
+ (f" 尺寸:{c['unit_note']}\n" if c["unit_note"] else "")
+ f" 计费重量 {c['charge']:g} kg(" + ",".join(parts) + ",取较大值)\n"
+ (f" 材积算法:{c['dim_rule']}\n" if c["dim_rule"] else "")
+ f" 适用档位:{c['tier']}KG+ 档,单价 {c['price']} 元/kg\n"
+ f" 运费:{c['formula']}\n"
)
if c["min_pcs_kg"] and c["charge"] < c["min_pcs_kg"]:
body += f" 注意:单件计费重不足 {c['min_pcs_kg']}kg,需按 {c['min_pcs_kg']}kg 起算或加收超轻费 {c['light_fee']} 元/kg\n"
if c["remote"]:
body += f" 偏远地址:额外 +{c['remote']} 元/kg,单票最低 250 元\n"
if c["note"]:
body += f" 其他:{c['note']}\n"
lines.append(body.rstrip("\n"))
return "\n\n".join(lines)
MAX_ANSWER_CHARS = 800 # 中文回答字数上限
MAX_ANSWER_CHARS_EN = 1400 # 英文回答字数上限(英文信息密度低,字符数约为中文的 2 倍)
# 压缩时若原文没有中文,要求压缩结果也用英文,避免把英文回答压成中文
EN_COMPRESS_RULE = (
"原文是英文,压缩后的结果必须仍然使用英文(English),不得翻译成中文。"
)
def has_chinese(t):
return any("\u4e00" <= ch <= "\u9fa5" for ch in t)
def compress_answer(text, limit=MAX_ANSWER_CHARS):
"""回答超长时,让模型压缩到规定字数内;失败则原样返回"""
if len(text) <= limit:
return text
try:
body = {
"model": MODEL_NAME,
"messages": [
{"role": "system",
"content": (EN_COMPRESS_RULE if has_chinese(text) is False else "") +
"你是文字精简助手。把用户给的内容压缩到 "
f"{limit} 字以内(必须严格满足,超出即为不合格),"
"必须保留所有数字、金额、时效和渠道名称,"
"删除重复表述、客套话和次要说明,结构保持不变。"},
{"role": "user", "content": text},
],
"stream": False,
"max_tokens": 800,
}
r = requests.post(URL,
headers={"Authorization": f"Bearer {DEEPSEEK_API_KEY}",
"Content-Type": "application/json"},
json=body, timeout=(5, 30))
if r.status_code == 200:
short = r.json()["choices"][0]["message"]["content"].strip()
print(f"[压缩] {len(text)} 字 -> {len(short)} 字")
return short
print(f"[压缩失败] 状态码 {r.status_code}")
except Exception as e:
print(f"[压缩异常] {e}")
return text
def gen_followups(user_msg, answer_text):
"""根据本轮问答,生成 3 个客户可能接着问的问题(失败返回空列表,不影响主流程)"""
if not DEEPSEEK_API_KEY:
return []
try:
body = {
"model": MODEL_NAME,
"messages": [
{"role": "system",
"content": "你是物流客服助手。根据用户问题和你的回答,生成3个客户最可能接着问的简短问题(每个不超过20字),只输出JSON字符串数组,不要任何解释。"},
{"role": "user",
"content": f"用户问:{user_msg}\n你的回答:{answer_text[:500]}\n\n请生成3个追问:"},
],
"stream": False,
"max_tokens": 200,
}
r = requests.post(URL,
headers={"Authorization": f"Bearer {DEEPSEEK_API_KEY}"},
json=body, timeout=(5, 15))
if r.status_code != 200:
return []
text = r.json()["choices"][0]["message"]["content"].strip()
m = re.search(r"\[.*\]", text, re.S)
if m:
arr = json.loads(m.group(0))
return [str(x).strip() for x in arr if str(x).strip()][:3]
except Exception:
pass
return []
@app.route("/api/feedback", methods=["POST"])
def feedback():
"""收集 👍/👎 反馈,存到 feedback 表"""
data = request.get_json(silent=True) or {}
conn = sqlite3.connect(DB)
conn.execute(
"INSERT INTO feedback(question, answer, vote, ctime) VALUES (?,?,?,?)",
(str(data.get("question", ""))[:500], str(data.get("answer", ""))[:2000],
str(data.get("vote", "")), now_str()))
conn.commit()
conn.close()
return jsonify({"ok": True})
@app.route("/stream_chat", methods=["GET", "POST"])
def stream_chat():
# POST:问题放在请求体里,不受 URL 长度限制(推荐,长文本用这个)
# GET:保留兼容旧链接与 EventSource 方式
if request.method == "POST":
data = request.get_json(silent=True) or {}
msg = (data.get("msg") or "").strip()
conv_id = data.get("conv_id")
try:
conv_id = int(conv_id) if conv_id else None
except (TypeError, ValueError):
conv_id = None
else:
msg = (request.args.get("msg") or "").strip()
conv_id = request.args.get("conv_id", type=int)
if not msg:
return Response(sse({"e": "问题不能为空"}), mimetype="text/event-stream")
# 纯图片消息:不调AI
if "[图片]" in msg and len(msg) < 300:
return Response(sse({"t": "收到您的图片,请描述一下您的问题。"}), mimetype="text/event-stream")
# 运单号:直接给轨迹卡片,不让模型编
track_no = extract_tracking_no(msg)
if track_no:
card = build_track(track_no)
def track_stream():
save_message(conv_id, "user", msg)
yield sse({"card": card})
save_message(conv_id, "assistant",
CARD_PREFIX + json.dumps(card, ensure_ascii=False))
yield sse({"d": 1})
return Response(stream_with_context(track_stream()),
mimetype="text/event-stream",
headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"})
if not DEEPSEEK_API_KEY:
return Response(sse({"e": "服务未配置 API Key,请设置环境变量 DEEPSEEK_API_KEY 后重启服务"}),
mimetype="text/event-stream")
know_text = search_knowledge(msg)
# 运费试算:话里带重量/尺寸就按规则算好,把明细作为权威资料交给模型
ship = parse_shipping_intent(msg)
ship["dim_unit"] = detect_dim_unit(msg)
freight_text = None
# 只是追问渠道(比如“那空运呢”)时,沿用上一条提问里的货物信息再算一遍
if not ship["weight"] and not ship["dims"] and ship["channel"]:
prev_q = last_user_question(conv_id)
if prev_q:
prev_ship = parse_shipping_intent(prev_q)
if prev_ship["weight"] or prev_ship["dims"]:
ship["weight"] = prev_ship["weight"]
ship["dims"] = prev_ship["dims"]
ship["dim_unit"] = detect_dim_unit(prev_q)
if ship["weight"] or ship["dims"]:
# 提到英国时,把渠道切到对应的英国专线
if re.search(r"英国|UK|伦敦|英格兰", msg, re.I):
uk_map = {"sea": "sea_uk", "rail": "rail_uk", "truck": "truck_uk", "air": "air_uk"}
ship["channel"] = uk_map.get(ship["channel"])
freight_text = freight_reply(ship)
if not ship["channel"]:
freight_text += "\n\n(客户没有指定渠道,上面已同时给出海运和空运,请在回答末尾询问更看重时效还是成本)"
# 扩展数据:卡派按方 / 产品附加费 / 岛屿邮编
extra_parts = []
# ① 卡派按方:提到 CBM 或方数时
cbm_m = re.search(r"(\d+(?:\.\d+)?)\s*(?:cbm|方|立方)", msg, re.I)
if cbm_m:
cbm = float(cbm_m.group(1))
ch = "海运"
if re.search(r"卡航|卡车", msg):
ch = "卡航"
elif re.search(r"铁路|班列", msg):
ch = "铁路"
elif re.search(r"普卡", msg):
ch = "普卡"
# 按提问里的国家挑选对应仓点分组
want = None
for kw, ctry in (("德国", "DE"), ("法国", "FR"), ("波兰", "PL"),
("捷克", "CZ"), ("西班牙", "ES"), ("马德里", "ES"), ("中国", "")):
if kw in msg:
want = ctry
break
card = calc_warehouse_card(cbm, ch, country=want)
if card:
store = card['warehouse'].splitlines()[0]
extra_parts.append(
"【海外仓卡派 · 按方计费】\n"
f" 仓点:{store}\n"
f" 计费:{card['note']}\n"
f" 渠道:{card['channel']},适用档位 {card['tier']}\n"
f" 运费:{card['formula']}\n"
" 说明:单箱实重超 35KG 加收 30 元/箱(封顶 3600 元);不接受仓储,特殊情况 85 元/箱/周"
)
# ② 产品附加费
sur = detect_surcharge(msg)
if sur:
extra_parts.append(surcharge_reply(sur[0], sur[1], ship.get("weight")))
# ③ 岛屿偏远邮编
pc_m = re.search(r"(?<![0-9])(\d{4,5})(?![0-9])", msg)
if pc_m and re.search(r"邮编|postcode|zip|偏远|岛屿", msg, re.I):
hit = check_island(pc_m.group(1))
if hit:
extra_parts.append(
f"【偏远判断】邮编 {pc_m.group(1)} 属于{hit[0]}({hit[1]})的岛屿/偏远地区,"
"需加收 DPD 岛屿附加费 35 欧元/件"
)
else:
extra_parts.append(
f"【偏远判断】邮编 {pc_m.group(1)} 未命中 DPD 岛屿偏远清单;"
"但仍需以 UPS/DPD 官网查询结果为准,偏远费以账单为准"
)
# 追问兜底:当前这句没命中关键词,就用上一句提问去检索
follow_src = None
if not know_text:
prev_q = last_user_question(conv_id)
if prev_q:
know_text = search_knowledge(prev_q)
if know_text:
follow_src = prev_q
# 检索完再落库(必须放在上面之后,否则会查到刚存进去的这句)
# 先看这个会话已有几条消息:0 条 = 新对话第一轮,才生成相关追问(省额度)
_c = sqlite3.connect(DB)
_msg_count = _c.execute("SELECT COUNT(*) FROM messages WHERE conv_id=?", (conv_id,)).fetchone()[0]
_c.close()
is_first_turn = (_msg_count == 0)
save_message(conv_id, "user", msg)
# 统一的长短约束:所有回答都遵守
LENGTH_RULE = (
"回答长度要求:全文不少于 120 字、不超过 350 字。"
"内容要完整、信息充分,把关键点讲清楚,可分 2~4 段或用短列表呈现。"
"可以补充相关的说明(例如适用范围、注意事项、下一步建议),让对方看完就能明白。"
"不要寒暄客套,不要重复同一个数字,不要罗列与问题无关的信息。"
)
# 语言要求放在最前面并强调,避免被后面的中文结构化指令盖过
is_en = bool(re.search(r"[A-Za-z]{3,}", msg)) and not re.search(r"[\u4e00-\u9fa5]", msg)
if is_en:
LANGUAGE_RULE = (
"【语言要求 · 最高优先级】用户用英文提问,你必须用英文(English)回答全文,"
"包括所有小节标题(如 Conclusion / Calculation / Notes),不要出现任何中文。"
"以下给出的中文结构与参考资料仅用于理解,输出必须翻译成英文。"
)
else:
LANGUAGE_RULE = "【语言要求】用户用中文提问,请用中文回答。"
if freight_text or extra_parts:
blocks = []
if freight_text:
blocks.append(freight_text)
blocks.extend(extra_parts)
body = "\n\n".join(blocks)
system_prompt = f"""{LANGUAGE_RULE}
你是柏悦物流智能客服。
下面是根据报价表计算好的结果,请严格按照这些数字向客户说明。
不要重新计算、不要改动任何数字,但要讲清楚是怎么算出来的。
{body}
请按以下结构输出(用英文回答时请把标题也翻译成英文):
【结论】先给结论:哪种方式、多少钱、多久到。
【计算过程】分点列出,最多 5 行,让客户看懂钱花在哪。
【补充说明】2~3 行:适用范围(如单件重量、尺寸要求)、偏远地区或敏感货附加费、下一步建议(如提供仓点或运单号可进一步核算)。
{LENGTH_RULE}
客户没指定渠道时,把主要几种渠道(海运/铁路/卡航/空运)列出来对比,并问客户更看重时效还是成本。
"""
elif know_text:
extra = f"\n(以上资料对应上一句提问:{follow_src})" if follow_src else ""
system_prompt = f"""{LANGUAGE_RULE}
你是柏悦物流智能客服,严格优先使用下面知识库内容回答,不要自己编造业务信息。
回答语句通顺自然,分段清晰,让人一眼看清重点。
要结合上文理解用户的追问(比如"那空运呢""这个多少钱"指的是什么)。
{LENGTH_RULE}
知识库内容(供你理解,输出需按语言要求翻译):{know_text}{extra}
"""
else:
system_prompt = f"""{LANGUAGE_RULE}
你是柏悦物流智能客服。
物流相关简单问题可以正常回答;不属于物流业务的问题礼貌说明无法解答,并建议联系人工客服。
回答语句通顺,表述自然。
{LENGTH_RULE}
如果用户在追问,请结合上文的对话继续回答,不要重复问已经说过的信息。
"""
# 取最近几条历史,和本次提问一起发给模型
context = load_context(conv_id, msg)
def generate():
headers = {
"Authorization": f"Bearer {DEEPSEEK_API_KEY}",
"Content-Type": "application/json"
}
body = {
"model": MODEL_NAME,
"messages": [{"role": "system", "content": system_prompt}] + context,
"stream": True
}
# 1) 连不上 / 超时:转成 SSE 事件,而不是让 Flask 直接 500
try:
resp = requests.post(URL, headers=headers, json=body,
stream=True, timeout=(5, 90))
except requests.exceptions.Timeout:
yield sse({"e": "模型响应超时,请稍后重试"})
return
except requests.exceptions.RequestException:
yield sse({"e": "无法连接模型服务,请检查网络后重试"})
return
except Exception:
yield sse({"e": "服务异常,请稍后重试"})
return
# 2) 状态码不是 200(Key 失效、余额不足、限流等)
if resp.status_code != 200:
detail = ""
try:
detail = (resp.json().get("error") or {}).get("message", "")
except Exception:
detail = (resp.text or "")[:120]
tips = {
401: "系统繁忙,请稍后再试",
402: "系统繁忙,请稍后再试",
429: "咨询量较大,请稍后再试",
}
tip = tips.get(resp.status_code, "系统繁忙,请稍后再试")
yield sse({"e": tip})
return
# 3) 流式读取,任何异常都变成可读的提示
parts = []
interrupted = False
try:
got_text = False
for raw in resp.iter_lines():
if not raw:
continue
line = raw.decode("utf-8", errors="ignore")
if not line.startswith("data:"):
continue
data = line[5:].strip()
if data == "[DONE]":
break
try:
j = json.loads(data)
chunk = j["choices"][0]["delta"].get("content") or ""
except Exception:
continue
if chunk:
got_text = True
parts.append(chunk)
yield sse({"t": chunk})
if not got_text:
yield sse({"e": "模型没有返回内容,请重试"})
return
full_text = "".join(parts)
# 中英文用不同的字数上限
limit = MAX_ANSWER_CHARS if has_chinese(full_text) else MAX_ANSWER_CHARS_EN
# 超过字数上限时自动精简(保留全部数字),并把精简结果推给前端
if len(full_text) > limit:
short = compress_answer(full_text, limit)
# 压缩后仍超标就再压一次(最多两轮)
if len(short) > limit:
short = compress_answer(short, limit - 60)
# 压缩后仍超上限:说明内容含大量数字(如多渠道对比),
# 此时保留完整内容,仅在末尾标注,不硬截断
if short != full_text:
full_text = short
yield sse({"replace": short})
save_message(conv_id, "assistant", full_text)
# 只在新对话第一轮生成 3 个相关追问(省额度,失败静默)
if is_first_turn:
try:
fu = gen_followups(msg, full_text)
if fu:
yield sse({"suggest": fu})
except Exception:
pass
yield sse({"d": 1})
except GeneratorExit:
# 浏览器刷新 / 关页面 / 切换会话:把已经生成的部分存下来
interrupted = True
raise
except Exception:
interrupted = True
yield sse({"e": "传输中断,请点击重试"})
finally:
text = "".join(parts)
# 太短的半截回答没有价值(比如刚输出"您好"就断了),直接丢弃,
# 会话会停在"只有提问"的状态,前端会给出"重新回答"入口
if interrupted and len(text) >= 20:
save_message(conv_id, "assistant",
text + "\n\n(回答中断,以上内容未生成完整)")
return Response(
stream_with_context(generate()),
mimetype="text/event-stream",
headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"}
)
@app.route("/api/new_chat", methods=["POST"])
def new_chat():
now = now_str()
conn = sqlite3.connect(DB)
cur = conn.cursor()
cur.execute("INSERT INTO chat_history(title, ctime, utime) VALUES (?,?,?)",
("新对话", now, now))
conv_id = cur.lastrowid
conn.commit()
conn.close()
return jsonify({"id": conv_id})
@app.route("/api/get_history")
def get_history():
conn = sqlite3.connect(DB)
rows = conn.execute(
"SELECT id, title, ctime, COALESCE(utime, ctime) FROM chat_history "
"ORDER BY COALESCE(utime, ctime) DESC, id DESC"
).fetchall()
conn.close()
return jsonify([{"id": r[0], "title": r[1], "ctime": r[2], "utime": r[3]}
for r in rows])
# ===== 管理员后台 =====
@app.route("/api/admin/stats")
def admin_stats():
conn = sqlite3.connect(DB)
total_conv = conn.execute("SELECT COUNT(*) FROM chat_history").fetchone()[0]
total_msg = conn.execute("SELECT COUNT(*) FROM messages").fetchone()[0]
good = conn.execute("SELECT COUNT(*) FROM feedback WHERE vote=1").fetchone()[0]
bad = conn.execute("SELECT COUNT(*) FROM feedback WHERE vote=-1").fetchone()[0]
today = datetime.now().strftime("%Y-%m-%d")
today_conv = conn.execute("SELECT COUNT(*) FROM chat_history WHERE date(ctime)=date(?)", (today,)).fetchone()[0]
conn.close()
return jsonify({"total_conv": total_conv, "total_msg": total_msg,
"good": good, "bad": bad, "today_conv": today_conv})
# ===== 人工客服 =====
@app.route("/api/human/request", methods=["POST"])
def human_request():
conv_id = request.json.get("conv_id")
if not conv_id:
return jsonify({"ok": False, "error": "no conv"}), 400
now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
conn = sqlite3.connect(DB)
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 jsonify({"ok": True, "session_id": sid})
@app.route("/api/human/poll")
def human_poll():
sid = request.args.get("sid", type=int)
after = request.args.get("after", 0, type=int)
conn = sqlite3.connect(DB)
msgs = conn.execute("SELECT id, side, content, ctime, read_at FROM human_messages WHERE session_id=? AND id>? ORDER BY id", (sid, after)).fetchall()
st = conn.execute("SELECT status 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 jsonify({"status": st[0] if st else "closed",
"typing": is_typing,
"messages": [{"id": m[0], "side": m[1], "content": m[2], "ctime": m[3], "read": bool(m[4])} for m in msgs]})
@app.route("/api/human/send", methods=["POST"])
def human_send():
sid = request.json.get("sid")
side = request.json.get("side", "customer")
content = (request.json.get("content") or "").strip()
if not sid or not content:
return jsonify({"ok": False}), 400
now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
conn = sqlite3.connect(DB)
conn.execute("INSERT INTO human_messages(session_id, side, content, ctime) VALUES(?,?,?,?)", (sid, side, content, now))
conn.commit()
conn.close()
return jsonify({"ok": True})
@app.route("/api/human/queue")
def human_queue():
conn = sqlite3.connect(DB)
rows = conn.execute("SELECT id, conv_id, status, ctime FROM human_sessions WHERE status='waiting' ORDER BY id").fetchall()
conn.close()
return jsonify([{"id": r[0], "conv_id": r[1], "ctime": r[3]} for r in rows])
@app.route("/api/human/active")
def human_active():
conn = sqlite3.connect(DB)
rows = conn.execute("SELECT id, conv_id FROM human_sessions WHERE status='active' ORDER BY id").fetchall()
conn.close()
return jsonify([{"id": r[0], "conv_id": r[1]} for r in rows])
@app.route("/api/human/accept", methods=["POST"])
def human_accept():
sid = request.json.get("sid")
conn = sqlite3.connect(DB)
conn.execute("UPDATE human_sessions SET status='active' WHERE id=?", (sid,))
conn.commit()
conn.close()
return jsonify({"ok": True})
@app.route("/api/human/close", methods=["POST"])
def human_close():
sid = request.json.get("sid")
conn = sqlite3.connect(DB)
conn.execute("UPDATE human_sessions SET status='closed' WHERE id=?", (sid,))
conn.commit()
conn.close()
return jsonify({"ok": True})
# 客服正在输入(简单内存标记)
typing_map = {}
@app.route("/api/human/typing", methods=["POST"])
def human_typing():
sid = request.json.get("sid")
typing_map[str(sid)] = time.time()
return jsonify({"ok": True})
@app.route("/api/human/stop_typing", methods=["POST"])
def human_stop_typing():
sid = request.json.get("sid")
typing_map.pop(str(sid), None)
return jsonify({"ok": True})
@app.route("/api/human/read", methods=["POST"])
def human_read():
sid = request.json.get("sid")
conn = sqlite3.connect(DB)
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 jsonify({"ok": True})
@app.route("/api/get_messages")
def get_messages():
conv_id = request.args.get("conv_id", type=int)
conn = sqlite3.connect(DB)
# 只取最近 100 条,避免超长对话一次渲染几百个气泡卡死
rows = conn.execute(
"SELECT role, content, ctime FROM messages WHERE conv_id=? "
"ORDER BY id DESC LIMIT 100",
(conv_id,)
).fetchall()
conn.close()
rows = rows[::-1] # 反序回来变成正序
return jsonify([{"role": r[0], "content": r[1], "ctime": r[2]} for r in rows])
@app.route("/api/del_history", methods=["DELETE"])
def del_history():
hid = request.args.get("id", type=int)
conn = sqlite3.connect(DB)
cur = conn.cursor()
cur.execute("DELETE FROM messages WHERE conv_id=?", (hid,))
cur.execute("DELETE FROM chat_history WHERE id=?", (hid,))
conn.commit()
conn.close()
return jsonify({"ok": True})
@app.route("/api/clear_all", methods=["DELETE"])
def clear_all():
conn = sqlite3.connect(DB)
cur = conn.cursor()
cur.execute("DELETE FROM messages")
cur.execute("DELETE FROM chat_history")
conn.commit()
conn.close()
return jsonify({"ok": True})
UPLOAD_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "uploads")
ALLOWED_EXT = {".png", ".jpg", ".jpeg", ".gif", ".webp", ".pdf", ".txt", ".csv"}
MAX_UPLOAD_MB = 10
@app.route("/api/upload", methods=["POST"])
def upload_file():
"""上传附件:保存到 uploads 目录,返回文件名与访问地址"""
if "file" not in request.files:
return jsonify({"ok": False, "error": "没有文件"}), 400
f = request.files["file"]
if not f or not f.filename:
return jsonify({"ok": False, "error": "文件名为空"}), 400
ext = os.path.splitext(f.filename)[1].lower()
if ext not in ALLOWED_EXT:
return jsonify({"ok": False, "error": f"不支持的格式 {ext},仅支持图片/PDF/TXT/CSV"}), 400
f.seek(0, os.SEEK_END)
size = f.tell()
f.seek(0)
if size > MAX_UPLOAD_MB * 1024 * 1024:
return jsonify({"ok": False, "error": f"文件超过 {MAX_UPLOAD_MB}MB"}), 400
os.makedirs(UPLOAD_DIR, exist_ok=True)
safe_name = f"{datetime.now().strftime('%Y%m%d%H%M%S')}_{random.randint(1000, 9999)}{ext}"
f.save(os.path.join(UPLOAD_DIR, safe_name))
is_image = ext in (".png", ".jpg", ".jpeg", ".gif", ".webp")
return jsonify({
"ok": True,
"name": f.filename,
"url": f"/uploads/{safe_name}",
"is_image": is_image,
})
@app.route("/uploads/<path:name>")
def uploaded_file(name):
return send_from_directory(UPLOAD_DIR, name)
@app.route("/")
def index():
return render_template("index.html")
@app.route("/admin/login", methods=["POST"])
def admin_login():
pwd = request.json.get("password", "")
if pwd == "123":
return jsonify({"ok": True})
return jsonify({"ok": False}), 401
@app.route("/admin")
def admin():
return render_template("admin_login.html")
@app.route("/admin/dashboard")
def admin_dash():
return render_template("admin.html")
if __name__ == "__main__":
# 0.0.0.0 = 接受局域网内所有设备访问(手机、其他电脑);
# 只在演示时开启,演示完建议改回 127.0.0.1,避免别人误用消耗 API 额度
HOST = "0.0.0.0"
PORT = 5000 # 若报 WinError 10013,改成 8000 等端口
# 启动时打印本机局域网 IP,方便直接发给别人访问
def local_ips():
import socket
ips = set()
try:
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s.connect(("8.8.8.8", 80))
ips.add(s.getsockname()[0])
s.close()
except Exception:
pass
try:
for info in socket.getaddrinfo(socket.gethostname(), None, socket.AF_INET):
ip = info[4][0]
if not ip.startswith("127."):
ips.add(ip)
except Exception:
pass
return sorted(ips)
try:
from waitress import serve
print(" * 柏悦物流智能客服已启动(多线程)")
print(f" 本机访问: http://127.0.0.1:{PORT}")
for ip in local_ips():
print(f" 局域网访问:http://{ip}:{PORT}")
serve(app, host=HOST, port=PORT, threads=8)
except ImportError:
# 没装 waitress 时回退到 Flask 自带服务器,保证仍能运行
print(f"本机访问: http://127.0.0.1:{PORT}")
for ip in local_ips():
print(f"局域网访问:http://{ip}:{PORT}")
app.run(host=HOST, port=PORT, debug=True, use_reloader=False)