| import gradio as gr |
| import sqlite3 |
| import pandas as pd |
| import os |
| import re |
| import base64 |
| import difflib |
| import traceback |
|
|
| |
| |
| |
| UPLOAD_DIR = "uploaded_dbs" |
| if not os.path.exists(UPLOAD_DIR): |
| os.makedirs(UPLOAD_DIR) |
|
|
| db_registry = [] |
|
|
| def convert_blob_to_html_img(blob_data): |
| if blob_data is None or pd.isna(blob_data): |
| return "" |
| 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: 40%; |
| max-height: 300px; |
| object-fit: contain; |
| display: block; |
| margin: 10px 0;"> |
| ''' |
| return str(blob_data) |
| except Exception: |
| return str(blob_data) |
|
|
| def decode_sqlite_text(x): |
| try: |
| return x.decode('utf-8') |
| except UnicodeDecodeError: |
| return x |
|
|
| def fetch_available_standards(): |
| global db_registry |
| db_registry = [] |
|
|
| for file_name in os.listdir(UPLOAD_DIR): |
| if file_name.endswith(".db"): |
| name = file_name.replace(".db", "") |
| parts = name.split("_") |
| if len(parts) >= 2: |
| db_registry.append({ |
| "path": os.path.join(UPLOAD_DIR, file_name), |
| "standard": parts[0], |
| "version": parts[1] |
| }) |
|
|
| return sorted(list(set([d["standard"] for d in db_registry]))) |
|
|
|
|
| |
| |
| |
|
|
| def load_initial_standards(): |
| return gr.Dropdown(choices=fetch_available_standards()) |
|
|
| def update_version_dropdown(standard): |
| if not standard: |
| return gr.Dropdown(choices=[]) |
| versions = sorted(set(d["version"] for d in db_registry if d["standard"] == standard)) |
| return gr.Dropdown(choices=versions) |
|
|
| |
| def update_base_category_dropdown(standard, version): |
| if not standard or not version: |
| return gr.update(choices=[], value=None, interactive=False), gr.update(value="") |
|
|
| choices = ["ALL"] |
| status_value = "" |
| |
| db_path = os.path.join(UPLOAD_DIR, f"{standard}_{version}.db") |
| if not os.path.exists(db_path): |
| return gr.update(choices=choices), gr.update(value="") |
|
|
| try: |
| conn = sqlite3.connect(db_path) |
| 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 |
|
|
| if main_table: |
| |
| try: |
| cols_check = pd.read_sql(f"PRAGMA table_info([{main_table}])", conn)['name'].tolist() |
| if "Status" in cols_check or "status" in cols_check: |
| status_df = pd.read_sql(f"SELECT Status FROM [{main_table}] WHERE Status IS NOT NULL AND Status != '' LIMIT 1", conn) |
| if not status_df.empty: |
| status_value = str(status_df.iloc[0]['Status']) |
| except Exception: |
| pass |
|
|
| cols = pd.read_sql(f"PRAGMA table_info([{main_table}])", conn)['name'].tolist() |
| lower_cols = [c.lower() for c in cols] |
| if 'chapter' in lower_cols and 'category' in lower_cols: |
| ch_col = cols[lower_cols.index('chapter')] |
| ca_col = cols[lower_cols.index('category')] |
| |
| df = pd.read_sql(f"SELECT DISTINCT [{ch_col}], [{ca_col}] FROM [{main_table}]", conn) |
| for _, row in df.iterrows(): |
| ch = str(row[ch_col]).strip() |
| ca = str(row[ca_col]).strip() |
| if ch and ca and ch.lower() not in ['none', 'nan'] and ca.lower() not in ['none', 'nan']: |
| choices.append(f"{ch}.{ca}") |
|
|
| pattern = re.compile(f"^{standard}[_\\s-]*{version}[_\\s-]*", re.IGNORECASE) |
| for t in valid_tables: |
| if t == main_table: continue |
| short_name = pattern.sub("", t).strip(" _") |
| if short_name and short_name not in choices: |
| choices.append(short_name) |
| elif t not in choices: |
| choices.append(t) |
|
|
| conn.close() |
| except Exception: |
| import traceback |
| traceback.print_exc() |
| |
| return gr.update(choices=choices, value=None, interactive=True), gr.update(value=status_value) |
|
|
| def update_comp_standard_dropdown(base_std, base_ver): |
| if not base_std or not base_ver: |
| return gr.Dropdown(choices=[], value=None, interactive=False) |
|
|
| try: |
| conn = sqlite3.connect(os.path.join(UPLOAD_DIR, "mapping.db")) |
| query = "SELECT DISTINCT TRIM(Comp_std) AS Comp_std FROM Mapping_registry WHERE TRIM(Base_std)=TRIM(?) AND TRIM(Base_ver)=TRIM(?)" |
| df = pd.read_sql(query, conn, params=[base_std, base_ver]) |
| conn.close() |
|
|
| mapped_stds = sorted(df['Comp_std'].dropna().unique().tolist()) if not df.empty else [] |
| return gr.update(choices=mapped_stds, value=None, interactive=bool(mapped_stds)) |
| except Exception: |
| return gr.update(choices=[], value=None, interactive=False) |
|
|
| def update_comp_version_dropdown(base_std, base_ver, comp_std): |
| if not all([base_std, base_ver, comp_std]): |
| return gr.update(choices=[], value=None, interactive=False) |
|
|
| try: |
| conn = sqlite3.connect(os.path.join(UPLOAD_DIR, "mapping.db")) |
| query = "SELECT DISTINCT TRIM(Comp_ver) AS Comp_ver FROM Mapping_registry WHERE TRIM(Base_std)=TRIM(?) AND TRIM(Base_ver)=TRIM(?) AND TRIM(Comp_std)=TRIM(?)" |
| df = pd.read_sql(query, conn, params=[base_std, base_ver, comp_std]) |
| conn.close() |
|
|
| mapped_vers = sorted(df['Comp_ver'].dropna().unique().tolist()) if not df.empty else [] |
| return gr.update(choices=mapped_vers, value=None, interactive=bool(mapped_vers)) |
| except Exception: |
| return gr.update(choices=[], value=None, interactive=False) |
|
|
| |
| def update_comp_category_dropdown(base_std, base_ver, base_cat, comp_std, comp_ver): |
| if not all([base_std, base_ver, base_cat, comp_std, comp_ver]): |
| return gr.update(choices=[], value=None, interactive=False), gr.update(value="") |
|
|
| base_type = "Main" if not base_cat or base_cat == "ALL" or "." in base_cat else base_cat |
| status_value = "" |
|
|
| try: |
| |
| comp_db_path = os.path.join(UPLOAD_DIR, f"{comp_std}_{comp_ver}.db") |
| if os.path.exists(comp_db_path): |
| c_conn = sqlite3.connect(comp_db_path) |
| c_tables = pd.read_sql("SELECT name FROM sqlite_master WHERE type='table';", c_conn)['name'].tolist() |
| c_valid_tables = [t for t in c_tables if t.lower() not in ['mapping_registry', 'mapping_table', 'table_config', 'sqlite_sequence']] |
| c_main_table = f"{comp_std}_{comp_ver}" |
| if c_main_table not in c_valid_tables: |
| c_main_table = c_valid_tables[0] if c_valid_tables else None |
| |
| if c_main_table: |
| try: |
| cols_check = pd.read_sql(f"PRAGMA table_info([{c_main_table}])", c_conn)['name'].tolist() |
| if "Status" in cols_check or "status" in cols_check: |
| status_df = pd.read_sql(f"SELECT Status FROM [{c_main_table}] WHERE Status IS NOT NULL AND Status != '' LIMIT 1", c_conn) |
| if not status_df.empty: |
| status_value = str(status_df.iloc[0]['Status']) |
| except Exception: |
| pass |
|
|
| conn = sqlite3.connect(os.path.join(UPLOAD_DIR, "mapping.db")) |
| query = """ |
| SELECT DISTINCT TRIM(Comp_Type) AS Comp_Type |
| FROM Mapping_registry |
| WHERE TRIM(Base_std)=TRIM(?) |
| AND TRIM(Base_ver)=TRIM(?) |
| AND TRIM(Base_Type)=TRIM(?) |
| AND TRIM(Comp_std)=TRIM(?) |
| AND TRIM(Comp_ver)=TRIM(?) |
| """ |
| df = pd.read_sql(query, conn, params=[base_std, base_ver, base_type, comp_std, comp_ver]) |
| conn.close() |
|
|
| allowed_types = df['Comp_Type'].dropna().tolist() |
| if not allowed_types: |
| return gr.update(choices=[], value=None), gr.update(value=status_value) |
|
|
| final_choices = [] |
| if "Main" in allowed_types: |
| final_choices.append("ALL") |
| if os.path.exists(comp_db_path): |
| |
| c_conn = sqlite3.connect(comp_db_path) |
| if c_main_table: |
| cols = pd.read_sql(f"PRAGMA table_info([{c_main_table}])", c_conn)['name'].tolist() |
| lower_cols = [c.lower() for c in cols] |
| if 'chapter' in lower_cols and 'category' in lower_cols: |
| ch_col = cols[lower_cols.index('chapter')] |
| ca_col = cols[lower_cols.index('category')] |
| |
| c_df = pd.read_sql(f"SELECT DISTINCT [{ch_col}], [{ca_col}] FROM [{c_main_table}]", c_conn) |
| for _, row in c_df.iterrows(): |
| ch = str(row[ch_col]).strip() |
| ca = str(row[ca_col]).strip() |
| if ch and ca and ch.lower() not in ['none', 'nan'] and ca.lower() not in ['none', 'nan']: |
| final_choices.append(f"{ch}.{ca}") |
| c_conn.close() |
| |
| for t in allowed_types: |
| if t != "Main": final_choices.append(t) |
|
|
| return gr.update(choices=final_choices, value=final_choices[0] if final_choices else None, interactive=True), gr.update(value=status_value) |
| |
| except Exception as e: |
| import traceback |
| traceback.print_exc() |
| return gr.update(choices=[], value=None), gr.update(value="") |
|
|
| def reset_base_selections(): |
| |
| return gr.update(value=None), gr.update(choices=[], value=None), gr.update(choices=[], value=None), gr.update(value="") |
|
|
| def reset_comp_selections(): |
| |
| return gr.update(value=None), gr.update(choices=[], value=None), gr.update(choices=[], value=None), gr.update(value="") |
|
|
|
|
| |
| |
| |
| 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") |
| |
| |
| |
| |
| 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(): |
| |
| 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): |
| if not cat or cat == "ALL": return "Main" |
| if "." in cat: return "Main" |
| return cat |
|
|
| type_b = get_type_by_cat(base_cat) |
| type_c = get_type_by_cat(comp_cat) |
| |
| def apply_visual_merge(df, cols): |
| if not df.empty and len(cols) > 1: |
| is_dup = pd.Series([True] * len(df), index=df.index) |
| for col in cols: |
| if col in df.columns: |
| curr = df[col].astype(str).str.strip() |
| match = (curr == curr.shift(1)) & (~curr.isin(["", "nan", "None", " "])) |
| is_dup = is_dup & match |
| df.loc[is_dup, col] = " " |
| return df |
|
|
| def combine_code_desc(df): |
| cols = list(df.columns) |
| new_cols = [] |
| processed = set() |
| for col in cols: |
| if col in processed: continue |
| if "_Code" in col: |
| desc_col = col.replace("_Code", "_Description") |
| if desc_col in cols: |
| new_col_name = col.replace("_Code", "") |
| def combine_cells(row): |
| c, d = str(row[col]).strip(), str(row[desc_col]).strip() |
| if c in ["nan", "None", "", " "]: return d |
| if d in ["nan", "None", "", " "]: return f"<span style='font-weight:bold; color:#1a73e8;'>{c}</span>" |
| return f"<span style='font-weight:bold; color:#1a73e8; display:block; margin-bottom:4px;'>{c}</span>{d}" |
| df[new_col_name] = df.apply(combine_cells, axis=1) |
| new_cols.append(new_col_name) |
| processed.update([col, desc_col]) |
| else: new_cols.append(col) |
| elif "_Description" in col: |
| if col.replace("_Description", "_Code") not in cols: new_cols.append(col) |
| else: new_cols.append(col) |
| return df[new_cols] |
|
|
| if base_std and base_ver and base_cat and (not comp_std or not comp_ver or not comp_cat): |
| df, _ = fetch_database_records(base_std, base_ver, base_cat, type_b) |
| if "Error" in df.columns: return df |
| return apply_visual_merge(df, df.columns) |
|
|
| if all([base_std, base_ver, base_cat, comp_std, comp_ver, comp_cat]): |
| df_base, real_anchors_b = fetch_database_records(base_std, base_ver, base_cat, type_b) |
| df_comp, real_anchors_c = fetch_database_records(comp_std, comp_ver, comp_cat, type_c) |
|
|
| if "Error" in df_base.columns: return df_base |
| if "Error" in df_comp.columns: return df_comp |
|
|
| for ra in real_anchors_b: |
| if ra not in df_base.columns: |
| return pd.DataFrame({"Error": [f"기준 열(Anchor) '{ra}'이(가) 기준 데이터에 존재하지 않습니다. Table_Config를 확인하세요."]}) |
| for ra in real_anchors_c: |
| if ra not in df_comp.columns: |
| return pd.DataFrame({"Error": [f"비교 열(Anchor) '{ra}'이(가) 비교 데이터에 존재하지 않습니다. Table_Config를 확인하세요."]}) |
|
|
| def clean_key_val(v): |
| s = str(v).strip() |
| if s.endswith('.0') and s[:-2].isdigit(): |
| s = s[:-2] |
| return s.replace(" ", "") |
|
|
| 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)=?" |
| 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)=?" |
| 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} |
| |
| if 'base_type' in cols_fw_lower and 'comp_type' in cols_fw_lower: |
| b_col = cols_fw_lower['base_type'] |
| c_col = cols_fw_lower['comp_type'] |
| |
| df_fw[b_col] = df_fw[b_col].fillna('Main').astype(str).str.strip().str.upper() |
| df_fw[c_col] = df_fw[c_col].fillna('Main').astype(str).str.strip().str.upper() |
| df_fw = df_fw[(df_fw[b_col] == type_b.strip().upper()) & (df_fw[c_col] == type_c.strip().upper())] |
| |
| if not df_rv.empty: |
| df_rv[b_col] = df_rv[b_col].fillna('Main').astype(str).str.strip().str.upper() |
| df_rv[c_col] = df_rv[c_col].fillna('Main').astype(str).str.strip().str.upper() |
| df_rv = df_rv[(df_rv[c_col] == type_b.strip().upper()) & (df_rv[b_col] == type_c.strip().upper())] |
|
|
| b_sec = cols_fw_lower.get('base_section', 'Base_section') |
| c_sec = cols_fw_lower.get('comp_section', 'Comp_section') |
| |
| df_fw = df_fw[[b_sec, c_sec]].rename(columns={b_sec: 'Base_section', c_sec: 'Comp_section'}) |
| if not df_rv.empty: |
| df_rv = df_rv[[b_sec, c_sec]].rename(columns={b_sec: 'Comp_section', c_sec: 'Base_section'}) |
| else: |
| df_rv = pd.DataFrame(columns=['Base_section', 'Comp_section']) |
|
|
| df_mapping = pd.concat([df_fw, df_rv], ignore_index=True) |
|
|
| if not df_mapping.empty: |
| df_mapping['Base_section'] = df_mapping['Base_section'].astype(str).str.replace('\n', ',').str.split(',') |
| df_mapping['Comp_section'] = df_mapping['Comp_section'].astype(str).str.replace('\n', ',').str.split(',') |
| df_mapping = df_mapping.explode('Base_section').explode('Comp_section') |
| |
| df_mapping['Base_section'] = df_mapping['Base_section'].apply(clean_key_val) |
| df_mapping['Comp_section'] = df_mapping['Comp_section'].apply(clean_key_val) |
| df_mapping = df_mapping.dropna().drop_duplicates() |
| else: |
| df_mapping = pd.DataFrame(columns=['Base_section', 'Comp_section']) |
|
|
| if base_std.strip().upper() == comp_std.strip().upper() and type_b.strip().upper() == type_c.strip().upper(): |
| implicit = pd.DataFrame({'Base_section': list(set(df_base['merge_key']) & set(df_comp['merge_key'])), |
| 'Comp_section': list(set(df_base['merge_key']) & set(df_comp['merge_key']))}) |
| bridge = pd.concat([df_mapping, implicit], ignore_index=True).drop_duplicates() |
| else: |
| bridge = df_mapping.copy() |
|
|
| rename_b = {c: f"{c}_{base_ver}" for c in df_base.columns if c != 'merge_key'} |
| df_base = df_base.rename(columns=rename_b) |
| |
| rename_c = {c: f"{c}_{comp_ver}" for c in df_comp.columns if c != 'merge_key'} |
| df_comp = df_comp.rename(columns=rename_c) |
|
|
| df_base['base_idx'], df_comp['comp_idx'] = range(len(df_base)), range(len(df_comp)) |
| merged = pd.merge(bridge, df_base, left_on='Base_section', right_on='merge_key', how='outer') |
| merged = pd.merge(merged, df_comp, left_on='Comp_section', right_on='merge_key', how='outer') |
| |
| merged['base_idx'] = merged['base_idx'].fillna(float('inf')) |
| merged['comp_idx'] = merged['comp_idx'].fillna(float('inf')) |
| merged = merged.sort_values(['base_idx', 'comp_idx']) |
|
|
| result_rows = [] |
| for _, row in merged.iterrows(): |
| 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') |
|
|
| if mapped_only and not (has_b and has_c): |
| continue |
| |
| 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 has_b and has_c: |
| for orig_col in rename_b.keys(): |
| if 'description' in orig_col.lower() and rename_c.get(orig_col) in row_dict: |
| b_v, c_v = row_dict[rename_b[orig_col]], row_dict[rename_c[orig_col]] |
| if b_v and c_v and "<img" not in b_v and "<img" not in c_v and b_v != c_v: |
| row_dict[rename_b[orig_col]], row_dict[rename_c[orig_col]] = generate_html_diff(b_v, c_v) |
| result_rows.append(row_dict) |
|
|
| final_df = combine_code_desc(pd.DataFrame(result_rows)) |
|
|
| 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] |
| |
| 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}")] |
|
|
| final_df = apply_visual_merge(final_df, b_cols_final) |
| final_df = apply_visual_merge(final_df, c_cols_final) |
|
|
| return final_df |
|
|
| return pd.DataFrame({"Info": ["조건을 선택하세요."]}) |
| except Exception as e: |
| import traceback |
| traceback.print_exc() |
| return pd.DataFrame({"Error": [f"시스템 오류 발생: {str(e)}"]}) |
|
|
|
|
| |
| |
| |
| with gr.Blocks() as demo: |
| gr.Markdown("# 📜 Regulation Viewer") |
|
|
| with gr.Row(): |
| with gr.Accordion("📌 기준 법규", open=True): |
| with gr.Column(): |
| base_standard = gr.Dropdown(label="Standard") |
| base_version = gr.Dropdown(label="Version") |
| base_status = gr.Textbox(label="Status", interactive=False, lines=1) |
| |
| with gr.Row(elem_classes="reset-row"): |
| base_category = gr.Dropdown(label="Category", scale=4) |
| |
| base_reset_btn = gr.Button("↺ 초기화", scale=1, elem_classes="reset-btn") |
| |
| with gr.Accordion("🔄 비교 법규", open=False): |
| with gr.Column(): |
| comp_standard = gr.Dropdown(label="Standard", choices=[]) |
| comp_version = gr.Dropdown(label="Version", choices=[]) |
| comp_status = gr.Textbox(label="Status", interactive=False, lines=1) |
| |
| with gr.Row(elem_classes="reset-row"): |
| comp_category = gr.Dropdown(label="Category", choices=[], scale=4) |
| |
| comp_reset_btn = gr.Button("↺ 초기화", scale=1, elem_classes="reset-btn") |
|
|
| with gr.Row(elem_id="search_row"): |
| search_btn = gr.Button("🔍 조회", variant="primary", scale=10) |
| |
| mapped_only_cb = gr.Checkbox(label="🔗 매핑된 항목만 보기", value=False, elem_id="mapped_cb_item", container=False, scale=1) |
| diff_filter_cb = gr.Checkbox(label="💡 변경된 내용만 보기", value=False, elem_id="diff_cb_item", container=False, scale=1) |
|
|
| output_df = gr.Dataframe(wrap=True, interactive=False, datatype="html", max_height=800) |
|
|
| demo.load(fn=load_initial_standards, inputs=None, outputs=base_standard) |
|
|
| base_standard.change(fn=update_version_dropdown, inputs=[base_standard], outputs=[base_version]) |
| base_version.change(fn=update_base_category_dropdown, inputs=[base_standard, base_version], outputs=[base_category, base_status]) |
|
|
| base_version.change(fn=update_comp_standard_dropdown, inputs=[base_standard, base_version], outputs=[comp_standard]) |
| comp_standard.change(fn=update_comp_version_dropdown, inputs=[base_standard, base_version, comp_standard], outputs=[comp_version]) |
|
|
| comp_change_triggers = [base_standard, base_version, base_category, comp_standard, comp_version] |
| |
| base_category.change(fn=update_comp_category_dropdown, inputs=comp_change_triggers, outputs=[comp_category, comp_status]) |
| comp_version.change(fn=update_comp_category_dropdown, inputs=comp_change_triggers, outputs=[comp_category, comp_status]) |
|
|
| search_btn.click( |
| fn=execute_unified_search, |
| inputs=[base_standard, base_version, base_category, comp_standard, comp_version, comp_category, mapped_only_cb, diff_filter_cb], |
| outputs=[output_df] |
| ) |
| |
| mapped_only_cb.change( |
| fn=execute_unified_search, |
| inputs=[base_standard, base_version, base_category, comp_standard, comp_version, comp_category, mapped_only_cb, diff_filter_cb], |
| outputs=[output_df] |
| ) |
|
|
| diff_filter_cb.change( |
| fn=execute_unified_search, |
| inputs=[base_standard, base_version, base_category, comp_standard, comp_version, comp_category, mapped_only_cb, diff_filter_cb], |
| outputs=[output_df] |
| ) |
|
|
| base_reset_btn.click(fn=reset_base_selections, inputs=None, outputs=[base_standard, base_version, base_category, base_status]) |
| comp_reset_btn.click(fn=reset_comp_selections, inputs=None, outputs=[comp_standard, comp_version, comp_category, comp_status]) |
|
|
|
|
| |
| |
| |
| css = """ |
| .reset-row { |
| align-items: flex-end !important; |
| margin-bottom: 5px !important; |
| } |
| .reset-btn { |
| margin-bottom: 10px !important; |
| } |
| |
| table { |
| table-layout: auto !important; |
| width: max-content !important; |
| min-width: 100% !important; |
| } |
| |
| th, td { |
| min-width: 150px; |
| } |
| |
| table:has(th:nth-last-child(2):first-child), |
| table:has(th:nth-last-child(3):first-child), |
| table:has(th:nth-last-child(4):first-child), |
| table:has(th:nth-last-child(5):first-child), |
| table:has(th:nth-last-child(6):first-child) { |
| table-layout: fixed !important; |
| width: 100% !important; |
| } |
| |
| table th:nth-last-child(2):first-child, table td:nth-last-child(2):first-child { width: 15% !important; min-width: 0 !important; } |
| table th:nth-last-child(1), table td:nth-last-child(1) { width: 85% !important; min-width: 0 !important; } |
| |
| table th:nth-child(1):nth-last-child(3), table td:nth-child(1):nth-last-child(3) { width: 20% !important; min-width: 0 !important; } |
| table th:nth-child(2):nth-last-child(2), table td:nth-child(2):nth-last-child(2) { width: 30% !important; min-width: 0 !important; } |
| table th:nth-child(3):nth-last-child(1), table td:nth-child(3):nth-last-child(1) { width: 50% !important; min-width: 0 !important; } |
| |
| table th:nth-child(1):nth-last-child(4), table td:nth-child(1):nth-last-child(4) { width: 10% !important; min-width: 0 !important; } |
| table th:nth-child(2):nth-last-child(3), table td:nth-child(2):nth-last-child(3) { width: 40% !important; min-width: 0 !important; } |
| table th:nth-child(3):nth-last-child(2), table td:nth-child(3):nth-last-child(2) { width: 10% !important; min-width: 0 !important; } |
| table th:nth-child(4):nth-last-child(1), table td:nth-child(4):nth-last-child(1) { width: 40% !important; min-width: 0 !important; } |
| |
| table th:nth-child(1):nth-last-child(5), table td:nth-child(1):nth-last-child(5) { width: 15% !important; min-width: 0 !important; } |
| table th:nth-child(2):nth-last-child(4), table td:nth-child(2):nth-last-child(4) { width: 15% !important; min-width: 0 !important; } |
| table th:nth-child(3):nth-last-child(3), table td:nth-child(3):nth-last-child(3) { width: 15% !important; min-width: 0 !important; } |
| table th:nth-child(4):nth-last-child(2), table td:nth-child(4):nth-last-child(2) { width: 15% !important; min-width: 0 !important; } |
| table th:nth-child(5):nth-last-child(1), table td:nth-child(5):nth-last-child(1) { width: 40% !important; min-width: 0 !important; } |
| |
| table th:nth-child(1):nth-last-child(6), table td:nth-child(1):nth-last-child(6) { width: 8% !important; min-width: 0 !important; } |
| table th:nth-child(2):nth-last-child(5), table td:nth-child(2):nth-last-child(5) { width: 15% !important; min-width: 0 !important; } |
| table th:nth-child(3):nth-last-child(4), table td:nth-child(3):nth-last-child(4) { width: 27% !important; min-width: 0 !important; } |
| table th:nth-child(4):nth-last-child(3), table td:nth-child(4):nth-last-child(3) { width: 8% !important; min-width: 0 !important; } |
| table th:nth-child(5):nth-last-child(2), table td:nth-child(5):nth-last-child(2) { width: 15% !important; min-width: 0 !important; } |
| table th:nth-child(6):nth-last-child(1), table td:nth-child(6):nth-last-child(1) { width: 27% !important; min-width: 0 !important; } |
| |
| thead th { |
| font-size: 18px !important; |
| position: sticky; |
| top: 0; |
| background: white; |
| z-index: 10; |
| } |
| .dataframe { |
| max-height: none !important; |
| overflow-y: visible !important; |
| overflow-x: auto !important; |
| display: block; |
| } |
| .dataframe > div { |
| max-height: none !important; |
| overflow: visible !important; |
| } |
| td { |
| font-size: 18px !important; |
| white-space: pre-wrap !important; |
| word-break: keep-all !important; |
| line-height: 1.6; |
| padding: 10px; |
| vertical-align: top !important; |
| text-align: left !important; |
| } |
| td img { |
| display: block; |
| max-width: none !important; |
| } |
| |
| #search_row { |
| align-items: center !important; |
| margin-bottom: 5px !important; |
| } |
| |
| #diff_cb_item, #mapped_cb_item { |
| margin-top: 0 !important; |
| padding-left: 15px !important; |
| width: max-content !important; |
| min-width: max-content !important; |
| flex-grow: 0 !important; |
| } |
| """ |
|
|
| if __name__ == "__main__": |
| demo.launch( |
| theme=gr.themes.Soft(), |
| share=True, |
| css=css |
| ) |