Update app.py

app.py

@@ -6,6 +6,8 @@ import re
 import base64
 import difflib
 import traceback
+from functools import lru_cache
+from threading import Lock
 
 # --------------------------
 # 1️⃣ 저장소 설정
@@ -17,39 +19,184 @@ if not os.path.exists(UPLOAD_DIR):
 db_registry = []
 
 # --------------------------
-# 유틸
+# 2️⃣ DB 연결 캐싱 (핵심 개선)
 # --------------------------
-def blob_to_base64_html(blob_data):
-    if blob_data is None or pd.isna(blob_data):
-        return ""
+_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):
+    """자주 조회되는 컬럼에 인덱스가 없으면 자동 생성."""
     try:
-        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)
+        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
+
+
+# --------------------------
+# 유틸
+# --------------------------
 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 []
-    return [int(t) if t.isdigit() else t.lower()
-            for t in re.split(r'(\d+)', str(s))]
+    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])
+
 
+# --------------------------
+# 레지스트리
+# --------------------------
 def refresh_registry_data():
     global db_registry
     db_registry = []
@@ -67,6 +214,7 @@ def refresh_registry_data():
 
     return sorted(list(set([d["standard"] for d in db_registry])))
 
+
 # --------------------------
 # 드롭다운
 # --------------------------
@@ -76,9 +224,7 @@ 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):
@@ -89,10 +235,7 @@ def update_category_dd(standard, version):
 
     try:
         target = next(d for d in db_registry if d["standard"] == standard and d["version"] == version)
-        conn = sqlite3.connect(target["path"])
-        
-        # 🚀 [핵심 추가] 카테고리 불러올 때도 이미지 에러 차단
-        conn.text_factory = safe_text_factory 
+        conn = get_connection(target["path"])
 
         tables = pd.read_sql("SELECT name FROM sqlite_master WHERE type='table';", conn)['name'].tolist()
         main_t = f"{standard}_{version}"
@@ -102,35 +245,26 @@ 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)
 
-        conn.close()
-
-    except Exception as e:
-        print(f"[Error in update_category_dd] standard: {standard}, version: {version}")
+    except Exception:
         traceback.print_exc()
 
     return gr.Dropdown(choices=choices)
 
+
 # --------------------------
 # 초기화
 # --------------------------
@@ -140,74 +274,35 @@ 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 = sqlite3.connect(target["path"])
-        
-        # 🚀 [핵심 추가] 데이터 조회 시 이미지 에러 완벽 차단!
-        conn.text_factory = safe_text_factory 
+        conn = get_connection(target["path"])
 
         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
@@ -217,31 +312,23 @@ 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)
 
-            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
-
-        # --------------------------
-        # 🔹 메인 테이블 구조
-        # --------------------------
+        # 메인 테이블 구조 파악
         cursor = conn.cursor()
         cursor.execute(f"PRAGMA table_info([{main_t}])")
         cols_info = cursor.fetchall()
@@ -255,85 +342,48 @@ 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']]
 
-        # 🔥 핵심: blob → 이미지 변환 (삭제 X)
-        for col in df.columns:
-            df[col] = df[col].apply(blob_to_base64_html)
+        df = convert_blob_columns(df)
 
-        conn.close()
-        return df
+        set_cached_df(ck, df)
+        return df.copy()
 
     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)
@@ -341,52 +391,32 @@ 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"
-        )
-
-        # NaN 값을 빈 문자열로 처리
+        merged = pd.merge(df_base, df_comp, left_on=base_sec, right_on=comp_sec, how="outer")
         merged = merged.fillna("")
 
-        # 최적화 포인트: iterrows() 대신 apply() 사용
         merged[[base_col, comp_col]] = merged.apply(
-            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]
+            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 = merged.sort_values(
             by="__sort_key",
-            key=lambda col: col.map(natural_sort_key)
+            key=natural_sort_key_list
         ).drop(columns="__sort_key")
 
         return merged
 
     return pd.DataFrame({"Info": ["선택 필요"]})
 
+
 # --------------------------
 # UI
 # --------------------------
@@ -430,6 +460,7 @@ 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
 # --------------------------
@@ -438,63 +469,27 @@ 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__":