from __future__ import annotations from collections import Counter from datetime import UTC, date, datetime from typing import Any 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, EmployeeImportResultRead, EmployeeMetaRead, EmployeeOrganizationRead, EmployeeRead, EmployeeRoleOptionRead, EmployeeStatusSummaryRead, EmployeeUpdate, ) from app.services.employee_import import EmployeeImportCoordinator from app.services.employee_serialization import serialize_employee from app.services.employee_spreadsheet import build_import_template_bytes from app.services.employee_seed import ( CANONICAL_DEPARTMENT_CODES, EMPLOYEE_DEFINITIONS, EMPLOYEE_PROFILE_REPAIRS, LEGACY_ORGANIZATION_UNIT_CODE_MAP, ORGANIZATION_DEFINITIONS, ROLE_DEFINITIONS, ROLE_DISPLAY_ORDER, ROLE_PERMISSION_MAP, normalize_organization_unit_code, ) from app.services.employee_time import ( format_date, format_datetime, format_history_datetime, normalize_optional_text, parse_date, parse_datetime, ) logger = get_logger("app.services.employee") DEFAULT_EMPLOYEE_PASSWORD = "123456" MAX_EMPLOYEE_CHANGE_LOGS = 5 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._normalize_legacy_employee_departments() 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 ] visible_role_codes = {item["role_code"] for item in ROLE_DEFINITIONS} 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()) if role.role_code in visible_role_codes ] canonical_department_codes = set(CANONICAL_DEPARTMENT_CODES) 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( ( unit for unit in self.repository.list_organization_units() if unit.unit_code in canonical_department_codes ), 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: organization_code = normalize_organization_unit_code(payload.organization_unit_code) employee.organization_unit = self.repository.get_organization_by_code(organization_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 = 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 = 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 = 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 = 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 = 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 = normalize_organization_unit_code( 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 = 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() return self._import_coordinator().export_employees(status=status, keyword=keyword) def import_employees(self, content: bytes, actor: str = "系统管理员") -> EmployeeImportResultRead: self.ensure_directory_ready() return self._import_coordinator().import_employees(content, actor=actor) def _import_coordinator(self) -> EmployeeImportCoordinator: return EmployeeImportCoordinator( self.db, self.repository, sorted_roles=self._sorted_roles, append_change_log=self._append_change_log, format_date=format_date, format_datetime=format_datetime, default_password=DEFAULT_EMPLOYEE_PASSWORD, ) def _seed_roles(self) -> None: existing_by_code = {role.role_code: role for role in self.repository.list_roles()} legacy_auditor = existing_by_code.get("auditor") if legacy_auditor is not None and "budget_monitor" not in existing_by_code: legacy_auditor.role_code = "budget_monitor" existing_by_code["budget_monitor"] = legacy_auditor existing_by_code.pop("auditor", None) 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 else: role.name = definition["name"] role.description = definition["description"] 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 else: organization.name = definition["name"] organization.unit_type = definition["unit_type"] organization.cost_center = definition.get("cost_center") organization.location = definition.get("location") organization.manager_name = definition.get("manager_name") 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"]] parent = existing_by_code.get(parent_code) if parent is not None and organization.parent_id != parent.id: organization.parent = parent self.db.flush() def _normalize_legacy_employee_departments(self) -> None: if not LEGACY_ORGANIZATION_UNIT_CODE_MAP: return organizations_by_code = { unit.unit_code: unit for unit in self.repository.list_organization_units() } for employee in self.repository.list(): current_code = ( employee.organization_unit.unit_code if employee.organization_unit else None ) next_code = normalize_organization_unit_code(current_code) if not next_code or next_code == current_code: continue organization = organizations_by_code.get(next_code) if organization is not None: employee.organization_unit = organization 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=parse_date(definition.get("birth_date")), phone=definition.get("phone"), join_date=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=parse_datetime(definition.get("last_sync_at")), updated_at=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)")) if "compliance_score" not in column_names: self.db.execute( text("ALTER TABLE employees ADD COLUMN compliance_score INTEGER DEFAULT 100 NOT NULL") ) self.db.flush() def _seed_employee_history(self, employee: Employee, definition: dict[str, Any]) -> None: existing_keys = { (item.action, item.owner, 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 = parse_datetime(history.get("occurred_at")) if occurred_at is None: continue identity = ( history["action"], history["owner"], 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: return serialize_employee( employee, sorted_roles=self._sorted_roles(list(employee.roles)), sorted_change_logs=self._sorted_change_logs(employee), format_date=format_date, format_datetime=format_datetime, format_history_datetime=format_history_datetime, role_permission_map=ROLE_PERMISSION_MAP, status_tone_map=STATUS_TONE_MAP, max_change_logs=MAX_EMPLOYEE_CHANGE_LOGS, ) 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))