shishengliang
65904c8270
Preserve August hydrogen revision and September audit evidence; calculate interrupted vehicle history across all prior months. Record remaining carry-policy and layout-baseline discrepancies.
111 lines
11 KiB
Python
111 lines
11 KiB
Python
"""September mileage selection and awards, with January-August settlements locked."""
|
||
from pathlib import Path
|
||
from collections import defaultdict,Counter
|
||
from decimal import Decimal,ROUND_HALF_UP
|
||
import json,sys,copy,re,hashlib
|
||
import openpyxl
|
||
P=Path(__file__).resolve().parent;REPO=P.parents[1];H=P.parent/'hydrogen-history-20260916'
|
||
sys.path.insert(0,str(REPO));import calc_engine as e
|
||
def money(v):return float(Decimal(str(v)).quantize(Decimal('.01'),rounding=ROUND_HALF_UP))
|
||
def load(p):return json.loads(p.read_text())
|
||
def dump(name,x):(P/name).write_text(json.dumps(x,ensure_ascii=False,indent=2,default=str))
|
||
prepared=load(P/'prepared_records.json');aug=load(H/'august_calculation.json')
|
||
confirmed=dict(test_rule='沿用8月：名单内每日仪表减GPS严格大于50公里为测试日，取GPS；非测试日优先纯氢（含0），无纯氢取仪表；名单内无法判定且无纯氢时兜底GPS',test_threshold_km=50,august_actual_paid=45065.47,overlaps='交接日计前段，后段次日起算',new_policy='沿用上月原始39辆名单，按9月车型档位匹配',missing='沿用8月按已有里程累计并单列缺口；不将缺失解释为实测0',history='1—6月修正后的里程继续承接；1—8月已发及扣减锁定，不重复补发、不追扣')
|
||
assert aug['stats']['amount']==45065.47
|
||
ap=H/'里程任务考核_8月核算_纯氢修正版.xlsx'
|
||
wb=openpyxl.load_workbook(ap,read_only=True,data_only=True)
|
||
assert wb['8月汇总（新版界面）']['M22'].value==45065.47
|
||
rr=list(wb['8月奖金发放记录'].values);wb.close()
|
||
aug_rows={(r[0],r[1],r[4]):r for r in rr[1:] if r[0] and r[0]!='合计' and isinstance(r[9],(int,float))}
|
||
for l in aug['ledger']:
|
||
r=aug_rows[l['plate'],l['sales'],l['category']];assert r[9]==l['amount'] and r[8]==l['withheld']
|
||
daily=prepared['daily'];basis=Counter();gaps=[];byrow=defaultdict(list)
|
||
for x in daily:
|
||
t,g,h=x['meter'],x['gps'],x['hydrogen'];flag=False
|
||
if x['test_candidate']:flag=(Decimal(str(t))-Decimal(str(g))>50) if t is not None and g is not None else None
|
||
if flag is True:mode,v='测试日GPS',g
|
||
elif h is not None:mode,v='非测试日纯氢',h
|
||
elif x['test_candidate'] and flag is None:mode,v='测试名单无纯氢GPS',g
|
||
elif t is not None:mode,v='非测试日仪表',t
|
||
else:mode,v='缺失',None
|
||
needed='GPS' if x['test_candidate'] and h is None and flag is None else '仪表'
|
||
x.update(test_flag=flag,comparison=money(t-g) if t is not None and g is not None else None,selection=mode if v is not None else '缺失',included=v,needed=needed if v is None else '',reason=('测试候选无纯氢且无法判定，需GPS' if needed=='GPS' else '无纯氢且无仪表，普通车辆不自动以GPS替代') if v is None else '')
|
||
if v is None:gaps.append(x)
|
||
basis[x['selection']]+=1;byrow[x['source_row']].append(x)
|
||
records=prepared['records']
|
||
for r in records:
|
||
if r['销售经理']=='刘念念':r['部门名称']='业务一部'
|
||
ds=byrow[r['源表行']];km=money(sum(Decimal(str(x['included'])) for x in ds if x['included'] is not None));gap=sum(x['included'] is None for x in ds)
|
||
target=r['应考核里程(km)'];reward=r['月奖励'];q=km>=target and target>0
|
||
r.update(月份=9,实际行驶里程_km=km,是否达标='达标' if q else '未达标',奖励基准_达标=reward,奖励基准_未达标=0,奖励基准_结转=reward,奖励基准_补发=reward,奖金=reward*r['考核天数']/30 if q else 0,多跑=max(0,km-target),可结转=int(max(0,km-target)//r['月度目标里程']),缺口天数=gap,状态='待补里程' if gap else '数据齐全',取数方式='；'.join(f'{k}{v}天' for k,v in Counter(x['selection'] for x in ds).items()))
|
||
r['实际行驶里程(km)']=km;r['完成率(%)']=money(km/target*100)
|
||
# Rebuild original history only for unchanged monthly target/reward/day metadata;
|
||
# then apply the saved corrected mileage and actual-settlement state.
|
||
D={m:e.read_file(REPO/f'租赁任务考核_2026年{m}月.xlsx' if m<7 else REPO/'7月绩效重算_多源考核里程_20260817/租赁任务考核_2026年7月.xlsx',m) for m in range(1,8)}
|
||
G={m:e.grp(D[m],m) for m in D};ad={};e.calc_jan_carryover(G[1])
|
||
for m in range(2,8):e.calc_month(m,G,ad,G[m-1],G[m])
|
||
for x in load(H/'comparison.json'):G[x['month']][(x['plate'],x['sales'])]['实际']=x['corrected']
|
||
states=load(H/'historical_states.json')
|
||
for s in states:
|
||
g=G[s['month']][(s['plate'],s['sales'])]
|
||
g.update(cum_t=s['cum_target'],cum_a=s['cum_actual'],cum_q=s['cum_actual']>=s['cum_target'] and s['cum_target']>0,累计已发结转=s['used'],可结转=s['available'])
|
||
settled={(r['assessment_month'],r['plate'],r['owner']) for r in load(H/'historical_paid.json')}
|
||
settled.update((l['assessment_month'],l['plate'],l['sales']) for l in aug['ledger'] if l['gross_amount']>0)
|
||
aug_carry={(l['plate'],l['sales']) for l in aug['ledger'] if l['category']=='结转' and l['amount']>0}
|
||
G[8]=e.grp(aug['records'],8)
|
||
for x in aug['groups']:
|
||
k=x['plate'],x['sales'];old=G[7].get(k,{})
|
||
if k not in G[8]:
|
||
assert x['virtual'];G[8][k]=dict(recs=[],应考核=x['target'],实际=x['actual'],奖金=0,天数=x['days'],有达标=False,目标km=old['目标km'],奖励达标=x['reward'],奖励未达标=0,奖励结转=x['reward'],奖励补发=x['reward'],部门=x['department'],销售=x['sales'],车牌=x['plate'],虚拟=True)
|
||
g=G[8][k]
|
||
g.update(cum_t=x['cum_target'],cum_a=x['cum_actual'],cum_q=x['cum_qualified'],可结转=x['carry_left'],累计已发结转=old.get('累计已发结转',0)+int(k in aug_carry))
|
||
for m in range(1,9):
|
||
for k,g in G[m].items():
|
||
g[f'{m}月已发']=(m,*k) in settled
|
||
for bm in range(1,m):g.pop(f'补发{bm}月',None)
|
||
before=copy.deepcopy(G)
|
||
G[9]=e.grp(records,9);data={};e.calc_month(9,G,data,G[8],G[9],cumulative_history=True)
|
||
policy=set(prepared['policy_plates']);price=aug['rules']['new_policy']['award_by_target']
|
||
for k,g in G[9].items():
|
||
if k[1]=='刘念念':g['部门']='业务一部'
|
||
if g.get('虚拟') and k[0] in policy:
|
||
last=next((r for m in range(8,0,-1) for r in (aug['records'] if m==8 else D[m]) if (r['车牌号'],r['销售经理'])==k),None)
|
||
assert last is not None;reward=price[last['考核目标']]
|
||
g.update(奖励达标=reward,奖励结转=reward,奖励补发=reward,结转=reward)
|
||
for item in data[9]['结转']:
|
||
if (item['车牌'],item['销售'])==k:item['额']=reward
|
||
departed=set(aug['departed']['people']);ledger=[];excluded=[]
|
||
cats=['当月','结转']+[f'补发{m}月' for m in range(1,9)]+['累计补发9月']
|
||
for cat in cats:
|
||
for item in data[9].get(cat,[]):
|
||
k=(item['车牌'],item['销售']);g=G[9][k];m=int(re.search(r'补发(\d+)月',cat).group(1)) if '补发' in cat else 9
|
||
if k[1] in departed:excluded.append(dict(plate=k[0],sales=k[1],category=cat,amount=money(item['额'])));continue
|
||
if cat=='结转':base,days,den=g['奖励结转'],1,1
|
||
elif cat=='当月':
|
||
rs=[r for r in g['recs'] if r['是否达标']=='达标'];assert len({r['奖励基准_达标'] for r in rs})==1
|
||
base,days,den=rs[0]['奖励基准_达标'],sum(r['考核天数'] for r in rs),30
|
||
else:base,days,den=G[m][k]['奖励补发'],G[m][k]['天数'],e.DAYS[m]
|
||
assert m==9 or (m,*k) not in settled
|
||
exact=base*days/den;assert abs(exact-item['额'])<1e-7
|
||
gap=sum(r['缺口天数'] for r in g['recs']);history_gap=sum(r.get('缺口天数',0) for r in aug['records'] if (r['车牌号'],r['销售经理'])==k)
|
||
ledger.append(dict(id=len(ledger)+1,plate=k[0],sales=k[1],department=g['部门'],category=cat,assessment_month=m,base=base,days=days,month_days=den,amount=money(exact),gross_amount=money(exact),withheld=0,unrounded=exact,virtual=bool(g.get('虚拟')),gap_days=gap,history_gap_days=history_gap,status='按现有里程计算' if gap or history_gap else '数据齐全'))
|
||
groups=[]
|
||
for k,g in sorted(G[9].items(),key=lambda kv:(kv[0][1],kv[0][0])):
|
||
if k[1] in departed:continue
|
||
prev=G[8].get(k,{});pays=[l for l in ledger if (l['plate'],l['sales'])==k]
|
||
groups.append(dict(plate=k[0],sales=k[1],department=g['部门'],days=g['天数'],target=money(g['应考核']),actual=money(g['实际']),monthly_target=g['目标km'],reward=g['奖励达标'],qualified=g['有达标'],cum_target=money(g['cum_t']),cum_actual=money(g['cum_a']),cum_qualified=g['cum_q'],previous_carry=prev.get('可结转',0),carry_left=g['可结转'],used_carry=g['累计已发结转'],virtual=bool(g.get('虚拟')),total=money(sum(l['amount'] for l in pays)),gap_days=sum(r['缺口天数'] for r in g['recs']),history_gap_days=sum(r.get('缺口天数',0) for r in aug['records'] if (r['车牌号'],r['销售经理'])==k),history=[[money(G[m][k]['应考核']),money(G[m][k]['实际'])] if k in G[m] else [None,None] for m in range(1,10)]))
|
||
people=sorted({g['sales'] for g in groups});total=money(sum(l['amount'] for l in ledger))
|
||
summary=dict(amount=total,unrounded_total=money(sum(l['unrounded'] for l in ledger)),records=len(records),vehicles=len({r['车牌号'] for r in records}),groups=len(groups),ledger_items=len(ledger),people=len(people),daily=len(daily),gap_days=len(gaps),gap_records=sum(r['缺口天数']>0 for r in records),gap_vehicles=len({x['plate'] for x in gaps}),wholly_missing_records=sum(r['缺口天数']==r['考核天数'] for r in records),basis_days=dict(basis),test_days=sum(x['test_flag'] is True for x in daily),unknown_test_days=sum(x['test_candidate'] and x['test_flag'] is None for x in daily),new_policy_vehicles=len(policy&{r['车牌号'] for r in records}),categories={c:money(sum(l['amount'] for l in ledger if l['category']==c)) for c in cats},excluded_departed_amount=money(sum(x['amount'] for x in excluded)))
|
||
assert len({(l['plate'],l['sales'],l['category']) for l in ledger})==len(ledger)
|
||
assert abs(sum(g['total'] for g in groups)-total)<1e-7
|
||
for l in ledger:
|
||
if l['category']=='结转':assert not any(y['plate']==l['plate'] and y['sales']==l['sales'] and y['category'] in ('当月','累计补发9月') for y in ledger)
|
||
for s in prepared['sources'].values():assert hashlib.sha256(Path(s['snapshot']).read_bytes()).hexdigest()==s['sha256']
|
||
paid_schedule=[dict(month=m,plate=k[0],sales=k[1],days=g['天数'],reward=g['奖励补发'],settled=(m,*k) in settled,target=money(g['应考核']),actual=money(g['实际'])) for m,gs in before.items() for k,g in gs.items()]
|
||
display_categories=['当月','结转','累计补发9月']+[f'补发{m}月' for m in range(8,0,-1)]
|
||
confirmed['september_department_overrides']={'刘念念':'业务一部'}
|
||
result={**prepared,**dict(confirmed_rules=confirmed,daily=daily,records=records,gaps=gaps,groups=groups,ledger=ledger,people=people,categories=display_categories,summary=summary,historical_months=paid_schedule,august_ledger=aug['ledger'],excluded=excluded)}
|
||
dump('september_calculation.json',result);dump('confirmed_rules.json',confirmed);dump('calculation_summary.json',summary)
|
||
print(json.dumps(summary,ensure_ascii=False,indent=2))
|
||
print('Missing groups:',json.dumps([dict(plate=r['车牌号'],sales=r['销售经理'],days=r['缺口天数'],known=r['实际行驶里程(km)']) for r in records if r['缺口天数']],ensure_ascii=False))
|