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): source_dates = getattr(row, "source_report_dates", []) or [ getattr(row, "source_report_date", None) or row.report_date ] source_dates = sorted({item for item in source_dates if item}) source_date_text = "、".join( item.isoformat() for item in source_dates ) 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(), source_date_text, 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, "是" if getattr(row, "has_over_limit", False) else "否", getattr(row, "over_limit_summary", None), getattr(row, "over_limit_reasons", None), "、".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=15).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=12).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)