# -*- coding: utf-8 -*-
+#
+# This file is part of FNP-Redakcja, licensed under GNU Affero GPLv3 or later.
+# Copyright © Fundacja Nowoczesna Polska. See NOTICE for more information.
+#
import os
import tempfile
import datetime
import mimetypes
import urllib
+import functools
+
+import logging
+logger = logging.getLogger('fnp.hazlenut.vstorage')
# Note: we have to set these before importing Mercurial
os.environ['HGENCODING'] = 'utf-8'
def urlquote(url, safe='/'):
- """Quotes URL
-
+ """Quotes URL
+
>>> urlquote(u'Za\u017c\xf3\u0142\u0107 g\u0119\u015bl\u0105 ja\u017a\u0144')
'Za%C5%BC%C3%B3%C5%82%C4%87_g%C4%99%C5%9Bl%C4%85_ja%C5%BA%C5%84'
"""
return urllib.quote(url.replace(' ', '_').encode('utf-8', 'ignore'), safe)
-
+
+
def urlunquote(url):
- """Unqotes URL
-
+ """Unqotes URL
+
# >>> urlunquote('Za%C5%BC%C3%B3%C5%82%C4%87_g%C4%99%C5%9Bl%C4%85_ja%C5%BA%C5%84')
# u'Za\u017c\xf3\u0142\u0107 g\u0119\u015bl\u0105 ja\u017a\u0144'
"""
return unicode(urllib.unquote(url), 'utf-8', 'ignore').replace('_', ' ')
+
def find_repo_path(path):
"""Go up the directory tree looking for a Mercurial repository (a directory containing a .hg subdirectory)."""
while not os.path.isdir(os.path.join(path, ".hg")):
return None
return path
-def locked_repo(func):
+
+def with_working_copy_locked(func):
"""A decorator for locking the repository when calling a method."""
- def new_func(self, *args, **kwargs):
+ @functools.wraps(func)
+ def wrapped(self, *args, **kwargs):
"""Wrap the original function in locks."""
-
wlock = self.repo.wlock()
+ try:
+ return func(self, *args, **kwargs)
+ finally:
+ wlock.release()
+ return wrapped
+
+
+def with_storage_locked(func):
+ """A decorator for locking the repository when calling a method."""
+
+ @functools.wraps(func)
+ def wrapped(self, *args, **kwargs):
+ """Wrap the original function in locks."""
lock = self.repo.lock()
try:
- func(self, *args, **kwargs)
+ return func(self, *args, **kwargs)
finally:
lock.release()
- wlock.release()
+ return wrapped
+
+
+def guess_mime(file_name):
+ """
+ Guess file's mime type based on extension.
+ Default of text/x-wiki for files without an extension.
+
+ >>> guess_mime('something.txt')
+ 'text/plain'
+ >>> guess_mime('SomePage')
+ 'text/x-wiki'
+ >>> guess_mime(u'ąęśUnicodePage')
+ 'text/x-wiki'
+ >>> guess_mime('image.png')
+ 'image/png'
+ >>> guess_mime('style.css')
+ 'text/css'
+ >>> guess_mime('archive.tar.gz')
+ 'archive/gzip'
+ """
- return new_func
+ mime, encoding = mimetypes.guess_type(file_name, strict=False)
+ if encoding:
+ mime = 'archive/%s' % encoding
+ if mime is None:
+ mime = 'text/x-wiki'
+ return mime
class DocumentNotFound(Exception):
if not os.path.exists(self.path):
os.makedirs(self.path)
self.repo_path = find_repo_path(self.path)
- try:
- self.ui = mercurial.ui.ui(report_untrusted=False,
- interactive=False, quiet=True)
- except TypeError:
- # Mercurial 1.3 changed the way we setup the ui object.
- self.ui = mercurial.ui.ui()
- self.ui.quiet = True
- self.ui._report_untrusted = False
- self.ui.setconfig('ui', 'interactive', False)
+
+ self.ui = mercurial.ui.ui()
+ self.ui.quiet = True
+ self.ui._report_untrusted = False
+ self.ui.setconfig('ui', 'interactive', False)
+
if self.repo_path is None:
self.repo_path = self.path
create = True
else:
create = False
+
self.repo_prefix = self.path[len(self.repo_path):].strip('/')
self.repo = mercurial.hg.repository(self.ui, self.repo_path,
create=create)
def reopen(self):
"""Close and reopen the repo, to make sure we are up to date."""
-
self.repo = mercurial.hg.repository(self.ui, self.repo_path)
def _file_path(self, title):
return urlunquote(name)
def __contains__(self, title):
- return os.path.exists(self._file_path(title))
+ return urlquote(title) in self.repo['tip']
def __iter__(self):
return self.all_pages()
self.repo.dirstate.setparents(parent_node)
node = self._commit([repo_file], text, user)
-
+
partial = lambda filename: repo_file == filename
-
+
# If p1 is equal to p2, there is no work to do. Even the dirstate is correct.
p1, p2 = self.repo[None].parents()[0], self.repo[tip_node]
if p1 == p2:
return text
-
- # TODO: Check if merge was successful
- mercurial.merge.update(self.repo, tip_node, True, False, partial)
+
+ try:
+ mercurial.merge.update(self.repo, tip_node, True, False, partial)
+ msg = 'merge of edit conflict'
+ except mercurial.util.Abort:
+ msg = 'failed merge of edit conflict'
self.repo.dirstate.setparents(tip_node, node)
# Mercurial 1.1 and later need updating the merge state
mercurial.merge.mergestate(self.repo).mark(repo_file, "r")
except (AttributeError, KeyError):
pass
- return u'merge of edit conflict'
+ return msg
- @locked_repo
- def save_file(self, title, file_name, author=u'', comment=u'', parent=None):
+ @with_working_copy_locked
+ @with_storage_locked
+ def save_file(self, title, file_name, **kwargs):
"""Save an existing file as specified page."""
- user = author.encode('utf-8') or u'anon'.encode('utf-8')
- text = comment.encode('utf-8') or u'comment'.encode('utf-8')
+ author = kwargs.get('author', u'anonymous').encode('utf-8')
+ comment = kwargs.get('comment', u'Empty comment.').encode('utf-8')
+ parent = kwargs.get('parent', None)
+
repo_file = self._title_to_file(title)
file_path = self._file_path(title)
mercurial.util.rename(file_name, file_path)
changectx = self._changectx()
+
try:
filectx_tip = changectx[repo_file]
current_page_rev = filectx_tip.filerev()
except mercurial.revlog.LookupError:
self.repo.add([repo_file])
current_page_rev = -1
- if parent is not None and current_page_rev != parent:
- msg = self.merge_changes(changectx, repo_file, text, user, parent)
- user = '<wiki>'
- text = msg.encode('utf-8')
- self._commit([repo_file], text, user)
+ if parent is not None and current_page_rev != parent:
+ msg = self.merge_changes(changectx, repo_file, comment, author, parent)
+ author = '<wiki>'
+ comment = msg.encode('utf-8')
- def _commit(self, files, text, user):
- try:
- return self.repo.commit(files=files, text=text, user=user,
- force=True, empty_ok=True)
- except TypeError:
- # Mercurial 1.3 doesn't accept empty_ok or files parameter
- match = mercurial.match.exact(self.repo_path, '', list(files))
- return self.repo.commit(match=match, text=text, user=user,
- force=True)
-
+ self._commit([repo_file], comment, author)
- def save_data(self, title, data, author=u'', comment=u'', parent=None):
+ def save_data(self, title, data, **kwargs):
"""Save data as specified page."""
-
try:
temp_path = tempfile.mkdtemp(dir=self.path)
file_path = os.path.join(temp_path, 'saved')
f = open(file_path, "wb")
f.write(data)
f.close()
- self.save_file(title, file_path, author, comment, parent)
+
+ return self.save_file(title=title, file_name=file_path, **kwargs)
finally:
try:
os.unlink(file_path)
except OSError:
pass
- def save_text(self, title, text, author=u'', comment=u'', parent=None):
+ def save_text(self, **kwargs):
"""Save text as specified page, encoded to charset."""
+ text = kwargs.pop('text')
+ return self.save_data(data=text.encode(self.charset), **kwargs)
- data = text.encode(self.charset)
- self.save_data(title, data, author, comment, parent)
-
- def page_text(self, title):
- """Read unicode text of a page."""
-
- data = self.open_page(title).read()
- text = unicode(data, self.charset, 'replace')
- return text
-
- def page_lines(self, page):
- for data in page:
- yield unicode(data, self.charset, 'replace')
+ def _commit(self, files, comment, user):
+ match = mercurial.match.exact(self.repo_path, '', list(files))
+ return self.repo.commit(match=match, text=comment, user=user, force=True)
- @locked_repo
+ @with_working_copy_locked
+ @with_storage_locked
def delete_page(self, title, author=u'', comment=u''):
user = author.encode('utf-8') or 'anon'
text = comment.encode('utf-8') or 'deleted'
self.repo.remove([repo_file])
self._commit([repo_file], text, user)
- def open_page(self, title):
+ def page_text(self, title, revision=None):
+ """Read unicode text of a page."""
+ ctx = self._find_filectx(title, revision)
+
+ if ctx is None:
+ raise DocumentNotFound(title)
+
+ return ctx.data().decode(self.charset, 'replace'), ctx.filerev()
+
+ def page_text_by_tag(self, title, tag):
+ """Read unicode text of a taged page."""
+ fname = self._title_to_file(title)
+ tag = u"{fname}#{tag}".format(**locals()).encode('utf-8')
+
try:
- return open(self._file_path(title), "rb")
- except IOError:
- raise DocumentNotFound()
+ ctx = self.repo[tag][fname]
+ return ctx.data().decode(self.charset, 'replace'), ctx.filerev()
+ except IndexError:
+ raise DocumentNotFound(fname)
+ @with_working_copy_locked
def page_file_meta(self, title):
"""Get page's inode number, size and last modification time."""
-
try:
- (st_mode, st_ino, st_dev, st_nlink, st_uid, st_gid, st_size,
- st_atime, st_mtime, st_ctime) = os.stat(self._file_path(title))
+ (_st_mode, st_ino, _st_dev, _st_nlink, _st_uid, _st_gid, st_size,
+ _st_atime, st_mtime, _st_ctime) = os.stat(self._file_path(title))
except OSError:
return 0, 0, 0
return st_ino, st_size, st_mtime
- def page_meta(self, title):
+ @with_working_copy_locked
+ def page_meta(self, title, revision=None):
"""Get page's revision, date, last editor and his edit comment."""
+ fctx = self._find_filectx(title, revision)
- filectx_tip = self._find_filectx(title)
- if filectx_tip is None:
- raise DocumentNotFound()
- #return -1, None, u'', u''
- rev = filectx_tip.filerev()
- filectx = filectx_tip.filectx(rev)
- date = datetime.datetime.fromtimestamp(filectx.date()[0])
- author = unicode(filectx.user(), "utf-8",
- 'replace').split('<')[0].strip()
- comment = unicode(filectx.description(), "utf-8", 'replace')
- return rev, date, author, comment
+ if fctx is None:
+ raise DocumentNotFound(title)
+
+ return {
+ "revision": fctx.filerev(),
+ "date": datetime.datetime.fromtimestamp(fctx.date()[0]),
+ "author": fctx.user().decode("utf-8", 'replace'),
+ "comment": fctx.description().decode("utf-8", 'replace'),
+ }
def repo_revision(self):
- return self._changectx().rev()
+ return self.repo['tip'].rev()
+
+ def _changectx(self):
+ return self.repo['tip']
def page_mime(self, title):
"""
- Guess page's mime type ased on corresponding file name.
+ Guess page's mime type based on corresponding file name.
Default ot text/x-wiki for files without an extension.
-
- # >>> page_mime('something.txt')
- # 'text/plain'
- # >>> page_mime('SomePage')
- # 'text/x-wiki'
- # >>> page_mime(u'ąęśUnicodePage')
- # 'text/x-wiki'
- # >>> page_mime('image.png')
- # 'image/png'
- # >>> page_mime('style.css')
- # 'text/css'
- # >>> page_mime('archive.tar.gz')
- # 'archive/gzip'
"""
+ return guess_mime(self._file_path(title))
- addr = self._file_path(title)
- mime, encoding = mimetypes.guess_type(addr, strict=False)
- if encoding:
- mime = 'archive/%s' % encoding
- if mime is None:
- mime = 'text/x-wiki'
- return mime
-
- def _changectx(self):
- """Get the changectx of the tip."""
- try:
- # This is for Mercurial 1.0
- return self.repo.changectx()
- except TypeError:
- # Mercurial 1.3 (and possibly earlier) needs an argument
- return self.repo.changectx('tip')
-
- def _find_filectx(self, title):
+ def _find_filectx(self, title, rev=None):
"""Find the last revision in which the file existed."""
-
repo_file = self._title_to_file(title)
- changectx = self._changectx()
+ changectx = self._changectx() # start with tip
+ visited = set()
+
stack = [changectx]
+ visited.add(changectx)
+
while repo_file not in changectx:
if not stack:
- return None
+ raise DocumentNotFound(title)
+
changectx = stack.pop()
for parent in changectx.parents():
- if parent != changectx:
+ if parent not in visited:
stack.append(parent)
- return changectx[repo_file]
+ visited.add(parent)
+
+ try:
+ fctx = changectx[repo_file]
+
+ if rev is not None:
+ fctx = fctx.filectx(rev)
+ fctx.filerev()
+
+ return fctx
+ except (IndexError, LookupError) as e:
+ raise DocumentNotFound(title)
def page_history(self, title):
"""Iterate over the page's history."""
filectx_tip = self._find_filectx(title)
- if filectx_tip is None:
- return
+
maxrev = filectx_tip.filerev()
minrev = 0
- for rev in range(maxrev, minrev-1, -1):
+ for rev in range(maxrev, minrev - 1, -1):
filectx = filectx_tip.filectx(rev)
date = datetime.datetime.fromtimestamp(filectx.date()[0])
- author = unicode(filectx.user(), "utf-8",
- 'replace').split('<')[0].strip()
- comment = unicode(filectx.description(), "utf-8", 'replace')
- yield rev, date, author, comment
+ author = filectx.user().decode('utf-8', 'replace')
+ comment = filectx.description().decode("utf-8", 'replace')
+ tags = [t.rsplit('#', 1)[-1] for t in filectx.changectx().tags() if '#' in t]
- def page_revision(self, title, rev):
- """Get unicode contents of specified revision of the page."""
+ yield {
+ "version": rev,
+ "date": date,
+ "author": author,
+ "description": comment,
+ "tag": tags,
+ }
- filectx_tip = self._find_filectx(title)
- if filectx_tip is None:
- raise DocumentNotFound()
- try:
- data = filectx_tip.filectx(rev).data()
- except IndexError:
- raise DocumentNotFound()
- return data
+ @with_working_copy_locked
+ def add_page_tag(self, title, rev, tag, user, doctag=True):
+ ctitle = self._title_to_file(title)
+
+ if doctag:
+ tag = u"{ctitle}#{tag}".format(**locals()).encode('utf-8')
- def revision_text(self, title, rev):
- data = self.page_revision(title, rev)
- text = unicode(data, self.charset, 'replace')
- return text
+ message = u"Assigned tag {tag!r} to version {rev!r} of {ctitle!r}".format(**locals()).encode('utf-8')
+
+ fctx = self._find_filectx(title, rev)
+ self.repo.tag(
+ names=tag, node=fctx.node(), local=False,
+ user=user, message=message, date=None,
+ )
def history(self):
"""Iterate over the history of entire wiki."""
changectx = self._changectx()
maxrev = changectx.rev()
minrev = 0
- for wiki_rev in range(maxrev, minrev-1, -1):
+ for wiki_rev in range(maxrev, minrev - 1, -1):
change = self.repo.changectx(wiki_rev)
date = datetime.datetime.fromtimestamp(change.date()[0])
- author = unicode(change.user(), "utf-8",
- 'replace').split('<')[0].strip()
- comment = unicode(change.description(), "utf-8", 'replace')
+ author = change.user().decode('utf-8', 'replace')
+ comment = change.description().decode("utf-8", 'replace')
for repo_file in change.files():
if repo_file.startswith(self.repo_prefix):
title = self._file_to_title(repo_file)
yield title, rev, date, author, comment
def all_pages(self):
+ tip = self.repo['tip']
"""Iterate over the titles of all pages in the wiki."""
-
- for filename in os.listdir(self.path):
- if (os.path.isfile(os.path.join(self.path, filename))
- and not filename.startswith('.')):
- yield urlunquote(filename)
+ return [urlunquote(filename) for filename in tip]
def changed_since(self, rev):
"""Return all pages that changed since specified repository revision."""
current = self.repo.lookup('tip')
status = self.repo.status(current, last)
modified, added, removed, deleted, unknown, ignored, clean = status
- for filename in modified+added+removed+deleted:
+ for filename in modified + added + removed + deleted:
if filename.startswith(self.repo_prefix):
yield self._file_to_title(filename)
fnp / redakcja.git / blobdiff