123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452 |
- import difflib
- from django.conf import settings
- from django.utils.translation import gettext as _
- import fstools
- import json
- import logging
- from pages import messages, url_page
- import mycreole
- import os
- import shutil
- import time
- from . import timestamp_to_datetime
-
- logger = logging.getLogger(settings.ROOT_LOGGER_NAME).getChild(__name__)
-
-
- SPLITCHAR = ":"
- HISTORY_FOLDER_NAME = 'history'
-
-
- def full_path_all_pages(expression="*"):
- system_pages = fstools.dirlist(settings.SYSTEM_PAGES_ROOT, expression=expression, rekursive=False)
- system_pages = [os.path.join(settings.PAGES_ROOT, os.path.basename(path)) for path in system_pages]
- pages = fstools.dirlist(settings.PAGES_ROOT, expression=expression, rekursive=False)
- return list(set(system_pages + pages))
-
-
- class meta_data(dict):
- META_FILE_NAME = 'meta.json'
- #
- KEY_CREATION_TIME = "creation_time"
- KEY_MODIFIED_TIME = "modified_time"
- KEY_MODIFIED_USER = "modified_user"
- KEY_TAGS = "tags"
-
- def __init__(self, path, history_version=None):
- self._path = path
- self._history_version = history_version
- #
- # Load data from disk
- try:
- with open(self.filename, 'r') as fh:
- super().__init__(json.load(fh))
- except (FileNotFoundError, json.decoder.JSONDecodeError) as e:
- super().__init__()
-
- @property
- def filename(self):
- if not self._history_version:
- return os.path.join(self._path, self.META_FILE_NAME)
- else:
- return self.history_filename(self._history_version)
-
- def history_filename(self, history_version):
- return os.path.join(self._path, HISTORY_FOLDER_NAME, "%05d_%s" % (history_version, self.META_FILE_NAME))
-
- def update_required(self, tags):
- return tags != self.get(self.KEY_TAGS)
-
- def update(self, username, tags):
- if self._history_version:
- logger.error("A history version %05d can not be updated!", self._history_version)
- return False
- else:
- if username:
- self[self.KEY_MODIFIED_TIME] = int(time.time())
- self[self.KEY_MODIFIED_USER] = username
- if self.KEY_CREATION_TIME not in self:
- self[self.KEY_CREATION_TIME] = self[self.KEY_MODIFIED_TIME]
- if tags:
- self[self.KEY_TAGS] = tags
- #
- if username or tags:
- self.save()
- return True
-
- def save(self):
- if self._history_version:
- logger.error("A history version %05d can not be updated!", self._history_version)
- return False
- else:
- with open(self.filename, 'w') as fh:
- json.dump(self, fh, indent=4)
- return True
-
- def store_to_history(self, history_number):
- history_filename = self.history_filename(history_number)
- fstools.mkdir(os.path.dirname(history_filename))
- shutil.copy(self.filename, history_filename)
-
-
- class page_data(object):
- PAGE_FILE_NAME = 'page'
-
- def __init__(self, path, history_version=None):
- self._history_version = history_version
- self._path = path
- self._raw_page_src = None
-
- def _load_page_src(self):
- if self._raw_page_src is None:
- try:
- with open(self.filename, 'r') as fh:
- self._raw_page_src = fh.read()
- except FileNotFoundError:
- self._raw_page_src = ""
-
- def update_required(self, page_txt):
- return page_txt.replace("\r\n", "\n") != self.raw_page_src
-
- def update_page(self, page_txt):
- if self._history_version:
- logger.error("A history version %05d can not be updated!", self._history_version)
- return False
- else:
- # save the new page content
- fstools.mkdir(os.path.dirname(self.filename))
- with open(self.filename, 'w') as fh:
- fh.write(page_txt)
- self._raw_page_src = page_txt
- return True
-
- @property
- def filename(self):
- if not self._history_version:
- return os.path.join(self._path, self.PAGE_FILE_NAME)
- else:
- return self.history_filename(self._history_version)
-
- def history_filename(self, history_version):
- return os.path.join(self._path, HISTORY_FOLDER_NAME, "%05d_%s" % (history_version, self.PAGE_FILE_NAME))
-
- @property
- def rel_path(self):
- return os.path.basename(self._path).replace(2*SPLITCHAR, "/")
-
- def is_available(self):
- is_a = os.path.isfile(self.filename)
- if not is_a:
- logger.debug("page.is_available: Not available - %s", self.filename)
- return is_a
-
- @property
- def title(self):
- return os.path.basename(self._path).split("::")[-1]
-
- @property
- def raw_page_src(self):
- self._load_page_src()
- return self._raw_page_src
-
- def store_to_history(self, history_number):
- history_filename = self.history_filename(history_number)
- fstools.mkdir(os.path.dirname(history_filename))
- shutil.copy(self.filename, history_filename)
-
-
- class page_django(page_data):
- FOLDER_ATTACHMENTS = "attachments"
-
- def __init__(self, request, path, history_version=None) -> None:
- self._request = request
- super().__init__(path, history_version=history_version)
-
- @property
- def attachment_path(self):
- return os.path.join(os.path.basename(self._path), self.FOLDER_ATTACHMENTS)
-
- def render_to_html(self):
- if self.is_available():
- return self.render_text(self._request, self.raw_page_src)
- else:
- messages.unavailable_msg_page(self._request, self.rel_path)
- return ""
-
- def history_numbers_list(self):
- history_folder = os.path.join(self._path, HISTORY_FOLDER_NAME)
- return list(set([int(os.path.basename(filename)[:5]) for filename in fstools.filelist(history_folder)]))
-
- def render_meta(self, ctime, mtime, user, tags):
- #
- # Page meta data
- #
- meta = f'=== {_("Meta data")}\n'
- meta += f'|{_("Created")}:|{timestamp_to_datetime(self._request, ctime)}|\n'
- meta += f'|{_("Modified")}:|{timestamp_to_datetime(self._request, mtime)}|\n'
- meta += f'|{_("Editor")}|{user}|\n'
- meta += f'|{_("Tags")}|{tags}|\n'
- #
- # List of hostory page versions
- #
- hnl = self.history_numbers_list()
- if hnl:
- meta += f'=== {_("History")}\n'
- meta += f'| ={_("Version")} | ={_("Date")} | ={_("Page")} | ={_("Meta data")} | \n'
- # Current
- name = _("Current")
- meta += f"| {name} \
- | {timestamp_to_datetime(self._request, mtime)} \
- | [[{url_page(self._request, self.rel_path)} | Page]] \
- | [[{url_page(self._request, self.rel_path, meta=None)} | Meta]]\n"
- # History
- for num in reversed(hnl):
- p = page_wrapped(self._request, self._path, history_version=num)
- meta += f"| {num} \
- | {timestamp_to_datetime(self._request, p.modified_time)} \
- | [[{url_page(self._request, p.rel_path, history=num)} | Page]] \
- | [[{url_page(self._request, p.rel_path, meta=None, history=num)} | Meta]] (with page changes)\n"
- # Diff
- html_diff = ""
- if self._history_version:
- meta += f'=== {_("Page differences")}\n'
- #
- c = page_django(self._request, self._path)
- left_lines = c.raw_page_src.splitlines()
- right_lines = self.raw_page_src.splitlines()
- html_diff = difflib.HtmlDiff(wrapcolumn=80).make_table(left_lines, right_lines)
- #
- return mycreole.render_simple(meta) + html_diff
-
- def render_text(self, request, txt):
- macros = {
- "subpages": self.macro_subpages,
- "allpages": self.macro_allpages,
- }
- return mycreole.render(request, txt, self.attachment_path, macros=macros)
-
- def macro_allpages(self, *args, **kwargs):
- kwargs["allpages"] = True
- return self.macro_subpages(*args, **kwargs)
-
- def macro_subpages(self, *args, **kwargs):
- allpages = kwargs.pop("allpages", False)
- #
-
- def parse_depth(s: str):
- try:
- return int(s)
- except ValueError:
- pass
-
- params = kwargs.get('', '')
- filter_str = ''
- depth = parse_depth(params)
- if depth is None:
- params = params.split(",")
- depth = parse_depth(params[0])
- if len(params) == 2:
- filter_str = params[1]
- elif depth is None:
- filter_str = params[0]
- #
- rv = ""
- # create a page_list
- if allpages:
- expression = "*"
- parent_rel_path = ""
- else:
- expression = os.path.basename(self._path) + 2 * SPLITCHAR + "*"
- parent_rel_path = self.rel_path
- #
- pl = page_list(
- self._request,
- [page_django(self._request, path) for path in full_path_all_pages(expression)]
- )
- return pl.html_list(depth=depth, filter_str=filter_str, parent_rel_path=parent_rel_path)
-
-
- class page_list(list):
- def __init__(self, request, *args, **kwargs):
- self._request = request
- return super().__init__(*args, **kwargs)
-
- def sort_basename(self):
- return list.sort(self, key=lambda x: os.path.basename(x.rel_path))
-
- def creole_list(self, depth=None, filter_str='', parent_rel_path=''):
- self.sort_basename()
- depth = depth or 9999 # set a random high value if None
- #
- parent_rel_path = parent_rel_path + "/" if len(parent_rel_path) > 0 else ""
- #
- rv = ""
- last_char = None
- for page in self:
- name = page.rel_path[len(parent_rel_path):]
- if name.startswith(filter_str) and name != filter_str:
- if name.count('/') < depth:
- first_char = os.path.basename(name)[0].upper()
- if last_char != first_char:
- last_char = first_char
- rv += f"=== {first_char}\n"
- rv += f"* [[{url_page(self._request, page.rel_path)} | {name} ]]\n"
- return rv
-
- def html_list(self, depth=9999, filter_str='', parent_rel_path=''):
- return mycreole.render_simple(self.creole_list(depth, filter_str, parent_rel_path))
-
-
- class page_wrapped(object):
- """
- This class holds different page and meta instances and decides which will be used in which case.
- """
-
- def __init__(self, request, path, history_version=None):
- """_summary_
-
- Args:
- request (_type_): The django request or None (if None, the page functionality is limited)
- path (_type_): A rel_path of the django page or the filesystem path to the page
- history_version (_type_, optional): The history version of the page to be created
- """
- self._request = request
- #
- page_path = self.__page_path__(path)
- system_page_path = self.__system_page_path__(path)
- # Page
- if request:
- self._page = page_django(request, page_path, history_version=history_version)
- else:
- self._page = page_data(page_path, history_version=history_version)
- self._page_meta = meta_data(page_path, history_version=history_version)
- # System page
- if request:
- self._system_page = page_django(request, system_page_path)
- else:
- self._system_page = page_data(system_page_path)
- self._system_meta_data = meta_data(system_page_path)
-
- def __page_path__(self, path):
- if path.startswith(settings.PAGES_ROOT):
- # must be a filesystem path
- return path
- else:
- # must be a relative url
- return os.path.join(settings.PAGES_ROOT, path.replace("/", 2*SPLITCHAR))
-
- def __system_page_path__(self, path):
- return os.path.join(settings.SYSTEM_PAGES_ROOT, os.path.basename(path))
-
- def __page_choose__(self):
- if not self._page.is_available():
- return self._system_page
- else:
- return self._page
-
- def __meta_choose__(self):
- if not self._page.is_available():
- return self._system_meta_data
- else:
- return self._page_meta
-
- def __store_history__(self):
- if self._page.is_available():
- try:
- history_number = max(self._page.history_numbers_list()) + 1
- except ValueError:
- history_number = 1 # no history yet
- self._page.store_to_history(history_number)
- self._page_meta.store_to_history(history_number)
-
- #
- # meta_data
- #
- @property
- def creation_time(self):
- meta = self.__meta_choose__()
- rv = meta.get(meta.KEY_CREATION_TIME)
- return rv
-
- @property
- def modified_time(self):
- meta = self.__meta_choose__()
- rv = meta.get(meta.KEY_MODIFIED_TIME)
- return rv
-
- @property
- def modified_user(self):
- meta = self.__meta_choose__()
- rv = meta.get(meta.KEY_MODIFIED_USER)
- return rv
-
- @property
- def tags(self):
- meta = self.__meta_choose__()
- rv = meta.get(meta.KEY_TAGS)
- return rv
-
- #
- # page
- #
- @property
- def attachment_path(self):
- page = self.__page_choose__()
- rv = page.attachment_path
- return rv
-
- @property
- def raw_page_src(self):
- page = self.__page_choose__()
- rv = page.raw_page_src
- return rv
-
- @property
- def rel_path(self):
- page = self.__page_choose__()
- rv = page.rel_path
- return rv
-
- def render_meta(self):
- page = self.__page_choose__()
- rv = page.render_meta(self.creation_time, self.modified_time, self.modified_user, self.tags)
- return rv
-
- def render_to_html(self):
- page = self.__page_choose__()
- rv = page.render_to_html()
- return rv
-
- def render_text(self, request, txt):
- page = self.__page_choose__()
- rv = page.render_text(request, txt)
- return rv
-
- @property
- def title(self):
- page = self.__page_choose__()
- rv = page.title
- return rv
-
- def update_page(self, txt, tags):
- if self._page.update_required(txt) or self._page_meta.update_required(tags):
- rv = False
- # Store history
- self.__store_history__()
- username = None
- if self._page.update_required(txt):
- # Update page
- rv |= self._page.update_page(txt)
- # Identify username, to update meta
- try:
- if self._request.user.is_authenticated:
- username = self._request.user.username
- else:
- logger.warning("Page edit without having a logged in user. This is not recommended. Check your access definitions!")
- except AttributeError:
- logger.exception("Page edit without having a request object. Check programming!")
- rv |= self._page_meta.update(username, tags)
- # Update search index
- from pages.search import update_item
- update_item(self)
- return rv
|