specify7/specifyweb/backend/export/dwca_utils.py at 104e151e998a98718e40c2febdedfcce6939804f · specify/specify7 · GitHub

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
"""Shared utilities for DwC archive generation."""
import re
from datetime import date
from uuid import uuid4
from xml.etree import ElementTree as ET


def sanitize_term_name(term_iri):
    """Extract the short name from a DwC term IRI.

    'http://rs.tdwg.org/dwc/terms/catalogNumber' -> 'catalogNumber'
    'http://purl.org/dc/terms/type' -> 'type'
    """
    if '/' in term_iri:
        term_iri = term_iri.rsplit('/', 1)[-1]
    if '#' in term_iri:
        term_iri = term_iri.rsplit('#', 1)[-1]
    return term_iri


def sanitize_column_name(name):
    """Sanitize a term IRI into a valid MySQL column name."""
    name = sanitize_term_name(name)
    name = re.sub(r'[^a-zA-Z0-9_]', '_', name)
    return name[:64]


# Known extension rowType URIs
EXTENSION_ROW_TYPES = {
    'MeasurementOrFact': 'http://rs.iobis.org/obis/terms/ExtendedMeasurementOrFact',
    'ResourceRelationship': 'http://rs.tdwg.org/dwc/terms/ResourceRelationship',
    'Identification': 'http://rs.tdwg.org/dwc/terms/Identification',
    'Multimedia': 'http://rs.gbif.org/terms/1.0/Multimedia',
}


def build_meta_xml(core_terms, ext_info_list):
    """Build meta.xml describing the DwC archive structure.

    core_terms: list of full term IRIs for the core file
    ext_info_list: list of dicts with 'filename' and 'terms' (full IRIs)
    """
    archive = ET.Element('archive')
    archive.set('xmlns', 'http://rs.tdwg.org/dwc/text/')
    archive.set('metadata', 'eml.xml')

    # Core
    core = ET.SubElement(archive, 'core')
    core.set('encoding', 'UTF-8')
    core.set('fieldsTerminatedBy', ',')
    core.set('linesTerminatedBy', '\\n')
    core.set('fieldsEnclosedBy', '"')
    core.set('ignoreHeaderLines', '1')
    core.set('rowType', 'http://rs.tdwg.org/dwc/terms/Occurrence')

    files = ET.SubElement(core, 'files')
    location = ET.SubElement(files, 'location')
    location.text = 'occurrence.csv'

    if core_terms:
        id_elem = ET.SubElement(core, 'id')
        id_elem.set('index', '0')

    for idx, term_iri in enumerate(core_terms):
        f = ET.SubElement(core, 'field')
        f.set('index', str(idx))
        f.set('term', term_iri)

    # Extensions
    for ext in ext_info_list:
        extension = ET.SubElement(archive, 'extension')
        extension.set('encoding', 'UTF-8')
        extension.set('fieldsTerminatedBy', ',')
        extension.set('linesTerminatedBy', '\\n')
        extension.set('fieldsEnclosedBy', '"')
        extension.set('ignoreHeaderLines', '1')
        row_type = ext.get('rowType', 'http://rs.tdwg.org/dwc/terms/MeasurementOrFact')
        extension.set('rowType', row_type)

        files = ET.SubElement(extension, 'files')
        location = ET.SubElement(files, 'location')
        location.text = ext['filename']

        coreid = ET.SubElement(extension, 'coreid')
        coreid.set('index', '0')

        for idx, term_iri in enumerate(ext['terms']):
            f = ET.SubElement(extension, 'field')
            f.set('index', str(idx))
            f.set('term', term_iri)

    return ET.tostring(archive, encoding='unicode', xml_declaration=True)


def build_eml_xml(export_dataset):
    """Build EML metadata. Returns custom EML if uploaded, else generates minimal EML."""
    if export_dataset.metadata:
        try:
            from specifyweb.specify.models import Spappresourcedata
            data = Spappresourcedata.objects.filter(
                spappresource=export_dataset.metadata
            ).first()
            if data and data.data:
                content = data.data
                if isinstance(content, bytes):
                    content = content.decode('utf-8')
                return content
        except Exception:
            pass

    eml = ET.Element('eml:eml')
    eml.set('xmlns:eml', 'eml://ecoinformatics.org/eml-2.1.1')
    eml.set('packageId', str(uuid4()))
    eml.set('system', 'http://specify.org')

    dataset = ET.SubElement(eml, 'dataset')
    title = ET.SubElement(dataset, 'title')
    title.text = export_dataset.exportname

    creator = ET.SubElement(dataset, 'creator')
    org = ET.SubElement(creator, 'organizationName')
    org.text = 'Specify Collection'

    pubdate = ET.SubElement(dataset, 'pubDate')
    pubdate.text = date.today().strftime('%Y-%m-%d')

    abstract = ET.SubElement(dataset, 'abstract')
    para = ET.SubElement(abstract, 'para')
    para.text = f'Darwin Core Archive export: {export_dataset.exportname}'

    return ET.tostring(eml, encoding='unicode', xml_declaration=True)