v6ole
1d0f412e4a
get_light_board 新增 user_id/role 参数: - director/leader: 看全部经理 + 未分配客户 - manager: 只看自己, 不显示未分配客户 - API 层从 current_user 提取身份信息透传
214 lines
7.4 KiB
Python
214 lines
7.4 KiB
Python
"""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,
|
|
user_id: UUID | None = None,
|
|
role: str = "manager",
|
|
) -> dict:
|
|
"""Get per-manager customer visit coverage for the light board.
|
|
|
|
Directors/leaders see all managers; managers see only themselves.
|
|
"""
|
|
ref = reference_date or today_cst()
|
|
|
|
# ── Managers (filtered by role) ──
|
|
managers_result = await db.execute(select(User).where(User.role == "manager"))
|
|
all_managers = managers_result.scalars().all()
|
|
# Filter: managers only see themselves
|
|
visible_managers = all_managers if role in ("director", "leader") else [
|
|
m for m in all_managers if m.id == user_id
|
|
]
|
|
|
|
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 visible_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) ──
|
|
# Only visible to directors/leaders — not actionable for individual managers
|
|
unassigned = []
|
|
if role in ("director", "leader"):
|
|
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)
|
|
)
|
|
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,
|
|
}
|