X-Financial/server/src/app/services/employee.py

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