'init'

sc-lktx-backend/scripts/import_employees.py

@@ -0,0 +1,181 @@
+"""
+从通讯录 Excel 提取四川凌空员工数据  调用 admin API 导入
+用法: python import_employees.py <xlsx文件路径> [--dry-run]
+
+需要管理员 JWT，三种方式：
+  方式1：环境变量  ADMIN_TOKEN="xxx"
+  方式2：环境变量  ADMIN_ACCOUNT="admin" ADMIN_PASSWORD="xxx"
+  方式3：都不设则提示手动输入账号密码
+"""
+
+import os
+import sys
+import json
+import requests
+import openpyxl
+
+API_BASE = os.getenv("API_BASE", "http://localhost:28175/v1/admin")
+TOKEN = os.getenv("ADMIN_TOKEN", "")
+
+def login_and_get_token():
+    account = os.getenv("ADMIN_ACCOUNT", "")
+    password = os.getenv("ADMIN_PASSWORD", "")
+    if not account or not password:
+        account = input("管理员账号: ").strip()
+        password = input("管理员密码: ").strip()
+    resp = requests.post(API_BASE.replace("/admin", "") + "/admin/login", json={
+        "account": account,
+        "password": password,
+    })
+    data = resp.json()
+    if data.get("code") != 200:
+        print(f"登录失败: {data.get('msg')}")
+        sys.exit(1)
+    token = data["data"]["token"]
+    print(f" 登录成功")
+    return token
+
+def get_headers():
+    global TOKEN
+    if not TOKEN:
+        TOKEN = login_and_get_token()
+    return {"Authorization": f"Bearer {TOKEN}"}
+
+def get_dept_map():
+    """从后台获取部门列表，建立 部门名称  ID 的映射"""
+    resp = requests.get(f"{API_BASE}/departments-all", headers=get_headers())
+    data = resp.json().get("data", [])
+    print(f"  找到 {len(data)} 个部门")
+    dept_map = {}
+    for d in data:
+        dept_map[d["name"]] = d["id"]
+    return dept_map
+
+def parse_excel(path):
+    """解析 Excel，返回所有员工列表（不过滤公司）"""
+    wb = openpyxl.load_workbook(path, data_only=True)
+    ws = wb.active
+
+    employees = []
+    for row in ws.iter_rows(min_row=4, values_only=True):
+        name = str(row[1] or "").strip()
+        phone = str(row[2] or "").strip()
+        dept_full = str(row[3] or "").strip()
+        position = str(row[4] or "").strip()
+
+        if not name or not phone:
+            continue
+
+        # 标准化手机号
+        phone_clean = phone.replace("+86-", "").replace("+86", "").strip()
+
+        # 取第一个部门路径
+        dept_first = dept_full.split(",")[0].strip() if dept_full else ""
+
+        employees.append({
+            "real_name": name,
+            "phone": phone_clean,
+            "department_full": dept_first,
+            "position": position,
+        })
+
+    return employees
+
+def match_department(dept_full, dept_map):
+    """尝试多种方式匹配部门"""
+    # 1. 精确匹配完整名称
+    if dept_full in dept_map:
+        return dept_full, dept_map[dept_full]
+
+    # 2. 带分隔符的：取最后一级名
+    for sep in ["-", "－", "", "/", ""]:
+        if sep in dept_full:
+            sub = dept_full.rsplit(sep, 1)[-1].strip()
+            if sub in dept_map:
+                return sub, dept_map[sub]
+
+    # 3. 模糊匹配：dept_map 中有包含关系的
+    for name, did in dept_map.items():
+        if name in dept_full or dept_full in name:
+            return name, did
+
+    return None, None
+
+def import_employees(employees, dept_map, dry_run=False):
+    """批量导入"""
+    items = []
+    skipped_no_dept = 0
+    for emp in employees:
+        matched_name, dept_id = match_department(emp["department_full"], dept_map)
+
+        if not dept_id:
+            print(f"   跳过 {emp['real_name']}: 未匹配部门 '{emp['department_full']}'")
+            skipped_no_dept += 1
+            continue
+
+        items.append({
+            "real_name": emp["real_name"],
+            "phone": emp["phone"],
+            "department_id": dept_id,
+            "department_name": matched_name,
+            "position": emp["position"],
+        })
+
+    if dry_run:
+        print(f"\n[dry-run] 将导入 {len(items)} 人，跳过 {skipped_no_dept} 人（无匹配部门）")
+        for item in items[:5]:
+            print(f"  {item['real_name']} | {item['phone']} | {item['department_name']} | {item['position']}")
+        if len(items) > 5:
+            print(f"  ... 共 {len(items)} 人")
+        return
+
+    batch_size = 50
+    total_imported = 0
+    for i in range(0, len(items), batch_size):
+        batch = items[i:i + batch_size]
+        resp = requests.post(f"{API_BASE}/employees/import", json={"employees": batch}, headers=get_headers())
+        result = resp.json()
+        if result.get("code") == 200:
+            data = result.get("data", {})
+            total_imported += data.get("imported", 0)
+            print(f"  批次 {i//batch_size + 1}: 导入 {data.get('imported')} / 跳过 {data.get('skipped')}")
+        else:
+            print(f"  批次 {i//batch_size + 1}: 失败 - {result.get('msg')}")
+
+    print(f"\n 共导入 {total_imported} 人")
+
+if __name__ == "__main__":
+    if len(sys.argv) < 2:
+        print("用法: python import_employees.py <xlsx文件路径> [--dry-run] [--region 关键词]")
+        print("  --region 关键词  只导入部门包含此关键词的员工（默认: 四川）")
+        sys.exit(1)
+
+    path = sys.argv[1]
+    dry_run = "--dry-run" in sys.argv
+
+    # 解析 region 参数
+    region_filter = None
+    for i, arg in enumerate(sys.argv):
+        if arg == "--region" and i + 1 < len(sys.argv):
+            region_filter = sys.argv[i + 1]
+    if region_filter is None:
+        region_filter = "四川"  # 默认只导入四川
+
+    print(f" 获取部门列表...")
+    dept_map = get_dept_map()
+
+    print("\n 解析 Excel...")
+    employees = parse_excel(path)
+    # 按区域过滤
+    if region_filter:
+        before = len(employees)
+        employees = [e for e in employees if region_filter in e["department_full"]]
+        print(f"   共 {before} 名员工，按区域 '{region_filter}' 过滤后: {len(employees)} 人")
+    else:
+        print(f"   提取到 {len(employees)} 名员工")
+
+    print("\n 导入中...")
+    import_employees(employees, dept_map, dry_run)
+
+    if dry_run:
+        print("\n 去掉 --dry-run 执行实际导入")