1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
| """Consolidate equipment categories and classify the ledger by equipment purpose."""
| import sys,json
| from pathlib import Path
| from collections import Counter
| P=Path(__file__).parent
| sys.stdout.reconfigure(encoding='utf8')
| sys.path.insert(0,str(P.parent/'iqc-import-20261012/deps'))
| import pymysql
| assert not (P/'result.json').exists(),'Already done; inspect instead of repeating.'
| # Exact names reviewed against the full ledger; avoid broad keyword mistakes
| # (e.g. process dryers vs compressed-air dryers; inline vs standalone tests).
| other_names={'燕窝成品柜（不锈钢）','海尔空调','双开门冰柜','螺杆式压缩机','冷冻式压缩空气干燥机','固定平台','组合式空调机组','除湿机','全自动洗衣机','单机除尘净化机组','空调系统','纯化水系统','冷却塔','循环水罐','叉车','电动拖地机','平衡重式叉车','密封性测试仪','电导率仪'}
| reviewed=json.loads((P/'before.json').read_text(encoding='utf8'))
| c=pymysql.connect(host='36.213.150.237',port=10000,user='root',password=sys.stdin.readline().strip(),database='mom-hm',charset='utf8mb4',autocommit=False,cursorclass=pymysql.cursors.DictCursor)
| try:
| with c.cursor() as q:
| q.execute("SET time_zone='+08:00'")
| q.execute('SELECT DATABASE() db,NOW() now,GET_LOCK(%s,10) locked',('equipment-types-20261014',));state=q.fetchone();assert state['db']=='mom-hm' and state['locked']==1;now=state['now']
| before={};schemas={}
| for t in ['mes_dv_machinery_type','mes_dv_machinery']:
| q.execute('SELECT * FROM '+t+' ORDER BY id FOR UPDATE');before[t]=q.fetchall()
| q.execute('SHOW COLUMNS FROM '+t);schemas[t]={x['Field'] for x in q.fetchall()}
| (P/'locked-before.json').write_text(json.dumps({'state':state,'tables':before},ensure_ascii=False,default=str,indent=2),encoding='utf8')
| live=lambda x:x['deleted'] in (0,b'\x00')
| machines=[x for x in before['mes_dv_machinery'] if live(x)]
| assert {(r['id'],r['name'],r['machinery_type_id']) for r in machines}=={(r['id'],r['name'],r['machinery_type_id']) for r in reviewed['mes_dv_machinery']['rows']},'Ledger changed after review'
| assert len(machines)==134
| q.execute('SELECT COUNT(*) n FROM system_users u JOIN system_user_role ur ON ur.user_id=u.id WHERE u.id=244 AND u.status=0 AND u.deleted=0 AND ur.role_id=166 AND ur.deleted=0');assert q.fetchone()['n']>0
| q.execute("SELECT table_name,column_name FROM information_schema.columns WHERE table_schema=DATABASE() AND (column_name LIKE '%machinery_type%' OR column_name LIKE '%equipment_type%')");refs=q.fetchall();assert {(x['TABLE_NAME'],x['COLUMN_NAME']) for x in refs}=={('mes_dv_machinery','machinery_type_id')},'New type reference requires migration'
| types={};new_type_ids=[]
| for label,code,sort in [('生产设备','SCSB',1),('其他设备','QTSB',2)]:
| existing=[x for x in before['mes_dv_machinery_type'] if live(x) and x['name']==label]
| if existing:
| target=min(existing,key=lambda x:x['id']);id=target['id']
| q.execute('UPDATE mes_dv_machinery_type SET parent_id=0,status=0,sort=%s,updater=%s,update_time=%s WHERE id=%s',(sort,'244',now,id))
| else:
| assert not any(x['code']==code for x in before['mes_dv_machinery_type']),('Code already exists',code)
| q.execute('INSERT INTO mes_dv_machinery_type (code,name,parent_id,status,sort,creator,updater,create_time,update_time,deleted) VALUES (%s,%s,0,0,%s,%s,%s,%s,%s,0)',(code,label,sort,'244','244',now,now));id=q.lastrowid;new_type_ids.append(id)
| types[label]=id
| plan=[]
| for old in machines:
| category='其他设备' if old['name'] in other_names else '生产设备'
| id=types[category]
| q.execute('UPDATE mes_dv_machinery SET machinery_type_id=%s,updater=%s,update_time=%s WHERE id=%s AND deleted=0',(id,'244',now,old['id']));assert q.rowcount==1
| plan.append({'id':old['id'],'code':old['code'],'name':old['name'],'old_type_id':old['machinery_type_id'],'type_id':id,'category':category})
| removed=[]
| for old in before['mes_dv_machinery_type']:
| if live(old) and old['id'] not in types.values():
| q.execute('UPDATE mes_dv_machinery_type SET deleted=1,updater=%s,update_time=%s WHERE id=%s AND deleted=0',('244',now,old['id']));assert q.rowcount==1;removed.append(old['id'])
| q.execute('SELECT * FROM mes_dv_machinery ORDER BY id');after=q.fetchall();expected={x['id']:x for x in plan};assert len(after)==len(before['mes_dv_machinery'])
| for old,new in zip(before['mes_dv_machinery'],after):
| assert old['id']==new['id'];diff={k for k in old if old[k]!=new[k]};assert diff<=({'machinery_type_id','updater','update_time'} if old['id'] in expected else set()),(old['id'],diff)
| if old['id'] in expected:assert new['machinery_type_id']==expected[old['id']]['type_id']
| q.execute('SELECT * FROM mes_dv_machinery_type ORDER BY id');after_types=q.fetchall();live_types=[x for x in after_types if live(x)];assert {(x['name'],x['id']) for x in live_types}==set(types.items()) and all(x['status']==0 and x['parent_id']==0 for x in live_types)
| assert all(x['machinery_type_id'] in types.values() for x in after if live(x))
| c.commit()
| q.execute('SELECT * FROM mes_dv_machinery ORDER BY id');assert q.fetchall()==after
| q.execute('SELECT * FROM mes_dv_machinery_type ORDER BY id');assert q.fetchall()==after_types
| result={'committed':True,'verified':True,'at':str(now),'types':types,'counts':dict(Counter(x['category'] for x in plan)),'total':len(plan),'new_type_ids':new_type_ids,'soft_deleted_old_type_ids':removed,'equipment_other_fields_unchanged':True,'mapping':plan}
| (P/'result.json').write_text(json.dumps(result,ensure_ascii=False,indent=2),encoding='utf8')
| print(json.dumps({k:v for k,v in result.items() if k!='mapping'},ensure_ascii=False))
| except Exception:c.rollback();raise
| finally:
| with c.cursor() as q:q.execute('SELECT RELEASE_LOCK(%s)',('equipment-types-20261014',))
| c.close()
|
|
