402 lines
16 KiB
Python
402 lines
16 KiB
Python
from jsonpath_ng.ext import parse
|
|
import json
|
|
from itertools import chain
|
|
import os
|
|
import pandas as pd # 추가된 import
|
|
import unicodedata # 상단에 import 추가
|
|
import re # 상단에 추가
|
|
from datetime import datetime
|
|
import logging
|
|
from logging_config import setup_logging # logging 설정을 위한 import
|
|
import traceback
|
|
|
|
setup_logging() # logging 설정 호출
|
|
|
|
# JSON 파일 읽기
|
|
def read_json(file_path):
|
|
with open(file_path, 'r', encoding='utf-8') as file:
|
|
return json.load(file)
|
|
|
|
# 요소 탐색 함수
|
|
def find_element(project_data, jsonpath_expr):
|
|
jsonpath_expr = parse(jsonpath_expr)
|
|
result = []
|
|
for match in jsonpath_expr.find(project_data):
|
|
value = match.value
|
|
result.append(value)
|
|
return result
|
|
|
|
# 요소 탐색 함수
|
|
def find_script_element(project_data, jsonpath_expr):
|
|
jsonpath_expr = parse(jsonpath_expr)
|
|
match = jsonpath_expr.find(project_data)
|
|
|
|
if not match:
|
|
return None
|
|
|
|
return match[0].value
|
|
|
|
# jsonpath_expr_list 로 넘어온 jsonpath들을 하나씩 parse 해주고 결과를 result 리스트로 반환
|
|
def find_list_element(data, jsonpath_expr_list):
|
|
result = []
|
|
|
|
for jsonpath_expr in jsonpath_expr_list:
|
|
jsonpath_expr = parse(jsonpath_expr)
|
|
result.append([match.value for match in jsonpath_expr.find(data)])
|
|
|
|
return result
|
|
|
|
# 스크립트 채점 진행 전 스크립트 블럭 순서가 when_run_button_click 1번째, when_clone_start 2번째 배열에 없으면
|
|
# 리스트 순서 스왑해서 각각 0, 1번 순서로 배치될 수 있도록 함
|
|
# when_scene_start 1번째, when_message_cast 2번째 동일한 순서로 리스트 순서 스왑해서 정렬
|
|
# when_scene_start = [장면이 시작되었을 때]
|
|
# when_message_cast = ["성공"신호를 받았을 때]
|
|
def swap_script(origin):
|
|
"""스크립트 블록 순서 정렬 함수"""
|
|
if not origin or len(origin) == 0:
|
|
return origin
|
|
|
|
# 스크립트 블록 분류를 위한 임시 저장소
|
|
run_button_blocks = []
|
|
key_press_blocks = []
|
|
clone_start_block = None
|
|
message_cast_block = []
|
|
other_blocks = []
|
|
|
|
# 각 블록을 타입별로 분류
|
|
for i, block in enumerate(origin):
|
|
if not block or len(block) == 0:
|
|
continue
|
|
|
|
block_type = block[0].get("type")
|
|
|
|
# # 기존 로직: run_button과 scene_start 처리
|
|
# if (block_type == "when_run_button_click" or block_type == "when_scene_start") and i > 0:
|
|
# print(f"swap script run button or scene start")
|
|
# swap = origin[0]
|
|
# origin[0] = origin[i]
|
|
# origin[i] = swap
|
|
|
|
# # 기존 로직: clone_start와 message_cast 처리
|
|
# elif (block_type == "when_clone_start" or block_type == "when_message_cast") and i > 0:
|
|
# print(f"swap script clone start or msg cast")
|
|
# swap = origin[1]
|
|
# origin[1] = origin[i]
|
|
# origin[i] = swap
|
|
|
|
# 새로운 로직: 키 이벤트 처리
|
|
if block_type == "when_run_button_click":
|
|
run_button_blocks.append(block)
|
|
elif block_type == "when_some_key_pressed":
|
|
# params[1]에서 키 값 확인 (49, 50, 51) 아스키코드 값이므로 필요하면 추가 가능
|
|
key_value = block[0].get("params")[1]
|
|
if key_value in ["49", "50", "51"]:
|
|
key_press_blocks.append((key_value, block))
|
|
elif block_type == "when_clone_start":
|
|
clone_start_block = block
|
|
elif block_type == "when_message_cast":
|
|
message_cast_block.append(block)
|
|
else:
|
|
other_blocks.append(block)
|
|
|
|
# 결과 배열 재구성
|
|
result = []
|
|
|
|
# 1. when_run_button_click 블록을 첫 번째로 추가
|
|
if run_button_blocks:
|
|
result.extend(run_button_blocks)
|
|
|
|
# 2. when_some_key_pressed 블록들을 키 값 순서대로 정렬하여 추가
|
|
key_press_blocks.sort(key=lambda x: int(x[0]) if x[0] else 0)
|
|
result.extend([block for _, block in key_press_blocks])
|
|
|
|
# 3. clone_start와 message_cast 블록 추가
|
|
if clone_start_block:
|
|
result.append(clone_start_block)
|
|
if message_cast_block:
|
|
result.extend(message_cast_block)
|
|
|
|
# 4. 나머지 블록 추가
|
|
result.extend(other_blocks)
|
|
|
|
# 결과가 비어있으면 원본 반환
|
|
return result if result else origin
|
|
|
|
def clean_string(text):
|
|
"""문자열 끝의 . 또는 ! 제거"""
|
|
if isinstance(text, str):
|
|
# 문자열 끝의 . 또는 ! 제거
|
|
return re.sub(r'', '', text.strip())
|
|
return text
|
|
|
|
def convert_to_str(value):
|
|
"""값을 문자열로 변환"""
|
|
if isinstance(value, (list, tuple)):
|
|
return [convert_to_str(v) for v in value]
|
|
return str(value)
|
|
|
|
# def process_project(project_data, scoring_data):
|
|
total_points = 0
|
|
point_list = []
|
|
|
|
for key, question_info in scoring_data.items():
|
|
element_path = question_info.get('ele')
|
|
type = question_info.get('type')
|
|
block_list = question_info.get('blocks')
|
|
answer = question_info.get('answer')
|
|
points = question_info.get('points')
|
|
|
|
print(f"example: {key}")
|
|
if type == "scene":
|
|
found_elements = find_element(project_data, element_path)
|
|
if found_elements:
|
|
print(f"🟨 Elements found for {element_path}")
|
|
# scene type의 경우 문자열 변환
|
|
found_elements = [convert_to_str(x) for x in found_elements]
|
|
if found_elements == answer:
|
|
total_points += points
|
|
point_list.append(points)
|
|
|
|
elif found_elements and answer == None:
|
|
total_points += points
|
|
point_list.append(points)
|
|
print(f"{element_path} found ")
|
|
else:
|
|
point_list.append(0)
|
|
print(f"{found_elements} not found")
|
|
else:
|
|
print(f"🟥 Element '{element_path}' not found")
|
|
point_list.append(0)
|
|
|
|
# 스크립트 타입의 경우
|
|
# 스크립트 블록을 찾고, 블록 내의 요소를 확인
|
|
# 블록 내의 요소를 확인하고, 정답과 비교하여 점수 계산
|
|
if type == "script":
|
|
script_raw = find_script_element(project_data, element_path)
|
|
|
|
if script_raw == None:
|
|
script_data = None
|
|
else:
|
|
script_data = json.loads(script_raw)
|
|
script_data = swap_script(script_data)
|
|
|
|
block_index = 1
|
|
for block in block_list:
|
|
block_type = block.get('type')
|
|
|
|
if script_data == None:
|
|
print(f"{key}-{block_index}: Script Not exist")
|
|
point_list.append("확인 필요")
|
|
continue
|
|
|
|
elif block_type == "list":
|
|
block_elements = find_list_element(script_data, block.get('ele'))
|
|
else:
|
|
block_elements = find_element(script_data, block.get('ele'))
|
|
|
|
answer = block.get('answer', None)
|
|
|
|
if block_elements and isinstance(answer, list):
|
|
# 리스트의 모든 요소를 문자열로 변환
|
|
flat_matches = [convert_to_str(x) for x in list(chain.from_iterable(block_elements))]
|
|
else:
|
|
if not block_elements:
|
|
flat_matches = None
|
|
else:
|
|
# 단일 값을 문자열로 변환
|
|
flat_matches = convert_to_str(block_elements[0])
|
|
|
|
if block_elements:
|
|
# answer도 문자열로 변환하여 비교
|
|
str_answer = convert_to_str(answer) if answer is not None else None
|
|
if answer is not None and str_answer != flat_matches:
|
|
print(f"{key}-{block_index}: {str_answer} != {flat_matches}")
|
|
point_list.append(0)
|
|
elif answer is not None and str_answer == flat_matches:
|
|
print(f"{key}-{block_index}: {str_answer} == {flat_matches}")
|
|
total_points += block.get('points')
|
|
point_list.append(block.get('points'))
|
|
elif answer is None and block_elements:
|
|
total_points += block.get('points')
|
|
point_list.append(block.get('points'))
|
|
print(f"{key}-{block_index}: exist ele: {block_elements}")
|
|
else:
|
|
print(f"No elements found for {block.get('ele')}")
|
|
point_list.append(0)
|
|
|
|
block_index = block_index + 1
|
|
|
|
point_list.append(total_points)
|
|
return point_list
|
|
|
|
def process_project(project_data, scoring_data):
|
|
total_points = 0
|
|
score_list = []
|
|
|
|
for question_key, question_info in scoring_data.items():
|
|
element_path = question_info.get('ele')
|
|
question_type = question_info.get('type')
|
|
block_list = question_info.get('blocks')
|
|
expected_answer = question_info.get('answer')
|
|
question_points = question_info.get('points')
|
|
|
|
print(f"▶ Processing question: {question_key}")
|
|
|
|
# ✅ SCENE TYPE 처리
|
|
if question_type == "scene":
|
|
scene_elements = find_element(project_data, element_path)
|
|
|
|
if scene_elements:
|
|
print(f"🟨 Found elements for '{element_path}'")
|
|
scene_elements = [convert_to_str(x) for x in scene_elements]
|
|
|
|
if scene_elements == expected_answer or (expected_answer is None and scene_elements):
|
|
total_points += question_points
|
|
score_list.append(question_points)
|
|
# print(f"✅ Matched: {scene_elements}")
|
|
else:
|
|
score_list.append(0)
|
|
print(f"❌ Mismatch: {scene_elements} vs {expected_answer}")
|
|
else:
|
|
print(f"🟥 Element '{element_path}' not found")
|
|
score_list.append(0)
|
|
|
|
# ✅ SCRIPT TYPE 처리
|
|
elif question_type == "script":
|
|
script_raw = find_script_element(project_data, element_path)
|
|
script_data_1 = json.loads(script_raw) if script_raw else None
|
|
script_data = swap_script(script_data_1) if script_data_1 else None
|
|
|
|
block_index = 1
|
|
for block in block_list:
|
|
block_type = block.get('type')
|
|
block_path = block.get('ele')
|
|
block_answer = block.get('answer', None)
|
|
block_points = block.get('points')
|
|
|
|
if script_data is None:
|
|
print(f"{question_key}-{block_index}: Script Not Found")
|
|
score_list.append("확인 필요")
|
|
block_index += 1
|
|
continue
|
|
|
|
# 블록 요소 검색
|
|
if block_type == "list":
|
|
block_elements = find_list_element(script_data, block_path)
|
|
else:
|
|
block_elements = find_element(script_data, block_path)
|
|
|
|
# 결과값 정리
|
|
if block_elements and isinstance(block_answer, list):
|
|
found_values = [convert_to_str(x) for x in chain.from_iterable(block_elements)]
|
|
else:
|
|
found_values = convert_to_str(block_elements[0]) if block_elements else None
|
|
|
|
expected_str = convert_to_str(block_answer) if block_answer is not None else None
|
|
|
|
# 비교 및 점수 처리
|
|
if block_elements:
|
|
if block_answer is not None and expected_str != found_values:
|
|
print(f"{question_key}-{block_index}: ❌ {expected_str} != {found_values}")
|
|
score_list.append(0)
|
|
elif block_answer is not None and expected_str == found_values:
|
|
print(f"{question_key}-{block_index}: ✅ {expected_str} == {found_values}")
|
|
total_points += block_points
|
|
score_list.append(block_points)
|
|
elif block_answer is None:
|
|
total_points += block_points
|
|
score_list.append(block_points)
|
|
print(f"{question_key}-{block_index}: Element Exists")
|
|
else:
|
|
print(f"{question_key}-{block_index}: No elements found for {block_path}")
|
|
score_list.append(0)
|
|
|
|
block_index += 1
|
|
|
|
# total_points = round(total_points, ndigits=0) # 총점 반올림
|
|
score_list.append(total_points)
|
|
return score_list
|
|
|
|
def normalize_path(path):
|
|
"""한글 경로명을 NFC 방식으로 정규화"""
|
|
|
|
return unicodedata.normalize('NFC', path)
|
|
|
|
def main():
|
|
# 파일 경로 설정
|
|
project_json_path = './output/2507_CAT_3_A/'
|
|
# project_json_path = './output/00_test/'
|
|
scoring_json_path = './correct/2507_CAT_3_A.json'
|
|
|
|
scoring_data = read_json(scoring_json_path)
|
|
student_score_list = []
|
|
|
|
# 컬럼명 생성
|
|
columns = ['학생명']
|
|
idx = 1
|
|
for key in scoring_data.keys():
|
|
if scoring_data[key].get('type') == 'scene':
|
|
columns.append(f'{idx}')
|
|
idx = idx + 1
|
|
elif scoring_data[key].get('type') == 'script':
|
|
for i in range(len(scoring_data[key].get('blocks', []))):
|
|
columns.append(f'{idx}')
|
|
idx = idx + 1
|
|
|
|
columns.append('총점')
|
|
|
|
# os.walk 결과를 리스트로 변환하고 정렬
|
|
walk_results = []
|
|
for root, dirs, files in os.walk(project_json_path):
|
|
# 디렉토리명 정규화
|
|
normalized_root = normalize_path(root)
|
|
normalized_dirs = [normalize_path(d) for d in dirs]
|
|
normalized_files = [normalize_path(f) for f in files]
|
|
|
|
normalized_dirs.sort() # 정규화된 디렉토리 정렬
|
|
walk_results.append((normalized_root, normalized_dirs, normalized_files))
|
|
|
|
# 정렬된 결과를 바탕으로 처리
|
|
for root, dirs, files in sorted(walk_results):
|
|
for file in sorted(files): # 파일도 정렬
|
|
if file == 'project.json':
|
|
full_path = os.path.join(root, file)
|
|
print(f"\nProcessing: {full_path}")
|
|
try:
|
|
# 디렉토리 패스 내에 학생 이름만 뽑아서 엑셀 컬럼 명으로 추가
|
|
# output/cas-000040-이지원/temp/project.json
|
|
# student_id = normalize_path(full_path.split('/')[3])
|
|
match = re.search(r'(\d{6}[-_][^\\/]+)[\\/]', full_path)
|
|
if match:
|
|
student_id = match.group(1)
|
|
else:
|
|
if '정답' in full_path:
|
|
student_id = '정답'
|
|
else:
|
|
student_id = '000000'
|
|
|
|
# project.json 파일 내용
|
|
project_data = read_json(full_path)
|
|
points = process_project(project_data, scoring_data)
|
|
points.insert(0, student_id)
|
|
student_score_list.append(points)
|
|
print(f"Total Points for {points}")
|
|
except Exception as e:
|
|
# print(traceback.format_exc())
|
|
logging.exception(f"🚫Error processing {full_path}: {str(e)}")
|
|
continue
|
|
|
|
# DataFrame 생성 및 엑셀 저장
|
|
df = pd.DataFrame(student_score_list, columns=columns).transpose()
|
|
df.columns = df.iloc[0]
|
|
df = df[1:]
|
|
|
|
timestamp = datetime.now().strftime("%y%m%d")
|
|
|
|
excel_path = f'{project_json_path}/{timestamp}_results.xlsx'
|
|
|
|
df.to_excel(excel_path, index=False)
|
|
print(f"\nResults saved to {excel_path}")
|
|
|
|
if __name__ == "__main__":
|
|
main() |