Update app.py

app.py

@@ -6,8 +6,6 @@ import re
 import base64
 import difflib
 import traceback
-from functools import lru_cache
-from threading import Lock
 
 # --------------------------
 # 1️⃣ 저장소 설정
@@ -19,184 +17,39 @@ if not os.path.exists(UPLOAD_DIR):
 db_registry = []
 
 # --------------------------
-# 2️⃣ DB 연결 캐싱 (핵심 개선)
+# 유틸
 # --------------------------
-_conn_cache: dict[str, sqlite3.Connection] = {}
-_conn_lock = Lock()
-
-def get_connection(db_path: str) -> sqlite3.Connection:
-    """
-    같은 경로의 DB는 연결을 재사용.
-    DB 파일이 변경되었을 경우(mtime 변화)를 감지해 자동으로 재연결.
-    """
-    mtime = os.path.getmtime(db_path)
-    cache_key = f"{db_path}::{mtime}"
-
-    with _conn_lock:
-        if cache_key not in _conn_cache:
-            # 기존 연결 정리 (같은 path의 구버전 연결 제거)
-            stale = [k for k in _conn_cache if k.startswith(db_path + "::")]
-            for k in stale:
-                try:
-                    _conn_cache[k].close()
-                except Exception:
-                    pass
-                del _conn_cache[k]
-
-            conn = sqlite3.connect(db_path, check_same_thread=False)
-            conn.text_factory = safe_text_factory
-
-            # 인덱스 자동 생성 (chapter, category, section 컬럼 대상)
-            _ensure_indexes(conn, db_path)
-
-            _conn_cache[cache_key] = conn
-
-        return _conn_cache[cache_key]
-
-
-def _ensure_indexes(conn: sqlite3.Connection, db_path: str):
-    """자주 조회되는 컬럼에 인덱스가 없으면 자동 생성."""
+def blob_to_base64_html(blob_data):
+    if blob_data is None or pd.isna(blob_data):
+        return ""
     try:
-        tables = pd.read_sql(
-            "SELECT name FROM sqlite_master WHERE type='table';", conn
-        )['name'].tolist()
-
-        for table in tables:
-            cols_df = pd.read_sql(f"PRAGMA table_info([{table}])", conn)
-            cols = [c.lower() for c in cols_df['name'].tolist()]
-
-            for target_col in ['chapter', 'category', 'section']:
-                if target_col in cols:
-                    real_col = cols_df['name'].tolist()[cols.index(target_col)]
-                    idx_name = f"idx_{table}_{target_col}"
-                    conn.execute(
-                        f"CREATE INDEX IF NOT EXISTS [{idx_name}] ON [{table}] ([{real_col}])"
-                    )
-        conn.commit()
-    except Exception:
-        pass  # 인덱스 생성 실패는 무시 (read-only DB 등 예외 상황 대응)
-
-
-# --------------------------
-# 3️⃣ 쿼리 결과 캐싱
-# --------------------------
-_query_cache: dict[str, pd.DataFrame] = {}
-
-def _cache_key(*args) -> str:
-    return "::".join(str(a) for a in args)
-
-def get_cached_df(key: str):
-    return _query_cache.get(key)
-
-def set_cached_df(key: str, df: pd.DataFrame):
-    # 캐시가 너무 커지지 않도록 50개 초과 시 가장 오래된 항목 제거
-    if len(_query_cache) >= 50:
-        oldest = next(iter(_query_cache))
-        del _query_cache[oldest]
-    _query_cache[key] = df
-
+        if isinstance(blob_data, (bytes, bytearray)):
+            encoded = base64.b64encode(blob_data).decode('utf-8')
+            return f'''
+            <img src="data:image/png;base64,{encoded}"
+                 style="width: 50%; max-height: 400px; object-fit: contain; 
+                        cursor: zoom-in; border: 1px solid #ccc; border-radius: 4px; 
+                        padding: 2px; background-color: white; margin: 5px 0; display: block;"
+                 onclick="window.open(this.src)">
+            '''
+        return str(blob_data)
+    except Exception as e:
+        return str(blob_data)
 
-# --------------------------
-# 유틸
-# --------------------------
+# 🚀 [핵심 추가] 텍스트 디코딩 에러 방지용 팩토리 함수
 def safe_text_factory(x):
+    """텍스트 디코딩을 시도하고, 실패하면(PNG 등 이미지일 경우) 원본 바이트를 그대로 반환"""
     try:
         return x.decode('utf-8')
     except UnicodeDecodeError:
         return x
 
-
-def blob_to_base64_html(blob_data):
-    if blob_data is None or (isinstance(blob_data, float) and pd.isna(blob_data)):
-        return ""
-    if isinstance(blob_data, (bytes, bytearray)):
-        encoded = base64.b64encode(blob_data).decode('utf-8')
-        return (
-            f'<img src="data:image/png;base64,{encoded}" '
-            f'style="width:50%;max-height:400px;object-fit:contain;'
-            f'cursor:zoom-in;border:1px solid #ccc;border-radius:4px;'
-            f'padding:2px;background-color:white;margin:5px 0;display:block;" '
-            f'onclick="window.open(this.src)">'
-        )
-    return str(blob_data)
-
-
-def convert_blob_columns(df: pd.DataFrame) -> pd.DataFrame:
-    """BLOB이 실제로 존재하는 컬럼만 변환 (불필요한 순회 제거)."""
-    for col in df.columns:
-        # 샘플 첫 행만 확인해서 bytes 여부 판단
-        sample = df[col].dropna().head(1)
-        if not sample.empty and isinstance(sample.iloc[0], (bytes, bytearray)):
-            df[col] = df[col].apply(blob_to_base64_html)
-    return df
-
-
-_sort_key_cache: dict[str, list] = {}
-
 def natural_sort_key(s):
-    """결과를 캐싱해 동일 값 재계산 방지."""
     if s is None:
         return []
-    s = str(s)
-    if s not in _sort_key_cache:
-        _sort_key_cache[s] = [
-            int(t) if t.isdigit() else t.lower()
-            for t in re.split(r'(\d+)', s)
-        ]
-    return _sort_key_cache[s]
-
-
-def natural_sort_key_list(series: pd.Series):
-    return series.map(natural_sort_key)
-
-
-# --------------------------
-# diff (최적화)
-# --------------------------
-def highlight_diff(base: str, comp: str):
-    """SequenceMatcher 기반으로 교체 — ndiff보다 빠름."""
-    b_words = base.split()
-    c_words = comp.split()
-
-    sm = difflib.SequenceMatcher(None, b_words, c_words, autojunk=False)
-    b_res, c_res = [], []
-
-    for tag, i1, i2, j1, j2 in sm.get_opcodes():
-        if tag == 'equal':
-            b_res.extend(b_words[i1:i2])
-            c_res.extend(c_words[j1:j2])
-        elif tag == 'replace':
-            for w in b_words[i1:i2]:
-                b_res.append(f"<span style='color:#ff4d4f;font-weight:600'>{w}</span>")
-            for w in c_words[j1:j2]:
-                c_res.append(f"<span style='color:#ff4d4f;font-weight:600'>{w}</span>")
-        elif tag == 'delete':
-            for w in b_words[i1:i2]:
-                b_res.append(f"<span style='color:#ff4d4f;font-weight:600'>{w}</span>")
-        elif tag == 'insert':
-            for w in c_words[j1:j2]:
-                c_res.append(f"<span style='color:#2ecc71;font-weight:600'>{w}</span>")
-
-    return " ".join(b_res), " ".join(c_res)
-
-
-def apply_diff_row(row, base_col, comp_col):
-    base_val = str(row.get(base_col, "") or "")
-    comp_val = str(row.get(comp_col, "") or "")
-
-    if "<img" in base_val or "<img" in comp_val:
-        return pd.Series([base_val, comp_val])
-    if base_val == comp_val:
-        return pd.Series([base_val, comp_val])
-    if base_val and comp_val:
-        b, c = highlight_diff(base_val, comp_val)
-        return pd.Series([b, c])
-    return pd.Series([base_val, comp_val])
-
+    return [int(t) if t.isdigit() else t.lower()
+            for t in re.split(r'(\d+)', str(s))]
 
-# --------------------------
-# 레지스트리
-# --------------------------
 def refresh_registry_data():
     global db_registry
     db_registry = []
@@ -214,7 +67,6 @@ def refresh_registry_data():
 
     return sorted(list(set([d["standard"] for d in db_registry])))
 
-
 # --------------------------
 # 드롭다운
 # --------------------------
@@ -224,7 +76,9 @@ def on_load():
 def update_version_dd(standard):
     if not standard:
         return gr.Dropdown(choices=[])
-    versions = sorted(set(d["version"] for d in db_registry if d["standard"] == standard))
+    versions = sorted(set(
+        d["version"] for d in db_registry if d["standard"] == standard
+    ))
     return gr.Dropdown(choices=versions)
 
 def update_category_dd(standard, version):
@@ -235,7 +89,10 @@ def update_category_dd(standard, version):
 
     try:
         target = next(d for d in db_registry if d["standard"] == standard and d["version"] == version)
-        conn = get_connection(target["path"])
+        conn = sqlite3.connect(target["path"])
+        
+        # 🚀 [핵심 추가] 카테고리 불러올 때도 이미지 에러 차단
+        conn.text_factory = safe_text_factory 
 
         tables = pd.read_sql("SELECT name FROM sqlite_master WHERE type='table';", conn)['name'].tolist()
         main_t = f"{standard}_{version}"
@@ -245,26 +102,35 @@ def update_category_dd(standard, version):
         cols = pd.read_sql(f"PRAGMA table_info([{main_t}])", conn)['name'].tolist()
         lower = [c.lower() for c in cols]
 
+        # 1. 챕터/카테고리 기준 로직 (기존 유지)
         if 'chapter' in lower and 'category' in lower:
             ch = cols[lower.index('chapter')]
             ca = cols[lower.index('category')]
+
             df = pd.read_sql(f"SELECT DISTINCT [{ch}], [{ca}] FROM [{main_t}]", conn)
+
             for _, r in df.iterrows():
                 choices.append(f"{str(r[ch]).strip()}.{str(r[ca]).strip()}")
 
+        # 2. 테이블 이름 깔끔하게 자르기 (강력한 정규식 적용)
         pattern = re.compile(f"^{standard}[_\\s-]*{version}[_\\s-]*", re.IGNORECASE)
+
         for t in tables:
             if t == main_t:
                 continue
+            
+            # 대소문자, 언더바, 공백을 찰떡같이 무시하고 기준 접두사를 날려버림
             short_name = pattern.sub("", t).strip(" _")
             choices.append(short_name)
 
-    except Exception:
+        conn.close()
+
+    except Exception as e:
+        print(f"[Error in update_category_dd] standard: {standard}, version: {version}")
         traceback.print_exc()
 
     return gr.Dropdown(choices=choices)
 
-
 # --------------------------
 # 초기화
 # --------------------------
@@ -274,35 +140,74 @@ def reset_base():
 def reset_comp():
     return None, None, None
 
+# --------------------------
+# diff (하이라이팅 로직)
+# --------------------------
+def highlight_diff(base, comp):
+    base = "" if pd.isna(base) else str(base)
+    comp = "" if pd.isna(comp) else str(comp)
+
+    b_words = base.split()
+    c_words = comp.split()
+
+    d = list(difflib.ndiff(b_words, c_words))
+
+    b_res, c_res = []
+    i = 0
+
+    while i < len(d):
+        code = d[i][0]
+        word = d[i][2:]
+
+        if code == ' ':
+            b_res.append(word)
+            c_res.append(word)
+
+        elif code == '-' and i+1 < len(d) and d[i+1][0] == '+':
+            new_word = d[i+1][2:]
+            b_res.append(f"<span style='color:#ff4d4f;font-weight:600'>{word}</span>")
+            c_res.append(f"<span style='color:#ff4d4f;font-weight:600'>{new_word}</span>")
+            i += 1
+
+        elif code == '-':
+            b_res.append(f"<span style='color:#ff4d4f;font-weight:600'>{word}</span>")
+
+        elif code == '+':
+            c_res.append(f"<span style='color:#2ecc71;font-weight:600'>{word}</span>")
+
+        i += 1
+
+    return " ".join(b_res), " ".join(c_res)
 
 # --------------------------
-# 데이터 조회 (캐싱 적용)
+# 데이터 조회
 # --------------------------
 def display_data(standard, version, selection):
     if not all([standard, version, selection]):
         return pd.DataFrame({"Info": ["선택 필요"]})
 
-    # 캐시 확인
-    ck = _cache_key(standard, version, selection)
-    cached = get_cached_df(ck)
-    if cached is not None:
-        return cached.copy()
-
     try:
         target = next(d for d in db_registry if d["standard"] == standard and d["version"] == version)
-        conn = get_connection(target["path"])
+        conn = sqlite3.connect(target["path"])
+        
+        # 🚀 [핵심 추가] 데이터 조회 시 이미지 에러 완벽 차단!
+        conn.text_factory = safe_text_factory 
 
         tables = pd.read_sql(
-            "SELECT name FROM sqlite_master WHERE type='table';", conn
+            "SELECT name FROM sqlite_master WHERE type='table';",
+            conn
         )['name'].tolist()
 
         main_t = f"{standard}_{version}"
         if main_t not in tables:
             main_t = tables[0]
 
-        # 테이블 reverse lookup
+        # --------------------------
+        # 🔹 테이블 reverse lookup
+        # --------------------------
         full_table_name = None
         pattern = re.compile(f"^{standard}[_\\s-]*{version}[_\\s-]*", re.IGNORECASE)
+
         for t in tables:
             if t == selection:
                 full_table_name = t
@@ -312,23 +217,31 @@ def display_data(standard, version, selection):
                 full_table_name = t
                 break
 
-        # TableA/B 선택
+        # --------------------------
+        # 🔹 TableA/B 선택
+        # --------------------------
         if full_table_name:
             cursor = conn.cursor()
             cursor.execute(f"SELECT * FROM [{full_table_name}]")
+
             rows = cursor.fetchall()
             cols = [desc[0] for desc in cursor.description]
 
             df = pd.DataFrame(rows, columns=cols)
+
             df.columns = [c.strip() for c in df.columns]
             df = df.drop(columns=[c for c in df.columns if c.lower() == "version"], errors="ignore")
             df.columns = [c.replace("_", " ").title() for c in df.columns]
-            df = convert_blob_columns(df)
 
-            set_cached_df(ck, df)
-            return df.copy()
+            for col in df.columns:
+                df[col] = df[col].apply(blob_to_base64_html)
 
-        # 메인 테이블 구조 파악
+            conn.close()
+            return df
+
+        # --------------------------
+        # 🔹 메인 테이블 구조
+        # --------------------------
         cursor = conn.cursor()
         cursor.execute(f"PRAGMA table_info([{main_t}])")
         cols_info = cursor.fetchall()
@@ -342,48 +255,85 @@ def display_data(standard, version, selection):
             elif lc == "category":
                 real_cat = c
 
+        # --------------------------
+        # 🔹 ALL 선택
+        # --------------------------
         if selection == "ALL":
+            # 🚀 [수정] 텍스트 팩토리로 에러를 잡았으므로, 불필요한 필터 조건 삭제
             cursor.execute(f"SELECT * FROM [{main_t}]")
 
+        # --------------------------
+        # 🔹 chapter.category
+        # --------------------------
         elif "." in selection and real_ch and real_cat:
             ch, ca = selection.split(".", 1)
+
             query = f"""
             SELECT * FROM [{main_t}]
             WHERE REPLACE(TRIM(CAST([{real_ch}] AS TEXT)), ' ', '') = ?
             AND REPLACE(TRIM(CAST([{real_cat}] AS TEXT)), ' ', '') = ?
             """
+
             cursor.execute(query, (ch.replace(" ", ""), ca.replace(" ", "")))
 
+        # --------------------------
+        # 🔹 fallback
+        # --------------------------
         else:
             cursor.execute(f"SELECT * FROM [{main_t}]")
 
         rows = cursor.fetchall()
         df = pd.DataFrame(rows, columns=cols)
+
+        # --------------------------
+        # 🔹 후처리
+        # --------------------------
         df.columns = [c.lower().strip() for c in df.columns]
 
         if 'section' in df.columns and 'description' in df.columns:
             df = df[['section', 'description']]
 
-        df = convert_blob_columns(df)
+        # 🔥 핵심: blob → 이미지 변환 (삭제 X)
+        for col in df.columns:
+            df[col] = df[col].apply(blob_to_base64_html)
 
-        set_cached_df(ck, df)
-        return df.copy()
+        conn.close()
+        return df
 
     except Exception as e:
+        print(f"[Error in display_data]")
         traceback.print_exc()
         return pd.DataFrame({"Error": [str(e)]})
 
-
 # --------------------------
-# 통합 조회
+# 통합 조회 (apply로 속도 최적화)
 # --------------------------
+def apply_diff_row(row, base_col, comp_col):
+    """Pandas apply를 위한 Diff 연산 래퍼 함수"""
+    base_val = str(row.get(base_col, ""))
+    comp_val = str(row.get(comp_col, ""))
+
+    # 1. 이미지가 포함된 경우 diff 생략
+    if "<img" in base_val or "<img" in comp_val:
+        return pd.Series([base_val, comp_val])
+
+    # 2. [핵심] 두 텍스트가 완전히 똑같으면 무거운 diff 연산 생략 (Fast-path)
+    if base_val == comp_val:
+        return pd.Series([base_val, comp_val])
+
+    # 3. 내용이 서로 다를 때만 단어 단위 diff 연산 수행
+    if base_val and comp_val:
+        b, c = highlight_diff(base_val, comp_val)
+        return pd.Series([b, c])
+
+    return pd.Series([base_val, comp_val])
+
+
 def unified_search(bs, bv, bc, cs, cv, cc):
-    # 단일 조회
     if bs and bv and bc and not (cs and cv and cc):
         df = display_data(bs, bv, bc)
         return df.rename(columns={"section": "Section", "description": "Description"})
 
-    # 비교 조회
     if bs and bv and bc and cs and cv and cc:
         df_base = display_data(bs, bv, bc)
         df_comp = display_data(cs, cv, cc)
@@ -391,32 +341,52 @@ def unified_search(bs, bv, bc, cs, cv, cc):
         if 'section' not in df_base.columns or 'section' not in df_comp.columns:
             return pd.DataFrame({"Error": ["section 없음"]})
 
-        df_base = df_base.rename(columns={"section": f"Section_{bv}", "description": f"Description_{bv}"})
-        df_comp = df_comp.rename(columns={"section": f"Section_{cv}", "description": f"Description_{cv}"})
+        df_base = df_base.rename(columns={
+            "section": f"Section_{bv}",
+            "description": f"Description_{bv}"
+        })
+
+        df_comp = df_comp.rename(columns={
+            "section": f"Section_{cv}",
+            "description": f"Description_{cv}"
+        })
 
         base_sec = f"Section_{bv}"
         comp_sec = f"Section_{cv}"
         base_col = f"Description_{bv}"
         comp_col = f"Description_{cv}"
 
-        merged = pd.merge(df_base, df_comp, left_on=base_sec, right_on=comp_sec, how="outer")
+        merged = pd.merge(
+            df_base,
+            df_comp,
+            left_on=base_sec,
+            right_on=comp_sec,
+            how="outer"
+        )
+
+        # NaN 값을 빈 문자열로 처리
         merged = merged.fillna("")
 
+        # 최적화 포인트: iterrows() 대신 apply() 사용
         merged[[base_col, comp_col]] = merged.apply(
-            lambda row: apply_diff_row(row, base_col, comp_col), axis=1
+            lambda row: apply_diff_row(row, base_col, comp_col),
+            axis=1
+        )
+
+        # 정렬 개선
+        merged["__sort_key"] = merged[base_sec].where(
+            merged[base_sec] != "", merged[comp_sec]
         )
 
-        merged["__sort_key"] = merged[base_sec].where(merged[base_sec] != "", merged[comp_sec])
         merged = merged.sort_values(
             by="__sort_key",
-            key=natural_sort_key_list
+            key=lambda col: col.map(natural_sort_key)
         ).drop(columns="__sort_key")
 
         return merged
 
     return pd.DataFrame({"Info": ["선택 필요"]})
 
-
 # --------------------------
 # UI
 # --------------------------
@@ -460,7 +430,6 @@ with gr.Blocks() as demo:
     base_reset_btn.click(reset_base, None, [base_standard, base_version, base_category])
     comp_reset_btn.click(reset_comp, None, [comp_standard, comp_version, comp_category])
 
-
 # --------------------------
 # CSS
 # --------------------------
@@ -469,27 +438,63 @@ table {
     table-layout: fixed !important; 
     width: 100% !important; 
 }
+
+/* -------------------------- */
+/* 2컬럼                        */
+/* -------------------------- */
 table th:nth-last-child(2):first-child, table td:nth-last-child(2):first-child { width: 15% !important; } 
 table th:nth-last-child(1), table td:nth-last-child(1) { width: 85% !important; }
+
+/* -------------------------- */
+/* 4컬럼                        */
+/* -------------------------- */
 table th:nth-child(1):nth-last-child(4), table td:nth-child(1):nth-last-child(4) { width: 7% !important; } 
 table th:nth-child(2):nth-last-child(3), table td:nth-child(2):nth-last-child(3) { width: 43% !important; } 
 table th:nth-child(3):nth-last-child(2), table td:nth-child(3):nth-last-child(2) { width: 7% !important; } 
 table th:nth-child(4):nth-last-child(1), table td:nth-child(4):nth-last-child(1) { width: 43% !important; }
+
+/* -------------------------- */
+/* 5컬럼                        */
+/* -------------------------- */
 table th:nth-child(1):nth-last-child(5), table td:nth-child(1):nth-last-child(5) { width: 15% !important; } 
 table th:nth-child(2):nth-last-child(4), table td:nth-child(2):nth-last-child(4) { width: 10% !important; } 
 table th:nth-child(3):nth-last-child(3), table td:nth-child(3):nth-last-child(3) { width: 30% !important; } 
 table th:nth-child(4):nth-last-child(2), table td:nth-child(4):nth-last-child(2) { width: 10% !important; } 
 table th:nth-child(5):nth-last-child(1), table td:nth-child(5):nth-last-child(1) { width: 35% !important; }
+
+/* -------------------------- */
+/* 6컬럼                        */
+/* -------------------------- */
 table th:nth-child(1):nth-last-child(6), table td:nth-child(1):nth-last-child(6) { width: 8% !important; } 
 table th:nth-child(2):nth-last-child(5), table td:nth-child(2):nth-last-child(5) { width: 25% !important; } 
 table th:nth-child(3):nth-last-child(4), table td:nth-child(3):nth-last-child(4) { width: 8% !important; } 
 table th:nth-child(4):nth-last-child(3), table td:nth-child(4):nth-last-child(3) { width: 25% !important; } 
 table th:nth-child(5):nth-last-child(2), table td:nth-child(5):nth-last-child(2) { width: 8% !important; } 
 table th:nth-child(6):nth-last-child(1), table td:nth-child(6):nth-last-child(1) { width: 26% !important; }
-thead th { position: sticky; top: 0; background: white; z-index: 10; } 
-.dataframe { max-height: none !important; overflow-y: visible !important; } 
-.dataframe > div { max-height: none !important; overflow: visible !important; } 
-td { white-space: pre-wrap !important; word-break: break-word !important; line-height: 1.6; padding: 10px; } 
+
+/* -------------------------- */
+/* 기본 테이블 & 스크롤 설정  */
+/* -------------------------- */
+thead th { 
+    position: sticky;
+    top: 0; 
+    background: white; 
+    z-index: 10;
+} 
+.dataframe { 
+    max-height: none !important; 
+    overflow-y: visible !important; 
+} 
+.dataframe > div { 
+    max-height: none !important; 
+    overflow: visible !important; 
+} 
+td { 
+    white-space: pre-wrap !important;
+    word-break: break-word !important;
+    line-height: 1.6; 
+    padding: 10px; 
+} 
 """
 
 if __name__ == "__main__":