"""User-authorized logical deletion of documents, retaining process master data and audit evidence."""
|
import sys,json,csv,hashlib
|
from pathlib import Path
|
P=Path(__file__).parent
|
sys.path.insert(0,str(P.parent/'iqc-import-20261012/deps'))
|
import pymysql
|
encode=lambda r:json.dumps(r,ensure_ascii=False,default=str,sort_keys=True)
|
snapshot=json.loads((P/'before-sales-production-delete.json').read_text(encoding='utf8'))
|
all_tables={r['TABLE_NAME'] for r in snapshot['columns']}
|
sales=['crm_sale_quotation_product','crm_sale_quotation','crm_contract_product','crm_contract','erp_sale_order_items','erp_sale_order']
|
production=['mes_pro_work_order_param_record_detail','mes_pro_work_order_param_record','mes_pro_task_pause_record',
|
'mes_pro_task_issue','mes_pro_task_assignee','mes_pro_feedback','mes_pro_card_process','mes_pro_card',
|
'mes_pro_andon_record','mes_pro_batch_worker','mes_pro_batch','mes_pro_bag_code_replace','mes_pro_bag_code',
|
'mes_pro_pallet','mes_pro_task','mes_pro_work_order_bom','mes_pro_work_order_process','mes_pro_work_order',
|
'mes_pro_mps_merge_item','mes_pro_mps']
|
warehouse=['mes_wm_product_issue_line','mes_wm_product_issue','mes_wm_return_issue_line','mes_wm_return_issue',
|
'mes_wm_item_consume_line','mes_wm_item_consume','mes_wm_product_produce_line','mes_wm_product_produce']
|
targets=[t for t in sales+production+warehouse if t in all_tables]
|
protected=[t for t in snapshot['tables'] if t not in targets]
|
protected += [t for t in all_tables if t.startswith(('mes_qc_','mdm_item','mes_md_item','mes_wm_material_stock'))]
|
protected += [t for t in ['crm_customer','mes_md_workstation','mes_wm_batch','mes_wm_transaction','system_operate_log'] if t in all_tables]
|
protected=sorted(set(protected))
|
args=dict(host='36.213.150.237',port=10000,user='root',password=sys.stdin.readline().strip(),database='mom-hm',charset='utf8mb4',cursorclass=pymysql.cursors.DictCursor,read_timeout=60)
|
c=pymysql.connect(**args,autocommit=False)
|
backups={};protected_hashes={};counts={};schemas={};scopes={}
|
try:
|
with c.cursor() as q:
|
q.execute('SELECT DATABASE() db,NOW() now');state=q.fetchone();assert state['db']=='mom-hm';now=state['now']
|
q.execute("SELECT GET_LOCK('codex-delete-sales-production-documents',0) locked");assert q.fetchone()['locked']==1
|
for table in targets:
|
q.execute('SHOW CREATE TABLE `'+table+'`');schemas[table]=q.fetchone()
|
assert '`deleted`' in schemas[table]['Create Table']
|
scope='deleted=0'+(' AND tenant_id=0' if '`tenant_id`' in schemas[table]['Create Table'] else '')
|
scopes[table]=(scope,[])
|
q.execute('SELECT * FROM `'+table+'` WHERE '+scope+' ORDER BY id FOR UPDATE');backups[table]=q.fetchall()
|
# Posted inventory operations are outside this cleanup; there are currently none in this scope.
|
assert all(not backups[t] for t in warehouse if t in backups), 'Inventory documents require separate handling'
|
for table in ['crm_contract','crm_sale_quotation','mes_pro_work_order','mes_pro_feedback']:
|
assert all(not r.get('process_instance_id') for r in backups[table]), 'Workflow instance requires handling: '+table
|
# Ensure finance/after-sales/sales warehouse have no live references into the documents being removed.
|
external={'crm_receivable':('contract_id','crm_contract'),'crm_receivable_plan':('contract_id','crm_contract'),
|
'crm_invoice':('contract_id','crm_contract'),'after_sale_ticket':('contract_id','crm_contract'),
|
'after_sale_return':('sale_order_id','erp_sale_order'),'mes_wm_product_sales':('sale_order_id','erp_sale_order'),
|
'mes_wm_return_sales':('sale_order_id','erp_sale_order'),'mes_wm_sales_notice':('sale_order_id','erp_sale_order')}
|
for table,(field,parent) in external.items():
|
ids=[r['id'] for r in backups[parent]]
|
if table not in all_tables or not ids:continue
|
q.execute('SELECT COUNT(*) n FROM `'+table+'` WHERE deleted=0 AND `'+field+'` IN ('+','.join('%s' for _ in ids)+')',ids)
|
assert q.fetchone()['n']==0, 'External business document reference: '+table
|
# Remove only permission/follow-up rows owned by these quotation and contract records.
|
quote_ids=[r['id'] for r in backups['crm_sale_quotation']];contract_ids=[r['id'] for r in backups['crm_contract']]
|
clauses=[];parameters=[]
|
for biz_type,ids in [(9,quote_ids),(5,contract_ids)]:
|
if ids:
|
clauses.append('(biz_type=%s AND biz_id IN ('+','.join('%s' for _ in ids)+'))');parameters += [biz_type]+ids
|
if clauses:
|
for table in ['crm_permission','crm_follow_up_record']:
|
if table not in all_tables:continue
|
scope='deleted=0 AND ('+' OR '.join(clauses)+')';scopes[table]=(scope,parameters)
|
q.execute('SHOW CREATE TABLE `'+table+'`');schemas[table]=q.fetchone()
|
q.execute('SELECT * FROM `'+table+'` WHERE '+scope+' ORDER BY id FOR UPDATE',parameters);backups[table]=q.fetchall()
|
# Attachment links are scoped to the deleted documents; shared physical files are preserved.
|
for table in ['storage_attachment','system_storage_attachment']:
|
if table not in all_tables:continue
|
parts=[];params=[]
|
for record_type,parent in [('crm_sale_quotation','crm_sale_quotation'),('crm_contract','crm_contract'),('erp_sale_order','erp_sale_order'),('mes_pro_work_order','mes_pro_work_order')]:
|
ids=[r['id'] for r in backups[parent]]
|
if ids:parts.append('(record_type=%s AND record_id IN ('+','.join('%s' for _ in ids)+'))');params += [record_type]+ids
|
if not parts:continue
|
scope='deleted=0 AND ('+' OR '.join(parts)+')';scopes[table]=(scope,params)
|
q.execute('SHOW CREATE TABLE `'+table+'`');schemas[table]=q.fetchone()
|
q.execute('SELECT * FROM `'+table+'` WHERE '+scope+' ORDER BY id FOR UPDATE',params);backups[table]=q.fetchall()
|
for table in protected:
|
q.execute('SELECT * FROM `'+table+'` ORDER BY id FOR SHARE');rows=q.fetchall()
|
protected_hashes[table]={'rows':len(rows),'sha256':hashlib.sha256(encode(rows).encode('utf8')).hexdigest()}
|
backup={'database':'mom-hm','logical_delete_only':True,'at':str(now),'schemas':schemas,'rows':backups,'protected':protected_hashes}
|
backup_path=P/'sales-production-documents-deletion-backup.json'
|
assert not backup_path.exists(),'This cleanup has already been backed up; inspect the prior result before rerunning'
|
backup_path.write_text(encode(backup),encoding='utf8')
|
updater='sales-pro-doc-clean-20261013'
|
for table,rows in backups.items():
|
if not rows:counts[table]=0;continue
|
fields=set(rows[0]);updates=['deleted=1'];params=[]
|
if 'updater' in fields:updates.append('updater=%s');params.append(updater)
|
if 'update_time' in fields:updates.append('update_time=%s');params.append(now)
|
ids=[r['id'] for r in rows]
|
q.execute('UPDATE `'+table+'` SET '+','.join(updates)+' WHERE deleted=0 AND id IN ('+','.join('%s' for _ in ids)+')',params+ids)
|
counts[table]=q.rowcount;assert q.rowcount==len(rows)
|
q.execute('SELECT * FROM `'+table+'` WHERE id IN ('+','.join('%s' for _ in ids)+') ORDER BY id',ids);after=q.fetchall()
|
assert len(after)==len(rows)
|
for old,new in zip(rows,after):
|
assert new['deleted'] in (1,True,b'\x01')
|
assert all(new[k]==v for k,v in old.items() if k not in {'deleted','updater','update_time'})
|
for table,(scope,params) in scopes.items():
|
q.execute('SELECT COUNT(*) n FROM `'+table+'` WHERE '+scope,params);assert q.fetchone()['n']==0
|
for table,h in protected_hashes.items():
|
q.execute('SELECT * FROM `'+table+'` ORDER BY id');rows=q.fetchall()
|
assert len(rows)==h['rows'] and hashlib.sha256(encode(rows).encode('utf8')).hexdigest()==h['sha256'], 'Protected data changed: '+table
|
c.commit()
|
result={'logical_delete_only':True,'deleted_by_table':counts,'total_deleted_rows':sum(counts.values()),
|
'production_master_data_preserved':True,'protected_tables_verified':len(protected_hashes),
|
'qc_ipqc_preserved':protected_hashes.get('mes_qc_ipqc',{}).get('rows'),
|
'backup_file':str(backup_path),'executed_at':str(now)}
|
(P/'sales-production-documents-deletion-result.json').write_text(encode(result),encoding='utf8')
|
with (P/'销售生产单据清理清单.csv').open('w',encoding='utf-8-sig',newline='') as f:
|
w=csv.writer(f);w.writerow(['数据表','逻辑删除条数','处理时间'])
|
for t,n in counts.items():w.writerow([t,n,str(now)])
|
finally:c.close()
|
# Verify committed state with a fresh connection.
|
c=pymysql.connect(**args)
|
try:
|
with c.cursor() as q:
|
for table,(scope,params) in scopes.items():
|
q.execute('SELECT COUNT(*) n FROM `'+table+'` WHERE '+scope,params);assert q.fetchone()['n']==0
|
for table,h in protected_hashes.items():
|
q.execute('SELECT * FROM `'+table+'` ORDER BY id');rows=q.fetchall()
|
assert len(rows)==h['rows'] and hashlib.sha256(encode(rows).encode('utf8')).hexdigest()==h['sha256']
|
result['verified_after_commit']=True
|
(P/'sales-production-documents-deletion-result.json').write_text(encode(result),encoding='utf8')
|
print(encode(result))
|
finally:c.close()
|
