-
Notifications
You must be signed in to change notification settings - Fork 10
/
segment.py
153 lines (145 loc) · 6.58 KB
/
segment.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
import boto3, ffmpy, os, json
s3 = boto3.resource('s3')
dynamo = boto3.resource('dynamodb')
table = dynamo.Table('FT_SegmentState')
sqs = boto3.resource('sqs')
# Uncomment for debug logging
#boto3.set_stream_logger(name='botocore')
# Triggered by write to FT_VideoConversions
def lambda_handler(event, context):
# Load triggering row from FT_VideoConversions and assign variables
try:
print event['Records'][0]['dynamodb']['NewImage']
Row = event['Records'][0]['dynamodb']['NewImage']
Bucket = Row['Bucket']['S']
ConversionID = Row['ConversionID']['S']
Filename, Extension = os.path.splitext(Row['Filename']['S'])
Path = Row['Path']['S']
StatusQueueMessageID = Row['QueueMessageID']['S']
RequestedFormats = Row['RequestedFormats']['M']
except KeyError:
print "DynamoDB records are incomplete! {}".format(KeyError)
else:
if Path == 'NULL':
S3Path = '{}{}'.format(Filename, Extension)
elif Path != 'NULL':
S3Path = '{}{}{}'.format(Path, Filename, Extension)
LocalPath = '/tmp/{}{}'.format(Filename, Extension)
print 'Bucket/ConversionID is {}, {}'.format(Bucket, ConversionID)
print 'StatusQueueMessageID is {}'.format(StatusQueueMessageID)
'''
statusqueue.send_message(
MessageBody='Downloading source from S3...',
MessageAttributes={
'ConversionID': {
'StringValue': ConversionID,
'DataType': 'String'
}
}
)'''
try:
print "Cleaning working directory"
for f in os.listdir('/tmp/'):
os.remove('/tmp/{}'.format(f))
except Exception as e:
raise Exception('Failed to clean temp dir')
try:
print "Downloading original file..."
s3.Bucket(Bucket).download_file(S3Path, LocalPath)
except Exception as e:
raise Exception('Failure downloading file from s3: {} to local: {} Exception: {}'.format(S3Path, LocalPath, e))
'''
statusqueue.send_message(
MessageBody='Segmenting video...',
MessageAttributes={
'ConversionID': {
'StringValue': ConversionID,
'DataType': 'String'
}
}
)'''
# Segment video with ffmpeg
segment(LocalPath)
print "Removing original file..."
os.remove(LocalPath)
'''
statusqueue.send_message(
MessageBody='Uploading segments to S3',
MessageAttributes={
'ConversionID': {
'StringValue': ConversionID,
'DataType': 'String'
}
}
)'''
# Upload each segment to S3
print 'Uploading segments and audio to s3...'
for filename in os.listdir('/tmp/'):
attempts = 0
while attempts < 3:
try:
print 'uploading {} to s3'.format(filename)
if Path != 'NULL':
s3.Bucket(Bucket).upload_file('/tmp/{}'.format(filename), '{}{}'.format(Path, filename))
break
elif Path == 'NULL':
s3.Bucket(Bucket).upload_file('/tmp/{}'.format(filename), '{}'.format(filename))
break
except Exception as UploadError:
attempts += 1
raise Exception('Failure reuploading segment /tmp/{}, attempt {}'.format(filename, attempts))
# Write to FT_SegmentState
if filename.endswith('mp3'):
SegmentID = '-1'
else:
segments = os.path.splitext(filename)[0].split('SEGMENT')
SegmentID = segments[len(segments) - 1] + "-" + ConversionID
print "preparing dynamo statement for SegmentID {}".format(SegmentID)
writeattempts = 0
while writeattempts < 3:
try:
print 'Writing information to DynamoDB...'
print "Bucket: {} ConversionID: {} Filename: {} Path: {} StatusQueueMessageID: {} RequestedFormats: {} SegmentID: {}".format(Bucket, ConversionID, filename, Path, StatusQueueMessageID, RequestedFormats, SegmentID)
response = table.put_item(
Item = {
'Bucket': Bucket,
'ConversionID': ConversionID,
'Completed': 0,
'Filename': filename,
'Path': Path,
'QueueMessageID': StatusQueueMessageID,
'RequestedFormats': RequestedFormats,
'SegmentID': SegmentID,
}
)
formattedresponse = json.dumps(response, indent=4)
print('PutItem succeeded: {}'.format(formattedresponse))
break
except Exception as DynamoError:
raise Exception('Failure writing data to dynamodb for segment {}, attempt {} Exception: {}'.format(SegmentID, attempts, DynamoError))
writeattempts += 1
# ffmpy invocation that SEGMENTs the video into chunks
def segment(path):
if path is not None:
try:
print 'Breaking out audio from video file...'
FilePath, Extension = os.path.splitext(path)
f = ffmpy.FFmpeg(
executable='./ffmpeg/ffmpeg',
inputs={path : None},
outputs={'{}.mp3'.format(FilePath): '-loglevel 100 -c:a mp3'}
)
f.run()
except Exception as ffmpegerror:
raise Exception('Failure during audio breakout. {}'.format(ffmpegerror))
try:
print 'Segmenting video file...'
ff = ffmpy.FFmpeg(
executable='./ffmpeg/ffmpeg',
inputs={path : None},
outputs={'{}SEGMENT{}d{}'.format(FilePath, '%', Extension): '-loglevel 100 -acodec copy -f segment -vcodec copy -reset_timestamps 1 -map 0 -segment_format_options movflags=faststart'}
)
ff.run()
print "Segmentation complete"
except Exception as ffmpegerror:
raise Exception('Failure during segmentation of video file. {}'.format(ffmpegerror))