import pathlib import re import pandas as pd ROOT = pathlib.Path(__file__).resolve().parents[1] OUTPUT = ROOT / "src" / "main" / "resources" / "db" / "employee_directory.sql" def clean(value): if value is None: return "" return str(value).strip() def normalize_phone(value): return re.sub(r"\D", "", clean(value)) def sql(value): value = clean(value) if not value: return "NULL" return "'" + value.replace("\\", "\\\\").replace("'", "''") + "'" def split_paths(value): value = clean(value) if not value: return [] return [part.strip() for part in re.split(r"[；;]", value) if part.strip()] def infer_role(xuebang_position): text = clean(xuebang_position) if any(keyword in text for keyword in ["集团管理员", "总裁"]): return { "code": "GROUP_ADMIN", "name": "集团管理员", "level": 0, "scope_type": "ALL_ORGANIZATION", "source": "XUEBANG_POSITION_RULE", } if any(keyword in text for keyword in ["城市总监", "城市总", "运动馆负责人"]): return { "code": "VENUE_OWNER", "name": "运动馆负责人", "level": 1, "scope_type": "MULTI_VENUE", "source": "XUEBANG_POSITION_RULE", } if any(keyword in text for keyword in ["馆长", "教学主管"]): return { "code": "VENUE_MANAGER", "name": "馆长/教学主管", "level": 2, "scope_type": "VENUE", "source": "XUEBANG_POSITION_RULE", } if any(keyword in text for keyword in ["教练", "课程顾问", "学管师", "班主任"]): return { "code": "COACH", "name": "教练/一线员工", "level": 3, "scope_type": "SELF_OR_ASSIGNED", "source": "XUEBANG_POSITION_RULE", } return { "code": "STAFF", "name": "普通员工", "level": 9, "scope_type": "SELF", "source": "DEFAULT_RULE", } def main(): xb = pd.read_excel(ROOT / "学邦通讯录.xlsx", dtype=str).fillna("") wx = pd.read_excel(ROOT / "企业微信通讯录.xlsx", sheet_name="成员列表", header=9, dtype=str).fillna("") xb_by_phone = {} for _, row in xb.iterrows(): phone = normalize_phone(row.get("联系电话")) if phone and phone not in xb_by_phone: xb_by_phone[phone] = row employees = [] matched_xb_accounts = set() for _, row in wx.iterrows(): wecom_user_id = clean(row.get("账号")) if not wecom_user_id: continue mobile = normalize_phone(row.get("手机")) xb_match = xb_by_phone.get(mobile) if mobile else None xuebang_account = clean(xb_match.get("登录帐号")) if xb_match is not None else "" if xuebang_account: matched_xb_accounts.add(xuebang_account) position = clean(row.get("职务")) xuebang_position = clean(xb_match.get("职位")) if xb_match is not None else "" xuebang_manage_scope = clean(xb_match.get("管理权限")) if xb_match is not None else "" wecom_department_path = clean(row.get("部门")) role = infer_role(xuebang_position) employees.append({ "name": clean(row.get("姓名")), "mobile": mobile, "gender": clean(row.get("性别")), "position": xuebang_position or position, "wecom_user_id": wecom_user_id, "wecom_alias": clean(row.get("别名")), "wecom_department_path": wecom_department_path, "wecom_email": clean(row.get("邮箱")) or clean(row.get("企业邮箱")), "wecom_active_status": clean(row.get("激活状态")), "wecom_disabled_status": clean(row.get("禁用状态")), "xuebang_account": xuebang_account, "xuebang_status": clean(xb_match.get("状态")) if xb_match is not None else "", "xuebang_employment_status": clean(xb_match.get("在职状态")) if xb_match is not None else "", "xuebang_user_type": clean(xb_match.get("用户类型")) if xb_match is not None else "", "xuebang_primary_org": clean(xb_match.get("归属架构")) if xb_match is not None else "", "xuebang_manage_scope": xuebang_manage_scope, "xuebang_important_user": clean(xb_match.get("重要用户")) if xb_match is not None else "", "xuebang_position": xuebang_position, "primary_role_code": role["code"], "primary_role_name": role["name"], "role_level": role["level"], "data_scope_type": role["scope_type"], "role_source": role["source"], "match_method": "PHONE" if xb_match is not None else "UNMATCHED", "source_flags": "WECOM,XUEBANG" if xb_match is not None else "WECOM", }) for _, row in xb.iterrows(): xuebang_account = clean(row.get("登录帐号")) if not xuebang_account or xuebang_account in matched_xb_accounts: continue xuebang_position = clean(row.get("职位")) xuebang_manage_scope = clean(row.get("管理权限")) role = infer_role(xuebang_position) employees.append({ "name": clean(row.get("姓名")), "mobile": normalize_phone(row.get("联系电话")), "gender": clean(row.get("性别")), "position": xuebang_position, "wecom_user_id": "", "wecom_alias": "", "wecom_department_path": "", "wecom_email": "", "wecom_active_status": "", "wecom_disabled_status": "", "xuebang_account": xuebang_account, "xuebang_status": clean(row.get("状态")), "xuebang_employment_status": clean(row.get("在职状态")), "xuebang_user_type": clean(row.get("用户类型")), "xuebang_primary_org": clean(row.get("归属架构")), "xuebang_manage_scope": xuebang_manage_scope, "xuebang_important_user": clean(row.get("重要用户")), "xuebang_position": xuebang_position, "primary_role_code": role["code"], "primary_role_name": role["name"], "role_level": role["level"], "data_scope_type": role["scope_type"], "role_source": role["source"], "match_method": "XUEBANG_ONLY", "source_flags": "XUEBANG", }) orgs = {} def add_org(source, org_path): org_path = clean(org_path) if not org_path: return None parts = [part.strip() for part in org_path.split("/") if part.strip()] if "/" in org_path else [org_path] current = [] parent_path = "" for part in parts: current.append(part) full_path = "/".join(current) key = (source, full_path) if key not in orgs: orgs[key] = { "id": len(orgs) + 1, "source": source, "path": full_path, "name": part, "parent_path": parent_path, "level": len(current), } parent_path = full_path return orgs[(source, "/".join(parts))]["id"] for employee in employees: for path in split_paths(employee["wecom_department_path"]): add_org("WECOM", path) if employee["xuebang_primary_org"]: add_org("XUEBANG", employee["xuebang_primary_org"]) lines = [ "CREATE DATABASE IF NOT EXISTS agent_foundation DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;", "USE agent_foundation;", "", "DROP TABLE IF EXISTS employee_data_permission;", "DROP TABLE IF EXISTS employee_org_relation;", "DROP TABLE IF EXISTS employee_directory;", "DROP TABLE IF EXISTS role_definition;", "DROP TABLE IF EXISTS org_unit;", "", """CREATE TABLE org_unit ( id BIGINT PRIMARY KEY AUTO_INCREMENT COMMENT '组织节点主键', source_system VARCHAR(32) NOT NULL COMMENT '来源系统：WECOM 企业微信，XUEBANG 学邦', org_path VARCHAR(512) NOT NULL COMMENT '组织完整路径，企业微信使用 / 分隔，学邦使用原始架构名称', org_name VARCHAR(128) NOT NULL COMMENT '当前层级组织名称', parent_path VARCHAR(512) NULL COMMENT '父级组织完整路径，根节点为空', level_no INT NOT NULL COMMENT '组织层级，从 1 开始', created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间', updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间', UNIQUE KEY uk_org_source_path (source_system, org_path), KEY idx_org_parent (source_system, parent_path) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='组织架构表，统一保存企业微信部门和学邦架构';""", "", """CREATE TABLE employee_directory ( id BIGINT PRIMARY KEY AUTO_INCREMENT COMMENT '员工主键', employee_name VARCHAR(128) NOT NULL COMMENT '员工姓名，优先取企业微信姓名，学邦独有员工取学邦姓名', mobile VARCHAR(32) NULL COMMENT '手机号，已去除非数字字符，用于两套通讯录自动匹配', gender VARCHAR(16) NULL COMMENT '性别', position_name VARCHAR(255) NULL COMMENT '职位名称，企业微信员工优先取企业微信职务，学邦独有员工取学邦职位', wecom_user_id VARCHAR(128) NULL COMMENT '企业微信账号/UserID，用于从回调 FromUserName 反查员工', wecom_alias VARCHAR(128) NULL COMMENT '企业微信别名', wecom_department_path VARCHAR(512) NULL COMMENT '企业微信部门路径，多个部门时保留原始分隔文本', wecom_email VARCHAR(255) NULL COMMENT '企业微信邮箱或企业邮箱', wecom_active_status VARCHAR(64) NULL COMMENT '企业微信激活状态', wecom_disabled_status VARCHAR(64) NULL COMMENT '企业微信禁用状态', xuebang_account VARCHAR(128) NULL COMMENT '学邦登录账号，用于访问学邦业务数据', xuebang_status VARCHAR(64) NULL COMMENT '学邦账号状态', xuebang_employment_status VARCHAR(64) NULL COMMENT '学邦在职状态', xuebang_user_type VARCHAR(64) NULL COMMENT '学邦用户类型', xuebang_primary_org VARCHAR(512) NULL COMMENT '学邦归属架构', xuebang_manage_scope TEXT NULL COMMENT '学邦管理权限原始文本，作为后续数据权限范围依据', xuebang_important_user VARCHAR(32) NULL COMMENT '学邦重要用户标记', xuebang_position VARCHAR(255) NULL COMMENT '学邦职位', primary_role_code VARCHAR(64) NOT NULL COMMENT '组织角色编码：GROUP_ADMIN、VENUE_OWNER、VENUE_MANAGER、COACH、STAFF', primary_role_name VARCHAR(128) NOT NULL COMMENT '组织角色名称，例如运动馆负责人、馆长/教学主管、教练/一线员工', role_level INT NOT NULL COMMENT '组织角色层级，数字越小权限越高：1 运动馆负责人，2 馆长，3 教练', data_scope_type VARCHAR(64) NOT NULL COMMENT '默认数据范围类型：ALL_ORGANIZATION、MULTI_VENUE、VENUE、SELF_OR_ASSIGNED、SELF', role_source VARCHAR(64) NOT NULL COMMENT '角色识别来源：XUEBANG_POSITION_RULE 学邦职位规则，DEFAULT_RULE 默认规则', match_method VARCHAR(32) NOT NULL COMMENT '账号匹配方式：PHONE 手机号匹配，UNMATCHED 未匹配，XUEBANG_ONLY 仅学邦存在', source_flags VARCHAR(64) NOT NULL COMMENT '员工来源标记：WECOM、XUEBANG 或 WECOM,XUEBANG', created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间', updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间', UNIQUE KEY uk_employee_wecom_user_id (wecom_user_id), UNIQUE KEY uk_employee_xuebang_account (xuebang_account), KEY idx_employee_mobile (mobile), KEY idx_employee_name (employee_name) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='员工账号映射主表，保存企业微信账号和学邦账号对应关系';""", "", """CREATE TABLE role_definition ( role_code VARCHAR(64) PRIMARY KEY COMMENT '角色编码', role_name VARCHAR(128) NOT NULL COMMENT '角色名称', role_level INT NOT NULL COMMENT '角色层级，数字越小权限越高', data_scope_type VARCHAR(64) NOT NULL COMMENT '该角色默认数据范围类型', description VARCHAR(512) NOT NULL COMMENT '角色权限说明' ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='组织角色定义表，用于区分运动馆负责人、馆长、教练等权限层级';""", "", """CREATE TABLE employee_org_relation ( id BIGINT PRIMARY KEY AUTO_INCREMENT COMMENT '员工组织关系主键', employee_id BIGINT NOT NULL COMMENT '员工 ID，关联 employee_directory.id', org_unit_id BIGINT NOT NULL COMMENT '组织节点 ID，关联 org_unit.id', source_system VARCHAR(32) NOT NULL COMMENT '来源系统：WECOM 企业微信，XUEBANG 学邦', relation_type VARCHAR(64) NOT NULL COMMENT '关系类型：WECOM_DEPARTMENT 企业微信部门，XUEBANG_PRIMARY_ORG 学邦归属架构', is_primary TINYINT(1) NOT NULL DEFAULT 0 COMMENT '是否主组织，1 是，0 否', created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间', UNIQUE KEY uk_employee_org_relation (employee_id, org_unit_id, relation_type), KEY idx_employee_org_employee (employee_id), KEY idx_employee_org_unit (org_unit_id), CONSTRAINT fk_employee_org_employee FOREIGN KEY (employee_id) REFERENCES employee_directory(id), CONSTRAINT fk_employee_org_org FOREIGN KEY (org_unit_id) REFERENCES org_unit(id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='员工与组织架构关系表，支持后续按组织做数据权限';""", "", """CREATE TABLE employee_data_permission ( id BIGINT PRIMARY KEY AUTO_INCREMENT COMMENT '数据权限主键', employee_id BIGINT NOT NULL COMMENT '员工 ID，关联 employee_directory.id', source_system VARCHAR(32) NOT NULL COMMENT '权限来源系统：WECOM 企业微信，XUEBANG 学邦', permission_type VARCHAR(64) NOT NULL COMMENT '权限类型：ALL_ORGANIZATION_SCOPE 全部组织，WECOM_DEPARTMENT_SCOPE 企业微信部门范围，XUEBANG_MANAGE_SCOPE 学邦管理权限范围', org_unit_id BIGINT NULL COMMENT '关联组织节点 ID；学邦管理权限原始文本无法拆成节点时为空', scope_path VARCHAR(512) NOT NULL COMMENT '权限范围完整路径或原始管理权限文本', scope_name VARCHAR(128) NOT NULL COMMENT '权限范围显示名称', created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间', UNIQUE KEY uk_employee_permission_scope (employee_id, source_system, permission_type, scope_path), KEY idx_permission_employee (employee_id), KEY idx_permission_org (org_unit_id), CONSTRAINT fk_permission_employee FOREIGN KEY (employee_id) REFERENCES employee_directory(id), CONSTRAINT fk_permission_org FOREIGN KEY (org_unit_id) REFERENCES org_unit(id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='员工数据权限范围表，用于从企业微信账号定位学邦账号和可见组织范围';""", "", ] for org in sorted(orgs.values(), key=lambda item: item["id"]): lines.append( "INSERT INTO org_unit (id, source_system, org_path, org_name, parent_path, level_no) " f"VALUES ({org['id']}, {sql(org['source'])}, {sql(org['path'])}, {sql(org['name'])}, {sql(org['parent_path'])}, {org['level']});" ) role_defs = [ ("GROUP_ADMIN", "集团管理员", 0, "ALL_ORGANIZATION", "集团级权限，可按学邦管理权限覆盖全部或大范围组织数据"), ("VENUE_OWNER", "运动馆负责人", 1, "MULTI_VENUE", "负责人层级，可查看其管理权限覆盖的多个运动馆或区域数据"), ("VENUE_MANAGER", "馆长/教学主管", 2, "VENUE", "场馆管理层级，默认查看所在场馆及下属教练相关数据"), ("COACH", "教练/一线员工", 3, "SELF_OR_ASSIGNED", "一线执行层级，默认查看本人或被分配学员/客户数据"), ("STAFF", "普通员工", 9, "SELF", "未命中明确岗位规则时使用，默认仅本人数据"), ] lines.append("") for role in role_defs: lines.append( "INSERT INTO role_definition (role_code, role_name, role_level, data_scope_type, description) " f"VALUES ({sql(role[0])}, {sql(role[1])}, {role[2]}, {sql(role[3])}, {sql(role[4])});" ) employee_cols = [ "employee_name", "mobile", "gender", "position_name", "wecom_user_id", "wecom_alias", "wecom_department_path", "wecom_email", "wecom_active_status", "wecom_disabled_status", "xuebang_account", "xuebang_status", "xuebang_employment_status", "xuebang_user_type", "xuebang_primary_org", "xuebang_manage_scope", "xuebang_important_user", "xuebang_position", "primary_role_code", "primary_role_name", "role_level", "data_scope_type", "role_source", "match_method", "source_flags", ] lines.append("") for employee_id, employee in enumerate(employees, start=1): values = [ employee["name"], employee["mobile"], employee["gender"], employee["position"], employee["wecom_user_id"], employee["wecom_alias"], employee["wecom_department_path"], employee["wecom_email"], employee["wecom_active_status"], employee["wecom_disabled_status"], employee["xuebang_account"], employee["xuebang_status"], employee["xuebang_employment_status"], employee["xuebang_user_type"], employee["xuebang_primary_org"], employee["xuebang_manage_scope"], employee["xuebang_important_user"], employee["xuebang_position"], employee["primary_role_code"], employee["primary_role_name"], employee["role_level"], employee["data_scope_type"], employee["role_source"], employee["match_method"], employee["source_flags"], ] lines.append( f"INSERT INTO employee_directory (id, {', '.join(employee_cols)}) " f"VALUES ({employee_id}, {', '.join(sql(value) for value in values)});" ) relation_id = 1 permission_id = 1 lines.append("") for employee_id, employee in enumerate(employees, start=1): if employee["primary_role_code"] == "GROUP_ADMIN": lines.append( "INSERT INTO employee_data_permission (id, employee_id, source_system, permission_type, org_unit_id, scope_path, scope_name) " f"VALUES ({permission_id}, {employee_id}, 'SYSTEM', 'ALL_ORGANIZATION_SCOPE', NULL, '*', '全部组织');" ) permission_id += 1 for index, path in enumerate(split_paths(employee["wecom_department_path"])): org_id = add_org("WECOM", path) if org_id: lines.append( "INSERT INTO employee_org_relation (id, employee_id, org_unit_id, source_system, relation_type, is_primary) " f"VALUES ({relation_id}, {employee_id}, {org_id}, 'WECOM', 'WECOM_DEPARTMENT', {1 if index == 0 else 0});" ) relation_id += 1 lines.append( "INSERT INTO employee_data_permission (id, employee_id, source_system, permission_type, org_unit_id, scope_path, scope_name) " f"VALUES ({permission_id}, {employee_id}, 'WECOM', 'WECOM_DEPARTMENT_SCOPE', {org_id}, {sql(path)}, {sql(path.split('/')[-1])});" ) permission_id += 1 if employee["xuebang_primary_org"]: org_id = add_org("XUEBANG", employee["xuebang_primary_org"]) lines.append( "INSERT INTO employee_org_relation (id, employee_id, org_unit_id, source_system, relation_type, is_primary) " f"VALUES ({relation_id}, {employee_id}, {org_id}, 'XUEBANG', 'XUEBANG_PRIMARY_ORG', 1);" ) relation_id += 1 if employee["xuebang_manage_scope"]: scope = employee["xuebang_manage_scope"] lines.append( "INSERT INTO employee_data_permission (id, employee_id, source_system, permission_type, org_unit_id, scope_path, scope_name) " f"VALUES ({permission_id}, {employee_id}, 'XUEBANG', 'XUEBANG_MANAGE_SCOPE', NULL, {sql(scope)}, {sql(scope)});" ) permission_id += 1 lines.extend([ "", """CREATE OR REPLACE VIEW v_employee_account_mapping AS SELECT id, employee_name, mobile, wecom_user_id, xuebang_account, xuebang_primary_org, xuebang_manage_scope, primary_role_code, primary_role_name, role_level, data_scope_type, match_method, source_flags FROM employee_directory;""", "", """CREATE OR REPLACE VIEW v_wecom_data_permission AS SELECT e.wecom_user_id, e.xuebang_account, e.employee_name, e.primary_role_code, e.primary_role_name, e.role_level, e.data_scope_type, p.source_system, p.permission_type, p.scope_path, p.scope_name FROM employee_directory e JOIN employee_data_permission p ON p.employee_id = e.id WHERE e.wecom_user_id IS NOT NULL;""", "", f"-- Summary: employees={len(employees)}, org_units={len(orgs)}, org_relations={relation_id - 1}, permissions={permission_id - 1}", "", ]) OUTPUT.parent.mkdir(parents=True, exist_ok=True) OUTPUT.write_text("\n".join(lines), encoding="utf-8") print(f"wrote {OUTPUT}") print(f"employees={len(employees)} org_units={len(orgs)} org_relations={relation_id - 1} permissions={permission_id - 1}") if __name__ == "__main__": main()