실행 명령과 작업 기록

예약보다 먼저 재현 가능한 명령을 만듭니다

매일 같은 파일을 처리할 작업이라도 누가 어느 폴더에서 실행했는지에 따라 결과가 달라지면 예약 실행이 어려워집니다. 자동 실행 환경에는 터미널에서 쓰던 현재 폴더와 활성화한 가상환경이 없을 수 있습니다. 이번 레슨은 OS 예약 등록 전에 명령의 출발점과 종료 상태를 고정합니다. run.sh 한 줄을 어떤 현재 폴더에서 호출해도 프로젝트 루트로 이동하고 같은 Python 진입점을 실행하게 만듭니다. 시간표보다 먼저 수동 재실행의 계약이 필요합니다.

셸 래퍼는 명령을 모으는 경계이며 데이터 정제의 새 구현 장소가 아닙니다. 미션의 순서는 기존 collect, transform, 새 merge로 Python 코드에 남습니다. run.sh는 자신의 경로를 기준으로 프로젝트 루트에 이동하고 python3 -m incremental.run을 실행합니다. 작업 순서를 셸과 Python 양쪽에 따로 적으면 한쪽만 바뀌어 실패 경계가 달라집니다. 래퍼가 할 일과 파이프라인이 할 일을 나눠야 실행 안내가 짧아져도 내부 단계는 검증 가능한 상태를 유지합니다.

경로와 종료 상태를 전달합니다

cd -- "$(dirname -- "$0")"는 호출한 래퍼 파일이 들어 있는 폴더로 이동합니다. 경로에 공백이 있어도 한 인자로 전달하도록 따옴표를 씁니다. 이 예제는 일반 파일로 ZIP을 풀어 bash로 실행하는 방식이며 심볼릭 링크를 통한 설치 경로 추적은 지원하지 않습니다. 상대 입력 경로는 ZIP 루트 기준이라는 계약을 README에 적습니다. 작성자 홈 폴더를 절대 경로로 넣지 않고 이동한 뒤 상대 경로로 설정을 읽습니다. 경로를 바꾼 테스트가 통과해야 편의가 실제 동작이 됩니다.

set -euo pipefail은 이 래퍼의 준비 명령 오류, 정의하지 않은 변수, 파이프라인 오류를 드러내는 데 사용합니다. 이것만 쓰면 모든 셸 오류를 자동으로 올바르게 처리한다고 일반화하지 않습니다. 마지막 exec는 현재 셸을 Python 명령으로 바꾸므로 Python 종료 코드가 호출자에게 그대로 전달됩니다. 별도 서버나 백그라운드 프로세스를 띄우지 않습니다. 이 모듈은 실패 시 자동 무한 재시도를 하지 않으며 호출자가 오류를 보고 다음 실행을 결정합니다.

정상 실행은 종료 코드 0, 처리 실패는 1로 구분합니다. stderr에 오류를 찍고 마지막에 exit 0을 붙이면 예약 도구는 성공으로 오해합니다. 성공 메시지 문자열의 존재만 검사하지 않고 subprocess.run의 returncode로 명령 계약을 검사합니다. 파일이 없는 경우 FileNotFoundError, JSON이 잘못된 경우 JSONDecodeError처럼 입력 단계 원인을 기록합니다. 실패 때문에 건수를 알 수 없으면 0과 error를 함께 남기며 이를 정상 빈 배치라고 해석하지 않습니다.

실행 ID와 입력 해시는 다른 질문에 답합니다

run_id는 각 실행 시도에 부여하는 식별자입니다. 같은 입력을 두 번 실행하면 실행 ID가 달라집니다. input_sha256은 실제 읽은 바이트의 SHA-256으로 같은 바이트라면 같습니다. 실행 ID를 파일 해시로 만들면 같은 파일의 두 번째 실패와 첫 번째 성공을 구분하기 어렵습니다. 반대로 해시 대신 실행 ID만 비교하면 동일 입력 재처리 여부를 확인할 수 없습니다. 두 필드를 함께 남겨 어떤 실행이 어떤 입력을 처리했는지 연결합니다.

started_at과 finished_at은 시간대가 포함된 UTC 시각으로 기록합니다. 교통 관측의 date와 실행 시각은 역할이 다릅니다. 어제 관측을 오늘 정정할 수 있으므로 실행 시각을 관측 키에 넣으면 같은 관측이 새 행으로 만들어집니다. 이 실습의 UUID와 시각은 실행마다 달라 기대 출력에 고정하지 않습니다. 테스트는 두 ID의 차이와 필드의 의미를 검사하고 화면 출력은 상태와 건수만 보여 줍니다. 실제로 달라지는 값을 억지로 고정해서 재현성을 주장하지 않습니다.

미션의 입력 해시는 설정 파일 이름이 아니라 collect가 보존한 payload 바이트에서 계산합니다. 그 해시가 수집 스냅샷 폴더와 실행 기록을 연결합니다. 같은 관측이라도 줄바꿈이나 공백이 바뀌면 파일 해시는 달라질 수 있습니다. 해시가 다르다는 이유만으로 관측이 추가되어야 하는 것은 아닙니다. 지역·날짜 키와 값 비교는 별도로 유지합니다. 해시를 안전한 입력이라는 보증으로도 사용하지 않습니다. 해시가 있어도 잘못된 날짜와 충돌 키는 스키마 검사를 통과할 수 없습니다.

건수와 상태를 함께 남깁니다

독립 셸 실습의 job.py는 JSON 배열 여부와 행 수만 검사하는 작은 명령 계약 예제입니다. total_vehicles 등의 실제 데이터 품질 검사는 미션에서 이어받습니다. input_rows는 입력 배열의 행 수이고 stored_rows는 DB 전체 행 수입니다. 변경분 한 행을 적재해도 기존 여섯 행이 남아 있으면 입력 건수는 1이고 저장 건수는 6입니다. 두 수치가 다르다고 오류라고 단정하지 않고 각각의 단위를 설명합니다. 새 행만 센 수치와 정정된 행 수를 혼합하지 않습니다.

미션의 감사 기록은 before와 after에 행 수, 정렬된 키, 통행량 합계, NULL 개수와 전체 행을 남깁니다. 동일 입력 재처리에서는 before와 after가 같고 상태는 success입니다. 실패 주입에서는 상태가 failed이며 두 상태가 같아야 합니다. 정상 빈 입력은 success이면서 input_rows가 0입니다. 실패 전에 입력을 해석하지 못한 실행은 failed이면서 input_rows가 0입니다. 건수 하나로 빈 데이터와 오류를 구분할 수 없으므로 종료 코드와 상태, 오류 이유를 함께 읽습니다.

run-audit.json은 배열에 실행 기록을 추가하여 두 번의 실행 증거를 보존합니다. 임시 파일에 쓰고 replace로 교체하여 정상 완료된 JSON을 읽도록 합니다. 그렇다고 DB 커밋과 JSON이 동시에 확정되는 것은 아니며 프로세스 강제 중단이나 파일 기록 실패 때 기록이 빠질 수 있습니다. 이번 구현은 한 번에 한 실행만 지원합니다. 동시 실행이 같은 임시 파일이나 기록 배열을 덮어쓸 수 있으므로 예약 간격만 길게 설정해 동시성 문제를 해결했다고 주장하지 않습니다.

예약할 때 정해야 하는 운영 조건

OS 예약 자체는 이 레슨에서 등록하지 않습니다. 실제 예약 전에는 프로젝트 절대 경로, Python 실행 경로, 관측 날짜 결정 방법, 시간대, 이전 작업이 아직 실행 중일 때의 처리와 로그 보관 기간을 인계서에 적습니다. 이미 실행 중이면 새 실행을 건너뛸지 대기할지 정책이 필요합니다. SQLite 쓰기 잠금은 DB 충돌을 일부 드러내지만 수집 파일과 감사 기록의 경쟁까지 해결하는 스케줄러 잠금은 아닙니다. 수동 명령과 종료 코드부터 검증한 뒤 예약 기능을 선택합니다.

같은 날짜를 다시 처리할 수 있는 명령은 장애 복구에도 사용됩니다. 입력을 보존하고 해시로 연결했다면 운영자는 실패 기록의 입력을 찾아 수동 재실행할 수 있습니다. 무조건 현재 최신 파일을 다시 내려받으면 실패했던 바이트와 달라져 원인 재현이 어려워집니다. 미션은 네트워크 없이 작업 사본의 fixture로 정상·정정·실패를 재현합니다. 실제 HTTP 수집의 기존 선택 테스트는 별도로 남기며 이 모듈의 완료를 외부 API 성공 여부에 의존시키지 않습니다.

실습 결과를 읽는 순서

starter의 run.sh는 현재 폴더를 고정하지 않고 Python 실패 뒤에도 0으로 종료합니다. tests의 다른 현재 폴더 검사와 missing.json 종료 코드 검사가 실패하는 이유를 각각 설명하고 래퍼를 고칩니다. job.py의 기록 형식을 바꾸거나 실패 입력을 삭제해 통과시키지 않습니다. bash check.sh가 성공하면 입력 해시, 실행 ID, 정상 빈 배열과 잘못된 JSON까지 확인된 것입니다. 실제 명령을 두 번 실행한 뒤 runs.json의 두 시도를 읽어 어떤 입력이 재처리되었는지 찾아봅니다.

프로젝트 미션에서는 bash run.sh를 두 번 호출하고 정정 fixture와 실패 fixture를 작업 사본에 적용합니다. 기존 80개 검사와 새 증분 검사를 함께 유지해 보고서와 정제의 계약도 보호합니다. 사람에게 전달할 완료 증거는 실행 ID가 있는 기록, 재처리 전후 같은 관측 상태, 정정 한 키의 변경, 실패 후 보존된 DB입니다. 더 읽기의 CLI·예약 장에서 운영체제별 등록 방법을 확인하되 이 레슨에서는 실행 가능한 명령과 실패 신호의 전달을 먼저 완성합니다.

import hashlib,uuid
payload=b'region,date,total_vehicles\nA,2026-09-01,100\n'
a={'run_id':uuid.uuid4().hex,'hash':hashlib.sha256(payload).hexdigest()}
b={'run_id':uuid.uuid4().hex,'hash':hashlib.sha256(payload).hexdigest()}
print('different_run_id',a['run_id']!=b['run_id'])
print('same_input_hash',a['hash']==b['hash'])

different_run_id True
same_input_hash True

import subprocess,sys
for code in [0,1]:
    result=subprocess.run([sys.executable,'-c',f'import sys; sys.exit({code})'])
    print('exit',result.returncode,'state','success' if result.returncode==0 else 'failed')

exit 0 state success
exit 1 state failed

import tempfile,subprocess
from pathlib import Path
with tempfile.TemporaryDirectory() as folder:
    root=Path(folder)/'work folder';root.mkdir()
    (root/'job.py').write_text("print('root_ready')\n")
    (root/'run.sh').write_text('#!/usr/bin/env bash\nset -euo pipefail\ncd -- "$(dirname -- "$0")"\nexec python3 job.py\n')
    p=subprocess.run(['bash',str(root/'run.sh')],cwd=folder,capture_output=True,text=True,check=True)
    print(p.stdout,end='')
    print('exit',p.returncode)

root_ready
exit 0

bash check.sh

bash check.sh