import mysql.connector from app.core.config import get_db_config, logger from datetime import timedelta from datetime import datetime, timedelta import openpyxl import subprocess import os from openpyxl.formatting.rule import DataBarRule import pandas as pd import re from datetime import datetime from app.core.config import get_gemini_model import glob # C_DB 설정 가져오기 C_DB_CONFIG = get_db_config("Sales_C_DB") Goalskill_login_DB_CONFIG = get_db_config("Sales_Goalskill_login") def get_real_chart_data(session_id: str): config = get_db_config("Sales_C_DB") periods = { "日次": "daily", "週次": "weekly", "月次": "monthly", "1/4期次": "quarterly" } category_map = { 2: "架電数", 3: "繋がるための行動", 4: "アポ", 5: "訪問", 6: "見積", 8: "受注" } target_categories = [2, 3, 4, 5, 6, 8] response_data = {} conn = None cursor = None try: conn = mysql.connector.connect(**config) cursor = conn.cursor(dictionary=True) for label, suffix in periods.items(): p_table = f"P_{suffix}" t_table = f"T_{suffix}" period_list = [] format_strings = ','.join(['%s'] * len(target_categories)) # 1. P(개인) 데이터 조회 sql_p = f""" SELECT category, achievement_rate, target, achievement FROM {p_table} WHERE session_id = %s AND category IN ({format_strings}) ORDER BY record_date DESC, id DESC """ # 2. [수정] T(팀) 데이터 조회: target, achievement 컬럼 추가 sql_t = f""" SELECT category, achievement_rate, target, achievement FROM {t_table} WHERE session_id = %s AND category IN ({format_strings}) ORDER BY record_date DESC, id DESC """ params = [session_id] + target_categories # P 실행 cursor.execute(sql_p, params) p_rows = cursor.fetchall() # T 실행 cursor.execute(sql_t, params) t_rows = cursor.fetchall() # 중복 제거 (딕셔너리 변환) p_dict = {} for row in p_rows: if row['category'] not in p_dict: p_dict[row['category']] = row t_dict = {} for row in t_rows: if row['category'] not in t_dict: t_dict[row['category']] = row # row 전체 저장 # 데이터 병합 for cat_id in target_categories: # P 데이터 처리 p_row = p_dict.get(cat_id) if p_row: tanaka_rate = p_row['achievement_rate'] p_tgt = int(p_row['target']) if p_row['target'] is not None else 0 p_ach = int(p_row['achievement']) if p_row['achievement'] is not None else 0 else: tanaka_rate = 0 p_tgt = 0; p_ach = 0 # T 데이터 처리 [수정] t_row = t_dict.get(cat_id) if t_row: team_rate = t_row['achievement_rate'] t_tgt = int(t_row['target']) if t_row['target'] is not None else 0 t_ach = int(t_row['achievement']) if t_row['achievement'] is not None else 0 else: team_rate = 0 t_tgt = 0; t_ach = 0 if team_rate is None: team_rate = 0 diff = team_rate - tanaka_rate period_list.append({ "title": category_map[cat_id], "tanaka": tanaka_rate, "team": team_rate, "diff": diff, "p_target": p_tgt, # 다나카 목표 "p_achievement": p_ach, # 다나카 실적 "t_target": t_tgt, # [추가] 팀 목표 "t_achievement": t_ach # [추가] 팀 실적 }) response_data[label] = period_list return response_data except Exception as e: logger.error(f"Real Chart Data Error: {e}") return {} finally: if cursor: cursor.close() if conn: conn.close() def get_am_chart_data(session_id: str): return _get_ampm_chart_data(session_id, "am") def get_pm_chart_data(session_id: str): return _get_ampm_chart_data(session_id, "pm") def _get_ampm_chart_data(session_id: str, time_type: str): config = get_db_config("Sales_C_DB") periods = { "日次": "daily", "週次": "weekly", "月次": "monthly", "1/4期次": "quarterly" } category_map = { 2: "架電数", 3: "繋がるための行動", 4: "アポ", 5: "訪問", 6: "見積", 8: "受注" } target_categories = [2, 3, 4, 5, 6, 8] response_data = {} conn = None cursor = None try: conn = mysql.connector.connect(**config) cursor = conn.cursor(dictionary=True) for label, suffix in periods.items(): table_name = f"today_sales_{time_type}_{suffix}" period_list = [] format_strings = ','.join(['%s'] * len(target_categories)) # [수정] target, achievement 추가 조회 sql = f""" SELECT category, achievement_rate, target, achievement FROM {table_name} WHERE session_id = %s AND category IN ({format_strings}) ORDER BY record_date DESC, id DESC """ params = [session_id] + target_categories cursor.execute(sql, params) rows = cursor.fetchall() data_dict = {} for row in rows: if row['category'] not in data_dict: data_dict[row['category']] = row for cat_id in target_categories: row = data_dict.get(cat_id) if row: rate = row['achievement_rate'] tgt = int(row['target']) if row['target'] is not None else 0 ach = int(row['achievement']) if row['achievement'] is not None else 0 else: rate = 0 tgt = 0 ach = 0 period_list.append({ "title": category_map[cat_id], "rate": rate, "target": tgt, # [추가] "achievement": ach # [추가] }) response_data[label] = period_list return response_data except Exception as e: logger.error(f"{time_type.upper()} Chart Data Error: {e}") return {} finally: if cursor: cursor.close() if conn: conn.close() def get_result_chart_data(session_id: str): """ Result 테이블(result_daily 등)에서 성과율뿐만 아니라 target(목표)과 achievement(실적) 값도 함께 조회하여 반환 """ config = get_db_config("Sales_C_DB") periods = { "日次": "daily", "週次": "weekly", "月次": "monthly", "1/4期次": "quarterly" } category_map = { 2: "架電数", 3: "繋がるための行動", 4: "アポ", 5: "訪問", 6: "見積", 8: "受注" } target_categories = [2, 3, 4, 5, 6, 8] response_data = {} conn = None cursor = None try: conn = mysql.connector.connect(**config) cursor = conn.cursor(dictionary=True) for label, suffix in periods.items(): table_name = f"result_{suffix}" # 예: result_daily period_list = [] format_strings = ','.join(['%s'] * len(target_categories)) # ★ target, achievement 컬럼 추가 조회 sql = f""" SELECT category, achievement_rate, target, achievement FROM {table_name} WHERE session_id = %s AND category IN ({format_strings}) ORDER BY record_date DESC, id DESC """ params = [session_id] + target_categories cursor.execute(sql, params) rows = cursor.fetchall() # 중복 제거 (최신 데이터만) data_dict = {} for row in rows: if row['category'] not in data_dict: data_dict[row['category']] = row # 결과 리스트 생성 for cat_id in target_categories: row = data_dict.get(cat_id) if row: rate = row['achievement_rate'] target_val = row['target'] achieve_val = row['achievement'] else: rate = 0 target_val = 0 achieve_val = 0 # Decimal 타입일 경우 float/int 변환 (JSON 직렬화 위해) if hasattr(target_val, 'to_integral_value'): target_val = int(target_val) if hasattr(achieve_val, 'to_integral_value'): achieve_val = int(achieve_val) period_list.append({ "title": category_map[cat_id], "rate": rate, "target": target_val, # 목표 수치 "achievement": achieve_val # 달성 수치 }) response_data[label] = period_list return response_data except Exception as e: logger.error(f"Result Chart Data Error: {e}") return {} finally: if cursor: cursor.close() if conn: conn.close() def execute_afternoon_adjustment(session_id: str): """ 오후 계획 자동 조정 (Sales_C_DB.P_daily의 target을 100% 기준으로 조회 -> AI 분석(70~75%) -> DB & Excel 업데이트) """ conn_b = None cursor_b = None conn_c = None cursor_c = None try: # ========================================================= # [Step 1] 데이터 수집 # ========================================================= # 1. Sales_B_DB 연결 (현재 상태 및 계획 ID 조회용) b_config = get_db_config("Sales_B_DB") conn_b = mysql.connector.connect(**b_config) cursor_b = conn_b.cursor(dictionary=True) # 최신 daily_plan 가져오기 sql_plan = """ SELECT id, morning_motivation FROM daily_plans WHERE session_id = %s ORDER BY date DESC, id DESC LIMIT 1 """ cursor_b.execute(sql_plan, (session_id,)) plan_row = cursor_b.fetchone() if not plan_row: return {"status": "error", "message": "No daily plan found in DB"} plan_id = plan_row['id'] motivation = plan_row['morning_motivation'] or "特になし" # 2. [수정됨] Sales_C_DB 연결 및 100% 기준 목표(P_daily.target) 가져오기 c_config = get_db_config("Sales_C_DB") conn_c = mysql.connector.connect(**c_config) cursor_c = conn_c.cursor(dictionary=True) target_categories = [2, 3, 4, 5, 6, 8] format_strings = ','.join(['%s'] * len(target_categories)) # P_daily에서 해당 세션의 최신 타겟 정보를 가져옵니다. # category 2(가전), 3(연결), 4(아포), 5(방문), 6(견적), 8(수주) sql_p_daily = f""" SELECT category, target FROM P_daily WHERE session_id = %s AND category IN ({format_strings}) ORDER BY record_date DESC, id DESC """ cursor_c.execute(sql_p_daily, [session_id] + target_categories) p_rows = cursor_c.fetchall() # { category_id : 100%기준값 } # 중복된 카테고리가 있다면 쿼리 정렬 순서상 최신 것이 덮어씌워집니다. standard_targets = {} for row in p_rows: cat = row['category'] # Decimal 등을 int로 변환 val = row['target'] standard_targets[cat] = int(val) if val is not None else 0 # Sales_C_DB 연결 종료 (데이터 확보 완료) cursor_c.close() conn_c.close() conn_c = None # finally 블록 중복 방지 # 3. Sales_B_DB에서 오늘의 당초 목표(original)와 현재 실적(current) 가져오기 sql_goals = f""" SELECT category, original_target_count, current_count FROM goals WHERE plan_id = %s AND category IN ({format_strings}) """ cursor_b.execute(sql_goals, [plan_id] + target_categories) goal_rows = cursor_b.fetchall() # ========================================================= # [Step 2] AI 프롬프트 데이터 구성 # ========================================================= cat_map_name = { 2: "架電数", 3: "繋がるための行動", 4: "アポ", 5: "訪問", 6: "見積", 8: "受注" } status_lines = [] ai_input_data = {} for row in goal_rows: cat_id = row['category'] name = cat_map_name.get(cat_id, "Unknown") # 1. 100% 기준 목표 (P_daily에서 가져온 값) std_target = standard_targets.get(cat_id, 0) # 2. 오늘의 당초 목표 (B_DB goals) plan_target = int(row['original_target_count']) if row['original_target_count'] is not None else 0 # 3. 현재 실적 (B_DB goals) achieved = int(row['current_count']) if row['current_count'] is not None else 0 # 만약 P_daily 값이 없거나 0이면, 안전장치로 현재 목표를 100%로 간주 if std_target == 0: std_target = plan_target logger.warning(f"No P_daily target for {name}, using plan_target ({plan_target}) as standard.") ai_input_data[name] = { "cat_id": cat_id, "std_target": std_target, "plan_target": plan_target, "achieved": achieved } status_lines.append( f"{name} -> 【100%基準(P_daily): {std_target}件】 / 本日当初計画: {plan_target}件 / 現在実績: {achieved}件" ) status_text = "\n".join(status_lines) # AI 프롬프트 (100% 기준의 70~75% 계산 지시) prompt = f""" あなたは敏腕営業マネージャーです。 部下（田中）の進捗状況を分析し、午後から達成すべき「今日1日の最終的な修正目標数値」を再設定してください。 【部下の現状】 モチベーション: {motivation} 【数値データ (重要)】 {status_text} 【計算ルール】 1. 各項目の「100%基準(P_daily)」がその日の完全達成ライン(MAX)です。**絶対にこれを超えてはいけません。** (例: 100%基準が20件なら、25件にするのは間違いです) 2. 体調や進捗を考慮し、**「100%基準」に対して「70% 〜 75%」の範囲**になるように目標を再計算してください。 - 計算式: (100%基準 × 0.70) 〜 (100%基準 × 0.75) - 計算結果が小数になる場合は、整数（四捨五入または切り捨て）にしてください。 3. もし現在実績が既に75%を超えている場合は、その実績値を維持、または無理のない範囲（MAX 80%程度）で微増させてください。 4. 出力は「今日1日のトータル目標値」です。 【重要】 ※ 金額ではなく、必ず「件数(件)」で出力してください。 【出力フォーマット】(数値のみ、単位「件」をつける) 架電数: N件 繋がるための行動: N件 アポ: N件 訪問: N件 見積: N件 受注: N件 """ client = get_gemini_model() response = client.models.generate_content( model="gemini-2.5-flash", contents=prompt ) ai_output = response.text logger.info(f"Gemini Plan Output: \n{ai_output}") # ========================================================= # [Step 3] AI 결과 파싱 및 DB 업데이트 (Sales_B_DB) # ========================================================= new_targets_map = {} for line in ai_output.strip().split('\n'): for name in cat_map_name.values(): if name in line: match = re.search(r'(\d+)', line) if match: new_targets_map[name] = int(match.group(1)) updated_db_counts = 0 for name, new_val in new_targets_map.items(): cat_data = ai_input_data.get(name) if cat_data: cat_id = cat_data["cat_id"] update_sql = """ UPDATE goals SET adjusted_target_count = %s WHERE plan_id = %s AND category = %s """ cursor_b.execute(update_sql, (new_val, plan_id, cat_id)) updated_db_counts += 1 conn_b.commit() # ========================================================= # [Step 4] 엑셀 업데이트 (Sales_NEW_GOAL) # ========================================================= base_dir_write = "/home/air/goalskill_sales/front/progress_xls/Sales_NEW_GOAL" target_date = "20260219" search_pattern_write = os.path.join(base_dir_write, f"{target_date}_Sales_today_NEW_GOAL_{session_id}.xlsx") files_write = glob.glob(search_pattern_write) if not files_write: return { "status": "success", "message": f"DB updated ({updated_db_counts}), but NEW_GOAL Excel not found.", "ai_output": ai_output } file_path_write = files_write[0] wb_val = openpyxl.load_workbook(file_path_write, data_only=True) ws_val = wb_val['Data入力'] wb_write = openpyxl.load_workbook(file_path_write, data_only=False) ws_write = wb_write['Data入力'] # Tanaka 찾기 tanaka_start_row = None for row in range(13, 50): val_a = str(ws_val.cell(row=row, column=1).value or "") val_b = str(ws_val.cell(row=row, column=2).value or "") if "田中" in val_a or "田中" in val_b: tanaka_start_row = row break if tanaka_start_row is None: return {"status": "error", "message": "Could not find '田中' section in Excel."} # Daily 행 찾기 target_row = None for row in range(tanaka_start_row, tanaka_start_row + 15): row_vals = [str(ws_val.cell(row=row, column=c).value) for c in range(1, 5)] if any("日次" in s for s in row_vals): target_row = row break if target_row: col_map_excel = { "架電数": 8, "繋がるための行動": 11, "アポ": 14, "訪問": 17, "見積": 20, "受注": 26 } for name, new_val in new_targets_map.items(): if name in col_map_excel: col_idx = col_map_excel[name] # 엑셀 기존값 읽기 current_val = ws_val.cell(row=target_row, column=col_idx).value if current_val is None: current_val = 0 try: current_val = int(current_val) except ValueError: current_val = 0 # 차이 계산 및 업데이트 diff = current_val - new_val ws_write.cell(row=target_row, column=col_idx).value = new_val # 상위 기간 연쇄 업데이트 if diff != 0: for offset in [1, 2, 3]: r_idx = target_row - offset old_upper_val = ws_val.cell(row=r_idx, column=col_idx).value try: old_upper_val = int(old_upper_val) if old_upper_val is not None else 0 except: old_upper_val = 0 ws_write.cell(row=r_idx, column=col_idx).value = old_upper_val - diff wb_write.save(file_path_write) excel_msg = "Excel updated successfully." else: excel_msg = "Could not find '日次' row." return { "status": "success", "message": f"Updated DB & Excel (Based on P_daily Standard). {excel_msg}", "ai_output": ai_output } except Exception as e: logger.error(f"Afternoon Plan Logic Error: {e}") return {"status": "error", "message": str(e)} finally: # 두 DB 연결 모두 안전하게 닫기 if 'cursor_b' in locals() and cursor_b: cursor_b.close() if 'conn_b' in locals() and conn_b: conn_b.close() if 'cursor_c' in locals() and cursor_c: cursor_c.close() if 'conn_c' in locals() and conn_c: conn_c.close() def execute_report_update(session_id: str, report_type: str): """ AM/PM 리포트 엑셀 업데이트 + Sales_today 종합 리포트 업데이트 함수 """ conn = None cursor = None try: # ========================================================= # [Step 1] DB 연결 및 데이터 조회 (공통) # ========================================================= b_config = get_db_config("Sales_B_DB") conn = mysql.connector.connect(**b_config) cursor = conn.cursor(dictionary=True) # 최신 daily_plan 찾기 sql_plan = """ SELECT id, date FROM daily_plans WHERE session_id = %s ORDER BY date DESC, id DESC LIMIT 1 """ cursor.execute(sql_plan, (session_id,)) plan_row = cursor.fetchone() if not plan_row: return {"status": "error", "message": "No daily plan found in DB"} plan_id = plan_row['id'] logger.info(f"Target Plan ID: {plan_id}") # 현재 실적(current_count) 가져오기 target_categories = [2, 3, 4, 5, 6, 8] format_strings = ','.join(['%s'] * len(target_categories)) sql_goals = f""" SELECT category, current_count FROM goals WHERE plan_id = %s AND category IN ({format_strings}) """ cursor.execute(sql_goals, [plan_id] + target_categories) goal_rows = cursor.fetchall() # 카테고리별 실적 매핑 (int 변환) current_counts_map = {} for row in goal_rows: try: cat_key = int(row['category']) val = row['current_count'] count_val = int(val) if val is not None else 0 current_counts_map[cat_key] = count_val except Exception as e: logger.error(f"Data conversion error: {e}") logger.info(f"Current Counts: {current_counts_map}") # --------------------------------------------------------- # 업데이트할 엑셀 컬럼 매핑 (공통) # H=8, K=11, N=14, Q=17, T=20, Z=26 # --------------------------------------------------------- cat_to_col = { 2: 8, # 架電数 3: 11, # 繋がる 4: 14, # アポ 5: 17, # 訪問 6: 20, # 見積 8: 26 # 受注 } target_date = "20260219" # 날짜 고정 # ========================================================= # [Step 2] 개별 리포트 (Sales_AM 또는 Sales_PM) 업데이트 # ========================================================= folder_name = f"Sales_{report_type.upper()}" base_dir = f"/home/air/goalskill_sales/front/progress_xls/{folder_name}" search_pattern = os.path.join(base_dir, f"{target_date}_Sales_today_report_{report_type.upper()}_{session_id}.xlsx") files = glob.glob(search_pattern) msg_part1 = "" if files: file_path = files[0] wb_val = openpyxl.load_workbook(file_path, data_only=True) ws_val = wb_val['Data入力'] wb_write = openpyxl.load_workbook(file_path, data_only=False) ws_write = wb_write['Data入力'] # Tanaka 찾기 tanaka_row = None for row in range(13, 50): val_a = str(ws_val.cell(row=row, column=1).value or "") val_b = str(ws_val.cell(row=row, column=2).value or "") if "田中" in val_a or "田中" in val_b: tanaka_row = row break if not tanaka_row: # 안전장치: 전체 재검색 for row in range(1, 50): val_a = str(ws_val.cell(row=row, column=1).value or "") val_b = str(ws_val.cell(row=row, column=2).value or "") if "田中" in val_a or "田中" in val_b: tanaka_row = row break if tanaka_row: # '日次' 행 찾기 target_row = None for row in range(tanaka_row, tanaka_row + 15): vals = [str(ws_val.cell(row=row, column=c).value) for c in range(1, 5)] if any("日次" in s for s in vals): target_row = row break if target_row: for cat_id, col_idx in cat_to_col.items(): cur_count = current_counts_map.get(cat_id, 0) # 일차 덮어쓰기 ws_write.cell(row=target_row, column=col_idx).value = cur_count # 상위 기간 더하기 for offset in [1, 2, 3]: r_idx = target_row - offset old_val = ws_val.cell(row=r_idx, column=col_idx).value try: old_val = int(old_val) if old_val is not None else 0 except: old_val = 0 ws_write.cell(row=r_idx, column=col_idx).value = old_val + cur_count wb_write.save(file_path) msg_part1 = f"{report_type} file updated." else: msg_part1 = f"{report_type} file found but 'Daily' row missing." else: msg_part1 = f"{report_type} file found but 'Tanaka' missing." else: msg_part1 = f"{report_type} file NOT found." # ========================================================= # [Step 3] 종합 리포트 (Sales_today) 업데이트 ★ 추가된 로직 # ========================================================= base_dir_today = "/home/air/goalskill_sales/front/progress_xls/Sales_today" search_pattern_today = os.path.join(base_dir_today, f"{target_date}_Sales_today_report_{session_id}.xlsx") # 파일명 주의 (sess_... 전후 확인) # 만약 파일명 패턴이 sess_ID가 맨 뒤라면 아래 패턴 사용 # 여기서는 요청주신 "/home/air/.../20260219_Sales_today_report_sess_....xlsx" 경로에 맞춤 files_today = glob.glob(search_pattern_today) msg_part2 = "" if files_today: file_path_today = files_today[0] # 읽기/쓰기 로드 wb_val_t = openpyxl.load_workbook(file_path_today, data_only=True) ws_val_t = wb_val_t['Data入力'] wb_write_t = openpyxl.load_workbook(file_path_today, data_only=False) ws_write_t = wb_write_t['Data入力'] # 업데이트해야 할 '日次' 행 번호 리스트 결정 # AM인 경우: 오전 실적 (Row 30) # PM인 경우: 오후 실적 (Row 40) AND 오늘 실적 (Row 20) target_daily_rows = [] if report_type == "AM": # 다나카 오전 실적: 일차=30 (27~30행) target_daily_rows.append(30) elif report_type == "PM": # 다나카 오후 실적: 일차=40 (37~40행) target_daily_rows.append(40) # 다나카 오늘 실적: 일차=20 (17~20행) target_daily_rows.append(20) updated_today_count = 0 for target_r in target_daily_rows: for cat_id, col_idx in cat_to_col.items(): cur_count = current_counts_map.get(cat_id, 0) # (1) 일차(Daily) 셀 덮어쓰기 # 지정해주신 행(30, 40, 20)이 정확히 '일차' 위치라고 가정하고 업데이트 ws_write_t.cell(row=target_r, column=col_idx).value = cur_count # (2) 주/월/분기 셀 더하기 (위로 1, 2, 3칸) for offset in [1, 2, 3]: upper_row = target_r - offset # 기존 값 읽기 (수식이 있을 수 있으니 값 모드에서) old_val = ws_val_t.cell(row=upper_row, column=col_idx).value try: old_val = int(old_val) if old_val is not None else 0 except: old_val = 0 # 더해서 쓰기 ws_write_t.cell(row=upper_row, column=col_idx).value = old_val + cur_count updated_today_count += 1 wb_write_t.save(file_path_today) msg_part2 = f"Sales_today file updated (Rows: {target_daily_rows})." else: msg_part2 = "Sales_today file NOT found." return { "status": "success", "message": f"{msg_part1} | {msg_part2}", "file_am_pm": os.path.basename(files[0]) if files else "None", "file_today": os.path.basename(files_today[0]) if files_today else "None" } except Exception as e: logger.error(f"Report Update Error: {e}") return {"status": "error", "message": str(e)} finally: if cursor: cursor.close() if conn: conn.close() def update_achievement_rate_db(session_id: str, time_type: str): """ AM/PM Daily 테이블의 달성률(achievement_rate) 계산 및 업데이트 Formula: (achievement / target) * 100 -> INT """ conn = None cursor = None try: # 1. DB 연결 (Sales_C_DB) c_config = get_db_config("Sales_C_DB") conn = mysql.connector.connect(**c_config) cursor = conn.cursor(dictionary=True) # 테이블명 결정 (today_sales_am_daily 또는 today_sales_pm_daily) table_name = f"today_sales_{time_type.lower()}_daily" # 2. 해당 세션의 데이터 조회 (id, target, achievement) sql_select = f""" SELECT id, category, target, achievement FROM {table_name} WHERE session_id = %s """ cursor.execute(sql_select, (session_id,)) rows = cursor.fetchall() updated_count = 0 # 3. 계산 및 업데이트 루프 for row in rows: record_id = row['id'] # None 체크 및 형변환 tgt = float(row['target']) if row['target'] is not None else 0.0 ach = float(row['achievement']) if row['achievement'] is not None else 0.0 # 달성률 계산 # 목표가 0이면 나눗셈 불가하므로 0% 처리 (혹은 무한대 개념이지만 보통 0으로 둠) if tgt > 0: rate_val = (ach / tgt) * 100 rate_int = int(rate_val) # 정수 변환 (버림) else: rate_int = 0 # DB 업데이트 sql_update = f""" UPDATE {table_name} SET achievement_rate = %s WHERE id = %s """ cursor.execute(sql_update, (rate_int, record_id)) updated_count += 1 conn.commit() return { "status": "success", "message": f"Updated achievement_rate for {table_name}. ({updated_count} rows)", "updated_rows": updated_count } except Exception as e: logger.error(f"Achievement Rate Update Error ({time_type}): {e}") return {"status": "error", "message": str(e)} finally: if cursor: cursor.close() if conn: conn.close() # ======================================== # デバッグ日誌 차트 데이터 (MBTI + SPI) # ======================================== def get_debug_chart_data(session_id: str): """ MBTI scores와 SPI 결과를 Sales_A_DB에서 조회하여 반환 """ import json a_config = get_db_config("Sales_A_DB") conn = None cursor = None result = { "mbti": None, "mbti_type": None, "spi": None, "ojt": None, "ojt_average": None } try: conn = mysql.connector.connect(**a_config) cursor = conn.cursor(dictionary=True, buffered=True) # 1. MBTI 데이터 조회 (최신 1건) sql_mbti = """ SELECT result, scores FROM mbti_Output WHERE session_id = %s ORDER BY created_at DESC LIMIT 1 """ cursor.execute(sql_mbti, (session_id,)) mbti_row = cursor.fetchone() if mbti_row and mbti_row.get("scores"): try: result["mbti"] = json.loads(mbti_row["scores"]) result["mbti_type"] = mbti_row.get("result", "") except json.JSONDecodeError: logger.error(f"MBTI scores JSON parse error for session: {session_id}") # 2. SPI 데이터 조회 (최신 1건) sql_spi = """ SELECT language_score, nonverbal_score, personality_score FROM spi_Output WHERE session_id = %s ORDER BY created_at DESC LIMIT 1 """ cursor.execute(sql_spi, (session_id,)) spi_row = cursor.fetchone() if spi_row: spi_data = { "language_score": float(spi_row["language_score"]) if spi_row["language_score"] is not None else 0, "nonverbal_score": float(spi_row["nonverbal_score"]) if spi_row["nonverbal_score"] is not None else 0, "personality": {} } # personality_score JSON 파싱 if spi_row.get("personality_score"): try: p_data = spi_row["personality_score"] if isinstance(p_data, str): p_data = json.loads(p_data) # {"積極性": {"score": 3.5, "level": "中"}, ...} 형태에서 score만 추출 for key, val in p_data.items(): if isinstance(val, dict) and "score" in val: spi_data["personality"][key] = val["score"] elif isinstance(val, (int, float)): spi_data["personality"][key] = val except (json.JSONDecodeError, TypeError) as e: logger.error(f"SPI personality_score parse error: {e}") result["spi"] = spi_data # 3. Human OS (OJT) 데이터 조회 (최신 1건) sql_ojt = """ SELECT Node_score, output_value FROM human_os_Output WHERE session_id = %s ORDER BY created_at DESC LIMIT 1 """ cursor.execute(sql_ojt, (session_id,)) ojt_row = cursor.fetchone() if ojt_row: if ojt_row.get("Node_score"): try: ojt_data = ojt_row["Node_score"] if isinstance(ojt_data, str): ojt_data = json.loads(ojt_data) result["ojt"] = ojt_data except (json.JSONDecodeError, TypeError) as e: logger.error(f"OJT Node_score parse error: {e}") if ojt_row.get("output_value") is not None: result["ojt_average"] = float(ojt_row["output_value"]) return result except Exception as e: logger.error(f"Debug Chart Data Error: {e}") return result finally: if cursor: cursor.close() if conn: conn.close()