117 lines
4.0 KiB
Python
117 lines
4.0 KiB
Python
# copier.py
|
|
# Funções para copiar ficheiros para o destino organizado
|
|
from modules.organizer import (
|
|
obter_data_ficheiro,
|
|
criar_pasta_destino,
|
|
categorizar_por_tipo,
|
|
)
|
|
from pathlib import Path
|
|
import shutil
|
|
from tqdm import tqdm
|
|
from concurrent.futures import ThreadPoolExecutor, as_completed
|
|
from modules.logger import logger
|
|
from collections.abc import Callable
|
|
|
|
|
|
def copiar_ficheiro_individual(
|
|
ficheiro: str,
|
|
base_destino: str,
|
|
categorizar_data: bool = False,
|
|
usar_criacao: bool = False,
|
|
formato_data: str = "%Y-%m-%d",
|
|
) -> None:
|
|
"""copiar um ficheiro para o destino indicado,
|
|
com base nos parametros passados
|
|
|
|
argumentos:
|
|
ficheiro (str) -> caminho (como str) do ficheiro a ser copiado
|
|
base_destino (str) -> directório de destino para onde será copiado o ficheiro
|
|
categorizar_data (bool) -> flag para activar categorização por data (defeito: False)
|
|
usar_criacao (bool) -> usar data de criação (defeito: False usar data de modificacao)
|
|
formato_data (str) -> formato de data para nome dos subdirectórios (defeito: '%d-%m-%Y')
|
|
|
|
retorna:
|
|
None
|
|
"""
|
|
try:
|
|
ficheiro_path = Path(ficheiro)
|
|
if not ficheiro_path.exists():
|
|
logger.warning(
|
|
"Ficheiro '%s' não encontrado. A ignorar.", ficheiro)
|
|
return
|
|
if categorizar_data:
|
|
data_ficheiro = obter_data_ficheiro(
|
|
ficheiro, usar_criacao=usar_criacao)
|
|
pasta_destino = criar_pasta_destino(
|
|
str(base_destino),
|
|
categorizar_por_tipo(ficheiro),
|
|
data_ficheiro,
|
|
formato_data,
|
|
)
|
|
else:
|
|
pasta_destino = criar_pasta_destino(
|
|
str(base_destino), categorizar_por_tipo(ficheiro)
|
|
)
|
|
destino_ficheiro = pasta_destino / ficheiro_path.name
|
|
shutil.copy2(ficheiro_path, destino_ficheiro)
|
|
except Exception as e:
|
|
logger.error("Erro a copiar '%s': %s", Path(ficheiro).name, e)
|
|
|
|
|
|
def copiar_ficheiros_para_destino(
|
|
lista_ficheiros: list[str],
|
|
base_destino: str,
|
|
categorizar_data: bool = False,
|
|
usar_criacao: bool = False,
|
|
formato_data: str = "%Y-%m-%d",
|
|
usar_threads: bool = True,
|
|
max_workers: int = 8,
|
|
progresso_callback: Callable | None = None,
|
|
) -> None:
|
|
"""
|
|
Copiar cada ficheiro na lista para o directório base, seguindo as regras de categorização
|
|
|
|
argumentos:
|
|
lista_ficheiros (list[str]) -> lista com o caminho dos ficheiros a serem copiados
|
|
base_destino (str) -> directorio base no qual serão criados os novos directórios com base nas regras de categorização
|
|
categorizar_data (bool) -> flag para activar categorização por data (defeito: False)
|
|
usar_criacao (bool) -> usar data de criação (defeito: False usar data de modificacao)
|
|
formato_data (str) -> formato de data para nome dos subdirectórios (defeito: '%d-%m-%Y')
|
|
|
|
retorna:
|
|
None
|
|
"""
|
|
if usar_threads:
|
|
with ThreadPoolExecutor(max_workers=max_workers) as executor:
|
|
futures = [
|
|
executor.submit(
|
|
copiar_ficheiro_individual,
|
|
ficheiro,
|
|
base_destino,
|
|
categorizar_data,
|
|
usar_criacao,
|
|
formato_data,
|
|
)
|
|
for ficheiro in lista_ficheiros
|
|
]
|
|
|
|
for future in as_completed(futures):
|
|
if progresso_callback:
|
|
progresso_callback()
|
|
|
|
# codigo comentado para integrar a funcionalidade na interface gui
|
|
# for _ in tqdm(
|
|
# as_completed(futures),
|
|
# total=len(futures),
|
|
# desc="A copiar ficheiros ",
|
|
# unit="ficheiro ",
|
|
# ):
|
|
# pass
|
|
else:
|
|
for ficheiro in tqdm(
|
|
lista_ficheiros, desc="A copiar ficheiros ", unit="ficheiro "
|
|
):
|
|
copiar_ficheiro_individual(
|
|
ficheiro, base_destino, categorizar_data, usar_criacao, formato_data
|
|
)
|