python - 怎麼寫一個腳本可以清除資料庫裡的某行測試數據

就是要寫個腳本資料庫有測試用的手機號碼註冊完資料庫裡會有一大堆相關的資料然後腳本就需要能清除這些資料或是將它們設定為null 這個是類似的一個腳本應該怎麼修改

from future import print_function
import 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):
if not entity['username']:
return False
if entity['telephone'] and not is_telephone(entity['telephone']):
return False
if entity['email'] and not is_email(entity['email']):
return False
if not (
(entity['telephone'] and entity[' telephone_verified']) 或 (entity['email'] and entity['email_verified'])):
return False
return True

def _map_to_correct(entity):
def _set_blank_to_none(name):
if not entity[name]:
entity[name] = None

_set_blank_to_none('telephone')
_set_blank_to_none('email')
if entity['telephone'] and not entity['telephone_verified']:
entity['telephone'] = None
if entity['email'] and not entity['email_verified']:
entity['email'] = None
_set_blank_to_none('password')
_set_blank_to_none('wechat_unionid')
n_set_blank_to_none('wechat_unionid')
n_bank_ ('wechat_app_openid')
_set_blank_to_none('wechat_public_openid')
_set_blank_to_none('wechat_site_openid')
_set_blank_to_none('q_openid')#q_i_##_set_blank_to_none('qn_openid')#q_i_i_i_cin_b提供」 ')
_set_blank_to_none('user_uuid')
if not entity['signup_service_id']:
entity['signup_service_id'] = 11300
if entity['signup_method'] == 1:
entity ['signup_method'] = 'MOBILE'
elif entity['signup_method'] == 2:
entity['signup_method'] = 'TABLET'
elif entity['signup_method'] == 3:
entity['signup_method'] = 'WE_CHAT_PUBLIC'
elif entity['signup_method'] == 4:
entity['signup_method'] = 'WE_CHAT_SITE'
elif entity['signup_method'] = 'WE_CHAT_SITE'
elif entity['signup_method'] == 12:
entity['signup_method'] = 'WE_CHAT_APP'
elif entity['signup_method'] == 6:
entity['signup_method'] = 'QQ_WEB'
elif entity['signup_method'] = 'QQ_WEB'
elif entity[ 'signup_method'] == 7:
entity['signup_method'] = 'QQ_WEBAPP'
else:
entity['signup_method'] = None
entity['is_active'] = 1 if entity ['is_active'] else 0

return entity###

def _dict_to_tuple(entity):
return tuple((
entity['電話'], 實體['電子郵件'], 實體['使用者名稱'], 實體['密碼'], 實體[' wechat_unionid'],
entity['wechat_app_openid'],
entity['wechat_public_openid'],entity['wechat_site_openid'],entity['qq_openid'],##int(entity['signup_service_idity」 ]) 、實體['date_joined']、實體['last_login']、實體['is_active']、
entity['signup_method']、實體['sign_url']、實體['user_uuid']
))

try:

with sg_connection.cursor() as 遊標, uc_connection.cursor() as uc_cursor:
sql = 'SELECT
is_active, p.telephone, email, 使用者名稱, 密碼, 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_eventid FROM account_profile p LEFT JOIN event_baseeventuuid FROM account_profile p LEFT JOIN event_base 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_for_record = itertools.tee(cursor)
cursor = itertools.ifilter(_filter_valid, 遊標)
cursor = itertools.imap(_map_to_ Correct, 遊標)##cursor )

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 into cas_user (phone, email, username1, password1, wechat_unionid, wechat_app_openid, wechat_public_openid, wechat_site_openid, qq_openbate, sign_url_g, sign_ sg_inviter_uid) VALUES ( %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)'
uc_cursor.executemany(insert_sql, list(cursor))
uc_connection.commit()

with open('not_migration_report.csv', 'wb') as 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) == unicode else a for a in i.values()])
finally:
sg_connection.close ( )
uc_connection.close()

阿神2739 天前940

python - 怎麼寫一個腳本可以清除資料庫裡的某行測試數據

全部回覆(1)我來回復