rodado
dfb6362e30
- PLANO §1/§7 com o estado real (ledger e bucket em 2026-09-30) e §9 com o visor do mural (7.1–7.2c), 7.3 e 7.4 pendentes - MANIFEST final com contagens fechadas e seção sobre a descrição NOBRADE - textos das 15 seções e do nível 1 conferidos na web, com fontes fora da descrição - rascunho do pedido de CODEARQ ao CONARQ - descricao/ enviada a s3://brasiliana/descricao/ com manifest-sha256.txt
501 lines
26 KiB
Python
501 lines
26 KiB
Python
"""Monta a descrição multinível (níveis 0–5) a partir do catálogo, do ledger do
|
||
bucket e dos posts do blog. Escreve descricao/unidades.jsonl,
|
||
descricao/autoridades.jsonl e as tabelas de revisão.
|
||
|
||
Cada unidade é um objeto com `id` (= código de referência), `pai`, `nivel` e os
|
||
elementos da NOBRADE pelo número: "1.1", "1.2", … "8.1".
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import ast
|
||
import json
|
||
import re
|
||
import sqlite3
|
||
from collections import Counter, defaultdict
|
||
from datetime import date, datetime
|
||
|
||
import polars as pl
|
||
|
||
from . import campos, datas, series
|
||
from .autoridades import Autoridades
|
||
from .config import (ACESSO, BUCKET, CATALOGO, COLECAO, COLECAO_TITULO, ENTIDADE_NOME, HANDLE_URL,
|
||
INSTITUICOES, LEDGER, ORIGEM_NOME, ORIGEM_SITE, POSTS, REGRAS, RESPONSAVEL, REVISAO,
|
||
SAIDA, TEXTOS, codigo)
|
||
|
||
NIVEIS = {0: 'acervo da entidade custodiadora', 1: 'coleção', 2: 'seção', 3: 'série',
|
||
4: 'dossiê', 5: 'item documental'}
|
||
MESES_ABNT = ['jan.', 'fev.', 'mar.', 'abr.', 'maio', 'jun.', 'jul.', 'ago.', 'set.', 'out.', 'nov.', 'dez.']
|
||
HOJE = date.today().isoformat()
|
||
|
||
|
||
# ── leitura das fontes ──────────────────────────────────────────────────────
|
||
def _valores(item: dict) -> dict[str, list[str]]:
|
||
vs = defaultdict(list)
|
||
for row in item.get('full_table') or []:
|
||
v = (row.get('value') or '').strip()
|
||
if v and v not in vs[row['field']]:
|
||
vs[row['field']].append(v)
|
||
return vs
|
||
|
||
|
||
def _ledger() -> dict[str, list[dict]]:
|
||
arquivos = defaultdict(list)
|
||
if not LEDGER.exists():
|
||
return arquivos
|
||
con = sqlite3.connect(LEDGER)
|
||
for h, nome, b, sha, quando, tipo in con.execute(
|
||
'select handle, nome, bytes, sha256, quando, tipo from enviados order by handle, tipo, nome'):
|
||
arquivos[h].append({'nome': nome, 'bytes': b, 'sha256': sha, 'quando': quando, 'tipo': tipo})
|
||
return arquivos
|
||
|
||
|
||
def _abnt(post: dict) -> str:
|
||
autor = (post.get('autor') or '').strip()
|
||
partes = autor.split()
|
||
autor_abnt = f'{partes[-1].upper()}, {" ".join(partes[:-1])}' if len(partes) > 1 else autor.upper()
|
||
d = datetime.fromisoformat(post['data_iso'][:10])
|
||
acesso = (post.get('fetched_at') or HOJE)[:10]
|
||
a = datetime.fromisoformat(acesso)
|
||
return (f"{autor_abnt}. {post['titulo']}. {ORIGEM_NOME}, Rio de Janeiro, "
|
||
f"{d.day} {MESES_ABNT[d.month - 1]} {d.year}. Disponível em: {post['url']}. "
|
||
f"Acesso em: {a.day} {MESES_ABNT[a.month - 1]} {a.year}.")
|
||
|
||
|
||
def _posts() -> tuple[dict[str, list[dict]], list[dict]]:
|
||
por_handle, todos = defaultdict(list), []
|
||
if not POSTS.exists():
|
||
return por_handle, todos
|
||
for linha in POSTS.open(encoding='utf-8'):
|
||
p = json.loads(linha)
|
||
if p.get('erro') or not p.get('titulo'):
|
||
continue
|
||
hs = p.get('handles_referenciados') or []
|
||
if isinstance(hs, str):
|
||
hs = ast.literal_eval(hs)
|
||
cats = p.get('categorias') or []
|
||
if isinstance(cats, str):
|
||
cats = ast.literal_eval(cats)
|
||
ref = {'titulo': p['titulo'], 'url': p['url'], 'autor': p.get('autor'), 'data': p['data_iso'][:10],
|
||
'categorias': cats, 'abnt': _abnt(p)}
|
||
todos.append(ref)
|
||
for h in hs:
|
||
por_handle[h].append(ref)
|
||
return por_handle, todos
|
||
|
||
|
||
def _textos(caminho) -> dict[str, str]:
|
||
"""Um .md com seções "## 2.3 História arquivística" → {'2.3': texto}."""
|
||
if not caminho.exists():
|
||
return {}
|
||
out, atual, buf = {}, None, []
|
||
for linha in caminho.read_text(encoding='utf-8').splitlines():
|
||
if m := re.match(r'^##\s+(\d\.\d)\b', linha):
|
||
if atual:
|
||
out[atual] = '\n'.join(buf).strip()
|
||
atual, buf = m[1], []
|
||
elif linha.startswith('#'): # outro título (### Fontes…) fecha a seção e fica fora da descrição
|
||
if atual:
|
||
out[atual] = '\n'.join(buf).strip()
|
||
atual, buf = None, []
|
||
elif atual:
|
||
buf.append(linha)
|
||
if atual:
|
||
out[atual] = '\n'.join(buf).strip()
|
||
return {k: v for k, v in out.items() if v}
|
||
|
||
|
||
def _mb(b: int) -> str:
|
||
return f'{b / 1e6:.1f} MB'.replace('.', ',') if b < 1e9 else f'{b / 1e9:.1f} GB'.replace('.', ',')
|
||
|
||
|
||
def _formato(nome: str) -> str:
|
||
ext = nome.rsplit('.', 1)[-1].lower() if '.' in nome else ''
|
||
return {'jpg': 'JPEG', 'jpeg': 'JPEG', 'png': 'PNG', 'tif': 'TIFF', 'tiff': 'TIFF', 'webp': 'WebP',
|
||
'pdf': 'PDF', 'gif': 'GIF'}.get(ext, ext.upper() or 'desconhecido')
|
||
|
||
|
||
def _limites(ds: list[dict]) -> dict:
|
||
ini = [d['inicio'] for d in ds if d and d.get('inicio')]
|
||
fim = [d['fim'] for d in ds if d and d.get('fim')]
|
||
if not ini and not fim:
|
||
return {'forma': '[s.d.]', 'iso': None, 'inicio': None, 'fim': None, 'qualidade': None,
|
||
'tipo': 'produção'}
|
||
a, b = min(ini or fim), max(fim or ini)
|
||
return {'forma': f'{a}-{b}' if a != b else str(a), 'iso': f'{a}/{b}' if a != b else str(a),
|
||
'inicio': a, 'fim': b, 'qualidade': 'datas-limite', 'tipo': 'produção'}
|
||
|
||
|
||
def _robustos(ds: list[dict]) -> dict:
|
||
"""Datas-limite para atribuir a dossiês sem data: o miolo (percentis 10–90)
|
||
das datas conhecidas, para que um punhado de extremos não vire [1800-1999?]."""
|
||
anos = sorted(d['inicio'] for d in ds if d and d.get('inicio'))
|
||
if not anos:
|
||
return {'inicio': None, 'fim': None}
|
||
return {'inicio': anos[int(len(anos) * .1)], 'fim': anos[min(len(anos) - 1, int(len(anos) * .9))]}
|
||
|
||
|
||
def _dimensao(especies: Counter, n_itens: int, bytes_: int) -> dict:
|
||
partes = [f'{n} {e}' for e, n in especies.most_common()]
|
||
return {'por_especie': dict(especies), 'arquivos_digitais': n_itens, 'bytes': bytes_,
|
||
'texto': f"iconográfico: {', '.join(partes)}; eletrônico: {n_itens} arquivos digitais ({_mb(bytes_)})"}
|
||
|
||
|
||
# ── montagem ────────────────────────────────────────────────────────────────
|
||
def monta():
|
||
SAIDA.mkdir(exist_ok=True)
|
||
REVISAO.mkdir(parents=True, exist_ok=True)
|
||
ledger = _ledger()
|
||
posts_de, posts = _posts()
|
||
|
||
itens = []
|
||
for linha in CATALOGO.open(encoding='utf-8'):
|
||
if linha.strip():
|
||
it = json.loads(linha)
|
||
it['_v'] = _valores(it)
|
||
itens.append(it)
|
||
|
||
# séries (nível 3)
|
||
fontes = [(it['handle'], it['community'], (it['_v']['dc.source'] or [None])[0]) for it in itens]
|
||
serie_de, series_ = series.monta(fontes)
|
||
|
||
# autoridades
|
||
aut = Autoridades()
|
||
for it in itens:
|
||
for f in ('dc.contributor', 'dc.contributor.author', 'dc.contributor.other'):
|
||
for v in it['_v'][f]:
|
||
aut.registra(v)
|
||
aut.resolve()
|
||
for reg in aut.registros.values(): # biografias no blog (3.4) → fontes da autoridade
|
||
if reg['tipo'] == 'pessoa' and ',' in reg['nome']:
|
||
sob, nome = [x.strip() for x in reg['nome'].split(',', 1)]
|
||
primeiro = nome.split()[0] if nome else ''
|
||
for p in posts:
|
||
if 'Biografia' in p['categorias'] and sob in p['titulo'] and primeiro in p['titulo']:
|
||
reg['fontes'].append(p['abnt'])
|
||
|
||
unidades: dict[str, dict] = {}
|
||
filhos = defaultdict(list)
|
||
rev_datas, rev_tipos, rev_direitos, rev_lugares = [], [], [], []
|
||
|
||
def add(u: dict):
|
||
unidades[u['id']] = u
|
||
if u['pai']:
|
||
filhos[u['pai']].append(u['id'])
|
||
|
||
# níveis 4 e 5
|
||
for it in itens:
|
||
v, h = it['_v'], it['handle']
|
||
inst = it['community']
|
||
sigla_i, nome_i, cidade_i = INSTITUICOES[inst]
|
||
sg, notacao = serie_de[h]
|
||
num = h.rsplit('/', 1)[-1]
|
||
cod = codigo(COLECAO, sigla_i, sg, num)
|
||
pasta = h.replace('/', '_')
|
||
|
||
titulo = (v['dc.title'] or [it.get('title') or ''])[0].strip()
|
||
titulo_atribuido = False
|
||
if not titulo: # proc. 1.2: título atribuído a partir da própria unidade, entre colchetes
|
||
base = next((x for x in v['dc.description'] + v['dc.subject'] + v['dc.coverage']
|
||
if not campos.corrompido(x)), 'Sem título')
|
||
titulo, titulo_atribuido = f"[{base.split('. ')[0][:90]}]", True
|
||
d = datas.interpreta((v['dc.date'] or [''])[0])
|
||
data = d.registro()
|
||
if len(v['dc.date']) > 1:
|
||
data['notas'] = (data['notas'] or []) + [f'outra data na origem: {x}' for x in v['dc.date'][1:]]
|
||
|
||
produtores = []
|
||
for f in ('dc.contributor', 'dc.contributor.author', 'dc.contributor.other'):
|
||
for nome in v[f]:
|
||
aid = aut.id(nome)
|
||
if aid and aid not in [p['id'] for p in produtores]:
|
||
produtores.append({'id': aid, 'forma': aut[aid]['forma_autorizada'],
|
||
'funcao': 'autor' if f != 'dc.contributor.other' else 'outro'})
|
||
if not produtores:
|
||
produtores = [{'id': None, 'forma': 'Autoria não identificada', 'funcao': 'autor'}]
|
||
|
||
descricoes = [x for x in v['dc.description'] if not campos.corrompido(x)]
|
||
fisico = [x for x in descricoes if campos.e_fisico(x)]
|
||
conteudo = [x for x in descricoes if x not in fisico]
|
||
esp = campos.especie((v['dc.type'] or [None])[0])
|
||
cor = campos.cor(fisico)
|
||
|
||
arquivos = ledger.get(h, [])
|
||
preserv = sorted((a for a in arquivos if a['tipo'] == 'preservacao'),
|
||
key=lambda a: (a['nome'] != 'imagem.jpg', a['nome']))
|
||
imagens = preserv or [a for a in arquivos if a['tipo'] == 'thumbnail']
|
||
total = sum(a['bytes'] or 0 for a in imagens)
|
||
|
||
direitos = campos.direitos(v['dc.rights'] + [str(it.get('rights') or '')], nome_i)
|
||
lugar = campos.lugar(v['dc.coverage'])
|
||
assuntos = [s for s in v['dc.subject'] if not campos.corrompido(s)]
|
||
codigo_origem = next((re.sub(r'\.\w{3,4}$', '', x) for x in v['dc.identifier'] + v['dc.description']
|
||
if re.match(r'BR ?[A-Z]{2}[A-Z]{2,}\s', x)), None)
|
||
sophia = next((x for x in v['dc.relation'] if 'sophia' in x), None)
|
||
objdigital = [x for x in v['dc.identifier'] if 'objdigital' in x]
|
||
|
||
notas = ['Título atribuído pelo arquivista: a origem não informa título.'] if titulo_atribuido else []
|
||
if d.original and data['forma'] != d.original:
|
||
notas.append(f'Data como consta na origem: {d.original}')
|
||
notas += data['notas'] or []
|
||
if esp['presumida']:
|
||
notas.append('Espécie documental não informada na origem; presumida fotografia'
|
||
+ (f" (valor na origem: {esp['original']})" if esp.get('original') else '') + '.')
|
||
if direitos['originais']:
|
||
notas.append('Direitos como constam na origem: ' + ' | '.join(direitos['originais']))
|
||
lixo = [f"{f}: {x[:60]}" for f, xs in v.items() for x in xs if campos.corrompido(x)
|
||
and f in ('dc.type', 'dc.rights', 'dc.source', 'dc.subject', 'dc.date', 'dc.coverage')]
|
||
if lixo:
|
||
notas.append('Valores corrompidos na origem, desconsiderados: ' + '; '.join(lixo))
|
||
|
||
conservacao = None
|
||
if not preserv:
|
||
conservacao = ('Arquivo de preservação indisponível na origem (o repositório responde HTTP 500 '
|
||
'ao bitstream; ver errors.md). A cópia guarda apenas a miniatura gerada pelo DSpace.')
|
||
|
||
dossie = {
|
||
'id': cod, 'pai': codigo(COLECAO, sigla_i, sg), 'nivel': 4, 'handle': h,
|
||
'1.1': cod,
|
||
'1.2': {'titulo': titulo, 'atribuido': titulo.startswith('[')},
|
||
'1.3': {'producao': data, 'topica': lugar},
|
||
'1.4': {'nivel': 4, 'nome': NIVEIS[4]},
|
||
'1.5': {'genero': esp['genero'], 'especie': esp['especie'], 'suporte': esp['suporte'], 'cor': cor,
|
||
'fisico': fisico or None, 'arquivos_digitais': len(imagens), 'bytes': total,
|
||
'texto': (f"1 {esp['especie']}" + (f' {cor}' if cor else '')
|
||
+ (f", {esp['suporte']}" if esp['suporte'] else '')
|
||
+ (f" ({'; '.join(fisico)})" if fisico else '')
|
||
+ f"; {len(imagens)} arquivo{'s' if len(imagens) != 1 else ''} digita"
|
||
+ f"{'is' if len(imagens) != 1 else 'l'} ({_mb(total)})")},
|
||
'2.1': produtores,
|
||
'2.3': (f"Registrado no repositório DSpace da {ORIGEM_NOME}"
|
||
+ (f" em {v['dc.date.accessioned'][0][:10]}" if v['dc.date.accessioned'] else '')
|
||
+ f". Cópia digital capturada em {it['fetched_at'][:10]} e armazenada em "
|
||
f"{BUCKET}/itens/{pasta}/ com SHA-256 de cada arquivo."),
|
||
'2.4': (f"{ORIGEM_NOME} ({ORIGEM_SITE}), repositório mantido pela Biblioteca Nacional; "
|
||
f"coleta combinada com a Biblioteca Nacional, em {it['fetched_at'][:10]}. Handle {h}."),
|
||
'3.1': ' '.join(conteudo) or None,
|
||
'4.2': direitos,
|
||
'4.3': campos.idiomas(v['dc.language']) or None,
|
||
'4.4': (f"Arquivo{'s' if len(imagens) > 1 else ''} digita{'is' if len(imagens) > 1 else 'l'} "
|
||
f"{', '.join(sorted({_formato(a['nome']) for a in imagens})) or '—'}; "
|
||
'abre em qualquer visualizador de imagem.'
|
||
+ (f" Original: {'; '.join(fisico)}." if fisico else '')),
|
||
'5.1': {'instituicao': nome_i, 'local': cidade_i, 'url': HANDLE_URL.format(h),
|
||
'codigo_na_origem': codigo_origem, 'notacao_na_origem': notacao, 'sophia': sophia,
|
||
'texto': f"Original sob a guarda de {nome_i}, {cidade_i}"
|
||
+ (f", código {codigo_origem}" if codigo_origem else '')
|
||
+ (f", notação {notacao}" if notacao else '') + f". {HANDLE_URL.format(h)}"},
|
||
'5.2': {'nesta_entidade': [{'tipo': a['tipo'], 'chave': (f"{BUCKET}/derivados/thumb256/{a['nome']}"
|
||
if a['tipo'] == 'derivado-thumb256'
|
||
else f"{BUCKET}/itens/{pasta}/{a['tipo']}/{a['nome']}"),
|
||
'bytes': a['bytes'], 'sha256': a['sha256']} for a in arquivos],
|
||
'em_outras_entidades': [{'entidade': 'Biblioteca Nacional (Brasil) — BNDigital', 'url': u}
|
||
for u in objdigital] or None},
|
||
'5.3': (v['dc.relation.ispartof'] + v['dcterms.relation']) or None,
|
||
'5.4': [p['abnt'] for p in posts_de.get(h, [])] or None,
|
||
'6.1': conservacao,
|
||
'6.2': notas or None,
|
||
'7.1': (f'Descrição gerada pelo pipeline/nobrade a partir dos metadados Dublin Core colhidos '
|
||
f'da {ORIGEM_NOME} ({it["source_url"]}); responsável: {RESPONSAVEL}.'),
|
||
'7.3': {'coleta': it['fetched_at'][:10], 'descricao': HOJE},
|
||
'8.1': {'assuntos': [s for s in assuntos if campos.idioma_termo(s) == 'por'] or None,
|
||
'assuntos_eng': [s for s in assuntos if campos.idioma_termo(s) == 'eng'] or None,
|
||
'lugares': [c for c in v['dc.coverage'] if not campos.corrompido(c)] or None,
|
||
'nomes': [p['forma'] for p in produtores if p['id']] or None,
|
||
'periodo': data['iso']},
|
||
'_sd': not d.ok, '_especie': esp['especie'], '_serie': (inst, sg),
|
||
}
|
||
add(dossie)
|
||
rev_datas.append({'handle': h, 'original': d.original, 'forma': data['forma'], 'iso': data['iso'],
|
||
'inicio': data['inicio'], 'fim': data['fim'], 'qualidade': data['qualidade'],
|
||
'motivo': data['motivo']})
|
||
rev_tipos.append({'instituicao': inst, 'dc_type': (v['dc.type'] or [''])[0], 'genero': esp['genero'],
|
||
'especie': esp['especie'], 'suporte': esp['suporte'] or '', 'presumida': esp['presumida']})
|
||
rev_direitos.append({'instituicao': inst, 'categoria': direitos['categoria'], 'detentor': direitos['detentor'],
|
||
'originais': ' | '.join(direitos['originais'] or [])})
|
||
rev_lugares.append({'coverage': ' | '.join(v['dc.coverage']), 'data_topica': lugar or ''})
|
||
|
||
for i, a in enumerate(imagens, 1):
|
||
ci = f'{cod}.{i:02d}'
|
||
add({
|
||
'id': ci, 'pai': cod, 'nivel': 5, 'handle': h,
|
||
'1.1': ci,
|
||
'1.2': {'titulo': f"{titulo} [imagem {i} de {len(imagens)}]", 'atribuido': True},
|
||
'1.3': {'producao': data},
|
||
'1.4': {'nivel': 5, 'nome': NIVEIS[5]},
|
||
'1.5': {'genero': 'eletrônico', 'especie': 'arquivo de imagem', 'bytes': a['bytes'],
|
||
'texto': f"1 arquivo digital {_formato(a['nome'])} ({_mb(a['bytes'] or 0)})"},
|
||
'2.1': produtores,
|
||
'4.4': f"{_formato(a['nome'])}; SHA-256 {a['sha256']}"
|
||
+ ('; miniatura — preservação indisponível na origem' if a['tipo'] == 'thumbnail' else ''),
|
||
'5.2': {'nesta_entidade': [{'tipo': a['tipo'], 'chave': f"{BUCKET}/itens/{pasta}/{a['tipo']}/{a['nome']}",
|
||
'bytes': a['bytes'], 'sha256': a['sha256']}]},
|
||
'7.3': {'descricao': HOJE},
|
||
})
|
||
|
||
# níveis 3, 2, 1, 0 — de baixo para cima, agregando os filhos
|
||
textos1 = _textos(TEXTOS / 'nivel1.md')
|
||
textos0 = _textos(TEXTOS / 'nivel0.md')
|
||
por_serie = defaultdict(list)
|
||
for u in unidades.values():
|
||
if u['nivel'] == 4:
|
||
por_serie[u['_serie']].append(u)
|
||
|
||
# datas atribuídas aos s.d. (proc. 1.3.3.2): período do fotógrafo, senão datas-limite da série
|
||
rev_por_handle = {r['handle']: r for r in rev_datas}
|
||
lim_secao = {inst: _robustos([u['1.3']['producao'] for (i, _), us in por_serie.items() if i == inst
|
||
for u in us if not u['_sd']]) for inst, _ in por_serie}
|
||
lim_serie = {k: _robustos([u['1.3']['producao'] for u in us if not u['_sd']]) for k, us in por_serie.items()}
|
||
for k, us in por_serie.items():
|
||
for u in us:
|
||
if not u['_sd']:
|
||
continue
|
||
per = next((aut[p['id']]['periodo_atividade'] for p in u['2.1']
|
||
if p['id'] and aut[p['id']]['periodo_atividade']), None)
|
||
if per:
|
||
nova = datas.atribuida(*per, 'período de atividade do autor')
|
||
elif lim_serie[k]['inicio']:
|
||
nova = datas.atribuida(lim_serie[k]['inicio'], lim_serie[k]['fim'], 'miolo das datas conhecidas da série (percentis 10–90)')
|
||
elif lim_secao[k[0]]['inicio']:
|
||
nova = datas.atribuida(lim_secao[k[0]]['inicio'], lim_secao[k[0]]['fim'],
|
||
'miolo das datas conhecidas da instituição de origem (percentis 10–90)')
|
||
else:
|
||
continue
|
||
nova['original'] = u['1.3']['producao']['original']
|
||
nova['notas'] = nova['notas'] + (u['1.3']['producao']['notas'] or [])
|
||
u['1.3']['producao'] = nova
|
||
u['8.1']['periodo'] = nova['iso']
|
||
for fid in filhos[u['id']]:
|
||
unidades[fid]['1.3']['producao'] = nova
|
||
r = rev_por_handle[u['handle']]
|
||
r.update(forma=nova['forma'], iso=nova['iso'], inicio=nova['inicio'], fim=nova['fim'],
|
||
qualidade='atribuída pelo arquivista')
|
||
|
||
def agrega(us: list[dict]) -> dict:
|
||
esp = Counter(u['_especie'] for u in us)
|
||
n = sum(u['1.5']['arquivos_digitais'] for u in us)
|
||
b = sum(u['1.5']['bytes'] for u in us)
|
||
autores = Counter(p['forma'] for u in us for p in u['2.1'] if p['id'])
|
||
lugares = Counter(u['1.3']['topica'] for u in us if u['1.3'].get('topica'))
|
||
return {'data': _limites([u['1.3']['producao'] for u in us]), 'dim': _dimensao(esp, n, b),
|
||
'autores': autores, 'lugares': lugares, 'dossies': len(us)}
|
||
|
||
def resumo(a: dict) -> str:
|
||
s = f"{a['dossies']} dossiês, {a['data']['forma']}."
|
||
if a['autores']:
|
||
s += ' Principais autores: ' + '; '.join(f'{k} ({n})' for k, n in a['autores'].most_common(5)) + '.'
|
||
if a['lugares']:
|
||
s += ' Principais lugares: ' + '; '.join(f'{k} ({n})' for k, n in a['lugares'].most_common(5)) + '.'
|
||
return s
|
||
|
||
rev_series = []
|
||
por_secao = defaultdict(list)
|
||
for (inst, sg), us in sorted(por_serie.items()):
|
||
sigla_i, nome_i, cidade_i = INSTITUICOES[inst]
|
||
info = series_[(inst, sg)]
|
||
cod = codigo(COLECAO, sigla_i, sg)
|
||
a = agrega(us)
|
||
t = _textos(TEXTOS / 'series' / f'{sigla_i}.{sg}.md')
|
||
titulo = info['titulo'] if sg != 'AV' else f'Avulsos — {nome_i}'
|
||
produtor = re.sub(r'(?i)^(cole[cç][aã]o( particular)?|fundo|arquivo|s[ée]rie|conv[eê]nio)\s+(d[aeo]s?\s+)?', '', titulo)
|
||
add({
|
||
'id': cod, 'pai': codigo(COLECAO, sigla_i), 'nivel': 3,
|
||
'1.1': cod, '1.2': {'titulo': titulo, 'atribuido': sg == 'AV'},
|
||
'1.3': {'producao': a['data']}, '1.4': {'nivel': 3, 'nome': NIVEIS[3]},
|
||
'1.5': a['dim'],
|
||
'2.1': [{'id': None, 'forma': nome_i if sg == 'AV' else produtor, 'funcao': 'colecionador'}],
|
||
'2.2': t.get('2.2'),
|
||
'3.1': t.get('3.1') or resumo(a),
|
||
'3.4': t.get('3.4') or ('Dossiês ordenados pelo número do handle na origem.' if sg != 'AV' else
|
||
'Reúne os dossiês sem coleção/fundo de origem identificável em dc.source; '
|
||
'a notação da origem, quando há, fica em 5.1 de cada dossiê.'),
|
||
'5.1': {'instituicao': nome_i, 'local': cidade_i,
|
||
'variantes_na_origem': dict(info['variantes'].most_common()) or None},
|
||
'7.3': {'descricao': HOJE},
|
||
})
|
||
por_secao[inst].extend(us)
|
||
for var, n in (info['variantes'].most_common() or [('', info['n'])]):
|
||
rev_series.append({'instituicao': inst, 'codigo': cod, 'titulo': titulo, 'dc_source': var, 'n': n})
|
||
|
||
todos = []
|
||
for inst, us in sorted(por_secao.items(), key=lambda kv: -len(kv[1])):
|
||
sigla_i, nome_i, cidade_i = INSTITUICOES[inst]
|
||
cod = codigo(COLECAO, sigla_i)
|
||
a = agrega(us)
|
||
t = _textos(TEXTOS / 'secoes' / f'{sigla_i}.md')
|
||
add({
|
||
'id': cod, 'pai': codigo(COLECAO), 'nivel': 2,
|
||
'1.1': cod, '1.2': {'titulo': f'Seção {nome_i}', 'atribuido': True},
|
||
'1.3': {'producao': a['data']}, '1.4': {'nivel': 2, 'nome': NIVEIS[2]},
|
||
'1.5': a['dim'],
|
||
'2.1': [{'id': None, 'forma': nome_i, 'funcao': 'instituição custodiadora dos originais'}],
|
||
'2.2': t.get('2.2'),
|
||
'3.1': t.get('3.1') or resumo(a),
|
||
'3.4': f"{len(filhos[cod])} séries, por coleção ou fundo de origem.",
|
||
'5.1': {'instituicao': nome_i, 'local': cidade_i},
|
||
'7.3': {'descricao': HOJE},
|
||
})
|
||
todos.extend(us)
|
||
|
||
a = agrega(todos)
|
||
c1 = codigo(COLECAO)
|
||
add({
|
||
'id': c1, 'pai': codigo(), 'nivel': 1,
|
||
'1.1': c1, '1.2': {'titulo': COLECAO_TITULO, 'atribuido': False},
|
||
'1.3': {'producao': a['data'], 'acumulacao': {'forma': '2026', 'iso': '2026'}},
|
||
'1.4': {'nivel': 1, 'nome': NIVEIS[1]}, '1.5': a['dim'],
|
||
'2.1': [{'id': None, 'forma': ENTIDADE_NOME, 'funcao': 'colecionador'}],
|
||
'2.2': textos1.get('2.2'), '2.3': textos1.get('2.3'), '2.4': textos1.get('2.4'),
|
||
'3.1': textos1.get('3.1') or resumo(a), '3.2': textos1.get('3.2'), '3.3': textos1.get('3.3'),
|
||
'3.4': textos1.get('3.4'),
|
||
'4.1': textos1.get('4.1') or ACESSO, '4.2': textos1.get('4.2'),
|
||
'4.3': textos1.get('4.3'), '4.4': textos1.get('4.4'), '4.5': textos1.get('4.5'),
|
||
'5.1': textos1.get('5.1'), '5.4': textos1.get('5.4'),
|
||
'6.2': textos1.get('6.2'),
|
||
'7.1': textos1.get('7.1'), '7.2': REGRAS, '7.3': {'descricao': HOJE},
|
||
'8.1': textos1.get('8.1'),
|
||
})
|
||
c0 = codigo()
|
||
add({
|
||
'id': c0, 'pai': None, 'nivel': 0,
|
||
'1.1': c0, '1.2': {'titulo': f'Acervo {ENTIDADE_NOME}', 'atribuido': False},
|
||
'1.3': {'producao': a['data']}, '1.4': {'nivel': 0, 'nome': NIVEIS[0]}, '1.5': a['dim'],
|
||
'2.1': [{'id': None, 'forma': ENTIDADE_NOME, 'funcao': 'entidade custodiadora'}],
|
||
'2.2': textos0.get('2.2'), '3.1': textos0.get('3.1'),
|
||
'4.1': textos0.get('4.1') or ACESSO, '7.2': REGRAS, '7.3': {'descricao': HOJE},
|
||
})
|
||
|
||
# 1.4: títulos dos níveis superiores (proc. 1.1.4)
|
||
for u in unidades.values():
|
||
cad, p = [], u['pai']
|
||
while p:
|
||
cad.append(unidades[p]['1.2']['titulo'])
|
||
p = unidades[p]['pai']
|
||
u['1.4']['hierarquia'] = list(reversed(cad)) or None
|
||
|
||
ordem = sorted(unidades.values(), key=lambda u: (u['nivel'], u['id']))
|
||
with (SAIDA / 'unidades.jsonl').open('w', encoding='utf-8') as f:
|
||
for u in ordem:
|
||
f.write(json.dumps({k: x for k, x in u.items() if not k.startswith('_') and x is not None},
|
||
ensure_ascii=False) + '\n')
|
||
with (SAIDA / 'autoridades.jsonl').open('w', encoding='utf-8') as f:
|
||
for r in aut.registros.values():
|
||
f.write(json.dumps(r, ensure_ascii=False) + '\n')
|
||
|
||
# tabelas de revisão
|
||
pl.DataFrame(rev_datas, infer_schema_length=None).filter(pl.col('qualidade').is_null() | (pl.col('qualidade') != 'exata')) \
|
||
.sort('motivo', 'original', nulls_last=True).write_csv(REVISAO / 'datas.tsv', separator='\t')
|
||
pl.DataFrame(rev_series, infer_schema_length=None).sort('codigo', 'n', descending=[False, True]) \
|
||
.write_csv(REVISAO / 'series.tsv', separator='\t')
|
||
pl.DataFrame(rev_tipos, infer_schema_length=None).group_by(pl.all()).len('n').sort('n', descending=True) \
|
||
.write_csv(REVISAO / 'tipos.tsv', separator='\t')
|
||
pl.DataFrame(rev_direitos, infer_schema_length=None).group_by(pl.all()).len('n').sort('n', descending=True) \
|
||
.write_csv(REVISAO / 'direitos.tsv', separator='\t')
|
||
pl.DataFrame(rev_lugares, infer_schema_length=None).group_by(pl.all()).len('n').sort('n', descending=True) \
|
||
.write_csv(REVISAO / 'lugares.tsv', separator='\t')
|
||
pl.DataFrame([{'id': r['id'], 'forma_autorizada': r['forma_autorizada'], 'tipo': r['tipo'],
|
||
'ocorrencias': r['ocorrencias'], 'variantes': ' | '.join(r['variantes'])}
|
||
for r in aut.registros.values()], infer_schema_length=None).write_csv(REVISAO / 'autoridades.tsv', separator='\t')
|
||
|
||
return unidades, aut
|