Upload 6 files

core/config.py

@@ -0,0 +1,52 @@
+"""
+설정 파일 및 경로 관리 모듈
+"""
+import os
+from datetime import datetime
+
+# 기본 파일 경로 설정
+INPUT_DIR = os.path.join(os.getcwd(), 'input')
+OUTPUT_DIR = os.path.join(os.getcwd(), 'output')
+MAPPING_DIR = os.path.join(os.getcwd(), 'mapping')
+TEMPLATE_DIR = os.path.join(os.getcwd(), 'templates')
+
+# 디렉토리가 없으면 생성
+for directory in [INPUT_DIR, OUTPUT_DIR, MAPPING_DIR, TEMPLATE_DIR]:
+    if not os.path.exists(directory):
+        os.makedirs(directory)
+
+# 현재 날짜 정보
+CURRENT_DATE = datetime.now().strftime("%Y%m%d")
+CURRENT_MONTH = datetime.now().strftime("%m")  # 현재 월 (01-12)
+
+# 렌탈사 설정 (추후 다른 렌탈사가 추가될 수 있음)
+RENTAL_COMPANIES = {
+    '한국렌탈': {
+        'input_file': os.path.join(INPUT_DIR, '한국렌탈_렌탈료.csv'),
+        'mapping_file': os.path.join(MAPPING_DIR, 'team_name_mapping.json'),
+        'erp_form_file': os.path.join(TEMPLATE_DIR, 'erp_form.csv'),
+        'output_csv': os.path.join(OUTPUT_DIR, f'자동전표_한국렌탈_{CURRENT_DATE}.csv'),
+        'output_excel': os.path.join(OUTPUT_DIR, f'자동전표_한국렌탈_{CURRENT_DATE}.xls'),
+        'partner_code': '101388',  # 거래처 코드 (한국렌탈: 101388)
+        'cost_center': '5020',     # 코스트센터(운영2)
+        'expense_acct': '53000',   # 기본 비용 계정
+        'payable_acct': '25300',   # 미지급금 계정
+        'cd_company': '1200',      # 회사 코드
+        'cd_pc': '1200',           # 회계단위
+        'cd_wdept': '1010',        # 작성부서
+        'amount_field': f'{CURRENT_MONTH}월렌탈료',  # 금액 필드명 (현재 월 기준)
+        'team_fields': [f'{CURRENT_MONTH}월 변경PJT', f'{int(CURRENT_MONTH)-1}월 PJT'],  # 팀 정보 필드명 (우선순위 순)
+        'note_prefix': '한국렌탈㈜_PC 렌탈료',  # 적요 접두어
+    }
+}
+
+# 기본 설정값
+DEFAULT_ENCODING = 'utf-8'
+CSV_OUTPUT_ENCODING = 'utf-8-sig'  # Excel에서 한글이 깨지지 않도록 BOM 포함
+
+# ERP 관련 설정
+ERP_DATA_ROW_START = 4  # 데이터 시작 행 (5행)
+ERP_DOCUMENT_TYPE = '11'  # 전표유형 (11: 일반)
+ERP_APPROVAL_STATUS = '1'  # 승인여부 (1: 미결/임시)
+ERP_PROCESS_STATUS = 'N'  # 전표처리결과 (N: 미처리/임시)
+ERP_DOCUMENT_GUBUN = '3'  # 전표구분 (3: 대체전표)

generators/korea_rental_gen.py

@@ -0,0 +1,171 @@
+"""
+ERP 데이터 생성 모듈
+"""
+import pandas as pd
+from typing import Dict, List, Any, Tuple
+from datetime import datetime
+import config as cfg
+
+
+def generate_erp_data(df_filtered: pd.DataFrame, company_config: Dict[str, Any]) -> pd.DataFrame:
+    """
+    ERP 업로드용 데이터프레임 생성
+    
+    Args:
+        df_filtered: 필터링된 데이터프레임
+        company_config: 렌탈사 설정 정보
+        
+    Returns:
+        ERP 업로드용 데이터프레임
+    """
+    print("ERP 업로드용 데이터프레임 생성 중...")
+    current_date = datetime.now().strftime("%Y%m%d")
+    document_number = f"FI{current_date[-8:]}{company_config.get('id_write', '00000')[-3:]}"
+    
+    # 1. 차변 데이터 생성 (각 팀별 계정별 비용)
+    debit_data = {
+        "ROW_ID": [document_number] * len(df_filtered), 
+        "ROW_NO": [str(i) for i in range(1, len(df_filtered)+1)],
+        "NO_TAX": ["*"] * len(df_filtered),
+        "CD_PC": [company_config['cd_pc']] * len(df_filtered),
+        "CD_WDEPT": [company_config['cd_wdept']] * len(df_filtered),
+        "NO_DOCU": [document_number] * len(df_filtered), 
+        "NO_DOLINE": [str(i) for i in range(1, len(df_filtered)+1)],
+        "CD_COMPANY": [company_config['cd_company']] * len(df_filtered),
+        "ID_WRITE": [company_config['id_write']] * len(df_filtered),
+        "CD_DOCU": [cfg.ERP_DOCUMENT_TYPE] * len(df_filtered),
+        "DT_ACCT": [current_date] * len(df_filtered),
+        "ST_DOCU": [cfg.ERP_APPROVAL_STATUS] * len(df_filtered),
+        "TP_DRCR": ["1"] * len(df_filtered),  # 차대구분 (1: 차변)
+        "CD_ACCT": df_filtered["CD_ACCT"].tolist(),  # 각 팀별 계정 코드
+        "AMT": df_filtered["금액"].apply(lambda x: str(int(x)) if pd.notnull(x) else "0").tolist(),
+        "CD_PARTNER": [company_config['partner_code']] * len(df_filtered),
+        "NM_NOTE": df_filtered["적요"].tolist(),
+        "TP_DOCU": [cfg.ERP_PROCESS_STATUS] * len(df_filtered),
+        "NO_ACCT": ["0"] * len(df_filtered),
+        "TP_GUBUN": [cfg.ERP_DOCUMENT_GUBUN] * len(df_filtered),
+    }
+    
+    # 2. 대변 데이터 생성 (미지급금으로 합계 금액)
+    total_amount = df_filtered["금액"].sum()
+    total_amount_str = str(total_amount)
+    
+    # 대변 데이터
+    credit_data = {
+        "ROW_ID": [document_number], 
+        "ROW_NO": [str(len(df_filtered) + 1)],  # 마지막 번호 다음
+        "NO_TAX": ["*"],
+        "CD_PC": [company_config['cd_pc']],
+        "CD_WDEPT": [company_config['cd_wdept']],
+        "NO_DOCU": [document_number], 
+        "NO_DOLINE": [str(len(df_filtered) + 1)],  # 마지막 라인 다음
+        "CD_COMPANY": [company_config['cd_company']],
+        "ID_WRITE": [company_config['id_write']],
+        "CD_DOCU": [cfg.ERP_DOCUMENT_TYPE],
+        "DT_ACCT": [current_date],
+        "ST_DOCU": [cfg.ERP_APPROVAL_STATUS],
+        "TP_DRCR": ["2"],  # 차대구분 (2: 대변)
+        "CD_ACCT": [company_config['payable_acct']],  # 미지급금 계정코드
+        "AMT": [total_amount_str],  # 전체 금액의 합계
+        "CD_PARTNER": [company_config['partner_code']],
+        "NM_NOTE": [f"{company_config['note_prefix']} 미지급금"],  # 적요
+        "TP_DOCU": [cfg.ERP_PROCESS_STATUS],
+        "NO_ACCT": ["0"],
+        "TP_GUBUN": [cfg.ERP_DOCUMENT_GUBUN],
+    }
+    
+    # 3. 차변과 대변 데이터프레임 생성
+    debit_df = pd.DataFrame(debit_data)
+    credit_df = pd.DataFrame(credit_data)
+    
+    # 4. 두 데이터프레임 합치기
+    erp_df = pd.concat([debit_df, credit_df], ignore_index=True)
+    
+    # 금액 필드 확인
+    print("\nAMT 필드 확인:")
+    print("차변 금액 합계:", df_filtered["금액"].sum())
+    print("대변 금액:", total_amount)
+    print("차변 건수:", len(debit_df))
+    print("대변 건수:", len(credit_df))
+    
+    return erp_df
+
+
+def prepare_erp_columns(erp_df: pd.DataFrame) -> pd.DataFrame:
+    """
+    ERP 표준 컬럼 구조로 데이터프레임 준비
+    
+    Args:
+        erp_df: ERP 데이터프레임
+        
+    Returns:
+        표준 컬럼 구조를 가진 ERP 데이터프레임
+    """
+    # 필드 순서 지정 - ERP 양식에 맞게 정확한 순서로 컬럼 정렬
+    erp_columns = [
+        "ROW_ID", "ROW_NO", "NO_TAX", "CD_PC", "CD_WDEPT", "NO_DOCU", "NO_DOLINE", 
+        "CD_COMPANY", "ID_WRITE", "CD_DOCU", "DT_ACCT", "ST_DOCU", "TP_DRCR", 
+        "CD_ACCT", "AMT", "CD_PARTNER", "DT_START", "DT_END", "AM_TAXSTD", 
+        "AM_ADDTAX", "TP_TAX", "NO_COMPANY", "NM_NOTE", "CD_BIZAREA", "CD_DEPT", 
+        "CD_CC", "CD_PJT", "CD_FUND", "CD_BUDGET", "NO_CASH", "ST_MUTUAL", 
+        "CD_CARD", "NO_DEPOSIT", "CD_BANK", "UCD_MNG1", "UCD_MNG2", "UCD_MNG3", 
+        "UCD_MNG4", "UCD_MNG5", "CD_EMPLOY", "CD_MNG", "NO_BDOCU", "NO_BDOLINE", 
+        "TP_DOCU", "NO_ACCT", "TP_TRADE", "NO_CHECK3", "NO_CHECK4", "CD_EXCH", 
+        "RT_EXCH", "CD_TRADE", "AM_EX", "TP_EXPORT", "NO_TO", "DT_SHIPPING", 
+        "TP_GUBUN", "NO_INVOICE", "NO_ITEM", "MD_TAX1", "NM_ITEM1", "NM_SIZE1", 
+        "QT_TAX1", "AM_PRC1", "AM_SUPPLY1", "AM_TAX1", "NM_NOTE1", "CD_BIZPLAN", 
+        "CD_BGACCT", "CD_MNGD1", "NM_MNGD1", "CD_MNGD2", "NM_MNGD2", "CD_MNGD3", 
+        "NM_MNGD3", "CD_MNGD4", "NM_MNGD4", "CD_MNGD5", "NM_MNGD5", "CD_MNGD6", 
+        "NM_MNGD6", "CD_MNGD7", "NM_MNGD7", "CD_MNGD8", "NM_MNGD8", "YN_ISS", 
+        "FINAL_STATUS", "NO_BILL", "NM_BIGO", "TP_BILL", "TP_RECORD", "TP_ETCACCT", 
+        "ST_GWARE", "SELL_DAM_NM", "SELL_DAM_EMAIL", "SELL_DAM_MOBIL", "SELL_DAM_TEL", 
+        "NM_PUMM", "JEONJASEND15_YN", "DT_WRITE", "ST_TAX", "MD_TAX2", "NM_ITEM2", 
+        "NM_SIZE2", "QT_TAX2", "AM_PRC2", "AM_SUPPLY2", "AM_TAX2", "NM_NOTE2", 
+        "MD_TAX3", "NM_ITEM3", "NM_SIZE3", "QT_TAX3", "AM_PRC3", "AM_SUPPLY3", 
+        "AM_TAX3", "NM_NOTE3", "MD_TAX4", "NM_ITEM4", "NM_SIZE4", "QT_TAX4", 
+        "AM_PRC4", "AM_SUPPLY4", "AM_TAX4", "NM_NOTE4", "NM_PTR", "EX_HP", 
+        "EX_EMIL", "NO_BIZTAX", "NO_ASSET", "TP_EVIDENCE", "NO_CAR", "NO_CARBODY", 
+        "CD_BIZCAR", "NM_PARTNER", "YN_IMPORT", "YN_FIXASSET"
+    ]
+    
+    # 나머지 열 추가 (빈 문자열로)
+    for col in erp_columns:
+        if col not in erp_df.columns:
+            erp_df[col] = [""] * len(erp_df)
+    
+    return erp_df[erp_columns]
+
+
+def set_management_items(erp_df: pd.DataFrame, df_filtered: pd.DataFrame, company_config: Dict[str, Any]) -> pd.DataFrame:
+    """
+    관리항목 설정
+    
+    Args:
+        erp_df: ERP 데이터프레임
+        df_filtered: 필터링된 데이터프레임
+        company_config: 렌탈사 설정 정보
+        
+    Returns:
+        관리항목이 설정된 ERP 데이터프레임
+    """
+    # 차변 행 설정
+    debit_rows = erp_df["TP_DRCR"] == "1"
+    erp_df.loc[debit_rows, "CD_CC"] = company_config['cost_center']  # 코스트센터
+    
+    # 부서코드 설정 (관리항목2) - 필수 항목으로 보임
+    if 'cd_wdept' in company_config:
+        erp_df.loc[debit_rows, "CD_DEPT"] = company_config['cd_wdept']  # 부서코드
+    
+    # CD_PJT를 정수형으로 확실하게 설정
+    pjt_codes = df_filtered["CD_PJT"].astype(int).tolist()
+    erp_df.loc[debit_rows, "CD_PJT"] = pjt_codes  # 프로젝트 코드
+    
+    # 대변 행 설정
+    credit_rows = erp_df["TP_DRCR"] == "2"
+    erp_df.loc[credit_rows, "CD_CC"] = company_config['cost_center']  # 코스트센터
+    
+    # 대변에도 부서코드 설정 필요
+    if 'cd_wdept' in company_config:
+        erp_df.loc[credit_rows, "CD_DEPT"] = company_config['cd_wdept']  # 부서코드
+    
+    return erp_df

mappers/mapping_utils.py

@@ -0,0 +1,99 @@
+"""
+팀명 매핑 관련 유틸리티 모듈
+"""
+import json
+import pandas as pd
+from typing import Dict, List, Any
+
+
+def load_mapping_file(mapping_file: str) -> Dict[str, Dict[str, str]]:
+    """
+    매핑 파일을 로드하여 딕셔너리 형태로 반환
+    
+    Args:
+        mapping_file: 매핑 파일 경로
+        
+    Returns:
+        매핑 딕셔너리: {팀명: {present: 현재팀명, CD_ACCT: 계정코드, CD_PJT: 프로젝트코드}}
+    """
+    try:
+        with open(mapping_file, 'r', encoding='utf-8') as f:
+            mapping_list = json.load(f)
+        
+        # 매핑 딕셔너리 생성
+        mapping_dict = {}
+        for item in mapping_list:
+            mapping_dict[item['past']] = {
+                'present': item['present'],
+                'CD_ACCT': item['CD_ACCT'],
+                'CD_PJT': item['CD_PJT']
+            }
+        
+        print(f"매핑 정보 로드 완료: {len(mapping_dict)}개 항목")
+        return mapping_dict
+    
+    except Exception as e:
+        print(f"매핑 파일 로드 중 오류 발생: {e}")
+        return {}
+
+
+def apply_mapping(team_name: str, mapping_dict: Dict[str, Dict[str, str]]) -> Dict[str, str]:
+    """
+    팀명에 매핑 정보 적용
+    
+    Args:
+        team_name: 원본 팀명
+        mapping_dict: 매핑 딕셔너리
+        
+    Returns:
+        매핑된 정보: {present: 현재팀명, CD_ACCT: 계정코드, CD_PJT: 프로젝트코드}
+    """
+    if pd.isna(team_name) or team_name == "":
+        return {"present": "", "CD_ACCT": "", "CD_PJT": ""}
+    
+    if team_name in mapping_dict:
+        return mapping_dict[team_name]
+    
+    # 없는 경우 빈 값 반환
+    return {"present": team_name, "CD_ACCT": "", "CD_PJT": ""}
+
+
+def get_unmapped_teams(df: pd.DataFrame) -> List[str]:
+    """
+    매핑되지 않은 팀명 목록 추출
+    
+    Args:
+        df: 데이터프레임
+        
+    Returns:
+        매핑되지 않은 팀명 목록
+    """
+    unmapped_df = df[(df["CD_ACCT"] == "") | (df["CD_PJT"] == "")]
+    return unmapped_df["원본팀명"].unique().tolist()
+
+
+def get_mapping_summary(df_filtered: pd.DataFrame, mapping_dict: Dict[str, Dict[str, str]]) -> Dict[str, Any]:
+    """
+    매핑 결과 요약 정보 생성
+    
+    Args:
+        df_filtered: 필터링된 데이터프레임
+        mapping_dict: 매핑 딕셔너리
+        
+    Returns:
+        매핑 요약 정보: {mapped_teams: 매핑된 팀명 목록, unmapped_teams: 매핑되지 않은 팀명 목록}
+    """
+    mapped_teams = []
+    for team in df_filtered["원본팀명"].unique():
+        mapped_info = mapping_dict.get(team, {})
+        mapped_teams.append({
+            'original': team,
+            'mapped': mapped_info.get('present', team),
+            'acct': mapped_info.get('CD_ACCT', ''),
+            'pjt': mapped_info.get('CD_PJT', '')
+        })
+    
+    return {
+        'mapped_teams': mapped_teams,
+        'mapped_count': len(mapped_teams)
+    }

processors/rental_processor.py

@@ -0,0 +1,270 @@
+"""
+데이터 전처리 및 가공 모듈
+"""
+import pandas as pd
+from typing import Dict, List, Any, Tuple
+import mapping_utils
+
+
+def load_and_preprocess_data(input_file: str, config: Dict[str, Any], mapping_dict: Dict[str, Dict[str, str]]) -> Tuple[pd.DataFrame, pd.DataFrame]:
+    """
+    데이터 로드 및 전처리
+    
+    Args:
+        input_file: 입력 파일 경로
+        config: 렌탈사 설정 정보
+        mapping_dict: 매핑 딕셔너리
+        
+    Returns:
+        전처리된 데이터프레임, 필터링된 데이터프레임
+    """
+    # CSV 파일 로드 - 다양한 인코딩 시도
+    print(f"'{input_file}' 파일 로딩 중...")
+    try:
+        rental_df = pd.read_csv(input_file, encoding='utf-8')
+    except UnicodeDecodeError:
+        try:
+            # UTF-8 실패 시 CP949 시도
+            rental_df = pd.read_csv(input_file, encoding='cp949')
+            print("CP949 인코딩으로 파일 로드 성공")
+        except UnicodeDecodeError:
+            try:
+                # EUC-KR 시도
+                rental_df = pd.read_csv(input_file, encoding='euc-kr') 
+                print("EUC-KR 인코딩으로 파일 로드 성공")
+            except Exception as e:
+                print(f"파일 로드 실패: {e}")
+                raise
+    
+    print(f"로딩 완료: {len(rental_df)}개 행 발견")
+    
+    # 컬럼명 양쪽 공백 제거 (더 엄격한 처리)
+    original_columns = rental_df.columns.tolist()
+    print("원본 컬럼명:")
+    for col in original_columns:
+        print(f"- '{col}'")
+
+    # 컬럼명에서 공백 제거 및 처리
+    rental_df.columns = [col.strip() for col in rental_df.columns]
+
+    # 처리된 컬럼명 출력
+    processed_columns = rental_df.columns.tolist()
+    print("처리 후 컬럼명:")
+    for i, col in enumerate(processed_columns):
+        orig = original_columns[i] if i < len(original_columns) else "?"
+        print(f"- '{orig}' -> '{col}'")
+
+    # 컬럼명 중복 체크 및 처리
+    if len(set(rental_df.columns)) != len(rental_df.columns):
+        print("경고: 공백 제거 후 중복된 컬럼명이 있습니다.")
+        duplicate_count = {}
+        new_columns = []
+        
+        for col in rental_df.columns:
+            if col in duplicate_count:
+                duplicate_count[col] += 1
+                new_col = f"{col}_{duplicate_count[col]}"
+                new_columns.append(new_col)
+                print(f"  중복 컬럼 처리: '{col}' -> '{new_col}'")
+            else:
+                duplicate_count[col] = 0
+                new_columns.append(col)
+        
+        rental_df.columns = new_columns
+    
+    # 필요한 필드 확인 및 조정
+    # 필요한 컬럼이 있는지 확인
+    column_exists = {}
+    required_columns = ["모델명", "영업분류", "관리부서", "거래처명", "관리지점"]
+    
+    for col in required_columns:
+        if col in rental_df.columns:
+            column_exists[col] = True
+        else:
+            column_exists[col] = False
+            print(f"경고: '{col}' 컬럼이 파일에 없습니다.")
+    
+    # 금액 필드 찾기 - 월별 자동 인식 패턴
+    amount_field = None
+    
+    # 1. 먼저 config에 설정된 필드 시도 (앞뒤 공백 제거 후 비교)
+    clean_amount_field = config['amount_field'].strip()
+    for col in rental_df.columns:
+        if col.strip() == clean_amount_field:
+            amount_field = col
+            print(f"금액 필드로 '{amount_field}'를 설정값에서 찾았습니다.")
+            break
+    
+    if not amount_field:
+        # 2. 'N월렌탈료' 패턴 찾기 - 공백 고려
+        import re
+        month_pattern = re.compile(r'^\s*(?:[0-9]{1,2})월렌탈료\s*$')
+        
+        for col in rental_df.columns:
+            if month_pattern.match(col):
+                amount_field = col
+                print(f"금액 필드로 '{amount_field}'를 자동 인식했습니다.")
+                break
+                
+        # 3. 렌탈료 포함 필드 찾기
+        if not amount_field:
+            for col in rental_df.columns:
+                if '렌탈료' in col:
+                    amount_field = col
+                    print(f"금액 필드로 '{amount_field}'를 사용합니다.")
+                    break
+    
+    if not amount_field:
+        # 4. 컬럼명에 '원'이나 '￦' 또는 '₩'가 포함된 것을 amount_field로 사용
+        for col in rental_df.columns:
+            if '원' in col or '￦' in col or '₩' in col:
+                amount_field = col
+                print(f"금액 필드로 '{amount_field}'를 사용합니다.")
+                break
+    
+    # 금액 필드를 찾을 수 없으면 오류 발생
+    if not amount_field:
+        raise ValueError("금액 필드를 찾을 수 없습니다. 파일 형식을 확인해주세요.")
+    
+    # 금액 필드 확인 출력
+    print(f"사용할 금액 필드: '{amount_field}'")
+    print(f"금액 필드 샘플 값: {rental_df[amount_field].head().tolist()}")
+    
+    # 팀 필드 찾기 - 월별 자동 인식 패턴
+    team_fields = []
+    
+    # 1. 먼저 config에 설정된 필드 시도
+    configured_team_fields = config.get('team_fields', [])
+    if isinstance(configured_team_fields, str):
+        configured_team_fields = [configured_team_fields]
+    
+    for field in configured_team_fields:
+        clean_field = field.strip()
+        for col in rental_df.columns:
+            if col.strip() == clean_field:
+                team_fields.append(col)
+                print(f"팀 필드로 '{col}'를 설정값에서 찾았습니다.")
+                break
+    
+    if not team_fields:
+        # 2. '[0-9]월 변경PJT' 패턴만 찾기 - 공백 허용
+        import re
+        # 공백 허용하고 '변경PJT'만 찾는 패턴
+        month_pjt_pattern = re.compile(r'^\s*(?:[0-9]{1,2})월\s*변경PJT\s*$')
+        
+        for col in rental_df.columns:
+            if month_pjt_pattern.match(col):
+                team_fields.append(col)
+                print(f"팀 필드로 '{col}'를 자동 인식했습니다 (변경PJT 패턴).")
+    
+    # 팀 필드를 찾을 수 없음 - 오류 발생
+    if not team_fields:
+        raise ValueError("팀 정보 필드를 찾을 수 없습니다. 파일 형식을 확인해주세요.")
+    
+    # 사용 가능한 컬럼만 선택
+    available_columns = []
+    for col in required_columns:
+        if column_exists.get(col, False):
+            available_columns.append(col)
+    
+    if amount_field:
+        available_columns.append(amount_field)
+    
+    available_columns.extend(team_fields)
+    
+    # 중복 제거
+    available_columns = list(dict.fromkeys(available_columns))
+    
+    print(f"사용할 컬럼: {available_columns}")
+    
+    # 필요한 필드만 선택 (존재하는 컬럼만)
+    df = rental_df[available_columns].copy()
+    
+    # 금액 필드 처리 - 간단한 방법으로 숫자만 추출
+    print(f"금액 필드 '{amount_field}' 데이터 처리 중...")
+    
+    # 숫자로 변환 가능한 값만 유효한 것으로 간주 (한 줄로 처리)
+    valid_amount_mask = pd.to_numeric(df[amount_field], errors='coerce').notna()
+    
+    # 유효하지 않은 행 수 출력
+    invalid_rows = (~valid_amount_mask).sum()
+    if invalid_rows > 0:
+        print(f"금액이 없거나 숫자가 아닌 행(반납 항목) {invalid_rows}개를 제외합니다.")
+    
+    # 유효한 행만 선택
+    df = df[valid_amount_mask].copy()
+    
+    # 금액 변환 - 단순화된 방법
+    df["금액"] = pd.to_numeric(df[amount_field], errors='coerce')
+    df["금액"] = df["금액"].astype(int)
+    print(f"금액 변환 성공: 샘플 값 = {df['금액'].head().tolist()}")
+    
+    # 팀명 처리 (우선순위에 따라)
+    if team_fields:
+        df["원본팀명"] = df[team_fields[0]].copy()
+        for field in team_fields[1:]:
+            df["원본팀명"] = df["원본팀명"].combine_first(df[field])
+    
+    # 매핑 적용
+    df["매핑정보"] = df["원본팀명"].apply(lambda x: mapping_utils.apply_mapping(x, mapping_dict))
+    
+    # 매핑 정보에서 필드 추출
+    df["팀명"] = df["매핑정보"].apply(lambda x: x["present"])
+    df["CD_ACCT"] = df["매핑정보"].apply(lambda x: x["CD_ACCT"])
+    
+    # CD_PJT를 정수형으로 변환하는 부분
+    df["CD_PJT"] = df["매핑정보"].apply(lambda x: x["CD_PJT"])
+    # 문자열이나 NaN 값 처리 후 정수형으로 변환
+    df["CD_PJT"] = pd.to_numeric(df["CD_PJT"], errors='coerce').fillna(1000).astype(int)
+    
+    # 적요 생성
+    df["적요"] = f"{config['note_prefix']}(" + df["팀명"] + ")"
+    
+    # MNG 코드 설정
+    df["CD_MNG1"] = config['cost_center']  # 코스트센터
+    df["CD_MNG3"] = config['partner_code']  # 거래처 코드
+    
+    # 매핑된 항목만 선택 (CD_ACCT와 CD_PJT가 있는 항목만)
+    df_filtered = df[(df["CD_ACCT"] != "") & (df["CD_PJT"] != "")].copy()
+    
+    # 매핑되지 않은 팀명 정보 출력
+    if len(df_filtered) < len(df):
+        unmapped_teams = df[~df.index.isin(df_filtered.index)]["원본팀명"].unique()
+        print(f"매핑되지 않은 팀명 {len(unmapped_teams)}개:")
+        for team in unmapped_teams:
+            print(f"- '{team}'")
+        
+        # 매핑되지 않은 항목이 있으면 경고 (전체 다 매핑 안 되는 경우만 오류)
+        if len(df_filtered) == 0:
+            raise ValueError("모든 팀명이 매핑되지 않았습니다. 매핑 파일을 확인해주세요.")
+    
+    print(f"매핑된 항목: {len(df_filtered)}개 / 전체 {len(df)}개")
+    
+    return df, df_filtered
+
+
+def summarize_data(df_filtered: pd.DataFrame, mapping_dict: Dict[str, Dict[str, str]]) -> Dict[str, Any]:
+    """
+    데이터 요약 정보 생성
+    
+    Args:
+        df_filtered: 필터링된 데이터프레임
+        mapping_dict: 매핑 딕셔너리
+        
+    Returns:
+        데이터 요약 정보
+    """
+    total_amount = df_filtered["금액"].sum()
+    
+    # 매핑 결과 요약
+    mapping_summary = mapping_utils.get_mapping_summary(df_filtered, mapping_dict)
+    
+    # 계정 사용 현황
+    account_counts = df_filtered['CD_ACCT'].value_counts().to_dict()
+    
+    return {
+        'total_count': len(df_filtered),
+        'total_amount': total_amount,
+        'account_counts': account_counts,
+        'mapping_summary': mapping_summary
+    }

utils/file_handler.py

@@ -0,0 +1,115 @@
+"""
+파일 입출력 관련 모듈
+"""
+import os
+import pandas as pd
+from typing import Dict, Any
+import config as cfg
+
+
+def load_erp_form_template(erp_form_file: str) -> pd.DataFrame:
+    """
+    ERP 양식 파일 로드
+    
+    Args:
+        erp_form_file: ERP 양식 파일 경로
+        
+    Returns:
+        ERP 양식 데이터프레임
+    """
+    try:
+        erp_form = pd.read_csv(erp_form_file, encoding=cfg.DEFAULT_ENCODING)
+        print(f"ERP 양식 파일 '{erp_form_file}'을 성공적으로 로드했습니다.")
+        return erp_form
+    except Exception as e:
+        print(f"ERP 양식 파일 로드 실패: {e}")
+        print("기본 양식 없이 진행합니다.")
+        return None
+
+
+def save_to_files(result_df: pd.DataFrame, output_csv: str, output_excel: str, erp_data_count: int) -> None:
+    """
+    결과를 CSV 및 Excel 파일로 저장
+    
+    Args:
+        result_df: 결과 데이터프레임
+        output_csv: CSV 출력 파일 경로
+        output_excel: Excel 출력 파일 경로
+        erp_data_count: ERP 데이터 행 수
+        
+    Returns:
+        None
+    """
+    # CSV 파일 저장
+    print(f"'{output_csv}'로 CSV 저장 중...")
+    try:
+        result_df.to_csv(output_csv, index=False, encoding=cfg.CSV_OUTPUT_ENCODING)
+        print(f"처리 완료: {erp_data_count}개 행이 '{output_csv}'에 저장됨 ({cfg.CSV_OUTPUT_ENCODING} 인코딩)")
+        print(f"데이터는 {cfg.ERP_DATA_ROW_START}행부터 시작합니다.")
+    except Exception as e:
+        print(f"CSV 파일 저장 중 오류 발생: {e}")
+    
+    # Excel 파일 저장 (.xls 형식으로 변경)
+    xls_output_excel = output_excel.replace('.xlsx', '.xls')
+    print(f"\n'{xls_output_excel}'로 엑셀 파일 저장 중...")
+    try:
+        # xlwt 엔진을 사용하여 Excel 97-2003 형식(.xls)으로 저장
+        result_df.to_excel(xls_output_excel, index=False, engine='xlwt')
+        print(f"처리 완료: {erp_data_count}개 행이 '{xls_output_excel}'에 저장됨")
+        print(f"엑셀 파일이 성공적으로 생성되었습니다: {os.path.abspath(xls_output_excel)}")
+    except Exception as e:
+        print(f"엑셀 파일 저장 중 오류 발생: {e}")
+        print("CSV 파일은 정상적으로 저장되었습니다.")
+        print("CSV 파일을 열 때는 Excel의 '데이터' 탭에서 '텍스트/CSV에서' 기능을 사용하시기 바랍니다.")
+        print(f"오류 내용: {e}")
+
+
+def prepare_file_with_template(erp_df: pd.DataFrame, erp_form: pd.DataFrame) -> pd.DataFrame:
+    """
+    ERP 양식을 적용하여 파일 준비
+    
+    Args:
+        erp_df: ERP 데이터프레임
+        erp_form: ERP 양식 데이터프레임
+        
+    Returns:
+        결과 데이터프레임
+    """
+    if erp_form is not None:
+        # 양식 파일의 컬럼 순서 사용
+        form_columns = erp_form.columns.tolist()
+        
+        # 결과 데이터프레임을 양식 컬럼 순서에 맞게 재정렬
+        for col in form_columns:
+            if col not in erp_df.columns:
+                erp_df[col] = ""
+        
+        erp_df = erp_df[form_columns]
+        
+        # erp_form 복사 (양식 파일의 처음 4행만 사용)
+        result_df = erp_form.copy()
+        
+        # 양식 파일이 4행보다 많으면 4행만 유지
+        if len(result_df) > cfg.ERP_DATA_ROW_START - 1:
+            result_df = result_df.iloc[:(cfg.ERP_DATA_ROW_START - 1)]
+        
+        # 빈 행 추가 (필요한 경우)
+        current_rows = len(result_df)
+        target_rows = cfg.ERP_DATA_ROW_START - 1  # 시작행 - 1 (인덱스는 0부터 시작하므로)
+        
+        # 현재 행 수가 타겟 행 수보다 적으면 빈 행 추가
+        if current_rows < target_rows:
+            empty_rows_needed = target_rows - current_rows
+            empty_df = pd.DataFrame([[""] * len(form_columns) for _ in range(empty_rows_needed)], columns=form_columns)
+            result_df = pd.concat([result_df, empty_df], ignore_index=True)
+        
+        # 처리된 데이터 추가 (ERP_DATA_ROW_START행부터 시작)
+        result_df = pd.concat([result_df, erp_df], ignore_index=True)
+        return result_df
+    else:
+        # 양식 파일이 없는 경우 빈 데이터프레임 생성 후 데이터 추가
+        # 필요한 빈 행 생성 (ERP_DATA_ROW_START-1개의 빈 행)
+        empty_rows = cfg.ERP_DATA_ROW_START - 1
+        empty_df = pd.DataFrame([[""] * len(erp_df.columns) for _ in range(empty_rows)], columns=erp_df.columns)
+        result_df = pd.concat([empty_df, erp_df], ignore_index=True)
+        return result_df

utils/reporter.py

@@ -0,0 +1,91 @@
+"""
+처리 결과 보고 모듈
+"""
+from typing import Dict, Any
+import pandas as pd
+
+
+def print_data_summary(summary: Dict[str, Any], company_config: Dict[str, Any]) -> None:
+    """
+    데이터 처리 결과 요약 출력
+    
+    Args:
+        summary: 데이터 요약 정보
+        company_config: 렌탈사 설정 정보
+        
+    Returns:
+        None
+    """
+    total_count = summary['total_count']
+    total_amount = summary['total_amount']
+    account_counts = summary['account_counts']
+    mapping_summary = summary['mapping_summary']
+    
+    print(f"\n총 처리 건수: {total_count + 1}건 (차변 {total_count}건, 대변 1건)")
+    print(f"총 금액: {total_amount:,.0f}원")
+    print(f"차변 계정: {len(account_counts)}개 계정 사용")
+    print(f"대변 계정: {company_config['payable_acct']} (미지급금) 1개 계정 사용")
+    
+    # 관리항목 설정 내용 출력
+    print("\n관리항목 설정 정보:")
+    print(f"- CD_CC (코스트센터): {company_config['cost_center']} (고정)")
+    print(f"- CD_PARTNER (거래처코드): {company_config['partner_code']} (고정)")
+    print(f"- CD_PJT (프로젝트코드): 각 팀별 매핑된 코드 사용")
+    
+    # 매핑 정보 요약
+    print("\n매핑 성공 팀명:")
+    mapped_teams = mapping_summary['mapped_teams']
+    for idx, team in enumerate(mapped_teams[:10]):
+        if len(mapped_teams) > 10 and idx == 9:
+            print(f"- {team['original']} ... 외 {len(mapped_teams)-10}개")
+        else:
+            print(f"- {team['original']} -> {team['mapped']} (ACCT: {team['acct']}, PJT: {team['pjt']})")
+
+
+def generate_report_file(summary: Dict[str, Any], erp_df: pd.DataFrame, report_file: str) -> None:
+    """
+    처리 결과 보고서 파일 생성
+    
+    Args:
+        summary: 데이터 요약 정보
+        erp_df: ERP 데이터프레임
+        report_file: 보고서 파일 경로
+        
+    Returns:
+        None
+    """
+    with open(report_file, 'w', encoding='utf-8') as f:
+        f.write("# ERP 전표 생성 결과 보고서\n\n")
+        
+        # 기본 정보
+        f.write("## 1. 기본 정보\n")
+        f.write(f"- 생성 일시: {pd.Timestamp.now().strftime('%Y-%m-%d %H:%M:%S')}\n")
+        f.write(f"- 처리 건수: 차변 {summary['total_count']}건, 대변 1건\n")
+        f.write(f"- 총 금액: {summary['total_amount']:,.0f}원\n\n")
+        
+        # 계정 분포
+        f.write("## 2. 계정 코드별 사용 현황\n")
+        for acct, count in summary['account_counts'].items():
+            f.write(f"- {acct}: {count}건\n")
+        f.write("\n")
+        
+        # 팀별 분포
+        f.write("## 3. 팀별 매핑 정보\n")
+        for team in summary['mapping_summary']['mapped_teams']:
+            f.write(f"- {team['original']} -> {team['mapped']} (계정: {team['acct']}, 프로젝트: {team['pjt']})\n")
+        
+        f.write("\n## 4. 전표 주요 정보\n")
+        # 차변 행 정보
+        debit_rows = erp_df[erp_df["TP_DRCR"] == "1"]
+        f.write(f"- 차변 건수: {len(debit_rows)}건\n")
+        f.write(f"- 차변 계정: {len(debit_rows['CD_ACCT'].unique())}개 계정 사용\n")
+        
+        # 대변 행 정보
+        credit_rows = erp_df[erp_df["TP_DRCR"] == "2"]
+        f.write(f"- 대변 건수: {len(credit_rows)}건\n")
+        f.write(f"- 대변 계정: {credit_rows['CD_ACCT'].iloc[0]} (미지급금)\n")
+        
+        # 전표 번호 정보
+        f.write(f"- 전표 번호: {erp_df['NO_DOCU'].iloc[0]}\n")
+        
+    print(f"\n처리 결과 보고서가 '{report_file}'에 저장되었습니다.")