"""HR leave / time-off mirror — accounts + postings + daily sync cron. Promotes SF EC Time-Off (TimeAccount / TimeAccountDetail) into Helix as first-class, self-service-visible data instead of the four ``identity_persons.annual_leave_*`` snapshot columns (which only ever covered 法定年假 + 福利年假 carry/used). Tables: * hr_leave_accounts — one row per SF TimeAccount (person, accountType, fiscal-year bucket); carries derived balance/entitled/used days. * hr_leave_postings — one row per SF TimeAccountDetail (ledger line); carries amount, running balance_after, creator, comment. Convention (mirrors migration 322): ``person_id`` is a plain id column (NO FK to identity_persons — TiDB-FK-safe), sized via ``id_type_for('identity_persons')``. The only real FK is the intra- feature posting → account ref; both tables share the parent charset via ``table_kw_for`` so that FK is type-compatible on TiDB. Also seeds the daily ``hr.leave_account_sync`` cron (single global row; the handler fans out across every enabled HRIS instance internally). Revision ID: 324 Revises: 323 """ from __future__ import annotations import json as _json import uuid import sqlalchemy as sa from alembic import op from app.core.migration_helpers import id_type_for, table_kw_for revision = "324" down_revision = "323" branch_labels = None depends_on = None ACCOUNTS = "hr_leave_accounts" POSTINGS = "hr_leave_postings" HANDLER_KEY = "hr.leave_account_sync" JOB_NAME = "hr.leave_account_sync" _DAYS = sa.Numeric(8, 2) _JOB_DESC = ( "每日从 SuccessFactors EC Time-Off 拉取所有员工的「假期余额」(TimeAccount)与" "「账户假期记录」(TimeAccountDetail),镜像进 hr_leave_accounts / hr_leave_postings," "并据此派生回 identity_persons 的四个 annual_leave_* 快照列(年假计算器仍可用)。" "数据驱动:不再只取法定年假/福利年假两种,而是同步该员工的全部假期账户类型" "(法定年假/福利年假/带薪病假/调休假等,逐账户独立成行)。单例任务,内部遍历所有" "enabled 的 HRIS 实例;按 external_code 幂等 upsert,SF 不再返回的行做软删除。" "EC Time-Off 未授权或鉴权失败时优雅降级为空,不影响其余同步。" ) def _table_exists(bind, name: str) -> bool: return bind.execute( sa.text( "SELECT 1 FROM information_schema.tables " "WHERE table_schema = DATABASE() AND table_name = :n" ), {"n": name}, ).scalar() is not None def upgrade() -> None: bind = op.get_bind() person_id_type = id_type_for("identity_persons") tkw = table_kw_for("identity_persons") if not _table_exists(bind, ACCOUNTS): op.create_table( ACCOUNTS, sa.Column("id", person_id_type, nullable=False), sa.Column("external_code", sa.String(64), nullable=False), sa.Column("person_id", person_id_type, nullable=False), sa.Column("hris_person_id", sa.String(64), nullable=True), sa.Column("hris_instance_id", sa.String(36), nullable=True), sa.Column("account_type", sa.String(64), nullable=False), sa.Column("account_type_label", sa.String(128), nullable=True), sa.Column("fiscal_year", sa.Integer, nullable=True), sa.Column("start_date", sa.Date, nullable=True), sa.Column("end_date", sa.Date, nullable=True), sa.Column("booking_start_date", sa.Date, nullable=True), sa.Column("booking_end_date", sa.Date, nullable=True), sa.Column("account_closed", sa.Boolean, nullable=False, server_default="0"), sa.Column("balance_days", _DAYS, nullable=True), sa.Column("entitled_days", _DAYS, nullable=True), sa.Column("used_days", _DAYS, nullable=True), sa.Column("is_current", sa.Boolean, nullable=False, server_default="0"), sa.Column("synced_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()), 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()), sa.Column("deleted_at", sa.DateTime(timezone=True), nullable=True), sa.PrimaryKeyConstraint("id"), sa.UniqueConstraint("external_code", name="uq_hr_leave_accounts_external_code"), **tkw, ) op.create_index("ix_hr_leave_accounts_person", ACCOUNTS, ["person_id"]) op.create_index("ix_hr_leave_accounts_hris_person", ACCOUNTS, ["hris_person_id"]) op.create_index("ix_hr_leave_accounts_type", ACCOUNTS, ["account_type"]) if not _table_exists(bind, POSTINGS): op.create_table( POSTINGS, sa.Column("id", person_id_type, nullable=False), sa.Column("external_code", sa.String(64), nullable=False), sa.Column("leave_account_id", person_id_type, nullable=False), sa.Column("account_external_code", sa.String(64), nullable=True), sa.Column("person_id", person_id_type, nullable=False), sa.Column("booking_type", sa.String(32), nullable=False), sa.Column("booking_amount", _DAYS, nullable=False), sa.Column("booking_unit", sa.String(16), nullable=True), sa.Column("booking_date", sa.Date, nullable=True), sa.Column("accrual_period_id", sa.String(64), nullable=True), sa.Column("balance_after", sa.Numeric(9, 2), nullable=True), sa.Column("created_by_source", sa.String(128), nullable=True), sa.Column("created_at_source", sa.DateTime(timezone=True), nullable=True), sa.Column("comment", sa.Text, nullable=True), sa.Column("synced_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()), 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()), sa.Column("deleted_at", sa.DateTime(timezone=True), nullable=True), sa.PrimaryKeyConstraint("id"), sa.UniqueConstraint("external_code", name="uq_hr_leave_postings_external_code"), # Real intra-feature FK — both ends share the parent charset # (table_kw_for) so TiDB accepts it. sa.ForeignKeyConstraint( ["leave_account_id"], [f"{ACCOUNTS}.id"], name="fk_hr_leave_postings_account", ondelete="CASCADE", ), **tkw, ) op.create_index("ix_hr_leave_postings_account", POSTINGS, ["leave_account_id"]) op.create_index("ix_hr_leave_postings_person_date", POSTINGS, ["person_id", "booking_date"]) # Seed the daily sync cron (idempotent). if _table_exists(bind, "scheduled_jobs"): bind.execute( sa.text( "INSERT INTO scheduled_jobs " "(id, name, display_name, description, handler_key, " " trigger_type, cron_expr, timezone, enabled, config, " " run_count, failure_count, created_at, updated_at) " "SELECT :id, :name, :display, :desc, :handler, 'cron', " ":cron, 'Asia/Shanghai', 1, :config, 0, 0, NOW(), NOW() " "FROM DUAL WHERE NOT EXISTS " "(SELECT 1 FROM scheduled_jobs WHERE handler_key = :handler)" ), { "id": str(uuid.uuid4()), "name": JOB_NAME, "display": "假期余额与记录 · 每日同步", "desc": _JOB_DESC, "handler": HANDLER_KEY, # 06:30 SH — after the early HRIS / identity ticks so the # person roster is fresh before leave data is keyed to it. "cron": "30 6 * * *", "config": _json.dumps({}), }, ) def downgrade() -> None: bind = op.get_bind() if _table_exists(bind, "scheduled_jobs"): bind.execute( sa.text("DELETE FROM scheduled_jobs WHERE handler_key = :k"), {"k": HANDLER_KEY}, ) if _table_exists(bind, POSTINGS): op.drop_table(POSTINGS) if _table_exists(bind, ACCOUNTS): op.drop_table(ACCOUNTS)