marco
e489b6a2e2
Add core Python agent with configuration management, environment setup, and deployment documentation. Includes user/agent profiles, startup scripts, and dependency management for the psagentia system.
220 lines
8.5 KiB
Python
Executable File
220 lines
8.5 KiB
Python
Executable File
"""Objects that implement reading and writing OPC packages."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import IO, TYPE_CHECKING, Iterator, cast
|
|
|
|
from docx.opc.constants import RELATIONSHIP_TYPE as RT
|
|
from docx.opc.packuri import PACKAGE_URI, PackURI
|
|
from docx.opc.part import PartFactory
|
|
from docx.opc.parts.coreprops import CorePropertiesPart
|
|
from docx.opc.pkgreader import PackageReader
|
|
from docx.opc.pkgwriter import PackageWriter
|
|
from docx.opc.rel import Relationships
|
|
from docx.shared import lazyproperty
|
|
|
|
if TYPE_CHECKING:
|
|
from typing_extensions import Self
|
|
|
|
from docx.opc.coreprops import CoreProperties
|
|
from docx.opc.part import Part
|
|
from docx.opc.rel import _Relationship # pyright: ignore[reportPrivateUsage]
|
|
|
|
|
|
class OpcPackage:
|
|
"""Main API class for |python-opc|.
|
|
|
|
A new instance is constructed by calling the :meth:`open` class method with a path
|
|
to a package file or file-like object containing one.
|
|
"""
|
|
|
|
def after_unmarshal(self):
|
|
"""Entry point for any post-unmarshaling processing.
|
|
|
|
May be overridden by subclasses without forwarding call to super.
|
|
"""
|
|
# don't place any code here, just catch call if not overridden by
|
|
# subclass
|
|
pass
|
|
|
|
@property
|
|
def core_properties(self) -> CoreProperties:
|
|
"""|CoreProperties| object providing read/write access to the Dublin Core
|
|
properties for this document."""
|
|
return self._core_properties_part.core_properties
|
|
|
|
def iter_rels(self) -> Iterator[_Relationship]:
|
|
"""Generate exactly one reference to each relationship in the package by
|
|
performing a depth-first traversal of the rels graph."""
|
|
|
|
def walk_rels(
|
|
source: OpcPackage | Part, visited: list[Part] | None = None
|
|
) -> Iterator[_Relationship]:
|
|
visited = [] if visited is None else visited
|
|
for rel in source.rels.values():
|
|
yield rel
|
|
if rel.is_external:
|
|
continue
|
|
part = rel.target_part
|
|
if part in visited:
|
|
continue
|
|
visited.append(part)
|
|
new_source = part
|
|
for rel in walk_rels(new_source, visited):
|
|
yield rel
|
|
|
|
for rel in walk_rels(self):
|
|
yield rel
|
|
|
|
def iter_parts(self) -> Iterator[Part]:
|
|
"""Generate exactly one reference to each of the parts in the package by
|
|
performing a depth-first traversal of the rels graph."""
|
|
|
|
def walk_parts(source, visited=[]):
|
|
for rel in source.rels.values():
|
|
if rel.is_external:
|
|
continue
|
|
part = rel.target_part
|
|
if part in visited:
|
|
continue
|
|
visited.append(part)
|
|
yield part
|
|
new_source = part
|
|
for part in walk_parts(new_source, visited):
|
|
yield part
|
|
|
|
for part in walk_parts(self):
|
|
yield part
|
|
|
|
def load_rel(self, reltype: str, target: Part | str, rId: str, is_external: bool = False):
|
|
"""Return newly added |_Relationship| instance of `reltype` between this part
|
|
and `target` with key `rId`.
|
|
|
|
Target mode is set to ``RTM.EXTERNAL`` if `is_external` is |True|. Intended for
|
|
use during load from a serialized package, where the rId is well known. Other
|
|
methods exist for adding a new relationship to the package during processing.
|
|
"""
|
|
return self.rels.add_relationship(reltype, target, rId, is_external)
|
|
|
|
@property
|
|
def main_document_part(self):
|
|
"""Return a reference to the main document part for this package.
|
|
|
|
Examples include a document part for a WordprocessingML package, a presentation
|
|
part for a PresentationML package, or a workbook part for a SpreadsheetML
|
|
package.
|
|
"""
|
|
return self.part_related_by(RT.OFFICE_DOCUMENT)
|
|
|
|
def next_partname(self, template: str) -> PackURI:
|
|
"""Return a |PackURI| instance representing partname matching `template`.
|
|
|
|
The returned part-name has the next available numeric suffix to distinguish it
|
|
from other parts of its type. `template` is a printf (%)-style template string
|
|
containing a single replacement item, a '%d' to be used to insert the integer
|
|
portion of the partname. Example: "/word/header%d.xml"
|
|
"""
|
|
partnames = {part.partname for part in self.iter_parts()}
|
|
for n in range(1, len(partnames) + 2):
|
|
candidate_partname = template % n
|
|
if candidate_partname not in partnames:
|
|
return PackURI(candidate_partname)
|
|
|
|
@classmethod
|
|
def open(cls, pkg_file: str | IO[bytes]) -> Self:
|
|
"""Return an |OpcPackage| instance loaded with the contents of `pkg_file`."""
|
|
pkg_reader = PackageReader.from_file(pkg_file)
|
|
package = cls()
|
|
Unmarshaller.unmarshal(pkg_reader, package, PartFactory)
|
|
return package
|
|
|
|
def part_related_by(self, reltype: str) -> Part:
|
|
"""Return part to which this package has a relationship of `reltype`.
|
|
|
|
Raises |KeyError| if no such relationship is found and |ValueError| if more than
|
|
one such relationship is found.
|
|
"""
|
|
return self.rels.part_with_reltype(reltype)
|
|
|
|
@property
|
|
def parts(self) -> list[Part]:
|
|
"""Return a list containing a reference to each of the parts in this package."""
|
|
return list(self.iter_parts())
|
|
|
|
def relate_to(self, part: Part, reltype: str):
|
|
"""Return rId key of new or existing relationship to `part`.
|
|
|
|
If a relationship of `reltype` to `part` already exists, its rId is returned. Otherwise a
|
|
new relationship is created and that rId is returned.
|
|
"""
|
|
rel = self.rels.get_or_add(reltype, part)
|
|
return rel.rId
|
|
|
|
@lazyproperty
|
|
def rels(self):
|
|
"""Return a reference to the |Relationships| instance holding the collection of
|
|
relationships for this package."""
|
|
return Relationships(PACKAGE_URI.baseURI)
|
|
|
|
def save(self, pkg_file: str | IO[bytes]):
|
|
"""Save this package to `pkg_file`.
|
|
|
|
`pkg_file` can be either a file-path or a file-like object.
|
|
"""
|
|
for part in self.parts:
|
|
part.before_marshal()
|
|
PackageWriter.write(pkg_file, self.rels, self.parts)
|
|
|
|
@property
|
|
def _core_properties_part(self) -> CorePropertiesPart:
|
|
"""|CorePropertiesPart| object related to this package.
|
|
|
|
Creates a default core properties part if one is not present (not common).
|
|
"""
|
|
try:
|
|
return cast(CorePropertiesPart, self.part_related_by(RT.CORE_PROPERTIES))
|
|
except KeyError:
|
|
core_properties_part = CorePropertiesPart.default(self)
|
|
self.relate_to(core_properties_part, RT.CORE_PROPERTIES)
|
|
return core_properties_part
|
|
|
|
|
|
class Unmarshaller:
|
|
"""Hosts static methods for unmarshalling a package from a |PackageReader|."""
|
|
|
|
@staticmethod
|
|
def unmarshal(pkg_reader, package, part_factory):
|
|
"""Construct graph of parts and realized relationships based on the contents of
|
|
`pkg_reader`, delegating construction of each part to `part_factory`.
|
|
|
|
Package relationships are added to `pkg`.
|
|
"""
|
|
parts = Unmarshaller._unmarshal_parts(pkg_reader, package, part_factory)
|
|
Unmarshaller._unmarshal_relationships(pkg_reader, package, parts)
|
|
for part in parts.values():
|
|
part.after_unmarshal()
|
|
package.after_unmarshal()
|
|
|
|
@staticmethod
|
|
def _unmarshal_parts(pkg_reader, package, part_factory):
|
|
"""Return a dictionary of |Part| instances unmarshalled from `pkg_reader`, keyed
|
|
by partname.
|
|
|
|
Side-effect is that each part in `pkg_reader` is constructed using
|
|
`part_factory`.
|
|
"""
|
|
parts = {}
|
|
for partname, content_type, reltype, blob in pkg_reader.iter_sparts():
|
|
parts[partname] = part_factory(partname, content_type, reltype, blob, package)
|
|
return parts
|
|
|
|
@staticmethod
|
|
def _unmarshal_relationships(pkg_reader, package, parts):
|
|
"""Add a relationship to the source object corresponding to each of the
|
|
relationships in `pkg_reader` with its target_part set to the actual target part
|
|
in `parts`."""
|
|
for source_uri, srel in pkg_reader.iter_srels():
|
|
source = package if source_uri == "/" else parts[source_uri]
|
|
target = srel.target_ref if srel.is_external else parts[srel.target_partname]
|
|
source.load_rel(srel.reltype, target, srel.rId, srel.is_external)
|