220 lines
8.2 KiB
Python
220 lines
8.2 KiB
Python
from io import BytesIO
|
||
|
||
from openpyxl import Workbook
|
||
from openpyxl.styles import PatternFill
|
||
from sqlalchemy import and_, not_, or_, select
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.models import PersonAttendancePoint, PersonRole, Personnel, Product, Role
|
||
from app.services.attendance_points import accessible_point_names
|
||
from app.services.common import ROLE_NAMES, as_float
|
||
from app.services.misc_work import MISC_WORK_PROCESS_NAME, MISC_WORK_PRODUCT_NAME, MISC_WORK_STAMPING_METHOD
|
||
from app.services.process_names import export_process_name
|
||
|
||
|
||
def _workbook_bytes(workbook: Workbook) -> bytes:
|
||
output = BytesIO()
|
||
workbook.save(output)
|
||
return output.getvalue()
|
||
|
||
|
||
def _optional_float(value):
|
||
return "" if value is None else as_float(value)
|
||
|
||
|
||
def export_products(db: Session, user=None) -> bytes:
|
||
workbook = Workbook()
|
||
sheet = workbook.active
|
||
sheet.title = "产品清单"
|
||
sheet.append([
|
||
"考勤点",
|
||
"项目号",
|
||
"型材号",
|
||
"产品名称",
|
||
"物料编码",
|
||
"物料名称",
|
||
"供应商",
|
||
"工序",
|
||
"冲压方式",
|
||
"操作人数",
|
||
"工序单价",
|
||
"不包含物料转运节拍",
|
||
"标准节拍",
|
||
"产品净重(kg)",
|
||
"产品毛重(kg)",
|
||
"允许报废率",
|
||
"废料单价(元/kg)",
|
||
])
|
||
query = select(Product).where(
|
||
Product.device_no == "",
|
||
or_(Product.stamping_method.is_(None), Product.stamping_method != MISC_WORK_STAMPING_METHOD),
|
||
not_(and_(Product.product_name == MISC_WORK_PRODUCT_NAME, Product.process_name == MISC_WORK_PROCESS_NAME)),
|
||
)
|
||
if user is not None:
|
||
query = query.where(Product.attendance_point_name.in_(accessible_point_names(db, user)))
|
||
products = db.scalars(query.order_by(Product.attendance_point_name, Product.project_no, Product.product_name, Product.process_name)).all()
|
||
for product in products:
|
||
sheet.append([
|
||
product.attendance_point_name,
|
||
product.project_no,
|
||
product.profile_no,
|
||
product.product_name,
|
||
product.material_code,
|
||
product.material_name,
|
||
product.supplier,
|
||
export_process_name(product.process_name),
|
||
product.stamping_method,
|
||
as_float(product.operator_count),
|
||
as_float(product.process_unit_price_yuan),
|
||
"",
|
||
as_float(product.standard_beat),
|
||
_optional_float(product.product_net_weight_kg),
|
||
_optional_float(product.product_gross_weight_kg),
|
||
_optional_float(product.scrap_loss_rate),
|
||
_optional_float(product.waste_price_yuan_per_kg),
|
||
])
|
||
return _workbook_bytes(workbook)
|
||
|
||
|
||
def export_people(db: Session, user=None) -> bytes:
|
||
workbook = Workbook()
|
||
sheet = workbook.active
|
||
sheet.title = "人员清单"
|
||
sheet.append(["考勤点", "电话号", "姓名", "角色"])
|
||
query = select(PersonRole).join(PersonRole.person).order_by(Personnel.name, PersonRole.role)
|
||
if user is not None and user.role == Role.admin:
|
||
point_names = accessible_point_names(db, user)
|
||
query = query.where(
|
||
PersonRole.role == Role.worker,
|
||
Personnel.attendance_points.any(PersonAttendancePoint.attendance_point_name.in_(point_names)),
|
||
)
|
||
people = db.scalars(query).all()
|
||
for item in people:
|
||
point_names = sorted({point.attendance_point_name for point in item.person.attendance_points})
|
||
sheet.append([
|
||
"、".join(point_names),
|
||
item.phone,
|
||
item.person.name,
|
||
ROLE_NAMES.get(str(item.role), str(item.role)),
|
||
])
|
||
return _workbook_bytes(workbook)
|
||
|
||
|
||
def export_dashboard_rows(rows: list) -> bytes:
|
||
workbook = Workbook()
|
||
sheet = workbook.active
|
||
sheet.title = "报工看板"
|
||
warn_fill = PatternFill(fill_type="solid", fgColor="FFFF6666")
|
||
void_fill = PatternFill(fill_type="solid", fgColor="FFD9D9D9")
|
||
sheet.append([
|
||
"考勤点",
|
||
"报工日期",
|
||
"员工姓名",
|
||
"员工电话",
|
||
"工种",
|
||
"项目号",
|
||
"产品名称",
|
||
"工序",
|
||
"有效工时(时)",
|
||
"班种分布",
|
||
"成品数量",
|
||
"不良数量",
|
||
"标准工作量",
|
||
"实际节拍",
|
||
"标准节拍",
|
||
"型材号",
|
||
"物料编码",
|
||
"物料名称",
|
||
"材料库存批次号",
|
||
"使用设备",
|
||
"冲压方式",
|
||
"换料次数",
|
||
"操作人数",
|
||
"工序单价",
|
||
"参考工资",
|
||
"报工标记",
|
||
"校正项",
|
||
"备注",
|
||
])
|
||
wage_summary: dict[tuple[str, str, str], dict[str, float | str]] = {}
|
||
for row_index, row in enumerate(rows, start=2):
|
||
report_flags = []
|
||
if getattr(row, "is_voided", False):
|
||
report_flags.append("作废")
|
||
if getattr(row, "is_multi_person", False):
|
||
report_flags.append("多人协作")
|
||
if getattr(row, "is_continuous_die", False):
|
||
report_flags.append("连续模")
|
||
correction_text = "；".join(
|
||
f"{item.get('label', '')} {item.get('old_value', '-')}→{item.get('new_value', '-')}"
|
||
for item in getattr(row, "correction_pairs", [])
|
||
)
|
||
sheet.append([
|
||
row.attendance_point_name,
|
||
row.report_date.isoformat(),
|
||
row.employee_name,
|
||
row.employee_phone,
|
||
row.worker_type,
|
||
row.project_no,
|
||
row.product_name,
|
||
row.process_name,
|
||
round(as_float(row.effective_minutes) / 60, 2),
|
||
row.shift_distribution_text,
|
||
row.total_good_qty,
|
||
row.total_defect_qty,
|
||
row.expected_workload,
|
||
row.actual_beat,
|
||
row.standard_beat,
|
||
row.profile_no,
|
||
row.material_code,
|
||
row.material_name,
|
||
row.raw_material_batch_no,
|
||
row.device_no,
|
||
row.stamping_method,
|
||
row.changeover_count,
|
||
row.operator_count,
|
||
row.process_unit_price_yuan,
|
||
row.reference_wage,
|
||
"、".join(report_flags),
|
||
correction_text,
|
||
getattr(row, "review_remark", None),
|
||
])
|
||
summary_key = (row.employee_phone, row.employee_name, row.worker_type)
|
||
summary = wage_summary.setdefault(
|
||
summary_key,
|
||
{
|
||
"employee_phone": row.employee_phone,
|
||
"employee_name": row.employee_name,
|
||
"worker_type": row.worker_type,
|
||
"total_good_qty": 0.0,
|
||
"reference_wage": 0.0,
|
||
"detail_count": 0.0,
|
||
},
|
||
)
|
||
summary["total_good_qty"] = as_float(summary["total_good_qty"]) + as_float(row.total_good_qty)
|
||
summary["reference_wage"] = as_float(summary["reference_wage"]) + as_float(row.reference_wage)
|
||
summary["detail_count"] = as_float(summary["detail_count"]) + 1
|
||
actual_beat = as_float(row.actual_beat)
|
||
standard_beat = as_float(row.standard_beat)
|
||
total_good_qty = as_float(row.total_good_qty)
|
||
expected_workload = as_float(row.expected_workload)
|
||
if standard_beat > 0 and (actual_beat > standard_beat or actual_beat < standard_beat * 0.7):
|
||
sheet.cell(row=row_index, column=14).fill = warn_fill
|
||
if expected_workload > 0 and (total_good_qty < expected_workload or total_good_qty > expected_workload * 1.3):
|
||
sheet.cell(row=row_index, column=11).fill = warn_fill
|
||
if getattr(row, "is_voided", False):
|
||
for cell in sheet[row_index]:
|
||
cell.fill = void_fill
|
||
summary_sheet = workbook.create_sheet("工资汇总")
|
||
summary_sheet.append(["员工姓名", "员工电话", "工种", "明细行数", "成品数量合计", "参考工资合计"])
|
||
for summary in sorted(wage_summary.values(), key=lambda item: (str(item["employee_name"]), str(item["employee_phone"]))):
|
||
summary_sheet.append([
|
||
summary["employee_name"],
|
||
summary["employee_phone"],
|
||
summary["worker_type"],
|
||
int(as_float(summary["detail_count"])),
|
||
round(as_float(summary["total_good_qty"]), 2),
|
||
round(as_float(summary["reference_wage"]), 2),
|
||
])
|
||
return _workbook_bytes(workbook)
|