importer.py 6.4 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217
  1. import json
  2. import os
  3. from tempfile import TemporaryDirectory
  4. from zipfile import BadZipFile, ZipFile
  5. from django.core.files.uploadedfile import UploadedFile
  6. from django.utils.translation import gettext as _, gettext_lazy
  7. from ..models import Theme
  8. from .css import create_css
  9. from .forms import (
  10. ThemeCssUrlManifest,
  11. ThemeManifest,
  12. create_css_file_manifest,
  13. create_media_file_manifest,
  14. )
  15. from .media import create_media
  16. from .tasks import build_theme_css, update_remote_css_size
  17. INVALID_MANIFEST_ERROR = gettext_lazy(
  18. '"manifest.json" contained by ZIP file is not a valid theme manifest file.'
  19. )
  20. class ThemeImportError(BaseException):
  21. pass
  22. class InvalidThemeManifest(ThemeImportError):
  23. def __init__(self):
  24. super().__init__(INVALID_MANIFEST_ERROR)
  25. def import_theme(name, parent, zipfile):
  26. with TemporaryDirectory() as tmp_dir:
  27. extract_zipfile_to_tmp_dir(zipfile, tmp_dir)
  28. validate_zipfile_contains_single_directory(tmp_dir)
  29. theme_dir = os.path.join(tmp_dir, os.listdir(tmp_dir)[0])
  30. cleaned_manifest = clean_theme_contents(theme_dir)
  31. theme = create_theme_from_manifest(name, parent, cleaned_manifest)
  32. try:
  33. create_css_from_manifest(theme_dir, theme, cleaned_manifest["css"])
  34. create_media_from_manifest(theme_dir, theme, cleaned_manifest["media"])
  35. except Exception:
  36. theme.delete()
  37. raise InvalidThemeManifest()
  38. else:
  39. for css in theme.css.filter(url__isnull=False):
  40. update_remote_css_size.delay(css.pk)
  41. build_theme_css.delay(theme.pk)
  42. return theme
  43. def extract_zipfile_to_tmp_dir(zipfile, tmp_dir):
  44. try:
  45. ZipFile(zipfile).extractall(tmp_dir)
  46. except BadZipFile:
  47. raise ThemeImportError(_("Uploaded ZIP file could not be extracted."))
  48. def validate_zipfile_contains_single_directory(tmp_dir):
  49. dir_contents = os.listdir(tmp_dir)
  50. if not dir_contents:
  51. raise ThemeImportError(_("Uploaded ZIP file is empty."))
  52. if len(dir_contents) > 1:
  53. raise ThemeImportError(_("Uploaded ZIP file should contain single directory."))
  54. if not os.path.isdir(os.path.join(tmp_dir, dir_contents[0])):
  55. raise ThemeImportError(_("Uploaded ZIP file didn't contain a directory."))
  56. def clean_theme_contents(theme_dir):
  57. manifest = read_manifest(theme_dir)
  58. return clean_manifest(theme_dir, manifest)
  59. def read_manifest(theme_dir):
  60. try:
  61. with open(os.path.join(theme_dir, "manifest.json")) as fp:
  62. return json.load(fp)
  63. except FileNotFoundError:
  64. raise ThemeImportError(
  65. _('Uploaded ZIP file didn\'t contain a "manifest.json".')
  66. )
  67. except json.decoder.JSONDecodeError:
  68. raise ThemeImportError(
  69. _('"manifest.json" contained by ZIP file is not a valid JSON file.')
  70. )
  71. def clean_manifest(theme_dir, manifest):
  72. if not isinstance(manifest, dict):
  73. raise InvalidThemeManifest()
  74. form = ThemeManifest(manifest)
  75. if not form.is_valid():
  76. raise InvalidThemeManifest()
  77. cleaned_manifest = form.cleaned_data.copy()
  78. cleaned_manifest["css"] = clean_css_list(theme_dir, manifest.get("css"))
  79. cleaned_manifest["media"] = clean_media_list(theme_dir, manifest.get("media"))
  80. return cleaned_manifest
  81. def clean_css_list(theme_dir, manifest):
  82. if not isinstance(manifest, list):
  83. raise InvalidThemeManifest()
  84. theme_css_dir = os.path.join(theme_dir, "css")
  85. if not os.path.isdir(theme_css_dir):
  86. raise InvalidThemeManifest()
  87. cleaned_data = []
  88. for item in manifest:
  89. cleaned_data.append(clean_css_list_item(theme_css_dir, item))
  90. return cleaned_data
  91. def clean_css_list_item(theme_css_dir, item):
  92. if not isinstance(item, dict):
  93. raise InvalidThemeManifest()
  94. if item.get("url"):
  95. return clean_css_url(item)
  96. if item.get("path"):
  97. return clean_css_file(theme_css_dir, item)
  98. raise InvalidThemeManifest()
  99. def clean_css_url(data):
  100. form = ThemeCssUrlManifest(data)
  101. if not form.is_valid():
  102. raise InvalidThemeManifest()
  103. return form.cleaned_data
  104. def clean_css_file(theme_css_dir, data):
  105. file_manifest = create_css_file_manifest(theme_css_dir)
  106. if data.get("path"):
  107. data["path"] = os.path.join(theme_css_dir, str(data["path"]))
  108. form = file_manifest(data)
  109. if not form.is_valid():
  110. raise InvalidThemeManifest()
  111. return form.cleaned_data
  112. def clean_media_list(theme_dir, manifest):
  113. if not isinstance(manifest, list):
  114. raise InvalidThemeManifest()
  115. theme_media_dir = os.path.join(theme_dir, "media")
  116. if not os.path.isdir(theme_media_dir):
  117. raise InvalidThemeManifest()
  118. cleaned_data = []
  119. for item in manifest:
  120. cleaned_data.append(clean_media_list_item(theme_media_dir, item))
  121. return cleaned_data
  122. def clean_media_list_item(theme_media_dir, data):
  123. if not isinstance(data, dict):
  124. raise InvalidThemeManifest()
  125. file_manifest = create_media_file_manifest(theme_media_dir)
  126. if data.get("path"):
  127. data["path"] = os.path.join(theme_media_dir, str(data["path"]))
  128. form = file_manifest(data)
  129. if not form.is_valid():
  130. raise InvalidThemeManifest()
  131. return form.cleaned_data
  132. def create_theme_from_manifest(name, parent, cleaned_data):
  133. return Theme.objects.create(
  134. name=name or cleaned_data["name"],
  135. parent=parent,
  136. version=cleaned_data.get("version") or None,
  137. author=cleaned_data.get("author") or None,
  138. url=cleaned_data.get("url") or None,
  139. )
  140. def create_css_from_manifest(tmp_dir, theme, manifest):
  141. for item in manifest:
  142. if "url" in item:
  143. theme.css.create(**item, order=theme.css.count())
  144. else:
  145. with open(item["path"], "rb") as fp:
  146. file_obj = UploadedFile(
  147. fp,
  148. name=item["name"],
  149. content_type="text/css",
  150. size=os.path.getsize(item["path"]),
  151. )
  152. create_css(theme, file_obj)
  153. def create_media_from_manifest(tmp_dir, theme, manifest):
  154. for item in manifest:
  155. with open(item["path"], "rb") as fp:
  156. file_obj = UploadedFile(
  157. fp,
  158. name=item["name"],
  159. content_type=item["type"],
  160. size=os.path.getsize(item["path"]),
  161. )
  162. create_media(theme, file_obj)