0. Summary

분석 데이터 전처리 및 검수 Tool 요약

목적

• 비정형 평가 로그를 분석 가능한 표준 데이터(Tidy Data)로 자동 변환하고, 평가자 간 의견 충돌이 발생한 지점(이상치)을 시각화하여 데이터 검수 효율을 극대화함

  핵심 기능

• 동적 데이터 정규화(Normalization): Wide Format의 로그를 분석 최적화된 Long Format으로 자동 전환하며, 가변적인 컬럼 구조에도 에러 없이 대응
• 마스킹 자동 해제(De-identification): 익명화된 모델 ID와 실제 모델 정보를 매핑 테이블을 통해 결합하여 분석의 데이터 정합성 확보
• 통계적 이상치 탐지(Outlier Detection): 동일 문항 내 점수 편차($\text{Max} - \text{Min} \ge 3$)를 기반으로 재검토가 필요한 문항을 실시간 선별
• 시각적 검수 자동화(Highlighting): 구글 시트 API 연동을 통해 이상치가 발생한 셀의 최댓값/최솟값을 자동으로 컬러링하여 가독성 증대

  기대 효과

• 데이터 클리닝 가속화: 수작업으로 진행하던 방대한 양의 전처리 과정을 자동화하여 분석 준비 시간 단축
• 품질 관리(QA) 정밀도 향상: 단순 일치도 확인을 넘어, 실제 갈등 지점을 시각적으로 즉각 파악하여 데이터 신뢰도 제고
• 확장성 있는 파이프라인: 다양한 평가 양식과 프로젝트 스키마에 유연하게 적용 가능한 견고한 전처리 구조 구축 (평가 지표명, 로그 상의 Column 명 등이 바뀌어도 대응할 수 있음)

1. Input Format

xlsx 포맷

통합 문서 구성

• 각각의 영역(Category) 별로 시트가 분리되어 있음

  평가 결과 로그

• 해당 영역(Category)에 속한 세부영역(Sub-category)에 해당하는 문항이 포함
• ID,script_itn,file_size와 같은 평가에 사용한 모델 음성 관련 정보 포함
• (모델 명): 응답열에 각 모델별 출력 응답, (모델 명): (평가항목)열에 각 모델별 출력 응답에 대한 평가자의 평가 점수가 위치, 상세 사유는 (모델 명): 비고열에 기록 위치함
• (모델 명): 이상여부에는 평가자가 모델 오류로 판단한 경우 표시하는 체크 박스의 값을 표시

2. 평가 결과 일치도 확인 툴

0) 필수 라이브러리 설치

import sys
import subprocess
import importlib

#@title 필수 라이브러리 목록 (pip 설치명 : 파이썬 모듈명)
required_packages = {
    'gspread': 'gspread',
    'gspread-dataframe': 'gspread_dataframe',
    'gspread-formatting': 'gspread_formatting',
    'statsmodels': 'statsmodels',
    'ipywidgets': 'ipywidgets'
}

print("--- 필수 라이브러리 점검 시작 ---")
for package_name, module_name in required_packages.items():
    try:
        importlib.import_module(module_name)
        print(f"[OK] {package_name} 라이브러리가 이미 설치되어 있습니다.")
    except ImportError:
        print(f"[INSTALL] {package_name} 모듈을 찾을 수 없어 설치를 진행합니다...")
        # 파이썬 내부에서 pip 명령어를 안전하게 호출하여 조용히(-q) 설치
        subprocess.check_call([sys.executable, "-m", "pip", "install", "-q", package_name])
        print(f" -> {package_name} 설치 완료.")
print("--- 필수 라이브러리 점검 완료 ---\n")

# 데이터 처리 및 GUI용 라이브러리 임포트
import pandas as pd
import numpy as np
import io
import ipywidgets as widgets
from IPython.display import display
from statsmodels.stats.inter_rater import fleiss_kappa

# 구글 시트 연동용 라이브러리 임포트
from google.colab import auth
from google.auth import default
import gspread
from gspread_dataframe import set_with_dataframe
from gspread_formatting import *

# Google 계정 인증 (실행 시 팝업창을 통해 권한 허용 필요)
try:
    auth.authenticate_user()
    creds, _ = default()
    gc = gspread.authorize(creds)
    print("Google 계정 인증이 완료되었습니다.")
except Exception as e:
    print(f"인증 과정에서 오류가 발생했습니다: {e}")

필수 라이브러리 목록

• gspread : Google Sheets API를 사용하여 구글 스프레드시트를 생성, 데이터 접근 및 제어 기능 제공
• gspread-dataframe : Pandas의 데이터프레임(DataFrame) 객체를 구글 시트로 즉시 전송, 또는 반대로 시트의 데이터를 데이터프레임으로 변환하는 기능 제공
• gspread-formatting : 구글 시트 내의 셀 배경색, 텍스트 스타일 등 조건부 서식을 파이썬 코드로 조작
• statsmodels : 고급 통계 모델링 및 검정 기능을 제공(Fleiss-kappa 지수를 연산을 위해 사용)
• ipywidgets : Google Colab 환경에서 텍스트 입력창, 파일 업로드 버튼, 실행 버튼 등 사용자가 직접 조작할 수 있는 대화형 GUI 인터페이스 제공

라이브러리 설치 여부 확인

print("--- 필수 라이브러리 점검 시작 ---")
for package_name, module_name in required_packages.items():
    try:
        importlib.import_module(module_name)
        print(f"[OK] {package_name} 라이브러리가 이미 설치되어 있습니다.")
    except ImportError:
        print(f"[INSTALL] {package_name} 모듈을 찾을 수 없어 설치를 진행합니다...")
        # 파이썬 내부에서 pip 명령어를 안전하게 호출하여 조용히(-q) 설치
        subprocess.check_call([sys.executable, "-m", "pip", "install", "-q", package_name])
        print(f" -> {package_name} 설치 완료.")
print("--- 필수 라이브러리 점검 완료 ---\n")

• 라이브러리 목록에 있는 라이브러리를 import문이 아닌 importlib.import_module() 함수를 통해 호출해, 현재 런타임 상에 존재하는지 확인
• 해당 호출에서 에러가 발생할 경우 try-except 구문을 활용해 내부 프로세스 호출을 통한 자동 설치로 분기
• 별도의 !pip install 명령어 없이 동적으로 미설치된 라이브러리 항목 설치를 진행

내부 프로세스 설치 과정 상세

subprocess.check_call([sys.executable, "-m", "pip", "install", "-q", package_name])

• subprocess : 파이썬 내부의 시스템 명령어를 실행하는 모듈
  • 종료 대기 여부를 설정할 수 있어 향후 비동기 기능 구현에 사용될 수 있음
• sys.executable : 실행 중인 파이썬 인터프리터의 절대 결로를 호출
• check_call : 입력한 명령어 실행이 완료될 때까지 메인 프로그램 실행을 일시정지(Blocking) → 정상적으로 실행이 완료된 이후 메인 프로그램 실행
• "-q" : 설치 과정에서 출력되는 로그를 출력하지 않도록 함

1) 입력 GUI 및 파일 업로드

#@title 1. 입력 GUI 및 파일 업로드 폼 구성
import pandas as pd
import io
import ipywidgets as widgets
from IPython.display import display

# 시트 접두사 입력 위젯 추가
prefix_input = widgets.Text(value='', description='시트 접두사:', placeholder='예: SLM_번역_1차_')
id_input = widgets.Text(value='id', description='ID 컬럼:')
rater_input = widgets.Text(value='tester', description='작업자 컬럼:')
# 그룹 컬럼 기본값을 'category, sub_category'로 변경
group_input = widgets.Text(value='category, sub_category', description='그룹 컬럼(쉼표 구분):', layout=widgets.Layout(width='50%'))
metric_input = widgets.Text(value='유창성, 정확성', description='평가지표 키워드(쉼표 구분):', layout=widgets.Layout(width='50%'))
threshold_input = widgets.FloatText(value=3.0, description='이상치 임계값:')

log_upload_widget = widgets.FileUpload(accept='.csv, .xlsx', multiple=False, description='로그 파일')
mask_upload_widget = widgets.FileUpload(accept='.csv, .xlsx', multiple=False, description='마스킹 파일')

load_button = widgets.Button(description='데이터 로드', button_style='info')
output_log = widgets.Output()

# GUI 화면 배치
display(prefix_input, id_input, rater_input, group_input, metric_input, threshold_input)
display(widgets.HBox([log_upload_widget, mask_upload_widget]), load_button, output_log)

df_raw = pd.DataFrame()
df_masking_raw = pd.DataFrame()

def load_file_to_df(upload_data):
    if isinstance(upload_data, dict):
        filename = list(upload_data.keys())[0]
        content = upload_data[filename]['content']
    else:
        uploaded_file = upload_data[0]
        filename = uploaded_file['name']
        content = uploaded_file['content']

    if filename.endswith('.csv'):
        df = pd.read_csv(io.BytesIO(content))
        # CSV의 경우 파일명을 'category'로 간주하여 추가
        df['category'] = filename.replace('.csv', '')
        return df

    elif filename.endswith(('.xls', '.xlsx')):
        all_sheets = pd.read_excel(io.BytesIO(content), sheet_name=None)
        df_list = []

        # 엑셀 내의 모든 시트를 순회하며 시트 이름을 'category' 컬럼으로 추가
        for sheet_name, df in all_sheets.items():
            df['category'] = sheet_name
            df_list.append(df)

        if len(df_list) > 1:
            return pd.concat(df_list, ignore_index=True)
        else:
            return df_list[0]

    return pd.DataFrame()

def on_load_button_clicked(b):
    global df_raw, df_masking_raw
    with output_log:
        output_log.clear_output()
        if not log_upload_widget.value or not mask_upload_widget.value:
            print("로그 파일과 마스킹 파일을 모두 업로드해주세요.")
            return

        try:
            df_raw = load_file_to_df(log_upload_widget.value)
            df_masking_raw = load_file_to_df(mask_upload_widget.value)
            print(f"로그 데이터({len(df_raw)}행) 및 마스킹 데이터({len(df_masking_raw)}행) 로드 완료.")
            if 'category' in df_raw.columns:
                print(f"-> 인식된 'category'(시트) 목록: {df_raw['category'].unique().tolist()}")
        except Exception as e:
            print(f"로드 중 오류 발생: {e}")

load_button.on_click(on_load_button_clicked)

입력 GUI 요소

# 시트 접두사 입력 위젯 추가
prefix_input = widgets.Text(value='', description='시트 접두사:', placeholder='예: SLM_번역_1차_')
id_input = widgets.Text(value='id', description='ID 컬럼:')
rater_input = widgets.Text(value='tester', description='작업자 컬럼:')
# 그룹 컬럼 기본값을 'category, sub_category'로 변경
group_input = widgets.Text(value='category, sub_category', description='그룹 컬럼(쉼표 구분):', layout=widgets.Layout(width='50%'))
metric_input = widgets.Text(value='유창성, 정확성', description='평가지표 키워드(쉼표 구분):', layout=widgets.Layout(width='50%'))
threshold_input = widgets.FloatText(value=3.0, description='이상치 임계값:')

log_upload_widget = widgets.FileUpload(accept='.csv, .xlsx', multiple=False, description='로그 파일')
mask_upload_widget = widgets.FileUpload(accept='.csv, .xlsx', multiple=False, description='마스킹 파일')

load_button = widgets.Button(description='데이터 로드', button_style='info')
output_log = widgets.Output()

• prefix_input : 구글 시트 이름 앞에 붙을 문자열 (파일 구분을 쉽게 하기 위함)
• id_input : 문항을 식별할 때 사용할 고유 번호가 위치한 Cloumn 명
• rater_input : 평가를 실시한 작업자를 표시하는 이름, ID가 위치한 Cloumn 명
• group_input : 일치도 분석을 실시할 계층과 포함할 로그 정보 목록
• metric_input : 평가 점수열을 찾기 위한 키워드 목록 (해당 목록에 속한 평가지표 명이 포함된 열을 추출해 통계 분석 대상으로 포함)
• threshold_input : 동일 평가 대상에 대해 점수 차이가 해당 입력값 이상일 경우 이상치로 판정

업로드 파일 처리 로직

df_raw = pd.DataFrame()
df_masking_raw = pd.DataFrame()

def load_file_to_df(upload_data):
    if isinstance(upload_data, dict):
        filename = list(upload_data.keys())[0]
        content = upload_data[filename]['content']
    else:
        uploaded_file = upload_data[0]
        filename = uploaded_file['name']
        content = uploaded_file['content']

    if filename.endswith('.csv'):
        df = pd.read_csv(io.BytesIO(content))
        # CSV의 경우 파일명을 'category'로 간주하여 추가
        df['category'] = filename.replace('.csv', '')
        return df

    elif filename.endswith(('.xls', '.xlsx')):
        all_sheets = pd.read_excel(io.BytesIO(content), sheet_name=None)
        df_list = []

        # 엑셀 내의 모든 시트를 순회하며 시트 이름을 'category' 컬럼으로 추가
        for sheet_name, df in all_sheets.items():
            df['category'] = sheet_name
            df_list.append(df)

        if len(df_list) > 1:
            return pd.concat(df_list, ignore_index=True)
        else:
            return df_list[0]

    return pd.DataFrame()

• isinstance(upload_data, dict) : 입력된 파일 형식이 딕셔너리(dict) 자료형인지 확인
  • 딕셔너리(dict) 자료형일 경우 : keys()를 통해 첫 번째 파일명을 가져오고, 해당 파일 내부 정보 추출
  • 딕셔너리(dict) 자료형이 아닌 경우 : 객체 배열로 간주하고, 첫 번째 요소의 'name'과 'content'를 가져와 정보를 추출
• 불러온 내부 정보를.csv, .xls, .xlsx 확장자에 따라 대응할 수 있도록 설계
  • .csv 확장자인 경우 : 파일명을 영역(category) 요소로 간주하고 내부 정보를 데이터 프레임으로 불러옴
  • .xls, .xlsx 확장자인 경우 : 엑셀 내 모든 시트를 순회하며 시트 이름을영역(category) 열에 추가

데이터 규모 및 데이터 분류 식별

print(f"로그 데이터({len(df_raw)}행) 및 마스킹 데이터({len(df_masking_raw)}행) 로드 완료.")
if 'category' in df_raw.columns:
    print(f"-> 인식된 'category'(시트) 목록: {df_raw['category'].unique().tolist()}")

• 업로드된 전체 데이터 규모를 확인하고 엑셀 파일의 모든 시트가 로드되었는지 시각적으로 확인할 수 있도록 함

2) 데이터 전처리 및 통합 데이터 프레임 구성

#@title 2. 전처리: script_itn 보존 및 통합 데이터 프레임 구성
if not df_raw.empty and not df_masking_raw.empty:
    ID_COLUMN = id_input.value.strip()
    RATER_COLUMN = rater_input.value.strip()
    GROUP_COLUMNS = [x.strip() for x in group_input.value.split(',') if x.strip()]
    TARGET_METRIC_KEYWORDS = [x.strip() for x in metric_input.value.split(',') if x.strip()]

    # [수정] script_itn 컬럼이 존재할 경우 필수 변수(essential_cols) 및 ID 변수(id_vars)에 포함
    REFERENCE_COLS = ['script_itn'] if 'script_itn' in df_raw.columns else []
    essential_cols = [ID_COLUMN, RATER_COLUMN] + GROUP_COLUMNS + REFERENCE_COLS

    # [1] 마스킹 데이터 정규화
    mask_id_col = df_masking_raw.columns[0]
    df_mask_lookup = df_masking_raw.melt(id_vars=mask_id_col, var_name='마스킹_모델명', value_name='모델')
    df_mask_lookup.rename(columns={mask_id_col: ID_COLUMN}, inplace=True)
    df_mask_lookup['모델'] = df_mask_lookup['모델'].str.strip()

    # [2] 평가 로그 컬럼 분류
    metric_vars = []
    text_vars = []
    TEXT_KEYWORDS = ["이상여부", "비고", "응답", "오류", "사유"]

    for col in df_raw.columns:
        if any(ex in col for ex in TEXT_KEYWORDS):
            text_vars.append(col)
        elif any(keyword in col for keyword in TARGET_METRIC_KEYWORDS):
            metric_vars.append(col)
            df_raw[col] = pd.to_numeric(df_raw[col], errors='coerce')

    # [3] Long Format 생성 (id_vars에 REFERENCE_COLS 추가하여 데이터 보존)
    id_vars = [ID_COLUMN, RATER_COLUMN] + [c for c in GROUP_COLUMNS if c in df_raw.columns] + REFERENCE_COLS

    # 점수 데이터 처리
    if metric_vars:
        melted_metrics = df_raw.melt(id_vars=id_vars, value_vars=metric_vars, var_name='raw_metric', value_name='score')
        melted_metrics[['모델', '지표']] = melted_metrics['raw_metric'].str.split(':', n=1, expand=True)
        melted_metrics['지표'] = melted_metrics['지표'].str.strip()
        melted_metrics['모델'] = melted_metrics['모델'].str.strip()
        df_scores = melted_metrics.pivot_table(index=id_vars + ['모델'], columns='지표', values='score', aggfunc='first').reset_index()
    else:
        df_scores = pd.DataFrame(columns=id_vars + ['모델'])

    # 텍스트 데이터 처리
    if text_vars:
        melted_texts = df_raw.melt(id_vars=id_vars, value_vars=text_vars, var_name='raw_text_metric', value_name='text_value')
        melted_texts[['모델', '유형']] = melted_texts['raw_text_metric'].str.split(':', n=1, expand=True)
        melted_texts['유형'] = melted_texts['유형'].str.strip()
        melted_texts['모델'] = melted_texts['모델'].str.strip()
        df_texts = melted_texts.pivot_table(index=id_vars + ['모델'], columns='유형', values='text_value', aggfunc='first').reset_index()
        df_final_raw = pd.merge(df_scores, df_texts, on=id_vars + ['모델'], how='outer')
    else:
        df_final_raw = df_scores

    # [4] 마스킹 정보 결합
    df_final = pd.merge(df_final_raw, df_mask_lookup, on=[ID_COLUMN, '모델'], how='left')

    # [5] 문항 번호 카운팅
    if GROUP_COLUMNS:
        unique_q = df_final[GROUP_COLUMNS + [ID_COLUMN]].drop_duplicates().sort_values(GROUP_COLUMNS + [ID_COLUMN])
        unique_q['문항 번호'] = unique_q.groupby(GROUP_COLUMNS).cumcount() + 1
        df_final = pd.merge(df_final, unique_q, on=GROUP_COLUMNS + [ID_COLUMN], how='left')

    # 컬럼 순서 조정 (script_itn을 ID와 문항 번호 근처로 배치)
    cols = df_final.columns.tolist()
    if 'script_itn' in cols:
        target_idx = cols.index('문항 번호') + 1 if '문항 번호' in cols else cols.index(ID_COLUMN) + 1
        cols.insert(target_idx, cols.pop(cols.index('script_itn')))
    df_final = df_final[cols]

    print(f"전처리 완료. 'script_itn'을 포함한 {len(df_final)}행의 통합 데이터를 구성했습니다.")
    df_wide = df_raw[essential_cols + metric_vars].copy()

사용자 입력값 변수화

ID_COLUMN = id_input.value.strip()
RATER_COLUMN = rater_input.value.strip()
GROUP_COLUMNS = [x.strip() for x in group_input.value.split(',') if x.strip()]
TARGET_METRIC_KEYWORDS = [x.strip() for x in metric_input.value.split(',') if x.strip()]

• GUI에서 입력한 값 변수화
  • ID와 평가자는 strip()으로 공백 제거
  • 그룹과 평가지표는 공백 제거 후 콤마( , )를 기준으로 분리해 리스트화

마스킹 데이터 정규화

• 업로드된 평가로그는 실제 모델 데이터를 제공하지 않으며 오로지 마스킹된 모델명만 제시함.
• 문항별 실제 모델 정보는 마스킹 정보 파일에 기록되어 있으므로 ID별 실제 모델 정보가 포함된 마스킹 파일을 함께 업로드해 문항의 응답별 실제 모델을 확인할 수 있게 함

  mask_id_col = df_masking_raw.columns[0]
  df_mask_lookup = df_masking_raw.melt(id_vars=mask_id_col, var_name='마스킹_모델명', value_name='모델')
  df_mask_lookup.rename(columns={mask_id_col: ID_COLUMN}, inplace=True)
  df_mask_lookup['모델'] = df_mask_lookup['모델'].str.strip()

• 마스킹 파일을 불러온 데이터 프레임이 첫번째 열을 ID 열로 간주
• 기존에 wide format으로 작성된df_masking_raw를 melt()해 long format으로 수정
  • wide format : 하나의 ID 행에 마스킹 모델 명을 헤더로 하는 열에 실제 모델명이 가로로 나열됨
  • long format : 하나의 ID 행에 마스킹 모델명, 실제 모델명이 나열
• 수정한 모델 마스킹 정보의 ID 열의 이름을 입력한 ID 열의 이름과 동일하게 변경

df_final = pd.merge(df_final_raw, df_mask_lookup, on=[ID_COLUMN, '모델'], how='left')

• 전체 평가 로그에서 ID와 마스킹 모델명 2가지를 통해 실제 모델명을 찾아 기록
• 이전에 전체 평가 로그의 모델 및 ID 열 이름과 마스킹 정보의 열 이름을 통일해두었기에, 쉽게 합칠 수 있음

평가 결과 전처리

metric_vars = []
text_vars = []
TEXT_KEYWORDS = ["이상여부", "비고", "응답", "오류", "사유"]

for col in df_raw.columns:
    if any(ex in col for ex in TEXT_KEYWORDS):
        text_vars.append(col)
    elif any(keyword in col for keyword in TARGET_METRIC_KEYWORDS):
        metric_vars.append(col)
        df_raw[col] = pd.to_numeric(df_raw[col], errors='coerce')

• raw 데이터의 열을 확인하며 입력했던 평가지표 키워드를 포함하는 열은 평가 점수 열로 간주해 숫자로 값을 가져오고, 텍스트 키워드에 해당하는 열은 텍스트 형태로 값을 불러옮

id_vars = [ID_COLUMN, RATER_COLUMN] + [c for c in GROUP_COLUMNS if c in df_raw.columns] + REFERENCE_COLS

• long format 구성시 사용할 고정 헤더 목록 설정
  • ID와 평가자 + 그룹 목록에서 실제 데이터에 존재하는 요소 + 참고자료 열
• id_vars +text_vars (텍스트 형태) + metric_vars (숫자 형태)로 최종 long format이 구성됨

3) 계층별 kappa 계산

#@title 3. Fleiss-kappa 계산 (다중 계층 통합 방식 적용)
from statsmodels.stats.inter_rater import fleiss_kappa
import numpy as np
import pandas as pd

def calculate_multi_level_kappa(df_source, group_cols, id_col, metric_columns):
    final_results = []

    # 그룹 컬럼 인식 (인덱스 0: Category, 인덱스 1: Sub-Category)
    cat_col = group_cols[0] if len(group_cols) > 0 else None
    subcat_col = group_cols[1] if len(group_cols) > 1 else None

    # --- 내부 헬퍼 함수: 특정 데이터 집합(Subset)에 대한 단일 Kappa 계산 ---
    def compute_kappa_for_subset(df_subset):
        group_freq_matrix = []
        for id_val, id_group in df_subset.groupby(id_col):
            for col in metric_columns:
                # 결측치만 제거하고 이상여부와 상관없이 모든 점수 반영
                vs = id_group[col].dropna()

                if not vs.empty:
                    counts = vs.value_counts()
                    group_freq_matrix.append([int(counts.get(i, 0)) for i in [1, 2, 3, 4, 5]])

        freq_array = np.array(group_freq_matrix)

        try:
            if len(freq_array) == 0:
                return np.nan, 0, 0
            elif np.all(freq_array == freq_array[0, :]):
                return 1.0, len(freq_array), len(freq_array)
            else:
                row_sums = freq_array.sum(axis=1)
                target_raters = np.max(row_sums)
                valid_freq_array = freq_array[row_sums == target_raters]
                valid_rows = len(valid_freq_array)

                if valid_rows > 0:
                    return fleiss_kappa(valid_freq_array), len(freq_array), valid_rows
                else:
                    return np.nan, len(freq_array), 0
        except:
            return np.nan, len(freq_array), 0

    # 1. 전체 통합 Kappa 연산
    print("1/4. 전체 통합 Kappa 계산 중...")
    k, total, valid = compute_kappa_for_subset(df_source)
    final_results.append({'계산_수준': '1. 전체 통합', 'Kappa': k, '총_추출_문항행수': total, '유효_연산_문항행수': valid})

    # 2. Category 기준 통합 Kappa 연산
    if cat_col and cat_col in df_source.columns:
        print(f"2/4. Category({cat_col}) 기준 통합 Kappa 계산 중...")
        for val, grp in df_source.groupby(cat_col):
            k, total, valid = compute_kappa_for_subset(grp)
            final_results.append({'계산_수준': '2. Category 통합', cat_col: val, 'Kappa': k, '총_추출_문항행수': total, '유효_연산_문항행수': valid})

    # 3. Sub-Category 기준 통합 Kappa 연산
    if cat_col and subcat_col and subcat_col in df_source.columns:
        print(f"3/4. Sub-Category({subcat_col}) 기준 통합 Kappa 계산 중...")
        for (c_val, s_val), grp in df_source.groupby([cat_col, subcat_col]):
            k, total, valid = compute_kappa_for_subset(grp)
            final_results.append({'계산_수준': '3. Sub-Category 통합', cat_col: c_val, subcat_col: s_val, 'Kappa': k, '총_추출_문항행수': total, '유효_연산_문항행수': valid})

    # 4. 개별 문항(ID)별 Kappa 연산
    print("4/4. 개별 문항(ID)별 Kappa 계산 중...")
    for id_val, grp in df_source.groupby(id_col):
        k, total, valid = compute_kappa_for_subset(grp)
        res = {'계산_수준': '4. 개별 문항별', id_col: id_val, 'Kappa': k, '총_추출_문항행수': total, '유효_연산_문항행수': valid}
        if cat_col and cat_col in grp.columns: res[cat_col] = grp[cat_col].iloc[0]
        if subcat_col and subcat_col in grp.columns: res[subcat_col] = grp[subcat_col].iloc[0]
        final_results.append(res)

    # 결과 데이터 프레임 구성 및 열 순서 재정렬
    result_df = pd.DataFrame(final_results)

    ordered_cols = ['계산_수준']
    if cat_col: ordered_cols.append(cat_col)
    if subcat_col: ordered_cols.append(subcat_col)
    ordered_cols.extend([id_col, 'Kappa', '기준미달_여부', '총_추출_문항행수', '유효_연산_문항행수'])

    final_cols = [c for c in ordered_cols if c in result_df.columns]
    result_df = result_df[final_cols]

    result_df['기준미달_여부'] = result_df['Kappa'].apply(lambda x: '미달 (<=0.2)' if pd.notna(x) and x <= 0.2 else '')

    print("모든 수준의 Kappa 연산이 완료되었습니다.")
    return result_df

if 'df_raw' in locals() and not df_raw.empty and 'metric_vars' in locals():
    df_kappa = calculate_multi_level_kappa(df_raw, GROUP_COLUMNS, ID_COLUMN, metric_vars)
else:
    print("먼저 1단계와 2단계를 완료하여 데이터를 로드해야 합니다.")

특정 집단 kappa 연산

    # --- 내부 헬퍼 함수: 특정 데이터 집합(Subset)에 대한 단일 Kappa 계산 ---
    def compute_kappa_for_subset(df_subset):
        group_freq_matrix = []
        for id_val, id_group in df_subset.groupby(id_col):
            for col in metric_columns:
                # 결측치만 제거하고 이상여부와 상관없이 모든 점수 반영
                vs = id_group[col].dropna()

                if not vs.empty:
                    counts = vs.value_counts()
                    group_freq_matrix.append([int(counts.get(i, 0)) for i in [1, 2, 3, 4, 5]])

        freq_array = np.array(group_freq_matrix)

        try:
            if len(freq_array) == 0:
                return np.nan, 0, 0
            elif np.all(freq_array == freq_array[0, :]):
                return 1.0, len(freq_array), len(freq_array)
            else:
                row_sums = freq_array.sum(axis=1)
                target_raters = np.max(row_sums)
                valid_freq_array = freq_array[row_sums == target_raters]
                valid_rows = len(valid_freq_array)

                if valid_rows > 0:
                    return fleiss_kappa(valid_freq_array), len(freq_array), valid_rows
                else:
                    return np.nan, len(freq_array), 0
        except:
            return np.nan, len(freq_array), 0

• 데이터를 ID를 기준으로 묶고, 하나로 묶인 평가 결과에 대해 여러 지표를 순회하며 빈도를 계산
  • counts = vs.value_counts()를 통해 1~5점 각각의 점수가 등장한 빈도가 계산됨
• freq_array = np.array(group_freq_matrix)를 통해 기존에 리스트 형태로 저장된 점수별 빈도 데이터를 행렬로 변환
  • [ 0,0,1,1,3 ], [ 1,0,0,2,2 ], ...] → (점수별 빈도가 저장된 2차원 행렬)
• 계산할 데이터가 없거나, 전부 동일한 경우 예외 처리
  • if len(freq_array) == 0: return np.nan, 0, 0 계산할 데이터가 없어 0을 반환
  • elif np.all(freq_array == freq_array[0, :]): return 1.0, len(freq_array), len(freq_array) 모든 분항에 대해 똑같은 점수를 준 경우 1을 반환(만점 처리)
• 메인 로직 계산시 target_raters = np.max(row_sums) 특정 행의 전체 빈도값의 합 중 최댓값을 기준으로 평가자 수를 설정
  • 모든 문항이 동일한 평가자 수를 가진다는 것을 전제로 하므로 이를 바탕으로 계산에 사용할 행을 필터링하는 기준으로 해당 행의 답변 수의 합이 평가자 수와 동일한지 확인
• 계산된 fleiss_kappa와 계산에 사용된 정보를 반환

계층적 kappa 계산

    # 1. 전체 통합 Kappa 연산
    print("1/4. 전체 통합 Kappa 계산 중...")
    k, total, valid = compute_kappa_for_subset(df_source)
    final_results.append({'계산_수준': '1. 전체 통합', 'Kappa': k, '총_추출_문항행수': total, '유효_연산_문항행수': valid})

    # 2. Category 기준 통합 Kappa 연산
    if cat_col and cat_col in df_source.columns:
        print(f"2/4. Category({cat_col}) 기준 통합 Kappa 계산 중...")
        for val, grp in df_source.groupby(cat_col):
            k, total, valid = compute_kappa_for_subset(grp)
            final_results.append({'계산_수준': '2. Category 통합', cat_col: val, 'Kappa': k, '총_추출_문항행수': total, '유효_연산_문항행수': valid})

    # 3. Sub-Category 기준 통합 Kappa 연산
    if cat_col and subcat_col and subcat_col in df_source.columns:
        print(f"3/4. Sub-Category({subcat_col}) 기준 통합 Kappa 계산 중...")
        for (c_val, s_val), grp in df_source.groupby([cat_col, subcat_col]):
            k, total, valid = compute_kappa_for_subset(grp)
            final_results.append({'계산_수준': '3. Sub-Category 통합', cat_col: c_val, subcat_col: s_val, 'Kappa': k, '총_추출_문항행수': total, '유효_연산_문항행수': valid})

    # 4. 개별 문항(ID)별 Kappa 연산
    print("4/4. 개별 문항(ID)별 Kappa 계산 중...")
    for id_val, grp in df_source.groupby(id_col):
        k, total, valid = compute_kappa_for_subset(grp)
        res = {'계산_수준': '4. 개별 문항별', id_col: id_val, 'Kappa': k, '총_추출_문항행수': total, '유효_연산_문항행수': valid}
        if cat_col and cat_col in grp.columns: res[cat_col] = grp[cat_col].iloc[0]
        if subcat_col and subcat_col in grp.columns: res[subcat_col] = grp[subcat_col].iloc[0]
        final_results.append(res)

• 평가 플랫폼에서 제공하는 kappa 값과 동일한 값을 도출하기 위해서는 영역, 세부영역, 문항 등 계층별로 kappa를 계산해야 했으므로 앞선 부분 집단 kappa 계산을 여러 단계로 적용해 계층적 kappa 계산을 실시
• 데이터를 영역, 세부영역 등 기준 단위로 분할하여 순회
  • 같은 구분명을 가진 단위로 데이터를 그룹화 예시) 기기_제어라는 영역을 가진 답변을 하나로 묶음
  • 그룹화한 답변 내용을 특정 집단 kappa 계산 함수에 입력해 해당 집단의 kappa를 계산

4) 이상치 확인

#@title 4. 이상치 연산, 통계 보고서 생성 및 모델 오류 목록 취합
if 'df_final' in locals() and not df_final.empty:
    print("이상치 연산 및 모델 오류 목록 취합 시작...")
    outlier_df = df_final.copy()

    # 1) 순수 수치 지표 추출 (script_itn 제외)
    exclude_keys = ['모델', '마스킹_모델명', '응답', '비고', '모델 오류', '오류', '이상여부', '사유', '문항 번호', 'script_itn']
    pure_metrics = [c for c in df_final.columns if c not in essential_cols and c not in exclude_keys]

    outlier_groups = set() # 이상치가 발생한 (ID, 모델) 그룹 저장용

    # 2) 통계적 이상치(Outlier) 계산: 최댓값 - 최솟값 >= 3
    for metric in pure_metrics:
        if metric not in outlier_df.columns: continue

        # 그룹별 max, min 계산
        grouped = outlier_df.groupby([ID_COLUMN, '모델'])[metric]
        metric_max = grouped.transform('max')
        metric_min = grouped.transform('min')

        # 판정 기준 적용
        is_out_group = (metric_max - metric_min) >= 3
        outlier_df[f'{metric}_이상치판정'] = is_out_group

        # 이상치 그룹 수집
        outlier_rows = outlier_df[is_out_group]
        for _, row in outlier_rows.iterrows():
            outlier_groups.add((row[ID_COLUMN], row['모델']))

    outlier_masks = [outlier_df[f'{m}_이상치판정'] for m in pure_metrics if f'{m}_이상치판정' in outlier_df.columns]
    outlier_df['총_이상치_발생수'] = pd.concat(outlier_masks, axis=1).sum(axis=1) if outlier_masks else 0

    # 3) 통계 보고서 생성
    analysis_dims = list(dict.fromkeys([RATER_COLUMN, ID_COLUMN, '모델'] + GROUP_COLUMNS))
    summary_list = []
    for dim in analysis_dims:
        if dim in outlier_df.columns:
            summary = outlier_df.groupby(dim).agg(총_평가=(ID_COLUMN, 'count'), 이상치_합=('총_이상치_발생수', 'sum')).reset_index()
            summary.columns = ['분석대상', '총_평가_건수', '이상치_발생_총합']
            summary.insert(0, '분석차원', dim)
            summary_list.append(summary)
    outlier_summary = pd.concat(summary_list, ignore_index=True)

    # 4) 모델 오류 목록 취합 (script_itn 위치 지정)
    error_col = next((c for c in df_final.columns if any(k in c for k in ['이상여부', '오류'])), None)
    df_error_list = pd.DataFrame()

    if error_col:
        is_error = outlier_df[error_col].map(lambda x: str(x).lower() in ['true', '1', '1.0', 't'])
        df_filtered = outlier_df[is_error].copy()

        # 열 순서 재배치: 세부영역, id, 문항 번호, script_itn, 테스터, 마스킹 모델명, 비고, 실제 모델명, 응답
        ordered_cols = []
        ordered_cols.extend([c for c in GROUP_COLUMNS if c in df_filtered.columns])
        if ID_COLUMN in df_filtered.columns: ordered_cols.append(ID_COLUMN)
        if '문항 번호' in df_filtered.columns: ordered_cols.append('문항 번호')
        if 'script_itn' in df_filtered.columns: ordered_cols.append('script_itn') # [추가]
        if RATER_COLUMN in df_filtered.columns: ordered_cols.append(RATER_COLUMN)
        if '마스킹_모델명' in df_filtered.columns: ordered_cols.append('마스킹_모델명')
        if '비고' in df_filtered.columns: ordered_cols.append('비고')
        if '모델' in df_filtered.columns: ordered_cols.append('모델')
        if '응답' in df_filtered.columns: ordered_cols.append('응답')

        extra_cols = [error_col]
        for c in df_filtered.columns:
            if any(k in str(c) for k in ['사유']) and c not in ordered_cols and c not in extra_cols:
                extra_cols.append(c)

        df_error_list = df_filtered[[c for c in ordered_cols + extra_cols if c in df_filtered.columns]].copy()
        df_error_list.rename(columns={'모델': '실제 모델명'}, inplace=True)

    print("4단계 완료.")
else:
    print("먼저 이전 단계를 완료해야 합니다.")

이상치 확인 방법

• 순수 평가 지표가 기록된 열만 골라냄(숫자가 기록된 영역만 확인하기 위함)
• ID와 모델명 조합을 통해 단일 평가 대상을 구분해 해당 대상에 대한 평가를 그룹화
• 그룹 내에서 grouped.transform('max'),grouped.transform('min')으로 최대, 최소 도출
• 도출된 최대 metric_max, 최소 metric_min의 차가 threshold_input보다 크거나 같으면 이상치로 판단
  • metric_max와 metric_min은 그룹 내에서 각각 최대,최소 값을 뽑아 원본과 동일한 크기로 생성한 것임
  • 동일한 대상에 대해 평가자 1은 2점, 평가자 2는 5점을 부여한 상황은 정상적이지 않은 평가로 간주
• 이상치를 포함한 문항 전체를 true로 표기할 수 있도록 is_out_group을 구성한 뒤, 전체 응답 목록에 (평가 지표명)_이상치판정열을 추가해 이상치 포함 여부를 기록

5) 이상치 목록 작성

#@title 5. 이상치 목록 작성
if 'outlier_df' in locals() and 'outlier_groups' in locals():
    print("이상치 목록 작성을 시작합니다...")
    df_outlier_list = pd.DataFrame()

    if outlier_groups:
        # 이상치 그룹에 해당하는 전체 행 필터링
        mask = outlier_df.apply(lambda x: (x[ID_COLUMN], x['모델']) in outlier_groups, axis=1)
        df_filtered = outlier_df[mask].copy()

        # 열 순서 정의 (세부영역, ID, 문항 번호, 작업자, 마스킹 정보, 실제 모델, 지표, 기타 응답)
        ordered_cols = []
        ordered_cols.extend([c for c in GROUP_COLUMNS if c in df_filtered.columns])
        if ID_COLUMN in df_filtered.columns: ordered_cols.append(ID_COLUMN)
        if '문항 번호' in df_filtered.columns: ordered_cols.append('문항 번호')
        if RATER_COLUMN in df_filtered.columns: ordered_cols.append(RATER_COLUMN)
        if '마스킹_모델명' in df_filtered.columns: ordered_cols.append('마스킹_모델명')
        if '모델' in df_filtered.columns: ordered_cols.append('모델')

        ordered_cols.extend(pure_metrics)

        extra_cols = []
        for c in df_filtered.columns:
            if any(k in str(c) for k in ['응답', '비고', '사유', '이상여부', '오류']) and c not in ordered_cols:
                extra_cols.append(c)

        final_target_cols = [c for c in ordered_cols + extra_cols if c in df_filtered.columns]
        df_outlier_list = df_filtered[final_target_cols].copy()

        # 정렬 및 명칭 변경
        df_outlier_list.sort_values(by=[ID_COLUMN, '모델', RATER_COLUMN], inplace=True)
        df_outlier_list.rename(columns={'모델': '실제 모델명'}, inplace=True)
        # 6단계 셀 서식 매핑을 위해 인덱스 초기화 필수
        df_outlier_list.reset_index(drop=True, inplace=True)

        print(f"이상치 목록 {len(df_outlier_list)}건 취합 완료 (총 {len(outlier_groups)}개 문항:모델 그룹).")
    else:
        print("이상치 조건(최댓값-최솟값 >= 3)을 만족하는 항목이 없습니다.")

    print("5단계 완료.")
else:
    print("4단계가 정상적으로 실행되지 않았습니다.")

• 앞선 과정에서 확인한 이상치를 따로 확인할 수 있도록 별도의 목록을 작성

  6) 서식 적용 및 최종 시트 저장

  #@title 6. 최종 구글 시트 저장
  from datetime import datetime
  

required_vars = ['outlier_df', 'outlier_summary', 'df_kappa', 'df_error_list', 'df_outlier_list'] missing_vars = [var for var in required_vars if var not in globals()]

if not missing_vars: current_time = datetime.now().strftime("%Y%m%d_%H%M%S")

# [수정] 사용자가 입력한 접두사 가져오기 및 시트명 적용
prefix = prefix_input.value.strip() if 'prefix_input' in globals() else ""
if prefix and not prefix.endswith('_'):
    prefix += " " # 자연스러운 연결을 위해 스페이싱 자동 추가

sheet_title = f"{prefix}평가로그_통합_결과_{current_time}"
print(f"'{sheet_title}' 시트 생성 및 데이터 기록 중...")

spreadsheet = gc.create(sheet_title)

# 공통 서식 설정
red_fmt = cellFormat(backgroundColor=color(1, 0.78, 0.8), textFormat=textFormat(bold=True, foregroundColor=color(0.61, 0, 0.02)))
max_fmt = cellFormat(backgroundColor=color(0.8, 0.9, 1.0), textFormat=textFormat(bold=True, foregroundColor=color(0.0, 0.2, 0.8)))
gray_bg_fmt = cellFormat(backgroundColor=color(0.95, 0.95, 0.95))

# [시트 1] 평가 결과 취합
ws_main = spreadsheet.sheet1
ws_main.update_title("평가_결과_취합")
set_with_dataframe(ws_main, outlier_df)

main_reqs = []
for i, col in enumerate(outlier_df.columns):
    if f'{col}_이상치판정' in outlier_df.columns:
        for r_idx, is_out in enumerate(outlier_df[f'{col}_이상치판정']):
            if is_out:
                main_reqs.append({"repeatCell": {"range": {"sheetId": ws_main.id, "startRowIndex": r_idx+1, "endRowIndex": r_idx+2, "startColumnIndex": i, "endColumnIndex": i+1}, "cell": {"userEnteredFormat": red_fmt.to_props()}, "fields": "userEnteredFormat"}})
if main_reqs: spreadsheet.batch_update({"requests": main_reqs})

# [시트 2] Kappa 결과
ws_kappa = spreadsheet.add_worksheet(title="Kappa_계산결과", rows="1000", cols="20")
set_with_dataframe(ws_kappa, df_kappa)

# [시트 3] 이상치 통계
ws_summary = spreadsheet.add_worksheet(title="이상치_통계보고서", rows="1000", cols="20")
set_with_dataframe(ws_summary, outlier_summary)

# [시트 4] 모델 오류 목록
ws_error = spreadsheet.add_worksheet(title="모델오류_목록", rows="1000", cols="20")
set_with_dataframe(ws_error, df_error_list)

# [시트 5] 이상치 목록
if not df_outlier_list.empty:
    ws_outlier = spreadsheet.add_worksheet(title="이상치_목록", rows=str(len(df_outlier_list)+100), cols="30")
    set_with_dataframe(ws_outlier, df_outlier_list)

    outlier_reqs = []

    # 1) 문항 그룹 교차 배경색
    current_group = None
    is_gray = False

    for row_idx, row in df_outlier_list.iterrows():
        group_key = (row[ID_COLUMN], row['실제 모델명'])
        if group_key != current_group:
            current_group = group_key
            is_gray = not is_gray

        if is_gray:
            outlier_reqs.append({
                "repeatCell": {
                    "range": {
                        "sheetId": ws_outlier.id,
                        "startRowIndex": row_idx + 1, "endRowIndex": row_idx + 2,
                        "startColumnIndex": 0, "endColumnIndex": len(df_outlier_list.columns)
                    },
                    "cell": {"userEnteredFormat": gray_bg_fmt.to_props()},
                    "fields": "userEnteredFormat.backgroundColor"
                }
            })

    # 2) 최댓값/최솟값 하이라이트
    grouped = df_outlier_list.groupby([ID_COLUMN, '실제 모델명'])
    for metric in pure_metrics:
        if metric not in df_outlier_list.columns: continue
        col_idx = df_outlier_list.columns.get_loc(metric)

        for _, group_df in grouped:
            metric_series = group_df[metric].dropna()
            if metric_series.empty: continue

            g_max = metric_series.max()
            g_min = metric_series.min()

            if (g_max - g_min) >= 3:
                for row_idx, val in metric_series.items():
                    if val == g_max:
                        outlier_reqs.append({"repeatCell": {"range": {"sheetId": ws_outlier.id, "startRowIndex": row_idx+1, "endRowIndex": row_idx+2, "startColumnIndex": col_idx, "endColumnIndex": col_idx+1}, "cell": {"userEnteredFormat": max_fmt.to_props()}, "fields": "userEnteredFormat"}})
                    elif val == g_min:
                        outlier_reqs.append({"repeatCell": {"range": {"sheetId": ws_outlier.id, "startRowIndex": row_idx+1, "endRowIndex": row_idx+2, "startColumnIndex": col_idx, "endColumnIndex": col_idx+1}, "cell": {"userEnteredFormat": red_fmt.to_props()}, "fields": "userEnteredFormat"}})

    if outlier_reqs:
        spreadsheet.batch_update({"requests": outlier_reqs})
else:
    ws_outlier = spreadsheet.add_worksheet(title="이상치_목록", rows="10", cols="10")

print(f"\n작업 완료! 구글 시트 URL: {spreadsheet.url}")

else: print(f"오류: 데이터 메모리 누락 -> {missing_vars}")

- 앞선 과정에서 구성한 데이터를 취합해 구글 시트로 구성
- 가시성 강화를 위해 색상 변경 및 서식 적용

#### 최댓값/최솟값 하이라이트

    # 2) 최댓값/최솟값 하이라이트
    grouped = df_outlier_list.groupby([ID_COLUMN, '실제 모델명'])
    for metric in pure_metrics:
        if metric not in df_outlier_list.columns: continue
        col_idx = df_outlier_list.columns.get_loc(metric)

        for _, group_df in grouped:
            metric_series = group_df[metric].dropna()
            if metric_series.empty: continue

            g_max = metric_series.max()
            g_min = metric_series.min()

            if (g_max - g_min) >= 3:
                for row_idx, val in metric_series.items():
                    if val == g_max:
                        outlier_reqs.append({"repeatCell": {"range": {"sheetId": ws_outlier.id, "startRowIndex": row_idx+1, "endRowIndex": row_idx+2, "startColumnIndex": col_idx, "endColumnIndex": col_idx+1}, "cell": {"userEnteredFormat": max_fmt.to_props()}, "fields": "userEnteredFormat"}})
                    elif val == g_min:
                        outlier_reqs.append({"repeatCell": {"range": {"sheetId": ws_outlier.id, "startRowIndex": row_idx+1, "endRowIndex": row_idx+2, "startColumnIndex": col_idx, "endColumnIndex": col_idx+1}, "cell": {"userEnteredFormat": red_fmt.to_props()}, "fields": "userEnteredFormat"}})

```

• 앞서 이상치 검색할 때처럼 Id-모델을 단위로 그룹화해 최댓값, 최솟값을 확인할 집단을 지정
• g_max와 g_min가 최댓값과 최솟값에 해당하는 셀로, 해당 셀을 outlier_reqs에 추가해 서식 적용 목록을 구성 (batch 단위로 업데이트)
• outlier_reqs의 셀들을 최댓값과 최솟값에 따라 정해진 서식으로 변경

3. 평가 로그 분석 구글 시트

1. 평가 로그 입력
2. 전처리 및 지표 계산
3. 통합 분석 구글 시트 생성

• 해당 기능을 통해 간편하게 평가 로그를 분석하고, 문제점을 확인할 수 있었음

0.Summary

분석 Tool 요약

• 목적: TTS(음성합성) 평가 데이터의 품질 지표(평균/표준편차)와 신뢰도(Kappa/Agreement)를 다차원으로 분석
• 핵심 기능:
  • 다차원 분석: 전체, 모델별, 문항(ID)별, 도메인/카테고리별 자동 그룹화 분석
  • 신뢰도 검증: Fleiss' Kappa 및 Agreement를 통한 평가자 간 합의 수준 측정
  • 의사결정 지원: 분석 수치에 따른 자동 조치 가이드(Action Guide) 생성
• 기대 효과: 평가 데이터의 객관성 확보 및 불성실 평가자/모호한 문항의 조기 선별

1. Input xlsx, csv Format

예시 이미지

입력 xlsx, csv 형식

• 구분자 헤더 : 문항, 평가자, 모델, 도메인 등 (녹색 영역)
• 구분자 입력값 : (노란색 영역)
• 평가지표 헤더 : 정확성, 발음, 속도, 끊어읽기 등 (파란색 영역)
• 평가지표 입력값 : (빨간색 영역)

2. Input Setting

입력값

• ID_COLUMN
• RATER_COLUMN
• GROUP_INPUT
  • 문항, 평가자 열의 이름과 그 외 구분자로 사용할 헤더 목록을 입력
  • 해당 목록에 포함되지 않은 헤더는 평가지표로 간주
• RESULT_FILE
  • 최종 분석 결과를 기록하여 저장할 파일명

3. Install Libraries

#@title 라이브러리 설치 및 확인
import importlib
import subprocess
import sys

def install_if_missing(package_name, import_name=None):
    if import_name is None:
        import_name = package_name

    try:
        importlib.import_module(import_name)
        # print(f"✅ {package_name} 이(가) 이미 설치되어 있습니다.")
    except ImportError:
        print(f"Installing {package_name}...")
        # Colab 환경에서 시스템 호출을 통해 설치
        subprocess.check_call([sys.executable, "-m", "pip", "install", package_name])
        print(f"✅ {package_name} 설치 완료.")

# 설치가 필요한 라이브러리 목록 (패키지명, 임포트명)
packages = [
    ("openpyxl", "openpyxl"),
    ("statsmodels", "statsmodels"),
    ("xlsxwriter", "xlsxwriter")
]

for pkg, imp in packages:
    install_if_missing(pkg, imp)

import pandas as pd
import numpy as np
import io

• 필수 라이브러리, 패키지 openptxl, statsmodel , xlsxwriter 설치
  • openptxl : 엑셀(xlsx) 파일 불러와 Data Frame으로 변환
  • statsmodel : 빈도 행렬을 통해 Fleiss-Kappa, Agreement 계산 함수 사용
  • xlsxwriter : 엑셀(xlsx) 파일 내용 수정, 쓰기
• 설치 확인 후 설치되지 않은 라이브러리, 패키지만 선택적으로 설치

4. Data Frame 변환

업로드 된 파일 불러오기

# 1. 파일 확장자 확인 및 데이터 로드
if FILE_PATH and os.path.exists(FILE_PATH):
    # 파일 확장자에 따라 읽기 방식 결정
    if FILE_PATH.endswith('.csv'):
        df = pd.read_csv(FILE_PATH)
    elif FILE_PATH.endswith(('.xls', '.xlsx')):
        df = pd.read_excel(FILE_PATH)
    else:
        print("❌ 지원하지 않는 파일 형식입니다. (CSV 또는 Excel 파일 필요)")
        df = pd.DataFrame() # 빈 데이터프레임 생성
else:
    print("❌ 파일을 업로드하거나 올바른 경로를 입력해주세요.")
    df = pd.DataFrame()

• 파일 확장자에 따라 파일 불러와 Data Frame(df) 형식으로 변환해 향후 활용에 용이하게 함

입력값 전처리

if not df.empty:
    # (1) 그룹 컬럼 리스트화 (쉼표 분리 및 공백 제거)
    GROUP_COLUMNS = [x.strip() for x in GROUP_INPUT.split(',') if x.strip()]

    # (2) 구분자(식별자) 목록 통합
    # 사용자가 입력한 ID, 평가자, 그룹 컬럼들을 모두 모음
    excluded_cols = [ID_COLUMN, RATER_COLUMN] + GROUP_COLUMNS

    # (3) 평가지표(Metrics) 자동 추출
    # 전체 컬럼 중 구분자 영역을 제외한 나머지를 모두 평가지표로 간주
    metric_cols = [col for col in df.columns if col not in excluded_cols and "Unnamed" not in col]

    # (4) 데이터 타입 정리 (평가지표는 숫자형으로 변환)
    for col in metric_cols:
        df[col] = pd.to_numeric(df[col], errors='coerce')

• 데이터 프레임을 정상적으로 불러온 경우 GROUP_COLUMNS 리스트화
  • 예시) GROUP_COLUMNS에 "모델, 도메인, 카테고리" 입력 리스트로 변형 : GROUP_COLUMNS = ["모델", "도메인", "카테고리"]
• ID_COLUMN, RATER_COLUMN에서 입력받은 값들도 모아 excluded_cols 구분자 리스트를 구성
  • 예시) ID_COLUMN = "문항", RATER_COLUMN = "평가자"로 입력 리스트로 변형 및 통합 : excluded_cols = ["문항","평가자","모델", "도메인", "카테고리"]
• 나머지 열의 요소는 모두 평가지표로 간주
  • 예시) 기존 헤더 목록 : [문항,평가자,모델,도메인,카테고리,정확성,발음,속도,끊어읽기...] 구분자 제외 후 평가지표만 추출한 목록 : [정확성,발음,속도,끊어읽기...]
• 평가지표 열의 요소를 전부 숫자형으로 변환

5. Fleiss-kappa & Agreement 측정

평균, 표준편차, kappa 산출 통합 함수

def get_combined_metrics(target_df, group_id_col, metric_col):
    try:
        # 데이터 정제
        valid_df = target_df[[group_id_col, metric_col]].dropna()
        if valid_df.empty: return np.nan, np.nan, np.nan, np.nan, 0

        # [통계] 평균 및 표준편차
        mean_v = round(valid_df[metric_col].mean(), 4)
        std_v = round(valid_df[metric_col].std(), 4)

        # [Kappa] 빈도 행렬 생성 (행: 평가대상, 열: 점수범주)
        count_matrix = pd.get_dummies(valid_df[metric_col]).groupby(valid_df[group_id_col]).sum().to_numpy()

        kappa = np.nan
        # 대상(Subject)이 2개 이상이고 점수 종류가 2개 이상일 때 Fleiss' Kappa 산출
        if count_matrix.shape[0] >= 2 and count_matrix.shape[1] >= 2:
            kappa = round(fleiss_kappa(count_matrix, method='fleiss'), 4)
        elif count_matrix.shape[0] >= 2 and count_matrix.shape[1] == 1:
            kappa = 1.0 # 모든 대상에 대해 모든 평가자가 동일 점수를 준 경우

        agreement = np.nan
        # 일치도(Agreement) 계산 로직
        # n_raters가 2명 이상이어야 쌍(pair)을 지어 일치도를 구할 수 있음
        if count_matrix.shape[0] >= 2:
          n_raters = count_matrix[0].sum()

          if n_raters >= 2:
            # 각 문항별 일치도 P_i 계산: (sum(n_ij^2) - n_i) / (n_i * (n_i - 1))
            numerator = np.sum(count_matrix**2, axis=1) - n_raters
            denominator = n_raters * (n_raters - 1)
            p_i = numerator / denominator
            agreement = round(np.mean(p_i), 4) # 전체 문항에 대한 평균 일치도


        return mean_v, std_v, kappa, agreement, len(valid_df)
    except Exception as e:
      print(f"Error in {metric_col} for {group_id_col}: {e}") # 에러 메시지 출력
      return np.nan, np.nan, np.nan, np.nan, 0

Fleiss-kappa 측정

# [Kappa] 빈도 행렬 생성 (행: 평가대상, 열: 점수범주)
count_matrix = pd.get_dummies(valid_df[metric_col]).groupby(valid_df[group_id_col]).sum().to_numpy()

kappa = np.nan
# 대상(Subject)이 2개 이상이고 점수 종류가 2개 이상일 때 Fleiss' Kappa 산출
if count_matrix.shape[0] >= 2 and count_matrix.shape[1] >= 2:
    kappa = round(fleiss_kappa(count_matrix, method='fleiss'), 4)
elif count_matrix.shape[0] >= 2 and count_matrix.shape[1] == 1:
    kappa = 1.0 # 모든 대상에 대해 모든 평가자가 동일 점수를 준 경우

• Fleiss-kappa 측정을 위해서는 빈도 행렬을 구성해야 함
• metric_col에 포함된 평가 지표에 해당하는 데이터를 순차적으로 가져옴
  • 예시) metric_col = [정확성,발음,속도,끊어읽기...]인 경우 정확성 항목에 관한 평가 데이터를 가져옴
• 점수를 열로 가지는 표로 재구성해, 각 평가자의 점수에 해당하는 열에 1로 표기
  • 예시) 정확성에 대한 점수 구성이 [4,2,1,3,5]인 경우 아래와 같이 행렬 구성 이후 묶고자 하는 구분자를 기준으로 변수들을 한 그룹으로 묶음

• 구분자를 기준으로 묶은 변수 그룹을 합쳐 해당 점수의 빈도 행렬로 변환
  • 예시) 문항을 기준으로 위의 행렬을 빈도 행렬로 변환한 결과

• statsmodels이 계산할 수 있도록 문자열 영역을 제거하고 숫자 값만 행렬로 전달
• 어떤 평가자가 몇 점을 주었는지는 중요하지 않고, 전체에서 각 점수가 어떤 빈도로 등장했는지가 중요함

  count_matrix.shape[0] >= 2 and count_matrix.shape[1] >= 2

• 구분자에 해당하는 평가 요소가 2개 이상인지, 발생한 점수의 종류가 2개 이상인지 확인 구분자 : 평가요소 = 문항 : [1번, 2번, 3번...]
  • 평가 요소가 하나라면 비교할 수 있는 다른 대상이 없어 kappa를 구할 수 없음
    • Fleiss-kappa는 평가 대상이 다름에도 평가가 일관되게 이루어졌는지를 확인하기 위함이므로 단일 항목에 대한 정보로는 평가가 일관되게 이루어졌는지를 정의할 수 없음
  • 점수의 종류가 1개라면 모두 통일된 상태이므로 kappa가 1임

Agreement 측정

agreement = np.nan
# 일치도(Agreement) 계산 로직
# n_raters가 2명 이상이어야 쌍(pair)을 지어 일치도를 구할 수 있음
if count_matrix.shape[0] >= 2:
n_raters = count_matrix[0].sum()

if n_raters >= 2:
    # 각 문항별 일치도 P_i 계산: (sum(n_ij^2) - n_i) / (n_i * (n_i - 1))
    numerator = np.sum(count_matrix**2, axis=1) - n_raters
    denominator = n_raters * (n_raters - 1)
    p_i = numerator / denominator
    agreement = round(np.mean(p_i), 4) # 전체 문항에 대한 평균 일치도

• 일치도 계산도 동일하게 빈도 행렬을 사용함
  • 점수 빈도 행렬의 첫번째 열의 모든 빈도 값을 합치면 전체 평가자 수를 알 수 있음

    numerator = np.sum(count_matrix**2, axis=1) - n_raters

• 각 빈도 값을 제곱한 뒤, 다 더해서 평가자 수를 뺌
  • 같은 점수를 선택한 사람들끼리 짝을 맺을 수 있는 경우의 수를 구함

    denominator = n_raters * (n_raters - 1)

• 발생할 수 있는 모든 쌍의 개수를 구함(가능한 모든 쌍이므로 순서/방향을 고려한 개수)

  agreement = round(np.mean(p_i), 4)

• 같은 점수 짝을 맺을 경우의 수/가능한 모든 쌍의 개수 의 평균을 구함
  • 평가 결과가 보여주는 평균적인 합의 수준을 보여줌

6. 분석 시나리오 세팅

# (차원 이름, 그룹화할 컬럼, Kappa 계산 시 기준이 될 대상 컬럼)
scenarios = [('전체', [], 'id_model_pair')] # 전체는 ID+모델 조합을 대상으로 봄

# 단일 구분자별 분석 (id, 모델, 도메인, 카테고리 등)
for col in [ID_COLUMN] + GROUP_COLUMNS:
    # ID별 분석일 때는 '모델'을 대상으로 일치도를 보고, 그 외에는 'ID'를 대상으로 봄
    sub_target = '모델' if col == ID_COLUMN else ID_COLUMN
    scenarios.append((f'{col}별', [col], sub_target))

# 조합별 분석 (사용자 입력 그룹 전체 조합)
if len(GROUP_COLUMNS) > 1:
    scenarios.append(('조합별(Group)', GROUP_COLUMNS, ID_COLUMN))

# 4. 분석 실행 루프
final_report = []

# 전체 분석용 임시 식별자 생성
df['id_model_pair'] = df[ID_COLUMN].astype(str) + "_" + df['모델'].astype(str)

for sc_name, sc_cols, sub_col in scenarios:
    if not sc_cols: # 전체 분석
        for m in metric_cols:
            mean, std, kap, agr, n = get_combined_metrics(df, sub_col, m)
            final_report.append({
                '분석차원': sc_name, '대상': 'Total', '지표': m,
                '평균': mean, '표준편차': std, 'Kappa': kap, '일치도': agr, '샘플수': n
            })
    else: # 그룹/구분자별 분석
        grouped = df.groupby(sc_cols)
        for name, group_df in grouped:
            label = name if isinstance(name, str) else "-".join(map(str, name))
            for m in metric_cols:
                mean, std, kap, agr, n = get_combined_metrics(group_df, sub_col, m)
                final_report.append({
                    '분석차원': sc_name, '대상': label, '지표': m,
                    '평균': mean, '표준편차': std, 'Kappa': kap, '일치도': agr, '샘플수': n
                })

# 5. 결과 정리 및 최종 검증
report_df = pd.DataFrame(final_report)

# 지표 열에 구분자 명칭이 들어간 행 강제 제거 (순수 지표만 남김)
report_df = report_df[~report_df['지표'].isin(all_identifiers)]

print(f"✅ 분석 완료!")
display(report_df.head(20))

전체 분석

• 전체 분석시에는 ID+모델 조합을 구분자로 사용해 지표를 구함
  • 하나의 문항에 대해 3개의 모델이 있는 조건이므로 하나의 고유한 요소는 ID+모델이 조합되어야 구분할 수 있음

    단일 구분자 분석

• 각 구분자를 기준으로 지표를 계산
  • 문항별 분석일 때는 '모델'을 대상으로 일치도를 보고, 그 외에는 '문항'을 대상으로 봄

    조합 구분자 분석

• 구분자 조합 전체를 합친 조합 구분자를 기준으로 지표를 계산

  분석 시나리오 분리 이유

• 각 구분자에 따라 Fleiss-kappa와 Agreement를 따로 계산함으로써, 어느 요소에서 평가 결과의 불일치가 발생하는지 파악하기 쉽게 하고자 함
• 특정 구분자에서 불일치가 반복된다면, 해당 구분자에 맞춘 대처가 가능
  • 모델 : 모델 성능이 월등하여 전반적으로 점수가 높게 치우친 경우, 아주 적은 특이값으로도 kappa가 낮게 나올 수 있음
  • 평가자 : 평가자가 다른 평가자와 다른 기준을 적용하고 있거나, 가이드 숙지가 미흡할 수 있음

7. 지표에 따른 가이드

def get_action_guide(row):
    kap = row.get('Kappa')
    agr = row.get('일치도')

    # 데이터가 없는 경우 처리
    if pd.isna(kap) or pd.isna(agr):
        return "-"

    # 1. 일치도 높음 / Kappa 음수 또는 매우 낮음 (Prevalence 현상)
    if agr >= 0.8 and kap <= 0.1:
        return "[특이값 확인] 대세와 다른 소수 의견 존재. 단순 실수인지 또는 타인이 놓친 결함 발견인지 해당 평가자 면담 필요."

    # 2. 일치도 중간 / Kappa가 0에 가까움 (Bimodal 분리)
    elif 0.35 <= agr <= 0.55 and kap <= 0.1:
        return "[가이드 수정] 의견이 두 집단으로 팽팽하게 갈림. 기준 모호성 확인 및 가이드라인 구체화 필요."

    # 3. 일치도 낮음 / Kappa가 어느 정도 존재 (난이도 높음)
    elif agr < 0.4 and 0.1 < kap <= 0.4:
        return "[난이도 높음] 전반적으로 의견이 분산되었으나 일부 합의 존재."

    # 4. 일치도 매우 낮음 / Kappa 0 이하 (무작위 수준)
    elif agr < 0.3 and kap <= 0:
        return "[신뢰 불가] 무작위 평가 수준. 데이터 분석 가치 낮음."

    return "정상"

# '조치 가이드' 컬럼 생성
report_df['조치 가이드'] = report_df.apply(get_action_guide, axis=1)

# 파일 저장
output_path = f"{RESULT_FILE}_통합분석.csv"
report_df.to_csv(output_path, index=False, encoding='utf-8-sig')

print(f"✅ 분석 및 가이드 생성 완료! '{output_path}' 파일에 조치사항이 포함되었습니다.")
display(report_df.head(20))

Fleiss-kappa / Agreement에 따른 가이드

• 각 지표에 따른 조치 가이드를 제공해 평가 결과를 반영한 의사결정을 도움

0. Summary

음성 데이터 일치도 검사 요약

1. 설정: STT 결과, 정답 스크립트, 점수 기록 열 지정 및 시트 연결
2. 필터링: 점수가 없고 비교 데이터(STT/스크립트)가 모두 있는 행만 추출
3. 전처리: 정규표현식을 활용해 특수문자와 공백을 제거하여 데이터 정제
4. 채점: Levenshtein Distance 기반 편집 거리 계산 및 100점 환산
5. 기록: 계산된 일치도 점수를 구글 시트에 자동 업데이트하여 완료

   1. 입력값 설정

   

   입력값

   SHEET_URL

• 검수할 음성 파일 url 또는 드라이브 파일 id가 포함된 Google Sheet url

  SHEET_NAME

• 전체 시트에서 작업 데이터가 위치한 세부 시트명
• 여러 시트를 포함하고 있는 경우 특정 시트를 찾아 검수 대상으로 선정

  RESULT_COLUMN

• STT 결과물이 기록된 열 문자

  SCRIPT_COLUMN

• STT 결과물과 비교할 정답 스크립트가 위치한 열 문자

  SCORE_COLUMN

• 비교 후 점수가 기록될 열 문자

확인 메시지 창

예시 이미지

실행 코드

from google.colab import output
import sys

# @title ⚠️ 설정값 최종 확인
confirm_msg = f"""[일치도 검사 실행 확인]

- STT 결과 열: {RESULT_COLUMN}
- 원본 스크립트 열: {SCRIPT_COLUMN}
- 점수 기록 열: {SCORE_COLUMN}

위 설정이 맞습니까? 작업을 진행하시려면 '확인'을 눌러주세요."""

# 파이썬의 줄바꿈(\n)을 자바스크립트 문자열 내 줄바꿈(\\n)으로 변환
# 또한 따옴표 충돌을 방지하기 위해 replace 처리
js_msg = confirm_msg.strip().replace('\n', '\\n').replace('"', '\\"')

# 자바스크립트 confirm 창 호출
res = output.eval_js(f'confirm("{js_msg}")')

if not res:
    print("🚫 사용자에 의해 작업이 중단되었습니다.")
    # 이후 셀들이 실행되지 않도록 에러를 발생시키거나 흐름 제어
    raise SystemExit("작업 중단")
else:
    print("✅ 확인 완료. 작업을 시작합니다.")

• STT 실행에서와 동일하게 입력값 확인 창 전시

2. 작업 시트 불러오기

Google Sheet 연결

#@title 📊 Google Sheet 연결
from google.colab import auth
import gspread
from google.auth import compute_engine
from google.colab import drive
import unicodedata
import re

# 구글 계정 인증 (시트 및 드라이브 접근 권한)
auth.authenticate_user()
from google.auth import default
creds, _ = default()
gc = gspread.authorize(creds)

# 시트 로드 실행
try:
    ws, all_rows = get_sheet_data(SHEET_URL, SHEET_NAME)
    print(f"✅ 시트 로드 완료: {len(all_rows)}개의 행을 발견했습니다.")
except Exception as e:
    print(f"❌ 시트 로드 실패: {e}")

• (STT 실행에서와 동일)

3. 일치도 검사

대상 필터링

#title 📥 작업 대상 필터링
# @title 📥 일치도 검사 대상 필터링
# 1. 설정된 열 문자를 인덱스로 변환
result_idx = column_to_index(RESULT_COLUMN)   # STT 결과 열
script_idx = column_to_index(SCRIPT_COLUMN)   # 원본 스크립트 열
score_idx = column_to_index(SCORE_COLUMN)     # 일치도 점수 기록 열

# 일치도 검사를 수행할 데이터 객체들을 담을 리스트
comparison_queue = []

print(f"🔍 '{SCORE_COLUMN}'열을 확인하여 미작업 대상을 필터링합니다...")

# all_rows 순회 (헤더 제외: [1:])
for i, row in enumerate(all_rows[1:]):
    try:
        # 1. 기존 점수가 있는지 확인 (중복 작업 방지)
        # 행의 길이가 점수 열 인덱스보다 짧거나, 해당 셀이 비어있어야만 작업 대상으로 분류
        has_score = len(row) > score_idx and str(row[score_idx]).strip() != ""

        if has_score:
            # 이미 점수가 기록된 행은 건너뜁니다.
            continue

        # 2. 비교에 필요한 데이터(STT 결과, 원본 스크립트)가 모두 존재하는지 확인
        if len(row) > max(result_idx, script_idx):
            stt_text = str(row[result_idx]).strip()
            script_text = str(row[script_idx]).strip()

            # 두 데이터 중 하나라도 비어있으면 비교가 불가능하므로 제외
            if stt_text and script_text:
                comparison_queue.append({
                    "row_index": i + 2,     # 시트의 실제 행 번호 (1-based + header)
                    "stt_text": stt_text,
                    "script_text": script_text
                })

    except Exception as e:
        print(f"⚠️ {i+2}행 데이터 확인 중 오류 발생: {e}")

print(f"✅ 필터링 완료: 총 {len(comparison_queue)}개의 일치도 검사 작업이 예약되었습니다.")

Score 열 확인

has_score = len(row) > score_idx and str(row[score_idx]).strip() != ""

• 점수 열에 이미 값이 있는 경우 작업이 이미 이루어진 것으로 보고 작업 대상에서 제외

필요 데이터 존재 여부 확인

# 두 데이터 중 하나라도 비어있으면 비교가 불가능하므로 제외
if stt_text and script_text:
    comparison_queue.append({
    "row_index": i + 2,     # 시트의 실제 행 번호 (1-based + header)
    "stt_text": stt_text,
    "script_text": script_text
})

• STT 결과와 정답 스크립트 둘 다 있어야 비교 가능하므로, 둘 중 하나라도 누락된 경우 작업 대상에서 제외

스크립트 일치도 측정

#@title 💯 스크립트 일치도 측정
import re

def python_levenshtein(s1, s2):
    """자바스크립트 levenshtein 함수를 파이썬으로 변환"""
    if len(s1) < len(s2):
        return python_levenshtein(s2, s1)

    if len(s2) == 0:
        return len(s1)

    previous_row = range(len(s2) + 1)
    for i, c1 in enumerate(s1):
        current_row = [i + 1]
        for j, c2 in enumerate(s2):
            insertions = previous_row[j + 1] + 1
            deletions = current_row[j] + 1
            substitutions = previous_row[j] + (c1 != c2)
            current_row.append(min(insertions, deletions, substitutions))
        previous_row = current_row

    return previous_row[-1]

def calculate_score(stt_text, script_text):
    """텍스트 전처리 후 일치도 점수(0~100) 계산"""
    # 1. 전처리: 공백 제거 및 소문자화 (정확한 비교를 위해 필요 시 선택)
    s1 = re.sub(r'[^가-힣a-zA-Z0-9]', '', stt_text)
    s2 = re.sub(r'[^가-힣a-zA-Z0-9]', '', script_text)

    if not s1 and not s2: return 100.0

    # 2. 편집 거리 계산
    distance = python_levenshtein(s1, s2)

    # 3. 점수 변환 (최대 길이 대비 일치율)
    max_len = max(len(s1), len(s2))
    score = (1 - distance / max_len) * 100
    return round(score, 2)

# --- 실제 작업 진행 ---
print(f"🚀 총 {len(comparison_queue)}건의 점수 채점을 시작합니다.")

for task in comparison_queue:
    try:
        # 점수 계산
        score = calculate_score(task['stt_text'], task['script_text'])

        # 구글 시트에 업데이트 (row_index는 1-based)
        # SCORE_COLUMN 문자를 인덱스로 변환하여 업데이트
        ws.update_cell(task['row_index'], column_to_index(SCORE_COLUMN) + 1, score)

        print(f"✅ {task['row_index']}행 채점 완료: {score}점")

    except Exception as e:
        print(f"❌ {task['row_index']}행 처리 중 오류 발생: {e}")

print("✨ 모든 작업이 완료되었습니다.")

levenshtein 함수

def python_levenshtein(s1, s2):
    """자바스크립트 levenshtein 함수를 파이썬으로 변환"""
    if len(s1) < len(s2):
        return python_levenshtein(s2, s1)

    if len(s2) == 0:
        return len(s1)

    previous_row = range(len(s2) + 1)
    for i, c1 in enumerate(s1):
        current_row = [i + 1]
        for j, c2 in enumerate(s2):
            insertions = previous_row[j + 1] + 1
            deletions = current_row[j] + 1
            substitutions = previous_row[j] + (c1 != c2)
            current_row.append(min(insertions, deletions, substitutions))
        previous_row = current_row

    return previous_row[-1]

• 기존 GAS 환경에서 사용하던 편집 거리 기반 일치도 측정 함수를 파이썬으로 변환

점수 계산

def calculate_score(stt_text, script_text):
    """텍스트 전처리 후 일치도 점수(0~100) 계산"""
    # 1. 전처리: 공백 제거 및 소문자화 (정확한 비교를 위해 필요 시 선택)
    s1 = re.sub(r'[^가-힣a-zA-Z0-9]', '', stt_text)
    s2 = re.sub(r'[^가-힣a-zA-Z0-9]', '', script_text)

    if not s1 and not s2: return 100.0

    # 2. 편집 거리 계산
    distance = python_levenshtein(s1, s2)

    # 3. 점수 변환 (최대 길이 대비 일치율)
    max_len = max(len(s1), len(s2))
    score = (1 - distance / max_len) * 100
    return round(score, 2)

• 텍스트 전처리 후 스크립트와 STT 결과를 비교해 일치도를 점수로 표현

결과 기록

for task in comparison_queue:
    try:
        # 점수 계산
        score = calculate_score(task['stt_text'], task['script_text'])

        # 구글 시트에 업데이트 (row_index는 1-based)
        # SCORE_COLUMN 문자를 인덱스로 변환하여 업데이트
        ws.update_cell(task['row_index'], column_to_index(SCORE_COLUMN) + 1, score)

        print(f"✅ {task['row_index']}행 채점 완료: {score}점")

    except Exception as e:
        print(f"❌ {task['row_index']}행 처리 중 오류 발생: {e}")

print("✨ 모든 작업이 완료되었습니다.")

• comparison_queue 내에 있는 스크립트와 STT 결과물을 채점 후 시트에 기록

0. Summary

음성 데이터 STT 검수 자동화 요약

1. 입력 설정: 구글 시트 URL 및 작업 대상 열(ID/URL) 지정
2. 필터링: 결과값이 없는 행만 추출하여 work_queue 생성
3. 정규화: Drive API로 다운로드 후, FFmpeg로 16k/Mono 규격 변환
4. STT 실행: Whisper AI를 통해 음성을 텍스트로 추출
5. 결과 기록: 변환된 텍스트를 시트에 자동 업데이트 및 로컬 파일 삭제

1. Input & Utilities Setting

입력값

SHEET_URL

• 검수할 음성 파일 url 또는 드라이브 파일 id가 포함된 Google Sheet url

  SHEET_NAME

• 전체 시트에서 작업 데이터가 위치한 세부 시트명
• 여러 시트를 포함하고 있는 경우 특정 시트를 찾아 검수 대상으로 선정

  INPUT_COLUMN

• 음성 파일 url 또는 드라이브 파일 id가 기록된 열 이름

  RESULT_COLUMN

• STT 결과를 기록할 열 이름

  (선택) LANGUAGE_COLUMN

• STT시 모델이 언어를 자동 감지하도록 하려면 미입력
• STT시 해당 음성에 대한 언어가 미리 정해져 있어, 이를 입력값에 포함하려면 열 이름 입력

확인 메시지 창

예시 이미지

실행 코드

from google.colab import output
import sys
# @title ⚠️ 설정값 최종 확인
confirm_msg = f"""
입력된 설정값을 확인하십시오:
- 입력 대상 열: {INPUT_COLUMN}
- 입력 형식: {INPUT_FORMAT}
- 결과 기록 열: {RESULT_COLUMN}
작업을 진행하시겠습니까?
"""
# 자바스크립트 confirm 창 호출
js_msg = confirm_msg.strip().replace('\n', '\\n').replace('"', '\\"')
# 자바스크립트 confirm 창 호출
res = output.eval_js(f'confirm("{js_msg}")')
if not res:
    print("🚫 사용자에 의해 작업이 중단되었습니다.")
    # 이후 셀들이 실행되지 않도록 에러를 발생시키거나 흐름 제어
    raise SystemExit("작업 중단")
else:
    print("✅ 확인 완료. 작업을 시작합니다.")

• 실행 전 입력된 설정값을 확인창으로 전시해, 사전에 입력값을 한번 더 확인하도록 함

helper 함수 세팅

column_to_index(col_str)

# 열 문자(A, B, C...)를 인덱스 숫자(0, 1, 2...)로 변환하는 유틸리티
def column_to_index(col_str):
    if(col_str == ""): return -1

    col_str = col_str.upper()
    index = 0
    for char in col_str:
        index = index * 26 + (ord(char) - ord('A') + 1)
    return index - 1

• INPUT_COLUMN과 RESULT_COLUMN 등을 열 문자로 입력하므로, 이를 숫자로 변환하는 함수
• A,B,C~X,Y,Z까지만 대응 (AA, AB, AC~ 등 이후 열 문자에는 대응하지 못함)

get_sheet_data(url, sheet_name)

def get_sheet_data(url, sheet_name):
    """지정한 URL과 시트명으로 데이터를 로드합니다."""
    sh = gc.open_by_url(url)
    worksheet = sh.worksheet(sheet_name)
    # 셀의 수식(Formula)까지 가져와서 RichText 링크를 추출할 수 있게 함
    data = worksheet.get_all_values(value_render_option='FORMULA')
    return worksheet, data

• 입력한 SHEET_URL과 SHEET_NAME을 통해 데이터 로드
• RichText를 추출해 내부에 포함된 링크까지 추출할 수 있게 함

  get_file_id(url), get_url(rich_text)

  def get_file_id(url):
    """URL에서 구글 드라이브 파일 ID 추출"""
    match = re.search(r'[-\w]{25,}', str(url))
    return match.group(0) if match else None
  

def get_url(rich_text): """Rich Text에서 구글 파일 공유 url 추출""" match = re.search(r'HYPERLINK("(.*?)",', str(rich_text)) return match

- `get_file_id(url)` : `파일 id`만 있으면 드라이브 상의 파일에 접근할 수 있으므로, url에서 `파일 id`를 추출하는 함수 
- `get_url(rich_text)` : `Rich Text`에서 `파일 url` 추출

#### `extract_clean_id(cell_content)`

def extract_clean_id(cell_content): """ URL, ID, Formula(RichText) 등 다양한 형태에서 순수 파일 ID만 추출합니다. INPUT_FORMAT의 값이 ID, URL, Rich_Text 인지에 따라 구분 """ if not cell_content: return None

#1. 입력 형식에 따라 cell_content 처리 분기
file_id = ""

if(INPUT_FORMAT == "id"):
  file_id = cell_content
elif(INPUT_FORMAT == "url"):
  file_id = get_file_id(cell_content)
elif(INPUT_FORMAT == "rich_text"):
  file_id = get_file_id(get_url(cell_content))
return file_id

- `INPUT_FORMAT` 값에 따라 셀 값에서 순수 `파일 id`를 추출

## 2. Load Sheet & Filtering
### 작업 대상 불러오기

#@title 📊 Google Sheet 연결 및 ID 추출 로직 from google.colab import auth import gspread from google.auth import compute_engine from google.colab import drive import unicodedata import re

구글 계정 인증 (시트 및 드라이브 접근 권한)

auth.authenticate_user() from google.auth import default creds, _ = default() gc = gspread.authorize(creds)

시트 로드 실행

try: ws, all_rows = get_sheet_data(SHEET_URL, SHEET_NAME) print(f"✅ 시트 로드 완료: {len(all_rows)}개의 행을 발견했습니다.") except Exception as e: print(f"❌ 시트 로드 실패: {e}")

- 시트 불러오기 및 작업물 개수(행 개수) 확인
### 작업 대상 필터링

#@title 📥 작업 대상 필터링

1. 설정된 열 문자를 인덱스로 변환

input_idx = column_to_index(INPUT_COLUMN) result_idx = column_to_index(RESULT_COLUMN) lang_idx = column_to_index(LANGUAGE_COLUMN)

작업을 수행할 데이터 객체들을 담을 리스트

work_queue = []

print(f"🔍 '{INPUT_COLUMN}'열에서 ID 추출 및 '{RESULT_COLUMN}'열 결과 확인 중...")

all_rows 순회(헤더 건너뛰기 설정)

for i, row in enumerate(all_rows[1:]): try: # 1. 기존 결과값이 있는지 확인 (중복 작업 방지) # 행의 길이가 결과 열 인덱스보다 짧거나, 해당 셀이 비어있어야만 작업 대상으로 분류 has_result = len(row) > result_idx and str(row[result_idx]).strip() != ""

    if has_result:
        # 이미 결과가 있는 행은 건너뜁니다.
        continue

    # 2. 입력 열에서 ID 추출
    if len(row) > input_idx:
        cell_val = row[input_idx]
        clean_id = extract_clean_id(cell_val)

        if clean_id:
          # 해당 행의 언어 설정 값 확인
          spec_lang = None

          if(lang_idx < 0): spec_lang = None
          elif (len(row) > lang_idx):
            val = str(row[lang_idx]).strip().lower()
            # 값이 비어있지 않다면 해당 값을 사용 (예: "ko", "en", "ja")
            spec_lang = val if val else None

    print(spec_lang)

    work_queue.append({
        "row_index": i + 2,
        "file_id": clean_id,
        "local_path": "",
        "stt_result": "",
        "language": spec_lang  # 언어 설정값 저장 (없으면 None)
        })

except Exception as e:
    print(f"⚠️ {i+1}행 데이터 확인 중 오류 발생: {e}")

print(f"✅ 필터링 완료: 총 {len(work_queue)}개의 새로운 작업이 예약되었습니다.")

#### 작업물 객체 구조

{ "row_index": i + 2, "file_id": clean_id, "local_path": "", "stt_result": "", "language": spec_lang # 언어 설정값 저장 (없으면 None) }

- `row_index` : 작업물이 위치한 행 번호, 향후 결과물 기록시 동일한 행에 기록
- `file_id` : `Rich Text`, `url` 등에서 추출된 `파일 id`
- `local_path` : 음성 파일 다운로드 후 저장한 로컬 파일 경로
- `stt_result` : `Whisper AI`를 통해 STT한 결과
- `language` : 음성 파일 언어 입력값 (없을 경우 `None`)

#### 작업대상 필터링

    # 1. 기존 결과값이 있는지 확인 (중복 작업 방지)
    # 행의 길이가 결과 열 인덱스보다 짧거나, 해당 셀이 비어있어야만 작업 대상으로 분류
    has_result = len(row) > result_idx and str(row[result_idx]).strip() != ""

    if has_result:
        # 이미 결과가 있는 행은 건너뜁니다.
        continue

- `RESULT_COLUMN`에 값이 없는 경우에만 작업 대상으로 선정해, 처리해야할 음성 파일 개수를 줄임
- 작업 대상으로 선정된 행에서 필요한 값을 추출해 `work_queue`에 추가

## 3. Download & Normalize
### 음성파일 다운로드
#### Google 드라이브 세팅

#@title 📁 작업용 드라이브 환경 세팅 #@markdown 작업용 임시 디렉토리 및 드라이브 api 서비스 빌드 import os import subprocess from googleapiclient.discovery import build from googleapiclient.http import MediaIoBaseDownload import io

1. 작업용 임시 디렉토리 생성

TEMP_DIR = "stt_workspace" os.makedirs(TEMP_DIR, exist_ok=True)

Drive API 서비스 빌드 (이미 인증된 서비스 객체가 있다고 가정)

drive_service = build('drive', 'v3', credentials=creds)

- 작업용 임시 디렉토리를 생성해 작업 대상인 음성 파일을 다운로드 및 저장할 공간으로 사용
#### 다운로드 및 변환

@title 🎙️음성 파일 다운로드 및 변환

print(f"🚀 총 {len(work_queue)}개의 작업에 대한 파일 처리를 시작합니다.")

for task in work_queue: file_id = task['file_id'] row_num = task['row_index']

# 임시 파일 경로 설정
download_path = os.path.join(TEMP_DIR, f"temp_{file_id}")
output_path = os.path.join(TEMP_DIR, f"audio_{row_num}.wav")

try:
    # A. 구글 드라이브에서 파일 다운로드
    request = drive_service.files().get_media(fileId=file_id)
    fh = io.FileIO(download_path, 'wb')
    downloader = MediaIoBaseDownload(fh, request)

    done = False
    while done is False:
        status, done = downloader.next_chunk()

    # B. FFmpeg를 이용한 포맷 변환 (Whisper 권장 사양: 16k, mono)
    # -y: 기존 파일 덮어쓰기, -ar: 샘플링 레이트, -ac: 채널 수
    command = [
        'ffmpeg', '-y', '-i', download_path,
        '-ar', '16000', '-ac', '1',
        output_path
    ]

    # subprocess를 사용하여 ffmpeg 실행
    result = subprocess.run(command, capture_output=True, text=True)

    if result.returncode == 0:
        # 성공 시 객체 업데이트
        task['local_path'] = output_path
        print(f"✅ [행 {row_num}] 변환 완료: {output_path}")
    else:
        print(f"❌ [행 {row_num}] FFmpeg 오류: {result.stderr}")

    # 다운로드한 원본 임시 파일 삭제 (용량 관리)
    if os.path.exists(download_path):
        os.remove(download_path)

except Exception as e:
    print(f"❌ [행 {row_num}] 처리 중 예외 발생: {e}")

print("\n📦 모든 파일 다운로드 및 변환 공정 완료.")

- `work_queue`를 순회하며 작업 대상을 다운로드 및 변환하고, 객체에 값을 설정

#### 다운로드

for task in work_queue: file_id = task['file_id'] row_num = task['row_index']

# 임시 파일 경로 설정
download_path = os.path.join(TEMP_DIR, f"temp_{file_id}")
output_path = os.path.join(TEMP_DIR, f"audio_{row_num}.wav")

try:
    # A. 구글 드라이브에서 파일 다운로드
    request = drive_service.files().get_media(fileId=file_id)
    fh = io.FileIO(download_path, 'wb')
    downloader = MediaIoBaseDownload(fh, request)

    done = False
    while done is False:
        status, done = downloader.next_chunk()

    ...

- `파일 id`를 통해 드라이브 상에 업로드되어 있는 음성 파일에 접근하고, 이를 다운로드 한 뒤, 행 번호를 통해 파일을 네이밍
#### 형식 변환

    ...

    # B. FFmpeg를 이용한 포맷 변환 (Whisper 권장 사양: 16k, mono)
    # -y: 기존 파일 덮어쓰기, -ar: 샘플링 레이트, -ac: 채널 수
    command = [
        'ffmpeg', '-y', '-i', download_path,
        '-ar', '16000', '-ac', '1',
        output_path
    ]

    # subprocess를 사용하여 ffmpeg 실행
    result = subprocess.run(command, capture_output=True, text=True)

    if result.returncode == 0:
        # 성공 시 객체 업데이트
        task['local_path'] = output_path
        print(f"✅ [행 {row_num}] 변환 완료: {output_path}")
    else:
        print(f"❌ [행 {row_num}] FFmpeg 오류: {result.stderr}")

    # 다운로드한 원본 임시 파일 삭제 (용량 관리)
    if os.path.exists(download_path):
        os.remove(download_path)

- `ffmpeg`를 사용해 `16k, mono, PCM` 사양에 맞춰 음성 파일을 변환해, 이전에 발생했던 [입력 형식 오류](https://velog.io/@allhoon_718/GAS-음성-파일-format-정규화)를 방지
- `output_path`를 객체에 저장하고, 기존 원본 음성파일을 삭제

## 4. STT
### STT 실행

#@title 🎧 Whisper 사용 STT 실행

import whisper import torch

모델 로드 (다중 언어 대응을 위해 turbo 또는 large-v3 권장)

model = whisper.load_model("turbo")

print(f"🎙️ 총 {len(work_queue)}개 작업에 대해 STT를 시작합니다.")

for task in work_queue: audio_path = task.get('local_path') target_lang = task.get('language') # 시트에서 가져온 언어 값 (None일 수 있음)

if audio_path and os.path.exists(audio_path):
    try:


        # 로그 출력: 언어 설정 여부에 따른 메시지 차별화
        has_language = target_lang is not None

        if(has_language):
          log_msg = f"🌐 언어 지정({target_lang})"
          result = model.transcribe(
            audio_path,
            language=target_lang,
            task="transcribe",
            beam_size=5,
            fp16=True if torch.cuda.is_available() else False)
        else:
          log_msg = "🔍 언어 자동 감지"
          result = model.transcribe(
            audio_path,
            task="transcribe",
            beam_size=5,
            fp16=True if torch.cuda.is_available() else False)

        print(f"📝 [행 {task['row_index']}] {log_msg} 처리 중...")

        task['stt_result'] = result['text'].strip()

        # 파일 관리: 변환 완료 후 로컬 파일 삭제
        os.remove(audio_path)

    except Exception as e:
        print(f"❌ [행 {task['row_index']}] STT 오류: {e}")
        task['stt_result'] = f"ERROR: {str(e)}"

- `work_queue`에 저장된 작업 대상의 STT 실행
- `language`값이 있는지, 없는지를 구분하여 실행
-  STT 결과물 출력 후 로컬의 음성파일 삭제

### 결과물 기록

#@title 📝 Google Sheet에 기록

RESULT_COLUMN을 인덱스로 변환 (예: "F" -> 6)

result_col_idx = column_to_index(RESULT_COLUMN) + 1

print(f"📊 '{RESULT_COLUMN}'열에 결과를 기록합니다...")

gspread의 update_cell은 호출이 많으면 느려지므로,

작업량이 많다면 범위를 지정해 한 번에 업데이트하는 것이 좋으나

여기서는 직관적인 row_index 기반 업데이트를 사용합니다.

success_count = 0

for task in work_queue: if task['stt_result']: try: # ws는 이전 단계에서 정의된 gspread 워크시트 객체 ws.update_cell(task['row_index'], result_col_idx, task['stt_result']) success_count += 1 except Exception as e: print(f"❌ [행 {task['row_index']}] 시트 기록 실패: {e}")

print(f"\n✅ 최종 완료: {success_count}개의 결과가 시트에 반영되었습니다.")

- `work_queue`에 종합되어 있는 정보를 통해 시트에 최종 결과물을 기록
- `RESULT_COLUMN`에 STT 결과를 기록

Summary

1. 문제: Android/iOS 기기별 음성 녹음 포맷 불일치로 인한 Whisper AI 인식 오류 발생
2. 해결: GCF(FFmpeg) 서버 방식을 거쳐, 최종적으로 브라우저(Web Audio API) 기반 재인코딩 파이프라인 구축
3. 성과: Web Audio API 기반의 브라우저 재인코딩 공정으로 전환하여, 서버 비용 없이 녹음본을 자동 검수할 수 있는 데이터 정규화 파이프라인을 구축

Problem

배경

• 불특정 다수의 사람에게 녹음할 스크립트(발화문)을 제공한 뒤, 녹음된 음성의 정확도를 검사하기 위해 OpenAI의 Whisper를 통한 STT 실시
• 기기별로, 또는 확장자에 따라 Whisper가 인식하지 못하는 경우가 발생

  에러 유형

  1. 모델이 인식할 수 없는 file format

     Error: Invalid file format. Supported formats: ['flac', 'm4a', 'mp3', 'mp4', 'mpeg', 'mpga', 'oga', 'ogg', 'wav', 'webm']

  2. 파일 확장자와 file format 불일치 voice_record.m4a ≠ audio/mpeg ( getContentType()으로 확인되는 파일 형식 )

원인 분석

녹음 환경별 차이

• 다양한 환경에서 녹음된 음성파일을 녹음된 기기의 운영체제로 구분했을 때, 위와 같은 패턴이 관찰됨
• IOS의 경우 모든 음성 파일이 에러를 발생시키지는 않았지만,audio/x-m4a 포맷은 에러를 발생시킴
• Android의 경우 명시된 확장자와 실제 mimeType이 일치하지 않았으며, Whisper에 입력시 에러를 발생시킴

예상 원인 및 목표

• 파일 포맷이 Supported format이 아닐 경우 에러 발생
• 파일의 확장자뿐만 아니라 내부 오디오 컨테이너의 정합성 검사시, Android 기기에서 녹음된 음성파일의 불일치 특성을 '사용 불가능한 포맷'으로 인식하여 처리를 거부

*→ 파일 형식을 변환하여 Whisper에 입력 가능하도록 할 방법을 고안해야함 *

Solution

시도 1. Blob 재구성을 통한 MIME 타입 정규화

const fileName = audioBlob.getName();
const extensionMatch = fileName.match(/\.([a-z0-9]+)$/i);
const extension = extensionMatch ? extensionMatch[1].toLowerCase() : "";

const mimeTypeMap = { 
"m4a": "audio/x-m4a",
"mp3": "audio/mpeg", 
"wav": "audio/wav", 
"mp4": "audio/mp4", 
"aac": "audio/aac" };

const newFileName = “input_audio.” + extension; 

const newAudioBlob = Utilities.newBlob(
    audioBlob.getBytes(), 
    mimeTypeMap[extension], 
    newFileName
  );

• 가장 처음 시도한 방법은 audioBlob을 재구성해, 파일명의 확장자와 동일한 mimeType을 갖도록 수정
• 파일명의 확장자를 정규식으로 확인 후 mimeTypeMap 을 통해 해당하는 mimeType을 찾아 입력

결과

• mimeType 을 일치시켰으나, 여전히 Whisper는 이를 사용 불가능한 포맷으로 인지함
• 새로운 Blob을 생성하는 것으로는 해결할 수 없으며, 재인코딩이 필요함을 확인함

시도 2. 자체 배포 API를 통한 파일 형식 변환

현재 작업 환경 개요

• 불특정 다수의 인원이 음성파일을 각각의 Google Drive 폴더에 업로드
• 파일들의 링크 또는 ID를 평가용 Google Sheet에 취합하여 기록
• GAS(Google App Script)를 통해 함수를 실행해 링크/ID를 통해 파일에 접근해, 로컬에 파일을 저장하지 않고 검수 작업을 자동화

→ 파일을 저장하지 않고 GAS 함수 실행 중에 음성 파일 형식 변환 후, Whisper에 입력하는 프로세스

전체 검수 프로세스 개요

검수 대상인 음성 링크를 취합 → 음성 링크에서 음성 파일 불러오기 
→ 음성 파일 포맷 변환 (현재 미구현) → Whisper 모델을 통한 녹음 파일 전사
→ 전사된 스크립트를 통해 녹음 정확도 확인

GCF를 통한 파일 포맷 변환

• 음성 파일 포맷 변환을 위해서는 ffmpeg 라이브러리 필요하나, GAS 환경에서는 외부 라이브러리를 직접 실행할 수 없었음
• GCF(Google Cloud Functions)는 Python 환경에서 ffmpeg를 자유롭게 호출할 수 있는 서버리스 환경을 제공하므로, ffmpeg를 실행하기 위한 도구로 적합해 ffmpeg활용한 파일 수신-변환-반환 함수를 작성해 배포함

import functions_framework
from flask import request, send_file
import subprocess
import io
import os
import tempfile
import imageio_ffmpeg as ff

@functions_framework.http
def audio_converter(request):
    # 1. 파일 수신 확인
    input_file = request.files.get('file')
    if not input_file:
        return "No file provided", 400

    input_data = input_file.read()
    print(f"DEBUG: Received data size: {len(input_data)} bytes")

    # 2. 임시 파일 생성 및 쓰기
    # 안드로이드의 partial file 에러를 방지하기 위해 데이터를 파일로 만듭니다.
    with tempfile.NamedTemporaryFile(delete=False, suffix=".m4a") as temp_in:
        temp_in.write(input_data)
        temp_in_path = temp_in.name

    # 3. 출력 데이터를 담을 경로 설정
    temp_out_path = temp_in_path + ".wav"

    try:
        ffmpeg_path = ff.get_ffmpeg_exe()

        # 4. FFmpeg 명령어 실행 (파이프 대신 파일 경로 사용)
        command = [
            ffmpeg_path,
            '-y',               # 기존 파일 덮어쓰기 허용
            '-i', temp_in_path, # 임시 입력 파일 경로
            '-ar', '16000',
            '-ac', '1',
            '-f', 'wav',
            '-acodec', 'pcm_s16le',
            temp_out_path      # 결과를 파일로 바로 저장
        ]

        process = subprocess.run(
            command, 
            capture_output=True, 
            text=True
        )

        if process.returncode != 0:
            print(f"FFmpeg Error: {process.stderr}")
            return f"Conversion failed: {process.stderr}", 500

        # 5. 변환된 파일 읽어서 반환
        with open(temp_out_path, 'rb') as f:
            wav_data = f.read()

        print(f"DEBUG: Final WAV size: {len(wav_data)} bytes")
        return send_file(io.BytesIO(wav_data), mimetype='audio/wav')

    except Exception as e:
        print(f"Exception: {str(e)}")
        return f"Internal Error: {str(e)}", 500

    finally:
        # 6. 사용한 임시 파일 즉시 삭제 (메모리 확보)
        if os.path.exists(temp_in_path):
            os.remove(temp_in_path)
        if os.path.exists(temp_out_path):
            os.remove(temp_out_path)

• 배포한 url을 GAS에서 호출해 음성 파일을 변환

  //GCF 변환 서버를 통해 안드로이드 오디오를 정규화된 WAV로 변환
  function getNormalizedWavBlob(originalBlob) {
  const gcfUrl = "(url)"; // GCF 콘솔에서 복사한 URL 입력
  
  // 파일명과 타입을 명시적으로 재지정하여 Blob 생성
  const fileToUpload = Utilities.newBlob(
    originalBlob.getBytes(),
    originalBlob.getContentType(),
    "input_audio.m4a" // 파일명을 명시해야 GCF가 파일로 인식함
  );
  
  const options = {
    method: "post",
    payload: {
      file: fileToUpload // GCF의 request.files['file']과 매칭
    },
    muteHttpExceptions: true
  };
  
  const response = UrlFetchApp.fetch(gcfUrl, options);
  
  if (response.getResponseCode() !== 200) {
    console.log(response.getBlob().getName());
    console.log(response.getBlob().getContentType());
    throw new Error("변환 서버 오류: " + response.getContentText());
  }
  
  return response.getBlob();
  }

  결과

• 파일을 변환해야하는 이벤트가 발생할 때마다, 해당 GAS 함수를 호출해 음성 파일 형식을 변환해 Whisper 모델을 활용한 안정적인 녹음본 검수가 가능해짐

한계점

• GCF에서 초기 제공하는 기본 크레딧이 충분해 작업해야하는 음성파일의 개수가 해당 크레딧 분량을 초과하지는 않음 하지만 음성 파일 개수가 늘어나거나, 향후에 사용할 수 있는 범용 툴로 사용하기 위해서는 GCF의 서버를 빌리는 것에서 발생하는 비용을 없앨 방안이 필요함
• 현재는 음성 파일의 길이가 1분을 초과하지 않아 변환 과정에 긴 시간이 소요되지 않으나, 용량이 큰 파일을 대량으로 다룰 경우 속도 저하 또는 서버 오류가 발생할 수 있음

→ 브라우저 자원을 활용해 파일 형식을 변환할 수 있는 방법을 고안

시도 3. 브라우저 기반 파일 포맷 변환

브라우저 기반 방식 고안

• 음성 파일 검수 과정에서 무음구간 계산 기능을 구현할 때,GAS에서 지정된 HTML 파일을 실행해 음성 파일을 base64로 인코딩해 전달해, 브라우저 기능을 활용해 무음 구간을 계산하는 기능을 사용
• 해당 방법에서 아이디어를 얻어, 클라이언트(브라우저) 내 자원을 활용한 파일 포맷 변환 방식을 고안함

  전체 프로세스 개요

  GAS에서 파일 변환 실행할 HTML 파일 호출 → HTML에 정의된 창(이하 Modal) 렌더링 → Modal내부의 'javaScript' 함수 'initProcess' 실행 → Modal이 GAS의 'getWorkIds' 함수를 호출해 작업할 파일 Id 배열 요청 → 작업 ID 목록을 순회하며 GAS의 'getBase64'함수에 Id 입력해, 음성 파일의 base64 인코딩 값 요청 → Web Audio API를 통해 base64 값을 audioBuffer로 변환 후 wav 형식으로 파일 변환 → 변환된 파일을 GAS의 Whisper 실행 함수에 전달해 STT 실행
• GAS에서 실행해야 하는 기능과 브라우저 상에서 실행해야 하는 기능을 구분하고, 서로 간에 필요한 경우 어떻게 정보를 주고 받을지 고려
• GAS와 브라우저 간에는 base64 형식은 가능하지만, Blob과 같은 복잡한 구조는 전달할 수 없는 점을 고려해야함

GAS 실행 함수

function getAudioBase64(fileId) {
  const f = DriveApp.getFileById(fileId);

  const blob = f.getBlob();
  const bytes = blob.getBytes();
  const b64 = Utilities.base64Encode(bytes);

  return {
    fileId,
    name: f.getName(),
    mimeType: blob.getContentType(),
    base64: b64
  };
}

• ** DriveApp 사용** : 파일 id를 통해 음성 파일을 불러올 때는 DriveApp 사용이 필요하나, 해당 기능은 구글 서버 사이드에서만 작동하므로 GAS 상에서 음성 파일을 찾아 필요한 정보를 객체로 묶어 Modal로 전달
• Base64 인코딩 : 음성파일을 직접 전달할 수 없기 때문에 base64 형태로 인코딩 해 음성파일을 전달

function writeResultToSheet(base64Data, index) {
  try {
    // 설정 정보 (실제 시트 및 컬럼 위치로 수정 필요)    

    (기록할 시트 정보)

    // Base64 문자열을 디코딩하여 Blob 생성
    const decodedData = Utilities.base64Decode(base64Data);
    const audioBlob = Utilities.newBlob(decodedData, "audio/wav", "audio_" + (index + 1) + ".wav");

    //Whisper를 통한 STT
    const result = transcribeWithWhisper(audioBlob)

    (시트에 결과 기록)

  } catch (e) {
    Logger.log("저장 실패: " + e.message);
    return { success: false, error: e.message };
  }
}

• Base64 디코딩 : 음성파일, Blob으로 입력받을 수 없으므로 Base64 형식을 입력받아, 새로운 Blob 생성

Modal 실행 함수

    async function base64ToAudioBuffer(base64) {
      const binary = atob(base64);
      const len = binary.length;
      const bytes = new Uint8Array(len);
      for (let i = 0; i < len; i++) bytes[i] = binary.charCodeAt(i);

      const audioCtx = new (window.AudioContext || window.webkitAudioContext)();
      return await audioCtx.decodeAudioData(bytes.buffer);
    }

• 입력받은 Base64 형식 값을 AudioBuffer로 변환
• 브라우저의 오디오 연산 환경인 AudioContext 객체를 생성
• AudioContext 객체가 입력된 데이터 포맷을 자동으로 해석해 AudioBuffer 객체를 반환

    function bufferToWav(buffer) {
      // buffer가 정상인지 체크
      if (!buffer) {
        addLog("오류: buffer가 비어있습니다.");
      }
      if (typeof buffer.getChannelData !== 'function') {
        addLog("오류: 전달된 객체가 AudioBuffer 형식이 아닙니다.");
      }

      const sampleRate = buffer.sampleRate;
      const length = buffer.length * 2;
      const view = new DataView(new ArrayBuffer(44 + length));
      const writeStr = (off, s) => {
        for (let i = 0; i < s.length; i++) view.setUint8(off + i, s.charCodeAt(i));
      };

      // 1. WAV 헤더 작성 단계
      addLog("WAV 헤더 작성 중...");
      writeStr(0, 'RIFF');
      view.setUint32(4, 36 + length, true);
      writeStr(8, 'WAVE');
      writeStr(12, 'fmt ');
      view.setUint32(16, 16, true);
      view.setUint16(20, 1, true); // PCM
      view.setUint16(22, 1, true); // Mono
      view.setUint32(24, sampleRate, true); //sampleRate 설정
      view.setUint32(28, sampleRate * 2, true);
      view.setUint16(32, 2, true);
      view.setUint16(34, 16, true);
      writeStr(36, 'data');
      view.setUint32(40, length, true);

      // 2. 오디오 데이터 작성 단계
      const channelData = buffer.getChannelData(0);
      for (let i = 0, off = 44; i < channelData.length; i++, off += 2) {
        const s = Math.max(-1, Math.min(1, channelData[i]));
        view.setInt16(off, s < 0 ? s * 0x8000 : s * 0x7FFF, true);
      }

      const uint8Array = new Uint8Array(view.buffer);

      // 바이트 배열을 이진 문자열로 변환 후 btoa 실행
      let binary = '';
      const len = uint8Array.byteLength;
      for (let i = 0; i < len; i++) {
          binary += String.fromCharCode(uint8Array[i]);
          }

      return window.btoa(binary); // 순수 Base64 문자열 반환
    }

• 리샘플링된 Audiobuffer를 라이브러리 없이 바이트 단위로 파일 헤더를 작성해 wav 포맷으로 패키징
• 44바이트의 헤더 작성 이후, wav 포맷에 맞춰 소수점 형태의 데이터를 정수형으로 바꾸면서도 클리핑 등 문제가 생기지 않도록 데이터 정제
• 새로 구성한 AudioBuffer를 이진 문자열로 변환해 Base64 형식으로 바꿔 GAS에 안정적으로 전달할 수 있게함

결론

• GCF를 거치지 않고 브라우저 내 기능을 활용해 음성 파일을 변환하고, 이렇게 변환한 음성 파일을 Whisper에 입력해 STT를 실행하는 프로세스를 구축할 수 있었음

Result

업무 관련

• 여러 환경에서 녹음된 다양한 확장자를 가진 음성파일을 매끄럽게 처리해 STT 검수를 실시하는 프로세스 구축

  학습 관련

• GCF를 활용한 자체 api 배포를 통해 ffmpeg 활용 음성 파일 변환 기능 구현
• 브라우저의 Web Audio API와 자체 헤더 및 채널 샘플링 과정을 통해 클라이언트 상에서의 음성 파일 변환 기능 구현
• 음성 파일 데이터의 byte 단위의 low-level 구조 학습
• GAS-브라우저 간 통신을 위한 Base64 개념 학습 및 활용

Keys

• key는 데이터를 구분하기 위한 요소

  구분

  1. Super Key

• 표에서 행을 구분할 수 있는 속성 값

  2. Candidate Key

• 단일하며 단순성을 충족하는 Super Key

  3. Primary Key

• Candidate Key 중에서 대표성을 가진 것을 선택
• null일 수 없고, 중복될 수 없으며 안정적이어야 함

  4. Alternate Key

• Candidate Key 중에서 Primary Key로 선택되지 않은 나머지

Map

• 검색을 위한 자료 구조
• key를 활용해 빨리 자료를 탐색할 수 있도록 설계됨
• key - value 쌍을 통해 빠르게 값을 찾을 수 있음
• 삽입, 검색, 삭제 기능을 가지는 것이 가장 중요
• Array, Binary Search Tree 등을 통해서도 구현할 수 있지만 Hashing을 통해 구현하는 것이 굉장히 빠른 검색 속도를 가짐
• 정렬되지 않은 배열을 사용할 경우 O(n), 정렬된 배열은 O(log₂n)의 시간 복잡도를 가짐
• 균형잡힌 이진 검색 트리 또한 O(log₂n)의 시간 복잡도를 가짐
• 그러나 Hash를 통하면 O(1)의 시간 복잡도로 검색 기능을 구현 가능. 이 때문에 데이터가 많을수록 Hash를 사용하는 것이 유리

7. Heap Sort

개념

• Heap 자료 구조를 활용한 정렬
• Heap Tree에 모든 자료를 입력한 후 이를 다시 제거해 정렬을 실시

  정리

• Unstable & External Sort
• Merge Sort보다 메모리 효율적이며 Quick Sort보다 예상가능한 실행 시간을 가짐
• Quick Sort보다 느릴 수 있지만 보다 안정적이고 예측가능하다는 장점

8. Radix Sort

개념

• 비교 연산 없이 자릿수를 통해 정렬을 실시
• 비교 기반 연산은 O(n * log₂n)보다 빠를 수 없지만 Radix Sort는 가능함
• 연산을 위해 추가적인 메모리 공간을 필요로 함

  구현

  1자리 숫자만 가진 배열일 경우

• 0~9까지의 배열을 구성한 후, 정렬하려는 배열의 각각의 값을 인덱스로 하는 위치에 값을 저장
• 이후 해당 배열을 그대로 출력하면 정렬된 상태를 유지 (자릿수가 늘어나면 같은 작업을 자릿수별로, 낮은 자릿수부터 실시)

  정리

• d의 자릿수를 가진 값들을 가진 배열이 n개의 요소를 가질 경우 O(d * n)의 시간 복잡도를 가짐 (일반적으로 d가 n보다 작은 경우가 많으므로 O(n))
• 그러나 사용할 수 있는 경우가 제한적; 숫자, 알파벳 등
• Stable & External Sort

9. Counting Sort

개념

• 최댓값이 정해진 요소들에 대해서 사용가능
• 각각의 값이 얼마나 자주 등장했는지를 통해 값을 정렬
• 비교 기반 정렬이 아니므로 O(n + k)의 복잡도를 가지며, 여기서 k는 가능한 값의 범위

  구현

1. 해당 배열의 최대 값을 찾음
2. 최대값을 크기로 하는 배열을 새로 구성
3. 이후 주어진 배열에서 각각의 값이 얼마나 자주 등장했는지를 최대값 크기 배열에 입력
4. 최대값 크기 배열을 누적된 형태로 연산
5. 해당 배열을 기준으로 정렬 실시

   정리

• 시간 복잡도와 메모리 공간 모두 O(n + k)
• 범위가 정해진 정수 배열에서 효과적
• Stable & External Sort
• k가 n보다 지나치게 큰 경우 메모리 공간 활용이 비효율적일 수 있음(빈 곳이 많아짐)

4. Shell Sort

개념

• Donald L.Shell에 의해서 제시되었으며 Insertion Sort가 부분적으로 정렬된 배열에서 매우 빠르다는 점에서 착안 (↔완전히 정렬되어 있지 않은 배열에서는 매우 느림)
• Insertion Sort의 단점은 인접한 곳으로 삽입되는 경우가 아닌, 먼 곳에 삽입될 경우 많은 요소들의 이동이 발생한다는 점
• Insertion Sort와 달리 한 번에 정렬을 실시하지 않고 gap 값에 따라 Sublist를 구성
• 단계마다 gap 값은 줄어들고 Sublist의 크기는 커짐 (일반적으로 gap은 처음에는 n/2로 시작해서 단계마다 절반이 됨)
• 마지막 단계에서 gap은 1 (하나의 Sublist만 남은 상태)
• 배열은 작은 단위로 나누고 Insertion Sort를 실시, 이후 부분적으로 정렬된 더 큰 Sublist를 다시 Insertion Sort, 해당 과정을 반복함으로 부분적으로 정렬된 곳에서 빠른 Insertion Sort의 장점을 활용

구현

static void sortIncInsertion (int list[], int first, int last, int gap)
{
    int i, j, key;
    for( i=first+gap ; i<=last ; i=i+gap){
        key = list[i];
        for( j=i-gap ; j>=first && key<list[j] ; j=j-gap )
            list[j+gap]=list[j];
        list[j+gap]=key;
    }
}

void shellSort ( int list[], int n )
{
    for( int gap=n/2; gap>0; gap = gap/2 ) {
        printArray( list, n, "Shell...." );

        if( (gap%2) == 0 ) gap++;
        for( int i=0 ; i<gap ; i++ )
            sortIncInsertion( list, i, n-1, gap );
    }
}

• gap만큼 서로 떨어진 요소들을 모아 Sublist 구성 (gap이 5라면, 0번과 5번 요소가 하나의 Sublist)
• 배열내에 많은 요소를 이동시키는 것을 최소화
• 주어진 데이터의 특성에 따라 gap을 설정해야하며 gap에 따라 효율성이 달라짐
• 최악의 경우에는 O(n^2), 그러나 일반적으로는 O(n^1.3~1.5)

정리

• Insertion Sort의 장점만 극대화
• 기본적인 Insertion Sort 비교와 이동 횟수 모두를 줄임
• Unstable & Internal Sort (매우 적은 일부 메모리를 추가적으로 요구함)
• 앞선 Insertion, Selection, Bubble Sort보다 효율적

5. Merge Sort

개념

• 주어진 배열을 같은 크기의 두 Sublist로 나누고, Sublist들을 각각 정렬한 후, 다시 합침
• divide and conquer 전략을 활용: 문제를 보다 작은 단위로 나누고 각각을 해결한 뒤 합치는 방법

  예시

  4 2 3 1 6 5

  1. Divide : 4 2 3, 1 6 5 - 2개의 Sublist로 나눔
  2. Conquer : 2 3 4, 1 5 6 - 각각을 정렬
  3. Combine : 1 2 3 4 5 6 - 다시 합쳐 하나의 배열로 만듦

구현

static void Merge(int list[], int left, int mid, int right)
{  
    int i, j, k = left;                
    static int sorted[MAX_SIZE];

    for( i=left, j=mid+1 ; i<=mid && j<=right ; )
        sorted[k++]    = (list[i]<=list[j])
                    ? list[i++]
                    : list[j++];

    if( i > mid )
        for( int l=j ; l<=right ; l++, k++ )
            sorted[k] = list[l];
    else
        for( int l=i ; l<=mid   ; l++, k++ )
            sorted[k] = list[l];

    for( int l=left ; l<=right ; l++ )
        list[l] = sorted[l];
}
void mergeSort ( int list[], int left, int right )
{  
    if( left<right ) {
        int mid = (left+right)/2;        
        mergeSort(list, left, mid);        
        mergeSort(list, mid+1, right);    
        Merge(list, left, mid, right);    
    }
}

• 크기가 1인 배열이 될 때까지 나누고 합칠 때는 양쪽의 크기를 비교하며 합침
• 재귀 호출을 통해 배열을 나누고 정렬을 실시하므로 배열의 크기n이 2의 몇번째 지수인지에 따라 깊이가 결정되므로 log₂n의 재귀 함수 호출이 발생
• 배열을 나누는 과정에서는 비교 연산이 없지만 이후 합치는 과정에서 비교 연산이 발생
• 재귀 레벨마다 n회의 비교가 발생해 전체 연산은 n * log₂n
• 합치는 과정에서 임시 배열에 값이 복사되었다가 다시 반환되는데 이 과정에서 2회 이동이 발생하므로 2n * log₂n이 됨. 따라서 시간 복잡도는 O(n * log₂n)

정리

• Divide & Conquer 전략을 기반으로 함
• 반복적으로 배열을 반으로 나누는 재귀 호출과 이를 정렬하며 합치는 과정으로 구성
• Stable & External Sort

6. Quick Sort

개념

• 평균적으로 가장 빠른 알고리즘
• Merge Sort와 마찬가지로 Divide & Conquer 전략을 사용
• 배열은 완전히 반으로 나누는 Merge Sort와 달리 불균형하게 배열을 나누기도 함

  작동 방식

1. Pivot을 배열에서 하나 선택함
2. Pivot보다 작은 값을 Pivot을 기준으로 왼쪽, 큰 값은 오른쪽으로 이동
3. Pivot을 기준으로 배열이 2개의 영역으로 나뉘게 됨
4. 해당 과정을 재귀적으로 반복

구현

static int partition( int list[], int left, int right )
{
    int low        = left;                 
    int high    = right+1;
    int pivot    = list[left];                        // pivot value
    do {
        do {                
            low++;
        } while(low<=right &&list[low]<pivot);

        do {
            high--;            
        } while(high>=left && list[high]>pivot);

        if( low<high )                                // do until low meets high
            swap(list[low], list[high]);            // swap two elements
    } while(low<high);        
    swap( list[left], list[high] );                 // put pivot into proper place
    return high;
}
// quick sort function: sorts list[left..right] in place ascending order
void quickSort (int list[], int left, int right)
{  
    if( left<right ){        
       int q=partition(list, left, right);
       quickSort (list, left, q-1);        
       quickSort (list, q+1, right);    
     }
}

• 배열을 반씩 나누기 때문에 해당 작업은 log₂n만큼 발생
• 비교 연산 또한 매 단계마다 n회 발생
• 값의 이동은 비교 연산에 비해 극히 작아 무시할 수 있음
• Pivot을 처음에 잘못 설정해 지나치게 불균형한 구조가 만들어질 경우에 가장 비효율적, 이 경우 나누는 작업이 n회 발생
• 이를 방지하기 위해 하나의 요소로 Pivot을 설정하는 것보다 여러 요소의 평균이나 중간값을 사용할 수 있으며 평균적인 시간 복잡도는 θ(n * log₂n)

정리

• Divide & Conquer에 근거한 매우 효과적인 정렬
• Pivot을 설정하고 이를 기준으로 배열을 나누는 방법을 사용
• 가장 빠른 Internal 정렬 방식이나 Unstable 함
• Merge Sort보다 일반적으로 빠르고 메모리 공간도 적게 사용함
• Pivot 선정 전략에 따라 효율성이 달라질 수 있음

개념

• 배열 내의 요소를 크기에 따라 오름차순 또는 내림차순 정렬하는 것
• 특정 요소를 검색하고자 할 때 정렬된 배열일 경우 효율성이 높아지는 경우가 있어 효율성에 있어 매우 중요함
• 데이터에 따라 효율적인 정렬 방법을 사용해야하먀 데이터에 따라 효율적인 정렬 방법을 사용해야함

구분

효율성 & 복잡도에 따른 구분

간단하지만 효율적이지 않은 정렬

• Insertion Sort
• Selection Sort
• Bubble Sort

복잡하지만 효율적인 정렬

• Quick Sort
• Heap Sort
• Merge Sort
• Radix Sort

메모리 공간에 따른 구분

Internal Sort

• 정렬시 모든 데이터를 메모리에 한번에 올림

External Sort

• 정렬시 대부분은 외부 저장 공간에 있고 일부분만 메모리에 올라감

Stability에 따른 구분

Stable Sort

• 같은 값을 가진 요소의 순서가 기존 상태와 비교해 바뀌지 않는 것을 보장
• Insertion sort, Bubble sort, Merge sort가 해당

Unstable Sort

• 정렬 수행 후 기존 상태와 비교해 같은 값을 가진 요소의 순서가 바뀔 수 있음 (반드시 바뀌는 것은 아님)
• Stable Sort를 제외한 나머지

1. Selection Sort

개념

• 정렬된 요소를 담을 빈 리스트 left와 정렬되지 않은 상태의 리스트 right를 이용
• right를 순회하며 규칙에 따라 요소를 선택해 left에 추가
• left에 요소가 추가될 때마다 right의 요소가 하나씩 줄어듦
• right 리스트가 빌 때 정렬이 완료됨리스트가 빌 때 정렬이 완료됨

구현

• 실제 구현시에는 left, right라는 2개의 리스트를 사용할 필요가 없음
• 하나의 배열을 순회하며 자리를 바꾸는 것으로 정렬을 실시하므로 추가적인 메모리 공간을 필요로 하지 않는 Internal Sort임

void selectionSort (int list[], int n)
{  
    int i, j, least;
    for( i=0 ; i<n-1 ; i++) { 
        least = i;
        for(j=i+1; j<n; j++)
            if(list[j]<list[least]) least = j;
        swap(list[i], list[least]); 
    }
}

• 주어진 코드에서는 오름차순 정렬을 실시
• 0번 인덱스의 값에서부터 시작해 이후에 있는 배열 요소를 이중 반복문으로 순회하며 최소 값의 인덱스를 찾아 Swap()실시
• 실행시 n-1 + n-2 + ... + 2 + 1 횟수의 비교를 실시하므로 전체 n(n-1)/2 비교 연산 실시 최종적으로 O(n^2)의 시간 복잡도를 가짐
• Swap()과정에서도 3번의 이동이 있으나 전체 시간 복잡도는 바뀌지 않음

정리

• Selection Sort는 배열에서 가장 작은(또는 가장 큰) 요소를 찾아 정렬
• 순회 과정에서 최솟값 탐색과 자리 바꾸기 과정이 실행
• 자리를 바꾸는 과정은 횟수가 적지만 비교 과정이 많이 요소가 많아질수록 비효율적
• Unstable & Internal Sort

2. Insertion Sort

개념

• 정렬되지 않은 요소를 반복적으로 정렬될 위치에 삽입
• Selection Sort처럼 추가적인 메모리 공간을 필요로 하지 않음

구현

void insertionSort (int list[], int n )
{
    for(int i=1; i<n; i++){
        int key = list[i];
        int j;
        for(j=i-1 ; j>=0 && list[j] > key ;j--)
            list[j+1]=list[j];                    // Shift element to the right
        list[j+1]=key;
    }
}

예시

1. 4 2 3 1 6 5 - key : 2 0번 인덱스에서부터 탐색 [4] 2 3 1 6 5 - j는 0이므로 0번 인덱스에서 2보다 큰 요소 확인 ([] : 탐색 범위) 4 4 3 1 6 5 - 1칸 뒤로 미루기

2 4 3 1 6 5 - 0번 위치에 기존 key 값 저장 2. 2 4 3 1 6 5 - key : 3 [2 4] 3 1 6 5 - 탐색 범위에서 4를 이동해야함을 발견 2 4 4 1 6 5 - 1칸 뒤로 미루기 2 3 4 1 6 5 - j가 0이므로 1번 위치에 key 값을 삽입 3. 2 3 4 1 6 5 - key : 1 [2 3 4] 1 6 5 - j는 2이므로 20번 위치를 탐색 2 2 3 4 6 5 - 02번 모두 한 칸씩 뒤로 이동 1 2 3 4 6 5 - j는 -1이므로 0번 위치에 key 값을 삽입 ...

• key 값을 지정한 후 key의 위치 앞에서부터 탐색해 더 큰 값을 한칸 뒤 이동한 후 빈 자리에 key 값을 다시 입력
• 이미 정렬된 상태일 경우 O(n)의 시간 복잡도를 가짐 (순서대로 값을 비교하기만 하고 미루는 과정이 없기 때문)
• 최악의 경우 전체 비교 횟수는 n(n-1)/2로 O(n^2)의 시간복잡도를 가짐

정리

• 간단한 방법이지만 반복적으로 이웃한 요소들을 뒤로 이동시키는 과정이 포함되어 비효율적임
• Stable & Internal Sort
• 이미 거의 정렬된 경우 효율적 (정렬된 상태를 빨리 파악하고 elarly exit 할 수 있다면)

3. Bubble Sort

개념

• 인접한 두 요소를 순서를 바꾸는 것을 반복
• 두 요소를 비교하고 더 큰 요소를 뒤로, 작은 요소를 앞에 오도록 Swap
• 1번 전체 배열을 위와 같은 방법으로 순회하면 가장 큰 요소가 배열의 맨 뒤에 위치함

구현

void bubbleSort (int list[], int n)
{  
    int i, j;
    for( i=n-1 ; i>0 ; i-- ){
        for( j=0 ; j<i ; j++ )
            if(list[j]>list[j+1])            // compare adjacent elements
                 swap(list[j], list[j+1]);    // swap if they are in wrong order
    }
}

예시

1. 4 2 3 1 6 5

2 4 3 1 6 5 - 4와 2를 비교하고 자리 바꿈 2 3 4 1 6 5 - 4와 3을 비교하고 자리 바꿈 2 3 1 4 6 5 - 4 와 1을 비교하고 자리 바꿈 2 3 1 4 6 5 - 4 와 6을 비교했지만 자리를 바꿀 필요 없음 2 3 1 4 5 6 - 5 와 6을 비교하고 자리 바꿈 (가장 큰 값이 6이 맨 뒤에 오게 됨) ...(자리 바꿈이 발생하지 않는 것으로 정렬이 완료된 것을 확인)

• 비교 횟수는 언제나 n(n-1)/2로 시간 복잡도는 O(n^2)
• 이미 정렬된 배열일 경우 자리 바꿈은 발생하지 않으나 최악의 경우 3회의 이동이 발생

정리

• 두 인접한 요소를 자리 바꾸는 매우 간단한 구조
• 잦은 Swap이 발생하여 처리 속도가 느림
• Stable & Internal Sort
• 이미 정렬된 배열이나, Early Exit이 적용되면 보다 효율적일 수 있음

개념

• 우선순위 Queue에 Heap을 사용하면 가장 높은 우선 순위를 가진 요소를 제거할 수 있는 점을 활용해 정렬을 실시
• Heap에 따라 값을 삽입O(n * log₂n)하고 반복적으로 Heap이 빌 때까지 제거O(n * log₂n)하면 정렬된 결과물을 얻을 수 있음 → 2 * O(n * log₂n) = O(n * log₂n)

구현

void heapSort(int a[], int n)
{
    MaxHeap h;
    for (int i = 0; i < n; i++)
        h.insert(a[i]);

    // In a max heap, the largest value is returned when deleted,
    // so to sort in ascending order, we fill the array from the end toward the front
    // with the deleted elements.
    for (int i = n - 1; i >= 0; i--)
        a[i] = h.remove().getKey();
}

• 정렬할 배열과 해당 배열의 크기를 입력
• 입력된 크기만큼 Heap에 값을 삽입하고 제거하며 정렬을 실행

Huffman Coding

개념

• 특정 문서에서 어떤 글자의 사용 빈도를 안다면, Binary Tree를 활용해 문서를 압축할 수 있음
• 기존 ASCII 방식은 모든 글자가 같은 개수의 비트를 할당 받으므로 압축에 효율적이지 않음
• Huffman 방식은 자주 사용된 글자에 더 적은 비트 개수를 할당함으로써 문서 용량을 줄일 수 있음
• ASCII와 같은 Fixed-Length Code는 읽을 때 일정한 간격에 따라 읽거나 쓸 수 있음
• 그러나 Huffman 방식과 같은 Variable-Length Code는 할당된 비트를 알려줄 Table을 따라 해석해야함(한 비트씩 확인하면서 일치하는 문자가 있는지 확인)

구현

1. 모든 글자의 사용 빈도를 파악. 각각의 글자는 독립적인 Tree의 root가 됨
2. 가장 적은 빈도를 가지는 2 글자를 결합해 새로운 Binary Tree를 생성. 새로운 root 노드가 만들어지고 해당 노드는 앞선 2 글자를 자식 노드로 취함
3. 새로 추가된 노드를 포함해서 가장 작은 값을 가지는 노드를 찾아 결합 (해당 과정을 반복)
4. 하나의 root를 통해 Binary Tree를 구성
5. 왼쪽 노드는 1을 오른쪽 노드는 0을 할당해 비트 값을 할당 (비트의 개수는 Tree 내에서 위치한 레벨에 따라 결정됨)

• 해당 과정에서 가장 적은 빈도를 가지는 노드들이 먼저 결합하므로 가장 낮은 레벨을 가지며 가장 큰 비트 개수를 할당받음

개념

• 특정 상황에서 저장된 값들의 우선순위가 중요한 경우가 있을 수 있음
• 이러한 우선순위 개념을 적용한 것 Priority Queue라는 자료구조
• 저장된 값들은 특정 우선순위를 가지며, 높은 우선순위를 가진 값이 먼저 제거됨
• 이러한 우선순위를 조절해 Priority Queue를 통해 Stack, Queue 모두 구현할 수 있음
• Array나 Linked List를 통해 구현할 수도 있지만 Heap 구조를 통해서도 구현 가능

구분

1. Min-priority Queue: 우선순위가 가장 낮은 요소가 먼저 제거됨
2. Max-priority Queue: 우선순위가 가장 높은 요소가 먼저 제거됨

구현

1. Array

• 삽입시 배열의 마지막에 추가하면되므로 O(1)의 시간복잡도가 소요 (장점)
• 삭제시, 가장 우선순위가 높은 요소를 찾아야하는데 이때 전체 배열을 순회해야하므로 O(n)의 시간복잡도가 소요 (단점)

2. Sorted Array

• 삽입시 정렬된 상태를 유지하기 위한 위치를 찾아 추가하므로 O(n)의 시간복잡도가 소요 (단점)
• 단순히 배열을 순회하거나, 이진 탐색 등의 방법을 사용할 수 있음
• 삭제시 정렬된 순서에 따라 가장 마지막의 요소를 제거하면 되므로 O(1)의 시간복잡도가 소요 (장점)

앞선 Array와 Sorted Array에서의 장단점은 Linked List와 Sorted Linked List에서 동일하게 작용

3. Heap

• Heap이란 Complete Binary Tree의 한 종류로, Priority Queue를 위해 만들어진 자료 구조
• 모든 요소가 완벽히 정렬되어 있지는 않지만, 그렇다고 완전히 정렬되지 않은 것도 아님 → 거의 정렬된 상태를 가짐
• 삽입과 삭제시 모두 O(log₂n)의 시간 복잡도를 가지므로 앞선 구현 방식에 비해 노드의 개수가 많아질수록 유리

Heap

개념

• 여러 값들 중 가장 큰 값이나 가장 작은 값을 빨리 찾기에 유리
• 자식 노드들이 부모 노드보다 작다는 규칙만 유지하면 됨
• Binary Search Tree에서는 중복되는 값을 서용하지 않았지만 Heap은 중복되는 값을 허용
• 큰 값은 높은 Level에 작은 값은 낮은 Level에 위치해 느슨하게 정렬된 상태를 달성
• 기본적으로 Complete Binary Tree이므로 마지막 Level을 제외하고는 모두 채워져 있어야하며 왼쪽에서 오른쪽 순서로 채워져야함

구분

1. Max heap: 각각의 부모 노드는 자식 노드보다 크거나 같음
2. Min heap: 각각의 부모 노드는 자식 노드보다 작거나 같음

구현

• Binary Tree는 Array로 구현할 경우 빈 공간이 생겨 비효율적이지만 Heap은 그 특성상 빈 공간이 발생하지 않아 유리함
• root노드는 배열의 1번에 저장됨
• 각 노드의 인덱스는 새로운 노드가 추가되어도 바뀌지 않음

부모, 자식 노드의 인덱스 계산

현재 인덱스 i 부모 노드의 인덱스 : i/2 왼쪽 자식 노드의 인덱스 : i * 2 오른쪽 자식 노드의 인덱스 : i * 2 + 1

Insertion

• Upheap 방식을 사용 (상향식으로 비교)

• 새로운 요소를 추가할 때 가장 마지막 노드에 추가하고 부모 노드와 비교하며 자리를 바꿈

• Complete Binary Tree의 높이 h는 log₂n이므로 삽입의 시간 복잡도는 O(log₂n)

    void insert(int key)
    {
        if (isFull()) return;       // When the heap is full
        int i = ++size;             // Start from the position corresponding to the incremented heap size
  
        // Traverse upward through the tree, comparing with parent nodes
        // While the key is greater than the parent
        while (i != 1 && key > getParent(i).getKey()) 
        {  
            node[i] = getParent(i).getKey() ; // Move the parent node down
            i /= 2;                 // Move up one level
        }
        node[i].setKey(key);        // Final position: copy the data
    }

• 탐색하는 노드가 root가 아니고 부모 노드의 값보다 크다면 현재 위치에 부모 노드의 값을 저장하고 윗 레벨로 이동

• 부모 노드가 추가하려는 값보다 크다면 반복을 종료하고 해당 위치에 값을 저장

Deletion

• Downheap 방식을 사용(하향식으로 비교)

• 가장 마지막 노드를 root로 올리고 자리를 바꾸며 내려옴 (root를 제거하므로)

• Insertion과 동일하게 O(log₂n)의 시간 복잡도를 가짐

    HeapNode remove() { 
        if( isEmpty() ) return NULL;
  
        HeapNode root = node[1];        // Root node (the element to be removed)
        HeapNode last = node[size--];    // The last node in the heap
  
        int parent    = 1;                // Start Index
        int    child    = 2;                // Start Index's child(left)
  
        while( child <= size ) // While not exceeding the heap boundary
        {            
              if( child < size
               && getLeft(parent).getKey() < getRight(parent).getKey())
                child++;                // Select the larger child node    
  
              if( last.getKey() >= node[child].getKey() ) break;
            // Move down one level
              node[parent] = node[child];
              parent = child;
              child *= 2;
        }
        node[parent] = last;            // Place the last node in its final position
        return root;                    // Return the root node
    }

• root 노드의 값을 root에, 마지막 노드의 값을 last에 저장

• root 노드는 1에서 시작하므로 parent는 1을 갖고 child는 왼쪽 Subtree를 기준으로 하여 2로 설정

• heap 영역 안에 있는 동안 while문 내부 로직을 반복

• child를 설정할 때 더 큰 자식 노드를 선택하기 위해 오른쪽 자식 노드가 더 크다면 기존 child 값에 1을 더함

• last 노드에 저장된 값이 선택된 자식 노드의 값보다 크거나 같다면 반복을 종료

• 그렇지 않다면 부모 위치에 자식 노드로 자리를 바꾸고 바꾼 자식 노드를 기준으로 반복을 실시

개념

• Binary Tree를 활용한 탐색
• 일정하게 정렬된 상태를 유지해 탐색에 효율적인 구조를 가짐
• 기존 Binary Tree와 동일하게 Traversal 등의 기능을 가지지만 추가적으로 Insertion, Deletion, Search 기능이 추가됨

조건

1. 모든 노드는 유일한 Key값을 가짐
2. 왼쪽 Subtree에 있는 값은 root에 있는 값보다 작음
3. 오른쪽 Subtree에 있는 값은 root에 있는 값보다 큼
4. 왼쪽, 오른쪽 Subtree 모두 Binary Search Tree임 (=앞선 조건을 만족함)

성능 분석

• Search와 Insertion 그리고 Deletion 모두 tree의 높이 h에 따라 결정되며 O(h)의 시간 복잡도를 가짐
• 일반적으로 균형잡힌 tree일 경우 h는 log₂n (n: 노드의 개수)
• 극단적으로 편향된 tree일 경우 h = n

관계형 데이터베이스의 구성

Field

• 데이터 베이스의 저장하는 가장 작은 단위
• 일반적으로 하나의 속성 값만을 저장

Record

• 다수의 Field로 이루어진 데이터의 단위
• 데이터 베이스의 행

Table

• 같은 형태를 가진 Record의 집합
• 전체 데이터베이스의 표

Key

• Record를 구분하거나 정렬, 검색하기 위한 Field
• 빨리 원하는 값을 찾는 것을 도움 Ex) 전체 학생 목록에서 학번은 특정 학생을 찾는 것을 도움

Primary Key

• 여러 Key들 중에서 전체 Record 중에서 중복되지 않는 것
• 반드시 중복되지 않아야하며 null 값을 가지지 않음

Search Operation

• 특정 Key값을 가진 노드를 찾기 위해 root에서부터 값을 비교하며 탐색
• 정렬된 구조가 아닐 경우 특정 값을 찾을 때 O(n)의 시간 복잡도 발생
• Recursive한 방법과 Iterative한 방법 모두를 활용해 구현 가능
• tree에 해당 기능을 구현할 수도 있고, node에서 구현할 수도 있음

탐색 과정

1. 현재 root와 찾고자 하는 값이 같다면 성공적으로 값을 찾은 것으로 간주
2. 찾고자하는 값이 현재 root보다 작다면, 왼쪽 Subtree를 탐색
3. 찾고자하는 값이 현재 root보다 크다면, 오른쪽 Subtree를 탐색 (leaf노드에 이를 때까지 1~3번 과정을 반복)

    BinaryNode* searchRecur( BinaryNode *n, int key ) {
        if( n == NULL ) return NULL;            // n is NULL

        if( key == n->getData() )                // (1) key == n->data
            return n;
        else if (key < n->getData() )            // (2) key < n->data
            return searchRecur( n->getLeft(), key );
        else                                     // (3) key > n->data
            return searchRecur( n->getRight(), key );
    }

    BinaryNode* searchIter( BinaryNode *n, int key ) {
        while( n != NULL ) {
            if( key == n->getData() ) return n;
            else if (key < n->getData() ) 
                n = n->getLeft();
            else n = n->getRight();
        }
        return NULL;
    }

Insert Operation

• Insertion을 위해서는 먼저 Search를 통해 삽입될 위치를 파악해야함
• Binary Search Tree에는 중복되는 key값은 존재할 수 없으므로 이미 존재하는 값은 추가될 수 없음
• Search에서 탐색을 실패한 지점이 새로운 노드를 추가할 지점
• 같은 값이어도 삽입 순서에 따라 Tree의 구조가 달라질 수 있음

구현

    void insertRecur( BinaryNode* r, BinaryNode* n ) {
        if( n->getData() == r->getData() ) 
            return;
        else if( n->getData() < r->getData() ) {
            if( r->getLeft() == NULL )
                r->setLeft(n);
            else
                insertRecur( r->getLeft(), n );
        }
        else {
            if( r->getRight() == NULL )
                r->setRight(n);
            else
                insertRecur( r->getRight(), n );
        }
    }

• 추가하려는 값과 현재 노드의 값을 비교해 값이 더 작을 경우 왼쪽 Subtree를 확인하고, 그때 해당 위치가 비어있다면 노드를 추가, 비어있지 않다면 Search를 이어서 수행 (값이 더 클 경우에는 오른쪽 Subtree에 대해 같은 작업을 수행)

Deletion Operation

• Binary Search Tree에서 가장 복잡한 작업
• 삭제될 노드가 어떤 노드인지 파악해야함

삭제될 노드의 경우의 수

1. leaf 노드인 경우
2. 해당 노드가 왼쪽 또는 오른쪽 Subtree만 가지는 경우
3. 해당 노드가 왼쪽과 오른쪽 Subtree를 모두 가지는 경우

1. leaf 노드인 경우

• 삭제할 노드의 부모노드를 찾아 삭제할 노드로의 연결을 끊음 (삭제를 위해서는 삭제할 노드와 해당 노드의 부모노드까지 알아야함)

2. 하나의 Subtree만 가지는 경우

• 삭제할 노드의 부모 노드에 해당 Subtree를 연결 (삭제할 노드, 부모 노드 그리고 추가로 자식 노드까지 알아야함)

3. 양쪽 Subtree를 가지는 경우

• 삭제할 노드를 양쪽 Subtree 중 하나의 값으로 교체

  가능한 옵션

1. 왼쪽 Subtree의 가장 큰 값으로 교체하기
2. 오른쪽 Subtree의 가장 작은 값으로 교체하기

구현

    void remove(BinaryNode *parent, BinaryNode *node) {

        // case 1
        if( node->isLeaf() ) {
            if( parent == NULL ) root = NULL;
            else {
                if( parent->getLeft() == node )
                    parent->setLeft(NULL);
                else
                    parent->setRight(NULL);
            }
        }
        // case 2
        else if( node->getLeft()== NULL|| node->getRight()==NULL ) {
            BinaryNode *child = (node->getLeft() != NULL )
                                ? node->getLeft()
                                : node->getRight();
            if( node == root )
                root = child;
            else {
                if( parent->getLeft() == node )
                    parent->setLeft(child); 
                else 
                    parent->setRight(child); 
            }
        }
        // case 3
        else {
            BinaryNode* succp = node;
            BinaryNode* succ = node->getRight();
            while (succ->getLeft() != NULL) {
                succp = succ;
                succ = succ->getLeft();
            }

            if( succp->getLeft() == succ )
                succp->setLeft(succ->getRight());
            else
                succp->setRight(succ->getRight());

            node->setData(succ->getData());

            node = succ;
        }
        delete node;
    }

• 해당 코드는 3번 경우에서 오른쪽 Subtree의 가장 작은 값을 사용
• 삭제할 노드뿐만 아니라 부모, 자식노드 등 전반에 대한 파악이 필요하므로 Tree에서 기능 구현

        // case 3
        else {
            BinaryNode* succp = node;
            BinaryNode* succ = node->getLeft();
            while (succ->getRight() != NULL) {
                succp = succ;
                succ = succ->getRight();
            }

            ...

• 왼쪽 Subtree의 가장 큰 값을 찾도록 변경할 수 있음
• 지속적으로 오른쪽 Subtree를 탐색하는 이유는 오른쪽에 root보다 큰 값이 저장되기 때문

Tree Operation

노드의 개수 세기

• 노드의 개수를 세기 위해서는 트리의 모든 노드를 순회해야함
• Traversal에서와 마찬가지로 자기 자신, left와 right노드를 재귀적으로 탐색

int getCount(BinaryNode *node)
{
    if(node == null) return 0;
    //자기 자신을 추가하기 위해 1을 더함
    return 1 + getCount(node->getLeft()) + getCount(node->getRight());
}

Leaf 노드 개수 세기

• leaf노드는 자식 노드를 갖지 않는 노드이므로 본인이 leaf 노드라면 1을 반환하고, 그렇지 않다면 재귀적으로 탐색을 이어감
• leaf노드 직전 노드에서 자신의 자식 노드를 탐색하면 자식 노드에서 1을 반환하므로 이를 leaf노드 개수에 추가

  int getLeafCount(BinaryNode *node)
  {
    if(node == null) return 0;
    if(node->isLeaf()) return 1;
    return getLeafCount(node->getLeft()) + getLeafCount(node->getRight());
  }

Tree 높이 계산

• Tree의 높이는 Subtree의 높이 + 1 (재귀적으로 적용하기 유리함)
• left와 right의 Subtree의 높이를 모두 구하되 둘 중 더 큰 값만 가져옴
• 높이가 1인 Subtree를 만날 때까지 재귀 탐색을 실시 후 연쇄적으로 값을 반환하며 종료됨

  높이가 3인 Perfect Binary Tree 예시

  1. root에서 출발
  2. getHeight(root->left) 호출 2-1) 자식 노드가 존재하므로 getHeight(left->left) 호출 2-1-1) 동일하게 getHeight(left->left) 호출하나 자식노드가 없어 바로 0이 반환 됨 2-1-2) getHeight(left->right) 또한 자식노드가 없으므로 0이 반환 2-1-3) return 1 + max(0,0) 이므로 1이 반환 2-2) getHeight(left->left)가 1을 반환 2-3) getHeight(left->right)를 호출 2-3-1) getHeight(right->left)를 호출하나 0을 반환 2-3-2) getHeight(right->right) 또한 0을 반환 ([2-1] 내부의 과정과 동일) 2-3-3) return 1 + max(0,0) 이므로 1이 반환 2-4) getHeight(left->right)가 1을 반환 2-5) 최종적으로 return 1 + max(1,1) 이므로 2가 반환
  3. 2번에서의 과정과 동일한 과정이 getHeight(root->right)에서 진행되었고, 최종적으로 2를 반환
  4. return 1 + max(2,2)이므로 최종적으로 3을 반환

Tree의 활용

Expression Tree

• 후위 연산자, 중위 연산자 등에서 사용했던 방식을 Tree를 통해 표현할 수 있음

• Preorder는 전위 연산, Inorder는 중위 연산, Postorder는 후위 연산에 대응

  예시

  Preorder : +ab Inorder : a+b Postorder : ab+

• 연산식에서 괄호는 Subtree로 표현할 수 있음

  예시

  a + (c * d) Preorder : + a * cd Inorder : a + c * d Postorder : acd*+

• root 노드는 연산자 역할을 하고 Subtree들이 피연산자역할을 수행

• 괄호를 고려한 연산 순서를 위해서는 Postorder를 사용한 연산이 필요

File Size Claculator

• 컴퓨터의 폴더 구조는 Tree를 통해 표현할 수 있음 (일반적으로 Binary Tree 구조는 아니지만 이번 경우는 이진 트리로 가정함)
• leaf노드에서부터 각각의 노드의 크기를 측정해 전부 더한 값이 전체 파일의 크기로 볼 수 있음
• 따라서 Postorder 순회를 통해 파일 크기 구조 계산이 가능 (Postorder는 root로 가기 전에 무조건 자식 노드들 전부 방문하므로)

Threaded Binary Tree

• Threaded Binary Tree란 기존에 재귀 방식을 사용하는 순회 방식과 다른 방법을 추가적으로 구현
• leaf노드의 null 링크들을 재활용함, null 링크들이 inorder predecessor 또는 successor를 가리키는데 사용됨
• 링크를 재활용하기 때문에 자식 노드를 가리키는 링크와 thread 필드를 가리키는 링크를 구분하는 것이 필요

Traversal

• Traversal이란 Tree 내부의 모든 노드를 1번씩 방문 하여 값을 확인할 수 있도록 하는 것
• 선형 자료구조는 단방향의 순회만 가능하지만 Tree 구조에서는 여러 방향의 순회 방법을 사용할 수 있음
• 각각의 노드가 각자의 left와 right에 해당하는 포인터를 갖고 있으므로 재귀 방식을 통해 단순한 코드로 순회를 구현할 수 있음
• Preorder, Inorder, Postorder 방법이 있으며 순서가 중요하지 않다면 셋 중 어떤 것을 사용해도 관계 없음

1. Preorder Traversal

• root → left → right 순서로 방문
• 자식 노드를 방문하기 전에 해당 노드의 부모 노드를 무조건 방문해야 하는 경우 사용 Ex) 모든 노드의 레벨을 확인할 때(자식 노드들은 부모 노드보다 레벨이 1 높기 때문)

2. Inorder Traversal

• left → root → right 순서로 방문

3. Postorder Traversal

• left → right → root순서로 방문
• 자식 노드를 먼저 방문해야 하는 경우에 사용 Ex) 저장공간 크기 측정, Binary Tree의 메모리 해제

순회 방식 구현

1. Tree Class에서 구현

• Tree 클래스에서 모든 노드들은 관망하며 관리
• 재귀 방식을 사용하지만 외부에서 관리됨 Like 선생님이 모든 학생의 이름을 순서대로 부르는 것

    void inorder(BinaryNode *node) {
        if( node != NULL ) {
            if( node->getLeft() != NULL ) inorder(node->getLeft());
            printf( " [%c] ", node->getData());
            if( node->getRight()!= NULL ) inorder(node->getRight());
        }
    }
    void preorder(BinaryNode *node) {
        if( node != NULL ) {
            printf( " [%c] ", node->getData());
            if( node->getLeft() != NULL ) preorder(node->getLeft());
            if( node->getRight()!= NULL ) preorder(node->getRight());
        }
    }
    void postorder(BinaryNode *node) {
        if( node != NULL ) {
            if( node->getLeft() != NULL ) postorder(node->getLeft());
            if( node->getRight()!= NULL ) postorder(node->getRight());
            printf( " [%c] ", node->getData());
        }
    }

• 매개변수로 시작 노드를 지정

2. Node Class에서 구현

• Node가 자신의 다음 노드들을 순차적으로 순회
• 모든 노드가 각각의 책임을 갖고 재귀적으로 다음 노드를 호출
• 캡슐화에 유리 Like 학생들이 서로 자신 다음에 있는 학생의 출석을 확인

    void inorder(BinaryNode *node)
    {
        if (node != NULL)
        {
            node->inorder();
        }
    }
    void preorder(BinaryNode *node)
    {
        if (node != NULL)
        {
            node->preorder();
        }
    }
    void postorder(BinaryNode *node)
    {
        if (node != NULL)
        {
            node->postorder();
        }
    }

• 구체적인 로직은 노드에 구현하고 트리에서는 함수를 호출하기만 함

추가

4. Level-Order Traversal

• 일반적인 순회 방법은 아니지만 BFS 탐색을 위해 필요
• 같은 레벨에 있는 노드들을 순회
• 앞선 3개의 순회 방법은 재귀 방식으로 구현할 수 있지만 (내부적으로 Stack을 사용) Level-Order는 Queue를 사용해야함
• 한 노드가 dequeue될 때, 해당 노드의 자식들을 enqueue
• 자식이 없다면 enqueue하지 않고 있다면 left-right순서로 처리
• 이러한 과정을 Queue가 빌 때까지 반복
• Tree에서 구현하는 것은 쉽지만 노드 내부에 Level-Order를 구현하는 것은 어려움 (전체 과정을 총괄하는 Queue가 필요하기 때문)

개념

• 가장 흔하게 사용되는 Tree의 형태
• 노드들은 2개의 Subtree를 가짐
• 2개의 Subtree들은 left와 right의 순서를 가짐
• 자식 노드를 하나도 가지지 않은 empty set 가능

구성요소

edge

• 노드의 개수가 n개라면 edge는 n-1개
• 모든 노드는 부모로 연결되는 edge(=link)를 갖지만 root는 가지지 않으므로 n-1 개

height

• 한 Binary Tree의 height가 h라면 가능한 노드의 최대 개수는 2^h -1개
• 모든 노드가 2개의 자식을 갖는다면 특정 레벨 l에서의 최대 노드의 개수는 2^(l-1)개 Ex) root에서 레벨이 1이므로 노드의 개수 2^0 = 1, 레벨이 3일 경우 2^2 = 4
• 높이가 h일 경우 1~h까지 모든 레벨에서의 노드 개수의 합이므로 2^h -1

예시

레벨 1 : 1개 = 2^(1-1) 레벨 2 : 2개 = 2^(2-1) 레벨 3 : 4개 = 2^(3-1) 총합 7개 = 2^3 -1

종류

Full Binary Tree

• 모든 노드가 0개 또는 2개의 자식 노드를 가짐
• 1개의 자식을 갖는 노드가 있으면 Full Binary Tree가 아님

Complet Binary Tree

• 마지막 레벨을 제외하고 모든 레벨이 전부 채워진 상태
• 마지막 레벨은 전부 채워질 필요는 없지만 left to right의 순서를 따라 채워져야함

Perfect Binary Tree

• 모든 노드는 2개의 자식을 가지며 모든 leaf 노드들은 동일한 레벨을 가짐
• Perfect Binary Tree일 경우 높이가 h일 경우 노드의 개수가 2^h -1을 보장

구현

Array를 통한 구현

• Array를 통해 Binary Tree를 구현하기 위해서는 높이가 h일 경우 2^h -1 크기를 할당해야함

• 각각의 값을 대응되는 인덱스에 저장

  root : 1번 인덱스에 저장 Level 1 : 2,3번 인덱스에 저장 Level 2 : 4~8번 인덱스에 저장 ... (각 레벨당 2^(l-1)개수의 메모리 공간을 할당)

• Complet Binary Tree나 Prefect Binary Tree에는 적합

• 편향된 Binary Tree일 경우 사용하지 않는 배열 영역이 많아져 공간 활용에 불리함

• 0번 인덱스는 사용하지 않음

  주변 노드의 인덱스

  특정 노드의 인덱스가 i라고 가정

1. 부모 노드의 인덱스 : i/2
2. 자식 노드 중 left 노드의 인덱스 : 2*i
3. 자식 노드 중 right 노드의 인덱스 : 2*i +1

Linked List를 통한 구현

• Link 파트와 Data 파트로 구성된 노드를 통해 구현
• 메모리 공간 상에 산재하나 포인터를 통해 연결됨

개념

• Stack, Queue와 같은 선형 자료구조는 연속적으로 값이 분포하지만 Tree는 비선형, 계층적 자료 구조를 가짐
• 노드들 간에 부모-자식 관계를 가짐
• Linked List와 유사한 구조를 가지지만 tree의 노드의 링크 필드는 자식과 연결됨
• 일반적인 경우 자식 노드의 개수에 제한은 없음
• 이진 트리(Binary Tree)는 최대 2개의 자식만 가짐

Level

• 부모-자식 관계를 1 단계 떨어진 관계로 볼때 tree의 층위의 개수
• 특정 노드가 루트 노드의 자식 노드라면 Level 2에 위치한 것임 (루트 노드는 Level 1)

  Height

• 전체 tree의 최대 Level

  Degree

• 특정 노드가 가진 자식 노드의 개수

  Forest

• 여러 tree들의 집합

구성 요소

Node

• 트리를 구성하는 기본 단위

  Parent Node

• 하나 이상의 자식 노드와 연결된 노드

  Child Node

• 부모 노드로부터 바로 연결된 노드

  Sibling Node

• 같은 부모를 공유하는 노드 집합

  Ancestor Node

• 특정 노드로부터 루트 노드까지 연속적으로 부모 관계에 있는 노드들의 집합

  Descendant Node

• 특정 노드로부터 리프 노드까지 연속적으로 자식관계에 있는 노드들의 집합

  Root

• 시작 지점 노드
• 다른 모든 노드들의 조상 노드

  Subtree

• 한 노드와 해당 노드의 모든 자식 노드

  Terminal Node (Leaf)

• tree의 마지막에 위치한 노드
• 자식 노드를 갖지 않는 노드

  Nonterminal Node

• 자식 노드를 가진 노드

Recursion을 사용한 연산

하노이탑

규칙

1. 한번에 한 디스크만 움직일 수 있음
2. 가장 위에 있는 디스크만 움직일 수 있음
3. 더 큰 디스크가 작은 디스크 위에 위치할 수 없음
4. 중간 기둥은 보조역할로 사용될 수 있음

• 해당 문제를 n개의 원반의 개수를 가질 때 풀 수 있도록 일반화

  매개변수 : 원반 n, 시작 지점 from, 중간 지점 temp, 목표 지점 to (n은 몇번째 원반인지를 나타냄)

  1. n이 1과 같으면 가장 밑에 있는 원반이므로 to 지점으로 이동 2-1. 그렇지 않은 경우 모든 원반을 temp으로 이동 2-2. 하나 남은 원반을 to로 이동 2-3. temp의 원반들을 to로 이동

코드로 구현

void hanoiTower(int n, char from, char temp, char to)
{
    if(n == 1) 
        //1번 원반을 from에서 to로 이동
    else
    {
        hanoiTower(n-1, from, to, temp)
        //n번 원반을 from에서 to로 이동
        hanoiTower(n-1, tmp, from, to)
    }

}

n = 4일 때의 예시 (from = A, tmp = B, to = C)

1. hanoiTower(3, A, C, B) 호출 1-1-1. hanoiTower(2, A, B, C) 호출 1-1-2. hanoiTower(1, A, C, B) 호출

    `n`이 1이므로 **1번 원반을 `A`에서 `B`로 이동** 후 `hanoiTower(2, A, B, C)`로 돌아감

   1-1-3. hanoiTower(2, A, B, C)에서 2번 원반을 A에서 C로 이동 1-1-4. hanoiTower(1, B, A, C) 호출 n이 1이므로 1번 원반을 B에서 C로 이동 후 hanoiTower(2, A, B, C)로 돌아간 후 함수가 종료되었으므로 hanoiTower(3, A, C, B)로 돌아감

   1-2-1. 3번 원반을 A에서 B로 이동

   1-3-1. hanoiTower(2, C, A, B) 호출 1-3-2. hanoiTower(1, C, B, A) 호출 n이 1이므로 1번 원반을 C에서 A로 이동 후 hanoiTower(2, C, A, B)로 돌아감 1-3-3. hanoiTower(2, C, A, B)에서 2번 원반을 C에서 B로 이동 1-3-4. hanoiTower(1, A, C, B) 호출 n이 1이므로 1번 원반을 A에서 B로 이동 후 hanoiTower(2, C, A, B)로 돌아감 돌아간 후 함수가 종료되었으므로 hanoiTower(3, A, C, B)로 돌아감

2. 4번 원반을 A에서 C로 이동

3. hanoiTower(3, B, A, C) 호출 3-1-1. hanoiTower(2, B, C, A) 호출 3-1-2. hanoiTower(1, B, A, C) 호출 n이 1이므로 1번 원반을 B에서 C로 이동 후 hanoiTower(2, B, C, A)로 돌아감 3-1-3. hanoiTower(2, B, C, A)에서 2번 원반을 B에서 A로 이동 3-1-4. hanoiTower(1, C, B, A) 호출 n이 1이므로 1번 원반을 C에서 A로 이동 후 hanoiTower(2, B, C, A)로 돌아감 돌아간 후 함수가 종료되었으므로 hanoiTower(3, B, A, C)로 돌아감

   3-2-1. 3번 원반을 B에서 C로 이동

   3-3-1. hanoiTower(2, A, B, C) 호출 3-3-2. hanoiTower(1, A, C, B) 호출 n이 1이므로 1번 원반을 A에서 B로 이동 후 hanoiTower(2, A, B, C)로 돌아감 3-3-3. hanoiTower(2, A, B, C)에서 2번 원반을 A에서 C로 이동 3-3-4. hanoiTower(1, B, A, C) 호출 n이 1이므로 1번 원반을 B에서 C로 이동 후 hanoiTower(2, A, B, C)로 돌아감 돌아간 후 함수가 종료되었으므로 hanoiTower(3, B, A, C)로 돌아감

4. 모든 과정이 종료되었으므로 hanoiTower(4, A, B, C) 함수를 종료

Recursion 연산의 종류

Linear Recursion(선형 재귀)

• 단일한 재귀 연산만 발생 Ex) 팩토리얼 연산, x^n(power) 연산

Binary Recursion(이진 재귀)

• 2개의 재귀 연산이 발생 Ex) 피보나치 연산, 하노이 탑 연산

Multiple Recursion(다중 재귀)

• 2개 이상의 재귀 연산이 발생
• 이진 재귀 연산도 이에 포함됨 Ex) Blob Coloring, Maze Traversal

Blob Coloring

• 서로 연결되어있지 않은 영역을 다른 색으로 색칠하는 것
• 이미지를 픽셀 단위로 스캔하다가 색 값을 가진 픽셀을 만나면 해당 픽셀의 이웃 4개 픽셀을 재귀 탐색
• 같은 색을 가진 연결된 영역들을 라벨링할 수 있음

Maze Traversal

• 깊이 우선 탐색으로 주변 4개 방향의 경로가 접근가능한지 탐색
• 4방향에 대한 다중 재귀 탐색으로 경로를 찾을 수 있음

정의

• 함수가 자기 자신을 호출하는 것
• Tree구조를 다루는데 유용함

예시

• 팩토리얼 계산의 예시 : n 팩토리얼을 계산하는 과정에서 n-1 팩토리얼의 계산 과정을 필요로 함

  n! = (n = 0) 1 || (n >= 1) n * (n-1)!

• 자기 자신을 호출하기 때문에 정지 조건을 명시하지 않으면 무한히 반복하게 됨

재귀 연산 과정

• factorial(3)을 연산하는 과정을 통해 재귀 연산 과정을 알아봄

1. factorial(3)에서 종료 조건 검사 → 3 == 0 이 false이므로 3 * factorial(3-1) 연산 과정에서 factorial(2) 호출
2. factorial(2)에서 종료 조건 검사 → 2 == 0 이 false이므로 2 * factorial(2-1) 연산 과정에서 factorial(2) 호출
3. factorial(1)에서 종료 조건 검사 → 1 == 0 이 false이므로 1 * factorial(1-1) 연산 과정에서 factorial(0) 호출
4. factorial(0)에서 종료 조건 검사 → 0 == 0 이 true이므로 1을 반환
5. factorial(0)이 1을 반환, factorial(1) 함수에서 1 * factorial(0)의 값이 1로 도출되어 반환됨
6. factorial(1)이 1을 반환, factorial(2) 함수에서 2 * factorial(1)의 값이 2로 도출되어 반환됨
7. factorial(2)의 값이 3을 반환, factorial(3) 함수에서 3 * factorial(2)의 값이 6으로 도출되어 최종적으로 값을 출력

구성

• 자기 자신을 호출하는 재귀 파트
• 종료 조건을 포함하는 종료 파트

예시

• 재귀 방식은 Divide and Conquer 방식을 기본적으로 사용하며 특정 작업을 작은 부분으로 쪼갠 뒤 각각을 해결하는 방식의 문제 풀이에 적합함
• 앞서 언급된 팩토리얼 외에도 피보나치 수열, 하노이 탑 문제 풀이에 사용하기에 적합
• 이진 트리와 이진 탐색에 사용 되는 점이 가장 핵심

Iteration과의 비교

• 같은 업무를 반복하여 실시한다는 점에서 Iteration과 비교될 수 있음

Iteration

• for, while을 통해 구현

• 반복 조건이 for, while문에 명시됨

• 반복문 내부는 반복 시 실시할 시퀀스가 명시됨

• 자료 구조에 따라 재귀 방식보다 비효율적일 경우가 있음

• Tree나 Graph에서 동일한 일이 점차 작은 업무로 분산되는 특성을 가지는 경우 재귀 방식을 사용하는 것이 Iteration 방식을 활용하는 것보다 자연스러움

• 그러나 재귀 방식은 function call(함수 호출)을 반드시 포함하기 때문에 일반적으로 Iteration 방식보다는 느림

• 두 방식은 서로 전환 될 수 있음 (재귀 방식으로 구현된 기능이 Iteration 방식으로도 구현될 수 있음)

팩토리얼 연산

• 둘다 시간복잡도는 O(n)으로 동일
• 재귀 방식은 추가적인 메모리 공간과 함수 호출을 위한 추가 비용이 발생해 실행 시간과 메모리 사용 측면에서 불리

x^n 연산

• Iteration 방식은 반복적으로 주어진 숫자 x를 n번 곱하므로 직관적이며 O(n)의 시간복잡도를 가짐
• Recursive 방식은 종료 조건: n == 0일 때 1을 반환 재귀 영역: n이 짝수일 때는 pow(x*x, n/2)를 호출 n이 홀수일 때는 x * pow(x*x, (n-1)/2)를 호출

1. pow(2,5)에서 5 == 0이 false이고 5가 홀수이므로 2 * pow(2 * 2, (5 -1)/2) 연산 과정에서 pow(4,2)를 호출
2. pow(4,2)에서 2 == 0이 false이고 4가 짝수이므로 pow(4 * 4, 2 / 2) 연산 과정에서 pow(16,1)을 호출
3. pow(16,1)에서 1 == 0이 false이고 1이 홀수이므로 16 * pow(16 * 16, (1 - 1) / 2 ) 연산 과정에서 pow(256, 0)을 호출
4. pow(256,0)은 종료 조건을 만족하므로 1을 반환
5. pow(16,1)은 16 * pow(256, 0)의 결과로 16을 반환
6. pow(4,2)는 pow(16,1)의 결과로 16을 반환
7. pow(2,5)는 2 * pow(4,2)의 결과인 32로 최종값을 도출

• 재귀 방식에서 함수 호출은 3번 발생
• 이와 같은 방식으로 x^n연산에서 재귀 방식은 O(log₂ n)의 복잡도를 가짐

피보나치 수열

• 피보나치 수열은 재귀 방식을 사용할 때 보다 간단하게 구현할 수 있는 연산 그러나 재귀 방식을 사용하면 비효율적임
• Recursive 방식으로 피보나치 수열 계산 종료 조건: 1) (n == 0) return 0 2) (n == 1) return 1

재귀 호출: fib(n - 1) + fib(n - 2)

• fib(6)을 연산하는 과정에서 fib(4)나 fib(3)이 반복적으로 등장하며 동일한 연산을 독립적인 작업 내에서 따로 따로 실시하는 비효율적인 방식으로 동작

• Iteration 방식은 O(n)의 시간복잡도를 가지지만 재귀 방식은 O(2^n)의 복잡도를 가져 보다 비효율적임

특징

• 일정한 순서를 가지는 요소들의 집합
• 배열은 일정 순서를 가지지 않지만 List는 일정한 순서를 유지해야함
• 선형 자료구조이며 정해지지 않은 위치에서 삽입, 삭제가 가능 (Stack이나 Queue는 맨앞 내지는 맨뒤에서만 발생)

구성요소

Insert(int index, T item) : index위치에 입력받은 값을 추가 Delete(int index) : index위치에 입력받은 값을 제거 [int index] : index위치에 있는 값을 반환 IsEmpty() : 리스트가 비어있는지 여부를 반환 IsFull() : 리스트가 가득 차 있는지 여부를 반환 Find(T item) : 해당 요소가 리스트에 존재하는지 여부를 확인 Replace(int index, T item) : index위치에 있는 값을 새로 입력한 item으로 교체 size : 리스트가 가지고 있는 요소의 개수

구현

• 배열을 사용해 구현하는 방법이 있고 Linked List를 사용해 구현할 수 있음

Private 필드

    int     data[MAX_LIST_SIZE];                
    int     length;     

• 데이터를 저장할 배열data와 현재 담고 있는 요소들의 갯수를 확인할 수 있는 length 변수 선언

insert()

    void insert( int pos, int e ) { 
        if( !isFull() && pos >= 0 && pos<=length ) {
            for( int i=length ; i>pos ; i-- )
                data[i] = data[i-1];
            data[pos] = e;
            length++;
        }
        else error("Overflow error or invalid insert position");
    }

• 입력받은 위치가 length 보다 작아야 배열 내에 추가할 수 있고, 배열이 가득 차있지 않아야 하므로 이를 먼저 확인
• 이후 끝 지점에서 pos까지 값을 1칸씩 뒤로 옮기고 pos 자리에 입력받은 값을 추가

Remove()

    void remove( int pos ) {
        if( !isEmpty() && 0<=pos && pos<length ) {
            for(int i=pos+1 ; i<length ; i++ )
                data[i-1] = data[i];
            length--;
        }
        else error("Underflow error or invalid delete position");
    }

• 배열이 비어있는지 확인하고 제거하고자 하는 위치가 배열이 값을 갖고 있는 범위 이내인지 확인
• 제거하고자 하는 위치인 pos에서 1칸 뒤에있는 값부터 앞으로 당겨 오고 전체 길이를 하나 줄여 마무리

find()

    bool find( int item ) {
        for( int i=0 ; i<length ; i++ )
            if( data[i] == item ) return true;
        return false;
    }

• 배열의 시작부터 하나씩 입력받은 값과 같은지 확인하다 같은 값을 찾았을 때 true 값을 반환

단점

• Linked List를 사용하지 않고 배열을 사용해서 List를 구현했을 때 값을 추가하거나 삭제할 때 모든 배열 내의 요소를 이동시켜야하는 번거로움이 존재
• 배열의 최대 크기만큼 큰 메모리 공간을 점유하고 있다는 단점이 존재

특징

• 물리적으로는 메모리 공간에 흩어져 있는 데이터를 집합적으로 연결하는 것
• 배열은 물리적으로 연속된 공간에 데이터가 위치하지만 Linked List는 떨어져 있지만 포인터를 통해 연결
• 생성시 크기를 정해야 하는 배열과 달리 동적으로 공간이 늘어나고 줄일 수 있음
• 각각의 노드는 값을 저장하는 데이터 파트와 앞선 노드들과의 연결을 괸리하는 링크 파트로 구성됨
• 기존의 Stack, Queue뿐만 아니라 Tree, Graph 등의 자료구조에 다양하게 사용할 수 있음
• 첫번째 노드부터 순서대로 연결되어 있으므로 특정 노드를 찾을 때 앞에서부터 연쇄적으로 접근해 찾을 수 있음

장점

• 고정되지 않아 동적으로 크기를 바꿀 수 있음
• 특정 위치에 삽입할 때 기존 배열은 해당 위치보다 뒤에 있는 것들을 전부 뒤로 밀고 삽입해야하나 그저 포인터의 지시 대상을 바꾸는 것으로 효율적이게 수행 가능
• 삭제하는 것 또한 동일하게 배열 내 요소들의 이동 없이 지시 대상을 바꾸는 것으로 실행 가능
• 필요할 때 메모리를 동적으로 할당해 공간을 확보하므로 미리 큰 공간을 확보해 둬야하는 배열보다 유리

단점

• 배열보다 복잡해 오류의 가능성이 높고 포인터를 잘 관리하지 못했을 때 문제가 발생
• 실제 저장할 데이터 외에도 포인터 등 노드의 연결을 관리할 요소가 필요해 메모리 공간을 더 요구함
• 순회, 삽입, 삭제가 일반 배열보다는 복잡함
• 자주 동적 할당이 이루어져 프로그램 속도를 늦출 수 있음

구성 요소

노드

• Data Field와 Link Field로 구성됨
• Data Field : 실제 데이터를 저장
• Link Field : 다음 노드의 주소를 저장. 해당 요소를 통해 노드들이 연결됨

Head Pointer

• 향후 다른 노드에 접근하기 위한 시작 지점
• 마지막 노드의 Link Field의 포인터를 NULL 값으로 설정됨

종류

Singly Linked List

• 노드들이 오직 한 방향으로만 연결됨
• 마지막 노드는 NULL 값을 지시함

Circular Linked List

• 한 방향으로 진행하는 것은 Singly Linked List와 동일
• 마지막 노드는 첫번째 노드를 지시해 순환하는 구조를 가짐

Doubly Linked List

• 각각의 노드들은 2개의 Limk Field를 가져 자신의 앞에 있는 데이터와 뒤에 있는 데이터 모두를 저장
• 양방향 순회가 가능

구성 요소

Node 클래스

data : 실제 값을 저장할 변수 next : 다음 노드의 포인터를 저장할 변수

LinkedList 클래스

headPointer : Linked List의 시작지점이 될 포인터 Add(T data) : 새로운 값을 Linked List에 추가 Insert(int index, T data) : index지점에 data를 추가 Delete(int index) : index지점의 요소를 제거 Clear() : Linked List내의 모든 요소를 제거 "LinkedList name"[int index] : index를 통해 해당 요소에 접근

구현

template <typename T>
class Node
{
public:
    Node(T value, Node *ptr) { data = value; next = ptr; }
    T data;
    Node *next;
};
template <typename T>
class LinkedList
{
private:
    Node<T> *head;

public:
    LinkedList() : head(nullptr) {}

    void Add(T data)
    {
        Node<T> *newNode = new Node<T>(data, nullptr);

        if (!head)
        {
            head = newNode;
        }
        else
        {
            Node<T> *temp = head;
            while (temp->next)
            {
                temp = temp->next;
            }
            temp->next = newNode;
        }
    }

    void Insert(int index, T data)
    {
        Node<T> *newNode = new Node<T>(data, nullptr);

        if (index == 0)
        {
            newNode->next = head;
            head = newNode;
            return;
        }

        Node<T> *temp = head;

        for (int i = 0; i < index - 1 && temp; i++)
        {
            temp = temp->next;
        }

        if (temp)
        {
            newNode->next = temp->next;
            temp->next = newNode;
        }
        else
        {
            throw "Index out of bounds";
        }
    }

    void RemoveAt(int index)
    {
        if (!head)
        {
            throw "List is empty";
        }

        if (index == 0)
        {
            Node<T> *temp = head;
            head = head->next;
            delete temp;
            return;
        }

        Node<T> *temp = head;

        for (int i = 0; i < index - 1 && temp; i++)
        {
            temp = temp->next;
        }

        if (temp && temp->next)
        {
            Node<T> *toDelete = temp->next;
            temp->next = toDelete->next;
            delete toDelete;
        }
        else
        {
            throw "Index out of bounds";
        }
    }

    T operator[](int index)
    {
        Node<T> *temp = head;

        for (int i = 0; i < index && temp; i++)
        {
            temp = temp->next;
        }

        if (temp)
        {
            return temp->data;
        }
        else
        {
            throw "Index out of bounds";
        }
    }

    void Clear()
    {
        Node<T> *temp = head;

        while (temp)
        {
            Node<T> *toDelete = temp;
            temp = temp->next;
            delete toDelete;
        }

        head = nullptr;
    }
};

Node 클래스

template <typename T>
class Node
{
public:
    Node(T value, Node *ptr) { data = value; next = ptr; }
    T data;
    Node *next;
};

• 노드의 기본 구성대로 Data Field와 Link Field로 구성되어 생성자에서 포인터와 저장할 데이터를 입력받록 함

Linked List 클래스

Private 필드

private:
    Node<T> *head;

• Private 하게 저장해야할 부분은 시작 지점이 될 head 포인터만 존재

생성자

    LinkedList() : head(nullptr) {}

• 생성시 head를 null 포인터를 생성해 연결

Add(T date) & Insert(int index, T data)

    void Add(T data)
    {
        Node<T> *newNode = new Node<T>(data, nullptr);

        if (!head)
        {
            head = newNode;
        }
        else
        {
            Node<T> *temp = head;
            while (temp->next)
            {
                temp = temp->next;
            }
            temp->next = newNode;
        }
    }

• 리스트의 가장 마지막에 값을 추가하는 함수
• 새로운 노드를 newNode로 생성하면서 메모리를 동적 할당
• 만약 head가 비어있다면 head에 새로운 노드를 연결하는 것으로 완료
• 만약 head가 비어있지 않다면 next 노드가 비어있는 것을 찾아 해당 노드의 next에 새로운 노드를 연결

    void Insert(int index, T data)
    {
        Node<T> *newNode = new Node<T>(data, nullptr);

        if (index == 0)
        {
            newNode->next = head;
            head = newNode;
            return;
        }

        Node<T> *temp = head;

        for (int i = 0; i < index - 1 && temp; i++)
        {
            temp = temp->next;
        }

        if (temp)
        {
            newNode->next = temp->next;
            temp->next = newNode;
        }
        else
        {
            throw "Index out of bounds";
        }
    }

• Add함수와 비슷하게 작동하나 특정 인덱스 위치에 값을 추가

        for (int i = 0; i < index - 1 && temp; i++)
        {
            temp = temp->next;
        }

• head에서 시작해 선택한 인덱스 지점까지 이동해 해당 위치의 포인터가 null인지 확인 하고 그렇지 않다면 해당 노드의 next를 추가하고자 하는 노드의 next로 연결하고, temp 노드의 next를 새로 추가하는 노드로 설정

RemoveAt(int index)

    void RemoveAt(int index)
    {
        if (!head)
        {
            throw "List is empty";
        }

        if (index == 0)
        {
            Node<T> *temp = head;
            head = head->next;
            delete temp;
            return;
        }

        Node<T> *temp = head;

        for (int i = 0; i < index - 1 && temp; i++)
        {
            temp = temp->next;
        }

        if (temp && temp->next)
        {
            Node<T> *toDelete = temp->next;
            temp->next = toDelete->next;
            delete toDelete;
        }
        else
        {
            throw "Index out of bounds";
        }
    }

• 입력받은 인덱스 위치 index에 있는 값을 삭제하기 위해 Insert 함수와 같이 해당 노드까지 순차적으로 이동을 실시
• 현재 있는 노드의 다음 노드를 삭제하기 위해 현재 노드의 다음 노드를 삭제하고자 하는 노드의 다음 노드로 연결하여 삭제하고자 하는 노드를 건너뛰도록 하고 삭제를 진행

Operator

    T operator[](int index)
    {
        Node<T> *temp = head;

        for (int i = 0; i < index && temp; i++)
        {
            temp = temp->next;
        }

        if (temp)
        {
            return temp->data;
        }
        else
        {
            throw "Index out of bounds";
        }
    }

• 배열에서 arr[2]와 같이 접근하는 것을 구현하기 위해 operator[]를 선언해 구현
• Insert, RemoveAt함수와 유사하게 특정 인덱스까지 이동하고 해당 위치의 값을 반환