python - 怎么写一个脚本可以清除数据库里的某行测试数据

问题描述

就是要写个脚本 数据库有测试用的手机号 注册完数据库里会有一大堆相关的数据 然后脚本就需要能清除这些数据 或者将它们设置为null 这个是类似的一个脚本 应该怎么修改

from future import print_functionimport itertoolsimport pymysql.cursorsimport 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 Falseif entity[’telephone’] and not is_telephone(entity[’telephone’]):return Falseif entity[’email’] and not is_email(entity[’email’]):return Falseif not ((entity[’telephone’] and entity[’telephone_verified’]) or (entity[’email’] and entity[’email_verified’])):return Falsereturn 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’] = Noneif entity[’email’] and not entity[’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 not entity[’signup_service_id’]:entity[’signup_service_id’] = 11300if 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’] == 12:entity[’signup_method’] = ’WE_CHAT_APP’elif entity[’signup_method’] == 6:entity[’signup_method’] = ’QQ_WEB’elif entity[’signup_method’] == 7:entity[’signup_method’] = ’QQ_WEBAPP’else:entity[’signup_method’] = Noneentity[’is_active’] = 1 if entity[’is_active’] else 0return entity

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’], entity[’wechat_site_openid’], entity[’qq_openid’],int(entity[’signup_service_id’]), entity[’date_joined’], entity[’last_login’], entity[’is_active’],entity[’signup_method’], entity[’sign_url’], entity[’user_uuid’]))

try:with sg_connection.cursor() as cursor, uc_connection.cursor() as 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)cursor = itertools.ifilter(_filter_valid, cursor)cursor = itertools.imap(_map_to_correct, cursor)cursor = itertools.imap(_dict_to_tuple, 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_openid, sign_up_service_id, date_joined, last_login, is_active, sign_up_device, sg_event_url, 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 = Falsecursor_for_record = itertools.ifilterfalse(_filter_valid, cursor_for_record)for i in cursor_for_record:if not init:init = Trueprint(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()

问题解答

如果对表有操作权限的话 只需要 增加一个标识状态的flag即可 如 flag = test

DELETE * FROM table_name WHERE flag = test