import os import shutil from typing import List, Union import cv2 import numpy as np from PIL import Image from reactor_core.analyzer import ReActorFaceAnalysis from reactor_core.face_objects import Face from reactor_core.inswap import INSwapper from reactor_core.hyperswap import HyperSwapper import torch import folder_paths import comfy.model_management as model_management from r_modules.shared import state from scripts.reactor_logger import logger from reactor_utils import ( move_path, get_image_md5hash, progress_bar, progress_bar_reset ) from scripts.r_faceboost import swapper, restorer import warnings np.warnings = warnings np.warnings.filterwarnings('ignore') # PROVIDERS try: if torch.cuda.is_available(): providers = ["CUDAExecutionProvider"] elif torch.backends.mps.is_available(): providers = ["CoreMLExecutionProvider"] elif hasattr(torch,'dml') or hasattr(torch,'privateuseone'): providers = ["ROCMExecutionProvider"] else: providers = ["CPUExecutionProvider"] except Exception as e: logger.debug(f"ExecutionProviderError: {e}.\nEP is set to CPU.") providers = ["CPUExecutionProvider"] models_path_old = os.path.join(os.path.dirname(os.path.dirname(__file__)), "models") insightface_path_old = os.path.join(models_path_old, "insightface") insightface_models_path_old = os.path.join(insightface_path_old, "models") models_path = folder_paths.models_dir insightface_path = os.path.join(models_path, "insightface") insightface_models_path = os.path.join(insightface_path, "models") reswapper_path = os.path.join(models_path, "reswapper") hyperswap_path = os.path.join(models_path, "hyperswap") if os.path.exists(models_path_old): move_path(insightface_models_path_old, insightface_models_path) move_path(insightface_path_old, insightface_path) move_path(models_path_old, models_path) if os.path.exists(insightface_path) and os.path.exists(insightface_path_old): shutil.rmtree(insightface_path_old) shutil.rmtree(models_path_old) FS_MODEL = None CURRENT_FS_MODEL_PATH = None ANALYSIS_MODELS = { "640": None, "320": None, } SOURCE_FACES = None SOURCE_IMAGE_HASH = None TARGET_FACES = None TARGET_IMAGE_HASH = None TARGET_FACES_LIST = [] TARGET_IMAGE_LIST_HASH = [] def unload_model(model): if model is not None: del model return None def unload_all_models(): global FS_MODEL, CURRENT_FS_MODEL_PATH FS_MODEL = unload_model(FS_MODEL) ANALYSIS_MODELS["320"] = unload_model(ANALYSIS_MODELS["320"]) ANALYSIS_MODELS["640"] = unload_model(ANALYSIS_MODELS["640"]) def get_current_faces_model(): global SOURCE_FACES return SOURCE_FACES def getAnalysisModel(det_size = (640, 640)): global ANALYSIS_MODELS ANALYSIS_MODEL = ANALYSIS_MODELS[str(det_size[0])] if ANALYSIS_MODEL is None: ANALYSIS_MODEL = ReActorFaceAnalysis( name="buffalo_l", providers=providers, root=insightface_path ) ANALYSIS_MODEL.prepare(ctx_id=0, det_size=det_size) ANALYSIS_MODELS[str(det_size[0])] = ANALYSIS_MODEL return ANALYSIS_MODEL def getFaceSwapModel(model_path: str): global FS_MODEL, CURRENT_FS_MODEL_PATH if FS_MODEL is None or CURRENT_FS_MODEL_PATH is None or CURRENT_FS_MODEL_PATH != model_path: CURRENT_FS_MODEL_PATH = model_path FS_MODEL = unload_model(FS_MODEL) model_filename = os.path.basename(model_path) if "hyperswap" in model_filename.lower(): # Если это Hyperswap model_path = os.path.join(folder_paths.models_dir, "hyperswap", model_filename) FS_MODEL = HyperSwapper(model_path, providers=providers) else: # Если это INSwapper / Reswapper if "reswapper" in model_filename.lower(): model_path = os.path.join(folder_paths.models_dir, "reswapper", model_filename) FS_MODEL = INSwapper(model_path, providers=providers) return FS_MODEL def sort_by_order(face, order: str): if order == "left-right": return sorted(face, key=lambda x: x.bbox[0]) if order == "right-left": return sorted(face, key=lambda x: x.bbox[0], reverse = True) if order == "top-bottom": return sorted(face, key=lambda x: x.bbox[1]) if order == "bottom-top": return sorted(face, key=lambda x: x.bbox[1], reverse = True) if order == "small-large": return sorted(face, key=lambda x: (x.bbox[2] - x.bbox[0]) * (x.bbox[3] - x.bbox[1])) # by default "large-small": return sorted(face, key=lambda x: (x.bbox[2] - x.bbox[0]) * (x.bbox[3] - x.bbox[1]), reverse = True) def get_face_gender( face, face_index, gender_condition, operated: str, order: str, ): # 1. Сортируем ВСЕ найденные лица (без фильтрации!) faces_sorted = sort_by_order(face, order) # 2. Проверяем, существует ли вообще лицо с таким визуальным индексом if face_index >= len(faces_sorted): logger.info("Requested face index (%s) is out of bounds (max available index is %s)", face_index, len(faces_sorted) - 1) return None, 0, None # 3. Берем конкретное лицо по его позиции на фото (например, второе справа) face_selected = faces_sorted[face_index] # Если фильтр по полу отключен (no) - сразу отдаем лицо в работу if gender_condition == 0: return face_selected, 0, face_index # 4. Проверяем пол выбранного лица # face.gender: 0 = female, 1 = male # gender_condition: 1 = female, 2 = male expected_gender = 0 if gender_condition == 1 else 1 actual_gender = getattr(face_selected, 'gender', -1) sel_gender_str = "Male" if actual_gender == 1 else "Female" if actual_gender == 0 else "Unknown" logger.info("%s Face %s: Detected Gender -%s-", operated, face_index, sel_gender_str) # Если пол не совпадает с тем, что заказал юзер if actual_gender != expected_gender: logger.info(f"{operated} Face {face_index}: WRONG gender ({sel_gender_str})") return face_selected, 1, face_index # 1 означает флаг wrong_gender = True (цикл его пропустит) # Если всё идеально return face_selected, 0, face_index def half_det_size(det_size): logger.status("Trying to halve 'det_size' parameter") return (det_size[0] // 2, det_size[1] // 2) def analyze_faces(img_data: np.ndarray, det_size=(640, 640)): face_analyser = getAnalysisModel(det_size) faces = [] try: faces = face_analyser.get(img_data) except Exception as e: # import traceback # traceback.print_exc() # logger.error(f"Error during face analysis: {e}") logger.error("No faces found") # Try halving det_size if no faces are found if len(faces) == 0 and det_size[0] > 320 and det_size[1] > 320: det_size_half = half_det_size(det_size) return analyze_faces(img_data, det_size_half) return faces def get_face_single(img_data: np.ndarray, face, face_index=0, det_size=(640, 640), gender_source=0, gender_target=0, order="large-small"): buffalo_path = os.path.join(insightface_models_path, "buffalo_l.zip") if os.path.exists(buffalo_path): os.remove(buffalo_path) if gender_source != 0: if len(face) == 0 and det_size[0] > 320 and det_size[1] > 320: det_size_half = half_det_size(det_size) return get_face_single(img_data, analyze_faces(img_data, det_size_half), face_index, det_size_half, gender_source, gender_target, order) return get_face_gender(face,face_index,gender_source,"Source", order) if gender_target != 0: if len(face) == 0 and det_size[0] > 320 and det_size[1] > 320: det_size_half = half_det_size(det_size) return get_face_single(img_data, analyze_faces(img_data, det_size_half), face_index, det_size_half, gender_source, gender_target, order) return get_face_gender(face,face_index,gender_target,"Target", order) if len(face) == 0 and det_size[0] > 320 and det_size[1] > 320: det_size_half = half_det_size(det_size) return get_face_single(img_data, analyze_faces(img_data, det_size_half), face_index, det_size_half, gender_source, gender_target, order) try: faces_sorted = sort_by_order(face, order) return faces_sorted[face_index], 0, face_index except IndexError: return None, 0, None def swap_face( source_img: Union[Image.Image, None], target_img: Image.Image, model: Union[str, None] = None, source_faces_index: List[int] = [0], faces_index: List[int] = [0], gender_source: int = 0, gender_target: int = 0, face_model: Union[Face, None] = None, faces_order: List = ["large-small", "large-small"], face_boost_enabled: bool = False, face_restore_model = None, face_restore_visibility: int = 1, codeformer_weight: float = 0.5, interpolation: str = "Bicubic", ): global SOURCE_FACES, SOURCE_IMAGE_HASH, TARGET_FACES, TARGET_IMAGE_HASH result_image = target_img bbox = [] swapped_indexes = [] if model is not None: if isinstance(source_img, str): # source_img is a base64 string import base64, io if 'base64,' in source_img: # check if the base64 string has a data URL scheme # split the base64 string to get the actual base64 encoded image data base64_data = source_img.split('base64,')[-1] # decode base64 string to bytes img_bytes = base64.b64decode(base64_data) else: # if no data URL scheme, just decode img_bytes = base64.b64decode(source_img) source_img = Image.open(io.BytesIO(img_bytes)) target_img = cv2.cvtColor(np.array(target_img), cv2.COLOR_RGB2BGR) if source_img is not None: source_img = cv2.cvtColor(np.array(source_img), cv2.COLOR_RGB2BGR) source_image_md5hash = get_image_md5hash(source_img) if SOURCE_IMAGE_HASH is None: SOURCE_IMAGE_HASH = source_image_md5hash source_image_same = False else: source_image_same = True if SOURCE_IMAGE_HASH == source_image_md5hash else False if not source_image_same: SOURCE_IMAGE_HASH = source_image_md5hash logger.info("Source Image MD5 Hash = %s", SOURCE_IMAGE_HASH) logger.info("Source Image the Same? %s", source_image_same) if SOURCE_FACES is None or not source_image_same: logger.status("Analyzing Source Image...") source_faces = analyze_faces(source_img) SOURCE_FACES = source_faces elif source_image_same: logger.status("Using Hashed Source Face(s) Model...") source_faces = SOURCE_FACES elif face_model is not None: source_faces_index = [0] logger.status("Using Loaded Source Face Model...") source_face_model = [face_model] source_faces = source_face_model else: logger.error("Cannot detect any Source") if source_faces is not None: target_image_md5hash = get_image_md5hash(target_img) if TARGET_IMAGE_HASH is None: TARGET_IMAGE_HASH = target_image_md5hash target_image_same = False else: target_image_same = True if TARGET_IMAGE_HASH == target_image_md5hash else False if not target_image_same: TARGET_IMAGE_HASH = target_image_md5hash logger.info("Target Image MD5 Hash = %s", TARGET_IMAGE_HASH) logger.info("Target Image the Same? %s", target_image_same) if TARGET_FACES is None or not target_image_same: logger.status("Analyzing Target Image...") target_faces = analyze_faces(target_img) TARGET_FACES = target_faces elif target_image_same: logger.status("Using Hashed Target Face(s) Model...") target_faces = TARGET_FACES if len(target_faces) == 0: logger.status("Cannot detect any Target, skipping swapping...") return result_image, bbox, swapped_indexes # --- НОВАЯ ИДЕАЛЬНАЯ ЛОГИКА СОРТИРОВКИ --- # 1. Заранее собираем список ТОЛЬКО ВАЛИДНЫХ исходных лиц valid_source_faces = [] if source_img is not None: for idx in source_faces_index: sf, src_wrong_gender, _ = get_face_single(source_img, source_faces, face_index=idx, gender_source=gender_source, order=faces_order[1]) if sf is not None and src_wrong_gender == 0: valid_source_faces.append(sf) else: sf, src_wrong_gender, _ = get_face_single(None, source_faces, face_index=source_faces_index[0], gender_source=gender_source, order=faces_order[1]) if sf is not None and src_wrong_gender == 0: valid_source_faces.append(sf) if len(valid_source_faces) == 0: logger.status("No valid source face(s) found in the provided Index after gender filter") else: result = target_img if "inswapper" in model: model_path = os.path.join(insightface_path, model) elif "reswapper" in model: model_path = os.path.join(reswapper_path, model) elif "hyperswap" in model: model_path = os.path.join(hyperswap_path, model) face_swapper = getFaceSwapModel(model_path) source_face_idx = 0 # 2. Идем по целевым лицам for face_num in faces_index: target_face, wrong_gender, target_face_index = get_face_single(target_img, target_faces, face_index=face_num, gender_target=gender_target, order=faces_order[0]) if target_face is not None and wrong_gender == 0: logger.status(f"Swapping...") # 3. Берем валидное лицо (если их меньше, чем целей — идем по кругу) source_face_to_use = valid_source_faces[source_face_idx % len(valid_source_faces)] if face_boost_enabled and "hyperswap" not in model: logger.status(f"Face Boost is enabled (inswapper/reswapper only)") bgr_fake, M = face_swapper.get(result, target_face, source_face_to_use, paste_back=False) bgr_fake, scale = restorer.get_restored_face(bgr_fake, face_restore_model, face_restore_visibility, codeformer_weight, interpolation) M *= scale result = swapper.in_swap(result, bgr_fake, M) else: result = face_swapper.get(result, target_face, source_face_to_use) bbox.append(tuple(map(float, target_face.bbox))) swapped_indexes.append(target_face_index) # Продвигаем индекс исходного лица ТОЛЬКО после УСПЕШНОГО применения if len(valid_source_faces) > 1: source_face_idx += 1 elif wrong_gender == 1: logger.status("Wrong target gender detected") continue else: logger.info(f"No target face found for {face_num}") result_image = Image.fromarray(cv2.cvtColor(result, cv2.COLOR_BGR2RGB)) else: logger.status("No source face(s) found") return result_image, bbox, swapped_indexes def swap_face_many( source_img: Union[Image.Image, None], target_imgs: List[Image.Image], model: Union[str, None] = None, source_faces_index: List[int] = [0], faces_index: List[int] = [0], gender_source: int = 0, gender_target: int = 0, face_model: Union[Face, None] = None, faces_order: List = ["large-small", "large-small"], face_boost_enabled: bool = False, face_restore_model = None, face_restore_visibility: int = 1, codeformer_weight: float = 0.5, interpolation: str = "Bicubic", ): global SOURCE_FACES, SOURCE_IMAGE_HASH, TARGET_FACES, TARGET_IMAGE_HASH, TARGET_FACES_LIST, TARGET_IMAGE_LIST_HASH result_images = target_imgs bbox = [] swapped_indexes = [] if model is not None: if isinstance(source_img, str): import base64, io if 'base64,' in source_img: base64_data = source_img.split('base64,')[-1] img_bytes = base64.b64decode(base64_data) else: img_bytes = base64.b64decode(source_img) source_img = Image.open(io.BytesIO(img_bytes)) target_imgs = [cv2.cvtColor(np.array(target_img), cv2.COLOR_RGB2BGR) for target_img in target_imgs] if source_img is not None: source_img = cv2.cvtColor(np.array(source_img), cv2.COLOR_RGB2BGR) source_image_md5hash = get_image_md5hash(source_img) if SOURCE_IMAGE_HASH is None: SOURCE_IMAGE_HASH = source_image_md5hash source_image_same = False else: source_image_same = True if SOURCE_IMAGE_HASH == source_image_md5hash else False if not source_image_same: SOURCE_IMAGE_HASH = source_image_md5hash logger.info("Source Image MD5 Hash = %s", SOURCE_IMAGE_HASH) logger.info("Source Image the Same? %s", source_image_same) if SOURCE_FACES is None or not source_image_same: logger.status("Analyzing Source Image...") source_faces = analyze_faces(source_img) SOURCE_FACES = source_faces elif source_image_same: logger.status("Using Hashed Source Face(s) Model...") source_faces = SOURCE_FACES elif face_model is not None: source_faces_index = [0] logger.status("Using Loaded Source Face Model...") source_face_model = [face_model] source_faces = source_face_model else: logger.error("Cannot detect any Source") if source_faces is not None: target_faces = [] pbar = progress_bar(len(target_imgs)) if len(TARGET_IMAGE_LIST_HASH) > 0: logger.status(f"Using Hashed Target Face(s) Model...") else: logger.status(f"Analyzing Target Image...") for i, target_img in enumerate(target_imgs): if state.interrupted or model_management.processing_interrupted(): logger.status("Interrupted by User") break target_image_md5hash = get_image_md5hash(target_img) if len(TARGET_IMAGE_LIST_HASH) == 0: TARGET_IMAGE_LIST_HASH = [target_image_md5hash] target_image_same = False elif len(TARGET_IMAGE_LIST_HASH) == i: TARGET_IMAGE_LIST_HASH.append(target_image_md5hash) target_image_same = False else: target_image_same = True if TARGET_IMAGE_LIST_HASH[i] == target_image_md5hash else False if not target_image_same: TARGET_IMAGE_LIST_HASH[i] = target_image_md5hash logger.info("(Image %s) Target Image MD5 Hash = %s", i, TARGET_IMAGE_LIST_HASH[i]) logger.info("(Image %s) Target Image the Same? %s", i, target_image_same) if len(TARGET_FACES_LIST) == 0: target_face = analyze_faces(target_img) TARGET_FACES_LIST = [target_face] elif len(TARGET_FACES_LIST) == i and not target_image_same: target_face = analyze_faces(target_img) TARGET_FACES_LIST.append(target_face) elif len(TARGET_FACES_LIST) != i and not target_image_same: target_face = analyze_faces(target_img) TARGET_FACES_LIST[i] = target_face elif target_image_same: target_face = TARGET_FACES_LIST[i] if target_face is not None: target_faces.append(target_face) pbar.update(1) progress_bar_reset(pbar) if len(target_faces) == 0: logger.status("Cannot detect any Target, skipping swapping...") return result_images, bbox, swapped_indexes # --- НОВАЯ ИДЕАЛЬНАЯ ЛОГИКА СОРТИРОВКИ --- valid_source_faces = [] if source_img is not None: for idx in source_faces_index: sf, src_wrong_gender, _ = get_face_single(source_img, source_faces, face_index=idx, gender_source=gender_source, order=faces_order[1]) if sf is not None and src_wrong_gender == 0: valid_source_faces.append(sf) else: sf, src_wrong_gender, _ = get_face_single(None, source_faces, face_index=source_faces_index[0], gender_source=gender_source, order=faces_order[1]) if sf is not None and src_wrong_gender == 0: valid_source_faces.append(sf) if len(valid_source_faces) == 0: logger.status("No valid source face(s) found in the provided Index after gender filter") else: results = target_imgs if "inswapper" in model: model_path = os.path.join(insightface_path, model) elif "reswapper" in model: model_path = os.path.join(reswapper_path, model) elif "hyperswap" in model: model_path = os.path.join(hyperswap_path, model) face_swapper = getFaceSwapModel(model_path) source_face_idx = 0 pbar = progress_bar(len(target_imgs)) logger.status(f"Swapping...") for face_num in faces_index: target_used_in_any_image = False for i, (target_img, target_face_list) in enumerate(zip(results, target_faces)): target_face_single, wrong_gender, target_face_index = get_face_single(target_img, target_face_list, face_index=face_num, gender_target=gender_target, order=faces_order[0]) if target_face_single is not None and wrong_gender == 0: target_used_in_any_image = True source_face_to_use = valid_source_faces[source_face_idx % len(valid_source_faces)] result = target_img if face_boost_enabled and "hyperswap" not in model: bgr_fake, M = face_swapper.get(target_img, target_face_single, source_face_to_use, paste_back=False) bgr_fake, scale = restorer.get_restored_face(bgr_fake, face_restore_model, face_restore_visibility, codeformer_weight, interpolation) M *= scale result = swapper.in_swap(target_img, bgr_fake, M) else: result = face_swapper.get(target_img, target_face_single, source_face_to_use) results[i] = result bbox.append(tuple(map(float, target_face_single.bbox))) swapped_indexes.append(target_face_index) pbar.update(1) elif wrong_gender == 1: logger.status("Wrong target gender detected") pbar.update(1) continue else: logger.info(f"{i}: No target face found for {face_num}") pbar.update(1) if target_used_in_any_image and len(valid_source_faces) > 1: source_face_idx += 1 progress_bar_reset(pbar) result_images = [Image.fromarray(cv2.cvtColor(result, cv2.COLOR_BGR2RGB)) for result in results] else: logger.status("No source face(s) found") return result_images, bbox, swapped_indexes