Python常用方法
1、遍歷目錄及文件方法
1）根據時間遍歷指定目錄下的文件
[root@04 b2c]# cat file_list.py #!/usr/bin/env python # coding: utf-8 import os def get_file_list(file_path): dir_list = os.listdir(file_path) if not dir_list: return else: # 注意，這里使用lambda表達式，將文件按照最后修改時間順序升序排列 # os.path.getmtime() 函數是獲取文件最后修改時間 # os.path.getctime() 函數是獲取文件最后創建時間 dir_list = sorted(dir_list, key=lambda x: os.path.getmtime(os.path.join(file_path, x))) # print(dir_list) return dir_list print get_file_list('/logarchive/b2c')
執行 [root@04 b2c]# python file_list.py ['2019-05-29.log', '2019-05-30.log', 'b2c_input.py', '2019-05-31.log', '2019-06-01.log', '2019-06-02.log', '2019-06-03.log', '2019-06-04.log', '2019-06-05.log', '2019-06-06.log', '2019-06-07.log', '2019-06-08.log', '2019-06-09.log', '2019-06-10.log', '2019-06-11.log', '2019-06-12.log', '2019-06-13.log', '2019-06-14.log', '2019-06-15.log', '2019-06-16.log', '2019-06-17.log', '2019-06-18.log', '2019-06-19.log', '2019-06-20.log', '2019-06-21.log', '2019-06-22.log', '2019-06-23.log', '2019-06-24.log', 'file.py', 'file_list.py']
2）如果過濾以log為結尾的文件
方法一： [root@04 b2c]# cat file_list.py #!/usr/bin/env python # coding: utf-8 import os def get_file_list(file_path): dir_list = os.listdir(file_path) if not dir_list: return else: # 注意，這里使用lambda表達式，將文件按照最后修改時間順序升序排列 # os.path.getmtime() 函數是獲取文件最后修改時間 # os.path.getctime() 函數是獲取文件最后創建時間 dir_list = sorted(dir_list, key=lambda x: os.path.getmtime(os.path.join(file_path, x))) for filename in dir_list: if filename.endswith(".log"): print filename get_file_list('/logarchive/b2c') [root@app04 b2c]# python file_list.py 2019-05-29.log 2019-05-30.log 2019-05-31.log 2019-06-01.log 2019-06-02.log 2019-06-03.log 2019-06-04.log 2019-06-05.log 2019-06-06.log 2019-06-07.log 2019-06-08.log 2019-06-09.log 2019-06-10.log 2019-06-11.log 2019-06-12.log 2019-06-13.log 2019-06-14.log 2019-06-15.log 2019-06-16.log 2019-06-17.log 2019-06-18.log 2019-06-19.log 2019-06-20.log 2019-06-21.log 2019-06-22.log 2019-06-23.log 2019-06-24.log 方法二： #!/usr/bin/env python # coding: utf-8 import os, glob, time # 定義要清理文件的目錄 root = '/logarchive/b2c' _list = [] for folder in glob.glob(root): for _log_name in glob.glob(folder + '/*.log'): _list.append(_log_name) filelist = sorted(_list) for log_name in filelist: print log_name f = open(log_name, 'r') content = f.readlines() f.close() _time = [] for item in content: time = int(item.strip().split('>>>')[-1].split('ms')[0]) _time.append(time) access_time = sorted(_time) print("b2c 運價訪問時間:") print("min: %s ms"% access_time[0]) print("max: %s ms"% access_time[-1]) sum = 0 access_count = len(_time) for i in _time: sum += i avg = sum/access_count print("avg: %s ms" % avg) print '-'*50 print filelist
3）網上收集的其他方法，python 過濾出某后綴名文件
方法1：
import glob
import os
os.chdir(“./”)
for file in glob.glob(“*.py”):
print file
方法2：
for file in os.listdir(“./”):
if file.endswith(“.py”):
print file
方法3：
for root, dirs, files in os.walk(“./”):
for file in files:
if file.endswith(“.py”):
print os.path.join(root, file)
4) os.walk()方法
import os def all_path(dirname): result = []#所有的文件 for maindir, subdir, file_name_list in os.walk(dirname): print("1:",maindir) #當前主目錄 print("2:",subdir) #當前主目錄下的所有目錄 print("3:",file_name_list) #當前主目錄下的所有文件 for filename in file_name_list: apath = os.path.join(maindir, filename)#合并成一個完整路徑 result.append(apath) return result print(all_path("E:\myTest"))
5) python遍歷目錄下的所有文件和目錄
目錄結構如下圖： test---a------d------g--------g.txt test---a------d------a.txt test---a------e --------b --------c --------1.txt --------2.txt
一、使用os.walk遍歷所有的目錄和文件 1、獲取test目錄下的所有文件 for root,dirs,files in os.walk(r"D:\test"): for file in files: #獲取文件所屬目錄 print(root) #獲取文件路徑 print(os.path.join(root,file))
2、獲取test目錄下的所有目錄 for root,dirs,files in os.walk(r"D:\test"): for dir in dirs: #獲取目錄的名稱 print(dir) #獲取目錄的路徑 print(os.path.join(root,dir))
二、利用os.listdir遞歸獲取所有的目錄路徑和文件路徑 def get_file_path(root_path,file_list,dir_list): #獲取該目錄下所有的文件名稱和目錄名稱 dir_or_files = os.listdir(root_path) for dir_file in dir_or_files: #獲取目錄或者文件的路徑 dir_file_path = os.path.join(root_path,dir_file) #判斷該路徑為文件還是路徑 if os.path.isdir(dir_file_path): dir_list.append(dir_file_path) #遞歸獲取所有文件和目錄的路徑 get_file_path(dir_file_path,file_list,dir_list) else: file_list.append(dir_file_path) if __name__ == "__main__": #根目錄路徑 root_path = r"D:\test" #用來存放所有的文件路徑 file_list = [] #用來存放所有的目錄路徑 dir_list = [] get_file_path(root_path,file_list,dir_list) print(file_list) print(dir_list)
轉自
python 過濾出某后綴名文件 - 快樂多巴胺 - 博客園 http://www.rzrgm.cn/pscc/p/10122517.html
python按時間排序目錄下的文件 - Mr番茄蛋的博客 - CSDN博客 https://blog.csdn.net/qq_35203425/article/details/80903348
python遍歷目錄下的所有文件和目錄詳細介紹 - 修煉之路 - CSDN博客 https://blog.csdn.net/sinat_29957455/article/details/82778306
Windows列出磁盤文件及文件夾大小
簡單版：提取指定盤和大小的文件及文件夾
#!/usr/bin/env python # coding: utf-8 # 版本：提取指定盤和大小的文件及文件夾 import os from os.path import join, getsize def get_paths_size(dirs, maxnum): # 提取指定文件夾和大小的函數 print("{} -> 文件夾內文件占用空間：".format(dirs)) size = 0 for root, dirs, files in os.walk(dirs): sums = sum([getsize(join(root, file)) for file in files]) // 1024 // 1024 if sums > maxnum: print('{:>8,d} MB -> {}'.format(sums, root)) size += sums print('{:>8,d} MB -> 總大小'.format(size)) def get_files_size(dirs, maxnum): # 提取指定文件夾內文件和大小的函數 print("{} -> files usage of disk:".format(dirs)) size = 0 for root, dirs, files in os.walk(dirs): for file in files: fpth = join(root, file) sums = getsize(fpth) // 1024 // 1024 if sums > maxnum: print('{:>8,d} MB -> {}'.format(sums, fpth)) size += sums print('{:>8,d} MB -> 總大小'.format(size)) def main(): paths = r'C:\\' numbs = 1024 # -> MB # paths = input(r'請輸入盤符(如：D:\Python\Python38：') # numbs = int(input(r'請)輸入大小單位MB(如：1000)：')) get_paths_size(paths, numbs) get_files_size(paths, numbs) if __name__ == '__main__': main()
高級版：直接提取C-I盤所有大于1GB（參數可以修改）的文件及文件夾，輸出到CSV文件
# -*- coding:utf-8 -*- import os from os.path import join, getsize import csv """ 版本：直接提取C-I盤所有大于1GB的文件及文件夾 并輸出CSV文件 盤符不存在則程序執行結束！ 文件夾：size_C_path.csv 文件：size_C_file.csv """ def get_dirs_size(dirs, maxnum): print(dirs) # CSV文件名后綴 fname = dirs.replace('\\', '_').replace(':', '').replace('/', '_') path_size = [] # 路徑大小列表 file_size = [] # 文件大小列表 size = 0 # 合計 for root, dirs, files in os.walk(dirs): for f in files: fp = join(root, f) try: su = getsize(fp) // 1024 // 1024 except: pass #continue # su = getsize(fp) // 1024 // 1024 if su > maxnum: file_size.append([su, fp]) print('{:>8,d} MB --> {}'.format(su, fp)) pass sums = sum([getsize(join(root, file)) for file in files]) // 1024 // 1024 size += sums if sums > maxnum: path_size.append([sums, root]) print('{:>8,d} MB --> {}'.format(sums, root)) pass print('{:>8,d} MB --> {}'.format(size, dirs)) # 調用導出CSV函數導出CSV文件 savecsvfile(path_size, ['大小', '文件夾'], 'size_{}path.csv'.format(fname)) savecsvfile(file_size, ['大小', '文件'], 'size_{}file.csv'.format(fname)) def savecsvfile(rows, header, csv_name): # 導出CSV文件函數 # if not os.path.exists(csv_name): with open(csv_name, 'w', newline='', encoding='utf-8') as f: fc = csv.writer(f) fc.writerow(header) fc.writerows(rows) print(csv_name, '導出成功！') def main(): # 所有盤符列表推導式：從C盤到I盤 paths = ["{}:/".format(x) for x in 'CDEFGHI'] for p in paths: if not os.path.exists(p): print('盤符 -> {} 不存在!'.format(p)) continue # 只提取大于1000MB的文件和文件夾 get_dirs_size(p, 1000) if __name__ == '__main__': main()
2、時間datetime、time模塊
# 時間格式轉換 time.strftime('%Y-%m-%d %H:%M:%S') '2016-11-12 01:20:54' # 獲取昨天日期
date = datetime.date.today()
>>> date.today() - timedelta(days=1) datetime.date(2016, 11, 11) >>> date.isoformat(date.today() - timedelta(days=1)) '2016-11-11'
# 獲取明天日期 >>> date.today() + timedelta(days=1) datetime.date(2016, 11, 13) >>> date.isoformat(date.today() + timedelta(days=1)) '2016-11-13'
import datetime # 獲取前100天日期 (datetime.datetime.now() - datetime.timedelta(days = 100)).strftime("%Y-%m-%d") Out[18]: '2019-12-13' # 獲取昨天日期 (datetime.datetime.now() - datetime.timedelta(days = 1)).strftime("%Y-%m-%d") Out[19]: '2020-03-21'
3、多進程
from multiprocessing import Pool, current_process import urllib2 urls = [ 'http://www.baidu.com', 'http://www.jd1.com', 'http://www.sina.com', 'http://www.163.com', ] def status_code(url): print( 'process name:', current_process().name) try: req = urllib2.urlopen(url, timeout=2) return req.getcode() except urllib2.URLError: return p = Pool(processes=4) for url in urls: r = p.apply_async(status_code, args=(url,)) #print r.get() #if r.get(timeout=1) == 200: if r.get() == 200: print("%s OK" %url) else: print("%s NO" %url)
轉自 博客地址：http://lizhenliang.blog.51cto.com and https://yq.aliyun.com/u/lizhenliang
4、如果文件不存在就創建
if not os.path.exists('host_info'): f = open('host_info', 'w') f.close()
5、壓縮文件
#!/usr/bin/env python # -*- coding: UTF-8 -*- import os import tarfile tar = tarfile.open("/root/test/test.tar.gz","w:gz") # 創建壓縮包名 for path,dir,files in os.walk("/root/test"): # 遞歸文件目錄 for file in files: fullpath = os.path.join(path,file) tar.add(fullpath) # 創建壓縮包 tar.close()
3種壓縮方法
gzip [解壓縮gzip 刪除原文件] #壓縮gzip import gzip f_in = open('file.log', 'rb') f_out = gzip.open('file.log.gz', 'wb') f_out.writelines(f_in) f_out.close() f_in.close() #壓縮gzip File = 'xuesong_18.log' g = gzip.GzipFile(filename="", mode='wb', compresslevel=9, fileobj=open((r'%s.gz' %File),'wb')) g.write(open(r'%s' %File).read()) g.close() #解壓gzip g = gzip.GzipFile(mode='rb', fileobj=open((r'xuesong_18.log.gz'),'rb')) open((r'xuesong_18.log'),'wb').write(g.read()) tarfile [歸檔壓縮tar.gz 保留原文件] # 壓縮tar.gz import os import tarfile tar = tarfile.open("/tmp/tartest.tar.gz","w:gz") # 創建壓縮包名 for path,dir,files in os.walk("/tmp/tartest"): # 遞歸文件目錄 for file in files: fullpath = os.path.join(path,file) tar.add(fullpath) # 創建壓縮包 tar.close() # 解壓tar.gz import tarfile tar = tarfile.open("/tmp/tartest.tar.gz") #tar.extract("/tmp") # 全部解壓到指定路徑 names = tar.getnames() # 包內文件名 for name in names: tar.extract(name,path="./") # 解壓指定文件 tar.close() zipfile [解壓縮zip 最大2G] # 壓縮zip import zipfile,os f = zipfile.ZipFile('filename.zip', 'w' ,zipfile.ZIP_DEFLATED) # ZIP_STORE 為默認表不壓縮. ZIP_DEFLATED 表壓縮 #f.write('file1.txt') # 將文件寫入壓縮包 for path,dir,files in os.walk("tartest"): # 遞歸壓縮目錄 for file in files: f.write(os.path.join(path,file)) # 將文件逐個寫入壓縮包 f.close() # 解壓zip if zipfile.is_zipfile('filename.zip'): # 判斷一個文件是不是zip文件 f = zipfile.ZipFile('filename.zip') for file in f.namelist(): # 返回文件列表 f.extract(file, r'/tmp/') # 解壓指定文件 #f.extractall() # 解壓全部
# 如果文件大于2G
# 也可以使用context manager
my_zip = zipfile.ZipFile('files.zip', 'w')
# 如果文件大于2G
zipfile.ZipFile('files.zip', 'w', zipfile.ZIP_DEFLATED, True)
python統計apache、nginx訪問日志IP訪問次數
前言：python統計apache、nginx訪問日志IP訪問次數并且排序（顯示前20條）。其實用awk+sort等命令可以實現，用awk數組也可以實現，這里只是用python嘗試下。
apache腳本：
ips = {}
with open("/root/mail_access_log-20180629") as fh:
for line in fh:
ip = line.split(" ")[0]
if 6 < len(ip) <=15:
ips[ip] = ips.get(ip, 0) + 1
ip_num = []
for ipaddr,num in ips.items():
ip_num.append((ipaddr,num))
ip_num.sort(key=lambda x: x[1], reverse=True)
for ipaddr,num in ip_num[:20]:
print('IP地址為{}, 訪問次數為{}'.format(ipaddr,num))
nginx腳本：
ips = {}
with open("/root/access.log-20180629") as fh:
for line in fh:
ip = line.split(" ")[0]
if 6 < len(ip) <=15:
ips[ip] = ips.get(ip, 0) + 1
ip_num = []
for ipaddr,num in ips.items():
ip_num.append((ipaddr,num))
ip_num.sort(key=lambda x: x[1], reverse=True)
for ipaddr,num in ip_num[:20]:
print('IP地址為{}, 訪問次數為{}'.format(ipaddr,num))
壓縮和解壓文件
# 壓縮tar.gz import os import tarfile tar = tarfile.open("/root/test/test.tar.gz","w:gz") # 創建壓縮包名 for path,dir,files in os.walk("/root/test"): # 遞歸文件目錄 for file in files: fullpath = os.path.join(path,file) tar.add(fullpath) # 創建壓縮包 tar.close()
# 解壓tar.gz import tarfile tar = tarfile.open("/root/test/test.tar.gz") #tar.extract("/tmp") # 全部解壓到指定路徑 names = tar.getnames() # 包內文件名 for name in names: tar.extract(name,path="./") # 解壓指定文件 tar.close()
gzip [解壓縮gzip 刪除原文件] #壓縮gzip import gzip f_in = open('file.log', 'rb') f_out = gzip.open('file.log.gz', 'wb') f_out.writelines(f_in) f_out.close() f_in.close() #壓縮gzip File = 'xuesong_18.log' g = gzip.GzipFile(filename="", mode='wb', compresslevel=9, fileobj=open((r'%s.gz' %File),'wb')) g.write(open(r'%s' %File).read()) g.close() #解壓gzip g = gzip.GzipFile(mode='rb', fileobj=open((r'xuesong_18.log.gz'),'rb')) open((r'xuesong_18.log'),'wb').write(g.read())
輸入參數判斷
try: textfile1=sys.argv[1] textfile2=sys.argv[2] except Exception,e: print "Error:"+str(e) print "Usage: simple3.py filename1 filename2" sys.exit()
判斷輸入參數是否為2個
if len(sys.argv) > 2: dir1=sys.argv[1] dir2=sys.argv[2] else: print "Usage: ", sys.argv[0], "datadir backupdir" sys.exit()
輸入去掉空白
_input = input('please input your opinion: ') ret = _input.strip()
Python執行linux命令并得到執行結果
subprocess
#!/usr/bin/env python # coding: utf-8 import subprocess def create_process(cmd): p = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT) result = p.stdout.read() code = p.wait() return code, result code,result = create_process('ls -l test.py') #print(code) #print(result) #根據返回碼判斷是否執行成功 if code: print("備份MySQL失敗") print(code, result) else: print("備份MySQL成功") print(code, result)
利用執行命令后的返回代碼判斷
利用執行命令后的返回代碼判斷 #coding:utf-8 import os import subprocess cmd = 'ls -l test.txt' retcode = subprocess.call(cmd, shell=True) result = p.returncode if retcode != 0: print("備份MySQL失敗") #print(ret) else: print("備份MySQL成功")
commands
#!/usr/bin/env python # coding: utf-8 import sys import os import commands status,output=commands.getstatusoutput('ls -l test.py') if(status!=0): print "command is failed, please check" print output sys.exit(1) else: print "command is sucess."
os.system
os.system('ls -l /root')
ftp客戶端ftplib
from ftplib import FTP
ftp = FTP('10.112.13.98') # 連接ftp地址 FTP(host,port,timeout)
ftp.login(test1,f0W1V7kw) # 使用默認anonymous登錄 login(user,passwd)
ftp.cwd('debian') # 切換到目錄debian
ftp.retrlines('LIST') # 打印目錄列表
ftp.retrbinary('RETR README', open('README', 'wb').write) # 下載文件寫到本地
ftp.storbinary("STOR "+filename, fp, bufsize) # 上傳目標文件
ftp.delete('filename') # 刪除ftp中文件
ftp.mkd('dirname') # 在ftp上創建目錄
ftp.size('filename') # 查看文件大小
ftp.quit()
示例
#!/usr/bin/env python # coding: utf-8 import time import os import ftplib from ftplib import FTP user='username' password='password' filename = 'u_ex'+ time.strftime('%y%m%d') + '.log' for root,dirs,files in os.walk(r"C:\inetpub\logs\LogFiles\W3SVC2"): for file in files: #獲取文件所屬目錄 # print(root) #獲取文件路徑 # print(os.path.join(root,file)) if filename in os.path.join(root,file): print(os.path.join(root,file)) local_file = os.path.join(root,file) # print(type(local_file)) bufsize = 1024 fp = open(local_file, 'rb') ftp = FTP('IP') # 連接ftp地址 FTP(host,port,timeout) ftp.login(user, password) # 使用默認anonymous登錄 login(user,passwd) ftp.cwd('debian') # 切換到目錄debian ftp.storbinary("STOR "+filename, fp, bufsize) # 上傳目標文件 ftp.retrlines('LIST') # 打印目錄列表 ftp.quit() fp.close() print('upload file success.')
python按行讀取文件，如何去掉換行符"\n"
for line in file.readlines(): line=line.strip('\n')
python 字典、列表、字符串 之間的轉換
1、列表與字符串轉換
列表轉字符串：
將列表中的內容拼接成一個字符串
將列表中的值轉成字符串
字符串轉列表：
用eval轉換
將字符串每個字符轉成列表中的值
將字符串按分割成列表
2、列表與字典轉換
列表轉字典：
將兩個列表轉成字典
將嵌套列表轉為字典
字典轉列表：
字典中鍵、值轉為列表
3、字典與字符串轉換
字符串轉字典：
用eval轉換
用json.loads 轉換
字典轉字符串：
用json.dumps 轉換
強轉換
列表轉換成字典
def GenPassword_reg(length=16, chars=string.ascii_letters + string.digits): return ''.join([choice(chars) for i in range(length)]) dic = {} # result 是個列表 for i in result: dic[i] = GenPassword_reg() print(dic)
shell與python間傳遞變量方法
python -> shell：1.環境變量 import os var=123或var=’123’ os.environ[’var’]=str(var) #environ的鍵值必須是字符串 os.system(’echo $var’) 2.字符串連接 import os path=’/root/a.txt’ var=[1] var=’bash’ os.system(’echo ’ + path) #注意echo后有空格 os.system(‘echo ‘ + str(var[0])) os.system(’echo ’ + var + ’ /root/c.sh’) #注意echo后和/root前有空格 3.通過管道 import os var=’123’ os.popen(’wc -c’, ’w’).write(var) 4.通過文件 output = open(‘/tmp/mytxt’, ‘w’) output.write(S) #把字符串S寫入文件 output.writelines(L) #將列表L中所有的行字符串寫到文件中 output.close() 5.通過重定向標準備輸出 buf = open(’/root/a.txt’, ’w’) print >> buf, ‘123\n’, ‘abc’ 或 print >> open(‘/root/a.txt’, ‘w’), ‘123\n’, ‘abc’ #寫入或生成文件 print >> open(‘/root/a.txt’, ‘a’), ‘123\n’, ‘abc’ #追加
shell -> python：
1.管道 import os var=os.popen(’echo -n 123’).read( ) print var 2. import commands var=commands.getoutput(’echo abc’) #輸出結果 var=commands.getstatusoutput(’echo abc’) #退出狀態和輸出結果 3.文件 input = open(‘/tmp/mytxt’, ‘r’) S = input.read( ) #把整個文件讀到一個字符串中 S = input.readline( ) #讀下一行（越過行結束標志） L = input.readlines( ) #讀取整個文件到一個行字符串的列表中
轉自：原文鏈接：https://blog.csdn.net/kpshare/article/details/7523511
python用變量拼接成 shell命令執行
def create_process(cmd):
p = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
result = p.stdout.read()
code = p.wait()
return code, result
mysql_trunc = '/usr/local/mysql/bin/mysql -uroot -p -h localhost -N -e "select * from moffice_hnair_schedule_log_his.' + str(final_list[-1]) + ' limit 1"' print(mysql_trunc) code, result = create_process(mysql_trunc) if code != 0: print("執行MySQL清除log_interface_result失敗") print(result) else: print("執行MySQL清除log_interface_result成功") print(result)
在windows下刪除文件報錯
WindowsError: [Error 123] : 'C:\\Users\\chh-huang\\Desktop\test\\2019-11-23.txt'
filename = r'C:\Users\chh-huang\Desktop\test\\' + name
需要在路徑前加r，這樣就沒問題了
filename = r'C:\Users\chh-huang\Desktop\test' + os.sep + name 等于 filename = r'C:\Users\chh-huang\Desktop\test\\' + name
其中，os.sep根據你所處的平臺，自動采用相應的分隔符號
>>> os.sep
'/'
Python刪除過期文件
# -*- coding:utf8 -*- #!/usr/bin/python import os import datetime import time import shutil for root, dirs, files in os.walk(r'C:\Users\chh-huang\Desktop\test'): for name in files: (y1, m1, d1) = (int(x) for x in name.split('.')[0].split('-')) # print(y1, m1, d1) date1 = datetime.date(y1, m1, d1) datenow = time.strftime('%Y%m%d') y2 = int(datenow[:4]) m2 = int(datenow[4:6]) d2 = int(datenow[6:]) date2 = datetime.date(y2, m2, d2) # print(date1, date2)
# 刪除大于等于2天的文件 if (date2 - date1).days >= 2: print 'Expired file! Deleting file... ', name print('os.remove(%s) ' % name) # filename = r'C:\Users\chh-huang\Desktop\test\\' + name filename = r'C:\Users\chh-huang\Desktop\test' + os.sep + name os.remove(filename)
import os import time import datetime import shutil filepath = r"xxx" timestamp = os.path.getctime(filepath) # 獲取文件創建的時間戳 timestruct = time.localtime(timestamp) # 將時間戳轉換成本地時間 file_create_time = time.strftime('%Y-%m-%d', timestruct) # 本地時間轉換成字符串 current_time= time.strftime('%Y-%m-%d') # 獲取當前時間 # 利用datetime來計算時間差 day1 = datetime.datetime.strptime(file_create_time, '%Y-%m-%d') day2 = datetime.datetime.strptime(current_time, '%Y-%m-%d') delta = day2 - day1 print(delta.days) # 獲取當前時間與文檔創建時間之差 if delta.days>3: # 刪除創建日期大于3天的文件 shutil.rmtree(filepath)
參考
(3條消息)python 根據時間間隔刪除文件夾內修改時間比較早的文件_weixin_38383877的博客-CSDN博客 https://blog.csdn.net/weixin_38383877/article/details/82897359
Python實現刪除文件夾內規定時間內的文件 - 愛我所艾 - 博客園 http://www.rzrgm.cn/wangchy0927/archive/2013/08/30/3291304.html
(3條消息)python 刪除指定時間間隔之前的文件_百惱專欄-CSDN博客 https://blog.csdn.net/zsy19881226/article/details/72638036
Python實現刪除文件夾內規定時間內的文件 - 愛我所艾 - 博客園 http://www.rzrgm.cn/wangchy0927/archive/2013/08/30/3291304.html
使用Python批量刪除文件前綴的操作方法-百度經驗 https://jingyan.baidu.com/article/1974b289eaab0eb4b0f7746a.html
清理以開頭過期日志文件
#!/usr/bin/env python # coding: utf-8 import os import shutil import datetime import time src_dir = r'C:\Windows\System32\LogFiles\Sum' dst_dir = r'G:\log-svcbak' for file in os.listdir(src_dir): # 找出以svc開頭的log文件 if file[:3] == 'Svc' and file[-3:] == 'log': # 排除Svctmp.log或Svc.log文件 if file == 'Svctmp.log' or file == 'Svc.log': continue # print('%s was been moved' % file) src_file = src_dir + '\\' + file # os.remove(src_file) # print(src_file) timestamp = os.path.getmtime(src_file) timestruct = time.localtime(timestamp) file_create_time = time.strftime('%Y-%m-%d', timestruct) current_time= time.strftime('%Y-%m-%d') day1 = datetime.datetime.strptime(file_create_time, '%Y-%m-%d') day2 = datetime.datetime.strptime(current_time, '%Y-%m-%d') delta = day2 - day1 # print(file_create_time, current_time, day1, day2, delta.days, src_file) # 刪除大于21天的文件 if delta.days > 21: print(src_file) # os.remove(src_file) # shutil.rmtree(filepath)
文件的讀、寫
#!/usr/bin/env python #coding:utf-8 # 讀取文件 f = open(r'2', 'r') content = f.readlines() f.close() print(content) # 寫入文件多行數據 with open('/etc/yum.repos.d/redhat7_nanshu.repo', 'w') as f: # 采用with語法結構可以讓程序自動關閉打開的文件，不用自己處理 f.writelines(['[base]\n', 'name=CentOS-7 - Base\n']) # 讀取打印多行文件 with open('text.txt') as rows: for row in rows: print(row, end='')
文件的寫入多行
lines = ["這是第一行內容\n", "這是第二行內容\n", "這是第三行內容\n"] file.writelines(lines)
文件的移動和復制
#!/usr/bin/env python #coding:utf-8 src_dir = '/etc/yum.repos.d' dst_dir = '/tmp/yum-bak' if not os.path.exists(dst_dir): os.makedirs(dst_dir) if os.path.exists(dst_dir): for root,dirs,files in os.walk(src_dir): for file in files: #獲取文件所屬目錄 print(root) #獲取文件路徑 print(os.path.join(root,file)) # 文件完整路徑 src_file = os.path.join(root, file) # 復制文件 shutil.copy(src_file, dst_dir) # 移動文件 shutil.move(src_file, dst_dir) # 移動文件并重命名 dst_file = os.path.join(dst_dir, file + time.strftime('%Y-%m-%d %H:%M:%S')) shutil.move(src_file, dst_file) code,result = create_process("cat /etc/redhat-release|awk '{print $1}'")
判斷文件是否存在
if not os.path.exists(dst_dir): os.makedirs(dst_dir) if os.path.exists(dst_dir):
判斷文件是否存在，如果存在輸出路徑
#!/usr/bin/env python # coding: utf-8 import os with open('lost_file.txt') as rows: for row in rows: row = row.strip('\n') dirname = row[:2] filename = "/data/dfd/source/" + dirname + "/" + row if not os.path.exists(filename): print(f"{row} is not exist.") else: print(f"{row} is exist.") print(filename) print("\n")
端口telnet
#!/usr/bin/env python
#coding:utf-8
import telnetlib
# 判斷端口是否通
def port_scan(ip):
try:
telnetlib.Telnet(host=ip, port=80, timeout=2)
print(ip, 80, 'is avaliable')
except Exception, e:
print(ip, 80, 'is not avaliable')
raise e
exit()
else:
pass
port_scan('www.baidu.com')
多線程
# coding:utf8 import time from concurrent.futures import ThreadPoolExecutor TEST_COUNT = 100 THREADS_SUM = 50 thread_pool = ThreadPoolExecutor(THREADS_SUM) def test(): time.sleep(1) print('run test') def runfunc(count): for i in range(count): thread_pool.submit(testi) runfunc(TEST_COUNT)
windows輸出中文亂碼
def paint(): # os.system('cls') print("=================================================") print("| The Points Menu: |") print("=================================================") print("| 1. gateway上傳變更包 |").decode("utf-8").encode("gbk")
print("| 2. shopping上傳變更包 |").decode("utf-8").encode("gbk") print("| 0. 退出 |").decode("utf-8").encode("gbk") print("================================================")
用.decode("utf-8").encode("gbk")來解決。
Python啟動windows服務或程序
啟動服務
#!/usr/bin/python # -*- coding: utf-8 -*- import os try: os.system('NET STOP MSSQLSERVER') except OSError: print "system error" os.system('NET START MSSQLSERVER')
啟動程序
import os def open_app(app_dir): os.startfile(app_dir) #os.startfile（）打開外部應該程序，與windows雙擊相同 if __name__ == "__main__": app_dir = r'G:\yeshen\Nox\bin\Nox.exe'#指定應用程序目錄 open_app(app_dir)
用enumerate對現有列表每項進行修改
對現有列表img_list，每項中添加新的字符串，不用新建列表，用enumerate就可以做到
for i,v in enumerate(img_list): img_list[i] = "http://www.xueshengmai.com" + v print(img_list)
Python爬蟲
實例1：爬取豆瓣圖書名
# coding: utf-8 import requests import re # 制作headers head = {'User-Agent':'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/60.0.3112.113 Safari/537.36'} page_obj = requests.get("https://book.douban.com/tag/%E5%B0%8F%E8%AF%B4", headers = head) page = page_obj.text # print(page) regex = '<a title="(.+?)"' data = re.findall(regex, page) for i in data: print(i.encode('gbk')) print(data)
實例2：爬取校花網站圖片
#!/usr/bin/env python # coding: utf-8 import requests import re from urllib import request page = requests.get("http://www.xueshengmai.com/hua/") content = page.text # print(content) # 正則表達式 regex = '<img width="210" alt=".+?" src="(.+?)" />' img_list = re.findall(regex, content) print(img_list) # 拼接圖片地址 for i,v in enumerate(img_list): img_list[i] = "http://www.xueshengmai.com" + v print(img_list) # 下載圖片 count = 0 for i in img_list: # i為下載地址 后面為保存的名字 request.urlretrieve(i, "美女%s.jpg" % count) print("第%s圖片下載完成" % (count + 1)) count += 1
with open寫入Excel CSV文件
將數據寫入guazi.csv文件
with open(r'guazi.csv', 'w') as f: for li in title_url: name = li.xpath('.//h2/text()').extract() year = li.xpath('.//div/text()').extract() price = li.xpath('.//div/p/text()').extract() price_real = li.xpath('.//em/text()').extract() img = li.xpath('.//img/@src').extract() print('車型：{}， 車況：{},{}, 價格：{}, 原價：{}, 圖片：{}'.format(name[0], year[0], year[1], price[0], price_real[0], img[0])) f.write('{},{},{},{},{}\n'.format(name[0], year[0], year[1], price[0], price_real[0]))
保存為中文csv文件
# 保存到CSV文件中, 保存為中文編碼選擇gb1803 with open(name+'.csv', 'w', encoding='gb18030') as f: f.write('職位名,公司名,薪資,工作地點,職位連接,公司鏈接,發布日期\n')
Tomcat配置文件注釋
#!/usr/bin/env python # coding: utf-8 # Tomcat配置文件注釋 import os import re with open(r'server.xml.20200226', 'r') as fr, open("server.xml.20200226.bak", 'w') as fw: for line in fr: regx = re.compile(r'\<Connector.*?protocol="AJP/1.3" redirectPort="8443" \/\>') # 匹配替換的內容 ret = ''.join(re.findall(regx, line)) # 列表轉成成字符串 if ret: print('匹配成功，{}'.format(ret)) ret1 = ''.join(ret.replace('<Connector', '<-- <Connector')) line = ''.join(ret1.replace('>', '> -->')) print('替換后，{}'.format(line)) fw.write(line)
執行效果
Python對文件修改，替換，刪除
一、Python使用replace修改原文件內容方式：
#!/usr/bin/env python
# -*- coding:utf8 -*-
old_str = "aaa" #老文件內容字段
new_str = "bbb" #要改成字段
file_data = ''
with open('/opt/1.txt', 'r', encoding='utf-8') as f:
for line in f:
if old_str in line:
line = line.replace(old_str, new_str)
file_data += line
with open('/opt/1.txt', 'w',encoding='utf-8') as f:
f.write(file_data)
二、python 使用正則表達式 替換文件內容 re.sub 方法替換
import re,os
def alter(file,old_str,new_str):
with open(file, "r", encoding="utf-8") as f1,open("%s.bak" % file, "w", encoding="utf-8") as f2:
for line in f1:
f2.write(re.sub(old_str,new_str,line))
os.remove(file)
os.rename("%s.bak" % file, file)
alter("file1", "admin", "password"
三、python 匹配IP及端口
#!/usr/bin/env python # -*- coding: utf-8 -*- # @Date : 2017-08-30 20:38:23 # @Author : EnderZhou (zptxwd@gmail.com) # @Link : http://www.rzrgm.cn/enderzhou/ # @Version : $Id$ import requests import re def open_url(url): req = requests.get(url=url,headers=headers) html = req.content return html def get_ipandport(html): #默認網頁內的IP地址位于端口號之前，并且中間至少隔了一個非數字的字符串 # (?:((?:\d|[1-9]\d|1\d{2}|2[0-5][0-5])\.(?:\d|[1-9]\d|1\d{2}|2[0-5][0-5])\.(?:\d|[1-9]\d|1\d{2}|2[0-5][0-5])\.(?:\d|[1-9]\d|1\d{2}|2[0-5][0-5])) 用于匹配IP地址 # (6[0-5]{2}[0-3][0-5]|[1-5]\d{4}|[1-9]\d{1,3}|[0-9]) 用于匹配端口號 注意端口號匹配規則應從大到校排序 # 使用 ([0-9]|[1-9]\d{1,3}|[1-5]\d{4}|6[0-5]{2}[0-3][0-5]) 替換即可觀察到原因。 # 使用\D+?匹配IP地址與端口號中間至少隔了一個非數字的字符串 p = r'(?:((?:\d|[1-9]\d|1\d{2}|2[0-5][0-5])\.(?:\d|[1-9]\d|1\d{2}|2[0-5][0-5])\.(?:\d|[1-9]\d|1\d{2}|2[0-5][0-5])\.(?:\d|[1-9]\d|1\d{2}|2[0-5][0-5]))\D+?(6[0-5]{2}[0-3][0-5]|[1-5]\d{4}|[1-9]\d{1,3}|[0-9]))' iplist = re.findall(p,html) for each in iplist: print(each) if __name__ == '__main__': url = 'http://www.66ip.cn/' headers = {'user-agent':'Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/60.0.3112.78 Safari/537.36'} get_ipandport(open_url(url))
對文件所有行前添加‘#’注釋
#!/usr/bin/env python # coding: utf-8 import os import re with open(r'server.xml.20200226', 'r') as fr, open("server.xml.20200226.bak", 'w') as fw: for line in fr: after = '#' + line fw.write(after)
對文件進行注釋#，空行不注釋
#!/usr/bin/env python # coding: utf-8 import os import re with open(r'server.xml.20200226', 'r') as fr, open("server.xml.20200226.bak", 'w') as fw: for line in fr: if not line.strip() or line.startswith('#'): #判斷是否是空行或注釋行 fw.write(line) else: after = '#' + line print(after) fw.write(after)
Python發送郵件
純文本郵件
import smtplib from email.mime.text import MIMEText def send_text_only(): # 1.發件人、授權碼，收件人信息 from_addr = "9071xxxxx@qq.com" pwd = "aqinixxxxxxxxxxx" to_addr = "sunxxxxx@efee.com.cn" # 發送給多個，直接["郵箱1","郵箱2","郵箱3"] # 2.發送內容 msg = MIMEText("hello, I am suner!!") msg["Subject"] = "測試報告" msg["From"] = from_addr msg["To"] = to_addr # 3.連接smtp服務器，登錄服務器并發送文本 smtp_server = "smtp.qq.com" server = smtplib.SMTP(smtp_server, 25) server.login(from_addr, pwd) server.sendmail(from_addr, to_addr, msg.as_string()) # as_string（）把MIMEText變成一個str server.close() if __name__ == '__main__': send_text_only()
多個附件
import smtplib from email.mime.text import MIMEText from email.mime.multipart import MIMEMultipart from email.mime.application import MIMEApplication def send_enclosure(): # 1.發件人、授權碼，收件人信息 from_addr = "907xxxxx@qq.com" pwd = "aqinirkcqyxubdbf" to_addr = "sunyingying@efee.com.cn" # 2.創建實例對象，設置主題等信息 msg = MIMEMultipart() msg["Subject"] = "收到一個文檔文件和圖片文件" msg["From"] = from_addr msg["To"] = to_addr # 郵件內容（按每個部分） part1 = MIMEText("這是一個word文件和圖片") msg.attach(part1) # 添加圖片附件 part2 = MIMEApplication(open("meimei.jpg", "rb").read()) part2.add_header("Content-Disposition", "attachment", filename="meimei.jpg") msg.attach(part2) # 添加word附件 part3 = MIMEApplication(open("1.docx","rb").read()) part3.add_header("Content-Disposition", "attachment", filename="1.docx") msg.attach(part3) # 3.連接smtp服務器，登錄服務器并發送文本 smtp_server = "smtp.qq.com" server = smtplib.SMTP(smtp_server,25) server.login(from_addr,pwd) server.sendmail(from_addr,to_addr,msg.as_string()) # as_string（）把MIMEText變成一個str server.close() if __name__ == '__main__': send_enclosure()
參考
(3條消息)Python基礎-46-郵件發送（帶附件）_Python_karry_孫二的博客-CSDN博客
https://blog.csdn.net/qq_39620483/article/details/90903782
Python輸入參數執行腳本
l = len(sys.argv) if l == 1: s = """ 請輸入參數 參數說明： up_his 更新歷史記錄表 up_hot 更新實時熱搜 up_det 更新詳細表 """ print(s) else: order = sys.argv[1] if order == "up_his": update_history() elif order == "up_det": update_details() elif order == "up_hot": get_baidu_hot() update_hot_search()
# 執行
python test.py up_his
使用python讀取文件數據并轉化為列表
#首先找到需要讀取的文件目錄，絕對路徑或者相對路徑均可 filename = r"processed.cleveland.data" #先聲明一下a a=[] try: #打開文件 fp=open(filename,"r") print('%s 文件打開成功' % filename) for line in fp.readlines(): ''' 當你讀取文件數據時會經常遇見一種問題， 那就是每行數據末尾都會多個換行符‘\n’， 所以我們需要先把它們去掉 ''' line=line.replace('\n','') #或者line=line.strip('\n') #但是這種只能去掉兩頭的，可以根據情況選擇使用哪一種 line=line.split(',') #以逗號為分隔符把數據轉化為列表 a.append(line) fp.close() print("文件內容為：") print(a) except IOError: print("文件打開失敗，%s文件不存在" % filename)
正則的使用和舉例
import re str = 'adbdaf124124d' pattern = re.compile(r'\d') print(re.findall(pattern, str)) ['1', '2', '4', '1', '2', '4'] r'\d’中的r表示字符串中的轉義 舉例
s1 = r'\n' print(s1) 輸出\n
pattern1 = re.compile(r'\d+') print(re.findall(pattern1, str)) ['124124'] # 匹配手機號 str = '13522600131\n0898-56156414' pattern = re.compile(r'\d{11}') # 手機號是11位 print(re.findall(pattern, str)) ['13522600131'] # 單詞結界\b str = 'verb people' pattern = re.compile(r'le\b') print(re.findall(pattern, str)) ['le'] str = 'verb peoper' pattern = re.compile(r'\w+er\b') print(re.findall(pattern, str)) ['peoper'] str = '13522600131,0898-56156414,17877777777,18888888888' pattern = re.compile(r'(1(35|88)\d{8})') print(re.findall(pattern, str)) [('13522600131', '35'), ('18888888888', '88')] # 分組（） pattern = re.compile(r'(1(35|88)(\d+))') print(re.findall(pattern, str)) [('13522600131', '35', '22600131'), ('18888888888', '88', '88888888')] 解釋 (1(35|88)(\d+))這個分組匹配到，13522600131、18888888888 (35|88) 這個分組匹配到，35、88 (\d+)這個分組匹配到，22600131、88888888 # re.I忽略大小寫 str = 'abc,ABC' pattern = re.compile(r'abc', re.I) print(re.findall(pattern, str)) ['abc', 'ABC'] # re.M多行匹配 pattern = re.compile(r'ab', re.I|re.M) print(re.findall(pattern, str)) ['ab', 'AB'] re.A、re.U str = 'abc,我是中國人' # 匹配ASCII碼 pattern = re.compile(r'\w', re.A) print(re.findall(pattern, str)) ['a', 'b', 'c'] # 匹配Unicode碼 pattern = re.compile(r'\w', re.U) print(re.findall(pattern, str)) ['a', 'b', 'c', '我', '是', '中', '國', '人'] # 貪婪與懶惰 str = 'abdcdefg' pattern = re.compile(r'\w+') print(re.findall(pattern, str)) ['abdcdefg'] # 懶惰模式 pattern = re.compile(r'\w+?') print(re.findall(pattern, str)) ['a', 'b', 'd', 'c', 'd', 'e', 'f', 'g']
python 判斷文件夾內文件數量以及文件夾數量
path = 'C:/Users/Dell/Desktop/test' print('dirnum:',len([lists for lists in os.listdir(path) if os.path.isdir(os.path.join(path, lists))])) print('filenum:',len([lists for lists in os.listdir(path) if os.path.isfile(os.path.join(path, lists))]))
python 異常處理
自己常用的方法
#!/usr/bin/env python # -*- coding: UTF-8 -*- import paramiko import time hostname='10.x.x.122' username='username' password='password' port = 22 try: t = paramiko.Transport((hostname, port)) t.connect(username=username, password=password) sftp = paramiko.SFTPClient.from_transport(t) dest_filename = "kams.war" + "." + time.strftime('%Y-%m-%d') src = '/update/kams.war' dest = '/kams/war120/' + dest_filename sftp.put(src, dest) # 上傳文件 print('kams.war is transfered to 122 successfully.') t.close(); except Exception, e: print str(e)
其他方法
假如我們知道異常可能是NameError或者TypeError，可以通過下面兩種方法來保證能夠捕獲： ----方法一--- # /usr/bin/env python try: a*2 except NameError as err: print(err) except TypeError as err: print(err) print('hello,world') ----方法二---- # /usr/bin/env python try: a*2 except (NameError, TypeError) as err: print(err) print('hello,world') 如果我們連具體的異常類型都不知道呢？這個時候，就可以使用Python中的異常基類Exception來進行捕獲： ----方法三--- # /usr/bin/env python try: a*2 except Exception as err: print(err) ----方法四----（缺省表示Exception） # /usr/bin/env python try: a*2 except: print('err') 需要注意：當一個程序中try后面跟有多個exception的時候，會匹配第一個可以匹配的異常類型
04 容易忽略的finally
import sys try: f = open('aaa.txt', 'r') # some data processing except OSError as err: print('OS error: {}'.format(err)) except: print('Unexpected error:', sys.exc_info()[0]) finally: f.close()
異常處理的使用需要根據場景來確定，不能不用，不用的話代碼穩定性不高；也不能濫用，濫用的話，會顯得代碼很冗余。
例如下面的2個小場景：
----場景一---- try: data = json.loads(raw_data) except JSONDecodeError as err: print('JSONDecodeError: {}'.format(err)) 解析json前，對json進行合法性判斷是有必要的， 否則json的解析就會報錯。 ---場景二---- d = {'name': 'jason', 'age': 20} try: value = d['dob'] except KeyError as err: print('KeyError: {}'.format(err)) 字典的key值解析這種寫法也可以，但是顯得有點繁瑣 最好的寫法是： if 'dob' in d: xxxxx
print輸出結果的三種方法
格式化字符串： print(“hello %s”%name) 保留2位小數點： calc = 100 / 88 print('計算結果：%.2f' %calc) 1）數據庫使用 print("姓名：%s, 年齡：%d, 身高: %.2f" % (name, age, height)) 2）SQL使用不安全 print("姓名：{}, 年齡：{}, 身高: {}".format(name, age, height)) 3）下面這倆種是相同的，形式不同 print(f"姓名：{name}, 年齡：{age}, 身高: {height}") 字符串格式化輸出 name = “aliang” age = 30 # 方法1 print("我的名字是%s,今年%s歲了。" % (name, age)) # 方法2 print(f"我的名字是{name},今年{age}歲了。")
字符串拼接 使用“+”可以對多個字符串進行拼接。 示例： str1 = "hello" str2 = “world" print(str1 + str2) 其他方式： ?格式化輸出 ?逗號 ?join() 字符串的內置函數，將序列中的元素拼接成一個字符串。后面講到
浙公網安備 33010602011771號