from __future__ import annotations from collections import Counter from datetime import UTC, date, datetime from typing import Any from sqlalchemy import select 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_bank_info import apply_default_bank_info from app.services.employee_schema import ensure_employee_schema 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()) ensure_employee_schema(self.db) self._prune_extra_seed_employees() self._seed_roles() self._seed_organization_units() self._seed_employees() self._normalize_legacy_employee_departments() self._backfill_employee_bank_info() 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, bank_name=normalize_optional_text(payload.bank_name), bank_account_no=normalize_optional_text(payload.bank_account_no), bank_account_name=normalize_optional_text(payload.bank_account_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), ) apply_default_bank_info(employee) 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 "bank_account_name" in payload.model_fields_set: bank_account_name = normalize_optional_text(payload.bank_account_name) if bank_account_name != employee.bank_account_name: employee.bank_account_name = bank_account_name changed_fields.append("银行户名") if "bank_name" in payload.model_fields_set: bank_name = normalize_optional_text(payload.bank_name) if bank_name != employee.bank_name: employee.bank_name = bank_name changed_fields.append("开户行") if "bank_account_no" in payload.model_fields_set: bank_account_no = normalize_optional_text(payload.bank_account_no) if bank_account_no != employee.bank_account_no: employee.bank_account_no = bank_account_no 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"), bank_name=definition.get("bank_name"), bank_account_no=definition.get("bank_account_no"), bank_account_name=definition.get("bank_account_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) apply_default_bank_info(employee) 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", "bank_name", "bank_account_no", "bank_account_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) apply_default_bank_info(employee) 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 _backfill_employee_bank_info(self) -> None: for employee in self.repository.list(): apply_default_bank_info(employee) 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))