YJKIM/WEHAGO_DB
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
3500d6c70f1168600b24c70827c3574830acc119
WEHAGO_DB/wehago_data_download_260421.py
T

1732 lines
67 KiB
Python
Executable File
Raw Blame History

"""
WEHAGO 계정별원장 엑셀 자동 다운로드
사용 흐름
1. 이 파일을 실행한다.
2. 자동으로 열린 Chrome에서 WEHAGO에 직접 로그인한다.
3. 계정별원장 화면까지 이동한 뒤 콘솔에서 Enter를 누른다.
4. 이후 왼쪽 계정 목록에서 계정을 선택하고, 우클릭 메뉴의 엑셀 다운로드와 파일명 변경을 자동 처리한다.
주의
- WEHAGO 화면의 HTML 구조는 업데이트될 수 있습니다.
- 처음 실행에서 요소를 찾지 못하면 아래 CONFIG 영역의 선택자를 README 안내에 따라 조정하세요.
"""
from __future__ import annotations
import argparse
import csv
import re
import subprocess
import sys
import time
import zipfile
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
from typing import Iterable
from xml.etree import ElementTree as ET
from openpyxl import Workbook, load_workbook
from openpyxl.styles import Alignment, Font, PatternFill
from selenium import webdriver
from selenium.common.exceptions import (
ElementClickInterceptedException,
NoSuchElementException,
TimeoutException,
WebDriverException,
)
from selenium.webdriver import ChromeOptions
from selenium.webdriver.chrome.service import Service
from selenium.webdriver.common.action_chains import ActionChains
from selenium.webdriver.common.by import By
from selenium.webdriver.common.keys import Keys
from selenium.webdriver.remote.webdriver import WebDriver
from selenium.webdriver.remote.webelement import WebElement
from webdriver_manager.chrome import ChromeDriverManager
# =============================================================================
# CONFIG: 이 영역만 상황에 맞게 수정하면 됩니다.
# =============================================================================
BASE_DIR = Path(__file__).resolve().parent
WEHAGO_START_URL = "https://www.wehago.com/"
# 계정별원장 화면 URL입니다.
# WEHAGO 내부 URL은 회사/기수/사용자 세션 쿼리값이 붙어야 정상 동작하는 경우가 많습니다.
# 잘못된 고정 URL을 넣으면 404가 나므로, 기본값은 비워둡니다.
ACCOUNT_LEDGER_URL = ""
LEDGER_URL_FILE = BASE_DIR / "wehago_account_ledger_url.txt"
USE_SAVED_LEDGER_URL = False
# Chrome 로그인 정보를 저장할 자동화 전용 프로필입니다.
# 첫 로그인 뒤에는 같은 프로필을 계속 사용합니다.
CHROME_USER_DATA_DIR = BASE_DIR / ".chrome-wehago-profile"
CHROME_PROFILE_NAME = "Default"
# 다운로드 폴더입니다. 결과 파일은 이 폴더 안에 계정코드_계정명.xlsx 형태로 저장됩니다.
DATA_DOWNLOAD_DIR = BASE_DIR / "data_download"
DOWNLOAD_DIR = DATA_DOWNLOAD_DIR / f"wehago_account_ledger_{datetime.now():%Y%m%d}"
DIAGNOSTIC_DIR = DATA_DOWNLOAD_DIR / "diagnostics"
FALLBACK_DOWNLOAD_DIR = Path.home() / "Downloads"
# 왼쪽 계정 목록 그리드 후보입니다.
ACCOUNT_LIST_SELECTORS = [
".realgrid",
".rg-root",
".rg-grid",
".grid-container",
".grid_container",
".gridWrap",
".grid-wrap",
"[class*='realgrid']",
"[class*='grid']",
]
# 우클릭할 계정별원장 상세 그리드 후보입니다. 오른쪽의 큰 그리드를 우선 선택합니다.
GRID_SELECTORS = [
".realgrid",
".rg-root",
".rg-grid",
".grid-container",
".grid_container",
".gridWrap",
".grid-wrap",
"[class*='realgrid']",
"[class*='grid']",
]
# 우클릭 메뉴에서 엑셀 다운로드 항목 후보입니다.
EXCEL_MENU_XPATHS = [
"//*[@id='context:엑셀변환']",
"//*[contains(normalize-space(.), '엑셀')]",
"//*[contains(normalize-space(.), 'Excel')]",
"//*[contains(normalize-space(.), 'EXCEL')]",
"//*[contains(normalize-space(.), '다운로드')]",
]
EXCEL_MENU_TEXTS = ["엑셀변환", "엑셀", "Excel", "EXCEL", "다운로드"]
WAIT_SECONDS = 20
DOWNLOAD_TIMEOUT_SECONDS = 90
DELAY_AFTER_ACCOUNT_CLICK_SECONDS = 0.2
ACCOUNT_SCROLL_TRIES = 180
ACCOUNT_SCROLL_PIXELS = 160
MAX_DOWNLOAD_ATTEMPTS = 3
PAUSE_ON_FAILURE = True
OVERLAY_WAIT_SECONDS = 30
DETAIL_ROW_WAIT_SECONDS = 15
DETAIL_CHANGE_WAIT_SECONDS = 3
DOWNLOAD_POLL_INTERVAL_SECONDS = 0.15
ACCOUNT_CODE_COLUMN = 8
ACCOUNT_NAME_COLUMN = 9
# 재실행 시 이미 같은 이름의 엑셀 파일이 있으면 건너뜁니다.
SKIP_ALREADY_DOWNLOADED = True
@dataclass(frozen=True)
class Account:
code: str
name: str
@property
def safe_stem(self) -> str:
return safe_filename(f"{self.code}_{self.name}")
@property
def safe_filename(self) -> str:
return f"{self.safe_stem}.xlsx"
# 이미지에 보이는 계정 목록입니다. 필요한 계정을 추가/삭제해도 됩니다.
ACCOUNTS: list[Account] = [
Account("103", "보통예금"),
Account("104", "국고보조금"),
Account("106", "기타예금"),
Account("108", "외상매출금"),
Account("109", "대손충당금"),
Account("110", "받을어음"),
Account("112", "공사미수금"),
Account("114", "단기대여금"),
Account("116", "미수수익"),
Account("120", "미수금"),
Account("123", "단기매매증권"),
Account("125", "받아볼어음"),
Account("131", "선급금"),
Account("133", "선급비용"),
Account("135", "부가세대급금"),
Account("136", "선납세금"),
Account("137", "주.종단기채권"),
Account("138", "전도금"),
Account("170", "미완성공사도급"),
Account("183", "매도가능증권"),
Account("192", "단체퇴직보험예금"),
Account("201", "토지"),
Account("202", "건물"),
Account("203", "감가상각누계액"),
Account("206", "기계장치"),
Account("207", "감가상각누계액"),
Account("208", "차량운반구"),
Account("209", "감가상각누계액"),
Account("210", "공구와기구"),
Account("211", "감가상각누계액"),
Account("212", "비품"),
Account("213", "감가상각누계액"),
Account("219", "시설장치"),
Account("220", "감가상각누계액"),
Account("221", "연구기자재"),
Account("222", "감가상각누계액"),
Account("223", "국고보조금"),
Account("231", "영업권"),
Account("234", "실용신안권"),
Account("241", "사용수익기부자산"),
Account("251", "외상매출금"),
Account("253", "미지급금"),
Account("254", "예수금"),
Account("255", "부가세예수금"),
Account("259", "선수금"),
Account("260", "단기차입금"),
Account("262", "미지급비용"),
Account("271", "공사선수금"),
Account("274", "예수국민연금"),
Account("275", "예수건강보험"),
Account("276", "예수고용보험"),
Account("290", "주.종단기차입금"),
Account("294", "임대보증금"),
Account("331", "자본금"),
Account("342", "감자차익"),
Account("351", "이익준비금"),
Account("375", "이월이익잉여금"),
Account("377", "미처분이익잉여금"),
Account("383", "자기주식"),
Account("400", "손익"),
Account("411", "설계용역수입"),
Account("412", "감리용역수입"),
Account("413", "임대료수입"),
Account("414", "주차료수입"),
Account("415", "안전점검수입"),
Account("417", "연구용역수입"),
Account("452", "도급공사매출원가"),
Account("602", "외주비"),
Account("604", "임금"),
Account("606", "잡금"),
Account("609", "퇴직급여"),
Account("611", "복리후생비"),
Account("612", "여비교통비"),
Account("614", "통신비"),
Account("615", "가스수도료"),
Account("616", "전력비"),
Account("617", "세금과공과금"),
Account("618", "감가상각비"),
Account("619", "지급임차료"),
Account("620", "수선비"),
Account("621", "보험료"),
Account("622", "차량유지비"),
Account("625", "교육훈련비"),
Account("626", "도서인쇄비"),
Account("629", "사무용품비"),
Account("630", "소모품비"),
Account("631", "지급수수료"),
Account("634", "접대비"),
Account("636", "광고선전비"),
Account("637", "관리비"),
Account("639", "보증수수료"),
Account("643", "해외출장비"),
Account("644", "행사비용"),
Account("645", "관리현장운영비"),
Account("646", "부서비"),
Account("650", "연구개발비"),
Account("802", "직원급여"),
Account("808", "퇴직급여"),
Account("811", "복리후생비"),
Account("812", "여비교통비"),
Account("813", "접대비(기업업무추진비)"),
Account("814", "통신비"),
Account("815", "수도광열비"),
Account("816", "전력비"),
Account("817", "세금과공과금"),
Account("818", "감가상각비"),
Account("819", "지급임차료"),
Account("820", "수선비"),
Account("821", "보험료"),
Account("822", "차량유지비"),
Account("823", "경상연구개발비"),
Account("825", "교육훈련비"),
Account("826", "도서인쇄비"),
Account("829", "사무용품비"),
Account("830", "소모품비"),
Account("831", "지급수수료"),
Account("835", "대손상각비"),
Account("837", "건물관리비"),
Account("846", "부서비"),
Account("901", "이자수익"),
Account("903", "배당금수익"),
Account("905", "투자주식평가이익"),
Account("906", "유가증권처분이익"),
Account("908", "대손충당금환입"),
Account("914", "유형자산처분이익"),
Account("930", "잡이익"),
Account("931", "이자비용"),
Account("933", "기부금"),
Account("935", "외화환산손실"),
Account("937", "투자주식평가손실"),
Account("960", "잡손실"),
Account("962", "민사보전금"),
Account("964", "기타보증금"),
Account("989", "매도가능증권평가이익"),
Account("998", "법인세등"),
]
def safe_filename(name: str) -> str:
cleaned = re.sub(r'[\\/:*?"<>|]', "_", name)
cleaned = re.sub(r"\s+", " ", cleaned).strip()
return cleaned
def log(message: str) -> None:
print(f"[{datetime.now():%H:%M:%S}] {message}", flush=True)
def save_diagnostics(driver: WebDriver, label: str, error: Exception | None = None) -> Path:
DIAGNOSTIC_DIR.mkdir(parents=True, exist_ok=True)
safe_label = safe_filename(label)
stamp = datetime.now().strftime("%Y%m%d_%H%M%S")
base_path = DIAGNOSTIC_DIR / f"{stamp}_{safe_label}"
png_path = base_path.with_suffix(".png")
txt_path = base_path.with_suffix(".txt")
html_path = base_path.with_suffix(".html")
try:
driver.save_screenshot(str(png_path))
except Exception:
pass
try:
title = driver.title
except Exception:
title = ""
try:
url = driver.current_url
except Exception:
url = ""
try:
html = driver.page_source
html_path.write_text(html, encoding="utf-8")
except Exception:
html_path = Path("")
message_lines = [
f"label={label}",
f"time={datetime.now():%Y-%m-%d %H:%M:%S}",
f"url={url}",
f"title={title}",
]
if error is not None:
message_lines.append(f"error={type(error).__name__}: {error}")
message_lines.extend(
[
f"screenshot={png_path.resolve() if png_path.exists() else ''}",
f"html={html_path.resolve() if html_path and html_path.exists() else ''}",
f"download_dir={DOWNLOAD_DIR.resolve()}",
f"fallback_download_dir={FALLBACK_DOWNLOAD_DIR.resolve()}",
]
)
txt_path.write_text("\n".join(message_lines), encoding="utf-8")
return txt_path
def build_driver(download_dir: Path, headless: bool = False) -> WebDriver:
download_dir.mkdir(parents=True, exist_ok=True)
options = ChromeOptions()
options.add_argument(f"--user-data-dir={CHROME_USER_DATA_DIR}")
options.add_argument(f"--profile-directory={CHROME_PROFILE_NAME}")
options.add_argument("--start-maximized")
options.add_argument("--disable-popup-blocking")
options.add_argument("--log-level=3")
options.add_argument("--disable-logging")
options.add_argument("--disable-features=WebUSB,WebBluetooth")
options.add_experimental_option("excludeSwitches", ["enable-logging"])
options.add_experimental_option("detach", True)
options.add_experimental_option(
"prefs",
{
"download.default_directory": str(download_dir.resolve()),
"download.prompt_for_download": False,
"download.directory_upgrade": True,
"safebrowsing.enabled": True,
"profile.default_content_setting_values.automatic_downloads": 1,
"profile.default_content_settings.popups": 0,
},
)
if headless:
options.add_argument("--headless=new")
options.add_argument("--window-size=1920,1080")
try:
service = Service(ChromeDriverManager().install(), log_output=subprocess.DEVNULL)
driver = webdriver.Chrome(service=service, options=options)
except WebDriverException as exc:
log("ChromeDriverManager로 ChromeDriver를 시작하지 못했습니다. Selenium Manager 방식으로 다시 시도합니다.")
log("그래도 실패하면 WSL Ubuntu에 Chrome 및 ChromeDriver 실행 라이브러리가 설치되어 있는지 확인해야 합니다.")
log("필요 패키지 예: google-chrome-stable, libnss3, libnspr4, libgtk-3-0, libgbm1")
log(f"첫 번째 오류: {exc}")
driver = webdriver.Chrome(options=options)
driver.execute_cdp_cmd(
"Page.setDownloadBehavior",
{"behavior": "allow", "downloadPath": str(download_dir.resolve())},
)
return driver
def switch_to_frame_containing(driver: WebDriver, selectors: Iterable[str]) -> None:
"""현재 문서와 iframe을 순회하며 후보 선택자가 있는 프레임으로 이동합니다."""
driver.switch_to.default_content()
if any(driver.find_elements(By.CSS_SELECTOR, selector) for selector in selectors):
return
frames = driver.find_elements(By.CSS_SELECTOR, "iframe, frame")
for frame in frames:
driver.switch_to.default_content()
driver.switch_to.frame(frame)
if any(driver.find_elements(By.CSS_SELECTOR, selector) for selector in selectors):
return
driver.switch_to.default_content()
def first_visible_by_css(driver: WebDriver, selectors: Iterable[str], timeout: int = WAIT_SECONDS) -> WebElement:
selector_list = list(selectors)
deadline = time.time() + timeout
while time.time() < deadline:
for selector in selector_list:
for element in driver.find_elements(By.CSS_SELECTOR, selector):
if element.is_displayed():
return element
time.sleep(0.25)
raise TimeoutException(f"요소를 찾지 못했습니다. 후보: {selector_list}")
def first_clickable_by_css(driver: WebDriver, selectors: Iterable[str], timeout: int = WAIT_SECONDS) -> WebElement:
selector_list = list(selectors)
deadline = time.time() + timeout
while time.time() < deadline:
for selector in selector_list:
for element in driver.find_elements(By.CSS_SELECTOR, selector):
if element.is_displayed() and element.is_enabled():
return element
time.sleep(0.25)
raise TimeoutException(f"클릭 가능한 요소를 찾지 못했습니다. 후보: {selector_list}")
def click_visible_text_button(driver: WebDriver, texts: Iterable[str], timeout: float = 5.0) -> bool:
"""Click a visible button-like element with one of the given labels."""
labels = list(texts)
deadline = time.time() + timeout
while time.time() < deadline:
for _ in contexts_with_default_first(driver):
for label in labels:
xpath = f"//*[normalize-space(.)='{label}']"
for element in driver.find_elements(By.XPATH, xpath):
if not element.is_displayed() or not element.is_enabled():
continue
rect = rect_of(driver, element)
if rect["width"] <= 0 or rect["height"] <= 0:
continue
try:
ActionChains(driver).move_to_element(element).click(element).perform()
except WebDriverException:
try:
element.click()
except WebDriverException:
js_click_element(driver, element)
time.sleep(0.1)
driver.switch_to.default_content()
return True
time.sleep(0.1)
driver.switch_to.default_content()
return False
def accept_download_complete_popup(driver: WebDriver, timeout: float = 1.5) -> bool:
"""Close WEHAGO's Excel-download complete confirmation popup if it appears."""
try:
alert = driver.switch_to.alert
alert.accept()
time.sleep(0.1)
return True
except Exception:
driver.switch_to.default_content()
return click_visible_text_button(driver, ["확인", "OK", "Ok", "예"], timeout=timeout)
def contexts_with_default_first(driver: WebDriver):
driver.switch_to.default_content()
yield
frames = driver.find_elements(By.CSS_SELECTOR, "iframe, frame")
for frame in frames:
driver.switch_to.default_content()
driver.switch_to.frame(frame)
yield
def rect_of(driver: WebDriver, element: WebElement) -> dict:
return driver.execute_script(
"""
const r = arguments[0].getBoundingClientRect();
return {left: r.left, top: r.top, width: r.width, height: r.height};
""",
element,
)
def viewport_width(driver: WebDriver) -> int:
return int(driver.execute_script("return window.innerWidth || document.documentElement.clientWidth;"))
def js_click_element(driver: WebDriver, element: WebElement) -> None:
"""Selenium click이 0 크기 자식요소에 걸릴 때를 피하기 위해 요소 중앙에 마우스 이벤트를 보냅니다."""
driver.execute_script(
"""
const source = arguments[0];
const el = source.closest('button, a, tr, li, div') || source;
const r = el.getBoundingClientRect();
const x = r.left + r.width / 2;
const y = r.top + r.height / 2;
for (const type of ['mouseover', 'mousemove', 'mousedown', 'mouseup', 'click']) {
el.dispatchEvent(new MouseEvent(type, {
bubbles: true,
cancelable: true,
view: window,
clientX: x,
clientY: y,
button: 0
}));
}
""",
element,
)
def has_blocking_overlay(driver: WebDriver) -> bool:
"""WEHAGO 로딩/모달 투명 레이어가 클릭을 가로막는지 확인합니다."""
try:
return bool(
driver.execute_script(
"""
const viewportArea = window.innerWidth * window.innerHeight;
const nodes = Array.from(document.querySelectorAll('div'));
return nodes.some((el) => {
const style = window.getComputedStyle(el);
if (style.display === 'none' || style.visibility === 'hidden') return false;
if (style.pointerEvents === 'none') return false;
const z = Number.parseInt(style.zIndex || '0', 10);
if (!Number.isFinite(z) || z < 1000) return false;
if (style.position !== 'fixed' && style.position !== 'absolute') return false;
const r = el.getBoundingClientRect();
if (r.width <= 0 || r.height <= 0) return false;
const area = r.width * r.height;
const coversScreen = area > viewportArea * 0.5;
const coversTopLeft = r.left <= 5 && r.top <= 5 && r.right >= window.innerWidth * 0.8 && r.bottom >= window.innerHeight * 0.8;
return coversScreen || coversTopLeft;
});
"""
)
)
except WebDriverException:
return False
def wait_for_blocking_overlay_gone(driver: WebDriver, timeout: int = OVERLAY_WAIT_SECONDS) -> None:
deadline = time.time() + timeout
while time.time() < deadline:
if not has_blocking_overlay(driver):
return
time.sleep(0.1)
diagnostic = save_diagnostics(driver, "blocking_overlay_timeout")
raise TimeoutException(f"화면을 가리는 로딩/팝업 레이어가 사라지지 않았습니다. 진단 파일: {diagnostic.resolve()}")
def close_open_menus(driver: WebDriver) -> None:
# Do not send ESC or click outside the grid here.
# In WEHAGO, global keyboard shortcuts can cancel the current ledger view
# and reset the searched account-ledger screen to an empty initial state.
return
def click_left_account_text(driver: WebDriver, account: Account) -> bool:
"""왼쪽 계정 목록에서 계정 코드 또는 계정명을 찾아 클릭합니다."""
width = viewport_width(driver)
left_limit = width * 0.35
candidates: list[tuple[float, float, WebElement]] = []
for _ in contexts_with_default_first(driver):
for text in (account.code, account.name):
xpath = f"//*[normalize-space(.)='{text}']"
for element in driver.find_elements(By.XPATH, xpath):
if not element.is_displayed():
continue
rect = rect_of(driver, element)
if rect["left"] < left_limit and rect["width"] > 0 and rect["height"] > 0:
candidates.append((rect["top"], rect["left"], element))
if candidates:
_, _, element = sorted(candidates, key=lambda item: (item[0], item[1]))[0]
driver.execute_script("arguments[0].scrollIntoView({block: 'center', inline: 'center'});", element)
time.sleep(0.2)
wait_for_blocking_overlay_gone(driver)
try:
element.click()
except ElementClickInterceptedException:
wait_for_blocking_overlay_gone(driver)
js_click_element(driver, element)
return True
driver.switch_to.default_content()
return False
def visible_text_exists_in_left_area(driver: WebDriver, text: str) -> bool:
width = viewport_width(driver)
left_limit = width * 0.35
for _ in contexts_with_default_first(driver):
xpath = f"//*[normalize-space(.)='{text}']"
for element in driver.find_elements(By.XPATH, xpath):
if not element.is_displayed():
continue
rect = rect_of(driver, element)
if rect["left"] < left_limit and rect["width"] > 0 and rect["height"] > 0:
return True
driver.switch_to.default_content()
return False
def account_list_has_any_expected_account(driver: WebDriver, accounts: list[Account]) -> bool:
for account in accounts[: min(10, len(accounts))]:
if visible_text_exists_in_left_area(driver, account.code) or visible_text_exists_in_left_area(driver, account.name):
return True
return False
def click_query_button(driver: WebDriver) -> bool:
"""상단 조건 영역의 조회 버튼을 누릅니다. 메뉴 검색 버튼과 혼동하지 않도록 위치를 제한합니다."""
viewport = viewport_width(driver)
candidates: list[tuple[float, WebElement]] = []
for _ in contexts_with_default_first(driver):
for text in ():
xpath = f"//*[normalize-space(.)='{text}']"
for element in driver.find_elements(By.XPATH, xpath):
if not element.is_displayed() or not element.is_enabled():
continue
rect = rect_of(driver, element)
if rect["width"] <= 0 or rect["height"] <= 0:
continue
# 계정별원장 상단 조회 버튼은 화면 상단 조건줄 오른쪽에 있습니다.
if 90 <= rect["top"] <= 230 and rect["left"] > viewport * 0.45:
candidates.append((rect["left"], element))
if candidates:
_, button = sorted(candidates, key=lambda item: item[0], reverse=True)[0]
js_click_element(driver, button)
time.sleep(0.8)
wait_for_blocking_overlay_gone(driver)
return True
driver.switch_to.default_content()
return False
def wait_for_account_list_data(driver: WebDriver, accounts: list[Account], timeout: int = 45) -> bool:
deadline = time.time() + timeout
while time.time() < deadline:
if has_blocking_overlay(driver):
time.sleep(0.5)
continue
if account_list_has_any_expected_account(driver, accounts):
return True
time.sleep(0.7)
return False
def ensure_ledger_data_loaded(driver: WebDriver, accounts: list[Account]) -> None:
if account_list_has_any_expected_account(driver, accounts):
return
if False:
log("계정 목록이 비어 있어 조회 버튼을 누르고 데이터 로딩을 기다립니다.")
else:
diagnostic = save_diagnostics(driver, "account_list_empty")
raise RuntimeError(
"왼쪽 계정 목록에 계정 데이터가 없습니다. "
"WEHAGO 화면에서 직접 조회를 눌러 계정 목록이 보이는 상태로 만든 뒤 다시 Enter를 누르세요. "
"자동 조회는 화면 조건을 초기화할 수 있어 기본으로 사용하지 않습니다. "
f"진단 파일: {diagnostic.resolve()}"
)
if click_query_button(driver):
if wait_for_account_list_data(driver, accounts):
log("계정 목록 데이터가 로딩되었습니다.")
return
diagnostic = save_diagnostics(driver, "account_list_empty_after_query")
raise RuntimeError(
"조회 후에도 왼쪽 계정 목록에 계정 데이터가 없습니다. "
"기간/회사/계정과목 조건을 확인하고, 화면에 계정 목록이 보이는 상태에서 다시 실행하세요. "
f"진단 파일: {diagnostic.resolve()}"
)
def find_left_scroll_container(driver: WebDriver) -> WebElement | None:
"""왼쪽 계정 목록을 담은 스크롤 가능한 영역을 찾습니다."""
width = viewport_width(driver)
left_limit = width * 0.35
for _ in contexts_with_default_first(driver):
best: tuple[float, WebElement] | None = None
for element in driver.find_elements(By.CSS_SELECTOR, "div, section, article"):
if not element.is_displayed():
continue
rect = rect_of(driver, element)
if rect["left"] >= left_limit or rect["height"] < 100 or rect["width"] < 100:
continue
scrollable = driver.execute_script(
"return arguments[0].scrollHeight > arguments[0].clientHeight + 20;",
element,
)
if not scrollable:
continue
score = rect["width"] * rect["height"]
if best is None or score > best[0]:
best = (score, element)
if best is not None:
return best[1]
driver.switch_to.default_content()
return None
def find_left_account_grid(driver: WebDriver) -> WebElement | None:
"""스크롤 컨테이너를 못 찾을 때 휠을 보낼 왼쪽 계정 그리드를 찾습니다."""
width = viewport_width(driver)
left_limit = width * 0.35
for _ in contexts_with_default_first(driver):
best: tuple[float, WebElement] | None = None
for selector in ACCOUNT_LIST_SELECTORS:
for element in driver.find_elements(By.CSS_SELECTOR, selector):
if not element.is_displayed():
continue
rect = rect_of(driver, element)
if rect["left"] >= left_limit or rect["height"] < 100 or rect["width"] < 100:
continue
score = rect["width"] * rect["height"]
if best is None or score > best[0]:
best = (score, element)
if best is not None:
return best[1]
driver.switch_to.default_content()
return None
def scroll_left_account_list(driver: WebDriver) -> bool:
container = find_left_scroll_container(driver)
if container is not None:
before = driver.execute_script("return arguments[0].scrollTop;", container)
driver.execute_script("arguments[0].scrollTop = arguments[0].scrollTop + arguments[1];", container, ACCOUNT_SCROLL_PIXELS)
time.sleep(0.35)
after = driver.execute_script("return arguments[0].scrollTop;", container)
if after != before:
return True
grid = find_left_account_grid(driver)
if grid is None:
return False
try:
ActionChains(driver).move_to_element(grid).scroll_by_amount(0, ACCOUNT_SCROLL_PIXELS).perform()
time.sleep(0.35)
return True
except WebDriverException:
return False
def reset_left_account_list_scroll(driver: WebDriver) -> None:
container = find_left_scroll_container(driver)
if container is not None:
driver.execute_script("arguments[0].scrollTop = 0;", container)
time.sleep(0.5)
return
grid = find_left_account_grid(driver)
if grid is None:
return
try:
ActionChains(driver).move_to_element(grid).scroll_by_amount(0, -10000).perform()
time.sleep(0.5)
except WebDriverException:
pass
def select_account_from_left_list(driver: WebDriver, account: Account) -> None:
"""계정 코드 입력칸 대신 화면 왼쪽 계정 목록의 행을 직접 선택합니다."""
if click_left_account_text(driver, account):
return
for reset_before_search in (False, True):
if reset_before_search:
reset_left_account_list_scroll(driver)
if click_left_account_text(driver, account):
return
for _ in range(ACCOUNT_SCROLL_TRIES):
if not scroll_left_account_list(driver):
break
if click_left_account_text(driver, account):
return
raise TimeoutException(f"왼쪽 계정 목록에서 {account.code} {account.name} 행을 찾지 못했습니다.")
def wait_until_download_finished(download_dir: Path, before_files: dict[Path, set[Path]]) -> Path:
deadline = time.time() + DOWNLOAD_TIMEOUT_SECONDS
while time.time() < deadline:
completed: list[Path] = []
temp_files: list[Path] = []
for folder, previous_files in before_files.items():
current_files = set(folder.glob("*"))
new_files = current_files - previous_files
temp_files.extend(p for p in current_files if p.suffix.lower() in {".crdownload", ".tmp"})
completed.extend(
p
for p in new_files
if p.is_file()
and p.suffix.lower() in {".xlsx", ".xls", ".csv"}
and not p.name.startswith("~$")
)
if completed and not temp_files:
return max(completed, key=lambda p: p.stat().st_mtime)
time.sleep(DOWNLOAD_POLL_INTERVAL_SECONDS)
scan_targets = ", ".join(str(folder.resolve()) for folder in before_files)
raise TimeoutException(f"{DOWNLOAD_TIMEOUT_SECONDS}초 동안 다운로드 완료 파일을 찾지 못했습니다. 확인 위치: {scan_targets}")
def click_excel_menu_in_current_context(driver: WebDriver, timeout: int = 3) -> bool:
deadline = time.time() + timeout
while time.time() < deadline:
menu_candidates: list[tuple[float, WebElement]] = []
for xpath in EXCEL_MENU_XPATHS:
for menu in driver.find_elements(By.XPATH, xpath):
if not menu.is_displayed():
continue
rect = rect_of(driver, menu)
if rect["width"] <= 0 or rect["height"] <= 0:
continue
text = (menu.text or "").strip()
element_id = menu.get_attribute("id") or ""
if not any(keyword in text or keyword in element_id for keyword in EXCEL_MENU_TEXTS):
continue
score = 0 if element_id == "context:엑셀변환" else rect["width"] * rect["height"]
menu_candidates.append((score, menu))
if menu_candidates:
_, menu = sorted(menu_candidates, key=lambda item: item[0])[0]
try:
ActionChains(driver).move_to_element(menu).click(menu).perform()
except WebDriverException:
try:
menu.click()
except WebDriverException:
js_click_element(driver, menu)
time.sleep(0.5)
return True
time.sleep(0.2)
return False
def find_detail_grid(driver: WebDriver) -> WebElement:
"""오른쪽 상세 내역 그리드를 찾습니다."""
width = viewport_width(driver)
right_start = width * 0.25
best: tuple[float, WebElement] | None = None
for _ in contexts_with_default_first(driver):
for selector in GRID_SELECTORS:
for element in driver.find_elements(By.CSS_SELECTOR, selector):
if not element.is_displayed():
continue
rect = rect_of(driver, element)
if rect["left"] < right_start or rect["width"] < 250 or rect["height"] < 120:
continue
score = rect["width"] * rect["height"]
if best is None or score > best[0]:
best = (score, element)
if best is not None:
return best[1]
driver.switch_to.default_content()
raise TimeoutException("오른쪽 상세 내역 그리드를 찾지 못했습니다.")
def find_detail_data_cell(driver: WebDriver) -> WebElement | None:
"""오른쪽 상세 그리드 안의 실제 데이터 셀을 찾습니다."""
width = viewport_width(driver)
right_start = width * 0.25
for _ in contexts_with_default_first(driver):
candidates: list[tuple[float, WebElement]] = []
for selector in [".rg-data-cell", "td[class*='rg-data-cell']", "[class*='rg-data-cell']"]:
for element in driver.find_elements(By.CSS_SELECTOR, selector):
if not element.is_displayed():
continue
rect = rect_of(driver, element)
if rect["left"] < right_start or rect["width"] <= 0 or rect["height"] <= 0:
continue
text = (element.text or "").strip()
# 금액 셀처럼 text가 비어 보이는 경우도 있어, 오른쪽 상세 영역의 보이는 데이터 셀 자체를 후보로 둡니다.
score = rect["top"] * 10000 + rect["left"]
if text or rect["top"] > 180:
candidates.append((score, element))
if candidates:
return sorted(candidates, key=lambda item: item[0])[0][1]
driver.switch_to.default_content()
return None
def detail_grid_signature(driver: WebDriver) -> str:
width = viewport_width(driver)
right_start = width * 0.25
parts: list[str] = []
for _ in contexts_with_default_first(driver):
elements = driver.find_elements(By.CSS_SELECTOR, ".rg-data-cell, td[class*='rg-data-cell'], [class*='rg-data-cell']")
for element in elements[:80]:
if not element.is_displayed():
continue
rect = rect_of(driver, element)
if rect["left"] < right_start or rect["width"] <= 0 or rect["height"] <= 0:
continue
text = (element.text or "").strip()
if text:
parts.append(text)
if len(parts) >= 20:
break
if parts:
break
driver.switch_to.default_content()
return "|".join(parts)
def wait_for_detail_change(driver: WebDriver, before_signature: str, timeout: int = DETAIL_CHANGE_WAIT_SECONDS) -> None:
deadline = time.time() + timeout
while time.time() < deadline:
wait_for_blocking_overlay_gone(driver)
current_signature = detail_grid_signature(driver)
if current_signature and current_signature != before_signature:
return
if find_detail_data_cell(driver) is not None and not before_signature:
return
time.sleep(0.15)
def wait_for_detail_data_cell(driver: WebDriver, timeout: int = DETAIL_ROW_WAIT_SECONDS) -> WebElement | None:
deadline = time.time() + timeout
while time.time() < deadline:
wait_for_blocking_overlay_gone(driver)
cell = find_detail_data_cell(driver)
if cell is not None:
return cell
time.sleep(0.15)
return None
def context_click_excel_download(driver: WebDriver) -> None:
close_open_menus(driver)
wait_for_blocking_overlay_gone(driver)
target = wait_for_detail_data_cell(driver) or find_detail_grid(driver)
ActionChains(driver).move_to_element(target).context_click(target).perform()
time.sleep(0.15)
if click_excel_menu_in_current_context(driver):
return
driver.switch_to.default_content()
if click_excel_menu_in_current_context(driver):
return
frames = driver.find_elements(By.CSS_SELECTOR, "iframe, frame")
for frame in frames:
driver.switch_to.default_content()
driver.switch_to.frame(frame)
if click_excel_menu_in_current_context(driver, timeout=2):
return
driver.switch_to.default_content()
raise NoSuchElementException("우클릭 메뉴에서 엑셀/다운로드 항목을 찾지 못했습니다.")
def rename_downloaded_file(downloaded_file: Path, account: Account, download_dir: Path) -> Path:
suffix = downloaded_file.suffix if downloaded_file.suffix.lower() in {".xlsx", ".xls", ".csv"} else ".xlsx"
target = download_dir / f"{account.safe_stem}{suffix.lower()}"
if target.exists():
target.unlink()
if downloaded_file.parent.resolve() == download_dir.resolve():
downloaded_file.rename(target)
else:
target.write_bytes(downloaded_file.read_bytes())
downloaded_file.unlink()
return target
def write_clean_xlsx_with_account_columns(path: Path, account: Account) -> None:
raw_rows = read_rows_from_xlsx_raw(path)
workbook = Workbook()
sheet = workbook.active
sheet.title = "Sheet1"
header_found = False
for _, _, values in raw_rows:
normalized = ["" if value is None else str(value).strip() for value in values]
if is_ledger_header(values):
header_found = True
break
in_table = False
for _, _, values in raw_rows:
row_values = list(values)
if header_found and is_ledger_header(row_values):
in_table = True
row_values = row_values[: ACCOUNT_CODE_COLUMN - 1]
while len(row_values) < ACCOUNT_CODE_COLUMN - 1:
row_values.append(None)
row_values.extend(["계정코드", "계정명"])
elif in_table and non_empty_row(row_values):
row_values = row_values[: ACCOUNT_CODE_COLUMN - 1]
while len(row_values) < ACCOUNT_CODE_COLUMN - 1:
row_values.append(None)
row_values.extend([account.code, account.name])
sheet.append(row_values)
workbook.save(path)
workbook.close()
def account_columns_already_in_h_i(path: Path, account: Account) -> bool:
rows = read_downloaded_rows(path)
found_header = False
checked_data_rows = 0
for _, _, values in rows:
normalized = normalize_row(values)
if is_ledger_header(values):
found_header = True
if len(normalized) < ACCOUNT_NAME_COLUMN:
return False
if normalized[ACCOUNT_CODE_COLUMN - 1 : ACCOUNT_NAME_COLUMN] != ["계정코드", "계정명"]:
return False
continue
if not found_header or not non_empty_row(values):
continue
if len(normalized) < ACCOUNT_NAME_COLUMN:
return False
if normalized[ACCOUNT_CODE_COLUMN - 1 : ACCOUNT_NAME_COLUMN] != [account.code, account.name]:
return False
checked_data_rows += 1
if checked_data_rows >= 20:
return True
return found_header
def add_account_columns_to_excel(path: Path, account: Account) -> None:
if path.suffix.lower() != ".xlsx":
return
try:
if account_columns_already_in_h_i(path, account):
return
except Exception:
pass
try:
write_clean_xlsx_with_account_columns(path, account)
return
except Exception:
pass
try:
workbook = load_workbook(path)
except Exception:
write_clean_xlsx_with_account_columns(path, account)
return
try:
for sheet in workbook.worksheets:
header_cell = None
for row in sheet.iter_rows():
for cell in row:
value = "" if cell.value is None else str(cell.value).strip()
if value == "전표번호":
header_cell = cell
break
if header_cell is not None:
break
if header_cell is None:
continue
row_index = header_cell.row
code_col = ACCOUNT_CODE_COLUMN
name_col = ACCOUNT_NAME_COLUMN
code_header = sheet.cell(row=row_index, column=code_col).value
name_header = sheet.cell(row=row_index, column=name_col).value
if str(code_header).strip() != "계정코드" or str(name_header).strip() != "계정명":
sheet.cell(row=row_index, column=code_col, value="계정코드")
sheet.cell(row=row_index, column=name_col, value="계정명")
for data_row in range(row_index + 1, sheet.max_row + 1):
values = [sheet.cell(row=data_row, column=col).value for col in range(1, sheet.max_column + 1)]
if non_empty_row(values):
sheet.cell(row=data_row, column=code_col, value=account.code)
sheet.cell(row=data_row, column=name_col, value=account.name)
workbook.save(path)
finally:
workbook.close()
def account_file_candidates(account: Account, download_dir: Path) -> list[Path]:
return [
download_dir / f"{account.safe_stem}.xlsx",
download_dir / f"{account.safe_stem}.xls",
download_dir / f"{account.safe_stem}.csv",
]
def find_account_file(account: Account, download_dir: Path) -> Path | None:
for candidate in account_file_candidates(account, download_dir):
if candidate.exists():
return candidate
return None
def download_scan_dirs(download_dir: Path) -> list[Path]:
dirs = [download_dir]
if FALLBACK_DOWNLOAD_DIR.resolve() != download_dir.resolve():
dirs.append(FALLBACK_DOWNLOAD_DIR)
return dirs
def snapshot_download_files(download_dir: Path) -> dict[Path, set[Path]]:
snapshots: dict[Path, set[Path]] = {}
for folder in download_scan_dirs(download_dir):
folder.mkdir(parents=True, exist_ok=True)
snapshots[folder] = set(folder.glob("*"))
return snapshots
def non_empty_row(values: Iterable[object]) -> bool:
return any(value is not None and str(value).strip() != "" for value in values)
def excel_column_number(cell_reference: str) -> int:
letters = "".join(ch for ch in cell_reference if ch.isalpha()).upper()
number = 0
for letter in letters:
number = number * 26 + (ord(letter) - ord("A") + 1)
return number
def read_shared_strings_from_xlsx(path: Path) -> list[str]:
namespace = {"x": "http://schemas.openxmlformats.org/spreadsheetml/2006/main"}
with zipfile.ZipFile(path) as archive:
if "xl/sharedStrings.xml" not in archive.namelist():
return []
root = ET.fromstring(archive.read("xl/sharedStrings.xml"))
strings: list[str] = []
for item in root.findall("x:si", namespace):
parts = [node.text or "" for node in item.findall(".//x:t", namespace)]
strings.append("".join(parts))
return strings
def read_rows_from_xlsx_raw(path: Path) -> list[tuple[str, int, tuple[object, ...]]]:
namespace = {"x": "http://schemas.openxmlformats.org/spreadsheetml/2006/main"}
shared_strings = read_shared_strings_from_xlsx(path)
rows: list[tuple[str, int, tuple[object, ...]]] = []
with zipfile.ZipFile(path) as archive:
sheet_names = [name for name in archive.namelist() if name.startswith("xl/worksheets/sheet") and name.endswith(".xml")]
for sheet_index, sheet_name in enumerate(sorted(sheet_names), start=1):
root = ET.fromstring(archive.read(sheet_name))
for row_node in root.findall(".//x:sheetData/x:row", namespace):
row_number = int(row_node.attrib.get("r", len(rows) + 1))
values_by_column: dict[int, object] = {}
for cell in row_node.findall("x:c", namespace):
reference = cell.attrib.get("r", "")
column_number = excel_column_number(reference)
cell_type = cell.attrib.get("t", "")
value_node = cell.find("x:v", namespace)
inline_node = cell.find("x:is", namespace)
value: object = None
if cell_type == "s" and value_node is not None and value_node.text is not None:
index = int(value_node.text)
value = shared_strings[index] if index < len(shared_strings) else value_node.text
elif cell_type == "inlineStr" and inline_node is not None:
value = "".join(node.text or "" for node in inline_node.findall(".//x:t", namespace))
elif value_node is not None:
value = value_node.text
if value is not None:
values_by_column[column_number] = value
if values_by_column:
max_column = max(values_by_column)
values = tuple(values_by_column.get(column) for column in range(1, max_column + 1))
if non_empty_row(values):
rows.append((f"Sheet{sheet_index}", row_number, values))
return rows
def read_rows_from_xlsx(path: Path) -> list[tuple[str, int, tuple[object, ...]]]:
rows: list[tuple[str, int, tuple[object, ...]]] = []
try:
workbook = load_workbook(path, read_only=True, data_only=True)
try:
for sheet in workbook.worksheets:
for row_number, row in enumerate(sheet.iter_rows(values_only=True), start=1):
values = tuple(row)
if non_empty_row(values):
rows.append((sheet.title, row_number, values))
finally:
workbook.close()
return rows
except Exception:
return read_rows_from_xlsx_raw(path)
def read_rows_from_csv(path: Path) -> list[tuple[str, int, tuple[object, ...]]]:
encodings = ["utf-8-sig", "cp949", "euc-kr"]
last_error: Exception | None = None
for encoding in encodings:
try:
rows: list[tuple[str, int, tuple[object, ...]]] = []
with path.open("r", encoding=encoding, newline="") as file:
reader = csv.reader(file)
for row_number, row in enumerate(reader, start=1):
values = tuple(row)
if non_empty_row(values):
rows.append(("CSV", row_number, values))
return rows
except UnicodeDecodeError as exc:
last_error = exc
raise UnicodeDecodeError("csv", b"", 0, 1, f"CSV 인코딩을 읽지 못했습니다: {last_error}")
def read_downloaded_rows(path: Path) -> list[tuple[str, int, tuple[object, ...]]]:
suffix = path.suffix.lower()
if suffix == ".xlsx":
return read_rows_from_xlsx(path)
if suffix == ".csv":
return read_rows_from_csv(path)
raise ValueError(".xls 파일은 현재 자동 취합에서 읽을 수 없습니다. WEHAGO 다운로드 형식을 .xlsx로 바꿔주세요.")
def normalize_row(values: Iterable[object]) -> list[str]:
return ["" if value is None else str(value).strip() for value in values]
def is_ledger_header(values: Iterable[object]) -> bool:
normalized = normalize_row(values)
return "전표번호" in normalized and ("일자" in normalized or "적요" in normalized)
def ensure_account_header_columns(values: tuple[object, ...]) -> tuple[object, ...]:
row = list(values)
while len(row) < ACCOUNT_NAME_COLUMN:
row.append(None)
row[ACCOUNT_CODE_COLUMN - 1] = "계정코드"
row[ACCOUNT_NAME_COLUMN - 1] = "계정명"
return tuple(row)
def ensure_account_data_columns(values: tuple[object, ...], account: Account) -> tuple[object, ...]:
row = list(values)
while len(row) < ACCOUNT_NAME_COLUMN:
row.append(None)
row[ACCOUNT_CODE_COLUMN - 1] = account.code
row[ACCOUNT_NAME_COLUMN - 1] = account.name
return tuple(row)
def extract_ledger_data_rows(
rows: list[tuple[str, int, tuple[object, ...]]],
account: Account,
) -> tuple[list[object], list[tuple[str, int, tuple[object, ...]]]]:
headers: list[object] = []
data_rows: list[tuple[str, int, tuple[object, ...]]] = []
found_header = False
for sheet_name, row_number, values in rows:
if is_ledger_header(values):
header = ensure_account_header_columns(values)
found_header = True
if len(header) > len(headers):
headers = list(header)
continue
if not found_header:
continue
if not non_empty_row(values):
continue
if is_ledger_header(values):
continue
row = ensure_account_data_columns(values, account)
data_rows.append((sheet_name, row_number, row))
return headers, data_rows
def autosize_columns(ws, max_width: int = 42) -> None:
for column_cells in ws.columns:
column_letter = column_cells[0].column_letter
values = [cell.value for cell in column_cells if cell.value is not None]
if not values:
continue
width = min(max(len(str(value)) for value in values) + 2, max_width)
ws.column_dimensions[column_letter].width = width
def style_header(ws, row: int = 1) -> None:
fill = PatternFill("solid", fgColor="D9EAF7")
font = Font(bold=True)
for cell in ws[row]:
cell.fill = fill
cell.font = font
cell.alignment = Alignment(horizontal="center")
def build_summary_rows(
accounts: list[Account],
download_dir: Path,
failures: list[tuple[Account, str]],
) -> list[dict[str, object]]:
failure_map = {account.code: message for account, message in failures}
rows: list[dict[str, object]] = []
for account in accounts:
path = find_account_file(account, download_dir)
if account.code in failure_map:
status = "오류"
message = failure_map[account.code]
elif path is None:
status = "누락"
message = "다운로드 파일을 찾지 못했습니다."
else:
status = "정상"
message = ""
rows.append(
{
"계정코드": account.code,
"계정명": account.name,
"상태": status,
"파일명": path.name if path else "",
"데이터행수": 0,
"메시지": message,
}
)
return rows
def consolidate_downloads(
accounts: list[Account],
download_dir: Path,
failures: list[tuple[Account, str]] | None = None,
) -> Path:
failures = failures or []
output_path = download_dir / f"통합_계정별원장_{datetime.now():%Y%m%d_%H%M%S}.xlsx"
summary_rows = build_summary_rows(accounts, download_dir, failures)
summary_by_code = {str(row["계정코드"]): row for row in summary_rows}
workbook = Workbook()
data_ws = workbook.active
data_ws.title = "취합데이터"
status_ws = workbook.create_sheet("계정별상태")
error_ws = workbook.create_sheet("오류_누락")
data_header = ["계정코드", "계정명", "원본파일", "원본시트", "원본행번호"]
data_ws.append(data_header)
max_value_columns = 0
source_headers: list[object] = []
for account in accounts:
path = find_account_file(account, download_dir)
if path is None:
continue
try:
add_account_columns_to_excel(path, account)
rows = read_downloaded_rows(path)
summary_by_code[account.code]["데이터행수"] = len(rows)
if not rows:
summary_by_code[account.code]["상태"] = "확인필요"
summary_by_code[account.code]["메시지"] = "파일은 있으나 읽을 수 있는 데이터 행이 없습니다."
continue
file_headers, ledger_rows = extract_ledger_data_rows(rows, account)
if file_headers and len(file_headers) > len(source_headers):
source_headers = file_headers
data_row_count = 0
for sheet_name, row_number, values in ledger_rows:
max_value_columns = max(max_value_columns, len(values))
data_ws.append([account.code, account.name, path.name, sheet_name, row_number, *values])
data_row_count += 1
summary_by_code[account.code]["데이터행수"] = data_row_count
if not ledger_rows:
summary_by_code[account.code]["상태"] = "확인필요"
summary_by_code[account.code]["메시지"] = "표 헤더는 찾았지만 취합할 거래 데이터 행이 없습니다."
except Exception as exc:
summary_by_code[account.code]["상태"] = "취합오류"
summary_by_code[account.code]["메시지"] = f"{type(exc).__name__}: {exc}"
for column_index in range(1, max_value_columns + 1):
header_value = source_headers[column_index - 1] if column_index <= len(source_headers) else f"원본열{column_index}"
data_ws.cell(row=1, column=len(data_header) + column_index, value=header_value)
status_ws.append(["계정코드", "계정명", "상태", "파일명", "데이터행수", "메시지"])
for row in summary_rows:
status_ws.append([row["계정코드"], row["계정명"], row["상태"], row["파일명"], row["데이터행수"], row["메시지"]])
error_ws.append(["계정코드", "계정명", "상태", "파일명", "메시지"])
for row in summary_rows:
if row["상태"] != "정상":
error_ws.append([row["계정코드"], row["계정명"], row["상태"], row["파일명"], row["메시지"]])
for ws in (data_ws, status_ws, error_ws):
style_header(ws)
ws.freeze_panes = "A2"
ws.auto_filter.ref = ws.dimensions
autosize_columns(ws)
workbook.save(output_path)
return output_path
def load_saved_ledger_url() -> str:
if not USE_SAVED_LEDGER_URL or not LEDGER_URL_FILE.exists():
return ""
return LEDGER_URL_FILE.read_text(encoding="utf-8").strip()
def save_ledger_url(url: str) -> None:
if USE_SAVED_LEDGER_URL and "smarta.wehago.com" in url and "SABK0107" in url:
LEDGER_URL_FILE.write_text(url, encoding="utf-8")
log(f"계정별원장 주소 저장: {LEDGER_URL_FILE.resolve()}")
def is_wehago_not_found_page(driver: WebDriver) -> bool:
try:
title = driver.title or ""
body_text = driver.find_element(By.TAG_NAME, "body").text
except Exception:
return False
checks = [title, body_text]
return any("페이지를 찾을 수 없습니다" in text or "404 Error" in text for text in checks)
def ledger_screen_ready(driver: WebDriver) -> bool:
if is_wehago_not_found_page(driver):
return False
try:
left_grid = find_left_account_grid(driver)
detail_grid = find_detail_grid(driver)
return left_grid is not None and detail_grid is not None
except Exception:
driver.switch_to.default_content()
return False
def open_wehago(driver: WebDriver) -> None:
saved_url = ""
target_url = ACCOUNT_LEDGER_URL.strip() or WEHAGO_START_URL
driver.get(target_url)
log("Chrome이 열렸습니다.")
log(f"계정별 다운로드 저장 위치: {DOWNLOAD_DIR.resolve()}")
log(f"보조 확인 위치: {FALLBACK_DOWNLOAD_DIR.resolve()}")
if saved_url:
log(f"저장된 계정별원장 주소를 열었습니다: {saved_url}")
elif ACCOUNT_LEDGER_URL.strip():
log(f"설정된 계정별원장 주소를 열었습니다: {ACCOUNT_LEDGER_URL.strip()}")
log("WEHAGO에 로그인하세요. 계정별원장 화면이 정상으로 열려 있으면 Enter를 누르세요.")
log("만약 404 화면이면 WEHAGO 메뉴에서 계정별원장 화면을 직접 연 뒤 Enter를 누르세요.")
input("계정별원장 화면 준비 후 Enter: ")
if is_wehago_not_found_page(driver):
log("현재 화면이 WEHAGO 404 페이지입니다. 고정 URL 또는 저장 URL이 현재 세션에서 유효하지 않습니다.")
log("브라우저에서 WEHAGO 메인으로 이동한 뒤, 메뉴로 계정별원장 화면을 직접 열어주세요.")
log("정상 계정별원장 화면이 보이면 다시 Enter를 누르세요.")
input("계정별원장 화면을 직접 연 뒤 Enter: ")
if not ledger_screen_ready(driver):
diagnostic = save_diagnostics(driver, "ledger_screen_not_ready")
raise RuntimeError(
"계정별원장 화면의 왼쪽 계정 목록 또는 오른쪽 상세 그리드를 찾지 못했습니다. "
f"진단 파일: {diagnostic.resolve()}"
)
log("계정별원장 화면 확인 완료. 사용자가 조회해 둔 현재 화면을 그대로 사용합니다.")
log("자동 조회, 저장 URL 재열기, 화면 초기화 동작은 수행하지 않습니다.")
def download_account_ledger(driver: WebDriver, account: Account, index: int, total: int) -> Path:
started_at = time.perf_counter()
close_open_menus(driver)
wait_for_blocking_overlay_gone(driver)
log(f"[{index}/{total}] {account.code} {account.name}: 왼쪽 계정 목록에서 선택")
before_files = snapshot_download_files(DOWNLOAD_DIR)
before_detail = detail_grid_signature(driver)
select_account_from_left_list(driver, account)
time.sleep(DELAY_AFTER_ACCOUNT_CLICK_SECONDS)
wait_for_blocking_overlay_gone(driver)
wait_for_detail_change(driver, before_detail, timeout=DETAIL_CHANGE_WAIT_SECONDS)
wait_for_detail_data_cell(driver, timeout=DETAIL_ROW_WAIT_SECONDS)
log(f"[{index}/{total}] {account.code} {account.name}: 우클릭 엑셀 다운로드")
context_click_excel_download(driver)
downloaded = wait_until_download_finished(DOWNLOAD_DIR, before_files)
if accept_download_complete_popup(driver):
log(f"[{index}/{total}] {account.code} {account.name}: 다운로드 완료 확인창 닫음")
renamed = rename_downloaded_file(downloaded, account, DOWNLOAD_DIR)
add_account_columns_to_excel(renamed, account)
elapsed = time.perf_counter() - started_at
log(f"[{index}/{total}] 완료: {renamed.name} ({elapsed:.1f}초)")
return renamed
def run(
accounts: list[Account],
start_index: int = 0,
end_index: int | None = None,
headless: bool = False,
merge: bool = True,
pause_on_failure: bool = PAUSE_ON_FAILURE,
) -> None:
selected_accounts = accounts[start_index:end_index]
if not selected_accounts:
raise ValueError("처리할 계정이 없습니다. start/end 옵션을 확인하세요.")
driver = build_driver(DOWNLOAD_DIR, headless=headless)
failures: list[tuple[Account, str]] = []
should_keep_browser_open = False
try:
open_wehago(driver)
total = len(selected_accounts)
pending_accounts = list(selected_accounts)
last_errors: dict[str, str] = {}
for attempt in range(1, MAX_DOWNLOAD_ATTEMPTS + 1):
if not pending_accounts:
break
if attempt > 1:
log(f"{attempt}회차 재시도를 시작합니다. 대상: {len(pending_accounts)}개 계정")
next_pending: list[Account] = []
for account in pending_accounts:
original_index = selected_accounts.index(account) + 1
existing_file = find_account_file(account, DOWNLOAD_DIR)
if SKIP_ALREADY_DOWNLOADED and existing_file is not None:
add_account_columns_to_excel(existing_file, account)
log(f"[{original_index}/{total}] {account.code} {account.name}: 이미 있음, 건너뜀")
last_errors.pop(account.code, None)
continue
try:
download_account_ledger(driver, account, original_index, total)
last_errors.pop(account.code, None)
except Exception as exc:
message = f"{type(exc).__name__}: {exc}"
last_errors[account.code] = message
diagnostic = save_diagnostics(driver, f"attempt{attempt}_failed_{account.code}_{account.name}", exc)
log(f"[실패 {attempt}/{MAX_DOWNLOAD_ATTEMPTS}] {account.code} {account.name}: {message}")
log(f"[진단 저장] {diagnostic.resolve()}")
if attempt < MAX_DOWNLOAD_ATTEMPTS:
next_pending.append(account)
pending_accounts = next_pending
failures = [
(account, last_errors.get(account.code, "3회 재시도 후에도 다운로드 파일을 만들지 못했습니다."))
for account in selected_accounts
if find_account_file(account, DOWNLOAD_DIR) is None
]
if failures:
should_keep_browser_open = True
for index, account in enumerate([], start=1):
existing_file = find_account_file(account, DOWNLOAD_DIR)
if SKIP_ALREADY_DOWNLOADED and existing_file is not None:
log(f"[{index}/{total}] {account.code} {account.name}: 이미 있음, 건너뜀")
continue
try:
close_open_menus(driver)
wait_for_blocking_overlay_gone(driver)
log(f"[{index}/{total}] {account.code} {account.name}: 왼쪽 계정 목록에서 선택")
before_files = snapshot_download_files(DOWNLOAD_DIR)
select_account_from_left_list(driver, account)
time.sleep(DELAY_AFTER_ACCOUNT_CLICK_SECONDS)
wait_for_blocking_overlay_gone(driver)
log(f"[{index}/{total}] {account.code} {account.name}: 우클릭 엑셀 다운로드")
context_click_excel_download(driver)
downloaded = wait_until_download_finished(DOWNLOAD_DIR, before_files)
if accept_download_complete_popup(driver):
log(f"[{index}/{total}] {account.code} {account.name}: 다운로드 완료 확인창 닫음")
renamed = rename_downloaded_file(downloaded, account, DOWNLOAD_DIR)
add_account_columns_to_excel(renamed, account)
log(f"[{index}/{total}] 완료: {renamed.name}")
except Exception as exc: # 계정 하나 실패해도 다음 계정으로 진행합니다.
message = f"{type(exc).__name__}: {exc}"
failures.append((account, message))
should_keep_browser_open = True
diagnostic = save_diagnostics(driver, f"failed_{account.code}_{account.name}", exc)
log(f"[실패] {account.code} {account.name}: {message}")
log(f"[진단 저장] {diagnostic.resolve()}")
log(f"작업 완료. 저장 위치: {DOWNLOAD_DIR.resolve()}")
if merge:
log("다운로드 파일 취합을 시작합니다.")
merged_file = consolidate_downloads(selected_accounts, DOWNLOAD_DIR, failures)
log(f"통합 엑셀 생성 완료: {merged_file.resolve()}")
if failures:
log("실패한 계정 목록")
for account, message in failures:
log(f"- {account.code} {account.name}: {message}")
should_keep_browser_open = True
if pause_on_failure and not headless:
log("실패가 있어 Chrome 창을 바로 닫지 않습니다.")
log("화면을 확인한 뒤 Enter를 누르면 스크립트만 종료되고 Chrome 창은 유지됩니다.")
input("확인 후 Enter: ")
raise RuntimeError("일부 계정 다운로드에 실패했습니다. 통합 파일의 오류_누락 시트를 확인하세요.")
finally:
if should_keep_browser_open and pause_on_failure and not headless:
log("진단을 위해 Chrome 창을 유지합니다. 필요하면 직접 닫으세요.")
else:
driver.quit()
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(description="WEHAGO 계정별원장 엑셀 자동 다운로드")
parser.add_argument("--start", type=int, default=0, help="0부터 시작하는 시작 순번")
parser.add_argument("--end", type=int, default=None, help="0부터 시작하는 끝 순번. 이 순번은 포함하지 않음")
parser.add_argument("--headless", action="store_true", help="브라우저 창 없이 실행. 로그인/화면 확인 뒤에는 권장")
parser.add_argument("--no-merge", action="store_true", help="다운로드 후 통합 엑셀을 만들지 않음")
parser.add_argument("--merge-only", action="store_true", help="브라우저 자동화 없이 기존 다운로드 파일만 취합")
parser.add_argument("--no-pause-on-failure", action="store_true", help="실패해도 화면 확인 대기 없이 종료")
parser.add_argument("--reset-ledger-url", action="store_true", help="저장된 계정별원장 주소를 삭제하고 WEHAGO 메인부터 시작")
return parser.parse_args()
if __name__ == "__main__":
args = parse_args()
try:
if args.reset_ledger_url and LEDGER_URL_FILE.exists():
LEDGER_URL_FILE.unlink()
log(f"저장된 계정별원장 주소를 삭제했습니다: {LEDGER_URL_FILE.resolve()}")
selected = ACCOUNTS[args.start:args.end]
if args.merge_only:
merged = consolidate_downloads(selected, DOWNLOAD_DIR)
log(f"통합 엑셀 생성 완료: {merged.resolve()}")
else:
run(
ACCOUNTS,
start_index=args.start,
end_index=args.end,
headless=args.headless,
merge=not args.no_merge,
pause_on_failure=not args.no_pause_on_failure,
)
except KeyboardInterrupt:
log("사용자가 중지했습니다.")
except RuntimeError as exc:
log(str(exc))
sys.exit(1)
except WebDriverException as exc:
log(f"Chrome/Selenium 오류: {exc}")
log("Chrome이 이미 실행 중이면 모두 닫고 다시 시도하세요.")
sys.exit(1)
Reference in New Issue View Git Blame Copy Permalink