407 lines
16 KiB
Python
407 lines
16 KiB
Python
from dataclasses import dataclass
|
||
from decimal import Decimal, InvalidOperation
|
||
from io import BytesIO
|
||
from typing import Any
|
||
|
||
from openpyxl import load_workbook
|
||
from sqlalchemy import select
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.models import PersonRole, Personnel, Product, Role
|
||
from app.services.attendance_points import (
|
||
DEFAULT_ATTENDANCE_POINT_NAME,
|
||
accessible_point_names,
|
||
attendance_point_names,
|
||
set_person_attendance_points,
|
||
)
|
||
from app.services.cleaning import LEGACY_CLEANING_PROCESS_NAME, is_cleaning_stamping_method
|
||
from app.services.misc_work import is_misc_stamping_method
|
||
from app.services.process_names import normalize_numeric_process_name
|
||
|
||
|
||
@dataclass
|
||
class ImportStats:
|
||
imported: int = 0
|
||
updated: int = 0
|
||
skipped: int = 0
|
||
errors: list[str] | None = None
|
||
|
||
def as_dict(self) -> dict[str, Any]:
|
||
return {
|
||
"imported": self.imported,
|
||
"updated": self.updated,
|
||
"skipped": self.skipped,
|
||
"errors": self.errors or [],
|
||
}
|
||
|
||
|
||
PRODUCT_HEADERS = {
|
||
"考勤点": "attendance_point_name",
|
||
"项目号": "project_no",
|
||
"型材号": "profile_no",
|
||
"产品名称": "product_name",
|
||
"物料编码": "material_code",
|
||
"物料名称": "material_name",
|
||
"供应商": "supplier",
|
||
"产品净重(kg)": "product_net_weight_kg",
|
||
"产品毛重(kg)": "product_gross_weight_kg",
|
||
"允许报废率": "scrap_loss_rate",
|
||
"废料单价(元/kg)": "waste_price_yuan_per_kg",
|
||
"工序": "process_name",
|
||
"冲压方式": "stamping_method",
|
||
"操作人数": "operator_count",
|
||
"工序单价": "process_unit_price_yuan",
|
||
"标准节拍": "standard_beat",
|
||
}
|
||
|
||
PRODUCT_AUX_FIELDS = (
|
||
("product_net_weight_kg", "产品净重(kg)"),
|
||
("product_gross_weight_kg", "产品毛重(kg)"),
|
||
("scrap_loss_rate", "允许报废率"),
|
||
("waste_price_yuan_per_kg", "废料单价(元/kg)"),
|
||
)
|
||
|
||
|
||
class ProductImportValidationError(ValueError):
|
||
pass
|
||
|
||
PERSON_HEADERS = {
|
||
"考勤点": "attendance_point_names",
|
||
"电话号": "phone",
|
||
"手机号": "phone",
|
||
"姓名": "name",
|
||
"角色": "role",
|
||
}
|
||
|
||
ROLE_MAP = {
|
||
"冲压工人": Role.worker,
|
||
"员工": Role.worker,
|
||
"worker": Role.worker,
|
||
"管理员": Role.admin,
|
||
"admin": Role.admin,
|
||
"经理": Role.manager,
|
||
"manager": Role.manager,
|
||
}
|
||
|
||
|
||
def _text(value: Any) -> str:
|
||
if value is None:
|
||
return ""
|
||
return str(value).strip()
|
||
|
||
|
||
def _split_names(value: Any) -> list[str]:
|
||
text = _text(value)
|
||
if not text:
|
||
return []
|
||
for sep in [",", "，", ";", "；", "/", "|"]:
|
||
text = text.replace(sep, "、")
|
||
return [item.strip() for item in text.split("、") if item.strip()]
|
||
|
||
|
||
def _number(value: Any, default: float = 0) -> float:
|
||
try:
|
||
if value is None or value == "":
|
||
return default
|
||
return float(value)
|
||
except (TypeError, ValueError):
|
||
return default
|
||
|
||
|
||
def _optional_number(value: Any) -> float | None:
|
||
if value is None or value == "":
|
||
return None
|
||
try:
|
||
return float(value)
|
||
except (TypeError, ValueError):
|
||
return None
|
||
|
||
|
||
def _normalized_aux_value(value: Any) -> str:
|
||
text = _text(value)
|
||
if text == "":
|
||
return ""
|
||
try:
|
||
return str(Decimal(text).normalize())
|
||
except InvalidOperation:
|
||
return text
|
||
|
||
|
||
def _rows_from_excel(content: bytes) -> tuple[list[str], list[dict[str, Any]]]:
|
||
wb = load_workbook(BytesIO(content), read_only=True, data_only=True)
|
||
ws = wb.active
|
||
rows = list(ws.iter_rows(values_only=True))
|
||
if not rows:
|
||
return [], []
|
||
headers = [_text(value) for value in rows[0]]
|
||
mapped_rows = []
|
||
for values in rows[1:]:
|
||
mapped_rows.append({headers[index]: value for index, value in enumerate(values) if index < len(headers)})
|
||
return headers, mapped_rows
|
||
|
||
|
||
def _validate_product_aux_consistency(db: Session, rows: list[dict[str, Any]]) -> None:
|
||
aux_headers = {label: field for field, label in PRODUCT_AUX_FIELDS}
|
||
if not any(header in row for row in rows for header in aux_headers):
|
||
return
|
||
|
||
grouped: dict[str, dict[str, tuple[int, str, str]]] = {}
|
||
errors: list[str] = []
|
||
for index, raw in enumerate(rows, start=2):
|
||
point_name = _text(raw.get("考勤点")) or DEFAULT_ATTENDANCE_POINT_NAME
|
||
product_name = _text(raw.get("产品名称"))
|
||
process_name = _text(raw.get("工序"))
|
||
if not product_name:
|
||
continue
|
||
values = tuple(
|
||
_normalized_aux_value(raw.get(label))
|
||
for _, label in PRODUCT_AUX_FIELDS
|
||
)
|
||
group_key = f"{point_name}||{product_name}"
|
||
first_by_product = grouped.setdefault(group_key, {})
|
||
previous = first_by_product.get("values")
|
||
current_text = "，".join(
|
||
f"{label}={value or '空'}"
|
||
for (_, label), value in zip(PRODUCT_AUX_FIELDS, values)
|
||
)
|
||
if previous is None:
|
||
first_by_product["values"] = (index, process_name, current_text)
|
||
first_by_product["raw_values"] = (-1, "", "|".join(values))
|
||
continue
|
||
previous_raw = first_by_product["raw_values"][2]
|
||
if previous_raw != "|".join(values):
|
||
first_row, first_process, first_text = previous
|
||
errors.append(
|
||
f"考勤点【{point_name}】产品【{product_name}】不同工序的ERP辅助字段不一致："
|
||
f"第{first_row}行({first_process or '未填写工序'})为【{first_text}】，"
|
||
f"第{index}行({process_name or '未填写工序'})为【{current_text}】"
|
||
)
|
||
|
||
if errors:
|
||
detail = "；".join(errors[:10])
|
||
if len(errors) > 10:
|
||
detail += f"；另有{len(errors) - 10}处不一致"
|
||
raise ProductImportValidationError(f"{detail}。本次导入已取消，数据库未写入。")
|
||
|
||
|
||
def _validate_point_names(db: Session, names: list[str], user=None, *, action: str) -> None:
|
||
known = set(attendance_point_names(db))
|
||
missing = [name for name in names if name not in known]
|
||
if missing:
|
||
raise ProductImportValidationError(f"没有该考勤点：{'、'.join(sorted(set(missing)))}。本次导入已取消，数据库未写入。")
|
||
if user is not None and user.role != Role.manager:
|
||
allowed = set(accessible_point_names(db, user))
|
||
denied = [name for name in names if name not in allowed]
|
||
if denied:
|
||
raise ProductImportValidationError(f"无该考勤点{action}权限：{'、'.join(sorted(set(denied)))}。本次导入已取消，数据库未写入。")
|
||
|
||
|
||
def _validate_product_process_names(rows: list[dict[str, Any]]) -> None:
|
||
errors: list[str] = []
|
||
for index, raw in enumerate(rows, start=2):
|
||
if not any(_text(value) for value in raw.values()):
|
||
continue
|
||
raw_process_name = _text(raw.get("工序"))
|
||
stamping_method = _text(raw.get("冲压方式"))
|
||
is_misc = is_misc_stamping_method(stamping_method)
|
||
if is_misc:
|
||
errors.append(f"第{index}行处理杂活由系统自动生成，不能通过产品清单导入")
|
||
continue
|
||
if raw_process_name == LEGACY_CLEANING_PROCESS_NAME:
|
||
errors.append(f"第{index}行工序不能填写清洗，请把清洗填写在冲压方式中")
|
||
continue
|
||
try:
|
||
normalize_numeric_process_name(raw_process_name)
|
||
except ValueError:
|
||
errors.append(f"第{index}行工序必须是数字")
|
||
continue
|
||
if errors:
|
||
detail = "；".join(errors[:10])
|
||
if len(errors) > 10:
|
||
detail += f"；另有{len(errors) - 10}处错误"
|
||
raise ProductImportValidationError(f"{detail}。本次导入已取消，数据库未写入。")
|
||
|
||
|
||
def import_products(db: Session, content: bytes, user=None) -> ImportStats:
|
||
_, rows = _rows_from_excel(content)
|
||
_validate_product_aux_consistency(db, rows)
|
||
_validate_product_process_names(rows)
|
||
point_values = [
|
||
_text(raw.get("考勤点")) or DEFAULT_ATTENDANCE_POINT_NAME
|
||
for raw in rows
|
||
if any(_text(value) for value in raw.values())
|
||
]
|
||
_validate_point_names(db, point_values, user, action="产品导入")
|
||
stats = ImportStats(errors=[])
|
||
last_project_no = ""
|
||
seen: dict[tuple[str, str, str, str, str], Product] = {}
|
||
aux_by_product: dict[tuple[str, str], tuple[float | None, float | None, float | None, float | None]] = {}
|
||
|
||
for index, raw in enumerate(rows, start=2):
|
||
data = {field: raw[header] for header, field in PRODUCT_HEADERS.items() if header in raw}
|
||
attendance_point_name = _text(data.get("attendance_point_name")) or DEFAULT_ATTENDANCE_POINT_NAME
|
||
project_no = _text(data.get("project_no")) or last_project_no
|
||
product_name = _text(data.get("product_name"))
|
||
device_no = ""
|
||
raw_process_name = _text(data.get("process_name"))
|
||
stamping_method = _text(data.get("stamping_method"))
|
||
is_cleaning = is_cleaning_stamping_method(stamping_method)
|
||
is_misc = is_misc_stamping_method(stamping_method)
|
||
standard_beat = _number(data.get("standard_beat"))
|
||
if project_no:
|
||
last_project_no = project_no
|
||
|
||
if is_misc:
|
||
stats.skipped += 1
|
||
stats.errors.append(f"第{index}行处理杂活由系统自动生成，不能通过产品清单导入")
|
||
continue
|
||
if raw_process_name == LEGACY_CLEANING_PROCESS_NAME:
|
||
stats.skipped += 1
|
||
stats.errors.append(f"第{index}行工序不能填写清洗，请把清洗填写在冲压方式中，工序只能填写数字")
|
||
continue
|
||
try:
|
||
process_name = normalize_numeric_process_name(raw_process_name)
|
||
except ValueError:
|
||
stats.skipped += 1
|
||
stats.errors.append(f"第{index}行工序必须是数字")
|
||
continue
|
||
|
||
if not project_no or not product_name or not process_name or (not is_cleaning and not is_misc and standard_beat <= 0):
|
||
stats.skipped += 1
|
||
stats.errors.append(f"第{index}行缺少项目号、产品名称、工序或标准节拍（冲压方式为清洗时标准节拍可空）")
|
||
continue
|
||
|
||
key = (attendance_point_name, project_no, product_name, device_no, process_name)
|
||
if key in seen:
|
||
stats.skipped += 1
|
||
stats.errors.append(f"第{index}行重复产品/工序：{attendance_point_name} / {project_no} / {product_name} / {process_name}")
|
||
continue
|
||
product = seen.get(key)
|
||
if product is None:
|
||
product = db.get(
|
||
Product,
|
||
{
|
||
"attendance_point_name": attendance_point_name,
|
||
"project_no": project_no,
|
||
"product_name": product_name,
|
||
"device_no": device_no,
|
||
"process_name": process_name,
|
||
},
|
||
)
|
||
if product is None:
|
||
product = Product(
|
||
attendance_point_name=attendance_point_name,
|
||
project_no=project_no,
|
||
product_name=product_name,
|
||
device_no=device_no,
|
||
process_name=process_name,
|
||
)
|
||
db.add(product)
|
||
stats.imported += 1
|
||
else:
|
||
stats.updated += 1
|
||
seen[key] = product
|
||
|
||
product.profile_no = _text(data.get("profile_no")) or None
|
||
product.material_code = _text(data.get("material_code")) or None
|
||
product.material_name = _text(data.get("material_name")) or None
|
||
product.supplier = _text(data.get("supplier")) or None
|
||
if "product_net_weight_kg" in data:
|
||
product.product_net_weight_kg = _optional_number(data.get("product_net_weight_kg"))
|
||
if "product_gross_weight_kg" in data:
|
||
product.product_gross_weight_kg = _optional_number(data.get("product_gross_weight_kg"))
|
||
if "scrap_loss_rate" in data:
|
||
product.scrap_loss_rate = _optional_number(data.get("scrap_loss_rate"))
|
||
if "waste_price_yuan_per_kg" in data:
|
||
product.waste_price_yuan_per_kg = _optional_number(data.get("waste_price_yuan_per_kg"))
|
||
aux_by_product[(attendance_point_name, product_name)] = (
|
||
product.product_net_weight_kg,
|
||
product.product_gross_weight_kg,
|
||
product.scrap_loss_rate,
|
||
product.waste_price_yuan_per_kg,
|
||
)
|
||
product.device_no = device_no
|
||
product.process_name = process_name
|
||
product.stamping_method = stamping_method or None
|
||
product.operator_count = _number(data.get("operator_count"), 1) or 1
|
||
product.process_unit_price_yuan = _number(data.get("process_unit_price_yuan"), 0)
|
||
product.standard_beat = 0 if (is_cleaning or is_misc) else standard_beat
|
||
product.standard_workload = 0
|
||
|
||
if stats.errors:
|
||
db.rollback()
|
||
detail = "；".join(stats.errors[:20])
|
||
if len(stats.errors) > 20:
|
||
detail += f"；另有{len(stats.errors) - 20}处错误"
|
||
raise ProductImportValidationError(f"{detail}。本次导入已取消，数据库未写入。")
|
||
|
||
for (attendance_point_name, product_name), values in aux_by_product.items():
|
||
for sibling in db.scalars(
|
||
select(Product).where(
|
||
Product.attendance_point_name == attendance_point_name,
|
||
Product.product_name == product_name,
|
||
Product.device_no == "",
|
||
)
|
||
).all():
|
||
sibling.product_net_weight_kg = values[0]
|
||
sibling.product_gross_weight_kg = values[1]
|
||
sibling.scrap_loss_rate = values[2]
|
||
sibling.waste_price_yuan_per_kg = values[3]
|
||
|
||
db.commit()
|
||
return stats
|
||
|
||
|
||
def import_people(db: Session, content: bytes, user=None) -> ImportStats:
|
||
_, rows = _rows_from_excel(content)
|
||
row_point_names: list[str] = []
|
||
for raw in rows:
|
||
row_point_names.extend(_split_names(raw.get("考勤点")) or [DEFAULT_ATTENDANCE_POINT_NAME])
|
||
_validate_point_names(db, row_point_names, user, action="人员导入")
|
||
stats = ImportStats(errors=[])
|
||
|
||
for index, raw in enumerate(rows, start=2):
|
||
data = {field: raw[header] for header, field in PERSON_HEADERS.items() if header in raw}
|
||
phone = _text(data.get("phone"))
|
||
name = _text(data.get("name"))
|
||
role_text = _text(data.get("role"))
|
||
role = ROLE_MAP.get(role_text)
|
||
point_names = _split_names(data.get("attendance_point_names")) or [DEFAULT_ATTENDANCE_POINT_NAME]
|
||
if not phone or not name or role is None:
|
||
stats.skipped += 1
|
||
stats.errors.append(f"第{index}行缺少电话号、姓名或有效角色")
|
||
continue
|
||
if user is not None and user.role == Role.admin and role != Role.worker:
|
||
stats.skipped += 1
|
||
stats.errors.append(f"第{index}行管理员只能导入冲压工人")
|
||
continue
|
||
|
||
person = db.get(Personnel, phone)
|
||
if person is None:
|
||
person = Personnel(phone=phone)
|
||
db.add(person)
|
||
stats.imported += 1
|
||
else:
|
||
stats.updated += 1
|
||
person.name = name
|
||
person.is_temporary = False
|
||
person.temporary_expires_at = None
|
||
person_role = db.get(PersonRole, {"phone": phone, "role": role})
|
||
if person_role is None:
|
||
db.add(PersonRole(phone=phone, role=role))
|
||
try:
|
||
set_person_attendance_points(db, phone, point_names)
|
||
except ValueError as exc:
|
||
stats.skipped += 1
|
||
stats.errors.append(f"第{index}行{exc}")
|
||
|
||
if stats.errors:
|
||
db.rollback()
|
||
detail = "；".join(stats.errors[:20])
|
||
if len(stats.errors) > 20:
|
||
detail += f"；另有{len(stats.errors) - 20}处错误"
|
||
raise ProductImportValidationError(f"{detail}。本次导入已取消，数据库未写入。")
|
||
|
||
db.commit()
|
||
return stats
|