用Python的Django框架完成視頻處理任務的教程
更新時間：2015年04月02日 15:44:08 作者：stickyworld
這篇文章主要介紹了用Python的Django框架完成視頻處理任務的教程,包括用戶的視頻上傳和播放以及下載功能的實現(xiàn),需要的朋友可以參考下
Stickyworld 的網(wǎng)頁應用已經(jīng)支持視頻撥放一段時間，但都是通過YouTube的嵌入模式實現(xiàn)。我們開始提供新的版本支持視頻操作，可以讓我們的用戶不用受制于YouTube的服務。
我過去曾經(jīng)參與過一個項目，客戶需要視頻轉碼功能，這實在不是個容易達成的需求。需要大量的讀取每一個視頻、音訊與視頻容器的格式再輸出符合網(wǎng)頁使用與喜好的視頻格式。
考慮到這一點，我們決定將轉碼的工作交給 Encoding.com 。這個網(wǎng)站可以免費讓你編碼1GB大小的視頻，超過1GB容量的文件將采取分級計價收費。
開發(fā)的代碼如下，我上傳了一個178KB容量的兩秒視頻來測試代碼是否成功運作。當測試過程沒有發(fā)生任何的例外錯誤后，我繼續(xù)測試其它更大的外部文件。
階段一：用戶上傳視頻文件
現(xiàn)在這的新的代碼段提供了一個基于 HTML5且可以快速上手的 的上傳機制。用CoffeeScript撰寫的代碼，可以從客戶端上傳文件到服務器端。
$scope.upload_slide = (upload_slide_form) ->
file = document.getElementById("slide_file").files[0]
reader = new FileReader()
reader.readAsDataURL file
reader.onload = (event) ->
result = event.target.result
fileName = document.getElementById("slide_file").files[0].name
$.post "/world/upload_slide",
data: result
name: fileName
room_id: $scope.room.id
(response_data) ->
if response_data.success? is not yes
console.error "There was an error uploading the file", response_data
else
console.log "Upload successful", response_data
reader.onloadstart = ->
console.log "onloadstart"
reader.onprogress = (event) ->
console.log "onprogress", event.total, event.loaded, (event.loaded / event.total) * 100
reader.onabort = ->
console.log "onabort"
reader.onerror = ->
console.log "onerror"
reader.onloadend = (event) ->
console.log "onloadend", event
最好可以通過 (“slide_file”).files 且經(jīng)由獨立的POST上傳每個文件，而不是由一個POST需求上傳所有文件。稍后我們會解釋這點。
階段二：驗證并上傳至 Amazon S3
后端我們運行了Django與RabbitMQ。主要的模塊如下：
$ pip install 'Django>=1.5.2' 'django-celery>=3.0.21' \
'django-storages>=1.1.8' 'lxml>=3.2.3' 'python-magic>=0.4.3'
我建立了兩個模塊：SlideUploadQueue 用來儲存每一次上傳的數(shù)據(jù)，SlideVideoMedia 則是用來儲存每個要上傳影片的數(shù)據(jù)。
class SlideUploadQueue(models.Model):
created_by = models.ForeignKey(User)
created_time = models.DateTimeField(db_index=True)
original_file = models.FileField(
upload_to=filename_sanitiser, blank=True, default='')
media_type = models.ForeignKey(MediaType)
encoding_com_tracking_code = models.CharField(
default='', max_length=24, blank=True)
STATUS_AWAITING_DATA = 0
STATUS_AWAITING_PROCESSING = 1
STATUS_PROCESSING = 2
STATUS_AWAITING_3RD_PARTY_PROCESSING = 5
STATUS_FINISHED = 3
STATUS_FAILED = 4
STATUS_LIST = (
(STATUS_AWAITING_DATA, 'Awaiting Data'),
(STATUS_AWAITING_PROCESSING, 'Awaiting processing'),
(STATUS_PROCESSING, 'Processing'),
(STATUS_AWAITING_3RD_PARTY_PROCESSING,
'Awaiting 3rd-party processing'),
(STATUS_FINISHED, 'Finished'),
(STATUS_FAILED, 'Failed'),
)
status = models.PositiveSmallIntegerField(
default=STATUS_AWAITING_DATA, choices=STATUS_LIST)
class Meta:
verbose_name = 'Slide'
verbose_name_plural = 'Slide upload queue'
def save(self, *args, **kwargs):
if not self.created_time:
self.created_time = \
datetime.utcnow().replace(tzinfo=pytz.utc)
return super(SlideUploadQueue, self).save(*args, **kwargs)
def __unicode__(self):
if self.id is None:
return 'new <SlideUploadQueue>'
return '<SlideUploadQueue> %d' % self.id
class SlideVideoMedia(models.Model):
converted_file = models.FileField(
upload_to=filename_sanitiser, blank=True, default='')
FORMAT_MP4 = 0
FORMAT_WEBM = 1
FORMAT_OGG = 2
FORMAT_FL9 = 3
FORMAT_THUMB = 4
supported_formats = (
(FORMAT_MP4, 'MPEG 4'),
(FORMAT_WEBM, 'WebM'),
(FORMAT_OGG, 'OGG'),
(FORMAT_FL9, 'Flash 9 Video'),
(FORMAT_THUMB, 'Thumbnail'),
)
mime_types = (
(FORMAT_MP4, 'video/mp4'),
(FORMAT_WEBM, 'video/webm'),
(FORMAT_OGG, 'video/ogg'),
(FORMAT_FL9, 'video/mp4'),
(FORMAT_THUMB, 'image/jpeg'),
)
format = models.PositiveSmallIntegerField(
default=FORMAT_MP4, choices=supported_formats)
class Meta:
verbose_name = 'Slide video'
verbose_name_plural = 'Slide videos'
def __unicode__(self):
if self.id is None:
return 'new <SlideVideoMedia>'
return '<SlideVideoMedia> %d' % self.id
我們的模塊皆使用 filename_sanitiser。FileField 自動的將文件名調(diào)整成 <model>/<uuid4>.<extention> 格式。整理每個文件名并確保其獨一性。我們采用了有時效性簽署的網(wǎng)址列讓我們可以掌控哪些使用者在使用我們的服務，使用了多久。
def filename_sanitiser(instance, filename):
folder = instance.__class__.__name__.lower()
ext = 'jpg'
if '.' in filename:
t_ext = filename.split('.')[-1].strip().lower()
if t_ext != '':
ext = t_ext
return '%s/%s.%s' % (folder, str(uuid.uuid4()), ext)
拿來測試的文件 testing.mov 將會轉換成以下網(wǎng)址：https://our-bucket.s3.amazonaws.com/slideuploadqueue/3fe27193-e87f-4244-9aa2-66409f70ebd3.mov 并經(jīng)由Django Storages 模塊上傳。
我們通過 Magic 驗證從使用者端瀏覽器上傳的文件。Magic可以從文件內(nèi)容偵測是何種類型的文件。
@verify_auth_token
@return_json
def upload_slide(request):
file_data = request.POST.get('data', '')
file_data = base64.b64decode(file_data.split(';base64,')[1])
description = magic.from_buffer(file_data)
如果文件類型符合MPEG v4 系統(tǒng)或是Apple QuickTime 電影，我們就知道該文件轉碼不會有太大問題。如果格式不是上述所提的幾種，我們會標志給用戶知悉。
接著，我們將通過SlideUploadQueue 模塊將視頻儲存到隊列并發(fā)送一個需求給 RabbitMQ。因為我們使用了Django Storages 模塊，文件將自動被上傳到 Amazon S3。
slide_upload = SlideUploadQueue()
...
slide_upload.status = SlideUploadQueue.STATUS_AWAITING_PROCESSING
slide_upload.save()
slide_upload.original_file.\
save('anything.%s' % file_ext, ContentFile(file_data))
slide_upload.save()
task = ConvertRawSlideToSlide()
task.delay(slide_upload)
階段3：發(fā)送視頻到第三方.
RabbitMQ 將控管 task.delay(slide_upload) 的呼叫。
我們現(xiàn)在只需要發(fā)送視頻檔網(wǎng)址與輸出格式給Encoding.com。該網(wǎng)站會回復我們一個工作碼讓我們檢查視頻轉碼的進度。
class ConvertRawSlideToSlide(Task):
queue = 'backend_convert_raw_slides'
...
def _handle_video(self, slide_upload):
mp4 = {
'output': 'mp4',
'size': '320x240',
'bitrate': '256k',
'audio_bitrate': '64k',
'audio_channels_number': '2',
'keep_aspect_ratio': 'yes',
'video_codec': 'mpeg4',
'profile': 'main',
'vcodecparameters': 'no',
'audio_codec': 'libfaac',
'two_pass': 'no',
'cbr': 'no',
'deinterlacing': 'no',
'keyframe': '300',
'audio_volume': '100',
'file_extension': 'mp4',
'hint': 'no',
}
webm = {
'output': 'webm',
'size': '320x240',
'bitrate': '256k',
'audio_bitrate': '64k',
'audio_sample_rate': '44100',
'audio_channels_number': '2',
'keep_aspect_ratio': 'yes',
'video_codec': 'libvpx',
'profile': 'baseline',
'vcodecparameters': 'no',
'audio_codec': 'libvorbis',
'two_pass': 'no',
'cbr': 'no',
'deinterlacing': 'no',
'keyframe': '300',
'audio_volume': '100',
'preset': '6',
'file_extension': 'webm',
'acbr': 'no',
}
ogg = {
'output': 'ogg',
'size': '320x240',
'bitrate': '256k',
'audio_bitrate': '64k',
'audio_sample_rate': '44100',
'audio_channels_number': '2',
'keep_aspect_ratio': 'yes',
'video_codec': 'libtheora',
'profile': 'baseline',
'vcodecparameters': 'no',
'audio_codec': 'libvorbis',
'two_pass': 'no',
'cbr': 'no',
'deinterlacing': 'no',
'keyframe': '300',
'audio_volume': '100',
'file_extension': 'ogg',
'acbr': 'no',
}
flv = {
'output': 'fl9',
'size': '320x240',
'bitrate': '256k',
'audio_bitrate': '64k',
'audio_channels_number': '2',
'keep_aspect_ratio': 'yes',
'video_codec': 'libx264',
'profile': 'high',
'vcodecparameters': 'no',
'audio_codec': 'libfaac',
'two_pass': 'no',
'cbr': 'no',
'deinterlacing': 'no',
'keyframe': '300',
'audio_volume': '100',
'file_extension': 'mp4',
}
thumbnail = {
'output': 'thumbnail',
'time': '5',
'video_codec': 'mjpeg',
'keep_aspect_ratio': 'yes',
'file_extension': 'jpg',
}
encoder = Encoding(settings.ENCODING_API_USER_ID,
settings.ENCODING_API_USER_KEY)
resp = encoder.add_media(source=[slide_upload.original_file.url],
formats=[mp4, webm, ogg, flv, thumbnail])
media_id = None
if resp is not None and resp.get('response') is not None:
media_id = resp.get('response').get('MediaID')
if media_id is None:
slide_upload.status = SlideUploadQueue.STATUS_FAILED
slide_upload.save()
log.error('Unable to communicate with encoding.com')
return False
slide_upload.encoding_com_tracking_code = media_id
slide_upload.status = \
SlideUploadQueue.STATUS_AWAITING_3RD_PARTY_PROCESSING
slide_upload.save()
return True
Encoding.com 推薦一些堪用的Python程序，可用來與它們的服務溝通。我修改了模塊一些地方，但還需要修改一些功能才能達到我滿意的狀態(tài)。以下是修改過后目前正在使用的程序代碼：
import httplib
from lxml import etree
import urllib
from xml.parsers.expat import ExpatError
import xmltodict
ENCODING_API_URL = 'manage.encoding.com:80'
class Encoding(object):
def __init__(self, userid, userkey, url=ENCODING_API_URL):
self.url = url
self.userid = userid
self.userkey = userkey
def get_media_info(self, action='GetMediaInfo', ids=[],
headers={'Content-Type': 'application/x-www-form-urlencoded'}):
query = etree.Element('query')
nodes = {
'userid': self.userid,
'userkey': self.userkey,
'action': action,
'mediaid': ','.join(ids),
}
query = self._build_tree(etree.Element('query'), nodes)
results = self._execute_request(query, headers)
return self._parse_results(results)
def get_status(self, action='GetStatus', ids=[], extended='no',
headers={'Content-Type': 'application/x-www-form-urlencoded'}):
query = etree.Element('query')
nodes = {
'userid': self.userid,
'userkey': self.userkey,
'action': action,
'extended': extended,
'mediaid': ','.join(ids),
}
query = self._build_tree(etree.Element('query'), nodes)
results = self._execute_request(query, headers)
return self._parse_results(results)
def add_media(self, action='AddMedia', source=[], notify='', formats=[],
instant='no',
headers={'Content-Type': 'application/x-www-form-urlencoded'}):
query = etree.Element('query')
nodes = {
'userid': self.userid,
'userkey': self.userkey,
'action': action,
'source': source,
'notify': notify,
'instant': instant,
}
query = self._build_tree(etree.Element('query'), nodes)
for format in formats:
format_node = self._build_tree(etree.Element('format'), format)
query.append(format_node)
results = self._execute_request(query, headers)
return self._parse_results(results)
def _build_tree(self, node, data):
for k, v in data.items():
if isinstance(v, list):
for item in v:
element = etree.Element(k)
element.text = item
node.append(element)
else:
element = etree.Element(k)
element.text = v
node.append(element)
return node
def _execute_request(self, xml, headers, path='', method='POST'):
params = urllib.urlencode({'xml': etree.tostring(xml)})
conn = httplib.HTTPConnection(self.url)
conn.request(method, path, params, headers)
response = conn.getresponse()
data = response.read()
conn.close()
return data
def _parse_results(self, results):
try:
return xmltodict.parse(results)
except ExpatError, e:
print 'Error parsing encoding.com response'
print e
return None
其他待完成事項包括通過HTTPS-only (加密聯(lián)機) 使用Encoding.com 嚴謹?shù)腟SL驗證，還有一些單元測試。
階段4：下載所有新的視頻檔格式
我們有個定期執(zhí)行的程序，通過RabbitMQ每15秒檢查視頻轉碼的進度：
class CheckUpOnThirdParties(PeriodicTask):
run_every = timedelta(seconds=settings.THIRD_PARTY_CHECK_UP_INTERVAL)
...
def _handle_encoding_com(self, slides):
format_lookup = {
'mp4': SlideVideoMedia.FORMAT_MP4,
'webm': SlideVideoMedia.FORMAT_WEBM,
'ogg': SlideVideoMedia.FORMAT_OGG,
'fl9': SlideVideoMedia.FORMAT_FL9,
'thumbnail': SlideVideoMedia.FORMAT_THUMB,
}
encoder = Encoding(settings.ENCODING_API_USER_ID,
settings.ENCODING_API_USER_KEY)
job_ids = [item.encoding_com_tracking_code for item in slides]
resp = encoder.get_status(ids=job_ids)
if resp is None:
log.error('Unable to check up on encoding.com')
return False
檢查Encoding.com的響應來驗證每個部分是否正確以利我們繼續(xù)下去。
if resp.get('response') is None:
log.error('Unable to get response node from encoding.com')
return False
resp_id = resp.get('response').get('id')
if resp_id is None:
log.error('Unable to get media id from encoding.com')
return False
slide = SlideUploadQueue.objects.filter(
status=SlideUploadQueue.STATUS_AWAITING_3RD_PARTY_PROCESSING,
encoding_com_tracking_code=resp_id)
if len(slide) != 1:
log.error('Unable to find a single record for %s' % resp_id)
return False
resp_status = resp.get('response').get('status')
if resp_status is None:
log.error('Unable to get status from encoding.com')
return False
if resp_status != u'Finished':
log.debug("%s isn't finished, will check back later" % resp_id)
return True
formats = resp.get('response').get('format')
if formats is None:
log.error("No output formats were found. Something's wrong.")
return False
for format in formats:
try:
assert format.get('status') == u'Finished', \
"%s is not finished. Something's wrong." % format.get('id')
output = format.get('output')
assert output in ('mp4', 'webm', 'ogg', 'fl9',
'thumbnail'), 'Unknown output format %s' % output
s3_dest = format.get('s3_destination')
assert 'http://encoding.com.result.s3.amazonaws.com/'\
in s3_dest, 'Suspicious S3 url: %s' % s3_dest
https_link = \
'https://s3.amazonaws.com/encoding.com.result/%s' %\
s3_dest.split('/')[-1]
file_ext = https_link.split('.')[-1].strip()
assert len(file_ext) > 0,\
'Unable to get file extension from %s' % https_link
count = SlideVideoMedia.objects.filter(slide_upload=slide,
format=format_lookup[output]).count()
if count != 0:
print 'There is already a %s file for this slide' % output
continue
content = self.download_content(https_link)
assert content is not None,\
'There is no content for %s' % format.get('id')
except AssertionError, e:
log.error('A format did not pass all assertions: %s' % e)
continue
到這里我們已確認所有事項皆正常，所以我們可以儲存所有的視頻檔了：
media = SlideVideoMedia()
media.format = format_lookup[output]
media.converted_file.save('blah.%s' % file_ext, ContentFile(content))
media.save()
階段5：經(jīng)由HTML5播放視頻檔
在我們的前端網(wǎng)頁已經(jīng)新增了一個有HTML5的影像單元的網(wǎng)頁。并采用對每個瀏覽器都有最佳支持的video.js來顯示視頻。
? bower install video.js bower caching git://github.com/videojs/video.js-component.git bower cloning git://github.com/videojs/video.js-component.git bower fetching video.js bower checking out video.js#v4.0.3 bower copying /home/mark/.bower/cache/video.js/5ab058cd60c5615aa38e8e706cd0f307 bower installing video.js#4.0.3
在我們的首頁有包含其他相依的文件：
!!! 5 html(lang="en", class="no-js") head meta(http-equiv='Content-Type', content='text/html; charset=UTF-8') ... link(rel='stylesheet', type='text/css', href='/components/video-js-4.1.0/video-js.css') script(type='text/javascript', src='/components/video-js-4.1.0/video.js')
在Angular.js/JADE-based 框架下的模塊，我們引入<video>卷標 與其<source>子卷標。每個視頻文件都會有縮圖通過<video>卷標的 poster 組件顯示，縮圖的圖像是由我們從視頻的前幾秒擷取下來。
#main.span12
video#example_video_1.video-js.vjs-default-skin(controls, preload="auto", width="640", height="264", poster="{{video_thumbnail}}", data-setup='{"example_option":true}', ng-show="videos")
source(ng-repeat="video in videos", src="{{video.src}}", type="{{video.type}}")
還會顯示出我們轉換的每個視頻文件格式，并使用在<source>標簽。Video.js 會根據(jù)使用者使用的瀏覽器決定播放哪種格式的視頻。
我們?nèi)匀挥性S多工作需要完成，建立單元測試與加強和Encoding.com服務溝通的程序。如果你對這些工作感興趣請與我連絡。
相關文章
這篇文章主要介紹了python之super的使用小結，小編覺得挺不錯的，現(xiàn)在分享給大家，也給大家做個參考。一起跟隨小編過來看看吧2018-08-08
在Linux中通過Python腳本訪問mdb數(shù)據(jù)庫的方法
這篇文章主要介紹了在Linux中通過Python腳本訪問mdb數(shù)據(jù)庫的方法,本文示例基于debian系的Linux系統(tǒng),需要的朋友可以參考下2015-05-05
源代碼目錄一般很詳細，每個代碼文件都不長，一般只占幾十行而已。在跟著書籍學習的過程中，可以打開源代碼運行之后查看效果，或者適當修改看看變化，這都是很好的掌握技術的好辦法。2011-03-03
這篇文章主要介紹了Pyqt5 關于流式布局和滾動條的綜合使用示例代碼，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧2020-03-03
這篇文章主要為大家詳細介紹了Django自定義分頁與bootstrap分頁結合使用的方法，具有一定的參考價值，感興趣的小伙伴們可以參考一下2017-05-05
今天小編就為大家分享一篇關于Numpy中的行向量和列向量詳解，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧2019-11-11
Python 如何創(chuàng)建一個簡單的REST接口
這篇文章主要介紹了Python 如何創(chuàng)建一個簡單的REST接口，文中講解非常細致，代碼幫助大家更好的理解和學習，感興趣的朋友可以了解下2020-07-07
這篇文章主要介紹了pytorch之關于PyTorch結構的使用方式,具有很好的參考價值,希望對大家有所幫助,如有錯誤或未考慮完全的地方,望不吝賜教2023-09-09
在Python中定義函數(shù)并調(diào)用的操作步驟
這篇文章主要介紹了在Python中如何定義函數(shù)并調(diào)用它,函數(shù)的定義和調(diào)用是Python編程中最基本也是最重要的概念之一,掌握它們對于進行有效的Python編程至關重要,需要的朋友可以參考下2024-01-01
Flask使用SQLAlchemy實現(xiàn)持久化數(shù)據(jù)
本文主要介紹了Flask使用SQLAlchemy實現(xiàn)持久化數(shù)據(jù)，文中通過示例代碼介紹的非常詳細，需要的朋友們下面隨著小編來一起學習學習吧2021-07-07