기술정리/Python

파이썬 프로그램 만들기

bingual 2024. 6. 14. 16:39
반응형

Flet

https://flet.dev/

 

Build multi-platform apps in Python powered by Flutter | Flet

Build multi-platform apps in Python powered by Flutter.

flet.dev

 

`pyqt`, `tkinter` 등등 과 같이 파이썬 프로그램을 만들기 위해 사용된다.

flutter 기반으로 만들어졌기 때문에 웹/앱 어디서든 배포가 가능하다. 
단순하고 빠르게 개발이 가능하기 때문에 선택한 라이브러리이다.



들어가기 앞서

  • 가능하면 최신적이면서 어느 정도 검증되었고 성능, 유지보수, 생산 이점을 챙기는 라이브러리 버전을 채택한다. 
  • 중복제거를 최대한 피하고 유지보수를 용이하게 하기 위해서 `class` 별로 코드를 작성한다.
  • 가능하면 최대한 클린 코드를 지향한다. 따라서 파이썬 타입힌트를 빠짐없이 작성하거나 코드를 일관성 있게 유지하는 편이다.
  • 함수는 전부 비동기로 작성되어있다. 확장성을 위해 동기적으로 작동하는 함수여도 `async/await`를 명시적으로 사용한다.

`flet` 공식 문서에서도 `class`로 유지보수에 용이하게 끔 코드를 작성하도록 권장하고 있지만 `docs` 예제는 전부 함수형이다.

공식 문서에서 특정 위젯을 어떻게 사용해야하는지 자세하게 설명하고 있으므로 위젯 사용법에 대한 것은 다루지 않는다.

 

해당 포스팅은 어떠한 방법으로 `flet`을 더 효율적으로 사용하며 관리할 수 있는지 그리고 해당 프로그램을 어떻게 패키징 하는지에 대한 방법만을 다룬다.

 

코드 작성

하나의 폴더 아래의 파일들을 생성 해주도록 한다.

myprogram라는 이름의 최상의 디렉터리가 존재한다면 해당 디렉터리 하위에 `main.py`, `utils.py`, `components.py`을 생성한다.

 

`main.py`

import asyncio

import flet as ft

from myprogram.components import ImageUpdateComponent
from myprogram.utils import (
    setup_asyncio,
    setup_logging,
)

setup_asyncio()


async def main(page: ft.Page):
    page.title = "이미지 관리 프로그램"

    page.window.width = 700
    page.window.height = 700

    page.vertical_alignment = ft.MainAxisAlignment.CENTER
    page.horizontal_alignment = ft.CrossAxisAlignment.CENTER

    image_update = ImageUpdateComponent(page)

    page.add(
        ft.Column(
            controls=[
                *image_update.get_controls(),
            ],
            expand=True,
        )
    )


if __name__ == "__main__":
    setup_logging()
    asyncio.run(ft.app(target=main))


프로그램 시작 도입 부이다. 

 

가능하면 `main.py`의 함수는 위와 같이 간결하게 작성해 놓고 기능들을 다른 `.py` 파일의 `class`에서 `import` 해서 가져오도록 한다.

 


`utils.py`

if getattr(sys, "frozen", False):
    # test.exe로 실행한 경우,test.exe를 보관한 디렉토리의 full path를 취득
    BASE_DIR = Path(sys.executable).parent
else:
    # python test.py로 실행한 경우,test.py를 보관한 디렉토리의 full path를 취득
    BASE_DIR = Path(__file__).resolve().parent

os.chdir(BASE_DIR)

# 윈도우 환경에 맞게 끔 이벤트 루프 설정
def setup_asyncio() -> None:
    nest_asyncio.apply()
    asyncio.set_event_loop_policy(asyncio.WindowsProactorEventLoopPolicy())


def setup_datetime(date_format: str = "%Y-%m-%d_%H_%M") -> str:
    now = datetime.now()
    timestamp = now.strftime(date_format)
    return timestamp


def setup_logging() -> None:
    output_path = BASE_DIR / "logs"
    output_path.mkdir(parents=True, exist_ok=True)

    current_date = setup_datetime("%Y-%m-%d")
    log_filename = f"log_{current_date}.log"
    log_file = output_path / log_filename

    logging.basicConfig(
        filename=log_file,
        level=logging.WARNING,
        format="[%(asctime)s] [%(levelname)s %(filename)s:%(lineno)s] >>\n%(message)s",
        encoding="utf-8-sig",
    )


async def get_logger() -> logging.Logger:
    return logging.getLogger(__name__)


async def get_error_message() -> str:
    return traceback.format_exc()


async def download_and_save_image(
    image_url: str,
    output_path: Path,
    filename: str,
    target_size: Tuple[int, int],
) -> None:
    headers = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) "
        "Chrome/91.0.4472.124 Safari/537.36"
    }
    try:
        async with aiohttp.ClientSession() as session:
            async with session.get(image_url, headers=headers) as resp:
                if resp.status == 200:
                    image_data = BytesIO(await resp.read())
                    image = Image.open(image_data)
                    image = image.resize(target_size)

                    output_image_data = BytesIO()
                    image.save(output_image_data, format="PNG")
                    output_image_data.seek(0)

                    async with aiofiles.open(output_path / filename, mode="wb") as f:
                        await f.write(output_image_data.read())
                else:
                    message = f"이미지 다운로드 실패: '{filename}', '{image_url}', 상태코드: '{resp.status}'"
                    logger = await get_logger()
                    logger.error(message)
                    print(message)

    except Exception as e:
        message = f"이미지 저장 중 예외 발생: '{filename}'\n{await get_error_message()}"
        logger = await get_logger()
        logger.error(message)
        print(message)
        # raise e


async def read_data_info_excel_and_download_images(
    file_path: str,
    sheet_name: str = DEFAULT_DIR_NAME,
    target_size: Tuple[int, int] = (800, 800),
) -> None:

    df = pd.read_excel(file_path, sheet_name=sheet_name, engine="openpyxl")

    if not df.empty:
        timestamp = setup_datetime("%Y-%m-%d_%H_%M")

        output_path = BASE_DIR / "이미지 업데이트" / "이미지" / timestamp
        output_path.mkdir(parents=True, exist_ok=True)
        extension = ".png"

        for _, row in tqdm(
            df.iterrows(), total=len(df), desc=f"'{sheet_name}' 이미지 다운로드 중"
        ):

            image_src = row["이미지소스"]
            image_src = image_src.split(";\n")
            if isinstance(image_src, list) and len(image_src) > 1:
                for i, src in enumerate(image_src):
                    await download_and_save_image(
                        image_url=src,
                        output_path=output_path,
                        filename=f"{row['상품번호']}_{i + 1}{extension}",
                        target_size=target_size,
                    )
            else:
                await download_and_save_image(
                    image_url=row["이미지소스"],
                    output_path=output_path,
                    filename=f"{row['상품번호']}{extension}",
                    target_size=target_size,
                )

 

사용자 정의 함수들을 정의하는 파일이며 기본적으로 `flet`과는 직접적인 관계는 없다.


`setup_asyncio()`은 윈도우에서 비동기 실행을 위해 세팅을 하기 위한 커스텀 함수이다.

`setup_logging()`은 로그 관련 설정을 위한 커스텀 함수이다.
`read_data_info_excel_and_download_images`은 엑셀 파일을 읽어 들여 이미지를 다운로드하는 함수이다.

 

중요한 부분 중 하나는 `BASE_DIR` 설정 부분인데 `pyinstaller`로 패키징된 `.exe` 파일을 실행할 때와 IDE 혹은 터미널을 통해서 프로그램을 실행할 때 디렉터리의 `fullpath`가 달라지기 때문에 어떠한 환경에서 실행하느냐에 따라 설정할 필요가 있다.

 

`components.py`

class FletUtilComponent:
    def __init__(self, page: ft.Page):
        self.page = page

        self.dialog_text = ft.Text()
        self.dlg = ft.AlertDialog(title=self.dialog_text)

        self.progress_text = ft.Text(
            visible=False, value="작업이 진행되는 동안 창을 닫지 마세요."
        )
        self.progress_bar = ft.ProgressBar(height=10, visible=False, expand=True)

        self.cancel_button = ft.ElevatedButton(
            text="작업 중지",
            on_click=self.cancel_task,
            expand=True,
            visible=False,
        )

        self.task = None  # 작업을 추적할 Task 객체

    async def start_task(self, task_func, start_button):
        self.progress_text.visible = True
        self.progress_bar.visible = True
        self.cancel_button.visible = True

        start_button.disabled = True  # 시작 버튼 비활성화
        self.page.update()

        self.page.dialog = self.dlg
        self.task = asyncio.create_task(task_func())

        try:
            await self.task
            message = "완료되었습니다."
            print(message)
            self.dialog_text.value = message
            self.dlg.open = True

        except asyncio.CancelledError:
            message = "작업이 중지되었습니다."
            print(message)
            self.dialog_text.value = message
            self.dlg.open = True

        except Exception as e:
            message = "에러가 발생했습니다."
            print(message)
            self.dialog_text.value = message
            self.dlg.open = True

            logger = await get_logger()
            logger.error(await get_error_message())
            raise e

        finally:
            self.progress_text.visible = False
            self.progress_bar.visible = False
            self.cancel_button.visible = False

            start_button.disabled = False  # 작업 완료 후 시작 버튼 활성화
            self.page.update()

    def cancel_task(self, e):
        if self.task:
            self.task.cancel()


메인이 되는 컴포넌트 클래스다.

 

프로그램을 실행한 뒤 이후에 각 작업을 별도의 태스크로 관리한다.

 

해당 유틸 컴포넌트에서는 `태스크 생성`, `태스크 취소`, `진행표시줄`, `로깅` 을 관리하고 있으며 이는 자주 사용되어야만 하는 객체의 집합이므로 다른 클래스에서 상속받아서 재사용할 수 있도록 한다.

 

class ImageUpdateComponent(FletUtilComponent):
    def __init__(self, page: ft.Page):
        super().__init__(page)
        self.image_update_label = ft.Text("이미지 업데이트", size=30)

        self.pick_files_dialog = ft.FilePicker(on_result=self.pick_files_result)
        self.page.overlay.append(self.pick_files_dialog)

        self.selected_files = ft.TextField(
            label="엑셀 파일", expand=True, read_only=True, helper_text="읽기 전용"
        )
        self.initialize_file = ft.ElevatedButton(
            "파일 초기화",
            icon=ft.icons.CLEAR,
            on_click=self.clear_files,
        )

        self.upload_file = ft.FilledButton(
            "엑셀 업로드",
            icon=ft.icons.UPLOAD_FILE,
            on_click=lambda _: self.pick_files_dialog.pick_files(
                allow_multiple=False, allowed_extensions=["xlsx"]
            ),
            expand=True,
        )
        self.update_image = ft.FilledButton(
            "이미지 업데이트 시작",
            on_click=self.start_image_update,
            expand=True,
        )

    async def pick_files_result(self, e: ft.FilePickerResultEvent):
        if e.files:
            self.selected_files.value = e.files[0].path
            self.selected_files.update()

    async def clear_files(self, e):
        self.selected_files.value = ""
        self.selected_files.update()

    async def is_valid(self) -> str | None:
        selected_files = self.selected_files.value

        return selected_files

    async def image_update_task(self):
        await read_data_info_excel_and_download_images(
            file_path=self.selected_files.value,
        )

    async def start_image_update(self, e):
        if await self.is_valid():
            await asyncio.create_task(
                self.start_task(self.image_update_task, self.update_image)
            )
        else:
            await show_snack_bar(
                self.page, "엑셀 파일을 업로드해주세요.", ft.colors.RED_400
            )

    def get_controls(self):
        return [
            ft.Row(controls=[self.image_update_label]),
            ft.Row(controls=[self.selected_files, self.initialize_file]),
            ft.Row(controls=[self.upload_file, self.update_image]),
            ft.Row(controls=[self.cancel_button]),
            ft.Row(controls=[self.progress_text]),
            ft.Row(controls=[self.progress_bar]),
        ]

 

예를 들어서 CSV파일의 이미지소스 칼럼을 읽어 들여서 이미지를 저장하는 로직을 구성한다고 해보자 그렇다면 이미지 업데이트 컴포넌트 클래스에서는 `파일 업로드`, `이미지 저장` 로직만을 관리하고 나머지는 위에서 정의했던 유틸 컴포넌트의 `태스크 생성`, `태스크 취소`, `진행표시줄`, `로깅` 기능을 재정의해서 재사용한다.

 

이렇게 함으로써 중복을 피하고 유지보수에 용이하게 코드를 작성하여 관리한다면 코드의 확장 및 수정을 빠르게 할 수 있다.

 

https://flet.dev/blog/file-picker-and-uploads/

업로드 관련한 기본적인 로직은 위에 공식 문서 링크에서 확인할 수 있다.

물론 나의 경우 해당 로직을 참고해서 다른 몇 가지 기능들을 추가해놓긴 했다.

 

 

프로그램 실행

 

main.py을 IDE 혹은 터미널로 실행한다면 위와 같은 GUI를 만나볼 수가 있다.

 

샘플.xlsx
0.04MB

 

 

playwright를 사용해 데이터를 추출한 엑셀 샘플이다. 엑셀 번호에 맞게 끔 해당 상품 이미지를 다운로드해주는 것을 확인 가능하다.

 

 

파이썬 프로그램 패키징 하기

 

pipenv run pyinstaller -c -F -n myprogram ./myprogram/main.py

 

파이썬은 인터프리터 언어이기 때문에 패키징 하는 게 매우 간단하다.

venv. pyenv 등의 가상환경을 사용해도 되지만 나는 파이썬에서 공식적으로 권장하는 가상 환경 라이브러리인 pipenv를 사용하기 때문에 해당 명령어 기준으로 설명한다.

pyinstaller를 설치해 준 다음 프로그램 진입 시작점이 되는 `main.py`을 위에 명령어를 통해서 경로 지정 후 빌드 해주기만 한 하면 끝이 난다.

 

-c, -F, -n은 차례대로 터미널 표시 여부, 하나의 파일로 패키징 여부, 패키징 파일 이름 설정 옵션이다.

 

 

마무리

크롤링 & 스크랩핑 하는 코드를 다듬어서 `flet`에 적용 할 수 도 있고 여러장의 이미지를 하나로 합쳐주는 프로그램도 비슷하게 작성할 수 있겠다.

 

pandas, openpyxl을 다듬어서 다른 방식으로 엑셀 파일을 생성할 수 도 있고 새 칼럼을 추가하는 방법도 있다. 또한 비동기적인 방법으로 파일을 다운로드하는 게 아니라 동기적으로 다운로드를 할 수 도 있고 임시파일에 저장해 놨다가 한꺼번에 파일을 다운로드할 수 도 있다.

 

다만 해당 포스팅에서는 깊게 다루지 않는다.

 

해당 포스팅의 모든 코드들은 온전히 나만의 스타일대로 작성된 코드 이기 때문에 더 좋은 대안이 되는 방법이 있을 수 있다.