"""Seed identity_sync_feishu integration + feishu_contact sync target. Revision ID: 215 Revises: 214 Create Date: 2026-05-29 What this migration does ------------------------ Idempotent seed so every environment (local, prod, fresh dev) lands fully configured the moment migration 215 runs — admin does NOT need to: * create the integration row by hand * paste app_id / app_secret * pick a preset * click "save target" The seed: 1. Looks up the existing ``feishu-primary`` integration (kind= ``chat_approval``, provider=``feishu``). When present + has credentials, re-uses its ``app_id`` and ``app_secret`` for the new identity-sync integration row — same self-built Feishu app backs both surfaces. 2. Creates ``feishu-contact-primary`` integration (kind= ``identity_sync_feishu``, provider=``feishu``, enabled=1 because the integration itself does nothing dangerous — only the target's ``enabled`` flag governs whether a sweep fires). 3. Creates ``feishu-contact`` identity_sync_target from the ``feishu_contact`` preset (defined in ``app.domains.identity_sync.presets``). **``enabled=0`` by default so no Feishu writes happen until admin explicitly turns it on from the UI** — this is the explicit user safety requirement ("开关关上 等我一个一个的测试好了再说"). Failure modes ------------- * If ``feishu-primary`` doesn't exist (clean dev DB), the integration row is created with NULL credentials — admin fills them in via the admin UI when ready. Less ideal but never blocks the migration. * If the existing ``feishu-primary`` row uses legacy no-AAD encryption (pre-migration-125), the decrypt path transparently falls back; we re-encrypt with the new AAD bound to the new row's id so the seeded row is on the modern path. * All inserts guard via ``slug``-uniqueness probes — re-running this migration on top of an already-seeded DB is a no-op. This is INTENTIONALLY a code-driven seed (not a fixture file) — the preset's mapping_rules are the source of truth and the seed re-reads them at migration time so future changes to the preset show up in fresh deploys without a follow-up migration. """ from __future__ import annotations import json import logging import uuid import sqlalchemy as sa from alembic import op logger = logging.getLogger(__name__) revision = "215" down_revision = "214" branch_labels = None depends_on = None _NEW_INTEGRATION_SLUG = "feishu-contact-primary" _NEW_TARGET_SLUG = "feishu-contact" def _row_exists(bind, table: str, slug: str) -> str | None: """Return the row id if the slug exists, else None.""" row = bind.execute( sa.text(f"SELECT id FROM {table} WHERE slug = :s LIMIT 1"), {"s": slug}, ).fetchone() return row[0] if row else None def _resolve_feishu_credentials(bind) -> tuple[str | None, str | None]: """Decrypt app_id + app_secret from the existing ``feishu-primary`` chat_approval integration so we can re-encrypt them under the new row's AAD. Returns ``(app_id, app_secret)`` — either or both may be None when the source row is missing / not yet configured. """ src = bind.execute( sa.text( "SELECT id, config_public, secrets_encrypted, secrets_nonce " "FROM integrations WHERE slug = 'feishu-primary' LIMIT 1" ) ).fetchone() if not src: return None, None src_id, config_public, ciphertext, nonce = src app_id: str | None = None if config_public: try: if isinstance(config_public, str): cp = json.loads(config_public) else: cp = config_public if isinstance(cp, dict): v = cp.get("app_id") if isinstance(v, str) and v: app_id = v except (ValueError, TypeError) as e: logger.warning( "migration 215: feishu-primary config_public is " "malformed (%s: %s) — seeded row will have empty " "app_id; admin must paste it in the UI.", type(e).__name__, e, ) app_secret: str | None = None if ciphertext and nonce: try: # Imported lazily — migrations should not import app code at # module level (alembic env may load them outside the FastAPI # process). At call time the credential_crypto helpers are # safe sync utilities. from app.domains.cmdb.credential_crypto import ( CredentialDecryptionError, decrypt_fields, ) fields = decrypt_fields( ciphertext, nonce, integration_id=str(src_id), ) if isinstance(fields, dict): v = fields.get("app_secret") if isinstance(v, str) and v: app_secret = v except CredentialDecryptionError as e: # Known crypto failure shape — log + leave app_secret=None. # Admin will paste credentials via the integration edit # form. Migration MUST NOT raise on this path. logger.warning( "migration 215: decrypt of feishu-primary secret failed " "(%s) — seeded integration row will land without creds; " "admin must paste app_secret in the UI.", e, ) except (ValueError, RuntimeError, TypeError) as e: # Surface unexpected decrypt-related errors loudly so an # operator can diagnose during a manual migration run. The # bare ``except Exception`` it replaced was hiding genuine # bugs (key misconfig, JSON parse errors, etc). logger.warning( "migration 215: unexpected decrypt error (%s: %s) — " "seeded integration row will land without creds.", type(e).__name__, e, ) return app_id, app_secret def _encrypt_app_secret(secret: str, integration_id: str) -> tuple[str | None, str | None]: try: from app.domains.cmdb.credential_crypto import encrypt_fields ct, nonce = encrypt_fields( {"app_secret": secret}, integration_id=integration_id, ) return ct, nonce except (ValueError, RuntimeError, TypeError) as e: logger.warning( "migration 215: app_secret re-encryption for integration " "%s failed (%s: %s) — seeded row will land WITHOUT " "encrypted creds and ``enabled=0``; admin must paste " "app_secret in the UI before enabling.", integration_id, type(e).__name__, e, ) return None, None def _build_preset_target_payload() -> dict[str, object]: """Read the live ``feishu_contact`` preset and serialise its mapping_rules / filter / schedule / runner_options to JSON-ready dicts for the INSERT. """ try: from app.domains.identity_sync.presets import get_preset p = get_preset("feishu_contact") if p is None: raise RuntimeError("preset feishu_contact not found") return { "mapping_config": {"rules": [dict(r) for r in p.mapping_rules]}, "filter_config": dict(p.filter_config), "schedule_config": dict(p.schedule_config), "runner_options": dict(p.runner_options), } except Exception as exc: # noqa: BLE001 # Migration MUST NOT raise. Fall back to a minimal-valid target # payload that the admin will customize. return { "mapping_config": {"rules": []}, "filter_config": { "include_offboarded": False, "departments": [], "include_subtree": True, "countries": [], "employment_statuses": ["active", "on_leave"], "job_title_includes": [], "email_suffixes": [], "person_ids": [], "mode": "and", }, "schedule_config": { "trigger": "cron", "cron_expr": "0 2 * * *", "timezone": "Asia/Shanghai", }, "runner_options": {}, "_seed_warning": f"preset load failed: {exc!r}", } def upgrade() -> None: bind = op.get_bind() # ── (1) Seed integration row (if absent) ───────────────────────── existing_integ_id = _row_exists(bind, "integrations", _NEW_INTEGRATION_SLUG) if existing_integ_id is None: new_integ_id = str(uuid.uuid4()) app_id, app_secret = _resolve_feishu_credentials(bind) ct, nonce = (None, None) if app_secret: ct, nonce = _encrypt_app_secret(app_secret, new_integ_id) config_public = { "app_id": app_id or "", } # Only flip enabled=1 when we actually inherited credentials # AND the re-encryption under the new row's AAD succeeded. # Either failure path → enabled=0 so a broken row doesn't get # picked up by "find primary identity_sync_feishu integration" # queries before admin pastes credentials manually. has_credentials = bool(app_id and app_secret and ct and nonce) if not has_credentials: logger.warning( "migration 215: feishu-primary credentials not " "available (fresh DB or chat_approval not configured) " "— seeding integration row with enabled=0; admin must " "paste app_id+app_secret and enable manually.", ) bind.execute( sa.text( "INSERT INTO integrations " "(id, kind, provider, name, slug, description, " " enabled, is_primary, " " config_public, secrets_encrypted, secrets_nonce, " " created_at, updated_at) " "VALUES " "(:id, 'identity_sync_feishu', 'feishu', :name, :slug, " " :desc, :enabled, 0, " " :public, :ct, :nonce, " " NOW(), NOW())" ), { "id": new_integ_id, "name": "飞书通讯录 (identity_sync)", "slug": _NEW_INTEGRATION_SLUG, "desc": ( "Outbound HRIS→飞书 通讯录同步。复用与「飞书审批」连接器同一个" "自建应用的 app_id / app_secret。需要的飞书 OAuth 权限请到本" "连接器的编辑表单 → 测试连接 跑一次 per-scope 探测,缺哪个" "scope 报告会指出。" ), "enabled": 1 if has_credentials else 0, "public": json.dumps(config_public, ensure_ascii=False), "ct": ct, "nonce": nonce, }, ) target_integration_id = new_integ_id else: target_integration_id = existing_integ_id # ── (2) Seed identity_sync_target (if absent) ──────────────────── existing_target_id = _row_exists( bind, "identity_sync_targets", _NEW_TARGET_SLUG, ) if existing_target_id is None: payload = _build_preset_target_payload() bind.execute( sa.text( "INSERT INTO identity_sync_targets " "(id, name, slug, description, kind, integration_id, " " enabled, on_offboard, " " mapping_config, filter_config, schedule_config, " " runner_options, " " created_at, updated_at) " "VALUES " "(:id, :name, :slug, :desc, 'feishu_contact', :integ, " " 0, 'deactivate', " " :mapping, :filter, :schedule, " " :runner_opts, " " NOW(), NOW())" ), { "id": str(uuid.uuid4()), "name": "飞书通讯录同步 (HRIS → 飞书)", "slug": _NEW_TARGET_SLUG, "desc": ( "Phase A 部门 + Phase B 用户 一体同步。默认 disabled," "确认 test_connection、dry-run、单人手动 sync 都正常后再 enable。" ), "integ": target_integration_id, "mapping": json.dumps( payload["mapping_config"], ensure_ascii=False, ), "filter": json.dumps( payload["filter_config"], ensure_ascii=False, ), "schedule": json.dumps( payload["schedule_config"], ensure_ascii=False, ), "runner_opts": json.dumps( payload["runner_options"], ensure_ascii=False, ), }, ) def downgrade() -> None: """Drop the seeded target + integration. Order matters: target references integration via FK, so target goes first. Keep this idempotent — if rows have already been removed by an admin (which would be unusual), the DELETE is a no-op. """ bind = op.get_bind() bind.execute( sa.text("DELETE FROM identity_sync_targets WHERE slug = :s"), {"s": _NEW_TARGET_SLUG}, ) bind.execute( sa.text("DELETE FROM integrations WHERE slug = :s"), {"s": _NEW_INTEGRATION_SLUG}, )