Python选择指定文件夹的文件然后复制出其中几个文件到新的文件夹

"""
要求：
1.读取cdm文件的所有子文件夹，然后每个文件夹里面是抽出一个一个mp3后缀的文件。
遍历所有的子文件，然后将这些mp3文件，保存到一个新的文件夹。文件夹后缀是当前文件的年代的
基础加后缀_single,例如：来自2000年文件夹就命名2000_single。
"""
import shutil
import pathlib
import pprint
from typing import List, Generator, Iterable, Tuple
from itertools import chain, islice
import os
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
from functools import partial
import traceback
import time
from multiprocessing import cpu_count

DIR_PATH = pathlib.Path("/Users/chennan/CDM")
MAX_CONCCURENT = cpu_count() * 4 - 2


def multiple_file_types(file_path: Generator, *patterns: Tuple[str])         -> Iterable:
    """
    因为glob不支持多个匹配规则，所以这里做一个适配。
    :param file_path: 文件的路径
    :param patterns: 匹配规则
    :return:
    """
    return (file_path.glob(pattern) for pattern in patterns)


def get_all_list_by_year(y: str) -> Generator:
    """
    :param y: 年代
    :return: 返回结果迭代器类型
    """
    file_path = DIR_PATH / y
    result = file_path.glob("*")
    return result


def get_mp3_list(cdm_path: str) -> Generator:
    """
    获取当前文件下面的所有音频文件，mp3,wav,ape,flac等
    :param cdm_path:专辑的路径
    :return:
    """
    mp3_gen = multiple_file_types(cdm_path, *("*.mp3", "*.ape", "*.wav", "*.flac"))
    return mp3_gen


def select_mp3(y: str) -> List[pathlib.Path]:
    """
    :param y:
    :return:
    """
    selected_mp3_list = []
    sa = selected_mp3_list.append
    cdm_list = get_all_list_by_year(y)
    for dir_path in cdm_list:
        mp3_gen = get_mp3_list(dir_path)
        try:
            single = mp3_gen.send(None)
            sa(list(single)[0])
        except StopIteration as e:
            pass
        except IndexError as e:
            pass
    return selected_mp3_list


def create_target_file(y):
    target = DIR_PATH / f"{y}_new"
    if not os.path.exists(target):
        os.makedirs(target)
    return target


def copy_file_to_new_path(source, y, target):
    """
    shutil复制文件
    :return:
    """
    try:
        shutil.copy(source, target / source.name)
    except Exception as e:
        print(traceback.format_exc())


if __name__ == '__main__':
    start = time.time()
    year = "2000"
    mp3_list = select_mp3(year)
    target = create_target_file(year)
    if target:
        with ThreadPoolExecutor(max_workers=MAX_CONCCURENT) as pool:
            pool.map(partial(copy_file_to_new_path, y=year, target=target), mp3_list)
    print(f"并发{MAX_CONCCURENT}次，用时", time.time() - start)


#关于并发的问题有待观察。。。
# 线程池并发结果
# 并发2次，用时 6.518146991729736
# 并发10次，用时 7.732945919036865
# 并发30次，用时 7.956831932067871
# 并发100次，用时 7.956831932067871

# 进程池并发结果
# 并发2次，用时 6.013077974319458
# 并发8次，用时 6.863225221633911
# 并发14次，用时 4.6188788414001465
# 并发30次，用时 4.858534812927246

Python选择指定文件夹的文件然后复制出其中几个文件到新的文件夹

原文：https://www.cnblogs.com/c-x-a/p/10977025.html