mirrors/facefusion
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add safer global named resolve_file_pattern() (#811)

Browse Source
This commit is contained in:
Henry Ruhs 2024-11-12 12:00:45 +01:00 committed by GitHub
parent 8385e199f4
commit 244df12ff8
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
4 changed files with 21 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
facefusion/core.py
View File

@ -1,4 +1,3 @@
import glob
import itertools
import shutil
import signal
@ -17,7 +16,7 @@ from facefusion.face_analyser import get_average_face, get_many_faces, get_one_f
from facefusion.face_selector import sort_and_filter_faces
from facefusion.face_store import append_reference_face, clear_reference_faces, get_reference_faces
from facefusion.ffmpeg import copy_image, extract_frames, finalize_image, merge_video, replace_audio, restore_audio
from facefusion.filesystem import filter_audio_paths, is_image, is_video, list_directory, resolve_relative_path
from facefusion.filesystem import filter_audio_paths, is_image, is_video, list_directory, resolve_file_pattern, resolve_relative_path
from facefusion.jobs import job_helper, job_manager, job_runner
from facefusion.jobs.job_list import compose_job_list
from facefusion.memory import limit_system_memory
@ -315,8 +314,8 @@ def process_bulk(args : Args) -> ErrorCode:
job_id = job_helper.suggest_job_id('bulk')
step_args = reduce_step_args(args)
job_args = reduce_job_args(args)
source_paths = sorted(glob.glob(job_args.get('source_pattern')))
target_paths = sorted(glob.glob(job_args.get('target_pattern')))
source_paths = resolve_file_pattern(job_args.get('source_pattern'))
target_paths = resolve_file_pattern(job_args.get('target_pattern'))
if job_manager.create_job(job_id):

13
facefusion/filesystem.py
View File

@ -1,3 +1,4 @@
import glob
import os
import shutil
from pathlib import Path
@ -127,12 +128,18 @@ def create_directory(directory_path : str) -> bool:
def list_directory(directory_path : str) -> Optional[List[str]]:
if is_directory(directory_path):
files = os.listdir(directory_path)
files = [ Path(file).stem for file in files if not Path(file).stem.startswith(('.', '__')) ]
return sorted(files)
file_paths = os.listdir(directory_path)
file_paths = [ Path(file_path).stem for file_path in file_paths if not Path(file_path).stem.startswith(('.', '__')) ]
return sorted(file_paths)
return None
def resolve_file_pattern(file_pattern : str) -> List[str]:
if in_directory(file_pattern):
return sorted(glob.glob(file_pattern))
return []
def remove_directory(directory_path : str) -> bool:
if is_directory(directory_path):
shutil.rmtree(directory_path, ignore_errors = True)

13
facefusion/jobs/job_manager.py
View File

@ -1,11 +1,10 @@
import glob
import os
from copy import copy
from typing import List, Optional
from facefusion.choices import job_statuses
from facefusion.date_helper import get_current_date_time
from facefusion.filesystem import create_directory, is_directory, is_file, move_file, remove_directory, remove_file
from facefusion.filesystem import create_directory, is_directory, is_file, move_file, remove_directory, remove_file, resolve_file_pattern
from facefusion.jobs.job_helper import get_step_output_path
from facefusion.json import read_json, write_json
from facefusion.typing import Args, Job, JobSet, JobStatus, JobStep, JobStepStatus
@ -86,12 +85,12 @@ def find_jobs(job_status : JobStatus) -> JobSet:
def find_job_ids(job_status : JobStatus) -> List[str]:
job_pattern = os.path.join(JOBS_PATH, job_status, '*.json')
job_files = glob.glob(job_pattern)
job_files.sort(key = os.path.getmtime)
job_paths = resolve_file_pattern(job_pattern)
job_paths.sort(key = os.path.getmtime)
job_ids = []
for job_file in job_files:
job_id, _ = os.path.splitext(os.path.basename(job_file))
for job_path in job_paths:
job_id, _ = os.path.splitext(os.path.basename(job_path))
job_ids.append(job_id)
return job_ids
@ -248,7 +247,7 @@ def find_job_path(job_id : str) -> Optional[str]:
if job_file_name:
for job_status in job_statuses:
job_pattern = os.path.join(JOBS_PATH, job_status, job_file_name)
job_paths = glob.glob(job_pattern)
job_paths = resolve_file_pattern(job_pattern)
for job_path in job_paths:
return job_path

5
facefusion/temp_helper.py
View File

@ -1,9 +1,8 @@
import glob
import os
from typing import List
from facefusion import state_manager
from facefusion.filesystem import create_directory, move_file, remove_directory
from facefusion.filesystem import create_directory, move_file, remove_directory, resolve_file_pattern
def get_temp_file_path(file_path : str) -> str:
@ -36,7 +35,7 @@ def clear_temp_directory(file_path : str) -> bool:
def get_temp_frame_paths(target_path : str) -> List[str]:
temp_frames_pattern = get_temp_frames_pattern(target_path, '*')
return sorted(glob.glob(temp_frames_pattern))
return resolve_file_pattern(temp_frames_pattern)
def get_temp_frames_pattern(target_path : str, temp_frame_prefix : str) -> str: