← back to KIM3310__districtpilot-ai

Function bodies 70 total

All specs Real LLM only Function bodies
add_bg function · python · L24-L28 (5 LOC)
generate_pptx.py
def add_bg(slide, color=NAVY):
    bg = slide.background
    fill = bg.fill
    fill.solid()
    fill.fore_color.rgb = color
add_bg function · python · L24-L28 (5 LOC)
generate_pptx.py
def add_bg(slide, color=NAVY):
    bg = slide.background
    fill = bg.fill
    fill.solid()
    fill.fore_color.rgb = color
add_text function · python · L30-L41 (12 LOC)
generate_pptx.py
def add_text(slide, text, x, y, w, h, size=18, color=WHITE, bold=False, align=PP_ALIGN.LEFT, font="Calibri"):
    txBox = slide.shapes.add_textbox(Inches(x), Inches(y), Inches(w), Inches(h))
    tf = txBox.text_frame
    tf.word_wrap = True
    p = tf.paragraphs[0]
    p.text = text
    p.font.size = Pt(size)
    p.font.color.rgb = color
    p.font.bold = bold
    p.font.name = font
    p.alignment = align
    return tf
add_text function · python · L30-L41 (12 LOC)
generate_pptx.py
def add_text(slide, text, x, y, w, h, size=18, color=WHITE, bold=False, align=PP_ALIGN.LEFT, font="Calibri"):
    txBox = slide.shapes.add_textbox(Inches(x), Inches(y), Inches(w), Inches(h))
    tf = txBox.text_frame
    tf.word_wrap = True
    p = tf.paragraphs[0]
    p.text = text
    p.font.size = Pt(size)
    p.font.color.rgb = color
    p.font.bold = bold
    p.font.name = font
    p.alignment = align
    return tf
add_multiline function · python · L43-L58 (16 LOC)
generate_pptx.py
def add_multiline(slide, lines, x, y, w, h, size=16, color=WHITE, bold=False, spacing=1.2, bullet=False):
    txBox = slide.shapes.add_textbox(Inches(x), Inches(y), Inches(w), Inches(h))
    tf = txBox.text_frame
    tf.word_wrap = True
    for i, line in enumerate(lines):
        if i == 0:
            p = tf.paragraphs[0]
        else:
            p = tf.add_paragraph()
        p.text = line
        p.font.size = Pt(size)
        p.font.color.rgb = color
        p.font.bold = bold
        p.font.name = "Calibri"
        p.space_after = Pt(size * spacing * 0.4)
    return tf
add_multiline function · python · L43-L58 (16 LOC)
generate_pptx.py
def add_multiline(slide, lines, x, y, w, h, size=16, color=WHITE, bold=False, spacing=1.2, bullet=False):
    txBox = slide.shapes.add_textbox(Inches(x), Inches(y), Inches(w), Inches(h))
    tf = txBox.text_frame
    tf.word_wrap = True
    for i, line in enumerate(lines):
        if i == 0:
            p = tf.paragraphs[0]
        else:
            p = tf.add_paragraph()
        p.text = line
        p.font.size = Pt(size)
        p.font.color.rgb = color
        p.font.bold = bold
        p.font.name = "Calibri"
        p.space_after = Pt(size * spacing * 0.4)
    return tf
add_rect function · python · L60-L65 (6 LOC)
generate_pptx.py
def add_rect(slide, x, y, w, h, color):
    shape = slide.shapes.add_shape(1, Inches(x), Inches(y), Inches(w), Inches(h))  # MSO_SHAPE.RECTANGLE
    shape.fill.solid()
    shape.fill.fore_color.rgb = color
    shape.line.fill.background()
    return shape
Provenance: Repobility (https://repobility.com) — every score reproducible from /scan/
add_rect function · python · L60-L65 (6 LOC)
generate_pptx.py
def add_rect(slide, x, y, w, h, color):
    shape = slide.shapes.add_shape(1, Inches(x), Inches(y), Inches(w), Inches(h))  # MSO_SHAPE.RECTANGLE
    shape.fill.solid()
    shape.fill.fore_color.rgb = color
    shape.line.fill.background()
    return shape
add_table_slide function · python · L67-L97 (31 LOC)
generate_pptx.py
def add_table_slide(slide, headers, rows, x, y, w, row_h=0.45):
    cols = len(headers)
    total_rows = len(rows) + 1
    table_shape = slide.shapes.add_table(total_rows, cols, Inches(x), Inches(y), Inches(w), Inches(row_h * total_rows))
    table = table_shape.table
    col_w = w / cols
    for i in range(cols):
        table.columns[i].width = Inches(col_w)
    # Headers
    for j, h in enumerate(headers):
        cell = table.cell(0, j)
        cell.text = h
        for p in cell.text_frame.paragraphs:
            p.font.size = Pt(13)
            p.font.color.rgb = WHITE
            p.font.bold = True
            p.font.name = "Calibri"
        cell.fill.solid()
        cell.fill.fore_color.rgb = RGBColor(0x29, 0x4D, 0x8F)
    # Data
    for i, row in enumerate(rows):
        for j, val in enumerate(row):
            cell = table.cell(i + 1, j)
            cell.text = str(val)
            for p in cell.text_frame.paragraphs:
                p.font.size = Pt(12)
                p.fo
add_table_slide function · python · L67-L97 (31 LOC)
generate_pptx.py
def add_table_slide(slide, headers, rows, x, y, w, row_h=0.45):
    cols = len(headers)
    total_rows = len(rows) + 1
    table_shape = slide.shapes.add_table(total_rows, cols, Inches(x), Inches(y), Inches(w), Inches(row_h * total_rows))
    table = table_shape.table
    col_w = w / cols
    for i in range(cols):
        table.columns[i].width = Inches(col_w)
    # Headers
    for j, h in enumerate(headers):
        cell = table.cell(0, j)
        cell.text = h
        for p in cell.text_frame.paragraphs:
            p.font.size = Pt(13)
            p.font.color.rgb = WHITE
            p.font.bold = True
            p.font.name = "Calibri"
        cell.fill.solid()
        cell.fill.fore_color.rgb = RGBColor(0x29, 0x4D, 0x8F)
    # Data
    for i, row in enumerate(rows):
        for j, val in enumerate(row):
            cell = table.cell(i + 1, j)
            cell.text = str(val)
            for p in cell.text_frame.paragraphs:
                p.font.size = Pt(12)
                p.fo
load_df function · python · L59-L61 (3 LOC)
streamlit_app_v7.py
def load_df(_session, sql: str) -> pd.DataFrame:
    """Cached SQL execution returning a pandas DataFrame."""
    return _session.sql(sql).to_pandas()
load_df function · python · L59-L61 (3 LOC)
streamlit_app_v7.py
def load_df(_session, sql: str) -> pd.DataFrame:
    """Cached SQL execution returning a pandas DataFrame."""
    return _session.sql(sql).to_pandas()
safe_read function · python · L64-L69 (6 LOC)
streamlit_app_v7.py
def safe_read(sql: str) -> pd.DataFrame:
    """Execute SQL with graceful error handling. Returns empty DataFrame on failure."""
    try:
        return load_df(session, sql)
    except Exception:  # noqa: BLE001 – intentional catch-all for resilient UI
        return pd.DataFrame()
safe_read function · python · L64-L69 (6 LOC)
streamlit_app_v7.py
def safe_read(sql: str) -> pd.DataFrame:
    """Execute SQL with graceful error handling. Returns empty DataFrame on failure."""
    try:
        return load_df(session, sql)
    except Exception:  # noqa: BLE001 – intentional catch-all for resilient UI
        return pd.DataFrame()
clean_variant function · python · L72-L77 (6 LOC)
streamlit_app_v7.py
def clean_variant(value) -> str:
    """Normalize VARIANT column values that may contain quotes or NULL strings."""
    if pd.isna(value):
        return "ALL"
    text = str(value).strip().strip('"')
    return "ALL" if text.upper() in {"NULL", "NONE", ""} else text
Repobility · code-quality intelligence platform · https://repobility.com
clean_variant function · python · L72-L77 (6 LOC)
streamlit_app_v7.py
def clean_variant(value) -> str:
    """Normalize VARIANT column values that may contain quotes or NULL strings."""
    if pd.isna(value):
        return "ALL"
    text = str(value).strip().strip('"')
    return "ALL" if text.upper() in {"NULL", "NONE", ""} else text
fmt_int function · python · L80-L82 (3 LOC)
streamlit_app_v7.py
def fmt_int(value) -> str:
    """Format numeric value as integer with comma separators."""
    return "-" if pd.isna(value) else f"{float(value):,.0f}"
fmt_int function · python · L80-L82 (3 LOC)
streamlit_app_v7.py
def fmt_int(value) -> str:
    """Format numeric value as integer with comma separators."""
    return "-" if pd.isna(value) else f"{float(value):,.0f}"
fmt_eok function · python · L85-L87 (3 LOC)
streamlit_app_v7.py
def fmt_eok(value) -> str:
    """Format value in 억원 (100M KRW) unit."""
    return "-" if pd.isna(value) else f"{float(value) / 1e8:,.1f}"
fmt_eok function · python · L85-L87 (3 LOC)
streamlit_app_v7.py
def fmt_eok(value) -> str:
    """Format value in 억원 (100M KRW) unit."""
    return "-" if pd.isna(value) else f"{float(value) / 1e8:,.1f}"
fmt_pct function · python · L90-L92 (3 LOC)
streamlit_app_v7.py
def fmt_pct(value) -> str:
    """Format as signed percentage string."""
    return "-" if pd.isna(value) else f"{float(value):+.1f}%"
fmt_pct function · python · L90-L92 (3 LOC)
streamlit_app_v7.py
def fmt_pct(value) -> str:
    """Format as signed percentage string."""
    return "-" if pd.isna(value) else f"{float(value):+.1f}%"
safe_float function · python · L95-L101 (7 LOC)
streamlit_app_v7.py
def safe_float(value, default=0.0) -> float:
    """Safely convert value to float."""
    try:
        result = float(value)
        return result if not pd.isna(result) else default
    except (ValueError, TypeError):
        return default
Want this analysis on your repo? https://repobility.com/scan/
safe_float function · python · L95-L101 (7 LOC)
streamlit_app_v7.py
def safe_float(value, default=0.0) -> float:
    """Safely convert value to float."""
    try:
        result = float(value)
        return result if not pd.isna(result) else default
    except (ValueError, TypeError):
        return default
latest_snapshot function · python · L153-L166 (14 LOC)
streamlit_app_v7.py
def latest_snapshot(df: pd.DataFrame) -> pd.DataFrame:
    """Return the most recent row per district."""
    if df.empty:
        return df
    out = df.copy()
    out["_ts"] = pd.to_datetime(
        out["YM"].astype(str).str.strip(), format="%Y%m", errors="coerce"
    )
    return (
        out.sort_values("_ts")
        .groupby("DISTRICT", as_index=False)
        .tail(1)
        .sort_values("DISTRICT")
    )
latest_snapshot function · python · L153-L166 (14 LOC)
streamlit_app_v7.py
def latest_snapshot(df: pd.DataFrame) -> pd.DataFrame:
    """Return the most recent row per district."""
    if df.empty:
        return df
    out = df.copy()
    out["_ts"] = pd.to_datetime(
        out["YM"].astype(str).str.strip(), format="%Y%m", errors="coerce"
    )
    return (
        out.sort_values("_ts")
        .groupby("DISTRICT", as_index=False)
        .tail(1)
        .sort_values("DISTRICT")
    )
build_allocation function · python · L177-L189 (13 LOC)
streamlit_app_v7.py
def build_allocation(df: pd.DataFrame):
    """Compute allocation percentages from first forecast timestamp."""
    if df.empty:
        return pd.DataFrame(), None
    out = df.copy()
    out["_ts"] = pd.to_datetime(out["TS"], errors="coerce")
    first_ts = out["_ts"].dropna().min()
    if pd.isna(first_ts):
        return pd.DataFrame(), None
    sub = out[out["_ts"] == first_ts].copy()
    total = sub["FORECAST"].sum()
    sub["ALLOC_PCT"] = (sub["FORECAST"] / total * 100).round(1) if total > 0 else 0.0
    return sub.sort_values("ALLOC_PCT", ascending=False), first_ts
build_allocation function · python · L177-L189 (13 LOC)
streamlit_app_v7.py
def build_allocation(df: pd.DataFrame):
    """Compute allocation percentages from first forecast timestamp."""
    if df.empty:
        return pd.DataFrame(), None
    out = df.copy()
    out["_ts"] = pd.to_datetime(out["TS"], errors="coerce")
    first_ts = out["_ts"].dropna().min()
    if pd.isna(first_ts):
        return pd.DataFrame(), None
    sub = out[out["_ts"] == first_ts].copy()
    total = sub["FORECAST"].sum()
    sub["ALLOC_PCT"] = (sub["FORECAST"] / total * 100).round(1) if total > 0 else 0.0
    return sub.sort_values("ALLOC_PCT", ascending=False), first_ts
build_overlay function · python · L195-L209 (15 LOC)
streamlit_app_v7.py
def build_overlay(df: pd.DataFrame) -> pd.DataFrame:
    """Pivot actual vs forecast for overlay chart."""
    if df.empty:
        return pd.DataFrame()
    out = df.copy()
    out["_ds"] = pd.to_datetime(out["DS"], errors="coerce")
    actual = (
        out.pivot_table(index="_ds", columns="DISTRICT", values="ACTUAL", aggfunc="max")
        .add_suffix(" Actual")
    )
    fcast = (
        out.pivot_table(index="_ds", columns="DISTRICT", values="FORECAST_VAL", aggfunc="max")
        .add_suffix(" Forecast")
    )
    return actual.join(fcast, how="outer").sort_index()
build_overlay function · python · L195-L209 (15 LOC)
streamlit_app_v7.py
def build_overlay(df: pd.DataFrame) -> pd.DataFrame:
    """Pivot actual vs forecast for overlay chart."""
    if df.empty:
        return pd.DataFrame()
    out = df.copy()
    out["_ds"] = pd.to_datetime(out["DS"], errors="coerce")
    actual = (
        out.pivot_table(index="_ds", columns="DISTRICT", values="ACTUAL", aggfunc="max")
        .add_suffix(" Actual")
    )
    fcast = (
        out.pivot_table(index="_ds", columns="DISTRICT", values="FORECAST_VAL", aggfunc="max")
        .add_suffix(" Forecast")
    )
    return actual.join(fcast, how="outer").sort_index()
build_eval_pivot function · python · L215-L234 (20 LOC)
streamlit_app_v7.py
def build_eval_pivot(df: pd.DataFrame) -> pd.DataFrame:
    """Pivot evaluation metrics to one row per series."""
    if df.empty:
        return pd.DataFrame()
    out = df.copy()
    if "SERIES" in out.columns:
        out["SERIES"] = out["SERIES"].apply(clean_variant)
    if "ERROR_METRIC" in out.columns:
        out["ERROR_METRIC"] = (
            out["ERROR_METRIC"].astype(str).str.replace('"', "", regex=False)
        )
    return (
        out.pivot_table(
            index="SERIES",
            columns="ERROR_METRIC",
            values="METRIC_VALUE",
            aggfunc="first",
        )
        .reset_index()
    )
About: code-quality intelligence by Repobility · https://repobility.com
build_eval_pivot function · python · L215-L234 (20 LOC)
streamlit_app_v7.py
def build_eval_pivot(df: pd.DataFrame) -> pd.DataFrame:
    """Pivot evaluation metrics to one row per series."""
    if df.empty:
        return pd.DataFrame()
    out = df.copy()
    if "SERIES" in out.columns:
        out["SERIES"] = out["SERIES"].apply(clean_variant)
    if "ERROR_METRIC" in out.columns:
        out["ERROR_METRIC"] = (
            out["ERROR_METRIC"].astype(str).str.replace('"', "", regex=False)
        )
    return (
        out.pivot_table(
            index="SERIES",
            columns="ERROR_METRIC",
            values="METRIC_VALUE",
            aggfunc="first",
        )
        .reset_index()
    )
normalize_fi function · python · L240-L249 (10 LOC)
streamlit_app_v7.py
def normalize_fi(df: pd.DataFrame) -> pd.DataFrame:
    """Normalize feature importance column names."""
    if df.empty:
        return df
    out = df.copy()
    if "SCORE" in out.columns and "IMPORTANCE_SCORE" not in out.columns:
        out = out.rename(columns={"SCORE": "IMPORTANCE_SCORE"})
    if "SERIES" in out.columns:
        out["SERIES"] = out["SERIES"].apply(clean_variant)
    return out
normalize_fi function · python · L240-L249 (10 LOC)
streamlit_app_v7.py
def normalize_fi(df: pd.DataFrame) -> pd.DataFrame:
    """Normalize feature importance column names."""
    if df.empty:
        return df
    out = df.copy()
    if "SCORE" in out.columns and "IMPORTANCE_SCORE" not in out.columns:
        out = out.rename(columns={"SCORE": "IMPORTANCE_SCORE"})
    if "SERIES" in out.columns:
        out["SERIES"] = out["SERIES"].apply(clean_variant)
    return out
build_context_json function · python · L259-L390 (132 LOC)
streamlit_app_v7.py
def build_context_json(
    scope: str,
    alloc: pd.DataFrame,
    lat: pd.DataFrame,
    fi: pd.DataFrame,
) -> str:
    """Build grounded context JSON for LLM calls with extended data."""
    # Allocation payload
    alloc_cols = ["DISTRICT", "FORECAST", "ALLOC_PCT"]
    alloc_payload = (
        alloc[[c for c in alloc_cols if c in alloc.columns]].to_dict("records")
        if not alloc.empty
        else []
    )

    # Snapshot payload — extended columns
    snap_cols = [
        "DISTRICT", "TOTAL_POP", "TOTAL_SALES", "NET_MOVE",
        "AVG_ASSET", "AVG_MEME_PRICE",
        "AGE_20_39_SHARE", "SENIOR_60P_SHARE",
        "TOURISM_DEMAND_IDX", "FOREIGN_VISITOR_IDX",
        "STABILITY_SCORE", "NET_STORE_CHANGE",
        "HOLIDAY_DAYS",
    ]
    if scope != "전체" and not lat.empty:
        snap = lat[lat["DISTRICT"] == scope]
    else:
        snap = lat
    snap_payload = (
        snap[[c for c in snap_cols if c in snap.columns]].to_dict("records")
        if not snap.empty
build_context_json function · python · L259-L390 (132 LOC)
streamlit_app_v7.py
def build_context_json(
    scope: str,
    alloc: pd.DataFrame,
    lat: pd.DataFrame,
    fi: pd.DataFrame,
) -> str:
    """Build grounded context JSON for LLM calls with extended data."""
    # Allocation payload
    alloc_cols = ["DISTRICT", "FORECAST", "ALLOC_PCT"]
    alloc_payload = (
        alloc[[c for c in alloc_cols if c in alloc.columns]].to_dict("records")
        if not alloc.empty
        else []
    )

    # Snapshot payload — extended columns
    snap_cols = [
        "DISTRICT", "TOTAL_POP", "TOTAL_SALES", "NET_MOVE",
        "AVG_ASSET", "AVG_MEME_PRICE",
        "AGE_20_39_SHARE", "SENIOR_60P_SHARE",
        "TOURISM_DEMAND_IDX", "FOREIGN_VISITOR_IDX",
        "STABILITY_SCORE", "NET_STORE_CHANGE",
        "HOLIDAY_DAYS",
    ]
    if scope != "전체" and not lat.empty:
        snap = lat[lat["DISTRICT"] == scope]
    else:
        snap = lat
    snap_payload = (
        snap[[c for c in snap_cols if c in snap.columns]].to_dict("records")
        if not snap.empty
call_ai_complete function · python · L393-L439 (47 LOC)
streamlit_app_v7.py
def call_ai_complete(question: str, context_json: str) -> dict:
    """AI_COMPLETE with structured output. Falls back to CORTEX.COMPLETE."""
    prompt = f"""당신은 DistrictPilot AI의 한국어 의사결정 보조 모델이다.

반드시 지킬 규칙:
1) 아래 CONTEXT 밖의 사실은 만들지 말 것.
2) 숫자는 CONTEXT에 있는 경우에만 사용할 것.
3) 확실하지 않으면 '현재 데이터만으로는 확정할 수 없습니다.'라고 말 것.
4) 답변은 짧고 실행 중심으로 작성할 것.
5) 한국어로 작성할 것.

CONTEXT:
{context_json}

사용자 질문: {question}""".strip()

    try:
        rows = session.sql(
            """
            SELECT AI_COMPLETE(
                model => ?,
                prompt => ?,
                model_parameters => {'temperature': 0, 'max_tokens': 600, 'guardrails': TRUE},
                response_format => TYPE OBJECT(
                    answer STRING,
                    recommended_district STRING,
                    allocation_pct FLOAT,
                    drivers ARRAY(STRING),
                    risk STRING,
                    next_action STRING
                ),
                show_details => TRUE
call_ai_complete function · python · L393-L439 (47 LOC)
streamlit_app_v7.py
def call_ai_complete(question: str, context_json: str) -> dict:
    """AI_COMPLETE with structured output. Falls back to CORTEX.COMPLETE."""
    prompt = f"""당신은 DistrictPilot AI의 한국어 의사결정 보조 모델이다.

반드시 지킬 규칙:
1) 아래 CONTEXT 밖의 사실은 만들지 말 것.
2) 숫자는 CONTEXT에 있는 경우에만 사용할 것.
3) 확실하지 않으면 '현재 데이터만으로는 확정할 수 없습니다.'라고 말 것.
4) 답변은 짧고 실행 중심으로 작성할 것.
5) 한국어로 작성할 것.

CONTEXT:
{context_json}

사용자 질문: {question}""".strip()

    try:
        rows = session.sql(
            """
            SELECT AI_COMPLETE(
                model => ?,
                prompt => ?,
                model_parameters => {'temperature': 0, 'max_tokens': 600, 'guardrails': TRUE},
                response_format => TYPE OBJECT(
                    answer STRING,
                    recommended_district STRING,
                    allocation_pct FLOAT,
                    drivers ARRAY(STRING),
                    risk STRING,
                    next_action STRING
                ),
                show_details => TRUE
_fallback_complete function · python · L442-L452 (11 LOC)
streamlit_app_v7.py
def _fallback_complete(prompt: str) -> dict:
    """Fallback to SNOWFLAKE.CORTEX.COMPLETE when AI_COMPLETE is unavailable."""
    try:
        safe_prompt = prompt.replace("'", "''").replace("\\", "\\\\")
        rows = session.sql(
            f"SELECT SNOWFLAKE.CORTEX.COMPLETE('{LLM_MODEL}', '{safe_prompt}') AS R"
        ).collect()
        answer = rows[0]["R"] if rows else "응답을 생성하지 못했습니다."
        return {"structured_output": {"answer": answer}}
    except Exception:  # noqa: BLE001
        return {"structured_output": {"answer": "AI 호출 실패: 일시적 오류가 발생했습니다. 잠시 후 다시 시도해주세요."}}
Provenance: Repobility (https://repobility.com) — every score reproducible from /scan/
_fallback_complete function · python · L442-L452 (11 LOC)
streamlit_app_v7.py
def _fallback_complete(prompt: str) -> dict:
    """Fallback to SNOWFLAKE.CORTEX.COMPLETE when AI_COMPLETE is unavailable."""
    try:
        safe_prompt = prompt.replace("'", "''").replace("\\", "\\\\")
        rows = session.sql(
            f"SELECT SNOWFLAKE.CORTEX.COMPLETE('{LLM_MODEL}', '{safe_prompt}') AS R"
        ).collect()
        answer = rows[0]["R"] if rows else "응답을 생성하지 못했습니다."
        return {"structured_output": {"answer": answer}}
    except Exception:  # noqa: BLE001
        return {"structured_output": {"answer": "AI 호출 실패: 일시적 오류가 발생했습니다. 잠시 후 다시 시도해주세요."}}
load_df function · python · L59-L61 (3 LOC)
submission/streamlit_app_v7.py
def load_df(_session, sql: str) -> pd.DataFrame:
    """Cached SQL execution returning a pandas DataFrame."""
    return _session.sql(sql).to_pandas()
load_df function · python · L59-L61 (3 LOC)
submission/streamlit_app_v7.py
def load_df(_session, sql: str) -> pd.DataFrame:
    """Cached SQL execution returning a pandas DataFrame."""
    return _session.sql(sql).to_pandas()
safe_read function · python · L64-L69 (6 LOC)
submission/streamlit_app_v7.py
def safe_read(sql: str) -> pd.DataFrame:
    """Execute SQL with graceful error handling. Returns empty DataFrame on failure."""
    try:
        return load_df(session, sql)
    except Exception:  # noqa: BLE001 – intentional catch-all for resilient UI
        return pd.DataFrame()
safe_read function · python · L64-L69 (6 LOC)
submission/streamlit_app_v7.py
def safe_read(sql: str) -> pd.DataFrame:
    """Execute SQL with graceful error handling. Returns empty DataFrame on failure."""
    try:
        return load_df(session, sql)
    except Exception:  # noqa: BLE001 – intentional catch-all for resilient UI
        return pd.DataFrame()
clean_variant function · python · L72-L77 (6 LOC)
submission/streamlit_app_v7.py
def clean_variant(value) -> str:
    """Normalize VARIANT column values that may contain quotes or NULL strings."""
    if pd.isna(value):
        return "ALL"
    text = str(value).strip().strip('"')
    return "ALL" if text.upper() in {"NULL", "NONE", ""} else text
clean_variant function · python · L72-L77 (6 LOC)
submission/streamlit_app_v7.py
def clean_variant(value) -> str:
    """Normalize VARIANT column values that may contain quotes or NULL strings."""
    if pd.isna(value):
        return "ALL"
    text = str(value).strip().strip('"')
    return "ALL" if text.upper() in {"NULL", "NONE", ""} else text
fmt_int function · python · L80-L82 (3 LOC)
submission/streamlit_app_v7.py
def fmt_int(value) -> str:
    """Format numeric value as integer with comma separators."""
    return "-" if pd.isna(value) else f"{float(value):,.0f}"
Repobility · code-quality intelligence platform · https://repobility.com
fmt_int function · python · L80-L82 (3 LOC)
submission/streamlit_app_v7.py
def fmt_int(value) -> str:
    """Format numeric value as integer with comma separators."""
    return "-" if pd.isna(value) else f"{float(value):,.0f}"
fmt_eok function · python · L85-L87 (3 LOC)
submission/streamlit_app_v7.py
def fmt_eok(value) -> str:
    """Format value in 억원 (100M KRW) unit."""
    return "-" if pd.isna(value) else f"{float(value) / 1e8:,.1f}"
fmt_eok function · python · L85-L87 (3 LOC)
submission/streamlit_app_v7.py
def fmt_eok(value) -> str:
    """Format value in 억원 (100M KRW) unit."""
    return "-" if pd.isna(value) else f"{float(value) / 1e8:,.1f}"
page 1 / 2next ›