import warnings
from dataclasses import dataclass
from enum import Enum
from typing import Dict
from typing import List
from typing import Optional
from typing import Tuple
from typing import Union
from xml.sax.handler import ContentHandler
from clarin.sru.xml.writer import XMLStreamWriterHelper
from clarin.sru.fcs.constants import FCS_NS
from clarin.sru.fcs.constants import FCS_PREFIX
from clarin.sru.fcs.constants import FCSDataViewNamespaces
# ---------------------------------------------------------------------------
# basic/HITS/KWIC dataview
[docs]class FCSRecordXMLStreamWriter:
"""This class provides several helper methods for writing records
in the CLARIN-FCS record schema. These methods **do not** cover
the full spectrum of all variations of records that are permitted
by the CLARIN-FCS specification.
See also:
- CLARIN FCS specification, section "Operation searchRetrieve"
"""
[docs] @staticmethod
def startResource(
writer: ContentHandler, pid: Optional[str] = None, ref: Optional[str] = None
) -> None:
"""Write the start of a resource (i.e. the ``<Resource>``
element). Calls to this method need to be balanced with calls
to the `endResource` method.
Args:
writer: the `xml.sax.handler.ContentHandler` to use
pid: the persistent identifier of this resource or ``None``, if not applicable. Defaults to None.
ref: the reference of this resource or ``None``, if not applicable. Defaults to None.
"""
if writer is None:
raise TypeError("writer is None")
attrs = dict()
if pid and not pid.isspace():
attrs[(None, "pid")] = pid
if ref and not ref.isspace():
attrs[(None, "ref")] = ref
writer.startPrefixMapping(FCS_PREFIX, FCS_NS)
writer.startElementNS((FCS_NS, "Resource"), None, attrs)
[docs] @staticmethod
def endResource(writer: ContentHandler) -> None:
"""Write the end of a resource (i.e. the ``</Resource>``
element). Calls to this method need to be balanced with calls
to the `startResource` method.
Args:
writer: the `xml.sax.handler.ContentHandler` to use
"""
if writer is None:
raise TypeError("writer is None")
writer.endElementNS((FCS_NS, "Resource"), None)
writer.endPrefixMapping(FCS_PREFIX)
[docs] @staticmethod
def startResourceFragment(
writer: ContentHandler, pid: Optional[str] = None, ref: Optional[str] = None
) -> None:
"""Write the start of a resource fragment (i.e. the
``<ResourceFragment>`` element). Calls to this method need to
be balanced with calls to the `endResourceFragment` method.
Args:
writer: the `xml.sax.handler.ContentHandler` to use
pid: the persistent identifier of this resource or ``None``, if not applicable. Defaults to None.
ref: the reference of this resource or ``None``, if not applicable. Defaults to None.
"""
if writer is None:
raise TypeError("writer is None")
attrs = dict()
if pid and not pid.isspace():
attrs[(None, "pid")] = pid
if ref and not ref.isspace():
attrs[(None, "ref")] = ref
writer.startElementNS((FCS_NS, "ResourceFragment"), None, attrs)
[docs] @staticmethod
def endResourceFragment(writer: ContentHandler) -> None:
"""Write the end of a resource fragment (i.e. the
``</ResourceFragment>`` element). Calls to this method need
to be balanced with calls to the `startResourceFragment`
method.
Args:
writer: the `xml.sax.handler.ContentHandler` to use
"""
if writer is None:
raise TypeError("writer is None")
writer.endElementNS((FCS_NS, "ResourceFragment"), None)
[docs] @staticmethod
def startDataView(writer: ContentHandler, mimetype: str) -> None:
"""Write the start of a data view (i.e. the ``<DataView>``
element). Calls to this method need to be balanced with calls
to the `endDataView` method.
Args:
writer: the `xml.sax.handler.ContentHandler` to use
mimetype: the MIME type of this data view applicable
"""
if writer is None:
raise TypeError("writer is None")
if mimetype is None:
raise TypeError("mimetype is None")
elif mimetype is None:
raise ValueError("mimetype is empty")
writer.startElementNS((FCS_NS, "DataView"), None, {(None, "type"): mimetype})
[docs] @staticmethod
def endDataView(writer: ContentHandler) -> None:
"""Write the end of a data view (i.e. the ``</DataView>``
element). Calls to this method need to be balanced with calls
to the `startDataView` method.
Args:
writer: the `xml.sax.handler.ContentHandler` to use
"""
if writer is None:
raise TypeError("writer is None")
writer.endElementNS((FCS_NS, "DataView"), None)
# ----------------------------------------------------
[docs] @staticmethod
def writeKWICDataView(
writer: ContentHandler, left: Optional[str], keyword: str, right: Optional[str]
) -> None:
"""[Deprecated] Use the HITS data view instead!
Convince method to write a KWIC data view. It automatically
performs the calls to `startDataView` and `endDataView`.
Args:
writer: the `xml.sax.handler.ContentHandler` to use
left: the left context of the KWIC or ``None`` if not applicable
keyword: the keyword of the KWIC
right: the right context of the KWIC or ``None`` if not applicable
"""
warnings.warn(
"'writeKWICDataView' is deprecated. Use the HITS data view instead!",
category=DeprecationWarning,
stacklevel=2,
)
if writer is None:
raise TypeError("writer is None")
if keyword is None:
raise TypeError("keyword is None")
writer = XMLStreamWriterHelper(writer)
ns = FCSDataViewNamespaces.KWIC
FCSRecordXMLStreamWriter.startDataView(writer, ns.mimetype)
# actual "kwic" data view
with writer.prefix(ns.prefix, ns.namespace), writer.element(
"kwic", ns.namespace
):
with writer.element("c", ns.namespace, attrs={"type": "left"}):
if left:
writer.characters(left)
with writer.element("kw", ns.namespace):
writer.characters(keyword)
with writer.element("c", ns.namespace, {"type": "right"}):
if right:
writer.characters(right)
FCSRecordXMLStreamWriter.endDataView(writer)
# ----------------------------------------------------
[docs] @staticmethod
def writeSingleHitHitsDataView(
writer: ContentHandler, left: Optional[str], hit: str, right: Optional[str]
) -> None:
"""Convince method to write a simple HITS data view. It
automatically performs the calls to `startDataView` and
`endDataView`.
Args:
writer: the `xml.sax.handler.ContentHandler` to use
left: the left context of the hit or ``None`` if not applicable
hit: the actual hit, that will be highlighted
right: the right context of the hit or ``None`` if not applicable
"""
if writer is None:
raise TypeError("writer is None")
if hit is None:
raise TypeError("hit is None")
writer = XMLStreamWriterHelper(writer)
ns = FCSDataViewNamespaces.HITS
FCSRecordXMLStreamWriter.startDataView(writer, ns.mimetype)
# actual "hits" data view
with writer.prefix(ns.prefix, ns.namespace), writer.element(
"Result", ns.namespace
):
if left and not left.isspace():
writer.characters(left)
with writer.element("Hit", ns.namespace):
writer.characters(hit)
if right and not right.isspace():
writer.characters(right)
FCSRecordXMLStreamWriter.endDataView(writer)
[docs] @staticmethod
def writeHitsDataView(
writer: ContentHandler,
text: str,
hits: List[Tuple[int, int]],
second_is_length: bool,
) -> None:
"""Convince method to write a simple HITS data view. It
automatically performs the calls to `startDataView` and
`endDataView`.
Args:
writer: the `xml.sax.handler.ContentHandler` to use
text: the text content of the hit
hits: a list containing tuples (of start offsets, end offset/length) for the hit markers in the ``text`` content
second_is_length: if ``True`` the second element of each tuple in this ``hits`` array is interpreted as an length; if ``False`` it is interpreted as an end-offset
"""
if writer is None:
raise TypeError("writer is None")
if text is None:
raise TypeError("text is None")
if hits is None:
raise TypeError("hits is None")
if not hits or not all(isinstance(e, tuple) and len(e) == 2 for e in hits):
raise ValueError("hits is empty or not all elements are 2-tuples")
writer = XMLStreamWriterHelper(writer)
ns = FCSDataViewNamespaces.HITS
FCSRecordXMLStreamWriter.startDataView(writer, ns.mimetype)
# actual "hits" data view
with writer.prefix(ns.prefix, ns.namespace), writer.element(
"Result", ns.namespace
):
pos = 0
for start, end in hits:
if start < 0 or start > len(text):
raise ValueError(f"start index out of bounds: {start=}")
if second_is_length:
if end < 1:
raise ValueError(f"length must be larger than 0: length={end}")
end += start
if start >= end:
raise ValueError(
f"end offset must be larger then start offset: {start=}, {end=}"
)
if start > pos:
writer.characters(text[pos:start])
writer.elementcontent("Hit", text[start:end], ns.namespace)
pos = end
if pos < len(text) - 1:
writer.characters(text[pos:])
FCSRecordXMLStreamWriter.endDataView(writer)
# ----------------------------------------------------
[docs] @staticmethod
def writeResourceWithSingleHitHitsDataView(
writer: ContentHandler,
pid: Optional[str],
ref: Optional[str],
left: Optional[str],
hit: str,
right: Optional[str],
):
"""Convince method to write a simple HITS data view. It
automatically performs the calls to `startResource` and
`endResource`.
The following code (arguments omitted) would accomplish the
same result::
...
FCSRecordXMLStreamWriter.startResource(...)
FCSRecordXMLStreamWriter.writeSingleHitHitsDataView(...)
FCSRecordXMLStreamWriter.endResource(...)
...
Args:
writer: the `xml.sax.handler.ContentHandler` to use
pid: the persistent identifier of this resource or ``None``, if not applicable.
ref: the reference of this resource or ``None``, if not applicable.
left: the left context of the hit or ``None`` if not applicable
hit: the actual hit, that will be highlighted
right: the right context of the hit or ``None`` if not applicable
Raises:
TypeError: if writer is None
"""
if writer is None:
raise TypeError("writer is None")
FCSRecordXMLStreamWriter.startResource(writer, pid=pid, ref=ref)
FCSRecordXMLStreamWriter.writeSingleHitHitsDataView(
writer, left=left, hit=hit, right=right
)
FCSRecordXMLStreamWriter.endResource(writer)
[docs] @staticmethod
def writeResourceWithHitsDataView(
writer: ContentHandler,
pid: Optional[str],
ref: Optional[str],
text: str,
hits: List[Tuple[int, int]],
second_is_length: bool,
):
"""Convince method to write a simple HITS data view. It
automatically performs the calls to `startResource` and
`endResource`.
The following code (arguments omitted) would accomplish the
same result::
...
FCSRecordXMLStreamWriter.startResource(...)
FCSRecordXMLStreamWriter.writeHitsDataView(...)
FCSRecordXMLStreamWriter.endResource(...)
...
Args:
writer: the `xml.sax.handler.ContentHandler` to use
pid: the persistent identifier of this resource or ``None``, if not applicable.
ref: the reference of this resource or ``None``, if not applicable.
text: the text content of the hit
hits: a list containing tuples (of start offsets, end offset/length) for the hit markers in the ``text`` content
second_is_length: if ``True`` the second element of each tuple in this ``hits`` array is interpreted as an length; if ``False`` it is interpreted as an end-offset
Raises:
TypeError: if writer is None
"""
if writer is None:
raise TypeError("writer is None")
FCSRecordXMLStreamWriter.startResource(writer, pid=pid, ref=ref)
FCSRecordXMLStreamWriter.writeHitsDataView(
writer, text=text, hits=hits, second_is_length=second_is_length
)
FCSRecordXMLStreamWriter.endResource(writer)
[docs] @staticmethod
def writeResourceWithKWICDataView(
writer: ContentHandler,
pid: Optional[str],
ref: Optional[str],
left: Optional[str],
keyword: str,
right: Optional[str],
) -> None:
"""[Deprecated]
Convince method for writing a record with a KWIC data view.
The following code (arguments omitted) would accomplish the
same result::
...
FCSRecordXMLStreamWriter.startResource(...)
FCSRecordXMLStreamWriter.writeKWICDataView(...)
FCSRecordXMLStreamWriter.endResource(...)
...
Args:
writer: the `xml.sax.handler.ContentHandler` to use
pid: the persistent identifier of this resource or ``None``, if not applicable.
ref: the reference of this resource or ``None``, if not applicable.
left: the left context of the KWIC or ``None`` if not applicable
keyword: the keyword of the KWIC
right: the right context of the KWIC or ``None`` if not applicable
Note:
Only use, if you want compatability to legacy FCS applications.
"""
if writer is None:
raise TypeError("writer is None")
FCSRecordXMLStreamWriter.startResource(writer, pid=pid, ref=ref)
FCSRecordXMLStreamWriter.writeKWICDataView(
writer, left=left, keyword=keyword, right=right
)
FCSRecordXMLStreamWriter.endResource(writer)
[docs] @staticmethod
def writeResourceWithHitsDataViewLegacy(
writer: ContentHandler,
pid: Optional[str],
ref: Optional[str],
left: Optional[str],
hit: str,
right: Optional[str],
) -> None:
"""[Deprecated]
Convince method for writing a record with a HITS and a
KWIC data view. This method is intended for applications that
want ensure computability to legacy CLARIN-FCS clients.
The following code (arguments omitted) would accomplish the
same result::
...
FCSRecordXMLStreamWriter.startResource(...)
FCSRecordXMLStreamWriter.writeSingleHitHitsDataView(...)
FCSRecordXMLStreamWriter.writeKWICDataView(...)
FCSRecordXMLStreamWriter.endResource(...)
...
Args:
writer: the `xml.sax.handler.ContentHandler` to use
pid: the persistent identifier of this resource or ``None``, if not applicable.
ref: the reference of this resource or ``None``, if not applicable.
left: the left context of the hit or ``None`` if not applicable
hit: the actual hit, that will be highlighted
right: the right context of the hit or ``None`` if not applicable
Note:
Only use, if you want compatability to legacy FCS applications.
"""
if writer is None:
raise TypeError("writer is None")
FCSRecordXMLStreamWriter.startResource(writer, pid=pid, ref=ref)
FCSRecordXMLStreamWriter.writeSingleHitHitsDataView(
writer, left=left, hit=hit, right=right
)
FCSRecordXMLStreamWriter.writeKWICDataView(
writer, left=left, keyword=hit, right=right
)
FCSRecordXMLStreamWriter.endResource(writer)
# ---------------------------------------------------------------------------
# ADV dataview
INITIAL_SEGMENT_ID = -1
NO_HIGHLIGHT = -1
[docs]class SpanOffsetUnit(str, Enum):
ITEM = "item"
TIMESTAMP = "timestamp"
[docs]@dataclass(eq=True) # (frozen=True)
class Segment:
id: Union[str, int]
start: int
end: int
# FIXME: add API to set reference
ref: Optional[str] = None
def __post_init__(self):
# object.__setattr__(self, "id", f"s{self.id:x}")
if isinstance(self.id, int):
self.id = f"s{self.id:x}"
[docs]@dataclass
class Span:
segment: Segment
value: Optional[str]
altValue: Optional[str]
highlight: Optional[Union[str, int]]
def __post_init__(self):
# object.__setattr__(self, "highlight", f"h{self.highlight:x}")
if self.highlight is not None and isinstance(self.highlight, int):
if self.highlight != NO_HIGHLIGHT:
self.highlight = f"h{self.highlight:x}"
else:
self.highlight = None
[docs]class AdvancedDataViewWriter:
"""Helper class for serializing Advanced Data Views. It can be
used for writing more than once, but it is **not thread-save**.
This helper can also serialize HITS Data Views.
"""
def __init__(self, unit: SpanOffsetUnit) -> None:
"""[Constructor]
Args:
unit: the unit to be used for span offsets
Raises:
TypeError: if unit is None
"""
if unit is None:
raise TypeError("unit is None")
self.unit = unit
"""the unit to be used for span offsets"""
self.next_segment_id = INITIAL_SEGMENT_ID
self.segments: List[Segment] = list()
self.layers: Dict[str, List[Span]] = dict()
[docs] def reset(self):
"""Reset the writer for writing a new data view (instance)."""
self.next_segment_id = INITIAL_SEGMENT_ID
[docs] def addSpan(
self,
layer_id: str,
start: int,
end: int,
value: Optional[str] = None,
altValue: Optional[str] = None,
highlight: Optional[int] = NO_HIGHLIGHT,
):
"""Add a span.
Args:
layer_id: the span's layer id
start: the span's start offset
end: the span's end offset
value: the span's content value or ``None``
altValue: the span's alternate value or ``None``
highlight: the span's alternate value or ``None``
"""
if layer_id is None:
raise TypeError("layer_id is None")
if start < 0:
raise ValueError("start < 0")
if end < start:
raise ValueError("end < start")
if highlight is None or highlight <= 0:
highlight = NO_HIGHLIGHT
# find segment or create a new one
segment: Segment
for seg in self.segments:
if seg.start == start and seg.end == end:
segment = seg
break
else:
# if not segment:
segment = Segment(self.next_segment_id, start, end)
self.segments.append(segment)
self.next_segment_id += 1
# find layer or create a new one
layer: Optional[List[Span]] = self.layers.get(layer_id, None)
if layer is None:
layer = list()
self.layers[layer_id] = layer
# sanity check (better overlap check?)
for span in layer:
if segment == span.segment:
# FIXME: better exception!
raise ValueError("segment already exists in layer")
layer.append(Span(segment, value, altValue, highlight))
# ----------------------------------------------------
[docs] def writeAdvancedDataView(self, writer: ContentHandler):
"""Write the Advanced Data View to the output stream.
Args:
writer: the `xml.sax.handler.ContentHandler` to use
"""
if writer is None:
raise TypeError("writer is None")
writer = XMLStreamWriterHelper(writer)
ns = FCSDataViewNamespaces.ADV
FCSRecordXMLStreamWriter.startDataView(writer, ns.mimetype)
with writer.prefix(ns.prefix, ns.namespace), writer.element(
"Advanced", ns.namespace, attrs={"unit": self.unit.value}
):
# segments
with writer.element("Segments", ns.namespace):
for segment in self.segments:
# FIXME: unit translation (long -> time)
attrs = {
"id": segment.id,
"start": str(segment.start),
"end": str(segment.end),
}
if segment.ref:
attrs["ref"] = segment.ref
writer.startElementNS((ns.namespace, "Segment"), attrs=attrs)
writer.endElementNS((ns.namespace, "Segment"))
# layers
with writer.element("Layers", ns.namespace):
for layer_id, layer in self.layers.items():
with writer.element("Layer", ns.namespace, attrs={"id": layer_id}):
for span in layer:
attrs = {"ref": span.segment.id}
if span.highlight is not None:
attrs["highlight"] = span.highlight
if span.altValue is not None:
attrs["alt-value"] = span.altValue
with writer.element("Span", ns.namespace, attrs=attrs):
if span.value and not span.value.isspace():
writer.characters(span.value)
FCSRecordXMLStreamWriter.endDataView(writer)
[docs] def writeHitsDataView(self, writer: ContentHandler, layer_id: str):
"""Convenience method to write HITS Data View.
Args:
writer: the `xml.sax.handler.ContentHandler` to use
layer_id: the layer id of the layer to be serialized as HITS Data View
"""
if writer is None:
raise TypeError("writer is None")
if layer_id is None:
raise TypeError("layer_id is None")
spans = self.layers.get(layer_id, None)
if not spans:
raise KeyError(f"layer with id '{layer_id}' does not exist")
writer = XMLStreamWriterHelper(writer)
ns = FCSDataViewNamespaces.HITS
FCSRecordXMLStreamWriter.startDataView(writer, ns.mimetype)
with writer.prefix(ns.prefix, ns.namespace), writer.element(
"Result", ns.namespace
):
need_space = False
for span in spans:
if need_space:
writer.characters(" ")
need_space = False
if span.highlight:
with writer.element("Hit", ns.namespace):
writer.characters(span.value)
need_space = True
else:
writer.characters(span.value)
if span.value and not span.value[-1].isspace():
need_space = True
FCSRecordXMLStreamWriter.endDataView(writer)
# ---------------------------------------------------------------------------