3.7 분석 결과를 MAM 등록용 레코드로 정규화하기
FFprobe 결과를 일관된 JSON 레코드로 바꾸고 경고를 붙이는 파이썬 분석 도구를 만들며, 정규화 설계 원칙과 MediaInfo와의 교차 검증 방법을 설명합니다.
#정규화가 필요한 이유
앞의 글에서 본 것처럼 같은 정보라도 파일 종류마다 위치와 표현이 다릅니다.
| 정보 | MXF | MOV | TS |
|---|---|---|---|
| 시작 타임코드 | 포맷 태그 | 스트림 태그(비디오, 데이터 트랙) | 보통 없음 |
| 컨테이너 이름 | mxf |
mov,mp4,m4a,3gp,3g2,mj2 |
mpegts |
| 프레임 수 | 있거나 N/A |
있음 | 보통 N/A |
| 필드 순서 | tt |
progressive |
progressive |
MAM의 DB와 검색 화면, 워크플로우 규칙은 이런 차이를 몰라야 합니다. 분석 결과를 하나의 일관된 레코드 형식으로 바꿔서 저장하는 단계가 필요한 이유입니다. 이것이 정규화(Normalization)입니다.
#설계 원칙
- 원시 값과 정규화 값을 모두 보관합니다. 정규화는 해석이 들어가므로, 나중에 규칙이 바뀌면 원시 분석 결과(ffprobe JSON 전체, MediaInfo JSON 전체)에서 다시 만들 수 있어야 합니다.
- 프레임레이트는 분수 문자열(
30000/1001)로 저장합니다. - 없는 값은
null로 두고, 빈 문자열이나 0으로 채우지 않습니다. 타임코드가 없는 TS가00:00:00:00으로 등록되면 이후 모든 구간 계산이 어긋납니다. - 경고(warnings)를 함께 저장합니다. 비표준 프레임레이트, 홀수 해상도, 오디오 없음 같은 항목은 등록을 막기보다 표시하고 사람이 판단하게 합니다.
- 분석 도구의 버전을 기록합니다. FFmpeg와 MediaInfo의 버전을 레코드에 남깁니다.
#구현
아래는 FFprobe JSON에서 레코드를 만드는 파이썬 스크립트입니다.
import json
import subprocess
import sys
from fractions import Fraction
KNOWN_FPS = {Fraction(24000, 1001), Fraction(24), Fraction(25), Fraction(30000, 1001),
Fraction(30), Fraction(50), Fraction(60000, 1001), Fraction(60)}
def ffprobe(path):
cmd = ['ffprobe', '-v', 'error', '-show_format', '-show_streams', '-of', 'json', path]
return json.loads(subprocess.run(cmd, capture_output=True, text=True, check=True).stdout)
def find_timecode(info):
tc = info.get('format', {}).get('tags', {}).get('timecode')
for s in info['streams']:
tc = tc or s.get('tags', {}).get('timecode')
return tc
def parse_fps(text):
"""'30000/1001' -> Fraction. 값이 없거나 'N/A', '0/0'처럼 쓸 수 없으면 None."""
try:
fps = Fraction(text)
except (TypeError, ValueError, ZeroDivisionError):
return None
return fps if fps > 0 else None
def normalize(path):
info = ffprobe(path)
fmt = info['format']
video = next((s for s in info['streams'] if s['codec_type'] == 'video'), None)
audio = [s for s in info['streams'] if s['codec_type'] == 'audio']
rec = {
'path': path,
'container': fmt['format_name'],
'duration_sec': float(fmt['duration']) if 'duration' in fmt else None,
'size_bytes': int(fmt['size']) if 'size' in fmt else None,
'timecode': find_timecode(info),
'warnings': [],
}
if video:
fps = parse_fps(video.get('r_frame_rate')) or parse_fps(video.get('avg_frame_rate'))
field = video.get('field_order', 'unknown')
w, h = video.get('width'), video.get('height')
rec['video'] = {
'codec': video.get('codec_name'),
'width': w, 'height': h,
'fps': f'{fps.numerator}/{fps.denominator}' if fps else None,
'scan': field,
'pix_fmt': video.get('pix_fmt'),
}
if fps is None:
rec['warnings'].append('프레임레이트를 확인할 수 없습니다')
elif fps not in KNOWN_FPS:
rec['warnings'].append(f'비표준 프레임레이트: {float(fps):.3f}')
if w and h and (w % 2 or h % 2):
rec['warnings'].append('가로/세로가 홀수입니다(4:2:0 변환 시 오류 가능)')
else:
rec['warnings'].append('비디오 스트림이 없습니다')
rec['audio'] = [{'codec': a.get('codec_name'), 'channels': a.get('channels'),
'sample_rate': int(a['sample_rate']) if 'sample_rate' in a else None} for a in audio]
if not audio:
rec['warnings'].append('오디오 스트림이 없습니다')
return rec
if __name__ == '__main__':
print(json.dumps(normalize(sys.argv[1]), ensure_ascii=False, indent=2))
핵심 부분은 세 가지입니다.
Fraction으로 프레임레이트를 다뤄 분수를 그대로 보존하고,30000/1001을 정확히 비교합니다.find_timecode가 포맷 태그와 모든 스트림 태그를 찾아서, 파일 종류에 관계없이 타임코드를 한 곳에서 얻습니다.- 경고는 판정이 아니라 표시입니다. 결과를 막지 않고
warnings목록에 남깁니다. parse_fps는0/0,N/A, 누락을 예외 없이None으로 바꿉니다.r_frame_rate가0/0인 스트림(일부 오디오 전용·이미지·손상 파일)에Fraction('0/0')을 그대로 쓰면ZeroDivisionError로 분석 전체가 중단됩니다.r_frame_rate를 쓸 수 없으면avg_frame_rate로 대체하고, 둘 다 안 되면fps를null로 두고 경고를 남깁니다.
#실행 결과
앞 글들에서 만든 세 종류의 파일을 분석한 결과입니다(경로는 생략).
{
"container": "mxf",
"duration_sec": 5.005,
"size_bytes": 23327289,
"timecode": "01:00:00:00",
"warnings": [],
"video": { "codec": "mpeg2video", "width": 1920, "height": 1080,
"fps": "30000/1001", "scan": "tt", "pix_fmt": "yuv422p" },
"audio": [ { "codec": "pcm_s24le", "channels": 1, "sample_rate": 48000 },
{ "codec": "pcm_s24le", "channels": 1, "sample_rate": 48000 } ]
}
{
"container": "mov,mp4,m4a,3gp,3g2,mj2",
"timecode": "00:59:58;00",
"video": { "codec": "prores", "width": 1280, "height": 720,
"fps": "30000/1001", "scan": "progressive", "pix_fmt": "yuv422p10le" }
}
{
"container": "mpegts",
"timecode": null,
"video": { "codec": "h264", "width": 1280, "height": 720,
"fps": "25/1", "scan": "progressive", "pix_fmt": "yuv420p" }
}
MXF, MOV, TS가 같은 구조의 레코드로 바뀌었습니다. TS는 타임코드가 없으므로 null입니다. 이 값이 null이라는 사실에서 인제스트 규칙(예: "타임코드가 없으면 00:00:00:00 시작으로 간주하고 timecode_source: assumed 표시")을 정할 수 있습니다.
#MediaInfo와의 교차 검증
FFprobe의 해석과 MediaInfo의 해석이 다르다면 파일에 문제가 있거나 한쪽이 잘못 읽은 것입니다. 대표적인 교차 검증 항목은 다음과 같습니다.
| 비교 항목 | 불일치의 의미 |
|---|---|
| 코덱·해상도 | 거의 일치해야 함. 다르면 손상 의심 |
| 프레임레이트 | TS 같은 포맷에서는 차이가 날 수 있음 |
| 길이 | 인덱스 없는 포맷에서는 소폭 차이 허용 |
IsTruncated |
Yes이면 FFprobe가 정상이어도 손상 의심 |
| 프레임 수 | -count_frames와 MediaInfo FrameCount 비교 |
불일치를 모두 오류로 처리하면 정상 파일도 걸리므로, 허용 오차와 포맷별 예외를 정해 두고 운영하면서 조정해 가는 편이 현실적입니다.
#확장 방향
- 큰 파일 대응:
-show_frames는 오래 걸리므로 의심 파일에서만 실행하고, 평소에는-show_format과-show_streams만 씁니다. - 병렬 처리: 분석은 I/O 중심이라 작업 큐(워커 여러 개)로 분산하기 좋습니다.
- 분석 이력 보관: 같은 파일을 다시 분석할 때 결과를 덮어쓰지 말고 버전과 시각을 붙여 남깁니다.
- 규격 프로파일 연동: 납품·송출 규격(예: "1080i 29.97, XDCAM HD422 MXF OP1a")을 데이터로 정의하고, 정규화 레코드가 규격에 맞는지 자동 판정합니다.
#정리
- 분석 결과는 파일 종류와 무관한 일관된 레코드로 정규화해 저장합니다.
- 원시 결과 보관, 분수 프레임레이트,
null표현, 경고 목록, 도구 버전 기록이 기본 원칙입니다. - 두 도구(FFprobe, MediaInfo)의 결과를 교차 검증하면 단일 도구로는 놓치는 손상을 잡을 수 있습니다.
다음 장에서는 분석한 파일을 FFmpeg로 변환하는 방법을 다룹니다.
#참고 자료
링크는 2026-10-28에 열람이 되는 것을 확인했습니다. 규격과 도구 문서는 개정될 수 있으므로 적용하기 전에 최신 판을 확인하세요.