src/librarian/parser.py

   1 # This file is part of Librarian, licensed under GNU Affero GPLv3 or later.
   2 # Copyright © Fundacja Wolne Lektury. See NOTICE for more information.
   3 #
   4 from collections import Counter
   5 import io
   6 import os
   7 import re
   8 from xml.parsers.expat import ExpatError
   9 from lxml import etree
  10 from lxml.etree import XMLSyntaxError, XSLTApplyError
  11 from librarian import ValidationError, NoDublinCore,  ParseError, NoProvider
  12 from librarian import RDFNS
  13 from librarian.cover import make_cover
  14 from librarian import dcparser
  15 from .elements import WL_ELEMENTS
  16
  17
  18 class WLElementLookup(etree.CustomElementClassLookup):
  19     def lookup(self, node_type, document, namespace, name):
  20         if node_type != 'element':
  21             return
  22         if namespace:
  23             return
  24         try:
  25             return WL_ELEMENTS[name]
  26         except KeyError:
  27             return
  28
  29
  30 parser = etree.XMLParser()
  31 parser.set_element_class_lookup(
  32     WLElementLookup()
  33 )
  34
  35
  36
  37 class WLDocument:
  38     """Legacy class, to be replaced with documents.WLDocument."""
  39     LINE_SWAP_EXPR = re.compile(r'/\s', re.MULTILINE | re.UNICODE)
  40     provider = None
  41
  42     def __init__(self, edoc, parse_dublincore=True, provider=None,
  43                  strict=False, meta_fallbacks=None):
  44         self.edoc = edoc
  45         self.provider = provider
  46
  47         root_elem = edoc.getroot()
  48
  49         dc_path = './/' + RDFNS('RDF')
  50
  51         if root_elem.tag != 'utwor':
  52             raise ValidationError(
  53                 "Invalid root element. Found '%s', should be 'utwor'"
  54                 % root_elem.tag
  55             )
  56
  57         if parse_dublincore:
  58             self.rdf_elem = root_elem.find(dc_path)
  59
  60             if self.rdf_elem is None:
  61                 raise NoDublinCore(
  62                     "Document must have a '%s' element." % RDFNS('RDF')
  63                 )
  64
  65             self.book_info = dcparser.BookInfo.from_element(
  66                 self.rdf_elem, fallbacks=meta_fallbacks, strict=strict)
  67         else:
  68             self.book_info = None
  69
  70     @classmethod
  71     def from_bytes(cls, xml, *args, **kwargs):
  72         return cls.from_file(io.BytesIO(xml), *args, **kwargs)
  73
  74     @classmethod
  75     def from_file(cls, xmlfile, *args, **kwargs):
  76
  77         # first, prepare for parsing
  78         if isinstance(xmlfile, str):
  79             file = open(xmlfile, 'rb')
  80             try:
  81                 data = file.read()
  82             finally:
  83                 file.close()
  84         else:
  85             data = xmlfile.read()
  86
  87         if not isinstance(data, str):
  88             data = data.decode('utf-8')
  89
  90         data = data.replace('\ufeff', '')
  91
  92         try:
  93             parser = etree.XMLParser(remove_blank_text=False)
  94             tree = etree.parse(io.BytesIO(data.encode('utf-8')), parser)
  95
  96             return cls(tree, *args, **kwargs)
  97         except (ExpatError, XMLSyntaxError, XSLTApplyError) as e:
  98             raise ParseError(e)
  99
 100     def swap_endlines(self):
 101         """Converts line breaks in stanzas into <br/> tags."""
 102         # only swap inside stanzas
 103         for elem in self.edoc.iter('strofa'):
 104             for child in list(elem):
 105                 if child.tail:
 106                     chunks = self.LINE_SWAP_EXPR.split(child.tail)
 107                     ins_index = elem.index(child) + 1
 108                     while len(chunks) > 1:
 109                         ins = etree.Element('br')
 110                         ins.tail = chunks.pop()
 111                         elem.insert(ins_index, ins)
 112                     child.tail = chunks.pop(0)
 113             if elem.text:
 114                 chunks = self.LINE_SWAP_EXPR.split(elem.text)
 115                 while len(chunks) > 1:
 116                     ins = etree.Element('br')
 117                     ins.tail = chunks.pop()
 118                     elem.insert(0, ins)
 119                 elem.text = chunks.pop(0)
 120
 121     def parts(self, pass_part_errors=False):
 122         if self.provider is None:
 123             raise NoProvider('No document provider supplied.')
 124         if self.book_info is None:
 125             raise NoDublinCore('No Dublin Core in document.')
 126         for part_uri in self.book_info.parts:
 127             try:
 128                 with self.provider.by_slug(part_uri.slug) as f:
 129                     yield self.from_file(f, provider=self.provider)
 130             except Exception as e:
 131                 if pass_part_errors:
 132                     yield e
 133                 else:
 134                     raise
 135
 136     def chunk(self, path):
 137         # convert the path to XPath
 138         expr = self.path_to_xpath(path)
 139         elems = self.edoc.xpath(expr)
 140
 141         if len(elems) == 0:
 142             return None
 143         else:
 144             return elems[0]
 145
 146     def path_to_xpath(self, path):
 147         parts = []
 148
 149         for part in path.split('/'):
 150             match = re.match(r'([^\[]+)\[(\d+)\]', part)
 151             if not match:
 152                 parts.append(part)
 153             else:
 154                 tag, n = match.groups()
 155                 parts.append("*[%d][name() = '%s']" % (int(n)+1, tag))
 156
 157         if parts[0] == '.':
 158             parts[0] = ''
 159
 160         return '/'.join(parts)
 161
 162     def transform(self, stylesheet, **options):
 163         return self.edoc.xslt(stylesheet, **options)
 164
 165     def update_dc(self):
 166         if self.book_info:
 167             parent = self.rdf_elem.getparent()
 168             parent.replace(self.rdf_elem, self.book_info.to_etree(parent))
 169
 170     def serialize(self):
 171         self.update_dc()
 172         return etree.tostring(self.edoc, encoding='unicode', pretty_print=True)
 173
 174     def merge_chunks(self, chunk_dict):
 175         unmerged = []
 176
 177         for key, data in chunk_dict.iteritems():
 178             try:
 179                 xpath = self.path_to_xpath(key)
 180                 node = self.edoc.xpath(xpath)[0]
 181                 repl = etree.fromstring(
 182                     "<%s>%s</%s>" % (node.tag, data, node.tag)
 183                 )
 184                 node.getparent().replace(node, repl)
 185             except Exception as e:
 186                 unmerged.append(repr((key, xpath, e)))
 187
 188         return unmerged
 189
 190     def clean_ed_note(self, note_tag='nota_red'):
 191         """ deletes forbidden tags from nota_red """
 192
 193         for node in self.edoc.xpath('|'.join(
 194                 '//%s//%s' % (note_tag, tag) for tag in
 195                 ('pa', 'pe', 'pr', 'pt', 'begin', 'end', 'motyw'))):
 196             tail = node.tail
 197             node.clear()
 198             node.tag = 'span'
 199             node.tail = tail
 200
 201     def fix_pa_akap(self):
 202         for pa in ('pa','pe','pr','pt'):
 203             for akap in self.edoc.findall(f'//{pa}/akap'):
 204                 akap.getparent().set('blocks', 'true')
 205                 if not akap.getparent().index(akap):
 206                     akap.set('inline', 'true')
 207
 208     def editors(self):
 209         """Returns a set of all editors for book and its children.
 210
 211         :returns: set of dcparser.Person objects
 212         """
 213         if self.book_info is None:
 214             raise NoDublinCore('No Dublin Core in document.')
 215         persons = set(self.book_info.editors
 216                       + self.book_info.technical_editors)
 217         for child in self.parts():
 218             persons.update(child.editors())
 219         if None in persons:
 220             persons.remove(None)
 221         return persons
 222
 223     # Converters
 224
 225     def as_pdf(self, *args, **kwargs):
 226         from librarian import pdf
 227         return pdf.transform(self, *args, **kwargs)
 228
 229     def as_fb2(self, *args, **kwargs):
 230         from librarian import fb2
 231         return fb2.transform(self, *args, **kwargs)
 232
 233     def as_cover(self, cover_class=None, *args, **kwargs):
 234         if cover_class is None:
 235             cover_class = make_cover
 236         return cover_class(self.book_info, *args, **kwargs).output_file()
 237
 238     # for debugging only
 239     def latex_dir(self, *args, **kwargs):
 240         kwargs['latex_dir'] = True
 241         from librarian import pdf
 242         return pdf.transform(self, *args, **kwargs)
 243
 244     def save_output_file(self, output_file, output_path=None,
 245                          output_dir_path=None, make_author_dir=False,
 246                          ext=None):
 247         if output_dir_path:
 248             save_path = output_dir_path
 249             if make_author_dir:
 250                 save_path = os.path.join(
 251                     save_path,
 252                     str(self.book_info.author).encode('utf-8')
 253                 )
 254             save_path = os.path.join(save_path, self.book_info.url.slug)
 255             if ext:
 256                 save_path += '.%s' % ext
 257         else:
 258             save_path = output_path
 259
 260         output_file.save_as(save_path)