feat: descrição arquivística NOBRADE multinível com exportação EAD 3, EAC-CPF e AtoM
- pipeline/nobrade: normalizadores de datas (ISO 8601-2), séries, autoridades, espécie/suporte, direitos, lugar e assuntos; montagem dos níveis 0–5 com código de referência BR RJXYZ.BF.<seção>.<série>.<handle>.<imagem> - validador dos elementos obrigatórios: 0 pendências em 26.454 unidades - EAD 3 e 457 EAC-CPF validados contra os XSD - mural: cartão com código, data normalizada, série e reprodução; filtro por série
This commit is contained in:
@@ -0,0 +1,352 @@
|
||||
"""Exportação da descrição: EAD 3 (descrições), EAC-CPF (autoridades) e CSV
|
||||
de importação do AtoM (modelo ISAD). A NOBRADE não impõe formato; recomenda o
|
||||
EAD para intercâmbio (nota 13).
|
||||
|
||||
Correspondência NOBRADE → EAD 3:
|
||||
1.1 unitid · 1.2 unittitle · 1.3 unitdate (@normal em ISO 8601-2) ·
|
||||
1.4 @level · 1.5 physdesc + physdescstructured · 2.1 origination ·
|
||||
2.2 bioghist · 2.3 custodhist · 2.4 acqinfo · 3.1 scopecontent ·
|
||||
3.2 appraisal · 3.3 accruals · 3.4 arrangement · 4.1 accessrestrict ·
|
||||
4.2 userestrict · 4.3 langmaterial · 4.4 phystech · 4.5 otherfindaid ·
|
||||
5.1 originalsloc · 5.2 altformavail + dao · 5.3 relatedmaterial ·
|
||||
5.4 bibliography · 6.1 odd[@localtype=conservacao] · 6.2 odd ·
|
||||
7.1 processinfo · 7.2 control/conventiondeclaration · 7.3 control/maintenancehistory ·
|
||||
8.1 controlaccess
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import csv
|
||||
import re
|
||||
import xml.etree.ElementTree as ET
|
||||
from collections import defaultdict
|
||||
from datetime import date
|
||||
|
||||
from .config import ENTIDADE, ENTIDADE_NOME, PAIS, REGRAS, SAIDA
|
||||
|
||||
EAD = 'http://ead3.archivists.org/schema/'
|
||||
CPF = 'urn:isbn:1-931666-33-4'
|
||||
XLINK = 'http://www.w3.org/1999/xlink'
|
||||
NIVEL_EAD = {1: ('collection', None), 2: ('otherlevel', 'seção'), 3: ('series', None),
|
||||
4: ('file', None), 5: ('item', None)}
|
||||
HOJE = date.today().isoformat()
|
||||
|
||||
|
||||
# ── texto plano de cada elemento (usado por EAD, AtoM e mural) ──────────────
|
||||
def texto(u: dict, el: str) -> str | None:
|
||||
v = u.get(el)
|
||||
if v is None:
|
||||
return None
|
||||
if el == '1.2':
|
||||
return v['titulo']
|
||||
if el == '1.3':
|
||||
p = v.get('producao') or {}
|
||||
return p.get('forma')
|
||||
if el == '1.5':
|
||||
return v.get('texto')
|
||||
if el == '2.1':
|
||||
return '; '.join(p['forma'] for p in v)
|
||||
if el == '4.2':
|
||||
return v['texto'] if isinstance(v, dict) else v
|
||||
if el == '4.3':
|
||||
return '; '.join(i['nome'] for i in v) if isinstance(v, list) else v
|
||||
if el == '5.1':
|
||||
return v.get('texto') or v.get('instituicao') if isinstance(v, dict) else v
|
||||
if el == '5.2':
|
||||
if isinstance(v, str):
|
||||
return v
|
||||
partes = [f"{c['tipo']}: {c['chave']} (SHA-256 {c['sha256']})" for c in v.get('nesta_entidade') or []]
|
||||
partes += [f"{c['entidade']}: {c['url']}" for c in v.get('em_outras_entidades') or []]
|
||||
return '; '.join(partes) or None
|
||||
if el in ('5.3', '5.4', '6.2') and isinstance(v, list):
|
||||
return '\n'.join(v)
|
||||
if el == '7.3':
|
||||
return '; '.join(f'{k}: {x}' for k, x in v.items())
|
||||
if el == '8.1' and isinstance(v, dict):
|
||||
return None
|
||||
return v if isinstance(v, str) else None
|
||||
|
||||
|
||||
# ── EAD 3 ────────────────────────────────────────────────────────────────────
|
||||
def _sub(pai, tag, txt=None, **attrs):
|
||||
e = ET.SubElement(pai, f'{{{EAD}}}{tag}', {k.rstrip('_'): str(x) for k, x in attrs.items() if x is not None})
|
||||
if txt is not None:
|
||||
e.text = txt
|
||||
return e
|
||||
|
||||
|
||||
def _paragrafos(pai, tag, txt, head=None, **attrs):
|
||||
if not txt:
|
||||
return
|
||||
e = _sub(pai, tag, **attrs)
|
||||
if head:
|
||||
_sub(e, 'head', head)
|
||||
for par in [p.strip() for p in re.split(r'\n\s*\n|\n(?=[-\d])', txt) if p.strip()]:
|
||||
_sub(e, 'p', re.sub(r'\s+', ' ', par))
|
||||
|
||||
|
||||
def _nome(pai, forma: str, aid: str | None, tipo: str | None, relator=None):
|
||||
tag = {'entidade coletiva': 'corpname', 'família': 'famname'}.get(tipo, 'persname')
|
||||
e = _sub(pai, tag, identifier=aid, relator=relator)
|
||||
_sub(e, 'part', forma)
|
||||
return e
|
||||
|
||||
|
||||
def _componente(pai, u: dict, aut: dict, tag='c'):
|
||||
nivel, outro = NIVEL_EAD[u['nivel']]
|
||||
c = pai if tag == 'archdesc' else _sub(pai, 'c', level=nivel, otherlevel=outro, id=_xmlid(u['id']))
|
||||
did = _sub(c, 'did')
|
||||
_sub(did, 'unitid', u['1.1'], countrycode=PAIS, repositorycode=f'{PAIS}-{ENTIDADE}')
|
||||
_sub(did, 'unittitle', u['1.2']['titulo'])
|
||||
p = u['1.3'].get('producao') or {}
|
||||
cert = {'aproximada': 'approximate', 'incerta': 'uncertain', 'atribuida': 'inferred',
|
||||
'atribuída pelo arquivista': 'inferred'}.get(p.get('qualidade'))
|
||||
_sub(did, 'unitdate', p.get('forma') or '[s.d.]', normal=p.get('iso'),
|
||||
datechar='publication' if p.get('tipo') == 'publicação' else 'creation', certainty=cert,
|
||||
unitdatetype='inclusive' if u['nivel'] < 4 else None)
|
||||
orig = _sub(did, 'origination')
|
||||
for pr in u['2.1']:
|
||||
_nome(orig, pr['forma'], pr.get('id'), (aut.get(pr.get('id')) or {}).get('tipo'),
|
||||
relator=pr.get('funcao'))
|
||||
dim = u['1.5']
|
||||
_sub(did, 'physdesc', dim['texto'])
|
||||
for esp, n in (dim.get('por_especie') or {}).items():
|
||||
ps = _sub(did, 'physdescstructured', physdescstructuredtype='materialtype', coverage='part')
|
||||
_sub(ps, 'quantity', str(n))
|
||||
_sub(ps, 'unittype', esp)
|
||||
if isinstance(u.get('4.3'), list):
|
||||
lm = _sub(did, 'langmaterial')
|
||||
for i in u['4.3']:
|
||||
_sub(lm, 'language', i['nome'], langcode=i['codigo'])
|
||||
if u['nivel'] == 5:
|
||||
for cp in (u.get('5.2') or {}).get('nesta_entidade') or []:
|
||||
_sub(did, 'dao', daotype='derived', href=cp['chave'], linktitle=cp['tipo'],
|
||||
identifier=f"sha256:{cp['sha256']}")
|
||||
|
||||
_paragrafos(c, 'bioghist', u.get('2.2'))
|
||||
_paragrafos(c, 'custodhist', u.get('2.3'))
|
||||
_paragrafos(c, 'acqinfo', u.get('2.4'))
|
||||
_paragrafos(c, 'scopecontent', u.get('3.1'))
|
||||
_paragrafos(c, 'appraisal', u.get('3.2'))
|
||||
_paragrafos(c, 'accruals', u.get('3.3'))
|
||||
_paragrafos(c, 'arrangement', u.get('3.4'))
|
||||
_paragrafos(c, 'accessrestrict', u.get('4.1'))
|
||||
_paragrafos(c, 'userestrict', texto(u, '4.2'))
|
||||
_paragrafos(c, 'phystech', u.get('4.4'))
|
||||
_paragrafos(c, 'otherfindaid', u.get('4.5'))
|
||||
_paragrafos(c, 'originalsloc', texto(u, '5.1'))
|
||||
if u['nivel'] != 5:
|
||||
_paragrafos(c, 'altformavail', texto(u, '5.2'))
|
||||
if isinstance(u.get('5.3'), list):
|
||||
rm = _sub(c, 'relatedmaterial')
|
||||
for x in u['5.3']:
|
||||
_sub(rm, 'p', x)
|
||||
if isinstance(u.get('5.4'), list):
|
||||
bib = _sub(c, 'bibliography')
|
||||
for x in u['5.4']:
|
||||
_sub(bib, 'bibref', x)
|
||||
elif u.get('5.4'):
|
||||
_paragrafos(c, 'bibliography', u['5.4'])
|
||||
_paragrafos(c, 'odd', u.get('6.1'), head='Notas sobre conservação', localtype='conservacao')
|
||||
_paragrafos(c, 'odd', texto(u, '6.2'), head='Notas gerais')
|
||||
_paragrafos(c, 'processinfo', u.get('7.1'))
|
||||
|
||||
ca = u.get('8.1')
|
||||
if isinstance(ca, dict):
|
||||
ctl = _sub(c, 'controlaccess')
|
||||
for s in ca.get('assuntos') or []:
|
||||
_sub(_sub(ctl, 'subject', lang='por'), 'part', s)
|
||||
for s in ca.get('assuntos_eng') or []:
|
||||
_sub(_sub(ctl, 'subject', lang='eng'), 'part', s)
|
||||
if u['1.3'].get('topica'):
|
||||
_sub(_sub(ctl, 'geogname', localtype='data tópica'), 'part', u['1.3']['topica'])
|
||||
for s in ca.get('lugares') or []:
|
||||
_sub(_sub(ctl, 'geogname'), 'part', s)
|
||||
for pr in u['2.1']:
|
||||
if pr.get('id'):
|
||||
_nome(ctl, pr['forma'], pr['id'], (aut.get(pr['id']) or {}).get('tipo'))
|
||||
if len(ctl) == 0:
|
||||
c.remove(ctl)
|
||||
elif ca:
|
||||
_paragrafos(c, 'processinfo', ca, head='Pontos de acesso e indexação')
|
||||
return c
|
||||
|
||||
|
||||
def _xmlid(cod: str) -> str:
|
||||
return 'u-' + re.sub(r'[^\w.-]', '_', cod)
|
||||
|
||||
|
||||
def ead(unidades: list[dict], aut: dict):
|
||||
filhos = defaultdict(list)
|
||||
por_id = {}
|
||||
for u in unidades:
|
||||
por_id[u['id']] = u
|
||||
if u.get('pai'):
|
||||
filhos[u['pai']].append(u['id'])
|
||||
col = next(u for u in unidades if u['nivel'] == 1)
|
||||
|
||||
ET.register_namespace('', EAD)
|
||||
raiz = ET.Element(f'{{{EAD}}}ead')
|
||||
ctrl = _sub(raiz, 'control', langencoding='iso639-2b', scriptencoding='iso15924',
|
||||
dateencoding='iso8601', countryencoding='iso3166-1', repositoryencoding='iso15511')
|
||||
_sub(ctrl, 'recordid', col['id'])
|
||||
fd = _sub(ctrl, 'filedesc')
|
||||
ts = _sub(fd, 'titlestmt')
|
||||
_sub(ts, 'titleproper', col['1.2']['titulo'])
|
||||
_sub(ctrl, 'maintenancestatus', value='new')
|
||||
ag = _sub(ctrl, 'maintenanceagency', countrycode=PAIS)
|
||||
_sub(ag, 'agencycode', f'{PAIS}-{ENTIDADE}')
|
||||
_sub(ag, 'agencyname', ENTIDADE_NOME)
|
||||
ld = _sub(ctrl, 'languagedeclaration')
|
||||
_sub(ld, 'language', 'português', langcode='por')
|
||||
_sub(ld, 'script', 'latino', scriptcode='Latn')
|
||||
cd = _sub(ctrl, 'conventiondeclaration')
|
||||
_sub(cd, 'citation', REGRAS)
|
||||
mh = _sub(ctrl, 'maintenancehistory')
|
||||
ev = _sub(mh, 'maintenanceevent')
|
||||
_sub(ev, 'eventtype', value='created')
|
||||
_sub(ev, 'eventdatetime', HOJE, standarddatetime=HOJE)
|
||||
_sub(ev, 'agenttype', value='machine')
|
||||
_sub(ev, 'agent', 'pipeline/nobrade')
|
||||
|
||||
arch = _sub(raiz, 'archdesc', level='collection')
|
||||
_componente(arch, col, aut, tag='archdesc')
|
||||
rep = arch.find(f'{{{EAD}}}did')
|
||||
r = _sub(rep, 'repository')
|
||||
_sub(_sub(r, 'corpname', identifier=f'{PAIS}-{ENTIDADE}'), 'part', ENTIDADE_NOME)
|
||||
dsc = _sub(arch, 'dsc')
|
||||
|
||||
def desce(pai_xml, cod):
|
||||
for fid in sorted(filhos[cod], key=lambda x: (len(x), x)):
|
||||
c = _componente(pai_xml, por_id[fid], aut)
|
||||
desce(c, fid)
|
||||
desce(dsc, col['id'])
|
||||
|
||||
(SAIDA / 'ead').mkdir(exist_ok=True)
|
||||
ET.indent(raiz, space=' ')
|
||||
ET.ElementTree(raiz).write(SAIDA / 'ead' / 'brasiliana.xml', encoding='utf-8', xml_declaration=True)
|
||||
|
||||
|
||||
# ── EAC-CPF ──────────────────────────────────────────────────────────────────
|
||||
def _c(pai, tag, txt=None, **attrs):
|
||||
e = ET.SubElement(pai, f'{{{CPF}}}{tag}', {k.rstrip('_'): str(x) for k, x in attrs.items() if x is not None})
|
||||
if txt is not None:
|
||||
e.text = txt
|
||||
return e
|
||||
|
||||
|
||||
def eac(autoridades: list[dict], usos: dict[str, list[str]]):
|
||||
ET.register_namespace('', CPF)
|
||||
ET.register_namespace('xlink', XLINK)
|
||||
pasta = SAIDA / 'eac'
|
||||
pasta.mkdir(exist_ok=True)
|
||||
for a in autoridades:
|
||||
raiz = ET.Element(f'{{{CPF}}}eac-cpf')
|
||||
ctrl = _c(raiz, 'control')
|
||||
_c(ctrl, 'recordId', f"{PAIS}-{ENTIDADE}-{a['id']}")
|
||||
_c(ctrl, 'maintenanceStatus', 'new')
|
||||
ag = _c(ctrl, 'maintenanceAgency')
|
||||
_c(ag, 'agencyCode', f'{PAIS}-{ENTIDADE}')
|
||||
_c(ag, 'agencyName', ENTIDADE_NOME)
|
||||
ld = _c(ctrl, 'languageDeclaration')
|
||||
_c(ld, 'language', 'português', languageCode='por')
|
||||
_c(ld, 'script', 'latino', scriptCode='Latn')
|
||||
cd = _c(ctrl, 'conventionDeclaration')
|
||||
_c(cd, 'citation', 'ISAAR(CPF) 2ª ed.; NOBRADE (CONARQ, 2006), proc. 8.1.3–8.1.4')
|
||||
mh = _c(ctrl, 'maintenanceHistory')
|
||||
ev = _c(mh, 'maintenanceEvent')
|
||||
_c(ev, 'eventType', 'created')
|
||||
_c(ev, 'eventDateTime', HOJE, standardDateTime=HOJE)
|
||||
_c(ev, 'agentType', 'machine')
|
||||
_c(ev, 'agent', 'pipeline/nobrade')
|
||||
if a['fontes']:
|
||||
src = _c(ctrl, 'sources')
|
||||
for f in a['fontes']:
|
||||
s = _c(src, 'source')
|
||||
_c(s, 'sourceEntry', f)
|
||||
|
||||
cpf = _c(raiz, 'cpfDescription')
|
||||
ident = _c(cpf, 'identity')
|
||||
_c(ident, 'entityType', {'entidade coletiva': 'corporateBody', 'família': 'family'}.get(a['tipo'], 'person'))
|
||||
ne = _c(ident, 'nameEntry', localType='autorizada')
|
||||
_c(ne, 'part', a['nome'])
|
||||
if a['datas_existencia']:
|
||||
_c(ne, 'useDates').append(_data_eac(a['datas_existencia']))
|
||||
_c(ne, 'authorizedForm', 'NOBRADE')
|
||||
for v in a['variantes']:
|
||||
nv = _c(ident, 'nameEntry', localType='variante')
|
||||
_c(nv, 'part', v)
|
||||
_c(nv, 'alternativeForm', 'NOBRADE')
|
||||
desc = _c(cpf, 'description')
|
||||
if a['datas_existencia']:
|
||||
ex = _c(desc, 'existDates')
|
||||
ex.append(_data_eac(a['datas_existencia']))
|
||||
bh = _c(desc, 'biogHist')
|
||||
_c(bh, 'p', f"Autor de {a['ocorrencias']} dossiês da Coleção Brasiliana Fotográfica (cópia digital)."
|
||||
+ (f" Tipo presumido: {a['tipo']}." if a['tipo'] == 'pessoa ou entidade' else ''))
|
||||
if usos.get(a['id']):
|
||||
rel = _c(cpf, 'relations')
|
||||
for cod in usos[a['id']][:500]:
|
||||
r = _c(rel, 'resourceRelation', resourceRelationType='creatorOf')
|
||||
_c(r, 'relationEntry', cod)
|
||||
ET.indent(raiz, space=' ')
|
||||
ET.ElementTree(raiz).write(pasta / f"{a['id']}.xml", encoding='utf-8', xml_declaration=True)
|
||||
|
||||
|
||||
def _data_eac(datas: str):
|
||||
anos = re.findall(r'\d{4}', datas)
|
||||
if len(anos) >= 2:
|
||||
e = ET.Element(f'{{{CPF}}}dateRange')
|
||||
_c(e, 'fromDate', datas.split('-')[0].strip(), standardDate=anos[0])
|
||||
_c(e, 'toDate', datas.split('-')[-1].strip(), standardDate=anos[1])
|
||||
return e
|
||||
e = ET.Element(f'{{{CPF}}}date', standardDate=anos[0] if anos else None)
|
||||
e.text = datas
|
||||
return e
|
||||
|
||||
|
||||
# ── AtoM (modelo CSV ISAD da versão 2.x) ────────────────────────────────────
|
||||
ATOM = ['legacyId', 'parentId', 'identifier', 'title', 'levelOfDescription', 'extentAndMedium',
|
||||
'repository', 'archivalHistory', 'acquisition', 'scopeAndContent', 'appraisal', 'accruals',
|
||||
'arrangement', 'accessConditions', 'reproductionConditions', 'language', 'physicalCharacteristics',
|
||||
'findingAids', 'locationOfOriginals', 'locationOfCopies', 'relatedUnitsOfDescription',
|
||||
'publicationNote', 'generalNote', 'subjectAccessPoints', 'placeAccessPoints', 'nameAccessPoints',
|
||||
'descriptionIdentifier', 'rules', 'archivistNote', 'eventDates', 'eventTypes', 'eventStartDates',
|
||||
'eventEndDates', 'eventActors', 'culture']
|
||||
NIVEL_ATOM = {1: 'Coleção', 2: 'Seção', 3: 'Série', 4: 'Dossiê', 5: 'Item documental'}
|
||||
|
||||
|
||||
def atom(unidades: list[dict]):
|
||||
with (SAIDA / 'atom.csv').open('w', encoding='utf-8', newline='') as f:
|
||||
w = csv.DictWriter(f, fieldnames=ATOM)
|
||||
w.writeheader()
|
||||
for u in unidades:
|
||||
if u['nivel'] == 0:
|
||||
continue
|
||||
p = u['1.3'].get('producao') or {}
|
||||
ca = u.get('8.1') if isinstance(u.get('8.1'), dict) else {}
|
||||
w.writerow({
|
||||
'legacyId': u['id'], 'parentId': u['pai'] if u['nivel'] > 1 else '',
|
||||
'identifier': u['1.1'].split('.')[-1] if u['nivel'] > 1 else u['1.1'],
|
||||
'title': u['1.2']['titulo'], 'levelOfDescription': NIVEL_ATOM[u['nivel']],
|
||||
'extentAndMedium': texto(u, '1.5'), 'repository': ENTIDADE_NOME,
|
||||
'archivalHistory': u.get('2.3'), 'acquisition': u.get('2.4'),
|
||||
'scopeAndContent': u.get('3.1'), 'appraisal': u.get('3.2'), 'accruals': u.get('3.3'),
|
||||
'arrangement': u.get('3.4'), 'accessConditions': u.get('4.1'),
|
||||
'reproductionConditions': texto(u, '4.2'),
|
||||
'language': '|'.join(i['codigo'] for i in u['4.3']) if isinstance(u.get('4.3'), list) else '',
|
||||
'physicalCharacteristics': u.get('4.4'), 'findingAids': u.get('4.5'),
|
||||
'locationOfOriginals': texto(u, '5.1'), 'locationOfCopies': texto(u, '5.2'),
|
||||
'relatedUnitsOfDescription': texto(u, '5.3'), 'publicationNote': texto(u, '5.4'),
|
||||
'generalNote': '\n'.join(x for x in [u.get('6.1'), texto(u, '6.2')] if x),
|
||||
'subjectAccessPoints': '|'.join(ca.get('assuntos') or []),
|
||||
'placeAccessPoints': '|'.join(ca.get('lugares') or []),
|
||||
'nameAccessPoints': '|'.join(ca.get('nomes') or []),
|
||||
'descriptionIdentifier': u['id'], 'rules': u.get('7.2') or '',
|
||||
'archivistNote': u.get('7.1'),
|
||||
'eventDates': p.get('forma'), 'eventTypes': 'Produção',
|
||||
'eventStartDates': f"{p['inicio']}-01-01" if p.get('inicio') else '',
|
||||
'eventEndDates': f"{p['fim']}-12-31" if p.get('fim') else '',
|
||||
'eventActors': '|'.join(x['forma'] for x in u['2.1'] if x.get('id')),
|
||||
'culture': 'pt_BR',
|
||||
})
|
||||
Reference in New Issue
Block a user