Update app.py

app.py

@@ -249,147 +249,6 @@ def reset_comp_selections():
     return gr.update(value=None), gr.update(choices=[], value=None), gr.update(choices=[], value=None), gr.update(value="")
 
 
-# ==========================================
-# 3. Core Logic & Data Processing
-# ==========================================
-def generate_html_diff(base_text, comp_text):
-    base_text = "" if pd.isna(base_text) else str(base_text)
-    comp_text = "" if pd.isna(comp_text) else str(comp_text)
-
-    b_words = base_text.split()
-    c_words = comp_text.split()
-
-    diff_generator = list(difflib.ndiff(b_words, c_words))
-    b_result, c_result = [], []
-    i = 0
-
-    while i < len(diff_generator):
-        code = diff_generator[i][0]
-        word = diff_generator[i][2:]
-
-        if code == ' ':
-            b_result.append(word)
-            c_result.append(word)
-        elif code == '-' and i+1 < len(diff_generator) and diff_generator[i+1][0] == '+':
-            new_word = diff_generator[i+1][2:]
-            b_result.append(f"<span style='color:#ff4d4f;font-weight:600'>{word}</span>")
-            c_result.append(f"<span style='color:#ff4d4f;font-weight:600'>{new_word}</span>")
-            i += 1
-        elif code == '-':
-            b_result.append(f"<span style='color:#ff4d4f;font-weight:600'>{word}</span>")
-        elif code == '+':
-            c_result.append(f"<span style='color:#2ecc71;font-weight:600'>{word}</span>")
-
-        i += 1
-
-    return " ".join(b_result), " ".join(c_result)
-
-def fetch_database_records(standard, version, selection, table_type):
-    if not all([standard, version, selection]):
-        return pd.DataFrame({"Info": ["선택 필요"]}), []
-
-    try:
-        db_path = os.path.join(UPLOAD_DIR, f"{standard}_{version}.db")
-        if not os.path.exists(db_path):
-            return pd.DataFrame({"Error": [f"파일을 찾을 수 없습니다: {db_path}"]}), []
-            
-        conn = sqlite3.connect(db_path)
-        conn.text_factory = decode_sqlite_text 
-
-        tables = pd.read_sql("SELECT name FROM sqlite_master WHERE type='table';", conn)['name'].tolist()
-        valid_tables = [t for t in tables if t.lower() not in ['mapping_registry', 'mapping_table', 'table_config', 'sqlite_sequence']]
-        
-        main_table = f"{standard}_{version}"
-        if main_table not in valid_tables:
-            main_table = valid_tables[0] if valid_tables else None
-
-        target_table = None
-        pattern = re.compile(f"^{standard}[_\\s-]*{version}[_\\s-]*", re.IGNORECASE)
-
-        for t in valid_tables:
-            if t == selection or pattern.sub("", t).strip(" _") == selection:
-                target_table = t
-                break
-
-        if target_table and target_table != main_table:
-            df = pd.read_sql(f"SELECT * FROM [{target_table}]", conn)
-        else:
-            if not main_table:
-                return pd.DataFrame({"Error": ["데이터베이스 내에서 메인 테이블을 찾을 수 없습니다."]}), []
-                
-            cursor = conn.cursor()
-            cursor.execute(f"PRAGMA table_info([{main_table}])")
-            cols = [c[1] for c in cursor.fetchall()]
-            lower_cols = [c.lower().strip() for c in cols]
-
-            real_ch, real_cat = None, None
-            for c, lc in zip(cols, lower_cols):
-                if lc == "chapter": real_ch = c
-                elif lc == "category": real_cat = c
-
-            if selection == "ALL":
-                cursor.execute(f"SELECT * FROM [{main_table}]")
-            elif "." in selection and real_ch and real_cat:
-                ch, ca = selection.split(".", 1)
-                query = f"SELECT * FROM [{main_table}] WHERE REPLACE(TRIM(CAST([{real_ch}] AS TEXT)), ' ', '') = ? AND REPLACE(TRIM(CAST([{real_cat}] AS TEXT)), ' ', '') = ?"
-                cursor.execute(query, (ch.replace(" ", ""), ca.replace(" ", "")))
-            else:
-                cursor.execute(f"SELECT * FROM [{main_table}]")
-            
-            df = pd.DataFrame(cursor.fetchall(), columns=cols)
-
-        df = df.drop(columns=[c for c in df.columns if c.lower() == "version"], errors="ignore")
-        
-        # =========================================================
-        # 💡 [스마트 스캔 로직] DB 컬럼을 보고 맞는 Main 설정을 알아서 찾습니다.
-        # =========================================================
-        conn_map = sqlite3.connect(os.path.join(UPLOAD_DIR, "mapping.db"))
-        config_df = pd.read_sql("SELECT Anchor_Column, Display_Columns FROM Table_Config WHERE Table_Type=?", conn_map, params=[table_type])
-        conn_map.close()
-
-        anchor_col_config = "Section"
-        display_setting = "Description"
-
-        if not config_df.empty:
-            df_cols_lower = [c.lower() for c in df.columns]
-            for _, row in config_df.iterrows():
-                # DB에 적힌 Anchor 열이 실제 데이터프레임에 존재하는지 확인
-                first_anchor = str(row['Anchor_Column']).split(',')[0].strip().lower()
-                if first_anchor in df_cols_lower:
-                    anchor_col_config = str(row['Anchor_Column'])
-                    raw_disp = row['Display_Columns']
-                    display_setting = str(raw_disp) if pd.notna(raw_disp) and str(raw_disp).strip() != "" else None
-                    break
-        # =========================================================
-
-        anchor_cols_config = [x.strip() for x in anchor_col_config.split(',')]
-
-        real_anchors = []
-        for ac in anchor_cols_config:
-            real_ac = next((c for c in df.columns if c.lower() == ac.lower()), ac)
-            real_anchors.append(real_ac)
-
-        if display_setting:
-            cols_to_show = [c.strip() for c in display_setting.split(',')]
-            actual_cols_to_show = [c for c in df.columns if next((True for req in cols_to_show if c.lower() == req.lower()), False)]
-            
-            for ra in reversed(real_anchors):
-                if ra in df.columns and ra not in actual_cols_to_show:
-                    actual_cols_to_show.insert(0, ra)
-            
-            if actual_cols_to_show:
-                df = df[actual_cols_to_show]
-
-        for col in df.columns:
-            df[col] = df[col].apply(convert_blob_to_html_img)
-
-        return df, real_anchors 
-
-    except Exception as e:
-        import traceback
-        traceback.print_exc()
-        return pd.DataFrame({"Error": [str(e)]}), []
-
 def execute_unified_search(base_std, base_ver, base_cat, comp_std, comp_ver, comp_cat, mapped_only, diff_only):
     try:
         def get_type_by_cat(cat):
@@ -463,14 +322,28 @@ def execute_unified_search(base_std, base_ver, base_cat, comp_std, comp_ver, com
             df_base['merge_key'] = df_base[real_anchors_b].apply(lambda row: '-'.join([clean_key_val(x) for x in row]), axis=1)
             df_comp['merge_key'] = df_comp[real_anchors_c].apply(lambda row: '-'.join([clean_key_val(x) for x in row]), axis=1)
 
+            # =========================================================
+            # 💡 [핵심 업그레이드] 매핑 라우터: 어떤 장부를 읽을지 결정합니다!
+            # =========================================================
             conn_map = sqlite3.connect(os.path.join(UPLOAD_DIR, "mapping.db"))
             
-            q_fw = "SELECT * FROM Mapping_table WHERE TRIM(Base_std)=? AND TRIM(Base_ver)=? AND TRIM(Comp_std)=? AND TRIM(Comp_ver)=?"
+            registry_query = """
+                SELECT Target_Table FROM Mapping_registry 
+                WHERE TRIM(Base_std)=? AND TRIM(Base_ver)=? AND TRIM(Comp_std)=? AND TRIM(Comp_ver)=?
+                LIMIT 1
+            """
+            reg_df = pd.read_sql(registry_query, conn_map, params=[base_std.strip(), base_ver.strip(), comp_std.strip(), comp_ver.strip()])
+            
+            # DB에 Target_Table이 적혀있으면 그걸 쓰고, 없거나 비어있으면 기본 장부(Mapping_table)를 씁니다.
+            target_table_name = reg_df.iloc[0]['Target_Table'] if not reg_df.empty and pd.notna(reg_df.iloc[0]['Target_Table']) else "Mapping_table"
+
+            q_fw = f"SELECT * FROM [{target_table_name}] WHERE TRIM(Base_std)=? AND TRIM(Base_ver)=? AND TRIM(Comp_std)=? AND TRIM(Comp_ver)=?"
             df_fw = pd.read_sql(q_fw, conn_map, params=[base_std.strip(), base_ver.strip(), comp_std.strip(), comp_ver.strip()])
             
-            q_rv = "SELECT * FROM Mapping_table WHERE TRIM(Comp_std)=? AND TRIM(Comp_ver)=? AND TRIM(Base_std)=? AND TRIM(Base_ver)=?"
+            q_rv = f"SELECT * FROM [{target_table_name}] WHERE TRIM(Comp_std)=? AND TRIM(Comp_ver)=? AND TRIM(Base_std)=? AND TRIM(Base_ver)=?"
             df_rv = pd.read_sql(q_rv, conn_map, params=[base_std.strip(), base_ver.strip(), comp_std.strip(), comp_ver.strip()])
             conn_map.close()
+            # =========================================================
 
             cols_fw_lower = {c.lower(): c for c in df_fw.columns}
             
@@ -535,17 +408,21 @@ def execute_unified_search(base_std, base_ver, base_cat, comp_std, comp_ver, com
                 row_dict = {}
                 has_b = not pd.isna(row.get('base_idx')) and row.get('base_idx') != float('inf')
                 has_c = not pd.isna(row.get('comp_idx')) and row.get('comp_idx') != float('inf')
-
+                
                 for c in rename_b.values(): row_dict[c] = str(row[c]) if has_b and not pd.isna(row[c]) else ""
                 for c in rename_c.values(): row_dict[c] = str(row[c]) if has_c and not pd.isna(row[c]) else ""
-
+                
+                # =======================================================
+                # 💡 [깐깐한 빈칸 검사] 양쪽 다 글자가 채워져 있는지 확인!
+                # =======================================================
                 if mapped_only:
                     b_has_val = any(str(row_dict[c]).strip() not in ["", "nan", "None", "&nbsp;"] for c in rename_b.values())
                     c_has_val = any(str(row_dict[c]).strip() not in ["", "nan", "None", "&nbsp;"] for c in rename_c.values())
                     
                     if not (b_has_val and c_has_val):
                         continue
-
+                # =======================================================
+                
                 if has_b and has_c:
                     for orig_col in rename_b.keys():
                         if ('description' in orig_col.lower() or '내용' in orig_col) and rename_c.get(orig_col) in row_dict:
@@ -562,6 +439,8 @@ def execute_unified_search(base_std, base_ver, base_cat, comp_std, comp_ver, com
             if diff_only:
                 mask = final_df.astype(str).apply(lambda col: col.str.contains('color:#ff4d4f|color:#2ecc71', case=False, regex=True)).any(axis=1)
                 final_df = final_df[mask]
+                if final_df.empty:
+                    return pd.DataFrame({"Info": ["💡 선택하신 조건 간에 변경된 내용이 없습니다. (100% 동일)"]})
             
             b_cols_final = [c for c in final_df.columns if c.endswith(f"_{base_ver}")]
             c_cols_final = [c for c in final_df.columns if c.endswith(f"_{comp_ver}")]