Python - データベース内のテストデータの特定の行をクリアするスクリプトの書き方

Question

スクリプトを作成する必要があります。データベースにはテスト用の携帯電話番号があります。登録後、データベースには多くの関連データが存在します。その後、スクリプトでこれらのデータをクリアまたは設定できる必要があります。 null に設定します。これは同様のスクリプトです。どのように変更すればよいですか?

##from

future import print_functionimport itertools
import pymysql.cursors
import csv

ユーティリティインポートから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):

エンティティ['ユーザー名']でない場合:
falseを返す
エンティティ['電話']かつis_telephone(エンティティ['電話'])でない場合:
return False
ifentity['email'] and not is_email(entity['email']):
return False
if not (
(entity['telephone'] andentity[' Telephone_verified']) または (entity['email'] およびentity['email_verified'])):
return False
return True

def _map_to_correct(entity):

def _set_blank_to_none(name):
entity[name] でない場合:
entity[name] = None

_set_blank_to_none('電話')

_set_blank_to_none('電子メール')
エンティティ['電話']がエンティティ['電話認証済み']ではない場合:
エンティティ['電話'] = なし
if エンティティ['email'] であり、エンティティ['email_verified'] ではない:
entity['email'] = None
_set_blank_to_none('password')
_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')
if エンティティ['signup_service_id']:
entity['signup_service_id'] = 11300
if エンティティ['signup_method'] == 1:
entity ['signup_method'] = 'MOBILE'
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'] = None
entity['is_active'] = 1 if エンティティ['is_active'] else 0
エンティティを返す

def _dict_to_tuple(entity):
return tuple((
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, メール、ユーザー名、パスワード、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)
カーソル、カーソル_レコード = itertools.tee(カーソル)
cursor = itertools.ifilter(_filter_valid, カーソル)
cursor = itertools.imap(_map_to_correct, カーソル)
cursor = itertools.imap(_dict_to_tuple, カーソル)

sql = '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 (電話、メール、ユーザー名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(cursor))
uc_connection.commit()

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)
cursor_for_record の i の場合:
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()

Answer 1

テーブルを操作する権限がある場合は、flag = testなどのステータスを識別するためのフラグを追加するだけです

リーリー