from audit import connect,dump,P
|
import json,sys,urllib.request
|
from decimal import Decimal as D
|
sys.stdout.reconfigure(encoding='utf-8')
|
r=json.loads((P/'result-applied.json').read_text(encoding='utf-8'));assert r['applied']
|
c=connect();checks=[]
|
try:
|
with c.cursor() as q:
|
for row in r['links']:
|
q.execute('select m.*,w.quantity wo_quantity,w.quantity_produced wo_produced,w.parent_id,w.status wo_status,w.quantity_scheduled,w.order_source_code,w.request_date wo_due,l.count sale_quantity,l.product_unit_id,l.mps_id from mes_pro_mps m join mes_pro_work_order w on w.id=m.work_order_id join erp_sale_order_items l on l.id=m.sale_order_item_id where m.id=%s and m.deleted=0',(row['mps_id'],));v=q.fetchone()
|
assert v['quantity']==v['wo_quantity']==v['sale_quantity']==D(str(row['quantity']))
|
assert v['quantity_produced']==v['wo_produced']==0 and v['parent_id']==0 and v['status']==v['wo_status']==1
|
assert v['mps_id']==v['id'] and v['sale_order_no']==v['order_source_code']==row['order_no']
|
assert v['unit_measure_id']==v['product_unit_id']==row['unit_id']
|
assert v['request_date']==v['wo_due'] and v['request_date'].weekday()<5
|
checks.append({'product':row['name'],'quantity':row['quantity'],'unit':row['unit'],'due':str(v['request_date']),'mps':row['mps_code'],'work_order':row['wo_code'],'scheduled_quantity':str(v['quantity_scheduled']),'bom_count':row['bom_count']})
|
total=D(0)
|
for g in r['groups'].values():
|
q.execute('select * from crm_contract where id=%s',(g['contract_id'],));contract=q.fetchone()
|
q.execute('select * from erp_sale_order where id=%s',(g['order_id'],));order=q.fetchone()
|
q.execute('select * from crm_sale_quotation where id=%s',(g['quotation_id'],));quote=q.fetchone()
|
q.execute('select * from crm_receivable_plan where id=%s',(g['receivable_plan_id'],));plan=q.fetchone()
|
amount=D(g['amount']);assert contract['total_price']==order['total_price']==quote['total_price']==plan['price']==amount
|
assert all(x['customer_id']==1613 for x in [contract,order,quote,plan])
|
assert quote['contract_id']==contract['id']==order['contract_id']==plan['contract_id']
|
assert contract['order_id']==order['id'] and order['out_count']==0 and order['production_status']==0
|
assert plan['return_time'].strftime('%Y-%m-%d')=='2026-11-30' and plan['receivable_id'] is None
|
for t,col in [('crm_sale_quotation_product','quotation_id'),('crm_contract_product','contract_id'),('erp_sale_order_items','order_id')]:
|
q.execute('select sum(total_price) amount from '+t+' where '+col+'=%s and deleted=0',({'quotation_id':quote['id'],'contract_id':contract['id'],'order_id':order['id']}[col],));assert q.fetchone()['amount']==amount
|
total+=amount
|
assert total==D('2062353.18')
|
for table,count in r['protected_counts'].items():q.execute('select count(*) n from `'+table+'`');assert q.fetchone()['n']==count
|
ids=r['counts']['mes_pro_task'];current=next(x for x in r['links'] if x.get('task_ids'))
|
q.execute('select * from mes_pro_task where work_order_id=%s and deleted=0 order by start_time',(current['wo_id'],));tasks=q.fetchall();assert len(tasks)==3
|
assert tasks[0]['status']==2 and tasks[0]['produced_quantity']==800 and all(x['produced_quantity']==0 for x in tasks[1:])
|
assert all(t['start_time'].weekday()<5 and t['end_time'].weekday()<5 for t in tasks)
|
assert all(tasks[i]['end_time']<=tasks[i+1]['start_time'] for i in range(len(tasks)-1))
|
q.execute('select sum(feedback_quantity) qty from mes_pro_feedback where work_order_id=%s and deleted=0',(current['wo_id'],));assert q.fetchone()['qty']==800
|
dump('verification.json',{'passed':True,'total':str(total),'plans':checks,'groups':4,'today_packaged':800,'finished_goods_produced':0,'no_actual_receipts_or_warehousing':True})
|
lines=['# 10月15日至月底生产及销售录入清单','','客户：总公司；合计：2,062,353.18元；4组销售单据，13项产品。','回款：11月30日一次月结，仅建立回款计划。','今天：蜜桃味营养素饮品包装部分完成800盒，尚未形成末道成品；其余12工单待排产，不提前报工。','','| 产品 | 数量 | 单位 | 计划开工 | 交期 | 工单 |','|---|---:|---|---|---|---|']
|
for row,check in zip(r['links'],checks):lines.append('|'+ '|'.join(map(str,[row['name'],row['quantity'],row['unit'],row['planned_start'][:10],check['due'][:10],row['wo_code']]))+'|')
|
lines+=['','以下产品现有BOM不完整，未套用其他规格或虚构投料：'+ '、'.join(row['name'] for row in r['links'] if not row['bom_count'])+'。','原表中的项目号、进度备注均保留在对应单据；168盒、90瓶、180瓶等小批量保持原数，没有为了凑数量上调。','价格按表列结算单价及金额执行，未额外假设税率或追加税款。','新增了表内原编码缺失的3款成品；规范了相关成品的包装单位和规格，历史单据数量金额未改动。','工作日校对参考：https://www.gov.cn/zhengce/content/202511/content_7047090.htm']
|
(P/'录入清单.md').write_text('\n'.join(lines)+'\n',encoding='utf-8')
|
finally:c.rollback();c.close()
|
api=[]
|
for path in ['/crm/contract/get?id='+str(g['contract_id']) for g in r['groups'].values()]+['/mes/pro/task/gantt-list?status=1&type=1']:
|
request=urllib.request.Request('http://127.0.0.1:48080/admin-api'+path,headers={'Authorization':'Bearer test1','tenant-id':'1'})
|
with urllib.request.urlopen(request,timeout=25) as response:result=json.load(response)
|
assert result['code']==0,(path,result)
|
if 'gantt-list' in path:
|
nodes=result['data'];all_ids={n['id'] for n in nodes};assert all(n.get('parent') in (None,'',0,'0') or n['parent'] in all_ids for n in nodes)
|
api.append({'path':path,'ok':True})
|
dump('api-verification.json',api)
|
print(json.dumps({'passed':True,'plans':13,'work_orders':13,'sales_chains':4,'amount':'2062353.18','today_partial_packaging':800,'api_checks':len(api)},ensure_ascii=False))
|
