[Python] 리눅스 환경에서 윈도우 환경으로 파일 저장 (SMB 프로토콜)

리눅스 환경에서 윈도우 환경으로 파일 혹은 디렉토리와 하위 디렉토리, 파일들을 모두 전송하는 코드이다.

파이썬의 pysmb 모듈을 사용하며 Connection 정보(IP 주소, 사용자 이름, 패스워드 등)를 명확히 입력해주어야 한다.

from typing import Tuple, Optional
import json
import os
from smb.SMBConnection import SMBConnection

DataGram = Optional[bytes]
MetaDataGram = Optional[bytes]

# 디렉토리 생성 함수
def create_directory_recursive(self, conn, target_folder, destination_path):
    
    # 경로에서 폴더 구조 추출
    folders = destination_path.split("/")

    # 첫 번째 폴더 중복되므로 삭제
    folders.remove(target_folder)

    # dir_path 초기화
    dir_path=folders[0]

    # 중복되는경우 에러 메시지는 뜨지만 건너 뜀
    for i in range(0, len(folders)):
        try:
            conn.createDirectory(target_folder, dir_path)
        except Exception as e:
            print(f"Error : {e}")
        if i>0:
            dir_path = os.path.join(dir_path, folders[i])

# 디렉토리 모든 내용 삭제 함수
def delete_contents(directory):
    """주어진 디렉토리의 모든 내용을 삭제합니다."""
    for item in os.listdir(directory):
        item_path = os.path.join(directory, item)
        if os.path.isdir(item_path):
            # 폴더인 경우 재귀적으로 내부 내용 삭제
            delete_contents(item_path)
            # 폴더를 삭제합니다.
            os.rmdir(item_path)
        else:
            # 파일인 경우 삭제합니다.
            os.remove(item_path)

# 하위 디렉토리 전체 복사 및 원본 데이터 삭제 함수
def copy_and_delete_source(self, source_dir, target_dir, file_extensions, username, password, my_name, server_name):
    """
    Copy files and folders from source directory to destination directory
    and then delete them from the source.
    """   
    is_send_done=False
    try:
        # Connect to the network share using username and password
        conn = SMBConnection(username, password, my_name=my_name, remote_name=server_name, use_ntlm_v2=True, is_direct_tcp=True)
        conn.connect(server_name, 445)
        
        for root, directories, files in os.walk(source_dir):
            # Get the relative path from the source directory
            relative_path = os.path.relpath(root, source_dir)

            # Replace underscores with pound signs in the relative path
            relative_path = relative_path.replace('_WaferInfo', '').replace('_', '#')

            # Construct the corresponding destination directory path using UNC path format
            dest_directory = os.path.join(target_dir, relative_path)

            # Check if the source directory contains any files
            if not files:
                continue  # Skip copying if the source directory is empty

            # 디렉토리 생성
            create_directory_recursive(self, conn, target_dir, dest_directory)

            # Copy files to the destination directory
            for file in files:
                if file.endswith(tuple(file_extensions)):
                    source_file = os.path.join(root, file)
                    dest_file = os.path.join(dest_directory, file)
                    create_directory_recursive(self, conn, target_dir, dest_file)
                    with open(source_file, 'rb') as src_file:
                        conn.storeFile(service_name=target_dir, path=dest_file.replace(f"{target_dir}/", ""), file_obj=src_file)
                        # conn.storeFile(service_name=share_name, path=dest_directory, file_obj=src_file)
                        
            # Delete files from the source directory after copying
            for file in files:
                source_file = os.path.join(root, file)
                if os.path.exists(source_file):
                    try:
                        os.remove(source_file)
                    except Exception as e:
                        print(f"Error : {e}")

            # Delete the directory from the source directory after copying
            if os.path.exists(root):
                try:
                    os.rmdir(root)
                except Exception as e:
                    print(f"Error : {e}")

        is_send_done=True
    except Exception as e:
         print(f"Error : {e}")
    
    # 전제 폴더 삭제
    if is_send_done:
        delete_contents(source_dir)
    
class MyExampleClass:
    def __init__(self, node_name: str, config: str):
        """Called once on node creation."""

        # The config is passed through from the "passthrough-config" field in the node config.
        self._config = json.loads(config)

        # 원본 소스 디렉토리
        self.source_directory = self._config["source_directory"]

        # 저장할 타겟 디렉토리
        self.target_directory=self._config["target_directory"]

        # 저장할 파일 확장자
        self.file_extensions=self._config["file_extensions"]

        # 윈도우 사용자 이름
        self.username=self._config["username"]

        # 윈도우 사용자 패스워드
        self.password=self._config["password"]

        # 이 파이썬 프로그램이 실행되는 컴퓨터 이름
        self.my_name=self._config["my_name"]

        # 저장할 타겟의 IP 주소
        self.server_name=self._config["server_name"]        

    def __call__(self, data: DataGram, meta: MetaDataGram) -> Tuple[DataGram, MetaDataGram]:
        """Called by the preceding source, can be multiple times per cycle."""

        """ test 정보 예시 (외부 config)"""
        # source_directory = "/filedata/test"      
        # target_directory = "public_folder"  
        # file_extensions = (".csv", ".xls")
        # username = "tester"
        # password = "test"
        # server_name = "192.168.5.22"
        # my_name="site-laptop"
        
        copy_and_delete_source(self, self.source_directory, self.target_directory, self.file_extensions, self.username, self.password, self.my_name, self.server_name)

        return data, meta

 

 

윈도우 경로에 파일이 저장되지 않는다면 방화벽 문제를 의심해봐야 한다.

방화벽 및 네트워크 보호 > 고급 설정 > 인바운드 규칙 에서 아래의 파일 및 프린터 공유(SMB-In)를 규칙 사용을 통해 활성화해야 한다.