Summarize expense categories

import argparse, ast, collections, difflib, fnmatch, gzip, hashlib, io, json, math, os, random, re, shutil, statistics, string, sys, tarfile, textwrap, time, unicodedata, zipfile
from pathlib import Path
import xml.etree.ElementTree as ET

def write_text(path, content):
    path=Path(path)
    path.parent.mkdir(parents=True,exist_ok=True)
    with path.open('x',encoding='utf-8',newline='') as h: h.write(content)
    print('Created',path)

def fresh_dir(path):
    path=Path(path); path.mkdir(parents=True,exist_ok=False); return path

def files_under(root):
    root=Path(root)
    if not root.is_dir(): raise ValueError('Input must be a directory')
    return sorted((p for p in root.rglob('*') if p.is_file() and not p.is_symlink()),key=lambda p:p.as_posix().casefold())

def digest(path):
    h=hashlib.sha256()
    with Path(path).open('rb') as f:
        for b in iter(lambda:f.read(1048576),b''): h.update(b)
    return h.hexdigest()

def positive(n,name):
    if n<=0: raise ValueError(name+' must be positive')

def destination(root,name):
    name=name.replace('\\','/')
    if name.startswith('/') or any(p=='..' or ':' in p for p in name.split('/')): raise ValueError('Unsafe path: '+name)
    p=(root/name).resolve()
    if p!=root.resolve() and root.resolve() not in p.parents: raise ValueError('Path escapes destination')
    return p

def zip_members(z,limit=500000000):
    seen=set(); total=0; result=[]
    for i in z.infolist():
        if i.is_dir(): continue
        destination(Path.cwd(),i.filename)
        key=i.filename.replace('\\','/').casefold()
        if key in seen: raise ValueError('Duplicate/case-colliding member: '+i.filename)
        seen.add(key)
        if ((i.external_attr>>16)&0o170000)==0o120000: raise ValueError('Symlinks unsupported')
        total+=i.file_size
        if total>limit: raise ValueError('Expanded data exceeds limit')
        result.append(i)
    return result

def copy_member(z,i,root):
    p=destination(root,i.filename); p.parent.mkdir(parents=True,exist_ok=True)
    with z.open(i) as src,p.open('xb') as dst: shutil.copyfileobj(src,dst)

def read_input(a): return a.input.read_text(encoding='utf-8-sig')
NS={'w':'http://schemas.openxmlformats.org/wordprocessingml/2006/main','r':'http://schemas.openxmlformats.org/officeDocument/2006/relationships'}
def docx_root(path,member='word/document.xml'):
    with zipfile.ZipFile(path) as z:
        i=z.getinfo(member)
        if i.file_size>30000000: raise ValueError('Document XML exceeds 30 MB')
        return ET.fromstring(z.read(i))
def pdf_reader(path):
    try: from pypdf import PdfReader
    except ImportError as e: raise ValueError('Install dependency: python -m pip install pypdf') from e
    r=PdfReader(path)
    if r.is_encrypted: raise ValueError('Encrypted PDF unsupported')
    return r
def pdf_writer():
    try: from pypdf import PdfWriter
    except ImportError as e: raise ValueError('Install dependency: python -m pip install pypdf') from e
    return PdfWriter()
def save_pdf(w,p):
    Path(p).parent.mkdir(parents=True,exist_ok=True)
    with Path(p).open('xb') as h: w.write(h)
    print('Created',p)

def md_headings(content):
    fence=None
    for number,line in enumerate(content.splitlines(),1):
        mark=re.match(r'^\s{0,3}(`{3,}|~{3,})',line)
        if mark:
            char=mark.group(1)[0]
            if fence is None: fence=char
            elif fence==char: fence=None
            continue
        if fence is None:
            m=re.match(r'^(#{1,6})\s+(.+?)\s*#*\s*$',line)
            if m: yield number,len(m.group(1)),m.group(2)

import calendar, csv
from datetime import date,datetime,timedelta,timezone
from zoneinfo import ZoneInfo

def csv_rows(path,required=()):
    with Path(path).open(encoding='utf-8-sig',newline='') as f:
        reader=csv.DictReader(f)
        if not reader.fieldnames or any(k not in reader.fieldnames for k in required): raise ValueError('Required columns: '+', '.join(required))
        return list(reader)

def csv_text(rows,columns):
    out=io.StringIO(newline='');writer=csv.DictWriter(out,fieldnames=columns);writer.writeheader();writer.writerows(rows);return out.getvalue()

def aware(text):
    value=datetime.fromisoformat(text.replace('Z','+00:00'))
    if value.tzinfo is None: raise ValueError('Timestamp must include UTC offset or Z: '+text)
    return value.astimezone(timezone.utc)

def date_range(start,end):
    if end<start: raise ValueError('End precedes start')
    if (end-start).days>366000: raise ValueError('Date range exceeds 1000 years')
    for i in range((end-start).days+1): yield start+timedelta(days=i)

def ical_events(path):
    lines=re.sub(r'\r?\n[ \t]','',Path(path).read_text(encoding='utf-8-sig')).splitlines();events=[];current=None
    for line in lines:
        if line=='BEGIN:VEVENT':
            if current is not None: raise ValueError('Nested VEVENT')
            current=[]
        elif line=='END:VEVENT':
            if current is None: raise ValueError('END without VEVENT')
            events.append(current);current=None
        elif current is not None: current.append(line)
    if current is not None: raise ValueError('Unclosed VEVENT')
    return events

def ical_value(lines,key):
    return next((line.split(':',1)[1] for line in lines if line.startswith(key+':') or line.startswith(key+';')), '')

def ical_escape(text): return text.replace('\\','\\\\').replace('\n','\\n').replace(';','\\;').replace(',','\\,')
def ical_fold(line):
    parts=[];current=''
    for char in line:
        if len((current+char).encode('utf-8'))>75: parts.append(current);current=' '+char
        else: current+=char
    parts.append(current);return '\r\n'.join(parts)

from decimal import Decimal,InvalidOperation,ROUND_HALF_UP

def dec(value):
    try: result=Decimal(str(value))
    except InvalidOperation as e: raise ValueError('Invalid decimal: '+str(value)) from e
    if not result.is_finite(): raise ValueError('Non-finite decimal')
    return result

def money(value): return format(value.quantize(Decimal('.01'),rounding=ROUND_HALF_UP),'.2f')

def main():
    parser=argparse.ArgumentParser(description='Count expense records and sum amounts for each exact category.')
    parser.add_argument('--input',type=Path,required=True,help='Source file or folder')
    parser.add_argument('--output',type=Path,required=True,help='New result path; must not exist')
    args=parser.parse_args()
    totals=collections.defaultdict(Decimal);counts=collections.Counter()
    for row in csv_rows(args.input,['category','amount']): totals[row['category']]+=dec(row['amount']);counts[row['category']]+=1
    write_text(args.output,'category\tentries\ttotal\n'+'\n'.join(f'{k}\t{counts[k]}\t{money(v)}' for k,v in sorted(totals.items())))

if __name__=='__main__':
    try: main()
    except (OSError,ValueError,KeyError,zipfile.BadZipFile,tarfile.TarError,ET.ParseError) as error:
        print('Error:',error,file=sys.stderr); sys.exit(2)