Files
wxsales/demo/web/server.py
T
freedakgmail 5926c73ff1 微信营销管理系统 - 益童宝销售管理平台
- 数据库: PostgreSQL schema + 3名销售/21客户/341消息/6成交
- 采集代理: mock_sync.py 模拟微信聊天同步
- AI分析: analyze.py 规则模式 + 千问LLM模式
- 后端: FastAPI 11个API接口
- 前端: 仪表盘/销售列表/客户列表/成交记录/交流分析/录入成交
- 交流分析: 全部客户对话概览 + LLM标准范式对话生成
- 部署: systemd + nginx, 已部署至 sale.all8ai.top
2026-07-14 22:31:29 +08:00

668 lines
24 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.
#!/usr/bin/env python3
"""微信营销管理系统 MVP — FastAPI 后端。
提供 11 个 API 接口 + 静态前端托管。
"""
from __future__ import annotations
import datetime as dt
import json
import pathlib
import re
import subprocess
import sys
import psycopg2
import psycopg2.pool
from fastapi import FastAPI, HTTPException, Query
from fastapi.staticfiles import StaticFiles
from fastapi.responses import RedirectResponse
from pydantic import BaseModel
# ---------------------------------------------------------------------------
# 配置
# ---------------------------------------------------------------------------
ROOT = pathlib.Path(__file__).resolve().parent
STATIC_DIR = ROOT / "static"
SOCKET_DIR = ROOT.parent / "db" / "socket"
DSN = f"host={SOCKET_DIR} port=5434 dbname=wxchat_sales"
app = FastAPI(title="微信营销管理系统 MVP")
def parse_pg_array(val):
"""将 PostgreSQL TEXT[] 的字符串表示解析为 Python list。"""
if val is None:
return []
if isinstance(val, list):
return val
if isinstance(val, str):
s = val.strip()
if s == "{}" or s == "":
return []
# 去掉首尾 { }
s = s[1:-1] if s.startswith("{") and s.endswith("}") else s
# 按逗号分割(简单处理,不考虑逗号在引号内的情况)
return [item.strip().strip('"') for item in s.split(",") if item.strip()]
return []
# 连接池
_pool: psycopg2.pool.SimpleConnectionPool | None = None
def get_pool():
global _pool
if _pool is None:
_pool = psycopg2.pool.SimpleConnectionPool(1, 10, DSN)
return _pool
def get_conn():
return get_pool().getconn()
def put_conn(conn):
get_pool().putconn(conn)
# ---------------------------------------------------------------------------
# Pydantic 模型
# ---------------------------------------------------------------------------
class DealCreate(BaseModel):
salesperson_id: int
customer_id: int | None = None
contact_id: int | None = None
product_name: str
amount: float
deal_date: str
notes: str | None = None
status: str = "closed"
class AnalyzeRequest(BaseModel):
mode: str = "rule"
force: bool = False
# ---------------------------------------------------------------------------
# API 路由
# ---------------------------------------------------------------------------
@app.get("/api/dashboard")
def dashboard():
conn = get_conn()
try:
with conn.cursor() as cur:
cur.execute("SELECT count(*) FROM message")
total_messages = cur.fetchone()[0]
cur.execute("SELECT count(*) FROM customer")
active_customers = cur.fetchone()[0]
cur.execute("""
SELECT COALESCE(sum(amount), 0) FROM deal
WHERE status = 'closed'
AND deal_date >= date_trunc('month', now())
""")
monthly_deal_amount = float(cur.fetchone()[0])
cur.execute("""
SELECT s.id, s.name, s.team,
count(DISTINCT m.id) AS msg_count,
count(DISTINCT cu.id) AS customer_count,
COALESCE(sd.deal_amount, 0) AS deal_amount,
max(conv.last_synced_at) AS last_synced
FROM salesperson s
LEFT JOIN message m ON m.salesperson_id = s.id
LEFT JOIN customer cu ON cu.salesperson_id = s.id
LEFT JOIN conversation conv ON conv.salesperson_id = s.id
LEFT JOIN LATERAL (
SELECT COALESCE(sum(d.amount), 0) AS deal_amount
FROM deal d WHERE d.salesperson_id = s.id AND d.status = 'closed'
) sd ON true
GROUP BY s.id, s.name, s.team, sd.deal_amount
ORDER BY s.id
""")
salespersons = [
{
"id": r[0], "name": r[1], "team": r[2],
"message_count": r[3], "customer_count": r[4],
"deal_amount": float(r[5]),
"last_synced_at": r[6].isoformat() if r[6] else None,
}
for r in cur.fetchall()
]
return {
"total_messages": total_messages,
"active_customers": active_customers,
"monthly_deal_amount": monthly_deal_amount,
"salespersons": salespersons,
}
finally:
put_conn(conn)
@app.get("/api/salespersons")
def list_salespersons():
conn = get_conn()
try:
with conn.cursor() as cur:
cur.execute("""
SELECT s.id, s.name, s.team, s.wx_account, s.device_id,
count(DISTINCT c.id) AS contacts,
count(DISTINCT cu.id) AS customers,
count(DISTINCT m.id) AS messages,
COALESCE(sd.deal_amount, 0) AS deal_amount,
max(conv.last_synced_at) AS last_synced
FROM salesperson s
LEFT JOIN contact c ON c.salesperson_id = s.id
LEFT JOIN customer cu ON cu.salesperson_id = s.id
LEFT JOIN message m ON m.salesperson_id = s.id
LEFT JOIN conversation conv ON conv.salesperson_id = s.id
LEFT JOIN LATERAL (
SELECT COALESCE(sum(d.amount), 0) AS deal_amount
FROM deal d WHERE d.salesperson_id = s.id AND d.status = 'closed'
) sd ON true
GROUP BY s.id, s.name, s.team, s.wx_account, s.device_id, sd.deal_amount
ORDER BY s.id
""")
return [
{
"id": r[0], "name": r[1], "team": r[2],
"wx_account": r[3], "device_id": r[4],
"contact_count": r[5], "customer_count": r[6],
"message_count": r[7], "deal_amount": float(r[8]),
"last_synced_at": r[9].isoformat() if r[9] else None,
}
for r in cur.fetchall()
]
finally:
put_conn(conn)
@app.get("/api/customers")
def list_customers(
salesperson_id: int | None = Query(None),
intent_level: str | None = Query(None),
):
conn = get_conn()
try:
with conn.cursor() as cur:
sql = """
SELECT cu.id, cu.salesperson_id, s.name AS salesperson_name,
cu.customer_name, cu.industry, cu.intent_level,
cu.key_needs, cu.stage, cu.last_analysis,
c.display_name AS contact_display_name,
max(m.created_at) AS last_message_at
FROM customer cu
JOIN salesperson s ON s.id = cu.salesperson_id
JOIN contact c ON c.id = cu.contact_id
LEFT JOIN conversation conv ON conv.contact_id = c.id
LEFT JOIN message m ON m.conversation_id = conv.id
"""
conditions = []
params = []
if salesperson_id is not None:
conditions.append("cu.salesperson_id = %s")
params.append(salesperson_id)
if intent_level is not None:
conditions.append("cu.intent_level = %s")
params.append(intent_level)
if conditions:
sql += " WHERE " + " AND ".join(conditions)
sql += """
GROUP BY cu.id, cu.salesperson_id, s.name, cu.customer_name,
cu.industry, cu.intent_level, cu.key_needs, cu.stage,
cu.last_analysis, c.display_name
ORDER BY cu.salesperson_id, cu.id
"""
cur.execute(sql, params)
return [
{
"id": r[0], "salesperson_id": r[1],
"salesperson_name": r[2],
"customer_name": r[3], "industry": r[4],
"intent_level": r[5], "key_needs": parse_pg_array(r[6]),
"stage": r[7],
"last_analysis": r[8].isoformat() if r[8] else None,
"contact_display_name": r[9],
"last_message_at": r[10].isoformat() if r[10] else None,
}
for r in cur.fetchall()
]
finally:
put_conn(conn)
@app.get("/api/customers/{customer_id}")
def get_customer(customer_id: int):
conn = get_conn()
try:
with conn.cursor() as cur:
cur.execute("""
SELECT cu.id, cu.salesperson_id, s.name AS salesperson_name,
cu.customer_name, cu.industry, cu.intent_level,
cu.key_needs, cu.reason, cu.summary, cu.stage,
cu.key_points, cu.objections, cu.next_action, cu.last_analysis,
c.display_name AS contact_display_name,
c.remark, c.nickname
FROM customer cu
JOIN salesperson s ON s.id = cu.salesperson_id
JOIN contact c ON c.id = cu.contact_id
WHERE cu.id = %s
""", (customer_id,))
r = cur.fetchone()
if not r:
raise HTTPException(status_code=404, detail="客户不存在")
return {
"id": r[0], "salesperson_id": r[1],
"salesperson_name": r[2],
"customer_name": r[3], "industry": r[4],
"intent_level": r[5], "key_needs": parse_pg_array(r[6]),
"reason": r[7], "summary": r[8], "stage": r[9],
"key_points": parse_pg_array(r[10]),
"objections": parse_pg_array(r[11]), "next_action": r[12],
"last_analysis": r[13].isoformat() if r[13] else None,
"contact_display_name": r[14],
"contact_remark": r[15], "contact_nickname": r[16],
}
finally:
put_conn(conn)
@app.get("/api/customers/{customer_id}/messages")
def get_customer_messages(
customer_id: int,
page: int = Query(1, ge=1),
page_size: int = Query(50, ge=1, le=200),
):
conn = get_conn()
try:
with conn.cursor() as cur:
# 获取 contact_id
cur.execute("SELECT contact_id FROM customer WHERE id = %s", (customer_id,))
row = cur.fetchone()
if not row:
raise HTTPException(status_code=404, detail="客户不存在")
contact_id = row[0]
# 获取会话
cur.execute("SELECT id FROM conversation WHERE contact_id = %s", (contact_id,))
conv_ids = [r[0] for r in cur.fetchall()]
if not conv_ids:
return {"messages": [], "total": 0, "page": page, "page_size": page_size}
# 总数
cur.execute(
"SELECT count(*) FROM message WHERE conversation_id = ANY(%s)",
(conv_ids,),
)
total = cur.fetchone()[0]
# 分页
offset = (page - 1) * page_size
cur.execute("""
SELECT id, sender_display_name, message_type, normalized_content, created_at
FROM message
WHERE conversation_id = ANY(%s)
ORDER BY created_at, id
LIMIT %s OFFSET %s
""", (conv_ids, page_size, offset))
messages = [
{
"id": r[0],
"sender_display_name": r[1],
"message_type": r[2],
"normalized_content": r[3],
"created_at": r[4].isoformat() if r[4] else None,
}
for r in cur.fetchall()
]
return {
"messages": messages,
"total": total,
"page": page,
"page_size": page_size,
}
finally:
put_conn(conn)
@app.get("/api/customers/{customer_id}/deals")
def get_customer_deals(customer_id: int):
conn = get_conn()
try:
with conn.cursor() as cur:
cur.execute("""
SELECT d.id, d.product_name, d.amount, d.deal_date, d.status, d.notes,
d.created_at, s.name AS salesperson_name
FROM deal d
JOIN salesperson s ON s.id = d.salesperson_id
WHERE d.customer_id = %s
ORDER BY d.deal_date DESC
""", (customer_id,))
return [
{
"id": r[0], "product_name": r[1],
"amount": float(r[2]),
"deal_date": r[3].isoformat() if r[3] else None,
"status": r[4], "notes": r[5],
"created_at": r[6].isoformat() if r[6] else None,
"salesperson_name": r[7],
}
for r in cur.fetchall()
]
finally:
put_conn(conn)
@app.post("/api/deals")
def create_deal(deal: DealCreate):
conn = get_conn()
try:
with conn.cursor() as cur:
# 如果没有 contact_id,从 customer 获取
contact_id = deal.contact_id
if contact_id is None and deal.customer_id:
cur.execute("SELECT contact_id FROM customer WHERE id = %s", (deal.customer_id,))
row = cur.fetchone()
if row:
contact_id = row[0]
cur.execute("""
INSERT INTO deal (salesperson_id, customer_id, contact_id, product_name,
amount, deal_date, status, notes)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s)
RETURNING id
""", (
deal.salesperson_id, deal.customer_id, contact_id,
deal.product_name, deal.amount, deal.deal_date,
deal.status, deal.notes,
))
deal_id = cur.fetchone()[0]
conn.commit()
return {"id": deal_id, "message": "成交记录已创建"}
except Exception:
conn.rollback()
raise
finally:
put_conn(conn)
@app.get("/api/deals")
def list_deals(
salesperson_id: int | None = Query(None),
start_date: str | None = Query(None),
end_date: str | None = Query(None),
):
conn = get_conn()
try:
with conn.cursor() as cur:
sql = """
SELECT d.id, d.salesperson_id, s.name AS salesperson_name,
d.customer_id, cu.customer_name,
d.product_name, d.amount, d.deal_date, d.status, d.notes
FROM deal d
JOIN salesperson s ON s.id = d.salesperson_id
LEFT JOIN customer cu ON cu.id = d.customer_id
"""
conditions = []
params = []
if salesperson_id is not None:
conditions.append("d.salesperson_id = %s")
params.append(salesperson_id)
if start_date:
conditions.append("d.deal_date >= %s")
params.append(start_date)
if end_date:
conditions.append("d.deal_date <= %s")
params.append(end_date)
if conditions:
sql += " WHERE " + " AND ".join(conditions)
sql += " ORDER BY d.deal_date DESC, d.id DESC"
cur.execute(sql, params)
return [
{
"id": r[0], "salesperson_id": r[1],
"salesperson_name": r[2],
"customer_id": r[3], "customer_name": r[4],
"product_name": r[5],
"amount": float(r[6]),
"deal_date": r[7].isoformat() if r[7] else None,
"status": r[8], "notes": r[9],
}
for r in cur.fetchall()
]
finally:
put_conn(conn)
@app.post("/api/analyze")
def trigger_analyze(req: AnalyzeRequest):
"""触发 AI 分析(调用 analyze.py 脚本)。"""
analyze_script = ROOT.parent / "ai" / "analyze.py"
cmd = [sys.executable, str(analyze_script), "--mode", req.mode]
if req.force:
cmd.append("--force")
try:
result = subprocess.run(
cmd, capture_output=True, text=True, timeout=120,
)
if result.returncode != 0:
raise HTTPException(status_code=500, detail=result.stderr)
return json.loads(result.stdout)
except subprocess.TimeoutExpired:
raise HTTPException(status_code=504, detail="分析超时")
@app.get("/api/conversations")
def list_conversations(
salesperson_id: int | None = Query(None),
stage: str | None = Query(None),
):
"""全部客户的交流对话分析概览。"""
conn = get_conn()
try:
with conn.cursor() as cur:
sql = """
SELECT cu.id, cu.customer_name, s.name AS salesperson_name,
cu.industry, cu.intent_level, cu.stage,
cu.summary, cu.key_points, cu.objections, cu.next_action,
cu.key_needs, cu.last_analysis,
(SELECT count(*) FROM message m
JOIN conversation conv ON conv.id = m.conversation_id
WHERE conv.contact_id = cu.contact_id) AS msg_count
FROM customer cu
JOIN salesperson s ON s.id = cu.salesperson_id
"""
conditions = []
params = []
if salesperson_id is not None:
conditions.append("cu.salesperson_id = %s")
params.append(salesperson_id)
if stage:
conditions.append("cu.stage = %s")
params.append(stage)
if conditions:
sql += " WHERE " + " AND ".join(conditions)
sql += " ORDER BY cu.salesperson_id, cu.id"
cur.execute(sql, params)
return [
{
"id": r[0], "customer_name": r[1],
"salesperson_name": r[2], "industry": r[3],
"intent_level": r[4], "stage": r[5],
"summary": r[6],
"key_points": parse_pg_array(r[7]),
"objections": parse_pg_array(r[8]),
"next_action": r[9],
"key_needs": parse_pg_array(r[10]),
"last_analysis": r[11].isoformat() if r[11] else None,
"msg_count": r[12],
}
for r in cur.fetchall()
]
finally:
put_conn(conn)
@app.post("/api/conversations/{customer_id}/paradigm")
def generate_paradigm(customer_id: int):
"""调用 LLM 生成标准范式对话。"""
import os as _os
import urllib.request as _urllib
import urllib.error as _urllib_err
api_key = _os.environ.get("DASHSCOPE_API_KEY")
if not api_key:
raise HTTPException(status_code=500, detail="未设置 DASHSCOPE_API_KEY 环境变量")
conn = get_conn()
try:
with conn.cursor() as cur:
# 获取客户信息
cur.execute("""
SELECT cu.customer_name, cu.industry, cu.stage, cu.summary,
cu.objections, cu.key_needs, s.name
FROM customer cu
JOIN salesperson s ON s.id = cu.salesperson_id
WHERE cu.id = %s
""", (customer_id,))
customer = cur.fetchone()
if not customer:
raise HTTPException(status_code=404, detail="客户不存在")
# 获取对话记录
cur.execute("SELECT contact_id FROM customer WHERE id = %s", (customer_id,))
contact_id = cur.fetchone()[0]
cur.execute("SELECT id FROM conversation WHERE contact_id = %s", (contact_id,))
conv_ids = [r[0] for r in cur.fetchall()]
if not conv_ids:
raise HTTPException(status_code=400, detail="该客户无对话记录")
cur.execute("""
SELECT sender_display_name, message_type, normalized_content, created_at
FROM message
WHERE conversation_id = ANY(%s)
ORDER BY created_at, id
LIMIT 100
""", (conv_ids,))
messages = cur.fetchall()
finally:
put_conn(conn)
# 构建对话文本
dialog_lines = []
for m in messages:
sender = m[0] or "未知"
content = m[2] or f"[{m[1]}]"
dialog_lines.append(f"{sender}: {content}")
dialog_text = "\n".join(dialog_lines)
if len(dialog_text) > 6000:
dialog_text = dialog_text[:3000] + "\n...(中间部分省略)...\n" + dialog_text[-3000:]
customer_name, industry, stage, summary, objections, key_needs, sp_name = customer
objections_str = "".join(parse_pg_array(objections)) if objections else ""
key_needs_str = "".join(parse_pg_array(key_needs)) if key_needs else "未明确"
prompt = f"""你是一个微信销售对话分析专家。请基于以下真实销售对话,生成一套标准范式对话模板。
## 产品背景
益童宝儿童益生菌粉:丹麦进口菌株,主打小儿抗过敏(湿疹、鼻炎、食物过敏),298元/盒,3盒套餐798元,6盒套餐1499元。目标客户是宝妈。
## 客户信息
- 客户: {customer_name}
- 行业: {industry}
- 销售人员: {sp_name}
- 当前阶段: {stage}
- 关键需求: {key_needs_str}
- 异议: {objections_str}
- 摘要: {summary or ''}
## 真实对话记录
{dialog_text}
## 输出要求
请基于上述真实对话,提炼并生成一套**标准范式对话模板**,即针对此类客户的最优销售话术流程。输出严格 JSON:
- customer_type: 客户类型描述(如"湿疹宝妈-价格敏感型"
- stages: 按销售阶段排列的对话步骤数组,每个元素包含:
- stage: 阶段名称(建立联系/需求发现/报价/异议处理/成交)
- goal: 该阶段目标
- sales_script: 销售话术(1-3句,具体可执行)
- expected_response: 预期客户回应
- tips: 该阶段技巧提示
- key_techniques: 核心销售技巧总结(数组,每条一句话)
- improvement_points: 真实对话中可改进的点(数组,每条一句话)
只输出 JSON,不要其他文字。"""
body = json.dumps({
"model": "qwen-plus",
"messages": [
{"role": "system", "content": "你是微信销售对话分析专家,擅长从真实对话中提炼标准范式。"},
{"role": "user", "content": prompt},
],
"response_format": {"type": "json_object"},
"temperature": 0.4,
}).encode("utf-8")
req = _urllib.Request(
"https://dashscope.aliyuncs.com/compatible-mode/v1/chat/completions",
data=body,
headers={
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json",
},
method="POST",
)
try:
with _urllib.urlopen(req, timeout=60) as resp:
result = json.loads(resp.read().decode("utf-8"))
content = result["choices"][0]["message"]["content"]
parsed = json.loads(content)
return parsed
except Exception as exc:
raise HTTPException(status_code=502, detail=f"LLM 调用失败: {exc}")
@app.get("/api/sync/status")
def sync_status():
conn = get_conn()
try:
with conn.cursor() as cur:
cur.execute("""
SELECT s.id, s.name,
max(conv.last_synced_at) AS last_synced,
count(DISTINCT m.id) AS message_count
FROM salesperson s
LEFT JOIN conversation conv ON conv.salesperson_id = s.id
LEFT JOIN message m ON m.salesperson_id = s.id
GROUP BY s.id, s.name
ORDER BY s.id
""")
return [
{
"salesperson_id": r[0],
"name": r[1],
"last_synced_at": r[2].isoformat() if r[2] else None,
"message_count": r[3],
}
for r in cur.fetchall()
]
finally:
put_conn(conn)
# ---------------------------------------------------------------------------
# 静态文件托管(前端)
# ---------------------------------------------------------------------------
if STATIC_DIR.exists():
app.mount("/", StaticFiles(directory=str(STATIC_DIR), html=True), name="static")