v6ole/qiji
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 客户亮灯表 + AI 周报摘要 — v0.2

Browse Source 
亮灯表:
- 四色覆盖矩阵: 绿亮灯/黄临期/红灭灯/灰未分配
- 按客户经理折叠卡片流, 覆盖率进度条, 团队总览统计
- 红灯客户显示连续未拜访月份, 未分配客户专区
- 后端: light_board.py service + GET /api/dashboard/light-board
- 前端: LightBoard.vue + 路由 /light-board + 汇总侧边栏

AI 周报摘要:
- 接入 OpenAI 兼容大模型, 注入拜访数据+亮灯表覆盖数据
- 四段式结构化输出: 概况/需求/覆盖分析/建议
- 一键生成+Markdown渲染+复制纯文本
- 支局长/分管领导专用, 支持配置内部模型
- 后端: ai_summary.py service + POST /api/ai/summary
- 前端: WeeklyReport 集成按钮+结果面板
- 新增配置: AI_API_URL / AI_API_KEY / AI_MODEL

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
v6ole
2026-06-24 12:03:00 +08:00
parent b79a5296c4
commit 305ee5df09
13 changed files with 1100 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/services/ai_summary.py
+190
View File
@@ -0,0 +1,190 @@
"""AI-powered weekly report summary using an OpenAI-compatible LLM."""
import json
import httpx
from app.config import settings
from app.services.dashboard import get_weekly_report, get_week_range
from app.services.light_board import get_light_board
from sqlalchemy.ext.asyncio import AsyncSession
from uuid import UUID
from datetime import date
SUMMARY_SYSTEM_PROMPT = """你是一位经验丰富的政企客户经理团队的周报分析助手。你的分析将被直接提交给支局长作为工作周报的文字摘要。
## 约束
- 严格基于提供的数据进行分析，绝不编造数据中不存在的信息
- 使用正式但不生硬的中文，适合放入政企工作周报
- 每条分析简洁有力，1-2句话即可，避免空泛套话
- 如果某个结论是基于数据推断的，请使用"数据显示""从本周情况看"等表述
- 对于覆盖不足的情况，请明确指出具体客户名称和负责人，方便支局长跟进
## 输出格式（使用 Markdown）
### 一、本周概况
[2-3句话，涵盖：拜访总量、覆盖客户数、团队参与情况、拜访方式分布]
### 二、拜访重点与客户需求
[2-3个值得关注的客户需求或沟通内容要点，有具体客户名称]
### 三、客户覆盖分析
[引用覆盖数据，明确指出：覆盖率、低于60%的经理、红灯客户名单、需要关注的客户]
### 四、下周建议
[2-3条针对性的工作建议，基于数据中暴露的问题和客户需求]"""
def build_summary_prompt(
weekly_report: dict,
light_board: dict,
period: str,
reference_date: str,
) -> str:
"""Build the user prompt with structured visit data for the LLM."""
# ── Summary stats ──
visits = weekly_report.get("visits", [])
daily_notes = weekly_report.get("daily_notes", [])
managers_involved: set[str] = set()
customers_visited: set[str] = set()
methods: dict[str, int] = {}
demands: list[str] = []
for v in visits:
managers_involved.add(v.get("manager_name", ""))
customers_visited.add(v.get("customer_name", ""))
method = v.get("visit_method", "")
methods[method] = methods.get(method, 0) + 1
demand = v.get("customer_demand", "")
if demand and demand.strip():
demands.append(f"{v.get('customer_name', '未知')}: {demand.strip()}")
# Manager breakdown
manager_visits: dict[str, list] = {}
for v in visits:
mn = v.get("manager_name", "未知")
if mn not in manager_visits:
manager_visits[mn] = []
manager_visits[mn].append({
"client": v.get("customer_name", ""),
"method": v.get("visit_method", ""),
"content": (v.get("communication_content", "") or "")[:120],
"demand": v.get("customer_demand", "") or "",
})
# Build the data block
data_block = f"""## 基本信息
- 分析周期：{period}
- 参考日期：{reference_date}
- 周范围：{weekly_report.get('week_start', '')} — {weekly_report.get('week_end', '')}
## 拜访总览
- 拜访记录总数：{len(visits)}
- 覆盖客户数：{len(customers_visited)}
- 参与经理数：{len(managers_involved)}
- 拜访方式分布：{json.dumps(methods, ensure_ascii=False)}
## 各客户经理拜访明细
"""
for mn, items in manager_visits.items():
data_block += f"\n### {mn}（{len(items)}条）\n"
for item in items[:10]: # cap per manager
data_block += f"- {item['method']}拜访 {item['client']}"
if item['content']:
data_block += f" — {item['content'][:100]}"
if item['demand']:
data_block += f" [需求: {item['demand'][:80]}]"
data_block += "\n"
# Customer demands
if demands:
data_block += "\n## 客户需求汇总\n"
for d in demands[:15]:
data_block += f"- {d[:200]}\n"
# Daily notes summary
notes_by_cat: dict[str, int] = {}
for n in daily_notes:
cat = n.get("category", "其他")
notes_by_cat[cat] = notes_by_cat.get(cat, 0) + 1
if notes_by_cat:
data_block += "\n## 纪要分类统计\n"
data_block += json.dumps(notes_by_cat, ensure_ascii=False) + "\n"
# Light board data
team = light_board.get("team_summary", {})
data_block += f"""
## 客户覆盖数据（亮灯表）
- 团队总客户数：{team.get('total_customers', 0)}
- 本月已拜访（绿灯）：{team.get('visited_this_month', 0)}
- 仅上月拜访（黄灯）：{team.get('visited_last_month_only', 0)}
- 连续未拜访（红灯）：{team.get('not_visited_2months', 0)}
- 未分配客户：{team.get('unassigned', 0)}
- 整体覆盖率：{team.get('coverage_rate', 0) * 100:.1f}%
### 各经理覆盖率
"""
for m in light_board.get("managers", []):
data_block += (
f"- {m['manager_name']}: {m['coverage_rate'] * 100:.0f}% "
f"({m['visited_this_month']}/{m['total_customers']}) "
f"🟢{m['visited_this_month']} 🟡{m['visited_last_month_only']} 🔴{m['not_visited_2months']}\n"
)
# List red customers
red_customers = [c for c in m.get("customers", []) if c["status"] == "red"]
if red_customers:
data_block += " 红灯客户：\n"
for rc in red_customers[:5]:
lvd = rc.get("last_visit_date") or "从未"
data_block += f" - {rc['customer_name']}（上次拜访: {lvd}）\n"
return data_block
async def generate_summary(
db: AsyncSession,
user_id: UUID,
role: str,
reference_date: date | None = None,
period: str = "week",
) -> str:
"""Generate an AI-powered weekly summary.
Raises ValueError if AI config is missing, httpx.HTTPError on API failure.
"""
if not settings.AI_API_URL:
raise ValueError("AI_API_URL not configured")
# Gather data
ref = reference_date or date.today()
weekly_report = await get_weekly_report(
db=db, user_id=user_id, role=role, reference_date=ref,
)
light_board = await get_light_board(db, ref)
user_prompt = build_summary_prompt(weekly_report, light_board, period, str(ref))
# Call LLM
headers = {"Content-Type": "application/json"}
if settings.AI_API_KEY:
headers["Authorization"] = f"Bearer {settings.AI_API_KEY}"
payload = {
"model": settings.AI_MODEL,
"messages": [
{"role": "system", "content": SUMMARY_SYSTEM_PROMPT},
{"role": "user", "content": user_prompt},
],
"max_tokens": settings.AI_MAX_TOKENS,
"temperature": 0.3,
}
async with httpx.AsyncClient(timeout=90.0) as client:
resp = await client.post(settings.AI_API_URL, json=payload, headers=headers)
resp.raise_for_status()
result = resp.json()
content = result.get("choices", [{}])[0].get("message", {}).get("content", "")
if not content:
raise ValueError("AI returned empty response")
return content
backend/app/services/light_board.py
+198
View File
@@ -0,0 +1,198 @@
"""Customer light board — visit coverage matrix per manager.
Status:
green — visited this month (亮灯)
yellow — visited last month but not this month (临期)
red — not visited in 2+ months, or never visited (灭灯+警示)
gray — customer has no assigned primary manager (未分配)
"""
from datetime import date, timedelta
from uuid import UUID
from sqlalchemy import select, func, and_
from sqlalchemy.ext.asyncio import AsyncSession
from app.models.user import User
from app.models.customer import Customer
from app.models.customer_assignment import CustomerAssignment
from app.models.visit import Visit
from app.utils.timezone import today_cst
def month_start(ref: date) -> date:
return ref.replace(day=1)
def month_end(ref: date) -> date:
nxt = ref.replace(day=28) + timedelta(days=4)
return nxt - timedelta(days=nxt.day)
def classify(last_visit_date: date | None, ref: date) -> tuple[str, int]:
"""Return (status, consecutive_missed_months)."""
if last_visit_date is None:
return ("red", 99) # never visited
this_month = month_start(ref)
last_month = month_start(this_month - timedelta(days=1))
if last_visit_date >= this_month:
return ("green", 0)
elif last_visit_date >= last_month:
return ("yellow", 0)
# Count how many consecutive months missed
cursor = month_start(ref)
missed = 0
while True:
cursor = month_start(cursor - timedelta(days=1))
if last_visit_date >= cursor:
break
missed += 1
if missed > 24: # safety cap
break
return ("red", missed)
async def get_light_board(db: AsyncSession, reference_date: date | None = None) -> dict:
"""Get per-manager customer visit coverage for the light board."""
ref = reference_date or today_cst()
# ── All managers ──
managers_result = await db.execute(select(User).where(User.role == "manager"))
managers = managers_result.scalars().all()
manager_map: dict[UUID, dict] = {
m.id: {
"manager_id": str(m.id),
"manager_name": m.name,
"total_customers": 0,
"visited_this_month": 0,
"visited_last_month_only": 0,
"not_visited_2months": 0,
"coverage_rate": 0.0,
"customers": [],
}
for m in managers
}
# ── All customers with their last visit per manager ──
# Subquery: latest visit date + method per (manager_id, customer_id)
latest_visit = (
select(
Visit.manager_id,
Visit.customer_id,
func.max(Visit.visit_date).label("last_visit_date"),
)
.group_by(Visit.manager_id, Visit.customer_id)
.subquery()
)
# Join: assignments → customers → latest_visit → visits (for method)
rows = await db.execute(
select(
CustomerAssignment.manager_id,
CustomerAssignment.customer_id,
Customer.id.label("c_id"),
Customer.name.label("c_name"),
Customer.industry,
Customer.in_use_services,
Customer.monthly_fee,
latest_visit.c.last_visit_date,
Visit.visit_method,
)
.join(Customer, Customer.id == CustomerAssignment.customer_id)
.outerjoin(latest_visit, and_(
latest_visit.c.manager_id == CustomerAssignment.manager_id,
latest_visit.c.customer_id == CustomerAssignment.customer_id,
))
.outerjoin(Visit, and_(
Visit.manager_id == CustomerAssignment.manager_id,
Visit.customer_id == CustomerAssignment.customer_id,
Visit.visit_date == latest_visit.c.last_visit_date,
))
.where(CustomerAssignment.role == "primary")
.order_by(Customer.name)
)
assigned_customer_ids: set[UUID] = set()
for row in rows.all():
mgr_id, cust_id, c_id, c_name, industry, services, fee, lvd, method = row
assigned_customer_ids.add(c_id)
status, missed = classify(lvd, ref)
cust_entry = {
"customer_id": str(c_id),
"customer_name": c_name,
"industry": industry or "",
"in_use_services": services or "",
"monthly_fee": str(fee) if fee else "",
"last_visit_date": str(lvd) if lvd else None,
"last_visit_method": method or "",
"status": status,
"consecutive_missed_months": missed,
}
mgr_entry = manager_map.get(mgr_id)
if mgr_entry:
mgr_entry["customers"].append(cust_entry)
mgr_entry["total_customers"] += 1
if status == "green":
mgr_entry["visited_this_month"] += 1
elif status == "yellow":
mgr_entry["visited_last_month_only"] += 1
else:
mgr_entry["not_visited_2months"] += 1
# ── Calculate coverage rates ──
for mgr_entry in manager_map.values():
total = mgr_entry["total_customers"]
if total > 0:
mgr_entry["coverage_rate"] = round(mgr_entry["visited_this_month"] / total, 3)
# Sort managers: lowest coverage first (most problematic first)
manager_list = sorted(manager_map.values(), key=lambda m: m["coverage_rate"])
# ── Unassigned customers (no primary manager) ──
unassigned_rows = await db.execute(
select(Customer)
.outerjoin(CustomerAssignment, and_(
CustomerAssignment.customer_id == Customer.id,
CustomerAssignment.role == "primary",
))
.where(CustomerAssignment.id == None)
.order_by(Customer.name)
)
unassigned = []
for c in unassigned_rows.scalars():
unassigned.append({
"customer_id": str(c.id),
"customer_name": c.name,
"industry": c.industry or "",
"in_use_services": c.in_use_services or "",
"monthly_fee": str(c.monthly_fee) if c.monthly_fee else "",
"last_visit_date": None,
"last_visit_method": "",
"status": "gray",
"consecutive_missed_months": 0,
})
# ── Team summary ──
all_total = sum(m["total_customers"] for m in manager_list)
all_green = sum(m["visited_this_month"] for m in manager_list)
all_yellow = sum(m["visited_last_month_only"] for m in manager_list)
all_red = sum(m["not_visited_2months"] for m in manager_list)
team_summary = {
"total_customers": all_total,
"visited_this_month": all_green,
"visited_last_month_only": all_yellow,
"not_visited_2months": all_red,
"unassigned": len(unassigned),
"coverage_rate": round(all_green / all_total, 3) if all_total > 0 else 0.0,
}
return {
"reference_month": ref.strftime("%Y-%m"),
"managers": manager_list,
"unassigned_customers": unassigned,
"team_summary": team_summary,
}