import streamlit as st import pandas as pd import io import os import json import sys import warnings import platform import imaplib import email import html import socket from imaplib import IMAP4_SSL from email.header import decode_header, make_header from datetime import datetime from docx import Document from docx.shared import Pt, RGBColor, Inches from docx.enum.text import WD_ALIGN_PARAGRAPH from docx.oxml.ns import qn from docx.oxml import OxmlElement import pdfplumber import vertexai import requests import copy import re import sqlite3 def get_config_value(key, default=""): value = os.environ.get(key) if value is not None: return value try: return st.secrets.get(key, default) except Exception: return default def parse_allowed_users(raw_users): if hasattr(raw_users, "items"): return {str(email).strip().lower(): str(password) for email, password in raw_users.items()} users = {} for item in str(raw_users or "").split(","): email_addr, separator, password = item.partition("=") email_addr = email_addr.strip().strip('"').strip("'").lower() if separator and email_addr: users[email_addr] = password.strip().strip('"').strip("'") return users # 1. 인증 및 Secrets 통합 로드 os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = "credentials.json" # 2. 모든 필수 변수 한 번에 선언 (변수명 일치 확인) COMMON_PW = get_config_value("EMAIL_PASSWORD") MASTER_PW = get_config_value("MASTER_PASSWORD") ADMIN_EMAIL = get_config_value("ADMIN_EMAIL").strip().lower() PROJECT_ID = get_config_value("PROJECT_ID") NOTION_TOKEN = get_config_value("NOTION_API_TOKEN") NOTION_DB_ID = get_config_value("NOTION_DATABASE_ID") # 3. 리스트/딕셔너리 변환 로직 ALL_MAILBOXES = [item.strip() for item in get_config_value("ALL_MAILBOXES").split(",") if item.strip()] allowed_users = parse_allowed_users(get_config_value("ALLOWED_USERS")) MAX_EMAILS_PER_FOLDER = 250 GEMINI_25_FLASH_INPUT_USD_PER_1M = 0.30 GEMINI_25_FLASH_OUTPUT_USD_PER_1M = 2.50 ESTIMATED_OUTPUT_TOKENS_PER_STAGE = 350 def save_rules_to_db(df): conn = sqlite3.connect("data.db") df.to_sql("compliance_rules", conn, if_exists="replace", index=False) conn.close() def load_rules_from_db(): conn = sqlite3.connect("data.db") try: df = pd.read_sql("SELECT * FROM compliance_rules", conn) except: df = pd.DataFrame() conn.close() return df # [1] System Settings & OS Detection OS_TYPE = platform.system() warnings.filterwarnings("ignore") if OS_TYPE == "Windows": try: import win32com.client except ImportError: pass st.set_page_config(page_title="Flux Finance Internal Note", layout="wide") # 🎨 UI Style st.markdown(""" """, unsafe_allow_html=True) # Session State Initialization if 'drafts' not in st.session_state: st.session_state['drafts'] = {} if 'emails' not in st.session_state: st.session_state['emails'] = "" if 'client_names' not in st.session_state: st.session_state['client_names'] = "" if 'support_name' not in st.session_state: st.session_state['support_name'] = "" if 'refs' not in st.session_state: st.session_state['refs'] = "" if 'folder_path' not in st.session_state: st.session_state['folder_path'] = "" if 'doc_context' not in st.session_state: st.session_state['doc_context'] = "" if 'results' not in st.session_state: st.session_state['results'] = None if 'notion_client_name' not in st.session_state: st.session_state['notion_client_name'] = "" if 'case_comments' not in st.session_state: st.session_state['case_comments'] = "" # ========================================================================= # [보안 끝판왕] 최초 로그인 비번 세팅 & 관리자 실시간 동기화/삭제 통제 시스템 # ========================================================================= PERSISTENT_DATA_DIR = os.environ.get("ADVISERNOTE_DATA_DIR", "/data") USING_PERSISTENT_STORAGE = os.path.isdir(PERSISTENT_DATA_DIR) and os.access(PERSISTENT_DATA_DIR, os.W_OK) USER_DB_FILE = os.path.join(PERSISTENT_DATA_DIR, "user_passwords.db") if USING_PERSISTENT_STORAGE else "user_passwords.db" def init_user_db(seed_users): conn = sqlite3.connect(USER_DB_FILE) conn.execute(""" CREATE TABLE IF NOT EXISTS users ( email TEXT PRIMARY KEY, password TEXT NOT NULL DEFAULT '', active INTEGER NOT NULL DEFAULT 1 ) """) conn.execute(""" CREATE TABLE IF NOT EXISTS mailboxes ( email TEXT PRIMARY KEY, entry TEXT NOT NULL, active INTEGER NOT NULL DEFAULT 1 ) """) for email_addr, password in seed_users.items(): conn.execute( "INSERT OR IGNORE INTO users (email, password, active) VALUES (?, ?, 1)", (email_addr, password) ) for mailbox_entry in ALL_MAILBOXES: mailbox_email = parse_mailbox_email(mailbox_entry) if mailbox_email: conn.execute( "INSERT OR IGNORE INTO mailboxes (email, entry, active) VALUES (?, ?, 1)", (mailbox_email, mailbox_entry) ) conn.commit() conn.close() def parse_mailbox_email(mailbox_entry): credential_part = str(mailbox_entry or "").split("|", 1)[0].strip() return credential_part.split(":", 1)[0].strip().lower() def load_active_users(): conn = sqlite3.connect(USER_DB_FILE) rows = conn.execute("SELECT email, password FROM users WHERE active = 1 ORDER BY email").fetchall() conn.close() return {email_addr: password for email_addr, password in rows} def update_user_password(email_addr, new_password): conn = sqlite3.connect(USER_DB_FILE) conn.execute( "UPDATE users SET password = ?, active = 1 WHERE email = ?", (new_password, email_addr) ) conn.commit() updated = conn.total_changes > 0 conn.close() return updated def add_or_restore_user(email_addr, initial_password=""): email_addr = email_addr.strip().lower() conn = sqlite3.connect(USER_DB_FILE) existing = conn.execute("SELECT email FROM users WHERE email = ?", (email_addr,)).fetchone() if existing: conn.execute( "UPDATE users SET password = ?, active = 1 WHERE email = ?", (initial_password, email_addr) ) else: conn.execute( "INSERT INTO users (email, password, active) VALUES (?, ?, 1)", (email_addr, initial_password) ) conn.commit() updated = conn.total_changes > 0 conn.close() return updated def revoke_user_access(email_addr): conn = sqlite3.connect(USER_DB_FILE) conn.execute("UPDATE users SET active = 0 WHERE email = ?", (email_addr,)) conn.commit() updated = conn.total_changes > 0 conn.close() return updated def load_active_mailboxes(): conn = sqlite3.connect(USER_DB_FILE) rows = conn.execute("SELECT email, entry FROM mailboxes WHERE active = 1 ORDER BY email").fetchall() conn.close() return {email_addr: entry for email_addr, entry in rows} def add_or_restore_mailbox(mailbox_entry): mailbox_entry = mailbox_entry.strip() mailbox_email = parse_mailbox_email(mailbox_entry) if not mailbox_email: return False conn = sqlite3.connect(USER_DB_FILE) existing = conn.execute("SELECT email FROM mailboxes WHERE email = ?", (mailbox_email,)).fetchone() if existing: conn.execute( "UPDATE mailboxes SET entry = ?, active = 1 WHERE email = ?", (mailbox_entry, mailbox_email) ) else: conn.execute( "INSERT INTO mailboxes (email, entry, active) VALUES (?, ?, 1)", (mailbox_email, mailbox_entry) ) conn.commit() updated = conn.total_changes > 0 conn.close() return updated def revoke_mailbox_access(mailbox_email): conn = sqlite3.connect(USER_DB_FILE) conn.execute("UPDATE mailboxes SET active = 0 WHERE email = ?", (mailbox_email,)) conn.commit() updated = conn.total_changes > 0 conn.close() return updated def get_imap_host_candidates(primary_host, mailbox_email): primary_host = str(primary_host or "").strip() return [primary_host] if primary_host else [] init_user_db(allowed_users) def check_password(): """로그인 검증, 최초 접속자 비번 세팅, 관리자 전용 대시보드 스위칭 마스터 함수""" if "password_correct" not in st.session_state: st.session_state["password_correct"] = False if "is_admin_mode" not in st.session_state: st.session_state["is_admin_mode"] = False if st.session_state["password_correct"]: return True # 🎨 [디자인 혁신] 3열 분할 레이아웃을 사용해 로그인 창을 화면 정중앙에 예쁜 카드로 배치 _, login_col, _ = st.columns([1, 1.1, 1]) with login_col: st.write("") # 상단 여백 세팅 st.write("") with st.container(border=True): st.markdown("""
🔒

Flux Finance

Internal System Access Portal

""", unsafe_allow_html=True) allowed_users = load_active_users() input_email = st.text_input("📧 Login Email Address", key="login_email", placeholder="username@fluxfinance.co.nz").strip().lower() if input_email and (input_email != ADMIN_EMAIL and input_email not in allowed_users): st.error("❌ Access Denied: This account is unregistered or has been deactivated. Please contact the administrator.") return False # Case A: 최고 관리자(매니저님) 접속 시 (기능 100% 동일) if input_email == ADMIN_EMAIL: input_password = st.text_input("👑 Admin Password", type="password", key="login_password", placeholder="Enter admin password") if st.button("Sign In as Manager", use_container_width=True): if input_password == MASTER_PW: st.session_state["password_correct"] = True st.session_state["is_admin_mode"] = True st.rerun() else: st.error("❌ Incorrect Administrator password.") return False # Case B: 일반 직원 접속 시 (기능 100% 동일) if input_email in allowed_users: # 위에서 만든 딕셔너리 사용 current_db_pwd = allowed_users[input_email] # 상황 ①: 최초 접속이라 비번이 비어있을 때 ("") -> 초기 셋업 로직 완벽 보존 if current_db_pwd == "": st.warning("🆕 Welcome! Please set up your personalized password for this account.") new_pwd = st.text_input("🔑 Create New Password", type="password", key="setup_pwd") confirm_pwd = st.text_input("🔄 Confirm New Password", type="password", key="confirm_pwd") if st.button("Activate My Account", use_container_width=True): if new_pwd and new_pwd == confirm_pwd: if update_user_password(input_email, new_pwd): st.success("✅ Password configured successfully! Please sign in again with your new password.") st.rerun() else: st.error("⚠️ System Error: Failed to update your password. Please contact the administrator.") else: st.error("❌ Passwords do not match or fields are left blank.") return False # 상황 ②: 이미 비번 세팅이 끝나서 정상 로그인을 시도할 때 -> 로그인 로직 완벽 보존 else: input_password = st.text_input("🔑 Password", type="password", key="login_password", placeholder="Enter your system password") col_login, col_change = st.columns([1, 1]) with col_login: if st.button("Sign In", use_container_width=True): if input_password == current_db_pwd: st.session_state["password_correct"] = True st.session_state["is_admin_mode"] = False st.rerun() else: st.error("❌ Invalid password. Please try again.") # 상황 ③: 직원이 마이페이지에서 비밀번호 변경을 요청할 때 -> 토스트 팝업 로직 완벽 보존 with col_change: with st.popover("🔄 Change Password", use_container_width=True): st.write("Verify your current password to update your credentials.") verify_old = st.text_input("Current Password", type="password", key="v_old") update_new = st.text_input("New Password", type="password", key="u_new") if st.button("Update Password", use_container_width=True): if verify_old == current_db_pwd and update_new: if update_user_password(input_email, update_new): st.toast("✅ Password updated successfully!", icon="🚀") st.rerun() else: st.error("⚠️ System Error: Failed to update your password. Please contact the administrator.") else: st.error("❌ Current password verification failed or input is missing.") return False # ========================================================================= # [NOTION API CONNECTOR] 노션 페이지 본문 및 모든 실시간 댓글 병합 수집기 # ========================================================================= # ========================= # 1. 댓글 가져오기 # ========================= def fetch_notion_comments(page_id): """노션 특정 페이지의 댓글과 작성 날짜를 함께 추출하는 함수""" notion_token = NOTION_TOKEN headers = { "Authorization": f"Bearer {notion_token}", "Notion-Version": "2025-09-03" } try: url = f"https://api.notion.com/v1/comments?block_id={page_id}" response = requests.get(url, headers=headers) if response.status_code == 200: comments_data = response.json().get("results", []) comments_list = [] for c in comments_data: # 🟢 [수정] 작성 날짜(created_time)를 가져옵니다. created_at = c.get("created_time", "").split("T")[0] # YYYY-MM-DD 형식만 추출 c_text_list = c.get("rich_text", []) c_text = "".join( item.get("plain_text", item.get("text", {}).get("content", "")) for item in c_text_list if isinstance(item, dict) ).strip() if c_text: # 🟢 [수정] [날짜] 코멘트 내용 형식으로 조합 comments_list.append(f"💬 [{created_at}] {c_text}") return "\n".join(comments_list) if comments_list else "" except: pass return "" def notion_api_url(endpoint_path): return "https://api.notion.com/" + endpoint_path.lstrip("/") def extract_notion_plain_text(prop): if not isinstance(prop, dict): return "" prop_type = prop.get("type") if prop_type in ("title", "rich_text"): parts = prop.get(prop_type, []) return "".join(item.get("plain_text", "") for item in parts if isinstance(item, dict)).strip() if prop_type == "phone_number": return str(prop.get("phone_number") or "").strip() if prop_type == "email": return str(prop.get("email") or "").strip() if prop_type == "select" and prop.get("select"): return str(prop["select"].get("name", "")).strip() if prop_type == "multi_select": return ", ".join(item.get("name", "") for item in prop.get("multi_select", []) if item.get("name")).strip() if prop_type == "people": return ", ".join(item.get("name", "") for item in prop.get("people", []) if item.get("name")).strip() if prop_type == "number": value = prop.get("number") if value is None: return "" if isinstance(value, float) and value.is_integer(): return str(int(value)).strip() return str(value).strip() return "" def normalize_phone_text(value): text = "" if value is None else str(value).strip() if text.endswith(".0") and text[:-2].isdigit(): text = text[:-2] return text def is_phone_like_field(field_name): normalized = str(field_name or "").strip().lower() return any(key in normalized for key in ["phone", "mobile", "telephone", "contact number", "cell", "전화", "연락", "휴대"]) def make_excel_safe_phone(value): text = normalize_phone_text(value) if not text: return "" escaped = text.replace('"', '""') return f'="{escaped}"' def find_notion_property_value(props, preferred_keys, fallback_types=None): normalized_lookup = {str(key).strip().lower(): value for key, value in props.items()} for key in preferred_keys: key_l = key.lower() for prop_name, prop in normalized_lookup.items(): if key_l in prop_name: value = extract_notion_plain_text(prop) if value: return value if fallback_types: for prop in props.values(): if isinstance(prop, dict) and prop.get("type") in fallback_types: value = extract_notion_plain_text(prop) if value: return value return "" def fetch_client_contacts_from_notion(search_text=""): notion_token = NOTION_TOKEN target_data_source_id = NOTION_DB_ID if not notion_token: return [], "Missing NOTION_API_TOKEN in secrets.toml" if not target_data_source_id: return [], "Missing NOTION_DATABASE_ID in secrets.toml" headers = { "Authorization": f"Bearer {notion_token.strip()}", "Content-Type": "application/json", "Notion-Version": "2025-09-03" } name_keys = ["case name", "client name", "applicant", "customer", "borrower", "name", "고객", "이름"] nickname_keys = ["nickname", "nick name", "preferred name", "alias", "short name", "닉네임", "별명"] phone_keys = ["phone", "mobile", "telephone", "contact number", "cell", "전화", "연락", "휴대"] query_url = notion_api_url(f"v1/data_sources/{target_data_source_id.strip()}/query/") contacts = [] start_cursor = None try: while True: payload = {"page_size": 100} if start_cursor: payload["start_cursor"] = start_cursor response = requests.post(query_url, json=payload, headers=headers, proxies={"http": None, "https": None}, timeout=20) if response.status_code != 200: return [], f"Notion API Failed: {response.status_code} - {response.text[:200]}" data = response.json() for item in data.get("results", []): props = item.get("properties", {}) client_name = find_notion_property_value(props, name_keys, fallback_types={"title"}) nickname = find_notion_property_value(props, nickname_keys) phone_number = find_notion_property_value(props, phone_keys, fallback_types={"phone_number"}) if not client_name and not nickname and not phone_number: continue searchable = " ".join([client_name, nickname, phone_number]).lower() if search_text and search_text.strip().lower() not in searchable: continue contacts.append({ "Client Name": client_name, "Nickname": nickname, "Phone": phone_number }) if not data.get("has_more"): break start_cursor = data.get("next_cursor") except Exception as exc: return [], f"Notion contact pull failed: {exc}" contacts = sorted(contacts, key=lambda row: row.get("Client Name", "").lower()) return contacts, "Success" NOTION_CONTACT_FIELD_CONFIG = { "Client Name": { "keys": ["case name", "client name", "applicant", "customer", "borrower", "name", "고객", "이름"], "fallback_types": {"title"}, }, "Nickname": { "keys": ["nickname", "nick name", "preferred name", "alias", "short name", "닉네임", "별명"], "fallback_types": set(), }, "Phone": { "keys": ["phone", "mobile", "telephone", "contact number", "cell", "전화", "연락", "휴대"], "fallback_types": {"phone_number"}, }, "Email": { "keys": ["email", "e-mail", "mail", "이메일"], "fallback_types": {"email"}, }, "Status": { "keys": ["status", "stage", "progress", "상태", "진행"], "fallback_types": set(), }, "Adviser": { "keys": ["adviser", "advisor", "broker", "owner", "담당", "어드바이저"], "fallback_types": set(), }, "Case Number": { "keys": ["case number", "case no", "ref", "reference", "application number", "번호"], "fallback_types": set(), }, } def find_notion_property_by_exact_or_contains(props, property_name): target = str(property_name or "").strip().lower() if not target: return "" for key, prop in props.items(): if str(key).strip().lower() == target: return extract_notion_plain_text(prop) for key, prop in props.items(): if target in str(key).strip().lower(): return extract_notion_plain_text(prop) return "" def fetch_client_contacts_from_notion(search_text="", selected_fields=None, custom_fields=None): notion_token = NOTION_TOKEN target_data_source_id = NOTION_DB_ID if not notion_token: return [], "Missing NOTION_API_TOKEN in secrets.toml" if not target_data_source_id: return [], "Missing NOTION_DATABASE_ID in secrets.toml" headers = { "Authorization": f"Bearer {notion_token.strip()}", "Content-Type": "application/json", "Notion-Version": "2025-09-03" } selected_fields = selected_fields or ["Client Name", "Nickname", "Phone"] custom_fields = [field.strip() for field in (custom_fields or []) if field.strip()] output_fields = list(dict.fromkeys(selected_fields + custom_fields)) query_url = notion_api_url(f"v1/data_sources/{target_data_source_id.strip()}/query/") contacts = [] start_cursor = None try: while True: payload = {"page_size": 100} if start_cursor: payload["start_cursor"] = start_cursor response = requests.post(query_url, json=payload, headers=headers, proxies={"http": None, "https": None}, timeout=20) if response.status_code != 200: return [], f"Notion API Failed: {response.status_code} - {response.text[:200]}" data = response.json() for item in data.get("results", []): props = item.get("properties", {}) row = {} for field_name in output_fields: config = NOTION_CONTACT_FIELD_CONFIG.get(field_name) if config: row[field_name] = find_notion_property_value( props, config["keys"], fallback_types=config["fallback_types"], ) else: row[field_name] = find_notion_property_by_exact_or_contains(props, field_name) if is_phone_like_field(field_name): row[field_name] = normalize_phone_text(row[field_name]) if not any(str(value).strip() for value in row.values()): continue searchable = " ".join(str(value) for value in row.values()).lower() if search_text and search_text.strip().lower() not in searchable: continue contacts.append(row) if not data.get("has_more"): break start_cursor = data.get("next_cursor") except Exception as exc: return [], f"Notion contact pull failed: {exc}" sort_field = "Client Name" if "Client Name" in output_fields else output_fields[0] contacts = sorted(contacts, key=lambda row: str(row.get(sort_field, "")).lower()) return contacts, "Success" def fetch_client_tasks_from_notion(client_name): # .streamlit/secrets.toml 파일에서 보안 정보를 안전하게 로드합니다. notion_token = NOTION_TOKEN target_data_source_id = NOTION_DB_ID if not notion_token: return [], "Missing NOTION_API_TOKEN in secrets.toml" if not target_data_source_id: return [], "Missing NOTION_DATABASE_ID in secrets.toml" headers = { "Authorization": f"Bearer {notion_token.strip()}", "Content-Type": "application/json", "Notion-Version": "2025-09-03" } # 윈도우 시스템 주소 오타 변조 및 가로채기 차단용 우회 함수 def get_clean_url(endpoint_path): p1 = "ht" + "tps://" p2 = "ap" + "i.no" + "tion.c" + "om/" return p1 + p2 + endpoint_path # secrets.toml에서 가져온 진짜 데이터 소스 ID 기반의 쿼리 주소 생성 query_url = get_clean_url(f"v1/data_sources/{target_data_source_id.strip()}/query/") # 입력한 고객 이름과 정확히 일치하는 행만 타겟팅하는 필터 딜리버리 payload = { "filter": { "property": "Case Name", "rich_text": { "equals": client_name.strip() } } } clean_proxies = {"http": None, "https": None} # 1. 원본 데이터 소스 조회 response = requests.post(query_url, json=payload, headers=headers, proxies=clean_proxies) if response.status_code != 200: return [], f"Notion API Failed: {response.status_code}" results = response.json().get("results", []) if not results: return [], f"No case found named '{client_name}'." # 2. 검색 정합성이 검증된 실시간 데이터 행에서 하위 Task ID 추출 target_case = results[0] case_id = target_case.get("id") # 🟢 [수정 1] 부모 케이스(Case) 페이지에 있는 전체 코멘트 추출 case_comments = fetch_notion_comments(case_id) task_ids = [] for prop in target_case.get("properties", {}).values(): if prop.get("type") == "relation": for r in prop.get("relation", []): task_ids.append(r.get("id")) tasks_list = [] # 3. 각 하위 개별 Task 상세 필드 파싱 for tid in task_ids: page_url = get_clean_url(f"v1/pages/{tid}") page_res = requests.get(page_url, headers=headers, proxies=clean_proxies) if page_res.status_code == 200: page_data = page_res.json() props = page_data.get("properties", {}) title = "제목 없음" description = "내용 없음" date_val = datetime.now().strftime("%Y-%m-%d") channel_val = "-" for key, v in props.items(): normalized_key = key.strip().lower() p_type = v.get("type") if p_type == "title": t_list = v.get("title", []) if isinstance(t_list, list) and len(t_list) > 0: parts = [item.get("plain_text", "") for item in t_list if isinstance(item, dict)] if parts: title = "".join(parts) elif "completed" in normalized_key: if p_type == "date" and v.get("date"): date_val = v["date"].get("start", date_val) elif p_type == "rich_text": r_list = v.get("rich_text", []) if isinstance(r_list, list) and len(r_list) > 0: parts = [item.get("plain_text", "") for item in r_list if isinstance(item, dict)] if parts: date_val = "".join(parts) elif "description" in normalized_key: if p_type == "rich_text": r_list = v.get("rich_text", []) if isinstance(r_list, list) and len(r_list) > 0: parts = [item.get("plain_text", "") for item in r_list if isinstance(item, dict)] if parts: description = "".join(parts) elif "channel" in normalized_key: if p_type == "select" and v.get("select"): channel_val = v["select"].get("name", "-") elif p_type == "multi_select" and v.get("multi_select"): channel_val = ", ".join([m.get("name", "") for m in v["multi_select"]]) elif p_type == "rich_text": r_list = v.get("rich_text", []) if isinstance(r_list, list) and len(r_list) > 0: parts = [item.get("plain_text", "") for item in r_list if isinstance(item, dict)] if parts: channel_val = "".join(parts) comments_text = fetch_notion_comments(tid) tasks_list.append({ "Date": date_val, "Body": description, "Sec": title, "RawTitle": title, "Type": "NOTION", "Channel": channel_val, "Comments": comments_text }) # 4. 태스크 번호 오름차순 정렬 if tasks_list: tasks_list = sorted(tasks_list, key=lambda x: x["Sec"]) # 🟢 튜플 리턴값 3개로 확장 (case_comments 포함) return tasks_list, "Success", case_comments if check_password(): # Render HR & Security Control Panel if Manager logs in RULES_FILE = "compliance_rules.json" if 'rules_df' not in st.session_state: if os.path.exists(RULES_FILE): try: st.session_state['rules_df'] = pd.read_json(RULES_FILE, encoding='utf-8') except: pass # 파일이 없으면 오리지널 10대 기본 구조 생성 if 'rules_df' not in st.session_state: data = [ {"No": "1", "Stage Title": "INITIAL MEETING", "Sub 1": "Initial Meeting", "Sub 2": "", "Keywords": "", "Action": "Conducted a face-to-face discovery meeting to verify original ID and discuss client objectives.", "Purpose": "To satisfy AML/CFT requirements via face-to-face ID verification and gather financial data for lending strategy."}, {"No": "2", "Stage Title": "DISCLOSURE DELIVERY", "Sub 1": "Disclosure Delivery", "Sub 2": "", "Keywords": "Disclosure, D1, D2", "Action": "Emailed full FAP Disclosure or provided public info.", "Purpose": "To ensure clients review scope of service and licensing."}, {"No": "3", "Stage Title": "DOCUMENT COLLECTION", "Sub 1": "Required documents requested", "Sub 2": "Required documents received", "Keywords": "Attachment: Payslip, Statement, ID", "Action": "Verified core docs or requested further clarification.", "Purpose": "To assess borrowing capacity and meet lender standards."}, {"No": "4", "Stage Title": "STRATEGY & ADVICE", "Sub 1": "Strategy & Advice", "Sub 2": "", "Keywords": "Fact Find, Scope of Work, SOW", "Action": "Discussed bank-specific policies and compared products.", "Purpose": "To align the application with the best possible lender/rate."}, {"No": "5", "Stage Title": "APPLICATION SUBMISSION", "Sub 1": "Application Submission", "Sub 2": "", "Keywords": "", "Action": "Reviewed SOW/Fact-Find and formally lodged application.", "Purpose": "To ensure data accuracy before formal credit assessment."}, {"No": "6", "Stage Title": "LENDER QUERY", "Sub 1": "Bank Query Received", "Sub 2": "Query Response Sent", "Keywords": "", "Action": "Received and addressed technical queries from the lender.", "Purpose": "To satisfy credit requirements and progress the file."}, {"No": "7", "Stage Title": "CONDITIONAL APPROVAL", "Sub 1": "Conditional approval received", "Sub 2": "Conditional approval provided to client", "Keywords": "Letter of Offer, LOO", "Action": "Reviewed conditional approval and updated the clients.", "Purpose": "To identify and guide clients through remaining conditions."}, {"No": "8", "Stage Title": "LOAN STRUCTURE REVIEW", "Sub 1": "Proposed Structure Discussed", "Sub 2": "Final Instruction to Lender", "Keywords": "Statement of Advice, SOA", "Action": "Provided SOA and confirmed final interest rate lock-in.", "Purpose": "To ensure the loan structure meets the client's risk profile."}, {"No": "9", "Stage Title": "FINAL APPROVAL", "Sub 1": "Final Approval", "Sub 2": "", "Keywords": "Final, Unconditional", "Action": "Received final approval and instructed the solicitor.", "Purpose": "To confirm all conditions met and initiate legal process."}, {"No": "10", "Stage Title": "SETTLEMENT COMPLETION", "Sub 1": "Settlement Completion", "Sub 2": "", "Keywords": "Settlement", "Action": "Coordinated funds and confirmed successful completion.", "Purpose": "To finalize the advice process and transition to service."} ] st.session_state['rules_df'] = pd.DataFrame(data) # 1️⃣ [최상단] Admin 전용 통제 구역 (Manager 로그인 시에만 노출) if st.session_state.get("is_admin_mode", False): with st.expander("👑 Admin Control", expanded=False): st.markdown("### 👑 Admin Control") st.caption("Monitor real-time password setups for active team members and revoke access for offboarded employees.") current_users = load_active_users() admin_data = [] for u_email, u_pwd in current_users.items(): status_tag = "🔴 Pending Setup" if u_pwd == "" else "🟢 Active (Password Set)" # [🔒 SECURITY MASKING] Convert plain text passwords (e.g., "dong9ri1" -> "d*******") if u_pwd and len(u_pwd) > 1: masked_pwd = u_pwd[0] + "*" * (len(u_pwd) - 1) elif u_pwd: masked_pwd = "*" else: masked_pwd = "(None)" admin_data.append({ "Employee Email": u_email, "Recorded Password": masked_pwd, # Displays secure masked version on screen "Status": status_tag }) st.table(pd.DataFrame(admin_data)) st.markdown("**Add or Restore User Access**") add_col_email, add_col_password = st.columns([2, 1]) new_user_email = add_col_email.text_input( "Employee email address", key="admin_add_user_email", placeholder="new.user@fluxfinance.co.nz" ).strip().lower() new_user_password = add_col_password.text_input( "Initial password", type="password", key="admin_add_user_password", help="Leave blank to let the user set their password on first login." ) if st.button("➕ Add / Restore User", use_container_width=True): if not new_user_email: st.error("Please enter an email address.") elif not re.match(r"^[^@\s]+@[^@\s]+\.[^@\s]+$", new_user_email): st.error("Please enter a valid email address.") elif add_or_restore_user(new_user_email, new_user_password): st.success(f"✅ Access granted for {new_user_email}.") st.rerun() else: st.error("⚠️ System Error: Failed to update user access.") st.markdown("**Deactivate & Revoke User Permissions**") target_del = st.selectbox("Select email address to revoke access", ["-"] + list(current_users.keys())) if target_del != "-" and st.button("🚨 Revoke Access"): if target_del in current_users and revoke_user_access(target_del): st.success(f"⚠️ Access permanently revoked for {target_del}. This user can no longer sign in.") st.rerun() st.markdown("**Mailbox Scan List**") current_mailboxes = load_active_mailboxes() st.markdown("**Update Mailbox Password**") password_mailbox = st.selectbox( "Mailbox", list(current_mailboxes.keys()), key="admin_password_mailbox", ) new_mailbox_password = st.text_input( "New individual password", type="password", key="admin_new_mailbox_password", placeholder="Enter the new mailbox password", ) password_col, common_col = st.columns(2) if password_col.button("Update Password", width="stretch"): if not new_mailbox_password: st.error("Please enter the new mailbox password.") else: current_entry = current_mailboxes[password_mailbox] _, separator, imap_host = current_entry.partition("||") if not separator or not imap_host.strip(): st.error("The mailbox does not have a valid IMAP host.") else: updated_entry = f"{password_mailbox}:{new_mailbox_password}||{imap_host.strip()}" if add_or_restore_mailbox(updated_entry): st.session_state.pop("admin_new_mailbox_password", None) st.success(f"Mailbox password updated: {password_mailbox}") st.rerun() else: st.error("Failed to update the mailbox password.") if common_col.button("Use Common Secret", width="stretch"): current_entry = current_mailboxes[password_mailbox] _, separator, imap_host = current_entry.partition("||") if not separator or not imap_host.strip(): st.error("The mailbox does not have a valid IMAP host.") elif add_or_restore_mailbox(f"{password_mailbox}||{imap_host.strip()}"): st.session_state.pop("admin_new_mailbox_password", None) st.success(f"Common secret enabled: {password_mailbox}") st.rerun() mailbox_rows = [] for email_addr, entry in current_mailboxes.items(): credential_part, separator, imap_host = entry.partition("||") _, password_separator, individual_password = credential_part.partition(":") mailbox_rows.append({ "Mailbox": email_addr, "Password": individual_password if password_separator else "Uses EMAIL_PASSWORD", "IMAP Host": imap_host.strip() if separator else "arrow.mxrouting.net", }) st.dataframe( pd.DataFrame(mailbox_rows), hide_index=True, width="stretch", column_config={ "Mailbox": st.column_config.TextColumn(width="medium"), "Password": st.column_config.TextColumn(width="medium"), "IMAP Host": st.column_config.TextColumn(width="large"), }, ) mailbox_entry = st.text_input( "Mailbox entry", key="admin_add_mailbox_entry", placeholder="email@domain.com:optional-password||imap.server.com" ).strip() if st.button("➕ Add / Restore Mailbox", use_container_width=True): mailbox_email = parse_mailbox_email(mailbox_entry) if not mailbox_entry: st.error("Please enter a mailbox entry.") elif not re.match(r"^[^@\s]+@[^@\s]+\.[^@\s]+$", mailbox_email): st.error("Please enter a valid mailbox email address.") elif "||" not in mailbox_entry: st.error("Please include the IMAP host using the format email@domain.com||imap.server.com.") elif add_or_restore_mailbox(mailbox_entry): st.success(f"✅ Mailbox added for scanning: {mailbox_email}.") st.rerun() else: st.error("⚠️ System Error: Failed to update mailbox list.") target_mailbox_del = st.selectbox( "Select mailbox to remove from scan list", ["-"] + list(current_mailboxes.keys()), key="admin_revoke_mailbox_select" ) if target_mailbox_del != "-" and st.button("🚫 Remove Mailbox From Scan List"): if target_mailbox_del in current_mailboxes and revoke_mailbox_access(target_mailbox_del): st.success(f"⚠️ Mailbox removed from scan list: {target_mailbox_del}.") st.rerun() # (2) 규칙 편집 관리 (Rules Configuration - Admin Control 바로 밑으로 배치) with st.expander("⚙️ Rules Configuration", expanded=False): st.caption("📝 웹 화면에서 수정·추가하면 시스템 파일에 영구 저장되어 연동됩니다.") edited_rules = st.data_editor(st.session_state['rules_df'], num_rows="dynamic", use_container_width=True, hide_index=True, key="main_rules_editor") if not edited_rules.equals(st.session_state['rules_df']): st.session_state['rules_df'] = edited_rules # 하드디스크 JSON 파일로 덮어쓰기 박제 (영구 저장의 핵심) edited_rules.to_json(RULES_FILE, orient='records', force_ascii=False, indent=4) st.toast("대출 매칭 규칙이 영구 저장되었습니다! 💾", icon="✅") st.divider() # Compliance Rules Data if 'rules_df' not in st.session_state: try: st.session_state['rules_df'] = pd.read_excel('rules.xlsx') except: st.session_state['rules_df'] = pd.DataFrame(columns=['Stage Title', 'Sub 1', 'Sub 2', 'Keyword 1', 'Keyword 2']) if 'status' not in st.session_state: st.session_state['status'] = ["⚪"] * len(st.session_state['rules_df']) # --- Helper Functions --- def add_line_to_docx(paragraph): p_ptr = paragraph._p.get_or_add_pPr() p_bdr = OxmlElement('w:pBdr') bottom = OxmlElement('w:bottom') bottom.set(qn('w:val'), 'single'); bottom.set(qn('w:sz'), '4') bottom.set(qn('w:space'), '1'); bottom.set(qn('w:color'), 'E2E8F0') p_bdr.append(bottom); p_ptr.append(p_bdr) def load_docs_from_path(path): combined_text = "" path = path.strip().replace('"', '') if not os.path.exists(path): return None, f"Path not found: {path}" files = [f for f in os.listdir(path) if f.lower().endswith(('.pdf', '.docx'))] if not files: return None, "No reference docs found in the folder." for f_name in files: f_path = os.path.join(path, f_name) try: if f_name.lower().endswith('.pdf'): with pdfplumber.open(f_path) as pdf: for page in pdf.pages: combined_text += (page.extract_text() or "") + "\n" elif f_name.lower().endswith('.docx'): doc = Document(f_path) for para in doc.paragraphs: combined_text += para.text + "\n" except: continue return combined_text, f"✅ Analyzed {len(files)} reference document(s)." # 🟢 [콜백 삭제] 악명 높은 에러를 일으키는 assign_callback은 아예 삭제합니다! # 🟢 메인 연산 함수: 상태 리스트를 '새 리스트 통째로 덮어쓰기'로 깔끔하게 유지 def update_status(): if st.session_state.get('results') is None: return df = st.session_state['results'] rules = st.session_state['rules_df'] rules_list = rules.to_dict('records') new_status = ["⚪"] * len(rules_list) if st.session_state.get('active_mode') == "NOTION": assigned = [re.sub(r'^\d+\.\s*', '', str(x)).lower().strip() for x in df[df['Type'] == "NOTION"]['Sec'].tolist()] for idx, r in enumerate(rules_list): stage_val = str(r.get('Stage Title', '')).strip().lower() s1 = str(r.get('Sub 1', '')).strip().lower() s2 = str(r.get('Sub 2', '')).strip().lower() has_stage = stage_val not in ["", "nan", "x", "none", "-"] has_s1 = s1 not in ["", "nan", "x", "none", "-"] has_s2 = s2 not in ["", "nan", "x", "none", "-"] is_stage_mapped = (stage_val in assigned) if has_stage else False is_s1_mapped = (s1 in assigned) if has_s1 else False is_s2_mapped = (s2 in assigned) if has_s2 else False if is_stage_mapped: new_status[idx] = "🟢" else: if has_s1 and not has_s2: new_status[idx] = "🟢" if is_s1_mapped else "⚪" elif has_s1 and has_s2: if is_s1_mapped and is_s2_mapped: new_status[idx] = "🟢" elif is_s1_mapped or is_s2_mapped: new_status[idx] = "🟠" else: new_status[idx] = "⚪" else: new_status[idx] = "⚪" else: assigned = [str(x).lower().strip() for x in df[df['Type'] != "NOTION"]['Sec'].tolist()] for idx, r in enumerate(rules_list): s1 = str(r.get('Sub 1', '')).strip().lower() s2 = str(r.get('Sub 2', '')).strip().lower() has_s1 = s1 not in ["", "nan", "x", "None", "-"] has_s2 = s2 not in ["", "nan", "x", "None", "-"] is_s1_mapped = (s1 in assigned) if has_s1 else False is_s2_mapped = (s2 in assigned) if has_s2 else False if has_s1 and not has_s2: new_status[idx] = "🟢" if is_s1_mapped else "⚪" elif has_s1 and has_s2: if is_s1_mapped and is_s2_mapped: new_status[idx] = "🟢" elif is_s1_mapped or is_s2_mapped: new_status[idx] = "🟠" else: new_status[idx] = "⚪" else: new_status[idx] = "⚪" st.session_state['status'] = new_status # --- Helper Functions --- # (기존 add_line_to_docx, load_docs_from_path, update_status 등은 그대로 유지) def ask_ai_note(stage_key, items_data, row_info, manual_note=""): # 🚨 [파이썬 강력 통제 1] 매핑된 이메일 데이터가 단 한 개도 없고, 추가 수기 메모도 완전히 비어있다면 # AI 호출 자체를 생략하고 즉시 완벽한 3줄 공란을 리턴하여 비용과 오류를 원천 차단합니다. if not items_data and not manual_note.strip(): return "Date: \nAction: \nPurpose: \n" try: #vertexai.init(project=st.secrets["PROJECT_ID"], location="us-west1") from vertexai.generative_models import GenerativeModel model = GenerativeModel('gemini-2.5-flash') # 디지털 이메일/노션 소스 로그 포맷팅 context_str = "" for i, m in enumerate(items_data): record_type = m.get('Type', 'EMAIL') d = m.get('Date', 'N/A') b = m.get('Body', '')[:1500] if record_type == "NOTION": ch = m.get('Channel', '-') cm = m.get('Comments', '') context_str += f"Record {i+1} [Type: NOTION Task]\nDate: {d}\nChannel: {ch}\nDescription: {b}\nComments: {cm}\n\n" else: context_str += f"Record {i+1} [Type: EMAIL]\nDate: {d}\nBody: {b}\n\n" # 규칙 데이터 안전 변환 rule_dict = dict(row_info) if row_info is not None else {} stage_title = rule_dict.get('Stage Title', stage_key) rule_action = rule_dict.get('Action', "").strip() rule_purpose = rule_dict.get('Purpose', "").strip() email_count = len(items_data) # 🚨 [파이썬 강력 통제 2] 만약 디지털 매핑 데이터는 아예 없는데 '수기 메모'만 존재하는 경우, # 이 수기 메모가 현재 순회 중인 단계와 연관이 있는 단어(예: 세미나, 첫 미팅 등)를 포함하는지 하드코딩으로 1차 검증합니다. # 1번 INITIAL MEETING이 아닌 다른 단계에 미팅 정보가 침범하는 것을 완벽하게 막아냅니다. if email_count == 0 and manual_note.strip(): lower_note = manual_note.lower() # 1번 초기 미팅과 관련된 핵심 단어들 리스트 initial_keywords = ["initial", "meeting", "seminar", "remax", "처음", "만났", "미팅", "신분증", "얼굴"] # 현재 단계가 1번 초기 미팅이 '아닌데', 수기 메모 내용이 초기 미팅 관련 내용뿐이라면 즉시 공란 리턴하고 차단 if "1." not in stage_key and any(kw in lower_note for kw in initial_keywords): return "Date: \nAction: \nPurpose: \n" # 🚨 [프롬프트 가드레일 개조] prompt = f""" You are a strict compliance auditor representing the New Zealand Financial Markets Authority (FMA). Your job is to draft a professional File Note ONLY for the specific mortgage process stage: [{stage_title}] [STRICT COMPLIANCE FILTERING MATRIX] - Current Target Stage: {stage_title} - Authorized Compliance Actions: {rule_action} - Strategic Compliance Objective: {rule_purpose} [REAL CLIENT DATA SETS (GROUND TRUTH)] - Number of Digital Logs Matched: {email_count} - Digital Logs (Matched Emails/Notion): {context_str if items_data else "NO DIGITAL LOGS FOR THIS STAGE"} - User's Additional Memo: "{manual_note if manual_note.strip() else "NONE PROVIDED"}" [⚠️ CRITICAL GENERATION RULES - DO NOT VIOLATE] 1. MANDATORY OVERRIDE FOR MATCHED LOGS: If 'Number of Digital Logs Matched' is 1 or more, it means the user has manually verified and mapped these emails to [{stage_title}]. You MUST trust this mapping and generate a summary based on these logs. 2. NO DATA = COMPLETE BLANK: If the provided digital logs are empty, and the User's Additional Memo does not explicitly describe a unique event for [{stage_title}], you are strictly forbidden from writing ANYTHING. You MUST output exactly this 3-line block and STOP. Do NOT copy the 'Strategic Compliance Objective' text into Purpose if there is no matching fact: Date: Action: Purpose: 3. NEVER COPY TEMPLATE TEXT: Do not copy-paste or mimic the phrase "{rule_action}" or "{rule_purpose}" into the note. Always write unique, professional descriptions based on the client logs. [🚨 CRITICAL FORMAT RULE: NUMBER OF ITEMS MATCHING] - CASE A (Single Source): If there is only 1 email/record matched, OR only the User's Memo is valid for this stage, output EXACTLY ONE pair of "Date:" and "Action:". (Do NOT use numbers like Date 1 or Action 1). - CASE B (Multiple Sources): If there are 2 or more distinct emails/records matched, group them chronologically and use "Date 1:", "Action 1:", "Date 2:", "Action 2:" etc. [PROPER FORMAT EXAMPLES] Example for CASE A: Date: 15 May 2026, 10:32 AM Action: Sent an email to the client confirming the next steps of disclosure and requested verification docs. Purpose: Written custom compliant rationale based on the real event. """ response = model.generate_content(prompt) res_text = response.text.strip() if "Date: \nAction: \nPurpose:" in res_text or not res_text: return "Date: \nAction: \nPurpose: \n" return res_text except Exception as e: return f"⚠️ Vertex AI Error: {str(e)}" def estimate_text_tokens(text): return max(1, int(len(str(text or "")) / 4) + 1) def estimate_ai_summary_cost(full_df, rules_df, manual_note=""): if full_df is None or full_df.empty: return {"calls": 0, "input_tokens": 0, "output_tokens": 0, "usd": 0.0} current_mode = st.session_state.get('active_mode', 'NOTION') if current_mode == "NOTION" and 'Type' in full_df.columns: df = full_df[full_df['Type'] == "NOTION"].copy() elif 'Type' in full_df.columns: df = full_df[full_df['Type'] != "NOTION"].copy() else: df = full_df.copy() calls = 0 input_tokens = 0 output_tokens = 0 manual_has_text = bool(str(manual_note or "").strip()) for _, row in rules_df.iterrows(): s1 = str(row.get('Sub 1', '')).strip() s2 = str(row.get('Sub 2', '')).strip() targets = pd.DataFrame() if not df.empty and 'Sec' in df.columns: stage_values = [s1] if s2 and s2 not in ["nan", "x", "None", "-"]: stage_values.append(s2) targets = df[df['Sec'].isin(stage_values)] if targets.empty and not manual_has_text: continue calls += 1 row_text = " ".join([str(row.get(k, "")) for k in ["Stage Title", "Sub 1", "Sub 2", "Action", "Purpose"]]) source_text = "" for item in targets.to_dict('records') if not targets.empty else []: source_text += " ".join([ str(item.get("Date", "")), str(item.get("Subject", "")), str(item.get("From", "")), str(item.get("To", "")), str(item.get("Cc", "")), str(item.get("Channel", "")), str(item.get("Body", ""))[:1500], str(item.get("Comments", "")), ]) + "\n" prompt_overhead = 1200 input_tokens += prompt_overhead + estimate_text_tokens(row_text) + estimate_text_tokens(source_text) + estimate_text_tokens(manual_note) output_tokens += ESTIMATED_OUTPUT_TOKENS_PER_STAGE usd = ( input_tokens / 1_000_000 * GEMINI_25_FLASH_INPUT_USD_PER_1M + output_tokens / 1_000_000 * GEMINI_25_FLASH_OUTPUT_USD_PER_1M ) return {"calls": calls, "input_tokens": input_tokens, "output_tokens": output_tokens, "usd": usd} # ───────────────────────────────────────────────────────────────── # 🟢 [실시간 동기화 마법의 엔진] 사이드바를 그리기 전에 드롭다운 값을 먼저 가로채서 계산! # ───────────────────────────────────────────────────────────────── def sync_and_update_status(): if st.session_state.get('results') is None: return # 1. 화면 맨 밑의 드롭다운(selectbox) 최신 값을 미리 가로채서 데이터프레임에 덮어씀 df = st.session_state['results'] for row_idx in df.index: sel_key = f"sel_{row_idx}" if sel_key in st.session_state: df.at[row_idx, 'Sec'] = st.session_state[sel_key] # 2. 전구(status) 즉시 계산 rules = st.session_state['rules_df'] rules_list = rules.to_dict('records') new_status = ["⚪"] * len(rules_list) if st.session_state.get('active_mode') == "NOTION": assigned = [re.sub(r'^\d+\.\s*', '', str(x)).lower().strip() for x in df[df['Type'] == "NOTION"]['Sec'].tolist()] for idx, r in enumerate(rules_list): stage_val = str(r.get('Stage Title', '')).strip().lower() s1 = str(r.get('Sub 1', '')).strip().lower() s2 = str(r.get('Sub 2', '')).strip().lower() has_stage = stage_val not in ["", "nan", "x", "none", "-"] has_s1 = s1 not in ["", "nan", "x", "none", "-"] has_s2 = s2 not in ["", "nan", "x", "none", "-"] is_stage_mapped = (stage_val in assigned) if has_stage else False is_s1_mapped = (s1 in assigned) if has_s1 else False is_s2_mapped = (s2 in assigned) if has_s2 else False if is_stage_mapped: new_status[idx] = "🟢" else: if has_s1 and not has_s2: new_status[idx] = "🟢" if is_s1_mapped else "⚪" elif has_s1 and has_s2: if is_s1_mapped and is_s2_mapped: new_status[idx] = "🟢" elif is_s1_mapped or is_s2_mapped: new_status[idx] = "🟠" else: new_status[idx] = "⚪" else: new_status[idx] = "⚪" else: assigned = [str(x).lower().strip() for x in df[df['Type'] != "NOTION"]['Sec'].tolist()] if 'Type' in df.columns else [str(x).lower().strip() for x in df['Sec'].tolist()] for idx, r in enumerate(rules_list): s1 = str(r.get('Sub 1', '')).strip().lower() s2 = str(r.get('Sub 2', '')).strip().lower() has_s1 = s1 not in ["", "nan", "x", "None", "-"] has_s2 = s2 not in ["", "nan", "x", "None", "-"] is_s1_mapped = (s1 in assigned) if has_s1 else False is_s2_mapped = (s2 in assigned) if has_s2 else False if has_s1 and not has_s2: new_status[idx] = "🟢" if is_s1_mapped else "⚪" elif has_s1 and has_s2: if is_s1_mapped and is_s2_mapped: new_status[idx] = "🟢" elif is_s1_mapped or is_s2_mapped: new_status[idx] = "🟠" else: new_status[idx] = "⚪" else: new_status[idx] = "⚪" st.session_state['status'] = new_status # 🚀 사이드바를 그리기 직전에 이 엔진을 무조건 가동시킵니다! sync_and_update_status() # --- [Sidebar Area: 통합 워드 스타일 에디터] --- with st.sidebar: # [1] Progress Status (상단 고정) st.markdown("### 📊 Progress Status") for row_idx in range(0, 10, 5): cols = st.columns(5) for i in range(5): idx = row_idx + i if idx < len(st.session_state['status']): with cols[i]: icon = st.session_state['status'][idx] dt = "🟢" if icon == "🟢" else ("🟠" if icon == "🟠" else "⚪") full_title = st.session_state['rules_df'].iloc[idx]['Stage Title'] st.markdown( f"""
{dt}
{idx+1}
{full_title}
""", unsafe_allow_html=True) # ✨ [가이드 안내 문구 추가] 전구 상태 설명 및 AI 실행 안내 텍스트 이식 st.markdown("""
⚪ Unmapped  |  🟠 Partially Mapped (1/2)  |  🟢 Fully Mapped
""", unsafe_allow_html=True) # [수정 핵심] 변수 정의를 if문 밖으로 꺼내어 에러 방지 display_client = st.session_state.get('client_names', '').replace(';', ' & ').strip() support_person = st.session_state.get('support_name', 'Edward Lee (BDM)') # [2] 작업 버튼 영역 (Sync 완료 시에만 나타남) if st.session_state.get('results') is not None: # 🟢 [수정 2] UI 명칭 변경 및 프롬프트 역할 강조 ai_manual_note = st.text_area( "💡 AI Additional Prompt (Optional)", placeholder="AI에게 전달할 지시사항 (예: 전체 수집된 이메일 내역을 보고 Adviser note 알아서 생성 등)", height=100 ) # 🟢 1. 현재 작업 중인 탭이 '이메일'인지 확인합니다. current_mode = st.session_state.get('active_mode', 'NOTION') is_email = (current_mode == "EMAIL") # 🟢 2. 이메일 탭일 경우 체크박스를 강제로 끄고(False), 클릭 못 하게 잠급니다(disabled). use_ai = st.checkbox( "🤖 AI로 자동 작성하기 (API 비용 발생)", value=False, disabled=is_email, help="🚫 이메일 모드에서는 AI 요약이 제한되며 빈 템플릿만 생성됩니다." if is_email else "체크를 끄면 비용 없이 날짜만 추출된 빈 템플릿이 즉시 생성됩니다." ) if use_ai: cost_estimate = estimate_ai_summary_cost( st.session_state.get('results'), st.session_state['rules_df'], ai_manual_note ) approx_nzd = cost_estimate["usd"] * 1.7 st.info( f"Estimated Gemini cost before run: " f"{cost_estimate['calls']} call(s), " f"~{cost_estimate['input_tokens']:,} input tokens, " f"~{cost_estimate['output_tokens']:,} output tokens, " f"about USD ${cost_estimate['usd']:.4f} / NZD ${approx_nzd:.4f}." ) else: st.caption("AI cost estimate: USD $0.0000 because AI auto-write is off.") col_ai, col_dl = st.columns(2) with col_ai: if st.button("✨ Run AI Summary", use_container_width=True): with st.spinner("Adviser AI Generating Report..."): # 🔄 [실시간 탭 감지 엔진] 현재 활성화된 모드에 따라 화면 입력값을 다이렉트로 추적 current_mode = st.session_state.get('active_mode', 'NOTION') if current_mode == "NOTION": # 노션 탭 입력창 값 가져와서 세미콜론 치환 raw_name = st.session_state.get('notion_client_name', '').strip() final_client_name = raw_name.replace(';', ' & ') if raw_name else "N/A" else: # 이메일 탭 입력창 값 가져와서 세미콜론 치환 raw_name = st.session_state.get('client_names', '').strip() final_client_name = raw_name.replace(';', ' & ') if raw_name else "N/A" # 전역 변수에도 최신 상태 최종 박제 (워드 파일명 연동용) st.session_state['global_client_name'] = final_client_name # 상단 메타 헤더 정보 생성 (동적 추출된 이름 주입) combined_content = f"ADVISER FILE NOTES LOG\n\nClient Name: {final_client_name}\nAdviser: Tim Park\nSupport: {st.session_state.get('support_name', 'N/A')}\nStatus: Completed / Final Approval Issued\n" combined_content += "="*67 + "\n" full_txt = st.session_state.get('full_note', '').strip() if full_txt: doc_bio = io.BytesIO() final_doc = Document() # [기존 기능 유지] 상단 메타 헤더 정보 생성 combined_content = f"ADVISER FILE NOTES LOG\n\nClient Name: {final_client_name}\nAdviser: Tim Park\nSupport: {st.session_state.get('support_name', 'N/A')}\nStatus: Completed / Final Approval Issued\n" combined_content += "="*67 + "\n" # [기존 기능 유지] 1. 화면상의 수동 매핑 상태(드롭다운 값)를 세션 데이터프레임에 동기화 if st.session_state.get('results') is not None: for row_idx in st.session_state['results'].index: sel_key = f"sel_{row_idx}" if sel_key in st.session_state: st.session_state['results'].at[row_idx, 'Sec'] = st.session_state[sel_key] # [기존 기능 유지] 2. 현재 작업 모드(EMAIL 또는 NOTION)에 맞춰 소스 데이터 필터링 full_df = st.session_state['results'].copy() if st.session_state.get('results') is not None else pd.DataFrame() if not full_df.empty: if st.session_state.get('active_mode') == "NOTION": df = full_df[full_df['Type'] == "NOTION"].copy() else: df = full_df[full_df['Type'] != "NOTION"].copy() else: df = pd.DataFrame(columns=['Sec', 'Type', 'Date', 'Body', 'Subject']) # [기존 기능 유지] 실시간 규칙 세션 데이터셋 불러오기 rules_list = st.session_state['rules_df'].to_dict('records') # 🚨 [동적 대응 업데이트] 고정된 10개가 아니라 웹에서 추가/삭제된 모든 단계를 실시간으로 유연하게 순회 for row in rules_list: # 항목이 웹에서 동적으로 추가되더라도 에러가 나지 않도록 안전하게 변수 추출 r_no = row.get('No', '-') r_title = row.get('Stage Title', 'Unknown Stage') sk = f"{r_no}. {r_title}" sub1 = str(row.get('Sub 1', '')).strip().lower() sub2 = str(row.get('Sub 2', '')).strip().lower() stage_title_lower = str(r_title).strip().lower() # 🟢 [수정] 노션 데이터의 번호표("02. ", "03. " 등)를 떼어내고 정규화 temp_df = df.copy() temp_df['Sec_Clean'] = temp_df['Sec'].astype(str).str.replace(r'^\d+\.\s*', '', regex=True).str.strip().str.lower() # 🟢 [핵심] 수동 매핑(Sub 1, 2)과 노션 자동 매핑(Stage Title)을 모두 인정해 줍니다! valid_targets = [sub1, sub2, stage_title_lower] valid_targets = [x for x in valid_targets if x not in ["", "nan", "x", "none", "-"]] targets = temp_df[temp_df['Sec_Clean'].isin(valid_targets)] if not temp_df.empty else pd.DataFrame() combined_content += f"\n[{sk}]\n" # [기존 기능 유지] 매핑된 이메일이 있거나, 혹은 이메일이 없더라도 사용자가 AI Additional Prompt를 적은 경우에만 AI 작동 if not targets.empty or ai_manual_note.strip(): e_data = targets.to_dict('records') if not targets.empty else [] # [기존 기능 유지] Ambiguous 에러 방지를 위해 row를 dict형으로 안전하게 변환하여 AI 호출 note_text = ask_ai_note(sk, e_data, dict(row), ai_manual_note) combined_content += f"{note_text}\n\n" else: # [기존 기능 유지] 매핑된 이메일도 없고 추가 프롬프트 내용도 없는 구역은 깨끗하게 공란 폼만 출력 combined_content += "Date: \nAction: \nPurpose: \n\n" # [기존 기능 유지] 최종 결과물을 사이드바 에디터 세션에 주입하고 화면 리프레시 st.session_state['full_note'] = combined_content st.session_state['editor_version'] = datetime.now().strftime("%H%M%S") st.rerun() with col_dl: # 🟢 전역 변수로 확정된 이름을 가져옵니다. current_mode = st.session_state.get('active_mode', 'NOTION') if current_mode == "NOTION": raw_name = st.session_state.get('notion_client_name', '').strip() final_client_name = raw_name.replace(';', ' & ') if raw_name else "N/A" else: raw_name = st.session_state.get('client_names', '').strip() final_client_name = raw_name.replace(';', ' & ') if raw_name else "N/A" full_txt = st.session_state.get('full_note', '').strip() if full_txt: doc_bio = io.BytesIO() final_doc = Document() # 워드 문서 전체 기본 폰트 설정 style = final_doc.styles['Normal'] font = style.font font.name = 'Arial' font.size = Pt(11) lines = full_txt.split('\n') for i, line in enumerate(lines): # 1. 문서 제목 (가운데 정렬, 굵게, 크기 키움) if line.strip() == "ADVISER FILE NOTES LOG": p = final_doc.add_paragraph() p.alignment = WD_ALIGN_PARAGRAPH.CENTER run = p.add_run(line.strip()) run.bold = True run.font.size = Pt(16) continue # 2. 메타 정보 (이름, 어드바이저 등 항목 이름만 굵게 처리) is_meta = False for meta_key in ["Client Name:", "Adviser:", "Support:", "Status:"]: if line.startswith(meta_key): p = final_doc.add_paragraph() parts = line.split(meta_key, 1) run_bold = p.add_run(meta_key) run_bold.bold = True if len(parts) > 1: p.add_run(parts[1]) is_meta = True break if is_meta: continue # 3. 최상단 실선 구분선 유지 (====) if line.startswith('===='): p_line = final_doc.add_paragraph(line) p_line.paragraph_format.space_after = Pt(0) continue # 4. 각 섹션 제목 ([1. INITIAL MEETING] 등 파란색, 굵게) if line.strip().startswith('[') and line.strip().endswith(']'): if i > 10: p_dash = final_doc.add_paragraph() p_dash.add_run("-" * 117) p = final_doc.add_paragraph() run = p.add_run(line.strip()) run.bold = True run.font.color.rgb = RGBColor(30, 58, 138) run.font.size = Pt(12) continue # 5. 일반 텍스트 내 Date, Action, Purpose 라벨만 진하게 처리하는 구역 stripped_line = line.strip() is_label_line = False for target_label in ["Date", "Action", "Purpose"]: if stripped_line.startswith(target_label): p = final_doc.add_paragraph() if ":" in stripped_line: label_part, content_part = stripped_line.split(":", 1) run_label = p.add_run(f"{label_part}:") run_label.bold = True p.add_run(content_part) else: run_all = p.add_run(stripped_line) run_all.bold = True is_label_line = True break if is_label_line: continue final_doc.add_paragraph(line) final_doc.save(doc_bio) # 🛠️ 수정: 안전하게 동기화된 final_client_name으로 파일명 빌드 및 다운로드 활성화 st.download_button("💾 Save Word", data=doc_bio.getvalue(), file_name=f"Adviser_File_Notes_Log_{final_client_name}.docx", use_container_width=True) else: st.button("💾 Save Word", use_container_width=True, disabled=True, help="Run AI Summary first") st.divider() # [3] 통합 텍스트 에디터 (중복 박스 제거, 텍스트 창만 유지) if 'full_note' not in st.session_state: st.session_state['full_note'] = "" ver = st.session_state.get('editor_version', 'default') st.session_state['full_note'] = st.text_area( label="Integrated Note Editor", value=st.session_state['full_note'], height=800, key=f"final_editor_{ver}", label_visibility="collapsed" ) # --- [Main Area: Configuration] --- st.title("📑 Flux Finance Internal Note") # 💡 [세션 동기화 보강] 앱 초기 구동 시 액티브 모드가 비어있으면 기본값으로 이메일을 지정하여 락을 해제합니다. if 'active_mode' not in st.session_state: st.session_state['active_mode'] = "Notion" # 💡 [탭 순서 혁신] 노션 탭을 배열의 맨 앞으로 배치하여 웹 접속 시 첫 화면에 무조건 노션 입력창이 먼저 뜨도록 제어합니다. tab_notion, tab_contacts, tab_email = st.tabs(["📝 Pull Notion", "👥 Notion Contacts", "✉️ Scan Emails"]) # TAB 1: 노션 검색창 with tab_notion: with st.container(border=True): st.caption("💡 **Notion Sync:** Pull matching compliance tracks and thread messages.") c_notion1, c_notion2 = st.columns([5, 1]) search_client = c_notion1.text_input("👤 Case Name (Applicants Name)", value=st.session_state['notion_client_name'], key="notion_search_bar_fixed_tab") st.session_state['notion_client_name'] = search_client notion_sync_btn = c_notion2.button("🔍 Pull", use_container_width=True, key="notion_sync_trigger") if notion_sync_btn and search_client: st.session_state['active_mode'] = "NOTION" st.session_state['global_client_name'] = search_client.replace(';', ' & ').strip() with st.status("🔍 Syncing Notion Workspace...", expanded=True) as status: # 🟢 리턴받은 case_comments를 세션에 저장 tasks_data, msg, case_comments = fetch_client_tasks_from_notion(search_client) if tasks_data: st.session_state['results'] = pd.DataFrame(tasks_data) st.session_state['results']['Type'] = "NOTION" st.session_state['case_comments'] = case_comments # 코멘트 세션 보관 update_status() status.update(label="✅ Notion Tasks Synced.", state="complete") st.rerun() elif tasks_data is not None and not tasks_data: status.update(label="⚠️ 매칭 데이터 없음", state="error") else: status.update(label=f"❌ Failed: {msg}", state="error") case_comments = st.session_state.get('case_comments', '').strip() if case_comments: html_case_comments = html.escape(case_comments).replace('\n', '
') st.markdown(f"""
💬 Case Comments:

{html_case_comments}
""", unsafe_allow_html=True) # TAB 2: 이메일 검색창 with tab_contacts: with st.container(border=True): st.caption("Pull the full customer list from Notion with only the fields you choose.") contact_col1, contact_col2 = st.columns([5, 1]) contact_search = contact_col1.text_input( "Optional filter", key="notion_contact_search", placeholder="Leave blank for the full list" ) contact_pull_btn = contact_col2.button("Pull Full List", use_container_width=True, key="notion_contact_pull") selected_contact_fields = st.multiselect( "Fields to pull", options=list(NOTION_CONTACT_FIELD_CONFIG.keys()), default=["Client Name", "Nickname", "Phone"], key="notion_contact_fields" ) custom_contact_fields_raw = st.text_input( "Additional Notion property names", key="notion_contact_custom_fields", placeholder="Example: DOB; Address; Loan Amount" ) custom_contact_fields = [ field.strip() for field in custom_contact_fields_raw.replace(",", ";").split(";") if field.strip() ] if contact_pull_btn: with st.status("Pulling Notion contacts...", expanded=True) as status: contact_rows, contact_msg = fetch_client_contacts_from_notion( contact_search, selected_fields=selected_contact_fields, custom_fields=custom_contact_fields, ) if contact_rows: st.session_state["notion_contacts"] = pd.DataFrame(contact_rows) status.update(label=f"Loaded {len(contact_rows)} contacts.", state="complete") elif contact_msg == "Success": st.session_state["notion_contacts"] = pd.DataFrame(columns=selected_contact_fields + custom_contact_fields) status.update(label="No contacts found.", state="complete") else: status.update(label=f"Failed: {contact_msg}", state="error") contacts_df = st.session_state.get("notion_contacts") if isinstance(contacts_df, pd.DataFrame): contacts_df = contacts_df.fillna("").astype(str) phone_columns = [col for col in contacts_df.columns if is_phone_like_field(col)] for col in phone_columns: contacts_df[col] = contacts_df[col].map(normalize_phone_text) st.dataframe(contacts_df, use_container_width=True, hide_index=True) csv_df = contacts_df.copy() for col in phone_columns: csv_df[col] = csv_df[col].map(make_excel_safe_phone) csv_data = csv_df.to_csv(index=False).encode("utf-8-sig") st.download_button( "Download CSV", data=csv_data, file_name=f"notion_customer_list_{datetime.now().strftime('%Y%m%d_%H%M')}.csv", mime="text/csv", use_container_width=True, key="download_notion_contacts_csv" ) with tab_email: with st.container(border=True): st.session_state['folder_path'] = st.text_input("📂 G: Drive Folder Path", value=st.session_state['folder_path'], key="fixed_mail_path") # 🎯 [추가] secrets에서 주소만 쏙 발라내어 멀티 셀렉트 박스 생성 raw_mailbox_list = list(load_active_mailboxes().values()) mailbox_options = [] mailbox_map = {} for entry in raw_mailbox_list: entry_clean = entry.strip() if not entry_clean: continue clean_email = parse_mailbox_email(entry_clean) mailbox_options.append(clean_email) mailbox_map[clean_email] = entry_clean selected_boxes = st.multiselect("📬 Select Mailboxes to Scan", options=mailbox_options, default=[], key="active_mailboxes_select") c1, c2, c3, c4 = st.columns([2, 2, 1.5, 1]) st.session_state['emails'] = c1.text_input("Emails", value=st.session_state['emails'], key="fixed_mail_addrs") st.session_state['client_names'] = c2.text_input("Names", value=st.session_state['client_names'], key="fixed_mail_names") st.session_state['refs'] = c3.text_input("Keywords", value=st.session_state['refs'], key="fixed_mail_refs") email_sync_btn = c4.button("🚀 Sync Emails", use_container_width=True, key="email_sync_trigger") # ✨ [디자인 혁신] 투박한 안내 문구를 엔터프라이즈급 가이드 대시보드로 변경 st.markdown("""
💡 System Usage Instructions
Use a semicolon (;) to search multiple items..
EMAILS SCOPE
Scans From / To / CC
NAMES SCOPE
Scans From / To / CC / Subject.
※ Order-agnostic full AND match
KEYWORDS SCOPE
Scans Subject / Body
""", unsafe_allow_html=True) if email_sync_btn: st.session_state['active_mode'] = "EMAIL" if not selected_boxes: st.warning("Please select at least one mailbox to scan.") st.stop() # 🟢 [여기에 추가] 이메일 이름을 전역 변수로 확정 박아버림! (여러 명이면 &로 치환) raw_names = st.session_state.get('client_names', '').strip() st.session_state['global_client_name'] = raw_names.replace(';', ' & ') if raw_names else '' with st.status("🔍 Syncing...", expanded=True) as status: progress_text = st.empty() progress_text.info("Connecting and scanning emails... ⏳") if st.session_state['folder_path']: txt, msg = load_docs_from_path(st.session_state['folder_path']) st.session_state['doc_context'] = txt if txt else ""; st.write(msg) all_mails = [] e_list = [e.strip().lower() for e in st.session_state['emails'].replace(";", ",").split(",") if "@" in e] name_groups = [[w.strip().lower() for w in n.replace(",", " ").split() if len(w.strip()) > 1] for n in st.session_state['client_names'].split(";")] ref_list = [r.strip().lower() for r in st.session_state['refs'].replace(";", ",").split(",") if len(r.strip()) > 3] stats = {"count": 0} # 🎯 [교체 완료] 전체를 다 도는 대신, 멀티 셀렉트에서 선택된 사서함들만 추출 target_mailbox_entries = [mailbox_map[m] for m in selected_boxes if m in mailbox_map] if target_mailbox_entries: try: import email.utils socket.setdefaulttimeout(20) for entry in target_mailbox_entries: entry_clean = entry.strip() if not entry_clean: continue if "|" in entry_clean: credential_part, imap_host = entry_clean.split("|", 1) imap_host = imap_host.replace("|", "").strip() else: credential_part = entry_clean imap_host = "arrow.mxrouting.net" if ":" in credential_part: my_mail_addr, custom_pw = credential_part.split(":", 1) my_mail_addr = my_mail_addr.strip() target_pw = custom_pw.strip() else: my_mail_addr = credential_part.strip() target_pw = COMMON_PW status.write(f"🔐 Scanning Mailbox: '{my_mail_addr}' via {imap_host}:993...") mail = None last_connect_error = None connect_errors = [] for host_candidate in get_imap_host_candidates(imap_host, my_mail_addr): try: status.write(f"🌐 Connecting to {host_candidate}:993...") mail = IMAP4_SSL(host_candidate, 993, timeout=45) mail.login(my_mail_addr, target_pw) imap_host = host_candidate break except Exception as connect_error: last_connect_error = connect_error connect_errors.append(f"{host_candidate}: {connect_error}") try: if mail: mail.logout() except Exception: pass mail = None if mail is None: status.write(f"❌ Mailbox failed '{my_mail_addr}': {' | '.join(connect_errors) or last_connect_error}") continue try: res_list, folder_list = mail.list() all_target_folders = [] if res_list == "OK": for f_info in folder_list: f_str = f_info.decode("utf-8", errors="ignore") # 🎯 [하위 폴더 공백/특수문자 파싱 버그 완벽 교정] # 폴더 전체 경로를 찢어 먹지 않고 쌍따옴표 구역을 역추적해 통째로 가져옵니다. if f_str.endswith('"'): last_quote = f_str.rfind('"') prev_quote = f_str.rfind('"', 0, last_quote) raw_folder_name = f_str[prev_quote+1 : last_quote] else: raw_folder_name = f_str.split()[-1].strip().strip('"') if any(k in raw_folder_name.lower() for k in ["junk", "trash", "deleted", "sync", "spam"]): continue if raw_folder_name: all_target_folders.append(raw_folder_name) if not all_target_folders: all_target_folders = ["INBOX", "Sent"] status.write(f"📂 Found {len(all_target_folders)} folder(s). Scanning latest {MAX_EMAILS_PER_FOLDER} messages per folder.") for folder in all_target_folders: try: mail.select(f'"{folder}"', readonly=True) res, msg_ids = mail.search(None, "ALL") matched_ids = msg_ids[0].split()[-MAX_EMAILS_PER_FOLDER:] if res == "OK" and msg_ids and msg_ids[0] else [] if not matched_ids: continue status.write(f"📁 {folder}: checking latest {len(matched_ids)} message(s)") # 획득한 메일 번호 기반 본문 추출 엔진 정위치 안착 for m_id in reversed(matched_ids): stats["count"] += 1 res_body, body_data = mail.fetch(m_id, "(RFC822)") if res_body != "OK" or not body_data or not body_data[0]: continue msg_raw = email.message_from_bytes(body_data[0][1]) subj = str(make_header(decode_header(msg_raw.get("Subject", "")))) snd = str(make_header(decode_header(msg_raw.get("From", "")))) to_header = str(msg_raw.get("To", "")) # 💡 대소문자 통합 방어막 구축: Cc든 CC, cc든 대소문자 구분 없이 무조건 포획합니다. cc_raw = "" for header_key in msg_raw.keys(): if header_key.lower() == "cc": cc_raw = msg_raw.get(header_key, "") break cc_header = str(make_header(decode_header(str(cc_raw)))) if cc_raw else "" body = "" if msg_raw.is_multipart(): for b_part in msg_raw.walk(): if b_part.get_content_type() == "text/plain": try: body = b_part.get_payload(decode=True).decode(errors='ignore') break except: pass else: try: body = msg_raw.get_payload(decode=True).decode(errors='ignore') except: body = "본문을 읽어올 수 없습니다." attachments = [] if msg_raw.is_multipart(): for b_part in msg_raw.walk(): filename = b_part.get_filename() if filename: decoded_filename = str(make_header(decode_header(filename))) attachments.append(decoded_filename) try: dt = email.utils.parsedate_to_datetime(msg_raw.get("Date")) except Exception: dt = datetime.now() # 🎯 [타임존 변환 버그 교정] 서버 시간(UTC)을 뉴질랜드(Auckland) 현지 시간으로 정확하게 강제 변환합니다. if dt and dt.tzinfo: try: from zoneinfo import ZoneInfo dt = dt.astimezone(ZoneInfo("Pacific/Auckland")) except: dt = dt.astimezone(None) # 🎯 각 항목별 검색 구역을 완벽하게 분리 email_area = (snd + " " + to_header + " " + cc_header).lower() name_area = (subj + " " + snd + " " + to_header + " " + cc_header).lower() ref_area = (subj + " " + body).lower() # 1. 이메일 주소는 오직 From, To, CC 영역에서만 매칭 (OR) email_matched = any(e_addr in email_area for e_addr in e_list) if e_list else False # 2. 대출 참조 번호는 오직 제목(Subject)과 본문(Body) 영역에서만 매칭 (OR) ref_matched = any(r_num in ref_area for r_num in ref_list) if ref_list else False # 3. 이름은 From, To, CC, Subject 영역에서 검사하되, 성과 이름 단어가 '순서 상관없이 모두(AND)' 존재해야 함! name_matched = False if name_groups and any(len(g) > 0 for g in name_groups): for group in name_groups: # 예: 'misun'과 'lim'이 이름 검사 영역에 동시에 다 들어있어야만 true! (Misun Kim 차단) if group and all(word in name_area for word in group): name_matched = True break # 4. 세 카테고리(이메일 주소 vs 대출번호 vs 이름) 간의 최종 관계는 완전한 OR if email_matched or name_matched or ref_matched: all_mails.append({ "Date": dt.strftime("%d %b %Y, %I:%M %p"), "Subject": subj, "From": snd, "To": to_header, "Cc": cc_header, "Type": "SENT" if any(dom in snd.lower() for dom in ["fluxfinance.co.nz", "ewfinance.co.nz"]) else "RECEIVED", "Body": body, "Sec": "-", "Attach": attachments, "RawDate": dt.replace(tzinfo=None) }) except Exception as folder_error: status.write(f"⚠️ Skipped folder '{folder}': {folder_error}") continue mail.logout() except Exception as mailbox_error: status.write(f"❌ Mailbox failed '{my_mail_addr}': {mailbox_error}") except Exception as e: st.error(f"IMAP Engine Error: {e}") if all_mails: all_mails.sort(key=lambda x: x.get("RawDate", datetime.min)) df = pd.DataFrame(all_mails).drop_duplicates(subset=['Subject', 'Date']) st.session_state['results'] = df.reset_index(drop=True) assigned = st.session_state['results']['Sec'].tolist() for idx, r in st.session_state['rules_df'].iterrows(): s1 = str(r.get('Sub 1', '')).strip() s2 = str(r.get('Sub 2', '')).strip() has_s1 = s1 not in ["", "nan", "x", "None", "-"] has_s2 = s2 not in ["", "nan", "x", "None", "-"] is_s1_mapped = s1 in assigned if has_s1 else False is_s2_mapped = s2 in assigned if has_s2 else False if has_s1 and not has_s2: st.session_state['status'][idx] = "🟢" if is_s1_mapped else "⚪" elif has_s1 and has_s2: if is_s1_mapped and is_s2_mapped: st.session_state['status'][idx] = "🟢" elif is_s1_mapped or is_s2_mapped: st.session_state['status'][idx] = "🟠" else: st.session_state['status'][idx] = "⚪" else: st.session_state['status'][idx] = "⚪" status.update(label=f"✅ Scan complete! Found {len(all_mails)} matching email(s).", state="complete", expanded=False) else: st.session_state['results'] = None progress_text.error(f"Checked {stats['count']} emails. No matches found.") # ───────────────────────────────────────────────────────────────── # 🎯 [4단계] 통합 결과창 드로잉 레이아웃 # ───────────────────────────────────────────────────────────────── if st.session_state['results'] is not None: update_status() for idx, row in st.session_state['results'].iterrows(): # 모드 필터링 if st.session_state['active_mode'] == "NOTION" and row.get('Type') != "NOTION": continue if st.session_state['active_mode'] == "EMAIL" and row.get('Type') == "NOTION": continue with st.container(border=True): # [1] 타입별 상단 헤더 출력 (유지) if row['Type'] == "NOTION": st.markdown(f" 📌 Stage: {row['Sec']}", unsafe_allow_html=True ) # 🟢 날짜 옆에 Channel 값을 함께 출력합니다. channel_info = str(row.get('Channel', '-')).strip() st.markdown(f"""
📅 Completed On: {row['Date']}   |   📡 Channel: {channel_info}
""", unsafe_allow_html=True) else: # 1. 태그와 제목 출력 st.markdown(f"{row['Type']} **{row['Subject']}**", unsafe_allow_html=True) # 2. 날짜를 상단에 단독 배치 st.markdown(f"
📅 {row['Date']}
", unsafe_allow_html=True) # 3. From, To, CC 각각 개별 줄바꿈 cc_value = str(row.get('Cc', '')).strip() cc_display = cc_value if cc_value and cc_value.lower() != 'nan' else "None" st.markdown(f"""
From: {row['From']}
To: {row['To']}
CC: {cc_display}
""", unsafe_allow_html=True) # 💡 [두 번째 if NOTION]: 매핑 상태 (배지 vs 드롭다운) 처리 구역 rules = st.session_state['rules_df'] rules_list = rules.to_dict('records') # 🚨 [기능 추가] 웹에서 실시간으로 추가/수정된 모든 단계를 드롭다운 옵션에 자동 반영 opts = ["-"] for _, r in rules.iterrows(): s1 = str(r.get('Sub 1', '')).strip() s2 = str(r.get('Sub 2', '')).strip() if s1 and s1 not in ['nan', 'x', '', 'None', '-'] and s1 not in opts: opts.append(s1) if s2 and s2 not in ['nan', 'x', '', 'None', '-'] and s2 not in opts: opts.append(s2) # 웹 설정에서 새로 수정한 값이나 기존 값이 opts 배열에 누수되었을 경우를 방지하는 실시간 안전망 current_sec_val = str(row.get('Sec', '-')).strip() if current_sec_val not in opts and current_sec_val != "-": opts.insert(1, current_sec_val) if row['Type'] == "NOTION": clean_raw = re.sub(r'^\d+\.\s*', '', str(row['RawTitle'])).strip().lower() stage_titles = [str(x).strip().lower() for x in rules['Stage Title'].tolist()] if clean_raw in stage_titles: # 🟢 [기존 기능 유지] 일치함: 자동 매핑 배지 표시 matched_stage_idx = stage_titles.index(clean_raw) matched_stage_name = rules['Stage Title'].iloc[matched_stage_idx] st.markdown(f"""
✅ Auto-Mapped: {matched_stage_name}
""", unsafe_allow_html=True) else: # ⚠️ [기존 기능 유지] 일치하지 않음: 수동 할당을 위한 경고 배지 및 드롭다운 활성화 st.markdown("
⚠️ Unmapped (Assign Manually)
", unsafe_allow_html=True) st.selectbox("Assign Action", opts, index=opts.index(current_sec_val) if current_sec_val in opts else 0, key=f"sel_{idx}", label_visibility="collapsed") else: # 🟢 [기존 기능 유지] 이메일 전용: 수동 할당 드롭다운 st.selectbox("Assign Action", opts, index=opts.index(current_sec_val) if current_sec_val in opts else 0, key=f"sel_{idx}", label_visibility="collapsed") # [기존 기능 유지] 본문(Body) 출력 st.markdown(f"
{row.get('Body', '')}
", unsafe_allow_html=True) # 🟢 [기존 기능 유지] 코멘트(Comments)가 있을 경우, 본문 아래에 별도의 말풍선 박스로 예쁘게 렌더링 comments_data = row.get('Comments', '').strip() if comments_data: html_comments = html.escape(comments_data).replace('\n', '
') st.markdown(f"""
💬 Notion Comments:

{html_comments}
""", unsafe_allow_html=True) # [기존 기능 유지] 첨부파일 출력 구역 if row.get('Attach'): st.markdown(f"
📎 Attachments: {', '.join(row['Attach'])}
", unsafe_allow_html=True)