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
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
| """Import the reviewed September ledger atomically; no inventory or production postings."""
| import sys,json,hashlib,re
| from pathlib import Path
| from decimal import Decimal
| from datetime import datetime,timedelta
| P=Path(__file__).parent
| sys.path.insert(0,str(P.parent/'iqc-import-20261012/deps'))
| import pymysql
| D=Decimal
| plan=json.loads((P/'import-plan.json').read_text(encoding='utf8'))
| source=Path(plan['source']);assert hashlib.sha256(source.read_bytes()).hexdigest()==plan['sha256']
| 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)
| tables=['crm_customer','mdm_item','mes_md_item','crm_sale_quotation','crm_sale_quotation_product','crm_contract','crm_contract_product','erp_sale_order','erp_sale_order_items','crm_invoice','crm_receivable','crm_receivable_plan','crm_permission']
| result={'source_sha256':plan['sha256'],'counts':plan['counts'],'totals':plan['totals'],'groups':[],'new_customers':{},'new_items':{},'excluded':plan['excluded'],'rules':plan['rules']}
| try:
| with c.cursor() as q:
| q.execute('SELECT DATABASE() db,NOW() now,GET_LOCK(%s,10) acquired',('sales-ledger-202609',));state=q.fetchone();assert state['db']=='mom-hm' and state['acquired']==1
| now=state['now'];result['imported_at']=str(now)
| q.execute('SELECT id,status FROM system_users WHERE id IN (244,291) AND deleted=0');users=q.fetchall();assert len(users)==2 and all(u['status']==0 for u in users)
| q.execute("SELECT COUNT(*) n FROM crm_sale_quotation WHERE no LIKE 'BJ-202609-L%%'");assert q.fetchone()['n']==0,'Already imported; inspect existing result instead of duplicating.'
| backup={'state':state,'tables':{}}
| schemas={}
| for t in tables:
| q.execute('SHOW CREATE TABLE `'+t+'`');ddl=q.fetchone()['Create Table']
| q.execute('SHOW COLUMNS FROM `'+t+'`');schemas[t]={v['Field']:v for v in q.fetchall()}
| q.execute('SELECT * FROM `'+t+'` ORDER BY id');records=q.fetchall()
| backup['tables'][t]={'ddl':ddl,'rows':records}
| (P/'locked-before-import.json').write_text(json.dumps(backup,ensure_ascii=False,default=str,indent=2),encoding='utf8')
| def insert(t,values,actor=291):
| values={**values,'creator':str(actor),'updater':str(actor),'create_time':now,'update_time':now,'deleted':0}
| if 'tenant_id' in schemas[t]:values['tenant_id']=0
| assert all(k in schemas[t] for k in values),(t,set(values)-set(schemas[t]))
| for k,col in schemas[t].items():
| assert k in values or col['Null']=='YES' or col['Default'] is not None or col['Extra']=='auto_increment',(t,'missing required',k)
| for k,v in values.items():
| m=re.match(r'varchar\((\d+)\)',schemas[t][k]['Type'])
| if m and v is not None:assert len(str(v))<=int(m[1]),(t,k,len(str(v)))
| keys=list(values);q.execute('INSERT INTO `'+t+'` ('+','.join('`'+k+'`' for k in keys)+') VALUES ('+','.join(['%s']*len(keys))+')',[values[k] for k in keys]);return q.lastrowid
| def permit(biz,id):return insert('crm_permission',{'biz_type':biz,'biz_id':id,'user_id':291,'level':1})
| def active(t):return [r for r in backup['tables'][t]['rows'] if r['deleted'] in (0,False,b'\x00')]
| customers={r['name']:r for r in active('crm_customer')}
| for name,x in plan['new_customers'].items():
| if name in customers:result['new_customers'][name]=customers[name]['id'];continue
| id=insert('crm_customer',{'name':name,'follow_up_status':0,'owner_user_id':291,'owner_time':now,'lock_status':0,'deal_status':1,'remark':x['remark']})
| permit(2,id);customers[name]={'id':id};result['new_customers'][name]=id
| itemrows=active('mdm_item');items={r['id']:r for r in itemrows};itemcodes={r['code']:r for r in itemrows}
| normalize=lambda s:re.sub(r'[\s\-（）()/*×]','',str(s or '').lower().replace('g','克'))
| for code,x in plan['new_items'].items():
| if code in itemcodes:result['new_items'][code]=itemcodes[code]['id'];continue
| equivalents=[r for r in itemrows if normalize(r['name'])==normalize(x['name']) and normalize(r['specification'])==normalize(x['specification'])]
| assert not equivalents,('Do not duplicate equivalent item',code,[r['id'] for r in equivalents])
| remark=f'九月销售台账缺项补充；来源项目号{code}；未推算库存或成本'
| id=insert('mdm_item',{'code':code,**x,'status':0,'is_batch_managed':1,'is_serial_managed':0,'high_value':0,'remark':remark},actor=244)
| q.execute('SELECT COUNT(*) n FROM mes_md_item WHERE id=%s',(id,));assert q.fetchone()['n']==0,'Shared item ID collision'
| insert('mes_md_item',{'id':id,'mdm_item_id':id,'code':code,'name':x['name'],'specification':x['specification'],'unit_measure_id':x['unit_measure_id'],'item_type_id':x['category_id'],'status':0,'safe_stock_flag':0,'high_value':0,'batch_flag':1,'remark':remark},actor=244)
| items[id]={'id':id,**x};result['new_items'][code]=id
| for g in plan['groups']:
| key=g['key'];cid=customers[g['customer_name']]['id'];gross=D(g['gross']);net=D(g['net']);tax=D(g['tax']);rounding=D(g['rounding_adjustment']);day=datetime.fromisoformat(g['day']);lines=g['lines']
| origin='/'.join(g['source_numbers']);base=f'九月台账补录{key}，行{g["rows"][0]}-{g["rows"][1]}；原单{origin}；报价和合同为反推，业务日期按开票/备注日期，非原签约日。'
| note=base+f'含税合并金额{gross}；净税额与合并金额尾差{rounding}元，按台账合并金额登记。'+g['original']
| contractno='HT-'+key;orderno='XSDD-'+key;quoteno='BJ-'+key
| contract=insert('crm_contract',{'name':'九月台账-'+g['customer_name']+'-'+key,'no':contractno,'customer_id':cid,'owner_user_id':291,'audit_status':20,'order_date':day,'total_product_price':gross,'discount_percent':0,'total_price':gross,'deposit_price':0,'remark':note[:500]});permit(5,contract)
| order=insert('erp_sale_order',{'no':orderno,'status':20,'customer_id':cid,'contract_id':contract,'contract_no':contractno,'sale_user_id':291,'order_time':day,'total_count':sum(D(a['quantity']) for a in lines),'total_price':gross,'total_product_price':net,'total_tax_price':tax,'discount_percent':0,'discount_price':rounding,'deposit_price':0,'remark':note[:255],'out_count':0,'out_status':0,'production_status':0})
| q.execute('UPDATE crm_contract SET order_id=%s,order_no=%s WHERE id=%s',(order,orderno,contract))
| # Source enum has only DRAFT=0 and CONVERTED=10; the database comment is stale.
| rates={D(a['tax_percent']) for a in lines}
| quote=insert('crm_sale_quotation',{'no':quoteno,'name':'九月台账反推报价-'+key,'customer_id':cid,'owner_user_id':291,'status':10,'quotation_time':day,'total_product_price':net,'discount_percent':0,'discount_price':rounding,'total_price':gross,'tax_rate':next(iter(rates)) if len(rates)==1 else None,'tax_price':tax,'contract_id':contract,'contract_no':contractno,'remark':note});permit(9,quote)
| rowresult={'key':key,'customer_id':cid,'contract_id':contract,'order_id':order,'quotation_id':quote,'invoice_id':None,'receivable_id':None,'plan_id':None,'rows':g['rows'],'lines':[]}
| for a in lines:
| item=a['item_id'] or result['new_items'][a['new_item_code']]
| assert item in items and items[item].get('status',0)==0
| count=D(a['quantity']);unit=a['unit_id'];remark=f'台账行{a["row"]}；原单{a["source_no"]}；原项目号{a["source_code"]}；原净额{a["net_raw"]}，税{a["tax_raw"]}，含税{a["gross_raw"]}'
| cp=insert('crm_contract_product',{'contract_id':contract,'item_id':item,'item_unit_id':unit,'item_price':D(a['gross_price']),'contract_price':D(a['gross_price']),'count':count,'total_price':D(a['contract_total'])})
| common={'item_id':item,'item_unit_id':unit,'item_price':D(a['net_price']),'quotation_price':D(a['net_price']),'count':count,'total_price':D(a['net']),'tax_percent':D(a['tax_percent']),'tax_price':D(a['tax']),'remark':remark}
| qp=insert('crm_sale_quotation_product',{'quotation_id':quote,**common})
| op=insert('erp_sale_order_items',{'order_id':order,'product_id':item,'product_unit_id':unit,'product_price':D(a['net_price']),'count':count,'total_price':D(a['net']),'tax_percent':D(a['tax_percent']),'tax_price':D(a['tax']),'remark':remark,'out_count':0,'need_production':0})
| rowresult['lines'].append({'source_row':a['row'],'item_id':item,'unit_id':unit,'contract_product_id':cp,'quotation_product_id':qp,'order_item_id':op})
| if g['invoice']:
| invnote=f'九月台账{key}；原票据未提供附件。'
| if len(g['invoice_numbers'])>1:invnote+='两张发票合并登记；只有合计金额，未分配单张金额，不代表单张发票。'
| invnote+=g['original']
| invoice=insert('crm_invoice',{'no':'KP-'+key,'customer_id':cid,'contract_id':contract,'owner_user_id':291,'invoice_type':g['invoice_type'],'invoice_no':'/'.join(g['invoice_numbers']),'invoice_title':g['customer_name'],'price':gross,'invoice_time':day,'remark':invnote[:255],'audit_status':20});permit(10,invoice);rowresult['invoice_id']=invoice
| paid=D(g['received']);remaining=D(g['outstanding'])
| if paid>0:
| receipt=insert('crm_receivable',{'no':'HK-'+key,'customer_id':cid,'contract_id':contract,'owner_user_id':291,'return_time':day,'return_type':8,'price':paid,'audit_status':20,'remark':f'九月台账{key}；{g["payment_status"]}；用户确认到账日期取开票日；无票电商取备注销售日。原票号'+('/'.join(g['invoice_numbers']) or '无需开票')+'。暂无原票附件，按系统允许的未关联开票回款登记，合同已关联；到账方式保留原备注，未推断转账渠道。'});permit(7,receipt);rowresult['receivable_id']=receipt
| if remaining>0:
| due=datetime.fromisoformat(g['plan_day']);rp=insert('crm_receivable_plan',{'period':1,'customer_id':cid,'contract_id':contract,'owner_user_id':291,'return_time':due,'return_type':8,'price':remaining,'remind_days':3,'remind_time':due-timedelta(days=3),'remark':f'九月台账{key}；尚未回款余额{remaining}元；已回款{paid}元。日期按开票/销售日后30天补录测算，非约定账期；台账状态截至2026-09-29，实际到账另行核对。'});permit(8,rp);rowresult['plan_id']=rp
| result['groups'].append(rowresult)
| # Check the actual persisted values, cross-module links, source quantity precision, permissions and finance limits before commit.
| for g,r in zip(plan['groups'],result['groups']):
| q.execute('SELECT * FROM crm_contract WHERE id=%s',(r['contract_id'],));ct=q.fetchone();assert ct['total_price']==D(g['gross']) and ct['order_id']==r['order_id'] and ct['customer_id']==r['customer_id']
| q.execute('SELECT SUM(total_price) amount FROM crm_contract_product WHERE contract_id=%s AND deleted=0',(ct['id'],));assert q.fetchone()['amount']==ct['total_price']
| q.execute('SELECT * FROM crm_sale_quotation WHERE id=%s',(r['quotation_id'],));qt=q.fetchone();assert qt['contract_id']==ct['id'] and qt['status']==10 and qt['total_product_price']+qt['tax_price']-qt['discount_price']==ct['total_price']
| q.execute('SELECT * FROM erp_sale_order WHERE id=%s',(r['order_id'],));o=q.fetchone();assert o['contract_id']==ct['id'] and o['total_product_price']+o['total_tax_price']-o['discount_price']==ct['total_price']
| q.execute('SELECT SUM(price) amount FROM crm_invoice WHERE contract_id=%s AND deleted=0',(ct['id'],));inv=q.fetchone()['amount'] or D(0);assert inv==(ct['total_price'] if g['invoice'] else D(0))
| q.execute('SELECT SUM(price) amount FROM crm_receivable WHERE contract_id=%s AND deleted=0',(ct['id'],));paid=q.fetchone()['amount'] or D(0);assert paid==D(g['received']) and paid<=ct['total_price']
| q.execute('SELECT SUM(price) amount FROM crm_receivable_plan WHERE contract_id=%s AND deleted=0',(ct['id'],));pending=q.fetchone()['amount'] or D(0);assert paid+pending==ct['total_price']
| for a,l in zip(g['lines'],r['lines']):
| for t,id in [('crm_contract_product',l['contract_product_id']),('crm_sale_quotation_product',l['quotation_product_id']),('erp_sale_order_items',l['order_item_id'])]:
| q.execute('SELECT `count` FROM `'+t+'` WHERE id=%s',(id,));assert q.fetchone()['count']==D(a['quantity'])
| for biz,id in [(5,ct['id']),(9,r['quotation_id']),(10,r['invoice_id']),(7,r['receivable_id']),(8,r['plan_id'])]:
| if id:q.execute('SELECT COUNT(*) n FROM crm_permission WHERE biz_type=%s AND biz_id=%s AND user_id=291 AND level=1 AND deleted=0',(biz,id));assert q.fetchone()['n']==1
| c.commit();result['committed']=True
| (P/'import-result.json').write_text(json.dumps(result,ensure_ascii=False,default=str,indent=2),encoding='utf8')
| print(json.dumps({'committed':True,'counts':result['counts'],'totals':result['totals'],'at':result['imported_at']},ensure_ascii=False))
| except Exception:
| c.rollback();raise
| finally:
| with c.cursor() as q:q.execute('SELECT RELEASE_LOCK(%s)',('sales-ledger-202609',))
| c.close()
|
|
