diff --git a/data_detact_260422.py b/data_detact_260422.py new file mode 100755 index 0000000..2ad447a --- /dev/null +++ b/data_detact_260422.py @@ -0,0 +1,1244 @@ +from __future__ import annotations + +import argparse +import re +import shutil +from collections import defaultdict +from copy import copy +from dataclasses import dataclass, field +from datetime import date, datetime +from difflib import SequenceMatcher +from pathlib import Path +from typing import Iterable + +from openpyxl import load_workbook + + +DEFAULT_DIR = Path("downloads") / "wehago_account_ledger_20260421" +DEFAULT_LEDGER = DEFAULT_DIR / "\ud1b5\ud569_\uacc4\uc815\ubcc4\uc6d0\uc7a5_20260422_edit_ver.2.xlsx" +DEFAULT_VOUCHER = DEFAULT_DIR / "\uc804\ud45c\ub0b4\uc5ed\uc870\ud68c_2025_20250101_20251231_260422.xlsx" + +LEDGER_OUT_NAME = "ledger_with_matched_voucher_detected_context60_v10_20260422.xlsx" +VOUCHER_OUT_NAME = "voucher_with_matched_ledger_detected_context60_v10_20260422.xlsx" + +LEDGER_HEADERS = [ + "matched_\uac00\uc804\ud45c\ubc88\ud638", + "matched_\uacc4\uc815\ucf54\ub4dc", + "matched_\uacc4\uc815\uba85\uce6d", + "matched_\ucc28\ubcc0\uacf5\uae09\uac00", + "matched_\ub300\ubcc0\uacf5\uae09\uac00", + "matched_\ud655\uc815\uc804\ud45c\ubc88\ud638", + "matched_\uac00\uc804\ud45c\uc791\uc131\uc77c", + "matched_\uc77c\uc790\ucc28\uc774\uc77c", + "matched_\uac80\uc99d\uadfc\uac70", + "review_flag", + "review_memo", +] + +VOUCHER_HEADERS = [ + "matched_\uacc4\uc815\ucf54\ub4dc", + "matched_\uacc4\uc815\uba85", + "matched_\uc77c\uc790", + "matched_\ucc28\ubcc0", + "matched_\ub300\ubcc0", + "matched_\uc804\ud45c\ubc88\ud638", + "matched_\uc77c\uc790\ucc28\uc774\uc77c", + "matched_\uac80\uc99d\uadfc\uac70", + "review_flag", + "review_memo", +] + +# Strict rules for balance-sheet, tax, revenue, and other accounts where +# counterparty lines must not be treated as the same accounting line. +EXACT_RULES = { + "10110501": {"103"}, + "10115301": {"135"}, + "20110101": {"251"}, + "20111103": {"253"}, + "20111105": {"253"}, + "20111108": {"253"}, + "20111109": {"253"}, + "20111111": {"253"}, + "20112901": {"255"}, +} + +PREFIX_RULES = [ + ("101101", {"101"}), + ("101105", {"103"}), + ("101106", {"104"}), + ("101107", {"106", "123"}), + ("101111", {"108", "251"}), + ("101115", {"108"}), + ("101119", {"120"}), + ("101129", {"114"}), + ("101137", {"131"}), + ("101139", {"133"}), + ("101153", {"135"}), + ("101155", {"136"}), + ("101157", {"137"}), + ("101159", {"138"}), + ("201101", {"251"}), + ("201111", {"253"}), + ("20111501", {"254"}), + ("20111503", {"254"}), + ("20111505", {"275"}), + ("20111507", {"274"}), + ("20111509", {"276"}), + ("20111511", {"276"}), + ("201115", {"254"}), + ("201129", {"255"}), + ("40110101", {"411"}), + ("40110102", {"412"}), + ("40110103", {"415"}), + ("40110301", {"413"}), + ("40110401", {"414"}), + ("501701", {"602"}), + ("701101", {"901"}), + ("701103", {"903"}), + ("701105", {"905"}), + ("701107", {"906"}), + ("701109", {"908"}), + ("701111", {"914"}), + ("701115", {"930"}), + ("703101", {"931"}), + ("703103", {"933"}), + ("703105", {"935"}), + ("703107", {"937"}), + ("703125", {"960"}), + ("901", {"998"}), +] + + +@dataclass +class VoucherLine: + id: int + row: int + group_id: str + draft_no: str + confirmed_no: str + draft_date: date | None + side: str + amount: int + account_code: str + account_name: str + desc: str + vendor: str + group_blob: str + debit_supply: int + credit_supply: int + tax_debit: int + tax_credit: int + norm_desc: str + norm_vendor: str + norm_blob: str + match_id: int | None = None + score: int = 0 + reason: str = "" + date_diff: int | None = None + match_ledger_ids: list[int] = field(default_factory=list) + + +@dataclass +class VoucherAggregate: + id: int + group_id: str + row: int + draft_no: str + confirmed_no: str + draft_date: date | None + side: str + amount: int + account_code: str + account_name: str + desc: str + vendor: str + group_blob: str + line_ids: list[int] + norm_desc: str + norm_vendor: str + norm_blob: str + match_id: int | None = None + score: int = 0 + reason: str = "" + date_diff: int | None = None + + +@dataclass +class LedgerLine: + id: int + row: int + ledger_date: date + side: str + amount: int + account_code: str + account_name: str + desc: str + vendor: str + debit: int + credit: int + voucher_no: str + source_row_no: str + norm_desc: str + norm_vendor: str + norm_blob: str + match_kind: str = "" + match_id: int | None = None + score: int = 0 + reason: str = "" + date_diff: int | None = None + + +@dataclass(order=True) +class Candidate: + sort_key: tuple + ledger_id: int = field(compare=False) + voucher_kind: str = field(compare=False) + voucher_id: int = field(compare=False) + score: int = field(compare=False) + reason: str = field(compare=False) + date_diff: int = field(compare=False) + + +def clean(value) -> str: + return "" if value is None else str(value).strip() + + +def norm(value) -> str: + text = clean(value).lower() + text = re.sub(r"\s+", "", text) + return re.sub(r"[\(\)\[\]\{\},._\-/\\:;*×%◇]", "", text) + + +def norm_party(value) -> str: + text = clean(value).lower() + text = text.replace("㈜", "") + text = text.replace("(주)", "").replace("（주）", "") + text = text.replace("주식회사", "").replace("주.", "") + return norm(text) + + +def amount(value) -> int: + if value is None or value == "": + return 0 + if isinstance(value, bool): + return int(value) + if isinstance(value, (int, float)): + return int(round(value)) + try: + return int(round(float(str(value).replace(",", "").strip()))) + except ValueError: + return 0 + + +def ymd(value) -> date | None: + if value is None or value == "": + return None + if isinstance(value, datetime): + return value.date() + if isinstance(value, date): + return value + text = str(value).strip().replace(".0", "") + match = re.match(r"^(\d{4})(\d{2})(\d{2})$", text) + if match: + return date(int(match.group(1)), int(match.group(2)), int(match.group(3))) + match = re.match(r"^(\d{4})[-/.](\d{1,2})[-/.](\d{1,2})$", text) + if match: + return date(int(match.group(1)), int(match.group(2)), int(match.group(3))) + return None + + +def parse_ledger_date(value) -> date | None: + if isinstance(value, datetime): + return value.date() + if isinstance(value, date): + return value + text = clean(value) + match = re.match(r"^(\d{1,2})[-/.](\d{1,2})$", text) + if match: + return date(2025, int(match.group(1)), int(match.group(2))) + return ymd(text) + + +def date_from_voucher_no(value) -> date | None: + match = re.search(r"-(\d{8})-", clean(value)) + return ymd(match.group(1)) if match else None + + +def mmdd(value: date | None) -> str: + return value.strftime("%m-%d") if value else "" + + +def group_no(value) -> str: + return re.sub(r"-\d+$", "", clean(value)) + + +def draft_group_no(value) -> str: + return group_no(value) + + +def setting_group_no(value: str) -> str: + match = re.search(r"(11-\d{8}-[A-Z0-9]+)-0*(\d+)-0*\d+", clean(value)) + if not match: + return "" + return f"{match.group(1)}-{int(match.group(2))}" + + +def setting_group_refs(value: str) -> list[str]: + refs = [] + for match in re.finditer(r"(11-\d{8}-[A-Z0-9]+)-0*(\d+)-0*\d+", clean(value)): + refs.append(f"{match.group(1)}-{int(match.group(2))}") + return list(dict.fromkeys(refs)) + + +def account_base(name: str) -> str: + text = clean(name) + if ")" in text and text.index(")") <= 3: + text = text.split(")", 1)[1] + text = re.sub(r"\(.*?\)|\[.*?\]", "", text) + return norm(text) + + +ACCOUNT_EQUIVALENT_GROUPS = [ + {"외상매출금", "용역미수금"}, + {"단기대여금", "관계회사단기대여금"}, +] + + +def equivalent_account_names(left_name: str, right_name: str) -> bool: + left = account_base(left_name) + right = account_base(right_name) + if not left or not right: + return False + if left in right or right in left: + return True + for group in ACCOUNT_EQUIVALENT_GROUPS: + normalized = {norm(item) for item in group} + if any(item in left for item in normalized) and any(item in right for item in normalized): + return True + return False + + +def balance_sheet_context_accounts(voucher, ledger: LedgerLine) -> bool: + voucher_code = clean(voucher.account_code) + return ledger.account_code in {"108", "114", "131", "251"} and voucher_code[:3] in {"101", "201"} + + +def economic_side_amount(side: str, line_amount: int) -> tuple[str, int, str]: + if line_amount < 0: + flipped = "credit" if side == "debit" else "debit" + return flipped, abs(line_amount), "negative_amount_side_flip" + return side, line_amount, "" + + +def reversal_context_accounts(voucher, ledger: LedgerLine) -> bool: + voucher_code = clean(voucher.account_code) + receivable_ledger = ledger.account_code in {"108", "251"} + payable_voucher = voucher_code.startswith("201111") or voucher_code.startswith("201101") + return receivable_ledger and payable_voucher + + +def negative_reversal_context(voucher, ledger: LedgerLine) -> bool: + ledger_side, ledger_amount, ledger_reason = economic_side_amount(ledger.side, ledger.amount) + voucher_side, voucher_amount, voucher_reason = economic_side_amount(voucher.side, voucher.amount) + if not ledger_reason and not voucher_reason: + return False + if (ledger_side, ledger_amount) != (voucher_side, voucher_amount): + return False + return reversal_context_accounts(voucher, ledger) + + +def source_line_for_multi_sum(line: VoucherLine) -> bool: + if line.account_code.startswith("2"): + return False + if line.account_code.startswith("101101") or line.account_code.startswith("101105"): + return False + return True + + +def allowed_codes(satis_code: str) -> set[str]: + code = clean(satis_code).replace(".0", "") + if code in EXACT_RULES: + return EXACT_RULES[code] + for prefix, allowed in PREFIX_RULES: + if code.startswith(prefix): + return allowed + return set() + + +def compatible_account(voucher, ledger: LedgerLine) -> tuple[bool, str]: + allowed = allowed_codes(voucher.account_code) + if allowed and ledger.account_code in allowed: + return True, "semantic_code_ok" + + # Cost/expense accounts often differ only by SATIS detail and WEHAGO's + # 6xx/8xx split. Use the normalized account-name base as a secondary rule. + if voucher.account_code[:1] in ("5", "6") and ledger.account_code[:1] in ("6", "8"): + if equivalent_account_names(voucher.account_name, ledger.account_name): + return True, "account_name_base_ok" + + if equivalent_account_names(voucher.account_name, ledger.account_name): + return True, "account_name_equivalent_ok" + + return False, "semantic_mismatch" + + +def ledger_vendor_key(vendor: str) -> str: + text = clean(vendor) + if "/" in text: + text = text.split("/")[-1] + return norm_party(text) + + +def vendor_status(ledger: LedgerLine, voucher) -> str: + ledger_key = ledger_vendor_key(ledger.vendor) + voucher_key = norm_party(voucher.vendor) + voucher_blob = voucher.norm_blob + if not ledger_key or not voucher_key: + return "missing" + if ledger_key in voucher_key or voucher_key in ledger_key or ledger_key in voucher_blob: + return "match" + if "/" in clean(ledger.vendor) and len(ledger_key) >= 2: + return "mismatch" + return "weak_mismatch" + + +def date_score(diff: int) -> int: + if diff == 0: + return 140 + if diff <= 3: + return 120 + if diff <= 7: + return 90 + if diff <= 15: + return 60 + if diff <= 31: + return 35 + if diff <= 60: + return 5 + if diff <= 90: + return -35 + if diff <= 180: + return -110 + return -190 + + +def related_labor_desc(ledger: LedgerLine, voucher) -> bool: + left = ledger.norm_desc + right = voucher.norm_blob + labor_cost = "인건비" in right + labor_work = any(token in left for token in ["근무대가", "야간근무", "휴일근무", "휴일및야간근무"]) + if not labor_cost or not labor_work: + return False + ratio = SequenceMatcher(None, ledger.norm_desc, voucher.norm_desc).ratio() if voucher.norm_desc else 0 + return ratio >= 0.45 or any(token in right for token in ["책임감리", "건설사업관리", "고속도로", "터널"]) + + +def best_voucher_date(voucher, ledger_date: date) -> tuple[int, str]: + dates = [] + if voucher.draft_date: + dates.append(("draft", voucher.draft_date)) + confirmed_date = date_from_voucher_no(voucher.confirmed_no) + if confirmed_date: + dates.append(("confirmed", confirmed_date)) + if not dates: + return 9999, "missing" + basis, best = min(dates, key=lambda item: abs((item[1] - ledger_date).days)) + return abs((best - ledger_date).days), basis + + +def strong_desc_similarity(ledger: LedgerLine, voucher) -> tuple[bool, str, int]: + if not ledger.norm_desc or not voucher.norm_desc: + return False, "", 0 + if ledger.norm_desc == voucher.norm_desc: + return True, "absolute_desc_exact", 100 + if ledger.norm_desc in voucher.norm_desc or voucher.norm_desc in ledger.norm_desc: + return True, "absolute_desc_contains", 85 + ratio = SequenceMatcher(None, ledger.norm_desc, voucher.norm_desc).ratio() + if ratio >= 0.82: + return True, f"absolute_desc_sim_{ratio:.2f}", int(70 * ratio) + if ledger.norm_desc in voucher.norm_blob: + return True, "absolute_ledger_desc_in_voucher_context", 70 + return False, "", 0 + + +def absolute_amount_desc_context(ledger: LedgerLine, voucher) -> tuple[bool, str, int]: + if ledger.amount == voucher.amount: + return False, "", 0 + if abs(ledger.amount) != abs(voucher.amount): + return False, "", 0 + if ledger.side != voucher.side: + return False, "", 0 + desc_ok, desc_reason, desc_bonus = strong_desc_similarity(ledger, voucher) + if not desc_ok: + return False, "", 0 + diff, _basis = best_voucher_date(voucher, ledger.ledger_date) + v_status = vendor_status(ledger, voucher) + if diff > 60 and v_status != "match": + return False, "", 0 + if v_status == "mismatch": + return False, "", 0 + bonus = 35 + desc_bonus + reason = f"absolute_amount_desc_similarity,{desc_reason}" + return True, reason, bonus + + +def score_pair(ledger: LedgerLine, voucher) -> tuple[int, str, int]: + ok, account_reason = compatible_account(voucher, ledger) + if not voucher.draft_date and not date_from_voucher_no(voucher.confirmed_no): + return -999, "missing_draft_date", 9999 + + v_status = vendor_status(ledger, voucher) + diff, date_basis = best_voucher_date(voucher, ledger.ledger_date) + ledger_account_base = account_base(ledger.account_name) + account_context_in_blob = bool( + ledger_account_base + and len(ledger_account_base) >= 2 + and ledger_account_base in voucher.norm_blob + ) + labor_desc_related = related_labor_desc(ledger, voucher) + desc_strong = bool( + account_context_in_blob + or labor_desc_related + or ( + ledger.norm_desc + and ( + ledger.norm_desc == voucher.norm_desc + or ledger.norm_desc in voucher.norm_desc + or voucher.norm_desc in ledger.norm_desc + or ledger.norm_desc in voucher.norm_blob + ) + ) + ) + negative_reversal = negative_reversal_context(voucher, ledger) + absolute_desc_context, absolute_desc_reason, absolute_desc_bonus = absolute_amount_desc_context(ledger, voucher) + context_vendor_override = ( + v_status == "mismatch" + and diff <= 60 + and desc_strong + and (ok or negative_reversal or voucher.account_code[:1] in ("5", "6")) + ) + account_context_override = ( + not ok + and diff <= 60 + and ( + (v_status == "match" and desc_strong) + or (negative_reversal and v_status == "match" and desc_strong) + or (absolute_desc_context and v_status in {"match", "missing", "weak_mismatch"}) + or (v_status == "match" and diff <= 7 and balance_sheet_context_accounts(voucher, ledger)) + or context_vendor_override + ) + ) + if v_status == "mismatch" and not context_vendor_override: + return -999, "vendor_mismatch", 9999 + + if not ok: + if not account_context_override: + return -999, account_reason, 9999 + account_reason = "receivable_payable_reversal_context" if negative_reversal else "context_override_account_mismatch" + + score = 35 + date_score(diff) + reasons = [account_reason, f"{date_basis}_diff_{diff}d"] + if negative_reversal: + score += 90 + ledger_flip = economic_side_amount(ledger.side, ledger.amount)[2] + voucher_flip = economic_side_amount(voucher.side, voucher.amount)[2] + if ledger_flip: + reasons.append(ledger_flip) + if voucher_flip: + reasons.append(f"voucher_{voucher_flip}") + if absolute_desc_context: + score += absolute_desc_bonus + reasons.append(absolute_desc_reason) + confirmed_date = date_from_voucher_no(voucher.confirmed_no) + if confirmed_date and abs((confirmed_date - ledger.ledger_date).days) == 0: + score += 12 + reasons.append("confirmed_date_exact") + + if ledger.norm_desc and voucher.norm_desc: + if ledger.norm_desc == voucher.norm_desc: + score += 100 + reasons.append("desc_exact") + elif ledger.norm_desc in voucher.norm_desc or voucher.norm_desc in ledger.norm_desc: + score += 75 + reasons.append("desc_contains") + else: + ratio = SequenceMatcher(None, ledger.norm_desc, voucher.norm_desc).ratio() + if ratio >= 0.72: + score += int(55 * ratio) + reasons.append(f"desc_sim_{ratio:.2f}") + + if ledger.norm_desc and ledger.norm_desc in voucher.norm_blob: + score += 55 + reasons.append("ledger_desc_in_voucher_group_context") + if voucher.norm_desc and voucher.norm_desc in ledger.norm_blob: + score += 30 + reasons.append("voucher_desc_in_ledger_context") + if account_context_in_blob: + score += 55 + reasons.append("ledger_account_in_voucher_context") + if labor_desc_related: + score += 70 + reasons.append("desc_related_labor_cost_terms") + + if v_status == "match": + score += 95 + reasons.append("vendor_strong_match") + elif context_vendor_override: + score += 35 + reasons.append("vendor_context_override_by_desc") + elif v_status == "weak_mismatch": + score -= 45 + reasons.append("vendor_weak_mismatch") + else: + score -= 10 + reasons.append("vendor_missing") + + if diff > 90 and v_status != "match": + score -= 100 + reasons.append("long_date_gap_without_vendor") + + if isinstance(voucher, VoucherAggregate): + score += 35 + reasons.append(f"group_sum_{len(voucher.line_ids)}lines") + if voucher.group_id.startswith("multi:"): + score += 45 + reasons.append("multi_voucher_sum_by_settlement_refs") + + return score, ",".join(reasons), diff + + +def load_vouchers(path: Path) -> tuple[list[VoucherLine], list[VoucherAggregate]]: + workbook = load_workbook(path, read_only=True, data_only=True) + sheet = workbook.worksheets[0] + next(sheet.iter_rows(min_row=1, max_row=1, values_only=True)) + + raw_rows = [] + groups = defaultdict(lambda: {"blob": [], "vendors": [], "descs": []}) + settlement_groups = defaultdict(lambda: {"refs": [], "rows": [], "blob": [], "vendors": [], "descs": [], "date": None, "confirmed": []}) + + for row_number, row in enumerate(sheet.iter_rows(min_row=2, values_only=True), 2): + values = list(row) + base = group_no(values[10]) + draft_base = draft_group_no(values[1]) + desc = clean(values[15]) or clean(values[16]) + vendor = clean(values[18]) + blob = " ".join( + clean(values[index]) + for index in [8, 9, 11, 12, 13, 14, 15, 16, 18, 21, 22] + if index < len(values) and values[index] is not None + ) + raw_rows.append( + { + "row": row_number, + "values": values, + "group": base, + "draft_group": draft_base, + "draft_date": date_from_voucher_no(values[1]), + "desc": desc, + "vendor": vendor, + "blob": blob, + } + ) + groups[base]["blob"].append(blob) + if vendor: + groups[base]["vendors"].append(vendor) + if desc: + groups[base]["descs"].append(desc) + settlement_refs = setting_group_refs(blob) + if settlement_refs: + settlement = settlement_groups[draft_base] + settlement["refs"].extend(settlement_refs) + settlement["rows"].append(row_number) + settlement["blob"].append(blob) + if vendor: + settlement["vendors"].append(vendor) + if desc: + settlement["descs"].append(desc) + if not settlement["date"]: + settlement["date"] = date_from_voucher_no(values[1]) + confirmed = clean(values[10]) + if confirmed: + settlement["confirmed"].append(confirmed) + + lines: list[VoucherLine] = [] + for raw in raw_rows: + values = raw["values"] + debit = amount(values[4]) + credit = amount(values[6]) + if not debit and not credit: + continue + side, line_amount = ("debit", debit) if debit and (not credit or abs(debit) >= abs(credit)) else ("credit", credit) + group_data = groups[raw["group"]] + vendor = raw["vendor"] or next(iter(group_data["vendors"]), "") + desc = raw["desc"] or next(iter(group_data["descs"]), "") + group_blob = " ".join(group_data["blob"] + [desc, vendor]) + lines.append( + VoucherLine( + id=len(lines), + row=raw["row"], + group_id=raw["group"], + draft_no=clean(values[1]), + confirmed_no=clean(values[10]), + draft_date=raw["draft_date"], + side=side, + amount=line_amount, + account_code=clean(values[2]).replace(".0", ""), + account_name=clean(values[3]), + desc=desc, + vendor=vendor, + group_blob=group_blob, + debit_supply=debit, + credit_supply=credit, + tax_debit=amount(values[5]), + tax_credit=amount(values[7]), + norm_desc=norm(desc), + norm_vendor=norm(vendor), + norm_blob=norm(group_blob), + ) + ) + + aggregates: list[VoucherAggregate] = [] + source_by_draft_group = defaultdict(list) + for line in lines: + source_by_draft_group[draft_group_no(line.draft_no)].append(line) + + by_group_key = defaultdict(list) + for line in lines: + if line.account_code.startswith("2"): + continue + key = (line.group_id, line.side, account_base(line.account_name), norm(line.vendor)) + by_group_key[key].append(line) + + for (group_id, side, _account_base, _vendor), group_lines in by_group_key.items(): + if len(group_lines) < 2: + continue + first = group_lines[0] + total = sum(item.amount for item in group_lines) + desc = " / ".join(dict.fromkeys(item.desc for item in group_lines if item.desc)) + aggregates.append( + VoucherAggregate( + id=len(aggregates), + group_id=group_id, + row=first.row, + draft_no="; ".join(item.draft_no for item in group_lines), + confirmed_no="; ".join(dict.fromkeys(item.confirmed_no for item in group_lines if item.confirmed_no)), + draft_date=first.draft_date, + side=side, + amount=total, + account_code=first.account_code, + account_name=first.account_name, + desc=desc or first.desc, + vendor=first.vendor, + group_blob=" ".join(item.group_blob for item in group_lines), + line_ids=[item.id for item in group_lines], + norm_desc=norm(desc or first.desc), + norm_vendor=norm(first.vendor), + norm_blob=norm(" ".join(item.group_blob for item in group_lines)), + ) + ) + + for settlement_group, data in settlement_groups.items(): + refs = [ref for ref in dict.fromkeys(data["refs"]) if ref != settlement_group] + if len(refs) < 2: + continue + source_lines = [ + line + for ref in refs + for line in source_by_draft_group.get(ref, []) + if source_line_for_multi_sum(line) + ] + by_multi_key = defaultdict(list) + for line in source_lines: + by_multi_key[(line.side, norm(line.vendor))].append(line) + + for (side, _vendor), group_lines in by_multi_key.items(): + if len(group_lines) < 2: + continue + first = group_lines[0] + total = sum(item.amount for item in group_lines) + if not total: + continue + desc = " / ".join(dict.fromkeys(item.desc for item in group_lines if item.desc)) + vendor = first.vendor or next(iter(data["vendors"]), "") + blob = " ".join(data["blob"] + [item.group_blob for item in group_lines]) + aggregates.append( + VoucherAggregate( + id=len(aggregates), + group_id=f"multi:{settlement_group}", + row=min(item.row for item in group_lines), + draft_no="; ".join(item.draft_no for item in group_lines), + confirmed_no="; ".join(dict.fromkeys(data["confirmed"] + [item.confirmed_no for item in group_lines if item.confirmed_no])), + draft_date=data["date"] or first.draft_date, + side=side, + amount=total, + account_code="; ".join(dict.fromkeys(item.account_code for item in group_lines if item.account_code)), + account_name="; ".join(dict.fromkeys(item.account_name for item in group_lines if item.account_name)), + desc=desc or " / ".join(dict.fromkeys(data["descs"])), + vendor=vendor, + group_blob=blob, + line_ids=[item.id for item in group_lines], + norm_desc=norm(desc or " / ".join(dict.fromkeys(data["descs"]))), + norm_vendor=norm(vendor), + norm_blob=norm(blob), + ) + ) + + return lines, aggregates + + +def load_ledgers(path: Path) -> list[LedgerLine]: + workbook = load_workbook(path, read_only=True, data_only=True) + sheet = workbook.worksheets[0] + next(sheet.iter_rows(min_row=1, max_row=1, values_only=True)) + ledgers: list[LedgerLine] = [] + + for row_number, row in enumerate(sheet.iter_rows(min_row=2, values_only=True), 2): + values = list(row) + ledger_date = parse_ledger_date(values[5]) + desc = clean(values[6]) + debit = amount(values[8]) + credit = amount(values[9]) + if not ledger_date or desc.startswith("[") or (not debit and not credit): + continue + # Avoid carrying opening balance lines into matching without relying on a Korean literal. + if not values[5] and not values[11]: + continue + side, line_amount = ("debit", debit) if debit and (not credit or abs(debit) >= abs(credit)) else ("credit", credit) + vendor = clean(values[7]) + blob = " ".join([desc, vendor, clean(values[1]), clean(values[11])]) + ledgers.append( + LedgerLine( + id=len(ledgers), + row=row_number, + ledger_date=ledger_date, + side=side, + amount=line_amount, + account_code=clean(values[0]).replace(".0", ""), + account_name=clean(values[1]), + desc=desc, + vendor=vendor, + debit=debit, + credit=credit, + voucher_no=clean(values[11]), + source_row_no=clean(values[4]), + norm_desc=norm(desc), + norm_vendor=norm(vendor), + norm_blob=norm(blob), + ) + ) + return ledgers + + +def build_candidates(ledgers: list[LedgerLine], voucher_lines: list[VoucherLine], aggregates: list[VoucherAggregate]) -> list[Candidate]: + line_by_amount = defaultdict(list) + line_by_abs_amount = defaultdict(list) + line_by_economic_amount = defaultdict(list) + aggregate_by_amount = defaultdict(list) + for voucher in voucher_lines: + line_by_amount[(voucher.side, voucher.amount)].append(voucher) + line_by_abs_amount[(voucher.side, abs(voucher.amount))].append(voucher) + economic_side, economic_amount, _reason = economic_side_amount(voucher.side, voucher.amount) + line_by_economic_amount[(economic_side, economic_amount)].append(voucher) + for aggregate in aggregates: + aggregate_by_amount[(aggregate.side, aggregate.amount)].append(aggregate) + + candidates: list[Candidate] = [] + for ledger in ledgers: + for voucher in line_by_amount.get((ledger.side, ledger.amount), []): + score, reason, diff = score_pair(ledger, voucher) + if score >= 135: + exact_desc = int(bool(ledger.norm_desc and ledger.norm_desc == voucher.norm_desc)) + strong_vendor = int("vendor_strong_match" in reason) + candidates.append( + Candidate( + sort_key=(score, strong_vendor, exact_desc, -diff), + ledger_id=ledger.id, + voucher_kind="line", + voucher_id=voucher.id, + score=score, + reason=reason, + date_diff=diff, + ) + ) + for voucher in line_by_abs_amount.get((ledger.side, abs(ledger.amount)), []): + if voucher.amount == ledger.amount: + continue + score, reason, diff = score_pair(ledger, voucher) + if "absolute_amount_desc_similarity" not in reason: + continue + if score >= 150: + exact_desc = int(bool(ledger.norm_desc and ledger.norm_desc == voucher.norm_desc)) + strong_vendor = int("vendor_strong_match" in reason) + candidates.append( + Candidate( + sort_key=(score, strong_vendor, exact_desc, -diff), + ledger_id=ledger.id, + voucher_kind="absline", + voucher_id=voucher.id, + score=score, + reason=reason, + date_diff=diff, + ) + ) + ledger_economic_side, ledger_economic_amount, ledger_economic_reason = economic_side_amount(ledger.side, ledger.amount) + if ledger_economic_reason: + for voucher in line_by_economic_amount.get((ledger_economic_side, ledger_economic_amount), []): + if (voucher.side, voucher.amount) == (ledger.side, ledger.amount): + continue + score, reason, diff = score_pair(ledger, voucher) + if "receivable_payable_reversal_context" not in reason: + continue + if score >= 155: + exact_desc = int(bool(ledger.norm_desc and ledger.norm_desc == voucher.norm_desc)) + strong_vendor = int("vendor_strong_match" in reason) + candidates.append( + Candidate( + sort_key=(score, strong_vendor, exact_desc, -diff), + ledger_id=ledger.id, + voucher_kind="line", + voucher_id=voucher.id, + score=score, + reason=reason, + date_diff=diff, + ) + ) + for aggregate in aggregate_by_amount.get((ledger.side, ledger.amount), []): + score, reason, diff = score_pair(ledger, aggregate) + if score >= 155: + exact_desc = int(bool(ledger.norm_desc and ledger.norm_desc == aggregate.norm_desc)) + strong_vendor = int("vendor_strong_match" in reason) + kind = "multi" if aggregate.group_id.startswith("multi:") else "aggregate" + candidates.append( + Candidate( + sort_key=(score, strong_vendor, exact_desc, -diff), + ledger_id=ledger.id, + voucher_kind=kind, + voucher_id=aggregate.id, + score=score, + reason=reason, + date_diff=diff, + ) + ) + candidates.sort(reverse=True) + return candidates + + +def assign_matches(ledgers: list[LedgerLine], voucher_lines: list[VoucherLine], aggregates: list[VoucherAggregate], candidates: list[Candidate]) -> None: + used_ledgers: set[int] = set() + used_voucher_lines: set[int] = set() + + ordered_candidates = ( + [item for item in candidates if item.voucher_kind not in {"multi", "absline"}] + + [item for item in candidates if item.voucher_kind == "multi"] + + [item for item in candidates if item.voucher_kind == "absline"] + ) + for candidate in ordered_candidates: + if candidate.ledger_id in used_ledgers: + continue + if candidate.voucher_kind in {"line", "absline"}: + voucher = voucher_lines[candidate.voucher_id] + line_ids = [voucher.id] + else: + voucher = aggregates[candidate.voucher_id] + line_ids = voucher.line_ids + + if any(line_id in used_voucher_lines for line_id in line_ids): + continue + + ledger = ledgers[candidate.ledger_id] + ledger.match_kind = candidate.voucher_kind + ledger.match_id = candidate.voucher_id + ledger.score = candidate.score + ledger.reason = candidate.reason + ledger.date_diff = candidate.date_diff + + if candidate.voucher_kind in {"line", "absline"}: + voucher.match_id = ledger.id + voucher.score = candidate.score + voucher.reason = candidate.reason + voucher.date_diff = candidate.date_diff + else: + voucher.match_id = ledger.id + voucher.score = candidate.score + voucher.reason = candidate.reason + voucher.date_diff = candidate.date_diff + for line_id in voucher.line_ids: + line = voucher_lines[line_id] + line.match_id = ledger.id + line.score = candidate.score + line.reason = candidate.reason + line.date_diff = candidate.date_diff + + used_ledgers.add(ledger.id) + used_voucher_lines.update(line_ids) + + +def compact_range(values: Iterable[str]) -> str: + numbers = [] + text_values = [] + for value in values: + text = clean(value) + if not text: + continue + try: + numbers.append(int(float(text))) + except ValueError: + text_values.append(text) + if numbers and not text_values: + numbers = sorted(dict.fromkeys(numbers)) + if len(numbers) > 1 and numbers == list(range(numbers[0], numbers[-1] + 1)): + return f"{numbers[0]}-{numbers[-1]}" + return "; ".join(str(item) for item in numbers) + return "; ".join(dict.fromkeys(text_values + [str(item) for item in numbers])) + + +def ledger_group_key(ledger: LedgerLine) -> tuple: + return ( + ledger.side, + ledger.ledger_date, + ledger.account_code, + account_base(ledger.account_name), + ledger_vendor_key(ledger.vendor), + ledger.norm_desc, + ) + + +def make_ledger_group_probe(group_lines: list[LedgerLine], total: int) -> LedgerLine: + first = group_lines[0] + voucher_numbers = "; ".join(item.voucher_no for item in group_lines if item.voucher_no) + source_rows = compact_range(item.source_row_no for item in group_lines) + blob = " ".join([first.desc, first.vendor, first.account_name, voucher_numbers, source_rows]) + return LedgerLine( + id=-1, + row=min(item.row for item in group_lines), + ledger_date=first.ledger_date, + side=first.side, + amount=total, + account_code=first.account_code, + account_name=first.account_name, + desc=first.desc, + vendor=first.vendor, + debit=total if first.side == "debit" else 0, + credit=total if first.side == "credit" else 0, + voucher_no=voucher_numbers, + source_row_no=source_rows, + norm_desc=first.norm_desc, + norm_vendor=first.norm_vendor, + norm_blob=norm(blob), + ) + + +def assign_ledger_group_matches(ledgers: list[LedgerLine], voucher_lines: list[VoucherLine]) -> int: + unmatched_groups = defaultdict(list) + for ledger in ledgers: + if ledger.match_id is None: + unmatched_groups[ledger_group_key(ledger)].append(ledger) + + available_vouchers = defaultdict(list) + for voucher in voucher_lines: + if voucher.match_id is None: + available_vouchers[(voucher.side, voucher.amount)].append(voucher) + + group_candidates = [] + for group_lines in unmatched_groups.values(): + if len(group_lines) < 2: + continue + total = sum(item.amount for item in group_lines) + if not total: + continue + probe = make_ledger_group_probe(group_lines, total) + for voucher in available_vouchers.get((probe.side, probe.amount), []): + score, reason, diff = score_pair(probe, voucher) + if score < 185: + continue + reason = ",".join( + [ + reason, + f"ledger_group_sum_to_single_voucher", + f"group_sum_{len(group_lines)}ledger_lines", + f"group_source_rows_{probe.source_row_no}", + f"group_vouchers_{probe.voucher_no}", + ] + ) + exact_desc = int(bool(probe.norm_desc and probe.norm_desc == voucher.norm_desc)) + strong_vendor = int("vendor_strong_match" in reason) + group_candidates.append((score, strong_vendor, exact_desc, -diff, group_lines, voucher, reason, diff)) + + matched_count = 0 + used_ledgers = {ledger.id for ledger in ledgers if ledger.match_id is not None} + used_vouchers = {voucher.id for voucher in voucher_lines if voucher.match_id is not None} + for score, _strong_vendor, _exact_desc, _neg_diff, group_lines, voucher, reason, diff in sorted( + group_candidates, key=lambda item: item[:4], reverse=True + ): + if voucher.id in used_vouchers: + continue + if any(ledger.id in used_ledgers for ledger in group_lines): + continue + ledger_ids = [ledger.id for ledger in group_lines] + for ledger in group_lines: + ledger.match_kind = "ledger_group" + ledger.match_id = voucher.id + ledger.score = score + ledger.reason = reason + ledger.date_diff = diff + used_ledgers.add(ledger.id) + matched_count += 1 + voucher.match_id = group_lines[0].id + voucher.match_ledger_ids = ledger_ids + voucher.score = score + voucher.reason = reason + voucher.date_diff = diff + used_vouchers.add(voucher.id) + return matched_count + + +def review(diff: int | None) -> tuple[str, str]: + if diff is None: + return "", "" + if diff > 90: + return "REVIEW_LONG_DATE_GAP", f"draft-ledger date diff {diff}d" + if diff >= 31: + return "REVIEW_DATE_GAP", f"draft-ledger date diff {diff}d" + return "", "" + + +def style_like(target, source) -> None: + if not source.has_style: + return + target.font = copy(source.font) + target.fill = copy(source.fill) + target.border = copy(source.border) + target.alignment = copy(source.alignment) + target.number_format = source.number_format + target.protection = copy(source.protection) + + +def write_ledger_output( + source_path: Path, + output_path: Path, + ledgers: list[LedgerLine], + voucher_lines: list[VoucherLine], + aggregates: list[VoucherAggregate], +) -> None: + ledger_by_row = {item.row: item for item in ledgers} + workbook = load_workbook(source_path) + sheet = workbook.worksheets[0] + + for offset, header in enumerate(LEDGER_HEADERS): + cell = sheet.cell(row=1, column=15 + offset, value=header) + style_like(cell, sheet.cell(row=1, column=14)) + sheet.column_dimensions[cell.column_letter].width = 22 if offset != 8 else 60 + + for row_number in range(2, sheet.max_row + 1): + ledger = ledger_by_row.get(row_number) + values = [None] * len(LEDGER_HEADERS) + if ledger and ledger.match_id is not None: + if ledger.match_kind in ("line", "absline", "ledger_group"): + match = voucher_lines[ledger.match_id] + debit_value = match.debit_supply + credit_value = match.credit_supply + else: + match = aggregates[ledger.match_id] + debit_value = match.amount if match.side == "debit" else 0 + credit_value = match.amount if match.side == "credit" else 0 + flag, memo = review(ledger.date_diff) + values = [ + match.draft_no, + match.account_code, + match.account_name, + debit_value, + credit_value, + match.confirmed_no, + mmdd(match.draft_date), + ledger.date_diff, + ledger.reason, + flag, + memo, + ] + for offset, value in enumerate(values): + sheet.cell(row=row_number, column=15 + offset, value=value) + + workbook.save(output_path) + + +def write_voucher_output(source_path: Path, output_path: Path, ledgers: list[LedgerLine], voucher_lines: list[VoucherLine]) -> None: + ledger_by_id = {item.id: item for item in ledgers} + voucher_by_row = {item.row: item for item in voucher_lines} + workbook = load_workbook(source_path) + sheet = workbook.worksheets[0] + + for offset, header in enumerate(VOUCHER_HEADERS): + cell = sheet.cell(row=1, column=24 + offset, value=header) + style_like(cell, sheet.cell(row=1, column=23)) + sheet.column_dimensions[cell.column_letter].width = 18 if offset != 7 else 60 + + for row_number in range(2, sheet.max_row + 1): + voucher = voucher_by_row.get(row_number) + values = [None] * len(VOUCHER_HEADERS) + if voucher and voucher.match_id is not None: + matched_ledgers = [ledger_by_id[item] for item in voucher.match_ledger_ids] if voucher.match_ledger_ids else [ledger_by_id[voucher.match_id]] + first = matched_ledgers[0] + flag, memo = review(voucher.date_diff) + values = [ + "; ".join(dict.fromkeys(item.account_code for item in matched_ledgers)), + "; ".join(dict.fromkeys(item.account_name for item in matched_ledgers)), + mmdd(first.ledger_date), + sum(item.debit for item in matched_ledgers) or None, + sum(item.credit for item in matched_ledgers) or None, + "; ".join(item.voucher_no for item in matched_ledgers if item.voucher_no), + voucher.date_diff, + voucher.reason, + flag, + memo, + ] + for offset, value in enumerate(values): + sheet.cell(row=row_number, column=24 + offset, value=value) + + workbook.save(output_path) + + +def run(ledger_path: Path, voucher_path: Path, output_dir: Path) -> tuple[Path, Path, dict]: + output_dir.mkdir(parents=True, exist_ok=True) + ledger_out = output_dir / LEDGER_OUT_NAME + voucher_out = output_dir / VOUCHER_OUT_NAME + + voucher_lines, aggregates = load_vouchers(voucher_path) + ledgers = load_ledgers(ledger_path) + candidates = build_candidates(ledgers, voucher_lines, aggregates) + assign_matches(ledgers, voucher_lines, aggregates, candidates) + ledger_group_matches = assign_ledger_group_matches(ledgers, voucher_lines) + + write_ledger_output(ledger_path, ledger_out, ledgers, voucher_lines, aggregates) + write_voucher_output(voucher_path, voucher_out, ledgers, voucher_lines) + + stats = { + "ledger_rows": len(ledgers), + "voucher_lines": len(voucher_lines), + "voucher_aggregates": len(aggregates), + "candidates": len(candidates), + "ledger_group_matches": ledger_group_matches, + "matched_ledgers": sum(1 for item in ledgers if item.match_id is not None), + "matched_voucher_lines": sum(1 for item in voucher_lines if item.match_id is not None), + } + return ledger_out, voucher_out, stats + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description="Match WEHAGO ledger rows against SATIS voucher rows.") + parser.add_argument("--ledger", type=Path, default=DEFAULT_LEDGER, help="Path to ledger xlsx.") + parser.add_argument("--voucher", type=Path, default=DEFAULT_VOUCHER, help="Path to voucher xlsx.") + parser.add_argument("--output-dir", type=Path, default=DEFAULT_DIR, help="Directory for output xlsx files.") + return parser.parse_args() + + +def main() -> None: + args = parse_args() + ledger_out, voucher_out, stats = run(args.ledger, args.voucher, args.output_dir) + print(f"ledger_out={ledger_out}") + print(f"voucher_out={voucher_out}") + for key, value in stats.items(): + print(f"{key}={value}") + + +if __name__ == "__main__": + main() diff --git a/requirements.txt b/requirements.txt index 1bde9ce..30e36c8 100644 --- a/requirements.txt +++ b/requirements.txt @@ -2,4 +2,5 @@ python-dotenv requests pandas openpyxl - +selenium +webdriver-manager diff --git a/wehago_data_download_260421.py b/wehago_data_download_260421.py new file mode 100755 index 0000000..10cfea7 --- /dev/null +++ b/wehago_data_download_260421.py @@ -0,0 +1,1723 @@ +""" +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 형태로 저장됩니다. +DOWNLOAD_DIR = BASE_DIR / "downloads" / f"wehago_account_ledger_{datetime.now():%Y%m%d}" +DIAGNOSTIC_DIR = BASE_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") + + service = Service(ChromeDriverManager().install(), log_output=subprocess.DEVNULL) + driver = webdriver.Chrome(service=service, 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)