from pathlib import Path
|
import sys,re,json,urllib.request
|
sys.stdout.reconfigure(encoding='utf-8')
|
sys.path.insert(0,str(Path('scripts/iqc-import-20261012/deps').resolve()))
|
import pymysql
|
s=Path('yudao-server/src/main/resources/application-local.yaml').read_text(encoding='utf-8')
|
x=s[s.index('jdbc:mysql://localhost:3306/'):]
|
pw=re.search(r'password:\s*([^\r\n#]+)',x).group(1).strip().strip('\"\'')
|
c=pymysql.connect(host='127.0.0.1',port=3306,user='root',password=pw,database='mom-hm',charset='utf8mb4',cursorclass=pymysql.cursors.DictCursor,autocommit=False)
|
out=Path('scripts/local-production-20261015')
|
try:
|
with c.cursor() as q:
|
q.execute('select id,code,parent_id,quantity_scheduled from mes_pro_work_order where id in (102,103,104) and deleted=0 for update')
|
before=q.fetchall();assert len(before)==3
|
backup=out/'gantt-roots-before.json'
|
if not backup.exists():backup.write_text(json.dumps(before,ensure_ascii=False,indent=2,default=str),encoding='utf-8')
|
for r in before:
|
assert r['code'] in ('WO_20261015001','WO_20261015002','WO_20261015003')
|
assert r['parent_id'] in (None,0)
|
q.execute('update mes_pro_work_order set parent_id=0 where id in (102,103,104) and parent_id is null and deleted=0')
|
c.commit()
|
finally:
|
c.rollback();c.close()
|
r=urllib.request.Request('http://127.0.0.1:48080/admin-api/mes/pro/task/gantt-list?status=1&type=1',headers={'Authorization':'Bearer test1','tenant-id':'1'})
|
with urllib.request.urlopen(r,timeout=20) as response:data=json.load(response)
|
assert data['code']==0,data
|
rows=data['data'];ids={r['id'] for r in rows}
|
orphans=[r['id'] for r in rows if r.get('parent') not in (None,0,'0','') and r['parent'] not in ids]
|
assert not orphans,orphans
|
roots=[r for r in rows if r.get('parent') in (None,0,'0','')]
|
tasks=[r for r in rows if r['type']==303]
|
assert len(roots)==3 and len(tasks)==9
|
assert all(r['startDate'] and r['endDate'] for r in tasks)
|
result={'root_count':len(roots),'task_count':len(tasks),'orphan_count':len(orphans),'all_tasks_have_dates':True}
|
(out/'gantt-verification.json').write_text(json.dumps(result,indent=2),encoding='utf-8')
|
print(json.dumps(result))
|
