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") # 补 note 列（会话备注） hs_cols = [r[1] for r in cur.execute("PRAGMA table_info(human_sessions)").fetchall()] if "note" not in hs_cols: cur.execute("ALTER TABLE human_sessions ADD COLUMN note TEXT DEFAULT ''") # 满意度评价表 cur.execute(''' CREATE TABLE IF NOT EXISTS ratings ( id INTEGER PRIMARY KEY AUTOINCREMENT, sid INTEGER NOT NULL, stars INTEGER NOT NULL, comment TEXT DEFAULT '', ctime TEXT NOT NULL ) ''') # 老数据只有标题，补一条 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: def _img_reply(): yield sse({"t": "收到您的图片，请描述一下您的问题。"}) yield sse({"d": 1}) return Response(stream_with_context(_img_reply()), 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"(? 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"} ) WELCOME_TEXT = "您好！欢迎咨询柏悦物流智能客服，请问有什么可以帮您？输入运单号可查轨迹，或点击下方快捷选项。" @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 cur.execute("INSERT INTO messages(conv_id, role, content, ctime) VALUES (?,?,?,?)", (conv_id, "assistant", WELCOME_TEXT, now)) conn.commit() conn.close() return jsonify({"id": conv_id, "welcome": WELCOME_TEXT}) @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] rating_avg = conn.execute("SELECT AVG(stars) FROM ratings").fetchone()[0] or 0 rating_count = conn.execute("SELECT COUNT(*) FROM ratings").fetchone()[0] conn.close() return jsonify({"total_conv": total_conv, "total_msg": total_msg, "good": good, "bad": bad, "today_conv": today_conv, "rating_avg": round(rating_avg, 1), "rating_count": rating_count}) # ===== 人工客服 ===== @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 # 工作时间 8:00-22:00，其他时间不接人工 _h = datetime.now().hour if _h < 8 or _h >= 22: return jsonify({"ok": False, "closed": True, "msg": "人工客服工作时间为每天8:00-22:00，当前为非工作时间，请留言"}), 200 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, COALESCE(note,'') 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", "note": st[1] if st else "", "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/recall", methods=["POST"]) def human_recall(): data = request.json or {} mid = data.get("mid") conn = sqlite3.connect(DB) row = conn.execute("SELECT session_id, side FROM human_messages WHERE id=?", (mid,)).fetchone() if row: sid, side = row conn.execute("UPDATE human_messages SET content='', side='system' WHERE id=?", (mid,)) conn.execute("INSERT INTO human_messages(session_id, side, content, ctime) VALUES(?,?,?,?)", (sid, "system", "该消息已撤回", datetime.now().strftime("%Y-%m-%d %H:%M:%S"))) 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}) @app.route("/api/human/transfer", methods=["POST"]) def human_transfer(): sid = request.json.get("sid") conn = sqlite3.connect(DB) conn.execute("UPDATE human_sessions SET status='waiting' WHERE id=?", (sid,)) conn.execute("INSERT INTO human_messages(session_id, side, content, ctime) VALUES(?,?,?,?)", (sid, "system", "客服已转接，请其他客服接入", datetime.now().strftime("%Y-%m-%d %H:%M:%S"))) conn.commit() conn.close() return jsonify({"ok": True}) @app.route("/api/agents/list") def agents_list(): conn = sqlite3.connect(DB) rows = conn.execute("SELECT id, username, name FROM agents ORDER BY id").fetchall() conn.close() return jsonify([{"id": r[0], "username": r[1], "name": r[2]} for r in rows]) @app.route("/api/rate", methods=["POST"]) def rate_chat(): data = request.json or {} sid = data.get("sid") stars = data.get("stars", 0) comment = data.get("comment", "") if not sid or not isinstance(stars, int) or stars < 1 or stars > 5: return jsonify({"ok": False}), 400 conn = sqlite3.connect(DB) conn.execute("INSERT INTO ratings(sid, stars, comment, ctime) VALUES(?,?,?,?)", (sid, stars, comment, datetime.now().strftime("%Y-%m-%d %H:%M:%S"))) conn.commit() conn.close() return jsonify({"ok": True}) @app.route("/api/human/note", methods=["POST"]) def human_note(): data = request.json or {} sid = data.get("sid") note = data.get("note", "") conn = sqlite3.connect(DB) conn.execute("UPDATE human_sessions SET note=? WHERE id=?", (note, 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/") def uploaded_file(name): return send_from_directory(UPLOAD_DIR, name) @app.route("/") def index(): return render_template("index.html") @app.route("/api/my_orders") def my_orders(): phone = request.args.get("phone", "").strip() if not phone: return jsonify({"orders": []}) conn = sqlite3.connect(DB) rows = conn.execute("SELECT track_no, status, created FROM orders WHERE phone=? ORDER BY id DESC", (phone,)).fetchall() conn.close() return jsonify({"orders": [{"track_no": r[0], "status": r[1], "created": r[2]} for r in rows]}) @app.route("/api/ticket", methods=["POST"]) def add_ticket(): data = request.get_json() or {} conn = sqlite3.connect(DB) conn.execute("INSERT INTO tickets(content, phone, ctime, status) VALUES(?,?,?,?)", (data.get("content",""), data.get("phone",""), datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "pending")) conn.commit(); conn.close() return jsonify({"ok": True}) @app.route("/api/tickets") def list_tickets(): conn = sqlite3.connect(DB) rows = conn.execute("SELECT id, content, phone, ctime FROM tickets WHERE status='pending' ORDER BY id DESC").fetchall() conn.close() return jsonify([{"id":r[0],"content":r[1],"phone":r[2],"ctime":r[3]} for r in rows]) @app.route("/admin/login", methods=["POST"]) def admin_login(): data = request.json or {} u = data.get("username", "") p = data.get("password", "") conn = sqlite3.connect(DB) row = conn.execute("SELECT name FROM agents WHERE username=? AND password=?", (u, p)).fetchone() conn.close() if row: return jsonify({"ok": True, "name": row[0]}) return jsonify({"ok": False}), 401 @app.route("/api/export_chat") def export_chat(): conn = sqlite3.connect(DB) rows = conn.execute("SELECT h.title, m.role, m.content, m.ctime FROM messages m JOIN chat_history h ON m.conv_id=h.id ORDER BY m.id").fetchall() conn.close() from openpyxl import Workbook from openpyxl.styles import Font, PatternFill, Alignment wb = Workbook() ws = wb.active ws.title = "聊天记录" headers = ["对话标题", "角色", "内容", "时间"] ws.append(headers) header_fill = PatternFill(start_color="2978e8", end_color="2978e8", fill_type="solid") for cell in ws[1]: cell.font = Font(bold=True, color="FFFFFF") cell.fill = header_fill cell.alignment = Alignment(horizontal="center") role_map = {"user": "客户", "assistant": "客服", "system": "系统"} for r in rows: ws.append([r[0], role_map.get(r[1], r[1]), (r[2] or ""), r[3]]) ws.column_dimensions["A"].width = 20 ws.column_dimensions["B"].width = 8 ws.column_dimensions["C"].width = 80 ws.column_dimensions["D"].width = 20 import io buf = io.BytesIO() wb.save(buf) buf.seek(0) return Response(buf.getvalue(), mimetype="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", headers={"Content-Disposition": "attachment;filename=chat_export.xlsx"}) @app.route("/api/search_chat") def search_chat(): kw = request.args.get("q", "").strip() if not kw: return jsonify({"results": []}) conn = sqlite3.connect(DB) rows = conn.execute("""SELECT h.title, m.role, m.content, m.ctime FROM messages m JOIN chat_history h ON m.conv_id=h.id WHERE m.content LIKE ? ORDER BY m.id DESC LIMIT 50""", ("%"+kw+"%",)).fetchall() conn.close() return jsonify([{"title":r[0],"role":r[1],"content":r[2],"ctime":r[3]} for r in rows]) @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)