selfrelease
1046 lines
39 KiB
Python
1046 lines
39 KiB
Python
#!/usr/bin/env python3
|
||
"""微信营销管理系统 MVP — FastAPI 后端。
|
||
|
||
提供 11 个 API 接口 + 静态前端托管。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import datetime as dt
|
||
import json
|
||
import os
|
||
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, StreamingResponse
|
||
from pydantic import BaseModel
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 配置
|
||
# ---------------------------------------------------------------------------
|
||
|
||
ROOT = pathlib.Path(__file__).resolve().parent
|
||
STATIC_DIR = ROOT / "static"
|
||
SOCKET_DIR = ROOT.parent / "db" / "socket"
|
||
|
||
# DSN：优先环境变量 DATABASE_URL，其次 DB_HOST/DB_PORT，最后默认本地 socket
|
||
if os.environ.get("DATABASE_URL"):
|
||
DSN = os.environ["DATABASE_URL"]
|
||
elif os.environ.get("DB_HOST"):
|
||
DSN = f"host={os.environ['DB_HOST']} port={os.environ.get('DB_PORT', '5432')} dbname={os.environ.get('DB_NAME', 'wxchat_sales')} user={os.environ.get('DB_USER', 'postgres')} password={os.environ.get('DB_PASSWORD', '')}"
|
||
else:
|
||
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_id: int | None = None
|
||
product_name: str
|
||
amount: float
|
||
deal_date: str
|
||
notes: str | None = None
|
||
status: str = "closed"
|
||
|
||
|
||
class ProductCreate(BaseModel):
|
||
"""新增产品请求模型。"""
|
||
name: str
|
||
category: str
|
||
spec: str | None = None
|
||
price: float | None = None
|
||
description: str | None = None
|
||
positioning: str | None = None
|
||
selling_points: list[str] | None = None
|
||
target_audience: str | None = None
|
||
status: str = "active"
|
||
|
||
|
||
class ProductUpdate(BaseModel):
|
||
"""编辑产品请求模型。"""
|
||
name: str | None = None
|
||
category: str | None = None
|
||
spec: str | None = None
|
||
price: float | None = None
|
||
description: str | None = None
|
||
positioning: str | None = None
|
||
selling_points: list[str] | None = None
|
||
target_audience: str | None = None
|
||
status: str | None = None
|
||
|
||
|
||
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()
|
||
]
|
||
|
||
# 按产品线统计本月销售额
|
||
cur.execute("""
|
||
SELECT p.category,
|
||
COALESCE(sum(d.amount), 0) AS amount,
|
||
count(*) AS deal_count
|
||
FROM deal d
|
||
LEFT JOIN product p ON p.id = d.product_id
|
||
WHERE d.status = 'closed'
|
||
AND d.deal_date >= date_trunc('month', now())
|
||
GROUP BY p.category
|
||
ORDER BY amount DESC
|
||
""")
|
||
product_stats = [
|
||
{
|
||
"category": r[0] or "未分类",
|
||
"amount": float(r[1]),
|
||
"deal_count": r[2],
|
||
}
|
||
for r in cur.fetchall()
|
||
]
|
||
|
||
return {
|
||
"total_messages": total_messages,
|
||
"active_customers": active_customers,
|
||
"monthly_deal_amount": monthly_deal_amount,
|
||
"salespersons": salespersons,
|
||
"product_stats": product_stats,
|
||
}
|
||
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),
|
||
product_id: int | None = Query(None),
|
||
):
|
||
conn = get_conn()
|
||
try:
|
||
with conn.cursor() as cur:
|
||
sql = """
|
||
SELECT DISTINCT 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,
|
||
string_agg(DISTINCT p.category, ', ') AS deal_categories,
|
||
string_agg(m.normalized_content, ' ') AS all_messages
|
||
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
|
||
LEFT JOIN deal d ON d.customer_id = cu.id
|
||
LEFT JOIN product p ON p.id = d.product_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 product_id is not None:
|
||
conditions.append("d.product_id = %s")
|
||
params.append(product_id)
|
||
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)
|
||
# 产品线关键词：通过对话内容推断产品线
|
||
diet_keywords = ["减肥", "减脂", "瘦身", "B420", "纤益", "代餐", "体脂", "产后减肥", "节食"]
|
||
child_keywords = ["湿疹", "过敏", "鼻炎", "腹泻", "便秘", "免疫力", "胀气",
|
||
"拉肚子", "打喷嚏", "流鼻涕", "感冒", "食物过敏", "牛奶蛋白",
|
||
"宝宝", "宝妈", "儿童", "丹麦", "鼠李糖", "LGG", "合生元"]
|
||
results = []
|
||
for r in cur.fetchall():
|
||
deal_cats = r[11]
|
||
all_msgs = r[12] or ""
|
||
# 优先用 deal 关联的产品线，其次通过对话内容推断
|
||
if deal_cats:
|
||
product_categories = deal_cats
|
||
elif any(kw in all_msgs for kw in diet_keywords):
|
||
product_categories = "女性减肥益生菌"
|
||
elif any(kw in all_msgs for kw in child_keywords):
|
||
product_categories = "儿童益生菌"
|
||
else:
|
||
product_categories = None
|
||
results.append({
|
||
"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,
|
||
"product_categories": product_categories,
|
||
})
|
||
return results
|
||
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]
|
||
|
||
# 如果有 product_id 但没有 product_name，从 product 表获取
|
||
product_name = deal.product_name
|
||
if deal.product_id and not product_name:
|
||
cur.execute("SELECT name, spec FROM product WHERE id = %s", (deal.product_id,))
|
||
prow = cur.fetchone()
|
||
if prow:
|
||
product_name = f"{prow[0]} {prow[1]}" if prow[1] else prow[0]
|
||
|
||
cur.execute("""
|
||
INSERT INTO deal (salesperson_id, customer_id, contact_id, product_id, product_name,
|
||
amount, deal_date, status, notes)
|
||
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)
|
||
RETURNING id
|
||
""", (
|
||
deal.salesperson_id, deal.customer_id, contact_id,
|
||
deal.product_id, 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),
|
||
product_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_id, 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 product_id is not None:
|
||
conditions.append("d.product_id = %s")
|
||
params.append(product_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_id": r[5],
|
||
"product_name": r[6],
|
||
"amount": float(r[7]),
|
||
"deal_date": r[8].isoformat() if r[8] else None,
|
||
"status": r[9], "notes": r[10],
|
||
}
|
||
for r in cur.fetchall()
|
||
]
|
||
finally:
|
||
put_conn(conn)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 产品管理 API
|
||
# ---------------------------------------------------------------------------
|
||
|
||
@app.get("/api/products")
|
||
def list_products(
|
||
category: str | None = Query(None),
|
||
status: str | None = Query(None),
|
||
):
|
||
"""获取产品列表，支持按产品线和状态筛选。"""
|
||
conn = get_conn()
|
||
try:
|
||
with conn.cursor() as cur:
|
||
sql = """
|
||
SELECT p.id, p.name, p.category, p.spec, p.price,
|
||
p.description, p.positioning, p.selling_points,
|
||
p.target_audience, p.status, p.created_at,
|
||
COALESCE(ds.deal_count, 0) AS deal_count,
|
||
COALESCE(ds.total_amount, 0) AS total_amount
|
||
FROM product p
|
||
LEFT JOIN LATERAL (
|
||
SELECT count(*) AS deal_count, COALESCE(sum(d.amount), 0) AS total_amount
|
||
FROM deal d WHERE d.product_id = p.id AND d.status = 'closed'
|
||
) ds ON true
|
||
"""
|
||
conditions = []
|
||
params = []
|
||
if category:
|
||
conditions.append("p.category = %s")
|
||
params.append(category)
|
||
if status:
|
||
conditions.append("p.status = %s")
|
||
params.append(status)
|
||
if conditions:
|
||
sql += " WHERE " + " AND ".join(conditions)
|
||
sql += " ORDER BY p.category, p.id"
|
||
cur.execute(sql, params)
|
||
return [
|
||
{
|
||
"id": r[0], "name": r[1], "category": r[2],
|
||
"spec": r[3], "price": float(r[4]) if r[4] else None,
|
||
"description": r[5], "positioning": r[6],
|
||
"selling_points": parse_pg_array(r[7]),
|
||
"target_audience": r[8],
|
||
"status": r[9],
|
||
"created_at": r[10].isoformat() if r[10] else None,
|
||
"deal_count": r[11],
|
||
"total_amount": float(r[12]),
|
||
}
|
||
for r in cur.fetchall()
|
||
]
|
||
finally:
|
||
put_conn(conn)
|
||
|
||
|
||
@app.post("/api/products")
|
||
def create_product(product: ProductCreate):
|
||
"""新增产品。"""
|
||
conn = get_conn()
|
||
try:
|
||
with conn.cursor() as cur:
|
||
cur.execute("""
|
||
INSERT INTO product (name, category, spec, price, description, positioning, selling_points, target_audience, status)
|
||
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)
|
||
RETURNING id
|
||
""", (
|
||
product.name, product.category, product.spec,
|
||
product.price, product.description, product.positioning,
|
||
product.selling_points, product.target_audience,
|
||
product.status,
|
||
))
|
||
product_id = cur.fetchone()[0]
|
||
conn.commit()
|
||
return {"id": product_id, "message": "产品已创建"}
|
||
except Exception:
|
||
conn.rollback()
|
||
raise
|
||
finally:
|
||
put_conn(conn)
|
||
|
||
|
||
@app.put("/api/products/{product_id}")
|
||
def update_product(product_id: int, product: ProductUpdate):
|
||
"""编辑产品信息。"""
|
||
conn = get_conn()
|
||
try:
|
||
with conn.cursor() as cur:
|
||
fields = []
|
||
params = []
|
||
for field in ["name", "category", "spec", "price", "description", "positioning", "selling_points", "target_audience", "status"]:
|
||
val = getattr(product, field)
|
||
if val is not None:
|
||
fields.append(f"{field} = %s")
|
||
params.append(val)
|
||
if not fields:
|
||
raise HTTPException(status_code=400, detail="没有需要更新的字段")
|
||
params.append(product_id)
|
||
cur.execute(
|
||
f"UPDATE product SET {', '.join(fields)} WHERE id = %s RETURNING id",
|
||
params,
|
||
)
|
||
row = cur.fetchone()
|
||
if not row:
|
||
raise HTTPException(status_code=404, detail="产品不存在")
|
||
conn.commit()
|
||
return {"id": product_id, "message": "产品已更新"}
|
||
except Exception:
|
||
conn.rollback()
|
||
raise
|
||
finally:
|
||
put_conn(conn)
|
||
|
||
|
||
@app.delete("/api/products/{product_id}")
|
||
def delete_product(product_id: int):
|
||
"""删除产品（软删除，设为 inactive）。"""
|
||
conn = get_conn()
|
||
try:
|
||
with conn.cursor() as cur:
|
||
cur.execute(
|
||
"UPDATE product SET status = 'inactive' WHERE id = %s RETURNING id",
|
||
(product_id,),
|
||
)
|
||
row = cur.fetchone()
|
||
if not row:
|
||
raise HTTPException(status_code=404, detail="产品不存在")
|
||
conn.commit()
|
||
return {"id": product_id, "message": "产品已下架"}
|
||
except Exception:
|
||
conn.rollback()
|
||
raise
|
||
finally:
|
||
put_conn(conn)
|
||
|
||
|
||
@app.get("/api/products/categories")
|
||
def list_product_categories():
|
||
"""获取所有产品线（去重）。"""
|
||
conn = get_conn()
|
||
try:
|
||
with conn.cursor() as cur:
|
||
cur.execute(
|
||
"SELECT DISTINCT category FROM product WHERE status = 'active' ORDER BY category"
|
||
)
|
||
return [r[0] 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,
|
||
string_agg(DISTINCT p.category, ', ') AS deal_categories,
|
||
string_agg(m2.normalized_content, ' ') AS all_messages
|
||
FROM customer cu
|
||
JOIN salesperson s ON s.id = cu.salesperson_id
|
||
LEFT JOIN deal d ON d.customer_id = cu.id
|
||
LEFT JOIN product p ON p.id = d.product_id
|
||
LEFT JOIN conversation conv2 ON conv2.contact_id = cu.contact_id
|
||
LEFT JOIN message m2 ON m2.conversation_id = conv2.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 += """
|
||
GROUP BY cu.id, cu.customer_name, s.name,
|
||
cu.industry, cu.intent_level, cu.stage,
|
||
cu.summary, cu.key_points, cu.objections, cu.next_action,
|
||
cu.key_needs, cu.last_analysis, cu.salesperson_id
|
||
ORDER BY cu.salesperson_id, cu.id
|
||
"""
|
||
|
||
cur.execute(sql, params)
|
||
diet_keywords = ["减肥", "减脂", "瘦身", "B420", "纤益", "代餐", "体脂", "产后减肥", "节食"]
|
||
child_keywords = ["湿疹", "过敏", "鼻炎", "腹泻", "便秘", "免疫力", "胀气",
|
||
"拉肚子", "打喷嚏", "流鼻涕", "感冒", "食物过敏", "牛奶蛋白",
|
||
"宝宝", "宝妈", "儿童", "丹麦", "鼠李糖", "LGG", "合生元"]
|
||
results = []
|
||
for r in cur.fetchall():
|
||
deal_cats = r[13]
|
||
all_msgs = r[14] or ""
|
||
if deal_cats:
|
||
product_categories = deal_cats
|
||
elif any(kw in all_msgs for kw in diet_keywords):
|
||
product_categories = "女性减肥益生菌"
|
||
elif any(kw in all_msgs for kw in child_keywords):
|
||
product_categories = "儿童益生菌"
|
||
else:
|
||
product_categories = None
|
||
results.append({
|
||
"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],
|
||
"product_categories": product_categories,
|
||
})
|
||
return results
|
||
finally:
|
||
put_conn(conn)
|
||
|
||
|
||
@app.post("/api/conversations/{customer_id}/paradigm")
|
||
def generate_paradigm(
|
||
customer_id: int,
|
||
product_id: int | None = Query(None),
|
||
):
|
||
"""调用 LLM 生成标准范式对话，基于产品详情动态构建 prompt。"""
|
||
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="客户不存在")
|
||
|
||
# 如果未指定 product_id，尝试从客户成交记录中推断
|
||
if product_id is None:
|
||
cur.execute("""
|
||
SELECT d.product_id FROM deal d
|
||
WHERE d.customer_id = %s AND d.product_id IS NOT NULL
|
||
ORDER BY d.deal_date DESC LIMIT 1
|
||
""", (customer_id,))
|
||
deal_row = cur.fetchone()
|
||
if deal_row:
|
||
product_id = deal_row[0]
|
||
|
||
# 从产品表获取产品详情；如果仍无 product_id，取所有 active 产品汇总
|
||
product_info_text = ""
|
||
if product_id is not None:
|
||
cur.execute("""
|
||
SELECT name, spec, price, description, positioning,
|
||
selling_points, target_audience, category
|
||
FROM product WHERE id = %s
|
||
""", (product_id,))
|
||
prow = cur.fetchone()
|
||
if prow:
|
||
sp_list = parse_pg_array(prow[4])
|
||
product_info_text = (
|
||
f"### 主推产品：{prow[0]}（{prow[1]}）\n"
|
||
f"- 产品线：{prow[7]}\n"
|
||
f"- 价格：¥{float(prow[2]):.2f}\n"
|
||
f"- 定位：{prow[4] or '未设定'}\n"
|
||
f"- 目标人群：{prow[6] or '未设定'}\n"
|
||
f"- 产品描述：{prow[3] or '未设定'}\n"
|
||
f"- 核心卖点：\n" + "\n".join(f" {i+1}. {sp}" for i, sp in enumerate(sp_list))
|
||
)
|
||
else:
|
||
cur.execute("""
|
||
SELECT name, spec, price, positioning, target_audience, category
|
||
FROM product WHERE status = 'active' ORDER BY category, id
|
||
""")
|
||
products = cur.fetchall()
|
||
if products:
|
||
lines = []
|
||
for p in products:
|
||
lines.append(
|
||
f"- {p[0]}（{p[1]}）¥{float(p[2]):.2f} | {p[5]} | 定位：{p[3] or '未设定'} | 目标：{p[4] or '未设定'}"
|
||
)
|
||
product_info_text = "### 全部在售产品\n" + "\n".join(lines)
|
||
else:
|
||
product_info_text = "暂无产品数据"
|
||
|
||
# 获取对话记录
|
||
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"""你是一个微信销售对话分析专家。请基于以下产品详情和真实销售对话，生成一套标准范式对话模板。
|
||
|
||
## 产品背景
|
||
{product_info_text}
|
||
|
||
## 客户信息
|
||
- 客户: {customer_name}
|
||
- 行业: {industry}
|
||
- 销售人员: {sp_name}
|
||
- 当前阶段: {stage}
|
||
- 关键需求: {key_needs_str}
|
||
- 异议: {objections_str}
|
||
- 摘要: {summary or '无'}
|
||
|
||
## 真实对话记录
|
||
{dialog_text}
|
||
|
||
## 输出要求
|
||
请基于上述产品详情（定位、卖点、目标人群）和真实对话，提炼并生成一套**标准范式对话模板**，即针对此类客户的最优销售话术流程。话术要自然融入产品的核心卖点和定位，不要生硬推销。
|
||
|
||
请用 Markdown 格式输出，结构如下：
|
||
|
||
## 客户类型
|
||
（描述客户类型，如"湿疹宝妈-价格敏感型"）
|
||
|
||
## 产品推荐理由
|
||
（结合产品定位和客户需求，1-2句话说明为什么推荐这个产品）
|
||
|
||
## 标准销售流程
|
||
|
||
### 1. 建立联系
|
||
- **目标**：xxx
|
||
- **销售话术**：xxx
|
||
- **预期回应**：xxx
|
||
- **技巧提示**：xxx
|
||
|
||
### 2. 需求发现
|
||
（同上格式）
|
||
|
||
### 3. 报价
|
||
（同上格式）
|
||
|
||
### 4. 异议处理
|
||
（同上格式）
|
||
|
||
### 5. 成交
|
||
（同上格式）
|
||
|
||
## 核心销售技巧
|
||
- 技巧1
|
||
- 技巧2
|
||
- ...
|
||
|
||
## 真实对话可改进的点
|
||
- 改进点1
|
||
- 改进点2
|
||
- ...
|
||
|
||
直接输出 Markdown 内容，不要输出 JSON。"""
|
||
|
||
body = json.dumps({
|
||
"model": "qwen-plus",
|
||
"messages": [
|
||
{"role": "system", "content": "你是微信销售对话分析专家，擅长从真实对话中提炼标准范式。"},
|
||
{"role": "user", "content": prompt},
|
||
],
|
||
"temperature": 0.4,
|
||
"stream": True,
|
||
"stream_options": {"include_usage": False},
|
||
}).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",
|
||
)
|
||
|
||
def stream_generator():
|
||
"""SSE 流式生成器，逐块推送 LLM content delta。"""
|
||
try:
|
||
resp = _urllib.urlopen(req, timeout=120)
|
||
buffer = b""
|
||
while True:
|
||
chunk = resp.read(4096)
|
||
if not chunk:
|
||
break
|
||
buffer += chunk
|
||
while b"\n" in buffer:
|
||
line, buffer = buffer.split(b"\n", 1)
|
||
line = line.strip()
|
||
if not line:
|
||
continue
|
||
if line.startswith(b"data:"):
|
||
data_str = line[5:].strip()
|
||
if data_str == b"[DONE]":
|
||
yield "data: [DONE]\n\n"
|
||
return
|
||
try:
|
||
chunk_obj = json.loads(data_str)
|
||
delta = chunk_obj.get("choices", [{}])[0].get("delta", {})
|
||
content = delta.get("content", "")
|
||
if content:
|
||
yield f"data: {json.dumps({'content': content}, ensure_ascii=False)}\n\n"
|
||
except (json.JSONDecodeError, IndexError, KeyError):
|
||
continue
|
||
resp.close()
|
||
except Exception as exc:
|
||
yield f"data: {json.dumps({'error': str(exc)}, ensure_ascii=False)}\n\n"
|
||
yield "data: [DONE]\n\n"
|
||
|
||
return StreamingResponse(stream_generator(), media_type="text/event-stream")
|
||
|
||
|
||
@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")
|