v6ole
efc200b6de
- 抽取 auto_complete_work_plans() 可复用函数到 visits service - POST/PUT 拜访均触发自动完成（更新场景解决补录问题） - Excel 导入旧周报时也触发自动完成 - 逾期计划自动取消：plan_date < today 且无匹配拜访→已取消 - edit_log 记录区分三种来源：拜访自动完成/拜访更新自动完成/旧周报导入自动完成/逾期自动取消 Co-Authored-By: Claude <noreply@anthropic.com>
397 lines
17 KiB
Python
397 lines
17 KiB
Python
import io
|
||
import uuid
|
||
import re
|
||
from datetime import datetime, date
|
||
from typing import Any
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
from sqlalchemy import select, update
|
||
from app.models.customer import Customer
|
||
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
|
||
|
||
import openpyxl
|
||
|
||
# Companion name separators: Chinese comma, English comma, Chinese semicolon, dun-hao
|
||
_COMPANION_SEP = re.compile(r'[，,；、;]')
|
||
|
||
|
||
def _parse_companions(raw: str, user_map: dict[str, uuid.UUID]) -> tuple[list[uuid.UUID], list[str]]:
|
||
"""Split a companion string into system user IDs and external names.
|
||
|
||
Returns (system_ids, external_names).
|
||
"""
|
||
if not raw or not raw.strip():
|
||
return [], []
|
||
|
||
parts = [p.strip() for p in _COMPANION_SEP.split(raw) if p.strip()]
|
||
system_ids: list[uuid.UUID] = []
|
||
external_names: list[str] = []
|
||
|
||
for name in parts:
|
||
uid = user_map.get(name)
|
||
if uid:
|
||
system_ids.append(uid)
|
||
else:
|
||
external_names.append(name)
|
||
|
||
return system_ids, external_names
|
||
|
||
|
||
def _fuzzy_match_customer(
|
||
cust_name: str,
|
||
customer_map: dict[str, uuid.UUID],
|
||
) -> uuid.UUID | None:
|
||
"""Try to match a customer name with fuzzy rules.
|
||
|
||
Rules (in order):
|
||
1. Exact match (already handled before calling this)
|
||
2. Normalize whitespace → exact match
|
||
3. One name fully contains the other
|
||
"""
|
||
norm = cust_name.replace(' ', '').replace(' ', '')
|
||
# Rule 2: whitespace-normalized match
|
||
for existing_name, cid in customer_map.items():
|
||
existing_norm = existing_name.replace(' ', '').replace(' ', '')
|
||
if norm == existing_norm:
|
||
return cid
|
||
|
||
# Rule 3: containment (longer name contains shorter)
|
||
for existing_name, cid in customer_map.items():
|
||
if len(norm) >= 4 and len(existing_name.replace(' ', '').replace(' ', '')) >= 4:
|
||
if norm in existing_name.replace(' ', '').replace(' ', '') or \
|
||
existing_name.replace(' ', '').replace(' ', '') in norm:
|
||
return cid
|
||
|
||
return None
|
||
|
||
|
||
async def import_from_excel(db: AsyncSession, file_bytes: bytes, manager_id: uuid.UUID) -> dict:
|
||
"""Parse old weekly report Excel and import data. Returns summary stats.
|
||
|
||
Features:
|
||
- Companion parsing: split by comma/dun-hao → system users + external names
|
||
- Auto-create customers: if customer not found, create from Excel data
|
||
- Fuzzy name matching: whitespace normalization + containment
|
||
- Daily notes import (Sheet 5)
|
||
"""
|
||
wb = openpyxl.load_workbook(io.BytesIO(file_bytes), data_only=True)
|
||
stats = {
|
||
"visits": 0, "work_plans": 0, "mini_business": 0, "key_visits": 0,
|
||
"daily_notes": 0, "skipped": 0, "skip_reasons": [],
|
||
"customers_created": 0, "customers_created_names": [],
|
||
"external_companions": 0, "name_corrections": [],
|
||
}
|
||
|
||
# ── Pre-fetch lookups ──
|
||
customers_result = await db.execute(select(Customer.id, Customer.name))
|
||
customer_map = {c.name: c.id for c in customers_result.all()}
|
||
|
||
users_result = await db.execute(select(User.id, User.name))
|
||
user_map = {u.name: u.id for u in users_result.all()}
|
||
|
||
# Customer → assigned primary manager lookup (not from Excel)
|
||
assign_result = await db.execute(
|
||
select(CustomerAssignment.customer_id, CustomerAssignment.manager_id)
|
||
.where(CustomerAssignment.role == "primary")
|
||
)
|
||
customer_manager_map: dict[uuid.UUID, uuid.UUID] = {}
|
||
for cid, mid in assign_result.all():
|
||
if cid not in customer_manager_map: # first primary wins
|
||
customer_manager_map[cid] = mid
|
||
|
||
def get_assigned_manager(customer_id: uuid.UUID) -> uuid.UUID:
|
||
"""Return the customer's assigned primary manager, or the importer as fallback."""
|
||
return customer_manager_map.get(customer_id, manager_id)
|
||
|
||
# Helper: resolve or create customer
|
||
async def resolve_customer(cust_name: str, mgr_name: str = "") -> tuple[uuid.UUID | None, str]:
|
||
"""Resolve customer by name. Auto-creates if not found. Returns (id, note)."""
|
||
if not cust_name or not cust_name.strip():
|
||
return None, ""
|
||
|
||
cust_name = cust_name.strip()
|
||
cid = customer_map.get(cust_name)
|
||
if cid:
|
||
return cid, ""
|
||
|
||
# Fuzzy match
|
||
fuzzy_id = _fuzzy_match_customer(cust_name, customer_map)
|
||
if fuzzy_id:
|
||
real_name = next((n for n, i in customer_map.items() if i == fuzzy_id), cust_name)
|
||
stats["name_corrections"].append(f"「{cust_name}」→「{real_name}」")
|
||
customer_map[cust_name] = fuzzy_id # cache for future rows
|
||
return fuzzy_id, ""
|
||
|
||
# Auto-create customer (no manager assignment — leave unassigned)
|
||
customer = Customer(name=cust_name, created_by=manager_id)
|
||
db.add(customer)
|
||
await db.flush()
|
||
customer_map[cust_name] = customer.id
|
||
stats["customers_created"] += 1
|
||
stats["customers_created_names"].append(cust_name)
|
||
|
||
return customer.id, ""
|
||
|
||
# ── Parse Sheet 1: 每日拜访记录 ──
|
||
if "每日拜访记录" in wb.sheetnames:
|
||
ws = wb["每日拜访记录"]
|
||
for row in ws.iter_rows(min_row=2, values_only=True):
|
||
if not row[0]:
|
||
continue
|
||
try:
|
||
cust_name, visit_date_str, visit_method, time_range, visitor_name, visitor_phone, \
|
||
content, demand, companions_str, mgr_str = (
|
||
row[0], str(row[1]) if row[1] else str(date.today()),
|
||
str(row[2]) if row[2] else "上门", str(row[3]) if row[3] else "",
|
||
str(row[4]) if row[4] else "", str(row[5]) if row[5] else "",
|
||
str(row[6]) if row[6] else "", str(row[7]) if row[7] else "",
|
||
str(row[8]) if row[8] else "", str(row[9]) if row[9] else "",
|
||
)
|
||
|
||
# Resolve customer (auto-create if needed)
|
||
customer_id, _ = await resolve_customer(cust_name, str(row[9]) if row[9] else "")
|
||
if not customer_id:
|
||
stats["skipped"] += 1
|
||
stats["skip_reasons"].append(f"客户名称为空，跳过")
|
||
continue
|
||
|
||
# Parse visit date
|
||
try:
|
||
visit_date = datetime.strptime(visit_date_str[:10], "%Y-%m-%d").date()
|
||
except ValueError:
|
||
visit_date = date.today()
|
||
|
||
# Use customer's assigned primary manager (not Excel column)
|
||
visit_manager_id = get_assigned_manager(customer_id)
|
||
|
||
# Check duplicate: same date + same manager + same customer
|
||
existing = await db.execute(
|
||
select(Visit).where(
|
||
Visit.visit_date == visit_date,
|
||
Visit.manager_id == visit_manager_id,
|
||
Visit.customer_id == customer_id,
|
||
)
|
||
)
|
||
if existing.scalar_one_or_none():
|
||
stats["skipped"] += 1
|
||
stats["skip_reasons"].append(f"重复：{cust_name} {visit_date} 已存在")
|
||
continue
|
||
|
||
# Parse companions into system users + external names
|
||
sys_companions, ext_names = _parse_companions(str(companions_str), user_map)
|
||
if ext_names:
|
||
stats["external_companions"] += len(ext_names)
|
||
|
||
visit = Visit(
|
||
customer_id=customer_id,
|
||
visit_date=visit_date,
|
||
visit_method=visit_method if visit_method in ["上门", "电话", "微信", "出差"] else "上门",
|
||
time_range=time_range,
|
||
visitor_name=visitor_name,
|
||
visitor_phone=visitor_phone,
|
||
communication_content=content,
|
||
customer_demand=demand,
|
||
companions=sys_companions,
|
||
companion_names=ext_names,
|
||
manager_id=visit_manager_id,
|
||
)
|
||
db.add(visit)
|
||
stats["visits"] += 1
|
||
|
||
# Auto-complete matching work plans
|
||
from app.services.visits import auto_complete_work_plans
|
||
await auto_complete_work_plans(
|
||
db, customer_id, visit_date,
|
||
mgr_name, "旧周报导入自动完成",
|
||
)
|
||
|
||
# Update customer's last_visit_date for light board
|
||
cust = await db.get(Customer, customer_id)
|
||
if cust and (not cust.last_visit_date or visit_date > cust.last_visit_date):
|
||
cust.last_visit_date = visit_date
|
||
cust.last_visit_manager_id = visit_manager_id
|
||
except Exception as e:
|
||
stats["skipped"] += 1
|
||
stats["skip_reasons"].append(f"拜访解析异常：{repr(e)[:120]}")
|
||
|
||
# ── Parse Sheet 2: 下周工作计划 ──
|
||
if "下周工作计划" in wb.sheetnames:
|
||
ws2 = wb["下周工作计划"]
|
||
for row in ws2.iter_rows(min_row=2, values_only=True):
|
||
if not row[0]:
|
||
continue
|
||
try:
|
||
cust_name = str(row[0]).strip() if row[0] else ""
|
||
plan_content = str(row[1]) if row[1] else ""
|
||
plan_date_str = str(row[2]) if row[2] else str(date.today())
|
||
mgr_name = str(row[3]).strip() if row[3] else ""
|
||
status = str(row[4]) if row[4] else "计划中"
|
||
|
||
customer_id, _ = await resolve_customer(cust_name, mgr_name)
|
||
if not customer_id:
|
||
stats["skipped"] += 1
|
||
stats["skip_reasons"].append(f"工作计划：客户「{cust_name}」无法解析")
|
||
continue
|
||
|
||
try:
|
||
plan_date = datetime.strptime(plan_date_str[:10], "%Y-%m-%d").date()
|
||
except ValueError:
|
||
plan_date = date.today()
|
||
|
||
plan_manager_id = get_assigned_manager(customer_id)
|
||
|
||
work_plan = WorkPlan(
|
||
customer_id=customer_id,
|
||
plan_content=plan_content,
|
||
plan_date=plan_date,
|
||
manager_id=plan_manager_id,
|
||
status=status if status in ["计划中", "已完成", "已取消"] else "计划中",
|
||
)
|
||
db.add(work_plan)
|
||
stats["work_plans"] += 1
|
||
except Exception:
|
||
stats["skipped"] += 1
|
||
|
||
# ── Parse Sheet 3: 小微业务商机 ──
|
||
if "小微业务商机" in wb.sheetnames:
|
||
ws3 = wb["小微业务商机"]
|
||
for row in ws3.iter_rows(min_row=2, values_only=True):
|
||
if not row[0]:
|
||
continue
|
||
try:
|
||
cust_name = str(row[0]).strip() if row[0] else ""
|
||
product_type = str(row[1]) if row[1] else ""
|
||
amount = str(row[2]) if row[2] else ""
|
||
follow_up = str(row[3]) if row[3] else ""
|
||
status = str(row[4]) if row[4] else "跟进中"
|
||
mgr_name = str(row[5]).strip() if row[5] else ""
|
||
expected_revenue = str(row[6]) if row[6] else ""
|
||
|
||
customer_id, _ = await resolve_customer(cust_name, mgr_name)
|
||
if not customer_id:
|
||
stats["skipped"] += 1
|
||
stats["skip_reasons"].append(f"商机：客户「{cust_name}」无法解析")
|
||
continue
|
||
|
||
biz_manager_id = get_assigned_manager(customer_id)
|
||
|
||
mini = MiniBusiness(
|
||
customer_id=customer_id,
|
||
product_type=product_type,
|
||
amount=amount,
|
||
follow_up_detail=follow_up,
|
||
status=status if status in ["跟进中", "已成交", "已流失"] else "跟进中",
|
||
manager_id=biz_manager_id,
|
||
expected_revenue_date=expected_revenue,
|
||
)
|
||
db.add(mini)
|
||
stats["mini_business"] += 1
|
||
except Exception:
|
||
stats["skipped"] += 1
|
||
|
||
# ── Parse Sheet 4: 要客拜访计划 ──
|
||
if "要客拜访计划" in wb.sheetnames:
|
||
ws4 = wb["要客拜访计划"]
|
||
for row in ws4.iter_rows(min_row=2, values_only=True):
|
||
if not row[0]:
|
||
continue
|
||
try:
|
||
cust_name = str(row[0]).strip() if row[0] else ""
|
||
urgency = str(row[1]) if row[1] else "一般"
|
||
description = str(row[2]) if row[2] else ""
|
||
progress = str(row[3]) if row[3] else "未开始"
|
||
planned_date = str(row[4]) if row[4] else ""
|
||
planned_visitor = str(row[5]) if row[5] else ""
|
||
visit_target = str(row[6]) if row[6] else ""
|
||
mgr_name = str(row[7]).strip() if row[7] else ""
|
||
|
||
customer_id, _ = await resolve_customer(cust_name, mgr_name)
|
||
if not customer_id:
|
||
stats["skipped"] += 1
|
||
stats["skip_reasons"].append(f"要客：客户「{cust_name}」无法解析")
|
||
continue
|
||
|
||
kv_manager_id = get_assigned_manager(customer_id)
|
||
valid_urgency = urgency if urgency in ["一般", "重要", "紧急"] else "一般"
|
||
valid_progress = progress if progress in ["未开始", "进行中", "已完成"] else "未开始"
|
||
|
||
key_visit = KeyVisit(
|
||
customer_id=customer_id,
|
||
urgency_level=valid_urgency,
|
||
description=description,
|
||
progress_status=valid_progress,
|
||
planned_date=planned_date,
|
||
planned_visitor=planned_visitor,
|
||
visit_target=visit_target,
|
||
manager_id=kv_manager_id,
|
||
)
|
||
db.add(key_visit)
|
||
stats["key_visits"] += 1
|
||
except Exception:
|
||
stats["skipped"] += 1
|
||
|
||
# ── Parse Sheet 5: 今日纪要 ──
|
||
if "今日纪要" in wb.sheetnames:
|
||
ws5 = wb["今日纪要"]
|
||
valid_categories = ["行政事务", "合同整理", "发票处理", "内部会议", "培训学习", "其他"]
|
||
for row in ws5.iter_rows(min_row=2, values_only=True):
|
||
if not row[0]:
|
||
continue
|
||
try:
|
||
note_date_str = str(row[0]) if row[0] else str(date.today())
|
||
category = str(row[1]) if row[1] else "其他"
|
||
content = str(row[2]) if row[2] else ""
|
||
time_range = str(row[3]) if row[3] else ""
|
||
mgr_name = str(row[4]).strip() if row[4] else ""
|
||
|
||
# Validate category
|
||
if category not in valid_categories:
|
||
category = "其他"
|
||
|
||
# Resolve manager by name
|
||
note_manager_id = manager_id # default to importer
|
||
if mgr_name:
|
||
mgr_id = user_map.get(mgr_name)
|
||
if mgr_id:
|
||
note_manager_id = mgr_id
|
||
else:
|
||
stats["skip_reasons"].append(f"纪要：客户经理「{mgr_name}」不存在，使用导入人")
|
||
|
||
try:
|
||
note_date = datetime.strptime(note_date_str[:10], "%Y-%m-%d").date()
|
||
except ValueError:
|
||
note_date = date.today()
|
||
|
||
# Skip duplicates: same manager, same date, same category
|
||
existing = await db.execute(
|
||
select(DailyNote).where(
|
||
DailyNote.note_date == note_date,
|
||
DailyNote.manager_id == note_manager_id,
|
||
DailyNote.category == category,
|
||
)
|
||
)
|
||
if existing.scalar_one_or_none():
|
||
stats["skipped"] += 1
|
||
stats["skip_reasons"].append(f"重复：{note_date} {category} 纪要已存在")
|
||
continue
|
||
|
||
daily_note = DailyNote(
|
||
manager_id=note_manager_id,
|
||
note_date=note_date,
|
||
category=category,
|
||
content=content,
|
||
time_range=time_range,
|
||
)
|
||
db.add(daily_note)
|
||
stats["daily_notes"] += 1
|
||
except Exception:
|
||
stats["skipped"] += 1
|
||
|
||
await db.commit()
|
||
return stats
|