鄭重聲明： 此文章內分析用到的數據集僅為學習交流，本文未提供任何下載鏈接且已在 24 小時內刪除，請勿用於非法用途否則產生的一切後果自行承擔
tb#
事情起因#
代辦事項之 - 整理手上的褲子之固定字符分割的 txt 導入 mysql 篇
txt 以四個減號----來分割郵箱和密碼 (請忽略我的魔幻打碼技術)：
本篇文章就來試試用 Python 將這些固定字符分割的 txt 批量導入本地的mysql數據庫
部分過程解釋如下#
關於文件編碼以及換行符#
文件編碼#
當然是所有的工具和文件都是 utf8 編碼的最好啦！
但是就像我剛才的這四個文本，下下來都是：
ASCII text, with CRLF line terminators
手動修改為 utf-8 是什麼樣子呢？
UTF-8 Unicode (with BOM) text, with CRLF line terminators
這兩種平時是最經常看到的，啊？你問我為什麼能在win用Linux的命令，因為我偶然發現git for windows裡面提供了非常多的Linux命令，git for windows在你下Virtualbox的時候會提示你要不要安裝，你也可以直接下载安装：傳送門
建議沒安的同學安一下，還是非常好用的 (這兩家公司看到請給我打廣告費謝謝)，後面我們也要用到iconv、sed等Linux命令
使用file xxx.txt命令即可查看文件編碼以及換行符
我們將ANSI編碼的txt導入utf-8編碼的數據庫就報錯了：
1300, "Invalid utf8 character string: '˹'"
我這裡將所有的文件轉換為UTF8編碼的，先用前面說的命令行的file命令來試一下，這個秒出結果
Python中命令行語句的執行我們也在前面的文章《Python3 打印出電腦上所有 wifi 賬號和密碼》介紹過，我們這裡直接上代碼：
import os
txt_name_list = ['260W.txt','316W.txt','440W.txt','445W.txt',]
for i in txt_name_list:
print(os.popen('file {}'.format(i)).read())
效果如下:
顯示第一個為UTF-8，其他的為ANSI，但實際上我這裡316W.txt也為UTF-8，可見file命令準確性不高
我們再試下 Python3 中的庫chardet，參考文章：python 實現文本文件的編碼檢測和轉換
for i in txt_name_list:
detector = UniversalDetector()
print("正在檢測{}的文件編碼".format(i))
detector.reset()
for line in open(i, 'rb').readlines():
detector.feed(line)
if detector.done:
break
detector.close()
print(detector.result)
大文件的檢測真的是超慢，不過等待了很久之後還是能檢測出316W.txt為UTF-8編碼
我們改一下，將 readlines () 中加上參數1000000，發現檢測不出來：
加到2000000行時又能檢測出來：
那我們就偷懶地以200W作為最大檢測界限吧。
那麼獲得了文件的編碼之後，我們判斷其是否為utf8，不是的話將其使用命令iconv轉換文件編碼並重命名：
src_enc = detector.result['encoding']
# 如果編碼不為utf-8，則用iconv將xx.txt轉換為utf8-xx.txt
if 'utf-8' not in src_enc.lower():
utf8_txt_name = "utf8-{}".format(i)
print("{}不是utf-8編碼，正在轉換為{}".format(i, utf8_txt_name))
try:
os.popen(
'iconv -c -f {} -t utf-8 {} > {}'.format(src_enc, i, utf8_txt_name))
except Exception as e:
print("轉換報錯：{} \n 程序退出！".format(e))
exit(0)
processed_txt_filename_list.append(utf8_txt_name)
else:
processed_txt_filename_list.append(i)
換行符#
CRLF 代表回車換行，CR 回車，LF 換行，一般默認 CRLF 即可。CRLF 代表的就是 \n\r，這是 Windows 默認換行方式；LF 代表的就是 \r，這是 Unix 體系默認換行方式
Unix 體系下的文件在 Windows 裡打開的話，所有文字會變成一行 (用 VSCODE 打開也能正常顯示，VSC 看到請打錢謝謝)；而 Windows 裡的文件在 Unix 下打開的話，在每行的結尾可能會多出一個 ^M 符號。
那我就不改了，全部使用CRLF, 如果你要改也可以使用 sed 命令把所有可能存在的\n\r都換為\n
sed ’s/^M//’ filename > tmp_filename
其他方法參考文章：linux 去掉 windows 下文件中的 \r
pymysql 基本操作#
使用pymysql連接數據庫並執行sql語句的基本用法如下：
# 連接數據庫,密碼後面可以加數據庫名稱
connect_mysql = pymysql.connect('localhost', 'root', 'root', charset='utf8')
# 得到一個可以執行SQL語句的光標對象，執行完畢返回的結果默認以元組顯示
cursor = connect_mysql.cursor()
# 執行sql語句
cursor.execute('select * from `users`;')
# 關閉數據庫連接
connect_mysql.close()
但是！在我使用pymysql.connect()函數連接數據庫執行後面的加載文件導入命令load data local infile xxx.txt時，發現會報錯：
1148, 'The used command is not allowed with this MySQL version'
在網上查詢說
出於安全考慮，默認是不允許從 client host 遠程通過 load data 命令導數據的。
在這篇文章中找到了解決辦法：
connect_mysql = pymysql.connections.Connection('localhost', 'root', 'root',charset='utf8',local_infile=True)
全代碼如下：#
註釋應該能看懂，有疑問的地方可以留言一起探討～::quyin:1huaji::
#!/usr/bin/python
# -*- coding: utf-8 -*-
'''
@author: soapffz
@fucntion: 批量導入固定字符分割的txt到mysql
@time: 19-04-06
'''
import os # 文本名稱獲取及cmd命令執行
import pymysql # Python3用pymysql,Python2用mysqldb
from chardet.universaldetector import UniversalDetector # 判斷文件編碼
import timeit # 計算用時
def create_db():
# 如果數據庫不存在則創建，字符集為utf8,排序規則為utf8_general_ci
sql_create_db = "CREATE DATABASE IF NOT EXISTS `soendb` default charset utf8 COLLATE utf8_general_ci;"
# 創建taobao這個表含有兩個字段mail和passwd
sql_create_table = """
CREATE TABLE IF NOT EXISTS `taobao` (
`mail` varchar(255) DEFAULT NULL,
`passwd` varchar(255) DEFAULT NULL
) ENGINE=MyISAM DEFAULT CHARSET=utf8;
"""
# 自行選擇需要創建哪些東西
try:
cursor.execute(sql_create_db)
cursor.execute('use soendb;') # 將當前數據庫設置為剛創建的數據庫
cursor.execute(sql_create_table)
except Exception as e: # 捕獲所有異常並打印，python2 是Exception,e
print("數據庫創建出錯：{}\n退出程序！".format(e))
connect_mysql.rollback() # 發生錯誤時回滾
connect_mysql.close() # 關閉數據庫連接
exit(0)
def textencoding_lineterminators_modifi(txt_name_list):
# 處理文本編碼及字符
processed_txt_filename_list = []
for i in txt_name_list:
detector = UniversalDetector()
print("正在檢測{}的文件編碼".format(i))
detector.reset() # 清空餵食池
for line in open(i, 'rb').readlines(20000000): # 迭代200W行，足以檢測出結果
detector.feed(line) # 餵食
if detector.done: # 如果有結果，退出
break
detector.close()
src_enc = detector.result['encoding']
# 如果編碼不為utf-8，則用iconv將xx.txt轉換為utf8-xx.txt
if 'utf-8' not in src_enc.lower():
utf8_txt_name = "utf8-{}".format(i)
print("{}不是utf-8編碼，正在轉換為{}".format(i, utf8_txt_name))
try:
os.popen(
'iconv -c -f {} -t utf-8 {} > {}'.format(src_enc, i, utf8_txt_name))
except Exception as e:
print("轉換報錯：{} \n 程序退出！".format(e))
exit(0)
processed_txt_filename_list.append(utf8_txt_name)
else:
processed_txt_filename_list.append(i)
return processed_txt_filename_list
def import_txt_to_mysql(txt_name):
# 將txt導入mysql
sql_insert_txt = """
LOAD DATA LOCAL INFILE "{}" INTO TABLE `taobao`
FIELDS TERMINATED BY '----' LINES TERMINATED BY '\r\n';
""".format(txt_name)
# --字段用'----'分隔(例子)，記錄用CRLF進行分隔(例子)
try:
cursor.execute('use soendb;')
cursor.execute(sql_insert_txt)
# 提交到數據庫執行
connect_mysql.commit()
except Exception as e:
print("導入數據出錯：{}\n退出程序！".format(e))
connect_mysql.rollback()
connect_mysql.close()
exit(0)
if __name__ == "__main__":
start_time = timeit.default_timer()
# 連接數據庫,密碼後面可以加數據庫名稱
try:
connect_mysql = pymysql.connections.Connection(
'localhost', 'root', 'root', charset='utf8', local_infile=True)
# 得到一個可以執行SQL語句的光標對象，執行完畢返回的結果默認以元組顯示
cursor = connect_mysql.cursor()
except Exception as e:
print("數據庫連接出錯 ：{}\n退出程序！".format(e))
exit(0)
print("數據庫連接成功")
create_db() # 可選，創建數據庫及表
# 獲取當前目錄所有txt文件的文件名
src_txt_filename_list = [filename for filename in os.listdir(
".") if filename.split(".")[-1] == 'txt']
if src_txt_filename_list: # 如果當前目錄txt名字列表不為空，則將列表拿去檢測編碼得到新txt名字列表
processed_txt_filename_list = textencoding_lineterminators_modifi(
src_txt_filename_list)
for txt_name in processed_txt_filename_list:
try:
print("正在導入{}".format(txt_name))
import_txt_to_mysql(txt_name)
except Exception as e:
print("導入數據出錯：{}\n退出程序！".format(e))
connect_mysql.rollback()
connect_mysql.close()
exit(0)
else:
print("未獲取到txt文件名稱，請重新檢查！")
end_time = timeit.default_timer()
print("數據庫導入完畢，總用時{}".format(end_time-start_time))
效果如下：
其實 200 多秒，99% 以上的時間都在驗證文件的編碼，但是就像我文中說的那樣，file命令準確性基本只適用於chardet檢測可信度為 100% 的編碼的文件，其他的基本就是瞎報，但是chardet速度又太慢了，儘管我已經投機取巧設置了 200W 行的上限
當然你也可以不進行驗證，所有文件都轉換為utf-8編碼
所以，有什麼好的在Windows下 (如果是Linux也最好能在Windows下找到) 的批量驗證並修改文件編碼的工具，歡迎大家留言討論交流～
Notes：
- 加載文件的路徑必須是 /，但是 win 複製的默認是 \，文中未處理因為此 py 文件默認是放到 txt 所在目錄執行的所以不存在路徑問題
- 對於插入、更新、刪除等操作，需要使用 connect_mysql.commit () 來提交到數據庫執行，對於查詢、創建數據庫和數據表的操作不需要此語句。
- 其實大部分代碼是重複的，但是為了更方便閱讀以及防止有些看官用不到某些功能，我這裡還是分開寫了
參考鏈接：
- python 實現文本文件的編碼檢測和轉換
- linux 去掉 windows 下文件中的 \r
- Python3 MySQL 數據庫連接 - PyMySQL 驅動
- PyMySQL 的基本使用
- PyMYSQL 模塊
BreachCompilation#
[19-4-10 更新] 每個都檢測編碼實在是太費勁了，我在按照安全脈搏的 14 億泄露數據整理文章整理 Freebuf 等網站發布的數據安全諮詢文章14 億郵箱泄露密碼明文信息查詢網站驚現網絡 時，這裡的數據都是這樣的：
有很多文件夾有子文件，數據文件沒有後綴名，我們主要使用cat * > 1.txt這條命令來把他們合併到指定文件夾
代碼快寫完的時候發現 shell 一條命令就能合併文件夾 (包括子文件夾) 內的所有文件，參考文章
> find . 遞歸從本文件夾開始查找文件
> -type f 指定文件類型為普通文件，還可以選用的項有：d 目錄、l 鏈接符號、c 字符設備、b 塊設備、s 套接字等
> -amin/-mmin/-cmin 可以指定文件的訪問時間/修改時間/改變時間。e.g. find . -type f -atime +7 -print 打印出訪問時間超過七天的所有文件
> -perm 根據文件權限查找文件
> -user 更具文件所有者查找文件
> -delete 將刪除查找到的文件
> -exec 對查找到的文件執行命令，格式為： -exec ./commands.sh {} \;
當時的內心是崩潰的::quyin:hematemesis::，幸好試了下win自帶的find命令只能搜索文件中的字符串，而安裝的Git即使選了用Unix覆蓋win自帶的部分指令這個選項，也不能用find, 所以內心又恢復正常::quyin:witty::
文本整理的主要核心命令：進入每個文件夾並執行cat指令，每條指令不阻塞，但是希望多線程並互不干擾
參考文章：Python 執行 cmd 的各種實現方法及優劣（subprocess.Popen, os.system 和 commands.getstatusoutput）
並在這篇文章上補充：
os.system () 這個函數是阻塞的但是不會返回報錯信息，每條指令必須執行完才會執行下一條
在 3.x 版本總，commands 庫的三個方法之一 getstatus () 方法被移除，getoutput () 和 getstatusoutput () 被放到了 subprocess 模塊中。(即 3.x 版本不再使用 commands 庫，而是 subprocess 庫)
以下為全代碼：
#!/usr/bin/python
# -*- coding: utf-8 -*-
'''
@author: soapffz
@fucntion: 批量導入固定字符分割的txt到mysql(粗魯編碼轉換)
@time: 19-04-11
'''
import os # 文本名稱獲取及cmd命令執行
import shutil # 文件及目錄相關處理
import subprocess # 執行cmd命令
import pymysql # Python3用pymysql,Python2用mysqldb
import timeit # 計算用時
import re # 用正則表達式查找分離文件名
class db_progress:
def __init__(self):
self.collections_dir = 'E:\Pants\BreachCompilation\collections'
self.connect_mysql()
self.create_db()
def connect_mysql(self):
# 連接數據庫,密碼後面可以加數據庫名稱
try:
self.mysql_conn = pymysql.connections.Connection(
'localhost', 'root', 'root', charset='utf8', local_infile=True)
# 得到一個可以執行SQL語句的光標對象，執行完畢返回的結果默認以元組顯示
self.cursor = self.mysql_conn.cursor()
print("數據庫連接成功")
except Exception as e:
print("數據庫連接出錯 ：{}\n退出程序！".format(e))
exit(0)
def create_db(self):
# 如果數據庫不存在則創建，字符集為utf8,排序規則為utf8_general_ci
sql_create_db = "CREATE DATABASE IF NOT EXISTS `14yi` default charset utf8 COLLATE utf8_general_ci;"
# 自行選擇需要創建哪些東西
try:
self.cursor.execute(sql_create_db)
self.cursor.execute('use 14yi;') # 將當前數據庫設置為剛創建的數據庫
except Exception as e: # 捕獲所有異常並打印，python2 是Exception,e
print("數據庫創建出錯：{}\n退出程序！".format(e))
self.mysql_conn.rollback() # 發生錯誤時回滾
self.mysql_conn.close() # 關閉數據庫連接
exit(0)
def import_txt_to_mysql(self, txt_name_list):
if not txt_name_list:
# 如果未獲取到txt_name_list，則說明文件編碼處理失敗
print("文件編碼可能錯誤，沒有獲取到txt名字列表，請重新檢查！\n程序退出！")
print(0)
else:
# 將當前工作目錄轉到collections文件夾
os.chdir(self.collections_dir)
# 將txt導入mysql
for txt_name in txt_name_list:
table_name = re.findall(".*data-(.*).txt", txt_name)[0]
# 創建一個表含有兩個字段mail和passwd
sql_create_table = """
CREATE TABLE IF NOT EXISTS `{}` (
`mail` varchar(255) DEFAULT NULL,
`passwd` varchar(255) DEFAULT NULL
) ENGINE=MyISAM DEFAULT CHARSET=utf8;
""".format(table_name)
# 將文件插入數據庫
sql_insert_txt = """
LOAD DATA LOCAL INFILE "{}" INTO TABLE `{}`
FIELDS TERMINATED BY ':' LINES TERMINATED BY '\n';
""".format(txt_name, table_name)
# --字段用':'分隔(例子)，記錄用CRLF進行分隔(例子)
try:
print("正在導入{}".format(txt_name))
self.cursor.execute('use 14yi;')
self.cursor.execute(sql_create_table)
self.cursor.execute(sql_insert_txt)
self.mysql_conn.commit() # 提交到數據庫執行
except Exception as e:
print("導入數據出錯：{}\n退出程序！".format(e))
self.mysql_conn.rollback()
self.mysql_conn.close()
exit(0)
class txt_progress:
def __init__(self):
self.root_dir = 'data'
self.collections_dir = 'E:\Pants\BreachCompilation\collections'
self.utf8_txt_name_list = []
if os.path.exists(self.collections_dir):
# 目錄存在則連子文件夾一起刪除
shutil.rmtree(self.collections_dir)
os.mkdir(self.collections_dir)
self.txt_collections(self.root_dir)
self.txt_coding_conv()
def txt_collections(self, path):
# 將每個文件夾內的文件用cat命令存到collections_dir文件夾中
for item in os.listdir(path):
subFile = os.path.join(path + "\\" + item)
if os.path.isdir(subFile):
try:
# 用正則表達式查找分離文件名
txt_name = os.path.join(self.collections_dir, re.split(
r'BreachCompilation\\', os.path.abspath(subFile))[-1].replace("\\", '-'))
print("正在轉換{}".format(txt_name))
# 進入每個文件夾，將每個文件夾內所有內容合併到collections的對應名字的txt內
cc = 'cd {} && cat * > {}.txt'.format(
os.path.abspath(subFile), txt_name)
subprocess.run(cc, shell=True, stdout=subprocess.PIPE)
except Exception as e:
print("轉換報錯：{} \n 程序退出！".format(e))
exit(0)
self.txt_collections(subFile)
def txt_coding_conv(self):
os.chdir(self.collections_dir)
# 將collections_dir文件夾中所有文件轉換為utf-8編碼並返回txt_name_list供insert_txt_to_mysql函數
for item in os.listdir("."):
try:
utf8_txt_name = "utf8{}".format(item) # 新名字
# 強制轉換所有txt到utf-8編碼
subprocess.run('iconv -c -f ISO-8859-1 -t utf-8 {} > {}'.format(
item, utf8_txt_name), shell=True, stdout=subprocess.PIPE)
self.utf8_txt_name_list.append(utf8_txt_name)
except Exception as e:
print("轉換報錯：{} \n 程序退出！".format(e))
exit(0)
return self.utf8_txt_name_list
if __name__ == "__main__":
start_time = timeit.default_timer()
progress_db = db_progress()
progress_txt = txt_progress()
# 將轉化 後的utf8-_txt_name_list傳給導入數據庫函數
progress_db.import_txt_to_mysql(progress_txt.utf8_txt_name_list)
end_time = timeit.default_timer()
print("程序執行完畢，總用時{}".format(end_time-start_time))
效果如下：
暫時無效果，還有點小問題
7x7x#
[19-04-13 更新] 有了模板之後，其他固定字符分割的 txt 導入 mysql 的問題就很簡單了，用模板改一下就行
這次是某國內遊戲網站 7x7x 的數據庫 (網上搜集來的)：
以\t作為分隔符這個數據庫的比較特殊的地方在於它是同時有郵箱\t密碼和用戶名\t密碼兩種數據存在的：
所以創建表的時候就要創建三個字段：email、passwd、username
檢測到此行不含@符號 (即不含郵箱時)，則拼接為\t密碼\t用戶名的形式，(因為email-passwd的數據占較大比例所以把username放在最後面)
全代碼如下：
#!/usr/bin/python
# -*- coding: utf-8 -*-
'''
@author: soapffz
@fucntion: 從7k7k數據庫批量導入固定字符分割的txt到mysql(編碼轉換)
@Comment: 7k7k這幾個文本用文本編碼精緻檢測方法檢測都是utf-8，故此程序中不添加編碼轉換
@comment: 但是由於有的是郵箱-密碼，有的是用戶名-密碼，故添加文本整理函數
@time: 19-04-13
'''
import os # 文本名稱獲取及cmd命令執行
import subprocess
import pymysql # Python3用pymysql,Python2用mysqldb
import timeit # 計算用時
import re # 用正則表達式查找分離文件名
class db_progress(object):
def __init__(self):
self.connect_mysql()
self.create_db()
def connect_mysql(self):
# 連接數據庫,密碼後面可以加數據庫名稱
try:
self.mysql_conn = pymysql.connections.Connection(
'localhost', 'root', 'root', charset='utf8', local_infile=True)
# 得到一個可以執行SQL語句的光標對象，執行完畢返回的結果默認以元組顯示
self.cursor = self.mysql_conn.cursor()
print("數據庫連接成功")
except Exception as e:
print("數據庫連接出錯 ：{}\n退出程序！".format(e))
exit(0)
def create_db(self):
# 如果數據庫不存在則創建，字符集為utf8,排序規則為utf8_general_ci
sql_create_db = "CREATE DATABASE IF NOT EXISTS `soendb` default charset utf8 COLLATE utf8_general_ci;"
# 檢測表是否存在，存在則刪除，然後創建
sql_table_detection = "DROP TABLE IF EXISTS `7k7k`;"
sql_create_table = """
CREATE TABLE `7k7k` (
`email` varchar(255) DEFAULT NULL,
`passwd` varchar(255) DEFAULT NULL,
`username` varchar(255) DEFAULT NULL
) ENGINE=MyISAM DEFAULT CHARSET=utf8;
"""
try:
self.cursor.execute(sql_create_db)
self.cursor.execute('use soendb;') # 將當前數據庫設置為剛創建的數據庫
self.cursor.execute(sql_table_detection) # 檢測表是否存在，存在則刪除，然後創建
self.cursor.execute(sql_create_table)
except Exception as e: # 捕獲所有異常並打印，python2 是Exception,e
print("數據庫創建出錯：{}\n退出程序！".format(e))
self.mysql_conn.rollback() # 發生錯誤時回滾
self.mysql_conn.close() # 關閉數據庫連接
exit(0)
def import_txt_to_mysql(self, txt_name_list):
if not txt_name_list:
# 如果未獲取到txt_name_list，則說明文件編碼處理失敗
print("文件編碼可能錯誤，沒有獲取到txt名字列表，請重新檢查！\n程序退出！")
exit(0)
else:
# 將txt導入mysql
for txt_name in txt_name_list:
sql_insert_txt = """
LOAD DATA LOCAL INFILE "{}" IGNORE INTO TABLE `7k7k`
FIELDS TERMINATED BY '\t' LINES TERMINATED BY '\r\n';
""".format(txt_name)
# --字段用'\t'分隔(例子)，記錄用LFCR進行分隔(例子)
try:
print("正在導入{}".format(txt_name))
self.cursor.execute('use soendb;')
self.cursor.execute(sql_insert_txt)
self.mysql_conn.commit() # 提交到數據庫執行
except Exception as e:
print("導入數據出錯：{}\n退出程序！".format(e))
self.mysql_conn.rollback()
self.mysql_conn.close()
exit(0)
class txt_progress:
def __init__(self):
self.processed_txt_filename_list = []
self.txt_process()
def txt_process(self):
# 整理同時含有郵箱-密碼和用戶名-密碼的文本
# 過濾特殊字符
special_characters = ['?', 'ú', 'ü']
for i in os.listdir("."):
if os.path.splitext(i)[1] == ".txt":
print("正在整理{}".format(i))
with open(i, 'r', encoding='utf-8') as f:
processed_txt_name = "processed-{}".format(i)
with open(processed_txt_name, 'a', encoding='utf-8') as k:
for j in f.readlines():
# 刪除含有特殊字符的行
if any(chs in j for chs in special_characters):
continue
split = j.split('\t')
# 如果賬號不存在或者密碼不存在，則刪除此行
if '' in split:
continue
# 如果密碼中含有中文，則刪除此行
if re.compile(u'[\u4e00-\u9fa5]').search(split[-1]):
continue
# 如果用戶名不是郵箱，重新排列為：\t密碼\t用戶名\n
if '@' not in split[0]:
j = '\t' + \
split[-1].strip('\n') + \
'\t' + split[0] + '\n'
k.write(j)
self.processed_txt_filename_list.append(processed_txt_name)
return self.processed_txt_filename_list
if __name__ == "__main__":
start_time = timeit.default_timer()
progress_db = db_progress()
progress_txt = txt_progress()
progress_db.import_txt_to_mysql(progress_txt.processed_txt_filename_list)
end_time = timeit.default_timer()
print("程序執行完畢，總用時{}".format(end_time-start_time))
運行效果如下：
鄭重聲明： 此文章內分析用到的數據集僅為學習交流，本文未提供任何下載鏈接且已在 24 小時內刪除，請勿用於非法用途否則產生的一切後果自行承擔