core/pipelines/pipeline_input_processor.py

import os
import gradio as gr
from typing import Dict, Any

from core.settings import LORA_DIR
from utils.app_utils import sanitize_filename, get_lora_path

def process_pipeline_inputs(ui_inputs: Dict[str, Any], progress: gr.Progress, workflow_model_type: str = "h3") -> Dict[str, Any]:
    active_loras_for_gpu, active_loras_for_meta = [], []
    lora_data = ui_inputs.get('lora_data', [])
    if lora_data:
        sources, ids, scales, files = lora_data[0::4], lora_data[1::4], lora_data[2::4], lora_data[3::4]
        for i, (source, lora_id, scale, _) in enumerate(zip(sources, ids, scales, files)):
            if scale > 0 and lora_id and lora_id.strip():
                lora_filename = None
                if source == "File":
                    lora_filename = sanitize_filename(lora_id)
                    local_path = os.path.join(LORA_DIR, lora_filename)
                    if not os.path.exists(local_path):
                        raise gr.Error(f"Uploaded LoRA file '{lora_id}' no longer exists on server. Please re-upload it.")
                elif source in ("Civitai", "Hugging Face"):
                    local_path, status = get_lora_path(source, lora_id, os.environ.get("CIVITAI_API_KEY", ""), progress)
                    if local_path:
                        lora_filename = os.path.basename(local_path)
                    else:
                        raise gr.Error(f"Failed to prepare LoRA {lora_id}: {status}")
                
                if lora_filename:
                    active_loras_for_gpu.append({"lora_name": lora_filename, "strength_model": scale, "strength_clip": scale})
                    active_loras_for_meta.append(f"{source} {lora_id}:{scale}")

    return {
        "active_loras_for_gpu": active_loras_for_gpu,
        "active_loras_for_meta": active_loras_for_meta,
        "temp_files_to_clean": []
    }