#!/usr/bin/python3
-import PIL
-import PIL.ImageFilter
import difflib
-import io
import logging
import os.path
import re
from . import util
from .decorators import *
+INTERWIKIS = {
+ "google" : ("https://www.google.com/search?q=%(url)s", None, "fab fa-google"),
+ "rfc" : ("https://tools.ietf.org/html/rfc%(name)s", "RFC %s", None),
+ "wp" : ("https://en.wikipedia.org/wiki/%(name)s", None, "fab fa-wikipedia-w"),
+}
+
class Wiki(misc.Object):
def _get_pages(self, query, *args):
res = self.db.query(query, *args)
if res:
return Page(self.backend, res.id, data=res)
+ def make_path(self, page, path):
+ # Nothing to do for absolute links
+ if path.startswith("/"):
+ pass
+
+ # Relative links (one-level down)
+ elif path.startswith("./"):
+ path = os.path.join(page, path)
+
+ # All other relative links
+ else:
+ p = os.path.dirname(page)
+ path = os.path.join(p, path)
+
+ # Normalise links
+ return os.path.normpath(path)
+
def get_page_title(self, page, default=None):
+ # Try to retrieve title from cache
+ title = self.memcache.get("wiki:title:%s" % page)
+ if title:
+ return title
+
+ # If the title has not been in the cache, we will
+ # have to look it up
doc = self.get_page(page)
if doc:
- return doc.title
+ title = doc.title
+ else:
+ title = os.path.basename(page)
- return default or os.path.basename(page)
+ # Save in cache for forever
+ self.memcache.set("wiki:title:%s" % page, title)
+
+ return title
def get_page(self, page, revision=None):
page = Page.sanitise_page_name(page)
def get_recent_changes(self, account, limit=None):
pages = self._get_pages("SELECT * FROM wiki \
- WHERE timestamp >= NOW() - INTERVAL '4 weeks' \
ORDER BY timestamp DESC")
for page in pages:
page = self._get_page("INSERT INTO wiki(page, author_uid, markdown, changes, address) \
VALUES(%s, %s, %s, %s, %s) RETURNING *", page, author.uid, content or None, changes, address)
+ # Update cache
+ self.memcache.set("wiki:title:%s" % page.page, page.title)
+
# Send email to all watchers
page._send_watcher_emails(excludes=[author])
"""
self.db.execute("REFRESH MATERIALIZED VIEW wiki_search_index")
+ def get_watchlist(self, account):
+ pages = self._get_pages(
+ "WITH pages AS (SELECT * FROM wiki_current \
+ LEFT JOIN wiki ON wiki_current.id = wiki.id) \
+ SELECT * FROM wiki_watchlist watchlist \
+ LEFT JOIN pages ON watchlist.page = pages.page \
+ WHERE watchlist.uid = %s",
+ account.uid,
+ )
+
+ return sorted(pages)
+
# ACL
def check_acl(self, page, account):
mimetype, blob_id, size) VALUES(%s, %s, %s, %s, %s, %s, %s) RETURNING *", path,
filename, author.uid, address, mimetype, blob.id, len(data))
- def find_image(self, path, filename):
- for p in (path, os.path.dirname(path)):
- file = self.get_file_by_path(os.path.join(p, filename))
+ def render(self, path, text):
+ r = WikiRenderer(self.backend, path)
- if file and file.is_image():
- return file
+ return r.render(text)
class Page(misc.Object):
if self.data.author_uid:
return self.backend.accounts.get_by_uid(self.data.author_uid)
- def _render(self, text):
- logging.debug("Rendering %s" % self)
-
- # Link images
- replacements = []
- for match in re.finditer(r"!\[(.*?)\]\((.*?)\)", text):
- alt_text, url = match.groups()
-
- # Skip any absolute and external URLs
- if url.startswith("/") or url.startswith("https://") or url.startswith("http://"):
- continue
-
- # Try to split query string
- url, delimiter, qs = url.partition("?")
-
- # Parse query arguments
- args = urllib.parse.parse_qs(qs)
-
- # Find image
- file = self.backend.wiki.find_image(self.page, url)
- if not file:
- continue
-
- # Scale down the image if not already done
- if not "s" in args:
- args["s"] = "768"
-
- # Format URL
- url = "%s?%s" % (file.url, urllib.parse.urlencode(args))
-
- replacements.append((match.span(), file, alt_text, url))
-
- # Apply all replacements
- for (start, end), file, alt_text, url in reversed(replacements):
- text = text[:start] + "[![%s](%s)](%s?action=detail)" % (alt_text, url, file.url) + text[end:]
-
- # Add wiki links
- patterns = (
- (r"\[\[([\w\d\/\-\.]+)(?:\|(.+?))\]\]", r"\1", r"\2", None, True),
- (r"\[\[([\w\d\/\-\.]+)\]\]", r"\1", r"\1", self.backend.wiki.get_page_title, True),
-
- # External links
- (r"\[\[((?:ftp|git|https?|rsync|sftp|ssh|webcal)\:\/\/.+?)(?:\|(.+?))\]\]",
- r"\1", r"\2", None, False),
-
- # Mail
- (r"\[\[([a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+)\]\]",
- r"\1", r"\1", None, False),
- )
-
- for pattern, link, title, repl, internal in patterns:
- replacements = []
-
- for match in re.finditer(pattern, text):
- l = match.expand(link)
- t = match.expand(title)
-
- if internal:
- # Allow relative links
- if not l.startswith("/"):
- l = os.path.join(self.page, l)
-
- # Normalise links
- l = os.path.normpath(l)
-
- if callable(repl):
- t = repl(l) or t
-
- replacements.append((match.span(), t or l, l))
-
- # Apply all replacements
- for (start, end), t, l in reversed(replacements):
- text = text[:start] + "[%s](%s)" % (t, l) + text[end:]
-
- # Borrow this from the blog
- return self.backend.blog._render_text(text, lang="markdown")
-
@property
def markdown(self):
return self.data.markdown or ""
@property
def html(self):
- return self._render(self.markdown)
+ return self.backend.wiki.render(self.page, self.markdown)
@property
def timestamp(self):
logging.debug("Excluding %s" % watcher)
continue
+ # Check permissions
+ if not self.backend.wiki.check_acl(self.page, watcher):
+ logging.debug("Watcher %s does not have permissions" % watcher)
+ continue
+
logging.debug("Sending watcher email to %s" % watcher)
# Compose message
return thumbnail
# Generate the thumbnail
- thumbnail = self._generate_thumbnail(size)
+ thumbnail = util.generate_thumbnail(self.blob, size)
# Put it into the cache for forever
self.memcache.set(cache_key, thumbnail)
return thumbnail
- def _generate_thumbnail(self, size, **args):
- image = PIL.Image.open(io.BytesIO(self.blob))
- # Remove any alpha-channels
- if image.format == "JPEG" and not image.mode == "RGB":
- # Make a white background
- background = PIL.Image.new("RGBA", image.size, (255,255,255))
+class WikiRenderer(misc.Object):
+ # Wiki links
+ wiki_link = re.compile(r"\[\[([\w\d\/\-\.]+)(?:\|(.+?))?\]\]")
+
+ # External links
+ external_link = re.compile(r"\[\[((?:ftp|git|https?|rsync|sftp|ssh|webcal)\:\/\/.+?)(?:\|(.+?))?\]\]")
+
+ # Interwiki links e.g. [[wp>IPFire]]
+ interwiki_link = re.compile(r"\[\[(\w+)>(.+?)(?:\|(.+?))?\]\]")
+
+ # Mail link
+ email_link = re.compile(r"\[\[([a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+)(?:\|(.+?))?\]\]")
+
+ # Images
+ images = re.compile(r"{{([\w\d\/\-\.]+)(?:\|(.+?))?}}")
+
+ def init(self, path):
+ self.path = path
+
+ def _render_wiki_link(self, m):
+ path, alias = m.groups()
+
+ path = self.backend.wiki.make_path(self.path, path)
+
+ return """<a href="%s">%s</a>""" % (
+ path,
+ alias or self.backend.wiki.get_page_title(path),
+ )
+
+ def _render_external_link(self, m):
+ url, alias = m.groups()
+
+ return """<a class="link-external" href="%s">%s</a>""" % (url, alias or url)
+
+ def _render_interwiki_link(self, m):
+ wiki = m.group(1)
+ if not wiki:
+ return
+
+ # Retrieve URL
+ try:
+ url, repl, icon = INTERWIKIS[wiki]
+ except KeyError:
+ logging.warning("Invalid interwiki: %s" % wiki)
+ return
+
+ # Name of the page
+ name = m.group(2)
+
+ # Expand URL
+ url = url % {
+ "name" : name,
+ "url" : urllib.parse.quote(name),
+ }
+
+ # Get alias (if present)
+ alias = m.group(3)
+
+ if not alias and repl:
+ alias = repl % name
+
+ # Put everything together
+ s = []
+
+ if icon:
+ s.append("<span class=\"%s\"></span>" % icon)
+
+ s.append("""<a class="link-external" href="%s">%s</a>""" % (url, alias or name))
+
+ return " ".join(s)
+
+ def _render_email_link(self, m):
+ address, alias = m.groups()
+
+ return """<a class="link-external" href="mailto:%s">%s</a>""" \
+ % (address, alias or address)
+
+ def _render_image(self, m):
+ url, text = m.groups()
- # Flatten both images together
- flattened_image = PIL.Image.alpha_composite(background, image)
+ # Skip any absolute and external URLs
+ if url.startswith("/") or url.startswith("https://") or url.startswith("http://"):
+ return """<img src="%s" alt="%s">""" % (url, text or "")
- # Remove the alpha channel
- image = flattened_image.convert("RGB")
+ # Try to split query string
+ url, delimiter, qs = url.partition("?")
- # Resize the image to the desired resolution
- image.thumbnail((size, size), PIL.Image.LANCZOS)
+ # Parse query arguments
+ args = urllib.parse.parse_qs(qs)
- if image.format == "JPEG":
- # Apply a gaussian blur to make compression easier
- image = image.filter(PIL.ImageFilter.GaussianBlur(radius=0.05))
+ # Build absolute path
+ url = self.backend.wiki.make_path(self.path, url)
- # Arguments to optimise the compression
- args.update({
- "subsampling" : "4:2:0",
- "quality" : 70,
- })
+ # Find image
+ file = self.backend.wiki.get_file_by_path(url)
+ if not file or not file.is_image():
+ return "<!-- Could not find image %s in %s -->" % (url, self.path)
- with io.BytesIO() as f:
- # If writing out the image does not work with optimization,
- # we try to write it out without any optimization.
- try:
- image.save(f, image.format, optimize=True, **args)
- except:
- image.save(f, image.format, **args)
+ # Scale down the image if not already done
+ if not "s" in args:
+ args["s"] = "768"
- return f.getvalue()
+ return """<a href="%s?action=detail"><img src="%s?%s" alt="%s"></a>""" \
+ % (url, url, urllib.parse.urlencode(args), text or "")
+
+ def render(self, text):
+ logging.debug("Rendering %s" % self.path)
+
+ # Handle wiki links
+ text = self.wiki_link.sub(self._render_wiki_link, text)
+
+ # Handle interwiki links
+ text = self.interwiki_link.sub(self._render_interwiki_link, text)
+
+ # Handle external links
+ text = self.external_link.sub(self._render_external_link, text)
+
+ # Handle email links
+ text = self.email_link.sub(self._render_email_link, text)
+
+ # Handle images
+ text = self.images.sub(self._render_image, text)
+
+ # Borrow this from the blog
+ return self.backend.blog._render_text(text, lang="markdown")