name = forms.CharField(widget=forms.HiddenInput)
text = forms.CharField(widget=forms.Textarea)
revision = forms.IntegerField(widget=forms.HiddenInput)
- author = forms.CharField()
comment = forms.CharField()
def __init__(self, *args, **kwargs):
def get_storage(self):
return storage
-
- def save(self):
+
+ def save(self, document_author = 'anonymous'):
document = Document(self.get_storage(), name=self.cleaned_data['name'], text=self.cleaned_data['text'])
- storage.put(document, self.cleaned_data['author'], self.cleaned_data['comment'],
- self.cleaned_data['revision'])
+
+ storage.put(document,
+ author = document_author,
+ comment = self.cleaned_data['comment'],
+ parent =self.cleaned_data['revision'] )
+
return storage.get(self.cleaned_data['name'])
return Document(self, name = name, text = text)
def put(self, document, author, comment, parent):
- self.vstorage.save_text(document.name, document.text, author, comment, parent)
+ self.vstorage.save_text(
+ title = document.name,
+ text = document.text,
+ author = author,
+ comment = comment,
+ parent = parent)
def delete(self, name, author, comment):
self.vstorage.delete_page(name, author, comment)
--- /dev/null
+from django.conf.urls.defaults import *
+
+urlpatterns = patterns('wiki.views',
+ url(r'^$',
+ 'document_list', name='wiki_doclist'),
+ url(r'^gallery/(?P<directory>[^/]+)$',
+ 'document_gallery', name="wiki_gallery"),
+ url(r'^(?P<name>[^/]+)/history$',
+ 'document_history', name="wiki_history"),
+ url(r'^(?P<name>[^/]+)/publish/(?P<version>\d+)$',
+ 'document_publish', name="wiki_publish"),
+ url(r'^(?P<name>[^/]+)/diff/(?P<revA>\d+)/(?P<revB>\d+)$',
+ 'document_diff', name="wiki_diff"),
+ url(r'^(?P<name>[^/]+)$',
+ 'document_detail', name="wiki_details"),
+
+)
+
from datetime import datetime
from django.utils.encoding import smart_unicode
-# import google_diff
-# import difflib
+import httplib2
+import poster
import nice_diff
import operator
def document_detail(request, name, template_name = 'wiki/document_details.html'):
+ print "Trying to get", repr(name)
try:
document = storage.get(name)
except DocumentNotFound:
del last_documents[oldest_key]
request.session['wiki_last_docs'] = last_documents
- if request.method == 'POST':
-
+ if request.method == 'POST':
form = DocumentForm(request.POST, instance = document)
if form.is_valid():
- document = form.save()
+ document = form.save(document_author = request.user.username)
return HttpResponse(json.dumps({'text': document.plain_text, 'meta': document.meta(), 'revision': document.revision()}))
else:
return HttpResponse(json.dumps({'errors': form.errors}))
docB.plain_text.splitlines()) )
-def document_history(reuqest, name):
- return HttpResponse( json.dumps(storage.history(name), cls=DateTimeEncoder), mimetype='application/json')
+def document_history(request, name):
+ return HttpResponse(
+ json.dumps(storage.history(name), cls=DateTimeEncoder),
+ mimetype='application/json')
+
+
+def document_publish(request, name, version):
+ # get the document
+ try:
+ document = storage.get(name, revision = int(version))
+ except DocumentNotFound:
+ raise Http404
+
+ poster.streaminghttp.register_openers()
+
+ # send request to WL
+ http = httplib2.Http()
+ http.add_credentials("test", "test")
+ http.follow_all_redirects = True
+ datagen, headers = poster.encode.multipart_encode({name: document.plain_text})
+
+ for key, value in headers.items():
+ headers[key] = unicode(value)
+
+ try:
+ resp, data = http.request("http://localhost:8000/api/books.json",
+ method = "POST", body = ''.join(datagen), headers = headers)
+
+ print resp, data
+
+ if resp.status == 201: # success
+ result = {"success": True}
+ else:
+ result = {"success": False, "errno": resp.status, "reason": resp.reason}
+ except Exception, e:
+ result = {"success": False, "errno": 500, "reason": unicode(e)}
+
+ return HttpResponse( json.dumps(result), mimetype='application/json')
\ No newline at end of file
import os
import tempfile
from nose.tools import *
+from nose.core import runmodule
import vstorage
title = u"test title"
author = u"test author"
comment = u"test comment"
- self.repo.save_text(title, text, author, comment, parent=-1)
+
+ self.repo.save_text(title = title,
+ text = text, author = author,
+ comment = comment, parent=-1)
+
saved = self.repo.open_page(title).read()
assert saved == text
title = u"test title"
author = u"test author"
comment = u"test comment"
- self.repo.save_text(title, text, author, comment, parent=None)
+
+ self.repo.save_text(title = title,
+ text = text, author = author,
+ comment = comment, parent=None)
+
saved = self.repo.open_page(title).read()
assert saved == text
title = u"test title"
author = u"test author"
comment = u"test comment"
- self.repo.save_text(title, text, author, comment, parent=-1)
- self.repo.save_text(title, text, author, comment, parent=-1)
+ self.repo.save_text(title = title,
+ text = text, author = author,
+ comment = comment, parent=-1)
+ self.repo.save_text(title = title,
+ text = text, author = author,
+ comment = comment, parent=-1)
saved = self.repo.open_page(title).read()
assert saved == text
title = u"test title"
author = u"test author"
comment = u"test comment"
- self.repo.save_text(title, text, author, comment, parent=-1)
- self.repo.save_text(title, text1, author, comment, parent=0)
- self.repo.save_text(title, text2, author, comment, parent=0)
+
+ self.repo.save_text(title = title,
+ text = text, author = author,
+ comment = comment, parent=-1)
+
+ self.repo.save_text(title = title,
+ text = text1, author = author,
+ comment = comment, parent=0)
+
+ self.repo.save_text(title = title,
+ text = text2, author = author,
+ comment = comment, parent=0)
+
saved = self.repo.open_page(title).read()
+
# Other conflict markers placement can also be correct
assert_equal(saved, u'''\
text
>>>>>>> other
''')
+
def test_delete(self):
text = u"text test"
title = u"test title"
author = u"test author"
comment = u"test comment"
- self.repo.save_text(title, text, author, comment, parent=-1)
+ self.repo.save_text(title = title,
+ text = text, author = author,
+ comment = comment, parent=-1)
+
assert title in self.repo
+
self.repo.delete_page(title, author, comment)
+
assert title not in self.repo
@raises(vstorage.DocumentNotFound)
self.repo.open_page(u'unknown entity')
def test_open_existing_repository(self):
- self.repo.save_text(u'Python!', u'ham and spam')
+ self.repo.save_text(title = u'Python!', text = u'ham and spam')
current_repo_revision = self.repo.repo_revision()
same_repo = vstorage.VersionedStorage(self.repo_path)
assert same_repo.repo_revision() == current_repo_revision
+
+if __name__ == '__main__':
+ runmodule()
\ No newline at end of file
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'
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()
- lock = self.repo.lock()
try:
- func(self, *args, **kwargs)
- finally:
- lock.release()
+ return func(self, *args, **kwargs)
+ finally:
wlock.release()
+ return wrapped
- return new_func
+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:
+ return func(self, *args, **kwargs)
+ finally:
+ lock.release()
+ return wrapped
def guess_mime(file_name):
"""
Guess file's mime type based on extension.
- Default ot text/x-wiki for files without an extension.
+ Default of text/x-wiki for files without an extension.
>>> guess_mime('something.txt')
'text/plain'
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):
name = filename[len(self.repo_prefix):].strip('/')
return urlunquote(name)
- def __contains__(self, title):
- return urlquote(title) in self.repo.dirstate
+ def __contains__(self, title):
+ return urlquote(title) in self.repo['tip']
def __iter__(self):
return self.all_pages()
pass
return msg
- @locked_repo
+ @with_working_copy_locked
+ @with_storage_locked
def save_file(self, title, file_name, author = u'', comment = u'', parent = None):
"""Save an existing file as specified page."""
-
- user = author.encode('utf-8') or u'anon'.encode('utf-8')
+ user = author.encode('utf-8') or u'anonymous'.encode('utf-8')
text = comment.encode('utf-8') or u'comment'.encode('utf-8')
+
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)
-
-
- 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)
-
-
- 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)
+ 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):
- """Save text as specified page, encoded to charset."""
+ def save_text(self, text, **kwargs):
+ """Save text as specified page, encoded to charset."""
+ self.save_data(data = text.encode(self.charset), **kwargs)
- data = text.encode(self.charset)
- self.save_data(title, data, author, comment, parent)
+ def _commit(self, files, text, user):
+ match = mercurial.match.exact(self.repo_path, '', list(files))
+ return self.repo.commit(match = match, text = text, user = user, force = True)
+
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
for data in page:
yield unicode(data, self.charset, 'replace')
- @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)
+ @with_working_copy_locked
def open_page(self, title):
if title not in self:
raise DocumentNotFound()
-
+
+ path = self._title_to_file(title)
+ logger.debug("Opening page %s", path)
try:
- return open(self._file_path(title), "rb")
+ return self.repo.wfile(path, 'rb')
except IOError:
- import traceback
- print traceback.print_exc()
+ logger.exception("Failed to open page %s", title)
raise DocumentNotFound()
+ @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))
except OSError:
return 0, 0, 0
return st_ino, st_size, st_mtime
-
+
+ @with_working_copy_locked
def page_meta(self, title):
"""Get page's revision, date, last editor and his edit comment."""
if not title in self:
return rev, date, author, comment
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 based on corresponding file name.
Default ot text/x-wiki for files without an extension.
"""
- return guess_type(self._file_path(title))
-
- 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')
+ return guess_mime(self._file_path(title))
def _find_filectx(self, title):
"""Find the last revision in which the file existed."""
yield title, rev, date, author, comment
def all_pages(self):
+ tip = self.repo['tip']
"""Iterate over the titles of all pages in the wiki."""
- return [ urlunquote(filename) for filename in self.repo.dirstate]
- #status = self.repo.status(self.repo[None], None, None, True, True, True)
- #clean_files = status[6]
- #for filename in clean_files:
- # yield
+ return [ urlunquote(filename) for filename in tip ]
def changed_since(self, rev):
"""Return all pages that changed since specified repository revision."""
--- /dev/null
+[loggers]
+keys=root,fnp
+
+[handlers]
+keys=console
+
+[formatters]
+keys=default
+
+[logger_root]
+level=DEBUG
+handlers=console
+
+[logger_fnp]
+level=DEBUG
+handlers=console
+qualname=fnp
+propagate=0
+
+[formatter_default]
+format=%(asctime)s %(name)s/%(levelname)s :: %(module)s.%(funcName)s:%(lineno)d :: %(message)s
+datefmt=
+class=logging.Formatter
+
+[handler_console]
+class=StreamHandler
+level=DEBUG
+formatter=default
+args=(sys.stderr, )
\ No newline at end of file
# -*- coding: utf-8 -*-
-from os import path
+import os.path
-PROJECT_ROOT = path.realpath(path.dirname(__file__))
+PROJECT_ROOT = os.path.realpath(os.path.dirname(__file__))
DEBUG = False
TEMPLATE_DEBUG = DEBUG
except ImportError:
pass
+try:
+ LOGGING_CONFIG_FILE
+except NameError:
+ LOGGING_CONFIG_FILE = os.path.join(PROJECT_ROOT,
+ ('logging.cfg' if not DEBUG else 'logging.cfg.dev'))
+
+try:
+ import logging
+ import logging.config
+
+ logging.config.fileConfig(LOGGING_CONFIG_FILE)
+except ImportError, exc:
+ import traceback
+ traceback.print_exc()
+ raise
+
return true;
}
- var ANNOT_ALLOWED = ['wyroznienie'];
+ var ANNOT_ALLOWED = ['wyroznienie', 'slowo_obce', 'osoba'];
function html2plainText(fragment) {
var text = "";
admin.autodiscover()
urlpatterns = patterns('',
- url(r'^$', 'wiki.views.document_list'),
- url(r'^gallery/(?P<directory>[^/]+)$', 'wiki.views.document_gallery'),
- url(r'^(?P<name>[^/]+)/history$', 'wiki.views.document_history'),
- url(r'^(?P<name>[^/]+)/diff/(?P<revA>\d+)/(?P<revB>\d+)$', 'wiki.views.document_diff'),
-
# Auth
url(r'^accounts/login/$', 'django_cas.views.login', name = 'login'),
url(r'^accounts/logout/$', 'django_cas.views.logout', name = 'logout'),
{'document_root': settings.MEDIA_ROOT, 'show_indexes': True}),
url(r'^%s(?P<path>.+)$' % settings.STATIC_URL[1:], 'django.views.static.serve',
{'document_root': settings.STATIC_ROOT, 'show_indexes': True}),
-
- url(r'^(?P<name>[^/]+)$', 'wiki.views.document_detail'),
+ url(r'^', include('wiki.urls')),
)
# debug stuff
django-nose>=0.0.3
django-debug-toolbar>=0.8
+
+
+poster==0.5
+httplib2
fnp / redakcja.git / commitdiff