Pickle to JSON 변환

너는 파이썬 프로그래밍 전문가야. python pickle로 저장된 데이터가 있어. 이걸 읽어서 그 안의 모든 객체 tree를 json 형태로 뱉도록 하고 싶어. 이 때 pickle은 커스텀 클래스를 담고 있어. 이걸 해내기 위해서 어떻게 접근해야할까? 최대한 에러가 없이 동작하도록 고민해줘. Step by Step으로 생각해줘.

좋아, “피클 → 안전하고 튼튼한 JSON”을 목표로 단계별로 설계해볼게. (코드 샘플도 아래에 한 방에 넣어뒀어.)

Step by Step 접근

1) 위협 모델/전략 먼저 결정

• pickle은 임의 코드 실행 위험 이 있어. 출처가 100% 신뢰되는 파일만 바로 pickle.load 하면 되고, 그 외에는 “제한 해제”를 절대 하지 말아야 해.

• 따라서 두 가지 모드를 추천:

  1. 신뢰 모드(trusted) : 그냥 로드 → 직렬화.

  2. 세이프 모드(safe/whitelist) : 허용 목록(모듈.클래스)을 지정한 RestrictedUnpickler로만 언피클.

2) 미리 “무엇이 들어있는지” 검사

• 언피클을 실행하지 않고 pickletools.genops로 피클 스트림의 GLOBAL/STACK_GLOBAL opcode를 스캔해서, 필요한 클래스/모듈 목록을 뽑아본다.

• 그 결과를 보고 화이트리스트 를 만든다. (예: {"mymodels.User", "numpy.ndarray"})

3) 안전한 로더 만들기

• pickle.Unpickler를 상속해서 find_class를 오버라이드.

• builtins의 기본 컨테이너 정도만 허용하고, 나머지는 화이트리스트에 있는 것만 통과.

• 화이트리스트에 없는 건 UnpicklingError로 막는다.

4) “JSON 가능 형태”로 변환기 설계

JSON은 다음을 그대로 표현 못 하거나 깨지기 쉬워:

• 순환/공유 참조(cycles/aliases), 사용자 정의 객체, set/tuple, bytes, datetime, Decimal/UUID, numpy/pandas/torch 등.

• 원칙 : “가능하면 손실 없이”, “그래도 항상 직렬화되게”.

  • 순환/공유 참조 : "$id"/"$ref" 프로토콜을 사용 (예: {"$id":"obj1", "value": ...}, 참조는 {"$ref":"obj1"}).

  • 커스텀 클래스 : 우선 순위대로

    1. __json__() → 딕트/리스트 등

    2. to_json() / to_dict() / model_dump()(pydantic)

    3. dataclass → asdict

    4. __getstate__() or __dict__

    5. 마지막 수단: repr(obj)에 타입 메타("$type").

  • 특수 타입들 (가능한 한 보존):

    • set/frozenset → {"$type":"set","items":[...]}

    • tuple → {"$type":"tuple","items":[...]}

    • bytes/bytearray → {"$type":"bytes","base64":"..."}

    • datetime/date/time → {"$type":"datetime","iso":"..."}

    • Decimal → {"$type":"decimal","value":"..."}

    • UUID → {"$type":"uuid","value":"..."}

    • Path → {"$type":"path","value":"..."}

    • re.Pattern → 패턴/플래그 보존

    • Enum → 이름/값 보존

    • Exception → 타입/args/str

    • NumPy : ndarray는 .tolist() + shape/dtype 메타

    • pandas : DataFrame/Series는 to_dict(orient="records") 등 + 메타

    • PyTorch/TensorFlow 텐서는 .detach().cpu().numpy().tolist() 등으로 값만 (가능하면 dtype/shape 메타 포함)

  • dict의 비문자열 키 : JSON은 키가 문자열만 되므로

    • 전부 문자열이면 그대로 유지

    • 아니면 {"$type":"dict","entries":[[key,value], ...]} 형태로 보존

  • 깊이 제한/메모 : 무한 재귀 방지 위해 seen(id→$id) 메모와 max_depth 도입.

  • 마지막 안전판 : 무엇이든 실패하면 repr + "$error" 메타를 달아 절대 실패하지 않게.

5) 덤프 옵션

• json.dumps(jsonable, ensure_ascii=False, indent=2, allow_nan=False)를 권장.

• 파일이 크면 indent=None로 용량 최소화.

• 덤프 후 한 번 json.loads로 역파싱 검증 하면 실수 줄일 수 있어.

6) 대안: 라이브러리 사용

• jsonpickle 은 자체적으로 클래스/참조를 잘 처리한다. 빠르게 되고 범용성이 좋지만, 언피클 단계의 보안은 별도 처리 가 필요하다(즉, RestrictedUnpickler는 여전히 필요).

• 직접 규칙을 통제하고 싶으면 아래 커스텀 직렬화기를 쓰는 게 안전/예측 가능.

---

견고한 예시 코드 (스크립트 1개로 동작)

#!/usr/bin/env python3
# pickle_to_json.py
import io, json, base64, math, re, types, sys
import pickle, pickletools, importlib
from dataclasses import is_dataclass, asdict
from decimal import Decimal
from uuid import UUID
from pathlib import Path
from collections import OrderedDict
from typing import Any

# --- Optional dependencies detection ---
try:
    import numpy as np
except Exception:
    np = None

try:
    import pandas as pd
except Exception:
    pd = None

# ---------- 1) 분석: 피클 안의 글로벌 심볼 스캔 ----------
def audit_pickle_globals(pkl_bytes: bytes):
    """언피클 없이 GLOBAL/STACK_GLOBAL 사용 내역을 (module.name) 집합으로 리턴."""
    refs = set()
    for op, arg, pos in pickletools.genops(pkl_bytes):
        if op.name in ("GLOBAL", "STACK_GLOBAL"):
            if isinstance(arg, tuple):
                mod, name = arg
            else:
                # For GLOBAL, arg like "module\nname\n"
                parts = str(arg).split("\n")
                mod, name = parts[0], parts[1]
            refs.add(f"{mod}.{name}")
    return refs

# ---------- 2) 화이트리스트 기반 Restricted Unpickler ----------
class RestrictedUnpickler(pickle.Unpickler):
    def __init__(self, file, allowed: set[str] | None = None):
        super().__init__(file)
        self.allowed = set(allowed or [])

    def find_class(self, module: str, name: str):
        full = f"{module}.{name}"
        # 빌트인 컨테이너/예외/기본 타입 일부 허용
        allowed_builtins = {
            "builtins.set", "builtins.frozenset",
            "builtins.bytes", "builtins.bytearray",
            "builtins.complex", "builtins.Exception",
            "builtins.slice", "builtins.range",
            "builtins.dict", "builtins.list", "builtins.tuple",
        }
        if full in allowed_builtins or full in self.allowed:
            mod = importlib.import_module(module)
            return getattr(mod, name)
        raise pickle.UnpicklingError(f"forbidden global: {full}")

def restricted_loads(pkl_bytes: bytes, allowed: set[str] | None = None):
    return RestrictedUnpickler(io.BytesIO(pkl_bytes), allowed=allowed).load()

# ---------- 3) JSON 직렬화기 ----------
JSON_PRIMITIVES = (type(None), bool, int, float, str)

def is_primitive(x: Any) -> bool:
    if isinstance(x, JSON_PRIMITIVES):
        # NaN/Inf는 allow_nan=False에 막히므로 문자열 처리
        if isinstance(x, float) and (math.isnan(x) or math.isinf(x)):
            return False
        # 너무 큰 int는 JS에서 안전하지 않지만 JSON 스펙엔 문제 없음. 필요시 문자열화 가능.
        return True
    return False

def _pattern_type():
    # Py>=3.8: re.Pattern, Py<3.8: type(re.compile(''))
    pat = getattr(re, "Pattern", type(re.compile("")))
    return pat

def to_jsonable(obj: Any, *, seen: dict[int, str] | None = None, max_depth: int = 1000, _depth: int = 0):
    """객체 그래프를 JSON 가능 구조로 변환(+ 순환/참조 보존). 절대 예외 던지지 않음."""
    if seen is None:
        seen = {}

    # 깊이 보호
    if _depth > max_depth:
        return {"$type": "depth_truncated"}

    # 기본형
    if is_primitive(obj):
        return obj

    oid = id(obj)
    # 불변이지만 큰 튜플/프리미티브 아닌 객체에도 id를 부여해 참조 보존
    if oid in seen:
        return {"$ref": seen[oid]}

    # 새 객체에 $id 할당
    sid = f"obj{len(seen) + 1}"
    seen[oid] = sid

    # bytes/bytearray
    if isinstance(obj, (bytes, bytearray, memoryview)):
        b = bytes(obj)
        return {"$id": sid, "$type": "bytes", "base64": base64.b64encode(b).decode("ascii"), "length": len(b)}

    # datetime-like
    import datetime as _dt
    if isinstance(obj, (_dt.datetime, _dt.date, _dt.time)):
        try:
            iso = obj.isoformat()
        except Exception:
            iso = str(obj)
        return {"$id": sid, "$type": obj.__class__.__name__.lower(), "iso": iso}

    # Decimal / UUID / Path
    if isinstance(obj, Decimal):
        return {"$id": sid, "$type": "decimal", "value": str(obj)}
    if isinstance(obj, UUID):
        return {"$id": sid, "$type": "uuid", "value": str(obj)}
    if isinstance(obj, Path):
        return {"$id": sid, "$type": "path", "value": str(obj)}

    # 정규식 패턴
    if isinstance(obj, _pattern_type()):
        return {"$id": sid, "$type": "regex", "pattern": obj.pattern, "flags": obj.flags}

    # Enum
    import enum as _enum
    if isinstance(obj, _enum.Enum):
        return {"$id": sid, "$type": f"enum:{obj.__class__.__name__}", "name": obj.name, "value": obj.value}

    # 예외
    if isinstance(obj, BaseException):
        return {
            "$id": sid, "$type": f"exception:{obj.__class__.__name__}",
            "args": [to_jsonable(a, seen=seen, max_depth=max_depth, _depth=_depth+1) for a in obj.args],
            "str": str(obj),
        }

    # NumPy
    if np is not None:
        if isinstance(obj, np.ndarray):
            return {
                "$id": sid, "$type": "numpy.ndarray",
                "shape": obj.shape, "dtype": str(obj.dtype),
                "data": obj.tolist(),
            }
        if isinstance(obj, np.generic):  # numpy scalar
            return obj.item()

    # pandas
    if pd is not None:
        if isinstance(obj, pd.DataFrame):
            return {
                "$id": sid, "$type": "pandas.DataFrame",
                "columns": obj.columns.tolist(),
                "records": obj.to_dict(orient="records"),
            }
        if isinstance(obj, pd.Series):
            return {
                "$id": sid, "$type": "pandas.Series",
                "name": obj.name,
                "values": obj.to_list(),
                "index": obj.index.tolist(),
            }

    # torch / tf 텐서 (가능하면 값만)
    try:
        import torch
        if isinstance(obj, torch.Tensor):
            arr = obj.detach().cpu().numpy()
            return {"$id": sid, "$type": "torch.Tensor", "shape": list(arr.shape), "dtype": str(arr.dtype), "data": arr.tolist()}
    except Exception:
        pass
    try:
        import tensorflow as tf
        if isinstance(obj, (tf.Tensor,)):
            arr = obj.numpy()
            return {"$id": sid, "$type": "tf.Tensor", "shape": list(arr.shape), "dtype": str(arr.dtype), "data": arr.tolist()}
    except Exception:
        pass

    # set / frozenset
    if isinstance(obj, (set, frozenset)):
        return {
            "$id": sid, "$type": obj.__class__.__name__,
            "items": [to_jsonable(x, seen=seen, max_depth=max_depth, _depth=_depth+1) for x in obj],
        }

    # tuple / list
    if isinstance(obj, (list, tuple)):
        items = [to_jsonable(x, seen=seen, max_depth=max_depth, _depth=_depth+1) for x in obj]
        if isinstance(obj, tuple):
            return {"$id": sid, "$type": "tuple", "items": items}
        return items  # list는 흔히 그대로

    # dict / mapping
    if isinstance(obj, dict):
        # 모든 키가 문자열이면 평범한 dict 유지
        if all(isinstance(k, str) for k in obj.keys()):
            return {k: to_jsonable(v, seen=seen, max_depth=max_depth, _depth=_depth+1) for k, v in obj.items()}
        # 아니면 entries로 보존
        return {
            "$id": sid, "$type": "dict",
            "entries": [
                [to_jsonable(k, seen=seen, max_depth=max_depth, _depth=_depth+1),
                 to_jsonable(v, seen=seen, max_depth=max_depth, _depth=_depth+1)]
                for k, v in obj.items()
            ],
        }

    if isinstance(obj, OrderedDict):
        return {
            "$id": sid, "$type": "OrderedDict",
            "entries": [
                [to_jsonable(k, seen=seen, max_depth=max_depth, _depth=_depth+1),
                 to_jsonable(v, seen=seen, max_depth=max_depth, _depth=_depth+1)]
                for k, v in obj.items()
            ],
        }

    # dataclass / attrs / pydantic / 사용자 정의 클래스
    try:
        # dataclass
        if is_dataclass(obj):
            payload = asdict(obj)
            return {"$id": sid, "$type": obj.__class__.__name__, "fields":
                    to_jsonable(payload, seen=seen, max_depth=max_depth, _depth=_depth+1)}
    except Exception:
        pass

    # attrs
    try:
        import attr
        if attr.has(obj.__class__):
            payload = attr.asdict(obj, recurse=False)
            return {"$id": sid, "$type": obj.__class__.__name__, "fields":
                    to_jsonable(payload, seen=seen, max_depth=max_depth, _depth=_depth+1)}
    except Exception:
        pass

    # pydantic
    if hasattr(obj, "model_dump") and callable(getattr(obj, "model_dump")):
        try:
            payload = obj.model_dump()
            return {"$id": sid, "$type": obj.__class__.__name__, "fields":
                    to_jsonable(payload, seen=seen, max_depth=max_depth, _depth=_depth+1)}
        except Exception:
            pass

    # 커스텀 메서드들
    for meth in ("__json__", "to_json", "to_dict"):
        fn = getattr(obj, meth, None)
        if callable(fn):
            try:
                payload = fn()
                return {"$id": sid, "$type": obj.__class__.__name__, "fields":
                        to_jsonable(payload, seen=seen, max_depth=max_depth, _depth=_depth+1)}
            except Exception:
                pass

    # __getstate__ / __dict__
    state = None
    if hasattr(obj, "__getstate__"):
        try:
            state = obj.__getstate__()
        except Exception:
            state = None
    if state is None and hasattr(obj, "__dict__"):
        try:
            state = dict(obj.__dict__)
        except Exception:
            state = None
    if state is not None:
        return {"$id": sid, "$type": obj.__class__.__name__, "state":
                to_jsonable(state, seen=seen, max_depth=max_depth, _depth=_depth+1)}

    # 마지막 안전판
    try:
        rep = repr(obj)
    except Exception:
        rep = f"<unreprable {obj.__class__.__name__}>"
    return {"$id": sid, "$type": obj.__class__.__name__, "repr": rep, "$error": "fallback_repr"}

# ---------- 4) 최종 유틸 ----------
def pickle_to_json_bytes(pkl_bytes: bytes, *, mode: str = "safe", allow: set[str] | None = None,
                         ensure_ascii=False, indent=2) -> bytes:
    if mode not in ("safe", "trusted"):
        raise ValueError("mode must be 'safe' or 'trusted'")

    if mode == "safe":
        # 사전 점검: 어떤 클래스가 필요한지
        needed = audit_pickle_globals(pkl_bytes)
        if allow is not None:
            allow = set(allow)
        else:
            allow = set()
        # numpy/pandas 등 자동 허용 (있을 때)
        if np is not None:
            allow.update({"numpy.ndarray", "numpy.dtype", "numpy.core.numeric.complexfloating"})
        if pd is not None:
            allow.update({"pandas.core.frame.DataFrame", "pandas.core.series.Series", "pandas.core.indexes.base.Index"})
        # 필요하지만 화이트리스트에 없는 것 있으면 경고(여기선 예외로 명확히)
        missing = needed - allow - {
            # 종종 피클 스트림에 등장하는 builtins/eval 보조들 (이미 Restricted에서 기본 거름)
        }
        # 실제로는 사용자에게 리스트를 보여주고 승인받는 게 가장 안전
        # 여기서는 '허용 안 된 심볼이 있다'면 실패 처리
        if missing:
            raise pickle.UnpicklingError(f"Unapproved classes in pickle: {sorted(missing)}")

        obj = restricted_loads(pkl_bytes, allowed=allow)
    else:
        # trusted
        obj = pickle.loads(pkl_bytes)

    jsonable = to_jsonable(obj)
    txt = json.dumps(jsonable, ensure_ascii=ensure_ascii, indent=indent, allow_nan=False)
    # 역파싱 검증
    json.loads(txt)
    return txt.encode("utf-8")

# ---------- 5) CLI ----------
def main():
    import argparse, os
    ap = argparse.ArgumentParser()
    ap.add_argument("input", help="input .pkl file")
    ap.add_argument("output", help="output .json file")
    ap.add_argument("--mode", choices=["safe", "trusted"], default="safe")
    ap.add_argument("--allow", action="append", default=[], help="extra allowed 'module.Class' (repeatable)")
    ap.add_argument("--no-indent", action="store_true")
    args = ap.parse_args()

    with open(args.input, "rb") as f:
        pkl = f.read()

    indent = None if args.no_indent else 2
    out = pickle_to_json_bytes(pkl, mode=args.mode, allow=set(args.allow), indent=indent)

    with open(args.output, "wb") as f:
        f.write(out)

    print(f"Wrote {args.output}")

if __name__ == "__main__":
    main()

사용 예

# 1) 먼저 어떤 클래스가 필요한지 확인만 해보려면 (스크립트 수정 없이 간단 확인)
python -c "import sys, pickletools; import pathlib; p=pathlib.Path(sys.argv[1]).read_bytes(); \
print({(a.split('\n')[0]+'.'+a.split('\n')[1]) for (op,a,_) in pickletools.genops(p) if op.name in ('GLOBAL','STACK_GLOBAL')})" mydata.pkl

# 2) 안전 모드에서 허용 클래스 지정 후 변환
python pickle_to_json.py mydata.pkl mydata.json --mode safe --allow mypkg.models.User --allow mypkg.models.Order

# 3) 신뢰 모드(출처가 확실할 때만)
python pickle_to_json.py mydata.pkl mydata.json --mode trusted

---

실무 팁 (에러 최소화 포인트)

1. 화이트리스트는 점진적으로 넓혀라
   먼저 “막고 → 필요한 것만 허용”이 안전하고, 에러도 찾기 쉽다. 허용할 때는 정확한 풀네임(module.Class) 로.

2. 커스텀 클래스에__json__ 또는 to_dict() 추가
   프로젝트 내부 클래스라면 여기에 직렬화 규칙을 넣어두면, 변환기가 알아서 호출하므로 실패율이 크게 준다.

3. 순환 참조가 많은 그래프는$id/$ref 스킴 필수
   JSON 순환은 불가하니, 위 스킴 같은 참조 표현 을 반드시 도입해야 데이터 손실/오류가 없다.

4. 거대한 배열/데이터프레임은 별도 파일 고려
   JSON은 용량/속도에 약하다. 정말 크면 .npy나 .parquet를 사이드카 로 저장하고 JSON에는 메타/경로만 넣는 전략이 좋다.

5. 덤프 전에 변환만 먼저 실행해보기
   to_jsonable(obj) 결과를 한 번 찍어보면 어떤 타입이 어떻게 나오는지 바로 파악 가능. 실패 지점이 드러난다.

6. 로스리스가 중요하면 타입 메타를 꼭 남겨라
   "$type", shape, dtype 같은 메타가 재구성에 필수다.

7. 테스트

   • 작은 샘플 피클로 시작 → 점진 확대

   • json.loads(dumps(...)) 역파싱 검증

   • 대표 케이스: set/tuple/bytes/datetime/enum/numpy/pandas/커스텀 클래스를 모두 포함하는 합성 테스트

---

원하면 위 스크립트를 네 상황에 맞게 더 최소/간결화하거나, jsonpickle을 활용한 초간단 버전도 만들어줄게.

1. 그냥 모든걸 trust로 하고 싶어. 모두 내가 만든 파일들이야. 2. 질문. 파이썬 버전은 상관 없을까? 3. jsonpickle을 사용해서 간단해진다면 그렇게 해줘

좋아! 네 3가지 조건(1. 전부 trust 모드, 2. 파이썬 버전 이슈, 3. jsonpickle로 간단히) 기준으로, 에러를 최소화하는 실전용 절차 + 바로 쓸 수 있는 스크립트 까지 준비했어.

Step by Step

1) 전략 요약 (Trust 모드)

• 전제 : 피클은 전부 네가 만든 파일 → pickle.load로 곧바로 로드하고, 그 객체를 jsonpickle 로 JSON 변환.

• 이점 : 커스텀 클래스/순환 참조/공유 참조를 jsonpickle이 자동 처리(py/object, py/id, py/ref 메타).

• 선택지 :

  • 타입 복원 메타 포함(기본): unpicklable=True → 로스리스에 가까운 표현.

  • 순수 JSON만(타입 메타 제거): unpicklable=False → 아주 깔끔하지만, 타입/참조 복원 불가.

2) 파이썬 버전 호환성 (필수 체크)

• 같거나 더 새 버전의 파이썬 에서 로드하자. (예: Py 3.8에서 만든 피클 → Py 3.8 이상에서 로드 권장)

• 피클 프로토콜 은 “신버전이 구버전 읽기 OK, 그 반대는 종종 불가”가 일반적.

• 클래스 import 가능성 : 피클 안에 저장된 모듈/클래스 경로가 현재 환경에서 import 되어야 언피클 성공.

  • 즉, 프로젝트 코드가 PYTHONPATH 에 있어야 하고, 모듈/클래스 이름이 바뀌지 않아야 해.

• (구피클) Python 2에서 만든 피클이라면 Py3에서 encoding='latin1' 옵션이 필요할 수 있음.

3) jsonpickle로 간단/튼튼하게 변환

• 핵심 한 줄 : jsonpickle.encode(obj, unpicklable=..., make_refs=..., max_depth=...)

• 핸들러 등록 : NumPy/Pandas가 있다면 jsonpickle.ext.numpy.register_handlers(), jsonpickle.ext.pandas.register_handlers() 호출로 배열/데이터프레임을 자연스럽게 직렬화.

• Strict JSON 필요하면: allow_nan=False (NaN/Infinity 금지) 를 백엔드 옵션으로 지정.

• 대용량 이면: 들여쓰기 None(압축), 필요 시 sort_keys=False 유지로 성능/용량 최소화.

---

바로 실행 가능한 스크립트

내가 trust 모드 전용 변환 스크립트 를 만들어 뒀어. 옵션으로 순수 JSON, 참조 트래킹, strict JSON, 들여쓰기 등 세밀 제어가 가능해.

Download: pickle_to_json_trusted.py

사용법 예시

# 1) 기본: 타입/참조 메타 유지 (권장: 로스리스에 가까움)
python pickle_to_json_trusted.py mydata.pkl mydata.json

# 2) 순수 JSON(메타 제거) — 사람이 읽기/타 시스템 교환에 좋음
python pickle_to_json_trusted.py mydata.pkl mydata.json --pure

# 3) 사이클이 없고 참조 추적이 불필요하다면 약간 더 단순화
python pickle_to_json_trusted.py mydata.pkl mydata.json --no-refs

# 4) Strict JSON (NaN/Infinity 금지), 컴팩트 출력
python pickle_to_json_trusted.py mydata.pkl mydata.json --strict --no-indent

# 5) NumPy/Pandas가 있다면 자동 핸들러 등록됨(별도 옵션 불필요)

# 6) 정보 출력(프로토콜/글로벌 심볼 훑어보기)
python pickle_to_json_trusted.py mydata.pkl mydata.json --info

# 7) (레거시) Python 2에서 만든 피클이라면
python pickle_to_json_trusted.py mydata.pkl mydata.json --py2-latin1

출력 모드에 따른 JSON 형태는 대략 다음과 같아:

• 기본(메타 포함) : 각 객체에 py/object, 참조엔 py/id/py/ref가 붙음 → 복원 친화적.

• --pure : 표준 JSON만 남김 → 최대 호환성, 대신 타입/참조 정보는 사라짐.

---

에러를 더 줄이는 팁

1. 런타임 정합성

   • 언피클 시 ModuleNotFoundError, AttributeError가 나면, 현재 환경에서 필요한 모듈/클래스가 그대로 import 가능 한지 확인.

   • 패키지/모듈 이름 변경, 클래스 이동이 있었다면 임시 호환 레이어(구이름 alias) 추가가 현실적.

2. 메모리/성능

   • 매우 큰 배열/프레임은 --no-indent로 용량 줄이기.

   • 정말 큰 수치 데이터는 JSON 대신 사이드카 포맷(예: .npy/.parquet) 고려 후 JSON에는 메타/경로만 담는 전략도 가능.

3. 안전/정합성 검증

   • 생성된 JSON을 한 번 json.loads로 역파싱하면 문법 실수를 조기에 발견.

---

요약 답변

1. Trust로 가자 → pickle.load 후 jsonpickle.encode 가 가장 간단/범용.

2. 파이썬 버전 은 “같거나 더 최신”을 쓰면 대부분 안전. (특히 Py ≥3.8이면 프로토콜 5 대응 OK.) 모듈/클래스 경로가 동일하게 import 가능해야 함.

3. jsonpickle 사용 이 가장 손쉬움. 위 스크립트로 바로 가능하고, 필요 시 --pure(메타 제거), --strict(NaN 금지), --no-indent(대용량) 등 옵션으로 상황 맞춤화하면 돼.

원하면 py/object 메타 포맷 대신, 아예 $type 같은 커스텀 스킴으로 바꾼 미니 직렬화기도 만들어줄 수 있어.

본문에 그냥 써주고, 커스텀 클래스 import 는 어떤 식으로 해야해?

좋아—trust 모드 + jsonpickle 로 “피클 → JSON”을 간단/튼튼 하게 처리하는 스크립트를 본문에 바로 넣어둘게.
그리고 커스텀 클래스 import 는 어떻게 해야 언피클이 잘 되는지, 가장 실전적인 방법들을 아래에 정리했어.

---

Step by Step

1. 전제 : 전부 네가 만든 피클 → pickle.load(trust)로 로드해도 됨.

2. 버전 : 피클을 만든 파이썬 동일 또는 더 최신 버전 에서 로드(특히 모듈 경로가 동일하게 import 가능해야 함).

3. 핵심 : 로드한 객체를 jsonpickle.encode(...)로 JSON 변환.

4. 대용량/호환성 옵션 :

   • --pure → 타입 메타 제거(아주 깔끔한 JSON), 대신 복원 불가.

   • --no-refs → 참조/사이클 추적 비활성화(간단하지만 공유참조 정보 사라짐).

   • --strict → NaN/Infinity 금지(정-표준 JSON).

   • --no-indent → 용량 최소화.

5. 커스텀 클래스 import 는 “현재 프로세스에서 해당 모듈을 정확히 import 가능 ”해야 함. 아래 “Import 요령” 참고.

---

신뢰 모드 전용 스크립트 (jsonpickle 사용)

아래 코드를 pickle_to_json_trusted.py로 저장해서 바로 써.

#!/usr/bin/env python3
# pickle_to_json_trusted.py
# Trust-mode: loads your own pickle files, then encodes to JSON with jsonpickle.

import argparse, json, pickle, sys, os, traceback, importlib
import pickletools

try:
    import jsonpickle
except Exception:
    print("ERROR: jsonpickle is required. Install with: pip install jsonpickle", file=sys.stderr)
    raise

def register_optional_exts():
    # Optional handlers for numpy/pandas if available
    try:
        from jsonpickle.ext import numpy as jsonpickle_numpy
        jsonpickle_numpy.register_handlers()
    except Exception:
        pass
    try:
        from jsonpickle.ext import pandas as jsonpickle_pandas
        jsonpickle_pandas.register_handlers()
    except Exception:
        pass

def peek_protocol(pkl_bytes: bytes):
    try:
        for op, arg, pos in pickletools.genops(pkl_bytes):
            if op.name == "PROTO":
                return arg
    except Exception:
        return None
    return None

def scan_globals(pkl_bytes: bytes):
    refs = set()
    try:
        for op, arg, pos in pickletools.genops(pkl_bytes):
            if op.name in ("GLOBAL", "STACK_GLOBAL"):
                if isinstance(arg, tuple):
                    mod, name = arg
                else:
                    parts = str(arg).split("\n")
                    if len(parts) >= 2:
                        mod, name = parts[0], parts[1]
                    else:
                        continue
                refs.add(f"{mod}.{name}")
    except Exception:
        pass
    return sorted(refs)

def main():
    ap = argparse.ArgumentParser(description="Trust-mode: Convert a Python pickle to JSON using jsonpickle.")
    ap.add_argument("input", help="Input .pkl file path")
    ap.add_argument("output", help="Output .json file path")
    # JSON shape options
    ap.add_argument("--pure", action="store_true",
                    help="Emit plain JSON (unpicklable=False) without py/object metadata.")
    ap.add_argument("--no-refs", action="store_true",
                    help="Do not track object references/cycles (make_refs=False).")
    ap.add_argument("--max-depth", type=int, default=None,
                    help="Limit traversal depth (None = unlimited).")
    ap.add_argument("--indent", type=int, default=2,
                    help="Pretty-print indent (use 0 or --no-indent for compact).")
    ap.add_argument("--no-indent", action="store_true",
                    help="Compact JSON (indent=None).")
    ap.add_argument("--ensure-ascii", action="store_true",
                    help="Escape non-ASCII characters.")
    ap.add_argument("--strict", action="store_true",
                    help="Disallow NaN/Infinity (allow_nan=False).")
    ap.add_argument("--sort-keys", action="store_true",
                    help="Sort JSON object keys.")
    ap.add_argument("--info", action="store_true",
                    help="Print pickle protocol & referenced globals before converting.")
    ap.add_argument("--py2-latin1", action="store_true",
                    help="If the pickle came from Python 2, try encoding='latin1'.")

    # Import help (for custom classes)
    ap.add_argument("--sys-path", action="append", default=[],
                    help="Prepend a path to sys.path (repeatable).")
    ap.add_argument("--preload", action="append", default=[],
                    help="Import module(s) before unpickling, e.g. mypkg.models (repeatable).")
    ap.add_argument("--alias", action="append", default=[],
                    help="Module alias mapping OLD=NEW (repeatable).")

    args = ap.parse_args()

    # sys.path adjustments for project importability
    for p in args.sys_path:
        sys.path.insert(0, os.path.abspath(p))

    # module aliasing (module path changed)
    for pair in args.alias:
        if "=" not in pair:
            print(f"WARNING: --alias expects OLD=NEW, got {pair}", file=sys.stderr)
            continue
        old, new = pair.split("=", 1)
        new_mod = importlib.import_module(new)
        sys.modules[old] = new_mod

    # preload modules so that pickle can resolve classes
    for m in args.preload:
        importlib.import_module(m)

    # read pickle bytes
    try:
        with open(args.input, "rb") as f:
            pkl_bytes = f.read()
    except FileNotFoundError:
        print(f"ERROR: file not found: {args.input}", file=sys.stderr)
        sys.exit(1)

    if args.info:
        proto = peek_protocol(pkl_bytes)
        globs = scan_globals(pkl_bytes)
        print(f"[info] protocol: {proto if proto is not None else 'unknown'}")
        if globs:
            print(f"[info] referenced globals ({len(globs)}):")
            for g in globs:
                print(f"  - {g}")
        else:
            print("[info] referenced globals: (none found)")

    # jsonpickle backend setup
    register_optional_exts()
    jsonpickle.set_preferred_backend('json')
    indent = None if args.no_indent else args.indent
    # encoder opts
    try:
        jsonpickle.set_encoder_options(
            'json',
            ensure_ascii=args.ensure_ascii,
            allow_nan=not args.strict,
            indent=indent,
            sort_keys=args.sort_keys,
        )
    except TypeError:
        jsonpickle.set_encoder_options('json', ensure_ascii=args.ensure_ascii, indent=indent)

    # unpickle (TRUST)
    try:
        import io
        bio = io.BytesIO(pkl_bytes)
        if args.py2_latin1:
            obj = pickle.load(bio, encoding='latin1', fix_imports=True)
        else:
            obj = pickle.load(bio)
    except ModuleNotFoundError as e:
        print("ERROR: Required module isn't importable.", file=sys.stderr)
        print(f"Detail: {e}", file=sys.stderr)
        print("Tip: Use --sys-path/--preload/--alias or set PYTHONPATH.", file=sys.stderr)
        sys.exit(2)
    except AttributeError as e:
        print("ERROR: Class/attribute moved or renamed.", file=sys.stderr)
        print(f"Detail: {e}", file=sys.stderr)
        sys.exit(2)
    except Exception:
        print("ERROR: Unpickling failed.", file=sys.stderr)
        traceback.print_exc()
        sys.exit(2)

    # encode with jsonpickle
    try:
        json_str = jsonpickle.encode(
            obj,
            unpicklable=not args.pure,
            make_refs=not args.no_refs,
            max_depth=args.max_depth,
        )
    except Exception:
        print("ERROR: jsonpickle encoding failed.", file=sys.stderr)
        traceback.print_exc()
        sys.exit(3)

    # write
    try:
        with open(args.output, "w", encoding="utf-8") as f:
            f.write(json_str)
    except Exception as e:
        print(f"ERROR: writing output failed: {e}", file=sys.stderr)
        sys.exit(4)

    print(f"Wrote {args.output}")

if __name__ == "__main__":
    main()

실행 예

# (가장 기본) 타입/참조 메타 유지: 복원 친화적
python pickle_to_json_trusted.py data.pkl data.json

# 순수 JSON (메타 제거)
python pickle_to_json_trusted.py data.pkl data.json --pure

# 사이클/참조 추적 끄기 + 컴팩트 + Strict JSON
python pickle_to_json_trusted.py data.pkl data.json --no-refs --no-indent --strict

# 피클 정보(프로토콜/글로벌): 디버깅용
python pickle_to_json_trusted.py data.pkl data.json --info

---

커스텀 클래스 import — 이렇게 하면 된다 (실전 레시피)

언피클이 커스텀 클래스를 복원하려면, 피클에 기록된 “모듈.클래스”가 현재 런타임에서 import 가능 해야 해.

1) 가장 깔끔: 프로젝트를 패키지로 설치

# 프로젝트 루트에서
pip install -e .
# 또는 PYTHONPATH로 임시 등록
PYTHONPATH=/path/to/project python pickle_to_json_trusted.py data.pkl out.json

• 이러면 mypkg.models.User 같은 경로를 파이썬이 자연스럽게 찾음.

2) 경로만 추가해서 import 가능하게 (스크립트 옵션)

프로젝트 루트를 sys.path 앞에 추가:

python pickle_to_json_trusted.py data.pkl out.json \
  --sys-path /absolute/path/to/project_root \
  --preload mypkg.models

• --preload는 해당 모듈을 미리 import해서 클래스 등록을 확실히 해줘.

3) 모듈 경로가 바뀐 경우(리팩터링) → 모듈 alias

예: 예전에는 oldpkg.models였는데 지금은 newpkg.models로 이동했다면:

python pickle_to_json_trusted.py data.pkl out.json \
  --sys-path /path/to/project \
  --alias oldpkg.models=newpkg.models \
  --preload newpkg.models

• 이러면 pickle.load가 oldpkg.models를 찾으려 할 때, 내부적으로 newpkg.models를 대신 사용.

4) 클래스가 다른 모듈로 이동했다면 → 런타임 shim(간단 예)

모듈 alias만으로 부족할 때(클래스 이름도 바뀌었거나 세부 구조가 바뀐 경우), 언피클 전에 임시 shim 모듈 을 만들어 매핑해 줄 수 있어:

# 예: OldClass가 oldpkg.models → newpkg.models.NewClass 로 이동/개명된 경우
import types, sys
import newpkg.models as newmod

shim = types.ModuleType("oldpkg.models")
# 예전 이름을 새 클래스에 매핑
shim.OldClass = newmod.NewClass
# 언피클이 찾아쓰도록 sys.modules 등록
sys.modules["oldpkg.models"] = shim

# 이제 pickle.load를 호출

위 shim 코드는 스크립트 최상단(언피클 직전)에 넣으면 되고, 경우에 따라 여러 클래스를 넣어 매핑하면 된다.

5) Python 2 → 3 피클일 때

python pickle_to_json_trusted.py data.pkl out.json --py2-latin1

• encoding='latin1', fix_imports=True로 과거 피클 호환을 시도.

---

작게 요약

• Trust OK → pickle.load + jsonpickle.encode가 가장 간단.

• 버전 은 “같거나 최신” 권장 + 모듈/클래스 경로가 현재 환경에서 그대로 import 가능해야 함.

• 커스텀 클래스 import 는

  1. 패키지로 설치(또는 PYTHONPATH),

  2. --sys-path / --preload,

  3. --alias 올드=뉴(모듈 경로 변경),

  4. 필요 시 shim 모듈 로 클래스 레벨 매핑.

원하면 shim을 옵션화해서 스크립트에 바로 넣는 버전(예: --shim oldpkg.models:OldClass=newpkg.models:NewClass)도 만들어줄게.