import os import io import json import pymysql from datetime import datetime from flask import Flask, render_template, request, send_file import openpyxl from openpyxl import Workbook from openpyxl.styles import Font, Alignment, PatternFill BASE_DIR = os.path.dirname(os.path.abspath(__file__)) TEMPLATE_DIR = os.path.join(BASE_DIR, 'templates') app = Flask(__name__, template_folder=TEMPLATE_DIR) def get_db_connection(): with open('/root/mes_v2/config.json', 'r', encoding='utf-8') as f: config = json.load(f) db_info = config['mes_db'] return pymysql.connect( host=db_info['host'], port=db_info['port'], user=db_info['user'], password=db_info['password'], db='belt1', charset='utf8mb4', cursorclass=pymysql.cursors.DictCursor ) def safe_float(val): if val is None: return None try: return float(val) except (ValueError, TypeError): return None def fetch_filtered_data(current_mode, start_date, end_date, num, master_num, bon, limit_val=1000): if current_mode == 'thickness' and not start_date: start_date = '2026-01-01' conn = get_db_connection() data_list = [] try: with conn.cursor() as cursor: params = [] if current_mode == 'thickness': sql = """ SELECT no, num, master_num, '' as lot_spec, meter, bon, Rope_left, Rope_Center, Rope_Right, worker_id, datetime, 'thickness' as d_type FROM thickness WHERE 1=1 """ if start_date: sql += " AND datetime >= %s" params.append(f"{start_date} 00:00:00") if end_date: sql += " AND datetime <= %s" params.append(f"{end_date} 23:59:59") if master_num: sql += " AND master_num LIKE %s" params.append(f"%{master_num}%") else: sql = """ SELECT no, num, tag_material_number as master_num, lot_spec, meter, bon, Rope_left, Rope_Center, Rope_Right, worker_id, datetime, 'thickness1' as d_type FROM thickness1 WHERE 1=1 """ if start_date: sql += " AND datetime >= %s" params.append(f"{start_date} 00:00:00") if end_date: sql += " AND datetime <= %s" params.append(f"{end_date} 23:59:59") if master_num: sql += " AND tag_material_number LIKE %s" params.append(f"%{master_num}%") if num: sql += " AND num = %s" params.append(num) if bon: sql += " AND bon = %s" params.append(int(bon)) sql += " ORDER BY datetime DESC" if limit_val > 0: sql += " LIMIT %s" params.append(limit_val) cursor.execute(sql, params) data_list = cursor.fetchall() finally: conn.close() return data_list, start_date def process_advanced_filters(data_list, current_mode, sensor_min, sensor_max, time_interval, dedup_flag): """ 백엔드 고성능 복합 정제 엔진 """ processed = [] last_datetime = None seen_records = set() # 시간 정순 분석을 위해 reverse 처리 for row_data in reversed(data_list): l_f = safe_float(row_data.get('Rope_left')) c_f = safe_float(row_data.get('Rope_Center')) r_f = safe_float(row_data.get('Rope_Right')) # 1) 센서 범위 임계치 필터 if sensor_min is not None: if (l_f is not None and l_f < sensor_min) or (c_f is not None and c_f < sensor_min) or (r_f is not None and r_f < sensor_min): continue if sensor_max is not None: if (l_f is not None and l_f > sensor_max) or (c_f is not None and c_f > sensor_max) or (r_f is not None and r_f > sensor_max): continue # 2) 밀집 연속 로그 제한 필터 current_dt = row_data.get('datetime') if time_interval > 0 and isinstance(current_dt, datetime) and last_datetime is not None: if abs((current_dt - last_datetime).total_seconds()) <= time_interval: continue m_num = row_data.get('master_num', '') l_spec = row_data.get('lot_spec', '') display_master = f"{m_num} / {l_spec}" if current_mode != 'thickness' and l_spec else (m_num if m_num else '-') # 3) 중복 로우 압축 알고리즘 if dedup_flag == 'Y': dedup_key = ( str(row_data.get('num', '-')), str(display_master), str(row_data.get('meter', '-')), str(row_data.get('bon', '-')), f"{l_f:.2f}" if l_f is not None else '-', f"{c_f:.2f}" if c_f is not None else '-', f"{r_f:.2f}" if r_f is not None else '-', str(row_data.get('worker_id', '-')) ) if dedup_key in seen_records: continue seen_records.add(dedup_key) if isinstance(current_dt, datetime): last_datetime = current_dt row_copy = dict(row_data) row_copy['master_num'] = display_master # 포맷 통일 processed.append(row_copy) processed.reverse() # 다시 최신순 정렬 원복 return processed @app.route('/') def index(): current_mode = request.args.get('mode', 'thickness') start_date = request.args.get('start_date', '') end_date = request.args.get('end_date', '') num = request.args.get('num', '').strip() master_num = request.args.get('master_num', '').strip() bon = request.args.get('bon', '').strip() # 상세 필터 인자 수집 view_limit_raw = request.args.get('view_limit', '1000').strip() try: view_limit = max(1000, int(view_limit_raw)) if view_limit_raw else 1000 except ValueError: view_limit = 1000 excluded_cols = request.args.getlist('ex_col') sensor_min = safe_float(request.args.get('sensor_min', '').strip()) sensor_max = safe_float(request.args.get('sensor_max', '').strip()) time_interval = request.args.get('time_interval', '').strip() time_interval = int(time_interval) if time_interval and time_interval.isdigit() else 0 dedup_flag = request.args.get('dedup_flag', 'N') excel_limit = request.args.get('excel_limit', '').strip() # 원시 덤프 조회 raw_list, adjusted_start_date = fetch_filtered_data(current_mode, start_date, end_date, num, master_num, bon, limit_val=view_limit) # 고급 정제 필터 엔진 결합 통과 data_list = process_advanced_filters(raw_list, current_mode, sensor_min, sensor_max, time_interval, dedup_flag) left_sum, center_sum, right_sum = 0.0, 0.0, 0.0 left_cnt, center_cnt, right_cnt = 0, 0, 0 chart_labels, chart_left, chart_center, chart_right = [], [], [], [] averages = {'left': 0.0, 'center': 0.0, 'right': 0.0} for row in reversed(data_list): l_val = safe_float(row['Rope_left']) c_val = safe_float(row['Rope_Center']) r_val = safe_float(row['Rope_Right']) if l_val is not None: left_sum += l_val; left_cnt += 1 if c_val is not None: center_sum += c_val; center_cnt += 1 if r_val is not None: right_sum += r_val; right_cnt += 1 if row['datetime'] and hasattr(row['datetime'], 'strftime'): dt_str = row['datetime'].strftime('%m-%d %H:%M') else: dt_str = str(row['datetime'])[5:16] if row['datetime'] else '' chart_labels.append(dt_str) chart_left.append(l_val if l_val is not None else 0.0) chart_center.append(c_val if c_val is not None else 0.0) chart_right.append(r_val if r_val is not None else 0.0) if left_cnt > 0: averages['left'] = round(left_sum / left_cnt, 2) if center_cnt > 0: averages['center'] = round(center_sum / center_cnt, 2) if right_cnt > 0: averages['right'] = round(right_sum / right_cnt, 2) search_vals = { 'start_date': adjusted_start_date, 'end_date': end_date, 'num': num, 'master_num': master_num, 'bon': bon, 'view_limit': view_limit, 'sensor_min': request.args.get('sensor_min', ''), 'sensor_max': request.args.get('sensor_max', ''), 'time_interval': request.args.get('time_interval', ''), 'dedup_flag': dedup_flag, 'excel_limit': excel_limit, 'ex_col': excluded_cols } return render_template( 'index.html', data_list=data_list, averages=averages, current_mode=current_mode, chart_data={'labels': chart_labels, 'left': chart_left, 'center': chart_center, 'right': chart_right}, search_vals=search_vals ) @app.route('/download_excel') def download_excel(): current_mode = request.args.get('mode', 'thickness') start_date = request.args.get('start_date', '') end_date = request.args.get('end_date', '') num = request.args.get('num', '').strip() master_num = request.args.get('master_num', '').strip() bon = request.args.get('bon', '').strip() excel_limit_raw = request.args.get('excel_limit', '').strip() excluded_cols = request.args.getlist('ex_col') sensor_min = safe_float(request.args.get('sensor_min', '').strip()) sensor_max = safe_float(request.args.get('sensor_max', '').strip()) time_interval = request.args.get('time_interval', '').strip() time_interval = int(time_interval) if time_interval and time_interval.isdigit() else 0 dedup_flag = request.args.get('dedup_flag', 'N') try: excel_limit = int(excel_limit_raw) if excel_limit_raw else 0 except ValueError: excel_limit = 0 # 엑셀 다운로드 요청 시 사용자가 지정한 excel_limit 로 분기 조회 처리 raw_list, _ = fetch_filtered_data(current_mode, start_date, end_date, num, master_num, bon, limit_val=excel_limit) data_list = process_advanced_filters(raw_list, current_mode, sensor_min, sensor_max, time_interval, dedup_flag) wb = Workbook() ws = wb.active ws.title = "ThicknessData" second_header = "QRcode" if current_mode == 'thickness' else "MaterialNumber_LotSpec" all_headers = [ {"key": "no", "label": "no"}, {"key": "num", "label": "num"}, {"key": "master_num", "label": second_header}, {"key": "meter", "label": "meter"}, {"key": "bon", "label": "bon"}, {"key": "Rope_left", "label": "Rope_left"}, {"key": "Rope_Center", "label": "Rope_Center"}, {"key": "Rope_Right", "label": "Rope_Right"}, {"key": "worker_id", "label": "worker_id"}, {"key": "datetime", "label": "datetime"} ] active_headers = [h for h in all_headers if h["key"] not in excluded_cols] ws.append([h["label"] for h in active_headers]) header_font = Font(name='Arial', size=11, bold=True, color='FFFFFF') header_fill = PatternFill(start_color='4E73DF', end_color='4E73DF', fill_type='solid') center_alignment = Alignment(horizontal='center', vertical='center') for col_idx in range(1, len(active_headers) + 1): cell = ws.cell(row=1, column=col_idx) cell.font = header_font cell.fill = header_fill cell.alignment = center_alignment for row_data in data_list: if row_data.get('datetime') and hasattr(row_data['datetime'], 'strftime'): dt_str = row_data['datetime'].strftime('%Y-%m-%d %H:%M:%S') else: dt_str = str(row_data.get('datetime')) if row_data.get('datetime') else '-' l_f = safe_float(row_data.get('Rope_left')) c_f = safe_float(row_data.get('Rope_Center')) r_f = safe_float(row_data.get('Rope_Right')) l_val = round(l_f, 2) if l_f is not None else '-' c_val = round(c_f, 2) if c_f is not None else '-' r_val = round(r_f, 2) if r_f is not None else '-' full_row_map = { "no": row_data.get('no', '-'), "num": str(row_data['num']) if row_data.get('num') else '-', "master_num": str(row_data.get('master_num', '-')), "meter": row_data['meter'] if row_data.get('meter') is not None else 0, "bon": row_data['bon'] if row_data.get('bon') is not None else 0, "Rope_left": l_val, "Rope_Center": c_val, "Rope_Right": r_val, "worker_id": str(row_data['worker_id']) if row_data.get('worker_id') else '-', "datetime": dt_str } ws.append([full_row_map[h["key"]] for h in active_headers]) total_rows = len(data_list) + 1 if total_rows > 1: for row in ws.iter_rows(min_row=2, max_row=total_rows): for cell in row: cell.alignment = center_alignment for col in ws.columns: max_len = max(len(str(cell.value or '')) for cell in col) first_cell = col col_letter = openpyxl.utils.get_column_letter(first_cell.column) ws.column_dimensions[col_letter].width = max(max_len + 3, 12) excel_stream = io.BytesIO() wb.save(excel_stream) excel_stream.seek(0) file_name = f"Belt_Thickness_Custom_{current_mode}.xlsx" return send_file( excel_stream, mimetype='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet', as_attachment=True, download_name=file_name ) if __name__ == '__main__': app.run(host='0.0.0.0', port=5006)