Spaces:
Sleeping
Sleeping
| 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() |