From 0da73f6c6ef86b9b6dd7c88979081e39935baa72 Mon Sep 17 00:00:00 2001 From: dat972 Date: Sun, 5 Jul 2026 22:38:38 -0500 Subject: [PATCH] Initial commit: FCCS extraction tool --- config.ini | 32 +++++ fccs_config.py | 70 ++++++++++ fccs_export.py | 320 +++++++++++++++++++++++++++++++++++++++++++++ fccs_folders.txt | 11 ++ fccs_reorganize.py | 142 ++++++++++++++++++++ fccs_scan.py | 39 ++++++ 6 files changed, 614 insertions(+) create mode 100644 config.ini create mode 100644 fccs_config.py create mode 100755 fccs_export.py create mode 100644 fccs_folders.txt create mode 100644 fccs_reorganize.py create mode 100644 fccs_scan.py diff --git a/config.ini b/config.ini new file mode 100644 index 0000000..43576e5 --- /dev/null +++ b/config.ini @@ -0,0 +1,32 @@ +[paths] +# Point these at the current engagement's directories +backup_dir = C:\Migration\FCCSBackup +export_dir = C:\Migration\Export +output_dir = C:\Migration\Output +drawer_id_file = C:\Migration\drawer_ids.txt +completed_file = C:\Migration\completed.txt +log_file = C:\Migration\run_log.txt +screenshot_dir = C:\Migration\screenshots +folder_list = fccs_folders.txt + +[timeouts] +nav_timeout = 15 +dialog_timeout = 15 +progress_appear = 30 +progress_finish = 1800 +settle = 1.0 +confirm_timeout = 30 + +[controls] +main_class = CSIFCAB +drawer_box_title = << Enter dra&wer ID or name >> +go_button_title = Go +dialog_title = Send To File Location +dialog_class = #32770 +select_btn_title = &Select -> +ok_btn_title = OK +progress_title = Exporting Documents +progress_class = #32770 +saved_title = "Send to" file saved +saved_class = #32770 +saved_ok_title = OK diff --git a/fccs_config.py b/fccs_config.py new file mode 100644 index 0000000..20b691d --- /dev/null +++ b/fccs_config.py @@ -0,0 +1,70 @@ +""" +Shared configuration, logging, and utilities for the FCCS extraction tool. +""" + +import os +import sys +import argparse +import configparser +from datetime import datetime + +DEFAULT_CONFIG = "config.ini" + + +def parse_args(): + """Parse the optional --config argument common to all scripts.""" + parser = argparse.ArgumentParser() + parser.add_argument( + "--config", + default=None, + help=f"Path to config file (default: {DEFAULT_CONFIG} in script directory)", + ) + return parser.parse_args() + + +def load_config(path=None): + """Read config.ini and return a ConfigParser object.""" + if path is None: + script_dir = os.path.dirname(os.path.abspath(sys.argv[0])) + path = os.path.join(script_dir, DEFAULT_CONFIG) + if not os.path.exists(path): + print(f"ERROR: config file not found: {path}") + sys.exit(1) + cfg = configparser.ConfigParser() + cfg.read(path, encoding="utf-8") + return cfg + + +def make_logger(log_file): + """Return a log() function that writes timestamped lines to console and file.""" + def log(msg): + line = f"[{datetime.now():%Y-%m-%d %H:%M:%S}] {msg}" + print(line) + try: + with open(log_file, "a", encoding="utf-8") as f: + f.write(line + "\n") + except Exception: + pass + return log + + +def load_folder_list(path): + """Load known FCCS folder names from file, one per line.""" + if not os.path.exists(path): + print(f"ERROR: folder list file not found: {path}") + sys.exit(1) + folders = [] + with open(path, "r", encoding="utf-8") as f: + for line in f: + s = line.strip() + if s and not s.startswith("#"): + folders.append(s) + return folders + + +def load_lines(path): + """Read non-blank, non-comment lines from a file.""" + if not os.path.exists(path): + return [] + with open(path, "r", encoding="utf-8") as f: + return [line.strip() for line in f if line.strip() and not line.strip().startswith("#")] diff --git a/fccs_export.py b/fccs_export.py new file mode 100755 index 0000000..cd047df --- /dev/null +++ b/fccs_export.py @@ -0,0 +1,320 @@ +""" +FileCabinet CS - Batch Export Automation (Step 2) +================================================= +Reads a list of drawer IDs from a file, then for each drawer: + 1. Navigates to it in the main window (type ID -> Go) + 2. Opens File > Send To > File + 3. Clicks "Select ->" (grabs all documents via the tree root) + 4. Clicks OK + 5. Waits for the "Exporting Documents" progress dialog to appear and finish + 6. Logs success/failure and moves to the next drawer + +Key safety features: + - RESUME: completed drawer IDs are appended to a progress log. On restart, + already-completed drawers are skipped. A multi-hour run that dies partway + can simply be re-run and it picks up where it left off. + - LOGGING: every action and error is timestamped to a log file. + - DEFENSIVE: unexpected states are caught, logged, screenshotted, and the + run continues with the next drawer rather than crashing. + +PREREQUISITES: + - Run from 32-bit Python (matches FileCabinet CS's architecture) + - FileCabinet CS already open, pointed at the correct data location + - Export options/checkboxes and destination already set MANUALLY once + (they persist between exports within a session) + - Neither FileCabinet nor this script running elevated (they must match) +""" + +import time +import sys +import os + +from pywinauto import Desktop, Application +from pywinauto import timings + +from fccs_config import parse_args, load_config, make_logger, load_lines + + +# --------------------------------------------------------------------------- +# HELPERS +# --------------------------------------------------------------------------- + +def load_completed(path): + """Read the set of already-completed drawer IDs for resume support.""" + if not os.path.exists(path): + return set() + with open(path, "r", encoding="utf-8") as f: + return {line.strip() for line in f if line.strip()} + + +def mark_completed(path, drawer_id): + """Append a drawer ID to the completed log (resume tracking).""" + with open(path, "a", encoding="utf-8") as f: + f.write(drawer_id + "\n") + + +def take_screenshot(window, drawer_id, tag, screenshot_dir, log): + """Save a screenshot of a window for later diagnosis of a failure.""" + from datetime import datetime + try: + os.makedirs(screenshot_dir, exist_ok=True) + fname = os.path.join( + screenshot_dir, + f"{drawer_id}_{tag}_{datetime.now():%Y%m%d_%H%M%S}.png" + ) + window.capture_as_image().save(fname) + log(f" Screenshot saved: {fname}") + except Exception as e: + log(f" Could not capture screenshot: {e}") + + +# --------------------------------------------------------------------------- +# CONNECTION +# --------------------------------------------------------------------------- + +def connect_main(ctrl, log): + """Find and connect to the running FileCabinet CS main window.""" + target = None + for w in Desktop(backend="win32").windows(): + if w.class_name() == ctrl["main_class"]: + target = w + break + if target is None: + log("ERROR: FileCabinet CS window not found. Is it open?") + sys.exit(1) + + app = Application(backend="win32").connect(handle=target.handle) + main = app.window(handle=target.handle) + log(f"Connected to: {main.window_text()}") + return app, main + + +# --------------------------------------------------------------------------- +# PER-DRAWER EXPORT +# --------------------------------------------------------------------------- + +def navigate_to_drawer(main, drawer_id, ctrl, timeouts, log): + """Type the drawer ID into the search box and click Go.""" + box = main.child_window(title=ctrl["drawer_box_title"], class_name="Edit") + box.wait("visible ready", timeout=timeouts["nav"]) + + box.set_edit_text(drawer_id) + time.sleep(timeouts["settle"]) + + main.child_window(title=ctrl["go_button_title"], class_name="Button").click() + + deadline = time.time() + timeouts["nav"] + while time.time() < deadline: + title = main.window_text() + if f"[{drawer_id}" in title: + log(f" Navigated to drawer {drawer_id}: {title}") + return True + time.sleep(0.3) + + log(f" WARNING: window title did not confirm drawer {drawer_id} " + f"(title is: {main.window_text()!r})") + return False + + +def open_send_to_file(main, app, ctrl, timeouts): + """Open File > Send To > File and return the dialog window object.""" + main.menu_select("File->Send To->File") + + dlg = app.window(title=ctrl["dialog_title"], class_name=ctrl["dialog_class"]) + dlg.wait("visible ready", timeout=timeouts["dialog"]) + return dlg + + +def perform_export(dlg, ctrl, timeouts): + """Click Select -> then OK to start the export.""" + dlg.child_window(title=ctrl["select_btn_title"], class_name="Button").click() + time.sleep(timeouts["settle"]) + + dlg.child_window(title=ctrl["ok_btn_title"], class_name="Button").click() + + +def wait_for_export(app, ctrl, timeouts, log): + """ + Wait for the full export lifecycle to complete: + 1. 'Exporting Documents' progress dialog appears (export running) + 2. Progress dialog disappears (export finished writing files) + 3. '"Send to" file saved' confirmation dialog appears (MODAL) + 4. Dismiss it by clicking OK + """ + progress = app.window(title=ctrl["progress_title"], class_name=ctrl["progress_class"]) + + try: + progress.wait("visible", timeout=timeouts["progress_appear"]) + log(" Export in progress...") + except timings.TimeoutError: + log(" NOTE: progress dialog not seen; checking for confirmation dialog " + "(drawer may have exported very quickly).") + + try: + progress.wait_not("visible", timeout=timeouts["progress_finish"]) + except timings.TimeoutError: + log(f" ERROR: export did not finish within {timeouts['progress_finish']}s.") + return False + + saved = app.window(title=ctrl["saved_title"], class_name=ctrl["saved_class"]) + try: + saved.wait("visible ready", timeout=timeouts["confirm"]) + log(" Confirmation dialog appeared - export succeeded.") + except timings.TimeoutError: + log(" ERROR: confirmation dialog never appeared. Export may have " + "failed or produced no output.") + return False + + try: + saved.child_window(title=ctrl["saved_ok_title"], class_name="Button").click() + saved.wait_not("visible", timeout=timeouts["confirm"]) + log(" Confirmation dismissed.") + except Exception as e: + log(f" ERROR dismissing confirmation dialog: {e}") + return False + + log(" Export complete.") + return True + + +def export_drawer(app, main, drawer_id, ctrl, timeouts, screenshot_dir, log): + """Full per-drawer sequence. Returns True on success.""" + log(f"Drawer {drawer_id}: starting") + + # 1. Navigate + if not navigate_to_drawer(main, drawer_id, ctrl, timeouts, log): + take_screenshot(main, drawer_id, "nav_fail", screenshot_dir, log) + return False + + # 2. Open Send To File dialog + try: + dlg = open_send_to_file(main, app, ctrl, timeouts) + except Exception as e: + log(f" ERROR opening Send To File dialog: {e}") + take_screenshot(main, drawer_id, "dialog_fail", screenshot_dir, log) + return False + + # 3 & 4. Select -> and OK + try: + perform_export(dlg, ctrl, timeouts) + except Exception as e: + log(f" ERROR during export selection: {e}") + take_screenshot(dlg, drawer_id, "export_fail", screenshot_dir, log) + try: + dlg.child_window(title="Cancel", class_name="Button").click() + except Exception: + pass + return False + + # 5. Monitor progress + ok = wait_for_export(app, ctrl, timeouts, log) + if not ok: + take_screenshot(main, drawer_id, "progress_fail", screenshot_dir, log) + return False + + time.sleep(timeouts["settle"]) + return True + + +# --------------------------------------------------------------------------- +# MAIN LOOP +# --------------------------------------------------------------------------- + +def main(): + args = parse_args() + cfg = load_config(args.config) + log = make_logger(cfg.get("paths", "log_file")) + + # Build lookup dicts from config + paths = { + "drawer_id_file": cfg.get("paths", "drawer_id_file"), + "completed_file": cfg.get("paths", "completed_file"), + "screenshot_dir": cfg.get("paths", "screenshot_dir"), + } + timeouts = { + "nav": cfg.getfloat("timeouts", "nav_timeout"), + "dialog": cfg.getfloat("timeouts", "dialog_timeout"), + "progress_appear": cfg.getfloat("timeouts", "progress_appear"), + "progress_finish": cfg.getfloat("timeouts", "progress_finish"), + "settle": cfg.getfloat("timeouts", "settle"), + "confirm": cfg.getfloat("timeouts", "confirm_timeout"), + } + ctrl = { + "main_class": cfg.get("controls", "main_class"), + "drawer_box_title": cfg.get("controls", "drawer_box_title"), + "go_button_title": cfg.get("controls", "go_button_title"), + "dialog_title": cfg.get("controls", "dialog_title"), + "dialog_class": cfg.get("controls", "dialog_class"), + "select_btn_title": cfg.get("controls", "select_btn_title"), + "ok_btn_title": cfg.get("controls", "ok_btn_title"), + "progress_title": cfg.get("controls", "progress_title"), + "progress_class": cfg.get("controls", "progress_class"), + "saved_title": cfg.get("controls", "saved_title"), + "saved_class": cfg.get("controls", "saved_class"), + "saved_ok_title": cfg.get("controls", "saved_ok_title"), + } + + log("=" * 60) + log("FileCabinet CS batch export - run starting") + log("=" * 60) + + all_ids = load_lines(paths["drawer_id_file"]) + if not all_ids: + log(f"ERROR: no drawer IDs found in {paths['drawer_id_file']}") + sys.exit(1) + + completed = load_completed(paths["completed_file"]) + + pending = [d for d in all_ids if d not in completed] + log(f"Total drawers in list : {len(all_ids)}") + log(f"Already completed : {len(completed)}") + log(f"Remaining to process : {len(pending)}") + + if not pending: + log("Nothing to do - all drawers already completed.") + return + + app, main_win = connect_main(ctrl, log) + + success_count = 0 + fail_count = 0 + failed_ids = [] + + for i, drawer_id in enumerate(pending, start=1): + log(f"--- [{i}/{len(pending)}] Drawer {drawer_id} ---") + try: + ok = export_drawer(app, main_win, drawer_id, ctrl, timeouts, + paths["screenshot_dir"], log) + except Exception as e: + log(f" UNEXPECTED ERROR on drawer {drawer_id}: {e}") + try: + take_screenshot(main_win, drawer_id, "unexpected", + paths["screenshot_dir"], log) + except Exception: + pass + ok = False + + if ok: + mark_completed(paths["completed_file"], drawer_id) + success_count += 1 + log(f" Drawer {drawer_id} DONE ({success_count} ok / " + f"{fail_count} failed so far)") + else: + fail_count += 1 + failed_ids.append(drawer_id) + log(f" Drawer {drawer_id} FAILED - will NOT be marked complete. " + f"Re-run later to retry.") + + log("=" * 60) + log("Run finished.") + log(f" Succeeded : {success_count}") + log(f" Failed : {fail_count}") + if failed_ids: + log(f" Failed drawer IDs: {', '.join(failed_ids)}") + log(" These were NOT marked complete; re-running the script " + "will retry them.") + log("=" * 60) + + +if __name__ == "__main__": + main() diff --git a/fccs_folders.txt b/fccs_folders.txt new file mode 100644 index 0000000..9143c6b --- /dev/null +++ b/fccs_folders.txt @@ -0,0 +1,11 @@ +# FCCS Folder Names +# ----------------- +# List one folder name per line, exactly as it appears in FileCabinet CS. +# Open FCCS > Tools > Options > Folder Structure to see all folder names. +# Blank lines and lines starting with # are ignored. +# +# Examples: +# Tax Documents +# Correspondence +# Financial Statements +# Payroll diff --git a/fccs_reorganize.py b/fccs_reorganize.py new file mode 100644 index 0000000..da470e2 --- /dev/null +++ b/fccs_reorganize.py @@ -0,0 +1,142 @@ +""" +Step 3: Reorganize flat exported files into a proper folder structure. + +Parses FCCS export filenames using known folder names and date anchors +to reconstruct the original drawer/folder hierarchy. + +Filename format: + {drawer_id}_{client_name}_{folder_name}_{MM-DD-YYYY}_{document_name}.ext + Example: 01069_SMITH, BOB_2007 Tax Documents_02-10-2009_2007 Form 1040A Page 1.pdf + +Output structure: + output/{drawer_id}_{client_name}/{folder_name}/{original_filename} + +Files that cannot be parsed go to output/_unparsed/ for manual review. +""" + +import os +import re +import shutil +import sys + +from fccs_config import parse_args, load_config, make_logger, load_folder_list + + +def parse_filename(filename, known_folders): + """ + Parse an FCCS export filename into (drawer_id, client_name, folder_name, + date, doc_name_with_ext) or return None if it cannot be parsed. + + Strategy: anchor on drawer_id (left), date MM-DD-YYYY (middle), and match + a known folder name between client_name and date. + """ + stem, ext = os.path.splitext(filename) + + # Drawer ID is always the first underscore-delimited token + sep = stem.find("_") + if sep == -1: + return None + drawer_id = stem[:sep] + rest = stem[sep + 1:] + + # Find all date-pattern occurrences (MM-DD-YYYY) in rest + # We search with a leading underscore context to ensure proper boundaries + date_pattern = re.compile(r"_(\d{2}-\d{2}-\d{4})_") + # Prepend underscore so the first potential date at position 0 of rest is caught + search_str = "_" + rest + matches = list(date_pattern.finditer(search_str)) + + if not matches: + return None + + # Try each date match; known_folders is already sorted longest-first + for m in matches: + date_str = m.group(1) + # Position in `rest` where date starts (adjust for prepended _) + date_start_in_rest = m.start() - 1 # -1 for the prepended _ + # But the match includes the leading _, so the actual content before date: + before_date = rest[:date_start_in_rest] + after_date = rest[date_start_in_rest + len(date_str) + 1:] # +1 for trailing _ + + # Try to match a known folder at the end of before_date + for folder in known_folders: + if before_date.endswith(folder): + # Check there's an underscore separator before the folder name + prefix_end = len(before_date) - len(folder) + if prefix_end > 0 and before_date[prefix_end - 1] == "_": + client_name = before_date[: prefix_end - 1] + doc_name = after_date + ext + return (drawer_id, client_name, folder, date_str, doc_name) + + return None + + +def main(): + args = parse_args() + cfg = load_config(args.config) + log = make_logger(cfg.get("paths", "log_file")) + + export_dir = cfg.get("paths", "export_dir") + output_dir = cfg.get("paths", "output_dir") + folder_list_path = cfg.get("paths", "folder_list") + + if not os.path.isdir(export_dir): + log(f"ERROR: export directory not found: {export_dir}") + sys.exit(1) + + known_folders = load_folder_list(folder_list_path) + if not known_folders: + log("WARNING: folder list is empty. All files will go to _unparsed/.") + # Sort longest-first to prevent partial matches + known_folders.sort(key=len, reverse=True) + + files = [ + f for f in os.listdir(export_dir) + if os.path.isfile(os.path.join(export_dir, f)) + ] + + log(f"Found {len(files)} files to reorganize") + + success = 0 + failed = 0 + + for filename in files: + result = parse_filename(filename, known_folders) + + if result is None: + log(f" UNPARSED: {filename}") + dest_dir = os.path.join(output_dir, "_unparsed") + failed += 1 + else: + drawer_id, client_name, folder_name, date, doc_name = result + dest_dir = os.path.join( + output_dir, + f"{drawer_id}_{client_name}", + folder_name, + ) + success += 1 + + os.makedirs(dest_dir, exist_ok=True) + src = os.path.join(export_dir, filename) + dst = os.path.join(dest_dir, filename) + + # Handle duplicate filenames + if os.path.exists(dst): + base, fext = os.path.splitext(filename) + counter = 1 + while os.path.exists(dst): + dst = os.path.join(dest_dir, f"{base}_{counter}{fext}") + counter += 1 + log(f" DUPLICATE renamed: {filename} -> {os.path.basename(dst)}") + + shutil.copy2(src, dst) + + log("=" * 60) + log(f"Reorganization complete: {success} organized, {failed} unparsed") + if failed: + log(f"Review unparsed files in: {os.path.join(output_dir, '_unparsed')}") + log("=" * 60) + + +if __name__ == "__main__": + main() diff --git a/fccs_scan.py b/fccs_scan.py new file mode 100644 index 0000000..3cf1d18 --- /dev/null +++ b/fccs_scan.py @@ -0,0 +1,39 @@ +""" +Step 1: Scan a restored FCCS backup directory for drawer IDs. + +Every subdirectory in the backup root corresponds to a drawer. +Writes the sorted list of drawer IDs to the configured drawer_id_file. +""" + +import os +import sys + +from fccs_config import parse_args, load_config, make_logger + + +def main(): + args = parse_args() + cfg = load_config(args.config) + log = make_logger(cfg.get("paths", "log_file")) + + backup_dir = cfg.get("paths", "backup_dir") + output_file = cfg.get("paths", "drawer_id_file") + + if not os.path.isdir(backup_dir): + log(f"ERROR: backup directory not found: {backup_dir}") + sys.exit(1) + + drawer_ids = sorted( + name for name in os.listdir(backup_dir) + if os.path.isdir(os.path.join(backup_dir, name)) + ) + + with open(output_file, "w", encoding="utf-8") as f: + for did in drawer_ids: + f.write(did + "\n") + + log(f"Found {len(drawer_ids)} drawers, written to {output_file}") + + +if __name__ == "__main__": + main()