병합 셀이 있는 양식에서 헤더와 데이터 행을 구분해 읽기

ZIP 크기: 140005바이트. 동봉한 MANIFEST.json에는 개별 파일의 크기와 SHA-256이 있습니다.

ZIP SHA-256: fce885792859cdc830b4e6787a8482dc5baeef11e63fb1e85a9ff5f749b35df6

"""P42: 명시한 헤더/데이터 경계에서 병합을 거부하고 새 JSON만 만든다."""
import argparse
import io
import json
import math
import sys
import xml.etree.ElementTree as ET
from pathlib import Path, PurePosixPath
from zipfile import BadZipFile, ZipFile
from zlib import error as ZlibError

from openpyxl import load_workbook
from openpyxl.utils import get_column_letter, range_boundaries
from openpyxl.worksheet.cell_range import CellRange

# 양식의 계약: 행을 추측하거나 빈 셀을 채우지 않는다.
SHEET = 'Data'
HEADER_ROW, FIRST_DATA_ROW, LAST_DATA_ROW = 2, 3, 5
FIRST_COL, LAST_COL = 1, 3
HEADERS = ('id', 'qty', 'memo')
LIMITS = {'input_bytes': 2 * 1024 * 1024, 'expanded_bytes': 10 * 1024 * 1024,
          'members': 100, 'sheets': 10, 'rows': 1000, 'columns': 50, 'cells': 50000}
NS = '{http://schemas.openxmlformats.org/spreadsheetml/2006/main}'


class Rejected(Exception):
    def __init__(self, errors):
        self.errors = errors
        super().__init__(errors[0]['reason'])


def refuse(code, reason, **detail):
    raise Rejected([{'code': code, 'reason': reason, **detail}])


def bounded(ref, part):
    try:
        left, top, right, bottom = range_boundaries(ref)
        valid = (1 <= left <= right <= LIMITS['columns'] and
                 1 <= top <= bottom <= LIMITS['rows'])
    except (TypeError, ValueError):
        valid = False
    if not valid:
        refuse('DIMENSION_LIMIT', '지원 범위를 벗어난 좌표입니다.', part=part, range=ref, limits=LIMITS)
    return (right - left + 1) * (bottom - top + 1)


def inspect_package(source):
    if source.suffix.lower() != '.xlsx':
        refuse('UNSUPPORTED_EXTENSION', '일반 .xlsx 파일만 지원합니다.')
    try:
        if source.stat().st_size > LIMITS['input_bytes']:
            refuse('INPUT_SIZE_LIMIT', '입력은 2 MiB 이하여야 합니다.', limits=LIMITS)
        content = source.read_bytes()
        if len(content) > LIMITS['input_bytes']:
            refuse('INPUT_SIZE_LIMIT', '입력은 2 MiB 이하여야 합니다.', limits=LIMITS)
    except OSError as exc:
        refuse('INPUT_UNAVAILABLE', '입력 파일을 읽을 수 없습니다.', exception=type(exc).__name__)
    if content.startswith(bytes.fromhex('D0CF11E0A1B11AE1')):
        refuse('ENCRYPTED_OR_OLE_UNSUPPORTED', '암호화 Office 또는 구형 OLE 파일은 지원하지 않습니다.')
    try:
        with ZipFile(io.BytesIO(content)) as package:
            members = package.infolist()
            if len(members) > LIMITS['members']:
                refuse('ARCHIVE_MEMBER_LIMIT', 'ZIP 항목은 100개 이하여야 합니다.', limits=LIMITS)
            if sum(item.file_size for item in members) > LIMITS['expanded_bytes']:
                refuse('ARCHIVE_EXPANSION_LIMIT', 'ZIP 해제 크기는 10 MiB 이하여야 합니다.', limits=LIMITS)
            names = [item.filename for item in members]
            if len(set(names)) != len(names) or any(PurePosixPath(n).is_absolute() or '..' in PurePosixPath(n).parts for n in names):
                refuse('INVALID_PACKAGE', '중복 또는 안전하지 않은 ZIP 경로입니다.')
            if any(item.flag_bits & 1 for item in members):
                refuse('ENCRYPTED_OR_OLE_UNSUPPORTED', '암호화 ZIP은 지원하지 않습니다.')
            lowered = [n.lower() for n in names]
            if any('vbaproject' in n or 'vbadata' in n or '/macrosheets/' in n or '/dialogsheets/' in n for n in lowered):
                refuse('MACROS_UNSUPPORTED', '확장자 변경 여부와 관계없이 매크로 패키지를 거부합니다.')
            if any('/externallinks/' in n for n in lowered):
                refuse('EXTERNAL_LINK_UNSUPPORTED', '외부 연결이 포함된 파일은 지원하지 않습니다.')
            if '[Content_Types].xml' not in names or 'xl/workbook.xml' not in names:
                refuse('INVALID_PACKAGE', '필수 XLSX 패키지 항목이 없습니다.')
            sheets, cell_area = 0, 0
            for name in names:
                raw = package.read(name)  # CRC도 검증한다.
                if not name.endswith(('.xml', '.rels')):
                    continue
                try:
                    raw.decode('utf-8-sig')
                    if b'\x00' in raw:
                        raise UnicodeError('NUL byte')
                except UnicodeError:
                    refuse('XML_ENCODING_UNSUPPORTED', 'XML은 UTF-8만 지원합니다.', part=name)
                if b'<!DOCTYPE' in raw.upper() or b'<!ENTITY' in raw.upper():
                    refuse('XML_DECLARATION_UNSUPPORTED', 'DTD/ENTITY 선언은 지원하지 않습니다.', part=name)
                try:
                    root = ET.fromstring(raw)
                except ET.ParseError:
                    refuse('INVALID_XML', '유효하지 않은 XML입니다.', part=name)
                if name == '[Content_Types].xml':
                    types = [node.get('ContentType', '').lower() for node in root]
                    if any('macroenabled' in t or 'vba' in t or 'macrosheet' in t for t in types):
                        refuse('MACROS_UNSUPPORTED', '매크로 콘텐츠 유형을 거부합니다.')
                    if 'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet.main+xml' not in types:
                        refuse('INVALID_PACKAGE', '일반 XLSX 통합 문서 유형이 아닙니다.')
                if name.endswith('.rels') and any(node.get('TargetMode', '').lower() == 'external' for node in root):
                    refuse('EXTERNAL_LINK_UNSUPPORTED', '외부 관계/하이퍼링크를 거부합니다.', part=name)
                if name.startswith('xl/worksheets/') and name.endswith('.xml'):
                    sheets += 1
                    dimension = root.find(NS + 'dimension')
                    if dimension is None:
                        refuse('DIMENSION_LIMIT', '명시된 시트 dimension이 필요합니다.', part=name)
                    cell_area += bounded(dimension.get('ref'), name)
                    for node in root.iter():
                        if node.tag in (NS + 'c', NS + 'mergeCell'):
                            bounded(node.get('r') if node.tag == NS + 'c' else node.get('ref'), name)
                        if node.tag == NS + 'row' and not 1 <= int(node.get('r', '0')) <= LIMITS['rows']:
                            refuse('DIMENSION_LIMIT', '지원 행 수를 초과했습니다.', part=name)
                        if node.tag == NS + 'col' and not 1 <= int(node.get('min', '0')) <= int(node.get('max', '0')) <= LIMITS['columns']:
                            refuse('DIMENSION_LIMIT', '지원 열 수를 초과했습니다.', part=name)
                        if node.tag == NS + 'f':
                            refuse('FORMULA_UNSUPPORTED', '수식 셀은 계산하거나 캐시값으로 대체하지 않습니다.', part=name)
            if sheets > LIMITS['sheets'] or cell_area > LIMITS['cells']:
                refuse('DIMENSION_LIMIT', '시트 수/전체 dimension 면적 제한을 초과했습니다.', limits=LIMITS)
    except (BadZipFile, ZlibError, ValueError, KeyError, RuntimeError, NotImplementedError) as exc:
        refuse('INVALID_PACKAGE', '유효한 일반 XLSX ZIP 패키지가 아닙니다.', exception=type(exc).__name__)
    return content


def extract(source):
    content = inspect_package(source)
    try:
        wb = load_workbook(io.BytesIO(content), read_only=False, data_only=False, keep_vba=False, keep_links=False)
    except Exception as exc:
        refuse('INVALID_WORKBOOK', 'openpyxl이 통합 문서를 읽지 못했습니다.', exception=type(exc).__name__)
    try:
        if SHEET not in wb.sheetnames:
            refuse('SHEET_MISSING', 'Data 시트가 없습니다.', expected=SHEET, actual=wb.sheetnames)
        ws = wb[SHEET]
        regions = {'header': CellRange(min_col=FIRST_COL, max_col=LAST_COL, min_row=HEADER_ROW, max_row=HEADER_ROW),
                   'data': CellRange(min_col=FIRST_COL, max_col=LAST_COL, min_row=FIRST_DATA_ROW, max_row=LAST_DATA_ROW)}
        errors = []
        for merged in sorted(ws.merged_cells.ranges, key=lambda r: (r.min_row, r.min_col)):
            hits = [name for name, region in regions.items() if not merged.isdisjoint(region)]
            if hits:
                errors.append({'code': 'MERGE_IN_BOUNDARY', 'range': str(merged),
                               'anchor': f'{get_column_letter(merged.min_col)}{merged.min_row}',
                               'intersects': hits, 'reason': '헤더/데이터 경계와 겹치는 병합은 값을 복제하지 않고 거부합니다.'})
        if errors:
            raise Rejected(errors)
        actual = tuple(ws.cell(HEADER_ROW, col).value for col in range(FIRST_COL, LAST_COL + 1))
        if actual != HEADERS:
            refuse('HEADER_MISMATCH', '헤더가 양식 계약과 다릅니다.', expected=list(HEADERS), actual_repr=[repr(value) for value in actual])
        rows = []
        for row in ws.iter_rows(min_row=FIRST_DATA_ROW, max_row=LAST_DATA_ROW, min_col=FIRST_COL, max_col=LAST_COL, values_only=True):
            identifier, quantity, memo = row
            if (not isinstance(identifier, str) or not identifier or type(quantity) not in (int, float) or
                    (type(quantity) is float and not math.isfinite(quantity)) or quantity < 0 or (memo is not None and not isinstance(memo, str))):
                refuse('VALUE_TYPE_UNSUPPORTED', 'id는 비어 있지 않은 문자열, qty는 유한한 0 이상 숫자, memo는 문자열/빈 셀이어야 합니다.', row=FIRST_DATA_ROW + len(rows))
            rows.append(dict(zip(HEADERS, row)))  # 0과 None을 그대로 보존한다.
        try:
            total = sum(row['qty'] for row in rows)
        except OverflowError:
            refuse('VALUE_TYPE_UNSUPPORTED', 'qty 합계 계산에서 숫자 범위를 벗어났습니다.')
        if type(total) is float and not math.isfinite(total):
            refuse('VALUE_TYPE_UNSUPPORTED', 'qty 합계가 유한한 숫자 범위를 벗어났습니다.')
        return {'schema_version': 1, 'status': 'ok', 'sheet': SHEET,
                'boundary': {'header_row': HEADER_ROW, 'first_data_row': FIRST_DATA_ROW, 'last_data_row': LAST_DATA_ROW,
                             'first_col': get_column_letter(FIRST_COL), 'last_col': get_column_letter(LAST_COL)},
                'headers': list(HEADERS), 'rows': rows, 'row_count': len(rows), 'column_count': len(HEADERS),
                'quantity_total': total}
    finally:
        wb.close()  # save()는 호출하지 않는다.


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('source', type=Path)
    parser.add_argument('--output', required=True, type=Path)
    parser.add_argument('--diagnostics', required=True, type=Path)
    args = parser.parse_args()
    output, diagnostic = args.output, args.diagnostics
    errors = []
    resolved = [p.resolve() for p in (args.source, output, diagnostic)]
    if len(set(resolved)) != 3 or any(p.exists() or p.is_symlink() for p in (output, diagnostic)):
        errors = [{'code': 'OUTPUT_CONFLICT', 'reason': '입력/출력 경로가 겹치거나 결과 파일이 이미 있습니다. 새 파일명을 지정하세요.'}]
    elif not all(p.parent.is_dir() for p in (output, diagnostic)):
        errors = [{'code': 'OUTPUT_UNAVAILABLE', 'reason': '출력 폴더가 없습니다. 기존 폴더와 새 파일명을 지정하세요.'}]
    if errors:
        print(json.dumps({'status': 'blocked', 'errors': errors}, ensure_ascii=False))
        return 3
    try:
        payload, destination, code = extract(args.source), output, 0
    except Rejected as exc:
        payload = {'schema_version': 1, 'status': 'rejected', 'extracted_rows': 0, 'errors': exc.errors}
        destination, code = diagnostic, 2
    try:
        encoded = json.dumps(payload, ensure_ascii=False, indent=2, allow_nan=False) + '\n'
        with destination.open('x', encoding='utf-8') as handle:  # 기존 파일을 절대 덮어쓰지 않는다.
            handle.write(encoded)
    except (OSError, TypeError, ValueError) as exc:
        print(json.dumps({'status': 'blocked', 'errors': [{'code': 'OUTPUT_UNAVAILABLE', 'reason': '새 결과 파일을 쓰지 못했습니다.', 'exception': type(exc).__name__}]}, ensure_ascii=False))
        return 3
    print(json.dumps({'status': payload['status'], 'file': str(destination)}, ensure_ascii=False))
    return code


if __name__ == '__main__':
    sys.exit(main())