# -*- coding: utf-8 -*-
"""
오너클랜 신규 알림 수집기 v5 - 로그인 셀렉터 고정 버전
v5 핵심 수정:
- 오너클랜 로그인 HTML 기준으로 정확한 셀렉터 사용
아이디: input#id 또는 input[name='id']
비번: input#passwd 또는 input[name='passwd']
폼: form[name='loginForm']
로그인 버튼: input.img_log[type='submit']
- 기존처럼 엉뚱한 검색창 input을 아이디칸으로 잡는 문제 방지
- 입력 후 submit 버튼 클릭 → 실패하면 form.submit() fallback
- 수집 결과는 프로그램 카드 화면에 표시
"""
import importlib
import subprocess
import sys
def _ensure_deps():
"""
다른 PC(파이썬만 설치된 환경)에서도 바로 실행되도록,
필요한 외부 패키지가 없으면 자동으로 pip 설치한다.
selenium/openpyxl(수집), pystray/pillow(트레이 상주) 모두 포함.
"""
if getattr(sys, "frozen", False):
return # 런처 EXE 에 라이브러리가 이미 포함됨
no_window = 0x08000000 if sys.platform == "win32" else 0
need = []
for mod, pip_name in [
("selenium", "selenium"),
("openpyxl", "openpyxl"),
("pystray", "pystray"),
("PIL", "pillow"),
("tkcalendar", "tkcalendar"),
]:
try:
importlib.import_module(mod)
except Exception:
need.append(pip_name)
if need:
try:
subprocess.run(
[sys.executable, "-m", "pip", "install", *need],
creationflags=no_window,
)
except Exception:
pass
_ensure_deps()
import csv
import json
import math
import struct
import threading
import time
import traceback
import wave
from concurrent.futures import ThreadPoolExecutor, as_completed
from dataclasses import dataclass, asdict
from datetime import datetime, timedelta
from pathlib import Path
import tkinter as tk
from tkinter import ttk, messagebox, simpledialog, filedialog
try:
import winsound
except Exception:
winsound = None
try:
import pystray
from PIL import Image, ImageDraw
except Exception:
pystray = None
Image = None
ImageDraw = None
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.common.keys import Keys
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.chrome.service import Service
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
CREATE_NO_WINDOW = 0x08000000 # chromedriver 콘솔창 숨김용
class ChromeProcessRegistry:
"""
이 프로그램이 띄운 chromedriver 프로세스를 추적합니다.
- 종료/재시작 시 '우리가 만든 chromedriver + 그 자식 Chrome' 만 정리하기 위한 등록부.
- 사용자가 직접 쓰는 일반 Chrome 은 우리 chromedriver 의 자식이 아니므로 절대 건드리지 않습니다.
- PID 재사용 사고를 막기 위해 종료 직전 tasklist 로 chromedriver.exe 인지 다시 확인합니다.
"""
def __init__(self):
self._lock = threading.Lock()
self._items = {} # pid -> {"kind": "background"|"visible", "created": ts}
@staticmethod
def driver_pid(driver):
try:
return int(driver.service.process.pid)
except Exception:
return None
def register(self, driver, kind):
pid = self.driver_pid(driver)
if pid:
with self._lock:
self._items[pid] = {"kind": kind, "created": time.time()}
return pid
def unregister_pid(self, pid):
if pid:
with self._lock:
self._items.pop(pid, None)
def snapshot(self):
with self._lock:
return dict(self._items)
@staticmethod
def _is_chromedriver_alive(pid):
if sys.platform != "win32":
return False
try:
out = subprocess.check_output(
["tasklist", "/FI", f"PID eq {int(pid)}", "/FO", "CSV", "/NH"],
creationflags=CREATE_NO_WINDOW, stderr=subprocess.DEVNULL, timeout=5,
).decode("mbcs", errors="ignore")
except Exception:
return False
return "chromedriver" in out.lower()
def kill_pid(self, pid, log=None):
"""등록된 chromedriver(pid)와 그 자식 Chrome 트리를 강제 종료. 이미 죽었으면 등록만 해제."""
if not pid:
return False
killed = False
try:
if self._is_chromedriver_alive(pid):
subprocess.run(
["taskkill", "/PID", str(int(pid)), "/T", "/F"],
creationflags=CREATE_NO_WINDOW, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=10,
)
killed = True
if log:
try:
log(f"[정리] 남은 chromedriver/Chrome 강제 종료 (pid {pid})")
except Exception:
pass
except Exception:
pass
self.unregister_pid(pid)
return killed
def kill_all(self, kinds=("background", "visible"), log=None):
n = 0
for pid, info in self.snapshot().items():
if info.get("kind") in kinds and self.kill_pid(pid, log=log):
n += 1
return n
CHROME_REGISTRY = ChromeProcessRegistry()
def quit_driver(driver, timeout=6.0, log=None):
"""
driver.quit() 를 최대 timeout 초만 기다리고, 그래도 안 끝나면 등록부 기준으로 강제 종료합니다.
UI 스레드가 아닌 곳에서 호출하는 것을 권장 (종료 시에는 별도 스레드에서 호출).
"""
if driver is None:
return
pid = CHROME_REGISTRY.driver_pid(driver)
def _q():
try:
driver.quit()
except Exception:
pass
t = threading.Thread(target=_q, daemon=True)
t.start()
t.join(timeout)
# 정상 종료됐으면 tasklist 에서 chromedriver 가 안 보여 kill 은 건너뛰고 등록만 해제됨
CHROME_REGISTRY.kill_pid(pid, log=log)
try:
from openpyxl import Workbook
except Exception:
Workbook = None
# 달력 위젯(할일/메모 알림의 '특정 일시' 입력). 설치 실패 시 None → 텍스트 입력으로 폴백.
try:
from tkcalendar import DateEntry
except Exception:
DateEntry = None
APP_DIR = Path(__file__).resolve().parent
CONFIG_PATH = APP_DIR / "accounts.json"
SEEN_PATH = APP_DIR / "seen_alerts.json"
SETTINGS_PATH = APP_DIR / "settings.json"
TODO_PATH = APP_DIR / "todo_memos.json" # 할일/메모 알림 저장
ALERTS_CACHE_PATH = APP_DIR / "alerts_cache.json" # 마지막 수집 결과(재시작 후에도 카드 표시)
OUTPUT_DIR = APP_DIR / "output"
DEBUG_DIR = APP_DIR / "debug"
ICON_ICO = APP_DIR / "ownerclan_icon.ico" # 창/작업표시줄 아이콘
ICON_PNG = APP_DIR / "ownerclan_icon.png" # 트레이 아이콘
OUTPUT_DIR.mkdir(exist_ok=True)
DEBUG_DIR.mkdir(exist_ok=True)
LOGIN_URL = "https://ownerclan.com/V2/member/loginform.php"
DEFAULT_MEMO_URL = "https://ownerclan.com/V2/service/alert_noti_list.php"
QNA_LIST_URL = "https://ownerclan.com/V2/service/qnaList.php"
ACCOUNT_FILTER_ALL = "전체 사업자" # 사업자 필터: 전체 보기 값
TODO_COMMON_LABEL = "(공통)" # 할일/메모 알림에서 특정 사업자 미지정(=공통) 표기
# =============================================================================
# 자동 업데이트 (Cloudflare Pages 에 올린 version.json / .py 를 시작 시 확인)
# - 프로그램을 껐다 켜면 새 버전이 있을 때 자동으로 내려받아 교체 후 재시작합니다.
# - settings.json 의 "auto_update": false 로 끌 수 있고, "update_url" 로 주소를 바꿀 수 있습니다.
# - 실행 인자 --no-update 가 있으면 이번 실행에서는 확인하지 않습니다(재시작 직후 중복 확인 방지).
# =============================================================================
__version__ = "22.0.6"
__released__ = "2026-09-04 14:58" # 배포 시각 (publish.py 가 자동 기록)
UPDATE_BASE_URL_DEFAULT = "https://ownerclan-update.pages.dev"
UPDATE_MANIFEST_NAME = "version.json"
UPDATE_TIMEOUT_SEC = 8
UPDATE_LOG_PATH = APP_DIR / "update.log"
BACKUP_DIR = APP_DIR / "backup"
SELF_PATH = Path(__file__).resolve()
def _update_log(msg):
line = f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] {msg}"
try:
with open(UPDATE_LOG_PATH, "a", encoding="utf-8") as f:
f.write(line + "\n")
except Exception:
pass
return line
def _parse_version(s):
"""'22.1.3' / 'v22.1' 같은 문자열을 (22, 1, 3) 튜플로. 비교용."""
parts = []
for p in str(s or "").strip().lstrip("vV").split("."):
digits = "".join(ch for ch in p if ch.isdigit())
parts.append(int(digits) if digits else 0)
while len(parts) < 3:
parts.append(0)
return tuple(parts[:3])
def _update_base_url():
settings = load_json(SETTINGS_PATH, {})
url = str(settings.get("update_url") or UPDATE_BASE_URL_DEFAULT).strip()
return url.rstrip("/")
def _http_get(url, timeout=UPDATE_TIMEOUT_SEC):
import urllib.request
req = urllib.request.Request(
url,
headers={
"User-Agent": f"ownerclan-updater/{__version__}",
"Cache-Control": "no-cache",
"Pragma": "no-cache",
},
)
with urllib.request.urlopen(req, timeout=timeout) as resp:
return resp.read()
def check_for_update(timeout=UPDATE_TIMEOUT_SEC):
"""
서버의 version.json 을 읽어 (manifest, error) 를 돌려줍니다.
manifest 예: {"version": "22.0.1", "file": "ownerclan_integrated_v22.py", "sha256": "...", "notes": "..."}
새 버전이 없어도 manifest 는 돌려주며, 호출한 쪽에서 버전을 비교합니다.
"""
base = _update_base_url()
url = f"{base}/{UPDATE_MANIFEST_NAME}?t={int(time.time())}"
try:
raw = _http_get(url, timeout=timeout)
manifest = json.loads(raw.decode("utf-8"))
if not isinstance(manifest, dict) or not manifest.get("version") or not manifest.get("file"):
return None, "version.json 형식이 올바르지 않습니다."
manifest["_base"] = base
return manifest, None
except Exception as e:
return None, f"{type(e).__name__}: {e}"
def is_newer_version(manifest):
try:
return _parse_version(manifest.get("version")) > _parse_version(__version__)
except Exception:
return False
def apply_update(manifest, log=None):
"""
새 .py 를 내려받아 sha256 검증 → 문법 검사 → 백업 → 현재 실행 파일을 교체합니다.
성공하면 True. 실패하면 False (기존 파일은 그대로 유지).
"""
import hashlib
import os
def _log(m):
line = _update_log(m)
if log:
try:
log(line)
except Exception:
pass
base = manifest.get("_base") or _update_base_url()
file_name = str(manifest.get("file"))
url = f"{base}/{file_name}?t={int(time.time())}"
try:
_log(f"업데이트 다운로드: {url}")
data = _http_get(url, timeout=max(UPDATE_TIMEOUT_SEC, 30))
except Exception as e:
_log(f"다운로드 실패: {type(e).__name__}: {e}")
return False
expected = str(manifest.get("sha256") or "").lower().strip()
if expected:
actual = hashlib.sha256(data).hexdigest()
if actual != expected:
_log(f"sha256 불일치 (기대 {expected[:12]}…, 실제 {actual[:12]}…) → 적용 중단")
return False
try:
src = data.decode("utf-8")
compile(src, file_name, "exec")
except Exception as e:
_log(f"내려받은 파일 문법 검사 실패: {type(e).__name__}: {e} → 적용 중단")
return False
if f'__version__ = "{manifest.get("version")}"' not in src:
_log("내려받은 파일의 __version__ 이 version.json 과 다릅니다 → 적용 중단")
return False
try:
BACKUP_DIR.mkdir(exist_ok=True)
backup = BACKUP_DIR / f"{SELF_PATH.stem}_{__version__}.py"
backup.write_bytes(SELF_PATH.read_bytes())
tmp = SELF_PATH.with_suffix(".py.new")
tmp.write_bytes(data)
try:
os.replace(str(tmp), str(SELF_PATH))
except Exception:
# 드물게 os.replace 가 막히면 직접 덮어쓰기
SELF_PATH.write_bytes(data)
try:
tmp.unlink()
except Exception:
pass
_log(f"업데이트 적용 완료: {__version__} → {manifest.get('version')} (백업: {backup.name})")
return True
except Exception as e:
_log(f"파일 교체 실패: {type(e).__name__}: {e}")
return False
def relaunch_self(skip_update=True):
"""현재 스크립트를 같은 파이썬으로 다시 실행하고, 이 프로세스는 즉시 종료합니다."""
import os
args = [sys.executable, str(SELF_PATH)] + (["--no-update"] if skip_update else [])
flags = 0
if sys.platform == "win32":
flags = 0x00000008 | 0x00000200 # DETACHED_PROCESS | CREATE_NEW_PROCESS_GROUP
try:
subprocess.Popen(args, cwd=str(APP_DIR), creationflags=flags, close_fds=True)
except Exception as e:
_update_log(f"재시작 실패: {type(e).__name__}: {e}")
return False
_update_log("새 버전으로 재시작")
os._exit(0)
def startup_update_flow():
"""
프로그램 시작 직후(메인 창 만들기 전) 호출.
새 버전이 있으면 작은 안내창을 띄우고 교체 후 재시작합니다. 실패/오프라인이면 그냥 현재 버전으로 계속.
"""
if "--no-update" in sys.argv:
return
settings = load_json(SETTINGS_PATH, {})
if not settings.get("auto_update", True):
return
manifest, err = check_for_update()
if manifest is None:
_update_log(f"업데이트 확인 실패(현재 {__version__} 으로 계속): {err}")
return
if not is_newer_version(manifest):
_update_log(f"최신 버전입니다: {__version__} (서버 {manifest.get('version')})")
return
splash = None
try:
splash = tk.Tk()
splash.title("오너클랜 업데이트")
splash.resizable(False, False)
try:
splash.attributes("-topmost", True)
except Exception:
pass
w, h = 380, 120
sw, sh = splash.winfo_screenwidth(), splash.winfo_screenheight()
splash.geometry(f"{w}x{h}+{(sw - w) // 2}+{(sh - h) // 2}")
tk.Label(
splash,
text=f"새 버전 {manifest.get('version')} 으로 업데이트 중...\n(현재 {__version__})\n잠시만 기다려 주세요.",
font=("맑은 고딕", 11),
justify="center",
).pack(expand=True, fill="both", padx=12, pady=12)
splash.update()
except Exception:
splash = None
ok = apply_update(manifest)
if splash is not None:
try:
splash.destroy()
except Exception:
pass
if ok:
relaunch_self()
# 실패 시: 기존 버전으로 그대로 계속 실행
@dataclass
class Account:
enabled: bool
label: str
user_id: str
password: str
memo_url: str
def is_order_alert(kind, atype):
"""주문/오너클랜·주문/공급사 알림인지 판정 (레드 강조 + 소리/팝업 대상)."""
return (kind or "").strip() == "주문" and (atype or "").strip() in ("오너클랜", "공급사")
def is_qna_bold_alert(row):
"""문의게시판(qnaList.php)에서 굵은 글제목(
)으로 잡힌 문의인지 판정."""
return (row or {}).get("source_type") == "qna_bold"
def is_notify_alert(row):
"""소리/팝업 알림 대상: 기존 주문 알림 + 문의게시판 굵은 글제목."""
return is_order_alert((row or {}).get("알림종류"), (row or {}).get("알림타입")) or is_qna_bold_alert(row)
def load_json(path, default):
try:
if path.exists():
return json.loads(path.read_text(encoding="utf-8"))
except Exception:
pass
return default
def save_json(path, data):
path.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")
class OwnerclanAlertCollector:
def __init__(self, log_func, stop_event, ask_manual_login_func):
self.log = log_func
self.stop_event = stop_event
self.ask_manual_login_func = ask_manual_login_func
def debug_dump(self, driver, account_label, step):
ts = datetime.now().strftime("%Y%m%d_%H%M%S")
safe = "".join(c for c in account_label if c.isalnum() or c in ("_", "-")) or "account"
try:
png = DEBUG_DIR / f"{safe}_{step}_{ts}.png"
driver.save_screenshot(str(png))
self.log(f"[디버그] 스크린샷 저장: {png}")
except Exception as e:
self.log(f"[디버그] 스크린샷 저장 실패: {e}")
try:
html = DEBUG_DIR / f"{safe}_{step}_{ts}.html"
html.write_text(driver.page_source, encoding="utf-8", errors="ignore")
self.log(f"[디버그] HTML 저장: {html}")
except Exception as e:
self.log(f"[디버그] HTML 저장 실패: {e}")
try:
self.log(f"[디버그] 현재 URL: {driver.current_url}")
self.log(f"[디버그] 페이지 제목: {driver.title}")
except Exception:
pass
# ------------------------------------------------------------------
# Chrome 생성: '배경(수집/확인 처리)' 과 '보이기(열기 버튼)' 를 완전히 분리합니다.
# - 배경용: --headless=new (창을 그리지 않음) + 화면 밖 위치 + 첫 실행 UI 억제.
# --disable-gpu 는 넣지 않습니다(일부 그래픽 드라이버에서 headless 창이 실제로 그려지는 원인).
# - 보이기용: 사용자가 보는 일반 Chrome. 기존 동작 그대로.
# - 둘 다 CHROME_REGISTRY 에 등록되어 종료/재시작 시 우리가 만든 프로세스만 정리됩니다.
# ------------------------------------------------------------------
def _base_options(self, fast_mode):
options = Options()
options.add_argument("--disable-popup-blocking")
options.add_argument("--disable-notifications")
options.add_argument("--lang=ko-KR")
options.add_argument("--disable-blink-features=AutomationControlled")
options.page_load_strategy = "eager"
if fast_mode:
# 이미지/폰트 로딩 줄여서 페이지 진입 속도 개선
prefs = {
"profile.managed_default_content_settings.images": 2,
"profile.default_content_setting_values.notifications": 2,
}
options.add_experimental_option("prefs", prefs)
options.add_experimental_option("excludeSwitches", ["enable-automation"])
options.add_experimental_option("useAutomationExtension", False)
return options
@staticmethod
def _make_service():
# chromedriver.exe 콘솔창이 뜨지 않도록 Service에 CREATE_NO_WINDOW 지정
service = Service()
try:
service.creationflags = CREATE_NO_WINDOW
except Exception:
pass
return service
def make_background_driver(self, fast_mode=True):
"""자동수집/확인 처리용. 바탕화면·작업표시줄에 보이면 안 됨."""
options = self._base_options(fast_mode)
options.add_argument("--headless=new")
options.add_argument("--window-size=1200,900")
options.add_argument("--window-position=-32000,-32000")
options.add_argument("--no-first-run")
options.add_argument("--no-default-browser-check")
options.add_argument("--disable-background-networking")
options.add_argument("--disable-sync")
options.add_argument("--no-sandbox")
options.add_argument("--disable-dev-shm-usage")
driver = webdriver.Chrome(service=self._make_service(), options=options)
CHROME_REGISTRY.register(driver, "background")
return driver
def make_visible_driver(self, window_mode="normal", fast_mode=True):
"""'열기' 버튼용. 사용자에게 보여야 함 (기존 동작 유지)."""
options = self._base_options(fast_mode)
options.add_argument("--disable-gpu")
driver = webdriver.Chrome(service=self._make_service(), options=options)
CHROME_REGISTRY.register(driver, "visible")
try:
if window_mode == "tiny":
driver.set_window_position(0, 0)
driver.set_window_size(1, 1)
elif window_mode == "small":
driver.set_window_position(0, 0)
driver.set_window_size(500, 420)
else:
driver.set_window_position(0, 0)
driver.set_window_size(1200, 900)
except Exception:
pass
return driver
def make_driver(self, window_mode="normal", fast_mode=True):
"""
호환용 진입점. window_mode 가 'headless'/'background'/'hidden' 이면 배경용, 그 외는 보이기용.
"""
if window_mode in ("headless", "background", "hidden"):
return self.make_background_driver(fast_mode=fast_mode)
return self.make_visible_driver(window_mode=window_mode, fast_mode=fast_mode)
def set_input_value(self, driver, elem, value):
"""
일반 send_keys가 안 먹는 경우를 대비해서 JS input 이벤트까지 같이 발생.
"""
try:
elem.click()
elem.clear()
elem.send_keys(value)
time.sleep(0.2)
except Exception:
pass
# 값이 제대로 안 들어갔으면 JS로 강제 세팅
current = elem.get_attribute("value") or ""
if current != value:
driver.execute_script("""
const el = arguments[0];
const val = arguments[1];
el.focus();
el.value = val;
el.dispatchEvent(new Event('input', {bubbles:true}));
el.dispatchEvent(new Event('change', {bubbles:true}));
""", elem, value)
time.sleep(0.2)
def auto_login(self, driver, account: Account, wait_sec=20):
self.log(f"[{account.label}] 자동 로그인 시도 - v5 고정 셀렉터")
driver.get(LOGIN_URL)
wait = WebDriverWait(driver, wait_sec)
# 오너클랜 로그인 HTML 기준 정확한 셀렉터
id_input = wait.until(EC.presence_of_element_located((By.CSS_SELECTOR, "form[name='loginForm'] input#id, form[name='loginForm'] input[name='id']")))
pw_input = wait.until(EC.presence_of_element_located((By.CSS_SELECTOR, "form[name='loginForm'] input#passwd, form[name='loginForm'] input[name='passwd']")))
# 화면 중앙으로 이동
try:
driver.execute_script("arguments[0].scrollIntoView({block:'center'});", id_input)
time.sleep(0.3)
except Exception:
pass
self.set_input_value(driver, id_input, account.user_id)
self.set_input_value(driver, pw_input, account.password)
self.log(f"[{account.label}] 입력값 세팅 완료: id={id_input.get_attribute('value')!r}, pw_len={len(pw_input.get_attribute('value') or '')}")
# 로그인 버튼 클릭
clicked = False
try:
submit_btn = driver.find_element(By.CSS_SELECTOR, "form[name='loginForm'] input.img_log[type='submit'], form[name='loginForm'] input[type='submit']")
driver.execute_script("arguments[0].scrollIntoView({block:'center'});", submit_btn)
time.sleep(0.2)
driver.execute_script("arguments[0].click();", submit_btn)
clicked = True
self.log(f"[{account.label}] 로그인 submit 버튼 클릭")
except Exception as e:
self.log(f"[{account.label}] submit 버튼 클릭 실패, Enter 시도: {e}")
try:
pw_input.send_keys(Keys.ENTER)
clicked = True
except Exception:
pass
# 그래도 안 되면 form submit
if not clicked:
self.log(f"[{account.label}] form.submit() fallback")
driver.execute_script("document.forms['loginForm'].submit();")
time.sleep(1.2 if account.memo_url else 1.5)
# alert 처리
try:
alert = driver.switch_to.alert
msg = alert.text
alert.accept()
raise RuntimeError(f"로그인 알림창: {msg}")
except Exception as e:
if "로그인 알림창" in str(e):
raise
self.log(f"[{account.label}] 로그인 후 URL: {driver.current_url}")
# 로그인 페이지에 그대로 있으면 실패로 판단
if "loginform.php" in driver.current_url.lower():
self.debug_dump(driver, account.label, "login_still_loginform")
raise RuntimeError("로그인 후에도 로그인 페이지에 머물러 있습니다. 아이디/비밀번호 또는 보안처리를 확인해주세요.")
def manual_login(self, driver, account: Account):
self.log(f"[{account.label}] 수동 로그인 모드")
driver.get(LOGIN_URL)
self.ask_manual_login_func(account.label)
time.sleep(1)
self.log(f"[{account.label}] 수동 로그인 확인됨 / 현재 URL: {driver.current_url}")
def find_and_click_memo_menu(self, driver, account: Account):
self.log(f"[{account.label}] 긴급/알림 메모 메뉴 자동 탐색")
candidate_pages = [
"https://ownerclan.com/",
"https://ownerclan.com/V2/",
"https://ownerclan.com/V2/member/mypage.php",
"https://ownerclan.com/V2/service/alert_noti_list.php",
]
menu_texts = [
"긴급/알림 메모",
"긴급 · 알림 메모",
"긴급·알림 메모",
"긴급/알림",
"알림 메모",
"알림/긴급 메세지",
"알림/긴급 메세지 바로 가기",
]
# 직접 후보 URL 먼저 시도
for page in candidate_pages:
if self.stop_event.is_set():
return False
try:
driver.get(page)
time.sleep(0.8)
body_text = driver.find_element(By.TAG_NAME, "body").text
if "알림내용" in body_text and "처리상태" in body_text:
self.log(f"[{account.label}] 메모 페이지 직접 진입 성공: {page}")
return True
except Exception:
pass
for page in candidate_pages[:3]:
if self.stop_event.is_set():
return False
try:
driver.get(page)
time.sleep(2)
for txt in menu_texts:
elems = driver.find_elements(By.XPATH, f"//*[contains(normalize-space(.), '{txt}')]")
for e in elems:
try:
if e.is_displayed():
driver.execute_script("arguments[0].scrollIntoView({block:'center'});", e)
time.sleep(0.3)
e.click()
time.sleep(0.8)
self.log(f"[{account.label}] 메뉴 클릭 성공: {txt}")
return True
except Exception:
continue
except Exception:
continue
return False
def go_memo_page(self, driver, account: Account):
url = (account.memo_url or "").strip()
if url and "loginform.php" not in url:
self.log(f"[{account.label}] 긴급/알림 메모 URL 이동: {url}")
driver.get(url)
time.sleep(0.8)
else:
self.log(f"[{account.label}] 메모 URL 없음 → 메뉴/후보 URL 자동 탐색")
ok = self.find_and_click_memo_menu(driver, account)
if not ok:
raise RuntimeError("긴급/알림 메모 메뉴를 자동으로 찾지 못했습니다. 실제 바로가기 URL을 넣어주세요.")
def collect_new_rows_from_current_page(self, driver, account: Account):
rows_out = []
try:
WebDriverWait(driver, 20).until(lambda d: len(d.find_elements(By.TAG_NAME, "table")) > 0)
except Exception:
self.debug_dump(driver, account.label, "no_table")
raise RuntimeError("현재 페이지에서 표(table)를 못 찾았습니다. debug 폴더 스크린샷을 확인해주세요.")
time.sleep(0.3)
target_table = None
for table in driver.find_elements(By.TAG_NAME, "table"):
txt = table.text
if "처리상태" in txt and "알림내용" in txt:
target_table = table
break
if target_table is None:
self.debug_dump(driver, account.label, "no_alert_table")
raise RuntimeError("표는 찾았지만 긴급/알림 메모 표가 아닙니다. 실제 메모 URL을 넣어주세요.")
trs = target_table.find_elements(By.TAG_NAME, "tr")
if not trs:
return rows_out
header_cells = trs[0].find_elements(By.XPATH, "./th|./td")
headers = [c.text.strip().replace("\n", " ") for c in header_cells]
def idx_of(name, default=None):
for i, h in enumerate(headers):
if name in h:
return i
return default
idx_no = idx_of("NO", 1)
idx_content = idx_of("알림내용", 2)
idx_kind = idx_of("알림종류", 3)
idx_type = idx_of("알림타입", 4)
idx_status = idx_of("처리상태", 5)
idx_registered = idx_of("등록일", 6)
idx_processed = idx_of("처리일", 7)
for tr in trs[1:]:
if self.stop_event.is_set():
break
cells = tr.find_elements(By.XPATH, "./td")
if not cells:
continue
vals = [c.text.strip().replace("\n", " ") for c in cells]
def get(i):
if i is None or i < 0 or i >= len(vals):
return ""
return vals[i]
status = get(idx_status).strip()
if status != "신규":
continue
no = get(idx_no)
content = get(idx_content)
registered = get(idx_registered)
alert_id = f"{account.user_id}|{no}|{registered}|{content[:80]}"
rows_out.append({
"alert_id": alert_id,
"계정명": account.label,
"아이디": account.user_id,
"NO": no,
"알림내용": content,
"알림종류": get(idx_kind),
"알림타입": get(idx_type),
"처리상태": status,
"등록일": registered,
"처리일": get(idx_processed),
"수집시간": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
})
return rows_out
def go_qna_list_page(self, driver, account: Account):
self.log(f"[{account.label}] 문의게시판 URL 이동: {QNA_LIST_URL}")
driver.get(QNA_LIST_URL)
time.sleep(0.8)
def _abs_ownerclan_url(self, href):
href = (href or "").strip()
if not href:
return QNA_LIST_URL
if href.startswith("http://") or href.startswith("https://"):
return href
if href.startswith("/"):
return "https://ownerclan.com" + href
return "https://ownerclan.com/V2/service/" + href
def collect_bold_qna_from_current_page(self, driver, account: Account):
"""
https://ownerclan.com/V2/service/qnaList.php 에서
글제목 칸이 | 제목 형태인 행만 수집한다.
일반 글씨 제목(td.left > a)은 수집하지 않는다.
"""
rows_out = []
try:
WebDriverWait(driver, 20).until(lambda d: len(d.find_elements(By.TAG_NAME, "table")) > 0)
except Exception:
self.debug_dump(driver, account.label, "qna_no_table")
raise RuntimeError("문의게시판 표(table)를 못 찾았습니다.")
time.sleep(0.3)
target_table = None
for table in driver.find_elements(By.TAG_NAME, "table"):
txt = table.text
if "글제목" in txt and "답변여부" in txt and "작성일" in txt:
target_table = table
break
if target_table is None:
self.debug_dump(driver, account.label, "qna_no_list_table")
raise RuntimeError("문의게시판 목록 표를 찾지 못했습니다.")
trs = target_table.find_elements(By.TAG_NAME, "tr")
if not trs:
return rows_out
header_cells = trs[0].find_elements(By.XPATH, "./th|./td")
headers = [c.text.strip().replace("\n", " ") for c in header_cells]
def idx_of(name, default=None):
for i, h in enumerate(headers):
if name in h:
return i
return default
idx_no = idx_of("no", 0)
idx_title = idx_of("글제목", 1)
idx_answer = idx_of("답변여부", 2)
idx_file = idx_of("파일", 3)
idx_registered = idx_of("작성일", 4)
idx_answered = idx_of("답변일", 5)
for tr in trs[1:]:
if self.stop_event.is_set():
break
cells = tr.find_elements(By.XPATH, "./td")
if not cells:
continue
def cell(i):
if i is None or i < 0 or i >= len(cells):
return None
return cells[i]
title_cell = cell(idx_title)
if title_cell is None:
continue
# 핵심 조건: 글제목 칸 안에 b a가 있어야 굵은 글제목으로 판단
bold_links = title_cell.find_elements(By.CSS_SELECTOR, "b a[href]")
if not bold_links:
continue
title_link = bold_links[0]
title = (title_link.text or title_cell.text or "").strip().replace("\n", " ")
href = title_link.get_attribute("href") or title_link.get_attribute("getAttribute('href')") or ""
detail_url = self._abs_ownerclan_url(href)
vals = [c.text.strip().replace("\n", " ") for c in cells]
def get(i):
if i is None or i < 0 or i >= len(vals):
return ""
return vals[i]
no = get(idx_no)
registered = get(idx_registered)
answer_status = get(idx_answer)
alert_id = f"qna|{account.user_id}|{no}|{registered}|{title[:80]}"
rows_out.append({
"source_type": "qna_bold",
"detail_url": detail_url,
"alert_id": alert_id,
"계정명": account.label,
"아이디": account.user_id,
"NO": no,
"알림내용": title,
"알림종류": "문의게시판",
"알림타입": "굵은글제목",
"처리상태": answer_status or "굵은글제목",
"등록일": registered,
"처리일": get(idx_answered),
"파일": get(idx_file),
"수집시간": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
})
return rows_out
def open_qna_visible(self, account: Account, alert_row: dict, login_mode="auto", fast_mode=True):
"""
문의게시판 굵은 글제목 카드의 '열기' 버튼:
보이는 크롬창으로 로그인 후 해당 문의 상세글(qnaView.php)까지 바로 연다.
사용자가 직접 볼 수 있도록 driver.quit() 하지 않고 반환한다.
"""
driver = self.make_driver(window_mode="normal", fast_mode=fast_mode)
try:
if login_mode == "manual":
self.manual_login(driver, account)
else:
self.auto_login(driver, account)
url = (alert_row.get("detail_url") or "").strip() or QNA_LIST_URL
driver.get(url)
time.sleep(1.2)
self.log(f"[{account.label}] 문의글 열기 완료: NO.{alert_row.get('NO','')}")
return driver
except Exception:
self.debug_dump(driver, account.label, "open_qna_error")
self.log(f"[{account.label}] 문의글 열기 중 오류, 현재 창은 유지합니다.")
return driver
def confirm_qna_on_site(self, account: Account, alert_row: dict, window_mode="headless", login_mode="auto", fast_mode=True):
"""
문의게시판 굵은 글제목 카드의 '확인' 버튼:
해당 문의 상세글을 한 번 열어 읽음 처리/볼드 해제를 유도한다.
성공하면 프로그램 알림목록에서는 seen 처리로 사라진다.
"""
driver = None
try:
driver = self.make_driver(window_mode=window_mode, fast_mode=fast_mode)
if login_mode == "manual":
self.manual_login(driver, account)
else:
self.auto_login(driver, account)
url = (alert_row.get("detail_url") or "").strip()
if not url:
# detail_url이 없으면 목록에서 NO로 다시 찾아 클릭
self.go_qna_list_page(driver, account)
target_no = str(alert_row.get("NO", "")).strip()
found = False
for tr in driver.find_elements(By.CSS_SELECTOR, "tr"):
cells = tr.find_elements(By.XPATH, "./td")
if not cells:
continue
vals = [c.text.strip() for c in cells]
if vals and vals[0] == target_no:
links = tr.find_elements(By.CSS_SELECTOR, "td.left b a[href], td.left a[href]")
if links:
driver.execute_script("arguments[0].scrollIntoView({block:'center'});", links[0])
time.sleep(0.2)
try:
links[0].click()
except Exception:
driver.execute_script("arguments[0].click();", links[0])
found = True
break
if not found:
# 목록에 해당 글이 없으면 이미 읽음/처리된 것으로 보고 프로그램에서도 확인 처리.
self.log(f"[{account.label}] 문의게시판 목록에 NO.{alert_row.get('NO','')} 없음 → 이미 처리됨으로 확인 처리")
return True
else:
driver.get(url)
time.sleep(1.5)
# 상세 페이지 로딩 확인: qnaView.php 또는 본문 존재 정도만 확인
try:
WebDriverWait(driver, 10).until(lambda d: "qnaView" in d.current_url or len(d.find_elements(By.TAG_NAME, "table")) > 0)
except Exception:
pass
self.log(f"[{account.label}] 문의글 확인 처리 완료: NO.{alert_row.get('NO','')}")
return True
except Exception:
if driver:
self.debug_dump(driver, account.label, "confirm_qna_error")
raise
finally:
if driver:
quit_driver(driver, log=self.log)
def open_memo_visible(self, account: Account, login_mode="auto", fast_mode=True):
"""
카드의 '열기' 버튼을 눌렀을 때:
해당 계정으로 보이는 크롬창을 열고 로그인 후 알림/긴급 메모 페이지까지 이동.
사용자가 직접 볼 수 있도록 driver.quit() 하지 않고 반환한다.
"""
driver = self.make_driver(window_mode="normal", fast_mode=fast_mode)
try:
if login_mode == "manual":
self.manual_login(driver, account)
else:
self.auto_login(driver, account)
driver.get(DEFAULT_MEMO_URL)
time.sleep(1.5)
self.log(f"[{account.label}] 알림 페이지 열기 완료")
return driver
except Exception:
self.debug_dump(driver, account.label, "open_memo_error")
self.log(f"[{account.label}] 알림 페이지 열기 중 오류, 현재 창은 유지합니다.")
return driver
def confirm_alert_on_site(self, account: Account, alert_row: dict, window_mode="headless", login_mode="auto", fast_mode=True):
"""
프로그램 카드의 '확인' 버튼을 눌렀을 때 실제 오너클랜 사이트에서
해당 행의 '클릭 후 처리' 버튼까지 눌러 처리완료로 만드는 기능.
"""
driver = None
try:
driver = self.make_driver(window_mode=window_mode, fast_mode=fast_mode)
if login_mode == "manual":
self.manual_login(driver, account)
else:
self.auto_login(driver, account)
self.go_memo_page(driver, account)
WebDriverWait(driver, 20).until(lambda d: len(d.find_elements(By.TAG_NAME, "table")) > 0)
time.sleep(0.5)
target_no = str(alert_row.get("NO", "")).strip()
target_content = str(alert_row.get("알림내용", "")).strip()
target_registered = str(alert_row.get("등록일", "")).strip()
tables = driver.find_elements(By.TAG_NAME, "table")
target_table = None
for table in tables:
txt = table.text
if "처리상태" in txt and "알림내용" in txt:
target_table = table
break
if target_table is None:
self.debug_dump(driver, account.label, "confirm_no_table")
raise RuntimeError("확인 처리할 알림 표를 찾지 못했습니다.")
trs = target_table.find_elements(By.TAG_NAME, "tr")
if not trs:
raise RuntimeError("확인 처리할 행이 없습니다.")
header_cells = trs[0].find_elements(By.XPATH, "./th|./td")
headers = [c.text.strip().replace("\n", " ") for c in header_cells]
def idx_of(name, default=None):
for i, h in enumerate(headers):
if name in h:
return i
return default
idx_no = idx_of("NO", 1)
idx_content = idx_of("알림내용", 2)
idx_status = idx_of("처리상태", 5)
idx_registered = idx_of("등록일", 6)
idx_processed = idx_of("처리일", 7)
best_tr = None
for tr in trs[1:]:
cells = tr.find_elements(By.XPATH, "./td")
if not cells:
continue
vals = [c.text.strip().replace("\n", " ") for c in cells]
def get(i):
if i is None or i < 0 or i >= len(vals):
return ""
return vals[i]
no_match = (not target_no) or get(idx_no).strip() == target_no
status_match = "신규" in get(idx_status).strip()
# 등록일/내용은 사이트 줄바꿈 때문에 완전일치보다 느슨하게 비교
reg_text = get(idx_registered).replace(" ", "")
target_reg = target_registered.replace(" ", "")
reg_match = (not target_reg) or target_reg in reg_text or reg_text in target_reg
row_content = get(idx_content)
content_match = True
if target_content:
key = target_content[:25].strip()
content_match = key in row_content or row_content[:25].strip() in target_content
if no_match and status_match and (reg_match or content_match):
best_tr = tr
break
if best_tr is None:
# 신규 행을 못 찾았다고 무조건 오류 처리하면, 오너클랜에서 이미 처리된 알림을
# 프로그램 리스트에서 못 지우는 문제가 생긴다. NO(고유키) 기준으로 현재 상태를 재분류한다.
if not target_no:
# NO가 없으면 안전하게 식별 불가 → 기존대로 오류 처리.
self.debug_dump(driver, account.label, "confirm_row_not_found")
raise RuntimeError("해당 알림 행을 찾지 못했고 NO도 없어 식별할 수 없습니다.")
no_status = None # None=목록에 NO 자체가 없음
for tr in trs[1:]:
cells = tr.find_elements(By.XPATH, "./td")
if not cells:
continue
vals = [c.text.strip().replace("\n", " ") for c in cells]
def getv(i):
if i is None or i < 0 or i >= len(vals):
return ""
return vals[i]
if getv(idx_no).strip() == target_no:
no_status = getv(idx_status).strip()
break
if no_status is not None and "신규" in no_status:
# 아직 '신규'인데 1차 보조매칭(내용/등록일)만 실패 → 그 행을 채택해 정상 처리 진행(숨기지 않음).
self.log(f"[{account.label}] NO.{target_no} 보조매칭 실패했지만 신규 상태 → 해당 행으로 처리 진행")
best_tr = tr
else:
# 비-신규(이미 처리완료) 또는 목록에서 사라짐 → 이미 처리됨으로 간주, 프로그램에서도 확인 처리.
state_desc = f"상태 '{no_status}'" if no_status is not None else "목록에 없음"
self.log(f"[{account.label}] NO.{target_no} {state_desc} → 이미 처리됨으로 확인 처리")
return True
# 해당 행 안에서 '클릭 후 처리' 버튼/링크 찾기
# v12는 아무 input/button까지 넓게 클릭해서 체크박스를 누를 가능성이 있었음.
# v13은 텍스트/value/title/onclick에 '클릭 후 처리'가 있는 요소만 클릭.
buttons = best_tr.find_elements(
By.XPATH,
".//*[contains(normalize-space(.), '클릭 후 처리') "
"or contains(@value, '클릭 후 처리') "
"or contains(@title, '클릭 후 처리') "
"or contains(@onclick, 'click') "
"or contains(@onclick, 'Click') "
"or contains(@onclick, 'process') "
"or contains(@onclick, 'Process')]"
)
# 처리일 칸 쪽에 있는 실제 버튼 우선
exact_buttons = []
for btn in buttons:
try:
txt = (btn.text or btn.get_attribute("value") or btn.get_attribute("title") or "").strip()
onclick = (btn.get_attribute("onclick") or "").strip()
tag = btn.tag_name.lower()
if "클릭 후 처리" in txt or "click" in onclick.lower() or "process" in onclick.lower():
if tag in ("input", "button", "a", "span", "div"):
exact_buttons.append(btn)
except Exception:
continue
clicked = False
click_error = None
for btn in exact_buttons:
try:
driver.execute_script("arguments[0].scrollIntoView({block:'center'});", btn)
time.sleep(0.2)
# disabled/안보임 방지
try:
if not btn.is_enabled():
continue
except Exception:
pass
# 일반 클릭 먼저
try:
btn.click()
except Exception:
driver.execute_script("arguments[0].click();", btn)
clicked = True
break
except Exception as e:
click_error = e
continue
if not clicked:
self.debug_dump(driver, account.label, "confirm_button_not_found")
raise RuntimeError(f"'클릭 후 처리' 버튼을 정확히 찾거나 클릭하지 못했습니다. {click_error or ''}")
time.sleep(1)
# confirm/alert 확인
try:
alert = driver.switch_to.alert
msg = alert.text
alert.accept()
self.log(f"[{account.label}] 확인 처리 알림창: {msg}")
time.sleep(1)
except Exception:
pass
# 처리 후 페이지 갱신 대기/검증
time.sleep(1)
try:
driver.refresh()
time.sleep(1.2)
except Exception:
pass
# 같은 NO가 아직 신규인지 재확인
still_new = False
try:
tables2 = driver.find_elements(By.TAG_NAME, "table")
for table2 in tables2:
txt2 = table2.text
if "처리상태" not in txt2 or "알림내용" not in txt2:
continue
for tr2 in table2.find_elements(By.TAG_NAME, "tr")[1:]:
cells2 = tr2.find_elements(By.XPATH, "./td")
vals2 = [c.text.strip().replace("\n", " ") for c in cells2]
def get2(i):
if i is None or i < 0 or i >= len(vals2):
return ""
return vals2[i]
if get2(idx_no).strip() == target_no and "신규" in get2(idx_status).strip():
still_new = True
break
if still_new:
break
except Exception:
# 검증 실패는 치명 오류로 보지는 않음
pass
if still_new:
self.debug_dump(driver, account.label, "confirm_still_new")
raise RuntimeError("클릭 후 처리 버튼은 눌렀지만, 새로고침 후에도 상태가 '신규'로 남아 있습니다.")
self.log(f"[{account.label}] 사이트 확인 처리 완료: NO.{target_no}")
return True
except Exception:
if driver:
self.debug_dump(driver, account.label, "confirm_error")
raise
finally:
if driver:
quit_driver(driver, log=self.log)
def collect_account(self, account: Account, window_mode="normal", login_mode="auto", fast_mode=True):
driver = None
try:
driver = self.make_driver(window_mode=window_mode, fast_mode=fast_mode)
if login_mode == "manual":
self.manual_login(driver, account)
else:
self.auto_login(driver, account)
if self.stop_event.is_set():
return []
self.go_memo_page(driver, account)
if self.stop_event.is_set():
return []
rows = []
# 1) 기존 긴급/알림 메모 신규 수집
try:
memo_rows = self.collect_new_rows_from_current_page(driver, account)
rows.extend(memo_rows)
self.log(f"[{account.label}] 긴급/알림 신규 {len(memo_rows)}건 발견")
except Exception as e:
self.log(f"[{account.label}] 긴급/알림 수집 오류: {e}")
# 2) 문의게시판(qnaList.php) 굵은 글제목 수집
if not self.stop_event.is_set():
try:
self.go_qna_list_page(driver, account)
qna_rows = self.collect_bold_qna_from_current_page(driver, account)
rows.extend(qna_rows)
self.log(f"[{account.label}] 문의게시판 굵은 글제목 {len(qna_rows)}건 발견")
except Exception as e:
self.log(f"[{account.label}] 문의게시판 굵은 글제목 수집 오류: {e}")
self.log(f"[{account.label}] 전체 신규/문의 {len(rows)}건 발견")
return rows
except Exception:
if driver:
self.debug_dump(driver, account.label, "error")
raise
finally:
if driver:
quit_driver(driver, log=self.log)
class ScrollableFrame(ttk.Frame):
def __init__(self, parent):
super().__init__(parent)
self.canvas = tk.Canvas(self, highlightthickness=0, bg="#f6f7fb")
self.scrollbar = ttk.Scrollbar(self, orient="vertical", command=self.canvas.yview)
self.inner = tk.Frame(self.canvas, bg="#f6f7fb")
self.inner.bind("", lambda e: self.update_scrollregion())
self.window_id = self.canvas.create_window((0, 0), window=self.inner, anchor="nw")
self.canvas.configure(yscrollcommand=self.scrollbar.set)
self.canvas.pack(side="left", fill="both", expand=True)
self.scrollbar.pack(side="right", fill="y")
self.canvas.bind("", self.on_canvas_configure)
self.canvas.bind_all("", self.on_mousewheel)
def update_scrollregion(self):
"""
스크롤 영역을 '내용 높이와 화면 높이 중 큰 값' 으로 잡습니다.
Tk 캔버스는 스크롤 영역이 화면보다 작으면 휠을 굴릴 때 내용이 아래로 밀려 위에 빈 공간이 생기므로,
영역을 최소한 화면 크기로 유지해 그 현상을 막습니다.
"""
try:
self.canvas.update_idletasks()
bbox = self.canvas.bbox("all")
content_h = (bbox[3] - bbox[1]) if bbox else 0
content_w = (bbox[2] - bbox[0]) if bbox else 0
h = max(content_h, self.canvas.winfo_height())
w = max(content_w, self.canvas.winfo_width())
self.canvas.configure(scrollregion=(0, 0, w, h))
# 내용이 화면보다 작아졌으면 항상 맨 위로
if content_h <= self.canvas.winfo_height():
self.canvas.yview_moveto(0.0)
except Exception:
pass
def content_fits(self):
try:
bbox = self.canvas.bbox("all")
content_h = (bbox[3] - bbox[1]) if bbox else 0
return content_h <= self.canvas.winfo_height()
except Exception:
return False
def on_canvas_configure(self, event):
self.canvas.itemconfigure(self.window_id, width=event.width)
self.update_scrollregion()
def on_mousewheel(self, event):
try:
if self.content_fits():
self.canvas.yview_moveto(0.0)
return
self.canvas.yview_scroll(int(-1 * (event.delta / 120)), "units")
except Exception:
pass
# =========================
# 주문/CS/문자 생성기 탭 v4
# =========================
ORDER_CS_TEMPLATE_FILE = APP_DIR / "order_cs_sms_templates.json"
ORDER_CS_MARKETS = [
"네이버 스마트스토어",
"11번가",
"지마켓",
"옥션",
"쿠팡",
"롯데온",
"인터파크",
"기타 직접입력",
]
ORDER_CS_SEND_TIMES = ["낮", "밤"]
ORDER_CS_SEND_DAYS = ["오늘", "내일", "모레", "직접입력"]
ORDER_CS_DEFAULT_TEMPLATES = {
"주문취소": (
"{고객명} 고객님 안녕하세요. {마켓명} 판매 담당자입니다.\n"
"{밤문구}"
"주문해주신 {상품명} 상품이 품절되어 문자드립니다.\n"
"해당 상품은 재입고 일정이 미정이라 부득이하게 주문취소로 진행해야 할 것 같습니다.\n"
"저희 쪽에서 주문취소 진행해놓겠습니다.\n"
"불편을 드려 죄송합니다."
),
"반품": (
"{고객명} 고객님 안녕하세요. {마켓명} 판매 담당자입니다.\n"
"{밤문구}"
"주문해주신 {상품명} 상품 반품신청 확인하였습니다.\n"
"저희가 택배사에 수거신청 해놓겠습니다.\n"
"반품 상품은 받으신 그대로 포장하셔서 택배기사님께 전달 부탁드립니다.\n"
"문 앞에 두시면 택배기사님이 수거해가실 예정입니다.\n"
"감사합니다."
),
"교환": (
"{고객명} 고객님 안녕하세요. {마켓명} 판매 담당자입니다.\n"
"{밤문구}"
"주문해주신 {상품명} 상품 교환 요청 확인하여 문자드립니다.\n"
"저희가 택배사에 수거신청 해놓겠습니다.\n"
"수거 상품은 받으신 그대로 포장하셔서 택배기사님께 전달 부탁드립니다.\n"
"문 앞에 두시면 택배기사님이 수거해가실 예정입니다.\n"
"불편을 드려 죄송합니다."
),
}
ORDER_CS_PLACEHOLDER_BUTTONS = ["{고객명}", "{마켓명}", "{계정명}", "{상품명}", "{날짜}"]
def bind_text_undo_shortcuts(text_widget: tk.Text):
"""Tk Text 위젯에서 Ctrl+Z / Ctrl+Y가 안정적으로 동작하게 합니다."""
def undo(_event=None):
try:
text_widget.edit_undo()
except tk.TclError:
pass
return "break"
def redo(_event=None):
try:
text_widget.edit_redo()
except tk.TclError:
pass
return "break"
text_widget.bind("", undo)
text_widget.bind("", undo)
text_widget.bind("", redo)
text_widget.bind("", redo)
class TemplateEditDialog(tk.Toplevel):
def __init__(self, master, template_name: str, template_text: str, on_save):
super().__init__(master)
self.template_name = template_name
self.on_save = on_save
self.title(f"양식 수정 - {template_name}")
self.geometry("780x620")
self.minsize(640, 480)
self.transient(master.winfo_toplevel())
self.grab_set()
self.protocol("WM_DELETE_WINDOW", self._close)
root = ttk.Frame(self, padding=12)
root.pack(fill="both", expand=True)
header = ttk.Frame(root)
header.pack(fill="x", pady=(0, 8))
ttk.Label(header, text="양식명:", font=("맑은 고딕", 13, "bold")).pack(side="left")
self.name_var = tk.StringVar(value=template_name)
self.name_entry = ttk.Entry(header, textvariable=self.name_var, font=("맑은 고딕", 12), width=24)
self.name_entry.pack(side="left", padx=(6, 0))
ttk.Label(header, text="(이름을 바꾼 뒤 저장하면 양식명이 변경됩니다)", foreground="#888888").pack(side="left", padx=(8, 0))
btn_row = ttk.Frame(root)
btn_row.pack(fill="x", pady=(0, 8))
ttk.Button(btn_row, text="저장", command=self._save).pack(side="left")
ttk.Button(btn_row, text="저장 후 닫기", command=self._save_and_close).pack(side="left", padx=6)
ttk.Button(btn_row, text="닫기", command=self._close).pack(side="left")
# 삽입 버튼 = 기본 치환값 + 사용자 정의 추가 치환값(있으면)
try:
ph_keys = master.placeholder_button_keys()
except Exception:
ph_keys = list(ORDER_CS_PLACEHOLDER_BUTTONS)
ph_row = ttk.LabelFrame(root, text="자동 치환값 삽입", padding=8)
ph_row.pack(fill="x", pady=(0, 8))
ttk.Label(ph_row, text="커서 위치에 삽입됩니다:").pack(side="left", padx=(0, 6))
for ph in ph_keys:
ttk.Button(ph_row, text=ph, command=lambda p=ph: self._insert_placeholder(p)).pack(side="left", padx=2)
help_text = (
"사용 가능 값: " + ", ".join(ph_keys) + "\n"
"※ {날짜}는 문자 탭에서 직접 입력한 값이 들어갑니다. 비워두면 오늘 날짜가 자동으로 들어갑니다.\n"
"※ 추가 치환값은 '주문/CS/문자' 화면의 '추가 치환값'에서 +추가/✕삭제 할 수 있습니다."
)
ttk.Label(root, text=help_text, foreground="#555555").pack(anchor="w", pady=(0, 6))
body = ttk.Frame(root)
body.pack(fill="both", expand=True)
self.text = tk.Text(body, wrap="word", font=("맑은 고딕", 11), undo=True, maxundo=-1, autoseparators=True)
self.text.pack(side="left", fill="both", expand=True)
scroll = ttk.Scrollbar(body, orient="vertical", command=self.text.yview)
scroll.pack(side="right", fill="y")
self.text.configure(yscrollcommand=scroll.set)
self.text.insert("1.0", template_text)
self.text.edit_reset()
bind_text_undo_shortcuts(self.text)
self.text.focus_set()
def _insert_placeholder(self, placeholder):
self.text.insert("insert", placeholder)
self.text.focus_set()
def _get_text(self):
return self.text.get("1.0", "end").strip()
def _save(self):
new_name = self.name_var.get().strip()
content = self._get_text()
if not new_name:
messagebox.showwarning("저장 불가", "양식명을 입력해주세요.", parent=self)
return False
if not content:
messagebox.showwarning("저장 불가", "양식 내용이 비어 있습니다.", parent=self)
return False
# on_save 는 성공 시 None, 실패(이름 중복 등) 시 오류 메시지(문자열) 반환
err = self.on_save(self.template_name, new_name, content)
if err:
messagebox.showwarning("저장 불가", err, parent=self)
return False
self.template_name = new_name # 이후 저장은 새 이름 기준
self.title(f"양식 수정 - {new_name}")
messagebox.showinfo("저장 완료", f"'{new_name}' 양식을 저장했습니다.", parent=self)
return True
def _save_and_close(self):
if self._save():
self.destroy()
def _close(self):
self.destroy()
class OrderCsSmsFrame(ttk.Frame):
def __init__(self, master, settings=None, save_settings=None, account_labels_provider=None):
super().__init__(master, padding=18)
self.templates = self._load_templates()
# 알림 수집기 탭과 공유하는 설정/계정 연동
self._settings = settings if settings is not None else {}
self._save_settings_cb = save_settings
self._account_labels_provider = account_labels_provider
self._callbacks = {}
# 사용자 정의 추가 치환값(항목명 목록) → 각각 문자 탭에 입력칸이 생기고 {항목명}으로 치환
self.custom_fields = [str(n).strip() for n in self._settings.get("order_cs_custom_fields", []) if str(n).strip()]
self.custom_field_vars = {}
self.customer_name_var = tk.StringVar()
# 저장된 선택 복원 (없으면 기본값). 저장된 마켓이 목록에 없으면 기본값으로 폴백.
saved_market = self._settings.get("order_cs_selected_market", ORDER_CS_MARKETS[0])
if saved_market not in ORDER_CS_MARKETS:
saved_market = ORDER_CS_MARKETS[0]
self.market_var = tk.StringVar(value=saved_market)
self.custom_market_var = tk.StringVar(value=self._settings.get("order_cs_custom_market", ""))
self.account_var = tk.StringVar(value=self._settings.get("order_cs_selected_account", ""))
self.product_name_var = tk.StringVar()
self.date_var = tk.StringVar(value=datetime.now().strftime("%Y/%m/%d"))
self.cs_type_var = tk.StringVar(value=self._first_template_name())
self.send_time_var = tk.StringVar(value="낮")
self.exchange_day_var = tk.StringVar(value="내일")
self.custom_exchange_day_var = tk.StringVar()
self._build_ui()
self._refresh_dynamic_fields()
self._generate_message()
def set_callback(self, name, func):
"""App 이 '업데이트 확인' / '재시작' 동작을 넣어줍니다."""
self._callbacks[name] = func
def _call_cb(self, name):
cb = self._callbacks.get(name)
if cb:
cb()
def _load_templates(self):
if ORDER_CS_TEMPLATE_FILE.exists():
try:
data = json.loads(ORDER_CS_TEMPLATE_FILE.read_text(encoding="utf-8"))
if isinstance(data, dict) and data:
return {str(k): str(v) for k, v in data.items()}
except Exception:
pass
self._save_templates(ORDER_CS_DEFAULT_TEMPLATES)
return dict(ORDER_CS_DEFAULT_TEMPLATES)
def _save_templates(self, templates=None):
data = templates if templates is not None else self.templates
ORDER_CS_TEMPLATE_FILE.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")
def _first_template_name(self):
return next(iter(self.templates.keys())) if self.templates else "주문취소"
def _build_ui(self):
title_row = ttk.Frame(self)
title_row.pack(fill="x", pady=(0, 2))
ttk.Label(title_row, text="주문/CS/문자 생성기", font=("맑은 고딕", 16, "bold")).pack(side="left")
# 우측 상단: 업데이트 확인 / 재시작 (App 이 콜백을 넣어줌)
self.update_btn_bar = tk.Frame(title_row)
self.update_btn_bar.pack(side="right")
self.restart_btn = tk.Button(
self.update_btn_bar, text="재시작", bg="#6f42c1", fg="white", relief="flat",
padx=12, pady=4, font=("맑은 고딕", 9, "bold"),
activebackground="#59359a", activeforeground="white",
command=lambda: self._call_cb("on_restart"),
)
self.restart_btn.pack(side="right", padx=(6, 0))
self.update_check_top_btn = tk.Button(
self.update_btn_bar, text="업데이트 확인", bg="#2f80ed", fg="white", relief="flat",
padx=12, pady=4, font=("맑은 고딕", 9, "bold"),
activebackground="#1f66c9", activeforeground="white",
command=lambda: self._call_cb("on_check_update"),
)
self.update_check_top_btn.pack(side="right")
form = ttk.LabelFrame(self, text="", padding=12)
form.pack(fill="x")
# 고정 창 크기에 맞춘 정렬형 4열 레이아웃
form.columnconfigure(0, weight=0, minsize=72)
form.columnconfigure(1, weight=1, minsize=210)
form.columnconfigure(2, weight=0, minsize=52)
form.columnconfigure(3, weight=0, minsize=130)
ttk.Label(form, text="고객성함").grid(row=0, column=0, sticky="w", padx=(0, 8), pady=5)
self.customer_entry = tk.Entry(form, textvariable=self.customer_name_var, width=18, font=("맑은 고딕", 10, "bold"), fg="#C00000")
self.customer_entry.grid(row=0, column=1, sticky="ew", padx=(0, 14), pady=5, ipady=2)
ttk.Label(form, text="마켓명").grid(row=0, column=2, sticky="e", padx=(0, 6), pady=5)
self.market_combo = ttk.Combobox(form, textvariable=self.market_var, values=ORDER_CS_MARKETS, state="readonly", width=16)
self.market_combo.grid(row=0, column=3, sticky="ew", pady=5)
self.market_combo.bind("<>", lambda _e: self._refresh_dynamic_fields())
ttk.Label(form, text="상품명").grid(row=1, column=0, sticky="w", padx=(0, 8), pady=5)
self.product_entry = tk.Entry(form, textvariable=self.product_name_var, width=28, font=("맑은 고딕", 10, "bold"), fg="#C00000")
self.product_entry.grid(row=1, column=1, sticky="ew", padx=(0, 14), pady=5, ipady=2)
ttk.Label(form, text="날짜").grid(row=1, column=2, sticky="e", padx=(0, 6), pady=5)
self.date_entry = tk.Entry(form, textvariable=self.date_var, width=12, font=("맑은 고딕", 10, "bold"), fg="#C00000")
self.date_entry.grid(row=1, column=3, sticky="ew", pady=5, ipady=2)
ttk.Label(form, text="계정명").grid(row=2, column=0, sticky="w", padx=(0, 8), pady=5)
acc_box = ttk.Frame(form)
acc_box.grid(row=2, column=1, columnspan=3, sticky="ew", pady=5)
acc_box.columnconfigure(0, weight=1)
self.account_combo = ttk.Combobox(acc_box, textvariable=self.account_var, values=[""], state="readonly")
self.account_combo.grid(row=0, column=0, sticky="ew")
ttk.Button(acc_box, text="선택 저장", command=self._save_selection_state, width=10).grid(row=0, column=1, padx=(8, 0))
ttk.Label(form, text="CS유형/양식").grid(row=4, column=0, sticky="w", padx=(0, 8), pady=5)
self.cs_combo = ttk.Combobox(form, textvariable=self.cs_type_var, values=list(self.templates.keys()), state="readonly", width=18)
self.cs_combo.grid(row=4, column=1, sticky="w", padx=(0, 14), pady=5)
self.cs_combo.bind("<>", lambda _e: self._on_template_selected())
self.custom_market_label = ttk.Label(form, text="직접입력 마켓명")
self.custom_market_entry = ttk.Entry(form, textvariable=self.custom_market_var)
for var in [
self.customer_name_var,
self.market_var,
self.custom_market_var,
self.account_var,
self.product_name_var,
self.date_var,
self.cs_type_var,
]:
var.trace_add("write", lambda *_: self._generate_message())
# 사용자 정의 추가 치환값 입력칸을 '입력 정보' 박스(form) 맨 아래 행에 넣는다.
self.custom_fields_frame = ttk.Frame(form)
self.custom_fields_frame.grid(row=6, column=0, columnspan=4, sticky="ew", pady=(6, 0))
self._rebuild_custom_fields()
# 사용법/양식 관리는 맨 아래(side=bottom). 미리보기보다 먼저 패킹해야 미리보기가 가운데를 확장.
tip = ttk.Label(
self,
text="사용법: 고객성함/상품명 입력 → CS유형 선택 → 필요 시 미리보기 수정 → 복사하기",
foreground="#555555",
)
tip.pack(side="bottom", anchor="w", pady=(6, 0))
template_manage = ttk.LabelFrame(self, text="", padding=10)
template_manage.pack(side="bottom", fill="x", pady=(8, 0))
for i in range(5):
template_manage.columnconfigure(i, weight=1)
ttk.Button(template_manage, text="선택 양식 수정", command=self._open_template_editor).grid(row=0, column=0, sticky="ew", padx=(0, 4), pady=2)
ttk.Button(template_manage, text="새 양식 추가", command=self._add_template).grid(row=0, column=1, sticky="ew", padx=4, pady=2)
ttk.Button(template_manage, text="양식 삭제", command=self._delete_template).grid(row=0, column=2, sticky="ew", padx=4, pady=2)
ttk.Button(template_manage, text="양식업로드", command=self._upload_templates_csv).grid(row=0, column=3, sticky="ew", padx=4, pady=2)
ttk.Button(template_manage, text="양식다운로드", command=self._download_templates_csv).grid(row=0, column=4, sticky="ew", padx=(4, 0), pady=2)
# 문자 미리보기 — 가운데를 가득 채워 더 많은 줄이 보이게(가로폭은 width 작게+expand로 균형)
preview_frame = ttk.LabelFrame(self, text="", padding=10)
preview_frame.pack(fill="both", expand=True)
btn_row = ttk.Frame(preview_frame)
btn_row.pack(fill="x", pady=(0, 8))
ttk.Button(btn_row, text="복사하기", command=self._copy_message, width=10).pack(side="left")
ttk.Button(btn_row, text="입력 초기화", command=self._clear_inputs, width=10).pack(side="left", padx=(8, 0))
body = ttk.Frame(preview_frame)
body.pack(fill="both", expand=True)
self.preview_text = tk.Text(body, width=10, height=12, wrap="word", font=("맑은 고딕", 11), undo=True, maxundo=-1)
self.preview_text.pack(side="left", fill="both", expand=True)
self.preview_text.tag_configure("input_value", foreground="#C00000", font=("맑은 고딕", 11, "bold"))
self.preview_text.tag_configure("night_value", foreground="#8B0000", font=("맑은 고딕", 11, "bold"))
preview_scroll = ttk.Scrollbar(body, orient="vertical", command=self.preview_text.yview)
preview_scroll.pack(side="right", fill="y")
self.preview_text.configure(yscrollcommand=preview_scroll.set)
bind_text_undo_shortcuts(self.preview_text)
self.customer_entry.focus_set()
def _on_template_selected(self):
self._refresh_dynamic_fields()
self._generate_message()
def _refresh_template_combo(self):
names = list(self.templates.keys())
self.cs_combo.configure(values=names)
if self.cs_type_var.get() not in self.templates and names:
self.cs_type_var.set(names[0])
self._refresh_dynamic_fields()
self._generate_message()
def _refresh_dynamic_fields(self):
for widget in [
self.custom_market_label,
self.custom_market_entry,
]:
widget.grid_remove()
row = 3
if self.market_var.get() == "기타 직접입력":
self.custom_market_label.grid(row=row, column=0, sticky="w", padx=(0, 8), pady=5)
self.custom_market_entry.grid(row=row, column=1, columnspan=3, sticky="ew", padx=(0, 0), pady=5)
row += 1
self._generate_message()
def _current_template_text(self):
return self.templates.get(self.cs_type_var.get(), "")
def _market_name(self):
# 빈 값이면 "선택한 마켓" 같은 필드명을 넣지 않고 공백으로 둠(자연스러운 문자)
if self.market_var.get() == "기타 직접입력":
return self.custom_market_var.get().strip()
return self.market_var.get().strip()
def _customer_name(self):
return self.customer_name_var.get().strip()
def _account_name(self):
return self.account_var.get().strip()
def _product_name(self):
return self.product_name_var.get().strip()
def _night_sentence(self):
return "늦은시간에 문자 드려서 죄송합니다.\n"
def _date_text(self):
return self.date_var.get().strip() or datetime.now().strftime("%Y/%m/%d")
def _render_template(self):
template = self._current_template_text()
replacements = {
"{고객명}": self._customer_name(),
"{마켓명}": self._market_name(),
"{계정명}": self._account_name(),
"{상품명}": self._product_name(),
"{밤문구}": self._night_sentence(),
"{날짜}": self._date_text(),
"{새상품발송일}": "",
}
# 사용자 정의 추가 치환값
for name, var in self.custom_field_vars.items():
replacements[f"{{{name}}}"] = var.get().strip()
message = template
for key, value in replacements.items():
message = message.replace(key, value)
# 빈 치환값으로 생긴 이중 공백/줄앞 공백 제거 (단어 사이 1칸만 유지, 줄바꿈 보존)
message = "\n".join(" ".join(line.split()) for line in message.split("\n"))
message = message.replace("\n\n\n", "\n\n")
return message.strip()
def _generate_message(self):
if not hasattr(self, "preview_text"):
return
message = self._render_template()
self.preview_text.configure(state="normal")
self.preview_text.delete("1.0", "end")
self.preview_text.insert("1.0", message)
self._highlight_input_values()
self.preview_text.edit_reset()
def _highlight_text(self, target: str, tag_name: str):
if not target:
return
start = "1.0"
while True:
idx = self.preview_text.search(target, start, stopindex="end")
if not idx:
break
end = f"{idx}+{len(target)}c"
self.preview_text.tag_add(tag_name, idx, end)
start = end
def _highlight_input_values(self):
self.preview_text.tag_remove("input_value", "1.0", "end")
self.preview_text.tag_remove("night_value", "1.0", "end")
values = [self._customer_name(), self._market_name(), self._account_name(),
self._product_name(), self._date_text()]
values += [v.get().strip() for v in self.custom_field_vars.values()]
for value in values:
if value:
self._highlight_text(value, "input_value")
night = self._night_sentence().strip()
if night:
self._highlight_text(night, "night_value")
def refresh_accounts(self):
"""알림 수집기 탭의 등록 계정 목록을 계정명 콤보에 동기화."""
labels = list(self._account_labels_provider()) if self._account_labels_provider else []
self.account_combo.configure(values=[""] + labels) # 첫 항목 = (선택 안 함)
cur = self.account_var.get()
if cur and cur not in labels: # 삭제된 계정이 선택돼 있으면 비움
self.account_var.set("")
self._generate_message()
def _save_selection_state(self):
"""계정명/마켓명 선택을 settings.json 에 저장 → 재실행 시 자동 복원."""
self._settings["order_cs_selected_account"] = self.account_var.get()
self._settings["order_cs_selected_market"] = self.market_var.get()
self._settings["order_cs_custom_market"] = self.custom_market_var.get()
if self._save_settings_cb:
self._save_settings_cb()
messagebox.showinfo("저장 완료", "현재 계정명/마켓명 선택을 저장했습니다.\n다음 실행 때 자동으로 불러옵니다.",
parent=self.winfo_toplevel())
# ---------------- 사용자 정의 추가 치환값 ----------------
BUILTIN_PLACEHOLDER_NAMES = {"고객명", "마켓명", "계정명", "상품명", "날짜", "밤문구", "새상품발송일"}
def placeholder_button_keys(self):
"""편집기 '자동 치환값 삽입' 버튼 목록 = 기본 + 사용자 정의."""
return list(ORDER_CS_PLACEHOLDER_BUTTONS) + [f"{{{n}}}" for n in self.custom_fields]
def _persist_custom_fields(self):
self._settings["order_cs_custom_fields"] = list(self.custom_fields)
if self._save_settings_cb:
self._save_settings_cb()
def _rebuild_custom_fields(self):
frame = self.custom_fields_frame
for child in frame.winfo_children():
child.destroy()
head = ttk.Frame(frame)
head.pack(fill="x")
ttk.Label(head, text="추가 치환값", font=("맑은 고딕", 9, "bold")).pack(side="left", padx=(0, 8))
ttk.Button(head, text="+ 추가", command=self._add_custom_field).pack(side="left")
ttk.Label(head, text=" 추가하면 입력칸이 생기고, 양식 수정 창의 {항목명} 버튼으로 넣을 수 있어요.",
foreground="#888888").pack(side="left")
for name in self.custom_fields:
if name not in self.custom_field_vars:
var = tk.StringVar()
var.trace_add("write", lambda *_: self._generate_message())
self.custom_field_vars[name] = var
var = self.custom_field_vars[name]
row = ttk.Frame(frame)
row.pack(fill="x", pady=2)
ttk.Label(row, text=name, width=12, anchor="w").pack(side="left")
tk.Entry(row, textvariable=var, font=("맑은 고딕", 10, "bold"), fg="#C00000").pack(
side="left", fill="x", expand=True, padx=(4, 6), ipady=2)
ttk.Button(row, text="✕", width=3,
command=lambda n=name: self._remove_custom_field(n)).pack(side="right")
def _add_custom_field(self):
name = simpledialog.askstring(
"치환값 추가", "새 치환값(항목명)을 입력하세요. 예: 송장번호",
parent=self.winfo_toplevel())
if not name:
return
name = name.strip().strip("{}").strip()
if not name:
return
if name in self.BUILTIN_PLACEHOLDER_NAMES:
messagebox.showwarning("추가 불가", f"'{name}'은(는) 기본 치환값이라 추가할 수 없습니다.",
parent=self.winfo_toplevel())
return
if name in self.custom_fields:
messagebox.showwarning("추가 불가", "이미 있는 치환값입니다.", parent=self.winfo_toplevel())
return
self.custom_fields.append(name)
self._persist_custom_fields()
self._rebuild_custom_fields()
self._generate_message()
def _remove_custom_field(self, name):
if name not in self.custom_fields:
return
if not messagebox.askyesno(
"치환값 삭제",
f"치환값 '{name}' 입력칸을 삭제할까요?\n(양식 안에 적어둔 {{{name}}} 글자는 그대로 남습니다)",
parent=self.winfo_toplevel(),
):
return
self.custom_fields.remove(name)
self.custom_field_vars.pop(name, None)
self._persist_custom_fields()
self._rebuild_custom_fields()
self._generate_message()
def _copy_message(self):
# 미리보기에서 사용자가 직접 수정한 내용이 다시 기본 양식으로 바뀌지 않도록
# 현재 미리보기 창에 보이는 텍스트 그대로 복사한다.
message = self.preview_text.get("1.0", "end").strip()
top = self.winfo_toplevel()
top.clipboard_clear()
top.clipboard_append(message)
top.update()
messagebox.showinfo("복사 완료", "현재 미리보기 내용 그대로 클립보드에 복사되었습니다.", parent=top)
def _open_template_editor(self):
name = self.cs_type_var.get().strip()
if not name:
messagebox.showwarning("수정 불가", "수정할 양식을 선택해주세요.", parent=self.winfo_toplevel())
return
TemplateEditDialog(self, name, self.templates.get(name, ""), self._save_template_from_dialog)
def _save_template_from_dialog(self, old_name, new_name, content):
# 성공 시 None, 실패 시 오류 메시지(문자열) 반환
# 이름을 바꾼 경우, 새 이름이 다른 양식과 중복이면 거부
if new_name != old_name and new_name in self.templates:
return f"'{new_name}' 이름의 양식이 이미 있습니다."
if new_name == old_name:
self.templates[old_name] = content
else:
# 표시 순서를 유지하며 키 이름만 교체
self.templates = {
(new_name if k == old_name else k): (content if k == old_name else v)
for k, v in self.templates.items()
}
self._save_templates()
self.cs_type_var.set(new_name) # 콤보 선택을 새 이름으로
self._refresh_template_combo()
return None
def _add_template(self):
name = simpledialog.askstring("새 양식 추가", "새 CS유형/양식 이름을 입력해주세요.", parent=self.winfo_toplevel())
if not name:
return
name = name.strip()
if not name:
return
if name in self.templates:
messagebox.showwarning("추가 불가", "이미 같은 이름의 양식이 있습니다.", parent=self.winfo_toplevel())
return
base = self.templates.get(self.cs_type_var.get(), "") or ORDER_CS_DEFAULT_TEMPLATES.get("주문취소", "")
self.templates[name] = base
self._save_templates()
self.cs_type_var.set(name)
self._refresh_template_combo()
self._open_template_editor()
def _download_templates_csv(self):
path = filedialog.asksaveasfilename(
title="양식 다운로드",
defaultextension=".csv",
filetypes=[("CSV 파일", "*.csv"), ("모든 파일", "*.*")],
initialfile="order_cs_sms_templates.csv",
parent=self.winfo_toplevel(),
)
if not path:
return
try:
with open(path, "w", newline="", encoding="utf-8-sig") as f:
writer = csv.writer(f)
writer.writerow(["양식명", "내용"])
for name, content in self.templates.items():
writer.writerow([name, content])
messagebox.showinfo("다운로드 완료", "현재 저장된 모든 양식을 CSV로 저장했습니다.", parent=self.winfo_toplevel())
except Exception as e:
messagebox.showerror("다운로드 실패", str(e), parent=self.winfo_toplevel())
def _upload_templates_csv(self):
path = filedialog.askopenfilename(
title="양식 CSV 업로드",
filetypes=[("CSV 파일", "*.csv"), ("모든 파일", "*.*")],
parent=self.winfo_toplevel(),
)
if not path:
return
try:
loaded = {}
last_err = None
for enc in ("utf-8-sig", "cp949", "utf-8"):
try:
with open(path, "r", newline="", encoding=enc) as f:
sample = f.read(2048)
f.seek(0)
try:
dialect = csv.Sniffer().sniff(sample) if sample else csv.excel
except Exception:
dialect = csv.excel
reader = csv.DictReader(f, dialect=dialect)
if reader.fieldnames and "양식명" in reader.fieldnames and "내용" in reader.fieldnames:
for row in reader:
name = (row.get("양식명") or "").strip()
content = row.get("내용") or ""
if name and content.strip():
loaded[name] = content
else:
f.seek(0)
reader2 = csv.reader(f, dialect=dialect)
rows = list(reader2)
for r in rows[1:] if rows and len(rows[0]) >= 2 else rows:
if len(r) >= 2:
name = str(r[0]).strip()
content = str(r[1])
if name and content.strip():
loaded[name] = content
last_err = None
break
except Exception as e:
last_err = e
if last_err:
raise last_err
if not loaded:
messagebox.showwarning("업로드 실패", "CSV에서 양식을 찾지 못했습니다.\n열 이름은 '양식명', '내용'으로 만들어주세요.", parent=self.winfo_toplevel())
return
if not messagebox.askyesno(
"양식 업로드",
f"CSV 양식 {len(loaded)}개로 전체 교체합니다.\n"
"순서도 CSV 기준으로 반영되고, CSV에 없는 기존 양식은 삭제됩니다.\n진행할까요?",
parent=self.winfo_toplevel()):
return
# CSV에 적힌 순서 그대로 전체 교체(CSV에 있는 것만 남김). loaded는 행 순서를 보존.
self.templates = dict(loaded)
self._save_templates()
self.cs_type_var.set(next(iter(self.templates.keys())))
self._refresh_template_combo()
self._generate_message()
messagebox.showinfo("업로드 완료", f"양식 {len(self.templates)}개로 교체했습니다(순서 반영).", parent=self.winfo_toplevel())
except Exception as e:
messagebox.showerror("업로드 실패", str(e), parent=self.winfo_toplevel())
def _delete_template(self):
name = self.cs_type_var.get().strip()
if len(self.templates) <= 1:
messagebox.showwarning("삭제 불가", "최소 1개의 양식은 남아 있어야 합니다.", parent=self.winfo_toplevel())
return
if not messagebox.askyesno("양식 삭제", f"'{name}' 양식을 삭제하시겠습니까?", parent=self.winfo_toplevel()):
return
self.templates.pop(name, None)
self._save_templates()
self.cs_type_var.set(self._first_template_name())
self._refresh_template_combo()
messagebox.showinfo("삭제 완료", "양식을 삭제했습니다.", parent=self.winfo_toplevel())
def _clear_inputs(self):
self.customer_name_var.set("")
self.market_var.set(ORDER_CS_MARKETS[0])
self.custom_market_var.set("")
self.product_name_var.set("")
self.date_var.set(datetime.now().strftime("%Y/%m/%d"))
self._refresh_dynamic_fields()
class App(tk.Tk):
def __init__(self):
super().__init__()
self.title(f"오너클랜 통합 관리 프로그램 v{__version__} (최종 업데이트 {__released__})")
# 창/작업표시줄 아이콘 = 오너클랜 아이콘 (설정창 등 모든 Toplevel 에도 기본 적용)
try:
if sys.platform == "win32":
import ctypes
ctypes.windll.shell32.SetCurrentProcessExplicitAppUserModelID("Ownerclan.AlertCollector")
except Exception:
pass
try:
if ICON_ICO.exists():
self.iconbitmap(default=str(ICON_ICO))
except Exception:
pass
# 창 크기는 화면 보기 모드(탭/좌우)에 따라 settings 로드 후 _apply_window_geometry()에서 설정.
self.stop_event = threading.Event()
self.worker = None
self.accounts = []
# 마지막 수집 결과를 불러와 재시작 직후에도 카드가 바로 보이게 함 (확인 처리된 건 seen 으로 걸러짐)
cached = load_json(ALERTS_CACHE_PATH, [])
self.rows = [r for r in cached if isinstance(r, dict) and r.get("alert_id")] if isinstance(cached, list) else []
self.seen_ids = set(load_json(SEEN_PATH, []))
self.open_drivers = []
# 저장된 설정 로드 (자동수집 자동시작 등)
self.settings = load_json(SETTINGS_PATH, {})
try:
saved_minutes = int(self.settings.get("auto_collect_minutes", 10))
except Exception:
saved_minutes = 10
saved_minutes = max(1, min(1440, saved_minutes))
# 화면 보기 모드: "tabs"(탭, 심플) / "split"(좌우 펼치기). 변경은 재시작 후 적용.
self.layout_mode = self.settings.get("view_layout", "split")
if self.layout_mode not in ("tabs", "split"):
self.layout_mode = "tabs"
self.layout_mode_var = tk.StringVar(value=self.layout_mode)
self._apply_window_geometry()
self.auto_collect_enabled = False
self.auto_collect_job = None
self.auto_collect_minutes_var = tk.IntVar(value=saved_minutes)
self.auto_collect_interval_ms = saved_minutes * 60 * 1000
self.auto_start_var = tk.BooleanVar(value=True)
self.progress_var = tk.DoubleVar(value=0)
self.progress_text_var = tk.StringVar(value="대기 중")
self.current_account_var = tk.StringVar(value="현재 계정: -")
self.status_var = tk.StringVar(value="상태: 대기")
self.fast_mode_var = tk.BooleanVar(value=True)
self.parallel_workers_var = tk.IntVar(value=4)
self.notify_var = tk.BooleanVar(value=True) # 주문 알림 소리+팝업 on/off
# 알림 팝업 닫힘 방식: "auto"=5초 후 자동, "manual"=닫기 버튼 누를 때까지 유지
self.popup_mode_var = tk.StringVar(value=self.settings.get("popup_dismiss_mode", "auto"))
self.alert_filter_var = tk.StringVar(value="all") # all / memo / qna / todo
self.alert_filter_buttons = {}
self.alert_search_var = tk.StringVar(value="")
self.alert_search_var.trace_add("write", lambda *args: self.on_alert_search_change())
# 사업자(계정) 필터: 모든 알림 탭 상위에서 사업자별로 골라 볼 수 있음
self.account_filter_var = tk.StringVar(value=ACCOUNT_FILTER_ALL)
self.account_filter_combo = None
# 할일/메모 알림 목록
self.todo_memos = load_json(TODO_PATH, [])
self.todo_job = None
self.settings_window = None
self.settings_content = None
self._sound_playing = False
self._toasts = [] # 우하단 토스트 스택
self.tray_icon = None # 시스템 트레이 아이콘
self.create_widgets()
self.protocol("WM_DELETE_WINDOW", self.on_close_request)
self.load_accounts()
self.update_count_label()
# 재시작 직후에도 캐시된 알림 + 할일/메모 카드를 바로 표시
self.after(300, lambda: self.render_cards(preserve_scroll=False))
# 업데이트는 프로그램 시작 시 자동 적용 + [업데이트 확인] 버튼으로만 확인합니다(주기적 확인 없음).
self.update_info = None
# 설정에 '실행 시 자동수집 켜기'가 켜져 있으면 잠시 후 자동 ON
if self.auto_start_var.get():
self.after(800, self.maybe_auto_start)
# 할일/메모 알림 스케줄러 시작 (메인 스레드 self.after 루프)
self.todo_job = self.after(2000, self.check_todo_reminders)
def save_settings(self):
self.settings["auto_start_collect"] = bool(self.auto_start_var.get())
try:
self.settings["auto_collect_minutes"] = max(1, min(1440, int(self.auto_collect_minutes_var.get())))
except Exception:
self.settings["auto_collect_minutes"] = 10
self.settings["popup_dismiss_mode"] = self.popup_mode_var.get()
self.settings["view_layout"] = self.layout_mode_var.get()
try:
self.settings["auto_update"] = bool(self.auto_update_var.get())
except Exception:
pass
try:
save_json(SETTINGS_PATH, self.settings)
except Exception:
pass
# ---------------- 업데이트 (메인 화면) ----------------
def _save_alerts_cache(self):
try:
save_json(ALERTS_CACHE_PATH, list(self.rows))
except Exception:
pass
def check_update_now(self):
"""[업데이트 확인] 버튼을 눌렀을 때만 서버를 확인합니다. 새 버전이면 상단 안내 패널, 최신이면 안내창."""
self.log("업데이트 확인 중...")
self._check_update_async(notify_none=True)
def _check_update_async(self, notify_none=False):
def work():
manifest, err = check_for_update()
self.after(0, lambda: self._show_update(manifest, err, notify_none))
threading.Thread(target=work, daemon=True).start()
def _show_update(self, manifest, err, notify_none=False):
if manifest is None:
self.log(f"업데이트 확인 실패: {err}")
if notify_none:
messagebox.showwarning("업데이트", f"업데이트 서버에 연결하지 못했습니다.\n\n{err}", parent=self)
return
server_ver = manifest.get("version")
if not is_newer_version(manifest):
self.update_info = None
try:
self.update_panel.pack_forget()
except Exception:
pass
self.log(f"최신 버전입니다: {__version__} (서버 {server_ver})")
if notify_none:
messagebox.showinfo("업데이트", f"현재 최신 버전입니다.\n\n현재 {__version__} / 서버 {server_ver}", parent=self)
return
self.update_info = manifest
txt = f"❗ 새 버전 {server_ver} 이(가) 있습니다 (현재 {__version__})"
if manifest.get("released_at"):
txt += f" {manifest.get('released_at')} 배포"
notes = str(manifest.get("notes") or "").strip()
if notes:
txt += "\n" + notes
self.update_panel_label.config(text=txt)
self.update_apply_btn.config(state="normal", text="재시작 & 업데이트")
try:
self.update_panel.pack(fill="x", padx=8, pady=(0, 6), before=self.alert_filter_bar)
except Exception:
self.update_panel.pack(fill="x", padx=8, pady=(0, 6))
self.log(f"새 버전 {server_ver} 발견 → 상단 [재시작 & 업데이트] 로 적용")
def apply_update_now(self):
"""안내 패널의 [재시작 & 업데이트]"""
if self.update_info is None:
self.check_update_now()
return
self.update_apply_btn.config(state="disabled", text="준비 중…")
self._apply_update_and_restart(self.update_info)
def restart_program(self):
"""[재시작] 버튼: 업데이트 여부와 상관없이 프로그램만 다시 시작 (시작 시 자동 업데이트는 그대로 동작)."""
if not messagebox.askyesno("재시작", "프로그램을 재시작할까요?\n(수집 중이던 작업은 중단됩니다)", parent=self):
return
self._restart_without_update()
def _restart_without_update(self):
self.log("재시작합니다")
sys.argv = [a for a in sys.argv if a != "--no-update"]
try:
self.auto_collect_enabled = False
if self.auto_collect_job is not None:
self.after_cancel(self.auto_collect_job)
self.stop_event.set()
if self.tray_icon is not None:
self.tray_icon.stop()
self.tray_icon = None
for drv in list(getattr(self, "open_drivers", [])):
quit_driver(drv, timeout=3.0)
CHROME_REGISTRY.kill_all()
except Exception:
pass
relaunch_self(skip_update=False)
# ---------------- 업데이트 (수동 확인) ----------------
def check_update_manual(self):
"""설정창 [업데이트 확인] 버튼. 서버 확인은 백그라운드 스레드, 결과 처리는 메인 스레드."""
try:
self.update_check_btn.config(state="disabled")
except Exception:
pass
self.update_status_var.set("확인 중...")
self.log("업데이트 확인 중...")
def worker():
manifest, err = check_for_update()
self.after(0, lambda: self._on_update_checked(manifest, err))
threading.Thread(target=worker, daemon=True).start()
def _on_update_checked(self, manifest, err):
try:
self.update_check_btn.config(state="normal")
except Exception:
pass
if manifest is None:
self.update_status_var.set("확인 실패")
self.log(f"업데이트 확인 실패: {err}")
messagebox.showwarning("업데이트", f"업데이트 서버에 연결하지 못했습니다.\n\n{err}", parent=self.settings_window)
return
server_ver = manifest.get("version")
if not is_newer_version(manifest):
self.update_status_var.set(f"최신 버전 (서버 {server_ver})")
self.log(f"최신 버전입니다: {__version__}")
messagebox.showinfo("업데이트", f"현재 최신 버전입니다.\n\n현재 {__version__} / 서버 {server_ver}", parent=self.settings_window)
return
self.update_status_var.set(f"새 버전 {server_ver} 있음")
notes = str(manifest.get("notes") or "").strip()
msg = f"새 버전 {server_ver} 이(가) 있습니다. (현재 {__version__})\n\n지금 업데이트하고 프로그램을 다시 시작할까요?"
if notes:
msg += f"\n\n변경 내용:\n{notes}"
if messagebox.askyesno("업데이트", msg, parent=self.settings_window):
self._apply_update_and_restart(manifest)
def _apply_update_and_restart(self, manifest):
self.log(f"업데이트 적용 중: {__version__} → {manifest.get('version')}")
ok = apply_update(manifest, log=self.log)
if not ok:
messagebox.showerror("업데이트", "업데이트 적용에 실패했습니다.\nupdate.log 를 확인하세요. (기존 버전은 그대로 유지됩니다)", parent=self.settings_window)
self.update_status_var.set("적용 실패")
try:
self.update_apply_btn.config(state="normal", text="재시작 & 업데이트")
except Exception:
pass
return
self.log("업데이트 완료 → 재시작합니다")
# 자동수집/크롬 정리 후 재시작
try:
self.auto_collect_enabled = False
if self.auto_collect_job is not None:
self.after_cancel(self.auto_collect_job)
self.stop_event.set()
if self.tray_icon is not None:
self.tray_icon.stop()
self.tray_icon = None
for drv in list(getattr(self, "open_drivers", [])):
quit_driver(drv, timeout=3.0)
CHROME_REGISTRY.kill_all()
except Exception:
pass
relaunch_self()
def maybe_auto_start(self):
if not self.auto_start_var.get():
return
if self.auto_collect_enabled:
return
enabled = [a for a in self.accounts if a.enabled]
if not enabled:
self.log("실행 시 자동수집: 사용 체크된 계정이 없어 건너뜀")
return
self.log("실행 시 자동수집 켜짐 → 자동수집 시작")
self.toggle_auto_collect()
def _apply_window_geometry(self):
"""화면 보기 모드에 맞춰 창 크기를 설정한다."""
if self.layout_mode == "split":
# 좌우 펼치기: 크기 조절 가능 + 시작 크기를 화면에 맞게 클램프(작은 화면에서 하단 잘림 방지)
try:
sw, sh = self.winfo_screenwidth(), self.winfo_screenheight()
except Exception:
sw, sh = 1280, 820
w = min(1280, max(1000, sw - 100))
h = min(880, max(600, sh - 120))
self.maxsize(10000, 10000)
self.minsize(1000, 600)
self.resizable(True, True)
self.geometry(f"{w}x{h}")
else:
# 탭(심플): 기존처럼 고정
self.geometry("640x760")
self.minsize(640, 760)
self.maxsize(640, 760)
self.resizable(False, False)
def _init_sash(self, paned):
# 좌우 분할 초기 경계 = 창 가로 절반(좌우 균등). 콘텐츠 최소폭에 막히면 근사 균등.
try:
w = paned.winfo_width()
if w <= 1:
# 아직 실제 폭이 잡히지 않음 → 잠시 후 다시 시도
self.after(50, lambda: self._init_sash(paned))
return
# 좌우 1:1(가운데 분할). 이후 사용자가 경계를 드래그해 비율 조절 가능.
paned.sashpos(0, w // 2)
except Exception:
pass
def on_layout_mode_change(self):
"""설정에서 화면 보기 모드를 바꿨을 때: 저장하고 재시작 후 적용 안내."""
self.save_settings()
messagebox.showinfo(
"화면 보기 설정",
"변경한 화면 보기는 프로그램을 다시 시작하면 적용됩니다.",
)
def create_widgets(self):
if self.layout_mode == "split":
# 좌우 펼치기: 드래그 가능한 분할창에 알림 수집기(좌) / 주문·CS·문자(우)를 동시에.
paned = ttk.PanedWindow(self, orient="horizontal")
paned.pack(fill="both", expand=True)
alert_parent = ttk.Frame(paned)
sms_parent = ttk.Frame(paned)
paned.add(alert_parent, weight=1)
paned.add(sms_parent, weight=1)
self.after_idle(lambda: self._init_sash(paned))
else:
# 탭(심플): 기존처럼 탭으로 구분.
notebook = ttk.Notebook(self)
notebook.pack(fill="both", expand=True)
alert_parent = ttk.Frame(notebook)
sms_parent = ttk.Frame(notebook)
notebook.add(alert_parent, text="알림 수집기")
notebook.add(sms_parent, text="주문/CS/문자")
self.create_alert_widgets(alert_parent)
self.order_cs_sms_frame = OrderCsSmsFrame(
sms_parent,
settings=self.settings,
save_settings=self.save_settings,
account_labels_provider=lambda: [a.label for a in self.accounts],
)
self.order_cs_sms_frame.pack(fill="both", expand=True)
self.order_cs_sms_frame.set_callback("on_check_update", self.check_update_now)
self.order_cs_sms_frame.set_callback("on_restart", self.restart_program)
def create_alert_widgets(self, parent):
style = ttk.Style()
try:
style.theme_use("clam")
except Exception:
pass
try:
# 진행률 막대가 또렷이 보이도록 색 지정 (thickness는 잘림 유발하므로 미지정)
style.configure(
"Collect.Horizontal.TProgressbar",
troughcolor="#e6e9f0",
background="#2f80ed",
)
except Exception:
pass
try:
# '당장 수집!!' 버튼 강조용 볼드 스타일
style.configure("Collect.TButton", font=("맑은 고딕", 10, "bold"))
except Exception:
pass
main = ttk.Frame(parent)
main.pack(fill="both", expand=True)
right = ttk.Frame(main)
self.alert_right = right
right.pack(fill="both", expand=True)
# 설정창을 미리 만들고 숨겨둔 뒤, 그 안에 계정설정/실행/로그 UI를 처음부터 생성합니다.
self.settings_window = tk.Toplevel(self)
self.settings_window.title("오너클랜 설정")
self.settings_window.geometry("520x680")
self.settings_window.minsize(460, 560)
self.settings_window.withdraw()
self.settings_window.protocol("WM_DELETE_WINDOW", self.hide_settings_window)
try:
self.settings_window.transient(self)
except Exception:
pass
left = ttk.Frame(self.settings_window)
self.settings_content = left
self.settings_content.pack(fill="both", expand=True)
top = ttk.LabelFrame(left, text="계정 설정")
top.pack(fill="x", expand=False, padx=8, pady=(6, 4))
ttk.Label(top, text="계정을 클릭하면 아래 칸에 내용이 채워집니다. 수정 후 '추가/수정 반영'.",
foreground="#888888", wraplength=390).pack(anchor="w", padx=6, pady=(4, 2))
self.tree = ttk.Treeview(top, columns=("enabled", "label", "user_id"), show="headings", height=5)
self.tree.heading("enabled", text="사용")
self.tree.heading("label", text="계정명")
self.tree.heading("user_id", text="아이디")
self.tree.column("enabled", width=45, anchor="center")
self.tree.column("label", width=115)
self.tree.column("user_id", width=145)
self.tree.pack(fill="x", padx=6, pady=2)
# 단일 클릭(선택)만으로 폼에 데이터 로드
self.tree.bind("<>", self.edit_selected)
form = ttk.Frame(top)
form.pack(fill="x", padx=6, pady=(2, 4))
form.columnconfigure(1, weight=1)
self.enabled_var = tk.BooleanVar(value=True)
ttk.Checkbutton(form, text="사용", variable=self.enabled_var).grid(row=0, column=0, columnspan=3, sticky="w")
ttk.Label(form, text="계정명").grid(row=1, column=0, sticky="w", pady=1)
self.label_var = tk.StringVar()
ttk.Entry(form, textvariable=self.label_var).grid(row=1, column=1, columnspan=2, sticky="we", pady=1)
ttk.Label(form, text="아이디").grid(row=2, column=0, sticky="w", pady=1)
self.id_var = tk.StringVar()
ttk.Entry(form, textvariable=self.id_var).grid(row=2, column=1, columnspan=2, sticky="we", pady=1)
ttk.Label(form, text="비밀번호").grid(row=3, column=0, sticky="w", pady=1)
self.pw_var = tk.StringVar()
self.pw_entry = ttk.Entry(form, textvariable=self.pw_var, show="*")
self.pw_entry.grid(row=3, column=1, sticky="we", pady=1)
self.pw_shown = False
self.pw_toggle_btn = ttk.Button(form, text="표시", width=5, command=self.toggle_password)
self.pw_toggle_btn.grid(row=3, column=2, sticky="e", padx=(4, 0), pady=1)
# 메모 URL은 내부 기본값을 사용하므로 화면에서 확인
self.url_var = tk.StringVar(value=DEFAULT_MEMO_URL)
form_btns = ttk.Frame(form)
form_btns.grid(row=4, column=0, columnspan=3, sticky="we", pady=(6, 0))
ttk.Button(form_btns, text="입력초기화", command=self.add_account).pack(side="left")
ttk.Button(form_btns, text="선택 삭제", command=self.delete_selected).pack(side="left", padx=4)
ttk.Button(form_btns, text="추가/수정 반영", command=self.apply_form).pack(side="right")
self.editing_item = None
runbox = ttk.LabelFrame(left, text="실행")
runbox.pack(fill="x", padx=8, pady=4)
# 로그인 방식은 자동 로그인으로 고정
self.login_mode = tk.StringVar(value="auto")
# 창 모드 + 속도/주문 옵션을 한 줄로 압축
self.window_mode = tk.StringVar(value="headless")
optrow = ttk.Frame(runbox)
optrow.pack(fill="x", padx=8, pady=(5, 1))
ttk.Label(optrow, text="창:").pack(side="left")
ttk.Radiobutton(optrow, text="숨김", variable=self.window_mode, value="headless").pack(side="left")
ttk.Radiobutton(optrow, text="보이기", variable=self.window_mode, value="normal").pack(side="left", padx=(0, 8))
ttk.Checkbutton(optrow, text="빠른수집", variable=self.fast_mode_var).pack(side="left")
opt2 = ttk.Frame(runbox)
opt2.pack(fill="x", padx=8, pady=1)
ttk.Checkbutton(opt2, text="주문 신규 시 소리+우하단 팝업", variable=self.notify_var).pack(side="left")
ttk.Label(opt2, text=" 동시").pack(side="left")
self.workers_spin = ttk.Spinbox(opt2, from_=1, to=4, textvariable=self.parallel_workers_var, width=4)
self.workers_spin.pack(side="left", padx=4)
popup_row = ttk.Frame(runbox)
popup_row.pack(fill="x", padx=8, pady=1)
ttk.Label(popup_row, text="알림팝업:").pack(side="left")
ttk.Radiobutton(popup_row, text="5초 후 자동 닫힘", variable=self.popup_mode_var,
value="auto", command=self.save_settings).pack(side="left")
ttk.Radiobutton(popup_row, text="닫기 누를 때까지 유지", variable=self.popup_mode_var,
value="manual", command=self.save_settings).pack(side="left", padx=(6, 0))
layout_row = ttk.Frame(runbox)
layout_row.pack(fill="x", padx=8, pady=1)
ttk.Label(layout_row, text="화면보기:").pack(side="left")
ttk.Radiobutton(layout_row, text="탭 형식(심플)", variable=self.layout_mode_var,
value="tabs", command=self.on_layout_mode_change).pack(side="left")
ttk.Radiobutton(layout_row, text="좌우 펼치기", variable=self.layout_mode_var,
value="split", command=self.on_layout_mode_change).pack(side="left", padx=(6, 0))
ttk.Label(layout_row, text="(재시작 후 적용)", foreground="#888888").pack(side="left", padx=(6, 0))
row = ttk.Frame(runbox)
row.pack(fill="x", padx=8, pady=(4, 2))
ttk.Button(row, text="중지", command=self.stop_collect).pack(side="left", fill="x", expand=True)
auto_row = ttk.Frame(runbox)
auto_row.pack(fill="x", padx=8, pady=2)
self.auto_collect_btn = tk.Button(
auto_row,
text="자동수집 OFF",
bg="#777777",
fg="white",
activebackground="#555555",
activeforeground="white",
relief="flat",
padx=10,
pady=4,
command=self.toggle_auto_collect
)
self.auto_collect_btn.pack(side="left", fill="x", expand=True)
ttk.Label(auto_row, text="간격").pack(side="left", padx=(8, 2))
self.auto_minutes_spin = ttk.Spinbox(
auto_row,
from_=1,
to=1440,
textvariable=self.auto_collect_minutes_var,
width=5,
command=self.on_interval_change,
)
self.auto_minutes_spin.pack(side="left")
# 간격은 자동수집 ON/OFF와 무관하게 항상 편집 가능. 입력 확정 시 즉시 반영/저장.
self.auto_minutes_spin.bind("", lambda e: self.on_interval_change())
self.auto_minutes_spin.bind("", lambda e: self.on_interval_change())
ttk.Label(auto_row, text="분").pack(side="left", padx=(2, 6))
self.auto_collect_label = ttk.Label(auto_row, text="자동수집 꺼짐")
self.auto_collect_label.pack(side="left", padx=4)
autostart_row = ttk.Frame(runbox)
autostart_row.pack(fill="x", padx=8, pady=(0, 2))
ttk.Checkbutton(
autostart_row,
text="프로그램 실행 시 자동수집 켜기",
variable=self.auto_start_var,
command=self.save_settings
).pack(side="left")
# ---- 프로그램 정보 / 업데이트 ----
updbox = ttk.LabelFrame(left, text="프로그램 업데이트")
updbox.pack(fill="x", padx=8, pady=4)
upd_row = ttk.Frame(updbox)
upd_row.pack(fill="x", padx=8, pady=(4, 2))
ttk.Label(upd_row, text=f"현재 버전: {__version__} | 최종 업데이트: {__released__}").pack(side="left")
self.update_status_var = tk.StringVar(value="")
ttk.Label(upd_row, textvariable=self.update_status_var, foreground="#888888").pack(side="left", padx=(8, 0))
self.update_check_btn = ttk.Button(upd_row, text="업데이트 확인", command=self.check_update_manual)
self.update_check_btn.pack(side="right")
upd_row2 = ttk.Frame(updbox)
upd_row2.pack(fill="x", padx=8, pady=(0, 4))
self.auto_update_var = tk.BooleanVar(value=bool(self.settings.get("auto_update", True)))
ttk.Checkbutton(
upd_row2,
text="프로그램 시작 시 자동 업데이트 (껐다 켜면 새 버전 적용)",
variable=self.auto_update_var,
command=self.save_settings,
).pack(side="left")
# '전체 확인' / '확인 초기화'는 메인 화면 헤더(밖)로 이동했습니다.
# 진행률 박스는 알림 카드 '아래'에 고정(side=bottom). 카드(expand)가 가운데를 채운다.
progress_box = ttk.LabelFrame(right, text="")
progress_box.pack(side="bottom", fill="x", padx=8, pady=(4, 8))
ttk.Label(progress_box, textvariable=self.progress_text_var).pack(anchor="w", padx=8, pady=(4, 1))
progress_row = ttk.Frame(progress_box)
progress_row.pack(fill="x", padx=8, pady=2)
self.progress_bar = ttk.Progressbar(
progress_row,
variable=self.progress_var,
maximum=100,
mode="determinate",
style="Collect.Horizontal.TProgressbar"
)
self.progress_bar.pack(side="left", fill="x", expand=True)
ttk.Button(progress_row, text="당장 수집!!", command=self.start_collect,
style="Collect.TButton").pack(side="right", padx=(6, 0))
prow = ttk.Frame(progress_box)
prow.pack(fill="x", padx=8, pady=(1, 4))
ttk.Label(prow, textvariable=self.current_account_var).pack(side="left")
ttk.Label(prow, textvariable=self.status_var).pack(side="right")
logbox = ttk.LabelFrame(left, text="로그")
logbox.pack(fill="both", expand=True, padx=8, pady=(4, 6))
self.log_text = tk.Text(logbox, height=4, wrap="word")
self.log_text.pack(fill="both", expand=True, padx=4, pady=4)
header = ttk.Frame(right)
# 상단 여백을 SMS 섹션 패딩(18)과 비슷하게 줘서 좌우 제목 높이를 맞춘다.
header.pack(fill="x", padx=8, pady=(14, 4))
# 헤더는 제목 + 수집 건수만(설정/전체확인/확인초기화 버튼은 아래 사업자 줄로 이동 → 건수 안 잘림)
ttk.Label(header, text="신규 알림", font=("맑은 고딕", 16, "bold")).pack(side="left")
self.count_label = ttk.Label(header, text="", font=("맑은 고딕", 11), anchor="w")
self.count_label.pack(side="left", padx=12, fill="x", expand=True)
self.auto_collect_top_btn = tk.Button(
header,
text="자동수집 OFF",
bg="#777777",
fg="white",
activebackground="#555555",
activeforeground="white",
relief="flat",
padx=10,
pady=4,
command=self.toggle_auto_collect,
)
self.auto_collect_top_btn.pack(side="right", padx=(8, 0))
# 사업자 필터(모든 탭 상위) + 우측에 설정/전체확인/확인초기화 버튼
account_bar = tk.Frame(right, bg="#f6f7fb")
account_bar.pack(fill="x", padx=8, pady=(2, 4))
tk.Label(account_bar, text="사업자", bg="#f6f7fb", fg="#444444",
font=("맑은 고딕", 10, "bold")).pack(side="left", padx=(2, 6), pady=3)
self.account_filter_combo = ttk.Combobox(
account_bar,
textvariable=self.account_filter_var,
state="readonly",
values=[ACCOUNT_FILTER_ALL],
width=22,
)
self.account_filter_combo.pack(side="left", pady=3)
ttk.Button(account_bar, text="확인 초기화", command=self.clear_seen).pack(side="right", padx=(0, 2), pady=2)
ttk.Button(account_bar, text="전체 확인", command=self.confirm_all_visible).pack(side="right", padx=(0, 4), pady=2)
ttk.Button(account_bar, text="설정", command=self.open_settings_window).pack(side="right", padx=(0, 4), pady=2)
self.account_filter_combo.bind("<>", lambda e: self.on_account_filter_change())
# 업데이트 안내 패널: 새 버전이 있을 때만 탭 줄 위에 나타남
self.update_panel = tk.Frame(right, bg="#fff4e5", highlightbackground="#f0b429", highlightthickness=1)
self.update_panel_label = tk.Label(
self.update_panel, text="", bg="#fff4e5", fg="#8a5a00",
font=("맑은 고딕", 10, "bold"), anchor="w", justify="left",
)
self.update_panel_label.pack(side="left", padx=10, pady=6, fill="x", expand=True)
self.update_apply_btn = tk.Button(
self.update_panel, text="재시작 & 업데이트", bg="#6f42c1", fg="white", relief="flat", padx=10, pady=4,
font=("맑은 고딕", 9, "bold"), activebackground="#59359a", activeforeground="white",
command=self.apply_update_now,
)
self.update_apply_btn.pack(side="right", padx=8, pady=6)
self.alert_filter_bar = tk.Frame(right, bg="#f6f7fb")
self.alert_filter_bar.pack(fill="x", padx=8, pady=(0, 6))
tab_wrap = tk.Frame(self.alert_filter_bar, bg="#f6f7fb")
tab_wrap.pack(side="left")
# 탭 4개(+검색)가 좁은 폭에도 들어가도록 고정 width 대신 글자크기에 맞춰 자동 폭(패딩 최소화).
for key, label in [("all", "전체"), ("memo", "신규 알림"), ("qna", "문의글"), ("todo", "할일/메모 알림")]:
btn = tk.Button(
tab_wrap,
text=label,
relief="flat",
padx=5,
pady=4,
font=("맑은 고딕", 9, "bold"),
command=lambda k=key: self.set_alert_filter(k),
)
btn.pack(side="left", padx=(0, 3))
self.alert_filter_buttons[key] = btn
search_wrap = tk.Frame(self.alert_filter_bar, bg="#f6f7fb")
search_wrap.pack(side="right", fill="x", expand=True, padx=(12, 0))
search_box = tk.Frame(search_wrap, bg="#ffffff", highlightbackground="#cfd7e6", highlightthickness=1, bd=0)
search_box.pack(side="right", fill="x", expand=True)
tk.Label(search_box, text="검색", bg="#ffffff", fg="#2f80ed", font=("맑은 고딕", 10, "bold")).pack(side="left", padx=(10, 6), pady=5)
self.alert_search_entry = tk.Entry(
search_box,
textvariable=self.alert_search_var,
relief="flat",
bd=0,
font=("맑은 고딕", 10),
fg="#222222",
insertbackground="#2f80ed",
)
self.alert_search_entry.pack(side="left", fill="x", expand=True, padx=(0, 6), pady=6)
self.alert_search_entry.bind("", lambda e: self.render_cards())
self.alert_search_entry.bind("", lambda e: self.clear_alert_search())
self.alert_search_clear_btn = tk.Button(
search_box,
text="지우기",
relief="flat",
bd=0,
padx=8,
pady=4,
bg="#ffffff",
fg="#666666",
activebackground="#eef4ff",
activeforeground="#2f80ed",
font=("맑은 고딕", 9),
command=self.clear_alert_search,
)
self.alert_search_clear_btn.pack(side="right", padx=(0, 6))
self.cards = ScrollableFrame(right)
self.cards.pack(fill="both", expand=True, padx=8, pady=(0, 8))
def open_settings_window(self):
"""계정설정/실행/로그 설정창을 표시합니다."""
try:
self.settings_window.deiconify()
self.settings_window.lift()
self.settings_window.focus_force()
except Exception:
pass
def hide_settings_window(self):
"""설정창을 닫지 않고 숨깁니다. 설정값은 저장합니다."""
try:
self.save_accounts()
self.save_settings()
except Exception:
pass
try:
self.settings_window.withdraw()
except Exception:
pass
def on_close_request(self):
"""X 버튼: 종료 / 트레이로 이동 / 취소 선택"""
choice = self.ask_close_choice()
if choice == "quit":
self.real_quit()
elif choice == "tray":
self.hide_to_tray()
# cancel: 아무것도 안 함
def ask_close_choice(self):
dlg = tk.Toplevel(self)
dlg.title("종료")
dlg.transient(self)
dlg.resizable(False, False)
dlg.grab_set()
result = {"v": "cancel"}
tk.Label(dlg, text="프로그램을 어떻게 할까요?", font=("맑은 고딕", 11),
padx=20, pady=14).pack()
if self.tray_icon is None and pystray is None:
tk.Label(dlg, text="(트레이 기능을 쓰려면 pystray 설치가 필요합니다)",
fg="#888888", font=("맑은 고딕", 9)).pack()
btns = tk.Frame(dlg)
btns.pack(padx=16, pady=(4, 16))
def choose(v):
result["v"] = v
dlg.destroy()
tk.Button(btns, text="종료", width=10, command=lambda: choose("quit")).pack(side="left", padx=4)
tray_state = "normal" if pystray is not None else "disabled"
tk.Button(btns, text="트레이로 이동", width=12, state=tray_state,
command=lambda: choose("tray")).pack(side="left", padx=4)
tk.Button(btns, text="취소", width=10, command=lambda: choose("cancel")).pack(side="left", padx=4)
# 화면 중앙 배치
dlg.update_idletasks()
x = self.winfo_x() + (self.winfo_width() - dlg.winfo_width()) // 2
y = self.winfo_y() + (self.winfo_height() - dlg.winfo_height()) // 3
dlg.geometry(f"+{max(0, x)}+{max(0, y)}")
dlg.protocol("WM_DELETE_WINDOW", lambda: choose("cancel"))
dlg.wait_window()
return result["v"]
def _make_tray_image(self):
try:
if ICON_PNG.exists():
return Image.open(str(ICON_PNG)).convert("RGBA").resize((64, 64))
except Exception:
pass
img = Image.new("RGB", (64, 64), "#2f80ed")
d = ImageDraw.Draw(img)
d.ellipse((16, 16, 48, 48), fill="#ffffff")
d.ellipse((26, 26, 38, 38), fill="#e74c3c")
return img
def hide_to_tray(self):
self.withdraw() # 창 숨김 + 작업표시줄에서도 제거
if self.tray_icon is None:
menu = pystray.Menu(
pystray.MenuItem("열기", self._tray_show, default=True),
pystray.MenuItem("종료", self._tray_quit),
)
self.tray_icon = pystray.Icon(
"ownerclan_alert", self._make_tray_image(),
"오너클랜 알림 수집기 (실행 중)", menu
)
threading.Thread(target=self.tray_icon.run, daemon=True).start()
self.log("트레이로 이동 (백그라운드 실행 계속)")
def _tray_show(self, icon=None, item=None):
self.after(0, self._restore_from_tray)
def _restore_from_tray(self):
if self.tray_icon is not None:
try:
self.tray_icon.stop()
except Exception:
pass
self.tray_icon = None
self.deiconify()
self.lift()
self.focus_force()
def _tray_quit(self, icon=None, item=None):
self.after(0, self.real_quit)
def real_quit(self):
"""
종료: UI 는 즉시 숨기고, Chrome/chromedriver/트레이 정리는 별도 스레드에서 수행합니다.
정리가 3초 안에 안 끝나도 프로세스를 확실히 내려 '응답 없음' 과 잔류 프로세스를 막습니다.
"""
if getattr(self, "_quitting", False):
return
self._quitting = True
# 1) 화면 즉시 숨김 (응답 없음 표시 자체가 안 뜨게)
try:
if self.settings_window is not None and self.settings_window.winfo_exists():
self.settings_window.withdraw()
except Exception:
pass
try:
self.withdraw()
except Exception:
pass
# 2) 예약 작업 취소 + 수집 중지 신호
self.auto_collect_enabled = False
for job_attr in ("auto_collect_job", "todo_job"):
job = getattr(self, job_attr, None)
if job is not None:
try:
self.after_cancel(job)
except Exception:
pass
setattr(self, job_attr, None)
self.stop_event.set()
# 3) 정리 작업은 UI 스레드 밖에서
open_drivers = list(getattr(self, "open_drivers", []))
try:
self.open_drivers.clear()
except Exception:
pass
tray = self.tray_icon
self.tray_icon = None
def cleanup():
if tray is not None:
try:
tray.stop()
except Exception:
pass
for drv in open_drivers:
quit_driver(drv, timeout=3.0)
# 수집/확인 처리 중이던 배경 Chrome 포함, 우리가 만든 것만 정리
CHROME_REGISTRY.kill_all()
t = threading.Thread(target=cleanup, daemon=True)
t.start()
deadline = time.time() + 3.0
def finish():
if t.is_alive() and time.time() < deadline:
self.after(100, finish)
return
try:
self.destroy()
except Exception:
pass
# 남은 워커 스레드(ThreadPoolExecutor 등) 때문에 프로세스가 붙잡히지 않도록 확실히 종료
import os
os._exit(0)
self.after(50, finish)
def ask_manual_login(self, account_label):
event = threading.Event()
def show_popup():
messagebox.showinfo(
"수동 로그인",
f"[{account_label}] 크롬창에서 직접 로그인하세요.\n\n"
"로그인이 끝나고 메인/마이페이지 화면이 보이면\n"
"이 창의 확인 버튼을 누르세요."
)
event.set()
self.after(0, show_popup)
while not event.is_set():
if self.stop_event.is_set():
break
time.sleep(0.2)
def log(self, msg):
line = f"{datetime.now().strftime('%H:%M:%S')} {msg}\n"
self.log_text.after(0, lambda: (self.log_text.insert("end", line), self.log_text.see("end")))
def refresh_tree(self):
for item in self.tree.get_children():
self.tree.delete(item)
for acc in self.accounts:
self.tree.insert("", "end", values=("Y" if acc.enabled else "N", acc.label, acc.user_id))
# 주문/CS/문자 탭의 계정명 콤보를 계정 목록 변경에 맞춰 동기화
if hasattr(self, "order_cs_sms_frame"):
self.order_cs_sms_frame.refresh_accounts()
# 상단 사업자 필터 콤보도 계정 목록 변경에 맞춰 동기화
self.refresh_account_filter()
def refresh_account_filter(self):
combo = getattr(self, "account_filter_combo", None)
if combo is None:
return
values = [ACCOUNT_FILTER_ALL] + [a.label for a in self.accounts]
try:
combo.configure(values=values)
except Exception:
pass
# 현재 선택이 사라진 계정이면 전체로 리셋
if self.account_filter_var.get() not in values:
self.account_filter_var.set(ACCOUNT_FILTER_ALL)
def toggle_password(self):
self.pw_shown = not self.pw_shown
if self.pw_shown:
self.pw_entry.config(show="")
self.pw_toggle_btn.config(text="숨김")
else:
self.pw_entry.config(show="*")
self.pw_toggle_btn.config(text="표시")
def add_account(self):
self.editing_item = None
# 트리 선택 하이라이트도 풀어 '새 계정 준비' 상태를 명확히
try:
self.tree.selection_remove(self.tree.selection())
except Exception:
pass
self.enabled_var.set(True)
self.label_var.set("")
self.id_var.set("")
self.pw_var.set("")
self.url_var.set(DEFAULT_MEMO_URL)
def apply_form(self):
acc = Account(
enabled=self.enabled_var.get(),
label=self.label_var.get().strip(),
user_id=self.id_var.get().strip(),
password=self.pw_var.get(),
memo_url=(self.url_var.get().strip() or DEFAULT_MEMO_URL),
)
if not acc.label:
messagebox.showwarning("확인", "계정명은 입력해야 합니다.")
return
if self.login_mode.get() == "auto" and (not acc.user_id or not acc.password):
messagebox.showwarning("확인", "자동 로그인은 아이디/비밀번호가 필요합니다.")
return
if not acc.user_id:
acc.user_id = acc.label
if self.editing_item is not None:
idx = self.tree.index(self.editing_item)
self.accounts[idx] = acc
self.editing_item = None
else:
self.accounts.append(acc)
self.refresh_tree()
self.save_accounts()
def edit_selected(self, event=None):
sel = self.tree.selection()
if not sel:
return
item = sel[0]
idx = self.tree.index(item)
acc = self.accounts[idx]
self.editing_item = item
self.enabled_var.set(acc.enabled)
self.label_var.set(acc.label)
self.id_var.set(acc.user_id)
self.pw_var.set(acc.password)
self.url_var.set(acc.memo_url)
def delete_selected(self):
sel = self.tree.selection()
if not sel:
return
idxs = sorted([self.tree.index(i) for i in sel], reverse=True)
for idx in idxs:
del self.accounts[idx]
self.refresh_tree()
self.save_accounts()
def save_accounts(self):
save_json(CONFIG_PATH, [asdict(a) for a in self.accounts])
self.log(f"계정 설정 저장: {CONFIG_PATH}")
def load_accounts(self):
data = load_json(CONFIG_PATH, [])
try:
self.accounts = [Account(**x) for x in data]
except Exception:
self.accounts = []
self.refresh_tree()
def get_auto_collect_minutes(self):
try:
minutes = int(self.auto_collect_minutes_var.get())
except Exception:
minutes = 30
minutes = max(1, min(1440, minutes))
self.auto_collect_minutes_var.set(minutes)
return minutes
def on_interval_change(self):
"""자동수집 간격을 바꿨을 때: 값 확정·저장하고, 켜져 있으면 새 간격으로 다음 예약을 갱신."""
minutes = self.get_auto_collect_minutes()
new_ms = minutes * 60 * 1000
changed = (new_ms != self.auto_collect_interval_ms)
self.auto_collect_interval_ms = new_ms
self.save_settings()
if changed and self.auto_collect_enabled:
self.log(f"자동수집 간격 {minutes}분으로 변경 → 다음 예약 갱신")
self.schedule_next_auto_collect()
def toggle_auto_collect(self):
if self.auto_collect_enabled:
self.auto_collect_enabled = False
if self.auto_collect_job is not None:
try:
self.after_cancel(self.auto_collect_job)
except Exception:
pass
self.auto_collect_job = None
self.auto_collect_btn.config(text="자동수집 OFF", bg="#777777", activebackground="#555555")
try:
self.auto_collect_top_btn.config(text="자동수집 OFF", bg="#777777", activebackground="#555555")
except Exception:
pass
self.auto_collect_label.config(text="자동수집 꺼짐")
self.log("자동수집 OFF")
self.set_progress(self.progress_var.get(), status="상태: 자동수집 꺼짐")
else:
minutes = self.get_auto_collect_minutes()
self.auto_collect_interval_ms = minutes * 60 * 1000
self.auto_collect_enabled = True
self.auto_collect_btn.config(text="자동수집 ON", bg="#1f9d55", activebackground="#167a40")
try:
self.auto_collect_top_btn.config(text="자동수집 ON", bg="#1f9d55", activebackground="#167a40")
except Exception:
pass
self.auto_collect_label.config(text=f"자동수집 켜짐 / {minutes}분마다")
self.save_settings() # 자동시작 설정 + 현재 간격 저장
self.log(f"자동수집 ON: {minutes}분마다")
self.set_progress(self.progress_var.get(), status=f"상태: 자동수집 켜짐 ({minutes}분마다)")
# 켜는 즉시 1회 수집하고, 이후 설정한 분마다 반복
self.start_collect(auto_trigger=True)
self.schedule_next_auto_collect()
def schedule_next_auto_collect(self):
if not self.auto_collect_enabled:
return
if self.auto_collect_job is not None:
try:
self.after_cancel(self.auto_collect_job)
except Exception:
pass
minutes = self.get_auto_collect_minutes()
self.auto_collect_interval_ms = minutes * 60 * 1000
self.auto_collect_label.config(text=f"다음 자동수집 예약 / {minutes}분 후")
self.auto_collect_job = self.after(self.auto_collect_interval_ms, self.run_auto_collect)
def run_auto_collect(self):
self.auto_collect_job = None
if not self.auto_collect_enabled:
return
minutes = self.get_auto_collect_minutes()
self.log(f"자동수집 실행 ({minutes}분 간격)")
self.auto_collect_label.config(text="자동수집 실행 중")
self.start_collect(auto_trigger=True)
self.schedule_next_auto_collect()
def start_collect(self, auto_trigger=False):
if self.worker and self.worker.is_alive():
if auto_trigger:
self.log("자동수집 건너뜀: 이미 수집 중")
else:
messagebox.showinfo("진행중", "이미 수집 중입니다.")
return
accounts = [a for a in self.accounts if a.enabled]
if not accounts:
messagebox.showwarning("확인", "사용 체크된 계정이 없습니다.")
return
self.save_accounts()
self.stop_event.clear()
self.set_progress(0, "0% 시작 준비", "현재 계정: -", "상태: 시작 준비")
# Tkinter 변수는 메인 스레드에서만 .get() 가능 → 워커 시작 전 미리 캡처해 전달
opts = {
"max_workers": 4,
"fast_mode": bool(self.fast_mode_var.get()),
"window_mode": self.window_mode.get(),
"login_mode": self.login_mode.get(),
"notify": bool(self.notify_var.get()),
}
self.worker = threading.Thread(target=self.collect_worker, args=(accounts, opts), daemon=True)
self.worker.start()
def stop_collect(self):
self.stop_event.set()
self.set_progress(self.progress_var.get(), f"{int(self.progress_var.get())}% 중지 요청", status="상태: 중지 요청됨")
self.log("중지 요청됨")
def set_progress(self, percent, text=None, current_account=None, status=None):
def update():
self.progress_var.set(max(0, min(100, percent)))
if text is not None:
self.progress_text_var.set(text)
if current_account is not None:
self.current_account_var.set(current_account)
if status is not None:
self.status_var.set(status)
self.after(0, update)
def collect_worker(self, accounts, opts):
collector = OwnerclanAlertCollector(self.log, self.stop_event, self.ask_manual_login)
found_rows = []
total = len(accounts)
completed = 0
max_workers = opts["max_workers"]
fast_mode = opts["fast_mode"]
window_mode = opts["window_mode"]
login_mode = opts["login_mode"]
self.log(f"수집 시작: {total}개 계정 / 동시 {max_workers}개 / 빠른모드 {fast_mode}")
def run_one(idx_acc):
idx, acc = idx_acc
if self.stop_event.is_set():
return idx, acc, [], "stopped", None
self.log(f"=== {idx}/{total} {acc.label} 시작 ===")
rows = collector.collect_account(
acc,
window_mode=window_mode,
login_mode=login_mode,
fast_mode=fast_mode
)
return idx, acc, rows, "ok", None
self.set_progress(0, f"0% 시작 ({total}개 계정)", "현재 계정: 준비", f"상태: 동시 {max_workers}개 수집")
if max_workers <= 1:
for i, acc in enumerate(accounts, start=1):
if self.stop_event.is_set():
break
base_percent = int(((i - 1) / total) * 100)
self.set_progress(base_percent, f"{base_percent}% 진행 중 ({i}/{total})", f"현재 계정: {acc.label}", "상태: 로그인/수집 중")
try:
_, _, rows, _, _ = run_one((i, acc))
found_rows.extend(rows)
completed += 1
done_percent = int((completed / total) * 100)
self.set_progress(done_percent, f"{done_percent}% 완료 ({completed}/{total})", f"현재 계정: {acc.label}", f"상태: 신규 {len(rows)}건 수집 완료")
except Exception as e:
completed += 1
self.log(f"[{acc.label}] 오류: {e}")
self.log(traceback.format_exc())
done_percent = int((completed / total) * 100)
self.set_progress(done_percent, f"{done_percent}% 진행 ({completed}/{total})", f"현재 계정: {acc.label}", "상태: 오류 발생, 다음 계정 진행")
else:
with ThreadPoolExecutor(max_workers=max_workers) as executor:
futures = {}
for idx, acc in enumerate(accounts, start=1):
if self.stop_event.is_set():
break
fut = executor.submit(run_one, (idx, acc))
futures[fut] = acc
for fut in as_completed(futures):
acc = futures[fut]
if self.stop_event.is_set():
break
try:
idx, acc, rows, status, err = fut.result()
found_rows.extend(rows)
completed += 1
done_percent = int((completed / total) * 100)
self.set_progress(done_percent, f"{done_percent}% 완료 ({completed}/{total})", f"최근 완료 계정: {acc.label}", f"상태: 신규 {len(rows)}건 수집 완료")
except Exception as e:
completed += 1
self.log(f"[{acc.label}] 오류: {e}")
self.log(traceback.format_exc())
done_percent = int((completed / total) * 100)
self.set_progress(done_percent, f"{done_percent}% 진행 ({completed}/{total})", f"최근 오류 계정: {acc.label}", "상태: 오류 발생")
# 기존 주문 알림 + 문의게시판 굵은 글제목은 소리+팝업으로 알림 (아직 확인 처리 안 된 것만)
order_new = [
r for r in found_rows
if is_notify_alert(r)
and r.get("alert_id") not in self.seen_ids
]
by_id = {r.get("alert_id"): r for r in self.rows}
for r in found_rows:
by_id[r.get("alert_id")] = r
self.rows = list(by_id.values())
self._save_alerts_cache()
if found_rows:
self.save_results(found_rows)
self.after(0, self.render_cards)
if not found_rows:
self.log("수집된 신규 알림이 없습니다.")
if order_new and opts["notify"]:
self.log(f"주문/문의 알림 {len(order_new)}건 → 소리+팝업 알림")
self.after(0, lambda rows=order_new: self.notify_order_alert(rows))
final_pct = int((completed / total) * 100) if total else 0
if self.stop_event.is_set():
self.set_progress(final_pct, f"{final_pct}% 중지됨 ({completed}/{total})", status="상태: 중지됨")
else:
self.set_progress(100, f"100% 완료 ({completed}/{total})", "현재 계정: 전체 완료", "상태: 수집 종료")
self.log("수집 종료")
def _unseen_rows(self):
return [r for r in self.rows if r.get("alert_id") not in self.seen_ids]
def _row_matches_filter(self, row, filter_key=None):
key = filter_key or self.alert_filter_var.get()
if key == "todo":
return False # 할일/메모 탭은 수집 알림이 아니라 메모를 보여줌(별도 렌더)
if key == "qna":
return is_qna_bold_alert(row)
if key == "memo":
return not is_qna_bold_alert(row)
return True
def _row_matches_account(self, row):
sel = self.account_filter_var.get()
if not sel or sel == ACCOUNT_FILTER_ALL:
return True
return str(row.get("계정명", "")) == sel
def _row_matches_search(self, row, keyword=None):
keyword = (keyword if keyword is not None else self.alert_search_var.get()).strip().lower()
if not keyword:
return True
haystack = " ".join([
str(row.get("계정명", "")),
str(row.get("NO", "")),
str(row.get("알림종류", "")),
str(row.get("알림타입", "")),
str(row.get("알림내용", "")),
str(row.get("등록일", "")),
str(row.get("처리상태", "")),
]).lower()
return keyword in haystack
def visible_rows(self, filter_key=None):
return [
r for r in self._unseen_rows()
if self._row_matches_filter(r, filter_key)
and self._row_matches_account(r)
and self._row_matches_search(r)
]
def on_account_filter_change(self):
# 사업자 필터 변경 → 현재 탭(알림/할일) 다시 렌더
self.render_cards()
def on_alert_search_change(self):
self.render_cards()
def clear_alert_search(self):
if self.alert_search_var.get():
self.alert_search_var.set("")
else:
self.render_cards()
def _alert_filter_counts(self):
# 사업자 필터를 반영한 미확인 알림 개수(탭 배지용)
unseen = [r for r in self._unseen_rows() if self._row_matches_account(r)]
qna_count = sum(1 for r in unseen if is_qna_bold_alert(r))
memo_count = len(unseen) - qna_count
return {
"all": len(unseen),
"memo": memo_count,
"qna": qna_count,
"todo": len(self._visible_todos()),
}
def set_alert_filter(self, filter_key):
self.alert_filter_var.set(filter_key)
self.render_cards()
def update_count_label(self):
counts = self._alert_filter_counts()
current = self.alert_filter_var.get()
if current == "todo":
self.count_label.config(text=f"할일/메모 {counts['todo']}건")
else:
current_count = len(self.visible_rows(current))
if current == "all":
current_count += len(self._todos_for_all_tab()) # 전체 탭은 할일/메모도 함께 보임
self.count_label.config(text=f"보이는 {current_count}건 / 미확인 {counts['all']}건 / 수집 {len(self.rows)}건")
# 카카오톡 알림처럼 탭에 미확인 개수를 같이 표시
labels = {"all": "전체", "memo": "신규 알림", "qna": "문의글", "todo": "할일/메모 알림"}
for key, btn in getattr(self, "alert_filter_buttons", {}).items():
count = counts.get(key, 0)
text = f"{labels[key]} {count}" if count else labels[key]
selected = key == current
if selected:
btn.configure(text=text, bg="#2f80ed", fg="white", activebackground="#1d65c1", activeforeground="white")
elif count:
btn.configure(text=text, bg="#fff0f0", fg="#d93025", activebackground="#ffe2e2", activeforeground="#d93025")
else:
btn.configure(text=text, bg="#eeeeee", fg="#555555", activebackground="#dddddd", activeforeground="#333333")
def render_cards(self, preserve_scroll=True):
# 할일/메모 알림 탭이면 수집 카드 대신 할일 메모 뷰를 그린다.
if self.alert_filter_var.get() == "todo":
self.render_todo_cards()
return
# 카드 목록 갱신 후에도 작업 중이던 스크롤 위치를 최대한 그대로 유지합니다.
scroll_pos = 0.0
try:
if preserve_scroll and hasattr(self, "cards"):
scroll_pos = float(self.cards.canvas.yview()[0])
except Exception:
scroll_pos = 0.0
for child in self.cards.inner.winfo_children():
child.destroy()
current = self.alert_filter_var.get()
visible = self.visible_rows()
# '전체' 탭에서는 수집 알림과 함께 할일/메모도 보여준다.
todos = self._todos_for_all_tab() if current == "all" else []
self.update_count_label()
def restore_scroll():
try:
self.cards.canvas.update_idletasks()
self.cards.update_scrollregion()
self.cards.canvas.yview_moveto(max(0.0, min(1.0, scroll_pos)))
except Exception:
pass
if not visible and not todos:
keyword = self.alert_search_var.get().strip()
if keyword:
msg_text = f"검색어 '{keyword}'에 해당하는 항목이 없습니다.\n다른 단어로 다시 검색해보세요."
elif current == "qna":
msg_text = "표시할 문의글이 없습니다.\n굵은 글제목 문의가 수집되면 여기에 표시됩니다."
elif current == "memo":
msg_text = "표시할 신규 알림이 없습니다.\n긴급/알림 메모가 수집되면 여기에 표시됩니다."
else:
msg_text = "표시할 항목이 없습니다.\n수집 시작을 누르거나 할일/메모 알림 탭에서 메모를 추가해보세요."
empty = tk.Label(
self.cards.inner,
text=msg_text,
font=("맑은 고딕", 13),
fg="#777777",
bg="#f6f7fb",
justify="center"
)
empty.pack(fill="both", expand=True, pady=120)
self.after_idle(restore_scroll)
return
for r in visible:
self.make_card(r)
# 전체 탭: 알림 카드 아래에 구분선 + 할일/메모 카드
if current == "all" and todos:
divider = tk.Label(
self.cards.inner, text="── 할일 / 메모 ──",
font=("맑은 고딕", 9, "bold"), fg="#1f9d55", bg="#f6f7fb",
)
divider.pack(fill="x", pady=(10, 2))
for memo in todos:
self._make_todo_card(memo)
self.after_idle(restore_scroll)
def make_card(self, r):
is_order = is_order_alert(r.get("알림종류", ""), r.get("알림타입", ""))
is_qna = is_qna_bold_alert(r)
# 색상 테마: 주문/공급사·오너클랜은 레드, 문의게시판 굵은 글제목은 보라, 그 외는 기존 파랑
if is_order:
border = "#e74c3c"
head_bg = "#fdecea"
head_fg = "#c0392b"
elif is_qna:
border = "#8e44ad"
head_bg = "#f5ecfb"
head_fg = "#6c3483"
else:
border = "#d8e0ec"
head_bg = "#f3f7ff"
head_fg = "#1f4e8c"
card = tk.Frame(
self.cards.inner, bg="#ffffff",
highlightbackground=border, highlightthickness=1, bd=0
)
card.pack(fill="x", padx=4, pady=3)
top = tk.Frame(card, bg=head_bg)
top.pack(fill="x")
title = tk.Label(
top,
text=f"● {r.get('계정명','')} NO.{r.get('NO','')} [{r.get('알림종류','')}/{r.get('알림타입','')}]",
bg=head_bg,
fg=head_fg,
font=("맑은 고딕", 10, "bold"),
anchor="w"
)
title.pack(side="left", fill="x", expand=True, padx=10, pady=4)
date_label = tk.Label(top, text=r.get("등록일", ""), bg=head_bg, fg="#555555", font=("맑은 고딕", 9))
date_label.pack(side="right", padx=8)
# 알림 내용은 드래그/복사 가능하도록 Text 위젯 사용. 기본 2줄로 접고 더보기로 펼침.
msg = tk.Text(
card,
height=2,
bg="#ffffff",
fg="#222222",
font=("맑은 고딕", 11),
wrap="word",
relief="flat",
bd=0,
padx=0,
pady=0
)
msg.insert("1.0", r.get("알림내용", ""))
msg.configure(state="disabled")
msg.pack(fill="x", padx=12, pady=(5, 2))
# 실제 줄 수 측정 (wrap 너비 확정 후라야 정확)
try:
msg.update_idletasks()
line_count = int(msg.count("1.0", "end", "displaylines")[0])
except Exception:
# 폴백: 글자수 기반 추정
line_count = max(1, len(r.get("알림내용", "")) // 40 + 1)
full_height = max(2, min(8, line_count))
if line_count > 2:
toggle_row = tk.Frame(card, bg="#ffffff")
toggle_row.pack(fill="x", padx=12, pady=(0, 2))
state = {"expanded": False}
def toggle():
if state["expanded"]:
msg.configure(height=2)
toggle_btn.configure(text="더보기 ▼")
state["expanded"] = False
else:
msg.configure(height=full_height)
toggle_btn.configure(text="접기 ▲")
state["expanded"] = True
toggle_btn = tk.Label(
toggle_row, text="더보기 ▼", fg=head_fg, bg="#ffffff",
font=("맑은 고딕", 9, "underline"), cursor="hand2"
)
toggle_btn.pack(side="left")
toggle_btn.bind("", lambda e: toggle())
bottom = tk.Frame(card, bg="#ffffff")
bottom.pack(fill="x", padx=10, pady=(0, 6))
meta = tk.Label(
bottom,
text=f"처리상태: {r.get('처리상태','')} | 수집시간: {r.get('수집시간','')}",
bg="#ffffff",
fg="#777777",
font=("맑은 고딕", 8),
anchor="w"
)
meta.pack(side="left", fill="x", expand=True)
btn = tk.Button(
bottom,
text="확인",
bg="#2f80ed",
fg="white",
activebackground="#1d65c1",
activeforeground="white",
relief="flat",
padx=10,
pady=2,
command=lambda row=r: self.confirm_alert(row)
)
btn.pack(side="right")
open_btn = tk.Button(
bottom,
text="열기",
bg="#555555",
fg="white",
activebackground="#333333",
activeforeground="white",
relief="flat",
padx=10,
pady=2,
command=lambda row=r: self.open_alert_page(row)
)
open_btn.pack(side="right", padx=(0, 6))
def open_alert_page(self, row):
"""
열기 버튼:
해당 알림 계정으로 로그인된 보이는 크롬창을 열고
오너클랜 알림/긴급 메모 페이지까지 보여준다.
"""
acc = self.find_account_for_row(row)
if not acc:
messagebox.showerror("오류", "이 알림에 해당하는 계정 설정을 찾지 못했습니다.")
return
# Tk 변수는 메인 스레드에서만 접근 가능 → 워커 시작 전 캡처
login_mode = self.login_mode.get()
cur_pct = self.progress_var.get()
def worker():
self.set_progress(cur_pct, f"{int(cur_pct)}% 알림 페이지 열기",
f"현재 계정: {acc.label}", "상태: 보이는 크롬창 열기")
try:
collector = OwnerclanAlertCollector(self.log, self.stop_event, self.ask_manual_login)
# '열기'는 사용자가 직접 보는 용도 → 빠른수집 설정과 무관하게 이미지 표시
if is_qna_bold_alert(row):
driver = collector.open_qna_visible(acc, row, login_mode=login_mode, fast_mode=False)
done_status = "상태: 문의글 열림"
else:
driver = collector.open_memo_visible(acc, login_mode=login_mode, fast_mode=False)
done_status = "상태: 알림 페이지 열림"
self.open_drivers.append(driver)
self.set_progress(cur_pct, f"{int(cur_pct)}% 알림 페이지 열기 완료",
f"현재 계정: {acc.label}", done_status)
except Exception as e:
self.log(f"[{acc.label}] 알림 페이지 열기 오류: {e}")
self.log(traceback.format_exc())
self.set_progress(cur_pct, f"{int(cur_pct)}% 알림 페이지 열기 실패",
f"현재 계정: {acc.label}", "상태: 알림 페이지 열기 실패")
self.after(0, lambda: messagebox.showerror("알림 페이지 열기 실패", str(e)))
threading.Thread(target=worker, daemon=True).start()
def find_account_for_row(self, row):
uid = row.get("아이디", "")
label = row.get("계정명", "")
for acc in self.accounts:
if acc.user_id == uid or acc.label == label:
return acc
return None
def confirm_alert(self, row):
"""
확인 버튼:
1) 해당 계정으로 백그라운드 로그인
2) 오너클랜 메모 페이지에서 해당 행의 클릭 후 처리 클릭
3) 성공하면 프로그램 화면에서도 확인 처리
"""
alert_id = row.get("alert_id")
acc = self.find_account_for_row(row)
if not acc:
messagebox.showerror("오류", "이 알림에 해당하는 계정 설정을 찾지 못했습니다.")
return
if is_qna_bold_alert(row):
confirm_msg = f"[{acc.label}] NO.{row.get('NO','')} 문의글을 열어서 볼드체를 해제하고 목록에서 지울까요?"
else:
confirm_msg = f"[{acc.label}] NO.{row.get('NO','')} 알림을 오너클랜에서도 확인 처리할까요?"
if not messagebox.askyesno("확인", confirm_msg):
return
# Tk 변수는 메인 스레드에서만 접근 가능 → 워커 시작 전 캡처
window_mode = self.window_mode.get()
login_mode = self.login_mode.get()
fast_mode = bool(self.fast_mode_var.get())
cur_pct = self.progress_var.get()
def worker():
self.set_progress(cur_pct, f"{int(cur_pct)}% 확인 처리 중",
f"현재 계정: {acc.label}", f"상태: NO.{row.get('NO','')} 클릭 후 처리 중")
try:
collector = OwnerclanAlertCollector(self.log, self.stop_event, self.ask_manual_login)
if is_qna_bold_alert(row):
collector.confirm_qna_on_site(acc, row, window_mode=window_mode,
login_mode=login_mode, fast_mode=fast_mode)
else:
collector.confirm_alert_on_site(acc, row, window_mode=window_mode,
login_mode=login_mode, fast_mode=fast_mode)
self.seen_ids.add(alert_id)
save_json(SEEN_PATH, sorted(self.seen_ids))
self.after(0, self.render_cards)
self.set_progress(cur_pct, f"{int(cur_pct)}% 확인 처리 완료",
f"현재 계정: {acc.label}", f"상태: NO.{row.get('NO','')} 확인 완료")
except Exception as e:
self.log(f"[{acc.label}] 확인 처리 오류: {e}")
self.log(traceback.format_exc())
self.set_progress(cur_pct, f"{int(cur_pct)}% 확인 처리 실패",
f"현재 계정: {acc.label}", "상태: 확인 처리 실패")
self.after(0, lambda: messagebox.showerror("확인 처리 실패", str(e)))
threading.Thread(target=worker, daemon=True).start()
# ---------------- 주문 알림: 소리 + 우하단 팝업 ----------------
def notify_order_alert(self, rows):
self.play_alert_sound()
n = len(rows)
first = rows[0]
acc = first.get("계정명", "")
summary = (first.get("알림내용", "") or "")[:40]
text = f"🔔 주문/문의 신규 {n}건\n[{acc}] {summary}"
self.show_toast(text)
def play_alert_sound(self):
if self._sound_playing:
return
self._sound_playing = True
def run():
try:
self._beep_sequence()
except Exception:
try:
self._play_wav_fallback()
except Exception:
pass
finally:
self._sound_playing = False
threading.Thread(target=run, daemon=True).start()
def _beep_sequence(self):
# 2.5초 내외의 독특한 패턴 (윈도우 기본음과 구분). 0은 쉼.
if not winsound:
raise RuntimeError("winsound 사용 불가")
seq = [(880, 250), (0, 80), (880, 250), (0, 80), (1175, 500),
(0, 120), (988, 400), (0, 100), (1318, 650)]
for freq, dur in seq:
if freq <= 0:
time.sleep(dur / 1000.0)
else:
winsound.Beep(freq, dur)
def _play_wav_fallback(self):
if not winsound:
return
path = APP_DIR / "alert.wav"
if not path.exists():
self._generate_wav(path)
winsound.PlaySound(str(path), winsound.SND_FILENAME | winsound.SND_ASYNC)
def _generate_wav(self, path):
framerate = 44100
notes = [(880, 0.4), (1175, 0.4), (988, 0.4), (1318, 0.7)]
frames = bytearray()
for freq, dur in notes:
n = int(framerate * dur)
for i in range(n):
env = min(1.0, i / 500.0, (n - i) / 500.0) # 페이드로 클릭음 방지
val = int(32767 * 0.5 * env * math.sin(2 * math.pi * freq * i / framerate))
frames += struct.pack("= max_toasts:
oldest = self._toasts[0]
try:
self._toasts.remove(oldest)
except ValueError:
pass
try:
oldest.destroy()
except Exception:
pass
toast = tk.Toplevel(self)
toast.overrideredirect(True)
toast.attributes("-topmost", True)
w, h = 330, 92
sw = self.winfo_screenwidth()
sh = self.winfo_screenheight()
idx = len(self._toasts)
margin = 20
x = sw - w - margin
# 작업표시줄 여유 + 위로 스택. 화면 위로 넘어가지 않도록 하한 clamp.
y = max(20, sh - h - margin - 40 - idx * (h + 10))
toast.geometry(f"{w}x{h}+{x}+{y}")
# accent: 강조(테두리/글자)색. 주문/문의=빨강(기본), 할일/메모=초록.
border_c = accent or "#c0392b"
body_bg = {"#c0392b": "#fdecea", "#1f9d55": "#e9f7ef"}.get(border_c, "#eef4ff")
outer = tk.Frame(toast, bg=border_c)
outer.pack(fill="both", expand=True)
inner = tk.Frame(outer, bg=body_bg)
inner.pack(fill="both", expand=True, padx=2, pady=2)
lbl = tk.Label(
inner, text=text, bg=body_bg, fg=border_c,
font=("맑은 고딕", 11, "bold"), justify="left", anchor="w",
wraplength=w - 40 # ✕ 닫기 버튼 자리 확보
)
lbl.pack(fill="both", expand=True, padx=12, pady=10)
# ✕ 닫기 버튼(우상단 오버레이). manual 모드에서 이 버튼/팝업 클릭으로 닫음.
close_btn = tk.Label(inner, text="✕", bg=body_bg, fg=border_c,
font=("맑은 고딕", 11, "bold"), cursor="hand2")
close_btn.place(relx=1.0, x=-6, y=2, anchor="ne")
self._toasts.append(toast)
toast._dedup_key = key # 중복 팝업 방지용 키(현재 떠있는 토스트와 비교)
def close():
try:
self._toasts.remove(toast)
except ValueError:
pass
try:
toast.destroy()
except Exception:
pass
def open_main(_event=None):
# 팝업 본문 클릭 → 프로그램 창을 앞으로(트레이/최소화/백그라운드 모두), 팝업은 닫음.
try:
self._restore_from_tray()
except Exception:
pass
close()
# 본문(toast/outer/inner/lbl) 클릭 → 창 열기 + 닫기. ✕ 버튼은 닫기만.
for wdg in (toast, outer, inner, lbl):
wdg.bind("", open_main)
close_btn.bind("", lambda e: close())
# auto 모드만 5초 후 자동 닫힘. manual 모드는 ✕ 버튼/팝업 클릭으로만 닫힘.
if self.popup_mode_var.get() != "manual":
toast.after(5000, close)
except Exception:
pass
def confirm_all_visible(self):
visible = list(self.visible_rows())
if not visible:
messagebox.showinfo("안내", "현재 보이는 알림이나 문의글이 없습니다.")
return
if not messagebox.askyesno("확인", f"현재 보이는 항목 {len(visible)}건을 모두 오너클랜에서 확인 처리할까요?"):
return
window_mode = self.window_mode.get()
login_mode = self.login_mode.get()
fast_mode = bool(self.fast_mode_var.get())
def worker():
total = len(visible)
success = 0
failed = 0
for idx, row in enumerate(visible, 1):
alert_id = row.get("alert_id")
acc = self.find_account_for_row(row)
no = row.get("NO", "")
title = row.get("알림내용", "")
pct = ((idx - 1) / total) * 100 if total else 0
if not acc:
failed += 1
self.log(f"[전체 확인] NO.{no} 계정 설정을 찾지 못해 건너뜁니다.")
continue
self.set_progress(pct, f"{int(pct)}% 전체 확인 처리 중",
f"현재 계정: {acc.label}", f"상태: {idx}/{total} NO.{no} 처리 중")
try:
collector = OwnerclanAlertCollector(self.log, self.stop_event, self.ask_manual_login)
if is_qna_bold_alert(row):
collector.confirm_qna_on_site(acc, row, window_mode=window_mode,
login_mode=login_mode, fast_mode=fast_mode)
else:
collector.confirm_alert_on_site(acc, row, window_mode=window_mode,
login_mode=login_mode, fast_mode=fast_mode)
self.seen_ids.add(alert_id)
save_json(SEEN_PATH, sorted(self.seen_ids))
success += 1
self.log(f"[전체 확인] 완료: [{acc.label}] NO.{no} {title}")
except Exception as e:
failed += 1
self.log(f"[전체 확인] 실패: [{acc.label}] NO.{no} {e}")
self.log(traceback.format_exc())
self.after(0, self.render_cards)
final_msg = f"성공 {success}건"
if failed:
final_msg += f" / 실패 {failed}건"
self.set_progress(100, "100% 전체 확인 완료", "현재 계정: 전체 완료", f"상태: {final_msg}")
if failed:
self.after(0, lambda: messagebox.showwarning("전체 확인 완료", f"전체 확인이 끝났습니다.\n{final_msg}"))
else:
self.after(0, lambda: messagebox.showinfo("전체 확인 완료", f"전체 확인이 끝났습니다.\n{final_msg}"))
threading.Thread(target=worker, daemon=True).start()
def mark_seen(self, alert_id):
if not alert_id:
return
self.seen_ids.add(alert_id)
save_json(SEEN_PATH, sorted(self.seen_ids))
self.render_cards()
def mark_all_seen(self):
visible = self.visible_rows()
if not visible:
return
if not messagebox.askyesno("확인", f"현재 보이는 알림 {len(visible)}건을 모두 확인 처리할까요?"):
return
for r in visible:
self.seen_ids.add(r.get("alert_id"))
save_json(SEEN_PATH, sorted(self.seen_ids))
self.render_cards()
def clear_seen(self):
if not messagebox.askyesno("확인", "확인 처리한 알림 기록을 초기화할까요?"):
return
self.seen_ids.clear()
save_json(SEEN_PATH, [])
self.render_cards()
def save_results(self, rows):
ts = datetime.now().strftime("%Y%m%d_%H%M%S")
csv_path = OUTPUT_DIR / f"ownerclan_new_alerts_{ts}.csv"
headers = ["계정명", "아이디", "NO", "알림내용", "알림종류", "알림타입", "처리상태", "등록일", "처리일", "수집시간", "source_type", "detail_url"]
with csv_path.open("w", newline="", encoding="utf-8-sig") as f:
writer = csv.DictWriter(f, fieldnames=headers, extrasaction="ignore")
writer.writeheader()
writer.writerows(rows)
if Workbook:
xlsx_path = OUTPUT_DIR / f"ownerclan_new_alerts_{ts}.xlsx"
wb = Workbook()
ws = wb.active
ws.title = "신규알림"
ws.append(headers)
for r in rows:
ws.append([r.get(h, "") for h in headers])
for col, width in {"A":16,"B":18,"C":8,"D":80,"E":12,"F":12,"G":12,"H":22,"I":22,"J":22}.items():
ws.column_dimensions[col].width = width
wb.save(xlsx_path)
self.log(f"백업 저장 완료: {csv_path.name}")
# ---------------- 할일/메모 알림 ----------------
@staticmethod
def _parse_dt(s):
if not s:
return None
for fmt in ("%Y-%m-%dT%H:%M:%S", "%Y-%m-%dT%H:%M",
"%Y-%m-%d %H:%M:%S", "%Y-%m-%d %H:%M"):
try:
return datetime.strptime(s, fmt)
except Exception:
continue
try:
return datetime.fromisoformat(s)
except Exception:
return None
def save_todos(self):
try:
save_json(TODO_PATH, self.todo_memos)
except Exception as e:
self.log(f"할일 저장 실패: {e}")
def _todo_account_display(self, memo):
acc = (memo.get("account") or "").strip()
return acc if acc else TODO_COMMON_LABEL
def _visible_todos(self):
sel = self.account_filter_var.get()
if not sel or sel == ACCOUNT_FILTER_ALL:
return list(self.todo_memos)
# 특정 사업자 선택 시 그 사업자 메모만 (공통(빈값)은 전체에서만 노출 — 알림 카드 규칙과 동일)
return [m for m in self.todo_memos if (m.get("account") or "").strip() == sel]
def _todos_for_all_tab(self):
"""'전체' 탭에 함께 보일 할일/메모(사업자 필터 + 검색어 반영)."""
todos = self._visible_todos()
kw = self.alert_search_var.get().strip().lower()
if kw:
todos = [m for m in todos
if kw in (str(m.get("text", "")) + " " + self._todo_account_display(m)).lower()]
return todos
def _current_collect_minutes(self):
# 스케줄러/표시용: get_auto_collect_minutes()의 set() 부작용을 피해 값만 읽는다.
try:
m = int(self.auto_collect_minutes_var.get())
except Exception:
m = 10
return max(1, min(1440, m))
def _todo_interval_minutes(self, memo):
mode = memo.get("mode", "default")
if mode == "default":
return self._current_collect_minutes()
if mode == "interval":
try:
return max(1, int(memo.get("interval_minutes", 30)))
except Exception:
return 30
return None # datetime
def _todo_mode_text(self, memo):
mode = memo.get("mode", "default")
if mode == "default":
return f"수집 간격마다({self._current_collect_minutes()}분)"
if mode == "interval":
return f"{self._todo_interval_minutes(memo)}분마다"
fa = self._parse_dt(memo.get("fire_at"))
return (fa.strftime("%Y-%m-%d %H:%M") + " 1회") if fa else "일시 미설정"
def _todo_next_fire_text(self, memo):
if not memo.get("enabled", True):
return "사용 안 함"
now = datetime.now()
if memo.get("mode", "default") == "datetime":
fa = self._parse_dt(memo.get("fire_at"))
if fa is None:
return "일시 미설정"
if memo.get("last_fired"):
return "알림 완료(1회)"
return "곧 알림(지난 일시)" if fa <= now else fa.strftime("%Y-%m-%d %H:%M")
interval = self._todo_interval_minutes(memo)
last = self._parse_dt(memo.get("last_fired")) or self._parse_dt(memo.get("created")) or now
nxt = last + timedelta(minutes=interval)
return "곧 알림" if nxt <= now else nxt.strftime("%m-%d %H:%M")
def render_todo_cards(self):
for child in self.cards.inner.winfo_children():
child.destroy()
self.update_count_label()
topbar = tk.Frame(self.cards.inner, bg="#f6f7fb")
topbar.pack(fill="x", padx=4, pady=(2, 6))
tk.Button(
topbar, text="+ 할일 추가", bg="#1f9d55", fg="white",
activebackground="#167a40", activeforeground="white",
relief="flat", padx=12, pady=5, font=("맑은 고딕", 10, "bold"),
command=self.add_todo,
).pack(side="left")
sel = self.account_filter_var.get()
scope = sel if (sel and sel != ACCOUNT_FILTER_ALL) else "전체"
tk.Label(topbar, text=f" 사업자: {scope}", bg="#f6f7fb", fg="#666666",
font=("맑은 고딕", 9)).pack(side="left")
memos = self._visible_todos()
if not memos:
tk.Label(
self.cards.inner,
text="등록된 할일/메모가 없습니다.\n위 '+ 할일 추가'로 새 알림 메모를 만들어보세요.",
font=("맑은 고딕", 13), fg="#777777", bg="#f6f7fb", justify="center",
).pack(fill="both", expand=True, pady=100)
return
for memo in memos:
self._make_todo_card(memo)
def _make_todo_card(self, memo):
enabled = memo.get("enabled", True)
border = "#1f9d55" if enabled else "#bbbbbb"
head_bg = "#e9f7ef" if enabled else "#eeeeee"
head_fg = "#167a40" if enabled else "#888888"
card = tk.Frame(self.cards.inner, bg="#ffffff",
highlightbackground=border, highlightthickness=1, bd=0)
card.pack(fill="x", padx=4, pady=3)
top = tk.Frame(card, bg=head_bg)
top.pack(fill="x")
tk.Label(top, text=f"📝 [{self._todo_account_display(memo)}] {self._todo_mode_text(memo)}",
bg=head_bg, fg=head_fg, font=("맑은 고딕", 10, "bold"),
anchor="w").pack(side="left", fill="x", expand=True, padx=10, pady=4)
tk.Label(top, text=f"다음: {self._todo_next_fire_text(memo)}", bg=head_bg, fg="#555555",
font=("맑은 고딕", 9)).pack(side="right", padx=8)
tk.Label(card, text=memo.get("text", ""), bg="#ffffff", fg="#222222",
font=("맑은 고딕", 11), justify="left", anchor="w",
wraplength=560).pack(fill="x", padx=12, pady=(5, 4))
bottom = tk.Frame(card, bg="#ffffff")
bottom.pack(fill="x", padx=10, pady=(0, 6))
mid = memo.get("id")
tk.Button(bottom, text=("끄기" if enabled else "켜기"), bg="#888888", fg="white",
activebackground="#666666", activeforeground="white",
relief="flat", padx=10, pady=2,
command=lambda m=mid: self.toggle_todo(m)).pack(side="left")
tk.Button(bottom, text="삭제", bg="#e74c3c", fg="white",
activebackground="#c0392b", activeforeground="white",
relief="flat", padx=10, pady=2,
command=lambda m=mid: self.delete_todo(m)).pack(side="right")
tk.Button(bottom, text="수정", bg="#2f80ed", fg="white",
activebackground="#1d65c1", activeforeground="white",
relief="flat", padx=10, pady=2,
command=lambda m=mid: self.edit_todo(m)).pack(side="right", padx=(0, 6))
def _find_todo(self, mid):
for m in self.todo_memos:
if m.get("id") == mid:
return m
return None
def add_todo(self):
default_acc = self.account_filter_var.get()
if default_acc == ACCOUNT_FILTER_ALL:
default_acc = "" # 공통
TodoMemoDialog(
self, memo=None,
account_labels=[a.label for a in self.accounts],
default_account=default_acc,
collect_minutes=self._current_collect_minutes(),
on_save=self._save_todo_from_dialog,
)
def edit_todo(self, mid):
memo = self._find_todo(mid)
if not memo:
return
TodoMemoDialog(
self, memo=memo,
account_labels=[a.label for a in self.accounts],
default_account=memo.get("account", ""),
collect_minutes=self._current_collect_minutes(),
on_save=self._save_todo_from_dialog,
)
def _save_todo_from_dialog(self, data, memo_id):
if memo_id:
memo = self._find_todo(memo_id)
if memo:
memo.update(data)
else:
now_iso = datetime.now().isoformat(timespec="seconds")
new = {
"id": datetime.now().strftime("%Y%m%d%H%M%S%f"),
"created": now_iso,
"last_fired": None,
"enabled": True,
}
new.update(data)
# 반복형은 추가 직후 즉시 안 뜨도록 last_fired를 생성시각으로 둔다.
if new.get("mode") in ("default", "interval"):
new["last_fired"] = now_iso
self.todo_memos.append(new)
self.save_todos()
self.render_cards()
def delete_todo(self, mid):
if not self._find_todo(mid):
return
if not messagebox.askyesno("삭제", "이 할일/메모를 삭제할까요?"):
return
self.todo_memos = [m for m in self.todo_memos if m.get("id") != mid]
self.save_todos()
self.render_cards()
def toggle_todo(self, mid):
memo = self._find_todo(mid)
if not memo:
return
memo["enabled"] = not memo.get("enabled", True)
# 다시 켤 때 반복형은 즉시 안 뜨도록 last_fired 갱신
if memo["enabled"] and memo.get("mode") in ("default", "interval"):
memo["last_fired"] = datetime.now().isoformat(timespec="seconds")
self.save_todos()
self.render_cards()
def check_todo_reminders(self):
try:
now = datetime.now()
changed = False
for memo in self.todo_memos:
try:
if not memo.get("enabled", True):
continue
if self._todo_is_due(memo, now):
self._fire_todo(memo, now)
changed = True
except Exception as e:
self.log(f"할일 알림 처리 오류: {e}")
if changed:
self.save_todos()
if self.alert_filter_var.get() == "todo":
self.render_cards()
finally:
# 어떤 경우에도 다음 주기 재예약 (스케줄러가 멈추지 않도록)
self.todo_job = self.after(20000, self.check_todo_reminders)
def _todo_is_due(self, memo, now):
if memo.get("mode", "default") == "datetime":
if memo.get("last_fired"):
return False # 1회만
fa = self._parse_dt(memo.get("fire_at"))
return fa is not None and now >= fa
interval = self._todo_interval_minutes(memo)
last = self._parse_dt(memo.get("last_fired")) or self._parse_dt(memo.get("created")) or now
return now >= last + timedelta(minutes=interval)
def _fire_todo(self, memo, now):
acc = self._todo_account_display(memo)
text = f"📝 할일 [{acc}]\n{memo.get('text','')}"
self.show_toast(text, accent="#1f9d55", key=f"todo:{memo.get('id')}")
self.play_alert_sound()
memo["last_fired"] = now.isoformat(timespec="seconds")
self.log(f"할일 알림: [{acc}] {memo.get('text','')[:30]}")
class TodoMemoDialog(tk.Toplevel):
"""할일/메모 알림 추가·수정 다이얼로그."""
def __init__(self, master, memo, account_labels, default_account, collect_minutes, on_save):
super().__init__(master)
self.on_save = on_save
self.memo_id = memo.get("id") if memo else None
self.collect_minutes = collect_minutes
self.title("할일/메모 알림 " + ("수정" if memo else "추가"))
self.resizable(False, False)
try:
self.transient(master)
self.grab_set()
except Exception:
pass
pad = {"padx": 10, "pady": 4}
# 메모 내용
ttk.Label(self, text="메모 내용").grid(row=0, column=0, sticky="nw", **pad)
self.text_widget = tk.Text(self, width=42, height=4, wrap="word", font=("맑은 고딕", 10))
self.text_widget.grid(row=0, column=1, sticky="we", **pad)
if memo:
self.text_widget.insert("1.0", memo.get("text", ""))
# 사업자
ttk.Label(self, text="사업자").grid(row=1, column=0, sticky="w", **pad)
self.account_combo = ttk.Combobox(self, state="readonly",
values=[TODO_COMMON_LABEL] + list(account_labels))
self.account_combo.grid(row=1, column=1, sticky="we", **pad)
cur_acc = (default_account or "").strip()
self.account_combo.set(cur_acc if cur_acc in account_labels else TODO_COMMON_LABEL)
# 알림 방식
box = ttk.LabelFrame(self, text="알림 방식")
box.grid(row=2, column=0, columnspan=2, sticky="we", padx=10, pady=(8, 4))
self.mode_main = tk.StringVar(value="default")
self.custom_kind = tk.StringVar(value="interval")
ttk.Radiobutton(box, text=f"기본 — 수집 간격마다 반복 (현재 {collect_minutes}분)",
variable=self.mode_main, value="default").grid(row=0, column=0, columnspan=3, sticky="w", padx=8, pady=2)
ttk.Radiobutton(box, text="직접 입력", variable=self.mode_main, value="custom").grid(
row=1, column=0, columnspan=3, sticky="w", padx=8, pady=(4, 0))
# 직접입력 - 간격
ttk.Radiobutton(box, text="간격마다 반복", variable=self.custom_kind, value="interval").grid(
row=2, column=0, sticky="w", padx=(28, 4), pady=2)
self.interval_spin = ttk.Spinbox(box, from_=1, to=1440, width=6)
self.interval_spin.set(int(memo.get("interval_minutes", 30)) if memo and memo.get("mode") == "interval" else 30)
self.interval_spin.grid(row=2, column=1, sticky="w")
ttk.Label(box, text="분마다").grid(row=2, column=2, sticky="w")
# 직접입력 - 특정 일시
ttk.Radiobutton(box, text="특정 일시 (1회)", variable=self.custom_kind, value="datetime").grid(
row=3, column=0, sticky="w", padx=(28, 4), pady=2)
dtrow = ttk.Frame(box)
dtrow.grid(row=3, column=1, columnspan=2, sticky="w")
now = datetime.now()
init_dt = None
if memo and memo.get("mode") == "datetime":
init_dt = App._parse_dt(memo.get("fire_at"))
if init_dt is None:
init_dt = now + timedelta(hours=1)
if DateEntry is not None:
try:
self.date_widget = DateEntry(dtrow, width=12, date_pattern="yyyy-MM-dd",
year=init_dt.year, month=init_dt.month, day=init_dt.day)
except Exception:
self.date_widget = None
else:
self.date_widget = None
if self.date_widget is None:
# 폴백: 텍스트 입력 (YYYY-MM-DD)
self.date_widget = ttk.Entry(dtrow, width=12)
self.date_widget.insert(0, init_dt.strftime("%Y-%m-%d"))
self.date_widget.pack(side="left", padx=(0, 4))
self.hour_combo = ttk.Combobox(dtrow, state="readonly", width=4,
values=[f"{h:02d}" for h in range(24)])
self.hour_combo.set(f"{init_dt.hour:02d}")
self.hour_combo.pack(side="left")
ttk.Label(dtrow, text="시").pack(side="left", padx=(2, 6))
self.minute_combo = ttk.Combobox(dtrow, state="readonly", width=4,
values=[f"{m:02d}" for m in range(60)])
self.minute_combo.set(f"{init_dt.minute:02d}")
self.minute_combo.pack(side="left")
ttk.Label(dtrow, text="분").pack(side="left", padx=(2, 0))
# 편집 모드 초기 선택 반영
if memo:
mode = memo.get("mode", "default")
if mode == "default":
self.mode_main.set("default")
else:
self.mode_main.set("custom")
self.custom_kind.set("datetime" if mode == "datetime" else "interval")
# 버튼
btns = ttk.Frame(self)
btns.grid(row=3, column=0, columnspan=2, sticky="e", padx=10, pady=(8, 10))
ttk.Button(btns, text="취소", command=self.destroy).pack(side="right", padx=(6, 0))
ttk.Button(btns, text="저장", command=self._save).pack(side="right")
self.columnconfigure(1, weight=1)
def _save(self):
text = self.text_widget.get("1.0", "end").strip()
if not text:
messagebox.showwarning("확인", "메모 내용을 입력하세요.", parent=self)
return
acc_disp = self.account_combo.get()
account = "" if acc_disp == TODO_COMMON_LABEL else acc_disp
data = {"text": text, "account": account}
if self.mode_main.get() == "default":
data["mode"] = "default"
elif self.custom_kind.get() == "interval":
try:
iv = max(1, min(1440, int(self.interval_spin.get())))
except Exception:
messagebox.showwarning("확인", "간격(분)을 숫자로 입력하세요.", parent=self)
return
data["mode"] = "interval"
data["interval_minutes"] = iv
else:
# 특정 일시
try:
if hasattr(self.date_widget, "get_date"):
d = self.date_widget.get_date()
date_str = d.strftime("%Y-%m-%d")
else:
date_str = self.date_widget.get().strip()
datetime.strptime(date_str, "%Y-%m-%d") # 형식 검증
hh = self.hour_combo.get() or "00"
mm = self.minute_combo.get() or "00"
fire_at = f"{date_str}T{hh}:{mm}"
datetime.strptime(fire_at, "%Y-%m-%dT%H:%M") # 최종 검증
except Exception:
messagebox.showwarning("확인", "날짜/시간을 올바르게 입력하세요. (예: 2026-06-22)", parent=self)
return
data["mode"] = "datetime"
data["fire_at"] = fire_at
try:
self.on_save(data, self.memo_id)
finally:
self.destroy()
if __name__ == "__main__":
# pythonw(콘솔 숨김)로 실행되므로 기동 실패 시 조용히 죽지 않도록
# 예외를 error.log + messagebox로 남긴다.
try:
startup_update_flow()
except Exception:
_update_log("startup_update_flow 오류:\n" + traceback.format_exc())
try:
app = App()
app.mainloop()
except Exception:
err = traceback.format_exc()
try:
(APP_DIR / "error.log").write_text(
f"{datetime.now()}\n{err}\n", encoding="utf-8"
)
except Exception:
pass
try:
import tkinter.messagebox as mb
root = tk.Tk()
root.withdraw()
mb.showerror("실행 오류", f"프로그램을 시작할 수 없습니다.\n\n{err}\n\nerror.log를 확인하세요.")
root.destroy()
except Exception:
pass
sys.exit(1)
| |