143 lines
4.7 KiB
Python
143 lines
4.7 KiB
Python
"""
|
|
Step 3: Reorganize flat exported files into a proper folder structure.
|
|
|
|
Parses FCCS export filenames using known folder names and date anchors
|
|
to reconstruct the original drawer/folder hierarchy.
|
|
|
|
Filename format:
|
|
{drawer_id}_{client_name}_{folder_name}_{MM-DD-YYYY}_{document_name}.ext
|
|
Example: 01069_SMITH, BOB_2007 Tax Documents_02-10-2009_2007 Form 1040A Page 1.pdf
|
|
|
|
Output structure:
|
|
output/{drawer_id}_{client_name}/{folder_name}/{original_filename}
|
|
|
|
Files that cannot be parsed go to output/_unparsed/ for manual review.
|
|
"""
|
|
|
|
import os
|
|
import re
|
|
import shutil
|
|
import sys
|
|
|
|
from fccs_config import parse_args, load_config, make_logger, load_folder_list
|
|
|
|
|
|
def parse_filename(filename, known_folders):
|
|
"""
|
|
Parse an FCCS export filename into (drawer_id, client_name, folder_name,
|
|
date, doc_name_with_ext) or return None if it cannot be parsed.
|
|
|
|
Strategy: anchor on drawer_id (left), date MM-DD-YYYY (middle), and match
|
|
a known folder name between client_name and date.
|
|
"""
|
|
stem, ext = os.path.splitext(filename)
|
|
|
|
# Drawer ID is always the first underscore-delimited token
|
|
sep = stem.find("_")
|
|
if sep == -1:
|
|
return None
|
|
drawer_id = stem[:sep]
|
|
rest = stem[sep + 1:]
|
|
|
|
# Find all date-pattern occurrences (MM-DD-YYYY) in rest
|
|
# We search with a leading underscore context to ensure proper boundaries
|
|
date_pattern = re.compile(r"_(\d{2}-\d{2}-\d{4})_")
|
|
# Prepend underscore so the first potential date at position 0 of rest is caught
|
|
search_str = "_" + rest
|
|
matches = list(date_pattern.finditer(search_str))
|
|
|
|
if not matches:
|
|
return None
|
|
|
|
# Try each date match; known_folders is already sorted longest-first
|
|
for m in matches:
|
|
date_str = m.group(1)
|
|
# Position in `rest` where date starts (adjust for prepended _)
|
|
date_start_in_rest = m.start() - 1 # -1 for the prepended _
|
|
# But the match includes the leading _, so the actual content before date:
|
|
before_date = rest[:date_start_in_rest]
|
|
after_date = rest[date_start_in_rest + len(date_str) + 1:] # +1 for trailing _
|
|
|
|
# Try to match a known folder at the end of before_date
|
|
for folder in known_folders:
|
|
if before_date.endswith(folder):
|
|
# Check there's an underscore separator before the folder name
|
|
prefix_end = len(before_date) - len(folder)
|
|
if prefix_end > 0 and before_date[prefix_end - 1] == "_":
|
|
client_name = before_date[: prefix_end - 1]
|
|
doc_name = after_date + ext
|
|
return (drawer_id, client_name, folder, date_str, doc_name)
|
|
|
|
return None
|
|
|
|
|
|
def main():
|
|
args = parse_args()
|
|
cfg = load_config(args.config)
|
|
log = make_logger(cfg.get("paths", "log_file"))
|
|
|
|
export_dir = cfg.get("paths", "export_dir")
|
|
output_dir = cfg.get("paths", "output_dir")
|
|
folder_list_path = cfg.get("paths", "folder_list")
|
|
|
|
if not os.path.isdir(export_dir):
|
|
log(f"ERROR: export directory not found: {export_dir}")
|
|
sys.exit(1)
|
|
|
|
known_folders = load_folder_list(folder_list_path)
|
|
if not known_folders:
|
|
log("WARNING: folder list is empty. All files will go to _unparsed/.")
|
|
# Sort longest-first to prevent partial matches
|
|
known_folders.sort(key=len, reverse=True)
|
|
|
|
files = [
|
|
f for f in os.listdir(export_dir)
|
|
if os.path.isfile(os.path.join(export_dir, f))
|
|
]
|
|
|
|
log(f"Found {len(files)} files to reorganize")
|
|
|
|
success = 0
|
|
failed = 0
|
|
|
|
for filename in files:
|
|
result = parse_filename(filename, known_folders)
|
|
|
|
if result is None:
|
|
log(f" UNPARSED: {filename}")
|
|
dest_dir = os.path.join(output_dir, "_unparsed")
|
|
failed += 1
|
|
else:
|
|
drawer_id, client_name, folder_name, date, doc_name = result
|
|
dest_dir = os.path.join(
|
|
output_dir,
|
|
f"{drawer_id}_{client_name}",
|
|
folder_name,
|
|
)
|
|
success += 1
|
|
|
|
os.makedirs(dest_dir, exist_ok=True)
|
|
src = os.path.join(export_dir, filename)
|
|
dst = os.path.join(dest_dir, filename)
|
|
|
|
# Handle duplicate filenames
|
|
if os.path.exists(dst):
|
|
base, fext = os.path.splitext(filename)
|
|
counter = 1
|
|
while os.path.exists(dst):
|
|
dst = os.path.join(dest_dir, f"{base}_{counter}{fext}")
|
|
counter += 1
|
|
log(f" DUPLICATE renamed: {filename} -> {os.path.basename(dst)}")
|
|
|
|
shutil.copy2(src, dst)
|
|
|
|
log("=" * 60)
|
|
log(f"Reorganization complete: {success} organized, {failed} unparsed")
|
|
if failed:
|
|
log(f"Review unparsed files in: {os.path.join(output_dir, '_unparsed')}")
|
|
log("=" * 60)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|