ERP_Preprocessor / main.py
hariqueen's picture
Update main.py
b2b08ea verified
import os
import argparse
from datetime import datetime
from core.config import RENTAL_COMPANIES, OUTPUT_DIR
from mappers.mapping_utils import load_mapping_file
from processors.rental_processor import load_and_preprocess_data, summarize_data
from generators.korea_rental_gen import generate_erp_data, prepare_erp_columns, set_management_items
from utils import (
load_erp_form_template, prepare_file_with_template, save_to_files,
print_data_summary, generate_report_file
)
import pandas as pd
from pyexcel_xls import save_data
from collections import OrderedDict
def process_rental_company(company_name: str):
"""
νŠΉμ • λ Œνƒˆμ‚¬μ˜ 데이터 처리 (CLI μ‹€ν–‰μš©)
Args:
company_name: μ²˜λ¦¬ν•  λ Œνƒˆμ‚¬ 이름
"""
if company_name not in RENTAL_COMPANIES:
print(f"였λ₯˜: '{company_name}' λ Œνƒˆμ‚¬ 섀정을 찾을 수 μ—†μŠ΅λ‹ˆλ‹€.")
return
company_config = RENTAL_COMPANIES[company_name]
print(f"'{company_name}' λ Œνƒˆμ‚¬ 데이터 처리 μ‹œμž‘...")
input_file = company_config['input_file']
mapping_file = company_config['mapping_file']
erp_form_file = company_config['erp_form_file']
output_csv = company_config['output_csv']
output_excel = company_config['output_excel']
report_file = os.path.join(OUTPUT_DIR, f'λ³΄κ³ μ„œ_{company_name}_{datetime.now().strftime("%Y%m%d")}.txt')
mapping_dict = load_mapping_file(mapping_file)
df, df_filtered = load_and_preprocess_data(input_file, company_config, mapping_dict)
summary = summarize_data(df_filtered, mapping_dict)
erp_df = generate_erp_data(df_filtered, company_config)
erp_df = prepare_erp_columns(erp_df)
erp_df = set_management_items(erp_df, df_filtered, company_config)
erp_form = load_erp_form_template(erp_form_file)
result_df = prepare_file_with_template(erp_df, erp_form)
save_to_files(result_df, output_csv, output_excel, len(erp_df))
print_data_summary(summary, company_config)
# generate_report_file(summary, erp_df, report_file)
print(f"\n'{company_name}' λ Œνƒˆμ‚¬ 데이터 처리 μ™„λ£Œ.")
def process_rental_company_with_voucher(uploaded_file_path, voucher_number, employee_number):
"""
νŠΉμ • λ Œνƒˆμ‚¬μ˜ 데이터 처리 (μ›Ή μΈν„°νŽ˜μ΄μŠ€μš©)
Args:
uploaded_file_path: μ—…λ‘œλ“œλœ 파일 경둜
voucher_number: μ „ν‘œλ²ˆν˜Έ
employee_number: μ‚¬μ›λ²ˆν˜Έ (ν•„μˆ˜)
Returns:
좜λ ₯ 파일 경둜
"""
# μ‚¬μ›λ²ˆν˜Έ ν•„μˆ˜ 검증
if not employee_number or not employee_number.strip():
raise ValueError("μ‚¬μ›λ²ˆν˜Έλ₯Ό μž…λ ₯ν•΄μ£Όμ„Έμš”. μ‚¬μ›λ²ˆν˜ΈλŠ” ν•„μˆ˜ μž…λ ₯κ°’μž…λ‹ˆλ‹€.")
company_name = "ν•œκ΅­λ Œνƒˆ"
company_config = RENTAL_COMPANIES[company_name].copy() # 섀정을 λ³΅μ‚¬ν•΄μ„œ μ‚¬μš©
# μ‚¬μ›λ²ˆν˜Έ μ„€μ • - μž…λ ₯된 κ°’ μ‚¬μš©
company_config['id_write'] = employee_number.strip()
mapping_file = company_config['mapping_file']
mapping_dict = load_mapping_file(mapping_file)
df, df_filtered = load_and_preprocess_data(uploaded_file_path, company_config, mapping_dict)
summary = summarize_data(df_filtered, mapping_dict)
erp_df = generate_erp_data(df_filtered, company_config)
erp_df = prepare_erp_columns(erp_df)
erp_df = set_management_items(erp_df, df_filtered, company_config)
# μ „ν‘œλ²ˆν˜Έ μ±„μ›Œλ„£κΈ°
if 'ROW_ID' in erp_df.columns:
erp_df['ROW_ID'] = voucher_number
if 'NO_DOCU' in erp_df.columns:
erp_df['NO_DOCU'] = voucher_number
# ERP 양식 λ‘œλ“œ
erp_form = load_erp_form_template(company_config['erp_form_file'])
# ERP 양식에 λ§žμΆ°μ„œ 데이터 μ€€λΉ„
result_df = prepare_file_with_template(erp_df, erp_form)
# μ €μž₯
output_filename = f"μžλ™μ „ν‘œ_μ™„μ„±νŒŒμΌ_{datetime.now().strftime('%Y%m%d')}.xls"
output_path = os.path.join(OUTPUT_DIR, output_filename)
os.makedirs(OUTPUT_DIR, exist_ok=True)
try:
# λ°μ΄ν„°ν”„λ ˆμž„μ„ 리슀트둜 λ³€ν™˜
headers = result_df.columns.tolist()
data = [headers] # 헀더λ₯Ό 첫 번째 ν–‰μœΌλ‘œ μΆ”κ°€
# λ°μ΄ν„°ν”„λ ˆμž„μ˜ 각 행을 리슀트둜 λ³€ν™˜ν•˜μ—¬ data에 μΆ”κ°€
for _, row in result_df.iterrows():
data.append(row.tolist())
# OrderedDict 생성 (Sheet1μ΄λΌλŠ” μ΄λ¦„μ˜ μ‹œνŠΈμ— 데이터 μ €μž₯)
data_dict = OrderedDict()
data_dict["Sheet1"] = data
# xls 파일둜 μ €μž₯
save_data(output_path, data_dict)
print(f"Excel 97-2003 ν˜•μ‹(.xls)으둜 파일 μ €μž₯ μ™„λ£Œ: {output_path}")
except Exception as e:
print(f"Excel 97-2003 ν˜•μ‹ μ €μž₯ 쀑 였λ₯˜ λ°œμƒ: {e}")
# 였λ₯˜ λ°œμƒ μ‹œ κΈ°μ‘΄ λ°©μ‹μœΌλ‘œ μ €μž₯
backup_path = output_path.replace('.xls', '.xlsx')
result_df.to_excel(backup_path, index=False, engine='openpyxl')
output_path = backup_path
print(f"λŒ€μ²΄ ν˜•μ‹(.xlsx)으둜 파일 μ €μž₯ μ™„λ£Œ: {output_path}")
return output_path
def main():
"""
메인 μ‹€ν–‰ ν•¨μˆ˜ (CLI μ‹€ν–‰)
"""
parser = argparse.ArgumentParser(description='ERP μžλ™ μ „ν‘œ 생성 ν”„λ‘œκ·Έλž¨')
parser.add_argument('-c', '--company', type=str, help='μ²˜λ¦¬ν•  λ Œνƒˆμ‚¬ 이름')
parser.add_argument('-a', '--all', action='store_true', help='λͺ¨λ“  λ Œνƒˆμ‚¬ 처리')
parser.add_argument('-e', '--employee', type=str, default='00616', help='μ‚¬μ›λ²ˆν˜Έ (κΈ°λ³Έκ°’: 00616)')
args = parser.parse_args()
if args.all:
for company_name in RENTAL_COMPANIES.keys():
process_rental_company(company_name)
print('-' * 80)
elif args.company:
process_rental_company(args.company)
else:
process_rental_company('ν•œκ΅­λ Œνƒˆ')
if __name__ == "__main__":
main()