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.
40 lines
2.8 KiB
Python
40 lines
2.8 KiB
Python
import json,openpyxl,re
|
|
from pathlib import Path
|
|
from collections import defaultdict,Counter
|
|
P=Path(__file__).resolve().parents[1];R=P.parents[1];H=P.parent/'hydrogen-history-20260916'
|
|
s=json.loads((P/'september_calculation.json').read_text());paid=json.loads((H/'historical_paid.json').read_text())
|
|
raw=[];dif=[];controls=[]
|
|
for m in range(1,8):
|
|
f=R/f'里程任务考核_{m}月核算.xlsx' if m<7 else R/'7月绩效重算_多源考核里程_20260817/里程任务考核_7月核算.xlsx'
|
|
w=openpyxl.load_workbook(f,data_only=True,read_only=True);rows=list(w[f'{m}月奖金发放记录'].values)
|
|
for rn,r in enumerate(rows[1:],2):
|
|
if not r[0] or not isinstance(r[9],(int,float)) or r[9]<=0:continue
|
|
q=dict(payment_month=m,assessment_month=int(re.search(r'补发(\d+)月',r[4]).group(1)) if '补发' in r[4] else m,plate=r[0],owner=r[1],category=r[4],amount=r[9],source=str(f),source_row=rn);raw.append(q)
|
|
found=[p for p in paid if p['payment_month']==m and p['source_row']==rn]
|
|
if not found or any(q[k]!=found[0][k] for k in ['plate','owner','category','amount','assessment_month']):dif.append(q)
|
|
controls.append(dict(month=m,raw_sum=round(sum(x['amount'] for x in raw if x['payment_month']==m),2),locked_sum=round(sum(x['amount'] for x in paid if x['payment_month']==m),2)))
|
|
w.close()
|
|
allpaid=raw+[dict(payment_month=8,assessment_month=l['assessment_month'],plate=l['plate'],owner=l['sales'],category=l['category'],amount=l['gross_amount']) for l in s['august_ledger'] if l['gross_amount']>0]
|
|
back=[]
|
|
for l in s['ledger']:
|
|
if l['assessment_month']<9:
|
|
prior=[p for p in allpaid if p['assessment_month']==l['assessment_month'] and p['plate']==l['plate']]
|
|
back.append(dict(**l,prior_same_plate_month=prior))
|
|
byplate=defaultdict(list)
|
|
for l in s['ledger']:byplate[l['plate']].append(l)
|
|
multi=[dict(plate=p,items=ls) for p,ls in byplate.items() if len({l['sales'] for l in ls})>1]
|
|
# Check complete history eligibility independently from engine, all record/group state used only as metadata.
|
|
elig=[]
|
|
for g in s['groups']:
|
|
if g['virtual']:continue
|
|
h=[h for h in s['historical_months'] if h['plate']==g['plate'] and h['sales']==g['sales']]
|
|
t=sum(x['target'] for x in h)+g['target'];a=sum(x['actual'] for x in h)+g['actual']
|
|
for x in h:
|
|
settled=any(p['assessment_month']==x['month'] and p['plate']==g['plate'] and p['owner']==g['sales'] for p in allpaid)
|
|
due=a>=t>0 and not settled
|
|
has=any(l['plate']==g['plate'] and l['sales']==g['sales'] and l['assessment_month']==x['month'] for l in s['ledger'])
|
|
if due!=has:elig.append(dict(group=g,month=x,due=due,has=has))
|
|
result=dict(controls=controls,raw_vs_locked_differences=dif,backpay_checks=back,multiple_owners_same_plate_paid=multi,eligibility_mismatches=elig)
|
|
(P/'audit_history/history_checks.json').write_text(json.dumps(result,ensure_ascii=False,indent=2))
|
|
print(json.dumps(result,ensure_ascii=False,indent=2))
|