# -*- coding: utf-8 -*-
"""산업별 노동수요(취업자) 전망 — 취업계수 방식, 계산 함수 모음.

노트북 01(입수)·02(분석)·03(결과)이 이 모듈의 함수만 호출한다. 경로·상수는 config.py에 둔다.

계산 구조
  1. 취업자     : 경제활동인구조사 산업별 취업자(ECOS 901Y027, 분기, 계절조정), 한국표준산업분류 대분류 A–S(19개)
                  연간 값 = 4개 분기 평균(천 명)
  2. 실질부가가치: 경제활동별 GDP(ECOS 200Y104, 계절조정·실질·분기), 36개 산업, 연간 값 = 4개 분기 합계(십억 원)
  3. 취업계수   : c(s,y) = 취업자(s,y) / Σ_{i∈s} 실질부가가치(i,y)  — 대분류 s에 속한 36개 산업의 부가가치 합계 기준
  4. 기준 수준  : b_s = Σ_{y=2023}^{2025} 취업자(s,y) / Σ_{y=2023}^{2025} 실질부가가치(s,y), 기준 연도는 중앙 연도 2024
  5. 추세       : g_s = ln c(s,y)를 연도에 회귀한 최소제곱 기울기(2013–2025), 연간 로그변화율
  6. 계수 경로  : c(s,t) = b_s · exp(g_s · (t − 2024)), t = 2026…2035
  7. 취업자 전망: L(i,t) = c(s(i),t) × 실질부가가치_시나리오(i,t), 시나리오 S1–S5
                  주 산출물(정비판)은 보고서1 5장 2026-10 정비판 최종 전망(연간)을, 이전 판(2026-10-08 판) 재현은 정비 이전 최종 전망(연간)을,
                  최초 판(v0) 재현은 정합화 이전 분기 전망(4개 분기 합계)을 투입한다. 계수 경로는 세 판이 같다.
  8. 2025년 실적: 산업별 취업자 = 대분류 취업자 × 대분류 안의 실질부가가치 비중(대분류가 산업 하나이면 실측치와 같다)
"""
from __future__ import annotations

import hashlib
import json
import os
import io
import urllib.error
import urllib.request
import zipfile

import numpy as np
import pandas as pd

SECTIONS = list("ABCDEFGHIJKLMNOPQRS")          # 19개 대분류(T 가구 내 고용·U 국제기관은 대응 부가가치가 없어 제외)
SCENARIOS = ["S1", "S2", "S3", "S4", "S5"]
SCENARIO_LABELS = {"S1": "S1 기준", "S2": "S2 제약+위기", "S3": "S3 제약완화", "S4": "S4 기술도약", "S5": "S5 지정학"}

# ECOS 901Y027(경제활동인구조사) 산업 대분류 항목 코드와 계열 구분 코드
ECOS_EMP_TABLE = "901Y027"
ECOS_EMP_ITEMS = {
    "A": "I61BAAA", "B": "I61BAABA", "C": "I61BAABB", "D": "I61BAABC", "E": "I61BAABD", "F": "I61BAABE",
    "G": "I61BAAEA", "H": "I61BAAEB", "I": "I61BAAEC", "J": "I61BAAED", "K": "I61BAAEE", "L": "I61BAAEF",
    "M": "I61BAAEG", "N": "I61BAAEH", "O": "I61BAAEI", "P": "I61BAAEJ", "Q": "I61BAAEK", "R": "I61BAAEL",
    "S": "I61BAAEM", "T": "I61BAAEN", "U": "I61BAAEO", "전체": "I61BA",
}
ECOS_ADJ = {"계절조정": "I28B", "원계열": "I28A"}
ECOS_VA_TABLE = "200Y104"                        # 경제활동별 GDP 및 GNI(계절조정, 실질, 분기)
ECOS_URL = "https://ecos.bok.or.kr/api/StatisticSearch/{key}/json/kr/1/100000/{table}/{cycle}/{start}/{end}{items}"


# ── 공통 ──────────────────────────────────────────────────────────────
def sha256(path: str) -> str:
    h = hashlib.sha256()
    with open(path, "rb") as f:
        for chunk in iter(lambda: f.read(1 << 20), b""):
            h.update(chunk)
    return h.hexdigest()


def sha256_bytes(b: bytes) -> str:
    return hashlib.sha256(b).hexdigest()


def year_of(quarter: str) -> int:
    return int(str(quarter)[:4])


def cagr(x0: float, x1: float, years: int) -> float:
    """연평균 증가율(%)."""
    return 100.0 * ((x1 / x0) ** (1.0 / years) - 1.0)


# ── 1. 입수: 공개 API·포털 ─────────────────────────────────────────────
def _ecos_rows(key: str, table: str, cycle: str, start: str, end: str, items: tuple = ()) -> list[dict]:
    """ECOS StatisticSearch 호출. 오류 메시지에 인증키가 섞이지 않도록 URL은 출력하지 않는다."""
    url = ECOS_URL.format(key=key, table=table, cycle=cycle, start=start, end=end,
                          items="".join("/" + i for i in items))
    try:
        with urllib.request.urlopen(url, timeout=120) as r:
            body = json.loads(r.read().decode("utf-8"))
    except Exception as e:  # noqa: BLE001
        raise RuntimeError(f"ECOS {table} 호출 실패({type(e).__name__})") from None
    if "StatisticSearch" not in body:
        msg = body.get("RESULT", {}).get("MESSAGE", "응답 형식 오류")
        raise RuntimeError(f"ECOS {table} 응답 오류: {msg}")
    return body["StatisticSearch"]["row"]


def fetch_employment(key: str, start: str = "2013Q1", end: str = "2025Q4") -> pd.DataFrame:
    """ECOS 901Y027 산업 대분류 취업자(분기, 계절조정·원계열) → long 표."""
    rows = []
    for adj, adj_code in ECOS_ADJ.items():
        for sec, code in ECOS_EMP_ITEMS.items():
            for r in _ecos_rows(key, ECOS_EMP_TABLE, "Q", start, end, (code, adj_code)):
                if r.get("DATA_VALUE") in (None, "", "-"):
                    continue
                rows.append({"분기": r["TIME"], "대분류": sec, "구분": adj, "취업자_천명": float(r["DATA_VALUE"])})
    return pd.DataFrame(rows).sort_values(["구분", "대분류", "분기"]).reset_index(drop=True)


def fetch_va36(key: str, codes: list[str], start: str = "2013Q1", end: str = "2025Q4") -> pd.DataFrame:
    """ECOS 200Y104 36개 산업 실질부가가치(계절조정, 분기) → long 표."""
    want = set(map(str, codes))
    rows = []
    for r in _ecos_rows(key, ECOS_VA_TABLE, "Q", start, end):
        if r.get("ITEM_CODE1") in want and r.get("DATA_VALUE") not in (None, "", "-"):
            rows.append({"분기": r["TIME"], "산업코드": r["ITEM_CODE1"], "실질VA_십억원": float(r["DATA_VALUE"])})
    return pd.DataFrame(rows).sort_values(["산업코드", "분기"]).reset_index(drop=True)


def download(url: str, path: str) -> str:
    """공개 포털 파일을 내려받아 저장하고 SHA-256을 반환한다."""
    with urllib.request.urlopen(url, timeout=120) as r, open(path, "wb") as f:
        f.write(r.read())
    return sha256(path)


def try_download(url: str, path: str) -> tuple[str | None, str]:
    """내려받기를 시도하고 (SHA-256 또는 None, 상태 설명)을 반환한다. 실패해도 예외를 내지 않는다."""
    try:
        return download(url, path), "내려받음"
    except urllib.error.HTTPError as e:
        return None, f"HTTP {e.code}"
    except Exception as e:  # noqa: BLE001
        return None, f"실패({type(e).__name__})"


def bundle_member(zip_path: str, member: str) -> bytes:
    """통합 번들(zip)에서 파일 하나를 읽는다."""
    with zipfile.ZipFile(zip_path) as z:
        return z.read(member)


def reconciled_subset(raw: bytes) -> pd.DataFrame:
    """정합화 전망 원본(forecast_annual_36.csv)에서 S1–S5·2026–2035만 추린다(고정 입력과 같은 형식)."""
    r = pd.read_csv(io.BytesIO(raw))
    return r[r["시나리오"].isin(SCENARIOS) & r["연도"].between(2026, 2035)].reset_index(drop=True)


def reconciled_subset_bytes(raw: bytes) -> bytes:
    """원본 파일의 줄을 그대로 두고 S1–S5·2026–2035 행만 남긴 바이트열(머리글·BOM·줄 끝 문자 유지).

    고정 입력 va36_reconciled_annual.csv는 이 함수로 만든다. 값의 표기를 다시 쓰지 않으므로 원본과 문자 단위로 같다.
    """
    eol = b"\r\n" if b"\r\n" in raw else b"\n"
    lines = raw.split(eol)
    keep = [lines[0]]
    for ln in lines[1:]:
        if not ln:
            continue
        scen, year = ln.decode("utf-8").split(",", 2)[:2]
        if scen in SCENARIOS and 2026 <= int(year) <= 2035:
            keep.append(ln)
    return eol.join(keep) + eol


# ── 1. 입수: 파일 읽기 ─────────────────────────────────────────────────
def read_employment(path: str, adj: str = "계절조정") -> pd.DataFrame:
    """취업자 long 표 → 행=분기, 열=대분류(A–U, 전체), 값=천 명."""
    e = pd.read_csv(path, dtype={"분기": str})
    e = e[e["구분"] == adj]
    return e.pivot(index="분기", columns="대분류", values="취업자_천명").sort_index()


def annual_employment(emp_q: pd.DataFrame) -> pd.DataFrame:
    """분기 취업자 → 연간 값(4개 분기 평균). 4개 분기가 모두 있는 연도만 남긴다."""
    g = emp_q.groupby(emp_q.index.map(year_of))
    return g.mean()[g.count().min(axis=1) == 4].rename_axis("연도")


def read_va_hist(path: str) -> pd.DataFrame:
    """36개 산업 실질부가가치 long 표 → 행=분기, 열=산업명, 값=십억 원."""
    v = pd.read_csv(path, dtype={"분기": str, "산업코드": str})
    return v.pivot(index="분기", columns="산업명", values="실질VA_십억원").sort_index()


def annual_va(va_q: pd.DataFrame) -> pd.DataFrame:
    """분기 실질부가가치 → 연간 값(4개 분기 합계). 4개 분기가 모두 있는 연도만 남긴다."""
    g = va_q.groupby(va_q.index.map(year_of))
    return g.sum()[g.count().min(axis=1) == 4].rename_axis("연도")


def read_scenario_json(path: str) -> dict[str, pd.DataFrame]:
    """포털 「산업별 시나리오 전망」 36개 산업 분기 자료 → {시나리오: 행=분기, 열=산업}."""
    with open(path, encoding="utf-8") as f:
        d = json.load(f)
    return {s: pd.DataFrame(v, index=d["periods"]) for s, v in d["scenarios"].items()}


def scenario_annual(scen_q: dict[str, pd.DataFrame]) -> dict[str, pd.DataFrame]:
    return {s: annual_va(q) for s, q in scen_q.items()}


def read_reconciled(path: str) -> dict[str, pd.DataFrame]:
    """정합화 후 36개 산업 연간 전망(주 투입·이전 판 투입) → {시나리오: 행=연도, 열=산업}."""
    r = pd.read_csv(path)
    r = r[r["시나리오"].isin(SCENARIOS)]
    return {s: g.pivot(index="연도", columns="산업", values="실질VA_십억원").sort_index() for s, g in r.groupby("시나리오")}


# ── 2. 분석 ────────────────────────────────────────────────────────────
def section_va(va_annual: pd.DataFrame, sec_of: dict[str, str]) -> pd.DataFrame:
    """36개 산업 연간 부가가치 → 대분류 합계(행=연도, 열=대분류)."""
    missing = set(va_annual.columns) - set(sec_of)
    if missing:
        raise KeyError(f"대응표에 없는 산업: {sorted(missing)}")
    return va_annual.T.groupby(pd.Series(sec_of)).sum().T[SECTIONS]


def coefficients(emp_a: pd.DataFrame, va_sec: pd.DataFrame, years: range) -> pd.DataFrame:
    """대분류별 취업계수 c(s,y) = 취업자 / 실질부가가치 (천 명/십억 원)."""
    y = list(years)
    return emp_a.loc[y, SECTIONS] / va_sec.loc[y, SECTIONS]


def fit_params(emp_a: pd.DataFrame, va_sec: pd.DataFrame, base_years=(2023, 2025), trend_years=(2013, 2025),
               base: str = "ratio", trend: str = "ols") -> pd.DataFrame:
    """대분류별 기준 수준 b와 연간 로그변화율 g.

    base : "ratio" Σ취업자/Σ부가가치(기준 기간) · "mean" 연도별 계수의 산술평균 · "geo" 기하평균
    trend: "ols" ln c의 최소제곱 기울기 · "endpoint" (ln c_끝 − ln c_처음)/연수
    """
    b0, b1 = base_years
    t0, t1 = trend_years
    c = coefficients(emp_a, va_sec, range(min(b0, t0), max(b1, t1) + 1))
    rows = {}
    for s in SECTIONS:
        cb = c.loc[b0:b1, s]
        if base == "ratio":
            bv = emp_a.loc[b0:b1, s].sum() / va_sec.loc[b0:b1, s].sum()
        elif base == "mean":
            bv = cb.mean()
        elif base == "geo":
            bv = float(np.exp(np.log(cb).mean()))
        else:
            raise ValueError(base)
        lc = np.log(c.loc[t0:t1, s])
        x = lc.index.values.astype(float)
        if trend == "ols":
            slope, icpt = np.polyfit(x, lc.values, 1)
            resid = lc.values - (icpt + slope * x)
            r2 = 1 - resid.var() / lc.values.var()
        elif trend == "endpoint":
            slope, r2 = (lc.iloc[-1] - lc.iloc[0]) / (x[-1] - x[0]), np.nan
        else:
            raise ValueError(trend)
        rows[s] = {"기준수준_b": bv, "연간로그변화율_g": slope, "추세_R2": r2}
    return pd.DataFrame(rows).T


def coef_path(params: pd.DataFrame, years, anchor: int = 2024) -> pd.DataFrame:
    """c(s,t) = b_s · exp(g_s · (t − 기준연도))."""
    t = np.asarray(list(years), dtype=float)
    return pd.DataFrame({s: params.loc[s, "기준수준_b"] * np.exp(params.loc[s, "연간로그변화율_g"] * (t - anchor))
                         for s in params.index}, index=list(years))


def project(va_scen: dict[str, pd.DataFrame], path: pd.DataFrame, sec_of: dict[str, str]) -> dict[str, pd.DataFrame]:
    """산업별 취업자(천 명) = 소속 대분류의 취업계수 × 시나리오 실질부가가치."""
    out = {}
    for s, va in va_scen.items():
        yrs = list(path.index)
        cols = sorted(va.columns)
        out[s] = pd.DataFrame({i: path.loc[yrs, sec_of[i]].values * va.loc[yrs, i].values for i in cols}, index=yrs)
    return out


def path_level(params: pd.DataFrame, va_sec: pd.DataFrame, year: int, anchor: int = 2024) -> float:
    """계수 경로를 실적 연도에 적용한 19개 대분류 취업자 합계(천 명) = Σ_s c(s, year) × 실적 부가가치(s, year)."""
    c = coef_path(params, [year], anchor).loc[year]
    return float((c * va_sec.loc[year, c.index]).sum())


def demand_changes(tot: pd.DataFrame, base_level: float, base_year: int, periods) -> pd.DataFrame:
    """시나리오별 전산업 취업자 증감(천 명). tot은 행=연도(전망), 열=시나리오.

    기간의 시작 연도가 base_year이면 base_level(실적 또는 계수 경로 기준 값)을, 그 밖에는 전망값을 시작 값으로 쓴다.
    """
    rows = {}
    for a, b in periods:
        start = pd.Series(base_level, index=tot.columns) if a == base_year else tot.loc[a]
        rows[f"{a}→{b}"] = tot.loc[b] - start
    return pd.DataFrame(rows).T


def changes_from_base(tot: pd.DataFrame, emp_a: pd.DataFrame, va_annual: pd.DataFrame, params: pd.DataFrame,
                      sec_of: dict[str, str], periods, base_year: int = 2024, anchor: int = 2024):
    """인력수급 전망 비교용 전산업 취업자 증감(천 명). 기준 연도 값을 두 가지로 둔다.

    실적 기준      : 기준 연도의 19개 대분류 실적 취업자 합계
    계수 경로 기준 : 계수 경로의 기준 연도 값 × 기준 연도 실적 부가가치(대분류별)의 합계
    반환: (실적 기준 증감, 계수 경로 기준 증감, 실적 값, 계수 경로 기준 값)
    """
    sections = list(params.index)
    l_act = float(emp_a.loc[base_year, sections].sum())
    l_path = path_level(params, section_va(va_annual, sec_of), base_year, anchor)
    return (demand_changes(tot, l_act, base_year, periods), demand_changes(tot, l_path, base_year, periods), l_act, l_path)


def actual_by_industry(emp_a: pd.DataFrame, va_annual: pd.DataFrame, sec_of: dict[str, str], year: int) -> pd.Series:
    """실적 연도의 산업별 취업자(천 명): 대분류 취업자를 대분류 안의 실질부가가치 비중으로 나눈다."""
    va = va_annual.loc[year]
    sec_tot = va.groupby(pd.Series(sec_of)).sum()
    return pd.Series({i: emp_a.loc[year, sec_of[i]] * va[i] / sec_tot[sec_of[i]] for i in sorted(va.index)})


def to_long(proj: dict[str, pd.DataFrame]) -> pd.DataFrame:
    rows = [p.stack().rename("취업자_천명").reset_index().assign(시나리오=s) for s, p in proj.items()]
    d = pd.concat(rows).rename(columns={"level_0": "연도", "level_1": "산업"})
    return d[["시나리오", "연도", "산업", "취업자_천명"]].reset_index(drop=True)


def viewer_json(proj: dict[str, pd.DataFrame]) -> dict:
    """포털 뷰어 자료 형식: 산업 이름 정렬, 소수 첫째 자리 반올림, 시나리오 합계는 반올림 전 값을 더한 뒤 반올림."""
    years = [int(y) for y in next(iter(proj.values())).index]
    return {
        "years": years,
        "scenarios": {s: {i: [round(float(v), 1) for v in p[i].values] for i in sorted(p.columns)} for s, p in proj.items()},
        "scenarios_total": {s: [round(float(v), 1) for v in p.sum(axis=1).values] for s, p in proj.items()},
    }


def dumps_viewer(obj: dict) -> bytes:
    """포털 파일과 같은 직렬화(UTF-8, 공백 없는 구분자)."""
    return json.dumps(obj, ensure_ascii=False, separators=(",", ":")).encode("utf-8")


def same_structure(a: dict, b: dict) -> bool:
    """두 뷰어 자료의 키·연도·시나리오·산업 이름과 순서·배열 길이가 같은지(값은 비교하지 않는다)."""
    if list(a) != list(b) or a["years"] != b["years"]:
        return False
    if list(a["scenarios"]) != list(b["scenarios"]) or list(a["scenarios_total"]) != list(b["scenarios_total"]):
        return False
    for s in b["scenarios"]:
        if list(a["scenarios"][s]) != list(b["scenarios"][s]):
            return False
        if any(len(a["scenarios"][s][i]) != len(b["scenarios"][s][i]) for i in b["scenarios"][s]):
            return False
    return all(len(a["scenarios_total"][s]) == len(b["scenarios_total"][s]) for s in b["scenarios_total"])


def compare_viewer(calc: dict, pub: dict) -> dict:
    """계산 자료와 공표 자료의 구조·값 차이."""
    out = {"연도 일치": calc["years"] == pub["years"],
           "시나리오 일치": sorted(calc["scenarios"]) == sorted(pub["scenarios"]),
           "산업 일치": all(sorted(calc["scenarios"][s]) == sorted(pub["scenarios"][s]) for s in pub["scenarios"])}
    d_ind = [abs(a - b) for s in pub["scenarios"] for i in pub["scenarios"][s]
             for a, b in zip(calc["scenarios"][s][i], pub["scenarios"][s][i])]
    d_tot = [abs(a - b) for s in pub["scenarios_total"] for a, b in zip(calc["scenarios_total"][s], pub["scenarios_total"][s])]
    out.update({"산업값 개수": len(d_ind), "산업값 최대절대오차": max(d_ind),
                "합계 개수": len(d_tot), "합계 최대절대오차": max(d_tot)})
    return out


def spec_grid(emp_by_adj: dict[str, pd.DataFrame], va_sec: pd.DataFrame, va_scen: dict, sec_of: dict, pub: dict,
              full: bool = False) -> pd.DataFrame:
    """재구성 근거: 대안 명세별로 공표 자료와의 차이를 계산한다(FULL이면 기준·추세 기간까지 넓힌다)."""
    bases = ["ratio", "mean", "geo"] if full else ["ratio", "mean"]
    windows = [(2013, 2025), (2014, 2025), (2015, 2025), (2013, 2024)] if full else [(2013, 2025)]
    years = pub["years"]
    rows = []
    for adj, emp_a in emp_by_adj.items():
        for base in bases:
            for trend in ["ols", "endpoint"]:
                for w in windows:
                    par = fit_params(emp_a, va_sec, (2023, 2025), w, base, trend)
                    for anchor in [2024, 2025]:
                        calc = viewer_json(project(va_scen, coef_path(par, years, anchor), sec_of))
                        d = np.array([abs(a - b) for s in pub["scenarios"] for i in pub["scenarios"][s]
                                      for a, b in zip(calc["scenarios"][s][i], pub["scenarios"][s][i])])
                        rows.append({"취업자 계열": adj, "기준 수준": base, "추세 추정": trend,
                                     "추세 기간": f"{w[0]}–{w[1]}", "기준연도": anchor,
                                     "최대절대오차(천명)": d.max(), "일치 비율": float((d < 0.05).mean())})
    return pd.DataFrame(rows).sort_values(["최대절대오차(천명)", "일치 비율"], ascending=[True, False]).reset_index(drop=True)