Compare commits

..

33 Commits

Author SHA1 Message Date
884c1610e3 added a helper folder and a script that helps us clean up SmartVaults crappy upload tool 2026-08-27 17:27:07 -05:00
79f36fc774 we handled the '/' character in reorganize. Filecabinet converts it to a '-' so we are now processing that. The folder templates file still should match exactly how it looks in FCCS 2026-08-24 23:14:56 -05:00
960248b825 added some more generic handling of TR products instead of hard coding UltraTax. We now have a list we loop through for things like Planner and Practice CS and can add more later 2026-08-24 21:50:30 -05:00
5e7630bf5c updated the reorganize script to handle nested folders inside of template files like 'YYYY' Income Documents > 2016 > Joe Shmo 2016 2026-08-24 21:43:43 -05:00
253e4fe419 tweaked output of the reorg report 2026-08-20 23:27:46 -05:00
3ab753276c tweaked line 147 in report_reorganize.py so that the output is more clear 2026-08-20 23:10:42 -05:00
e03f4ad8d9 added a py file to audit the folder reorganization and help us fix the files we couldnt parse 2026-08-20 22:44:21 -05:00
71ffc1fbb5 reconfigured the Folder Templates for git so that the engagement specific template file doesnt cause Master branch sync issues 2026-08-20 22:07:55 -05:00
8a6083e177 added handling for FCCS stripping . out of drawer names 2026-08-18 15:34:17 -05:00
ca2c49019f updated scan to ignore $ folders 2026-08-18 14:54:16 -05:00
5e0fbdbf70 updated the config files to a tempalte/local standard format 2026-08-18 13:53:28 -05:00
07472711e7 had to error handle the reog script to add safe handling of weird folder paths with odd characaters 2026-07-21 07:24:50 -05:00
9116d1c04a updated the organizer logic so that unparased files at least still get put in the correct client folder 2026-07-20 23:48:21 -05:00
3b950cf886 updated the report so that we will flag drawers that had errors and also crashed 2026-07-20 19:44:34 -05:00
b270ca9186 added fccs_report.py this generates a report of all the errors in a clean html format 2026-07-19 21:47:31 -05:00
3b2263b529 tweaked the verify output so it logs to its own file 2026-07-11 12:30:16 -05:00
fec7745092 updated our verify logic 2026-07-11 12:27:40 -05:00
2b04210a23 added a check utility so we can scan the export folder and check if the folder that failed during the export run actually failed or managed to complete succesfully 2026-07-11 12:12:41 -05:00
bda12518dd we encoutered a crash due to a corruption in FCCS. Added error handling for that scenario which included a crashed list and modifying the export function. Need to test it on a known crashed folder 2026-07-10 22:28:22 -05:00
7921e91c2c added and removed a delay from the controls dump 2026-07-09 20:06:52 -07:00
705e270004 updated how we are doing manifests so they are more clean and easier to check file formats 2026-07-09 21:52:19 -05:00
1b300eacb3 reverted how we compile the ignore list. The longer numbers will actually handle fine so we dont need to ignore. For example 01230 we need to handle but 01230TS we dont. Also we created a separate script to handle clashing numbers specifically as well as helper function that will dump control id's 2026-07-09 20:33:07 -05:00
b5862a7ea7 tweaked the ignore list so it includes all files not just prefix, so 02800 and 02800TP now 2026-07-09 16:48:07 -05:00
acd985432e auto generate the ignore list then we can manually adjust for locked drawers 2026-07-09 16:44:19 -05:00
60cc36893e updated scan.py to help us error handle the weird searches and alos build an ignore list 2026-07-09 16:21:20 -05:00
041ffd1c3f dropped the wait timer on the search box causing an error 2026-07-08 21:43:52 -05:00
a9fa2c4916 updated code to find search box by finding the box next to the go button since the control id was ambiguous 2026-07-08 21:31:20 -05:00
9303bcf0b5 added manifests to create a list of expected file outputs as well as a verification script. 2026-07-08 20:33:44 -05:00
d02333f9f9 increased time out for file export 2026-07-07 14:05:35 -05:00
0c264aaa9c fixed bug that caused us not to click ok in send to file 2026-07-07 13:22:20 -05:00
1f4555bff6 hopefully fixed the success dialog close issue 2026-07-07 13:05:58 -05:00
9e303d01c9 dumb changes 2026-07-07 05:38:17 -07:00
6a79bc8715 made a typo and now its a change 2026-07-07 05:37:10 -07:00
17 changed files with 2075 additions and 182 deletions

10
.gitignore vendored
View File

@@ -1 +1,11 @@
references/ references/
# Per-machine working config (created from config.template.ini on first run)
config.ini
# Python bytecode cache
__pycache__/
*.pyc
# Per-engagement folder list (created from fccs_folders.template.txt on first run)
fccs_folders.txt

3
.vscode/settings.json vendored Normal file
View File

@@ -0,0 +1,3 @@
{
"python-envs.defaultEnvManager": "ms-python.python:system"
}

View File

@@ -13,17 +13,24 @@ Extracts and organizes documents from FileCabinet CS (Thomson Reuters) using GUI
| File | Purpose | | File | Purpose |
|------|---------| |------|---------|
| `config.ini` | All paths, timeouts, and FCCS control identifiers | | `config.template.ini` | Tracked template for all paths, timeouts, and FCCS control identifiers |
| `fccs_folders.txt` | FCCS folder templates (manually populated per engagement) | | `config.ini` | Per-machine working config (git-ignored; auto-created from the template) |
| `fccs_folders.template.txt` | Tracked template for FCCS folder names |
| `fccs_folders.txt` | Per-engagement folder list (git-ignored; auto-created from the template) |
| `fccs_config.py` | Shared config loading, logging, folder pattern building | | `fccs_config.py` | Shared config loading, logging, folder pattern building |
| `fccs_scan.py` | Step 1: Scan backup directory for drawer IDs | | `fccs_scan.py` | Step 1: Scan backup directory for drawer IDs |
| `fccs_export.py` | Step 2: Automate FCCS GUI to export all drawers | | `fccs_export.py` | Step 2: Automate FCCS GUI to export all drawers |
| `fccs_reorganize.py` | Step 3: Parse filenames and rebuild folder structure | | `fccs_reorganize.py` | Step 3: Parse filenames and rebuild folder structure |
| `fccs_verify.py` | Step 4 (optional): Compare manifests against exported files |
| `fccs_check.py` | Utility: Interactively check if specific drawers exported completely |
| `fccs_report.py` | Utility: Generate a clean client-facing HTML progress report (issues only) |
| `fccs_report_reorganize.py` | Utility: Console report of `_unparsed` leftovers after reorganizing, with suggested missing folder templates |
| `fccs_dump_controls.py` | Utility: Dump control identifiers of an on-screen dialog |
## Setup Per Engagement ## Setup Per Engagement
1. Edit `config.ini` -- point `backup_dir`, `export_dir`, and `output_dir` at the client's directories. 1. Edit `config.ini` -- point `backup_dir`, `export_dir`, and `output_dir` at the client's directories. This file is created automatically from `config.template.ini` the first time you run any script (it's git-ignored, so your machine-specific paths never drift the repo). To reset a machine, delete `config.ini` and re-run. Keep shared/default changes in `config.template.ini` (the tracked file).
2. Populate `fccs_folders.txt` -- open FCCS > System Configuration > Document Folders and list each folder template exactly as shown, one per line. Keep the `YYYY` prefix on recurring folders. UltraTax CS folders are detected automatically and do not need to be listed. 2. Populate `fccs_folders.txt` -- open FCCS > System Configuration > Document Folders and list each folder template exactly as shown, one per line (names containing `/`, like `Foreign Bank/Income`, can be copied verbatim — the code converts `/` to `-` automatically, matching how FCCS writes it into filenames). Keep the `YYYY` prefix on recurring folders. Thomson Reuters product folders (UltraTax CS, Planner CS, Practice CS) are detected automatically and do not need to be listed. Like `config.ini`, this file is auto-created from `fccs_folders.template.txt` on first use and is git-ignored — edit it freely per engagement; put only broadly-useful defaults in the tracked template.
## Workflow ## Workflow
@@ -33,7 +40,19 @@ Extracts and organizes documents from FileCabinet CS (Thomson Reuters) using GUI
python fccs_scan.py python fccs_scan.py
``` ```
Reads the restored FCCS backup directory and writes all drawer IDs (subfolder names) to `drawer_ids.txt`. Reads the FCCS data directory (`backup_dir`) and writes all drawer IDs (subfolder names) to `drawer_ids.txt`. It also:
- **Skips non-drawer entries** — only subdirectories are treated as drawers. When `backup_dir` points at FCCS's live data directory (the Restore directory), that folder also contains system folders whose names start with `$` and miscellaneous loose files; both are ignored.
- **Normalizes drawer IDs** — FileCabinet CS ignores `.` characters in drawer IDs, so a folder named `A123.TJ` on disk is searched and displayed in the UI as `A123TJ`. The scan strips dots from folder names when writing `drawer_ids.txt` so the ID matches what FCCS expects (searching the dotted form returns no results). This is done at the source because FCCS embeds the same dot-free ID as the prefix of exported filenames, which the reorganize/verify/report tools all key off. If stripping dots collapses two distinct folders onto one ID, the scan logs a **collision warning** rather than silently dropping a drawer.
- **Flags prefix clashes** — if one drawer ID is a prefix of another (e.g. `02218` and `02218A`), searching the **base** ID in FCCS pops up a selection box that breaks plain automated navigation. These clashes are reported, and the base (shorter) IDs are auto-seeded into `ignore.txt`. The longer, more-specific IDs (`02218A`) search fine and export normally. (Clash detection runs on the dot-normalized IDs, since that's what FCCS actually searches.)
- **Reports ignored drawers** — any IDs listed in `ignore.txt` that exist in this backup are shown as ones the export will skip.
**Ignoring drawers:** The scan creates `ignore.txt` (at `ignore_file`, default `C:\Migration\ignore.txt`) if it doesn't exist and pre-fills it with the clash base IDs — searching those in FCCS shows a selection box that stalls the plain export, so they're skipped by the main export. Open the file and:
- **Delete or comment out** any clash base you'd rather handle fully by hand.
- **Add** any other drawers to skip (e.g. password-protected folders), one ID per line.
Re-running the scan never overwrites your edits — it only appends newly-discovered clashes. `drawer_ids.txt` stays a full inventory; the export skips anything active in the ignore list. Lines starting with `#` are comments.
### Step 2: Export Documents ### Step 2: Export Documents
@@ -46,6 +65,7 @@ Requires FCCS to be open with export destination already configured. Automates t
- **Resumable** -- tracks completed drawers in `completed.txt`; safe to restart - **Resumable** -- tracks completed drawers in `completed.txt`; safe to restart
- **Screenshots** -- captures failure states for diagnosis - **Screenshots** -- captures failure states for diagnosis
- **Defensive** -- one bad drawer won't crash the entire run - **Defensive** -- one bad drawer won't crash the entire run
- **Crash recovery** -- some documents (e.g. UltraTax "Diagnostics" files) crash FCCS's converter (`FileConversionEngine::convert() failed`), which aborts that drawer's export. The script detects the error dialog, screenshots and logs the crashing document, dismisses it, and records the drawer in `crashed.txt` so it's skipped on future runs instead of stalling. Handle crashed drawers manually (export them excluding the poison document); delete a line from `crashed.txt` to retry after fixing.
### Step 3: Reorganize Files ### Step 3: Reorganize Files
@@ -69,16 +89,68 @@ output/
2008 Form 1040 Filing Instructions.doc 2008 Form 1040 Filing Instructions.doc
Permanent File/ Permanent File/
Driver's License.pdf Driver's License.pdf
_unparsed/
(files for this client that couldn't be fully parsed, kept for review)
_unparsed/ _unparsed/
(files that couldn't be parsed go here for manual review) (only files whose client couldn't be recovered from the filename)
``` ```
Exported filenames follow the format `{drawer_id}_{client_name}_{folder_name}_{creation_date}_{document_name}.ext`. The parser uses folder templates from `fccs_folders.txt` (with `YYYY` expanded via regex) and the creation date (`MM-DD-YYYY`) as anchors to reliably split the underscore-delimited fields. Folder names are decomposed into nested paths that match the FCCS UI structure (e.g. `2025 Tax Documents` becomes `Tax Documents/2025/`). UltraTax CS folders are matched by a built-in pattern. Exported filenames follow the format `{drawer_id}_{client_name}_{folder_name}_{creation_date}_{document_name}.ext`. The parser uses folder templates from `fccs_folders.txt` (with `YYYY` expanded via regex) and the creation date (`MM-DD-YYYY`) as anchors to reliably split the underscore-delimited fields. Folder names are decomposed into nested paths that match the FCCS UI structure (e.g. `2025 Tax Documents` becomes `Tax Documents/2025/`). Thomson Reuters product folders — `{Product} MM-DD-YYYY`, e.g. UltraTax CS, Planner CS, Practice CS — are matched by a built-in pattern and become `{Product}/{date}/`; to support another product, add its name to `TR_PRODUCT_FOLDERS` in `fccs_config.py`.
FCCS drawers can (rarely) contain **nested subfolders** under a template folder; the export encodes these in square brackets appended to the folder field, e.g. `2016 Income Documents[4201 N Beach Street, LLC]`. The reorganizer recreates them as deeper nesting — that example becomes `Income Documents/2016/4201 N Beach Street, LLC/` — as long as the parent (`YYYY Income Documents`) is a listed template; no bracket entries are needed in `fccs_folders.txt`.
Files that can't be fully parsed still keep their client: the drawer ID and client name are the first two underscore-delimited tokens and stay recoverable even when the folder/date parse fails, so those files are filed under `{client_name}/_unparsed/` (retaining their original filename). Only files whose client can't be recovered at all fall back to the top-level `_unparsed/`.
Folder names are sanitized for Windows before use — trailing spaces and periods are stripped from each path component, since Windows can't create a directory ending in a space or dot (e.g. a client named `FARR GROUP, P.L.` becomes `FARR GROUP, P.L`). Reorganization is also resilient per file: if one file can't be placed for any reason, the error is logged and counted (reported as `errored` in the summary) and the run continues with the rest rather than aborting.
**Checking for folder template gaps:** After a reorganize run, get a quick internal summary of what didn't parse:
```
python fccs_report_reorganize.py
```
This walks `output_dir`, lists every client that has an `_unparsed` subfolder (plus the top-level `_unparsed`), and — because unparsed files keep their original export filename — recovers the folder field from each name and aggregates them into **suggested template lines** (years generalized to `YYYY`) that can be pasted into `fccs_folders.txt`. Add the missing templates and re-run `fccs_reorganize.py`. Console-only output; unparsed files with no recoverable folder field are counted separately (oddball names, not template gaps).
### Step 4 (optional): Verify Export Completeness
During export, each drawer's document list is captured from the FCCS dialog and saved as a manifest (in `manifest_dir`). These tools compare the manifests against the files actually in the export folder to confirm nothing was missed.
Both compare at the **document level** and share identical matching logic. They account for:
- **Page-splitting** -- a multi-page document exported as `Name Page 1`, `Name Page 2`, … counts as that one document being present.
- **Filename sanitization** -- document titles containing characters illegal in Windows filenames (e.g. `:` `/` `?`) still match the exported files.
Batch-check every drawer that has a manifest:
```
python fccs_verify.py
```
Reports each drawer as `OK` or `INCOMPLETE` (listing the missing documents), plus a summary and any exported drawers that have no manifest. The full report is written to its own file (`verify_report`, default `C:\Migration\verify_report.txt`) as well as the console — separate from the export's `run_log.txt`.
Spot-check specific drawers interactively (e.g. ones the log marked failed, to see whether they actually finished exporting in the background):
```
python fccs_check.py
Drawer ID(s): 08097 18430
```
> Note: because page-splitting means the number of files can't be mapped one-to-one to documents, completeness is judged by document *presence* (is each manifest document represented by at least one exported file), not by exact file counts.
**Client-facing progress report:** For a clean summary to share with the client, generate an HTML report of outstanding work only:
```
python fccs_report.py
```
This runs the same document-level check as `fccs_verify.py` but writes a self-contained, print-friendly HTML file (`report_html`, default `C:\Migration\progress_report.html`) that lists **only** the drawers with missing documents, each with the client name and the specific documents still outstanding. Fully-migrated drawers and drawers with no manifest are omitted (a headline shows how many are done). Open it in any browser and print to PDF to send.
Any outstanding drawer that also appears in `crashed.txt` (its export was aborted by an FCCS converter crash) is badged **CRASHED** and sorted to the top, and counted in the header. These are the genuine failures worth spot-checking first — as opposed to benign false positives, where a collapsed "container" document's children exported fine but the container's name lands in the filename's folder field rather than the document field, so it reads as missing.
## Config Reference ## Config Reference
All scripts read from `config.ini` (or specify `--config path\to\config.ini`). All scripts read from `config.ini` (or specify `--config path\to\config.ini`). `config.ini` is the git-ignored, per-machine copy auto-created from the tracked `config.template.ini`; edit `config.ini` locally.
- **`[paths]`** -- backup_dir, export_dir, output_dir, drawer_id_file, completed_file, log_file, screenshot_dir, folder_list - **`[paths]`** -- backup_dir, export_dir, output_dir, drawer_id_file, completed_file, ignore_file, crashed_file, log_file, verify_report, report_html, screenshot_dir, manifest_dir, folder_list
- **`[timeouts]`** -- nav_timeout, dialog_timeout, progress_appear, progress_finish, settle, confirm_timeout - **`[timeouts]`** -- nav_timeout, dialog_timeout, progress_appear, progress_finish, settle, confirm_timeout
- **`[controls]`** -- FCCS window class names and button titles (rarely need changing) - **`[controls]`** -- FCCS window class names and button titles (rarely need changing)

View File

@@ -1,3 +1,12 @@
# ---------------------------------------------------------------------------
# TEMPLATE CONFIG — tracked in git. Do NOT edit for a specific machine.
#
# On first run the scripts copy this file to `config.ini` (which is git-ignored)
# and use that. Edit `config.ini` for this machine's paths/settings so your
# local changes never drift the repo. To reset a machine, delete `config.ini`
# and re-run; it will be recreated from this template.
# ---------------------------------------------------------------------------
[paths] [paths]
# Point these at the current engagement's directories # Point these at the current engagement's directories
backup_dir = C:\Migration\Backups backup_dir = C:\Migration\Backups
@@ -5,15 +14,20 @@ export_dir = C:\Migration\Export
output_dir = C:\Migration\Output output_dir = C:\Migration\Output
drawer_id_file = C:\Migration\drawer_ids.txt drawer_id_file = C:\Migration\drawer_ids.txt
completed_file = C:\Migration\completed.txt completed_file = C:\Migration\completed.txt
ignore_file = C:\Migration\ignore.txt
crashed_file = C:\Migration\crashed.txt
log_file = C:\Migration\run_log.txt log_file = C:\Migration\run_log.txt
verify_report = C:\Migration\verify_report.txt
report_html = C:\Migration\progress_report.html
screenshot_dir = C:\Migration\screenshots screenshot_dir = C:\Migration\screenshots
manifest_dir = C:\Migration\manifests
folder_list = fccs_folders.txt folder_list = fccs_folders.txt
[timeouts] [timeouts]
nav_timeout = 15 nav_timeout = 15
dialog_timeout = 15 dialog_timeout = 15
progress_appear = 30 progress_appear = 30
progress_finish = 1800 progress_finish = 7200
settle = 1.0 settle = 1.0
select_timeout = 180 select_timeout = 180
confirm_timeout = 60 confirm_timeout = 60
@@ -31,3 +45,5 @@ progress_class = #32770
saved_title = "Send to" file saved saved_title = "Send to" file saved
saved_class = #32770 saved_class = #32770
saved_ok_title = OK saved_ok_title = OK
error_title = FileCabinet CS
error_class = #32770

80
control_dump.txt Normal file
View File

@@ -0,0 +1,80 @@
========================================================================
WINDOW handle=121439734 class='ThumbnailDeviceHelperWnd' title=''
========================================================================
(could not dump this window: 'DialogWrapper' object has no attribute 'print_control_identifiers')
========================================================================
WINDOW handle=1507838 class='#32768' title=''
========================================================================
(could not dump this window: 'PopupMenuWrapper' object has no attribute 'print_control_identifiers')
========================================================================
WINDOW handle=65744 class='Shell_TrayWnd' title=''
========================================================================
(could not dump this window: 'DialogWrapper' object has no attribute 'print_control_identifiers')
========================================================================
WINDOW handle=66772 class='CSIFCAB' title='FileCabinet CS [04245 DELATORRE, LUIS M. & ANGELA H.]'
========================================================================
(could not dump this window: 'DialogWrapper' object has no attribute 'print_control_identifiers')
========================================================================
WINDOW handle=3737076 class='Chrome_WidgetWin_1' title='fccs_dump_controls.py - FCCS - Visual Studio Code'
========================================================================
(could not dump this window: 'DialogWrapper' object has no attribute 'print_control_identifiers')
========================================================================
WINDOW handle=5899176 class='ConsoleWindowClass' title='Administrator: Command Prompt'
========================================================================
(could not dump this window: 'DialogWrapper' object has no attribute 'print_control_identifiers')
========================================================================
WINDOW handle=787222 class='XamlExplorerHostIslandWindow_WASDK' title=''
========================================================================
(could not dump this window: 'DialogWrapper' object has no attribute 'print_control_identifiers')
========================================================================
WINDOW handle=197392 class='Windows.UI.Core.CoreWindow' title=''
========================================================================
(could not dump this window: 'DialogWrapper' object has no attribute 'print_control_identifiers')
========================================================================
WINDOW handle=262176 class='ApplicationFrameWindow' title='Settings'
========================================================================
(could not dump this window: 'DialogWrapper' object has no attribute 'print_control_identifiers')
========================================================================
WINDOW handle=65880 class='DummyDWMListenerWindow' title=''
========================================================================
(could not dump this window: 'DialogWrapper' object has no attribute 'print_control_identifiers')
========================================================================
WINDOW handle=65872 class='EdgeUiInputTopWndClass' title=''
========================================================================
(could not dump this window: 'DialogWrapper' object has no attribute 'print_control_identifiers')
========================================================================
WINDOW handle=65844 class='DummyDWMListenerWindow' title=''
========================================================================
(could not dump this window: 'DialogWrapper' object has no attribute 'print_control_identifiers')
========================================================================
WINDOW handle=65842 class='DummyDWMListenerWindow' title=''
========================================================================
(could not dump this window: 'DialogWrapper' object has no attribute 'print_control_identifiers')
========================================================================
WINDOW handle=65840 class='DummyDWMListenerWindow' title=''
========================================================================
(could not dump this window: 'DialogWrapper' object has no attribute 'print_control_identifiers')
========================================================================
WINDOW handle=65838 class='DummyDWMListenerWindow' title=''
========================================================================
(could not dump this window: 'DialogWrapper' object has no attribute 'print_control_identifiers')
========================================================================
WINDOW handle=65782 class='Progman' title='Program Manager'
========================================================================
(could not dump this window: 'DialogWrapper' object has no attribute 'print_control_identifiers')

98
fccs_check.py Normal file
View File

@@ -0,0 +1,98 @@
"""
Utility: Check whether specific drawers actually finished exporting.
Interactive — prompts for one or more drawer IDs, then for each drawer compares
its manifest (the documents FCCS said it would export, captured during Step 2)
against the files actually sitting in the export folder, and reports any
missing documents.
Handles two quirks of the FCCS export (see fccs_config.evaluate_drawer):
1. Page-splitting: a single manifest document (e.g. "Donations") is exported
as one file if it's a single page, or as "Donations Page 1",
"Donations Page 2", ... for a multi-page document. All of those count as
that one document being present.
2. Filename sanitization: FCCS strips characters that are illegal in Windows
filenames (: / \\ ? * " < > |) from document titles, so comparison is done
on a normalized key so these still match.
fccs_verify.py applies this same logic in batch across every drawer; this tool
is for spot-checking specific drawers (e.g. ones the log marked failed).
USAGE
-----
python fccs_check.py
Drawer ID(s): 08097 18430
"""
import os
import re
import sys
from fccs_config import (
parse_args, load_config, evaluate_drawer, index_files_by_drawer,
)
def report_drawer(drawer_id, files, manifest_dir, out):
"""Evaluate one drawer and print a human-readable completeness report."""
r = evaluate_drawer(drawer_id, files, manifest_dir)
out("")
if not r["has_manifest"]:
out(f"[{drawer_id}] NO MANIFEST at {r['manifest_path']} — cannot verify "
"(was this drawer exported by the tool?)")
return
expected = r["expected"]
missing = r["missing"]
out(f"[{drawer_id}] manifest lists {len(expected)} document(s); "
f"{r['file_count']} file(s) in export folder.")
if missing:
out(f" INCOMPLETE — {len(missing)} document(s) missing from export:")
for m in missing:
out(f" - {m}")
else:
out(f" COMPLETE — all {len(expected)} manifest document(s) present.")
if r["extras"]:
out(f" Note: {len(r['extras'])} exported document(s) not in the manifest:")
for e in r["extras"]:
out(f" - {e}")
if r["unparsed"]:
out(f" Note: {r['unparsed']} file(s) had no recognizable date anchor "
"and were skipped.")
def main():
args = parse_args()
cfg = load_config(args.config)
export_dir = cfg.get("paths", "export_dir")
manifest_dir = cfg.get("paths", "manifest_dir")
if not os.path.isdir(export_dir):
print(f"ERROR: export directory not found: {export_dir}")
sys.exit(1)
files_by_drawer = index_files_by_drawer(export_dir)
print("FCCS export completeness check")
print(f" export folder : {export_dir}")
print(f" manifests : {manifest_dir}")
print("Enter drawer ID(s) separated by spaces or commas (blank to quit).")
while True:
try:
raw = input("\nDrawer ID(s): ").strip()
except EOFError:
break
if not raw:
break
ids = [i for i in re.split(r"[\s,]+", raw) if i]
for drawer_id in ids:
report_drawer(drawer_id, files_by_drawer.get(drawer_id, []),
manifest_dir, print)
if __name__ == "__main__":
main()

View File

@@ -5,11 +5,31 @@ Shared configuration, logging, and utilities for the FCCS extraction tool.
import os import os
import re import re
import sys import sys
import shutil
import argparse import argparse
import configparser import configparser
from datetime import datetime from datetime import datetime
DEFAULT_CONFIG = "config.ini" DEFAULT_CONFIG = "config.ini"
TEMPLATE_CONFIG = "config.template.ini"
def ensure_from_template(path, template):
"""Create a per-machine working file from its git-tracked template.
If `path` doesn't exist but `template` does, copy template -> path and
announce it. Returns True if the file exists (or was just created).
Used for config.ini and fccs_folders.txt so local edits never drift the
repo — the working copies are git-ignored, only the templates are tracked.
"""
if os.path.exists(path):
return True
if os.path.exists(template):
shutil.copyfile(template, path)
print(f"Created {os.path.basename(path)} from "
f"{os.path.basename(template)}. Edit it for this machine: {path}")
return True
return False
def parse_args(): def parse_args():
@@ -24,12 +44,26 @@ def parse_args():
def load_config(path=None): def load_config(path=None):
"""Read config.ini and return a ConfigParser object.""" """Read the config file and return a ConfigParser object.
Config is split into two files:
- config.template.ini : tracked in git, the pristine template.
- config.ini : git-ignored, this machine's working copy.
When no explicit --config path is given, the per-machine config.ini is used.
If it doesn't exist yet, it's created from config.template.ini so a freshly
cloned/pulled repo works out of the box without editing the tracked file.
Edit config.ini locally; your changes never drift the repo.
"""
if path is None: if path is None:
script_dir = os.path.dirname(os.path.abspath(sys.argv[0])) script_dir = os.path.dirname(os.path.abspath(sys.argv[0]))
path = os.path.join(script_dir, DEFAULT_CONFIG) path = os.path.join(script_dir, DEFAULT_CONFIG)
# Auto-create the per-machine config from the template on first run.
ensure_from_template(path, os.path.join(script_dir, TEMPLATE_CONFIG))
if not os.path.exists(path): if not os.path.exists(path):
print(f"ERROR: config file not found: {path}") print(f"ERROR: config file not found: {path}")
print(f" Expected {DEFAULT_CONFIG} or a --config path "
f"(template: {TEMPLATE_CONFIG}).")
sys.exit(1) sys.exit(1)
cfg = configparser.ConfigParser() cfg = configparser.ConfigParser()
cfg.read(path, encoding="utf-8") cfg.read(path, encoding="utf-8")
@@ -50,9 +84,17 @@ def make_logger(log_file):
def load_folder_list(path): def load_folder_list(path):
"""Load known FCCS folder names from file, one per line.""" """Load known FCCS folder names from file, one per line.
if not os.path.exists(path):
Like config.ini, the folder list is split into a tracked template
(fccs_folders.template.txt) and a git-ignored per-engagement working copy
(fccs_folders.txt) — auto-created from the template on first use.
"""
root, ext = os.path.splitext(path)
template = root + ".template" + ext
if not ensure_from_template(path, template):
print(f"ERROR: folder list file not found: {path}") print(f"ERROR: folder list file not found: {path}")
print(f" (no template found at {template} to create it from)")
sys.exit(1) sys.exit(1)
folders = [] folders = []
with open(path, "r", encoding="utf-8") as f: with open(path, "r", encoding="utf-8") as f:
@@ -63,32 +105,46 @@ def load_folder_list(path):
return folders return folders
# Thomson Reuters product integrations that auto-generate folders named
# "{Product} MM-DD-YYYY" in FCCS. Matched built-in (no fccs_folders.txt entry
# needed); add newly-discovered products here.
TR_PRODUCT_FOLDERS = ("UltraTax CS", "Planner CS", "Practice CS")
def build_folder_patterns(templates): def build_folder_patterns(templates):
""" """
Convert folder template strings into regex patterns for matching. Convert folder template strings into regex patterns for matching.
Templates use YYYY as a year placeholder (e.g. 'YYYY Tax Documents'). Templates use YYYY as a year placeholder (e.g. 'YYYY Tax Documents').
Non-recurring folders (no YYYY) are matched literally. Non-recurring folders (no YYYY) are matched literally.
A built-in pattern for 'UltraTax CS MM-DD-YYYY' is always included. Built-in patterns for '{Product} MM-DD-YYYY' Thomson Reuters product
folders (TR_PRODUCT_FOLDERS, e.g. UltraTax CS) are always included.
Returns a list of (compiled_regex, template_name, folder_type) tuples, Returns a list of (compiled_regex, template_name, folder_type) tuples,
sorted longest-first to prevent partial matches. sorted longest-first to prevent partial matches.
folder_type is one of: folder_type is one of:
"yyyy" — recurring folder with year prefix (captured in group 1) "yyyy" — recurring folder with year prefix (captured in group 1)
"ultratax"UltraTax CS folder with date suffix (captured in group 1) "ultratax"TR product folder with date suffix (captured in group 1)
"static" — non-recurring folder, no decomposition needed "static" — non-recurring folder, no decomposition needed
""" """
patterns = [] patterns = []
# Built-in: UltraTax CS folders (auto-generated by UltraTax integration) # Built-in: TR product folders (auto-generated by each product's integration)
patterns.append(( for product in TR_PRODUCT_FOLDERS:
re.compile(r"UltraTax CS (\d{2}-\d{2}-\d{4})$"), patterns.append((
"UltraTax CS", re.compile(re.escape(product) + r" (\d{2}-\d{2}-\d{4})$"),
"ultratax", product,
)) "ultratax",
))
for tmpl in templates: for tmpl in templates:
# FCCS converts "/" to "-" when embedding the folder name in the flat
# export filename ("/" is illegal in Windows filenames), so templates
# copied verbatim from the FCCS UI (e.g. "YYYY Foreign Bank/Income")
# must be normalized the same way to match — and to be usable as an
# output directory component.
tmpl = tmpl.replace("/", "-")
if "YYYY" in tmpl: if "YYYY" in tmpl:
# Replace YYYY with captured 4-digit year pattern, escape the rest # Replace YYYY with captured 4-digit year pattern, escape the rest
parts = tmpl.split("YYYY") parts = tmpl.split("YYYY")
@@ -114,8 +170,9 @@ def decompose_folder_path(match, folder_type, template_name):
Examples: Examples:
yyyy: "YYYY Tax Documents" matched "2025 Tax Documents" yyyy: "YYYY Tax Documents" matched "2025 Tax Documents"
→ ("Tax Documents", "2025") → ("Tax Documents", "2025")
ultratax: "UltraTax CS 12-31-2008" ultratax: any TR product folder, e.g. "UltraTax CS 12-31-2008"
→ ("UltraTax CS", "12-31-2008") → ("UltraTax CS", "12-31-2008"),
"Planner CS 12-31-2016" → ("Planner CS", "12-31-2016")
static: "Permanent File" static: "Permanent File"
→ ("Permanent File",) → ("Permanent File",)
""" """
@@ -126,7 +183,7 @@ def decompose_folder_path(match, folder_type, template_name):
return (base_name, year) return (base_name, year)
elif folder_type == "ultratax": elif folder_type == "ultratax":
date = match.group(1) date = match.group(1)
return ("UltraTax CS", date) return (template_name, date)
else: else:
return (template_name,) return (template_name,)
@@ -137,3 +194,185 @@ def load_lines(path):
return [] return []
with open(path, "r", encoding="utf-8") as f: with open(path, "r", encoding="utf-8") as f:
return [line.strip() for line in f if line.strip() and not line.strip().startswith("#")] return [line.strip() for line in f if line.strip() and not line.strip().startswith("#")]
def check_for_clashes(drawer_ids):
"""Find drawer IDs that are a prefix of another drawer ID.
FCCS searches by prefix, so searching the shorter ID (e.g. '02218')
pops up a selection box when a longer ID exists (e.g. '02218A'),
which breaks the automated navigation.
Returns a list of (short_id, [longer_ids...]) tuples, sorted by short_id.
Empty list means no clashes.
"""
ids = sorted(set(drawer_ids))
clashes = []
for short in ids:
matches = [other for other in ids
if other != short and other.startswith(short)]
if matches:
clashes.append((short, matches))
return clashes
# ---------------------------------------------------------------------------
# MANIFEST / EXPORT-COMPLETENESS HELPERS
# ---------------------------------------------------------------------------
# Shared by fccs_verify.py (batch) and fccs_check.py (interactive) so both
# judge completeness identically.
# Trailing " Page N" (optionally "Page N of M") appended to multi-page exports.
_PAGE_RE = re.compile(r"\s*Page\s+\d+(?:\s+of\s+\d+)?\s*$", re.IGNORECASE)
# Creation-date field ("_MM-DD-YYYY_") that precedes the document name.
_DATE_ANCHOR = re.compile(r"_\d{2}-\d{2}-\d{4}_")
def load_manifest(path):
"""Load a manifest file and return one row (list of cells) per document.
Handles both formats:
- New: one document per line (tab-separated columns).
- Old: a raw ListView dump led by 'List1', then row-major cells with
3 columns per document (Drawer ID, Page Title, Application), reshaped
so the document count is correct.
"""
with open(path, "r", encoding="utf-8") as f:
lines = [line.rstrip("\n") for line in f if line.strip()]
if lines and lines[0].strip() == "List1":
cells = lines[1:]
rows = [cells[i:i + 3] for i in range(0, len(cells), 3)]
return [r for r in rows if len(r) == 3]
return [line.split("\t") for line in lines]
def manifest_doc_names(path, drawer_id):
"""Return the expected document (Page Title) names from a manifest file."""
rows = load_manifest(path)
names = []
for row in rows:
if len(row) >= 3 and row[0].strip() == drawer_id:
names.append(row[1]) # old format: DrawerID, PageTitle, Application
elif row:
names.append(row[0]) # new format: PageTitle, Application
return names
def match_key(name):
"""Normalize a document name for tolerant comparison.
Strips a trailing 'Page N' page-split suffix, then reduces to lowercase
alphanumerics so punctuation and filename-sanitization differences (FCCS
strips characters illegal in Windows filenames) don't cause false
mismatches.
"""
base = _PAGE_RE.sub("", name)
return re.sub(r"[^a-z0-9]+", "", base.lower())
def exported_doc_name(filename):
"""Extract the document-name portion from an exported filename, or None.
Format: {drawer}_{client}_{folder}_{MM-DD-YYYY}_{docname}.ext
The creation-date field is a reliable anchor; the doc name follows the last
one (client/folder fields don't carry an "_MM-DD-YYYY_" pattern).
"""
stem = os.path.splitext(filename)[0]
anchors = list(_DATE_ANCHOR.finditer(stem))
if not anchors:
return None
return stem[anchors[-1].end():]
def client_name_from_filename(filename):
"""Best-effort client name from a single export filename, or None.
Filenames are '{drawer}_{client}_{folder}_{MM-DD-YYYY}_{doc}.ext', so the
client name is the second underscore-delimited token. Client names carry
commas/spaces but not underscores, so this token is reliable even when the
fuller parse (folder/date matching) fails.
"""
parts = filename.split("_")
if len(parts) >= 2 and parts[1].strip():
return parts[1].strip()
return None
def client_name_from_files(files):
"""Best-effort client name from a drawer's export filenames, or None.
The most common per-file client name is returned to shrug off any oddball
filename.
"""
counts = {}
for f in files:
name = client_name_from_filename(f)
if name:
counts[name] = counts.get(name, 0) + 1
if not counts:
return None
return max(counts, key=counts.get)
def index_files_by_drawer(export_dir):
"""Map each export file to its leading drawer-ID token (before first '_').
The underscore boundary keeps clashing IDs separate (04289 vs 04289TS).
Returns {drawer_id: [filenames]}.
"""
index = {}
for f in os.listdir(export_dir):
if not os.path.isfile(os.path.join(export_dir, f)):
continue
token = f.split("_", 1)[0]
index.setdefault(token, []).append(f)
return index
def evaluate_drawer(drawer_id, files, manifest_dir):
"""Compare a drawer's manifest against its exported files.
`files` is the list of export filenames belonging to this drawer. Accounts
for page-splitting (a document exported as 'Name Page 1/2/...' counts as
present) and filename sanitization (via match_key).
Returns a dict:
has_manifest, manifest_path, expected (list), missing (list),
extras (list), unparsed (int), file_count (int)
"""
manifest_path = os.path.join(manifest_dir, drawer_id + ".txt")
result = {
"has_manifest": os.path.exists(manifest_path),
"manifest_path": manifest_path,
"expected": [],
"missing": [],
"extras": [],
"unparsed": 0,
"file_count": len(files),
}
if not result["has_manifest"]:
return result
expected = manifest_doc_names(manifest_path, drawer_id)
result["expected"] = expected
# Group exported files by normalized doc name; page-splits collapse together.
exported = {} # key -> list of full doc names (one entry per file/page)
unparsed = 0
for f in files:
doc = exported_doc_name(f)
if doc is None:
unparsed += 1
continue
exported.setdefault(match_key(doc), []).append(doc)
result["missing"] = [name for name in expected
if match_key(name) not in exported]
expected_keys = {match_key(n) for n in expected}
result["extras"] = [names[0] for k, names in exported.items()
if k not in expected_keys]
result["unparsed"] = unparsed
return result

95
fccs_dump_controls.py Normal file
View File

@@ -0,0 +1,95 @@
"""
Utility: Dump control identifiers of on-screen windows.
Use this to discover the control identifiers of a dialog we don't have mapped
yet — e.g. the drawer-selection box that FCCS shows when you search a base ID
that clashes with a longer one (02218 -> 02218A).
USAGE
-----
1. In FCCS, get the target screen on-screen (e.g. type the clashing base ID and
click Go so the selection box is showing).
2. From another cmd window (with the venv active), run:
python fccs_dump_controls.py
This lists every visible top-level window and writes the full control
identifiers of each to control_dump.txt.
To narrow it down, filter by title or class:
python fccs_dump_controls.py --title Select
python fccs_dump_controls.py --class #32770
3. Open control_dump.txt, find the selection box, and copy its identifiers
into the references folder (like references/send_to_file_dialog.txt).
Notes:
- Run from 32-bit Python (same as the rest of the tool).
- Running this from a separate cmd window means the FCCS dialog stays open;
this script does not need focus.
"""
import argparse
import sys
from pywinauto import Desktop
def main():
ap = argparse.ArgumentParser(description=__doc__)
ap.add_argument("--title", default=None,
help="only dump windows whose title contains this text")
ap.add_argument("--class", dest="cls", default=None,
help="only dump windows with this exact class name")
ap.add_argument("--out", default="control_dump.txt",
help="output file (default: control_dump.txt)")
args = ap.parse_args()
targets = []
for w in Desktop(backend="win32").windows():
try:
if not w.is_visible():
continue
title = w.window_text()
cls = w.class_name()
except Exception:
continue
if args.title and args.title.lower() not in title.lower():
continue
if args.cls and args.cls != cls:
continue
targets.append(w)
print(f"Found {len(targets)} matching visible window(s):")
for w in targets:
print(f" handle={w.handle} class={w.class_name()!r} "
f"title={w.window_text()!r}")
if not targets:
print("Nothing to dump. Is the target window visible? "
"Try without filters, or adjust --title/--class.")
return
with open(args.out, "w", encoding="utf-8") as f:
old_stdout = sys.stdout
sys.stdout = f
try:
for w in targets:
print("=" * 72)
print(f"WINDOW handle={w.handle} class={w.class_name()!r} "
f"title={w.window_text()!r}")
print("=" * 72)
try:
w.print_control_identifiers()
except Exception as e:
print(f" (could not dump this window: {e})")
print()
finally:
sys.stdout = old_stdout
print(f"\nControl identifiers written to: {args.out}")
if __name__ == "__main__":
main()

View File

@@ -35,6 +35,16 @@ from pywinauto import timings
from fccs_config import parse_args, load_config, make_logger, load_lines from fccs_config import parse_args, load_config, make_logger, load_lines
# Per-drawer outcomes.
STATUS_SUCCESS = "success" # exported cleanly -> completed.txt
STATUS_CRASHED = "crashed" # FCCS converter crash -> crashed.txt (skip on rerun)
STATUS_FAILED = "failed" # navigation/timeout/other -> retried next run
# Substrings that mark the FCCS crash error dialog body (case-insensitive),
# used to distinguish a converter crash from other "FileCabinet CS" prompts.
_ERROR_SIGNATURES = ("failed", "access violation", "convert")
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
# HELPERS # HELPERS
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
@@ -53,6 +63,12 @@ def mark_completed(path, drawer_id):
f.write(drawer_id + "\n") f.write(drawer_id + "\n")
def mark_crashed(path, drawer_id):
"""Append a drawer ID to the crashed log (skipped on future runs)."""
with open(path, "a", encoding="utf-8") as f:
f.write(drawer_id + "\n")
def take_screenshot(window, drawer_id, tag, screenshot_dir, log): def take_screenshot(window, drawer_id, tag, screenshot_dir, log):
"""Save a screenshot of a window for later diagnosis of a failure.""" """Save a screenshot of a window for later diagnosis of a failure."""
from datetime import datetime from datetime import datetime
@@ -93,15 +109,54 @@ def connect_main(ctrl, log):
# PER-DRAWER EXPORT # PER-DRAWER EXPORT
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
def navigate_to_drawer(main, drawer_id, ctrl, timeouts, log): def find_search_box(main, ctrl, timeouts, log):
"""Type the drawer ID into the search box and click Go.""" """Locate the drawer search box reliably using the Go button as an anchor.
# Use best_match name "Edit" instead of title — the title is placeholder
# text that changes once the user types in the box.
box = main["Edit"]
box.wait("visible ready", timeout=timeouts["nav"])
box.set_edit_text("") The main window has several Edit controls (notably the data-location box
box.set_edit_text(drawer_id) inside a ComboBox). Their best_match names ("Edit", "Edit0"...) are
unstable and the placeholder title only exists in a fresh session, so we
anchor off the Go button instead: the search box sits on the same row
(same top Y) as Go, immediately to its left. Call once before the drawer
loop — the returned wrapper stays valid for the window's lifetime.
"""
go = main.child_window(title=ctrl["go_button_title"], class_name="Button")
go.wait("visible ready", timeout=timeouts["nav"])
go_rect = go.rectangle()
edits = main.children(class_name="Edit")
if not edits:
edits = main.descendants(class_name="Edit")
tolerance = 10
candidates = [
e for e in edits
if abs(e.rectangle().top - go_rect.top) <= tolerance
]
if not candidates:
raise RuntimeError(
"Could not find the drawer search box (no Edit control on the "
"Go button's row). FCCS may not be on the main drawer view."
)
# If several share the row, pick the one whose right edge is closest to
# (and left of) the Go button — that's the box directly beside Go.
left_of_go = [e for e in candidates if e.rectangle().right <= go_rect.left]
pool = left_of_go if left_of_go else candidates
box = max(pool, key=lambda e: e.rectangle().right)
r = box.rectangle()
log(f" Search box located at {(r.left, r.top, r.right, r.bottom)} "
f"(Go button top={go_rect.top})")
return box
def navigate_to_drawer(main, search_box, drawer_id, ctrl, timeouts, log):
"""Type the drawer ID into the search box and click Go."""
# search_box is a control wrapper (from find_search_box), not a
# WindowSpecification, so it has no .wait() — it was already confirmed
# present/visible when located. set_edit_text works directly on it.
search_box.set_edit_text("")
search_box.set_edit_text(drawer_id)
time.sleep(timeouts["settle"]) time.sleep(timeouts["settle"])
main.child_window(title=ctrl["go_button_title"], class_name="Button").click() main.child_window(title=ctrl["go_button_title"], class_name="Button").click()
@@ -129,58 +184,189 @@ def open_send_to_file(main, app, ctrl, timeouts):
return dlg return dlg
def perform_export(dlg, ctrl, timeouts): def read_manifest(dlg, drawer_id, log):
"""Click Select -> then OK to start the export.""" """Read the selected documents from the ListView into clean rows.
# Wait for the Select button to be ready — the dialog may still be
# populating its controls after appearing. The ListView (`List1`) has columns Drawer ID, Page Title (the document
name) and Application. `texts()` returns a flat, row-major dump led by the
control's own name ('List1'), which is why the raw capture looked messy.
This parses it into one row per document and drops the redundant Drawer ID
column (always equal to drawer_id) and any empty columns, leaving the
document name (and Application) per line. Returns a list of rows (each a
list of cell strings), [] if empty, or None if it couldn't be read.
"""
try:
lv = dlg.child_window(title="List1", class_name="SysListView32")
n_rows = lv.item_count()
except Exception as e:
log(f" WARNING: could not read manifest list: {e}")
return None
if n_rows == 0:
log(" WARNING: ListView is empty after Select")
return []
try:
n_cols = lv.column_count()
except Exception:
n_cols = 0
if not n_cols or n_cols < 1:
n_cols = 3 # observed columns: Drawer ID, Page Title, Application
# Read every cell. Primary: get_item(row, col). Fallback: reshape texts().
try:
grid = [[lv.get_item(r, c).text() for c in range(n_cols)]
for r in range(n_rows)]
except Exception as e:
log(f" NOTE: get_item failed ({e}); falling back to texts().")
raw = lv.texts()
if raw and raw[0] == lv.window_text():
raw = raw[1:] # drop the control's own name ('List1')
grid = [raw[i:i + n_cols] for i in range(0, len(raw), n_cols)]
grid = [row for row in grid if len(row) == n_cols]
# Keep only columns that carry real, non-redundant info.
keep_cols = []
for c in range(n_cols):
vals = [(row[c].strip() if c < len(row) else "") for row in grid]
if all(v == drawer_id for v in vals):
continue # redundant Drawer ID column
if all(v == "" for v in vals):
continue # empty column
keep_cols.append(c)
rows = [[(row[c] if c < len(row) else "") for c in keep_cols]
for row in grid]
log(f" Manifest captured: {len(rows)} documents selected")
return rows
def save_manifest(drawer_id, items, manifest_dir, log):
"""Save the list of expected export items to a manifest file."""
os.makedirs(manifest_dir, exist_ok=True)
path = os.path.join(manifest_dir, f"{drawer_id}.txt")
with open(path, "w", encoding="utf-8") as f:
for row in items:
if isinstance(row, (list, tuple)):
f.write("\t".join(str(c) for c in row) + "\n")
else:
f.write(str(row) + "\n")
log(f" Manifest saved: {path}")
def perform_export(dlg, drawer_id, ctrl, timeouts, log):
"""Click Select -> then OK to start the export. Returns manifest rows."""
select_btn = dlg.child_window(title=ctrl["select_btn_title"], class_name="Button") select_btn = dlg.child_window(title=ctrl["select_btn_title"], class_name="Button")
select_btn.wait("visible ready enabled", timeout=timeouts["dialog"]) select_btn.wait("visible ready enabled", timeout=timeouts["dialog"])
select_btn.click() select_btn.click()
# Wait for the OK button to become ready — transferring files to the
# selected side can take a while on large drawers.
ok_btn = dlg.child_window(title=ctrl["ok_btn_title"], class_name="Button") ok_btn = dlg.child_window(title=ctrl["ok_btn_title"], class_name="Button")
ok_btn.wait("visible ready enabled", timeout=timeouts["select"]) ok_btn.wait("visible ready enabled", timeout=timeouts["select"])
manifest = read_manifest(dlg, drawer_id, log)
ok_btn.click() ok_btn.click()
return manifest
def dismiss_any_dialog(app, ctrl, log): def dismiss_any_dialog(app, ctrl, log):
"""Try to find and dismiss any stale dialogs left from a previous failure.""" """Try to find and dismiss any stale dialogs left from a previous failure."""
# Dismiss a lingering converter-crash error dialog (OK)
w = _find_error_dialog(ctrl)
if w:
try:
_click_dialog_button(w, "OK", log)
log(" Dismissed stale crash error dialog.")
except Exception:
pass
# Dismiss "Send to" file saved confirmation dialog (OK)
w = _find_desktop_dialog(ctrl["saved_title"])
if w:
try:
_click_dialog_button(w, "OK", log)
log(" Dismissed stale confirmation dialog.")
except Exception:
pass
# Dismiss Send To File Location dialog (Cancel)
w = _find_desktop_dialog(ctrl["dialog_title"])
if w:
try:
_click_dialog_button(w, "Cancel", log)
log(" Dismissed stale Send To File dialog.")
except Exception:
pass
def _find_desktop_dialog(title_match):
"""Search all top-level windows for a visible #32770 dialog matching title."""
for w in Desktop(backend="win32").windows():
try:
if w.class_name() == "#32770" and w.is_visible() and title_match in w.window_text():
return w
except Exception:
continue
return None
def _click_dialog_button(dialog, button_title, log):
"""Connect to a desktop dialog and click a button on it."""
app = Application(backend="win32").connect(handle=dialog.handle)
dlg = app.window(handle=dialog.handle)
dlg.child_window(title=button_title, class_name="Button").click()
dlg.wait_not("visible", timeout=10)
def _find_error_dialog(ctrl):
"""Find the FCCS converter-crash error dialog, if present.
Matches a visible #32770 titled 'FileCabinet CS' whose body text carries a
crash signature (failed / access violation / convert), so we don't confuse
it with other 'FileCabinet CS' prompts. Returns the window or None.
"""
try: try:
for w in app.windows(): windows = Desktop(backend="win32").windows()
if not (w.class_name() == "#32770" and w.is_visible()): except Exception:
return None
for w in windows:
try:
if w.class_name() != ctrl["error_class"] or not w.is_visible():
continue continue
title = w.window_text() if ctrl["error_title"] not in w.window_text():
# Dismiss the Send To File dialog (Cancel)
if title == ctrl["dialog_title"]:
try:
w.child_window(title="Cancel", class_name="Button").click()
log(f" Dismissed stale Send To File dialog.")
time.sleep(0.5)
except Exception:
pass
continue continue
# Dismiss any other dialog (OK) body = " ".join(s.window_text()
try: for s in w.children(class_name="Static")).lower()
w.child_window(title="OK", class_name="Button").click() if any(sig in body for sig in _ERROR_SIGNATURES):
log(f" Dismissed stale dialog: {title!r}") return w
time.sleep(0.5) except Exception:
except Exception: continue
pass return None
def _read_progress_doc(app, ctrl):
"""Read the document name from the 'Exporting Documents' progress dialog."""
try:
progress = app.window(title=ctrl["progress_title"],
class_name=ctrl["progress_class"])
for st in progress.children(class_name="Static"):
t = st.window_text()
if t and "document" in t.lower():
return t.strip()
except Exception: except Exception:
pass pass
return "(unknown document)"
def wait_for_export(app, ctrl, timeouts, log): def wait_for_export(app, drawer_id, ctrl, timeouts, screenshot_dir, log):
""" """
Wait for the full export lifecycle to complete: Wait for the export lifecycle to complete, watching for two outcomes:
1. 'Exporting Documents' progress dialog appears (export running) - '"Send to" file saved' confirmation dialog -> STATUS_SUCCESS
2. Progress dialog disappears (export finished writing files) - 'FileCabinet CS' converter-crash error -> STATUS_CRASHED
3. '"Send to" file saved' confirmation dialog appears (MODAL) A drawer that shows neither before the deadline -> STATUS_FAILED.
4. Dismiss it by clicking OK
""" """
progress = app.window(title=ctrl["progress_title"], class_name=ctrl["progress_class"]) progress = app.window(title=ctrl["progress_title"], class_name=ctrl["progress_class"])
saved = app.window(title=ctrl["saved_title"], class_name=ctrl["saved_class"])
try: try:
progress.wait("visible", timeout=timeouts["progress_appear"]) progress.wait("visible", timeout=timeouts["progress_appear"])
@@ -189,71 +375,61 @@ def wait_for_export(app, ctrl, timeouts, log):
log(" NOTE: progress dialog not seen; checking for confirmation dialog " log(" NOTE: progress dialog not seen; checking for confirmation dialog "
"(drawer may have exported very quickly).") "(drawer may have exported very quickly).")
# Poll until the confirmation dialog appears OR we time out. # Poll the entire desktop each cycle for either the success dialog or a
# We do NOT rely on wait_not for the progress dialog because it can # converter-crash error. app.window() can miss dialogs it doesn't own.
# briefly disappear and reappear during long exports.
deadline = time.time() + timeouts["progress_finish"] deadline = time.time() + timeouts["progress_finish"]
saved_win = None
while time.time() < deadline: while time.time() < deadline:
# Check if the confirmation dialog has appeared # Read the current document BEFORE checking for the crash, so if the
if saved.exists(timeout=0): # error dialog has already replaced the progress text we still logged it.
current_doc = _read_progress_doc(app, ctrl)
err_win = _find_error_dialog(ctrl)
if err_win is not None:
log(f" CRASH: FCCS converter failed on {current_doc}. "
f"This aborts the drawer export.")
take_screenshot(err_win, drawer_id, "crash", screenshot_dir, log)
try:
_click_dialog_button(err_win, "OK", log)
log(" Crash dialog dismissed; FCCS returned to home screen.")
except Exception as e:
log(f" WARNING: could not dismiss crash dialog: {e}")
return STATUS_CRASHED
saved_win = _find_desktop_dialog(ctrl["saved_title"])
if saved_win is not None:
break break
time.sleep(2) time.sleep(2)
else:
log(f" ERROR: export did not finish within {timeouts['progress_finish']}s.") if saved_win is None:
return False log(f" ERROR: no confirmation or crash within {timeouts['progress_finish']}s.")
return STATUS_FAILED
log(f" Confirmation dialog found: {saved_win.window_text()!r}")
try: try:
saved.wait("visible ready", timeout=timeouts["confirm"]) _click_dialog_button(saved_win, "OK", log)
log(" Confirmation dialog appeared - export succeeded.")
except timings.TimeoutError:
# Fallback: look for any #32770 dialog that appeared (title may differ)
log(" WARNING: expected confirmation dialog not found by title. "
"Checking for any popup dialog...")
found = False
try:
for w in app.windows():
if w.class_name() == "#32770" and w.is_visible():
log(f" Found dialog: {w.window_text()!r}")
try:
w.child_window(title="OK", class_name="Button").click()
w.wait_not("visible", timeout=10)
log(" Dismissed via fallback.")
found = True
break
except Exception:
pass
except Exception:
pass
if not found:
log(" ERROR: confirmation dialog never appeared. Export may have "
"failed or produced no output.")
return False
log(" Export complete.")
return True
try:
saved.child_window(title=ctrl["saved_ok_title"], class_name="Button").click()
saved.wait_not("visible", timeout=timeouts["confirm"])
log(" Confirmation dismissed.") log(" Confirmation dismissed.")
except Exception as e: except Exception as e:
log(f" WARNING: could not dismiss confirmation dialog: {e}") log(f" WARNING: could not dismiss confirmation dialog: {e}")
log(" Export succeeded but dialog remains — will be dismissed next iteration.") log(" Export succeeded but dialog remains — will be dismissed next iteration.")
log(" Export complete.") log(" Export complete.")
return True return STATUS_SUCCESS
def export_drawer(app, main, drawer_id, ctrl, timeouts, screenshot_dir, log): def export_drawer(app, main, search_box, drawer_id, ctrl, timeouts,
"""Full per-drawer sequence. Returns True on success.""" screenshot_dir, manifest_dir, log):
"""Full per-drawer sequence. Returns a STATUS_* outcome."""
log(f"Drawer {drawer_id}: starting") log(f"Drawer {drawer_id}: starting")
# 0. Dismiss any stale dialogs left from a previous failure # 0. Dismiss any stale dialogs left from a previous failure
dismiss_any_dialog(app, ctrl, log) dismiss_any_dialog(app, ctrl, log)
# 1. Navigate # 1. Navigate
if not navigate_to_drawer(main, drawer_id, ctrl, timeouts, log): if not navigate_to_drawer(main, search_box, drawer_id, ctrl, timeouts, log):
take_screenshot(main, drawer_id, "nav_fail", screenshot_dir, log) take_screenshot(main, drawer_id, "nav_fail", screenshot_dir, log)
return False return STATUS_FAILED
# 2. Open Send To File dialog # 2. Open Send To File dialog
try: try:
@@ -261,11 +437,13 @@ def export_drawer(app, main, drawer_id, ctrl, timeouts, screenshot_dir, log):
except Exception as e: except Exception as e:
log(f" ERROR opening Send To File dialog: {e}") log(f" ERROR opening Send To File dialog: {e}")
take_screenshot(main, drawer_id, "dialog_fail", screenshot_dir, log) take_screenshot(main, drawer_id, "dialog_fail", screenshot_dir, log)
return False return STATUS_FAILED
# 3 & 4. Select -> and OK # 3 & 4. Select -> and OK (captures manifest before clicking OK)
try: try:
perform_export(dlg, ctrl, timeouts) manifest = perform_export(dlg, drawer_id, ctrl, timeouts, log)
if manifest is not None:
save_manifest(drawer_id, manifest, manifest_dir, log)
except Exception as e: except Exception as e:
log(f" ERROR during export selection: {e}") log(f" ERROR during export selection: {e}")
take_screenshot(dlg, drawer_id, "export_fail", screenshot_dir, log) take_screenshot(dlg, drawer_id, "export_fail", screenshot_dir, log)
@@ -273,16 +451,18 @@ def export_drawer(app, main, drawer_id, ctrl, timeouts, screenshot_dir, log):
dlg.child_window(title="Cancel", class_name="Button").click() dlg.child_window(title="Cancel", class_name="Button").click()
except Exception: except Exception:
pass pass
return False return STATUS_FAILED
# 5. Monitor progress # 5. Monitor progress (success / crash / timeout)
ok = wait_for_export(app, ctrl, timeouts, log) status = wait_for_export(app, drawer_id, ctrl, timeouts, screenshot_dir, log)
if not ok: if status == STATUS_CRASHED:
return STATUS_CRASHED
if status != STATUS_SUCCESS:
take_screenshot(main, drawer_id, "progress_fail", screenshot_dir, log) take_screenshot(main, drawer_id, "progress_fail", screenshot_dir, log)
return False return STATUS_FAILED
time.sleep(timeouts["settle"]) time.sleep(timeouts["settle"])
return True return STATUS_SUCCESS
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
@@ -298,7 +478,10 @@ def main():
paths = { paths = {
"drawer_id_file": cfg.get("paths", "drawer_id_file"), "drawer_id_file": cfg.get("paths", "drawer_id_file"),
"completed_file": cfg.get("paths", "completed_file"), "completed_file": cfg.get("paths", "completed_file"),
"ignore_file": cfg.get("paths", "ignore_file"),
"crashed_file": cfg.get("paths", "crashed_file"),
"screenshot_dir": cfg.get("paths", "screenshot_dir"), "screenshot_dir": cfg.get("paths", "screenshot_dir"),
"manifest_dir": cfg.get("paths", "manifest_dir"),
} }
timeouts = { timeouts = {
"nav": cfg.getfloat("timeouts", "nav_timeout"), "nav": cfg.getfloat("timeouts", "nav_timeout"),
@@ -322,6 +505,8 @@ def main():
"saved_title": cfg.get("controls", "saved_title"), "saved_title": cfg.get("controls", "saved_title"),
"saved_class": cfg.get("controls", "saved_class"), "saved_class": cfg.get("controls", "saved_class"),
"saved_ok_title": cfg.get("controls", "saved_ok_title"), "saved_ok_title": cfg.get("controls", "saved_ok_title"),
"error_title": cfg.get("controls", "error_title"),
"error_class": cfg.get("controls", "error_class"),
} }
log("=" * 60) log("=" * 60)
@@ -334,10 +519,17 @@ def main():
sys.exit(1) sys.exit(1)
completed = load_completed(paths["completed_file"]) completed = load_completed(paths["completed_file"])
ignored = set(load_lines(paths["ignore_file"]))
crashed = set(load_lines(paths["crashed_file"]))
pending = [d for d in all_ids if d not in completed] pending = [d for d in all_ids
if d not in completed
and d not in ignored
and d not in crashed]
log(f"Total drawers in list : {len(all_ids)}") log(f"Total drawers in list : {len(all_ids)}")
log(f"Already completed : {len(completed)}") log(f"Already completed : {len(completed)}")
log(f"Ignored (skipped) : {len(ignored & set(all_ids))}")
log(f"Crashed (skipped) : {len(crashed & set(all_ids))}")
log(f"Remaining to process : {len(pending)}") log(f"Remaining to process : {len(pending)}")
if not pending: if not pending:
@@ -346,15 +538,24 @@ def main():
app, main_win = connect_main(ctrl, log) app, main_win = connect_main(ctrl, log)
try:
search_box = find_search_box(main_win, ctrl, timeouts, log)
except Exception as e:
log(f"ERROR: {e}")
sys.exit(1)
success_count = 0 success_count = 0
fail_count = 0 fail_count = 0
crash_count = 0
failed_ids = [] failed_ids = []
crashed_ids = []
for i, drawer_id in enumerate(pending, start=1): for i, drawer_id in enumerate(pending, start=1):
log(f"--- [{i}/{len(pending)}] Drawer {drawer_id} ---") log(f"--- [{i}/{len(pending)}] Drawer {drawer_id} ---")
try: try:
ok = export_drawer(app, main_win, drawer_id, ctrl, timeouts, status = export_drawer(app, main_win, search_box, drawer_id, ctrl,
paths["screenshot_dir"], log) timeouts, paths["screenshot_dir"],
paths["manifest_dir"], log)
except Exception as e: except Exception as e:
log(f" UNEXPECTED ERROR on drawer {drawer_id}: {e}") log(f" UNEXPECTED ERROR on drawer {drawer_id}: {e}")
try: try:
@@ -362,13 +563,20 @@ def main():
paths["screenshot_dir"], log) paths["screenshot_dir"], log)
except Exception: except Exception:
pass pass
ok = False status = STATUS_FAILED
if ok: if status == STATUS_SUCCESS:
mark_completed(paths["completed_file"], drawer_id) mark_completed(paths["completed_file"], drawer_id)
success_count += 1 success_count += 1
log(f" Drawer {drawer_id} DONE ({success_count} ok / " log(f" Drawer {drawer_id} DONE ({success_count} ok / "
f"{fail_count} failed so far)") f"{fail_count} failed / {crash_count} crashed so far)")
elif status == STATUS_CRASHED:
mark_crashed(paths["crashed_file"], drawer_id)
crash_count += 1
crashed_ids.append(drawer_id)
log(f" Drawer {drawer_id} CRASHED - added to crashed list; will be "
f"skipped on re-run. Handle manually (export excluding the "
f"poison document).")
else: else:
fail_count += 1 fail_count += 1
failed_ids.append(drawer_id) failed_ids.append(drawer_id)
@@ -379,10 +587,15 @@ def main():
log("Run finished.") log("Run finished.")
log(f" Succeeded : {success_count}") log(f" Succeeded : {success_count}")
log(f" Failed : {fail_count}") log(f" Failed : {fail_count}")
log(f" Crashed : {crash_count}")
if failed_ids: if failed_ids:
log(f" Failed drawer IDs: {', '.join(failed_ids)}") log(f" Failed drawer IDs: {', '.join(failed_ids)}")
log(" These were NOT marked complete; re-running the script " log(" These were NOT marked complete; re-running the script "
"will retry them.") "will retry them.")
if crashed_ids:
log(f" Crashed drawer IDs: {', '.join(crashed_ids)}")
log(f" These were added to {paths['crashed_file']} and will be skipped "
"on re-run. Export them manually, excluding the crashing document.")
log("=" * 60) log("=" * 60)

35
fccs_folders.template.txt Normal file
View File

@@ -0,0 +1,35 @@
# FCCS Folder Templates — TEMPLATE (tracked in git; do NOT edit per-engagement)
# ---------------------
# On first run the scripts copy this file to fccs_folders.txt (git-ignored) and
# use that. Edit fccs_folders.txt for the current engagement so local changes
# never drift the repo. Delete fccs_folders.txt to reset from this template.
#
# List folder names exactly as shown in FCCS System Configuration > Document Folders.
# Keep the YYYY prefix for recurring folders. Non-recurring folders go as-is.
# Names containing "/" (e.g. "YYYY Foreign Bank/Income") can be copied verbatim —
# the code converts "/" to "-" automatically, matching how FCCS writes filenames.
# Thomson Reuters product folders (UltraTax CS, Planner CS, Practice CS) are
# handled automatically — do not list them here.
# Blank lines and lines starting with # are ignored.
#
# Examples:
# YYYY Tax Documents
# YYYY Billing & Invoices
# YYYY PAYROLL
# YYYY Financial Stmt Documents
# CORRESPONDENCE
# Permanent File
CORRESPONDENCE
Correspondence - Client
Correspondence - Taxing Authorities
Permanent File
YYYY Billing & Invoices
YYYY BILLING/Invoices
YYYY Fin Stmt DOCS
YYYY Financial Stmt Documents
YYYY PAYROLL
YYYY Payroll & Sales Tax
YYYY Tangible Tax Returns
YYYY TAX DOCS
YYYY Tax Documents
YYYY W-2 Forms

View File

@@ -1,14 +0,0 @@
# FCCS Folder Templates
# ---------------------
# List folder names exactly as shown in FCCS System Configuration > Document Folders.
# Keep the YYYY prefix for recurring folders. Non-recurring folders go as-is.
# UltraTax CS folders are handled automatically — do not list them here.
# Blank lines and lines starting with # are ignored.
#
# Examples:
# YYYY Tax Documents
# YYYY Billing & Invoices
# YYYY PAYROLL
# YYYY Financial Stmt Documents
# CORRESPONDENCE
# Permanent File

View File

@@ -13,11 +13,14 @@ Examples:
01069_ABRAHAM, REBEKAH L._UltraTax CS 12-31-2008_02-12-2009_2008 Form 1040 Filing Instructions.doc 01069_ABRAHAM, REBEKAH L._UltraTax CS 12-31-2008_02-12-2009_2008 Form 1040 Filing Instructions.doc
Output structure (recreates FCCS UI nesting): Output structure (recreates FCCS UI nesting):
output/{drawer_id}_{client_name}/Tax Documents/2025/{original_filename} output/{client_name}/Tax Documents/2025/{original_filename}
output/{drawer_id}_{client_name}/UltraTax CS/12-31-2008/{original_filename} output/{client_name}/UltraTax CS/12-31-2008/{original_filename}
output/{drawer_id}_{client_name}/Permanent File/{original_filename} output/{client_name}/Permanent File/{original_filename}
Files that cannot be parsed go to output/_unparsed/ for manual review. Files that cannot be parsed are still filed under their client:
output/{client_name}/_unparsed/{original_filename}
Only files whose client can't be recovered from the leading tokens fall back to
the top-level output/_unparsed/ for manual review.
""" """
import os import os
@@ -28,10 +31,15 @@ import sys
from fccs_config import ( from fccs_config import (
parse_args, load_config, make_logger, parse_args, load_config, make_logger,
load_folder_list, build_folder_patterns, decompose_folder_path, load_folder_list, build_folder_patterns, decompose_folder_path,
client_name_from_filename,
) )
# Regex for the creation date field (MM-DD-YYYY) bounded by underscores # Regex for the creation date field (MM-DD-YYYY) bounded by underscores
_DATE_RE = re.compile(r"_(\d{2}-\d{2}-\d{4})_") _DATE_RE = re.compile(r"_(\d{2}-\d{2}-\d{4})_")
# Trailing "[Subfolder Name]" on the folder field — FCCS encodes nested
# subfolders (e.g. Income Documents > 2016 > "4201 N Beach Street, LLC") as
# "2016 Income Documents[4201 N Beach Street, LLC]" in the export filename.
_SUBFOLDER_RE = re.compile(r"\[([^\[\]]+)\]$")
def parse_filename(filename, folder_patterns): def parse_filename(filename, folder_patterns):
@@ -77,22 +85,47 @@ def parse_filename(filename, folder_patterns):
after_date_pos = m.start() - 1 + len("_") + len(date_str) + len("_") after_date_pos = m.start() - 1 + len("_") + len(date_str) + len("_")
doc_name = rest[after_date_pos:] doc_name = rest[after_date_pos:]
# Try each folder pattern against the end of before_date # Peel any trailing [Subfolder] groups off the folder field so the
# remainder can match a template; each becomes one more nested path
# component (in order, so "a[b][c]" nests as a/b/c).
folder_text = before_date
sub_parts = []
while True:
sm = _SUBFOLDER_RE.search(folder_text)
if not sm:
break
sub_parts.insert(0, sm.group(1))
folder_text = folder_text[: sm.start()]
# Try each folder pattern against the end of the folder text
for pattern, template_name, folder_type in folder_patterns: for pattern, template_name, folder_type in folder_patterns:
folder_match = pattern.search(before_date) folder_match = pattern.search(folder_text)
if folder_match and folder_match.end() == len(before_date): if folder_match and folder_match.end() == len(folder_text):
# Folder matched at the end — check for underscore separator before it # Folder matched at the end — check for underscore separator before it
folder_start = folder_match.start() folder_start = folder_match.start()
if folder_start > 0 and before_date[folder_start - 1] == "_": if folder_start > 0 and folder_text[folder_start - 1] == "_":
client_name = before_date[: folder_start - 1] client_name = folder_text[: folder_start - 1]
folder_parts = decompose_folder_path( folder_parts = decompose_folder_path(
folder_match, folder_type, template_name folder_match, folder_type, template_name
) ) + tuple(sub_parts)
return (drawer_id, client_name, folder_parts, date_str, doc_name + ext) return (drawer_id, client_name, folder_parts, date_str, doc_name + ext)
return None return None
def safe_component(name):
"""Make a single path component safe to create on Windows.
Windows cannot create a directory whose name ends in a space or a period —
os.makedirs raises FileNotFoundError (WinError 3). Client names such as
'FARR GROUP, P.L.' (trailing period) or ones with a stray trailing space
trigger this, so strip trailing dots/spaces. Falls back to '_' if the name
is entirely dots/spaces.
"""
cleaned = name.rstrip(" .")
return cleaned if cleaned else "_"
def main(): def main():
args = parse_args() args = parse_args()
cfg = load_config(args.config) cfg = load_config(args.config)
@@ -108,7 +141,8 @@ def main():
folder_templates = load_folder_list(folder_list_path) folder_templates = load_folder_list(folder_list_path)
if not folder_templates: if not folder_templates:
log("WARNING: folder list is empty. Only built-in patterns (UltraTax CS) will match.") log("WARNING: folder list is empty. Only built-in patterns (UltraTax CS, "
"Planner CS, Practice CS) will match.")
folder_patterns = build_folder_patterns(folder_templates) folder_patterns = build_folder_patterns(folder_templates)
log(f"Loaded {len(folder_patterns)} folder patterns") log(f"Loaded {len(folder_patterns)} folder patterns")
@@ -121,44 +155,75 @@ def main():
success = 0 success = 0
failed = 0 failed = 0
errored = 0
for filename in files: for filename in files:
result = parse_filename(filename, folder_patterns) # Defensive per-file: a single bad path (e.g. an un-creatable name) must
# not abort the whole run — log it and move on to the next file.
try:
result = parse_filename(filename, folder_patterns)
if result is None: if result is None:
log(f" UNPARSED: {filename}") # Still recover the client from the leading tokens so the file
dest_dir = os.path.join(output_dir, "_unparsed") # lands in that client's folder (in an _unparsed subfolder to
failed += 1 # flag it), rather than a single top-level bucket. Fall back to
else: # the top-level _unparsed only when even the client is unknown.
drawer_id, client_name, folder_parts, date, doc_name = result client_name = client_name_from_filename(filename)
dest_dir = os.path.join( if client_name:
output_dir, dest_dir = os.path.join(output_dir,
client_name, safe_component(client_name),
*folder_parts, "_unparsed")
) log(f" UNPARSED (filed under {client_name}): {filename}")
else:
dest_dir = os.path.join(output_dir, "_unparsed")
log(f" UNPARSED (no client): {filename}")
dest_filename = filename # keep original name for review
parsed = False
else:
drawer_id, client_name, folder_parts, date, doc_name = result
dest_dir = os.path.join(
output_dir,
safe_component(client_name),
*(safe_component(p) for p in folder_parts),
)
dest_filename = doc_name
parsed = True
os.makedirs(dest_dir, exist_ok=True)
src = os.path.join(export_dir, filename)
dst = os.path.join(dest_dir, dest_filename)
# Handle duplicate filenames
if os.path.exists(dst):
base, fext = os.path.splitext(dest_filename)
counter = 1
while os.path.exists(dst):
dst = os.path.join(dest_dir, f"{base}_{counter}{fext}")
counter += 1
log(f" DUPLICATE renamed: {dest_filename} -> {os.path.basename(dst)}")
shutil.copy2(src, dst)
except Exception as e:
log(f" ERROR processing {filename}: {e}")
errored += 1
continue
# Count only after a successful copy so the totals stay accurate.
if parsed:
success += 1 success += 1
else:
os.makedirs(dest_dir, exist_ok=True) failed += 1
src = os.path.join(export_dir, filename)
# For parsed files, use just the document name; for unparsed, keep original
dest_filename = doc_name if result else filename
dst = os.path.join(dest_dir, dest_filename)
# Handle duplicate filenames
if os.path.exists(dst):
base, fext = os.path.splitext(dest_filename)
counter = 1
while os.path.exists(dst):
dst = os.path.join(dest_dir, f"{base}_{counter}{fext}")
counter += 1
log(f" DUPLICATE renamed: {dest_filename} -> {os.path.basename(dst)}")
shutil.copy2(src, dst)
log("=" * 60) log("=" * 60)
log(f"Reorganization complete: {success} organized, {failed} unparsed") log(f"Reorganization complete: {success} organized, {failed} unparsed, "
f"{errored} errored")
if failed: if failed:
log(f"Review unparsed files in: {os.path.join(output_dir, '_unparsed')}") log("Review unparsed files in each client's _unparsed subfolder "
f"(and {os.path.join(output_dir, '_unparsed')} for any without a "
"detectable client).")
if errored:
log(f"{errored} file(s) could not be placed due to errors — see the "
"ERROR lines above; these were left in the export folder.")
log("=" * 60) log("=" * 60)

239
fccs_report.py Normal file
View File

@@ -0,0 +1,239 @@
"""
Utility: Generate a client-facing progress report (HTML).
Runs the same document-level completeness check as fccs_verify.py, but produces
a clean, self-contained HTML file listing ONLY the drawers with outstanding
items and, for each, the specific documents still missing from the export. It's
meant as a "check-in progress" document you can send to the client.
Drawers that are fully exported or that have no manifest are left out — the
report shows outstanding work. A headline shows how many drawers are fully
migrated so overall progress is clear.
Any outstanding drawer that also appears in crashed.txt (an FCCS converter crash
aborted its export) is badged CRASHED and sorted to the top: those are genuine
failures worth attention, as opposed to benign name-match false positives (e.g.
a collapsed "container" document whose children exported fine but whose name
lands in the filename's folder field rather than the document field).
Output goes to `report_html` (default C:\\Migration\\progress_report.html).
USAGE
-----
python fccs_report.py
"""
import os
import sys
import html
from datetime import datetime
from fccs_config import (
parse_args, load_config, load_lines,
evaluate_drawer, index_files_by_drawer, client_name_from_files,
)
PAGE_TEMPLATE = """<!doctype html>
<html lang="en">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>Migration Progress Report</title>
<style>
* {{ box-sizing: border-box; }}
body {{
font-family: -apple-system, "Segoe UI", Roboto, Helvetica, Arial, sans-serif;
color: #1f2933; background: #f5f7fa; margin: 0; padding: 2.5rem 1rem;
line-height: 1.5;
}}
.sheet {{
max-width: 820px; margin: 0 auto; background: #fff; border-radius: 10px;
box-shadow: 0 1px 3px rgba(16,24,40,.1), 0 1px 2px rgba(16,24,40,.06);
overflow: hidden;
}}
header {{
padding: 2rem 2.25rem 1.5rem; border-bottom: 1px solid #e4e7eb;
}}
h1 {{ margin: 0 0 .25rem; font-size: 1.5rem; font-weight: 650; }}
.date {{ color: #7b8794; font-size: .9rem; }}
.summary {{
display: flex; gap: 2rem; flex-wrap: wrap;
padding: 1.25rem 2.25rem; background: #fafbfc; border-bottom: 1px solid #e4e7eb;
}}
.stat .num {{ font-size: 1.75rem; font-weight: 680; line-height: 1; }}
.stat .lbl {{ color: #7b8794; font-size: .8rem; text-transform: uppercase;
letter-spacing: .03em; margin-top: .35rem; }}
.stat.good .num {{ color: #2e7d32; }}
.stat.flag .num {{ color: #c0392b; }}
.stat.crash .num {{ color: #b9770e; }}
.body {{ padding: 1.5rem 2.25rem 2.25rem; }}
.all-clear {{
text-align: center; padding: 2.5rem 1rem; color: #2e7d32; font-size: 1.1rem;
}}
.drawer {{
border: 1px solid #e4e7eb; border-radius: 8px; padding: 1rem 1.25rem;
margin-bottom: 1rem;
}}
.drawer:last-child {{ margin-bottom: 0; }}
.drawer.crashed {{ border-left: 4px solid #c0392b; background: #fdf6f5; }}
.drawer h2 {{
margin: 0 0 .1rem; font-size: 1.05rem; font-weight: 620;
}}
.drawer .id {{ color: #486581; font-variant-numeric: tabular-nums; }}
.badge {{
display: inline-block; margin-left: .5rem; padding: .1rem .5rem;
font-size: .72rem; font-weight: 700; letter-spacing: .04em;
text-transform: uppercase; color: #fff; background: #c0392b;
border-radius: 4px; vertical-align: middle;
}}
.drawer .meta {{ color: #7b8794; font-size: .82rem; margin-bottom: .6rem; }}
ul.docs {{ margin: 0; padding: 0; list-style: none; }}
ul.docs li {{
padding: .3rem 0 .3rem 1.5rem; position: relative; font-size: .93rem;
border-top: 1px solid #f0f2f5;
}}
ul.docs li:first-child {{ border-top: none; }}
ul.docs li::before {{
content: "\\2717"; color: #c0392b; position: absolute; left: 0; font-weight: 700;
}}
footer {{
padding: 1rem 2.25rem; border-top: 1px solid #e4e7eb; color: #9aa5b1;
font-size: .78rem;
}}
@media print {{
body {{ background: #fff; padding: 0; }}
.sheet {{ box-shadow: none; max-width: none; }}
}}
</style>
</head>
<body>
<div class="sheet">
<header>
<h1>Migration Progress Report</h1>
<div class="date">{date}</div>
</header>
<div class="summary">
<div class="stat good"><div class="num">{complete}</div>
<div class="lbl">Drawers migrated</div></div>
<div class="stat flag"><div class="num">{incomplete}</div>
<div class="lbl">Drawers outstanding</div></div>
<div class="stat crash"><div class="num">{crashed}</div>
<div class="lbl">Crashed in export</div></div>
<div class="stat"><div class="num">{total}</div>
<div class="lbl">Total drawers</div></div>
</div>
<div class="body">
{content}
</div>
<footer>Generated {date} &middot; Lists documents not yet present in the
export. Drawers not shown are fully migrated. A
<span class="badge">Crashed</span> badge marks drawers whose export was
interrupted by an error.</footer>
</div>
</body>
</html>
"""
def build_drawer_block(drawer_id, client, missing, crashed=False):
"""Return the HTML for one outstanding drawer."""
head = f'<span class="id">{html.escape(drawer_id)}</span>'
if client:
head += f' &mdash; {html.escape(client)}'
if crashed:
head += ' <span class="badge">Crashed</span>'
n = len(missing)
docs = "\n".join(
f" <li>{html.escape(m)}</li>" for m in missing
)
cls = "drawer crashed" if crashed else "drawer"
return (
f' <div class="{cls}">\n'
f' <h2>{head}</h2>\n'
f' <div class="meta">{n} document{"s" if n != 1 else ""} '
'outstanding</div>\n'
f' <ul class="docs">\n{docs}\n </ul>\n'
' </div>'
)
def main():
args = parse_args()
cfg = load_config(args.config)
export_dir = cfg.get("paths", "export_dir")
manifest_dir = cfg.get("paths", "manifest_dir")
report_html = cfg.get("paths", "report_html")
crashed_set = set(load_lines(cfg.get("paths", "crashed_file")))
if not os.path.isdir(manifest_dir):
print(f"ERROR: manifest directory not found: {manifest_dir}")
print("Run fccs_export.py first to generate manifests.")
sys.exit(1)
if not os.path.isdir(export_dir):
print(f"ERROR: export directory not found: {export_dir}")
sys.exit(1)
drawer_ids = sorted(
os.path.splitext(f)[0]
for f in os.listdir(manifest_dir)
if f.endswith(".txt")
)
if not drawer_ids:
print("No manifest files found.")
sys.exit(1)
files_by_drawer = index_files_by_drawer(export_dir)
complete = 0
outstanding = [] # (drawer_id, client, missing_list, crashed)
for drawer_id in drawer_ids:
files = files_by_drawer.get(drawer_id, [])
r = evaluate_drawer(drawer_id, files, manifest_dir)
if r["missing"]:
client = client_name_from_files(files)
outstanding.append((drawer_id, client, r["missing"],
drawer_id in crashed_set))
else:
complete += 1
# Crashed drawers are the genuine failures — surface them first.
outstanding.sort(key=lambda o: (not o[3], o[0]))
crashed_ids = [d for d, _, _, cr in outstanding if cr]
total = len(drawer_ids)
date_str = f"{datetime.now():%B %d, %Y}"
if outstanding:
content = "\n".join(
build_drawer_block(d, c, m, cr) for d, c, m, cr in outstanding
)
else:
content = (' <div class="all-clear">All drawers are fully '
"migrated &mdash; no outstanding documents.</div>")
page = PAGE_TEMPLATE.format(
date=html.escape(date_str),
complete=complete,
incomplete=len(outstanding),
crashed=len(crashed_ids),
total=total,
content=content,
)
with open(report_html, "w", encoding="utf-8") as f:
f.write(page)
print(f"Progress report written to {report_html}")
print(f" {complete}/{total} drawers fully migrated, "
f"{len(outstanding)} outstanding.")
if outstanding:
print(" Outstanding: " + ", ".join(d for d, _, _, _ in outstanding))
if crashed_ids:
print(f" Outstanding AND crashed ({len(crashed_ids)}) "
"— spot-check these first: " + ", ".join(crashed_ids))
if __name__ == "__main__":
main()

162
fccs_report_reorganize.py Normal file
View File

@@ -0,0 +1,162 @@
"""
Utility: Analyze the results of fccs_reorganize.py.
Internal-use console report. Walks the configured output_dir looking for
_unparsed folders (files the reorganizer couldn't match against a folder
template) and summarizes what it finds, so gaps in fccs_folders.txt are easy
to spot.
Because unparsed files keep their original export filename
({drawer}_{client}_{folder}_{MM-DD-YYYY}_{doc}.ext), the folder field can
still be recovered from the name. The report aggregates those folder fields
(with years generalized back to YYYY) into suggested template lines you can
paste straight into fccs_folders.txt, then re-run the reorganizer.
Usage:
python fccs_report_reorganize.py [--config path\\to\\config.ini]
"""
import os
import re
import sys
from fccs_config import parse_args, load_config
# Creation-date field that terminates the folder portion of the filename.
_DATE_RE = re.compile(r"_(\d{2}-\d{2}-\d{4})_")
# A standalone year inside a folder name, e.g. "2013 Tax Documents".
_YEAR_RE = re.compile(r"\b(19|20)\d{2}\b")
# Trailing "[Subfolder]" group(s) — FCCS's encoding for nested subfolders.
_SUBFOLDER_RE = re.compile(r"(?:\[[^\[\]]+\])+$")
UNPARSED_DIRNAME = "_unparsed"
def folder_field_from_filename(filename):
"""Recover the folder field from an unparsed export filename, or None.
Format: {drawer}_{client}_{folder}_{MM-DD-YYYY}_{doc}.ext
Drawer and client are the first two underscore tokens (client names never
contain underscores); the folder field runs from there to the first
creation-date anchor.
"""
stem = os.path.splitext(filename)[0]
parts = stem.split("_", 2)
if len(parts) < 3:
return None
rest = parts[2] # folder + date + doc
m = _DATE_RE.search("_" + rest)
if not m:
return None
folder = rest[: max(m.start() - 1, 0)].strip("_").strip()
return folder or None
def suggest_template(folder_field):
"""Generalize a concrete folder field into an fccs_folders.txt line.
e.g. "2013 Tax Documents" -> "YYYY Tax Documents". Fields without a year
are suggested as-is (static folders). Trailing [Subfolder] groups (FCCS's
nested-subfolder encoding) are dropped so the suggestion is the parent
template line, which is what fccs_folders.txt actually takes.
"""
stripped = _SUBFOLDER_RE.sub("", folder_field).strip()
if stripped:
folder_field = stripped
return _YEAR_RE.sub("YYYY", folder_field)
def main():
args = parse_args()
cfg = load_config(args.config)
output_dir = cfg.get("paths", "output_dir")
if not os.path.isdir(output_dir):
print(f"ERROR: output directory not found: {output_dir}")
sys.exit(1)
clients_total = 0
clients_with_unparsed = [] # (client_name, [filenames])
top_level_unparsed = [] # filenames in output/_unparsed
for name in sorted(os.listdir(output_dir)):
path = os.path.join(output_dir, name)
if not os.path.isdir(path):
continue
if name == UNPARSED_DIRNAME:
top_level_unparsed = sorted(
f for f in os.listdir(path)
if os.path.isfile(os.path.join(path, f))
)
continue
clients_total += 1
up = os.path.join(path, UNPARSED_DIRNAME)
if os.path.isdir(up):
files = sorted(
f for f in os.listdir(up)
if os.path.isfile(os.path.join(up, f))
)
if files:
clients_with_unparsed.append((name, files))
print("=" * 64)
print("REORGANIZE RESULTS REPORT")
print(f"Output directory : {output_dir}")
print(f"Client folders : {clients_total}")
print(f"With _unparsed : {len(clients_with_unparsed)}")
print(f"Top-level _unparsed (no client recovered): {len(top_level_unparsed)}")
print("=" * 64)
if not clients_with_unparsed and not top_level_unparsed:
print("CLEAN — no _unparsed folders found. All files matched a "
"folder template.")
return
# Aggregate folder fields across every unparsed file to point at the
# template gaps directly.
suggestions = {} # suggested template line -> file count
unrecovered = 0 # unparsed files whose folder field couldn't be read
# One line per affected client folder — the files themselves are on disk
# in that client's _unparsed subfolder if a closer look is needed.
print()
print("Client folders with _unparsed files:")
for client, files in clients_with_unparsed:
print(f" {client} ({len(files)} unparsed)")
for f in files:
field = folder_field_from_filename(f)
if field:
key = suggest_template(field)
suggestions[key] = suggestions.get(key, 0) + 1
else:
unrecovered += 1
if top_level_unparsed:
print(f" {UNPARSED_DIRNAME}/ (top level — client unknown, "
f"{len(top_level_unparsed)} files)")
for f in top_level_unparsed:
field = folder_field_from_filename(f)
if field:
key = suggest_template(field)
suggestions[key] = suggestions.get(key, 0) + 1
else:
unrecovered += 1
print()
if suggestions:
print("-" * 64)
print("Possible missing folder templates (add to fccs_folders.txt,")
print("then re-run fccs_reorganize.py):")
for tmpl, count in sorted(suggestions.items(),
key=lambda kv: (-kv[1], kv[0])):
print(f" {count:4d}x{tmpl}")
if unrecovered:
print(f"\n{unrecovered} unparsed file(s) had no recoverable folder "
"field (no date anchor in the name) — likely oddball names, "
"not template gaps.")
if __name__ == "__main__":
main()

View File

@@ -8,7 +8,65 @@ Writes the sorted list of drawer IDs to the configured drawer_id_file.
import os import os
import sys import sys
from fccs_config import parse_args, load_config, make_logger from fccs_config import (
parse_args, load_config, make_logger, load_lines, check_for_clashes,
)
_IGNORE_HEADER = [
"# Drawer IDs to SKIP during export (one ID per line).",
"# Lines starting with # are comments and are ignored.",
"#",
"# Add password-protected or otherwise-excluded drawers here.",
"#",
"# The IDs auto-added below are prefix clashes: searching the base ID in",
"# FCCS pops a selection box that breaks the plain export. Only the base",
"# (shorter) ID is listed — the longer, more-specific IDs export normally.",
"# The base IDs are exported separately by fccs_export_clashes.py, which",
"# handles the selection box. DELETE or comment out any you'd rather skip.",
"",
]
def update_ignore_file(ignore_file, clashes, log):
"""Create ignore.txt if missing and seed it with clash base (prefix) IDs.
Only the shorter (prefix) ID of each clash is added — that's the one that
triggers the FCCS selection box and needs special handling. The longer,
more-specific IDs (e.g. '02218A') search fine and export normally.
Never overwrites existing entries — only appends prefixes not already
present. Returns the list of IDs added.
"""
existing = set(load_lines(ignore_file))
file_exists = os.path.exists(ignore_file)
to_add = [(short, matches) for short, matches in clashes
if short not in existing]
# Existing file already covers every clash — leave it untouched.
if file_exists and not to_add:
return []
lines = []
if not file_exists:
lines.extend(_IGNORE_HEADER)
for short, matches in to_add:
lines.append(f"# clashes with: {', '.join(matches)}")
lines.append(short)
mode = "a" if file_exists else "w"
with open(ignore_file, mode, encoding="utf-8") as f:
if file_exists:
f.write("\n") # separate the new block from prior content
f.write("\n".join(lines) + "\n")
added = [short for short, _ in to_add]
if not file_exists:
log(f"Created ignore file: {ignore_file}")
if added:
log(f"Added {len(added)} clash base ID(s) to ignore list: "
f"{', '.join(added)}")
return added
def main(): def main():
@@ -18,15 +76,33 @@ def main():
backup_dir = cfg.get("paths", "backup_dir") backup_dir = cfg.get("paths", "backup_dir")
output_file = cfg.get("paths", "drawer_id_file") output_file = cfg.get("paths", "drawer_id_file")
ignore_file = cfg.get("paths", "ignore_file")
if not os.path.isdir(backup_dir): if not os.path.isdir(backup_dir):
log(f"ERROR: backup directory not found: {backup_dir}") log(f"ERROR: backup directory not found: {backup_dir}")
sys.exit(1) sys.exit(1)
drawer_ids = sorted( # Only subdirectories are drawers. When pointed at FCCS's live data dir
# (e.g. the Restore directory) there are also system folders that start with
# "$" and miscellaneous loose files — skip both. The isdir check drops the
# files; the "$" prefix check drops the system folders.
raw_dirs = [
name for name in os.listdir(backup_dir) name for name in os.listdir(backup_dir)
if os.path.isdir(os.path.join(backup_dir, name)) if not name.startswith("$")
) and os.path.isdir(os.path.join(backup_dir, name))
]
# FileCabinet CS ignores "." in drawer IDs: a folder named "A123.TJ" on disk
# is searched and displayed in the UI as "A123TJ", so searching the dotted
# form returns nothing. Everything downstream uses the UI form — export types
# the ID into the search box, and FCCS embeds the same dot-free ID as the
# prefix of exported filenames (which manifest names and verify/report keys
# are matched against) — so normalize by stripping dots here at the source.
normalized = {}
for name in raw_dirs:
did = name.replace(".", "")
normalized.setdefault(did, []).append(name)
drawer_ids = sorted(normalized)
with open(output_file, "w", encoding="utf-8") as f: with open(output_file, "w", encoding="utf-8") as f:
for did in drawer_ids: for did in drawer_ids:
@@ -34,6 +110,41 @@ def main():
log(f"Found {len(drawer_ids)} drawers, written to {output_file}") log(f"Found {len(drawer_ids)} drawers, written to {output_file}")
# Warn if stripping dots collapsed two distinct folders onto one FCCS ID —
# that would silently drop a drawer from the inventory otherwise.
collisions = {did: names for did, names in normalized.items() if len(names) > 1}
if collisions:
log("-" * 60)
log(f"WARNING: {len(collisions)} drawer ID(s) collide after removing "
"'.' — multiple folders map to a single FCCS ID:")
for did, names in sorted(collisions.items()):
log(f" {did} <- {', '.join(sorted(names))}")
log("-" * 60)
# Flag prefix clashes: FCCS search on the shorter ID pops a selection box.
clashes = check_for_clashes(drawer_ids)
if clashes:
log("-" * 60)
log(f"WARNING: {len(clashes)} potential drawer ID clash(es) found.")
log("Searching the shorter ID in FCCS may show a selection box "
"instead of navigating directly, which breaks the export.")
for short, matches in clashes:
log(f" {short} -> also matches: {', '.join(matches)}")
# Seed ignore.txt with the clash prefixes for review (never overwrites).
update_ignore_file(ignore_file, clashes, log)
log(f"Review the ignore list: {ignore_file}")
log("-" * 60)
else:
log("No drawer ID prefix clashes found.")
# Report which drawers will be skipped by the export.
ignored = set(load_lines(ignore_file))
if ignored:
present = sorted(ignored & set(drawer_ids))
log(f"Ignore list ({ignore_file}): {len(ignored)} IDs listed, "
f"{len(present)} present in this backup — these will be SKIPPED "
f"by fccs_export.py: {', '.join(present) if present else '(none present)'}")
if __name__ == "__main__": if __name__ == "__main__":
main() main()

95
fccs_verify.py Normal file
View File

@@ -0,0 +1,95 @@
"""
Step 4 (optional): Verify exported files against manifests.
Batch check across EVERY drawer that has a manifest: compares each drawer's
manifest (captured during Step 2) against the files in the export directory and
reports which drawers are complete vs missing documents.
Matching is document-level and identical to fccs_check.py (via
fccs_config.evaluate_drawer): it accounts for page-splitting (a document
exported as 'Name Page 1', 'Name Page 2', ... counts as present) and for
filename sanitization (titles with characters illegal in filenames still
match). Use fccs_check.py to spot-check individual drawers interactively.
"""
import os
import sys
from fccs_config import (
parse_args, load_config, make_logger,
evaluate_drawer, index_files_by_drawer,
)
def main():
args = parse_args()
cfg = load_config(args.config)
log = make_logger(cfg.get("paths", "verify_report"))
export_dir = cfg.get("paths", "export_dir")
manifest_dir = cfg.get("paths", "manifest_dir")
if not os.path.isdir(manifest_dir):
log(f"ERROR: manifest directory not found: {manifest_dir}")
log("Run fccs_export.py first to generate manifests.")
sys.exit(1)
if not os.path.isdir(export_dir):
log(f"ERROR: export directory not found: {export_dir}")
sys.exit(1)
drawer_ids = sorted(
os.path.splitext(f)[0]
for f in os.listdir(manifest_dir)
if f.endswith(".txt")
)
if not drawer_ids:
log("No manifest files found.")
sys.exit(1)
files_by_drawer = index_files_by_drawer(export_dir)
log("=" * 60)
log("Export Verification Report")
log("=" * 60)
log(f"Manifests loaded : {len(drawer_ids)} drawers")
log("")
complete_ids = []
incomplete = [] # (drawer_id, missing_list)
for drawer_id in drawer_ids:
r = evaluate_drawer(drawer_id, files_by_drawer.get(drawer_id, []),
manifest_dir)
n_expected = len(r["expected"])
missing = r["missing"]
if not missing:
complete_ids.append(drawer_id)
log(f" {drawer_id}: OK ({n_expected} docs, {r['file_count']} files)")
else:
incomplete.append((drawer_id, missing))
log(f" {drawer_id}: INCOMPLETE — {len(missing)}/{n_expected} "
f"document(s) missing:")
for m in missing:
log(f" - {m}")
# Exported files whose drawer has no manifest at all.
orphan_drawers = sorted(set(files_by_drawer) - set(drawer_ids))
orphan_count = sum(len(files_by_drawer[d]) for d in orphan_drawers)
log("")
log("-" * 60)
log(f"Complete drawers : {len(complete_ids)}")
log(f"Incomplete drawers : {len(incomplete)}")
if incomplete:
log(f" Incomplete IDs: {', '.join(d for d, _ in incomplete)}")
if orphan_drawers:
log(f"No manifest for : {', '.join(orphan_drawers)} "
f"({orphan_count} files)")
if not incomplete and not orphan_drawers:
log("All drawers complete.")
log("=" * 60)
if __name__ == "__main__":
main()

374
helpers/rerun_failed_uploads.py Executable file
View File

@@ -0,0 +1,374 @@
#!/usr/bin/env python3
"""
rerun_failed_uploads.py - parse a SmartVault bulk-upload log, work out which
files still need uploading, and optionally copy them (mirroring the relative
folder structure) to a new root so the upload tool can be re-run against it.
Default behaviour is a DRY RUN: it prints a summary and writes a report.
Add --copy to actually copy files.
What counts as "needs re-upload":
* http-error : a file line in the error section (502/503/504 etc.)
* folder-error : "All files under subdirectory: X [Can't Upload Folder]"
-> files sitting DIRECTLY in X (non-recursive)
* after-crash : the run aborted; the tool walks ~alphabetically, so every
top-level folder sorting >= the crash file's top-level folder
(inclusive) is included entirely. Disable with --no-after-crash.
Files in the "renamed due to invalid character" section DID upload and are
only reported, never copied.
Paths inside the log are always Windows paths (G:\\...). They are parsed with
ntpath regardless of the OS this script runs on, and the drive letter is
translated to a local mount via DRIVE_MAP (see below) so the script works on
Linux with the drive mounted under /run/media.
"""
from __future__ import annotations
import argparse
import ntpath
import os
import re
import shutil
import sys
from dataclasses import dataclass, field
# ---------------------------------------------------------------------------
# MACHINE-SPECIFIC SETTINGS - EDIT THESE WHEN MOVING BETWEEN LINUX AND WINDOWS
# ---------------------------------------------------------------------------
# The log file always contains Windows paths (G:\Source_Files_Organized\...).
# DRIVE_MAP tells the script where that drive letter is actually mounted on the
# machine running this script. Only the drive letter is translated; the rest of
# the path is kept as-is.
#
# Linux (Fedora, drive auto-mounted under /run/media):
# DRIVE_MAP = {"G:": "/run/media/way/Kirk H Fritschen CPA"}
#
# Windows (drive mapped as G:):
# DRIVE_MAP = {"G:": "G:\\"}
#
# Instead of editing this you can also pass --source-root on the command line,
# e.g. --source-root "G:\Source_Files_Organized" on Windows.
DRIVE_MAP = {
"G:": "/run/media/way/Kirk H Fritschen CPA",
}
# Where the files that need re-uploading get mirrored to (relative folder
# structure is preserved underneath it). It is kept on the same drive so the
# Windows upload tool can afterwards be pointed at G:\Rerun_Source_Files.
#
# Linux: DEFAULT_DEST_ROOT = "/run/media/way/Kirk H Fritschen CPA/Rerun_Source_Files"
# Windows: DEFAULT_DEST_ROOT = r"G:\Rerun_Source_Files"
#
# Can also be overridden per run with --dest-root.
DEFAULT_DEST_ROOT = "/run/media/way/Kirk H Fritschen CPA/Rerun_Source_Files"
# ---------------------------------------------------------------------------
HDR_RENAMED = "contain an invalid character"
HDR_ERRORS = "were skipped due to an error"
HDR_LASTFILE = "Last file uploaded was:"
RE_SUBDIR = re.compile(r"^All files under subdirectory:\s+(?P<path>.+?)\s{2,}\[", re.I)
RE_UPLOADSET = re.compile(r"^Upload Set#:", re.I)
# Path is greedy: file/folder names may themselves contain double spaces, so the
# split must happen at the LAST run of 2+ spaces. The message never contains a
# backslash or a double space.
RE_FILE_ERR = re.compile(r"^(?P<path>(?:[A-Za-z]:\\|\\\\).+)\s{2,}(?P<msg>[^\\]+?)\s*$")
RE_RENAMED = re.compile(r"^(?P<path>.+?)\s{2,}\[(?P<kind>file|dir\w*)\]\s*$", re.I)
@dataclass
class LogData:
last_file: str | None = None
renamed: list[str] = field(default_factory=list)
failed_files: list[tuple[str, str]] = field(default_factory=list) # (path, msg)
failed_dirs: list[str] = field(default_factory=list)
upload_set_errors: int = 0
unparsed: list[str] = field(default_factory=list)
def parse_log(log_path: str) -> LogData:
data = LogData()
section = None # None | "renamed" | "errors"
expect_last = False
with open(log_path, encoding="utf-8", errors="replace") as fh:
for raw in fh:
line = raw.rstrip("\r\n")
s = line.strip()
if not s:
continue
if expect_last:
data.last_file = s
expect_last = False
continue
if s.startswith(HDR_LASTFILE):
expect_last = True
continue
if HDR_RENAMED in s:
section = "renamed"
continue
if HDR_ERRORS in s:
section = "errors"
continue
if not line.startswith(" "): # other summary/header lines
continue
if section == "renamed":
m = RE_RENAMED.match(s)
if m:
data.renamed.append(m.group("path"))
else:
data.unparsed.append(s)
elif section == "errors":
m = RE_SUBDIR.match(s)
if m:
data.failed_dirs.append(m.group("path"))
continue
if RE_UPLOADSET.match(s):
data.upload_set_errors += 1
continue
m = RE_FILE_ERR.match(s)
if m:
data.failed_files.append((m.group("path"), m.group("msg")))
else:
data.unparsed.append(s)
return data
def detect_source_root(data: LogData) -> str | None:
paths = [p for p, _ in data.failed_files] + data.failed_dirs + data.renamed
if data.last_file:
paths.append(data.last_file)
if not paths:
return None
# Directory-level common prefix (path components, case-insensitive).
# Log paths are Windows paths -> always use ntpath, whatever the host OS.
split = [ntpath.normpath(p).split("\\") for p in paths]
common: list[str] = []
for parts in zip(*split):
if all(x.casefold() == parts[0].casefold() for x in parts):
common.append(parts[0])
else:
break
if not common:
return None
root = "\\".join(common)
if len(common) == 1: # "G:" -> "G:\\"
root += "\\"
return root
def log_path_to_local(path: str) -> str:
"""Translate a Windows path from the log (G:\\a\\b) to a local path using DRIVE_MAP."""
drive, rest = ntpath.splitdrive(ntpath.normpath(path))
base = DRIVE_MAP.get(drive.upper())
if base is None:
if os.name == "nt":
base = drive + "\\"
else:
raise SystemExit(f"ERROR: no DRIVE_MAP entry for drive '{drive}' (path: {path})")
parts = [x for x in rest.split("\\") if x]
return os.path.join(base, *parts) if parts else base
def _norm(p: str) -> str:
return os.path.normcase(os.path.normpath(p))
def _remap(path: str, log_root: str, real_root: str) -> str:
"""Translate a Windows path from the log (under log_root) to the real local root."""
rel = ntpath.relpath(ntpath.normpath(path), ntpath.normpath(log_root))
if rel == ".":
return real_root
return os.path.join(real_root, *rel.split("\\"))
def resolve_targets(data: LogData, log_root: str, source_root: str,
include_after_crash: bool, warn) -> dict[str, str]:
targets: dict[str, str] = {} # normcase path -> reason
display: dict[str, str] = {} # normcase path -> original-case path
def add(path: str, reason: str) -> None:
key = _norm(path)
if key not in targets:
targets[key] = reason
display[key] = os.path.normpath(path)
# 1. single-file errors
for p, _msg in data.failed_files:
rp = _remap(p, log_root, source_root)
if os.path.isfile(rp):
add(rp, "http-error")
else:
warn(f"missing file (http-error): {rp}")
# 2. folder errors, non-recursive
for d in dict.fromkeys(data.failed_dirs): # unique, keep order
rd = _remap(d, log_root, source_root)
if not os.path.isdir(rd):
warn(f"missing folder (folder-error): {rd}")
continue
try:
with os.scandir(rd) as it:
for e in it:
if e.is_file(follow_symlinks=False):
add(e.path, "folder-error")
except OSError as ex:
warn(f"cannot scan {rd}: {ex}")
# 3. everything from the crash point onward
if include_after_crash and data.last_file:
rel = ntpath.relpath(ntpath.normpath(data.last_file), ntpath.normpath(log_root))
crash_top = rel.split("\\")[0]
if crash_top in (".", ".."):
warn("crash file is not under the source root; skipping after-crash inclusion")
elif not os.path.isdir(source_root):
warn(f"source root not found, cannot enumerate after-crash folders: {source_root}")
else:
ct = crash_top.casefold()
tops = sorted(
(e for e in os.scandir(source_root) if e.is_dir(follow_symlinks=False)),
key=lambda e: e.name.casefold(),
)
chosen = [e for e in tops if e.name.casefold() >= ct]
for e in chosen:
for dirpath, _dirs, files in os.walk(e.path):
for f in files:
add(os.path.join(dirpath, f), "after-crash")
warn(f"after-crash: crash top-level folder = '{crash_top}', "
f"{len(chosen)} of {len(tops)} top-level folders included")
return {display[k]: v for k, v in targets.items()}
def _long(p: str) -> str:
if os.name == "nt" and len(p) > 240 and not p.startswith("\\\\?\\"):
p = os.path.abspath(p)
return "\\\\?\\UNC\\" + p[2:] if p.startswith("\\\\") else "\\\\?\\" + p
return p
def copy_targets(targets: dict[str, str], source_root: str, dest_root: str,
do_copy: bool, skip_existing: bool):
copied = skipped = 0
errors: list[tuple[str, str]] = []
total = len(targets)
for i, src in enumerate(targets, 1):
rel = os.path.relpath(src, source_root)
dst = os.path.join(dest_root, rel)
if not do_copy:
continue
try:
if skip_existing and os.path.isfile(dst) and \
os.path.getsize(dst) == os.path.getsize(src):
skipped += 1
continue
os.makedirs(_long(os.path.dirname(dst)), exist_ok=True)
shutil.copy2(_long(src), _long(dst))
copied += 1
except OSError as ex:
errors.append((src, str(ex)))
if i % 500 == 0 or i == total:
print(f" [{i}/{total}] copied={copied} skipped={skipped} errors={len(errors)}",
flush=True)
return copied, skipped, errors
def main(argv=None) -> int:
ap = argparse.ArgumentParser(description=__doc__,
formatter_class=argparse.RawDescriptionHelpFormatter)
ap.add_argument("log", help="SmartVault upload log file")
ap.add_argument("--source-root", help="LOCAL path of the root the tool was pointed at "
"(default: auto-detected from the log paths, translated via DRIVE_MAP)")
ap.add_argument("--dest-root", default=DEFAULT_DEST_ROOT,
help="where to mirror the files to (default: %(default)s)")
ap.add_argument("--copy", action="store_true",
help="actually copy files (default is dry run)")
ap.add_argument("--no-after-crash", action="store_true",
help="do NOT include folders from the crash point onward")
ap.add_argument("--skip-existing", action="store_true",
help="skip files already in dest with the same size")
ap.add_argument("--report", help="report path (default: rerun_report.txt next to the log)")
args = ap.parse_args(argv)
warnings: list[str] = []
def warn(msg: str) -> None:
warnings.append(msg)
print("WARN:", msg, file=sys.stderr)
data = parse_log(args.log)
log_root = detect_source_root(data)
if log_root is None:
print("ERROR: no paths found in log; cannot determine source root", file=sys.stderr)
return 2
source_root = os.path.normpath(args.source_root) if args.source_root \
else log_path_to_local(log_root)
dest_root = os.path.normpath(args.dest_root)
report_path = args.report or os.path.join(
os.path.dirname(os.path.abspath(args.log)), "rerun_report.txt")
print(f"Log : {args.log}")
print(f"Log root : {log_root}")
print(f"Source root : {source_root}")
print(f"Dest root : {dest_root}")
print(f"Crash file : {data.last_file}")
print(f"Parsed : {len(data.failed_files)} file errors, "
f"{len(data.failed_dirs)} folder errors ({len(set(map(_norm, data.failed_dirs)))} unique), "
f"{len(data.renamed)} renamed (uploaded OK), "
f"{data.upload_set_errors} upload-set lines ignored, "
f"{len(data.unparsed)} unparsed lines")
for u in data.unparsed[:10]:
print(" unparsed:", u)
targets = resolve_targets(data, log_root, source_root,
not args.no_after_crash, warn)
by_reason: dict[str, int] = {}
total_bytes = 0
for p, r in targets.items():
by_reason[r] = by_reason.get(r, 0) + 1
try:
total_bytes += os.path.getsize(p)
except OSError:
pass
print(f"\nFiles to re-upload: {len(targets)} ({total_bytes / 1e9:.3f} GB)")
for r, n in sorted(by_reason.items()):
print(f" {r:13s} {n}")
copied = skipped = 0
errors: list[tuple[str, str]] = []
if args.copy:
print(f"\nCopying to {dest_root} ...")
copied, skipped, errors = copy_targets(targets, source_root, dest_root,
True, args.skip_existing)
print(f"Done: copied={copied} skipped={skipped} errors={len(errors)}")
else:
print("\nDRY RUN - nothing copied. Re-run with --copy to copy.")
with open(report_path, "w", encoding="utf-8") as fh:
fh.write(f"log: {args.log}\nsource_root: {source_root}\ndest_root: {dest_root}\n")
fh.write(f"crash_file: {data.last_file}\nmode: {'COPY' if args.copy else 'DRY RUN'}\n")
fh.write(f"files_to_reupload: {len(targets)}\ntotal_bytes: {total_bytes}\n")
for r, n in sorted(by_reason.items()):
fh.write(f" {r}: {n}\n")
fh.write(f"renamed_uploaded_ok: {len(data.renamed)}\n")
fh.write(f"upload_set_lines_ignored: {data.upload_set_errors}\n")
if args.copy:
fh.write(f"copied: {copied}\nskipped: {skipped}\ncopy_errors: {len(errors)}\n")
if warnings:
fh.write("\n# warnings\n")
fh.writelines(f"{w}\n" for w in warnings)
if errors:
fh.write("\n# copy errors\n")
fh.writelines(f"{p}\t{e}\n" for p, e in errors)
fh.write("\n# targets (reason<TAB>path)\n")
for p, r in sorted(targets.items(), key=lambda kv: kv[0].casefold()):
fh.write(f"{r}\t{p}\n")
print(f"Report written : {report_path}")
return 1 if errors else 0
if __name__ == "__main__":
sys.exit(main())