182 lines
5.9 KiB
Python
182 lines
5.9 KiB
Python
"""
|
||
从通讯录 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 执行实际导入")
|