import logging
import os.path
import re
-import tornado.gen
import urllib.parse
from . import misc
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 __iter__(self):
+ return self._get_pages(
+ "SELECT wiki.* FROM wiki_current current \
+ LEFT JOIN wiki ON current.id = wiki.id \
+ WHERE current.deleted IS FALSE \
+ ORDER BY page",
+ )
+
def make_path(self, page, path):
# Nothing to do for absolute links
if path.startswith("/"):
# Normalise links
return os.path.normpath(path)
+ def page_exists(self, path):
+ page = self.get_page(path)
+
+ # Page must have been found and not deleted
+ return page and not page.was_deleted()
+
def get_page_title(self, page, default=None):
# Try to retrieve title from cache
title = self.memcache.get("wiki:title:%s" % page)
return ret
def search(self, query, account=None, limit=None):
- query = util.parse_search_query(query)
-
res = self._get_pages("SELECT wiki.* FROM wiki_search_index search_index \
LEFT JOIN wiki ON search_index.wiki_id = wiki.id \
- WHERE search_index.document @@ to_tsquery('english', %s) \
- ORDER BY ts_rank(search_index.document, to_tsquery('english', %s)) DESC",
+ WHERE search_index.document @@ websearch_to_tsquery('english', %s) \
+ ORDER BY ts_rank(search_index.document, websearch_to_tsquery('english', %s)) DESC",
query, query)
pages = []
# If user is in a matching group, we grant permission
for group in row.groups:
- if group in account.groups:
+ if account.is_member_of_group(group):
return True
# Otherwise access is not permitted
return list(files)
- def get_file_by_path(self, path):
+ def get_file_by_path(self, path, revision=None):
path, filename = os.path.dirname(path), os.path.basename(path)
+ if revision:
+ # Fetch a specific revision
+ return self._get_file("SELECT * FROM wiki_files \
+ WHERE path = %s AND filename = %s AND created_at <= %s \
+ ORDER BY created_at DESC LIMIT 1", path, filename, revision)
+
+ # Fetch latest version
+ return self._get_file("SELECT * FROM wiki_files \
+ WHERE path = %s AND filename = %s AND deleted_at IS NULL",
+ path, filename)
+
+ def get_file_by_path_and_filename(self, path, filename):
return self._get_file("SELECT * FROM wiki_files \
- WHERE path = %s AND filename = %s AND deleted_at IS NULL", path, filename)
+ WHERE path = %s AND filename = %s AND deleted_at IS NULL",
+ path, filename)
def upload(self, path, filename, data, mimetype, author, address):
+ # Replace any existing files
+ file = self.get_file_by_path_and_filename(path, filename)
+ if file:
+ file.delete(author)
+
# Upload the blob first
- blob = self.db.get("INSERT INTO wiki_blobs(data) VALUES(%s) RETURNING id", data)
+ blob = self.db.get("INSERT INTO wiki_blobs(data) VALUES(%s) \
+ ON CONFLICT (digest(data, %s)) DO UPDATE SET data = EXCLUDED.data \
+ RETURNING id", data, "MD5")
# Create entry for file
return self._get_file("INSERT INTO wiki_files(path, filename, author_uid, address, \
# Find first H1 headline in markdown
markdown = self.markdown.splitlines()
- m = re.match(r"^# (.*)( #)?$", markdown[0])
+ m = re.match(r"^#\s*(.*)( #)?$", markdown[0])
if m:
return m.group(1)
return self.data.timestamp
def was_deleted(self):
- return self.markdown is None
+ return not self.markdown
@lazy_property
def breadcrumbs(self):
return self.backend.wiki.make_breadcrumbs(self.page)
+ def is_latest_revision(self):
+ return self.get_latest_revision() == self
+
def get_latest_revision(self):
revisions = self.get_revisions()
# Compose message
self.backend.messages.send_template("wiki/messages/page-changed",
- recipients=[watcher], page=self, priority=-10)
+ account=watcher, page=self, priority=-10)
+
+ def restore(self, author, address, comment=None):
+ changes = "Restore to revision from %s" % self.timestamp.isoformat()
+
+ # Append comment
+ if comment:
+ changes = "%s: %s" % (changes, comment)
+
+ return self.backend.wiki.create_page(self.page,
+ author, self.markdown, changes=changes, address=address)
class File(misc.Object):
self.id = id
self.data = data
+ def __eq__(self, other):
+ if isinstance(other, self.__class__):
+ return self.id == other.id
+
@property
def url(self):
return os.path.join(self.path, self.filename)
def created_at(self):
return self.data.created_at
+ def delete(self, author=None):
+ self.db.execute("UPDATE wiki_files SET deleted_at = NOW(), deleted_by = %s \
+ WHERE id = %s", author.uid if author else None, self.id)
+
+ @property
+ def deleted_at(self):
+ return self.data.deleted_at
+
+ def get_latest_revision(self):
+ revisions = self.get_revisions()
+
+ # Return first object
+ for rev in revisions:
+ return rev
+
+ def get_revisions(self):
+ revisions = self.backend.wiki._get_files("SELECT * FROM wiki_files \
+ WHERE path = %s AND filename = %s ORDER BY created_at DESC", self.path, self.filename)
+
+ return list(revisions)
+
def is_pdf(self):
return self.mimetype in ("application/pdf", "application/x-pdf")
def is_image(self):
return self.mimetype.startswith("image/")
+ def is_vector_image(self):
+ return self.mimetype in ("image/svg+xml",)
+
+ def is_bitmap_image(self):
+ return self.is_image() and not self.is_vector_image()
+
@lazy_property
def blob(self):
res = self.db.get("SELECT data FROM wiki_blobs \
return bytes(res.data)
def get_thumbnail(self, size):
+ assert self.is_bitmap_image()
+
cache_key = "-".join((self.path, util.normalize(self.filename), self.created_at.isoformat(), "%spx" % size))
# Try to fetch the data from the cache
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-.]+)(?:\|(.+?))?\]\]")
+ schemas = (
+ "ftp://",
+ "git://",
+ "http://",
+ "https://",
+ "rsync://",
+ "sftp://",
+ "ssh://",
+ "webcal://",
+ )
+
+ # Links
+ links = re.compile(r"<a href=\"(.*?)\">(.*?)</a>")
# Images
- images = re.compile(r"{{([\w\d\/\-\.]+)(?:\|(.+?))?}}")
+ images = re.compile(r"<img alt(?:=\"(.*?)\")? src=\"(.*?)\" (?:title=\"(.*?)\" )?/>")
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()
+ def _render_link(self, m):
+ url, text = m.groups()
- return """<a class="link-external" href="%s">%s</a>""" % (url, alias or url)
+ # External Links
+ for schema in self.schemas:
+ if url.startswith(schema):
+ return """<a class="link-external" href="%s">%s</a>""" % \
+ (url, text 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
+ # Emails
+ if "@" in url:
+ # Strip mailto:
+ if url.startswith("mailto:"):
+ url = url[7:]
- # Put everything together
- s = []
+ return """<a class="link-external" href="mailto:%s">%s</a>""" % \
+ (url, text or url)
- if icon:
- s.append("<span class=\"%s\"></span>" % icon)
+ # Everything else must be an internal link
+ path = self.backend.wiki.make_path(self.path, url)
- 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)
+ return """<a href="%s">%s</a>""" % \
+ (path, text or self.backend.wiki.get_page_title(path))
def _render_image(self, m):
- url, caption = m.groups()
+ alt_text, url, caption = m.groups()
# Skip any absolute and external URLs
if url.startswith("/") or url.startswith("https://") or url.startswith("http://"):
return """<figure class="figure"><img src="%s" class="figure-img img-fluid rounded" alt="%s">
<figcaption class="figure-caption">%s</figcaption></figure>
- """ % (url, url, caption or "")
+ """ % (url, alt_text, caption or "")
# Try to split query string
url, delimiter, qs = url.partition("?")
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)
+ # Borrow this from the blog
+ text = self.backend.blog._render_text(text, lang="markdown")
- # Handle email links
- text = self.email_link.sub(self._render_email_link, text)
+ # Postprocess links
+ text = self.links.sub(self._render_link, text)
- # Handle images
+ # Postprocess images to <figure>
text = self.images.sub(self._render_image, text)
- # Borrow this from the blog
- return self.backend.blog._render_text(text, lang="markdown")
+ return text
projects / ipfire.org.git / blobdiff