"""Sales item need-production = yes; native draft MPS per item, idempotent links."""
|
import sys,json,csv,collections
|
from pathlib import Path
|
from decimal import Decimal
|
P=Path(__file__).parent
|
sys.path.insert(0,str(P.parent/'iqc-import-20261012/deps'))
|
import pymysql
|
s=json.loads((P/'before-sales-mps-snapshot.json').read_text(encoding='utf8'))
|
def norm(v):return json.loads(json.dumps(v,default=str))
|
def dump(n,v):(P/n).write_text(json.dumps(v,ensure_ascii=False,indent=2,default=str),encoding='utf8')
|
c=pymysql.connect(host='36.213.150.237',port=10000,user='root',password=sys.stdin.readline().strip(),database='mom-hm',charset='utf8mb4',cursorclass=pymysql.cursors.DictCursor,connect_timeout=10,read_timeout=60)
|
log=[]
|
try:
|
with c.cursor() as q:
|
q.execute('SELECT DATABASE() db');assert q.fetchone()['db']=='mom-hm'
|
q.execute("SELECT GET_LOCK('sales-mps-20261012',10) ok");assert q.fetchone()['ok']==1
|
c.begin();before={}
|
for t in ['erp_sale_order','erp_sale_order_items','mes_pro_mps','mes_pro_mps_merge_item','mes_md_item','mdm_item','mdm_unit_measure']:
|
q.execute('SELECT ENGINE FROM information_schema.tables WHERE table_schema=DATABASE() AND table_name=%s',(t,));assert q.fetchone()['ENGINE']=='InnoDB'
|
q.execute('SELECT * FROM '+t+' WHERE deleted=0 ORDER BY id FOR UPDATE');before[t]=list(q.fetchall())
|
assert norm(before[t])==s[t],f'{t} changed; re-review required'
|
dump('before-sales-mps-import.json',before)
|
orders={o['id']:o for o in before['erp_sale_order']}
|
assert all(o['status']==20 and o['production_status']==0 for o in orders.values())
|
mes=collections.defaultdict(list)
|
for i in before['mes_md_item']:mes[i['mdm_item_id']].append(i)
|
active=collections.defaultdict(list)
|
for m in before['mes_pro_mps']:
|
if m['status']!=3:active[m['sale_order_item_id']].append(m)
|
values=[];reused=[];codes={m['code'] for m in before['mes_pro_mps']};unit_issues=[]
|
for item in before['erp_sale_order_items']:
|
assert item['order_id'] in orders and Decimal(item['count'])>0
|
assert len(mes[item['product_id']])==1
|
product=mes[item['product_id']][0];order=orders[item['order_id']]
|
assert product['status']==0
|
if item['product_unit_id']!=product['unit_measure_id']:unit_issues.append(dict(sale_order_item_id=item['id'],sale_order_no=order['no'],product_code=product['code'],product_name=product['name'],specification=product['specification'],sale_unit_id=item['product_unit_id'],mes_unit_id=product['unit_measure_id'],quantity=str(item['count']),note='按原生MPS生成规则采用MES产品单位，数量为整个产品SKU数量；订单包装单位不同，后续下发前需核实主数据单位口径。'))
|
existing=active[item['id']];assert len(existing)<=1
|
if existing:
|
old=existing[0];assert old['product_id']==product['id'] and old['quantity']==item['count'];reused.append(old);continue
|
code='MPS20261012-SO-'+str(item['id']).zfill(6);assert code not in codes;codes.add(code)
|
values.append(dict(code=code,status=0,sale_order_id=order['id'],sale_order_no=order['no'],sale_order_item_id=item['id'],product_id=product['id'],quantity=item['count'],quantity_produced=Decimal(0),unit_measure_id=product['unit_measure_id'],client_id=order['customer_id'],request_date=order['order_time'],merge_flag=0,remark='由销售订单生成：'+order['no']+'；对应明细'+str(item['id'])+'；需要生产改为是；保留订单数量，待计划下发。'))
|
if values:
|
for v in values:v['creator']='1';v['updater']='1'
|
cols=list(values[0]);sql='INSERT INTO mes_pro_mps (`'+'`,`'.join(cols)+'`) VALUES ('+','.join(['%s']*len(cols))+')'
|
args=[tuple(v[k] for k in cols) for v in values];log.extend(q.mogrify(sql,a)+';' for a in args);q.executemany(sql,args)
|
# Scope to the reviewed order IDs; do not change approval, prices, quantities or shipment data.
|
ids=list(orders);marks=','.join(['%s']*len(ids))
|
sql="UPDATE erp_sale_order_items i JOIN mes_pro_mps m ON m.sale_order_item_id=i.id AND m.deleted=0 AND m.status<>3 SET i.need_production=1,i.mps_id=m.id,i.updater='1' WHERE i.deleted=0 AND i.order_id IN ("+marks+')'
|
log.append(q.mogrify(sql,ids)+';');q.execute(sql,ids);changed=q.rowcount
|
q.execute('SELECT * FROM mes_pro_mps WHERE deleted=0 ORDER BY id');after_mps=list(q.fetchall())
|
q.execute('SELECT * FROM erp_sale_order_items WHERE deleted=0 ORDER BY id');after_items=list(q.fetchall())
|
final={m['sale_order_item_id']:m for m in after_mps if m['status']!=3}
|
assert len(final)==len(before['erp_sale_order_items']) and len(after_mps)==len(before['mes_pro_mps'])+len(values)
|
for item,old in zip(after_items,before['erp_sale_order_items']):
|
assert item['id']==old['id'] and item['need_production']==1
|
m=final[item['id']];o=orders[item['order_id']];prd=mes[item['product_id']][0]
|
assert item['mps_id']==m['id'] and m['sale_order_id']==o['id'] and m['sale_order_no']==o['no'] and m['product_id']==prd['id'] and m['quantity']==item['count'] and m['unit_measure_id']==prd['unit_measure_id'] and m['client_id']==o['customer_id'] and m['request_date']==o['order_time']
|
if m['code'].startswith('MPS20261012-SO-'):assert m['status']==0 and m['quantity_produced']==0 and m['work_order_id'] is None and m['issue_date'] is None
|
for k in old:
|
if k not in ['need_production','mps_id','updater','update_time']:assert old[k]==item[k]
|
q.execute('SELECT * FROM erp_sale_order WHERE deleted=0 ORDER BY id');assert list(q.fetchall())==before['erp_sale_order']
|
for t in ['mes_pro_mps_merge_item','mes_md_item','mdm_item','mdm_unit_measure']:
|
q.execute('SELECT * FROM '+t+' WHERE deleted=0 ORDER BY id');assert list(q.fetchall())==before[t]
|
result={'database':'mom-hm','sale_orders':len(orders),'sales_items_need_production_yes':len(after_items),'new_draft_mps':len(values),'existing_mps_reused':len(reused),'sales_items_updated':changed,'unit_difference_items':len(unit_issues),'production_status':'未完成','mps_status':'草稿','duplicate_active_mps':0,'quantities_and_product_mapping_verified':True,'sales_financial_and_approval_fields_preserved':True,'verified':True}
|
c.commit();dump('sales-mps-import-result.json',result);dump('sales-mps-generated.json',after_mps);dump('sales-mps-unit-differences.json',unit_issues)
|
(P/'sales-mps-executed.sql').write_text('START TRANSACTION;\n'+'\n'.join(log)+'\nCOMMIT;\n',encoding='utf8')
|
units={u['id']:u['name'].strip() for u in before['mdm_unit_measure']};prodids={p['id']:p for p in before['mes_md_item']}
|
with (P/'销售订单生产主计划对应.csv').open('w',encoding='utf-8-sig',newline='') as f:
|
w=csv.writer(f);w.writerow(['销售订单号','订单明细ID','主计划ID','主计划编码','产品编码','产品名称','规格','计划数量','计划单位','状态','需求日期','客户ID'])
|
for m in after_mps:
|
if m['sale_order_item_id'] not in final:continue
|
prd=prodids[m['product_id']];w.writerow([m['sale_order_no'],m['sale_order_item_id'],m['id'],m['code'],prd['code'],prd['name'],prd['specification'],m['quantity'],units.get(m['unit_measure_id']),'草稿' if m['status']==0 else m['status'],m['request_date'],m['client_id']])
|
with (P/'销售生产单位口径待核实.csv').open('w',encoding='utf-8-sig',newline='') as f:
|
w=csv.writer(f);w.writerow(['销售订单号','明细ID','产品编码','产品名称','规格','销售单位','MES及MPS单位','计划数量','说明'])
|
for a in unit_issues:w.writerow([a['sale_order_no'],a['sale_order_item_id'],a['product_code'],a['product_name'],a['specification'],units.get(a['sale_unit_id']),units.get(a['mes_unit_id']),a['quantity'],a['note']])
|
print(json.dumps(result,ensure_ascii=False));q.execute("SELECT RELEASE_LOCK('sales-mps-20261012')")
|
except:c.rollback();raise
|
finally:c.close()
|
