@ -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/ |
@ -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 | |||
``` |
@ -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()? |
@ -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() |