TradeMate Dev
7b62c2f8b4
## H5 底部导航修复 (Bug #10) - 精简 App.vue，移除重复 tabbar，仅保留全局样式 - uni-page 设置 height: calc(100% - 50px) + overflow-y: auto - 内容区域精确停在底部导航上方，独立滚动不再叠加 - 恢复 custom-tab-bar 组件 ## 项目进度文档 - PROGRESS.md 更新至 10 个 Bug 修复 - 新增 H5 底部导航修复记录 - 新增历史变更条目
334 lines
13 KiB
Python
334 lines
13 KiB
Python
from typing import Dict, Any, Optional, List
|
|
from datetime import datetime, timedelta
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy import select, func, and_, desc
|
|
from app.models.customer import Customer, Message, Conversation
|
|
from app.models.quotation import Quotation
|
|
import logging
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
DEAL_SIGNAL_KEYWORDS = [
|
|
"moq", "minimum order", "sample", "certification", "certificate",
|
|
"fob", "cif", "lead time", "delivery time", "shipping",
|
|
"payment term", "tt", "lc", "deposit", "price", "quotation",
|
|
"order", "purchase", "buy", "interested", "inquiry", "rfq",
|
|
]
|
|
|
|
POSITIVE_WORDS = ["yes", "interested", "good", "great", "perfect", "thanks", "thank you", "proceed", "confirm", "agree"]
|
|
NEGATIVE_WORDS = ["no", "not interested", "too expensive", "high price", "over budget", "not now", "later", "maybe later"]
|
|
|
|
|
|
class CustomerHealthService:
|
|
def __init__(self, db: AsyncSession):
|
|
self.db = db
|
|
|
|
async def get_health_overview(self, user_id: str) -> Dict[str, Any]:
|
|
customers_result = await self.db.execute(
|
|
select(Customer.id, Customer.status, Customer.last_contact_at).where(
|
|
Customer.user_id == user_id
|
|
)
|
|
)
|
|
rows = customers_result.all()
|
|
total = len(rows)
|
|
active = 0
|
|
watch = 0
|
|
critical = 0
|
|
for row in rows:
|
|
score = self._calculate_silence_score(row.last_contact_at)
|
|
status_weight = self._status_weight(row.status)
|
|
combined = score * 0.7 + status_weight * 0.3
|
|
if combined >= 70:
|
|
active += 1
|
|
elif combined >= 40:
|
|
watch += 1
|
|
else:
|
|
critical += 1
|
|
return {
|
|
"total": total,
|
|
"active": active,
|
|
"watch": watch,
|
|
"critical": critical,
|
|
}
|
|
|
|
async def get_customer_health(self, user_id: str, customer_id: str) -> Optional[Dict[str, Any]]:
|
|
result = await self.db.execute(
|
|
select(Customer).where(
|
|
and_(Customer.id == customer_id, Customer.user_id == user_id)
|
|
)
|
|
)
|
|
customer = result.scalar_one_or_none()
|
|
if not customer:
|
|
return None
|
|
return await self._compute_full_health(user_id, customer)
|
|
|
|
async def get_all_health_scores(self, user_id: str) -> List[Dict[str, Any]]:
|
|
customers_result = await self.db.execute(
|
|
select(Customer).where(Customer.user_id == user_id).order_by(Customer.updated_at.desc())
|
|
)
|
|
customers = customers_result.scalars().all()
|
|
results = []
|
|
for c in customers:
|
|
health = await self._compute_full_health(user_id, c)
|
|
results.append(health)
|
|
return results
|
|
|
|
async def _compute_full_health(self, user_id: str, customer: Customer) -> Dict[str, Any]:
|
|
response_trend = await self._calc_response_trend(customer.id)
|
|
sentiment = await self._calc_sentiment(customer.id)
|
|
inquiry_depth = await self._calc_inquiry_depth(customer.id)
|
|
silence_score = self._calculate_silence_score(customer.last_contact_at)
|
|
business_value = await self._calc_business_value(customer.id)
|
|
|
|
silence_days = self._silence_days(customer.last_contact_at)
|
|
dimensions = {
|
|
"response_trend": response_trend,
|
|
"sentiment": sentiment,
|
|
"inquiry_depth": inquiry_depth,
|
|
"silence": {"score": silence_score, "days": silence_days},
|
|
"business_value": business_value,
|
|
}
|
|
result = self.calc_total_score(dimensions)
|
|
|
|
return {
|
|
"customer_id": str(customer.id),
|
|
"customer_name": customer.name,
|
|
"status": customer.status,
|
|
"total_score": result["total_score"],
|
|
"grade": result["grade"],
|
|
"dimensions": dimensions,
|
|
"suggestion": self._suggestion(result["grade"], customer),
|
|
}
|
|
|
|
async def _calc_response_trend(self, customer_id: str) -> Dict[str, Any]:
|
|
now_7d_ago = datetime.utcnow() - timedelta(days=7)
|
|
prev_7d_ago = datetime.utcnow() - timedelta(days=14)
|
|
|
|
recent_result = await self.db.execute(
|
|
select(func.avg(
|
|
func.extract("epoch", Message.created_at) -
|
|
func.extract("epoch", func.lag(Message.created_at).over(order_by=Message.created_at))
|
|
)).where(
|
|
and_(
|
|
Message.conversation_id == select(Conversation.id).where(
|
|
Conversation.customer_id == customer_id
|
|
).limit(1).scalar_subquery(),
|
|
Message.direction == "inbound",
|
|
Message.created_at >= now_7d_ago,
|
|
)
|
|
)
|
|
)
|
|
|
|
previous_result = await self.db.execute(
|
|
select(func.avg(
|
|
func.extract("epoch", Message.created_at) -
|
|
func.extract("epoch", func.lag(Message.created_at).over(order_by=Message.created_at))
|
|
)).where(
|
|
and_(
|
|
Message.conversation_id == select(Conversation.id).where(
|
|
Conversation.customer_id == customer_id
|
|
).limit(1).scalar_subquery(),
|
|
Message.direction == "inbound",
|
|
Message.created_at >= prev_7d_ago,
|
|
Message.created_at < now_7d_ago,
|
|
)
|
|
)
|
|
)
|
|
|
|
recent_avg = recent_result.scalar()
|
|
prev_avg = previous_result.scalar()
|
|
|
|
recent_hours = (recent_avg / 3600) if recent_avg else None
|
|
prev_hours = (prev_avg / 3600) if prev_avg else None
|
|
return self.calc_response_score(recent_hours, prev_hours)
|
|
|
|
async def _calc_sentiment(self, customer_id: str) -> Dict[str, Any]:
|
|
conv_result = await self.db.execute(
|
|
select(Conversation.id).where(
|
|
Conversation.customer_id == customer_id
|
|
).order_by(Conversation.created_at.desc()).limit(1)
|
|
)
|
|
conv_id = conv_result.scalar_one_or_none()
|
|
if not conv_id:
|
|
return {"score": 50, "label": "neutral", "last_messages": []}
|
|
|
|
msg_result = await self.db.execute(
|
|
select(Message.content).where(
|
|
and_(
|
|
Message.conversation_id == conv_id,
|
|
Message.direction == "inbound",
|
|
)
|
|
).order_by(desc(Message.created_at)).limit(3)
|
|
)
|
|
messages = list(msg_result.scalars().all())
|
|
return self.calc_sentiment_score(messages)
|
|
|
|
async def _calc_inquiry_depth(self, customer_id: str) -> Dict[str, Any]:
|
|
conv_result = await self.db.execute(
|
|
select(Conversation.id).where(
|
|
Conversation.customer_id == customer_id
|
|
).order_by(Conversation.created_at.desc()).limit(1)
|
|
)
|
|
conv_id = conv_result.scalar_one_or_none()
|
|
if not conv_id:
|
|
return {"score": 0, "signals_found": [], "signal_count": 0}
|
|
|
|
msg_result = await self.db.execute(
|
|
select(Message.content).where(
|
|
and_(
|
|
Message.conversation_id == conv_id,
|
|
Message.direction == "inbound",
|
|
)
|
|
).order_by(desc(Message.created_at)).limit(20)
|
|
)
|
|
messages = list(msg_result.scalars().all())
|
|
return self.calc_inquiry_depth_score(messages)
|
|
|
|
@staticmethod
|
|
def calculate_silence_score(last_contact_at: Optional[datetime]) -> float:
|
|
days = CustomerHealthService.silence_days(last_contact_at)
|
|
return max(0, min(100, 100 - (days / 14) * 100))
|
|
|
|
@staticmethod
|
|
def silence_days(last_contact_at: Optional[datetime]) -> int:
|
|
if not last_contact_at:
|
|
return 999
|
|
return (datetime.utcnow() - last_contact_at).days
|
|
|
|
@staticmethod
|
|
def status_weight(status: Optional[str]) -> float:
|
|
mapping = {"customer": 100, "negotiating": 70, "lead": 40, "lost": 10}
|
|
return mapping.get(status, 40)
|
|
|
|
@staticmethod
|
|
def grade(score: float) -> str:
|
|
if score >= 80:
|
|
return "active"
|
|
elif score >= 50:
|
|
return "watch"
|
|
else:
|
|
return "critical"
|
|
|
|
@staticmethod
|
|
def calc_response_score(recent_hours: Optional[float], prev_hours: Optional[float]) -> Dict[str, Any]:
|
|
if recent_hours is None and prev_hours is None:
|
|
return {"score": 50, "recent_avg_hours": None, "trend": "stable"}
|
|
if recent_hours is None:
|
|
return {"score": 30, "recent_avg_hours": None, "trend": "declining"}
|
|
if prev_hours is None or prev_hours == 0:
|
|
score = max(0, min(100, 100 - recent_hours * 5))
|
|
return {"score": round(score), "recent_avg_hours": round(recent_hours, 1), "trend": "stable"}
|
|
if recent_hours < prev_hours:
|
|
score = max(0, min(100, 100 - recent_hours * 5))
|
|
return {"score": round(score), "recent_avg_hours": round(recent_hours, 1), "trend": "improving"}
|
|
else:
|
|
score = max(0, min(100, 80 - recent_hours * 3))
|
|
return {"score": round(score), "recent_avg_hours": round(recent_hours, 1), "trend": "declining"}
|
|
|
|
@staticmethod
|
|
def calc_sentiment_score(messages: List[str]) -> Dict[str, Any]:
|
|
if not messages:
|
|
return {"score": 50, "label": "neutral", "last_messages": []}
|
|
positive = 0
|
|
negative = 0
|
|
for msg in messages:
|
|
lower = msg.lower()
|
|
if any(w in lower for w in POSITIVE_WORDS):
|
|
positive += 1
|
|
if any(w in lower for w in NEGATIVE_WORDS):
|
|
negative += 1
|
|
if positive > negative:
|
|
return {"score": 80, "label": "positive", "last_messages": messages}
|
|
elif negative > positive:
|
|
return {"score": 20, "label": "negative", "last_messages": messages}
|
|
else:
|
|
return {"score": 50, "label": "neutral", "last_messages": messages}
|
|
|
|
@staticmethod
|
|
def calc_inquiry_depth_score(messages: List[str]) -> Dict[str, Any]:
|
|
found_signals = []
|
|
for msg in messages:
|
|
lower = msg.lower()
|
|
for kw in DEAL_SIGNAL_KEYWORDS:
|
|
if kw in lower and kw not in found_signals:
|
|
found_signals.append(kw)
|
|
count = len(found_signals)
|
|
if count >= 5:
|
|
score = 100
|
|
elif count >= 3:
|
|
score = 75
|
|
elif count >= 1:
|
|
score = 50
|
|
else:
|
|
score = 0
|
|
return {"score": score, "signals_found": found_signals, "signal_count": count}
|
|
|
|
@staticmethod
|
|
def calc_business_value_score(total_value: float) -> Dict[str, Any]:
|
|
if total_value >= 100000:
|
|
score = 100
|
|
elif total_value >= 50000:
|
|
score = 80
|
|
elif total_value >= 10000:
|
|
score = 60
|
|
elif total_value >= 1000:
|
|
score = 40
|
|
elif total_value > 0:
|
|
score = 20
|
|
else:
|
|
score = 0
|
|
return {"score": score, "total_value": round(total_value, 2)}
|
|
|
|
@staticmethod
|
|
def calc_total_score(dimensions: Dict[str, Any]) -> Dict[str, Any]:
|
|
total = (
|
|
dimensions.get("response_trend", {}).get("score", 0) * 0.25
|
|
+ dimensions.get("sentiment", {}).get("score", 0) * 0.20
|
|
+ dimensions.get("inquiry_depth", {}).get("score", 0) * 0.20
|
|
+ dimensions.get("silence", {}).get("score", 0) * 0.20
|
|
+ dimensions.get("business_value", {}).get("score", 0) * 0.15
|
|
)
|
|
return {"total_score": round(total, 1), "grade": CustomerHealthService.grade(total)}
|
|
|
|
@staticmethod
|
|
def suggestion(grade: str, silence_days: int, status: Optional[str]) -> str:
|
|
if grade == "active":
|
|
return "保持正常跟进，客户状态良好"
|
|
elif grade == "watch":
|
|
if silence_days >= 3:
|
|
return f"客户已沉默{silence_days}天，建议3天内安排跟进"
|
|
return "客户活跃度下降，建议关注"
|
|
else:
|
|
if status in ("lead", "negotiating"):
|
|
return f"客户已沉默{silence_days}天，建议立即跟进，提供优惠或新产品信息"
|
|
return f"客户已沉默{silence_days}天，建议重新激活"
|
|
|
|
def _calculate_silence_score(self, last_contact_at: Optional[datetime]) -> float:
|
|
return self.calculate_silence_score(last_contact_at)
|
|
|
|
def _silence_days(self, last_contact_at: Optional[datetime]) -> int:
|
|
return self.silence_days(last_contact_at)
|
|
|
|
def _status_weight(self, status: Optional[str]) -> float:
|
|
return self.status_weight(status)
|
|
|
|
def _grade(self, score: float) -> str:
|
|
return self.grade(score)
|
|
|
|
def _suggestion(self, grade: str, customer: Customer) -> str:
|
|
return self.suggestion(grade, self._silence_days(customer.last_contact_at), customer.status)
|
|
|
|
async def _calc_business_value(self, customer_id: str) -> Dict[str, Any]:
|
|
result = await self.db.execute(
|
|
select(func.sum(Quotation.total)).where(
|
|
and_(
|
|
Quotation.customer_id == customer_id,
|
|
Quotation.status.in_(["sent", "accepted"]),
|
|
)
|
|
)
|
|
)
|
|
total_value = result.scalar() or 0
|
|
return self.calc_business_value_score(total_value)
|
|
|
|
|