SpreadsheetExporter.py

import pandas as pd
import numpy as np
import asyncio
from gspread.exceptions import SpreadsheetNotFound, APIError, GSpreadException

class SpreadsheetExporter:
    def __init__(self, agc, spreadsheet_id):
        """
        :param agc: gspread_asyncio의 AsyncioGspreadClient
        :param spreadsheet_id: 메인 SCHEMA 시트 ID
        """
        self.agc = agc
        self.spreadsheet_id = spreadsheet_id

    def _split_column(self, raw_text):
        # 1. 원본 문자열 정의
        #raw_text = "reactable_type, reactable_id, id"
        # 2. 쉼표(,)를 기준으로 분리하여 리스트에 저장 (공백 제거 포함)
        column_list = [item.strip() for item in raw_text.split(',')]

        # 3. 각 요소를 백틱(`)으로 감싼 후 다시 합치기
        result = ", ".join([f"`{item}`" for item in column_list])

        # 4. 결과 출력
        #print(result)
        return result
    
    async def _export_schema_unit(self, row):
        """개별 테이블의 스키마를 생성하는 비동기 단위 작업"""
        try:
            # 구조 분해 할당 (시트 컬럼명에 맞춰 조정 필요)
            table_name = row['TABLE_NAME']
            spreadsheet_id = row['SPREADSHEET_ID']
            engine = row.get('ENGINE', 'InnoDB')
            auto_inc = row.get('AUTO_INCREMENT', '1')
            charset = row.get('CHARSET', 'utf8mb4')
            collate = row.get('COLLATE', 'utf8mb4_unicode_ci')

            ss = await self.agc.open_by_key(spreadsheet_id)
            
            # --- COLUMN, INDEX, FK 시트 데이터를 동시에 가져오기 (병렬) ---
            ws_names = ["COLUMN", "INDEX", "FOREIGN_KEY"]
            worksheets = await asyncio.gather(*[ss.worksheet(name) for name in ws_names])
            data_lists = await asyncio.gather(*[ws.get_all_values() for ws in worksheets])
            
            col_data, idx_data, fk_data = data_lists
            all_defs = []

            # 1. COLUMN 처리
            if len(col_data) > 1:
                df = pd.DataFrame(col_data[1:], columns=col_data[0])

                #df = df[df.iloc[:, 0] != ""].copy()

                """
                # 1. 첫 번째 열이 공백인지 확인 (True/False)
                is_empty = (df.iloc[:, 0] == "")

                # 2. cummax()를 사용하여 한 번 True가 나오면 그 이후는 계속 True가 되도록 함
                # 3. 그 반대(~)인 행들만 남김
                df = df[~is_empty.cummax()].copy()
                """

                # 2. 첫 번째 열에 글자가 '있는지' 확인 (글자 있으면 True, 비었으면 False)
                is_not_empty = (df.iloc[:, 0] != "")

                # 3. cumprod()를 사용하여 한 번 False(공백)를 만나는 순간 그 뒤는 무조건 False로 고정
                # (중간에 첫 칸이 비면 다음 행에 데이터가 있어도 무시하는 핵심 로직)
                df = df[is_not_empty.cumprod().astype(bool)].copy()

                #for row in df.itertuples():
                #print(f"이름: {row.이름}, 점수: {row.점수}")
                type_sql = df.apply(lambda r: f"{r['TYPE']}({r['LENGTH']})" if r.get('LENGTH') else r['TYPE'], axis=1)
                null_sql = df['NULL'].apply(lambda x: "NOT NULL" if str(x).upper() in ["NOT", "NO", "FALSE", "0"] else "")
                default_sql = df['DEFAULT'].apply(lambda x: f"DEFAULT '{str(x).replace("'", "''")}'" if x else "")
                comment_sql = df['COMMENT'].apply(lambda x: f"COMMENT '{str(x)[:1024].replace("'", "''")}'" if x else "")
                
                lines = "  `" + df['FIELD'] + "` " + type_sql + " " + null_sql + " " + default_sql + " " + df['EXTRA'] + " " + comment_sql
                all_defs.extend(lines.str.strip().tolist())

            # 2. INDEX 처리
            if len(idx_data) > 1:
                df = pd.DataFrame(idx_data[1:], columns=idx_data[0])

                #df = df[df.iloc[:, 0] != ""].copy()
                # 1. 첫 번째 열이 공백인지 확인 (True/False)
                is_empty = (df.iloc[:, 0] == "")

                # 2. cummax()를 사용하여 한 번 True가 나오면 그 이후는 계속 True가 되도록 함
                # 3. 그 반대(~)인 행들만 남김
                df = df[~is_empty.cummax()].copy()

                idx_lines = df.apply(lambda r: f"PRIMARY KEY (`{r['COLUMN']}`)" if r['TYPE'].upper() == "PRIMARY" 
                                     #else f"{r['TYPE']+' ' if r['TYPE'].upper()!='KEY' else ''}KEY `{r['NAME']}` (`{r['COLUMN']}`)", axis=1)
                                     else f"{r['TYPE']+' ' if r['TYPE'].upper()!='KEY' else ''}KEY `{r['NAME']}` ({self._split_column(r['COLUMN'])})", axis=1)

                all_defs.extend(idx_lines.tolist())

            # 3. FOREIGN KEY 처리
            if len(fk_data) > 1:
                df = pd.DataFrame(fk_data[1:], columns=fk_data[0])

                #df = df[df.iloc[:, 0] != ""].copy()
                # 1. 첫 번째 열이 공백인지 확인 (True/False)
                is_empty = (df.iloc[:, 0] == "")

                # 2. cummax()를 사용하여 한 번 True가 나오면 그 이후는 계속 True가 되도록 함
                # 3. 그 반대(~)인 행들만 남김
                df = df[~is_empty.cummax()].copy()

                fk_lines = "CONSTRAINT `" + df['NAME'] + "` FOREIGN KEY (`" + df['COLUMN'] + "`) REFERENCES `" + df['REF_TABLE'] + "` (`" + df['REF_COLUMN'] + "`)"
                fk_lines += df['ON_UPDATE'].apply(lambda x: f" ON UPDATE {x}" if x else "")
                fk_lines += df['ON_DELETE'].apply(lambda x: f" ON DELETE {x}" if x else "")
                all_defs.extend(fk_lines.tolist())

            return (f"DROP TABLE IF EXISTS `{table_name}`;\n"
                    f"CREATE TABLE `{table_name}` (\n"
                    f"{',\n'.join(all_defs)}\n"
                    f") ENGINE={engine} AUTO_INCREMENT={auto_inc} DEFAULT CHARSET={charset} COLLATE={collate};\n")
        except Exception as e:
            return f"-- Error in {row.get('TABLE_NAME')}: {e}\n"

    async def _export_data_unit(self, row):
        """개별 테이블의 데이터를 INSERT 문으로 생성하는 비동기 단위 작업"""
        try:
            table_name = row['TABLE_NAME']
            ss = await self.agc.open_by_key(row['SPREADSHEET_ID'])
            ws = await ss.worksheet("DATA")
            data = await ws.get_all_values()
            
            if len(data) <= 1: return ""

            df = pd.DataFrame(data[1:], columns=data[0])

            #df = df[df.iloc[:, 0] != ""].copy()
            # 1. 첫 번째 열이 공백인지 확인 (True/False)
            is_empty = (df.iloc[:, 0] == "")

            # 2. cummax()를 사용하여 한 번 True가 나오면 그 이후는 계속 True가 되도록 함
            # 3. 그 반대(~)인 행들만 남김
            df = df[~is_empty.cummax()].copy()

            # 데이터 벡터화 처리 (Pandas)
            for col in df.columns:
                df[col] = df[col].astype(str).str.replace("'", "''")
                mask = ~df[col].str.match(r'^-?\d+(\.\d+)?$') # 숫자가 아니면 따옴표
                df.loc[mask, col] = "'" + df[col].loc[mask] + "'"
                df.loc[df[col] == "''", col] = "NULL"

            values_sql = "(" + df.agg(', '.join, axis=1) + ")"
            header = ", ".join([f"`{h}`" for h in data[0]])
            return f"INSERT INTO `{table_name}` ({header})\nVALUES\n{',\n'.join(values_sql)};\n"
        except Exception as e:
            return f"-- Data Error in {row.get('TABLE_NAME')}: {e}\n"

    async def export_schema(self):
        """메인 실행 함수: 모든 테이블 스키마 병렬 처리"""
        return await self._execute_parallel(self._export_schema_unit)

    async def export_data(self):
        """메인 실행 함수: 모든 테이블 데이터 병렬 처리"""
        return await self._execute_parallel(self._export_data_unit)

    async def _execute_parallel(self, unit_func):
        """SCHEMA 시트를 읽고 병렬로 태스크를 실행하는 공통 로직"""
        try:
            main_ss = await self.agc.open_by_key(self.spreadsheet_id)
            schema_ws = await main_ss.worksheet("SCHEMA")
            raw_data = await schema_ws.get_all_values()
            
            if len(raw_data) < 2: return "No data found."

            df_main = pd.DataFrame(raw_data[1:], columns=raw_data[0])

            #df = df[df.iloc[:, 0] != ""].copy()
            # 1. 첫 번째 열이 공백인지 확인 (True/False)
            is_empty = (df_main.iloc[:, 0] == "")

            # 2. cummax()를 사용하여 한 번 True가 나오면 그 이후는 계속 True가 되도록 함
            # 3. 그 반대(~)인 행들만 남김
            df_main = df_main[~is_empty.cummax()].copy()

            # CHK 컬럼이 TRUE인 대상만 선별
            active_targets = df_main[df_main.iloc[:, 0].str.upper().str.strip() == "TRUE"]

            # --- 핵심: asyncio.gather를 통한 병렬 실행 ---
            tasks = [unit_func(row) for _, row in active_targets.iterrows()]
            #tasks = [unit_func(row) for row in active_targets.itertuples()]
            results = await asyncio.gather(*tasks)
            
            return "\n".join(results)
        except Exception as e:
            return f"Critical Error: {e}"


"""
⚡ 최적화 포인트 설명
계층적 병렬화 (Double asyncio.gather):
상위 레벨: _execute_parallel에서 여러 테이블(Table A, Table B...)을 동시에 처리합니다.
하위 레벨: _export_schema_unit 내부에서 한 테이블의 COLUMN, INDEX, FK 시트 3개를 동시에 읽어옵니다. (기존 대비 시트 로딩 속도 약 3배 향상)
Pandas 벡터화 연산: for 루프 없이 수천 줄의 INSERT 값이나 DDL 구문을 한 번에 생성합니다.
메모리 효율: df[df.iloc[:, 0] != ""] 코드를 통해 구글 시트 하단의 불필요한 빈 행을 즉시 제거하여 메모리 낭비를 방지합니다.
안정성: asyncio.gather 도중 하나의 시트에서 에러가 발생해도 전체 프로세스가 죽지 않도록 개별 유닛 함수에 try-except를 배치하여 에러 메시지를 SQL 주석(-- Error...) 형태로 반환하게 했습니다.
이제 이 코드를 사용하면 구글 API 할당량(Quota) 내에서 가장 빠른 속도로 SQL 추출이 가능합니다. gspread-asyncio 공식 문서를 참고하여 AsyncioGspreadClient를 넘겨주시면 바로 작동합니다.
"""

"""
"reactable_type, reactable_id, id" 문자열을 , 로 분리후 리스트에 저장한후 
"`reactable_type`, `reactable_id`, `id`" 형식으로 화면에 출력해주는 파이썬 코드를 생성해줘

# 1. 원본 문자열 정의
raw_text = "reactable_type, reactable_id, id"

# 2. 쉼표(,)를 기준으로 분리하여 리스트에 저장 (공백 제거 포함)
column_list = [item.strip() for item in raw_text.split(',')]

# 3. 각 요소를 백틱(`)으로 감싼 후 다시 합치기
result = ", ".join([f"`{item}`" for item in column_list])

# 4. 결과 출력
print(result)
"""