commit c4e5bbb910c3296d916adf7227c2c2b0baad4f49 Author: Zykino Date: Sun Aug 30 23:47:04 2020 +0200 First commit with early library diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..ecdedf9 --- /dev/null +++ b/.gitignore @@ -0,0 +1,140 @@ +# Copied gitignore file from https://github.com/github/gitignore/blob/master/Python.gitignore + +# Byte-compiled / optimized / DLL files +__pycache__/ +*.py[cod] +*$py.class + +# C extensions +*.so + +# Distribution / packaging +.Python +build/ +develop-eggs/ +dist/ +downloads/ +eggs/ +.eggs/ +lib/ +lib64/ +parts/ +sdist/ +var/ +wheels/ +share/python-wheels/ +*.egg-info/ +.installed.cfg +*.egg +MANIFEST + +# PyInstaller +# Usually these files are written by a python script from a template +# before PyInstaller builds the exe, so as to inject date/other infos into it. +*.manifest +*.spec + +# Installer logs +pip-log.txt +pip-delete-this-directory.txt + +# Unit test / coverage reports +htmlcov/ +.tox/ +.nox/ +.coverage +.coverage.* +.cache +nosetests.xml +coverage.xml +*.cover +*.py,cover +.hypothesis/ +.pytest_cache/ +cover/ + +# Translations +*.mo +*.pot + +# Django stuff: +*.log +local_settings.py +db.sqlite3 +db.sqlite3-journal + +# Flask stuff: +instance/ +.webassets-cache + +# Scrapy stuff: +.scrapy + +# Sphinx documentation +docs/_build/ + +# PyBuilder +.pybuilder/ +target/ + +# Jupyter Notebook +.ipynb_checkpoints + +# IPython +profile_default/ +ipython_config.py + +# pyenv +# For a library or package, you might want to ignore these files since the code is +# intended to run in multiple environments; otherwise, check them in: +# .python-version + +# pipenv +# According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control. +# However, in case of collaboration, if having platform-specific dependencies or dependencies +# having no cross-platform support, pipenv may install dependencies that don't work, or not +# install all needed dependencies. +#Pipfile.lock + +# PEP 582; used by e.g. github.com/David-OConnor/pyflow +__pypackages__/ + +# Celery stuff +celerybeat-schedule +celerybeat.pid + +# SageMath parsed files +*.sage.py + +# Environments +.env +.venv +env/ +venv/ +ENV/ +env.bak/ +venv.bak/ + +# Spyder project settings +.spyderproject +.spyproject + +# Rope project settings +.ropeproject + +# mkdocs documentation +/site + +# mypy +.mypy_cache/ +.dmypy.json +dmypy.json + +# Pyre type checker +.pyre/ + +# pytype static type analyzer +.pytype/ + +# Cython debug symbols +cython_debug/ diff --git a/README.md b/README.md new file mode 100644 index 0000000..618ad5a --- /dev/null +++ b/README.md @@ -0,0 +1,19 @@ +# Prismedia AutoUpload +This the implementation for the file handling of prismedia's autoupload feature. + +It consist of one file: `autoupload.py`. There is also a test file as `autoupload_test.py` which can be helpfull has to how to use this object. + +I am not a python expert and everything may change when integrating this project in Prismedia: function/variable name/case, interface, ... + +# Dependencies +```sh +pip3 install toml +``` + +# Tests +Since there is no tests done for Prismedia, I choose `unittest` to do tests for this lib since it is present in the python standard library. + +Launch tests with +```sh +python3 autoupload_test.py +``` diff --git a/autoupload.py b/autoupload.py new file mode 100644 index 0000000..6cfffcb --- /dev/null +++ b/autoupload.py @@ -0,0 +1,72 @@ +#!/usr/bin/env python3 + +import toml +import pathlib +from datetime import datetime + +class AutoUpload(object): + """AutoUpload is a class handling the state of videos automatically uploaded by prismedia""" + + _currentVideo = None + _videoSuffix = ".mp4" + + _urlSuffix = "-url" + _errorSuffix = "-error" + _publishSuffix = "-publish" + _lastUpdateTimeKey = "update-time" + + def __init__(self, autouploadFile): + super(AutoUpload, self).__init__() + + self._autouploadFile = autouploadFile + self._basePath = pathlib.Path(autouploadFile).resolve().parent + self._autoUploadConfig = toml.load(autouploadFile) + + # print("content of file") + # with open(autouploadFile, "r") as file: + # print(file.readlines()) + + # TODO: remove me / should only be here for debug + # print("setting autouplad config") + # print(self._autouploadFile) + # print(self._autoUploadConfig) + + def nextVideo(self, platform): + """Get the path to the next video to upload for a specific platform""" + for video in self._autoUploadConfig["videos"]: + if platform + self._urlSuffix not in self._autoUploadConfig["videos"][video]: + self._currentVideo = video + return (self._basePath / video).with_suffix(self._videoSuffix).resolve().as_posix() + + self._currentVideo = None + return False + + def success(self, platform, url, publishDate): + """Last video asked was successfully uploaded""" + updateTime = datetime.today() + + self._autoUploadConfig["videos"][self._currentVideo].pop(platform + self._errorSuffix, None) + + + self._autoUploadConfig["videos"][self._currentVideo][platform + self._urlSuffix] = url + self._autoUploadConfig["videos"][self._currentVideo][platform + self._publishSuffix] = publishDate + + self._autoUploadConfig["videos"][self._currentVideo][self._lastUpdateTimeKey] = updateTime + + self._write() + + def error(self, platform, errorString): + """There was an error on upload of last video""" + updateTime = datetime.today() + + self._autoUploadConfig["videos"][self._currentVideo][platform + self._errorSuffix] = errorString + + self._autoUploadConfig["videos"][self._currentVideo][self._lastUpdateTimeKey] = updateTime + + self._write() + + def _write(self): + """Private helper function + Write current autoupload state on file(s)""" + with open(self._autouploadFile, 'w', encoding="utf-8", errors="strict") as f: + toml.dump(self._autoUploadConfig, f, encoder=toml.TomlPreserveInlineDictEncoder()) # can we also inherit from toml.TomlPreserveCommentEncoder()? diff --git a/autoupload_test.py b/autoupload_test.py new file mode 100644 index 0000000..34ae9fd --- /dev/null +++ b/autoupload_test.py @@ -0,0 +1,110 @@ +#!/usr/bin/env python3 + +import autoupload +from datetime import datetime +import os +import tempfile +import unittest + +class TestVideoMethods(unittest.TestCase): + + platform1 = "platform" + platform2 = "otherplatform" + + def test_nextVideo(self): + with TestFileContent( +"""[videos] +Episode1 = {} +""" + ) as videoFile: + auto = autoupload.AutoUpload(videoFile.filename) + + self.assertEqual(auto.nextVideo(self.platform1), "/tmp/Episode1.mp4") + self.assertEqual(auto.nextVideo(self.platform2), "/tmp/Episode1.mp4") + + def test_success(self): + with TestFileContent( +"""[videos] +Episode1 = {} +""" + ) as videoFile: + auto = autoupload.AutoUpload(videoFile.filename) + + auto.nextVideo(self.platform1) + auto.success(self.platform1, "https://platform/url", datetime(2020, 8, 28, 17, 54, 31)) + self.assertEqual(auto.nextVideo(self.platform1), False) + self.assertEqual(auto.nextVideo(self.platform2), "/tmp/Episode1.mp4") + # TODO: read file and check formatting and content? + + # print("content of file") + # with open(videoFile.filename, "r") as file: + # print(file.readlines()) + + def test_error(self): + with TestFileContent( +"""[videos] +Episode1 = {} +""" + ) as videoFile: + auto = autoupload.AutoUpload(videoFile.filename) + + auto.nextVideo(self.platform1) + auto.error(self.platform1, "https://platform/url") + self.assertEqual(auto.nextVideo(self.platform1), "/tmp/Episode1.mp4") + self.assertEqual(auto.nextVideo(self.platform2), "/tmp/Episode1.mp4") + # TODO: read file and check formatting and content? + + # print("content of file") + # with open(videoFile.filename, "r") as file: + # print(file.readlines()) + + def test_errorThenSuccess(self): + with TestFileContent( +"""[videos] +Episode1 = {} +""" + ) as videoFile: + auto = autoupload.AutoUpload(videoFile.filename) + + auto.nextVideo(self.platform1) + auto.error(self.platform1, "https://platform/url") + self.assertEqual(auto.nextVideo(self.platform1), "/tmp/Episode1.mp4") + self.assertEqual(auto.nextVideo(self.platform2), "/tmp/Episode1.mp4") + # TODO: read file and check formatting and content? + + print("content of file") + with open(videoFile.filename, "r") as file: + print(file.readlines()) + + + auto.success(self.platform1, "https://platform/url", datetime(2020, 8, 28, 17, 54, 31)) + self.assertEqual(auto.nextVideo(self.platform1), False) + self.assertEqual(auto.nextVideo(self.platform2), "/tmp/Episode1.mp4") + # TODO: read file and check formatting and content? + # TODO: check the key "platform-error" is not present after a success + + print("content of file") + with open(videoFile.filename, "r") as file: + print(file.readlines()) + +# https://stackoverflow.com/a/54053967 +class TestFileContent: + def __init__(self, content): + + self.file = tempfile.NamedTemporaryFile(mode="w", delete=False) + + with self.file as f: + f.write(content) + + @property + def filename(self): + return self.file.name + + def __enter__(self): + return self + + def __exit__(self, type, value, traceback): + os.unlink(self.filename) + +if __name__ == '__main__': + unittest.main()