from __future__ import annotations from collections import Counter from datetime import UTC, date, datetime from typing import Any from zoneinfo import ZoneInfo from sqlalchemy import inspect, select, text from sqlalchemy.orm import Session from app.core.config import get_settings from app.core.logging import get_logger from app.core.security import hash_password from app.db.base import Base from app.db.session import get_session_factory from app.models.employee import Employee from app.models.employee_change_log import EmployeeChangeLog from app.models.organization import OrganizationUnit from app.models.role import Role from app.repositories.employee import EmployeeRepository from app.schemas.employee import ( EmployeeCreate, EmployeeHistoryRead, EmployeeImportErrorRead, EmployeeImportResultRead, EmployeeImportSummaryRead, EmployeeMetaRead, EmployeeOrganizationRead, EmployeeRead, EmployeeRoleOptionRead, EmployeeStatusSummaryRead, EmployeeUpdate, ) from app.services.employee_spreadsheet import ( EmployeeImportRow, EmployeeSpreadsheetError, build_export_workbook_bytes, build_import_template_bytes, parse_employee_workbook, ) from app.services.employee_seed import ( EMPLOYEE_DEFINITIONS, EMPLOYEE_PROFILE_REPAIRS, ORGANIZATION_DEFINITIONS, ROLE_DEFINITIONS, ROLE_DISPLAY_ORDER, ROLE_PERMISSION_MAP, ) logger = get_logger("app.services.employee") DEFAULT_EMPLOYEE_PASSWORD = "123456" MAX_EMPLOYEE_CHANGE_LOGS = 5 DISPLAY_TIMEZONE = ZoneInfo("Asia/Shanghai") STATUS_TONE_MAP = { "在职": "success", "试用中": "warning", "停用": "neutral", } STATUS_ORDER = ["全部员工", "在职", "试用中", "停用"] SEEDED_EMPLOYEE_DEFINITIONS = EMPLOYEE_DEFINITIONS[:30] EXTRA_SEED_EMPLOYEE_NOS = {item["employee_no"] for item in EMPLOYEE_DEFINITIONS[30:]} def prepare_employee_directory() -> None: settings = get_settings() if not settings.setup_completed: logger.info("Employee directory bootstrap skipped because setup is incomplete") return session_factory = get_session_factory() with session_factory() as db: service = EmployeeService(db) service.ensure_directory_ready() service.apply_profile_repairs() class EmployeeService: def __init__(self, db: Session) -> None: self.db = db self.repository = EmployeeRepository(db) def ensure_directory_ready(self) -> None: try: Base.metadata.create_all(bind=self.db.get_bind()) self._ensure_employee_schema() self._prune_extra_seed_employees() self._seed_roles() self._seed_organization_units() self._seed_employees() self.db.commit() except Exception: self.db.rollback() logger.exception("Failed to prepare employee directory") raise def list_employees(self, status: str | None = None, keyword: str | None = None) -> list[EmployeeRead]: self.ensure_directory_ready() employees = self.repository.list(status=status, keyword=keyword) logger.info("Listed employees (count=%d, status=%s, keyword=%s)", len(employees), status, keyword) return [self._serialize_employee(item) for item in employees] def get_employee(self, employee_id: str) -> EmployeeRead | None: self.ensure_directory_ready() employee = self.repository.get(employee_id) if employee is None: logger.warning("Employee not found id=%s", employee_id) return None logger.info("Fetched employee id=%s name=%s", employee_id, employee.name) return self._serialize_employee(employee) def get_employee_meta(self) -> EmployeeMetaRead: self.ensure_directory_ready() employees = self.repository.list() status_counter = Counter(item.employment_status for item in employees) status_summary = [ EmployeeStatusSummaryRead( id=status, label=status, count=len(employees) if status == "全部员工" else status_counter.get(status, 0), ) for status in STATUS_ORDER ] role_options = [ EmployeeRoleOptionRead( id=role.role_code, code=role.role_code, label=role.name, desc=role.description, permissions=list(ROLE_PERMISSION_MAP.get(role.role_code, [])), ) for role in self._sorted_roles(self.repository.list_roles()) ] organization_options = [ EmployeeOrganizationRead( id=unit.id, code=unit.unit_code, name=unit.name, unitType=unit.unit_type, costCenter=unit.cost_center, location=unit.location, managerName=unit.manager_name, ) for unit in sorted( self.repository.list_organization_units(), key=lambda item: item.name, ) ] return EmployeeMetaRead( totalEmployees=len(employees), statusSummary=status_summary, roleOptions=role_options, organizationOptions=organization_options, ) def create_employee(self, payload: EmployeeCreate) -> EmployeeRead: self.ensure_directory_ready() if self.repository.get_by_employee_no(payload.employee_no): raise ValueError(f"员工编号 {payload.employee_no} 已存在") if self.repository.get_by_email(str(payload.email)): raise ValueError(f"邮箱 {payload.email} 已存在") employee = Employee( employee_no=payload.employee_no, name=payload.name, email=str(payload.email), gender=payload.gender, birth_date=payload.parsed_birth_date(), phone=payload.phone, join_date=payload.parsed_join_date(), location=payload.location, position=payload.position, grade=payload.grade, cost_center=payload.cost_center, finance_owner_name=payload.finance_owner_name, employment_status=payload.employment_status, sync_state=payload.sync_state, spotlight=payload.spotlight, password_hash=hash_password(DEFAULT_EMPLOYEE_PASSWORD), last_sync_at=datetime.now(UTC), ) if payload.organization_unit_code: employee.organization_unit = self.repository.get_organization_by_code(payload.organization_unit_code) if payload.manager_employee_no: employee.manager = self.repository.get_by_employee_no(payload.manager_employee_no) roles = [ role for code in payload.role_codes if (role := self.repository.get_role_by_code(code)) is not None ] employee.roles = self._sorted_roles(roles) created = self.repository.create(employee) logger.info( "Created employee id=%s no=%s name=%s", created.id, created.employee_no, created.name ) hydrated = self.repository.get(created.id) return self._serialize_employee(hydrated or created) def update_employee(self, employee_id: str, payload: EmployeeUpdate) -> EmployeeRead: self.ensure_directory_ready() employee = self.repository.get(employee_id) if employee is None: raise LookupError("Employee not found") changed_fields: list[str] = [] password_changed = False if "name" in payload.model_fields_set and payload.name is not None: name = payload.name.strip() if not name: raise ValueError("员工姓名不能为空") if name != employee.name: employee.name = name changed_fields.append("姓名") if "gender" in payload.model_fields_set: gender = self._normalize_optional_text(payload.gender) if gender != employee.gender: employee.gender = gender changed_fields.append("性别") if "birth_date" in payload.model_fields_set: birth_date = payload.parsed_birth_date() if payload.birth_date else None if birth_date != employee.birth_date: employee.birth_date = birth_date changed_fields.append("出生日期") if "phone" in payload.model_fields_set: phone = self._normalize_optional_text(payload.phone) if phone != employee.phone: employee.phone = phone changed_fields.append("手机号") if "email" in payload.model_fields_set and payload.email is not None: email = str(payload.email) existing = self.repository.get_by_email(email) if existing is not None and existing.id != employee.id: raise ValueError(f"邮箱 {email} 已存在") if email != employee.email: employee.email = email changed_fields.append("邮箱") if "join_date" in payload.model_fields_set: join_date = payload.parsed_join_date() if payload.join_date else None if join_date != employee.join_date: employee.join_date = join_date changed_fields.append("入职日期") if "location" in payload.model_fields_set: location = self._normalize_optional_text(payload.location) if location != employee.location: employee.location = location changed_fields.append("办公地点") if "position" in payload.model_fields_set and payload.position is not None: position = payload.position.strip() if not position: raise ValueError("岗位不能为空") if position != employee.position: employee.position = position changed_fields.append("岗位") if "grade" in payload.model_fields_set and payload.grade is not None: grade = payload.grade.strip() if not grade: raise ValueError("职级不能为空") if grade != employee.grade: employee.grade = grade changed_fields.append("职级") if "cost_center" in payload.model_fields_set: cost_center = self._normalize_optional_text(payload.cost_center) if cost_center != employee.cost_center: employee.cost_center = cost_center changed_fields.append("成本中心") if "finance_owner_name" in payload.model_fields_set: finance_owner_name = self._normalize_optional_text(payload.finance_owner_name) if finance_owner_name != employee.finance_owner_name: employee.finance_owner_name = finance_owner_name changed_fields.append("财务归口") if "organization_unit_code" in payload.model_fields_set: organization_code = self._normalize_optional_text(payload.organization_unit_code) current_code = ( employee.organization_unit.unit_code if employee.organization_unit else None ) if organization_code != current_code: if organization_code: organization = self.repository.get_organization_by_code(organization_code) if organization is None: raise ValueError(f"部门编码 {organization_code} 不存在") employee.organization_unit = organization else: employee.organization_unit = None changed_fields.append("所属部门") if "manager_employee_no" in payload.model_fields_set: manager_employee_no = self._normalize_optional_text(payload.manager_employee_no) current_manager_no = employee.manager.employee_no if employee.manager else None if manager_employee_no: if manager_employee_no == employee.employee_no: raise ValueError("直属上级不能是员工本人") manager = self.repository.get_by_employee_no(manager_employee_no) if manager is None: raise ValueError(f"直属上级工号 {manager_employee_no} 不存在") if manager_employee_no != current_manager_no: employee.manager = manager changed_fields.append("直属上级") elif current_manager_no is not None: employee.manager = None changed_fields.append("直属上级") role_changed = False sorted_roles: list[Role] = [] if "role_codes" in payload.model_fields_set and payload.role_codes is not None: requested_codes = list(dict.fromkeys(payload.role_codes)) roles: list[Role] = [] invalid_codes: list[str] = [] for code in requested_codes: role = self.repository.get_role_by_code(code) if role is None: invalid_codes.append(code) continue roles.append(role) if invalid_codes: raise ValueError(f"角色不存在：{'、'.join(invalid_codes)}") sorted_roles = self._sorted_roles(roles) next_role_codes = [role.role_code for role in sorted_roles] current_role_codes = [role.role_code for role in self._sorted_roles(list(employee.roles))] if next_role_codes != current_role_codes: employee.roles = sorted_roles role_changed = True if "password" in payload.model_fields_set and payload.password: password = payload.password.strip() if len(password) < 5: raise ValueError("员工密码至少需要 5 位") employee.password_hash = hash_password(password) password_changed = True if not changed_fields and not password_changed and not role_changed: return self._serialize_employee(employee) now = datetime.now(UTC) employee.last_sync_at = now employee.sync_state = "已同步" if changed_fields: self._append_change_log( employee, action=f"更新员工信息（{'、'.join(changed_fields)}）", occurred_at=now, ) if role_changed: role_labels = "、".join(role.name for role in sorted_roles) self._append_change_log( employee, action=f"更新系统角色（{role_labels}）", occurred_at=now, ) if password_changed: self._append_change_log(employee, action="重置员工登录密码", occurred_at=now) hydrated = self._save_employee_and_reload(employee) logger.info( "Updated employee id=%s fields=%s role_changed=%s", employee.id, ",".join(changed_fields), role_changed, ) return self._serialize_employee(hydrated) def disable_employee(self, employee_id: str) -> EmployeeRead: self.ensure_directory_ready() employee = self.repository.get(employee_id) if employee is None: raise LookupError("Employee not found") if employee.employment_status == "停用": return self._serialize_employee(employee) now = datetime.now(UTC) employee.employment_status = "停用" employee.sync_state = "已同步" employee.last_sync_at = now employee.spotlight = False self._append_change_log(employee, action="停用员工账号", occurred_at=now) hydrated = self._save_employee_and_reload(employee) logger.info("Disabled employee id=%s no=%s", employee.id, employee.employee_no) return self._serialize_employee(hydrated) def enable_employee(self, employee_id: str) -> EmployeeRead: self.ensure_directory_ready() employee = self.repository.get(employee_id) if employee is None: raise LookupError("Employee not found") if employee.employment_status != "停用": return self._serialize_employee(employee) now = datetime.now(UTC) employee.employment_status = "在职" employee.sync_state = "已同步" employee.last_sync_at = now self._append_change_log(employee, action="启用员工账号", occurred_at=now) hydrated = self._save_employee_and_reload(employee) logger.info("Enabled employee id=%s no=%s", employee.id, employee.employee_no) return self._serialize_employee(hydrated) def build_import_template(self) -> bytes: self.ensure_directory_ready() return build_import_template_bytes() def export_employees(self, status: str | None = None, keyword: str | None = None) -> bytes: self.ensure_directory_ready() employees = self.repository.list(status=status, keyword=keyword) rows: list[list[str]] = [] for employee in employees: organization = employee.organization_unit role_codes = ",".join(role.role_code for role in self._sorted_roles(list(employee.roles))) rows.append( [ employee.employee_no, employee.name, employee.email, employee.gender or "", self._format_date(employee.birth_date) or "", employee.phone or "", self._format_date(employee.join_date) or "", employee.location or "", employee.position, employee.grade, organization.unit_code if organization else "", employee.manager.employee_no if employee.manager else "", employee.finance_owner_name or "", employee.cost_center or "", employee.employment_status, role_codes, ] ) return build_export_workbook_bytes(rows) def import_employees(self, content: bytes, actor: str = "系统管理员") -> EmployeeImportResultRead: self.ensure_directory_ready() parsed_rows, parse_errors = parse_employee_workbook(content) if parse_errors: return self._build_import_failure(parse_errors, total_rows=len(parsed_rows)) validation_errors = self._validate_import_rows(parsed_rows) if validation_errors: return self._build_import_failure(validation_errors, total_rows=len(parsed_rows)) try: summary = self._apply_import_rows(parsed_rows, actor=actor) except Exception: self.db.rollback() logger.exception("Employee import failed during database write") raise imported_at = self._format_datetime(datetime.now(UTC)) or "" message = f"导入成功：新增 {summary['created']} 人，更新 {summary['updated']} 人。" logger.info( "Imported employees created=%d updated=%d total=%d", summary["created"], summary["updated"], len(parsed_rows), ) return EmployeeImportResultRead( success=True, message=message, summary=EmployeeImportSummaryRead( totalRows=len(parsed_rows), created=summary["created"], updated=summary["updated"], errorCount=0, ), errors=[], importedAt=imported_at, ) def _validate_import_rows( self, rows: list[EmployeeImportRow] ) -> list[EmployeeSpreadsheetError]: errors: list[EmployeeSpreadsheetError] = [] employee_nos_in_file: dict[str, int] = {} emails_in_file: dict[str, int] = {} roles_by_code = {role.role_code: role for role in self.repository.list_roles()} organizations_by_code = { unit.unit_code: unit for unit in self.repository.list_organization_units() } employees_by_no = { employee.employee_no: employee for employee in self.repository.list() } import_employee_nos = {row.employee_no for row in rows} for row in rows: if row.employee_no in employee_nos_in_file: errors.append( EmployeeSpreadsheetError( row=row.row_number, column="员工编号*", employee_no=row.employee_no, message=f"员工编号 {row.employee_no} 在文件中重复。", ) ) else: employee_nos_in_file[row.employee_no] = row.row_number if row.email in emails_in_file: errors.append( EmployeeSpreadsheetError( row=row.row_number, column="邮箱*", employee_no=row.employee_no, message=f"邮箱 {row.email} 在文件中重复。", ) ) else: emails_in_file[row.email] = row.row_number existing_by_email = self.repository.get_by_email(row.email) if existing_by_email is not None and existing_by_email.employee_no != row.employee_no: errors.append( EmployeeSpreadsheetError( row=row.row_number, column="邮箱*", employee_no=row.employee_no, message=( f"邮箱 {row.email} 已被员工 " f"{existing_by_email.employee_no} 使用。" ), ) ) if row.organization_unit_code and row.organization_unit_code not in organizations_by_code: errors.append( EmployeeSpreadsheetError( row=row.row_number, column="部门编码", employee_no=row.employee_no, message=f"部门编码 {row.organization_unit_code} 不存在。", ) ) if row.manager_employee_no: manager_exists = ( row.manager_employee_no in employees_by_no or row.manager_employee_no in import_employee_nos ) if not manager_exists: errors.append( EmployeeSpreadsheetError( row=row.row_number, column="直属上级工号", employee_no=row.employee_no, message=f"直属上级工号 {row.manager_employee_no} 不存在。", ) ) if row.manager_employee_no == row.employee_no: errors.append( EmployeeSpreadsheetError( row=row.row_number, column="直属上级工号", employee_no=row.employee_no, message="直属上级不能是员工本人。", ) ) invalid_role_codes = [ code for code in row.role_codes if code not in roles_by_code ] if invalid_role_codes: errors.append( EmployeeSpreadsheetError( row=row.row_number, column="角色编码", employee_no=row.employee_no, message=f"角色不存在：{'、'.join(invalid_role_codes)}。", ) ) return errors def _apply_import_rows( self, rows: list[EmployeeImportRow], *, actor: str, ) -> dict[str, int]: roles_by_code = {role.role_code: role for role in self.repository.list_roles()} organizations_by_code = { unit.unit_code: unit for unit in self.repository.list_organization_units() } employees_by_no = { employee.employee_no: employee for employee in self.repository.list() } created = 0 updated = 0 now = datetime.now(UTC) try: for row in rows: employee = employees_by_no.get(row.employee_no) is_new = employee is None if is_new: employee = Employee( employee_no=row.employee_no, name=row.name, email=row.email, password_hash=hash_password(DEFAULT_EMPLOYEE_PASSWORD), ) self.db.add(employee) employees_by_no[row.employee_no] = employee created += 1 else: updated += 1 employee.name = row.name employee.email = row.email employee.gender = row.gender employee.birth_date = row.birth_date employee.phone = row.phone employee.join_date = row.join_date employee.location = row.location employee.position = row.position employee.grade = row.grade employee.finance_owner_name = row.finance_owner_name employee.cost_center = row.cost_center employee.employment_status = row.employment_status employee.sync_state = "已同步" employee.last_sync_at = now if row.organization_unit_code: employee.organization_unit = organizations_by_code[row.organization_unit_code] else: employee.organization_unit = None employee.roles = self._sorted_roles( [roles_by_code[code] for code in row.role_codes if code in roles_by_code] ) action = ( "通过 Excel 导入新建员工档案" if is_new else "通过 Excel 导入更新员工档案" ) self._append_change_log(employee, action=action, owner=actor, occurred_at=now) self.db.flush() for row in rows: employee = employees_by_no[row.employee_no] if row.manager_employee_no: employee.manager = employees_by_no.get(row.manager_employee_no) else: employee.manager = None self.db.commit() except Exception: self.db.rollback() raise return {"created": created, "updated": updated} def _build_import_failure( self, errors: list[EmployeeSpreadsheetError], *, total_rows: int, ) -> EmployeeImportResultRead: error_reads = [ EmployeeImportErrorRead( row=item.row, column=item.column, employeeNo=item.employee_no, message=item.message, ) for item in errors ] return EmployeeImportResultRead( success=False, message=( f"导入未执行：共发现 {len(error_reads)} 处错误，请修正后重新导入。" "原有员工数据未变更。" ), summary=EmployeeImportSummaryRead( totalRows=total_rows, created=0, updated=0, errorCount=len(error_reads), ), errors=error_reads, importedAt=None, ) def _seed_roles(self) -> None: existing_by_code = {role.role_code: role for role in self.repository.list_roles()} for definition in ROLE_DEFINITIONS: role = existing_by_code.get(definition["role_code"]) if role is None: role = Role( role_code=definition["role_code"], name=definition["name"], description=definition["description"], ) self.db.add(role) existing_by_code[role.role_code] = role self.db.flush() def _seed_organization_units(self) -> None: existing_by_code = { unit.unit_code: unit for unit in self.repository.list_organization_units() } for definition in ORGANIZATION_DEFINITIONS: organization = existing_by_code.get(definition["unit_code"]) if organization is None: organization = OrganizationUnit( unit_code=definition["unit_code"], name=definition["name"], unit_type=definition["unit_type"], cost_center=definition.get("cost_center"), location=definition.get("location"), manager_name=definition.get("manager_name"), ) self.db.add(organization) existing_by_code[organization.unit_code] = organization self.db.flush() for definition in ORGANIZATION_DEFINITIONS: parent_code = definition.get("parent_code") if not parent_code: continue organization = existing_by_code[definition["unit_code"]] if organization.parent_id: continue parent = existing_by_code.get(parent_code) if parent is not None: organization.parent = parent self.db.flush() def _seed_employees(self) -> None: employees_by_no = { employee.employee_no: employee for employee in self.repository.list() } roles_by_code = {role.role_code: role for role in self.repository.list_roles()} organizations_by_code = { unit.unit_code: unit for unit in self.repository.list_organization_units() } for definition in SEEDED_EMPLOYEE_DEFINITIONS: employee_no = definition["employee_no"] if employee_no in employees_by_no: continue employee = Employee( employee_no=employee_no, name=definition["name"], email=definition["email"], gender=definition.get("gender"), birth_date=self._parse_date(definition.get("birth_date")), phone=definition.get("phone"), join_date=self._parse_date(definition.get("join_date")), location=definition.get("location"), position=definition.get("position", "员工"), grade=definition.get("grade", "P3"), cost_center=definition.get("cost_center"), finance_owner_name=definition.get("finance_owner_name"), employment_status=definition.get("employment_status", "在职"), sync_state=definition.get("sync_state", "已同步"), spotlight=bool(definition.get("spotlight")), last_sync_at=self._parse_datetime(definition.get("last_sync_at")), updated_at=self._parse_datetime(definition.get("updated_at")), ) self.db.add(employee) employees_by_no[employee_no] = employee self.db.flush() for definition in SEEDED_EMPLOYEE_DEFINITIONS: employee = employees_by_no[definition["employee_no"]] organization_code = definition.get("organization_unit_code") manager_employee_no = definition.get("manager_employee_no") if employee.organization_unit_id is None and organization_code: employee.organization_unit = organizations_by_code.get(organization_code) if employee.manager_id is None and manager_employee_no: employee.manager = employees_by_no.get(manager_employee_no) if not employee.password_hash: employee.password_hash = hash_password(DEFAULT_EMPLOYEE_PASSWORD) if not employee.roles: employee.roles = self._sorted_roles( [ roles_by_code[role_code] for role_code in definition.get("role_codes", []) if role_code in roles_by_code ] ) self._seed_employee_history(employee, definition) self.db.flush() def apply_profile_repairs(self) -> None: """Apply one-off demo profile repairs. Intended for startup/bootstrap only.""" try: self._repair_employee_profiles() self._trim_all_employee_change_logs() self.db.commit() except Exception: self.db.rollback() logger.exception("Failed to apply employee profile repairs") raise def _repair_employee_profiles(self) -> None: if not EMPLOYEE_PROFILE_REPAIRS: return employees = self.repository.list() employees_by_email = {employee.email.lower(): employee for employee in employees if employee.email} employees_by_no = {employee.employee_no: employee for employee in employees if employee.employee_no} roles_by_code = {role.role_code: role for role in self.repository.list_roles()} organizations_by_code = { unit.unit_code: unit for unit in self.repository.list_organization_units() } for definition in EMPLOYEE_PROFILE_REPAIRS: email = str(definition.get("email") or "").strip().lower() employee_no = str(definition.get("employee_no") or "").strip() employee = employees_by_email.get(email) or employees_by_no.get(employee_no) if employee is None: continue for field_name in ( "position", "grade", "location", "cost_center", "finance_owner_name", "employment_status", "sync_state", ): value = definition.get(field_name) if value: setattr(employee, field_name, value) organization_code = definition.get("organization_unit_code") if organization_code: employee.organization_unit = organizations_by_code.get(organization_code) manager_employee_no = definition.get("manager_employee_no") if manager_employee_no: employee.manager = employees_by_no.get(manager_employee_no) if not employee.password_hash: employee.password_hash = hash_password(DEFAULT_EMPLOYEE_PASSWORD) role_codes = [item for item in definition.get("role_codes", []) if item in roles_by_code] if role_codes: merged_roles = {role.role_code: role for role in employee.roles} for role_code in role_codes: merged_roles[role_code] = roles_by_code[role_code] employee.roles = self._sorted_roles(list(merged_roles.values())) self.db.flush() def _prune_extra_seed_employees(self) -> None: if not EXTRA_SEED_EMPLOYEE_NOS: return for employee_no in EXTRA_SEED_EMPLOYEE_NOS: employee = self.repository.get_by_employee_no(employee_no) if employee is not None: self.db.delete(employee) def _ensure_employee_schema(self) -> None: bind = self.db.get_bind() inspector = inspect(bind) if "employees" not in inspector.get_table_names(): return column_names = {column["name"] for column in inspector.get_columns("employees")} if "password_hash" not in column_names: self.db.execute(text("ALTER TABLE employees ADD COLUMN password_hash VARCHAR(255)")) self.db.flush() def _seed_employee_history(self, employee: Employee, definition: dict[str, Any]) -> None: existing_keys = { (item.action, item.owner, self._format_datetime(item.occurred_at)) for item in employee.change_logs } history_items = list(definition.get("history", [])) if not history_items: history_items = [ { "action": "初始化员工档案", "owner": "系统初始化任务", "occurred_at": definition.get("updated_at") or definition.get("last_sync_at"), } ] for history in history_items: occurred_at = self._parse_datetime(history.get("occurred_at")) if occurred_at is None: continue identity = ( history["action"], history["owner"], self._format_datetime(occurred_at), ) if identity in existing_keys: continue self.db.add( EmployeeChangeLog( employee=employee, action=history["action"], owner=history["owner"], occurred_at=occurred_at, ) ) existing_keys.add(identity) def _save_employee_and_reload(self, employee: Employee) -> Employee: saved = self.repository.save(employee) self._trim_employee_change_logs(saved.id) self.db.commit() return self.repository.get(saved.id) or saved def _append_change_log( self, employee: Employee, action: str, owner: str = "系统管理员", occurred_at: datetime | None = None, ) -> None: self.db.add( EmployeeChangeLog( employee=employee, action=action, owner=owner, occurred_at=occurred_at or datetime.now(UTC), ) ) def _trim_all_employee_change_logs(self) -> None: for employee in self.repository.list(): self._trim_employee_change_logs(employee.id) def _sorted_change_logs(self, employee: Employee) -> list[EmployeeChangeLog]: return sorted(employee.change_logs, key=lambda item: item.occurred_at, reverse=True) def _trim_employee_change_logs(self, employee_id: str) -> None: stmt = ( select(EmployeeChangeLog) .where(EmployeeChangeLog.employee_id == employee_id) .order_by(EmployeeChangeLog.occurred_at.desc()) ) logs = list(self.db.execute(stmt).scalars().all()) if len(logs) <= MAX_EMPLOYEE_CHANGE_LOGS: return for stale in logs[MAX_EMPLOYEE_CHANGE_LOGS:]: self.db.delete(stale) def _serialize_employee(self, employee: Employee) -> EmployeeRead: organization = employee.organization_unit roles = self._sorted_roles(list(employee.roles)) role_labels = [role.name for role in roles] role_codes = [role.role_code for role in roles] history = [ EmployeeHistoryRead( action=item.action, owner=item.owner, time=self._format_history_datetime(item.occurred_at), occurredAt=self._format_history_datetime(item.occurred_at), ) for item in self._sorted_change_logs(employee)[:MAX_EMPLOYEE_CHANGE_LOGS] ] return EmployeeRead( id=employee.id, avatar=(employee.name or "?")[:1], name=employee.name, employeeNo=employee.employee_no, department=organization.name if organization else "", position=employee.position, grade=employee.grade, manager=employee.manager.name if employee.manager else "CEO", managerEmployeeNo=employee.manager.employee_no if employee.manager else None, financeOwner=employee.finance_owner_name or "", roles=role_labels, roleCodes=role_codes, status=employee.employment_status, statusTone=STATUS_TONE_MAP.get(employee.employment_status, "neutral"), gender=employee.gender, age=self._calculate_age(employee.birth_date), birthDate=self._format_date(employee.birth_date), email=employee.email, phone=employee.phone, joinDate=self._format_date(employee.join_date), location=employee.location, costCenter=employee.cost_center, updatedAt=self._format_datetime(employee.updated_at or employee.created_at), lastSync=self._format_datetime(employee.last_sync_at), syncState=employee.sync_state, spotlight=employee.spotlight, permissions=self._collect_permissions(role_codes), history=history, organization=( EmployeeOrganizationRead( id=organization.id, code=organization.unit_code, name=organization.name, unitType=organization.unit_type, costCenter=organization.cost_center, location=organization.location, managerName=organization.manager_name, ) if organization else None ), ) def _collect_permissions(self, role_codes: list[str]) -> list[str]: permissions: list[str] = [] seen: set[str] = set() for role_code in role_codes: for permission in ROLE_PERMISSION_MAP.get(role_code, []): if permission in seen: continue permissions.append(permission) seen.add(permission) return permissions def _sorted_roles(self, roles: list[Role]) -> list[Role]: return sorted(roles, key=lambda item: (ROLE_DISPLAY_ORDER.get(item.role_code, 999), item.name)) @staticmethod def _normalize_optional_text(value: str | None) -> str | None: if value is None: return None text_value = value.strip() return text_value or None @staticmethod def _parse_date(value: str | None) -> date | None: if not value: return None return datetime.strptime(value, "%Y-%m-%d").date() @staticmethod def _parse_datetime(value: str | datetime | None) -> datetime | None: if value is None: return None if isinstance(value, datetime): return value return datetime.strptime(value, "%Y-%m-%d %H:%M") @staticmethod def _format_date(value: date | None) -> str | None: if value is None: return None return value.strftime("%Y-%m-%d") @staticmethod def _to_display_datetime(value: datetime) -> datetime: if value.tzinfo is None: normalized = value.replace(tzinfo=UTC) else: normalized = value.astimezone(UTC) return normalized.astimezone(DISPLAY_TIMEZONE) @staticmethod def _format_datetime(value: datetime | None) -> str | None: if value is None: return None local = EmployeeService._to_display_datetime(value) return local.strftime("%Y-%m-%d %H:%M") @staticmethod def _format_history_datetime(value: datetime | None) -> str: if value is None: return "" local = EmployeeService._to_display_datetime(value) return ( f"{local.year}年{local.month}月{local.day}日" f"{local.hour}时{local.minute}分" ) @staticmethod def _calculate_age(birth_date: date | None) -> int | None: if birth_date is None: return None today = date.today() age = today.year - birth_date.year if (today.month, today.day) < (birth_date.month, birth_date.day): age -= 1 return age