import sys,json
|
from pathlib import Path
|
P=Path(__file__).parent
|
sys.stdout.reconfigure(encoding='utf8')
|
sys.path.insert(0,str(P.parent/'iqc-import-20261012/deps'))
|
import pymysql
|
def encode(x):return json.dumps(x,ensure_ascii=False,default=str,sort_keys=True)
|
def save(name,obj):
|
P.mkdir(exist_ok=True);(P/name).write_text(encode(obj),encoding='utf8')
|
def connect():return pymysql.connect(host='36.213.150.237',port=10000,user='root',password=sys.stdin.readline().strip(),database='mom-hm',charset='utf8mb4',cursorclass=pymysql.cursors.DictCursor,autocommit=False,connect_timeout=15,read_timeout=120)
|
def get_user(q):
|
q.execute("SELECT id,username,nickname,status,dept_id FROM system_users WHERE deleted=0 AND TRIM(nickname)=%s",('罗军',))
|
users=q.fetchall();assert len(users)==1 and users[0]['status']==0,encode(users)
|
return users[0]
|
def read():
|
c=connect()
|
try:
|
with c.cursor() as q:
|
q.execute("SET time_zone='+08:00'");q.execute('SELECT DATABASE() db,NOW() now');state=q.fetchone();assert state['db']=='mom-hm'
|
user=get_user(q)
|
q.execute('SHOW COLUMNS FROM crm_permission');schema=q.fetchall()
|
q.execute('SELECT id,name,owner_user_id FROM crm_customer WHERE deleted=0 ORDER BY id');customers=q.fetchall()
|
q.execute('SELECT * FROM crm_permission WHERE biz_type=2 AND user_id=%s AND deleted=0 ORDER BY id',(user['id'],));members=q.fetchall()
|
existing={r['biz_id'] for r in members}
|
missing=[r for r in customers if r['id'] not in existing]
|
plan={'state':state,'user':user,'permission_schema':schema,'customers':customers,'existing_memberships':members,'missing_customers':missing,
|
'summary':{'user_id':user['id'],'user_name':user['nickname'].strip(),'active_customers':len(customers),'already_joined':len(customers)-len(missing),'to_add':len(missing),'new_permission_level':2,'new_permission_label':'只读'}}
|
save('plan.json',plan);print(encode(plan['summary']))
|
finally:c.close()
|
def apply():
|
assert not (P/'result.json').exists(),'Already applied'
|
plan=json.loads((P/'plan.json').read_text(encoding='utf8'));c=connect()
|
try:
|
with c.cursor() as q:
|
q.execute("SET time_zone='+08:00'");q.execute('SELECT DATABASE() db,NOW() now');state=q.fetchone();assert state['db']=='mom-hm'
|
q.execute("SELECT GET_LOCK('codex-crm-customer-team-luojun',0) locked");assert q.fetchone()['locked']==1
|
user=get_user(q);assert user['id']==plan['user']['id']
|
q.execute('SELECT * FROM crm_customer WHERE deleted=0 ORDER BY id FOR SHARE');customers=q.fetchall()
|
assert encode([{k:r[k] for k in ('id','name','owner_user_id')} for r in customers])==encode(plan['customers']),'Customers changed'
|
q.execute('SELECT * FROM crm_permission ORDER BY id FOR UPDATE');before=q.fetchall()
|
backup='locked-before-'+state['now'].strftime('%Y%m%d-%H%M%S')+'.json'
|
save(backup,{'customers':customers,'permissions':before})
|
memberships=[r for r in before if r['biz_type']==2 and r['user_id']==user['id'] and r['deleted'] in (0,b'\x00')]
|
existing={r['biz_id'] for r in memberships};ids={r['id'] for r in customers}
|
missing=[r for r in customers if r['id'] not in existing];inserted=[]
|
assert all(customer['owner_user_id']!=user['id'] for customer in missing),'Owner missing permission requires separate review'
|
if missing:
|
q.executemany('INSERT INTO crm_permission (biz_type,biz_id,user_id,level,creator,updater,create_time,update_time,deleted) VALUES (2,%s,%s,2,%s,%s,%s,%s,0)',
|
[(customer['id'],user['id'],'1','1',state['now'],state['now']) for customer in missing])
|
assert q.rowcount==len(missing)
|
q.execute('SELECT * FROM crm_permission ORDER BY id');after={r['id']:r for r in q.fetchall()}
|
before_ids={r['id'] for r in before};customer_map={r['id']:r for r in customers}
|
inserted=[{'id':id,'customer_id':r['biz_id'],'customer_name':customer_map[r['biz_id']]['name']} for id,r in after.items() if id not in before_ids]
|
assert all(encode(after[r['id']])==encode(r) for r in before),'Existing permissions changed'
|
assert len(after)==len(before)+len(inserted)
|
active=[r for r in after.values() if r['biz_type']==2 and r['user_id']==user['id'] and r['deleted'] in (0,b'\x00')]
|
assert ids<={r['biz_id'] for r in active},'Coverage incomplete'
|
for x in inserted:
|
r=after[x['id']];assert r['level']==2 and r['biz_id']==x['customer_id'] and r['user_id']==user['id']
|
assert sum(r['biz_id']==x['customer_id'] for r in active)==1,'Duplicate new membership'
|
c.commit()
|
result={'user_id':user['id'],'user_name':user['nickname'].strip(),'inserted':len(inserted),'active_customers':len(customers),
|
'already_joined':len(customers)-len(inserted),'new_permission':'只读','existing_permission_rows_unchanged':len(before),
|
'backup':str(P/backup),'committed_at':str(state['now']),'inserted_records':inserted}
|
save('result.json',result);print(encode({k:v for k,v in result.items() if k!='inserted_records'}))
|
except BaseException:c.rollback();raise
|
finally:c.close()
|
def verify():
|
result=json.loads((P/'result.json').read_text(encoding='utf8'));before=json.loads(Path(result['backup']).read_text(encoding='utf8'));c=connect()
|
try:
|
with c.cursor() as q:
|
q.execute("SET time_zone='+08:00'");q.execute('SELECT NOW() now');now=q.fetchone()['now']
|
q.execute('SELECT * FROM crm_customer WHERE deleted=0 ORDER BY id');customers=q.fetchall()
|
assert encode(customers)==encode(before['customers']),'Customer data changed'
|
q.execute('SELECT * FROM crm_permission ORDER BY id');permissions={r['id']:r for r in q.fetchall()}
|
assert all(encode(permissions[r['id']])==encode(r) for r in before['permissions'])
|
active=[r for r in permissions.values() if r['biz_type']==2 and r['user_id']==result['user_id'] and r['deleted'] in (0,b'\x00')]
|
missing=[r['id'] for r in customers if r['id'] not in {x['biz_id'] for x in active}];assert not missing
|
for r in result['inserted_records']:
|
assert permissions[r['id']]['level']==2
|
assert sum(x['biz_id']==r['customer_id'] for x in active)==1
|
checked={'passed':True,'user_id':result['user_id'],'active_customers':len(customers),'added':result['inserted'],
|
'missing_memberships':len(missing),'customers_and_existing_permissions_unchanged':True,'new_permissions_read_only':True,'verified_at':str(now)}
|
save('verification.json',checked);print(encode(checked))
|
finally:c.close()
|
if __name__=='__main__':
|
{'read':read,'apply':apply,'verify':verify}[sys.argv[1] if len(sys.argv)>1 else 'read']()
|
