返回教程check_stages.py
下载原文件
"""离线机械核对：人数、工作日与记录差异。不会决定处置或写回系统。
python3 scripts/check_stages.py 工作目录 审批字段.csv --rules scripts/核对口径.json --output-dir 核对结果
审批字段列：截图文件,申请编号,截图状态,最后节点时间,当前节点。
审批字段须先与原图核对；本脚本不识图。退出0只表示运算完成，不代表没有业务问题；退出2表示未完成。
"""
import argparse
import csv
import json
import re
import sys
from collections import Counter, defaultdict
from datetime import date, datetime, timedelta
from pathlib import Path
import openpyxl
from identity import group_people, person_key
from read_eml import read_messages
STAGES = ['简历筛选','一面','二面','用人经理复核','Offer审批','Offer已接受']
def parse_date(value):
if not value: return None
if isinstance(value, (date, datetime)): return value.date() if isinstance(value, datetime) else value
return date.fromisoformat(str(value)[:10])
class Calendar:
def __init__(self, config):
self.stat = parse_date(config['统计日'])
self.holidays = {parse_date(x) for x in config['休息日']}
self.extra = {parse_date(x) for x in config['调休工作日']}
self.limits = config['时限_工作日']
if any(not isinstance(n,int) or n < 1 for n in self.limits.values()): raise ValueError('时限必须为正整数')
def is_workday(self, value):
return value in self.extra or (value.weekday()<5 and value not in self.holidays)
def elapsed(self, start, end=None):
end = end or self.stat; count = 0
while start < end:
start += timedelta(days=1)
count += self.is_workday(start)
return count
def due(self, start, days):
while days:
start += timedelta(days=1)
if self.is_workday(start): days -= 1
return start
def read_rows(work):
files = list(Path(work).glob('*.csv'))
files = [f for f in files if f.name.startswith('招聘系统导出')]
if len(files)!=1: raise ValueError('工作目录须有唯一的招聘系统导出CSV')
with files[0].open(encoding='utf-8-sig') as f:
rows = list(csv.DictReader(f))
if not rows: raise ValueError('系统导出没有数据')
required = ['申请编号','候选人','手机号','招聘需求编号','当前阶段','阶段进入日期','面试时间','反馈完成时间']
if any(k not in rows[0] for k in required): raise ValueError('系统导出缺必要列')
ids = [r['申请编号'] for r in rows]
if not all(ids) or len(set(ids))!=len(ids): raise ValueError('申请编号为空或同一申请多行，先核对核心身份')
for i,row in enumerate(rows,2):
row['_line']=i
if row['当前阶段'] not in STAGES: raise ValueError('未知阶段：'+row['当前阶段'])
if not row['阶段进入日期']: raise ValueError('缺阶段进入日期：'+row['申请编号'])
for k in ['阶段进入日期','面试时间','反馈完成时间']:
if row[k]: parse_date(row[k])
return files[0].name,rows
def analyze(work, approval_file, rules):
work=Path(work); cal=Calendar(rules); csv_name,rows=read_rows(work); by_id={r['申请编号']:r for r in rows}
wb=openpyxl.load_workbook(work/'招聘计划.xlsx',data_only=True)
req={r[0].value:{'status':r[4].value,'owner':r[5].value,'line':r[0].row,'note':r[6].value or ''}for r in wb['需求计划'].iter_rows(min_row=5)if r[0].value}
plan={r[0].value:{'date':r[4].value,'line':r[0].row,'candidate':r[1].value}for r in wb['面试排期'].iter_rows(min_row=5)if r[0].value}
messages=read_messages(work/'回执邮件'); mails=defaultdict(list); unresolved=[]
for msg in messages:
ids=msg['正文编号'] or msg['文件名编号']
if not ids: unresolved.append(msg['文件']);continue
for aid in ids:
# 只有文件名编号时，再用正文完整号码代号交叉核对，避免凭文件名认人。
known=bool(msg['正文编号']) or (aid in by_id and by_id[aid]['手机号'] in msg['号码代号'])
if known: mails[aid].append(msg)
else: unresolved.append(msg['文件']+'（编号归属待核）')
approvals={}
if approval_file:
with Path(approval_file).open(encoding='utf-8-sig') as f:
for row in csv.DictReader(f):
if not all(k in row for k in ['截图文件','申请编号','截图状态','最后节点时间','当前节点']): raise ValueError('审批字段CSV列名不完整')
aid=row['申请编号']
if aid in approvals: raise ValueError('审批字段有重复申请编号：'+aid)
if row['截图状态'] not in ['已通过','审批中']: raise ValueError('无法识别审批状态：'+aid)
parse_date(row['最后节点时间']);approvals[aid]=row
groups=group_people(rows); raw=Counter(r['当前阶段']for r in rows);counts=Counter();multi=[];unknown=[]
for key,rs in groups.items():
stages={r['当前阶段']for r in rs}
if key.startswith('待核:'):unknown.append([r['申请编号']for r in rs])
if len(stages)==1: counts[next(iter(stages))]+=1
else:counts['多阶段待确认']+=1;multi.append([r['申请编号']for r in rs])
findings=[]
def add(aid,kind,detail,sources,key=None):
identity=key or person_key(by_id[aid])
findings.append({'申请编号':aid,'候选身份':identity,'类型':kind,'事项':detail,'来源':'; '.join(sources)})
for row in rows:
aid=row['申请编号'];stage=row['当前阶段'];src=f'{csv_name} 第{row["_line"]}行';enter=parse_date(row['阶段进入日期']);ms=mails[aid]
request=req.get(row['招聘需求编号'])
if request and request['status']=='暂停' and stage!='Offer已接受':
add(aid,'待确认','需求暂停；是否已有负责人继续办理的书面确认须回原文核实，不从面试发生日推断安排创建日',[src,f'招聘计划.xlsx[需求计划]第{request["line"]}行'])
if stage in ['一面','二面']:
iv=parse_date(row['面试时间']);fb=parse_date(row['反馈完成时间']);fb_mail=[m for m in ms if m['类型']in ['反馈已录','反馈未录']];unknown_mail=[m for m in ms if m['类型'].startswith('未识别')]
if iv:
dates=([fb]if fb else[])+[parse_date(m['时间'])for m in fb_mail]
if dates:
effective=min(dates)
if not fb and fb_mail:add(aid,'记录待同步','系统反馈时间为空，面试官邮件已有反馈；不能算作无反馈',[src,*['回执邮件/'+m['文件']for m in fb_mail]])
if cal.elapsed(iv,effective)>cal.limits['面试反馈']:add(aid,'历史超时待核',f'最早已知反馈距面试超过{cal.limits["面试反馈"]}个工作日；现已有反馈，不归为当前无反馈',[src,*['回执邮件/'+m['文件']for m in fb_mail]])
elif unknown_mail or unresolved:add(aid,'待确认','尚有未识别或未归属邮件，不能直接认定无反馈',[src,*['回执邮件/'+m['文件']for m in unknown_mail]])
elif cal.elapsed(iv)>cal.limits['面试反馈']:
add(aid,'规则超时',f'材料未见有效反馈，面试后{cal.elapsed(iv)}个工作日，超过{cal.limits["面试反馈"]}；自动回复不算反馈，承接人和时间待确认',[src,*['回执邮件/'+m['文件']for m in ms]])
else:
scheduling=[m for m in ms if m['类型']in ['改期请求','排期确认']]
if aid in plan or scheduling:
sources=[src]+([f'招聘计划.xlsx[面试排期]第{plan[aid]["line"]}行']if aid in plan else[])+['回执邮件/'+m['文件']for m in scheduling]
add(aid,'排期待核/待同步','系统面试时间为空；计划或邮件已有排期线索，改期请求不等于已确认，不能沿用未排期时限直接判超时',sources)
elif cal.elapsed(enter)>cal.limits['未排期']:
add(aid,'规则超时',f'系统未记面试时间且本包未见排期线索，按未排期口径已{cal.elapsed(enter)}个工作日',[src])
elif stage in ['简历筛选','用人经理复核']:
limit=cal.limits[stage]
if cal.elapsed(enter)>limit:add(aid,'规则超时',f'{stage}已{cal.elapsed(enter)}个工作日，超过{limit}',[src])
elif stage=='Offer审批':
ar=approvals.get(aid)
if not ar:add(aid,'待确认','审批字段未提供，不能将系统阶段等同于审批实际未结束',[src]);continue
source=[src,'审批截图/'+ar['截图文件']]
if ar['截图状态']=='已通过':
last=parse_date(ar['最后节点时间'])
if last is None:raise ValueError('审批通过却缺最后节点时间：'+aid)
due=cal.due(last,cal.limits['审批后更新'])
if cal.stat>due:add(aid,'规则超时',f'审批已通过，系统阶段仍未同步，已过按日期计算的更新截止日{due}',source)
elif cal.stat==due:add(aid,'截止当日待核',f'审批已通过，系统仍在审批阶段；{due}为截止当日，缺导出时刻/日内口径，不判已逾期',source)
else:add(aid,'期限内待同步',f'审批已通过，系统仍在审批阶段，更新截止日{due}尚未到',source)
elif cal.elapsed(enter)>cal.limits['Offer审批']:
add(aid,'规则超时',f'审批字段仍为审批中，阶段已{cal.elapsed(enter)}个工作日，超过{cal.limits["Offer审批"]}',source)
# 已接受是本课招聘终点；不据最后更新字段反推审批后的阶段历史。
for key,rs in groups.items():
if len(rs)>1:
aid=rs[0]['申请编号'];ids='/'.join(r['申请编号']for r in rs)
add(aid,'重复申请',f'{ids}同一完整号码代号，人数按一人；合并及主申请由HR确认',[f'{csv_name} 第'+str(r['_line'])+'行'for r in rs],key)
if len({r['当前阶段']for r in rs})>1:add(rs[0]['申请编号'],'待确认','同一人跨多个阶段，规则没有指定主申请，不擅选第一条归属',[csv_name],key)
for aid,item in plan.items():
if aid not in by_id:
ms=mails[aid];tokens={t for m in ms for t in m['号码代号']};key=next(iter(tokens))if len(tokens)==1 else'待核:'+aid
add(aid,'系统未录',f'计划有排期{item["date"]}，本次导出无该申请；核实身份后由HR补录，不并入系统人数',[f'招聘计划.xlsx[面试排期]第{item["line"]}行',*['回执邮件/'+m['文件']for m in ms]],key)
return {'统计日':str(cal.stat),'规则来源':rules['来源'],'完成说明':'机械核对完成，不代表业务无异常或已获批准','系统原始申请数':len(rows),'系统去重后身份数':len(groups),'身份待核':unknown,'多阶段待确认':multi,'阶段人数':dict(counts),'阶段原始行数':dict(raw),'系统外申请':[a for a in plan if a not in by_id],'涉及问题身份数':len({f['候选身份']for f in findings}),'机械核对事项':findings,'未归属邮件':sorted(set(unresolved)),'邮件分类':[{'文件':m['文件'],'类型':m['类型']}for m in messages]}
def main():
p=argparse.ArgumentParser(description=__doc__);p.add_argument('work');p.add_argument('approvals',nargs='?');p.add_argument('--rules',type=Path,default=Path(__file__).with_name('核对口径.json'));p.add_argument('--output-dir',type=Path);args=p.parse_args()
result=analyze(args.work,args.approvals,json.loads(args.rules.read_text(encoding='utf-8')))
if args.output_dir:
if args.output_dir.exists():raise ValueError('输出目录已存在，请换新名称')
args.output_dir.mkdir(parents=True)
(args.output_dir/'核对结果.json').write_text(json.dumps(result,ensure_ascii=False,indent=2)+'\n',encoding='utf-8')
with(args.output_dir/'核对事项.csv').open('w',encoding='utf-8-sig',newline='')as f:
writer=csv.DictWriter(f,fieldnames=['申请编号','候选身份','类型','事项','来源']);writer.writeheader();writer.writerows(result['机械核对事项'])
print(json.dumps(result,ensure_ascii=False,indent=2))
if __name__=='__main__':
try:main()
except Exception as exc:
print(f'核对未完成：{exc}',file=sys.stderr);sys.exit(2)