# This file is part of ZTM. # # ZTM is free software; you can redistribute it and/or modify # it under the terms of the GNU General Public License as published by # the Free Software Foundation; either version 2 of the License, or # (at your option) any later version. # # ZTM is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with ZTM; if not, write to the Free Software # Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA #TODO: Variants support #TODO: Define a decorator API. #TODO: Unicode handling of Locators. (Not a huge problem since they are almost always ASCII internally.) from cStringIO import StringIO from codecs import utf_8_encode, latin_1_decode, unicode_internal_encode from zLOG import * from Acquisition import aq_base from XMLWriter import XMLWriter from random import randint from time import time import xtm from Products.CMFCore.WorkflowCore import WorkflowException from Products.ZTM2.Utils import generateRandomId, wrapObject from Products.ZTM2.ZTopic import ZTopic as Topic from Products.ZTM2.ZBaseName import ZBaseName as BaseName from Products.ZTM2.ZOccurrence import ZOccurrence as Occurrence from Products.ZTM2.ZTopicMap import ZTopicMap as TopicMap from Products.ZTM2.ZAssociation import ZAssociation as Association from Products.ZTM2.ZAssociationRole import ZAssociationRole as AssociationRole from Products.ZTM2.Locator import Locator def getXTMId(context): """ Return id for XTM purposes """ #XXX: This should attempt to use the getId method for all topipcs. # Now it only uses it for the system topicmaps. #TODO: We should also support Ontopias ID scheme with a starting 'a' for associations # and a starting 't' for topics. if getattr(aq_base(context), '_occurrences', 0): return "t%s"%context.tm_serial elif getattr(aq_base(context), '_roles', 0): return "a%s"%context.tm_serial else: return "tmo%s"%context.tm_serial def unicodify(data): return latin_1_decode( data )[0] class TypedBaseName: """ """ def __init__(self, id, path): self.id = id self.path = path def __getattr__(self, name): if name == 'tm_serial': return str(time())+str(randint(0,1000)) raise AttributeError, name def getData(self): return self.id def getType(self): return SubjectIdentifierTopic(Locator('http://psi.emnekart.no/ztm/core/#x-zope-id')) def getScope(self): return [] def getVariants(self): return [TypedVariant(self.path)] def getXTMId(self): return '' class TypedParameter: """ """ def __init__(self, scope): self.scope = scope def getScope(self): return [self.scope] def getXTMId(self): return '' def __getattr__(self, name): if name == 'tm_serial': return str(time())+str(randint(0,1000)) raise AttributeError, name class TypedVariant: """ """ def __init__(self, path): self.path = path def getData(self): return self.path def getLocator(self): return None def getParameters(self): return [ TypedParameter( SubjectIdentifierTopic(Locator('http://psi.emnekart.no/ztm/core/#x-zope-path')) ) ] def getVariants(self): return [] def getXTMId(self): return '' def __getattr__(self, name): if name == 'tm_serial': return str(time())+str(randint(0,1000)) raise AttributeError, name class TypedBaseName: """ """ def __init__(self, id, path): self.id = id self.path = path def getData(self): return self.id def getType(self): return SubjectIdentifierTopic(Locator('http://psi.emnekart.no/ztm/core/#x-zope-id')) def getScope(self): return [] def getVariants(self): return [TypedVariant(self.path)] def getXTMId(self): return '' class TypedParameter: """ """ def __init__(self, scope): self.scope = scope def getScope(self): return [self.scope] def getXTMId(self): return '' class TypedVariant: """ """ def __init__(self, path): self.path = path def getData(self): return self.path def getLocator(self): return None #return Locator(self.path) def getParameters(self): return [ TypedParameter( SubjectIdentifierTopic(Locator('http://psi.emnekart.no/ztm/core/#x-zope-path')) ) ] def getVariants(self): return [] def getXTMId(self): return '' class DecorateTopic: def __init__(self, topic): self._topic = topic def __getattr__(self, name): return getattr(self._topic, name) def getBaseNames(self): basenames = self._topic.getBaseNames()[:] basenames.append( TypedBaseName(self.getId(), '/'.join(self._topic.getPhysicalPath()) )) return basenames def getOccurrences(self): extra_occurrences = [] topic = self._topic # Set current workflow-state pw = getattr(self._topic, 'portal_workflow', None) if pw: state = pw.getInfoFor(self._topic, 'review_state', None) if state: occ = ExtraOccurrence() #TODO: Set the correct workflow occ.setType(SubjectIdentifierTopic(Locator('http://psi.forskning.no/workflow/#ztmdefault'))) occ.setData(state) extra_occurrences.append(occ) #TODO: Workflow history if pw: history = pw.getInfoFor(topic, 'review_history', []) for event in history: tmp = {'time': '' ,'action': '', 'review_state':'', 'actor':'','comments':''} tmp.update(event) data = "%(time)s:-:%(action)s:-:%(review_state)s:-:%(actor)s:-:%(comments)s"%tmp occ = ExtraOccurrence() occ.setData(data) occ.setType(SubjectIdentifierTopic(Locator('http://psi.forskning.no/workflow/#history'))) extra_occurrences.append(occ) #TODO: Effective Date effective = topic.EffectiveDate() if effective: occ = ExtraOccurrence() occ.setData(effective) occ.setType(SubjectIdentifierTopic(Locator('http://purl.org/dc/elements/1.1/effectivedate'))) extra_occurrences.append(occ) #TODO: Expiration Date expiration = topic.ExpirationDate() if expiration: occ = ExtraOccurrence() occ.setData(expiration) occ.setType(SubjectIdentifierTopic(Locator('http://purl.org/dc/elements/1.1/expirationdate'))) extra_occurrences.append(occ) #TODO: Creation Date creation = topic.CreationDate() if creation: occ = ExtraOccurrence() occ.setData(creation) occ.setType(SubjectIdentifierTopic(Locator('http://purl.org/dc/elements/1.1/creationdate'))) extra_occurrences.append(occ) #TODO: Creator & Owners try: creator = topic.Creator() except: udb, creator = aq_get(topic, '_owner', None, 1) if creator: occ = ExtraOccurrence() occ.setData(creator) occ.setType(SubjectIdentifierTopic(Locator('http://purl.org/dc/elements/1.1/creator'))) extra_occurrences.append(occ) #TODO: ModificationDate description = topic.ModificationDate() if description: occ = ExtraOccurrence() occ.setData(description) occ.setType(SubjectIdentifierTopic(Locator('http://purl.org/dc/elements/1.1/modificationdate'))) extra_occurrences.append(occ) #TODO: Description description = topic.Description() if description: occ = ExtraOccurrence() occ.setData(description) occ.setType(SubjectIdentifierTopic(Locator('http://purl.org/dc/elements/1.1/description'))) extra_occurrences.append(occ) #TODO: templates if getattr(aq_base( topic ), 'condensed_view_template', None): pass if getattr(aq_base( topic ), 'collapsed_view_template', None): pass if getattr(aq_base( topic ), 'full_view_template', None): pass return self._topic.getOccurrences()[:] + extra_occurrences class Locator: def __init__(self, url): self._url = url def __repr__(self): return ""%self._url def __str__(self): return self._url def getAddress(self): return self._url class ExtraOccurrence: def __init__(self): self._type = None self._data = '' self._locator = None self._scope = [] def isLocator(self): return self._locator def setType(self, occtype): self._type = occtype def getType(self): return self._type def setData(self, data): self._data = data self._locator = None def getData(self): return self._data def setLocator(self, locator): self._locator = locator def getLocator(self): return self._locator def setScope(self, scope): self._scope = scope def getScope(self, wrap=0): return self._scope class SubjectIdentifierTopic: tm_serial = False def __init__(self, si): self._si = si def getSubjectIdentifiers(self): return [self._si] def getSubjectLocator(self): return None def getSourceLocators(self): return [] MARKER = [] class XTMExport: def __init__(self, topicmap, output=None, encoding='utf-8'): """ Initializes an XTM export. """ #By default we print to Standard Out. if not output: output = sys.stdout self._topicmap = topicmap self._encoding = encoding self._writer = XMLWriter(output, encoding) self._writer.doctype(xtm.root, xtm.pubid, xtm.sysid) def export(self, topics=MARKER, associations=MARKER): """ Exports the topic in XTM 1.0 format. If 'topics' not defined (or MARKER), the method will default to all topics in the topicmap. If 'topics' is an empty list or sequence, no topics will be included. If 'topics' is a sequence filled with topics, then only those topics will be exported. if 'associations' is not defined, the method will default to all topics in the topicmap. If 'associations' is an empty list or sequence, no associations will be included. If 'associations' is a sequence filled with associations, then only those associations will be exported. If 'associations' is None, only associations found in 'topics' will be used. #XXX: The last one may not be intuitive. """ topicdict = {} assocdict = {} # If the topics or associations is not specified, that means we will # include all topics or all associations in the topicmap. if topics is MARKER: #XXX: Should this perhaps be wrapped? #XXX: We ought to use a generator... topics = self._topicmap.getTopics() if associations is MARKER: #XXX: Should this perhaps be wrapped? #XXX: We ought to consider generators associations = self._topicmap.getAssociations() # First we set up the root object reifier = self._topicmap.getReifyingTopic() if reifier: #TODO: Quote basename topicmapid = reifier.getBaseName() else: topicmapid = u"ztmtopicmap" self._writer.push(u'topicMap', xtm.topicmap_attrs) # We loop through all topics, and build the topic list. # This way we partially support XTM-Fragments, for use like exporting # the system-ontology. #XXX: This step can be avoided with some clever use of internal BTrees. for topic in topics: topicdict[topic.tm_serial] = topic # Since associations is None, we if associations is None: for assoc in topic.getAssociations(): assocdict[assoc.tm_serial] = assoc # We decorate the topic so that we can control its look. # This is done to add data that is not regularly kept in the topicmap, but that still is interesting to export. # It can also be done to make system-specific inferencing explicit to other TM processors. for topic in topics: topic = wrapObject(self._topicmap, topic) #DecorateTopic(wrapObject(self._topicmap, topic)) self.writeTopic(topic, topicdict) # Finally we output all associations. if assocdict: for association in associations.values(): self.writeAssociations(association, topicdict) else: for association in associations: self.writeAssociations(association, topicdict) self._writer.pop() # def writeTopic(self, topic, topics): """ The method writes a full XTM representation for a Topic. """ # We attemtp to retrieve a nice id. xtmid = getXTMId( topic ) self._writer.push(u'topic', {'id':xtmid}) # # First we output the topics types as an optional and repeatable list of elements types = topic.getTypes() for topictype in types: self._writer.push(u'instanceOf') # if topictype.tm_serial in topics: # We use topicRef for internal links typeXTMId = getXTMId(topictype) self._writer.empty(u'topicRef', {'xlink:href':"#%s"%typeXTMId}) else: # In a normal topicmap, this should never be needed. # It may however be useful for exporting XTM-fragments # We use subjectIndicatorRef for external types. sis = topictype.getSubjectIdentifiers() if sis: si = sis[0] self._writer.empty(u'subjectIndicatorRef', {'xlink:href':si.getAddress()}) else: # Hmm, there is no SI or SA... Now what? # We could include the type, but we fail for now, # since this is a reminder that system-topics should have a PSI. # Garshol thinks that XTM-Fragments need another method to # exchange this anyway, so our XTM-fragment approach can # be considered broken. (Useful though :-) raise AssertionError, "No PSI available, topic (%r) would loose type (%r)."%(topic, topictype) self._writer.pop() # # Subject identity self._writer.push(u'subjectIdentity') #XXX: Not sure if we treat the source locator correctly here. # When reading the spec, it seems that topicRef should refer to topics that must be merged with this one. # Source locators for sl in topic.getSourceLocators(): self._writer.empty(u'topicRef', {'xlink:href':sl.getAddress()}) # First we verify that SI is not set when there is an SA and vice versa si = topic.getSubjectIdentifiers() sa = topic.getSubjectLocator() if si and sa: raise TopicMapError, "Topic %r has both subject identifiers and a subject locator."%(topic) #Subject identifiers for si in topic.getSubjectIdentifiers(): self._writer.empty(u'subjectIndicatorRef', {'xlink:href':si.getAddress()}) # Subject Address if sa: self._writer.empty(u'resourceRef', {'xlink:href':sa.getAddress()}) self._writer.pop() # # Basenames for basename in topic.getBaseNames(): self._writer.push(u'baseName', {'id':getXTMId(basename)}) self.writeScope([wrapObject(self._topicmap, obj) for obj in basename.getScope(wrap=0)], topics) self._writer.elem(u'baseNameString', unicodify(basename.getData())) for variant in basename.getVariants(): self.writeVariant(variant, topics) self._writer.pop() # # Occurrences for occurrence in topic.getOccurrences(): self._writer.push(u'occurrence') #TODO: Add id and sourcelocator... # instanceOf occtype = occurrence.getType() if occtype: self._writer.push(u'instanceOf') if occtype.tm_serial in topics: # We use topicRef for internal links typextmid = getXTMId(occtype) self._writer.empty(u'topicRef', {'xlink:href':"#%s"%typextmid}) else: # We use subjectIndicatorRef for external types. #XXX:Same issues as a topics type. sis = occtype.getSubjectIdentifiers() if sis: si = sis[0] try: self._writer.empty(u'subjectIndicatorRef', {'xlink:href':si.getAddress()}) except AttributeError: raise AssertionError, repr(occtype) else: raise AssertionError("No SI available on occurrencetype %s, topic %s would loose type."(occtype,self._topic)) self._writer.pop() # # scope self.writeScope([wrapObject(self._topicmap, obj) for obj in occurrence.getScope(wrap=0)], topics) # resourceRef or resourceData if occurrence.isLocator(): self._writer.empty(u'resourceRef', {'xlink:href':occurrence.getLocator().getAddress()}) else: #TODO: We need to find out how to treat this. It is supposed to be PCData IIRC. self._writer.elem(u'resourceData', unicodify(occurrence.getData())) self._writer.pop() # self._writer.pop() # def writeAssociations(self, association, topics): roles = association.getRoles(wrap=1) if not roles: # We do not include associations without roles. #XXX: Are associations without roles even allowed in TMDM? Member is mandatory in XTM 1.0 return associd = getXTMId(association) self._writer.push(u'association', {u'id': associd}) # # instanceOf assoctype = association.getType() if assoctype: self._writer.push(u'instanceOf') if assoctype.tm_serial in topics: # We use topicRef for internal links typextmid = getXTMId(assoctype) self._writer.empty(u'topicRef', {'xlink:href':"#%s"%typextmid}) else: # We use subjectIndicatorRef for external types. sis = assoctype.getSubjectIdentifiers() if sis: si = sis[0] self.writer.empty(u'subjectIndicatorRef', {'xlink:href':si.getAddress()}) else: raise AssertionError, "No PSI available, topic would loose type." self._writer.pop() # instanceOf # scope self.writeScope(association.getScope(), topics) # member for role in roles: self._writer.push(u'member') # rolespec = role.getRoleSpec() if rolespec: self._writer.push(u'roleSpec') # if rolespec.tm_serial in topics: # We use topicRef for internal links typextmid = getXTMId(rolespec) self._writer.empty(u'topicRef', {'xlink:href':"#%s"%typextmid}) else: # We use subjectIndicatorRef for external types. sis = rolespec.getSubjectIdentifiers() if sis: si = sis[0] self.writer.empty(u'subjectIndicatorRef', {'xlink:href':si.getAddress()}) else: raise AssertionError, "No PSI available, topic would loose type." self._writer.pop() # player = role.getPlayer() if player: if player.tm_serial in topics: # We use topicRef for internal links typextmid = getXTMId(player) self._writer.empty(u'topicRef', {'xlink:href':"#%s"%typextmid}) else: # We use subjectIndicatorRef for external types. sis = player.getSubjectIdentifiers() if sis: si = sis[0] self._writer.empty(u'subjectIndicatorRef', {'xlink:href':si.getAddress()}) else: sa = player.getSubjectLocator() if sa: self._writer.empty(u'resourceRef', {'xlink:href': sa.getAddress()}) else: # Hmm, there is no SI or SA... Now what? raise AssertionError, "No PSI available, topic (%r) would loose type (%r)."%(topic, topictype) self._writer.pop() # self._writer.pop() # def writeVariant(self, variant, topics): #TODO: Variant output needs to be checked, since we've never used variants. self._writer.push(u'variant', {'id':getXTMId(variant)}) self._writer.push(u'parameters') variantname = variant.getData() for parameter in variant.getScope(full=False): for theme in parameter.getScope(): if theme.tm_serial in topics: # We use topicRef for internal links typextmid = getXTMId(theme) self._writer.empty(u'topicRef', {'xlink:href':"#%s"%typextmid}) else: # We use subjectIndicatorRef for external types. for si in theme.getSubjectIdentifiers(): self.writer.empty(u'subjectIndicatorRef', {'xlink:href':si.getAddress()}) if variant.isLocator(): self._writer.empty(u'resourceRef', {'xlink:href':variant.getLocator().getAddress()}) else: self._writer.elem(u'resourceData', variant.getData() or u'') # It seems that according to XTM 1.0 variants can be deeply nested structures. # for variant in variant.getVariants(): # self.writeVariant(variant, topics) self._writer.pop() # self._writer.pop() # def writeScope(self, scopeset, topics): for theme in scopeset: self._writer.push(u'scope') # if theme.tm_serial in topics: # We use topicRef for internal links typextmid = getXTMId(theme) self._writer.empty(u'topicRef', {'xlink:href':"#%s"%typextmid}) else: # We use subjectIndicatorRef for external types. sis = theme.getSubjectIdentifiers() if sis: si = sis[0] self.writer.empty(u'subjectIndicatorRef', {'xlink:href':si.getAddress()}) else: sa = theme.getSubjectLocator(self) if sa: self.writer.empty(u'resourceRef', {'xlink:href':sa.getAddress()}) else: # Hmm, there is no SI or SA... Now what? raise AssertionError, "No PSI available, topic would loose type." self._writer.pop() #