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
| import json,sys
| from pathlib import Path
| P=Path('scripts/quotation-import')
| target=sys.argv[1] if len(sys.argv)>1 else 'local'
| assert target in ['local','remote']
| plan=json.loads((P/f'{target}-plan.json').read_text(encoding='utf-8'))
| def qs(v):return 'NULL' if v is None else ('CONVERT(0x'+str(v).encode().hex()+' USING utf8mb4)' if str(v) else "''")
| sql=[]
| def table(name,columns):sql.append(f'CREATE TEMPORARY TABLE {name}({columns}) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci')
| def insert(table,rows):
| for start in range(0,len(rows),80):sql.append('INSERT INTO '+table+' VALUES '+','.join('('+','.join(r)+')' for r in rows[start:start+80]))
| table('stage_quote_customer','name VARCHAR(255) PRIMARY KEY')
| table('stage_quote_material','code VARCHAR(50) PRIMARY KEY,name VARCHAR(200),specification VARCHAR(200),unit_id BIGINT,item_type INT')
| table('stage_quote_header','no VARCHAR(32) PRIMARY KEY,name VARCHAR(128),customer_name VARCHAR(255),quotation_time DATETIME,valid_until DATETIME,total DECIMAL(24,6),remark VARCHAR(1024)')
| table('stage_quote_line','no VARCHAR(32),line_no INT,item_code VARCHAR(50),expected_id BIGINT,expected_name VARCHAR(200),expected_base_unit BIGINT,unit_id BIGINT,item_price DECIMAL(24,6),quotation_price DECIMAL(24,6),count DECIMAL(24,6),total DECIMAL(24,6),remark VARCHAR(256),PRIMARY KEY(no,line_no),UNIQUE(no,item_code)')
| insert('stage_quote_customer',[[qs(name)] for name in dict.fromkeys(q['customer'] for q in plan['quotes'])])
| insert('stage_quote_material',[[qs(x['code']),qs(x['name']),qs(x['specification']),str(x['unit_id']),str(x['item_type'])] for x in plan['new_products']])
| insert('stage_quote_header',[[qs(q[k]) for k in ['no','name','customer','quotation_time','valid_until']]+[q['total'],qs(q['remark'])] for q in plan['quotes']])
| rows=[]
| for q in plan['quotes']:
| assert len({x['item_code'] for x in q['items']})==len(q['items']),q['no']
| for x in q['items']:rows.append([qs(q['no']),str(x['line']),qs(x['item_code']),str(x['item_id']),qs(x['item_name']),str(x['expected_base_unit']),str(x['item_unit_id']),x['original_price'],x['price'],x['count'],x['total'],qs(x['remark'])])
| insert('stage_quote_line',rows)
| (P/f'{target}-import.sql').write_text(';\n'.join(sql)+';\n',encoding='utf-8')
| print('Generated staging SQL for',len(plan['quotes']),'quotes and',len(rows),'lines')
|
|
