467 lines
14 KiB
Python
467 lines
14 KiB
Python
import os
|
|
import time
|
|
|
|
from django.contrib.auth.models import User
|
|
from django.contrib.gis.db import models
|
|
from django.conf import settings
|
|
from django.urls import reverse
|
|
from django.utils.translation import gettext_lazy as _
|
|
from django.core.signing import Signer
|
|
from django.template.defaultfilters import slugify
|
|
from django.core.files.base import File
|
|
|
|
from .managers import PublicManager
|
|
|
|
|
|
# Did not find a clean way to do this in Django
|
|
# - creating a Proxy model would mean replacing get_user_model by this proxy model
|
|
# in every template
|
|
# - extending User model woulc mean a non trivial migration
|
|
def display_name(self):
|
|
return settings.USER_DISPLAY_NAME.format(**self.__dict__)
|
|
|
|
|
|
def get_user_url(self):
|
|
identifier = getattr(self, settings.USER_URL_FIELD)
|
|
return reverse(settings.USER_MAPS_URL, kwargs={"identifier": identifier})
|
|
|
|
|
|
def get_user_stars_url(self):
|
|
identifier = getattr(self, settings.USER_URL_FIELD)
|
|
return reverse("user_stars", kwargs={"identifier": identifier})
|
|
|
|
|
|
User.add_to_class("__str__", display_name)
|
|
User.add_to_class("get_url", get_user_url)
|
|
User.add_to_class("get_stars_url", get_user_stars_url)
|
|
|
|
|
|
def get_default_share_status():
|
|
return settings.UMAP_DEFAULT_SHARE_STATUS or Map.PUBLIC
|
|
|
|
|
|
def get_default_edit_status():
|
|
return settings.UMAP_DEFAULT_EDIT_STATUS or Map.OWNER
|
|
|
|
|
|
class NamedModel(models.Model):
|
|
name = models.CharField(max_length=200, verbose_name=_("name"))
|
|
|
|
class Meta:
|
|
abstract = True
|
|
ordering = ("name",)
|
|
|
|
def __unicode__(self):
|
|
return self.name
|
|
|
|
def __str__(self):
|
|
return self.name
|
|
|
|
|
|
def get_default_licence():
|
|
"""
|
|
Returns a default Licence, creates it if it doesn't exist.
|
|
Needed to prevent a licence deletion from deleting all the linked
|
|
maps.
|
|
"""
|
|
return Licence.objects.get_or_create(
|
|
# can't use ugettext_lazy for database storage, see #13965
|
|
name=getattr(settings, "UMAP_DEFAULT_LICENCE_NAME", "No licence set")
|
|
)[0]
|
|
|
|
|
|
class Licence(NamedModel):
|
|
"""
|
|
The licence one map is published on.
|
|
"""
|
|
|
|
details = models.URLField(
|
|
verbose_name=_("details"),
|
|
help_text=_("Link to a page where the licence is detailed."),
|
|
)
|
|
|
|
@property
|
|
def json(self):
|
|
return {"name": self.name, "url": self.details}
|
|
|
|
|
|
class TileLayer(NamedModel):
|
|
url_template = models.CharField(
|
|
max_length=200, help_text=_("URL template using OSM tile format")
|
|
)
|
|
minZoom = models.IntegerField(default=0)
|
|
maxZoom = models.IntegerField(default=18)
|
|
attribution = models.CharField(max_length=300)
|
|
rank = models.SmallIntegerField(
|
|
blank=True, null=True, help_text=_("Order of the tilelayers in the edit box")
|
|
)
|
|
# See https://wiki.openstreetmap.org/wiki/TMS#The_Y_coordinate
|
|
tms = models.BooleanField(default=False)
|
|
|
|
@property
|
|
def json(self):
|
|
return dict(
|
|
(field.name, getattr(self, field.name)) for field in self._meta.fields
|
|
)
|
|
|
|
@classmethod
|
|
def get_default(cls):
|
|
"""
|
|
Returns the default tile layer (used for a map when no layer is set).
|
|
"""
|
|
return cls.objects.order_by("rank")[0] # FIXME, make it administrable
|
|
|
|
@classmethod
|
|
def get_list(cls):
|
|
default = cls.get_default()
|
|
l = []
|
|
for t in cls.objects.all():
|
|
fields = t.json
|
|
if default and default.pk == t.pk:
|
|
fields["selected"] = True
|
|
l.append(fields)
|
|
return l
|
|
|
|
class Meta:
|
|
ordering = ("rank", "name")
|
|
|
|
|
|
class Map(NamedModel):
|
|
"""
|
|
A single thematical map.
|
|
"""
|
|
|
|
ANONYMOUS = 1
|
|
EDITORS = 2
|
|
OWNER = 3
|
|
PUBLIC = 1
|
|
OPEN = 2
|
|
PRIVATE = 3
|
|
BLOCKED = 9
|
|
EDIT_STATUS = (
|
|
(ANONYMOUS, _("Everyone")),
|
|
(EDITORS, _("Editors only")),
|
|
(OWNER, _("Owner only")),
|
|
)
|
|
SHARE_STATUS = (
|
|
(PUBLIC, _("Everyone (public)")),
|
|
(OPEN, _("Anyone with link")),
|
|
(PRIVATE, _("Editors only")),
|
|
(BLOCKED, _("Blocked")),
|
|
)
|
|
slug = models.SlugField(db_index=True)
|
|
description = models.TextField(blank=True, null=True, verbose_name=_("description"))
|
|
center = models.PointField(geography=True, verbose_name=_("center"))
|
|
zoom = models.IntegerField(default=7, verbose_name=_("zoom"))
|
|
locate = models.BooleanField(
|
|
default=False, verbose_name=_("locate"), help_text=_("Locate user on load?")
|
|
)
|
|
licence = models.ForeignKey(
|
|
Licence,
|
|
help_text=_("Choose the map licence."),
|
|
verbose_name=_("licence"),
|
|
on_delete=models.SET_DEFAULT,
|
|
default=get_default_licence,
|
|
)
|
|
created_at = models.DateTimeField(auto_now_add=True)
|
|
modified_at = models.DateTimeField(auto_now=True)
|
|
owner = models.ForeignKey(
|
|
settings.AUTH_USER_MODEL,
|
|
blank=True,
|
|
null=True,
|
|
related_name="owned_maps",
|
|
verbose_name=_("owner"),
|
|
on_delete=models.PROTECT,
|
|
)
|
|
editors = models.ManyToManyField(
|
|
settings.AUTH_USER_MODEL, blank=True, verbose_name=_("editors")
|
|
)
|
|
edit_status = models.SmallIntegerField(
|
|
choices=EDIT_STATUS,
|
|
default=get_default_edit_status,
|
|
verbose_name=_("edit status"),
|
|
)
|
|
share_status = models.SmallIntegerField(
|
|
choices=SHARE_STATUS,
|
|
default=get_default_share_status,
|
|
verbose_name=_("share status"),
|
|
)
|
|
settings = models.JSONField(
|
|
blank=True, null=True, verbose_name=_("settings"), default=dict
|
|
)
|
|
|
|
objects = models.Manager()
|
|
public = PublicManager()
|
|
|
|
def get_absolute_url(self):
|
|
return reverse("map", kwargs={"slug": self.slug or "map", "map_id": self.pk})
|
|
|
|
def get_anonymous_edit_url(self):
|
|
signer = Signer()
|
|
signature = signer.sign(self.pk)
|
|
path = reverse("map_anonymous_edit_url", kwargs={"signature": signature})
|
|
return settings.SITE_URL + path
|
|
|
|
def is_anonymous_owner(self, request):
|
|
if not request or self.owner:
|
|
# edit cookies are only valid while map hasn't owner
|
|
return False
|
|
key, value = self.signed_cookie_elements
|
|
try:
|
|
has_anonymous_cookie = int(request.get_signed_cookie(key, False)) == value
|
|
except ValueError:
|
|
has_anonymous_cookie = False
|
|
return has_anonymous_cookie
|
|
|
|
def can_edit(self, user=None, request=None):
|
|
"""
|
|
Define if a user can edit or not the instance, according to his account
|
|
or the request.
|
|
|
|
In owner mode:
|
|
- only owner by default (OWNER)
|
|
- any editor if mode is EDITORS
|
|
- anyone otherwise (ANONYMOUS)
|
|
In anonymous owner mode:
|
|
- only owner (has ownership cookie) by default (OWNER)
|
|
- anyone otherwise (ANONYMOUS)
|
|
"""
|
|
can = False
|
|
if request and not self.owner:
|
|
if settings.UMAP_ALLOW_ANONYMOUS and self.is_anonymous_owner(request):
|
|
can = True
|
|
if self.edit_status == self.ANONYMOUS:
|
|
can = True
|
|
elif user is None:
|
|
can = False
|
|
elif user == self.owner:
|
|
can = True
|
|
elif self.edit_status == self.EDITORS and user in self.editors.all():
|
|
can = True
|
|
return can
|
|
|
|
def can_view(self, request):
|
|
if self.share_status == self.BLOCKED:
|
|
can = False
|
|
elif self.owner is None:
|
|
can = True
|
|
elif self.share_status in [self.PUBLIC, self.OPEN]:
|
|
can = True
|
|
elif request.user == self.owner:
|
|
can = True
|
|
else:
|
|
can = not (
|
|
self.share_status == self.PRIVATE
|
|
and request.user not in self.editors.all()
|
|
)
|
|
return can
|
|
|
|
@property
|
|
def signed_cookie_elements(self):
|
|
return ("anonymous_owner|%s" % self.pk, self.pk)
|
|
|
|
def get_tilelayer(self):
|
|
return self.tilelayer or TileLayer.get_default()
|
|
|
|
def clone(self, **kwargs):
|
|
new = self.__class__.objects.get(pk=self.pk)
|
|
new.pk = None
|
|
new.name = "%s %s" % (_("Clone of"), self.name)
|
|
if "owner" in kwargs:
|
|
# can be None in case of anonymous cloning
|
|
new.owner = kwargs["owner"]
|
|
new.save()
|
|
for editor in self.editors.all():
|
|
new.editors.add(editor)
|
|
for datalayer in self.datalayer_set.all():
|
|
datalayer.clone(map_inst=new)
|
|
return new
|
|
|
|
|
|
class Pictogram(NamedModel):
|
|
"""
|
|
An image added to an icon of the map.
|
|
"""
|
|
|
|
attribution = models.CharField(max_length=300)
|
|
category = models.CharField(max_length=300, null=True, blank=True)
|
|
pictogram = models.FileField(upload_to="pictogram")
|
|
|
|
@property
|
|
def json(self):
|
|
return {
|
|
"id": self.pk,
|
|
"attribution": self.attribution,
|
|
"name": self.name,
|
|
"category": self.category,
|
|
"src": self.pictogram.url,
|
|
}
|
|
|
|
|
|
# Must be out of Datalayer for Django migration to run, because of python 2
|
|
# serialize limitations.
|
|
def upload_to(instance, filename):
|
|
if instance.pk:
|
|
return instance.upload_to()
|
|
name = "%s.geojson" % slugify(instance.name)[:50] or "untitled"
|
|
return os.path.join(instance.storage_root(), name)
|
|
|
|
|
|
class DataLayer(NamedModel):
|
|
"""
|
|
Layer to store Features in.
|
|
"""
|
|
|
|
INHERIT = 0
|
|
ANONYMOUS = 1
|
|
EDITORS = 2
|
|
OWNER = 3
|
|
EDIT_STATUS = (
|
|
(INHERIT, _("Inherit")),
|
|
(ANONYMOUS, _("Everyone")),
|
|
(EDITORS, _("Editors only")),
|
|
(OWNER, _("Owner only")),
|
|
)
|
|
|
|
map = models.ForeignKey(Map, on_delete=models.CASCADE)
|
|
description = models.TextField(blank=True, null=True, verbose_name=_("description"))
|
|
geojson = models.FileField(upload_to=upload_to, blank=True, null=True)
|
|
display_on_load = models.BooleanField(
|
|
default=False,
|
|
verbose_name=_("display on load"),
|
|
help_text=_("Display this layer on load."),
|
|
)
|
|
rank = models.SmallIntegerField(default=0)
|
|
settings = models.JSONField(
|
|
blank=True, null=True, verbose_name=_("settings"), default=dict
|
|
)
|
|
edit_status = models.SmallIntegerField(
|
|
choices=EDIT_STATUS,
|
|
default=INHERIT,
|
|
verbose_name=_("edit status"),
|
|
)
|
|
|
|
class Meta:
|
|
ordering = ("rank",)
|
|
|
|
def save(self, force_insert=False, force_update=False, **kwargs):
|
|
is_new = not bool(self.pk)
|
|
super(DataLayer, self).save(force_insert, force_update, **kwargs)
|
|
|
|
if is_new:
|
|
force_insert, force_update = False, True
|
|
filename = self.upload_to()
|
|
old_name = self.geojson.name
|
|
new_name = self.geojson.storage.save(filename, self.geojson)
|
|
self.geojson.storage.delete(old_name)
|
|
self.geojson.name = new_name
|
|
super(DataLayer, self).save(force_insert, force_update, **kwargs)
|
|
self.purge_gzip()
|
|
self.purge_old_versions()
|
|
|
|
def upload_to(self):
|
|
root = self.storage_root()
|
|
name = "%s_%s.geojson" % (self.pk, int(time.time() * 1000))
|
|
return os.path.join(root, name)
|
|
|
|
def storage_root(self):
|
|
path = ["datalayer", str(self.map.pk)[-1]]
|
|
if len(str(self.map.pk)) > 1:
|
|
path.append(str(self.map.pk)[-2])
|
|
path.append(str(self.map.pk))
|
|
return os.path.join(*path)
|
|
|
|
def metadata(self, user=None, request=None):
|
|
# Retrocompat: minimal settings for maps not saved after settings property
|
|
# has been introduced
|
|
obj = self.settings or {
|
|
"name": self.name,
|
|
"displayOnLoad": self.display_on_load,
|
|
}
|
|
obj["id"] = self.pk
|
|
obj["permissions"] = {"edit_status": self.edit_status}
|
|
obj["editMode"] = "advanced" if self.can_edit(user, request) else 'disabled'
|
|
return obj
|
|
|
|
def clone(self, map_inst=None):
|
|
new = self.__class__.objects.get(pk=self.pk)
|
|
new.pk = None
|
|
if map_inst:
|
|
new.map = map_inst
|
|
new.geojson = File(new.geojson.file.file)
|
|
new.save()
|
|
return new
|
|
|
|
def is_valid_version(self, name):
|
|
return name.startswith("%s_" % self.pk) and name.endswith(".geojson")
|
|
|
|
def version_metadata(self, name):
|
|
els = name.split(".")[0].split("_")
|
|
return {
|
|
"name": name,
|
|
"at": els[1],
|
|
"size": self.geojson.storage.size(self.get_version_path(name)),
|
|
}
|
|
|
|
def get_versions(self):
|
|
root = self.storage_root()
|
|
names = self.geojson.storage.listdir(root)[1]
|
|
names = [name for name in names if self.is_valid_version(name)]
|
|
names.sort(reverse=True) # Recent first.
|
|
return names
|
|
|
|
@property
|
|
def versions(self):
|
|
names = self.get_versions()
|
|
return [self.version_metadata(name) for name in names]
|
|
|
|
def get_version(self, name):
|
|
path = self.get_version_path(name)
|
|
with self.geojson.storage.open(path, "r") as f:
|
|
return f.read()
|
|
|
|
def get_version_path(self, name):
|
|
return "{root}/{name}".format(root=self.storage_root(), name=name)
|
|
|
|
def purge_old_versions(self):
|
|
root = self.storage_root()
|
|
names = self.get_versions()[settings.UMAP_KEEP_VERSIONS :]
|
|
for name in names:
|
|
try:
|
|
self.geojson.storage.delete(os.path.join(root, name))
|
|
except FileNotFoundError:
|
|
pass
|
|
|
|
def purge_gzip(self):
|
|
root = self.storage_root()
|
|
names = self.geojson.storage.listdir(root)[1]
|
|
for name in names:
|
|
if name.startswith(f'{self.pk}_') and name.endswith(".gz"):
|
|
self.geojson.storage.delete(os.path.join(root, name))
|
|
|
|
def can_edit(self, user=None, request=None):
|
|
"""
|
|
Define if a user can edit or not the instance, according to his account
|
|
or the request.
|
|
"""
|
|
if self.edit_status == self.INHERIT:
|
|
return self.map.can_edit(user, request)
|
|
can = False
|
|
if not self.map.owner:
|
|
if settings.UMAP_ALLOW_ANONYMOUS and self.map.is_anonymous_owner(request):
|
|
can = True
|
|
if self.edit_status == self.ANONYMOUS:
|
|
can = True
|
|
elif user is not None and user == self.map.owner:
|
|
can = True
|
|
elif self.edit_status == self.EDITORS and user in self.map.editors.all():
|
|
can = True
|
|
return can
|
|
|
|
|
|
class Star(models.Model):
|
|
at = models.DateTimeField(auto_now=True)
|
|
map = models.ForeignKey(Map, on_delete=models.CASCADE)
|
|
by = models.ForeignKey(
|
|
settings.AUTH_USER_MODEL, related_name="stars", on_delete=models.CASCADE
|
|
)
|