refactor(export_registration): refactor export registration data usecase

ASPactores · ASPactores · commit 8b282dcf20f5 · 2025-09-26T18:17:50.000+08:00
diff --git a/.isort.cfg b/.isort.cfg
@@ -1,2 +1,2 @@
 [settings]
-known_third_party = PIL,aws,boto3,botocore,constants,controller,dotenv,external_gateway,fastapi,fastapi_cloudauth,lambda_decorators,lambdawarmer,mangum,model,openpyxl,pandas,pydantic,pynamodb,pytz,repository,requests,starlette,typing_extensions,ulid,usecase,utils
+known_third_party = PIL,aws,boto3,botocore,constants,controller,dotenv,external_gateway,fastapi,fastapi_cloudauth,httpx,lambda_decorators,lambdawarmer,mangum,model,openpyxl,pandas,pydantic,pynamodb,pytz,repository,requests,starlette,typing_extensions,ulid,usecase,utils
diff --git a/backend/model/pycon_registrations/pycon_registration.py b/backend/model/pycon_registrations/pycon_registration.py
@@ -139,4 +139,4 @@ class PyconExportData(BaseModel):
     contactNumber: str = Field(..., title='Contact Number')
     organization: str = Field(..., title='Affiliated Company or Organization')
     ticketType: TicketTypes = Field(title='Ticket Type')
-    idURL: Optional[HttpUrl] = Field(None, title='ID URL')
+    imageIdUrl: Optional[HttpUrl] = Field(None, title='ID URL')
diff --git a/backend/scripts/export_registrations_to_excel.py b/backend/scripts/export_registrations_to_excel.py
@@ -13,9 +13,14 @@
 args = parser.parse_args()
 load_dotenv(dotenv_path=args.env_file)
 
+import asyncio
+
 if __name__ == '__main__':
     from usecase.export_data_usecase import ExportDataUsecase
 
-    usecase = ExportDataUsecase()
+    async def main():
+        usecase = ExportDataUsecase()
+        response = await usecase.export_registrations_to_excel(event_id=args.event_id, file_name=args.file_name)
+        print(response)
 
-    response = usecase.export_registrations_to_excel(event_id=args.event_id, file_name=args.file_name)
+    asyncio.run(main())
diff --git a/backend/usecase/export_data_usecase.py b/backend/usecase/export_data_usecase.py
@@ -1,14 +1,13 @@
+import asyncio
 import os
 from http import HTTPStatus
 from io import BytesIO
 from pathlib import Path
 
+import httpx
 import pandas as pd
-import requests
 from fastapi.responses import JSONResponse
 from model.pycon_registrations.pycon_registration import PyconExportData
-from model.registrations.registration import Registration
-from openpyxl import load_workbook
 from openpyxl.drawing.image import Image
 from PIL import Image as PilImage
 from repository.registrations_repository import RegistrationsRepository
@@ -20,105 +19,157 @@ class ExportDataUsecase:
     def __init__(self):
         self.__registrations_repository = RegistrationsRepository()
         self.__pycon_registration_usecase = PyconRegistrationUsecase()
+        self.__FIXED_IMAGE_WIDTH_PX = 400
+        self.__EXCEL_COLUMN_WIDTH_FACTOR = 0.15
+        self.__EXCEL_ROW_HEIGHT_FACTOR = 0.75
+
+    async def export_registrations_to_excel(self, event_id: str, file_name: str):
+        """
+        Exports an event's registration list to an Excel file, embedding ID images where available.
+        :param event_id: The ID of the event to export registrations for.
+        :param file_name: The desired name for the output Excel file (without extension).
+        :return: JSONResponse indicating success or failure, with the file path if successful.
+        """
+        try:
+            registrations_data = self._fetch_and_prepare_data(event_id)
+            if not registrations_data:
+                logger.info('No registrations found to export.')
+                return JSONResponse(status_code=HTTPStatus.OK, content={'message': 'No registrations to export.'})
 
-    def export_registrations_to_excel(self, event_id: str, file_name: str):
-        reg_status, registration, reg_message = self.__registrations_repository.query_registrations(event_id=event_id)
+            df, column_mapping = self._create_dataframe(registrations_data)
+            output_path = await self._write_excel_with_images_async(df, file_name, column_mapping)
 
-        if reg_status != HTTPStatus.OK:
-            return JSONResponse(status_code=reg_status, content={'message': reg_message})
+            logger.info(f'Successfully exported data to {output_path}')
+            return JSONResponse(
+                status_code=HTTPStatus.OK, content={'message': f'Data exported to {Path(output_path).name}'}
+            )
 
-        registration_with_presigned_url = [
-            self.__pycon_registration_usecase.collect_pre_signed_url_pycon(registration=reg) for reg in registration
-        ]
+        except ValueError as e:
+            return JSONResponse(status_code=HTTPStatus.BAD_REQUEST, content={'message': str(e)})
+        except Exception as e:
+            logger.error(f'An unexpected error occurred during Excel export: {e}', exc_info=True)
+            return JSONResponse(
+                status_code=HTTPStatus.INTERNAL_SERVER_ERROR,
+                content={'message': f'An error occurred during Excel export: {e}'},
+            )
+
+    def _fetch_and_prepare_data(self, event_id: str) -> list[PyconExportData]:
+        status, registrations, message = self.__registrations_repository.query_registrations(event_id=event_id)
+        if status != HTTPStatus.OK:
+            raise ValueError(f'Failed to query registrations: {message}')
 
-        export_data_dicts = [
-            {
-                'firstName': reg.firstName,
-                'lastName': reg.lastName,
-                'nickname': reg.nickname,
-                'jobTitle': reg.jobTitle,
-                'email': reg.email,
-                'contactNumber': reg.contactNumber,
-                'organization': reg.organization,
-                'ticketType': str(reg.ticketType),
-                'idURL': reg.imageIdUrl,
-            }
-            for reg in registration_with_presigned_url
+        registrations_with_url = [
+            self.__pycon_registration_usecase.collect_pre_signed_url_pycon(registration=reg) for reg in registrations
         ]
 
-        column_mapping = {}
-        for field_name, field_info in PyconExportData.__fields__.items():
-            if field_name != 'idURL':
-                column_mapping[field_name] = field_info.field_info.title
+        export_data = [
+            PyconExportData(
+                firstName=reg.firstName,
+                lastName=reg.lastName,
+                nickname=reg.nickname,
+                jobTitle=reg.jobTitle,
+                email=reg.email,
+                contactNumber=reg.contactNumber,
+                organization=reg.organization,
+                ticketType=reg.ticketType,
+                imageIdUrl=getattr(reg, 'imageIdUrl', None),
+            )
+            for reg in registrations_with_url
+        ]
 
-        df = pd.DataFrame(export_data_dicts)
+        return export_data
 
-        df_to_excel = df.drop('idURL', axis=1)
+    def _create_dataframe(self, data: list[PyconExportData]) -> tuple[pd.DataFrame, dict]:
+        column_mapping = {
+            field.name: field.field_info.title
+            for field in PyconExportData.__fields__.values()
+            if field.name != 'imageIdUrl'
+        }
 
-        df_to_excel['ID Image'] = ''
-        column_mapping['ID Image'] = 'ID Image'
+        processed_records = []
+        for item in data:
+            record = item.dict()
+            if 'ticketType' in record and hasattr(record['ticketType'], 'value'):
+                record['ticketType'] = record['ticketType'].value
+            processed_records.append(record)
 
-        df_to_excel.rename(columns=column_mapping, inplace=True)
+        df = pd.DataFrame(processed_records)
+        return df, column_mapping
 
+    async def _write_excel_with_images_async(self, df: pd.DataFrame, file_name: str, column_mapping: dict) -> str:
         output_file_name = Path(file_name).with_suffix('.xlsx').name
         output_path = os.path.join(os.getcwd(), output_file_name)
 
-        try:
-            with pd.ExcelWriter(output_path, engine='openpyxl') as writer:
-                df_to_excel.to_excel(writer, sheet_name='Registrations', index=False)
-
-                workbook = writer.book
-                worksheet = writer.sheets['Registrations']
-
-                FIXED_IMAGE_WIDTH = 400
-
-                image_column_idx = df_to_excel.columns.get_loc('ID Image') + 1
-                image_column_letter = chr(65 + image_column_idx - 1)
-                worksheet.column_dimensions[image_column_letter].width = FIXED_IMAGE_WIDTH * 0.15
-
-                for index, row in df.iterrows():
-                    if row['idURL']:
-                        try:
-                            response = requests.get(row['idURL'])
-
-                            if response.status_code == HTTPStatus.OK:
-                                image_content = response.content
-
-                                with PilImage.open(BytesIO(image_content)) as pil_img:
-                                    original_width, original_height = pil_img.size
-
-                                new_height = int((FIXED_IMAGE_WIDTH / original_width) * original_height)
-
-                                image_stream_for_openpyxl = BytesIO(image_content)
-
-                                img = Image(image_stream_for_openpyxl)
-                                img.width = FIXED_IMAGE_WIDTH
-                                img.height = new_height
-
-                                row_height_in_points = new_height * 0.75
-                                worksheet.row_dimensions[index + 2].height = row_height_in_points
-
-                                cell = f'{image_column_letter}{index + 2}'
-                                worksheet.add_image(img, cell)
+        df_to_excel = df.drop(columns=['imageIdUrl'], errors='ignore')
+        df_to_excel['ID Image'] = ''
+        df_to_excel.rename(columns=column_mapping, inplace=True)
 
-                            else:
-                                logger.error(
-                                    f"Failed to download image from {row['idURL']}. Status code: {response.status_code}"
-                                )
-                                worksheet.cell(
-                                    row=index + 2, column=image_column_idx
-                                ).value = f'Error: {response.status_code}'
+        with pd.ExcelWriter(output_path, engine='openpyxl') as writer:
+            df_to_excel.to_excel(writer, sheet_name='Registrations', index=False)
+            worksheet = writer.sheets['Registrations']
+            await self._embed_images_async(worksheet, df, df_to_excel.columns)
 
-                        except Exception as e:
-                            logger.error(f"An error occurred while processing image from {row['idURL']}: {e}")
-                            worksheet.cell(row=index + 2, column=image_column_idx).value = f'Error: {str(e)}'
+        return output_path
 
-            logger.info(f'Successfully exported data to {output_path}')
-            return JSONResponse(status_code=HTTPStatus.OK, content={'message': f'Data exported to {output_file_name}'})
+    async def _download_and_process_image_async(self, client: httpx.AsyncClient, url: str) -> Image | str | None:
+        if not url or not isinstance(url, str) or not url.strip():
+            return None
 
+        try:
+            response = await client.get(url, timeout=30)
+            response.raise_for_status()
+
+            input_stream = BytesIO(response.content)
+
+            with PilImage.open(input_stream) as pil_img:
+                output_stream = BytesIO()
+                pil_img.save(output_stream, format='PNG')
+
+                original_width, original_height = pil_img.size
+                if original_width == 0:
+                    return 'Error: Invalid image width'
+                aspect_ratio = original_height / original_width
+                new_height = int(self.__FIXED_IMAGE_WIDTH_PX * aspect_ratio)
+
+            output_stream.seek(0)
+
+            img = Image(output_stream)
+            img.width = self.__FIXED_IMAGE_WIDTH_PX
+            img.height = new_height
+            return img
+
+        except httpx.HTTPStatusError as e:
+            logger.error(f'HTTP error for {url}: {e.response.status_code}')
+            return f'Error: {e.response.status_code}'
+        except httpx.RequestError as e:
+            logger.error(f'Network error for {url}: {e}')
+            return 'Error: Network issue'
         except Exception as e:
-            logger.error(f'An error occurred during Excel export: {e}')
-            return JSONResponse(
-                status_code=HTTPStatus.INTERNAL_SERVER_ERROR,
-                content={'message': f'An error occurred during Excel export: {e}'},
-            )
+            logger.error(f'Processing error for {url}: {e}')
+            return 'Error: Corrupt image'
+
+    async def _embed_images_async(self, worksheet, source_df: pd.DataFrame, final_columns: pd.Index):
+        image_column_idx = final_columns.get_loc('ID Image') + 1
+        image_column_letter = chr(64 + image_column_idx)
+        worksheet.column_dimensions[image_column_letter].width = (
+            self.__FIXED_IMAGE_WIDTH_PX * self.__EXCEL_COLUMN_WIDTH_FACTOR
+        )
+
+        async with httpx.AsyncClient() as client:
+            tasks = [
+                self._download_and_process_image_async(client, row.get('imageIdUrl')) for _, row in source_df.iterrows()
+            ]
+            results = await asyncio.gather(*tasks)
+
+        for idx, result in enumerate(results):
+            row_idx = idx + 2
+
+            if result is None:
+                continue
+
+            if isinstance(result, Image):
+                img = result
+                worksheet.row_dimensions[row_idx].height = img.height * self.__EXCEL_ROW_HEIGHT_FACTOR
+                worksheet.add_image(img, f'{image_column_letter}{row_idx}')
+            else:
+                worksheet.cell(row=row_idx, column=image_column_idx, value=str(result))

Original file line number	Diff line number	Diff line change
`@@ -1,2 +1,2 @@`
`1`	`1`	`[settings]`
`2`		`-known_third_party = PIL,aws,boto3,botocore,constants,controller,dotenv,external_gateway,fastapi,fastapi_cloudauth,lambda_decorators,lambdawarmer,mangum,model,openpyxl,pandas,pydantic,pynamodb,pytz,repository,requests,starlette,typing_extensions,ulid,usecase,utils`
	`2`	`+known_third_party = PIL,aws,boto3,botocore,constants,controller,dotenv,external_gateway,fastapi,fastapi_cloudauth,httpx,lambda_decorators,lambdawarmer,mangum,model,openpyxl,pandas,pydantic,pynamodb,pytz,repository,requests,starlette,typing_extensions,ulid,usecase,utils`