src/streamlit_app.py

from pathlib import Path

import pandas as pd
import streamlit as st

st.set_page_config(
    page_title="계량기 교체 후보 검토 앱",
    layout="wide",
)

DATA_FILE = Path("/04_계량기_교체대상_자동추출_예시데이터.xlsx")
SHEET_NAME = "계량기_교체대상"

REQUIRED_COLUMNS = [
    "계량기ID",
    "고객ID",
    "서비스센터",
    "지역",
    "설치일자",
    "사용연수",
    "배터리잔량_pct",
    "최근오류건수",
    "통신상태",
    "부품재고여부",
    "참고_교체점수",
    "참고_교체등급",
    "교체사유",
    "권장조치",
]


@st.cache_data
def load_default_excel(file_path: str, sheet_name: str) -> pd.DataFrame:
    return pd.read_excel(file_path, sheet_name=sheet_name)


def load_excel(uploaded_file, default_file: Path, sheet_name: str) -> pd.DataFrame:
    if uploaded_file is not None:
        return pd.read_excel(uploaded_file, sheet_name=sheet_name)

    if default_file.exists():
        return load_default_excel(str(default_file), sheet_name)

    return pd.DataFrame()


def prepare_data(df: pd.DataFrame) -> pd.DataFrame:
    df = df.copy()

    df["설치일자"] = pd.to_datetime(df["설치일자"], errors="coerce")

    number_columns = [
        "사용연수",
        "배터리잔량_pct",
        "최근오류건수",
        "참고_교체점수",
    ]

    for col in number_columns:
        df[col] = pd.to_numeric(df[col], errors="coerce")

    return df


def get_missing_columns(df: pd.DataFrame) -> list:
    return [col for col in REQUIRED_COLUMNS if col not in df.columns]


def to_csv_bytes(df: pd.DataFrame) -> bytes:
    return df.to_csv(index=False).encode("utf-8-sig")


def has_stock(value) -> bool:
    text = str(value).strip().upper()
    return text in ["예", "Y", "YES", "O", "OK", "있음", "있다", "가능", "보유", "재고있음"]


st.title("계량기 교체 후보 검토 앱")
st.caption(
    "계량기 교체 대상 데이터를 기준값으로 필터링하고, "
    "담당자가 선택한 교체 후보를 다운로드하는 업무형 Streamlit 앱입니다."
)

st.sidebar.header("데이터 업로드")

uploaded_file = st.sidebar.file_uploader(
    "계량기 교체 대상 엑셀",
    type=["xlsx"],
)

meter = load_excel(uploaded_file, DATA_FILE, SHEET_NAME)

if meter.empty:
    st.warning(
        "데이터를 불러오지 못했습니다. "
        "`data/04_계량기_교체대상_자동추출_예시데이터.xlsx` 파일이 있는지 확인하거나 "
        "엑셀 파일을 업로드해 주세요."
    )
    st.stop()

missing_columns = get_missing_columns(meter)

if missing_columns:
    st.error("필수 컬럼이 누락되어 앱을 실행할 수 없습니다.")
    st.write("누락된 컬럼:", missing_columns)
    st.write("현재 엑셀 컬럼:", meter.columns.tolist())
    st.stop()

meter = prepare_data(meter)

center_options = sorted(meter["서비스센터"].dropna().unique().tolist())
grade_options = sorted(meter["참고_교체등급"].dropna().unique().tolist())
status_options = sorted(meter["통신상태"].dropna().unique().tolist())

st.sidebar.header("조건 조정")

with st.sidebar.form("criteria_form"):
    selected_centers = st.multiselect(
        "서비스센터",
        options=center_options,
        default=[],
    )

    selected_grades = st.multiselect(
        "교체등급",
        options=grade_options,
        default=[],
    )

    selected_status = st.multiselect(
        "통신상태",
        options=status_options,
        default=[],
    )

    min_replace_score = st.slider(
        "교체 최소 점수",
        min_value=0,
        max_value=100,
        value=70,
        step=5,
    )

    max_battery = st.slider(
        "배터리 잔량 기준 이하(%)",
        min_value=0,
        max_value=100,
        value=20,
        step=5,
    )

    stock_only = st.checkbox(
        "부품 재고가 있는 건만 보기",
        value=True,
    )

    daily_target = st.number_input(
        "오늘 처리 목표 건수",
        min_value=1,
        max_value=100,
        value=20,
        step=1,
    )

    st.form_submit_button("조건 적용")

candidates = meter.copy()

if selected_centers:
    candidates = candidates[candidates["서비스센터"].isin(selected_centers)]

if selected_grades:
    candidates = candidates[candidates["참고_교체등급"].isin(selected_grades)]

if selected_status:
    candidates = candidates[candidates["통신상태"].isin(selected_status)]

if stock_only:
    candidates = candidates[candidates["부품재고여부"].apply(has_stock)]

candidates = candidates[
    (candidates["참고_교체점수"] >= min_replace_score)
    & (candidates["배터리잔량_pct"] <= max_battery)
].copy()

candidates = candidates.sort_values(
    by=["참고_교체점수", "배터리잔량_pct", "최근오류건수", "사용연수"],
    ascending=[False, True, False, False],
)

replace_count = int(len(candidates))
immediate_count = int(candidates["참고_교체등급"].isin(["즉시교체"]).sum())
avg_replace_score = candidates["참고_교체점수"].mean()
stock_count = int(candidates["부품재고여부"].apply(has_stock).sum())

m1, m2, m3, m4 = st.columns(4)

m1.metric("교체 후보 수", f"{replace_count:,}건")
m2.metric("즉시교체 후보 수", f"{immediate_count:,}건")
m3.metric(
    "평균 교체 점수",
    f"{avg_replace_score:.1f}점" if replace_count else "0.0점",
)
m4.metric("재고 있는 후보 수", f"{stock_count:,}건")

candidate_tab, selected_tab, guide_tab = st.tabs(
    ["계량기 교체 후보", "선택 결과", "데이터 설명"]
)

with candidate_tab:
    st.subheader("계량기 교체 후보")
    st.write("조건을 통과한 계량기 교체 후보를 확인하고, 담당자가 직접 선택 여부와 메모를 입력합니다.")

    display_columns = [
        "계량기ID",
        "고객ID",
        "서비스센터",
        "지역",
        "설치일자",
        "사용연수",
        "배터리잔량_pct",
        "최근오류건수",
        "통신상태",
        "부품재고여부",
        "참고_교체점수",
        "참고_교체등급",
        "교체사유",
        "권장조치",
    ]

    candidate_table = candidates[display_columns].copy()
    candidate_table.insert(0, "선택", False)
    candidate_table["담당메모"] = ""

    if candidate_table.empty:
        st.info("현재 조건에 맞는 계량기 교체 후보가 없습니다. 기준을 낮추거나 필터 조건을 조정해 주세요.")
        edited_candidates = candidate_table
    else:
        edited_candidates = st.data_editor(
            candidate_table,
            use_container_width=True,
            hide_index=True,
            num_rows="fixed",
            disabled=[
                col
                for col in candidate_table.columns
                if col not in ["선택", "담당메모"]
            ],
            column_config={
                "선택": st.column_config.CheckboxColumn(
                    "선택",
                    help="최종 교체 후보로 선택할 경우 체크합니다.",
                ),
                "담당메모": st.column_config.TextColumn(
                    "담당메모",
                    help="교체 전 확인사항이나 담당자 의견을 입력합니다.",
                ),
                "설치일자": st.column_config.DateColumn(
                    "설치일자",
                    format="YYYY-MM-DD",
                ),
                "사용연수": st.column_config.NumberColumn(
                    "사용연수",
                    format="%.1f년",
                ),
                "배터리잔량_pct": st.column_config.NumberColumn(
                    "배터리잔량_pct",
                    format="%.1f%%",
                ),
                "최근오류건수": st.column_config.NumberColumn(
                    "최근오류건수",
                    format="%d건",
                ),
                "참고_교체점수": st.column_config.NumberColumn(
                    "참고_교체점수",
                    format="%.0f점",
                ),
            },
        )

    st.download_button(
        "현재 교체 후보 CSV 다운로드",
        data=to_csv_bytes(candidates),
        file_name="meter_replacement_candidates.csv",
        mime="text/csv",
        disabled=candidates.empty,
    )

with selected_tab:
    st.subheader("선택 결과")

    if "선택" in edited_candidates.columns:
        selected_rows = edited_candidates[edited_candidates["선택"] == True].copy()
    else:
        selected_rows = pd.DataFrame()

    selected_count = int(len(selected_rows))
    progress_rate = min(selected_count / daily_target, 1.0)

    st.progress(
        progress_rate,
        text=f"오늘 처리 목표 {daily_target}건 중 {selected_count}건 선택",
    )

    if selected_rows.empty:
        st.info("아직 선택한 교체 후보가 없습니다. `계량기 교체 후보` 탭에서 후보를 선택해 주세요.")
    else:
        st.dataframe(
            selected_rows,
            use_container_width=True,
            hide_index=True,
        )

    st.download_button(
        "선택 결과 CSV 다운로드",
        data=to_csv_bytes(selected_rows),
        file_name="selected_meter_replacement.csv",
        mime="text/csv",
        disabled=selected_rows.empty,
    )

with guide_tab:
    st.subheader("데이터 설명")

    st.markdown("#### 주요 컬럼 설명")

    column_description = pd.DataFrame(
        [
            ["계량기ID", "교체 대상 계량기를 구분하는 ID", "현장 확인 및 결과 관리"],
            ["고객ID", "고객을 구분하는 ID", "고객 단위 검토"],
            ["서비스센터", "담당 서비스센터", "센터별 후보 필터링"],
            ["지역", "고객 또는 계량기가 속한 지역", "지역별 후보 확인"],
            ["설치일자", "계량기가 설치된 날짜", "노후도 판단"],
            ["사용연수", "계량기 사용 기간", "장기 사용 여부 판단"],
            ["배터리잔량_pct", "계량기 배터리 잔량", "배터리 부족 대상 선별"],
            ["최근오류건수", "최근 오류 발생 건수", "이상 징후 판단"],
            ["통신상태", "최근 원격 검침 또는 통신 상태", "통신 이상 여부 확인"],
            ["부품재고여부", "교체 부품 재고 보유 여부", "즉시 처리 가능 여부 판단"],
            ["참고_교체점수", "자동 산정된 교체 우선순위 점수", "후보 추출 기준"],
            ["참고_교체등급", "교체 추천 등급", "즉시교체 후보 구분"],
            ["교체사유", "교체 후보로 추천된 사유", "담당자 검토 참고"],
            ["권장조치", "권장되는 조치 내용", "최종 조치 판단"],
        ],
        columns=["컬럼명", "설명", "활용 방식"],
    )

    st.dataframe(
        column_description,
        use_container_width=True,
        hide_index=True,
    )

    st.markdown("#### 기준값 조정에 따른 변화")
    st.write("- 교체 최소 점수를 높이면 교체 후보 수는 줄고, 평균 교체 점수는 높아질 수 있습니다.")
    st.write("- 배터리 잔량 기준을 낮추면 배터리가 더 부족한 계량기 중심으로 후보가 줄어듭니다.")
    st.write("- 부품 재고가 있는 건만 보기 옵션을 켜면 즉시 처리 가능한 후보 중심으로 목록이 줄어듭니다.")
    st.write("- 서비스센터, 교체등급, 통신상태를 선택하면 해당 조건에 맞는 후보만 표시됩니다.")

    show_raw_data = st.toggle("원본 데이터 미리보기", value=False)

    if show_raw_data:
        st.dataframe(
            meter.head(50),
            use_container_width=True,
            hide_index=True,
        )