影刀RPA在HR领域：招聘、入职、考勤与薪酬自动化

影刀RPA在HR领域：招聘、入职、考勤与薪酬自动化

作者：林焱 | 发布平台：CSDN / 公众号 / 掘金

前言：HR工作的痛点

HR部门承担着大量重复性事务——筛选简历、录入员工信息、统计考勤、计算薪酬……一个50人规模的公司，HR每月光薪酬计算和考勤核对就需要2-3天，且稍有不慎就会引发员工投诉。

影刀RPA能够将HR事务性工作自动化，让HR聚焦在战略性人力资源管理上。

店群矩阵自动化突破运营极限！

场景一：招聘简历自动筛选

多平台简历自动下载与分类

import shadowbot as sb
import re
from datetime import datetime

def download_resumes_from_platforms():
"""从各招聘平台下载简历"""
all_resumes = []

# 从BOSS直聘下载
page = sb.browser.get_page("https://www.zhipin.com/web/geek/job")
# 登录…

# 点击"求职者管理"
page.click(".nav-job-manage")
page.click(".resume-list-item")

# 下载每份简历
resume_items = page.find_elements(".resume-item")

for item in resume_items:
name = item.find_element(".name").text
page.click(item.find_element(".download-btn"))
page.wait_for_download(timeout=15)

all_resumes.append({
"平台": "BOSS直聘",
"姓名": name,
"文件": sb.download.get_latest(),
"下载时间": datetime.now().isoformat(),
})

return all_resumes

def auto_screen_resumes(resumes, job_requirements):
"""自动筛选简历"""

screening_results = []

for resume in resumes:
try:
# 使用OCR读取简历文本
text = sb.ocr.recognize(resume["文件"])

score = 0
matched_keywords = []
missing_keywords = []

# 检查必要条件
for req in job_requirements.get("必须", []):
if req.lower() in text.lower():
score += 20
matched_keywords.append(req)
else:
missing_keywords.append(f"缺少：{req}")

# 检查加分项
for req in job_requirements.get("加分", []):
if req.lower() in text.lower():
score += 10
matched_keywords.append(f"✓{req}")

# 提取工作年限
exp_match = re.search(r"工作经验[：:]\\s*(\\d+)\\s*年", text)
if exp_match:
years = int(exp_match.group(1))
min_years = job_requirements.get("最低年限", 0)
if years >= min_years:
score += 15
else:
missing_keywords.append(f"经验不足（{years}年<{min_years}年）")

# 提取学历
edu_levels = ["博士", "硕士", "本科", "大专"]
required_edu = job_requirements.get("最低学历", "本科")
required_idx = edu_levels.index(required_edu) if required_edu in edu_levels else 2

for i, level in enumerate(edu_levels):
if level in text:
if i <= required_idx:
score += 15
break

# 判断结果
recommendation = "A类-直接面试" if score >= 70 else "B类-复核" if score >= 50 else "C类-不合适"

screening_results.append({
"姓名": resume["姓名"],
"平台": resume["平台"],
"综合评分": score,
"推荐结论": recommendation,
"匹配项": "、".join(matched_keywords),
"不足项": "、".join(missing_keywords),
"文件路径": resume["文件"],
})

except Exception as e:
sb.log.error(f"筛选失败：{resume.get('姓名', '')} – {e}")

# 按评分排序
screening_results.sort(key=lambda x: x["综合评分"], reverse=True)

# 保存筛选结果
sb.excel.write_rows("简历筛选结果.xlsx", screening_results)

a_class = [r for r in screening_results if "A类" in r["推荐结论"]]
sb.log.info(f"筛选完成：A类 {len(a_class)} 份，共 {len(screening_results)} 份")

return screening_results

# 执行
job_reqs = {
"必须": ["Python", "数据分析", "SQL"],
"加分": ["机器学习", "Spark", "Hadoop"],
"最低年限": 3,
"最低学历": "本科",
}

resumes = download_resumes_from_platforms()
results = auto_screen_resumes(resumes, job_reqs)

场景二：员工入职自动化

入职手续全流程自动办理

def onboarding_automation(new_employee_info):
"""新员工入职自动化"""
employee = new_employee_info

results = {}

# 1. 在HR系统中创建员工档案
page = sb.browser.get_page("http://hris.company.local/employee/create")

page.input("#emp-name", employee["姓名"])
page.input("#emp-id", employee["身份证号"])
page.input("#emp-phone", employee["手机号"])
page.input("#emp-email", employee["邮箱"])
page.select("#department", employee["部门"])
page.select("#position", employee["职位"])
page.input("#start-date", employee["入职日期"])
page.input("#salary", employee["基本薪资"])

page.click("#save-btn")
employee_id = page.get_text(".new-emp-id")
results["employee_id"] = employee_id

# 2. 创建企业邮箱
import requests
email_response = requests.post(
"http://mail-admin.company.local/api/create-mailbox",
headers={"Authorization": f"Bearer {sb.config.get('mail_admin_token')}"},
json={
"username": f"{employee['英文名']}.{employee['部门缩写']}",
"display_name": employee["姓名"],
"password": generate_initial_password(),
}
)
results["email"] = email_response.json().get("email")

# 3. 钉钉/企业微信开通账号
wecom_response = requests.post(
"https://qyapi.weixin.qq.com/cgi-bin/user/create",
params={"access_token": get_wecom_token()},
json={
"userid": employee["英文名"],
"name": employee["姓名"],
"mobile": employee["手机号"],
"email": results.get("email", ""),
"department": [get_dept_id(employee["部门"])],
}
)
results["wecom_userid"] = employee["英文名"]

# 4. 添加门禁权限
add_access_permission(employee_id, employee["办公区域"])
results["access"] = "已开通"

# 5. 分配设备（在资产管理系统登记）
register_equipment(employee_id, employee.get("分配设备", []))

# 6. 发送入职欢迎邮件
welcome_email = f"""
亲爱的{employee['姓名']}，

欢迎加入我们！🎉

您的账号信息如下：
– 工号：{employee_id}
– 企业邮箱：{results.get('email', '')}
– 企业微信：已发送二维码至您的手机
– 入职日期：{employee['入职日期']}
– 所属部门：{employee['部门']}

第一天需要携带：
✓ 身份证原件
✓ 学历证书
✓ 上家公司离职证明

如有疑问，请联系HR：hr@company.com

期待您的到来！
""".strip()

send_email(employee["个人邮箱"], "欢迎加入！您的入职信息", welcome_email)
results["welcome_email"] = "已发送"

sb.log.info(f"入职办理完成：{employee['姓名']}（工号：{employee_id}）")
return results

# 批量处理入职
new_employees = sb.excel.read_all("本月入职名单.xlsx")
for emp in new_employees:
result = onboarding_automation(emp)
sb.log.info(f"入职完成：{emp['姓名']} – {result}")

场景三：考勤数据自动汇总

多设备考勤数据合并与异常检测

def attendance_consolidation():
"""多来源考勤数据汇总"""
import pandas as pd
from datetime import datetime, timedelta

# 1. 从门禁系统导出考勤数据
page = sb.browser.get_page("http://access.company.local/attendance")
page.select("#month", datetime.now().strftime("%Y-%m"))
page.click("#export-btn")
page.wait_for_download(timeout=30)
access_df = pd.read_excel(sb.download.get_latest())

# 2. 从钉钉导出考勤数据
page2 = sb.browser.get_page("https://attend.dingtalk.com")
page2.navigate("/checkin/export")
page2.click("#export-current-month")
page2.wait_for_download(timeout=30)
dingding_df = pd.read_excel(sb.download.get_latest())

# 3. 读取请假记录
leave_records = pd.read_excel("请假记录.xlsx")

# 合并考勤数据
# 统一字段名
access_df.rename(columns={"姓名": "name", "打卡时间": "check_time", "打卡类型": "type"}, inplace=True)
dingding_df.rename(columns={"员工": "name", "时间": "check_time", "结果": "type"}, inplace=True)

combined = pd.concat([access_df, dingding_df], ignore_index=True)
combined["date"] = pd.to_datetime(combined["check_time"]).dt.date

# 每人每天取最早打卡（上班）和最晚打卡（下班）
daily_attendance = combined.groupby(["name", "date"]).agg(
first_check=("check_time", "min"),
last_check=("check_time", "max"),
).reset_index()

# 工作时间规则
work_start = "09:00"
work_end = "18:00"

anomalies = []

for _, row in daily_attendance.iterrows():
first = pd.to_datetime(row["first_check"])
last = pd.to_datetime(row["last_check"])

issues = []

# 检查迟到（超过09:15算迟到）
if first.time() > pd.Timestamp("09:15").time():
delay_min = (first – first.replace(hour=9, minute=0)).seconds // 60
issues.append(f"迟到{delay_min}分钟")

# 检查早退（18:00前离开算早退）
if last.time() < pd.Timestamp("18:00").time():
early_min = (last.replace(hour=18, minute=0) – last).seconds // 60
issues.append(f"早退{early_min}分钟")

# 检查是否请假
leave = leave_records[
(leave_records["姓名"] == row["name"]) &
(pd.to_datetime(leave_records["请假日期"]).dt.date == row["date"])
]

if not leave.empty:
issues = [] # 有请假则清除异常

if issues:
anomalies.append({
"姓名": row["name"],
"日期": str(row["date"]),
"上班打卡": str(first.time()),
"下班打卡": str(last.time()),
"异常情况": "、".join(issues),
})

# 保存异常报告
sb.excel.write_rows("考勤异常报告.xlsx", anomalies)
sb.log.info(f"考勤汇总完成：发现 {len(anomalies)} 条异常记录")

return daily_attendance, anomalies

daily, anomalies = attendance_consolidation()

场景四：薪酬自动计算

temu店群自动化报活动案例

基于考勤+绩效自动生成工资单

def calculate_monthly_salary():
"""月度薪酬自动计算"""
import pandas as pd

# 读取员工基本薪资
employees = pd.read_excel("员工信息.xlsx")

# 读取考勤汇总（已处理的）
attendance = pd.read_excel("本月考勤汇总.xlsx")

# 读取绩效评分
performance = pd.read_excel("本月绩效考核.xlsx")

# 读取请假记录
leaves = pd.read_excel("本月请假.xlsx")

salary_results = []

# 社保公积金比例
insurance_rates = {
"养老": 0.08,
"医疗": 0.02,
"失业": 0.005,
"公积金": 0.12,
}

for _, emp in employees.iterrows():
emp_id = emp["工号"]
base_salary = float(emp["基本薪资"])

# 获取考勤数据
emp_attend = attendance[attendance["工号"] == emp_id]

# 计算应出勤天数（本月工作日）
import calendar
from datetime import datetime
today = datetime.now()
_, days_in_month = calendar.monthrange(today.year, today.month)

# 计算实际出勤天数
actual_days = int(emp_attend["实出勤天数"].values[0]) if not emp_attend.empty else 0
required_days = int(emp_attend["应出勤天数"].values[0]) if not emp_attend.empty else 22

# 日薪
daily_salary = base_salary / required_days

# 计算迟到早退扣款
late_count = int(emp_attend["迟到次数"].values[0]) if not emp_attend.empty else 0
early_leave_count = int(emp_attend["早退次数"].values[0]) if not emp_attend.empty else 0

deduct_late = late_count * 50 # 每次迟到扣50元
deduct_early = early_leave_count * 50

# 计算请假扣款
emp_leaves = leaves[leaves["工号"] == emp_id]

deduct_leave = 0
for _, leave in emp_leaves.iterrows():
leave_type = leave.get("请假类型", "")
leave_days = float(leave.get("天数", 0))

if leave_type == "事假":
deduct_leave += daily_salary * leave_days
elif leave_type == "病假":
deduct_leave += daily_salary * leave_days * 0.8 # 病假按80%计
# 年假、婚假等不扣钱

# 获取绩效奖金
emp_perf = performance[performance["工号"] == emp_id]

perf_score = float(emp_perf["绩效得分"].values[0]) if not emp_perf.empty else 80

# 绩效系数
if perf_score >= 95:
perf_coef = 1.3
elif perf_score >= 85:
perf_coef = 1.1
elif perf_score >= 75:
perf_coef = 1.0
elif perf_score >= 60:
perf_coef = 0.8
else:
perf_coef = 0.6

performance_bonus = float(emp.get("绩效基数", 0)) * perf_coef

# 计算税前合计
gross_salary = (
base_salary
+ performance_bonus
– deduct_late
– deduct_early
– deduct_leave
)

# 计算个人社保公积金
total_insurance = sum(base_salary * rate for rate in insurance_rates.values())

# 计算个人所得税（简化版，按月累进税率）
taxable = gross_salary – total_insurance – 5000 # 5000元起征点

if taxable <= 0:
personal_tax = 0
elif taxable <= 3000:
personal_tax = taxable * 0.03
elif taxable <= 12000:
personal_tax = taxable * 0.10 – 210
elif taxable <= 25000:
personal_tax = taxable * 0.20 – 1410
else:
personal_tax = taxable * 0.25 – 2660

# 实发工资
net_salary = gross_salary – total_insurance – max(0, personal_tax)

salary_results.append({
"工号": emp_id,
"姓名": emp["姓名"],
"部门": emp["部门"],
"基本薪资": base_salary,
"绩效奖金": round(performance_bonus, 2),
"迟到扣款": round(deduct_late, 2),
"请假扣款": round(deduct_leave, 2),
"税前工资": round(gross_salary, 2),
"个人社保": round(total_insurance, 2),
"个人所得税": round(max(0, personal_tax), 2),
"实发工资": round(net_salary, 2),
"绩效评分": perf_score,
})

# 保存工资单
salary_df = pd.DataFrame(salary_results)
month_str = datetime.now().strftime("%Y年%m月")
salary_file = f"工资单_{month_str}.xlsx"
salary_df.to_excel(salary_file, index=False)

total_payroll = salary_df["实发工资"].sum()
sb.log.info(f"{month_str}工资计算完成：{len(salary_results)} 人，合计发放 {total_payroll:,.2f} 元")

# 发送工资条（每人单独邮件）
for _, row in salary_df.iterrows():
send_payslip_email(row)

return salary_file

calculate_monthly_salary()

效果总结

影刀RPA已成为HR数字化转型的标配工具，建议从薪酬计算自动化开始，因为其价值最直观，每月固定节省大量时间。

作者：林焱 | 转载请注明出处