1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298
|
# encoding: utf-8
"""
Provides a low-level, read-only API to a serialized Open Packaging Convention
(OPC) package.
"""
from __future__ import absolute_import
from .constants import RELATIONSHIP_TARGET_MODE as RTM
from .oxml import parse_xml
from .packuri import PACKAGE_URI, PackURI
from .phys_pkg import PhysPkgReader
from .shared import CaseInsensitiveDict
class PackageReader(object):
"""
Provides access to the contents of a zip-format OPC package via its
:attr:`serialized_parts` and :attr:`pkg_srels` attributes.
"""
def __init__(self, content_types, pkg_srels, sparts):
super(PackageReader, self).__init__()
self._pkg_srels = pkg_srels
self._sparts = sparts
@staticmethod
def from_file(pkg_file):
"""
Return a |PackageReader| instance loaded with contents of *pkg_file*.
"""
phys_reader = PhysPkgReader(pkg_file)
content_types = _ContentTypeMap.from_xml(phys_reader.content_types_xml)
pkg_srels = PackageReader._srels_for(phys_reader, PACKAGE_URI)
sparts = PackageReader._load_serialized_parts(
phys_reader, pkg_srels, content_types
)
phys_reader.close()
return PackageReader(content_types, pkg_srels, sparts)
def iter_sparts(self):
"""
Generate a 4-tuple `(partname, content_type, reltype, blob)` for each
of the serialized parts in the package.
"""
for s in self._sparts:
yield (s.partname, s.content_type, s.reltype, s.blob)
def iter_srels(self):
"""
Generate a 2-tuple `(source_uri, srel)` for each of the relationships
in the package.
"""
for srel in self._pkg_srels:
yield (PACKAGE_URI, srel)
for spart in self._sparts:
for srel in spart.srels:
yield (spart.partname, srel)
@staticmethod
def _load_serialized_parts(phys_reader, pkg_srels, content_types):
"""
Return a list of |_SerializedPart| instances corresponding to the
parts in *phys_reader* accessible by walking the relationship graph
starting with *pkg_srels*.
"""
sparts = []
part_walker = PackageReader._walk_phys_parts(phys_reader, pkg_srels)
for partname, blob, reltype, srels in part_walker:
content_type = content_types[partname]
spart = _SerializedPart(
partname, content_type, reltype, blob, srels
)
sparts.append(spart)
return tuple(sparts)
@staticmethod
def _srels_for(phys_reader, source_uri):
"""
Return |_SerializedRelationships| instance populated with
relationships for source identified by *source_uri*.
"""
rels_xml = phys_reader.rels_xml_for(source_uri)
return _SerializedRelationships.load_from_xml(
source_uri.baseURI, rels_xml)
@staticmethod
def _walk_phys_parts(phys_reader, srels, visited_partnames=None):
"""
Generate a 4-tuple `(partname, blob, reltype, srels)` for each of the
parts in *phys_reader* by walking the relationship graph rooted at
srels.
"""
if visited_partnames is None:
visited_partnames = []
for srel in srels:
if srel.is_external:
continue
partname = srel.target_partname
if partname in visited_partnames:
continue
visited_partnames.append(partname)
reltype = srel.reltype
part_srels = PackageReader._srels_for(phys_reader, partname)
blob = phys_reader.blob_for(partname)
yield (partname, blob, reltype, part_srels)
next_walker = PackageReader._walk_phys_parts(
phys_reader, part_srels, visited_partnames
)
for partname, blob, reltype, srels in next_walker:
yield (partname, blob, reltype, srels)
class _ContentTypeMap(object):
"""
Value type providing dictionary semantics for looking up content type by
part name, e.g. ``content_type = cti['/ppt/presentation.xml']``.
"""
def __init__(self):
super(_ContentTypeMap, self).__init__()
self._overrides = CaseInsensitiveDict()
self._defaults = CaseInsensitiveDict()
def __getitem__(self, partname):
"""
Return content type for part identified by *partname*.
"""
if not isinstance(partname, PackURI):
tmpl = "_ContentTypeMap key must be <type 'PackURI'>, got %s"
raise KeyError(tmpl % type(partname))
if partname in self._overrides:
return self._overrides[partname]
if partname.ext in self._defaults:
return self._defaults[partname.ext]
tmpl = "no content type for partname '%s' in [Content_Types].xml"
raise KeyError(tmpl % partname)
@staticmethod
def from_xml(content_types_xml):
"""
Return a new |_ContentTypeMap| instance populated with the contents
of *content_types_xml*.
"""
types_elm = parse_xml(content_types_xml)
ct_map = _ContentTypeMap()
for o in types_elm.overrides:
ct_map._add_override(o.partname, o.content_type)
for d in types_elm.defaults:
ct_map._add_default(d.extension, d.content_type)
return ct_map
def _add_default(self, extension, content_type):
"""
Add the default mapping of *extension* to *content_type* to this
content type mapping.
"""
self._defaults[extension] = content_type
def _add_override(self, partname, content_type):
"""
Add the default mapping of *partname* to *content_type* to this
content type mapping.
"""
self._overrides[partname] = content_type
class _SerializedPart(object):
"""
Value object for an OPC package part. Provides access to the partname,
content type, blob, and serialized relationships for the part.
"""
def __init__(self, partname, content_type, reltype, blob, srels):
super(_SerializedPart, self).__init__()
self._partname = partname
self._content_type = content_type
self._reltype = reltype
self._blob = blob
self._srels = srels
@property
def partname(self):
return self._partname
@property
def content_type(self):
return self._content_type
@property
def blob(self):
return self._blob
@property
def reltype(self):
"""
The referring relationship type of this part.
"""
return self._reltype
@property
def srels(self):
return self._srels
class _SerializedRelationship(object):
"""
Value object representing a serialized relationship in an OPC package.
Serialized, in this case, means any target part is referred to via its
partname rather than a direct link to an in-memory |Part| object.
"""
def __init__(self, baseURI, rel_elm):
super(_SerializedRelationship, self).__init__()
self._baseURI = baseURI
self._rId = rel_elm.rId
self._reltype = rel_elm.reltype
self._target_mode = rel_elm.target_mode
self._target_ref = rel_elm.target_ref
@property
def is_external(self):
"""
True if target_mode is ``RTM.EXTERNAL``
"""
return self._target_mode == RTM.EXTERNAL
@property
def reltype(self):
"""Relationship type, like ``RT.OFFICE_DOCUMENT``"""
return self._reltype
@property
def rId(self):
"""
Relationship id, like 'rId9', corresponds to the ``Id`` attribute on
the ``CT_Relationship`` element.
"""
return self._rId
@property
def target_mode(self):
"""
String in ``TargetMode`` attribute of ``CT_Relationship`` element,
one of ``RTM.INTERNAL`` or ``RTM.EXTERNAL``.
"""
return self._target_mode
@property
def target_ref(self):
"""
String in ``Target`` attribute of ``CT_Relationship`` element, a
relative part reference for internal target mode or an arbitrary URI,
e.g. an HTTP URL, for external target mode.
"""
return self._target_ref
@property
def target_partname(self):
"""
|PackURI| instance containing partname targeted by this relationship.
Raises ``ValueError`` on reference if target_mode is ``'External'``.
Use :attr:`target_mode` to check before referencing.
"""
if self.is_external:
msg = ('target_partname attribute on Relationship is undefined w'
'here TargetMode == "External"')
raise ValueError(msg)
# lazy-load _target_partname attribute
if not hasattr(self, '_target_partname'):
self._target_partname = PackURI.from_rel_ref(self._baseURI,
self.target_ref)
return self._target_partname
class _SerializedRelationships(object):
"""
Read-only sequence of |_SerializedRelationship| instances corresponding
to the relationships item XML passed to constructor.
"""
def __init__(self):
super(_SerializedRelationships, self).__init__()
self._srels = []
def __iter__(self):
"""Support iteration, e.g. 'for x in srels:'"""
return self._srels.__iter__()
@staticmethod
def load_from_xml(baseURI, rels_item_xml):
"""
Return |_SerializedRelationships| instance loaded with the
relationships contained in *rels_item_xml*. Returns an empty
collection if *rels_item_xml* is |None|.
"""
srels = _SerializedRelationships()
if rels_item_xml is not None:
rels_elm = parse_xml(rels_item_xml)
for rel_elm in rels_elm.Relationship_lst:
srels._srels.append(_SerializedRelationship(baseURI, rel_elm))
return srels
|