python - Comment écrire un script pour effacer une certaine ligne de données de test dans la base de données
Il est nécessaire d'écrire un script. La base de données a un numéro de téléphone mobile pour les tests. Après l'enregistrement, il y aura beaucoup de données associées dans la base de données. Ensuite, le script doit pouvoir effacer ces données ou les définir sur null. . Il s'agit d'un script similaire. Comment doit-il être modifié ?
fromfuture import print_functionimport itertools
import pymysql.cursors
import csv
if
name == '__main__':sg_connection = pymysql.connect(
host='localhost',
user='root',
password='root',
db='sellergrowth',
charset=' utf8',
cursorclass=pymysql.cursors.DictCursor
)
host='localhost',
user='root',
password='root',
db='sguc',
charset='utf8',
cursorclass=pymysql.cursors.DictCursor
)
if pas entité['nom d'utilisateur']:
return False
if entité['telephone'] et non is_telephone(entity['telephone']):
return False
if entité['email' ] et non is_email(entity['email']):
return False
sinon (
(entity['telephone'] etentity['telephone_verified']) ou (entity['email'] etentity['email_verified' ])):
retourner Faux
retourner Vrai
def _set_blank_to_none(name):
si ce n'est pas le cas d'entité[nom]:
entity[name] = Aucun
_set_blank_to_none('email')
if entité['telephone'] et non entité['telephone_verified']:
entity['telephone'] = None
if entité['email'] et non entité['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')
sinon entité['signup_service_id']:
entity['signup_service_id ' ] = 11300
si entité['signup_method'] == 1:
entity['signup_method'] = 'MOBILE'
elif entité['signup_method'] == 2:
entity['signup_method'] = 'TABLET'
elif entité['signup_method '] == 3:
entité['signup_method'] = 'WE_CHAT_PUBLIC'
elif entité['signup_method'] == 4:
entity['signup_method'] = 'WE_CHAT_SITE'
elif entité['signup_method'] == 12:
entity['signup_method'] = 'WE_CHAT_APP'
elifentity['signup_method'] == 6:
entity['signup_method'] = 'QQ_WEB'
elifentity['signup_method'] == 7:
entity ['signup_method'] = 'QQ_WEBAPP'
else:
entity['signup_method'] = Aucun
entity['is_active'] = 1 si entité['is_active'] else 0
retour d'entité
def _dict_to_tuple(entity):
return tuple((
entité['téléphone'], entité['email'], entité['nom d'utilisateur'], entité['mot de passe'], entité['wechat_unionid'],
entité ['wechat_app_openid'],
entité['wechat_public_openid'], entité['wechat_site_openid'], entité['qq_openid'],
int(entité['signup_service_id']), entité['date_joined'], entité['last_login '], entité['is_active'],
entité['signup_method'], entité['sign_url'], entité['user_uuid']
))
essayez :
avec sg_connection.cursor() comme curseur, uc_connection.cursor() comme 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, curseur_for_record = itertools .tee(curseur)
curseur = itertools.ifilter(_filter_valid, curseur)
curseur = itertools.imap(_map_to_correct, curseur)
curseur = itertools.imap(_dict_to_tuple, curseur)
sql = 'INSÉRER dans 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 = 'INSÉRER dans cas_user (téléphone, email, nom d'utilisateur1, mot de passe1, 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) VALEURS ( %s, %s, %s , %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)'
uc_cursor.executemany(insert_sql, list( curseur))
uc_connection.commit()
avec open('not_migration_report.csv', 'wb') comme fichier csv:
spamwriter = csv.writer(csvfile, delimiter=',', quotechar='|', quoting=csv.QUOTE_MINIMAL)
init = False
cursor_for_record = itertools.ifilterfalse(_filter_valid, curseur_for_record)
pour moi dans curseur_for_record:
si ce n'est pas init:
init = True
print(i.keys())
spamwriter.writerow(
[a.encode("utf8") si type (a) == unicode sinon a pour a dans i.values()])
finalement :
sg_connection.close()
uc_connection.close()
