Python练习1

一、linux，基于文件大小，创建时间，修改时间，文件内容，文件名称等进行查找汇总和输出

2019-01-04

只操作文本文件

#!/usr/bin/env python
# -*- coding: utf-8 -*
# Created by YangYongming at 2018/12/10 17:16
# FileName: main.py

import os
import time
import zipfile

fileslist = []  # 文件列表
if not os.path.exists("temporary"):
    os.popen("mkdir temporary &> /dev/null")


def zip_dir(dirname, zipfilename):
    filelist = []
    if os.path.isfile(dirname):
        filelist.append(dirname)
    else:
        for root, dirs, files in os.walk(dirname):
            for name in files:
                filelist.append(os.path.join(root, name))

    zf = zipfile.ZipFile(zipfilename, "w", zipfile.zlib.DEFLATED)
    for tar in filelist:
        arcname = tar[len(dirname):]
        zf.write(tar, arcname)
    zf.close()


def Get_Cur_time():
    """
    # 获取当前日期和时间
    :return:
    """
    return time.strftime("%Y-%m-%d-%H-%M-%S")


def Judge_filetype(filename):
    """
    # 判断一个文件是都是文本文件
    :param filename: 文件名称
    :return:
    """
    res = os.popen("file %s" % filename)
    if "text" in res.read():
        return True
    else:
        return False


def Get_FilesList(path):
    """
    # 获取所有文件列表
    :param path: 文件夹目录
    :return: 所有文件绝对路径组成的列表
    """
    global fileslist
    for root, dir, files in os.walk(path):
        for i in files:
            file = os.path.join(root,i)  # 拿到每个文件的绝对路径
            if os.path.isfile(file):  # 判断一下是否是文件
                if Judge_filetype(file):  # 判断是否是文本文件
                    if os.access(file, os.R_OK):  # 判断文件是都可读
                        fileslist.append(file)  # 添加到文件列表中
    return fileslist


def Content_Keyword(keyword):
    """
    # 基于文件内容关键字查询
    :param keyword: 关键字
    :return:
    """
    # 创建存储目录
    global number
    dir = "%s[Content:%s]" % (curtime, keyword)
    os.popen("cd temporary && mkdir %s" % dir)
    for i in fileslist:
        # 基于linux命令拿到匹配到文本的行号
        result_obj_line = os.popen("grep -n %s %s | cut -d: -f1" % (keyword, i))
        # 读取行号字符串
        result_line_seq = result_obj_line.read()
        # 把行号转换成序列,并且将序列内容连接到一起
        result_line = ''.join(result_line_seq.split())
        # 判断是否都为数字,如果不是全数字，说明匹配到的文本可能是二进制文件或者其他格式文件
        if result_line.isdigit():
            # 输出文件名称和具体的行号,使用split()把一个文件中匹配到的所有行放在一个数组中,防止字符串错乱显示
            print("33[0;32;40m%s  Line:%s33[0m" % (i, result_line_seq.split()))
            number += 1
            # 匹配到的文件-p拷贝到指定文件夹
            os.popen("cd temporary && cp -p %s %s" % (i, dir))
            time.sleep(0.03)

    return dir, number


def Name_Type_Keyword(keyword):
    """
    # 基于文件名称查找
    :param keyword: 关键字
    :return:
    """
    # 创建存储目录
    global number
    dir = "%s[NameType:%s]" % (curtime, keyword)
    os.popen("cd temporary && mkdir %s" % dir)
    for i in fileslist:
        # 拿到文件的基名
        basename = os.path.basename(i)
        # 在基名中查找keyword
        result_basename_obj = os.popen("echo %s | grep %s" % (basename, keyword))
        # 读取查找返回值
        result_basename = result_basename_obj.read()
        # 判断是否有匹配到的内容
        if result_basename.strip():
            # 输出文件名
            print("33[0;32;40m%s33[0m" % i)
            # copy到指定的目录
            os.popen("cd temporary && cp -p %s %s" % (i, dir))
            number += 1
            time.sleep(0.03)
    return dir, number


def Time_Keyword(start_time, end_time, timetype):
    """
    # 按照文件创建时间查询
    :param start_time: 起始时间:20181230103050
    :param end_time: 截止时间
    :timetype: ctime文件创建时间,mtime:文件修改时间
    :return:
    """
    global number
    # 创建存储目录
    dir = "%s[%s:%s-%s]" % (curtime, timetype, start_time.strip(), end_time.strip())
    os.popen("cd temporary && mkdir %s" % dir)
    # 拿到起始时间的时间数组
    start_timeArray = time.strptime(start_time, "%Y%m%d%H%M%S")
    end_timeArray = time.strptime(end_time, "%Y%m%d%H%M%S")
    # 拿到截止时间的时间戳
    strt_timeStamp = int(time.mktime(start_timeArray))
    end_timeStamp = int(time.mktime(end_timeArray))
    for i in fileslist:
        # 拿到文件时间戳
        if timetype == "ctime":
            time_Stamp = os.path.getctime(i)
        elif timetype == "mtime":
            time_Stamp = os.path.getmtime(i)
        # 文件创建时间的时间戳在起止时间的时间戳内则匹配
        if strt_timeStamp <= int(time_Stamp) <= end_timeStamp:
            # 拿到文件创建时间数组
            ctimeArray = time.localtime(time_Stamp)
            # 按照指定的时间格式 拿到文件创建时间
            otherStyleTime = time.strftime("%Y-%m-%d %H:%M:%S", ctimeArray)
            # 输出匹配的文件名和创建时间
            print("33[0;32;40m%s  Time:%s33[0m" % (i, otherStyleTime))
            # copy到指定的目录
            os.popen("cd temporary && cp -p %s %s" % (i, dir))
            number += 1
            time.sleep(0.03)
    return dir, number


def Size_Keyword(ge_size, le_size):
    """
    # 按照文件大小查找
    :param ge_size: 文件大小范围最小值[KB]
    :param le_size: 文件大小范围最大值[KB]
    :return:
    """
    # 创建存储目录
    global number
    dir = "%s[Size:%sKB-%sKB]" % (curtime, ge_size, le_size)
    os.popen("cd temporary && mkdir %s" % dir)
    for i in fileslist:
        size_B = os.path.getsize(i)
        size_KB = size_B / float(1024)
        if float(ge_size) <= size_KB <= float(le_size):
            # 输出文件名和文件大小
            print("33[0;32;40m%s  Size:%.2f KB33[0m" % (i, size_KB))
            # copy到指定目录
            os.popen("cd temporary && cp -p %s %s" % (i, dir))
            number += 1
            time.sleep(0.03)
    return dir, number


if __name__ == '__main__':
    number = 0
    curtime = Get_Cur_time()
    os.popen("clear")
    print("33[1;31;40m
Note: Only text files can be manipulated
33[0m")
    path = input("33[0;36;40mplease input absolute path:33[0m")
    if not os.path.exists(path):
        print("33[0;36;40m%s:Path does not exist!33[0m" % path)
        exit(10)
    res_fileslist = Get_FilesList(os.path.abspath(path))
    print("33[0;36;40mNumber of documents found: %s" % len(fileslist))
    msg = "1.Search based on file content
" 
          "2.Search based on file name or type
" 
          "3.Search based on file creation time
" 
          "4.Search based on file modification time
" 
          "5.Search based on file size
"
    num = input("33[0;36;40m%splease choose:33[0m" % msg)
    if num == "1":
        keyword = input("33[0;36;40mPlease enter keyword:33[0m")
        back = Content_Keyword(keyword.strip())
        zip_dir("temporary/%s" % back[0], "temporary/%s.zip" % back[0])


    elif num == "2":
        keyword = input("33[0;36;40mPlease enter name keyword or type:33[0m")
        back = Name_Type_Keyword(keyword.strip())
        zip_dir("temporary/%s" % back[0], "temporary/%s.zip" % back[0])


    elif num == "3":
        start_time = input("33[0;36;40mPlease enter a start time:33[0m")
        end_time = input("33[0;36;40mPlease enter a end time:33[0m")
        if bool(end_time) == False:
            end_time = time.strftime("%Y%m%d%H%M%S", time.localtime())
        if start_time.isdigit() and end_time.isdigit() and len(start_time) == 14 and len(end_time) == 14:
            back = Time_Keyword(start_time, end_time, timetype="ctime")
            zip_dir("temporary/%s" % back[0], "temporary/%s.zip" % back[0])
        else:
            print("33[0;31;40mInput error!33[0m")
            exit(3)


    elif num == "4":
        start_time = input("33[0;36;40mPlease enter a start time:33[0m")
        end_time = input("33[0;36;40mPlease enter a end time:33[0m")
        if bool(end_time) == False:
            end_time = time.strftime("%Y%m%d%H%M%S", time.localtime())
        if start_time.isdigit() and end_time.isdigit() and len(start_time) == 14 and len(end_time) == 14:
            back = Time_Keyword(start_time, end_time, timetype="mtime")
            zip_dir("temporary/%s" % back[0], "temporary/%s.zip" % back[0])
        else:
            print("33[0;31;40mInput error!33[0m")
            exit(4)


    elif num == "5":
        ge_size = input("33[0;36;40mPlease enter a minimum file size[KB]:33[0m")
        le_size = input("33[0;36;40mPlease enter a maximum file size[KB]:33[0m")
        if ge_size.isdigit() and le_size.isdigit():
            back = Size_Keyword(ge_size, le_size)
            zip_dir("temporary/%s" % back[0], "temporary/%s.zip" % back[0])
        else:
            print("33[0;31;40mInput error!33[0m")
            exit(5)

    else:
        print("33[0;31;40mInput error!33[0m")
        exit(1)

    try:
        print("33[0;36;40mTotal number of documents: %s33[0m" % back[1])
    except:
        pass

运行截图：

二、windows批量pingIP地址/地址段，

sfle:每行一个IP或者网段 

upip：存放up的ip

downip: 存放down的ip

#!/usr/bin/env python
# -*- coding: utf-8 -*
# Created by YangYongming at 2018/11/25 12:26
# FileName: ping.py

import os
import IPy
import queue
import threading
import time


class MyThread(threading.Thread):
    def __init__(self, queue):
        threading.Thread.__init__(self)
        self.queue = queue

    def run(self):  # 定义每个线程要运行的函数
        global aliveip
        global downip
        while True:
            host = self.queue.get(timeout=2)  # 从队列中取Ip
            host = str(host)
            res = os.popen("ping -n 3 %s" % host)
            os.popen("exit
")
            if "ms" not in res.read():
                print(">>>   %s is Down" % host)
                downip.append(host)
            else:
                print(">>>   %s is UP" % host)
                aliveip.append(host)
            self.queue.task_done()
            if self.queue.empty():  # 当队列为空时，终止该线程
                break


if __name__ == '__main__':
    start = time.time()  # 程序开始时间
    print("
   Yongming working for you......
")
    print("
   start......
")
    with open("upip", "w", encoding="utf-8") as tmp1, open("downip", "w", encoding="utf-8") as tmp2:
        pass  # 清空文件
    threadlist = []  # 线程列表
    queue = queue.Queue()  # 队列
    aliveip = []  # UP IP列表
    downip = []  # DOWN IP列表
    num = 0  # 线程数量
    with open("sfile", "r", encoding="utf-8") as f1:
        for line in f1.readlines():  # 读取IP文件
            ip = IPy.IP(line)  # 获取每个网段的IP地址列表
            for x in ip:
                num += 1  # 计算有多少个IP地址，决定开多少线程
                queue.put(x)  # 向队列里面放IP
    if num >= 100:
        num = 100  # 设置线程数量不超过100
    for i in range(num):
        t = MyThread(queue)  # 建立线程，传入队列
        t.setDaemon(False)  # 主线程执行完成，等待所有的前台线程执行完毕，默认[等待]False
        t.start()  # 线程准备就绪，等待CPU调度

    queue.join()  # 队列为空在进行以下操作

    with open("downip", "a", encoding="utf-8") as f1:
        for i in downip:
            f1.write(i + '
')  # DOWN的IP写入文件
    with open("upip", "a", encoding="utf-8") as f2:
        for i in aliveip:
            f2.write(i + '
')  # UP的IP写入文件

    end = time.time()  # 结束时间
    elapsed = end - start  # 计算总耗时
    print("
UP:%s" % len(aliveip))  # 输出UP的IP数量
    print("DOWN:%s" % len(downip))  # 输出DOWN的IP数量
    print("
Time taken: %d seconds
" % elapsed)  # 输出总耗时
    input("
Enter to Quit:")
    time.sleep(30)  # 等待10s退出

 运行截图：

 三，比较AB两个IP列表的不同

Afile：存储AIP列表

Bfile：存储BIP列表

#!/usr/bin/env python
# -*- coding: utf-8 -*
# Created by YangYongming at 2018/12/04 14:53
# FileName: ipCompar.py


def Check_Ip(filename):
    with open(filename, "r", encoding="utf-8") as file1:
        for i in file1.readlines():
            try:
                IP(i.strip())
            except Exception:
                if i.strip():
                    print("[Err]IP address format: %s in %s" % (i.strip(), filename))
                else:
                    print("[Err]There are blank lines: in %s" % filename)
                exit(100)


def Contrast_Ip(Afile, Bfile):
    excle = xlwt.Workbook(encoding="utf-8")
    sheet = excle.add_sheet('sheet 1')

    style0 = xlwt.XFStyle()
    al = xlwt.Alignment()
    al.horz = 0x02
    al.vert = 0x01
    style0.alignment = al
    pattern = xlwt.Pattern()
    pattern.pattern = xlwt.Pattern.SOLID_PATTERN
    pattern.pattern_fore_colour = xlwt.Style.colour_map['gray40']
    style0.pattern = pattern

    style1 = xlwt.XFStyle()
    al = xlwt.Alignment()
    al.horz = 0x02
    al.vert = 0x01
    style1.alignment = al

    for i in range(0, 3):
        col = sheet.col(i)
        col.width = 256 * 20

    sheet.write(0, 0, 'Only_IN_%s' % Afile, style0)
    sheet.write(0, 1, 'Only_IN_%s' % Bfile, style0)
    sheet.write(0, 2, 'Both', style0)

    with open(Afile, "r+") as file1:
        A_ip = file1.readlines()
        for i in range(0, len(A_ip)):
            A_ip[i] = A_ip[i].strip()
    A_IPSET = set(A_ip)

    with open(Bfile, "r+") as file2:
        B_ip = file2.readlines()
        for i in range(0, len(B_ip)):
            B_ip[i] = B_ip[i].strip()
    B_IPSET = set(B_ip)

    for i, k in enumerate(A_IPSET.difference(B_IPSET), 1):
        sheet.write(i, 0, str(k), style1)

    for i, k in enumerate(B_IPSET.difference(A_IPSET), 1):
        sheet.write(i, 1, str(k), style1)

    for i, k in enumerate(B_IPSET.intersection(A_IPSET), 1):
        sheet.write(i, 2, str(k), style1)

    excle.save("Ming[%s].xls" % Time)

    return True


if __name__ == '__main__':
    import xlwt
    import time
    from IPy import IP

    Time = time.strftime("%H-%M-%S")
    print("
Yongming working for you......
")
    time.sleep(3)

    Check_Ip(filename="Afile")
    Check_Ip(filename="Bfile")

    if Contrast_Ip("Afile", "Bfile"):
        print("Success, please check Ming[%s.xls]" % Time)
        time.sleep(0.5)
    else:
        print("[Err] in Contrast_Ip()")

    input("
Enter to Quit:")

    print("Quit after 3 seconds......")

    time.sleep(3)

 运行结果：会在当前目录生成一个excle文件，如下：

四：我的方法

#!/usr/bin/env python
# -*- coding: utf-8 -*
# Created by YangYongming at 2018/11/19 17:46
# FileName: ming.py
import os
import sys
import zipfile
import threading
import queue
import IPy


def Function_Validation(origin_func):
    """
    对类的方法进行验证
    :param origin_func:源函数名称
    :return:源函数的返回值
    """

    def wrapper(self, *args, **kwargs):
        """
        :param self: 可以直接调用类中的字段和方法
        :param args: 参数
        :param kwargs: 参数
        :return:
        """
        import hashlib
        md5 = hashlib.md5()
        # 对类识别码取MD5加密数据
        md5.update(bytes(str(self.password), encoding="utf-8"))
        if md5.hexdigest() != yangym.PASS:
            print("Error yangym()类识别码错误")
            exit(100)
        try:
            # 执行源函数
            u = origin_func(self, *args, **kwargs)
            # 返回原函数的返回值给外层函数
            return u
        except Exception:
            # self.revive() #不用顾虑，直接调用原来的类的方法
            return 'origin_func an Exception raised.'

    return wrapper


class yangym():
    # 类入口验证码，实例化yangym时，必须携带PASS[password]字段，只有匹配才能执行类中的方法

    PASS = "05a7319bcb20e06fa52a3dc3685f5f84"

    def __init__(self, password):
        # password 实例化类的验证码。
        self.password = password

    @Function_Validation
    def Del_Blank_Line(self, filename):

        """
        清除文件空白行空白行
        :param filename: 文件名称
        :return: True 成功；False 失败
        """
        try:
            with open(filename, "r+", encoding="utf-8") as infp:
                lines = infp.readlines()  # 把源文件内容读出来保存在lines中
            with open(filename, "w+", encoding="utf-8") as outfp:
                for li in lines:
                    if li.split():  # 判断是否为空白行
                        outfp.writelines(li)  # 将操作后的源文件覆盖写回
        except FileNotFoundError as e:
            print("[Err] No such file or directory: %s " % filename)
            return False
        except IOError:
            print("[Err] Permission deny: %s" % filename)
            return False
        except Exception as e:
            print("[Err:Del_Blank_Line()] %s" % e)
            return False
        else:
            return True

    @Function_Validation
    def Get_IpList(self, filename, repeat=False):
        """
        在文件中获取合法的IP地址
        :param filename: 文件名称
        :param repeat: 去除重复行，True去除，False不去除
        :return: 返回ip地址序列
        """
        import re
        try:
            with open(filename, "r", encoding="utf-8") as file1:
                line = file1.read()
                line = re.sub(r"[!@#$%^&*-+_~?/|\]", " ", line)
                pattern = re.compile(
                    r"(?:(?:d{1,2}|1d{2}|2[0-4]d|25[0-5]).){3}(?:d{1,2}|1d{2}|2[0-4]d|25[0-5])")
                list_ip = pattern.findall(line)
                if len(list_ip) == 0:
                    return list_ip
        except FileNotFoundError:
            print("[Err] No such file or directory: %s " % filename)
            return False
        except IOError:
            print("[Err] Permission deny: %s" % filename)
            return False
        except Exception as e:
            print("[Err:Get_IpList()] %s" % e)
            return False
        else:
            if repeat == True:
                return set(list_ip)
            elif repeat == False:
                return list_ip

    @Function_Validation
    def Get_CheckCode(self, n=6):
        """
        获取有大小写字母、数字组成的随机n位验证码
        :param num: 验证码位数，默认为6
        :return: 返回n位验证码
        """
        import random
        check_code = str()
        code = str()
        for i in range(n):
            ret = random.randint(0, 9)
            if ret == 0 or ret == 1 or ret == 4 or ret == 7:
                code = str(ret)
            elif ret == 2 or ret == 5 or ret == 8:
                code = chr(random.randint(65, 90))
            elif ret == 3 or ret == 6 or ret == 9:
                code = chr(random.randint(97, 122))
            check_code = check_code + code
        return check_code

    @Function_Validation
    def SendMail(self, SenderName, SenderMail, Password, ReceList, Theme, Text, CcList=[], Appendix=None, Port=25, ):
        """
        发送邮件
        :param SenderName: 发送者昵称
        :param SenderMail: 发送者邮箱
        :param Password: 邮箱密码/授权吗
        :param ReceList: 收件人列表
        :param Theme: 邮件主题
        :param Text: 邮件正文
        :param CcList: 抄送人列表[可选参数]
        :param Appendix: 附件全路径,不可以使用中文[可选参数]
        :param Port: 端口,[可选参数]，默认"25"
        :return True:发送成功，False:发送失败
        """
        ret = True
        try:
            import smtplib, os
            from email.mime.text import MIMEText
            from email.utils import formataddr
            from email.mime.multipart import MIMEMultipart
            from email.mime.application import MIMEApplication

            msg = MIMEMultipart('alternative')

            msgText = MIMEText(Text, "Plain", "utf-8")

            msg["from"] = formataddr([SenderName, SenderMail])
            msg["To"] = ",".join(ReceList)
            msg["Cc"] = ",".join(CcList)
            msg["Subject"] = Theme

            if Appendix != None:
                attachName = os.path.basename(Appendix)
                part = MIMEApplication(open(Appendix, 'rb').read())
                part.add_header('Content-Disposition', 'attachment', filename=attachName)
                msg.attach(part)

            msg.attach(msgText)

            if Port == 25:
                server = smtplib.SMTP("smtp.163.com", 25)
            elif Port == 465:
                server = smtplib.SMTP_SSL("smtp.163.com", 465)

            server.login(SenderMail, Password)
            server.sendmail(SenderMail, ReceList, msg.as_string())
            server.quit()
        except ModuleNotFoundError as e1:
            print("[Err] %s" % e1)
            ret = False
        except Exception as e2:
            print("[Err:SendMail()] %s" % e2)
        return ret

    @Function_Validation
    def Copy_AToB_File(self, fileA, fileB):
        """
        把文件A的内容追加到文件B中
        :param fileA: 源文件
        :param fileB: 目标文件
        :return: 成功返回True, 失败返回False
        """
        try:
            with open(fileA, "r", encoding="utf-8") as A, open(fileB, "r+", encoding="utf-8") as B:
                Afile = A.readlines()
                B.seek(0, 2)
                for i in Afile:
                    B.writelines(i.strip() + "
")
        except FileNotFoundError as e:
            print("[Err] %s" % e)
            return False
        else:
            return True

    @Function_Validation
    def IpCount(self, ipfile):
        """
        统计每个IP地址出现的次数
        :param ipfile: 包含纯IP地址的文件，每行一个，不要出现空白行
        :return: 返回一个字典{IP地址：次数，}
        """
        sip = []
        dic = {}
        try:
            with open(ipfile, "r+", encoding="utf-8") as f1:
                for i in f1.readlines():
                    sip.append(i.strip())
        except FileNotFoundError:
            print("[Err] No such file or directory: %s " % ipfile)
            return False
        else:
            setip = set(sip)
            for i in setip:
                num = sip.count(i)
                dic[i] = num
            return dic

    @Function_Validation
    def IpQuery(self, ip):
        """
        :param ip: IP地址
        :return: 执行成功返回归属地址信息，执行失败返回"False"
        """
        import requests
        from bs4 import BeautifulSoup
        url = 'http://m.ip138.com/ip.asp?ip='
        kv = {'User-Agent': 'Mozilla/5.0'}
        link = url + str(ip)
        try:
            r = requests.get(link, headers=kv)
            r.raise_for_status()
            r.encoding = r.apparent_encoding
            soup = BeautifulSoup(r.text, 'lxml')
            result = soup.select('p[class="result"]')[0].string
            return result
        except requests.HTTPError:
            return False

    @Function_Validation
    def Get_FilesList(self, dirpath):
        fileslist = list()
        for root, dir, files in os.walk(dirpath):
            for i in files:
                file = os.path.join(root, i)  # 拿到每个文件的绝对路径
                if os.path.isfile(file):  # 判断一下是否是文件
                    fileslist.append(file)  # 添加到文件列表中
        return fileslist

    @Function_Validation
    def zip_dir(self, dirname, zipfilename):
        """
        :param dirname: 需要打包的文件或目录名称
        :param zipfilename: 目标文件名
        :return:
        """
        filelist = list()
        if os.path.isfile(dirname):
            filelist.append(dirname)
        else:
            for root, dirs, files in os.walk(dirname):
                for name in files:
                    filelist.append(os.path.join(root, name))

        zf = zipfile.ZipFile(zipfilename, "w", zipfile.zlib.DEFLATED)
        for tar in filelist:
            arcname = tar[len(dirname):]
            zf.write(tar, arcname)
        zf.close()