데이터파이프 라인에서 리스트 타입 지원
list:int, list:bool 등
This commit is contained in:
@@ -1,5 +1,5 @@
|
||||
# DataTools/generate_csharp_classes.py
|
||||
"""노션 스키마 → Unity C# ScriptableObject 클래스 생성"""
|
||||
"""노션 스키마 → Unity C# ScriptableObject 클래스 생성 (리스트 지원 버전)"""
|
||||
|
||||
import json
|
||||
from pathlib import Path
|
||||
@@ -10,7 +10,18 @@ UNITY_OUTPUT_DIR = Path(__file__).parent.parent / "Assets" / "Data" / "Scripts"
|
||||
|
||||
|
||||
def get_csharp_type(field_type):
|
||||
"""Python 타입 → C# 타입 변환"""
|
||||
"""Python 타입 → C# 타입 변환 (list 지원)"""
|
||||
# 리스트 여부 확인 (예: list:int, list:string)
|
||||
is_list = field_type.lower().startswith('list')
|
||||
|
||||
# 기본 타입 추출
|
||||
base_type = field_type.lower()
|
||||
if is_list and ':' in base_type:
|
||||
base_type = base_type.split(':')[1]
|
||||
elif is_list:
|
||||
# 'list'만 적혀있을 경우 기본적으로 string 리스트로 간주
|
||||
base_type = 'string'
|
||||
|
||||
type_map = {
|
||||
'int': 'int',
|
||||
'float': 'float',
|
||||
@@ -19,87 +30,54 @@ def get_csharp_type(field_type):
|
||||
'bool': 'bool',
|
||||
'boolean': 'bool'
|
||||
}
|
||||
return type_map.get(field_type.lower(), 'string')
|
||||
|
||||
csharp_base = type_map.get(base_type, 'string')
|
||||
|
||||
if is_list:
|
||||
return f"List<{csharp_base}>"
|
||||
return csharp_base
|
||||
|
||||
|
||||
def to_camel_case(snake_str):
|
||||
"""
|
||||
snake_case → camelCase
|
||||
|
||||
예시:
|
||||
- tower_type → towerType
|
||||
- damage → damage
|
||||
- attack_speed → attackSpeed
|
||||
"""
|
||||
"""snake_case → camelCase"""
|
||||
components = snake_str.split('_')
|
||||
return components[0] + ''.join(x.title() for x in components[1:])
|
||||
|
||||
|
||||
def to_pascal_case(snake_str):
|
||||
"""
|
||||
snake_case → PascalCase
|
||||
|
||||
예시:
|
||||
- tower_type → TowerType
|
||||
- damage → Damage
|
||||
"""
|
||||
"""snake_case → PascalCase"""
|
||||
return ''.join(x.title() for x in snake_str.split('_'))
|
||||
|
||||
|
||||
def group_fields_by_condition(schema):
|
||||
"""
|
||||
필드를 조건별로 그룹화
|
||||
|
||||
반환:
|
||||
{
|
||||
'common': [공통 필드들],
|
||||
'attack': [attack 타입 필드들],
|
||||
'defense': [defense 타입 필드들],
|
||||
...
|
||||
}
|
||||
"""
|
||||
"""필드를 조건별로 그룹화"""
|
||||
groups = defaultdict(list)
|
||||
|
||||
for field in schema:
|
||||
if field.get('condition'):
|
||||
condition_value = field['condition']['value']
|
||||
groups[condition_value].append(field)
|
||||
else:
|
||||
groups['common'].append(field)
|
||||
|
||||
return groups
|
||||
|
||||
|
||||
def generate_class(schema_name, schema):
|
||||
"""
|
||||
스키마 → C# 클래스 코드 생성
|
||||
|
||||
schema: [
|
||||
{'name': 'id', 'type': 'int', 'condition': None, 'description': '...'},
|
||||
{'name': 'damage', 'type': 'float', 'condition': {...}, 'description': '...'},
|
||||
...
|
||||
]
|
||||
"""
|
||||
"""스키마 → C# 클래스 코드 생성"""
|
||||
class_name = schema_name + "Data"
|
||||
|
||||
# 필드 그룹화
|
||||
field_groups = group_fields_by_condition(schema)
|
||||
|
||||
# 조건 필드명 찾기 (예: tower_type, enemy_type)
|
||||
condition_field = None
|
||||
for field in schema:
|
||||
if field.get('condition'):
|
||||
condition_field = field['condition']['field']
|
||||
break
|
||||
|
||||
# C# 코드 생성 시작
|
||||
lines = []
|
||||
|
||||
# 파일 헤더
|
||||
lines.append("// 이 파일은 자동 생성되었습니다. 직접 수정하지 마세요!")
|
||||
lines.append("// 생성 스크립트: DataTools/generate_csharp_classes.py")
|
||||
lines.append("")
|
||||
lines.append("using UnityEngine;")
|
||||
lines.append("using System.Collections.Generic; // 리스트 지원을 위해 추가")
|
||||
lines.append("")
|
||||
lines.append("namespace Northbound.Data")
|
||||
lines.append("{")
|
||||
@@ -121,10 +99,8 @@ def generate_class(schema_name, schema):
|
||||
|
||||
header_name = condition_value.capitalize()
|
||||
lines.append(f" [Header(\"{header_name} 전용\")]")
|
||||
|
||||
for field in fields:
|
||||
add_field(lines, field, nullable=True)
|
||||
|
||||
lines.append("")
|
||||
|
||||
# 유틸리티 메서드
|
||||
@@ -132,30 +108,27 @@ def generate_class(schema_name, schema):
|
||||
lines.append(" // ===== 유틸리티 메서드 =====")
|
||||
lines.append("")
|
||||
|
||||
# 타입 체크 프로퍼티
|
||||
for condition_value in field_groups.keys():
|
||||
if condition_value == 'common':
|
||||
continue
|
||||
|
||||
property_name = f"Is{to_pascal_case(condition_value)}"
|
||||
field_name = to_camel_case(condition_field)
|
||||
|
||||
lines.append(f' public bool {property_name} => {field_name} == "{condition_value}";')
|
||||
|
||||
lines.append("")
|
||||
|
||||
# Nullable 필드 Getter 메서드
|
||||
for condition_value, fields in field_groups.items():
|
||||
if condition_value == 'common':
|
||||
continue
|
||||
|
||||
for field in fields:
|
||||
field_name = to_camel_case(field['name'])
|
||||
method_name = f"Get{to_pascal_case(field['name'])}"
|
||||
field_type = get_csharp_type(field['type'])
|
||||
|
||||
# 기본값 결정
|
||||
if field_type == 'int':
|
||||
# 리스트 여부에 따른 기본값 처리
|
||||
if "List<" in field_type:
|
||||
default_value = f"new {field_type}()"
|
||||
elif field_type == 'int':
|
||||
default_value = '0'
|
||||
elif field_type == 'float':
|
||||
default_value = '0f'
|
||||
@@ -166,7 +139,6 @@ def generate_class(schema_name, schema):
|
||||
|
||||
lines.append(f" public {field_type} {method_name}() => {field_name} ?? {default_value};")
|
||||
|
||||
# 클래스 종료
|
||||
lines.append(" }")
|
||||
lines.append("}")
|
||||
|
||||
@@ -177,15 +149,18 @@ def add_field(lines, field, nullable=False):
|
||||
"""필드 정의 추가"""
|
||||
field_name = to_camel_case(field['name'])
|
||||
field_type = get_csharp_type(field['type'])
|
||||
|
||||
# 주석 (설명이 있으면)
|
||||
is_list = "List<" in field_type
|
||||
|
||||
if field.get('description'):
|
||||
# 줄바꿈을 공백으로 변환 (C# 주석은 한 줄)
|
||||
description = field['description'].replace('\\n', ' ').replace('\n', ' ')
|
||||
lines.append(f" /// <summary>{description}</summary>")
|
||||
|
||||
# 필드 선언
|
||||
if nullable:
|
||||
if is_list:
|
||||
# 리스트는 인스펙터에서 바로 보이도록 초기화
|
||||
lines.append(f" public {field_type} {field_name} = new {field_type}();")
|
||||
elif nullable and field_type not in ['string']:
|
||||
# string은 원래 nullable이므로 제외, 값 타입만 ? 적용
|
||||
lines.append(f" public {field_type}? {field_name};")
|
||||
else:
|
||||
lines.append(f" public {field_type} {field_name};")
|
||||
@@ -193,65 +168,34 @@ def add_field(lines, field, nullable=False):
|
||||
|
||||
def generate_all_classes():
|
||||
"""모든 스키마에 대해 C# 클래스 생성"""
|
||||
|
||||
# 출력 폴더 생성
|
||||
UNITY_OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
# 스키마 파일 찾기
|
||||
schema_files = list(GAMEDATA_DIR.glob(".*_schema.json"))
|
||||
|
||||
if not schema_files:
|
||||
print("⚠️ 스키마 파일을 찾을 수 없습니다.")
|
||||
print("💡 먼저 sync_from_notion.py를 실행하세요.")
|
||||
print("⚠️ 스키마 파일을 찾을 수 없습니다.")
|
||||
return
|
||||
|
||||
print("=" * 60)
|
||||
print("🔧 C# 클래스 자동 생성")
|
||||
print("🔧 C# 클래스 자동 생성 (List 지원)")
|
||||
print("=" * 60)
|
||||
print()
|
||||
|
||||
generated_count = 0
|
||||
|
||||
for schema_file in schema_files:
|
||||
# ".Towers_schema.json" → "Towers"
|
||||
schema_name = schema_file.stem.replace("_schema", "").lstrip(".")
|
||||
|
||||
print(f"📋 {schema_name} 처리 중...")
|
||||
|
||||
# 스키마 로드
|
||||
try:
|
||||
with open(schema_file, 'r', encoding='utf-8') as f:
|
||||
schema = json.load(f)
|
||||
except Exception as e:
|
||||
print(f" ❌ 스키마 로드 실패: {e}")
|
||||
continue
|
||||
|
||||
# C# 코드 생성
|
||||
try:
|
||||
csharp_code = generate_class(schema_name, schema)
|
||||
except Exception as e:
|
||||
print(f" ❌ 코드 생성 실패: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
continue
|
||||
|
||||
# 파일 저장
|
||||
output_file = UNITY_OUTPUT_DIR / f"{schema_name}Data.cs"
|
||||
try:
|
||||
output_file = UNITY_OUTPUT_DIR / f"{schema_name}Data.cs"
|
||||
with open(output_file, 'w', encoding='utf-8') as f:
|
||||
f.write(csharp_code)
|
||||
|
||||
print(f" ✅ 생성: {schema_name}Data.cs ({len(schema)}개 필드)")
|
||||
print(f" ✅ 생성: {schema_name}Data.cs")
|
||||
generated_count += 1
|
||||
|
||||
except Exception as e:
|
||||
print(f" ❌ 파일 저장 실패: {e}")
|
||||
print(f" ❌ {schema_name} 실패: {e}")
|
||||
|
||||
print()
|
||||
print("=" * 60)
|
||||
print(f"🎉 완료: {generated_count}개 클래스 생성")
|
||||
print(f"📂 위치: {UNITY_OUTPUT_DIR}")
|
||||
print("=" * 60)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
@@ -1,9 +1,8 @@
|
||||
# DataTools/sync_from_notion.py
|
||||
"""노션 스키마 → CSV 동기화 (자동 발견)"""
|
||||
|
||||
import os
|
||||
import re
|
||||
import json
|
||||
import pandas as pd
|
||||
import shutil
|
||||
from notion_client import Client
|
||||
from pathlib import Path
|
||||
from datetime import datetime
|
||||
@@ -11,8 +10,6 @@ from datetime import datetime
|
||||
# ===== 설정 =====
|
||||
NOTION_API_KEY = "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"
|
||||
notion = Client(auth=NOTION_API_KEY) if NOTION_API_KEY else None
|
||||
|
||||
# ⭐ 부모 페이지 ID만 설정 (1회)
|
||||
SCHEMA_PARENT_PAGE_ID = "2f494d45b1a3818fa9fceb4f9e17d905"
|
||||
|
||||
SCRIPT_DIR = Path(__file__).parent
|
||||
@@ -20,456 +17,162 @@ GAMEDATA_DIR = SCRIPT_DIR.parent / "GameData"
|
||||
BACKUP_DIR = GAMEDATA_DIR / "Backups"
|
||||
BACKUP_DIR.mkdir(exist_ok=True)
|
||||
|
||||
|
||||
# ===== 유틸리티 함수 =====
|
||||
|
||||
def clean_page_title(title):
|
||||
"""
|
||||
노션 페이지 제목 → 스키마 이름 변환
|
||||
|
||||
규칙:
|
||||
1. 이모지, 특수문자 제거
|
||||
2. "스키마", "Schema" 제거
|
||||
3. 공백 제거
|
||||
|
||||
예시:
|
||||
- "🏰 타워 스키마" → "타워"
|
||||
- "Tower Schema" → "Tower"
|
||||
- "Enemies" → "Enemies"
|
||||
"""
|
||||
# 1. 이모지 및 특수문자 제거
|
||||
cleaned = re.sub(r'[^\w\s가-힣]', '', title).strip()
|
||||
|
||||
# 2. "스키마", "Schema" 제거
|
||||
cleaned = re.sub(r'\s*(스키마|Schema)\s*', '', cleaned, flags=re.IGNORECASE).strip()
|
||||
|
||||
# 3. 공백 제거
|
||||
cleaned = cleaned.replace(' ', '')
|
||||
|
||||
# 4. 비어있으면 원본 반환
|
||||
if not cleaned:
|
||||
return title.replace(' ', '')
|
||||
|
||||
return cleaned
|
||||
return cleaned if cleaned else title.replace(' ', '')
|
||||
|
||||
def check_page_has_table(page_id):
|
||||
try:
|
||||
blocks_response = notion.blocks.children.list(block_id=page_id)
|
||||
return any(block.get('type') == 'table' for block in blocks_response.get('results', []))
|
||||
except:
|
||||
return False
|
||||
|
||||
def discover_schema_pages(parent_id=None, depth=0, max_depth=3):
|
||||
"""
|
||||
부모 페이지의 하위 페이지들을 재귀적으로 탐색하여 스키마 발견
|
||||
|
||||
Args:
|
||||
parent_id: 탐색할 부모 페이지 ID (None이면 SCHEMA_PARENT_PAGE_ID 사용)
|
||||
depth: 현재 깊이 (0부터 시작)
|
||||
max_depth: 최대 탐색 깊이 (기본 3단계)
|
||||
|
||||
반환:
|
||||
{
|
||||
"타워": "page_id_1",
|
||||
"적유닛": "page_id_2",
|
||||
...
|
||||
}
|
||||
"""
|
||||
if not notion:
|
||||
raise ValueError("Notion API 클라이언트가 초기화되지 않았습니다.")
|
||||
|
||||
if parent_id is None:
|
||||
if not SCHEMA_PARENT_PAGE_ID or SCHEMA_PARENT_PAGE_ID == "노션_데이터_스키마_정의_페이지_ID":
|
||||
raise ValueError(
|
||||
"SCHEMA_PARENT_PAGE_ID가 설정되지 않았습니다.\n"
|
||||
"sync_from_notion.py 파일에서 부모 페이지 ID를 설정하세요."
|
||||
)
|
||||
parent_id = SCHEMA_PARENT_PAGE_ID
|
||||
|
||||
# 최대 깊이 체크
|
||||
if depth > max_depth:
|
||||
return {}
|
||||
if not notion: raise ValueError("API Key missing")
|
||||
parent_id = parent_id or SCHEMA_PARENT_PAGE_ID
|
||||
if depth > max_depth: return {}
|
||||
|
||||
indent = " " * depth
|
||||
|
||||
if depth == 0:
|
||||
print("🔍 스키마 페이지 자동 발견 중...")
|
||||
if depth == 0: print("🔍 스키마 페이지 자동 발견 중...")
|
||||
|
||||
try:
|
||||
# 부모 페이지의 자식 블록 가져오기
|
||||
children = notion.blocks.children.list(block_id=parent_id)
|
||||
|
||||
schemas = {}
|
||||
|
||||
for block in children['results']:
|
||||
if block['type'] == 'child_page':
|
||||
page_id = block['id']
|
||||
page_title = block['child_page']['title']
|
||||
|
||||
# 제목 정리
|
||||
schema_name = clean_page_title(page_title)
|
||||
|
||||
print(f"{indent}📋 발견: '{page_title}'", end="")
|
||||
|
||||
# 이 페이지에 테이블이 있는지 확인
|
||||
has_table = check_page_has_table(page_id)
|
||||
|
||||
if has_table:
|
||||
# 테이블이 있으면 스키마로 등록
|
||||
if check_page_has_table(page_id):
|
||||
schemas[schema_name] = page_id
|
||||
print(f" → {schema_name} ✅")
|
||||
else:
|
||||
# 테이블이 없으면 하위 페이지 탐색
|
||||
print(f" (폴더)")
|
||||
child_schemas = discover_schema_pages(page_id, depth + 1, max_depth)
|
||||
schemas.update(child_schemas)
|
||||
|
||||
if depth == 0 and not schemas:
|
||||
print(" ⚠️ 하위 페이지를 찾을 수 없습니다.")
|
||||
print(f" 💡 노션에서 부모 페이지 하위에 스키마 페이지를 추가하세요.")
|
||||
|
||||
schemas.update(discover_schema_pages(page_id, depth + 1, max_depth))
|
||||
return schemas
|
||||
|
||||
except Exception as e:
|
||||
print(f"{indent}❌ 탐색 실패: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
return {}
|
||||
|
||||
|
||||
def check_page_has_table(page_id):
|
||||
"""
|
||||
페이지에 테이블 블록이 있는지 확인
|
||||
|
||||
Args:
|
||||
page_id: 확인할 페이지 ID
|
||||
|
||||
반환:
|
||||
True: 테이블 있음
|
||||
False: 테이블 없음
|
||||
"""
|
||||
try:
|
||||
blocks_response = notion.blocks.children.list(block_id=page_id)
|
||||
blocks = blocks_response.get('results', [])
|
||||
|
||||
for block in blocks:
|
||||
if block.get('type') == 'table':
|
||||
return True
|
||||
|
||||
return False
|
||||
|
||||
except Exception as e:
|
||||
# 에러 발생 시 테이블 없음으로 간주
|
||||
return False
|
||||
|
||||
def parse_condition(condition_str):
|
||||
"""
|
||||
사용 조건 파싱
|
||||
|
||||
빈 문자열 → None (항상 사용)
|
||||
"tower_type=attack" → {'field': 'tower_type', 'op': '=', 'value': 'attack'}
|
||||
"""
|
||||
if not condition_str or condition_str.strip() == "":
|
||||
return None
|
||||
|
||||
condition_str = condition_str.strip()
|
||||
|
||||
# 단순 조건: "tower_type=attack"
|
||||
match = re.match(r'(\w+)\s*(=|!=|>|<|>=|<=)\s*(.+)', condition_str)
|
||||
if not condition_str or condition_str.strip() == "": return None
|
||||
match = re.match(r'(\w+)\s*(=|!=|>|<|>=|<=)\s*(.+)', condition_str.strip())
|
||||
if match:
|
||||
return {
|
||||
'field': match.group(1),
|
||||
'op': match.group(2),
|
||||
'value': match.group(3).strip()
|
||||
}
|
||||
|
||||
return {'field': match.group(1), 'op': match.group(2), 'value': match.group(3).strip()}
|
||||
return None
|
||||
|
||||
|
||||
def parse_notion_table(page_id):
|
||||
"""노션 테이블 파싱"""
|
||||
blocks = notion.blocks.children.list(block_id=page_id).get('results', [])
|
||||
table_block = next((b for b in blocks if b['type'] == 'table'), None)
|
||||
if not table_block: raise ValueError("테이블을 찾을 수 없습니다.")
|
||||
|
||||
if not notion:
|
||||
raise ValueError("Notion API 클라이언트가 초기화되지 않았습니다.")
|
||||
|
||||
try:
|
||||
# 1. 블록 가져오기
|
||||
blocks_response = notion.blocks.children.list(block_id=page_id)
|
||||
blocks = blocks_response.get('results', [])
|
||||
|
||||
# 2. 테이블 찾기
|
||||
table_block = None
|
||||
for block in blocks:
|
||||
if block.get('type') == 'table':
|
||||
table_block = block
|
||||
break
|
||||
|
||||
if not table_block:
|
||||
raise ValueError(f"테이블을 찾을 수 없습니다.")
|
||||
|
||||
print(f" 📋 테이블 발견")
|
||||
|
||||
# 3. 행 가져오기
|
||||
table_id = table_block['id']
|
||||
rows_response = notion.blocks.children.list(block_id=table_id)
|
||||
rows = rows_response.get('results', [])
|
||||
|
||||
if len(rows) < 2:
|
||||
raise ValueError("테이블에 데이터가 없습니다.")
|
||||
|
||||
# 4. 파싱
|
||||
schema = []
|
||||
|
||||
def extract_text(cell, preserve_newlines=False):
|
||||
"""
|
||||
셀에서 텍스트 추출
|
||||
|
||||
Args:
|
||||
cell: 노션 셀 데이터
|
||||
preserve_newlines: True면 줄바꿈 보존, False면 공백으로 변환
|
||||
"""
|
||||
if not cell or len(cell) == 0:
|
||||
return ""
|
||||
|
||||
text_parts = []
|
||||
for content in cell:
|
||||
if content.get('type') == 'text':
|
||||
text_content = content.get('text', {}).get('content', '')
|
||||
text_parts.append(text_content)
|
||||
|
||||
if preserve_newlines:
|
||||
# 줄바꿈 보존 (\\n으로 이스케이프)
|
||||
result = ''.join(text_parts)
|
||||
# CSV에서 안전하게 저장하기 위해 실제 줄바꿈을 \\n으로 변환
|
||||
result = result.replace('\n', '\\n')
|
||||
return result.strip()
|
||||
else:
|
||||
# 줄바꿈을 공백으로 변환
|
||||
return ''.join(text_parts).strip()
|
||||
|
||||
for row_idx, row in enumerate(rows[1:], start=2):
|
||||
if row.get('type') != 'table_row':
|
||||
continue
|
||||
|
||||
cells = row['table_row']['cells']
|
||||
|
||||
# 4개 컬럼: 필드명, 타입, 사용 조건, 설명
|
||||
field_name = extract_text(cells[0]) if len(cells) > 0 else ""
|
||||
field_type = extract_text(cells[1]) if len(cells) > 1 else "string"
|
||||
condition_str = extract_text(cells[2]) if len(cells) > 2 else ""
|
||||
# ⭐ 설명 컬럼만 줄바꿈 보존
|
||||
description = extract_text(cells[3], preserve_newlines=True) if len(cells) > 3 else ""
|
||||
|
||||
if not field_name:
|
||||
continue
|
||||
|
||||
# 조건 파싱
|
||||
condition = parse_condition(condition_str)
|
||||
|
||||
if condition:
|
||||
print(f" 📌 {field_name}: {condition['field']}={condition['value']}일 때 사용")
|
||||
|
||||
schema.append({
|
||||
'name': field_name,
|
||||
'type': field_type.lower(),
|
||||
'condition': condition,
|
||||
'description': description
|
||||
})
|
||||
|
||||
if len(schema) == 0:
|
||||
raise ValueError("파싱된 스키마가 비어있습니다.")
|
||||
|
||||
return schema
|
||||
|
||||
except Exception as e:
|
||||
print(f" ❌ 파싱 오류: {e}")
|
||||
raise
|
||||
rows = notion.blocks.children.list(block_id=table_block['id']).get('results', [])
|
||||
schema = []
|
||||
|
||||
def extract_text(cell, preserve_newlines=False):
|
||||
if not cell: return ""
|
||||
text = "".join([c.get('text', {}).get('content', '') for c in cell if c.get('type') == 'text'])
|
||||
return text.replace('\n', '\\n').strip() if preserve_newlines else text.strip()
|
||||
|
||||
for row in rows[1:]: # 헤더 제외
|
||||
if row.get('type') != 'table_row': continue
|
||||
cells = row['table_row']['cells']
|
||||
field_name = extract_text(cells[0])
|
||||
if not field_name: continue
|
||||
|
||||
# 필드 파싱
|
||||
field_type = extract_text(cells[1]).lower()
|
||||
condition_str = extract_text(cells[2])
|
||||
description = extract_text(cells[3], preserve_newlines=True) if len(cells) > 3 else ""
|
||||
|
||||
schema.append({
|
||||
'name': field_name,
|
||||
'type': field_type, # "list:int" 형태 그대로 보존
|
||||
'condition': parse_condition(condition_str),
|
||||
'description': description
|
||||
})
|
||||
return schema
|
||||
|
||||
def get_default_value(field_type, has_condition):
|
||||
"""
|
||||
기본값 결정
|
||||
"""기본값 결정 (List 대응)"""
|
||||
if has_condition: return None
|
||||
|
||||
조건부 필드 → None (빈 칸)
|
||||
공통 필드 → 타입별 기본값
|
||||
"""
|
||||
# 조건부 필드는 빈 칸
|
||||
if has_condition:
|
||||
return None
|
||||
|
||||
# 공통 필드는 타입별 기본값
|
||||
if field_type == "int":
|
||||
return 0
|
||||
elif field_type in ["float", "number"]:
|
||||
return 0.0
|
||||
elif field_type in ["bool", "boolean"]:
|
||||
return False
|
||||
elif field_type == "string":
|
||||
return ""
|
||||
else:
|
||||
return None
|
||||
|
||||
f_type = field_type.lower()
|
||||
if 'list' in f_type: return [] # 리스트는 빈 리스트 객체
|
||||
if f_type == "int": return 0
|
||||
if f_type in ["float", "number"]: return 0.0
|
||||
if f_type in ["bool", "boolean"]: return False
|
||||
return ""
|
||||
|
||||
def merge_schema_and_data(schema, existing_data):
|
||||
"""스키마와 데이터 병합"""
|
||||
|
||||
schema_columns = [f['name'] for f in schema]
|
||||
|
||||
if existing_data is None or existing_data.empty:
|
||||
print(" 새 파일 생성")
|
||||
example_row = {}
|
||||
for field in schema:
|
||||
has_condition = field.get('condition') is not None
|
||||
example_row[field['name']] = get_default_value(field['type'], has_condition)
|
||||
example_row = {f['name']: get_default_value(f['type'], f.get('condition') is not None) for f in schema}
|
||||
# CSV 저장을 위해 리스트는 문자열로 변환 (빈 값)
|
||||
for k, v in example_row.items():
|
||||
if isinstance(v, list): example_row[k] = ""
|
||||
return pd.DataFrame([example_row])
|
||||
|
||||
print(f" 기존 데이터: {len(existing_data)}행")
|
||||
|
||||
new_df = pd.DataFrame()
|
||||
|
||||
for field in schema:
|
||||
col_name = field['name']
|
||||
if col_name in existing_data.columns:
|
||||
print(f" ✓ {col_name}: 유지")
|
||||
new_df[col_name] = existing_data[col_name]
|
||||
col = field['name']
|
||||
if col in existing_data.columns:
|
||||
new_df[col] = existing_data[col]
|
||||
else:
|
||||
has_condition = field.get('condition') is not None
|
||||
default_val = get_default_value(field['type'], has_condition)
|
||||
|
||||
if default_val is None:
|
||||
print(f" + {col_name}: 추가 (조건부 필드, 빈 칸)")
|
||||
else:
|
||||
print(f" + {col_name}: 추가 (기본값: {default_val})")
|
||||
|
||||
new_df[col_name] = default_val
|
||||
|
||||
val = get_default_value(field['type'], field.get('condition') is not None)
|
||||
new_df[col] = "" if isinstance(val, list) else val
|
||||
return new_df
|
||||
|
||||
|
||||
def sync_single_schema(data_name, page_id):
|
||||
"""단일 스키마 동기화 (CSV 버전)"""
|
||||
print(f"\n{'='*60}")
|
||||
print(f"📋 {data_name} 동기화")
|
||||
print(f"{'='*60}")
|
||||
|
||||
print(f"\n🔄 {data_name} 동기화 시작...")
|
||||
try:
|
||||
# 1. 스키마 읽기
|
||||
print("1️⃣ 스키마 읽기...")
|
||||
schema = parse_notion_table(page_id)
|
||||
print(f" ✅ {len(schema)}개 필드")
|
||||
|
||||
# 2. 스키마를 JSON으로 저장 (검증용)
|
||||
import json
|
||||
schema_json_path = GAMEDATA_DIR / f".{data_name}_schema.json"
|
||||
with open(schema_json_path, 'w', encoding='utf-8') as f:
|
||||
# 1. 스키마 JSON 저장 (generate_all_classes.py가 읽을 파일)
|
||||
schema_path = GAMEDATA_DIR / f".{data_name}_schema.json"
|
||||
with open(schema_path, 'w', encoding='utf-8') as f:
|
||||
json.dump(schema, f, ensure_ascii=False, indent=2)
|
||||
print(f" 💾 스키마 저장: {schema_json_path.name}")
|
||||
|
||||
# 3. 기존 파일 확인
|
||||
# 2. CSV 업데이트
|
||||
csv_path = GAMEDATA_DIR / f"{data_name}.csv"
|
||||
print(f"\n2️⃣ 기존 파일: {csv_path}")
|
||||
existing_data = pd.read_csv(csv_path) if csv_path.exists() else None
|
||||
|
||||
existing_data = None
|
||||
if csv_path.exists():
|
||||
# 백업
|
||||
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
||||
backup_path = BACKUP_DIR / f"{data_name}_{timestamp}.csv"
|
||||
import shutil
|
||||
shutil.copy2(csv_path, backup_path)
|
||||
print(f" 💾 백업: {backup_path.name}")
|
||||
if existing_data is not None:
|
||||
shutil.copy2(csv_path, BACKUP_DIR / f"{data_name}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv")
|
||||
|
||||
existing_data = pd.read_csv(csv_path)
|
||||
|
||||
# 4. 병합
|
||||
print(f"\n3️⃣ 병합 중...")
|
||||
merged_df = merge_schema_and_data(schema, existing_data)
|
||||
|
||||
# 5. 저장 (CSV)
|
||||
print(f"\n4️⃣ 저장...")
|
||||
merged_df.to_csv(csv_path, index=False, encoding='utf-8-sig')
|
||||
print(f" ✅ 완료: {csv_path}")
|
||||
|
||||
print(f" ✅ 완료: {data_name} (스키마 및 CSV 업데이트)")
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
print(f"\n❌ 오류: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
print(f" ❌ 실패: {e}")
|
||||
return False
|
||||
|
||||
|
||||
def main():
|
||||
print("=" * 60)
|
||||
print("🔄 Notion → CSV 동기화 (자동 발견)")
|
||||
print("=" * 60)
|
||||
print()
|
||||
|
||||
if not NOTION_API_KEY:
|
||||
print("❌ NOTION_API_KEY 환경변수가 없습니다")
|
||||
print("💡 설정 방법:")
|
||||
print(' $env:NOTION_API_KEY = "your_key"')
|
||||
return
|
||||
|
||||
print(f"📂 데이터 폴더: {GAMEDATA_DIR}")
|
||||
print(f"💾 백업 폴더: {BACKUP_DIR}")
|
||||
print()
|
||||
|
||||
# ⭐ 스키마 자동 발견
|
||||
print("🚀 Notion Schema Sync Start")
|
||||
try:
|
||||
SCHEMA_PAGE_IDS = discover_schema_pages()
|
||||
page_ids = discover_schema_pages()
|
||||
if not page_ids: return
|
||||
|
||||
schemas = list(page_ids.keys())
|
||||
for idx, name in enumerate(schemas, 1):
|
||||
print(f" {idx}. {name}")
|
||||
print(f" {len(schemas) + 1}. 전체")
|
||||
|
||||
choice = input("\n번호 선택: ").strip()
|
||||
selected = schemas if choice == str(len(schemas) + 1) else [schemas[int(choice)-1]]
|
||||
|
||||
for name in selected:
|
||||
sync_single_schema(name, page_ids[name])
|
||||
|
||||
except Exception as e:
|
||||
print(f"\n❌ 스키마 발견 실패: {e}")
|
||||
return
|
||||
|
||||
if not SCHEMA_PAGE_IDS:
|
||||
print("\n❌ 발견된 스키마 페이지가 없습니다.")
|
||||
return
|
||||
|
||||
print()
|
||||
print("=" * 60)
|
||||
print("동기화할 스키마를 선택하세요:")
|
||||
|
||||
schemas = list(SCHEMA_PAGE_IDS.keys())
|
||||
for idx, name in enumerate(schemas, 1):
|
||||
print(f" {idx}. {name}")
|
||||
print(f" {len(schemas) + 1}. 전체")
|
||||
print()
|
||||
|
||||
try:
|
||||
choice = input("선택 (번호 입력): ").strip()
|
||||
|
||||
if choice == str(len(schemas) + 1):
|
||||
selected = schemas
|
||||
else:
|
||||
idx = int(choice) - 1
|
||||
if 0 <= idx < len(schemas):
|
||||
selected = [schemas[idx]]
|
||||
else:
|
||||
print("❌ 잘못된 선택입니다.")
|
||||
return
|
||||
|
||||
except (ValueError, KeyboardInterrupt):
|
||||
print("\n⚠️ 취소되었습니다.")
|
||||
return
|
||||
|
||||
# 동기화 실행
|
||||
print()
|
||||
success_count = 0
|
||||
|
||||
for schema_name in selected:
|
||||
page_id = SCHEMA_PAGE_IDS[schema_name]
|
||||
|
||||
if sync_single_schema(schema_name, page_id):
|
||||
success_count += 1
|
||||
|
||||
# 최종 결과
|
||||
print()
|
||||
print("=" * 60)
|
||||
print(f"✅ 완료: {success_count}/{len(selected)} 성공")
|
||||
print("=" * 60)
|
||||
|
||||
if success_count > 0:
|
||||
print()
|
||||
print("💡 다음 단계:")
|
||||
print(" 1. GameData 폴더에서 CSV 파일 확인")
|
||||
print(" 2. 데이터 수정")
|
||||
print(" 3. Git 커밋:")
|
||||
print(" git add GameData/*.csv")
|
||||
print(' git commit -m "Update data from Notion"')
|
||||
|
||||
print(f"오류 발생: {e}")
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user