Python常用功能函数系列

Posted 2020-12-14 超级英雄拯救世界之前成长的日子

本节目录

• 常用函数一：获取指定文件夹内所有文件

• 常用函数二：文件合并

• 常用函数三：将文件按时间划分

• 常用函数四：数据去重

写在前面

写代码也有很长时间了，总觉得应该做点什么有价值的事情，写代码初始阶段觉得做更多的项目，积累更多的经验是自己应该做的事情，这样可以使自己短时间内技术水平获得较大的提升。随着代码量和项目的增加，确实体会到了自身水平的进步，但同时由原来的尽可能多的做项目，学知识，逐渐转变为了尽可能精尽可能专的投入，更准确的说是当有了一定的知识基础和技术积累，追求不仅仅是知识和项目的广度，更注重的是自身的深度，学确实自己现在或以后有可能会用到的技术和知识，做自己认为有价值的项目，我想这就是我想表达的。然而，除了不断的学习，不断的前进，另外一点就是需要经常做回顾和记录，我想这就是博客和代码仓库的存在的价值，博客可以存储我们学过的知识，仓库可以存储我们写过的代码。有了这些东西，我们实现了随时随地可以找到我们自己保存的东西。我目前的困扰就是自己积累了大量的经验和代码，但是还没有形成一个系统的体系，很多重复性的工作会经常做，虽然实现过一次之后使用的时候会节省不少时间和精力，但是有时候由于=一些原因，会把时间浪费在找之前写的代码过程中，尤其代码越来越多之后，没有一个很好的存储目录结构，实际上是效率极低的，毕竟自己造的轮子或者搜集的轮子再次使用的时候因为找不到轮子在哪或者速度很慢不是我们追求的结果，这个Python常用函数系列博客就应运而生。

常用函数一：获取指定文件夹内所有文件

• 方式一：采用os.listdur方式

def print_directory_contents(dir_path, file_list):
    """
    这个函数接收文件夹的名称作为输入参数
    返回该文件夹中文件的路径
    以及其包含文件夹中文件的路径
    """
    import os
    for file in os.listdir(dir_path):
        file_path = os.path.join(dir_path, file)
        if os.path.isdir(file_path):
            print_directory_contents(file_path, file_list)
        else:
            file_list.append(file_path)


if __name__ == \'__main__\':
    file_list = []
    print_directory_contents(\'G:/programming/interview_question\', file_list)
    print(file_list)

• 方式二：采用os.walk方式

def print_directory_contents(dir_path):
    """
    这个函数接收文件夹的名称作为输入参数
    返回该文件夹中文件的路径
    以及其包含文件夹中文件的路径
    """
    import os
    for base_path, folders, files in os.walk(dir_path):
        for file in files:
            file_path = os.path.join(base_path, file)
            yield file_path


if __name__ == \'__main__\':
    file_list = print_directory_contents(\'G:/programming/interview_question\')
    for file in file_list:
        print(file)

os.listdir和os.walk各适合什么场景？

经验分享：两种方式均能实现想要的功能，若读取单层目录，选择os.listdir；若读取多层目录，选择os.walk

常用函数二：文件合并

文件合并是数据处理中会经常用到，按合并类型划分为横向合并和纵向合并，此功能在pandas中有很好的实现，包含纵向合并append，横向合并concat和join，横纵向合并concat。下面的代码是我基于这四个函数编写的合并文件通用代码，基本上这套代码可以实现你所有的合并文件的需求。

 

# -*- coding: utf-8 -*-

"""
Datetime: 2020/07/05
Author: Zhang Yafei
Description: 合并文件
"""
from pandas import read_csv, read_excel, merge, concat, DataFrame


def read_file(file_path, on):
    if file_path.endswith(\'.csv\'):
        return read_csv(file_path)
    if file_path.endswith(\'.xls\') or file_path.endswith(\'xlsx\'):
        return read_excel(file_path)


def df_to_file(df: DataFrame, file_path: str, index: bool = True, encoding: str = \'utf_8_sig\'):
    if file_path.endswith(\'.csv\'):
        df.to_csv(file_path, index=index, encoding=encoding)
    if file_path.endswith(\'.xls\') or file_path.endswith(\'xlsx\'):
        df.to_excel(file_path, index=index)


def merge_two_data(file1: str, file2: str, on: str = None, left_on: str = None, right_on: str = None,
                   how: str = \'inner\', to_file: str = None):
    """
    横向合并两个文件
    @param file1:
    @param file2:
    @param on:
    @param left_on:
    @param right_on:
    @param how:
    @param to_file:
    @return:
    """
    df1 = read_file(file1)
    df2 = read_file(file2)
    merge_df = merge(df1, df2, on=on, how=how, left_on=left_on, right_on=right_on)
    if to_file:
        if to_file.endswith(\'.csv\'):
            merge_df.to_csv(to_file, encoding=\'utf_8_sig\', index=False)
        elif to_file.endswith(\'xls\') or to_file.endswith(\'xlsx\'):
            merge_df.to_excel(to_file, index=False)
    else:
        return merge_df


def append_two_file(file1: str, file2: str, ignore_index: bool = True, to_file: str = None):
    """
    纵向合并两个文件
    @param file1:
    @param file2:
    @param to_file:
    @return:
    """
    df1 = read_file(file1)
    df2 = read_file(file2)
    df3 = df1.append(df2, ignore_index=ignore_index)
    if to_file:
        df_to_file(df3, to_file, index=False)
    else:
        return df3


def join_two_file(file1: str, file2: str, on=None, left_on=None, right_on=None, how: str = \'left\', to_file: str = None):
    """
    横向联接两个表格文件
    @param file1:
    @param file2:
    @param on:
    @param left_on:
    @param right_on:
    @param how:
    @param to_file:
    @return: None or DataFrame
    """
    if on:
        df1 = read_file(file1).set_index(keys=on)
        df2 = read_file(file2).set_index(keys=on)
    elif left_on or right_on:
        df1 = read_file(file1).set_index(keys=on) if left_on else read_file(file1)
        df2 = read_file(file2).set_index(keys=on) if right_on else read_file(file2)
    else:
        df1 = read_file(file1)
        df2 = read_file(file2)
    df3 = df1.join(df2, how=how)
    if to_file:
        df_to_file(df3, to_file, index=False)
    else:
        return df3


def concat_mul_file(axis: int = 0, on=None, to_file: str = None, encoding: str = \'utf_8_sig\', *files):
    """
    多个表格文件合并
    @param axis: 0/index 1/column 若axis=1, 默认基于索引将多个文件合并
    @param on: 当axis=1时，指定索引列/索引列
    @param to_file: 导出文件路径
    @param encoding: 导出文件编码
    @param files: 合并文件路径
    @return:
    """
    if len(files) > 1:
        if axis == 1 and on:
            objs = [read_file(file).set_index(keys=on) for file in files]
        else:
            objs = [read_file(file) for file in files]
        merge_data = concat(objs=objs, axis=axis)
        if to_file:
            df_to_file(merge_data, to_file, index=False, encoding=encoding)
        else:
            return merge_data
    else:
        raise Exception(\'合并的文件个数小于2，不能进行合并，请输入大于等于两个文件路径\')

 

　经验分享：若为两个文件合并，横向合并可以选择merge，join,concat，纵向合并可以选择append,concat，若为多个文件合并（大于2）,只能选择concat。

常用函数三：将文件按时间划分

    一个文件若有一列是时间类型，那么处理此类数据时经常会有一个需求就是按照时间点划分此文件为多个文件，从实现上来说，读取文件之后，把时间那一列转换为时间类型，然后可以利用时间类型可以直接比较的，那么这样就可以实现划分的目的。以下是具体实现通用代码。

# -*- coding: utf-8 -*-

"""
Datetime: 2020/06/25
Author: Zhang Yafei
Description: 将文件按时间段划分
"""
import pandas as pd
from datetime import datetime


def data_part(file_path, col, time_list):
    df = pd.read_excel(file_path)
    df[col] = pd.to_datetime(df[col])
    for t in range(len(time_list)):
        if t != len(time_list) - 1:
            data = df[(df[col] >= time_list[t]) & (df[col] <= time_list[t+1])]
        else:
            data = df[(df[col] >= time_list[t])]
        data.to_excel(f"{file_path}_{t}.xlsx", index=False)


if __name__ == \'__main__\':
    time_list = [datetime(2020, 1, 21), datetime(2020, 1, 24), datetime(2020, 2, 3)]
    data_part(\'山西政策.xlsx\', col=\'发布时间\', time_list=time_list)

经验分享：此功能函数可以直接拿来用，但是简单的几行代码，对于其应用场景和使用熟练的运用可以帮助我们实现更复杂的功能。

常用函数四：数据去重

# -*- coding: utf-8 -*-

"""
Datetime: 2020/06/29
Author: Zhang Yafei
Description: 04_数据去重
"""
from pandas import read_csv, read_excel


def data_drop_duplicate(file: str, to_file: str, columns: list = None, keep: str = \'first\'):
    """
    :param file: 要去重的文件路径
    :param to_file: 去重之后保存的文件路径
    :param columns: 哪些列重复的去重
    :param keep: 重复的情况下，保留方式，默认 \'first\'
    """
    if file.endswith(\'csv\'):
        df = read_csv(file)
    else:
        df = read_excel(file)
    if columns:
        df.drop_duplicates(subset=columns, keep=keep, inplace=True)
    else:
        df.drop_duplicates(keep=keep, inplace=True)
    if to_file.endswith(\'csv\'):
        df.to_csv(to_file, index=False)
    elif to_file.endswith(\'xlsx\') or to_file.endswith(\'xls\'):
        df.to_excel(to_file, index=False)


if __name__ == \'__main__\':
    # 修改参数 file 文件名 columns 去重的列  to_file 去重之后新文件名
    data_drop_duplicate(file=\'data.xlsx\', columns=[\'id\', \'title\'], to_file=\'new_data.xlsx\')

    经验分享：这个功能Python实现起来是非常的方便，基本上关键功能只需一行代码，之前觉得这个功能很简单，而且我用的时候这只是其中很小的一个步骤，没必要单独记录下来。直到后来有同学经常单独那这个功能来问我，我就觉得有必要记录下来了，我相信这也是这个系列博客存在的意义，记录过往，记录曾经，不重复造轮子，但造出来的轮子随时可以找到拿来用，所以这系列的博客定位就是利用轮子仓库帮助理解实际应用的知识，从而更好的提高自己的逻辑能力，理解代码的能力和具体写代码的能力，不积跬步无以至千里，不积小流无以成江海，核心道理就是积少成多，直至某一天完成质变。