import ffmpeg
import cv2
import os
from pathlib import Path

class VideoPipeline:
"""영상 전처리 & 후처리 파이프라인"""

def __init__(self, input_path: str, output_dir: str):
self.input_path = input_path
self.output_dir = Path(output_dir)
self.output_dir.mkdir(exist_ok=True)
self.metadata = self._extract_metadata()

def _extract_metadata(self) -> dict:
"""영상 메타데이터 추출"""
probe = ffmpeg.probe(self.input_path)
video_stream = next(
(s for s in probe['streams'] if s['codec_type'] == 'video'), None
)
return {
'width': int(video_stream['width']),
'height': int(video_stream['height']),
'fps': eval(video_stream['r_frame_rate']),
'duration': float(probe['format']['duration']),
'codec': video_stream['codec_name']
}

def preprocess(self, target_resolution=(1920, 1080), target_fps=30):
"""전처리: 해상도, FPS 정규화"""
preprocessed_path = self.output_dir / 'preprocessed.mp4'

stream = ffmpeg.input(self.input_path)
stream = ffmpeg.filter(stream, 'scale', 
target_resolution[0], target_resolution[1],
force_original_aspect_ratio='decrease')
stream = ffmpeg.filter(stream, 'pad', 
target_resolution[0], target_resolution[1],
'(ow-iw)/2', '(oh-ih)/2')
stream = ffmpeg.filter(stream, 'fps', fps=target_fps)

ffmpeg.output(stream, str(preprocessed_path),
vcodec='libx264', crf=18, 
acodec='aac', ar=48000).run(overwrite_output=True)

self.preprocessed_path = str(preprocessed_path)
print(f"✅ 전처리 완료: {preprocessed_path}")
return self

def denoise(self):
"""노이즈 제거"""
denoised_path = self.output_dir / 'denoised.mp4'

stream = ffmpeg.input(self.preprocessed_path)
stream = ffmpeg.filter(stream, 'hqdn3d', 4, 3, 6, 4.5)
ffmpeg.output(stream, str(denoised_path),
vcodec='libx264', crf=18).run(overwrite_output=True)

self.preprocessed_path = str(denoised_path)
print(f"✅ 노이즈 제거 완료")
return self

def color_grade(self, lut_path=None):
"""색보정 (LUT 적용 또는 기본 보정)"""
graded_path = self.output_dir / 'color_graded.mp4'

stream = ffmpeg.input(self.preprocessed_path)

if lut_path:
stream = ffmpeg.filter(stream, 'lut3d', file=lut_path)
else:
stream = ffmpeg.filter(stream, 'eq', 
brightness=0.02, contrast=1.05, saturation=1.1)

ffmpeg.output(stream, str(graded_path),
vcodec='libx264', crf=18).run(overwrite_output=True)

self.processed_path = str(graded_path)
print(f"✅ 색보정 완료")
return self

def add_watermark(self, watermark_text="© 재능넷"):
"""워터마크 삽입"""
watermarked_path = self.output_dir / 'watermarked.mp4'

stream = ffmpeg.input(self.processed_path)
stream = ffmpeg.drawtext(stream, 
text=watermark_text,
fontsize=20,
fontcolor='white@0.6',
x='W-tw-10', y='H-th-10')

ffmpeg.output(stream, str(watermarked_path),
vcodec='libx264', crf=18).run(overwrite_output=True)

self.processed_path = str(watermarked_path)
print(f"✅ 워터마크 삽입 완료")
return self

def normalize_audio(self):
"""오디오 볼륨 정규화 (-23 LUFS)"""
normalized_path = self.output_dir / 'audio_normalized.mp4'

stream = ffmpeg.input(self.processed_path)
audio = ffmpeg.filter(stream.audio, 'loudnorm', 
I=-23, TP=-1.5, LRA=11)

ffmpeg.output(stream.video, audio, str(normalized_path),
vcodec='copy').run(overwrite_output=True)

self.processed_path = str(normalized_path)
print(f"✅ 오디오 정규화 완료")
return self

def export(self, output_name='final_output.mp4', crf=23):
"""최종 출력"""
final_path = self.output_dir / output_name

stream = ffmpeg.input(self.processed_path)
ffmpeg.output(stream, str(final_path),
vcodec='libx264', crf=crf,
acodec='aac', ar=48000,
movflags='faststart').run(overwrite_output=True)

print(f"🎉 최종 출력 완료: {final_path}")
return str(final_path)


# 파이프라인 실행 예시
pipeline = VideoPipeline('input_video.mp4', './output')
result = (pipeline
.preprocess(target_resolution=(1920, 1080), target_fps=30)
.denoise()
.color_grade()
.add_watermark("© 내 채널")
.normalize_audio()
.export('final.mp4', crf=20)
)

# AWS Lambda 함수 예시 (S3 업로드 시 자동 트랜스코딩)
import boto3
import json

def lambda_handler(event, context):
mediaconvert = boto3.client('mediaconvert', 
endpoint_url='https://your-endpoint.mediaconvert.region.amazonaws.com')

# S3 이벤트에서 파일 정보 추출
bucket = event['Records'][0]['s3']['bucket']['name']
key = event['Records'][0]['s3']['object']['key']

input_url = f"s3://{bucket}/{key}"
output_url = f"s3://output-bucket/processed/"

job_settings = {
"Inputs": [{
"FileInput": input_url,
"AudioSelectors": {"Audio Selector 1": {"DefaultSelection": "DEFAULT"}},
"VideoSelector": {}
}],
"OutputGroups": [{
"Name": "File Group",
"OutputGroupSettings": {
"Type": "FILE_GROUP_SETTINGS",
"FileGroupSettings": {"Destination": output_url}
},
"Outputs": [{
"VideoDescription": {
"CodecSettings": {
"Codec": "H_264",
"H264Settings": {"RateControlMode": "QVBR", "QvbrSettings": {"QvbrQualityLevel": 7}}
},
"Width": 1920, "Height": 1080
},
"AudioDescriptions": [{
"CodecSettings": {"Codec": "AAC", "AacSettings": {"Bitrate": 128000, "SampleRate": 48000}}
}],
"ContainerSettings": {"Container": "MP4"}
}]
}]
}

response = mediaconvert.create_job(
Role='arn:aws:iam::account:role/MediaConvertRole',
Settings=job_settings
)

return {'statusCode': 200, 'body': json.dumps('트랜스코딩 작업 시작!')}

from scenedetect import VideoManager, SceneManager
from scenedetect.detectors import ContentDetector

def detect_scenes(video_path):
video_manager = VideoManager([video_path])
scene_manager = SceneManager()
scene_manager.add_detector(ContentDetector(threshold=30.0))

video_manager.start()
scene_manager.detect_scenes(frame_source=video_manager)
scene_list = scene_manager.get_scene_list()

print(f"총 {len(scene_list)}개 장면 감지됨")
for i, scene in enumerate(scene_list):
print(f"장면 {i+1}: {scene[0].get_timecode()} ~ {scene[1].get_timecode()}")

return scene_list

import whisper

def auto_subtitle(video_path, output_srt_path):
"""Whisper로 자동 자막 생성"""
model = whisper.load_model("large")  # 모델 크기: tiny/base/small/medium/large

result = model.transcribe(video_path, language='ko')

# SRT 파일 생성
with open(output_srt_path, 'w', encoding='utf-8') as f:
for i, segment in enumerate(result['segments']):
start = segment['start']
end = segment['end']
text = segment['text'].strip()

# SRT 형식으로 변환
f.write(f"{i+1}\n")
f.write(f"{format_time(start)} --> {format_time(end)}\n")
f.write(f"{text}\n\n")

print(f"✅ 자막 파일 생성 완료: {output_srt_path}")

def format_time(seconds):
h = int(seconds // 3600)
m = int((seconds % 3600) // 60)
s = int(seconds % 60)
ms = int((seconds % 1) * 1000)
return f"{h:02d}:{m:02d}:{s:02d},{ms:03d}"

# FFmpeg으로 VMAF 품질 측정
ffmpeg -i original.mp4 -i compressed.mp4 \
-lavfi "[0:v][1:v]libvmaf=log_fmt=json:log_path=vmaf_result.json" \
-f null -

# SSIM 측정
ffmpeg -i original.mp4 -i compressed.mp4 \
-lavfi "[0:v][1:v]ssim=stats_file=ssim_result.txt" \
-f null -

# PSNR 측정
ffmpeg -i original.mp4 -i compressed.mp4 \
-lavfi "[0:v][1:v]psnr=stats_file=psnr_result.txt" \
-f null -

import json
import subprocess

def quality_check(original_path, processed_path, vmaf_threshold=70):
"""자동 품질 검사"""
result = subprocess.run([
'ffmpeg', '-i', original_path, '-i', processed_path,
'-lavfi', f'[0:v][1:v]libvmaf=log_fmt=json:log_path=/tmp/vmaf.json',
'-f', 'null', '-'
], capture_output=True)

with open('/tmp/vmaf.json', 'r') as f:
vmaf_data = json.load(f)

vmaf_score = vmaf_data['pooled_metrics']['vmaf']['mean']

if vmaf_score >= vmaf_threshold:
print(f"✅ 품질 검사 통과! VMAF: {vmaf_score:.2f}")
return True
else:
print(f"❌ 품질 검사 실패! VMAF: {vmaf_score:.2f} (기준: {vmaf_threshold})")
return False