#!/usr/bin/python3
import logging
-import markdown2
import os.path
import re
from . import misc
+from . import util
from .decorators import *
-# Used to automatically link some things
-link_patterns = (
- # Find bug reports
- (re.compile(r"(?:#(\d+))", re.I), r"https://bugzilla.ipfire.org/show_bug.cgi?id=\1"),
-
- # Email Addresses
- (re.compile(r"([a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+)"), r"mailto:\1"),
-
- # CVE Numbers
- (re.compile(r"(?:CVE)[\s\-](\d{4}\-\d+)"), r"https://cve.mitre.org/cgi-bin/cvename.cgi?name=\1"),
-)
-
class Wiki(misc.Object):
def _get_pages(self, query, *args):
res = self.db.query(query, *args)
for row in res:
yield Page(self.backend, row.id, data=row)
+ def _get_page(self, query, *args):
+ res = self.db.get(query, *args)
+
+ if res:
+ return Page(self.backend, res.id, data=res)
+
def get_page_title(self, page, default=None):
doc = self.get_page(page)
if doc:
return doc.title
- return default
+ return default or os.path.basename(page)
def get_page(self, page, revision=None):
page = Page.sanitise_page_name(page)
assert page
if revision:
- res = self.db.get("SELECT * FROM wiki WHERE page = %s \
+ return self._get_page("SELECT * FROM wiki WHERE page = %s \
AND timestamp = %s", page, revision)
else:
- res = self.db.get("SELECT * FROM wiki WHERE page = %s \
+ return self._get_page("SELECT * FROM wiki WHERE page = %s \
ORDER BY timestamp DESC LIMIT 1", page)
- if res:
- return Page(self.backend, res.id, data=res)
-
def get_recent_changes(self, limit=None):
return self._get_pages("SELECT * FROM wiki \
WHERE timestamp >= NOW() - INTERVAL '4 weeks' \
ORDER BY timestamp DESC LIMIT %s", limit)
- def create_page(self, page, author, markdown):
+ def create_page(self, page, author, content, changes=None, address=None):
page = Page.sanitise_page_name(page)
- res = self.db.get("INSERT INTO wiki(page, author_id, markdown) \
- VALUES(%s, %s, %s) RETURNING id", page, author.id, markdown)
-
- if res:
- return self.get_page_by_id(res.id)
+ return 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)
- def delete_page(self, page, author):
+ def delete_page(self, page, author, **kwargs):
# Do nothing if the page does not exist
if not self.get_page(page):
return
# Just creates a blank last version of the page
- self.create_page(page, author, None)
+ self.create_page(page, author=author, content=None, **kwargs)
- @staticmethod
- def _split_url(url):
+ def make_breadcrumbs(self, url):
+ # Split and strip all empty elements (double slashes)
parts = list(e for e in url.split("/") if e)
- num_parts = len(parts)
- for i in range(num_parts):
- yield "/".join(parts[:i])
+ ret = []
+ for part in ("/".join(parts[:i]) for i in range(1, len(parts))):
+ ret.append(("/%s" % part, self.get_page_title(part, os.path.basename(part))))
- def make_breadcrumbs(self, url):
- for part in self._split_url(url):
- title = self.get_page_title(part, os.path.basename(part))
+ return ret
+
+ def search(self, query, 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 \
+ LIMIT %s", query, query, limit)
+
+ return list(res)
+
+ def refresh(self):
+ """
+ Needs to be called after a page has been changed
+ """
+ self.db.execute("REFRESH MATERIALIZED VIEW wiki_search_index")
+
+ # Files
+
+ def _get_files(self, query, *args):
+ res = self.db.query(query, *args)
+
+ for row in res:
+ yield File(self.backend, row.id, data=row)
+
+ def _get_file(self, query, *args):
+ res = self.db.get(query, *args)
+
+ if res:
+ return File(self.backend, res.id, data=res)
+
+ def get_files(self, path):
+ files = self._get_files("SELECT * FROM wiki_files \
+ WHERE path = %s AND deleted_at IS NULL ORDER BY filename", path)
+
+ return list(files)
+
+ def get_file_by_path(self, path):
+ path, filename = os.path.dirname(path), os.path.basename(path)
+
+ return self._get_file("SELECT * FROM wiki_files \
+ WHERE path = %s AND filename = %s AND deleted_at IS NULL", path, filename)
- yield ("/%s" % part, title)
+ def upload(self, path, filename, data, mimetype, author, address):
+ # Upload the blob first
+ blob = self.db.get("INSERT INTO wiki_blobs(data) VALUES(%s) RETURNING id", data)
+
+ # Create entry for file
+ return self._get_file("INSERT INTO wiki_files(path, filename, author_uid, address, \
+ mimetype, blob_id, size) VALUES(%s, %s, %s, %s, %s, %s, %s) RETURNING *", path,
+ filename, author.uid, address, mimetype, blob.id, len(data))
class Page(misc.Object):
def _render(self, text):
logging.debug("Rendering %s" % self)
- return markdown2.markdown(text, link_patterns=link_patterns,
- extras=["footnotes", "link-patterns", "wiki-tables"])
+ patterns = (
+ (r"\[\[([\w\d\/]+)(?:\|([\w\d\s]+))\]\]", r"/\1", r"\2", None, None),
+ (r"\[\[([\w\d\/\-]+)\]\]", r"/\1", r"\1", self.backend.wiki.get_page_title, r"\1"),
+ )
+
+ for pattern, link, title, repl, args in patterns:
+ replacements = []
+
+ for match in re.finditer(pattern, text):
+ l = match.expand(link)
+ t = match.expand(title)
+
+ if callable(repl):
+ t = repl(match.expand(args)) 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.backend.wiki.make_breadcrumbs(self.page)
def get_latest_revision(self):
- return self.backend.wiki.get_page(self.page)
+ revisions = self.get_revisions()
+
+ # Return first object
+ for rev in revisions:
+ return rev
+
+ def get_revisions(self):
+ return self.backend.wiki._get_pages("SELECT * FROM wiki \
+ WHERE page = %s ORDER BY timestamp DESC", self.page)
+
+ @property
+ def changes(self):
+ return self.data.changes
# Sidebar
parts = self.page.split("/")
while parts:
- sidebar = self.backend.wiki.get_page(os.path.join(*parts, "sidebar"))
+ sidebar = self.backend.wiki.get_page("%s/sidebar" % os.path.join(*parts))
if sidebar:
return sidebar
parts.pop()
+
+
+class File(misc.Object):
+ def init(self, id, data):
+ self.id = id
+ self.data = data
+
+ @property
+ def url(self):
+ return os.path.join(self.path, self.filename)
+
+ @property
+ def path(self):
+ return self.data.path
+
+ @property
+ def filename(self):
+ return self.data.filename
+
+ @property
+ def mimetype(self):
+ return self.data.mimetype
+
+ @property
+ def size(self):
+ return self.data.size
+
+ def is_image(self):
+ return self.mimetype.startswith("image/")
+
+ @lazy_property
+ def blob(self):
+ res = self.db.get("SELECT data FROM wiki_blobs \
+ WHERE id = %s", self.data.blob_id)
+
+ if res:
+ return bytes(res.data)
projects / ipfire.org.git / blobdiff