Nemotron-Personas-Korea 이해 보고서

"""10단계: 생성 원리 미니 실습. 확률적 그래프 모델(PGM)을 손으로 만들어 본다.

NVIDIA의 실제 모델은 비공개다. 여기서는 같은 '원리'를 변수 4개로 줄여 흉내 낸다.

    성별 -> 연령대 -> 혼인상태 -> 가구유형

순서
  1) 실제 데이터에서 조건부 확률표(CPT)를 계산한다.      데이터 -> 확률표
  2) 그 확률표로 조상 샘플링을 해서 가상 인물을 뽑는다.   확률표 -> 새 데이터
  3) 화살표를 모두 끊고(독립 샘플링) 뽑은 것과 비교한다.
  4) 뽑힌 한 사람을 LLM에게 줄 프롬프트 문자열로 바꾼다. (LLM은 부르지 않는다)

numpy와 파이썬 기본 자료형(딕셔너리)만 쓴다. polars는 데이터를 읽고 세는 데만 쓴다.
"""
import json
import os
from collections import Counter

import numpy as np
import polars as pl

import config as C

plt = C.setup_matplotlib()
rng = np.random.default_rng(C.SEED)  # 시드를 고정해 몇 번을 돌려도 같은 결과가 나온다
TOY_N = int(os.environ.get("NPK_TOY_N", "10000"))  # 뽑을 가상 인물 수

# ---------------------------------------------------------------- 0. 재료 준비
df = (
    pl.scan_parquet(C.parquet_glob())
    .select("sex", "age", "marital_status", "family_type")
    .collect()
    .with_columns(((pl.col("age") // 10) * 10).cast(pl.String).add("대").alias("age_band"))  # 23 -> "20대"
    .with_columns(pl.when(pl.col("age") < 20).then(pl.lit("10대")).otherwise(pl.col("age_band")).alias("age_band"))
)
# 가구유형은 39개나 된다. 많은 순으로 6개만 남기고 나머지는 '기타'로 합쳐 표를 작게 만든다.
top_family = df.group_by("family_type").len().sort("len", descending=True).head(6)["family_type"].to_list()
df = df.with_columns(
    pl.when(pl.col("family_type").is_in(top_family)).then(pl.col("family_type")).otherwise(pl.lit("기타")).alias("family")
).rename({"marital_status": "marital"})
rows = df.select("sex", "age_band", "marital", "family").rows()  # [(성별, 연령대, 혼인, 가구), ...]
print("원본 행 수:", len(rows))


# ---------------------------------------------------------------- 1. 데이터 -> 확률표
def marginal(values):
    """값들의 비율. 예: {'남자': 0.497, '여자': 0.503}"""
    c = Counter(values)
    total = sum(c.values())
    return {k: v / total for k, v in c.items()}


def conditional(pairs):
    """(부모 값, 자식 값) 쌍에서 조건부 확률표를 만든다. 결과: {부모 값: {자식 값: 확률}}"""
    by_parent = {}
    for parent, child in pairs:
        by_parent.setdefault(parent, []).append(child)
    return {parent: marginal(children) for parent, children in by_parent.items()}


P_SEX = marginal(r[0] for r in rows)                          # P(성별)
P_AGE_GIVEN_SEX = conditional((r[0], r[1]) for r in rows)     # P(연령대 | 성별)
P_MARITAL_GIVEN_AGE = conditional((r[1], r[2]) for r in rows)  # P(혼인 | 연령대)
P_FAMILY_GIVEN_MARITAL = conditional((r[2], r[3]) for r in rows)  # P(가구 | 혼인)

print("\n[확률표 예시] P(혼인 | 연령대)")
for band in ["20대", "50대", "80대"]:
    print(" ", band, {k: round(v, 3) for k, v in sorted(P_MARITAL_GIVEN_AGE[band].items(), key=lambda x: -x[1])})


# ---------------------------------------------------------------- 2. 확률표 -> 새 데이터 (조상 샘플링)
def draw(dist):
    """확률표 한 줄({값: 확률})에서 값 하나를 뽑는다."""
    keys = list(dist)
    return keys[rng.choice(len(keys), p=[dist[k] for k in keys])]


def sample_pgm():
    """부모부터 차례로 뽑는다. 앞에서 뽑힌 값이 다음 확률표의 '줄'을 고른다."""
    sex = draw(P_SEX)
    age = draw(P_AGE_GIVEN_SEX[sex])             # 성별에 맞는 줄에서 연령대를 뽑는다
    marital = draw(P_MARITAL_GIVEN_AGE[age])      # 연령대에 맞는 줄에서 혼인을 뽑는다
    family = draw(P_FAMILY_GIVEN_MARITAL[marital])  # 혼인에 맞는 줄에서 가구를 뽑는다
    return (sex, age, marital, family)


# 비교용: 화살표를 모두 끊는다. 변수마다 전체 비율만 보고 따로따로 뽑는다.
P_AGE = marginal(r[1] for r in rows)
P_MARITAL = marginal(r[2] for r in rows)
P_FAMILY = marginal(r[3] for r in rows)


def sample_independent():
    return (draw(P_SEX), draw(P_AGE), draw(P_MARITAL), draw(P_FAMILY))


pgm = [sample_pgm() for _ in range(TOY_N)]
ind = [sample_independent() for _ in range(TOY_N)]


# ---------------------------------------------------------------- 3. 비교
def tvd(p, q):
    """총변동거리. 두 분포가 얼마나 다른지. 0이면 같고 1이면 완전히 다르다."""
    keys = set(p) | set(q)
    return 0.5 * sum(abs(p.get(k, 0.0) - q.get(k, 0.0)) for k in keys)


def joint(data, i, j):
    """두 변수의 결합분포. 예: ('20대', '미혼')이 전체의 몇 %인가."""
    return marginal((r[i], r[j]) for r in data)


def odd_rate(data):
    """현실에서 거의 없는 조합의 비율: 10~20대인데 사별 또는 이혼."""
    n = sum(1 for r in data if r[1] in ("10대", "20대") and r[2] in ("사별", "이혼"))
    return n / len(data) * 100


IDX = {"sex": 0, "age": 1, "marital": 2, "family": 3}
result = {"toy_n": TOY_N, "marginal_tvd": {}, "joint_tvd": {}, "odd_rate_pct": {}}
for name, i in IDX.items():  # 변수 하나씩의 분포는 두 방법 모두 원본과 비슷해야 한다
    orig = marginal(r[i] for r in rows)
    result["marginal_tvd"][name] = {
        "pgm": round(tvd(orig, marginal(r[i] for r in pgm)), 4),
        "independent": round(tvd(orig, marginal(r[i] for r in ind)), 4),
    }
# 변수 '쌍'의 분포에서 차이가 난다. 마지막 쌍(연령대x가구)은 우리 모형에 직접 화살표가 없는 쌍이다.
for a, b in [("age", "marital"), ("marital", "family"), ("age", "family")]:
    orig = joint(rows, IDX[a], IDX[b])
    result["joint_tvd"][f"{a} x {b}"] = {
        "pgm": round(tvd(orig, joint(pgm, IDX[a], IDX[b])), 4),
        "independent": round(tvd(orig, joint(ind, IDX[a], IDX[b])), 4),
    }
result["odd_rate_pct"] = {"original": round(odd_rate(rows), 3), "pgm": round(odd_rate(pgm), 3),
                          "independent": round(odd_rate(ind), 3)}

print("\n[변수 하나씩] 원본과의 거리(TVD):", result["marginal_tvd"])
print("[변수 쌍] 원본과의 거리(TVD):", result["joint_tvd"])
print("[10~20대인데 사별·이혼인 비율 %]:", result["odd_rate_pct"])

# 그림: 연령대별 미혼 비율. 의존 구조를 지키면 원본 곡선을 따라가고, 끊으면 평평해진다.
bands = sorted(P_AGE, key=lambda b: int(b[:-1]))


def single_rate(data):
    out = []
    for b in bands:
        grp = [r for r in data if r[1] == b]
        out.append(sum(1 for r in grp if r[2] == "미혼") / len(grp) * 100 if grp else 0.0)
    return out


curves = {"원본 데이터": single_rate(rows), "모형 A (화살표 구조 유지)": single_rate(pgm),
          "독립 샘플링 (화살표 없음)": single_rate(ind)}
result["single_rate_by_age"] = {"bands": bands, **{k: [round(v, 1) for v in vals] for k, vals in curves.items()}}
fig, ax = plt.subplots(figsize=(8, 4.6))
styles = [(C.COLOR_MUTED, "-", 4.0), (C.COLOR_1, "-", 2.0), (C.COLOR_2, "--", 2.0)]  # 색과 선 모양을 함께 달리한다
for (label, vals), (color, ls, lw) in zip(curves.items(), styles):
    ax.plot(bands, vals, color=color, linestyle=ls, linewidth=lw, marker="o", markersize=5, label=label)
ax.set_ylabel("미혼 비율 (%)")
ax.set_ylim(0, 105)
ax.set_title("연령대별 미혼 비율", loc="left", fontweight="bold")
ax.legend(frameon=False)
fig.savefig(C.FIG_DIR / "toy_pgm_single_rate.png")
plt.close(fig)


# ---------------------------------------------------------------- 4. 인구통계 변수 -> 프롬프트
def build_prompt(person):
    """샘플링된 인구통계 변수를 LLM에게 줄 지시문으로 바꾼다. 실제 파이프라인의 'LLM 컬럼'이 하는 일이다.

    NVIDIA가 쓴 실제 프롬프트는 비공개다. 이것은 원리를 보여 주는 예시일 뿐이다.
    """
    sex, age, marital, family = person
    return (
        "다음 조건의 한국인 한 사람을 상상해 직업 페르소나를 두 문장으로 써라.\n"
        f"- 성별: {sex}\n- 연령대: {age}\n- 혼인 상태: {marital}\n- 가구 유형: {family}\n"
        "조건과 모순되는 내용은 쓰지 말고, 장점 하나와 인간적인 약점 하나를 넣어라."
    )


example_people = pgm[:3]
result["example_people"] = [list(p) for p in example_people]
result["example_prompt"] = build_prompt(example_people[0])
result["cpt_marital_given_age"] = {b: {k: round(v, 3) for k, v in P_MARITAL_GIVEN_AGE[b].items()} for b in bands}
print("\n[뽑힌 가상 인물 3명]", example_people)
print("\n[첫 번째 인물의 프롬프트]\n" + result["example_prompt"])

# ---------------------------------------------------------------- 5. 확률표가 만들어지는 과정을 숫자로 남긴다
# 확률표의 한 줄은 "그 부모 값을 가진 행만 모아서, 자식 값별로 센 뒤, 그 줄의 합으로 나눈 것"이다.
FAMILY_ORDER = sorted(P_FAMILY, key=lambda k: -P_FAMILY[k])
MARITAL_ORDER = ["미혼", "배우자있음", "이혼", "사별"]
worked = {}
for band in ["20대", "80대"]:
    cnt = Counter(r[2] for r in rows if r[1] == band)  # 이 연령대인 사람들의 혼인 상태를 센다
    total = sum(cnt.values())
    worked[band] = {"n": total, "counts": {m: cnt.get(m, 0) for m in MARITAL_ORDER},
                    "probs": {m: round(cnt.get(m, 0) / total, 4) for m in MARITAL_ORDER}}
result["worked_cpt"] = worked
result["original_preview"] = [list(r) for r in rows[:5]]  # 원본이 어떻게 생겼는지 보여 줄 다섯 행
result["n_original"] = len(rows)
result["cpt_sex"] = {k: round(v, 4) for k, v in P_SEX.items()}
result["cpt_age_given_sex"] = {s: {b: round(P_AGE_GIVEN_SEX[s].get(b, 0.0), 4) for b in bands} for s in P_SEX}
result["cpt_family_given_marital"] = {m: {f: round(P_FAMILY_GIVEN_MARITAL[m].get(f, 0.0), 4) for f in FAMILY_ORDER}
                                      for m in MARITAL_ORDER}
result["family_order"] = FAMILY_ORDER

# ---------------------------------------------------------------- 6. 한 사람이 뽑히는 과정을 따라가 본다
# draw()가 안에서 하는 일을 풀어 쓴 것이다. 0과 1 사이의 난수 u를 하나 뽑고,
# 확률을 차례로 쌓아 만든 구간 중 u가 떨어진 구간의 값을 고른다. 확률이 큰 값일수록 구간이 넓어 자주 뽑힌다.
trace_rng = np.random.default_rng(C.SEED + 1)


def draw_traced(dist, order):
    u = float(trace_rng.random())
    lo = 0.0
    intervals, chosen = [], None
    for k in order:
        p = dist.get(k, 0.0)
        hi = lo + p
        hit = chosen is None and u < hi
        if hit:
            chosen = k
        intervals.append({"value": k, "p": round(p, 4), "lo": round(lo, 4), "hi": round(hi, 4), "hit": hit})
        lo = hi
    return chosen or order[-1], round(u, 4), intervals


trace = []
v_sex, u, iv = draw_traced(P_SEX, list(P_SEX))
trace.append({"step": "성별", "table": "P(성별)", "row": "—", "u": u, "intervals": iv, "chosen": v_sex})
v_age, u, iv = draw_traced(P_AGE_GIVEN_SEX[v_sex], bands)
trace.append({"step": "연령대", "table": "P(연령대 | 성별)", "row": v_sex, "u": u, "intervals": iv, "chosen": v_age})
v_mar, u, iv = draw_traced(P_MARITAL_GIVEN_AGE[v_age], MARITAL_ORDER)
trace.append({"step": "혼인 상태", "table": "P(혼인 | 연령대)", "row": v_age, "u": u, "intervals": iv, "chosen": v_mar})
v_fam, u, iv = draw_traced(P_FAMILY_GIVEN_MARITAL[v_mar], FAMILY_ORDER)
trace.append({"step": "가구 유형", "table": "P(가구 | 혼인)", "row": v_mar, "u": u, "intervals": iv, "chosen": v_fam})
result["trace"] = trace
result["trace_prompt"] = build_prompt((v_sex, v_age, v_mar, v_fam))
print("\n[한 사람 추적]", [(t["step"], t["u"], t["chosen"]) for t in trace])

# ---------------------------------------------------------------- 7. 화살표를 하나 더 넣으면 어떻게 되나
# 모형 A: 성별 -> 연령대 -> 혼인 -> 가구               (위에서 쓴 것)
# 모형 B: 모형 A + '연령대 -> 가구' 화살표. 가구의 부모가 (혼인, 연령대) 둘이 된다.
P_FAMILY_GIVEN_MARITAL_AGE = conditional(((r[2], r[1]), r[3]) for r in rows)  # P(가구 | 혼인, 연령대)


def sample_pgm_b():
    sex = draw(P_SEX)
    age = draw(P_AGE_GIVEN_SEX[sex])
    marital = draw(P_MARITAL_GIVEN_AGE[age])
    family = draw(P_FAMILY_GIVEN_MARITAL_AGE[(marital, age)])  # 부모 두 개의 값으로 줄을 고른다
    return (sex, age, marital, family)


pgm_b = [sample_pgm_b() for _ in range(TOY_N)]
n_sex, n_age, n_mar, n_fam = len(P_SEX), len(bands), len(MARITAL_ORDER), len(FAMILY_ORDER)
cells_a = n_sex + n_sex * n_age + n_age * n_mar + n_mar * n_fam
cells_b = n_sex + n_sex * n_age + n_age * n_mar + n_mar * n_age * n_fam
cells_full = n_sex * n_age * n_mar * n_fam
orig_af = joint(rows, IDX["age"], IDX["family"])
# 같은 크기의 표본을 원본에서 그냥 뽑아도 0이 되지는 않는다. 그 값이 '표본 오차만으로 생기는 거리'의 기준이 된다.
resample = [rows[i] for i in rng.integers(0, len(rows), TOY_N)]
result["model_compare"] = {
    "cells": {"independent": n_sex + n_age + n_mar + n_fam, "model_a": cells_a, "model_b": cells_b, "full_joint": cells_full},
    "age_x_family_tvd": {
        "independent": result["joint_tvd"]["age x family"]["independent"],
        "model_a": result["joint_tvd"]["age x family"]["pgm"],
        "model_b": round(tvd(orig_af, joint(pgm_b, IDX["age"], IDX["family"])), 4),
        "resample_floor": round(tvd(orig_af, joint(resample, IDX["age"], IDX["family"])), 4),
    },
}


def alone_rate(data, band):
    """특정 연령대에서 '혼자 거주'의 비율(%)."""
    grp = [r for r in data if r[1] == band]
    return round(sum(1 for r in grp if r[3] == "혼자 거주") / len(grp) * 100, 1) if grp else 0.0


result["alone_by_age"] = {
    "bands": bands,
    "original": [alone_rate(rows, b) for b in bands],
    "model_a": [alone_rate(pgm, b) for b in bands],
    "model_b": [alone_rate(pgm_b, b) for b in bands],
}
print("[모형 비교] 칸 수:", result["model_compare"]["cells"])
print("[모형 비교] 연령대 x 가구의 거리:", result["model_compare"]["age_x_family_tvd"])

fig, ax = plt.subplots(figsize=(8, 4.6))
series = [("원본 데이터", result["alone_by_age"]["original"], C.COLOR_MUTED, "-", 4.0, "o"),
          ("모형 A (연령대 → 가구 화살표 없음)", result["alone_by_age"]["model_a"], C.COLOR_2, "--", 2.0, "s"),
          ("모형 B (화살표 추가)", result["alone_by_age"]["model_b"], C.COLOR_1, "-", 2.0, "o")]
for label, vals, color, ls, lw, mk in series:
    ax.plot(bands, vals, color=color, linestyle=ls, linewidth=lw, marker=mk, markersize=5, label=label)
ax.set_ylabel("혼자 거주 비율 (%)")
ax.set_ylim(bottom=0)
ax.set_title("연령대별 '혼자 거주' 비율", loc="left", fontweight="bold")
ax.legend(frameon=False)
fig.savefig(C.FIG_DIR / "toy_pgm_alone_rate.png")
plt.close(fig)

out = C.PROCESSED_DIR / "summary_toy_pgm.json"
out.write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8")
print("\n저장:", out)

"""11단계(학습용): 조건부 독립을 이 데이터에서 직접 숫자로 확인한다.

질문: "두 변수가 관련 있어 보인다"와 "두 변수가 직접 연결돼 있다"는 어떻게 다른가?

도구는 두 가지다.
  상호정보량        I(X;Y)   : X를 알면 Y에 대해 얼마나 알게 되나. 0이면 독립이다.
  조건부 상호정보량 I(X;Y|Z) : Z를 이미 아는 상태에서, X가 Y에 대해 '추가로' 알려 주는 양.
                               0에 가까우면 "Z를 알면 X와 Y는 독립"이다 (조건부 독립, X ⊥ Y | Z).

선행 독립 감사(eopchang)의 3단계가 쓴 방법과 같다. 단위는 nats(자연로그)다.
"""
import json

import numpy as np
import polars as pl

import config as C

COLS = ["sex", "age", "marital_status", "family_type", "housing_type",
        "education_level", "bachelors_field", "district", "province"]
df = pl.scan_parquet(C.parquet_glob()).select(COLS).collect()

# 나이는 감사 보고서와 같은 8개 구간으로 나눈다: 19-24 / 25-34 / ... / 85+
df = df.with_columns(
    pl.col("age").cut([24, 34, 44, 54, 64, 74, 84],
                      labels=["19-24", "25-34", "35-44", "45-54", "55-64", "65-74", "75-84", "85+"])
    .cast(pl.String).alias("age_bin")
)
N = df.height


def entropy(cols):
    """결합 엔트로피 H(cols). 값들의 조합이 얼마나 다양하게 흩어져 있는지. 조합별 비율 p에 대해 -sum(p log p)."""
    counts = df.group_by(cols).len()["len"].to_numpy().astype(float)
    p = counts / counts.sum()
    return float(-(p * np.log(p)).sum())


def mi(x, y):
    """I(X;Y) = H(X) + H(Y) - H(X,Y)"""
    return entropy([x]) + entropy([y]) - entropy([x, y])


def cmi(x, y, z):
    """I(X;Y|Z) = H(X,Z) + H(Y,Z) - H(X,Y,Z) - H(Z).  z는 변수 이름의 리스트다."""
    return entropy([x] + z) + entropy([y] + z) - entropy([x, y] + z) - entropy(z)


# (X, Y, Z, 구조 이름, 풀이)
CASES = [
    ("marital_status", "education_level", ["age_bin"], "분기 (공통 원인)",
     "혼인과 학력은 관련 있어 보인다. 하지만 둘 다 나이의 결과다. 나이를 알면 관계가 거의 사라진다."),
    ("age_bin", "bachelors_field", ["education_level"], "연쇄",
     "나이 → 학력 → 전공. 나이는 학력을 거쳐서만 전공에 영향을 준다. 학력을 알면 나이는 더 알려 줄 게 없다."),
    ("housing_type", "province", ["district"], "연쇄 (지리 계층)",
     "시도 → 시군구 → 주거. 시군구를 알면 시도는 군더더기다."),
    ("age_bin", "marital_status", ["education_level"], "직접 연결 (대조군)",
     "나이와 혼인은 직접 이어져 있다. 다른 변수를 알아도 관계가 그대로 남는다."),
    ("marital_status", "family_type", ["age_bin"], "직접 연결 (대조군)",
     "혼인과 가구 유형도 직접 이어져 있다. 나이를 알아도 관계가 남는다."),
    ("housing_type", "family_type", [], "애초에 독립",
     "주거 형태와 가구 유형은 조건을 걸기 전부터 거의 독립이다. 이 데이터의 약점이다."),
]

rows = []
for x, y, z, kind, story in CASES:
    m = mi(x, y)
    c = cmi(x, y, z) if z else None
    drop = None if c is None else round((1 - c / m) * 100, 1)
    rows.append({"x": x, "y": y, "z": z, "kind": kind, "story": story,
                 "mi": round(m, 4), "cmi": None if c is None else round(c, 4), "drop_pct": drop})
    print(f"[{kind}] I({x}; {y}) = {m:.4f}" + (f"  ->  I(.. | {', '.join(z)}) = {c:.4f}  ({drop}% 감소)" if z else ""))

# 범주가 많으면 상호정보량이 실제보다 크게 추정된다. 대략 (k-1)(m-1)/(2N) 만큼이다 (Miller-Madow 근사).
k_occ = pl.scan_parquet(C.parquet_glob()).select(pl.col("occupation").n_unique()).collect().item()
k_dist = df["district"].n_unique()
bias = {"k_occupation": k_occ, "k_district": k_dist, "n": N,
        "approx_bias_nats": round((k_occ - 1) * (k_dist - 1) / (2 * N), 3)}
print("직업 x 시군구의 과대 추정 근사:", bias)

# 결합분포를 통째로 적으려면 칸이 몇 개 필요한가 (11개 변수, 나이는 8구간)
cards = {"sex": 2, "age_bin": 8, "marital_status": 4, "military_status": 2,
         "family_type": df["family_type"].n_unique(), "housing_type": df["housing_type"].n_unique(),
         "education_level": df["education_level"].n_unique(), "bachelors_field": df["bachelors_field"].n_unique(),
         "occupation": k_occ, "district": k_dist, "province": df["province"].n_unique()}
cells = 1
for v in cards.values():
    cells *= v
print(f"결합분포의 칸 수: {cells:,} (데이터는 {N:,}행)")

out = C.PROCESSED_DIR / "summary_cond_indep.json"
out.write_text(json.dumps({"cases": rows, "bias": bias, "cardinalities": cards, "joint_cells": cells},
                          ensure_ascii=False, indent=2), encoding="utf-8")
print("저장:", out)

"""2단계: 26개 필드가 각각 무엇인지 한 장의 표로 만든다.

컬럼마다 자료형, 결측 수, 고유값 수, 예시 값 3개, 평균 글자 수를 구한다.
메모리를 아끼려고 컬럼을 하나씩 읽는다.
"""
import csv

import polars as pl

import config as C

lf = pl.scan_parquet(C.parquet_glob())  # 아직 읽지 않는다. 읽을 계획만 세운다.
schema = lf.collect_schema()
print("컬럼 수:", len(schema))

rows = []
for name, dtype in schema.items():
    s = lf.select(name).collect().to_series()  # 이 컬럼 하나만 메모리에 올린다
    n_null = s.null_count()
    is_list = isinstance(dtype, pl.List)
    # 리스트형은 고유값을 세기 어렵기 때문에 문자열로 바꿔서 센다
    s_cmp = s.cast(pl.List(pl.String)).list.join("|") if is_list else s
    n_unique = s_cmp.n_unique()
    examples = [str(v) for v in s_cmp.drop_nulls().head(3).to_list()]
    mean_len = None
    if s_cmp.dtype == pl.String:
        mean_len = round(s_cmp.str.len_chars().mean(), 1)  # 평균 글자 수
    rows.append(
        {
            "column": name,
            "dtype": str(dtype),
            "n_null": n_null,
            "n_unique": n_unique,
            "mean_chars": mean_len,
            "example_1": examples[0][:120] if len(examples) > 0 else "",
            "example_2": examples[1][:120] if len(examples) > 1 else "",
            "example_3": examples[2][:120] if len(examples) > 2 else "",
        }
    )
    print(f"{name:32s} {str(dtype):14s} null={n_null:<6d} unique={n_unique:<8d} len={mean_len}")

out = C.PROCESSED_DIR / "schema.csv"
with open(out, "w", encoding="utf-8-sig", newline="") as f:  # utf-8-sig: 엑셀에서 한글이 안 깨진다
    w = csv.DictWriter(f, fieldnames=list(rows[0]))
    w.writeheader()
    w.writerows(rows)
print("저장:", out)

"""3단계: 12개 인구통계·지리 변수의 기초통계.

만드는 것
- 변수별 빈도표 (data/processed/freq_*.csv)
- 나이 요약, 인구 피라미드
- 교차표 4개와 히트맵
- 선행 감사가 지적한 약점 2개를 빈도표로 확인
- 위 결과를 모은 요약 JSON (보고서가 읽는다)
"""
import json

import polars as pl

import config as C

plt = C.setup_matplotlib()

COLS = [
    "sex", "age", "marital_status", "military_status", "family_type", "housing_type",
    "education_level", "bachelors_field", "occupation", "district", "province", "country",
]

# 텍스트 컬럼은 빼고 12개 컬럼만 읽는다. 100만 행이어도 수백 MB면 된다.
df = pl.scan_parquet(C.parquet_glob()).select(COLS).collect()
N = df.height
print("행 수:", N)
summary = {"n_rows": N}


def save_csv(frame: pl.DataFrame, name: str) -> None:
    """엑셀에서 한글이 깨지지 않도록 BOM이 붙은 UTF-8로 저장한다."""
    path = C.PROCESSED_DIR / name
    with open(path, "w", encoding="utf-8-sig", newline="") as f:
        f.write(frame.write_csv())


# ---------------------------------------------------------------- 1. 변수별 빈도표
TOP_K = 30  # 범주가 아주 많은 변수는 상위 30개만 보이고 나머지는 합친다
freq_summary = {}
values = {}
MAX_FULL, TOP_VALUES = 60, 40
for col in COLS:
    if col == "age":
        continue
    vc = (
        df.group_by(col).len()
        .sort("len", descending=True)
        .with_columns((pl.col("len") / N * 100).round(2).alias("pct"))
        .rename({"len": "count"})
    )
    n_cat = vc.height
    # 스키마 표에 '어떤 값이 들어 있는지' 보여 주기 위한 목록. 범주가 60개 이하면 전부, 많으면 상위 40개만 담는다.
    values[col] = vc.head(n_cat if n_cat <= MAX_FULL else TOP_VALUES).to_dicts()
    if n_cat > TOP_K:
        rest = vc.slice(TOP_K)
        vc = pl.concat([
            vc.head(TOP_K),
            pl.DataFrame({col: [f"(나머지 {rest.height}개)"], "count": [rest["count"].sum()],
                          "pct": [round(rest["count"].sum() / N * 100, 2)]},
                         schema=vc.schema),
        ])
    save_csv(vc, f"freq_{col}.csv")
    freq_summary[col] = {"n_categories": n_cat, "top": vc.head(12).to_dicts()}
    print(f"{col:18s} 범주 {n_cat:5d}개  1위: {vc[col][0]} ({vc['pct'][0]}%)")
summary["freq"] = freq_summary
summary["values"] = values

# ---------------------------------------------------------------- 2. 나이
age = df["age"]
summary["age"] = {
    "mean": round(age.mean(), 2), "std": round(age.std(), 2), "min": int(age.min()),
    "q25": age.quantile(0.25), "median": age.median(), "q75": age.quantile(0.75), "max": int(age.max()),
}
print("나이:", summary["age"])

# 인구 피라미드: 5세 구간별로 남자는 왼쪽(음수), 여자는 오른쪽(양수)에 그린다
pyr = (
    df.with_columns(C.age_band_expr(5))
    .group_by("age_band", "sex").len()
    .pivot(on="sex", index="age_band", values="len")
    .sort("age_band")
    .fill_null(0)
)
save_csv(pyr, "age_pyramid.csv")
bands = pyr["age_band"].to_list()
male = [v / N * 100 for v in pyr["남자"].to_list()]
female = [v / N * 100 for v in pyr["여자"].to_list()]
fig, ax = plt.subplots(figsize=(8, 6))
ax.barh(bands, [-m for m in male], color=C.COLOR_1, height=0.8, label="남자")
ax.barh(bands, female, color=C.COLOR_2, height=0.8, label="여자")
ax.axvline(0, color=C.COLOR_MUTED, linewidth=0.8)
ax.set_xlabel("전체 인구 중 비율 (%)")
ax.set_title("성별·연령 인구 피라미드", loc="left", fontweight="bold")
ticks = ax.get_xticks()
ax.set_xticks(ticks)
ax.set_xticklabels([f"{abs(t):.0f}" for t in ticks])  # 왼쪽도 양수로 표시한다
ax.grid(axis="y", visible=False)
ax.legend(frameon=False, loc="upper right")
fig.savefig(C.FIG_DIR / "age_pyramid.png")
plt.close(fig)


# ---------------------------------------------------------------- 3. 교차표
def crosstab(row: str, col: str, frame: pl.DataFrame, top_cols: int = 0) -> pl.DataFrame:
    """행 기준 비율(%) 교차표. 각 행의 합이 100이 된다. '이 나이대에서 혼인 상태가 어떻게 나뉘나'를 읽는 표다."""
    ct = frame.group_by(row, col).len()
    if top_cols:  # 열 범주가 너무 많으면 상위 몇 개만 남기고 나머지는 '기타'로 합친다
        keep = frame.group_by(col).len().sort("len", descending=True).head(top_cols)[col].to_list()
        ct = ct.with_columns(
            pl.when(pl.col(col).is_in(keep)).then(pl.col(col)).otherwise(pl.lit("기타")).alias(col)
        ).group_by(row, col).agg(pl.col("len").sum())
    wide = ct.pivot(on=col, index=row, values="len").fill_null(0).sort(row)
    value_cols = [c for c in wide.columns if c != row]
    total = wide.select(pl.sum_horizontal(value_cols)).to_series()
    return wide.with_columns([(pl.col(c) / total * 100).round(1) for c in value_cols])


def heatmap(table: pl.DataFrame, row: str, title: str, fname: str) -> None:
    """교차표를 색의 진하기로 보여 준다. 한 가지 색의 밝기만 쓴다(값이 클수록 진하다)."""
    from matplotlib.colors import LinearSegmentedColormap

    cmap = LinearSegmentedColormap.from_list("seq", [C.COLOR_SURFACE, C.COLOR_SEQ_LIGHT, C.COLOR_SEQ_DARK])
    cols = [c for c in table.columns if c != row]
    data = table.select(cols).to_numpy()
    fig, ax = plt.subplots(figsize=(max(7, len(cols) * 1.15), max(3.5, table.height * 0.36)))
    im = ax.imshow(data, cmap=cmap, aspect="auto", vmin=0, vmax=100)
    ax.set_xticks(range(len(cols)))
    ax.set_xticklabels(cols, rotation=30, ha="right")
    ax.set_yticks(range(table.height))
    ax.set_yticklabels(table[row].to_list())
    ax.grid(False)
    for i in range(data.shape[0]):
        for j in range(data.shape[1]):
            v = data[i, j]
            if v >= 1:  # 1% 미만은 숫자를 생략해 표를 덜 복잡하게 한다
                ax.text(j, i, f"{v:.0f}", ha="center", va="center", fontsize=8,
                        color="white" if v > 55 else C.COLOR_INK)
    ax.set_title(title + "  (행 기준 %, 각 행의 합 = 100)", loc="left", fontweight="bold")
    fig.colorbar(im, ax=ax, fraction=0.025, pad=0.02)
    fig.savefig(C.FIG_DIR / fname)
    plt.close(fig)


banded = df.with_columns(C.age_band_expr(5))
ct_age_marital = crosstab("age_band", "marital_status", banded)
ct_age_edu = crosstab("age_band", "education_level", banded)
# 전공은 대학 이상만 의미가 있으므로 '해당없음'은 뺀다
ct_field_sex = crosstab("bachelors_field", "sex", df.filter(pl.col("bachelors_field") != "해당없음"))
ct_family_housing = crosstab("family_type", "housing_type", df)
# 가구유형은 39개라 많은 순으로 12개만 그림에 쓴다
top_family = df.group_by("family_type").len().sort("len", descending=True).head(12)["family_type"].to_list()
ct_family_housing_top = ct_family_housing.filter(pl.col("family_type").is_in(top_family))

for table, name in [
    (ct_age_marital, "ct_age_marital.csv"), (ct_age_edu, "ct_age_education.csv"),
    (ct_field_sex, "ct_field_sex.csv"), (ct_family_housing, "ct_family_housing.csv"),
]:
    save_csv(table, name)

EDU_ORDER = ["무학", "초등학교", "중학교", "고등학교", "2~3년제 전문대학", "4년제 대학교", "대학원"]
MARITAL_ORDER = ["미혼", "배우자있음", "이혼", "사별"]
heatmap(ct_age_marital.select(["age_band"] + MARITAL_ORDER), "age_band", "연령대별 혼인 상태", "ct_age_marital.png")
heatmap(ct_age_edu.select(["age_band"] + EDU_ORDER), "age_band", "연령대별 학력", "ct_age_education.png")
heatmap(ct_field_sex.sort("남자", descending=True), "bachelors_field", "전공별 성별 구성", "ct_field_sex.png")
heatmap(ct_family_housing_top, "family_type", "가구 유형별 주거 형태 (상위 12개 가구 유형)", "ct_family_housing.png")

# 주거형태가 가구 유형과 얼마나 무관한지 숫자 하나로: 가구 유형별 아파트 비율의 최솟값과 최댓값
apt = ct_family_housing_top["아파트"]
summary["housing_by_family"] = {"apt_pct_min": apt.min(), "apt_pct_max": apt.max(),
                                "overall_apt_pct": round(df.filter(pl.col("housing_type") == "아파트").height / N * 100, 1)}

# ---------------------------------------------------------------- 4. 선행 감사가 지적한 약점 확인
unemployed = df.filter(pl.col("occupation") == "무직").height / N * 100
# 병역: '현역'인 사람들의 직업이 무엇인지, 그리고 그 직업을 가진 사람 중 현역이 아닌 사람이 있는지 본다
active = df.filter(pl.col("military_status") == "현역")
active_occ = active.group_by("occupation").len().sort("len", descending=True)
occ_of_active = active_occ["occupation"].to_list()
same_occ_not_active = df.filter(pl.col("occupation").is_in(occ_of_active) & (pl.col("military_status") != "현역")).height
summary["audit_check"] = {
    "unemployed_pct": round(unemployed, 1),
    "active_duty_n": active.height,
    "active_duty_pct": round(active.height / N * 100, 2),
    "active_duty_occupations": active_occ.head(10).to_dicts(),
    "n_occupations_of_active": len(occ_of_active),
    "same_occupation_but_not_active": same_occ_not_active,
    "active_sex": active.group_by("sex").len().to_dicts(),
}
save_csv(active_occ, "active_duty_occupations.csv")
print("무직 비율:", summary["audit_check"]["unemployed_pct"], "%")
print("현역의 직업 종류:", len(occ_of_active), "/ 같은 직업인데 현역이 아닌 행:", same_occ_not_active)

# ---------------------------------------------------------------- 5. 직업 상위 막대 그림 (무직 제외)
occ = (df.filter(pl.col("occupation") != "무직").group_by("occupation").len()
       .sort("len", descending=True).head(20).sort("len"))
fig, ax = plt.subplots(figsize=(8, 6.5))
ax.barh(occ["occupation"].to_list(), [v / N * 100 for v in occ["len"].to_list()], color=C.COLOR_1, height=0.7)
ax.set_xlabel("전체 행 중 비율 (%)")
ax.set_title("직업 상위 20개 ('무직' 제외)", loc="left", fontweight="bold")
ax.grid(axis="y", visible=False)
fig.savefig(C.FIG_DIR / "occupation_top20.png")
plt.close(fig)

# 시도별 분포
prov = df.group_by("province").len().sort("len")
fig, ax = plt.subplots(figsize=(8, 5.5))
ax.barh(prov["province"].to_list(), [v / N * 100 for v in prov["len"].to_list()], color=C.COLOR_1, height=0.7)
ax.set_xlabel("전체 행 중 비율 (%)")
ax.set_title("시도별 분포", loc="left", fontweight="bold")
ax.grid(axis="y", visible=False)
fig.savefig(C.FIG_DIR / "province.png")
plt.close(fig)

out = C.PROCESSED_DIR / "summary_structured.json"
out.write_text(json.dumps(summary, ensure_ascii=False, indent=2, default=str), encoding="utf-8")
print("저장:", out)

"""4단계: LLM이 쓴 13개 텍스트 필드의 기초통계.

만드는 것
- 필드별 길이(글자 수, 어절 수)와 완전 중복률
- 필드별 첫 어절·마지막 어절 상위 20개 (LLM 문장의 반복 틀을 보기 위한 것)
- 리스트 필드 2개의 항목 수 분포와 상위 50개 항목
- 이름이 어디에 어떤 형태로 들어 있는지, 성씨 상위 10개
- 인구통계 변수(지역, 나이대)가 텍스트에 글자 그대로 언급되는 비율
- 26개 필드 전체를 보여 주는 예시 5건
"""
import json

import polars as pl

import config as C

plt = C.setup_matplotlib()

PERSONA_FIELDS = [
    "persona", "professional_persona", "sports_persona", "arts_persona",
    "travel_persona", "culinary_persona", "family_persona",
]
ATTR_FIELDS = ["cultural_background", "skills_and_expertise", "hobbies_and_interests", "career_goals_and_ambitions"]
LIST_FIELDS = ["skills_and_expertise_list", "hobbies_and_interests_list"]

lf = pl.scan_parquet(C.parquet_glob())
N = lf.select(pl.len()).collect().item()
print("행 수:", N)
summary = {"n_rows": N}

# 페르소나 문장은 "홍길동 씨는 ..."으로 시작한다. 이름 부분을 떼어 내는 정규식이다.
NAME_PREFIX = r"^\S+\s씨[는은의가와]?\s*"

# ---------------------------------------------------------------- 1. 길이, 중복, 반복 틀
length_rows = []
patterns = {}
for col in PERSONA_FIELDS + ATTR_FIELDS:
    s = lf.select(col).collect().to_series()  # 한 번에 한 컬럼만 메모리에 올린다
    chars = s.str.len_chars()
    words = s.str.split(" ").list.len()  # 어절 = 띄어쓰기로 나눈 덩어리
    n_dup = N - s.n_unique()  # 글자 하나까지 똑같은 문장이 몇 개나 겹치나
    length_rows.append({
        "field": col,
        "chars_mean": round(chars.mean(), 1), "chars_median": chars.median(),
        "chars_p05": chars.quantile(0.05), "chars_p95": chars.quantile(0.95),
        "words_mean": round(words.mean(), 1), "words_median": words.median(),
        "exact_dup_pct": round(n_dup / N * 100, 3),
    })
    body = s.str.replace(NAME_PREFIX, "")  # 이름을 뗀 본문
    first = body.str.extract(r"^(\S+)", 1).value_counts(sort=True).head(20)
    last = s.str.extract(r"(\S+)\s*$", 1).value_counts(sort=True).head(20)
    patterns[col] = {
        "first": [{"token": r[0], "pct": round(r[1] / N * 100, 2)} for r in first.iter_rows()],
        "last": [{"token": r[0], "pct": round(r[1] / N * 100, 2)} for r in last.iter_rows()],
    }
    print(f"{col:28s} 평균 {length_rows[-1]['chars_mean']:6.1f}자  중복 {length_rows[-1]['exact_dup_pct']}%  "
          f"가장 흔한 끝 어절: {patterns[col]['last'][0]}")
    del s, chars, words, body

summary["length"] = length_rows
summary["patterns"] = patterns
pl.DataFrame(length_rows).write_csv(C.PROCESSED_DIR / "text_length.csv", include_bom=True)

# 필드별 평균 글자 수 막대 그림
lr = sorted(length_rows, key=lambda r: r["chars_mean"])
fig, ax = plt.subplots(figsize=(8, 5))
ax.barh([r["field"] for r in lr], [r["chars_mean"] for r in lr], color=C.COLOR_1, height=0.7)
for i, r in enumerate(lr):
    ax.text(r["chars_mean"] + 2, i, f"{r['chars_mean']:.0f}", va="center", fontsize=9, color=C.COLOR_INK)
ax.set_xlabel("평균 글자 수")
ax.set_title("텍스트 필드별 평균 길이", loc="left", fontweight="bold")
ax.grid(axis="y", visible=False)
fig.savefig(C.FIG_DIR / "text_length.png")
plt.close(fig)

# ---------------------------------------------------------------- 2. 리스트 필드
# 값은 "['항목1', '항목2']" 모양의 문자열이다. 따옴표 안의 내용을 모두 뽑아 진짜 리스트로 바꾼다.
list_summary = {}
for col in LIST_FIELDS:
    s = lf.select(col).collect().to_series()
    items = s.str.extract_all(r"'([^']+)'").list.eval(pl.element().str.strip_chars("'"))
    n_items = items.list.len()
    dist = n_items.value_counts().sort(col)
    flat = items.explode().drop_nulls()
    top = flat.value_counts(sort=True).head(50)
    list_summary[col] = {
        "items_mean": round(n_items.mean(), 2),
        "items_dist": [{"n": r[0], "pct": round(r[1] / N * 100, 1)} for r in dist.iter_rows()],
        "n_items_total": flat.len(),
        "n_items_unique": flat.n_unique(),
        "top50": [{"item": r[0], "count": r[1], "pct_rows": round(r[1] / N * 100, 2)} for r in top.iter_rows()],
    }
    top.write_csv(C.PROCESSED_DIR / f"top_items_{col}.csv", include_bom=True)
    print(f"{col}: 행당 평균 {list_summary[col]['items_mean']}개, 고유 항목 {flat.n_unique():,}개 / 전체 {flat.len():,}개")
    del s, items, flat
summary["lists"] = list_summary

# ---------------------------------------------------------------- 3. 이름
# 이름은 독립된 필드가 없다. persona 문장 맨 앞의 "OOO 씨"에서 뽑는다.
base = lf.select("persona", "province", "district", "age", "sex").collect()
names = base["persona"].str.extract(r"^(\S+)\s씨", 1)
has_name = names.is_not_null().sum()
TWO_CHAR_SURNAMES = ["남궁", "황보", "제갈", "사공", "선우", "서문", "독고", "동방"]  # 두 글자 성씨
surname = (
    pl.when(names.str.slice(0, 2).is_in(TWO_CHAR_SURNAMES) & (names.str.len_chars() > 2))
    .then(names.str.slice(0, 2)).otherwise(names.str.slice(0, 1))
)
sn = base.select(surname.alias("surname")).drop_nulls().to_series()
sn_top = sn.value_counts(sort=True).head(10)
full_top = names.drop_nulls().value_counts(sort=True).head(10)
CARD_SURNAME = {"김": 21.5, "이": 14.7, "박": 8.5, "정": 4.8, "최": 4.7}  # 카드에 적힌 값
summary["names"] = {
    "rows_with_name_pct": round(has_name / N * 100, 2),
    "n_unique_full_names": names.n_unique(),
    "n_unique_surnames": sn.n_unique(),
    "surname_top10": [{"surname": r[0], "pct": round(r[1] / sn.len() * 100, 2), "card_pct": CARD_SURNAME.get(r[0])}
                      for r in sn_top.iter_rows()],
    "full_name_top10": [{"name": r[0], "count": r[1]} for r in full_top.iter_rows()],
}
print("이름이 뽑힌 행:", summary["names"]["rows_with_name_pct"], "%  고유 이름:", names.n_unique())
print("성씨 상위:", [(d["surname"], d["pct"]) for d in summary["names"]["surname_top10"][:5]])

# ---------------------------------------------------------------- 4. 인구통계 변수가 텍스트에 그대로 나오는 비율
# 검증이 아니라 '데이터가 어떻게 생겼나'를 보는 단순 집계다. 언급이 없다고 틀린 것은 아니다.
base = base.with_columns(
    pl.col("district").str.split("-").list.last().alias("district_short"),  # "광주-서구" -> "서구"
    (pl.col("age") // 10 * 10).cast(pl.String).add("대").alias("decade"),  # 74 -> "70대"
)
mention = {}
for col in ["persona", "professional_persona", "family_persona", "cultural_background"]:
    t = base["persona"] if col == "persona" else lf.select(col).collect().to_series()
    tmp = base.select("province", "district_short", "decade").with_columns(t.alias("t"))
    mention[col] = {
        "province_pct": round(tmp.select(pl.col("t").str.contains(pl.col("province"), literal=True).sum()).item() / N * 100, 1),
        "district_pct": round(tmp.select(pl.col("t").str.contains(pl.col("district_short"), literal=True).sum()).item() / N * 100, 1),
        "decade_pct": round(tmp.select(pl.col("t").str.contains(pl.col("decade"), literal=True).sum()).item() / N * 100, 1),
    }
    print("언급률", col, mention[col])
summary["mention"] = mention

# ---------------------------------------------------------------- 5. 전체 행 예시 5건
# 나이대와 성별이 서로 다른 다섯 사람을 고른다. 시드를 고정해 매번 같은 사람이 나온다.
PICKS = [("여자", 19, 29), ("남자", 30, 44), ("여자", 45, 59), ("남자", 60, 74), ("여자", 75, 99)]
first_file = sorted(C.DATA_DIR.rglob("*.parquet"))[0]
one = pl.read_parquet(first_file)  # 예시는 첫 파일에서만 고르면 충분하다
examples = []
for sex, lo, hi in PICKS:
    pool = one.filter((pl.col("sex") == sex) & pl.col("age").is_between(lo, hi) & (pl.col("occupation") != "무직"))
    examples.append(pool.sample(1, seed=C.SEED).row(0, named=True))
summary["examples"] = examples

out = C.PROCESSED_DIR / "summary_text.json"
out.write_text(json.dumps(summary, ensure_ascii=False, indent=2, default=str), encoding="utf-8")
print("저장:", out)

"""5단계(탐색): 이 데이터로 '수준별 직무 페르소나'(신입-대리-과장 식)를 만들 수 있을까?

확인하려는 것
  1) 직급이나 경력 연수를 나타내는 변수가 있는가?           -> 없다. 텍스트에 흔적이 있는지 센다.
  2) 한 직업 안에 나이가 고르게 퍼져 있는가?                -> 나이를 경력의 대리 지표로 쓸 수 있는지 본다.
  3) 나이가 달라지면 직업 서술의 숙련도 표현도 달라지는가?  -> 연령대별로 숙련도를 나타내는 낱말의 빈도를 센다.
"""
import json

import polars as pl

import config as C

plt = C.setup_matplotlib()

COLS = ["age", "sex", "education_level", "bachelors_field", "occupation",
        "professional_persona", "skills_and_expertise", "skills_and_expertise_list", "career_goals_and_ambitions"]
df = pl.scan_parquet(C.parquet_glob()).select(COLS).collect()
N = df.height
work = df.filter(~pl.col("occupation").str.contains("무직|구직"))  # 현재 일하는 사람만
W = work.height
res = {"n_rows": N, "n_employed": W, "n_occupations": df["occupation"].n_unique()}
print("전체", N, "취업자", W, "직업 종류", res["n_occupations"])

# ---------------------------------------------------------------- 1. 직급·연차 표현이 텍스트에 얼마나 나오나
RANK_WORDS = ["신입", "사원", "주임", "대리", "과장", "차장", "부장", "팀장", "임원", "주니어", "시니어"]
LEVEL_WORDS = ["베테랑", "수십 년", "노하우", "후배", "막내", "초년", "새내기", "배우", "은퇴"]
YEARS_PATTERN = r"\d{1,2}\s*년\s*(?:차|간|째|동안|이상|넘게|가까이)"  # "15년간", "3년 차" 같은 숫자 경력


def pct_contains(frame, col, pattern):
    return round(frame.filter(pl.col(col).str.contains(pattern)).height / frame.height * 100, 2)


res["rank_words"] = [
    {"word": w, "professional_pct": pct_contains(work, "professional_persona", w),
     "career_goals_pct": pct_contains(work, "career_goals_and_ambitions", w)} for w in RANK_WORDS + LEVEL_WORDS
]
any_rank = "|".join(RANK_WORDS)
res["any_rank_word_pct"] = pct_contains(work, "professional_persona", any_rank)
res["numeric_years_pct"] = round(
    work.filter(pl.col("professional_persona").str.contains(YEARS_PATTERN)
                | pl.col("skills_and_expertise").str.contains(YEARS_PATTERN)).height / W * 100, 2)
print("직급 낱말이 하나라도 있는 직업 페르소나:", res["any_rank_word_pct"], "%")
print("숫자로 된 경력 연수 표현:", res["numeric_years_pct"], "%")

# ---------------------------------------------------------------- 2. 한 직군 안의 나이 분포 (예: 프로그래머)
GROUPS = {
    "프로그래머": "프로그래머",
    "간호사": "간호사",
    "회계·경리 사무원": "회계 사무원|경리 사무원",
    "초·중등 교사": "초등학교 교사|중·고등학교 교사",
}
work = work.with_columns(((pl.col("age") // 10) * 10).alias("decade"))
res["groups"] = {}
for label, pat in GROUPS.items():
    g = work.filter(pl.col("occupation").str.contains(pat))
    dist = g.group_by("decade").len().sort("decade")
    res["groups"][label] = {
        "n": g.height,
        "occupations": g.group_by("occupation").len().sort("len", descending=True).head(8).to_dicts(),
        "age_dist": [{"decade": r[0], "n": r[1], "pct": round(r[1] / g.height * 100, 1)} for r in dist.iter_rows()],
    }
    print(label, g.height, "명", [(d["decade"], d["pct"]) for d in res["groups"][label]["age_dist"]])

# ---------------------------------------------------------------- 3. 나이에 따라 숙련도 표현이 달라지나 (전체 취업자)
SIGNALS = {
    "초기 경력 표현 (신입·막내·주니어·새내기·초년)": "신입|막내|주니어|새내기|초년|사회초년",
    "숙련 표현 (베테랑·노하우·수십 년·시니어)": "베테랑|노하우|수십 년|시니어|오랜",
    "관리·전수 표현 (후배·팀장·관리·이끌)": "후배|팀장|관리자|이끌|총괄|임원",
}
by_age = []
for dec in sorted(work["decade"].unique().to_list()):
    g = work.filter(pl.col("decade") == dec)
    if g.height < 200:
        continue
    row = {"decade": dec, "n": g.height}
    for name, pat in SIGNALS.items():
        row[name] = pct_contains(g, "professional_persona", pat)
    by_age.append(row)
res["signals_by_age"] = by_age
for r in by_age:
    print(r)

fig, ax = plt.subplots(figsize=(8, 4.6))
xs = [f"{r['decade']}대" for r in by_age]
styles = [(C.COLOR_1, "-", "o"), (C.COLOR_2, "--", "s"), ("#1baf7a", ":", "^")]
for (name, _), (color, ls, mk) in zip(SIGNALS.items(), styles):
    ax.plot(xs, [r[name] for r in by_age], color=color, linestyle=ls, marker=mk, linewidth=2, markersize=6, label=name)
ax.set_ylabel("해당 낱말이 나오는 직업 페르소나 비율 (%)")
ax.set_title("나이에 따라 직업 서술의 숙련도 표현이 달라진다", loc="left", fontweight="bold")
ax.legend(frameon=False, fontsize=9)
fig.savefig(C.FIG_DIR / "job_level_signals.png")
plt.close(fig)

# ---------------------------------------------------------------- 4. 프로그래머의 연령대별 예시
prog = work.filter(pl.col("occupation").str.contains("프로그래머"))
examples = []
for dec in (20, 30, 40, 50):
    pool = prog.filter(pl.col("decade") == dec)
    if pool.height:
        r = pool.sample(1, seed=C.SEED).row(0, named=True)
        examples.append({k: r[k] for k in ["age", "sex", "education_level", "occupation", "professional_persona",
                                           "skills_and_expertise_list", "career_goals_and_ambitions"]})
res["programmer_examples"] = examples

out = C.PROCESSED_DIR / "summary_job_level.json"
out.write_text(json.dumps(res, ensure_ascii=False, indent=2, default=str), encoding="utf-8")
print("저장:", out)