python - 데이터베이스에서 특정 테스트 데이터 행을 지우는 스크립트를 작성하는 방법

Question

스크립트를 작성해야 합니다. 데이터베이스에 테스트용 휴대폰 번호가 있어야 합니다. 등록 후에는 데이터베이스에 관련 데이터가 많이 있을 것입니다. 그러면 스크립트에서 이러한 데이터를 지우거나 null로 설정할 수 있어야 합니다. . 이것은 유사한 스크립트입니다. 어떻게 수정해야 합니까?

from

future import print_functionimport itertools
import pymysql.cursors
import csv

from utils import is_telephone, is_email

if

name == '__main__':sg_connection = pymysql.connect(
host='localhost',
user='root',
password='root',
db='sellergrowth',
charset=' utf8',
cursorclass=pymysql.cursors.DictCursor
)

uc_connection = pymysql.connect(

host='localhost',
user='root',
password='root',
db='sguc',
charset='utf8',
cursorclass=pymysql.cursors.DictCursor
)

def _filter_valid(entity):

entity['username']가 아닌 경우:
return False
entity['telephone']가 아닌 경우 is_telephone(entity['telephone']):
return False
ifentity['email' ]이고 is_email(entity['email'])이 아님:
return False
(
(entity['telephone'] 및 엔티티['telephone_verified']) 또는 (entity['email'] 및 엔티티['email_verified'')가 아닌 경우 ])):
False를 반환
True를 반환

def _map_to_correct(엔티티):

def _set_blank_to_none(이름):
엔티티[이름]이 아닌 경우:
엔티티[이름] = 없음

_set_blank_to_none('telephone')

_set_blank_to_none('email')
if 엔터티['telephone'] 및 not 엔터티['telephone_verified']:
entity['telephone'] = 없음
if 엔터티['email'] 및 not 엔터티['email_verified']:
entity['email'] = 없음
_set_blank_to_none('비밀번호')
_set_blank_to_none('wechat_unionid')
_set_blank_to_none('wechat_app_openid')
_set_blank_to_none('wechat_public_openid')
_set_blank_to_none('wechat_site_openid' )
_set_blank_to_none('qq_openid')
_set_blank_to_none('last_login')
_set_blank_to_none('sign_url')
_set_blank_to_none('user_uuid')
entity['signup_service_id']:
entity['signup_service_id ' ] = 11300
if 엔터티['signup_method'] == 1:
entity['signup_method'] = '모바일'
elif 엔터티['signup_method'] == 2:
entity['signup_method'] = 'TABLET'
elif 엔터티['signup_method '] == 3:
entity['signup_method'] = 'WE_CHAT_PUBLIC'
elif 엔터티['signup_method'] == 4:
entity['signup_method'] = 'WE_CHAT_SITE'
elif 엔터티['signup_method'] == 12:
entity['signup_method'] = 'WE_CHAT_APP'
elif 엔터티['signup_method'] == 6:
entity['signup_method'] = 'QQ_WEB'
elif 엔터티['signup_method'] == 7:
entity ['signup_method'] = 'QQ_WEBAPP'
else:
entity['signup_method'] = 없음
entity['is_active'] = 1 if 엔터티['is_active'] else 0
엔티티 반환

def _dict_to_tuple(entity):
return tuple((
entity['telephone'],entity['email'],entity['username'],entity['password'],entity['wechat_unionid'],
entity ['wechat_app_openid'],
entity['wechat_public_openid'], 엔터티['wechat_site_openid'], 엔터티['qq_openid'],
int(entity['signup_service_id']), 엔터티['date_joined'], 엔터티['last_login '], 엔터티['is_active'],
entity['signup_method'], 엔터티['sign_url'], 엔터티['user_uuid']
))

try:
sg_connection.cursor()를 커서로, uc_connection.cursor()를 uc_cursor:
sql = 'SELECT is_active, p.telephone, email, username, password, p.wechat_unionid, p.wechat_app_openid, p.wechat_public_openid, p.wechat_site_openid, p.qq_openid, p.signup_service_id, p.telephone_verified, p.email_verified, date_joined, last_login, p.signup_method, e.sign_url, p2.user_uuid FROM account_profile p LEFT JOIN event_baseevent e ON e.id=p.signup_event_id LEFT JOIN account_profile p2 ON p2.user_id=p.inviter_id, auth_user u WHERE p.user_id=u.id'
cursor.execute(sql)
cursor,cursor_for_record = itertools .tee(커서)
cursor = itertools.ifilter(_filter_valid, 커서)
cursor = itertools.imap(_map_to_corrent, 커서)
cursor = itertools.imap(_dict_to_tuple, 커서)

sql = 'INSERT into cas_service (service_id, service_group) VALUES (%s, %s)'
service_list = ((11300, 'SG'), (12300, 'SGASK'), (13300, 'YQSXY'), ( 14300, 'CHUANGLAN'))
uc_cursor.executemany(sql, service_list)
uc_connection.commit()

insert_sql = 'INSERT cas_user(전화, 이메일, 사용자 이름1, 비밀번호1, wechat_unionid, wechat_app_openid, wechat_public_openid, wechat_site_openid, qq_openid, sign_up_service_id, date_joined, last_login, is_active, sign_up_device, sg_event_url, sg_inviter_uid) (%s, %s, %s , %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)'
uc_cursor.executemany(insert_sql, list( 커서))
uc_connection.commit()

open('not_migration_report.csv', 'wb')을 csvfile로 사용:
spamwriter = csv.writer(csvfile, delimiter=',', quotechar='|', quoting=csv.QUOTE_MINIMAL)
init = False
cursor_for_record = itertools.ifilterfalse(_filter_valid,cursor_for_record)
for i incursor_for_record:
init가 아닌 경우:
init = True
print(i.keys())
spamwriter.writerow(
[a.encode("utf8")) if type (a) == 유니코드 else a for a in i.values()])
finally:
sg_connection.close()
uc_connection.close()

Answer 1

테이블 조작 권한이 있는 경우 플래그 = 테스트와 같이 상태를 식별할 수 있는 플래그만 추가하면 됩니다

으아악