v6ole
4afa796ddc
- mgr_map 由字典推导式改为按 customer_id 聚合 list,再用顿号连接 - 修复同一客户多个 primary 经理时,后写入覆盖先写入导致只显示一个名字 数据修复: 大化瑶族自治县公安局交通警察大队 移除韦柳柏的 primary 分配,唐晓静为唯一 primary Co-Authored-By: Claude <noreply@anthropic.com>
768 lines
30 KiB
Python
768 lines
30 KiB
Python
import io
|
|
import uuid as uuid_mod
|
|
from uuid import uuid4
|
|
from typing import Optional
|
|
from fastapi import APIRouter, Depends, HTTPException, Query, UploadFile, File
|
|
from fastapi.responses import StreamingResponse
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy import select, or_, update
|
|
from sqlalchemy.orm import selectinload
|
|
from pydantic import BaseModel
|
|
import openpyxl
|
|
from app.database import get_db
|
|
from app.middleware.auth import get_current_user, require_director, require_any_role
|
|
from app.models.customer import Customer
|
|
from app.models.customer_contact import CustomerContact
|
|
from app.models.customer_assignment import CustomerAssignment
|
|
from app.models.visit import Visit
|
|
from app.models.work_plan import WorkPlan
|
|
from app.models.mini_business import MiniBusiness
|
|
from app.models.key_visit import KeyVisit
|
|
from app.models.daily_note import DailyNote
|
|
from app.models.user import User
|
|
from app.utils.audit import log_audit
|
|
from app.schemas.customer import (
|
|
CustomerCreate, CustomerUpdate, CustomerOut, CustomerListOut, CustomerListResponse,
|
|
ContactCreate, ContactOut, AssignmentCreate, AssignmentOut, BatchAssignRequest,
|
|
)
|
|
|
|
router = APIRouter(prefix="/customers", tags=["Customers"])
|
|
|
|
|
|
def _split_fee(fee: str) -> tuple[str, str]:
|
|
"""Split '5000元/月' into ('5000', '元/月')."""
|
|
if not fee:
|
|
return ("", "")
|
|
for u in ["元/月", "元/年"]:
|
|
if fee.endswith(u):
|
|
return (fee[:-len(u)].strip(), u)
|
|
# Custom unit: separate trailing non-digit+non-space chars
|
|
m = __import__('re').match(r'^(.+?)\s*([^\d\s]+)$', fee)
|
|
if m:
|
|
return (m.group(1).strip(), m.group(2).strip())
|
|
return (fee, "")
|
|
|
|
|
|
# ══════ Fixed-path routes (must come before /{customer_id}) ══════
|
|
|
|
@router.get("/", response_model=CustomerListResponse)
|
|
async def list_customers(
|
|
search: Optional[str] = Query(None),
|
|
industry: Optional[str] = Query(None),
|
|
service: Optional[str] = Query(None),
|
|
manager_id: Optional[str] = Query(None),
|
|
customer_type: Optional[str] = Query(None),
|
|
page: int = Query(1, ge=1),
|
|
page_size: int = Query(100, ge=1, le=1000),
|
|
current_user: dict = Depends(get_current_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
"""List customers with filters and pagination. All roles see all customers."""
|
|
from sqlalchemy import func
|
|
|
|
base_query = select(Customer)
|
|
|
|
if customer_type:
|
|
base_query = base_query.where(Customer.customer_type == customer_type)
|
|
if industry:
|
|
base_query = base_query.where(Customer.industry.ilike(f"%{industry}%"))
|
|
if service:
|
|
base_query = base_query.where(Customer.in_use_services.ilike(f"%{service}%"))
|
|
if manager_id:
|
|
assign_subq = select(CustomerAssignment.customer_id).where(
|
|
CustomerAssignment.manager_id == manager_id
|
|
)
|
|
base_query = base_query.where(Customer.id.in_(assign_subq))
|
|
if search:
|
|
contact_subq = select(CustomerContact.customer_id).where(
|
|
or_(
|
|
CustomerContact.name.ilike(f"%{search}%"),
|
|
CustomerContact.phone.ilike(f"%{search}%"),
|
|
)
|
|
)
|
|
base_query = base_query.where(or_(
|
|
Customer.name.ilike(f"%{search}%"),
|
|
Customer.industry.ilike(f"%{search}%"),
|
|
Customer.address.ilike(f"%{search}%"),
|
|
Customer.id.in_(contact_subq),
|
|
))
|
|
|
|
# Count total
|
|
count_query = select(func.count()).select_from(base_query.subquery())
|
|
total = (await db.execute(count_query)).scalar() or 0
|
|
|
|
# Paginate
|
|
offset = (page - 1) * page_size
|
|
query = base_query.order_by(Customer.customer_type, Customer.name).offset(offset).limit(page_size)
|
|
result = await db.execute(query)
|
|
items = result.scalars().all()
|
|
|
|
# Enrich with manager names
|
|
if items:
|
|
mgr_result = await db.execute(
|
|
select(CustomerAssignment.customer_id, User.name)
|
|
.join(User, CustomerAssignment.manager_id == User.id)
|
|
.where(CustomerAssignment.customer_id.in_([c.id for c in items]), CustomerAssignment.role == "primary")
|
|
.order_by(CustomerAssignment.customer_id)
|
|
)
|
|
mgr_map: dict[str, list] = {}
|
|
for cid, name in mgr_result.all():
|
|
mgr_map.setdefault(str(cid), []).append(name)
|
|
for item in items:
|
|
names = mgr_map.get(str(item.id), [])
|
|
item.primary_manager_name = "、".join(names) if names else None
|
|
|
|
return CustomerListResponse(
|
|
items=items,
|
|
total=total,
|
|
page=page,
|
|
page_size=page_size,
|
|
)
|
|
|
|
|
|
@router.get("/export")
|
|
async def export_customers(
|
|
current_user: dict = Depends(get_current_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
"""Export all customers (name, industry, address, services, fee, contacts) as Excel."""
|
|
from openpyxl import Workbook
|
|
result = await db.execute(select(Customer).options(selectinload(Customer.contacts)))
|
|
customers = result.scalars().all()
|
|
|
|
# Build manager lookup
|
|
mgr_result = await db.execute(
|
|
select(CustomerAssignment.customer_id, User.name)
|
|
.join(User, CustomerAssignment.manager_id == User.id)
|
|
.where(CustomerAssignment.role == "primary")
|
|
)
|
|
mgr_map = {str(cid): name for cid, name in mgr_result.all()}
|
|
|
|
wb = Workbook()
|
|
ws = wb.active
|
|
ws.title = "客户档案"
|
|
ws.append(["单位名称", "所属行业", "单位地址", "在用业务", "收支费用-金额", "收支费用-单位", "客户经理", "备注", "联系人姓名", "联系人电话", "联系人角色"])
|
|
for c in customers:
|
|
mgr_name = mgr_map.get(str(c.id), "")
|
|
amt, unit = _split_fee(c.monthly_fee)
|
|
if c.contacts:
|
|
for ct in c.contacts:
|
|
ws.append([c.name, c.industry, c.address, c.in_use_services, amt, unit, mgr_name, c.remarks or "", ct.name, ct.phone, ct.role_desc])
|
|
else:
|
|
ws.append([c.name, c.industry, c.address, c.in_use_services, amt, unit, mgr_name, c.remarks or "", "", "", ""])
|
|
for col_cells in ws.columns:
|
|
ws.column_dimensions[col_cells[0].column_letter].width = 22
|
|
|
|
output = io.BytesIO()
|
|
wb.save(output)
|
|
output.seek(0)
|
|
return StreamingResponse(
|
|
output,
|
|
media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
|
|
headers={"Content-Disposition": "attachment; filename=customers.xlsx"},
|
|
)
|
|
|
|
|
|
@router.get("/template")
|
|
async def download_import_template():
|
|
"""Download a blank customer import template (public, no auth required for download)."""
|
|
from openpyxl import Workbook
|
|
wb = Workbook()
|
|
ws = wb.active
|
|
ws.title = "客户档案导入模板"
|
|
ws.append(["单位名称*", "所属行业", "单位地址", "在用业务", "收支费用-金额", "收支费用-单位", "客户经理", "备注", "联系人姓名", "联系人电话", "联系人角色"])
|
|
ws.append(["XX科技有限公司", "信息技术", "XX市XX路100号", "云桌面、专线", "5000", "元/月", "韦柳柏", "重点客户，季度回访", "张三", "13800000000", "技术负责人"])
|
|
for col_cells in ws.columns:
|
|
ws.column_dimensions[col_cells[0].column_letter].width = 22
|
|
|
|
output = io.BytesIO()
|
|
wb.save(output)
|
|
output.seek(0)
|
|
return StreamingResponse(
|
|
output,
|
|
media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
|
|
headers={"Content-Disposition": "attachment; filename=customer_import_template.xlsx"},
|
|
)
|
|
|
|
|
|
@router.post("/import")
|
|
async def import_customers(
|
|
file: UploadFile = File(...),
|
|
current_user: dict = Depends(require_director),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
"""Import customers from Excel file. Director only."""
|
|
import uuid as uuid_mod
|
|
content = await file.read()
|
|
try:
|
|
wb = openpyxl.load_workbook(io.BytesIO(content), data_only=True)
|
|
except Exception as e:
|
|
raise HTTPException(status_code=400, detail=f"Excel 解析失败: {str(e)}")
|
|
|
|
ws = wb.active
|
|
created, updated, skipped = 0, 0, 0
|
|
reasons = []
|
|
errors = []
|
|
|
|
# Build user name → id lookup (all users, not just managers)
|
|
user_rows = await db.execute(select(User.name, User.id))
|
|
user_map = {name: uid for name, uid in user_rows.all()}
|
|
default_user_id = uuid_mod.UUID(current_user["user_id"])
|
|
|
|
for row_idx, row in enumerate(ws.iter_rows(min_row=2, values_only=True), start=2):
|
|
if not row or not row[0]:
|
|
continue
|
|
name = str(row[0]).strip() if row[0] else ""
|
|
if not name:
|
|
continue
|
|
industry = str(row[1]).strip() if len(row) > 1 and row[1] else ""
|
|
address = str(row[2]).strip() if len(row) > 2 and row[2] else ""
|
|
services = str(row[3]).strip() if len(row) > 3 and row[3] else ""
|
|
fee_amt = str(row[4]).strip() if len(row) > 4 and row[4] else ""
|
|
fee_unit = str(row[5]).strip() if len(row) > 5 and row[5] else ""
|
|
fee = (fee_amt + fee_unit).strip() if fee_amt else ""
|
|
mgr_name = str(row[6]).strip() if len(row) > 6 and row[6] else ""
|
|
remarks = str(row[7]).strip() if len(row) > 7 and row[7] else ""
|
|
contact_name = str(row[8]).strip() if len(row) > 8 and row[8] else ""
|
|
contact_phone = str(row[9]).strip() if len(row) > 9 and row[9] else ""
|
|
contact_role = str(row[10]).strip() if len(row) > 10 and row[10] else ""
|
|
|
|
assignee_id = user_map.get(mgr_name, default_user_id)
|
|
|
|
existing_result = await db.execute(select(Customer).where(Customer.name == name))
|
|
existing = existing_result.scalar_one_or_none()
|
|
|
|
try:
|
|
if existing:
|
|
# Update existing customer
|
|
existing.industry = industry or existing.industry
|
|
existing.address = address or existing.address
|
|
existing.in_use_services = services or existing.in_use_services
|
|
existing.monthly_fee = fee or existing.monthly_fee
|
|
existing.remarks = remarks or existing.remarks
|
|
# Update or create primary assignment if manager changed
|
|
if mgr_name:
|
|
assign_rows = await db.execute(
|
|
select(CustomerAssignment).where(CustomerAssignment.customer_id == existing.id, CustomerAssignment.role == "primary")
|
|
)
|
|
first_assign = assign_rows.first()
|
|
if first_assign:
|
|
first_assign[0].manager_id = assignee_id
|
|
else:
|
|
db.add(CustomerAssignment(customer_id=existing.id, manager_id=assignee_id, role="primary", assigned_by=default_user_id))
|
|
updated += 1
|
|
reasons.append(f"更新「{name}」的信息")
|
|
else:
|
|
customer = Customer(name=name, industry=industry, address=address, in_use_services=services, monthly_fee=fee, remarks=remarks, created_by=default_user_id)
|
|
db.add(customer)
|
|
await db.flush()
|
|
if contact_name:
|
|
db.add(CustomerContact(customer_id=customer.id, name=contact_name, phone=contact_phone, role_desc=contact_role))
|
|
db.add(CustomerAssignment(customer_id=customer.id, manager_id=assignee_id, role="primary", assigned_by=default_user_id))
|
|
created += 1
|
|
reasons.append(f"新建「{name}」")
|
|
except Exception as e:
|
|
errors.append(f"第{row_idx}行({name}): {str(e)}")
|
|
|
|
await db.commit()
|
|
return {"created": created, "updated": updated, "skipped": skipped, "reasons": reasons, "errors": errors}
|
|
|
|
|
|
@router.get("/check-duplicate/{name}")
|
|
async def check_duplicate(
|
|
name: str,
|
|
current_user: dict = Depends(get_current_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
"""Check for duplicate customer names before creating."""
|
|
result = await db.execute(
|
|
select(Customer.id, Customer.name, Customer.industry)
|
|
.where(Customer.name.ilike(f"%{name}%"))
|
|
.limit(10)
|
|
)
|
|
matches = [{"id": str(r.id), "name": r.name, "industry": r.industry} for r in result.all()]
|
|
return {"matches": matches}
|
|
|
|
|
|
@router.post("/batch-assign")
|
|
async def batch_assign(
|
|
data: BatchAssignRequest,
|
|
current_user: dict = Depends(require_director),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
"""Batch transfer customers to a new manager."""
|
|
import uuid as uuid_mod
|
|
for cid in data.customer_ids:
|
|
result = await db.execute(
|
|
select(CustomerAssignment).where(
|
|
CustomerAssignment.customer_id == cid,
|
|
CustomerAssignment.role == "primary",
|
|
)
|
|
)
|
|
all_rows = result.all()
|
|
if all_rows:
|
|
# Update the first one, delete any duplicates
|
|
first = all_rows[0][0]
|
|
first.manager_id = data.manager_id
|
|
first.assigned_by = uuid_mod.UUID(current_user["user_id"])
|
|
for dup in all_rows[1:]:
|
|
await db.delete(dup[0])
|
|
else:
|
|
db.add(CustomerAssignment(
|
|
customer_id=cid, manager_id=data.manager_id,
|
|
role="primary", assigned_by=uuid_mod.UUID(current_user["user_id"]),
|
|
))
|
|
await db.commit()
|
|
return {"detail": f"Assigned {len(data.customer_ids)} customers"}
|
|
|
|
|
|
@router.post("/quick-create")
|
|
async def quick_create_customer(
|
|
name: str = Query(...),
|
|
current_user: dict = Depends(require_any_role),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
"""Quick-create a customer with just a name. Returns the new customer."""
|
|
import uuid
|
|
customer = Customer(name=name, created_by=uuid.UUID(current_user["user_id"]))
|
|
db.add(customer)
|
|
await db.flush()
|
|
db.add(CustomerAssignment(
|
|
customer_id=customer.id, manager_id=uuid.UUID(current_user["user_id"]),
|
|
role="primary", assigned_by=uuid.UUID(current_user["user_id"]),
|
|
))
|
|
await db.commit()
|
|
return {"id": str(customer.id), "name": customer.name}
|
|
|
|
|
|
# ══════ Parameterized routes (/{customer_id}) ══════
|
|
|
|
@router.get("/{customer_id}", response_model=CustomerOut)
|
|
async def get_customer(
|
|
customer_id: str,
|
|
current_user: dict = Depends(get_current_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
result = await db.execute(
|
|
select(Customer).where(Customer.id == customer_id).options(selectinload(Customer.contacts))
|
|
)
|
|
customer = result.scalar_one_or_none()
|
|
if not customer:
|
|
raise HTTPException(status_code=404, detail="Customer not found")
|
|
|
|
# Load primary manager name
|
|
from sqlalchemy.orm import selectinload as sl
|
|
assign_result = await db.execute(
|
|
select(CustomerAssignment, User.name).join(User, CustomerAssignment.manager_id == User.id)
|
|
.where(CustomerAssignment.customer_id == customer_id, CustomerAssignment.role == "primary")
|
|
)
|
|
row = assign_result.first()
|
|
manager_name = row[1] if row else None
|
|
|
|
# Attach to response via a dict
|
|
out = {
|
|
"id": customer.id, "name": customer.name, "industry": customer.industry,
|
|
"address": customer.address, "in_use_services": customer.in_use_services,
|
|
"monthly_fee": customer.monthly_fee, "remarks": customer.remarks or "",
|
|
"created_by": customer.created_by,
|
|
"created_at": customer.created_at, "updated_at": customer.updated_at,
|
|
"contacts": customer.contacts, "primary_manager_name": manager_name,
|
|
}
|
|
return out
|
|
|
|
|
|
@router.post("/", response_model=CustomerOut)
|
|
async def create_customer(
|
|
data: CustomerCreate,
|
|
current_user: dict = Depends(require_any_role),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
if current_user["role"] == "leader":
|
|
raise HTTPException(status_code=403, detail="分管领导无法创建客户")
|
|
"""Create a new customer with optional contacts and manager assignment."""
|
|
import uuid
|
|
customer = Customer(
|
|
name=data.name, industry=data.industry, address=data.address,
|
|
in_use_services=data.in_use_services, monthly_fee=data.monthly_fee,
|
|
remarks=data.remarks,
|
|
created_by=uuid.UUID(current_user["user_id"]),
|
|
)
|
|
db.add(customer)
|
|
await db.flush()
|
|
|
|
for contact_data in data.contacts:
|
|
if not contact_data.name.strip():
|
|
continue
|
|
db.add(CustomerContact(
|
|
customer_id=customer.id,
|
|
name=contact_data.name.strip(),
|
|
phone=contact_data.phone.strip(),
|
|
role_desc=contact_data.role_desc.strip(),
|
|
))
|
|
|
|
assignee_id = data.assignee_id or uuid.UUID(current_user["user_id"])
|
|
db.add(CustomerAssignment(
|
|
customer_id=customer.id, manager_id=assignee_id,
|
|
role="primary", assigned_by=uuid.UUID(current_user["user_id"]),
|
|
))
|
|
await db.commit()
|
|
|
|
# Audit log
|
|
await log_audit(db, "customer", customer.id, customer.name, "create", current_user["user_id"], current_user["name"])
|
|
await db.commit()
|
|
|
|
result = await db.execute(
|
|
select(Customer).where(Customer.id == customer.id).options(selectinload(Customer.contacts))
|
|
)
|
|
return result.scalar_one()
|
|
|
|
|
|
@router.put("/{customer_id}", response_model=CustomerOut)
|
|
async def update_customer(
|
|
customer_id: str, data: CustomerUpdate,
|
|
current_user: dict = Depends(require_any_role),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
if current_user["role"] == "leader":
|
|
raise HTTPException(status_code=403, detail="分管领导无法编辑客户")
|
|
result = await db.execute(
|
|
select(Customer).where(Customer.id == customer_id).options(selectinload(Customer.contacts))
|
|
)
|
|
customer = result.scalar_one_or_none()
|
|
if not customer:
|
|
raise HTTPException(status_code=404, detail="Customer not found")
|
|
|
|
update_data = data.model_dump(exclude_unset=True)
|
|
assignee_id = update_data.pop("assignee_id", None) # Handle separately
|
|
|
|
# ── Name collision detection (for merge) ──
|
|
new_name = update_data.get("name")
|
|
if new_name and new_name.strip() != customer.name:
|
|
existing = await db.execute(
|
|
select(Customer).where(
|
|
Customer.name == new_name.strip(),
|
|
Customer.id != customer.id,
|
|
)
|
|
)
|
|
target = existing.scalar_one_or_none()
|
|
if target:
|
|
# Return 409 with merge preview — frontend should show merge dialog
|
|
preview = await _build_merge_preview(db, str(customer.id), str(target.id))
|
|
raise HTTPException(status_code=409, detail={
|
|
"message": "名称冲突 — 同名客户已存在",
|
|
"source_id": str(customer.id),
|
|
"source_name": customer.name,
|
|
"target_id": str(target.id),
|
|
"target_name": target.name,
|
|
"preview": preview,
|
|
})
|
|
|
|
for key, value in update_data.items():
|
|
setattr(customer, key, value)
|
|
|
|
# Update primary manager assignment if changed (director only)
|
|
if assignee_id:
|
|
if current_user["role"] != "director":
|
|
raise HTTPException(status_code=403, detail="Only director can change manager assignment")
|
|
assign_result = await db.execute(
|
|
select(CustomerAssignment).where(
|
|
CustomerAssignment.customer_id == customer.id,
|
|
CustomerAssignment.role == "primary",
|
|
)
|
|
)
|
|
existing_a = assign_result.scalar_one_or_none()
|
|
if existing_a:
|
|
existing_a.manager_id = assignee_id
|
|
existing_a.assigned_by = uuid_mod.UUID(current_user["user_id"])
|
|
else:
|
|
db.add(CustomerAssignment(
|
|
customer_id=customer.id, manager_id=assignee_id,
|
|
role="primary", assigned_by=uuid_mod.UUID(current_user["user_id"]),
|
|
))
|
|
|
|
await db.commit()
|
|
|
|
# Audit log
|
|
await log_audit(db, "customer", uuid_mod.UUID(customer_id), customer.name, "update", current_user["user_id"], current_user["name"])
|
|
await db.commit()
|
|
|
|
result = await db.execute(
|
|
select(Customer).where(Customer.id == customer.id).options(selectinload(Customer.contacts))
|
|
)
|
|
return result.scalar_one()
|
|
|
|
|
|
@router.delete("/{customer_id}")
|
|
async def delete_customer(
|
|
customer_id: str,
|
|
current_user: dict = Depends(require_director),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
result = await db.execute(select(Customer).where(Customer.id == customer_id))
|
|
customer = result.scalar_one_or_none()
|
|
if not customer:
|
|
raise HTTPException(status_code=404, detail="Customer not found")
|
|
cust_name = customer.name
|
|
await db.delete(customer)
|
|
await db.commit()
|
|
# Audit log
|
|
await log_audit(db, "customer", uuid_mod.UUID(customer_id), cust_name, "delete", current_user["user_id"], current_user["name"])
|
|
await db.commit()
|
|
return {"detail": "deleted"}
|
|
|
|
|
|
# ── Customer Merge ──
|
|
|
|
|
|
class MergePreviewOut(BaseModel):
|
|
source_id: str
|
|
source_name: str
|
|
target_id: str
|
|
target_name: str
|
|
visits: int = 0
|
|
work_plans: int = 0
|
|
mini_business: int = 0
|
|
key_visits: int = 0
|
|
contacts: int = 0
|
|
assignments: int = 0
|
|
note: str = ""
|
|
|
|
|
|
class MergeRequest(BaseModel):
|
|
target_id: str
|
|
|
|
|
|
async def _build_merge_preview(db: AsyncSession, source_id: str, target_id: str) -> dict:
|
|
"""Count all records that would be migrated from source to target."""
|
|
sid = uuid_mod.UUID(source_id)
|
|
tid = uuid_mod.UUID(target_id)
|
|
|
|
visits = (await db.execute(
|
|
select(select(Visit).where(Visit.customer_id == sid).subquery()).with_only_columns(
|
|
__import__('sqlalchemy').func.count()
|
|
)
|
|
)).scalar() or 0
|
|
# Simpler: count directly
|
|
from sqlalchemy import func as sa_func
|
|
visits = (await db.execute(select(sa_func.count()).select_from(Visit).where(Visit.customer_id == sid))).scalar() or 0
|
|
plans = (await db.execute(select(sa_func.count()).select_from(WorkPlan).where(WorkPlan.customer_id == sid))).scalar() or 0
|
|
mini = (await db.execute(select(sa_func.count()).select_from(MiniBusiness).where(MiniBusiness.customer_id == sid))).scalar() or 0
|
|
kv = (await db.execute(select(sa_func.count()).select_from(KeyVisit).where(KeyVisit.customer_id == sid))).scalar() or 0
|
|
contacts = (await db.execute(select(sa_func.count()).select_from(CustomerContact).where(CustomerContact.customer_id == sid))).scalar() or 0
|
|
assignments = (await db.execute(select(sa_func.count()).select_from(CustomerAssignment).where(CustomerAssignment.customer_id == sid))).scalar() or 0
|
|
|
|
# Check for assignment conflicts
|
|
target_assigns = await db.execute(
|
|
select(CustomerAssignment.manager_id).where(CustomerAssignment.customer_id == tid)
|
|
)
|
|
target_managers = {str(row[0]) for row in target_assigns.all()}
|
|
source_assigns = await db.execute(
|
|
select(CustomerAssignment).where(CustomerAssignment.customer_id == sid)
|
|
)
|
|
conflict_note = ""
|
|
for sa in source_assigns.scalars().all():
|
|
if str(sa.manager_id) in target_managers:
|
|
conflict_note = f"目标客户已有相同经理的分配，将去重"
|
|
|
|
return {
|
|
"source_id": source_id,
|
|
"source_name": "",
|
|
"target_id": target_id,
|
|
"target_name": "",
|
|
"visits": visits,
|
|
"work_plans": plans,
|
|
"mini_business": mini,
|
|
"key_visits": kv,
|
|
"contacts": contacts,
|
|
"assignments": assignments,
|
|
"note": conflict_note,
|
|
}
|
|
|
|
|
|
@router.get("/{customer_id}/merge-preview")
|
|
async def get_merge_preview(
|
|
customer_id: str,
|
|
target_id: str = Query(...),
|
|
current_user: dict = Depends(require_director),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
"""Preview merge: show what data would be moved from source to target."""
|
|
source = (await db.execute(select(Customer).where(Customer.id == customer_id))).scalar_one_or_none()
|
|
target = (await db.execute(select(Customer).where(Customer.id == target_id))).scalar_one_or_none()
|
|
if not source or not target:
|
|
raise HTTPException(status_code=404, detail="客户不存在")
|
|
if customer_id == target_id:
|
|
raise HTTPException(status_code=400, detail="不能合并到自身")
|
|
|
|
preview = await _build_merge_preview(db, customer_id, target_id)
|
|
preview["source_name"] = source.name
|
|
preview["target_name"] = target.name
|
|
return preview
|
|
|
|
|
|
@router.post("/{customer_id}/merge")
|
|
async def execute_merge(
|
|
customer_id: str,
|
|
body: MergeRequest,
|
|
current_user: dict = Depends(require_director),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
"""Merge source customer into target. All related data is migrated, source is deleted."""
|
|
source_id = uuid_mod.UUID(customer_id)
|
|
target_id = uuid_mod.UUID(body.target_id)
|
|
|
|
if customer_id == body.target_id:
|
|
raise HTTPException(status_code=400, detail="不能合并到自身")
|
|
|
|
source = (await db.execute(select(Customer).where(Customer.id == source_id))).scalar_one_or_none()
|
|
target = (await db.execute(select(Customer).where(Customer.id == target_id))).scalar_one_or_none()
|
|
if not source or not target:
|
|
raise HTTPException(status_code=404, detail="客户不存在")
|
|
|
|
# Build preview for response
|
|
preview = await _build_merge_preview(db, customer_id, body.target_id)
|
|
|
|
# ── Transaction: migrate all FK references ──
|
|
for model, fk_col in [
|
|
(Visit, Visit.customer_id),
|
|
(WorkPlan, WorkPlan.customer_id),
|
|
(MiniBusiness, MiniBusiness.customer_id),
|
|
(KeyVisit, KeyVisit.customer_id),
|
|
]:
|
|
await db.execute(
|
|
update(model).where(fk_col == source_id).values(customer_id=target_id)
|
|
)
|
|
|
|
# Contacts: migrate, skip duplicates
|
|
source_contacts = (await db.execute(
|
|
select(CustomerContact).where(CustomerContact.customer_id == source_id)
|
|
)).scalars().all()
|
|
target_contacts = (await db.execute(
|
|
select(CustomerContact).where(CustomerContact.customer_id == target_id)
|
|
)).scalars().all()
|
|
existing_contact_keys = {(c.name, c.phone) for c in target_contacts}
|
|
for c in source_contacts:
|
|
if (c.name, c.phone) in existing_contact_keys:
|
|
await db.delete(c) # Skip duplicate
|
|
else:
|
|
c.customer_id = target_id
|
|
db.add(c)
|
|
|
|
# Assignments: migrate, skip same (manager_id, role) pairs
|
|
target_assigns = (await db.execute(
|
|
select(CustomerAssignment).where(CustomerAssignment.customer_id == target_id)
|
|
)).scalars().all()
|
|
existing_assign_keys = {(str(a.manager_id), a.role) for a in target_assigns}
|
|
source_assigns = (await db.execute(
|
|
select(CustomerAssignment).where(CustomerAssignment.customer_id == source_id)
|
|
)).scalars().all()
|
|
for a in source_assigns:
|
|
if (str(a.manager_id), a.role) in existing_assign_keys:
|
|
await db.delete(a) # Skip duplicate
|
|
else:
|
|
a.customer_id = target_id
|
|
db.add(a)
|
|
|
|
# Update target's last_visit_date to the max of both
|
|
if source.last_visit_date:
|
|
if not target.last_visit_date or source.last_visit_date > target.last_visit_date:
|
|
target.last_visit_date = source.last_visit_date
|
|
target.last_visit_manager_id = source.last_visit_manager_id
|
|
|
|
# Delete source customer
|
|
source_name = source.name
|
|
target_name = target.name
|
|
await db.delete(source)
|
|
await db.commit()
|
|
|
|
preview["result"] = f"已将「{source_name}」合并到「{target_name}」"
|
|
return preview
|
|
|
|
|
|
@router.post("/{customer_id}/contacts", response_model=ContactOut)
|
|
async def add_contact(
|
|
customer_id: str, data: ContactCreate,
|
|
current_user: dict = Depends(require_any_role),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
contact = CustomerContact(customer_id=customer_id, name=data.name, phone=data.phone, role_desc=data.role_desc)
|
|
db.add(contact)
|
|
await db.commit()
|
|
await db.refresh(contact)
|
|
return contact
|
|
|
|
|
|
@router.delete("/{customer_id}/contacts/{contact_id}")
|
|
async def delete_contact(
|
|
customer_id: str, contact_id: str,
|
|
current_user: dict = Depends(require_any_role),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
result = await db.execute(
|
|
select(CustomerContact).where(CustomerContact.id == contact_id, CustomerContact.customer_id == customer_id)
|
|
)
|
|
contact = result.scalar_one_or_none()
|
|
if not contact:
|
|
raise HTTPException(status_code=404, detail="Contact not found")
|
|
await db.delete(contact)
|
|
await db.commit()
|
|
return {"detail": "deleted"}
|
|
|
|
|
|
# ── Assignments ──
|
|
|
|
@router.get("/{customer_id}/assignments", response_model=list[AssignmentOut])
|
|
async def list_assignments(
|
|
customer_id: str,
|
|
current_user: dict = Depends(get_current_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
result = await db.execute(select(CustomerAssignment).where(CustomerAssignment.customer_id == customer_id))
|
|
return result.scalars().all()
|
|
|
|
|
|
@router.post("/{customer_id}/assignments", response_model=AssignmentOut)
|
|
async def assign_manager(
|
|
customer_id: str, data: AssignmentCreate,
|
|
current_user: dict = Depends(require_director),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
assignment = CustomerAssignment(
|
|
customer_id=customer_id, manager_id=data.manager_id,
|
|
role=data.role, assigned_by=current_user["user_id"],
|
|
)
|
|
db.add(assignment)
|
|
await db.commit()
|
|
await db.refresh(assignment)
|
|
return assignment
|
|
|
|
|
|
@router.get("/{customer_id}/overdue-plans")
|
|
async def get_overdue_plans(
|
|
customer_id: str,
|
|
current_user: dict = Depends(get_current_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
"""Return the latest overdue (plan_date < today) and still '计划中' plan for a customer.
|
|
Used by the frontend to pre-fill visit forms with pending plan content."""
|
|
from app.utils.timezone import today_cst
|
|
today = today_cst()
|
|
result = await db.execute(
|
|
select(WorkPlan).where(
|
|
WorkPlan.customer_id == uuid_mod.UUID(customer_id),
|
|
WorkPlan.status == "计划中",
|
|
WorkPlan.plan_date < today,
|
|
).order_by(WorkPlan.plan_date.desc()).limit(1)
|
|
)
|
|
plan = result.scalar_one_or_none()
|
|
if not plan:
|
|
return {"plan": None}
|
|
return {
|
|
"plan": {
|
|
"id": str(plan.id),
|
|
"plan_content": plan.plan_content,
|
|
"plan_date": str(plan.plan_date),
|
|
"manager_id": str(plan.manager_id),
|
|
}
|
|
}
|