""" 메인 페이지 라우트 """ import pandas as pd import uuid from flask import request, render_template, redirect, url_for, session, flash, send_from_directory, jsonify from werkzeug.utils import secure_filename from api.file import clean_excel_header_newlines from utils.common import get_db_connection, clean_dataframe_data from utils.file_validation import validate_excel_file, excel_to_clean_df, perform_detailed_validation from utils.decorators import login_required from utils.upload_count import check_and_use_upload_count from . import main_bp @main_bp.route('/') @main_bp.route('/index', methods=['GET', 'POST']) @login_required def index(): if 'user_id' not in session: flash('로그인이 필요합니다.') return redirect(url_for('auth.login')) """메인 페이지 - 파일 업로드 및 목록""" if request.method == 'POST': if 'file' in request.files: file = request.files['file'] if file.filename == '': flash('파일이 선택되지 않았습니다.', 'error') return redirect(request.url) if file and file.filename.endswith('.xlsx'): # 파일 업로드 횟수 확인 (업로드 횟수 또는 프로 횟수가 있어야 업로드 가능) user_id = session.get('user_id') or session.get('email') from utils.upload_count import can_upload_file, get_upload_count from utils.pro_trial import can_use_pro_trial, get_pro_trial_count # 업로드 횟수 또는 프로 횟수가 있어야 업로드 가능 has_upload_count = can_upload_file(user_id) has_pro_trial = can_use_pro_trial(user_id) if not has_upload_count and not has_pro_trial: upload_remaining = get_upload_count(user_id) pro_remaining = get_pro_trial_count(user_id) purchase_url = 'https://infrasmart.co.kr/order.php?plan=pro' # Pro 플랜 구매 링크 flash( f'파일 업로드 횟수와 Pro 무료체험 횟수가 모두 소진되었습니다. (남은 업로드 횟수: {upload_remaining}회, 남은 Pro 횟수: {pro_remaining}회)
' f'추가 사용을 위해 Pro 플랜을 구독해주세요.', 'error' ) return redirect(request.url) try: cleaned_file = clean_excel_header_newlines(file) # 파일 검증 및 정제 엑셀 헤더가 유효한지 확인 특히 곤지암교의 경우 헤더가 병합되어 오류 발생 수정 df_result = excel_to_clean_df(cleaned_file, file) # 검증 실패 시 오류 반환 if hasattr(df_result, 'is_valid'): # FileValidationResult 객체인 경우 (오류 발생) error_messages = '
'.join(df_result.errors) flash(f'파일 검증 실패:
{error_messages}', 'error') return redirect(request.url) # 정상적으로 DataFrame과 header_row를 받은 경우 df, header_row = df_result # DataFrame 데이터 정리 및 trim 처리 df = clean_dataframe_data(df) # 필수 컬럼만 필터링 # required_columns = ['부재명', '부재위치', '손상내용', '손상물량', '개소', '단위'] # available_columns = [col for col in required_columns if col in df.columns] # if len(available_columns) < len(required_columns): # missing_cols = [col for col in required_columns if col not in df.columns] # flash(f'필수 컬럼이 누락되었습니다: {", ".join(missing_cols)}', 'error') # return redirect(request.url) # 상세 검증 수행 (손상물량 계산 검증 포함) validation_result = perform_detailed_validation(df) # 검증 오류가 있으면 중단 if validation_result.get('error_rows'): error_count = len(validation_result['error_rows']) error_messages = [] # 최대 10개의 오류만 표시 for error_row in validation_result['error_rows'][:10]: row_num = error_row['row_index'] errors = ', '.join(error_row['errors']) error_messages.append(f"행 {row_num}: {errors}") if error_count > 10: error_messages.append(f"... 외 {error_count - 10}개 오류") flash(f'파일 검증 실패 (총 {error_count}개 오류):
' + '
'.join(error_messages), 'error') return redirect(request.url) # 검증을 통과한 정제된 데이터 사용 cleaned_data = validation_result.get('cleaned_data', []) if not cleaned_data: flash('유효한 데이터가 없습니다.', 'error') return redirect(request.url) # 데이터베이스에 저장 conn = get_db_connection() cur = conn.cursor() # 정제된 데이터를 JSON으로 변환 (NaN 처리) import json import math def convert_nan_to_none(obj): """NaN과 Infinity를 None으로 변환""" if isinstance(obj, float): if math.isnan(obj) or math.isinf(obj): return None elif isinstance(obj, dict): return {k: convert_nan_to_none(v) for k, v in obj.items()} elif isinstance(obj, list): return [convert_nan_to_none(item) for item in obj] return obj cleaned_data = convert_nan_to_none(cleaned_data) file_data = json.dumps(cleaned_data, ensure_ascii=False) # 파일 정보 저장 bridge_name = request.form.get('bridge_name', '') structure_type = request.form.get('structure_type', '') span_count = request.form.get('span_count', '') length = request.form.get('length', '') width = request.form.get('width', '') expansion_joint_location = request.form.get('expansion_joint_location', '') # 숫자형 데이터 변환 try: span_count = int(span_count) if span_count else None length = float(length) if length else None width = float(width) if width else None except (ValueError, TypeError): span_count = None length = None width = None # 사용자의 현재 플랜 정보 가져오기 if '@' in str(user_id): cur.execute('SELECT plan FROM users WHERE email = %s', (user_id,)) else: cur.execute('SELECT plan FROM users WHERE id = %s', (user_id,)) plan_result = cur.fetchone() user_plan = plan_result[0] if plan_result and plan_result[0] else 'free' # plan_type 컬럼이 있는지 확인 cur.execute(""" SELECT column_name FROM information_schema.columns WHERE table_name = 'uploaded_files' AND column_name = 'plan_type' """) has_plan_type = cur.fetchone() is not None if has_plan_type: cur.execute( ''' INSERT INTO uploaded_files (user_id, filename, original_filename, file_data, bridge_name, structure_type, span_count, length, width, expansion_joint_location, plan_type) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) ON CONFLICT (user_id, filename) DO UPDATE SET file_data = EXCLUDED.file_data, bridge_name = EXCLUDED.bridge_name, structure_type = EXCLUDED.structure_type, span_count = EXCLUDED.span_count, length = EXCLUDED.length, width = EXCLUDED.width, expansion_joint_location = EXCLUDED.expansion_joint_location, plan_type = EXCLUDED.plan_type ''', ( session['user_id'], str(uuid.uuid4()), # UUID로 파일명 생성 file.filename, file_data, bridge_name, structure_type, span_count, length, width, expansion_joint_location, user_plan ) ) else: cur.execute( ''' INSERT INTO uploaded_files (user_id, filename, original_filename, file_data, bridge_name, structure_type, span_count, length, width, expansion_joint_location) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s) ON CONFLICT (user_id, filename) DO UPDATE SET file_data = EXCLUDED.file_data, bridge_name = EXCLUDED.bridge_name, structure_type = EXCLUDED.structure_type, span_count = EXCLUDED.span_count, length = EXCLUDED.length, width = EXCLUDED.width, expansion_joint_location = EXCLUDED.expansion_joint_location ''', ( session['user_id'], str(uuid.uuid4()), # UUID로 파일명 생성 file.filename, file_data, bridge_name, structure_type, span_count, length, width, expansion_joint_location ) ) # 업로드 횟수와 프로 횟수 차감 (같은 트랜잭션 내에서 처리) # 현재 횟수 확인 if '@' in str(user_id): cur.execute('SELECT upload_count, pro_trial_count FROM users WHERE email = %s', (user_id,)) else: cur.execute('SELECT upload_count, pro_trial_count FROM users WHERE id = %s', (user_id,)) count_result = cur.fetchone() if count_result: upload_count = count_result[0] if count_result[0] is not None else 0 pro_trial_count = count_result[1] if count_result[1] is not None else 0 # 횟수 차감 (둘 다 차감) new_upload_count = max(0, upload_count - 1) if upload_count > 0 else upload_count new_pro_trial_count = max(0, pro_trial_count - 1) if pro_trial_count > 0 else pro_trial_count # 업데이트 if '@' in str(user_id): cur.execute( 'UPDATE users SET upload_count = %s, pro_trial_count = %s WHERE email = %s', (new_upload_count, new_pro_trial_count, user_id) ) else: cur.execute( 'UPDATE users SET upload_count = %s, pro_trial_count = %s WHERE id = %s', (new_upload_count, new_pro_trial_count, user_id) ) # 결과 메시지 생성 message = f'파일이 성공적으로 업로드되었습니다. (남은 업로드 횟수: {new_upload_count}회, 남은 Pro 무료체험 횟수: {new_pro_trial_count}회)' flash(message, 'success') else: flash('파일이 업로드되었지만 사용자 정보를 찾을 수 없습니다.', 'warning') conn.commit() cur.close() conn.close() return redirect(url_for('main.index')) except Exception as e: # 에러 발생 시 데이터베이스 연결 정리 및 rollback if 'conn' in locals(): try: conn.rollback() except: pass try: if 'cur' in locals(): cur.close() conn.close() except: pass import logging logging.error(f"파일 처리 오류: {e}", exc_info=True) flash(f'파일 처리 중 오류가 발생했습니다: {str(e)}', 'error') return redirect(request.url) else: flash('올바른 Excel 파일(.xlsx)을 업로드해주세요.', 'error') return redirect(request.url) # 사용자의 최근 업로드 파일 목록 가져오기 conn = get_db_connection() cur = conn.cursor() # 사용자의 현재 플랜 정보 가져오기 user_id = session.get('user_id') cur.execute('SELECT plan FROM users WHERE id = %s', (user_id,)) user_plan_result = cur.fetchone() current_plan = user_plan_result[0] if user_plan_result and user_plan_result[0] else 'free' # 모든 컬럼 가져오기 cur.execute(""" SELECT column_name FROM information_schema.columns WHERE table_name = 'uploaded_files' """) columns = [row[0] for row in cur.fetchall()] # plan_type 컬럼이 있는지 확인 has_plan_type = 'plan_type' in columns # plan_type 컬럼이 없으면 추가 if not has_plan_type: try: cur.execute(""" ALTER TABLE uploaded_files ADD COLUMN IF NOT EXISTS plan_type VARCHAR(10) DEFAULT NULL """) conn.commit() columns.append('plan_type') has_plan_type = True except Exception as e: print(f"plan_type 컬럼 추가 실패: {e}") # 데이터 가져오기 cur.execute(f""" SELECT {', '.join(columns)} FROM uploaded_files WHERE user_id = %s ORDER BY upload_date DESC """, (user_id,)) # 데이터를 files에 담기 files = [] for row in cur.fetchall(): file_dict = dict(zip(columns, row)) # plan_type이 없으면 'free'로 표시 (업로드 당시 플랜 정보가 없는 경우) if not file_dict.get('plan_type'): file_dict['plan_type'] = 'free' files.append(file_dict) cur.close() conn.close() return render_template('files.html', files=files) @main_bp.route('/index_re') @login_required def index_re(): """새로운 인덱스 페이지""" from utils.slab_processing import process_slab_damage_data conn = get_db_connection() cursor = conn.cursor() # 부재별 집계표 데이터 조회 cursor.execute(''' SELECT 부재명, 손상내용, 단위, SUM(손상물량) as 총손상물량, SUM(개소) as 총개소 FROM file_damage_details WHERE user_id = %s GROUP BY 부재명, 손상내용, 단위 ORDER BY 부재명, 손상내용 ''', (session['user_id'],)) component_data = cursor.fetchall() # 데이터 가공 processed_data = [] for row in component_data: processed_data.append({ '부재명': row[0], '손상내용': row[1], '단위': row[2], '손상물량': row[3] if row[3] is not None else 0, '개소': row[4] if row[4] is not None else 0 }) # 사용자 정의 손상 매핑 설정 custom_damage_mapping = { '균열': { '1방향': { 'keywords': ['균열부백태', '균열'], 'length_factor': 0.25 }, '2방향': { 'keywords': ['망상균열', '균열'], 'length_factor': 0.25 } }, '누수': { 'keywords': ['누수', '백태'] }, '표면손상': { 'keywords': ['박리', '박락', '파손'] }, '철근부식': { 'keywords': ['철근노출'] } } # 콘크리트 바닥판 데이터 처리 slab_data = process_slab_damage_data(processed_data, custom_damage_mapping) conn.close() return render_template('index_re.html', slab_data=slab_data) @main_bp.route('/favicon.ico') def favicon(): return send_from_directory('static', 'favicon.ico') @main_bp.route('/style.css') def style(): return send_from_directory('static/css', 'style.css') @main_bp.route('/js/') def serve_js(filename): return send_from_directory('static/js', filename) @main_bp.route('/download/') def download_file(filename): """파일 다운로드""" return send_from_directory('data', filename, as_attachment=True) @main_bp.route('/pricing') @login_required def pricing(): """가격 정책 페이지""" return render_template('pricing.html') @main_bp.route('/files/delete_file/', methods=['POST', 'GET']) @login_required def delete_file(filename): """파일 삭제 및 JSON 응답""" conn = get_db_connection() cur = conn.cursor() cur.execute( "DELETE FROM uploaded_files WHERE user_id = %s AND filename = %s", (session['user_id'], filename) ) conn.commit() cur.close() conn.close() return jsonify({"success": True})