"""合作方通讯录:名册镜像 + 打标结果 + 每日同步 cron。 ## 这是在解决什么 我们这边有一批账号,系统认不出它们属于谁,而它们其实都属于同一家公司 (CBG)。上线这张表之前,生产的形状是: | 面 | 认不出的行 | |---|---| | `google_orphan_accounts`(存活) | 60,其中 **32 个在 OU `/PingKai`** | | `software_external_holder`(category='external') | 535 | | `github_org_memberships`(person_id IS NULL, active) | **331**,`external_confirmed_at` 一条都没有 | 而 `external_account_registry` 里已经有 **58 条 `external_guest`** (界面上就叫「CBG保留」)—— 全是人一条条点出来的,还有 8 个 `/PingKai` 的孤儿账号没来得及点。这个迁移把那件手工活换成对方通讯录的一次每日 只读同步。 ## 为什么必须跨域按邮箱名匹配 对方员工在自己的 Lark 租户里用的是 `@pingcap.cn`,在我们这边持有的是 `@pingcap.com`。生产实测 `@pingcap.cn` 的分布: google_orphan_accounts 0 external_account_registry 0 software_external_holder 11 identity_persons 2 也就是说 Gmail 那 32 个人,**按完整邮箱串一个都对不上**。所以 `partner_directory_members.email_name` 单独存一列(`@` 前那一段,小写) 并建索引 —— 跨域匹配就落在这一列上。现算 `SUBSTRING_INDEX` 会让每天 一次的匹配变成对两张表的全表扫描。 ## 为什么打标结果要单独一张表 最终标签写的是 `external_account_registry`(按邮箱)和 `github_org_memberships.external_confirmed_at`(按 handle),两处都只 留得下结论。而「邮箱名相同 ⇒ 同一个人」是一条**推断**;有人问「凭什么 说这个 Google 账号是 CBG 的」时,得答得出依据、匹配方式和写入时刻。 `partner_directory_links` 就是那份依据,顺带承担弱匹配的待确认队列。 Revision ID: 587 Revises: 586 (这个 PR 开发期间号被抢了**两次**:先是 585 被 `585_github_org_bot_classification` 抢走,改成 586 之后 586 又被 `586_github_handle_labels` 抢走。重复 revision id ⇒ 两个 head ⇒ 每个后端测试分片和镜像构建全红。**迁移号要对着 `origin/main` 抢,而且每次 push 之前都要重新抢一次** —— 一个跑了几小时的 PR 期间 main 会 前进好几次。本仓前两次踩它是 462 和 582。) """ from __future__ import annotations import uuid import sqlalchemy as sa from alembic import op from app.core.migration_helpers import id_type_for revision = "587" down_revision = "586" branch_labels = None depends_on = None MEMBERS = "partner_directory_members" LINKS = "partner_directory_links" JOB_KEY = "partner_directory.sync" _JOB_DESC = ( "每天只读拉取合作方飞书租户的员工名册(姓名 / 企业邮箱 / GitHub ID / " "部门 / 在职状态),再用它给我们这边认不出人的账号打上归属标签:" "Google 孤儿账号与各厂商 external 坐席写 external_account_registry " "的 external_guest 分类,GitHub 上没关联到人的 handle 写「已确认是" "外部人员」。\n" "匹配只认三种确切口径 —— 完整邮箱串、跨域邮箱名(对方 @<合作方域> " "的 zhang.san 认作我们的 zhang.san@<本司域>)、对方通讯录里登记的 " "GitHub handle;仅靠姓名对上的进待确认队列,不自动写。\n" "**只对无主行生效**:已经关联到在职员工的行永远不被覆盖,管理员" "手工写过分类的注册表行也不被覆盖。一次读不全的同步(权限过期 / " "部门读不到)不会退役任何名册行,以免把已经打上的标签变成僵尸。" ) def _has_table(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() integration_fk = id_type_for("integrations") person_fk = id_type_for("identity_persons") if not _has_table(bind, MEMBERS): op.create_table( MEMBERS, sa.Column("id", sa.String(36), primary_key=True), sa.Column( "integration_id", integration_fk, sa.ForeignKey("integrations.id", ondelete="CASCADE"), nullable=False, ), # 对方目录的稳定用户 id(飞书 open_id)。不用邮箱做键: # 对方把离职者的邮箱给了新人时,按邮箱做键会让新人静默 # 继承前任身上的全部标签。 sa.Column("external_id", sa.String(128), nullable=False), sa.Column("union_id", sa.String(128)), sa.Column("partner_user_id", sa.String(128)), sa.Column("employee_no", sa.String(64)), sa.Column("name", sa.String(255)), sa.Column("en_name", sa.String(255)), sa.Column("email", sa.String(255)), sa.Column("secondary_email", sa.String(255)), # 跨域匹配的载体 —— 见模块文档。 sa.Column("email_name", sa.String(255)), sa.Column("job_title", sa.String(255)), sa.Column("department_ids", sa.JSON()), sa.Column("department_path", sa.JSON()), sa.Column("github_login", sa.String(64)), sa.Column("github_login_lower", sa.String(64)), sa.Column( "is_active", sa.Boolean(), nullable=False, server_default="1", ), sa.Column( "is_suspended", sa.Boolean(), nullable=False, server_default="0", ), sa.Column("status_raw", sa.JSON()), sa.Column("custom_attrs", sa.JSON()), sa.Column( "first_seen_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), ), sa.Column( "last_seen_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), ), sa.Column("deleted_at", sa.DateTime(timezone=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(), ), sa.UniqueConstraint( "integration_id", "external_id", name="uq_partner_directory_members_ext", ), ) op.create_index( "ix_partner_directory_members_email_name", MEMBERS, ["email_name"], ) op.create_index( "ix_partner_directory_members_email", MEMBERS, ["email"], ) op.create_index( "ix_partner_directory_members_github", MEMBERS, ["github_login_lower"], ) op.create_index( "ix_partner_directory_members_live", MEMBERS, ["integration_id", "deleted_at", "is_active"], ) if not _has_table(bind, LINKS): op.create_table( LINKS, sa.Column("id", sa.String(36), primary_key=True), sa.Column( "integration_id", integration_fk, sa.ForeignKey("integrations.id", ondelete="CASCADE"), nullable=False, ), # 父表就在上面这个迁移里建的,两边本来就同一套表默认;但 FK # 列的类型一律向 id_type_for 要 —— TiDB 3780 只认「和父列 # 完全一致」,而 CI 的 TiDB parity 闸对手写 String(36)+FK # 一票否决(它就是为 462/472 两次生产卡链加的)。 sa.Column( "member_id", id_type_for(MEMBERS), sa.ForeignKey( f"{MEMBERS}.id", ondelete="SET NULL", ), ), sa.Column("surface", sa.String(32), nullable=False), sa.Column("target_key", sa.String(255), nullable=False), sa.Column("target_label", sa.String(255)), sa.Column("match_method", sa.String(32), nullable=False), sa.Column( "state", sa.String(24), nullable=False, server_default="pending_review", ), sa.Column("applied_at", sa.DateTime(timezone=True)), sa.Column("apply_skipped_reason", sa.String(255)), sa.Column( "reviewed_by", person_fk, sa.ForeignKey("identity_persons.id", ondelete="SET NULL"), ), sa.Column("reviewed_at", sa.DateTime(timezone=True)), sa.Column("note", sa.Text()), sa.Column( "last_seen_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.UniqueConstraint( "integration_id", "surface", "target_key", name="uq_partner_directory_links_target", ), ) op.create_index( "ix_partner_directory_links_state", LINKS, ["state", "surface"], ) op.create_index( "ix_partner_directory_links_member", LINKS, ["member_id"], ) if not _has_table(bind, "scheduled_jobs"): return 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_KEY, "display": "合作方通讯录 · 每日同步与打标", "desc": _JOB_DESC, "handler": JOB_KEY, # 06:40(上海)。放在各家席位对账 cron 之前,这样对账面板 # 当天读到的 external_guest 分类已经是今天的名册算出来的。 "cron": "40 6 * * *", "config": "{}", }, ) def downgrade() -> None: bind = op.get_bind() if _has_table(bind, "scheduled_jobs"): bind.execute( sa.text("DELETE FROM scheduled_jobs WHERE handler_key = :h"), {"h": JOB_KEY}, ) if _has_table(bind, LINKS): op.drop_table(LINKS) if _has_table(bind, MEMBERS): op.drop_table(MEMBERS)