CSV 전체 구문과 행별 오류를 나눠 검토하는 모듈

"""Independent synthetic CSV review. No file I/O, callbacks, storage or network."""
import re
from decimal import Decimal

class CsvSyntaxError(ValueError):
    pass

def parse_records(text):
    if type(text) is not str:
        raise TypeError('TEXT_REQUIRED')
    if len(text)>12000:
        raise ValueError('TEXT_LIMIT')
    text=text.removeprefix('\ufeff')
    rows=[];row=[];cell=[];mode='start';index=0
    def end_cell():
        nonlocal cell,mode
        row.append(''.join(cell));cell=[];mode='start'
    def end_row():
        nonlocal row
        end_cell()
        if any(part.strip() for part in row):rows.append(row)
        row=[]
    while index<len(text):
        char=text[index]
        if mode=='quoted':
            if char=='"':
                if text[index+1:index+2]=='"':cell.append('"');index+=1
                else:mode='closed'
            else:cell.append(char)
        elif char==',':end_cell()
        elif char in '\r\n':
            if char=='\r' and text[index+1:index+2]=='\n':index+=1
            end_row()
        elif mode=='closed':
            if char not in ' \t':raise CsvSyntaxError('AFTER_QUOTE')
        elif char=='"':
            if mode!='start':raise CsvSyntaxError('QUOTE_IN_PLAIN_CELL')
            mode='quoted'
        else:cell.append(char);mode='plain'
        index+=1
    if mode=='quoted':raise CsvSyntaxError('UNCLOSED_QUOTE')
    end_row()
    return rows

def amount_minor(text):
    if type(text) is not str:raise ValueError('AMOUNT_TEXT')
    text=text.strip()
    if not re.fullmatch(r'(?:[0-9]+|[0-9]{1,3}(?:,[0-9]{3})+)(?:\.[0-9]{1,2})?',text):
        raise ValueError('AMOUNT_FORMAT')
    value=Decimal(text.replace(',',''))
    if not 0<value<=Decimal('999999.99'):raise ValueError('AMOUNT_RANGE')
    return int(value*100)

def review_csv(text):
    """The input is a small, explicitly two-decimal synthetic contract.

    record numbers are logical nonblank CSV records, never physical file lines.
    Even clean rows are only candidates. There is no automatic commit policy.
    """
    base={'file_ok':False,'all_rows_valid':False,'candidates':[],
          'issues':[],'imported_count':0,'review_only':True,'performs_actions':False}
    try:rows=parse_records(text)
    except CsvSyntaxError as exc:return {**base,'phase':'syntax','issues':[{'record':None,'code':str(exc)}]}
    except ValueError as exc:return {**base,'phase':'limit','issues':[{'record':None,'code':str(exc)}]}
    if not rows or [part.strip() for part in rows[0]]!=['label','amount','note']:
        return {**base,'phase':'header','issues':[{'record':1,'code':'HEADER'}]}
    if len(rows)>101:return {**base,'phase':'limit','issues':[{'record':None,'code':'RECORD_LIMIT'}]}
    candidates=[];issues=[]
    for number,row in enumerate(rows[1:],2):
        if len(row)!=3:
            issues.append({'record':number,'code':'WIDTH'});continue
        label,raw,note=[part.strip() for part in row]
        codes=[]
        if not label or len(label)>80:codes.append('LABEL')
        if len(note)>200:codes.append('NOTE')
        try:value=amount_minor(raw)
        except ValueError:codes.append('AMOUNT')
        if codes:issues.extend({'record':number,'code':code} for code in codes)
        else:candidates.append({'label':label,'amount_minor':value,'note':note})
    if len(rows)==1:issues.append({'record':None,'code':'EMPTY_BATCH'})
    return {**base,'file_ok':True,'phase':'rows','candidates':candidates,
            'issues':issues,'all_rows_valid':bool(candidates) and not issues}