품질 기준과 공개 차단

적재 전에 멈추는 경로를 만듭니다

보고서가 배포된 뒤 오류를 찾으면 이미 수치를 본 사람에게 정정을 알려야 합니다. 저장 전에 실패 배치를 멈추면 잘못된 자료가 정상 결과로 바뀌는 경로를 줄일 수 있습니다. 이번 로컬 실습은 m08의 검증된 solution 전체를 포함하고 새 quality.run 경로를 추가합니다. 기존 incremental.run은 보존합니다. 공개용 실행자는 새 명령을 사용하며 검사 실패가 호출자에게 비정상 종료로 전달되는지 확인합니다.

게이트의 역할은 검사와 다음 작업 사이의 통과 조건을 연결하는 것입니다. 오류 목록이 비어 있을 때만 정제, 저장, 공개 순서로 진행합니다. 오류를 출력하고도 merge를 계속 호출하면 로그에는 실패가 있고 DB에는 잘못된 값이 남는 모순이 생깁니다. 예외를 올리는 위치는 저장 호출보다 앞이어야 합니다. 함수를 작은 단계로 나누면 테스트에서 실패 시점과 이후에 실행되지 않아야 할 작업을 확인하기 쉽습니다.

검사 위치를 원본과 연결합니다

새 실행 경로는 collect_file로 입력 바이트를 보존하고 그 보존본을 candidate로 읽습니다. CSV 열 이름과 열 수를 확인한 뒤 정규화한 행 목록을 inspect에 보냅니다. 기존 transform은 같은 보존본을 정제하고 merge는 지역·날짜 키로 병합합니다. 검사 후 변경될 수 있는 joined.csv를 다시 읽지 않는 점이 중요합니다. 품질 판정의 해시와 실제 처리한 바이트가 연결되어야 재현 가능한 증거가 됩니다.

원본 음수 문자열은 기존 normalize에서 INTEGER 오류로 거절됩니다. 이번 운영 규칙의 NEGATIVE_TRAFFIC 코드를 남기기 위해 candidate는 음수 정수 표현을 먼저 확인합니다. 중복 검사는 transform이 동일 행을 제거하기 전 목록에서 수행합니다. 음수와 중복을 정제 후에만 확인하면 기대한 업무 오류가 형식 오류로 바뀌거나 사라질 수 있습니다. 검사 순서를 바꾸면 코드뿐 아니라 오류 계약도 바뀐다는 점을 테스트로 고정합니다.

예시 임계값의 근거와 범위를 적습니다

이 실습의 한계는 incoming batch 전체에 적용됩니다. 통행량 결측률은 0.2 초과, 강수 결측률은 0.4 초과일 때 차단합니다. 기존 학습 표본을 유지하면서 새 결측 증가를 감지하는 연습 기준이며 기관 운영 기준이 아닙니다. 누적 DB 전체의 결측률까지 이 규칙이 검사하는 것은 아닙니다. 신규 배치와 저장 전체를 같은 분모로 부르지 않도록 policy에 scope를 적고 실제 운영 전에는 누적 품질 정책도 검토합니다.

조건을 변경할 때는 통과시키고 싶은 특정 파일만 보지 않습니다. 비교 질문, 필요한 유효 쌍의 수, 누락의 지역별 집중, 공급자의 수집 상태를 함께 검토합니다. 한계값 변경은 이전 실패를 숨기는 임시 수단이 되어서는 안 됩니다. 기준과 승인 근거를 문서에 기록하고 과거 표본을 다시 평가합니다. 같은 코드라도 정책이 달라지면 동일 입력의 판정이 달라질 수 있으므로 정책을 계보 자료에도 포함합니다.

DB와 공개 파일의 경계를 구분합니다

merge는 한 배치를 SQLite 트랜잭션으로 저장합니다. 첫 쓰기 뒤 예외를 주입하면 변경을 롤백하고 이전 DB를 유지합니다. 정상 커밋 후에는 snapshot의 전체 저장 행을 사용해 보고서를 계산합니다. 새 배치의 합계를 보고서 전체 합계처럼 쓰지 않습니다. 정정 한 행이 들어와도 이전 날짜·지역은 DB에 남기 때문에 입력 행 수와 보고서 후보 행 수가 다를 수 있습니다. 검사 결과와 저장 결과의 단위를 각각 기록합니다.

공개 보고서는 두 측정값이 모두 있는 지역·날짜 행만 포함합니다. 초기 표본 여섯 행의 통행량 전체 합계는 390이지만 강수와 함께 비교할 네 행의 합계는 300입니다. 이는 오류를 숨긴 보정이 아니라 명시한 제외 조건의 결과입니다. report.json에 행 수, 날짜 목록, 단위와 합계를 적고 성공 시도의 lineage를 함께 묶습니다. 독자가 보고서 수치를 원본 전체 수치와 대조할 때 필터 차이를 확인할 수 있습니다.

파일은 report.json.tmp에 완전히 쓴 뒤 report.json으로 교체합니다. 쓰는 도중 기존 보고서를 덮어 반쪽 JSON을 공개하는 위험을 줄이는 방식입니다. 새 자료가 품질 검사에 실패하거나 DB 적재가 롤백되면 공개 파일 교체까지 진행하지 않습니다. 실패 뒤에도 파일 바이트의 SHA-256이 이전 정상 보고서와 같은지 확인합니다. 내용 일부만 비교하면 합계는 같지만 계보나 날짜가 바뀐 오류를 놓칠 수 있습니다.

파일 교체와 DB 커밋이 한 트랜잭션이라는 뜻은 아닙니다. PUBLISH 오류가 커밋 후 발생하면 DB는 새 값이고 보고서는 이전 값일 수 있습니다. 이때 db_committed가 true인지 확인하고 같은 승인 입력을 재처리하여 보고서를 회복합니다. 증분 UPSERT가 절대값을 갱신하므로 재처리에서 수치가 더해지지 않습니다. 디스크 자체가 감사 파일 쓰기도 막는 경우에는 기록이 없을 수 있어 DB와 보존본을 별도로 확인해야 합니다.

실패 자료와 마지막 성공 자료를 분리합니다

lineage.json은 마지막 시도의 상태를 담고 quality-audit.json은 시도 배열을 보관합니다. 마지막 시도가 실패했다고 해서 이전 report.json까지 실패한 보고서라고 부르지 않습니다. 성공 보고서 안에 들어 있는 lineage가 그 보고서의 생성 근거입니다. 최신 실패 lineage는 새 배치가 왜 공개되지 않았는지 설명하는 자료입니다. 상태를 파일 역할과 연결하면 최신 실행 시각만 보고 기존 결과를 새 결과로 오해하는 일을 줄일 수 있습니다.

실패 원본은 collected의 내용 해시 폴더에 유지합니다. 그 원본을 공개 폴더로 옮기지는 않습니다. 오류 코드와 입력 해시를 찾아 정제 규칙 또는 공급자의 정정본을 확인한 뒤 승인된 자료를 다시 실행합니다. 자동 재시도를 반복한다고 결측률이 낮아지지는 않습니다. 입력이 바뀌지 않은 품질 오류는 담당자의 판단이 필요한 사건이며, 일시적인 파일 저장 오류와 구분해 복구 절차를 정합니다.

테스트 실패를 구현 위치로 연결합니다

starter의 두 TODO는 중복 키와 통행량 결측률 검사입니다. test_duplicate가 ValueError not raised로 실패하면 예상한 차단 예외가 나오지 않은 것입니다. test_missing도 같은 방식으로 결측률 경로를 확인합니다. test_failure_preserves_and_recovers는 개별 판정뿐 아니라 마지막 성공 파일 유지까지 봅니다. 테스트 파일이나 기대 합계를 고쳐 통과시키지 말고 quality/checks.py에서 분모와 비교 연산 및 중복 판정을 고칩니다.

정답에서는 새 품질 검사 11개와 기존 검사들을 함께 실행합니다. 전체 108개 중 기존 선택 HTTP 통합 9개는 기본 실행에서 건너뜁니다. 이는 품질 실습에 네트워크가 필요하다는 뜻이 아닙니다. 솔루션은 나머지를 통과하고 starter는 새 검사 세 건이 실패해야 합니다. 장애 주입, 정확한 임계값, 빈 입력, 공개 실패 후 재실행까지 포함하여 정상 입력 한 번만 성공하는 구현과 반복 운영 가능한 구현을 구분합니다.

작업자는 report.json의 300이라는 숫자만 보고 완료하지 않습니다. 원본 바이트가 유지되는지, 실패 감사에 코드가 있는지, 저장 행 수가 여섯인지, 공개 비교 행이 네 개인지 확인합니다. 승인되지 않은 오래된 정정본을 재생하면 값이 과거로 돌아갈 수 있고 동시 실행은 감사 JSON 경쟁을 만들 수 있습니다. 현재 실습은 단일 실행과 최신 정정본의 올바른 제공을 전제로 하며, 이런 한계를 운영 안내에 분명히 남깁니다.

events=[]
codes=['DUPLICATE_KEY']
if codes:
    events.append('blocked')
else:
    events.extend(['load','publish'])
print(','.join(events))

blocked

import tempfile,hashlib
from pathlib import Path
with tempfile.TemporaryDirectory() as d:
    report=Path(d)/'report.json'
    report.write_text('{"sum_vehicles":300}\n')
    before=hashlib.sha256(report.read_bytes()).hexdigest()
    try:
        raise ValueError('TRAFFIC_MISSING')
    except ValueError as e:
        print('blocked',e)
    print('same_hash',before==hashlib.sha256(report.read_bytes()).hexdigest())

blocked TRAFFIC_MISSING
same_hash True

rows=[(100,3),(80,0),(None,None),(120,10),(0,0),(90,None)]
pairs=[r for r in rows if r[0] is not None and r[1] is not None]
print('stored_rows',len(rows))
print('stored_sum',sum(r[0] for r in rows if r[0] is not None))
print('report_rows',len(pairs))
print('report_sum',sum(r[0] for r in pairs))

stored_rows 6
stored_sum 390
report_rows 4
report_sum 300

python3 -m quality.run

quality=PASS stages=5

python3 -m unittest discover -s tests -v