"""仅修复 application-local 对应的本地库，并回滚验证 380/500 字符可写入。"""
|
from pathlib import Path
|
import sys,re,json
|
sys.stdout.reconfigure(encoding='utf-8')
|
sys.path.insert(0,str(Path('scripts/iqc-import-20261012/deps').resolve()))
|
import pymysql
|
config=Path('yudao-server/src/main/resources/application-local.yaml').read_text(encoding='utf-8')
|
section=config[config.index('jdbc:mysql://localhost:3306/'):]
|
password=re.search(r'password:\s*([^\r\n#]+)',section).group(1).strip().strip('\"\'')
|
c=pymysql.connect(host='127.0.0.1',port=3306,user='root',password=password,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('SHOW CREATE TABLE mes_pro_work_order_param_record_detail')
|
before=q.fetchone()
|
backup=out/'snapshot-detail-schema-before.json'
|
if not backup.exists(): backup.write_text(json.dumps(before,ensure_ascii=False,indent=2),encoding='utf-8')
|
q.execute("select character_maximum_length n from information_schema.columns where table_schema=database() and table_name='mes_pro_work_order_param_record_detail' and column_name='remark'")
|
width=q.fetchone()['n'];assert width in (255,500),width
|
if width==255:
|
sql=Path('sql/mysql/mes_pro_param_snapshot_remark_20261015.sql').read_text(encoding='utf-8')
|
q.execute(sql)
|
q.execute('select id,remark from mes_pd_process_param_template_detail where template_id=11 and deleted=0 limit 1')
|
source=q.fetchone();assert source
|
for remark in (source['remark'],'检'*500):
|
q.execute('SAVEPOINT snapshot_remark_probe')
|
q.execute('insert into mes_pro_work_order_param_record_detail(record_id,param_name,remark) select id,%s,%s from mes_pro_work_order_param_record where deleted=0 limit 1',('备注长度校验',remark))
|
assert q.rowcount==1
|
q.execute('select remark from mes_pro_work_order_param_record_detail where id=%s',(q.lastrowid,))
|
assert q.fetchone()['remark']==remark
|
q.execute('ROLLBACK TO SAVEPOINT snapshot_remark_probe')
|
c.rollback()
|
q.execute('select w.code,w.quantity_scheduled,count(t.id) tasks from mes_pro_work_order w left join mes_pro_task t on t.work_order_id=w.id and t.deleted=0 where w.id in (102,103,104) group by w.id')
|
result={'new_width':500,'source_remark_length':len(source['remark']),'round_trip_500_chinese_characters':True,'probe_rows_rolled_back':True,'orders':q.fetchall()}
|
(out/'snapshot-remark-verification.json').write_text(json.dumps(result,ensure_ascii=False,indent=2,default=str),encoding='utf-8')
|
print(json.dumps(result,ensure_ascii=False,default=str))
|
finally:
|
c.rollback();c.close()
|
