import sys,json,csv,collections
|
from decimal import Decimal
|
from pathlib import Path
|
sys.path.insert(0,str(Path('scripts/iqc-import-20261012/deps').resolve()))
|
import pymysql
|
apply='--apply' in sys.argv
|
conn=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=15,read_timeout=60,autocommit=False)
|
out=Path('scripts/mps-link-data-20261014');out.mkdir(exist_ok=True)
|
def save(name,data): (out/name).write_text(json.dumps(data,ensure_ascii=False,indent=2,default=str),encoding='utf8')
|
with conn.cursor() as q:
|
q.execute('SELECT p.*,u.name AS plan_unit_name FROM mes_pro_mps p LEFT JOIN mes_md_unit_measure u ON u.id=p.unit_measure_id AND u.deleted=0 WHERE p.deleted=0 ORDER BY p.id')
|
plans=q.fetchall()
|
q.execute('SELECT i.*,o.no,o.customer_id,o.order_time,u.name AS sale_unit_name FROM erp_sale_order_items i JOIN erp_sale_order o ON o.id=i.order_id AND o.deleted=0 LEFT JOIN mdm_unit_measure u ON u.id=i.product_unit_id AND u.deleted=0 WHERE i.deleted=0')
|
items=q.fetchall()
|
q.execute('SELECT id,code,mdm_item_id,name FROM mes_md_item WHERE deleted=0')
|
products={x['id']:x for x in q.fetchall()}
|
q.execute('SELECT id,code FROM mdm_item WHERE deleted=0')
|
mdm_by_code=collections.defaultdict(list)
|
for x in q.fetchall():mdm_by_code[x['code']].append(x['id'])
|
save('strict-audit-before.json',{'plans':plans,'items':items,'products':products})
|
rows=[];matches=[]
|
for plan in plans:
|
product=products.get(plan['product_id'],{})
|
mdm_id=product.get('mdm_item_id')
|
if not mdm_id and len(mdm_by_code[product.get('code')])==1:mdm_id=mdm_by_code[product['code']][0]
|
choices=[x for x in items if mdm_id and x['product_id']==mdm_id]
|
qty=[x for x in choices if Decimal(str(x['count']))==Decimal(str(plan['quantity']))]
|
unit=[x for x in qty if plan['plan_unit_name'] and x['sale_unit_name'] and plan['plan_unit_name'].strip()==x['sale_unit_name'].strip()]
|
month=[x for x in unit if plan['request_date'] and x['order_time'] and plan['request_date'].strftime('%Y-%m')==x['order_time'].strftime('%Y-%m')]
|
available=[x for x in month if not x['mps_id'] or x['mps_id']==plan['id']]
|
reason=('产品无对应销售明细' if not choices else '数量不一致' if not qty else '单位不一致或缺失' if not unit else '月份不一致或缺失' if not month else '销售明细已关联其他主计划' if not available else '多条符合条件的销售明细' if len(available)>1 else '唯一候选')
|
if plan['sale_order_id'] or plan['sale_order_no'] or plan['sale_order_item_id']:reason='已有来源关联，保持原样'
|
row={'主计划ID':plan['id'],'主计划编码':plan['code'],'产品编码':product.get('code',''),'产品名称':product.get('name',''),'计划数量':plan['quantity'],'计划单位':plan['plan_unit_name'],'需求日期':plan['request_date'],'同产品候选数':len(choices),'同数量候选数':len(qty),'同单位候选数':len(unit),'同月份候选数':len(month),'候选销售单号':'；'.join(sorted(set(x['no'] for x in available))),'结果':reason}
|
rows.append(row)
|
if reason=='唯一候选':matches.append((plan,available[0],row))
|
reverse=collections.Counter(x[1]['id'] for x in matches)
|
for plan,item,row in matches:
|
if reverse[item['id']]>1:row['结果']='同一销售明细匹配多个主计划，暂不关联'
|
matches=[x for x in matches if reverse[x[1]['id']]==1]
|
save('confirmed-candidates.json',[{'mps_id':p['id'],'mps_code':p['code'],'order_id':i['order_id'],'order_no':i['no'],'item_id':i['id']} for p,i,r in matches])
|
if apply and matches:
|
try:
|
with conn.cursor() as q:
|
backups=[]
|
for plan,item,row in matches:
|
q.execute('SELECT * FROM mes_pro_mps WHERE id=%s AND deleted=0 FOR UPDATE',(plan['id'],));old_plan=q.fetchone()
|
q.execute('SELECT * FROM erp_sale_order_items WHERE id=%s AND deleted=0 FOR UPDATE',(item['id'],));old_item=q.fetchone()
|
assert old_plan and old_item and not any(old_plan[x] for x in ['sale_order_id','sale_order_no','sale_order_item_id'])
|
assert not old_item['mps_id'] or old_item['mps_id']==plan['id']
|
backups.append({'plan':old_plan,'item':old_item})
|
save('updated-records-backup.json',backups)
|
for plan,item,row in matches:
|
q.execute('UPDATE mes_pro_mps SET sale_order_id=%s,sale_order_no=%s,sale_order_item_id=%s,update_time=update_time WHERE id=%s AND deleted=0',(item['order_id'],item['no'],item['id'],plan['id']));assert q.rowcount==1
|
if not item['mps_id']:
|
q.execute('UPDATE erp_sale_order_items SET mps_id=%s,update_time=update_time WHERE id=%s AND deleted=0 AND (mps_id IS NULL OR mps_id=0)',(plan['id'],item['id']));assert q.rowcount==1
|
row['结果']='已关联'
|
for backup in backups:
|
q.execute('SELECT * FROM mes_pro_mps WHERE id=%s',(backup['plan']['id'],));now_plan=q.fetchone()
|
q.execute('SELECT * FROM erp_sale_order_items WHERE id=%s',(backup['item']['id'],));now_item=q.fetchone()
|
changed=[k for k,v in backup['plan'].items() if k not in ['sale_order_id','sale_order_no','sale_order_item_id'] and now_plan[k]!=v]
|
assert not changed, 'Unexpected plan fields changed: '+','.join(changed)
|
assert all(now_item[k]==v for k,v in backup['item'].items() if k!='mps_id')
|
assert now_item['mps_id']==now_plan['id'] and now_plan['sale_order_item_id']==now_item['id'] and now_plan['sale_order_id']==now_item['order_id']
|
conn.commit()
|
except Exception:
|
conn.rollback();raise
|
else:conn.rollback()
|
with (out/'生产主计划销售订单严格匹配结果.csv').open('w',encoding='utf-8-sig',newline='') as f:
|
w=csv.DictWriter(f,fieldnames=list(rows[0]));w.writeheader();w.writerows(rows)
|
summary={'total_plans':len(plans),'unique_strict_matches':len(matches),'applied':len(matches) if apply else 0,'results':dict(collections.Counter(x['结果'] for x in rows))}
|
save('strict-match-summary.json',summary)
|
print(json.dumps(summary,ensure_ascii=False))
|
conn.close()
|
