SMB 공유의 파일 수집과 미디어 집합

SMB 커넥터는 Windows File Server나 Samba 같은 공유 서버와 Foundry 데이터셋 사이에서 파일을 전달합니다. SMB 프로토콜 버전 2와 3을 지원하며, 원천 탐색, 대량 가져오기, 증분 수집과 파일 내보내기는 일반 제공(GA)입니다.

파일 형식을 보존하고 스키마는 자동 적용하지 않습니다. 출력 데이터셋에 스키마를 설정하거나 하위 변환에서 파일을 해석합니다. 파일 크기 자체의 한도는 없지만 대규모 전송은 네트워크 문제로 실패할 수 있고, 이틀 넘게 실행되는 직접 클라우드 동기화는 중단됩니다. 작은 파일과 실행당 제한된 파일 수를 사용하고 스케줄을 더 자주 실행하는 구성을 검토합니다.

소스 구성

Data Connection의 +New Source에서 Protocol sources 아래 SMB를 선택합니다.

연결 주체가 Hostname의 Port에 도달할 수 있어야 합니다. 직접 연결 외부 통신 정책을 사용할 때에는 TCP 수준의 허용 목록을 구성합니다.

파일 수집과 미디어 처리의 구분

기본 동기화는 파일 기반 인터페이스를 사용하고 일반 데이터셋에 파일을 저장합니다. SMB 커넥터는 미디어 집합 동기화를 직접 지원하지 않습니다.

Pipeline Builder의 미디어 변환이나 PDF 텍스트 추출 등을 사용하려면 먼저 파일을 데이터셋에 수집한 뒤 Python 변환에서 미디어 집합에 업로드합니다. 동기화 데이터셋에서 직접 만든 미디어 참조는 이 미디어 변환이 받는 참조 형식과 다릅니다.

파일 내보내기

소스에서 내보내기를 활성화하고 새 내보내기를 생성합니다. 필수 Directory path의 기본값은 /이며 공유 안의 대상 폴더를 가리킵니다.

최종 파일 위치는 <Share>/<Directory Path>/<Exported File Path>로 계산합니다. 공유 이름과 폴더 경로, 내보낼 논리 파일 경로를 구분해 설정합니다.

코드에서 파일과 메타데이터 읽기

외부 변환에 SMB 소스를 가져오고 smbclient, pandas 의존성을 준비합니다. 예제의 YOUR_..._RID, 도메인, 사용자와 SMB 경로는 실제 구성으로 바꿉니다. Password 비밀 정보는 소스에서 읽습니다. 아래의 사용자 이름 구성은 AD의 DOMAIN\USERNAME 형식을 가정합니다.

첫 예제는 디렉터리의 하위 폴더를 제외한 파일을 읽습니다. 파일 내용과 별도로 이름, 안전하게 변환한 이름, 크기와 원격 경로를 메타데이터 출력에 기록합니다. 문제가 될 수 있는 파일 이름 문자는 밑줄로 바꿉니다. 변환한 이름이 서로 충돌하지 않는 입력 범위인지 확인합니다.

import smbclient
import pandas as pd
from transforms.api import Output, transform
from transforms.external.systems import external_systems, Source

DOMAIN = "<smb_domain>"
USERNAME = "<smb_username>"
SMB_PATH = "<smb_path>"

@external_systems(
    smb_source=Source("ri.magritte..source.YOUR_SMB_SOURCE_RID")
)
@transform(
    file_metadata=Output("ri.foundry.main.dataset.YOUR_METADATA_OUTPUT_RID"),
    output_files_dataset=Output("ri.foundry.main.dataset.YOUR_FILES_OUTPUT_RID")

)
def read_smb_files(ctx, smb_source, file_metadata, output_files_dataset):
    """
    Read files from an SMB share and output files and metadata.
    """
    # Configure SMB client with credentials from source
    username = f"{DOMAIN}\\{USERNAME}"
    password = smb_source.get_secret("Password")
    smbclient.ClientConfig(username=username, password=password)

    # List files in a directory
    files_info = []

    for item in smbclient.scandir(SMB_PATH):
        if not item.is_dir():
            # Get file info
            stat = item.stat()

            file_path = f"{SMB_PATH}\\{item.name}"
            files_info.append({
                "filename": item.name,
                "safe_filename": create_safe_filename(item.name),
                "size_bytes": stat.st_size,
                "path": file_path
            })
            with smbclient.open_file(file_path, mode="rb") as f:
                content = f.read()
                safe_foundry_file_name = create_safe_filename(item.name)
                with output_files_dataset.filesystem().open(safe_foundry_file_name, 'wb') as fileobj:
                    fileobj.write(content)

    # Write metadata to output
    if files_info:
        df = ctx.spark_session.createDataFrame(pd.DataFrame(files_info))
        file_metadata.write_dataframe(df)


def create_safe_filename(filepath: str) -> str:
    """
    Create a safe filename for storage by removing problematic characters.

    Args:
        filepath: Original file path

    Returns:
        Sanitized filename safe for storage
    """
    import os
    filename = os.path.basename(filepath)
    # Replace characters that might cause issues in storage
    safe_chars = str.maketrans({"\\": "_", "/": "_", ":": "_", "?": "_", "*": "_", "<": "_", ">": "_", "|": "_"})
    return filename.translate(safe_chars)

목록에 파일이 없으면 이 예제는 메타데이터 출력을 쓰는 분기를 수행하지 않습니다. 수집된 파일이 있다는 전제를 사용하는 하위 처리에서는 빈 디렉터리의 동작을 고려합니다.

미디어 집합으로 분류하는 변환

다음 예제의 입력 데이터셋은 path 열에 처리할 디렉터리 경로를 포함합니다. transform_directory_path는 예시 마운트 접두사를 제거하고 SMB UNC 경로를 구성합니다. 호스트, 공유 이름, 계정과 로컬 마운트 접두사는 실제 환경에 맞게 바꿉니다.

출력은 문서, 이미지, 스프레드시트용 미디어 집합과 파일별 메타데이터 데이터셋입니다.

미디어 집합에 저장할 수 있다는 사실이 모든 파일 형식에 동일한 후속 파서가 있다는 뜻은 아닙니다. 필요한 미디어 변환의 형식 지원을 따로 확인합니다.

변환은 하위 디렉터리를 재귀 탐색하고 이름, 전체 경로, 디렉터리, 크기, 생성 및 수정 시각, 확장자, 분류와 처리 시각을 기록합니다. 다운로드 실패는 download_error, 파일 처리 실패나 디렉터리 처리 실패는 processing_error에 남깁니다. 분류가 error 또는 directory_error인 행도 결과에 포함될 수 있습니다.

하위 디렉터리 탐색 함수는 예외를 로그에 남긴 뒤 반환하므로, 탐색하지 못한 모든 파일이 개별 오류 행으로 기록된다고 보장하지 않습니다. 최종 처리 건수와 분류별 건수는 다운로드 성공 전에 증가하므로 성공적인 미디어 업로드 수와 다를 수 있습니다.

"""
SMB File Processing Transform Template

This template demonstrates how to connect to an SMB share
from a Foundry Python transform to process files and organize them by type.

Key concepts covered:
- External systems integration with SMB sources
- Recursive directory traversal
- File categorization by extension
- MediaSet outputs for different file types
- Structured metadata collection
- Error handling and logging

Prerequisites:
- SMB source configured in Data Connection
- Input dataset with directory paths
- Output datasets/mediasets configured
- smbclient library available in the repository
"""

import logging
import os
from datetime import datetime

import pandas as pd
import smbclient
from transforms.api import Input, Output, transform
from transforms.external.systems import ResolvedSource, Source, external_systems
from transforms.mediasets import MediaSetOutput

# Configure logging

logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s")
logger = logging.getLogger(__name__)


@external_systems(
    smb_source=Source("ri.magritte..source.YOUR_SMB_SOURCE_RID")  # Replace with your SMB source RID
)
@transform(
    input_directories=Input("ri.foundry.main.dataset.YOUR_INPUT_DATASET_RID"),  # Dataset containing directory paths
    documents_mediaset=MediaSetOutput("ri.mio.main.media-set.YOUR_DOCUMENTS_MEDIASET_RID"),  # For PDF, DOC, PPT files
    images_mediaset=MediaSetOutput("ri.mio.main.media-set.YOUR_IMAGES_MEDIASET_RID"),  # For image files
    spreadsheets_mediaset=MediaSetOutput("ri.mio.main.media-set.YOUR_SPREADSHEETS_MEDIASET_RID"),  # For Excel files
    file_metadata=Output("ri.foundry.main.dataset.YOUR_METADATA_OUTPUT_RID"),  # Structured metadata output
)
def smb_file_processor(
    ctx,
    smb_source: ResolvedSource,
    input_directories,
    documents_mediaset,
    images_mediaset,
    spreadsheets_mediaset,
    file_metadata,
):
    """
    Process files from SMB directories and categorize them into different outputs.

    This transform:
    1. Reads directory paths from input dataset
    2. Connects to SMB share using configured source
    3. Recursively scans directories for files
    4. Categorizes files by extension
    5. Downloads and stores files in appropriate MediaSets
    6. Creates structured metadata for all processed files

    Args:
        ctx: Transform context
        smb_source: Configured SMB source from Data Connection
        input_directories: Dataset with 'path' column containing directory paths to process
        documents_mediaset: MediaSet for document files (PDF, DOC, PPT, TXT)
        images_mediaset: MediaSet for image files (JPG, PNG, etc.)
        spreadsheets_mediaset: MediaSet for spreadsheet files (XLSX)
        file_metadata: Structured dataset for file metadata and processing results
    """

    # Define file categories by extension
    DOCUMENT_EXTENSIONS = {".pdf", ".pptx", ".docx", ".txt", ".ppt", ".doc"}
    IMAGE_EXTENSIONS = {".jpg", ".jpeg", ".png", ".bmp", ".tiff", ".tif", ".webp", ".gif"}
    SPREADSHEET_EXTENSIONS = {".xlsx", ".xls", ".csv"}

    # SMB connection configuration
    # These values should match your SMB source configuration in Data Connection
    smb_config = {
        "hostname": "your-smb-server.domain.com",  # Replace with your SMB server hostname
        "share": "your-share-name",  # Replace with your SMB share name
        "username": "your-service-account",  # Replace with your SMB username
        "password": smb_source.get_secret("Password"),  # Password stored securely in source
        "domain": "your.domain.com",  # Replace with your domain (optional)
    }

    logger.info("=" * 80)
    logger.info("SMB FILE PROCESSOR - STARTING")
    logger.info("=" * 80)
    logger.info(f"Server: {smb_config['hostname']}")
    logger.info(f"Share: {smb_config['share']}")
    logger.info(f"Username: {smb_config['domain']}\\{smb_config['username']}")
    logger.info("=" * 80)

    def transform_directory_path(input_path: str) -> str:
        """
        Transform input directory path to SMB UNC format.

        Modify this function based on your specific path transformation needs.
        Example: Convert from local mount path to UNC path format.

        Args:
            input_path: Original directory path from input dataset

        Returns:
            SMB UNC path format (\\\\server\\share\\path)
        """
        # Example transformation - customize for your environment
        # Remove local mount prefix and convert to UNC format
        local_prefix = "/mnt/your-mount/path/"  # Replace with your local mount prefix

        if input_path.startswith(local_prefix):
            relative_path = input_path[len(local_prefix) :]
        else:
            # Handle alternative path formats
            logger.warning(f"Unexpected path format: {input_path}")
            relative_path = input_path

        # Convert to UNC format
        unc_path = f"\\\\{smb_config['hostname']}\\{smb_config['share']}\\{relative_path}"
        return unc_path.replace("/", "\\")  # Ensure Windows path separators

    def get_file_category(filename: str) -> str:
        """
        Determine file category based on file extension.

        Args:
            filename: Name of the file

        Returns:
            Category string: 'document', 'image', 'spreadsheet', or 'other'
        """
        ext = os.path.splitext(filename)[1].lower()

        if ext in DOCUMENT_EXTENSIONS:
            return "document"
        elif ext in IMAGE_EXTENSIONS:
            return "image"
        elif ext in SPREADSHEET_EXTENSIONS:
            return "spreadsheet"
        else:
            return "other"

    def create_safe_filename(filepath: str) -> str:
        """
        Create a safe filename for storage by removing problematic characters.

        Args:
            filepath: Original file path

        Returns:
            Sanitized filename safe for storage
        """
        filename = os.path.basename(filepath)
        # Replace characters that might cause issues in storage
        safe_chars = str.maketrans({"\\": "_", "/": "_", ":": "_", "?": "_", "*": "_", "<": "_", ">": "_", "|": "_"})
        return filename.translate(safe_chars)

    def get_all_files_recursive(directory_path: str) -> list:
        """
        Recursively scan directory and return all files with their metadata.

        Args:
            directory_path: SMB directory path to scan

        Returns:
            List of dictionaries containing file information
        """
        all_files = []

        try:
            # Scan directory contents
            items = list(smbclient.scandir(directory_path))

            for item in items:
                if item.is_dir():
                    # Recursively process subdirectories
                    subdirectory_path = f"{directory_path}\\{item.name}"
                    logger.info(f"Scanning subdirectory: {subdirectory_path}")
                    all_files.extend(get_all_files_recursive(subdirectory_path))
                else:
                    # Add file information
                    file_info = {"item": item, "directory_path": directory_path}
                    all_files.append(file_info)

        except Exception as e:
            logger.error(f"Error scanning directory {directory_path}: {str(e)}")

        return all_files

    # Initialize tracking variables
    metadata_records = []
    total_files_processed = 0
    files_by_category = {"document": 0, "image": 0, "spreadsheet": 0, "other": 0}

    try:
        # Configure SMB client authentication
        username = f"{smb_config['domain']}\\{smb_config['username']}"
        smbclient.ClientConfig(username=username, password=smb_config["password"])

        # Read input directories from dataset
        input_df = input_directories.dataframe()
        directory_rows = input_df.select("path").collect()
        directory_paths = [row.path for row in directory_rows]

        logger.info(f"Processing {len(directory_paths)} directories")

        # Process each directory
        for idx, directory_path in enumerate(directory_paths, 1):
            logger.info(f"[{idx}/{len(directory_paths)}] Processing: {directory_path}")

            try:
                # Transform to SMB path format
                smb_directory_path = transform_directory_path(directory_path)
                logger.info(f"SMB path: {smb_directory_path}")

                # Get all files recursively
                all_files_info = get_all_files_recursive(smb_directory_path)
                logger.info(f"Found {len(all_files_info)} files in directory tree")

                # Process each file
                for file_info in all_files_info:
                    file_item = file_info["item"]
                    file_directory_path = file_info["directory_path"]
                    filename = file_item.name
                    file_path = f"{file_directory_path}\\{filename}"

                    try:
                        # Get file statistics
                        stat_info = file_item.stat()
                        file_size = stat_info.st_size
                        created_time = datetime.fromtimestamp(stat_info.st_ctime)
                        modified_time = datetime.fromtimestamp(stat_info.st_mtime)

                        # Categorize file
                        category = get_file_category(filename)
                        files_by_category[category] += 1
                        total_files_processed += 1

                        # Create metadata record
                        metadata_record = {
                            "filename": filename,
                            "full_path": file_path,
                            "directory_path": file_directory_path,
                            "file_size_bytes": file_size,
                            "created_date": created_time,
                            "modified_date": modified_time,
                            "file_extension": os.path.splitext(filename)[1].lower(),
                            "category": category,
                            "processed_timestamp": pd.Timestamp.now(),
                        }

                        # Download and store files based on category
                        if category in ["document", "image", "spreadsheet"]:
                            try:
                                # Download file content
                                with smbclient.open_file(file_path, mode="rb") as smb_file:
                                    file_content = smb_file.read()

                                # Create safe filename for storage
                                safe_filename = create_safe_filename(file_path)

                                # Store in appropriate MediaSet
                                import io

                                file_stream = io.BytesIO(file_content)

                                if category == "document":
                                    documents_mediaset.put_media_item(file_stream, safe_filename)
                                elif category == "image":
                                    images_mediaset.put_media_item(file_stream, safe_filename)
                                elif category == "spreadsheet":
                                    spreadsheets_mediaset.put_media_item(file_stream, safe_filename)

                                logger.info(f"✓ Downloaded {category}: {filename} ({file_size:,} bytes)")

                            except Exception as download_error:
                                logger.error(f"✗ Failed to download {filename}: {str(download_error)}")
                                metadata_record["download_error"] = str(download_error)
                        else:
                            logger.info(f"⚠ Skipped {filename} (category: {category})")

                        metadata_records.append(metadata_record)

                    except Exception as file_error:
                        logger.error(f"✗ Error processing file {filename}: {str(file_error)}")
                        # Add error record
                        error_record = {
                            "filename": filename,
                            "full_path": file_path,
                            "directory_path": file_directory_path,
                            "file_size_bytes": 0,
                            "created_date": None,
                            "modified_date": None,
                            "file_extension": os.path.splitext(filename)[1].lower(),
                            "category": "error",
                            "processed_timestamp": pd.Timestamp.now(),
                            "processing_error": str(file_error),
                        }
                        metadata_records.append(error_record)

            except Exception as dir_error:
                logger.error(f"✗ Failed to process directory {directory_path}: {str(dir_error)}")
                # Add directory error record
                error_record = {
                    "filename": None,
                    "full_path": None,
                    "directory_path": directory_path,
                    "file_size_bytes": 0,
                    "created_date": None,
                    "modified_date": None,
                    "file_extension": None,
                    "category": "directory_error",
                    "processed_timestamp": pd.Timestamp.now(),
                    "processing_error": str(dir_error),
                }
                metadata_records.append(error_record)

        # Write metadata to output dataset
        if metadata_records:
            metadata_df = ctx.spark_session.createDataFrame(pd.DataFrame(metadata_records))
            file_metadata.write_dataframe(metadata_df)
            logger.info(f"✓ Written {len(metadata_records)} metadata records")
        else:
            logger.warning("No metadata records to write")

    except Exception as e:
        logger.error(f"✗ CRITICAL FAILURE: {str(e)}")
        raise

    # Log final summary
    logger.info("\n" + "=" * 80)
    logger.info("PROCESSING SUMMARY")
    logger.info("=" * 80)
    logger.info(f"Total files processed: {total_files_processed:,}")
    logger.info(f"Documents: {files_by_category['document']:,}")
    logger.info(f"Images: {files_by_category['image']:,}")
    logger.info(f"Spreadsheets: {files_by_category['spreadsheet']:,}")
    logger.info(f"Other/Skipped: {files_by_category['other']:,}")
    logger.info(f"Metadata records created: {len(metadata_records):,}")
    logger.info("=" * 80)

디렉터리 목록과 읽은 파일 내용을 메모리에 보관하는 예제이므로 파일 수와 크기에 맞는 실행 자원을 사용합니다. 메타데이터 목록이 비어 있으면 출력 기록을 건너뛰고 경고 로그를 남깁니다. 최외곽에서 발생한 오류는 로그를 남긴 뒤 다시 발생시킵니다.