commit 45818068a86bd0c6cb30e3de53e7ad7ad7d888c3
parent 616c746b100620cd96f2456f694e156fced617fe
Author: lash <dev@holbrook.no>
Date: Sun, 6 Sep 2026 15:34:46 -0600
Complete parsing to attachment
Diffstat:
5 files changed, 79 insertions(+), 86 deletions(-)
diff --git a/adesse/attachment.py b/adesse/attachment.py
@@ -30,6 +30,12 @@ class Attachment(Entity):
raise NotImplementedError()
+ def set_sum_from_urn(self, v):
+ if v[:11] != 'URN:sha256:':
+ raise ValueError('only sha256 urn supported')
+ return self.set_sum(bytes.fromhex(v[11:]))
+
+
def apply(self, g):
uu = to_uuid_uri(self.uu)
g.add((uu, DC.format, Literal(self.mime)))
diff --git a/adesse/ns/__init__.py b/adesse/ns/__init__.py
@@ -7,3 +7,4 @@ from .inc import *
from .wot import WOT
from .df import DF
from .pss import PSS
+from .xmp import XMP
diff --git a/adesse/ns/xmp.py b/adesse/ns/xmp.py
@@ -0,0 +1,10 @@
+from .inc import *
+
+
+class XMP(DefinedNamespace):
+
+ CreateDate: URIRef
+ CreatorTool: URIRef
+ ModifyDate: URIRef
+
+ _NS = Namespace('http://ns.adobe.com/xap/1.0/')
diff --git a/adesse/util/xmp.py b/adesse/util/xmp.py
@@ -1,43 +1,16 @@
import os
import logging
import hashlib
+import mimetypes
+import datetime
import lxml.etree
from rdflib import Graph, Namespace, RDF, URIRef, DC, Literal
from adesse import Attachment
+from adesse.ns import XMP
logg = logging.getLogger('util.xmp')
-EXIFTOOL_XMP = Namespace('http://ns.exiftool.org/XMP/XMP-xmp/1.0/')
-EXIFTOOL_XMPPDF = Namespace('http://ns.exiftool.org/XMP/XMP-pdf/1.0/')
-EXIFTOOL_XMPDC = Namespace('http://ns.exiftool.org/XMP/XMP-dc/1.0/')
-EXIFTOOL_SYSTEM = Namespace('http://ns.exiftool.org/File/System/1.0/')
-
-
-
-def _from_xml(item, o, exiftool=False):
- parsens = DC
- if exiftool:
- parsens = EXIFTOOL_XMPDC
- title = o.value(item, parsens.term('title'))
- if title == None:
- title = o.value(item, parsens.term('Title'))
- logg.debug('have title {} {} {}'.format(item, parsens, title))
- desc = o.value(item, parsens.term('subject'))
- if desc == None:
- desc = o.value(item, parsens.term('Subject'))
- logg.debug('have desc {} {} {}'.format(item, parsens, desc))
-# o = Attachment(title)
-
-
-def check_file(fp, strict=False):
- logg.debug('checi file {}'.format(fp))
- os.stat(fp)
- if not strict:
- return fp
- return fp
-
-
def _parse_xmp(xmp_file, resource_file, search_dir=None):
o = lxml.etree.parse(xmp_file)
@@ -66,8 +39,10 @@ def _parse_xmp(xmp_file, resource_file, search_dir=None):
for v in r.findall('rdf:Description', namespaces={'rdf': str(RDF)}):
v.set('{{{}}}about'.format(str(RDF)), uu)
+ mime = mimetypes.guess_type(resource_file)
+
s = lxml.etree.tostring(r)
- return (uu, s,)
+ return (uu, mime, s,)
@@ -76,7 +51,7 @@ def rdfn(v):
return URIRef(s)
-def _get_alt_lang(sbj, prd, o):
+def _get_alt_lang(sbj, prd, o, lang=None):
v = o.value(sbj, prd)
typ = o.value(v, RDF.type)
r = None
@@ -85,7 +60,7 @@ def _get_alt_lang(sbj, prd, o):
while i:
v = o.value(v, rdfn(i))
if isinstance(v, Literal):
- if v.language == 'x-default':
+ if v.language == 'x-default' or (v.language != None and v.language == lang):
r = v.value
i = 0
else:
@@ -95,12 +70,41 @@ def _get_alt_lang(sbj, prd, o):
return r
-def parse_title(sbj, o):
- return _get_alt_lang(sbj, DC.title, o)
+def _get_bag(sbj, prd, o):
+ v = o.value(sbj, prd)
+ typ = o.value(v, RDF.type)
+ r = []
+ if typ == RDF.Bag:
+ i = 1
+ while i:
+ v = o.value(v, rdfn(i))
+ if v == None:
+ break
+ r.append(v.value)
+ i += 1
+ return r
+
+
+
+def parse_title(sbj, o, lang=None):
+ return _get_alt_lang(sbj, DC.title, o, lang=lang)
+
+def parse_description(sbj, o, lang=None):
+ return _get_alt_lang(sbj, DC.description, o, lang=lang)
-def parse_description(sbj, o):
- return _get_alt_lang(sbj, DC.description, o)
+
+def parse_date(sbj, o):
+ v = o.value(sbj, XMP.CreateDate)
+ return datetime.datetime.fromisoformat(v.value)
+
+
+def scan_lang(sbj, o):
+ r = None
+ v = _get_bag(sbj, DC.language, o)
+ if len(v) == 1:
+ r = v[0]
+ return (r, v,)
def attachment_from_xmp(xml_file, resource_file=None, search_dir=None):
@@ -108,50 +112,18 @@ def attachment_from_xmp(xml_file, resource_file=None, search_dir=None):
search_dir = ['.']
r = _parse_xmp(xml_file, resource_file=resource_file, search_dir=search_dir)
o = Graph()
- sbj = URIRef(r[0])
- o.parse(r[1], format='xml')
- title = parse_title(sbj, o)
- desc = parse_description(sbj, o)
- logg.debug('have title {} desc {}'.format(title, desc))
-
-
-#def attachment_from_xmp(filepath, search_dir=None):
-# if search_dir == None:
-# search_dir = ['.']
-# state = {
-# 'exif': False,
-# 'xmp': False,
-# }
-# items = []
-# o = Graph()
-# r = o.parse(filepath, format='xml')
-# #o.bind('XMP-dc', DC, replace=True)
-# for v in o.namespace_manager.namespaces():
-# logg.debug('have namespace {}'.format(v))
-# k = v[0]
-# if 'ExifTool' in k:
-# state['exif'] = True
-# if 'xmp' in k:
-# state['xmp'] = True
-# if not state['xmp']:
-# raise AttributeError('not an xmp doument')
-# for v in o.subjects():
-# fp = None
-# for d in search_dir:
-# fp = os.path.join(d, v)
-# try:
-# fp = check_file(fp)
-# break
-# except FileNotFoundError:
-# fp = None
-# if fp == None:
-# continue
-#
-# if v not in items:
-# items.append(v)
-# logg.debug('added subject {} {}'.format(v, fp))
-#
-# r = []
-# for v in items:
-# r.append(_from_xml(v, o, exiftool=state['exif']))
-# return r
+ z = r[0]
+ sbj = URIRef(z)
+ mime = r[1]
+ o.parse(r[2], format='xml')
+ (default_lang, langs) = scan_lang(sbj, o)
+ title = parse_title(sbj, o, lang=default_lang)
+ desc = parse_description(sbj, o, lang=default_lang)
+ date = parse_date(sbj, o)
+ # TODO: prefer dc:format
+ mime = str(r[1][0])
+ logg.debug('have title {} desc {} mime {}'.format(title, desc, mime))
+ slug = os.path.splitext(xml_file)[0]
+ v = Attachment(title, desc, date, mime, lang=default_lang)
+ v.set_sum_from_urn(z)
+ return v
diff --git a/test/test_xmp.py b/test/test_xmp.py
@@ -5,9 +5,10 @@ import os
import logging
from rdflib.term import URIRef
+from rdflib import Graph
-from adesse.ns import XMP
from adesse.util import xmp
+from adesse import Serializer
logging.basicConfig(level=logging.DEBUG)
@@ -22,7 +23,10 @@ class TestXmpLoad(unittest.TestCase):
def test_parse(self):
fp = os.path.join(thisdir, 'import.xmp')
- xmp.attachment_from_xmp(fp, 'import.pdf', search_dir=[thisdir])
+ r = xmp.attachment_from_xmp(fp, 'import.pdf', search_dir=[thisdir])
+ z = Serializer()
+ r.apply(z)
+ print(z)
if __name__ == '__main__':
