"""工号分配台账 — hr_employee_numbers + one-time seed from identity_persons. Revision ID: 249 Revises: 248 Create Date: 2026-06-04 Adds the 工号系统 (employee-number allocation) feature: * ``hr_employee_numbers`` — standalone allocation ledger. ``employee_no`` is the user-facing unique key (format ``P`` + zero-padded-5, e.g. ``P02388``); ``seq`` is the integer behind it that drives ``MAX(seq)+1`` allocation. NOT linked to a person — HR allocates a number to a name before the person ever exists in HRIS. See the model docstring in ``app/domains/hr/models.py`` for the full rationale. * One-time seed: every distinct ``identity_persons.employee_no`` (non-NULL, non-hard-deleted) is copied in as a ``source='seed'`` row so the ledger starts as the complete history and ``MAX(seq)+1`` continues from the largest existing ``P`` value. Values that don't match ``^P\\d+$`` (legacy formats) are copied verbatim with ``seq=NULL`` — kept for the record, excluded from the allocation math. Idempotent: the seed only runs when the table is empty, so re-running the migration (or a later manual top-up) never double-inserts. Idempotent: table creation is guarded with information_schema; an earlier VARCHAR-id cut of the table is dropped so the corrected CHAR(36) CREATE converges (mirrors migration 235's TiDB-FK cleanup). Downgrade drops the table. """ from __future__ import annotations import re import uuid import sqlalchemy as sa from alembic import op from app.core.migration_helpers import table_kw_for revision = "249" down_revision = "248" branch_labels = None depends_on = None _P_NUMERIC = re.compile(r"^P(\d+)$") def _table_exists(table: str) -> bool: conn = op.get_bind() result = conn.execute( sa.text( "SELECT COUNT(*) FROM information_schema.tables " "WHERE table_schema = DATABASE() AND table_name = :t" ), {"t": table}, ) return (result.scalar() or 0) > 0 def _has_varchar_id(table: str) -> bool: """Detect a prior-bad cut where id was declared VARCHAR(36). Local CI MySQL accepts FK VARCHAR(36)→CHAR(36); prod TiDB rejects it (3780). On a clean deploy the table doesn't exist → returns False.""" conn = op.get_bind() row = conn.execute( sa.text( "SELECT DATA_TYPE FROM information_schema.columns " "WHERE table_schema = DATABASE() AND table_name = :t " "AND column_name = 'id' LIMIT 1" ), {"t": table}, ).scalar() return bool(row) and str(row).lower() == "varchar" def upgrade() -> None: # ── 0. TiDB-FK-strictness cleanup ───────────────────────────── # Table is empty on first deploy + DB is the only data source, so a # drop loses nothing. The CREATE below uses CHAR(36) matching # identity_persons.id on prod TiDB. if _has_varchar_id("hr_employee_numbers"): op.execute("DROP TABLE IF EXISTS `hr_employee_numbers`") # ── 1. hr_employee_numbers ──────────────────────────────────── if not _table_exists("hr_employee_numbers"): op.create_table( "hr_employee_numbers", sa.Column("id", sa.CHAR(length=36), nullable=False), sa.Column("employee_no", sa.String(length=16), nullable=False), sa.Column("seq", sa.Integer(), nullable=True), sa.Column("name", sa.String(length=128), nullable=True), sa.Column("hire_date", sa.Date(), nullable=True), sa.Column("remark", sa.String(length=255), nullable=True), sa.Column( "source", sa.String(length=16), nullable=False, server_default="manual", ), sa.Column( "created_by", sa.CHAR(length=36), sa.ForeignKey("identity_persons.id"), nullable=True, ), sa.Column( "created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), ), sa.Column( "updated_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), server_onupdate=sa.func.now(), ), sa.Column("deleted_at", sa.DateTime(timezone=True), nullable=True), sa.PrimaryKeyConstraint("id"), **table_kw_for("identity_persons"), ) op.create_index( "uq_hr_employee_numbers_no", "hr_employee_numbers", ["employee_no"], unique=True, ) op.create_index("ix_hr_employee_numbers_seq", "hr_employee_numbers", ["seq"]) op.create_index( "ix_hr_employee_numbers_created_at", "hr_employee_numbers", ["created_at"] ) op.create_index( "ix_hr_employee_numbers_hire_date", "hr_employee_numbers", ["hire_date"] ) op.create_index( "ix_hr_employee_numbers_source", "hr_employee_numbers", ["source"] ) # ── 2. One-time seed from identity_persons ──────────────────── # Only when the ledger is empty, so re-running this migration (or a # later manual top-up) never double-inserts. conn = op.get_bind() already = conn.execute( sa.text("SELECT COUNT(*) FROM hr_employee_numbers") ).scalar() or 0 if already == 0: # Distinct employee_no across non-hard-deleted persons. MIN(...) on # the descriptive columns just picks a deterministic representative # if (defensively) a number ever maps to more than one row. rows = conn.execute( sa.text( "SELECT employee_no, " " MIN(display_name) AS display_name, " " MIN(hire_date) AS hire_date " "FROM identity_persons " "WHERE employee_no IS NOT NULL AND employee_no <> '' " " AND deleted_at IS NULL " "GROUP BY employee_no" ) ).fetchall() seen_nos: set[str] = set() for row in rows: raw_no = str(row.employee_no).strip() if not raw_no: continue m = _P_NUMERIC.match(raw_no) if m: seq = int(m.group(1)) # Canonicalize to the padded form (P2388 -> P02388) so the # ledger's unique key matches what the service's # canonicalize() produces — otherwise an unpadded seed value # and a later canonical allocation would be two rows sharing # one seq. employee_no = f"P{seq:05d}" else: seq = None employee_no = raw_no # Dedupe within the seed itself: two source values that # canonicalize to the same number (e.g. 'P2388' + 'P02388') # would collide on the unique index — keep the first. if employee_no in seen_nos: continue seen_nos.add(employee_no) conn.execute( sa.text( "INSERT INTO hr_employee_numbers " "(id, employee_no, seq, name, hire_date, remark, source, " " created_by, created_at, updated_at) " "VALUES " "(:id, :no, :seq, :name, :hire_date, NULL, 'seed', " " NULL, NOW(), NOW())" ), { "id": str(uuid.uuid4()), "no": employee_no, "seq": seq, "name": row.display_name, "hire_date": row.hire_date, }, ) def downgrade() -> None: if _table_exists("hr_employee_numbers"): op.drop_table("hr_employee_numbers")