sysadmin/paquetes-sprpcb
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

adición de función

Browse Source
This commit is contained in:
root
2026-05-12 12:45:20 +00:00
parent b68e4e5dbb
commit 7013a4558b
3 changed files with 0 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

461
extensions/functions/datos-iniciales/datos-iniciales.py Normal file
View File

@@ -0,0 +1,461 @@
import uuid
from django.core.exceptions import ValidationError
from arches.app.functions.base import BaseFunction
from arches.app.models.system_settings import settings
from arches.app.models import models
from arches.app.models.tile import Tile
from django.db import connection, transaction
import re
import json
import logging
details = {
"name": "Datos iniciales",
"type": "node",
"description": "Establece los datos iniciales de las instancias de recursos",
"defaultconfig": {
"lugar": {"ndeId": "", "ndegrpId": ""},
"lugar_amplio": {"ndeId": "", "ndegrpId": ""},
"alcance": {"ndeId": "", "ndegrpId": ""},
"eta": {"ndeId": "", "ndegrpId": ""},
"codigo": {"ndeId": "", "ndegrpId": ""},
"nombre": {"ndeId": "", "ndegrpId": ""},
"meta_ubicacion": {"ndeId": "", "ndegrpId": ""},
"meta_identificacion": {"ndeId": "", "ndegrpId": ""},
"meta_marco_legal": {"ndeId": "", "ndegrpId": ""},
"meta_caracteristicas": {"ndeId": "", "ndegrpId": ""},
"valor_ndeId": "",
"valor_ponderado_ndeId": "",
"valor_ajuste_ndeId": "",
"valor_rango_ndeId": "",
"triggering_nodegroups": [],
},
"classname": "DatosInicio",
"component": "views/components/functions/datos-iniciales",
"functionid": "6ec500db-51cc-4cc8-a796-ee46a56f6998",
}
class DatosInicio(BaseFunction):
def get(self):
raise NotImplementedError
def componer_codigo(self, tile, request, is_function_save_method=True):
"""
Función que construye el código de identificación de ficha, en base a
la ubicación, ámbito y una secuencia numérica
Args:
self: Objeto código.
tile: Tile al que se adjuntará/modificará el código.
request: Solicitud WSGI utilizada para verificar que la llamada sea resultado de una acción del usuario.
Nota: La función devuelve el valor si está vacío.
is_function_save_method: Un valor booleano que indica si la función que lo llama es la función de guardado.
jbti - 20.9.25
"""
#Comprueba que sea un llamado válido
if request is None and is_function_save_method == True:
return
def actualizar_descriptors(str_ficha_id):
try:
qry = f"""
UPDATE resource_instances
SET descriptors = jsonb_set(
descriptors,
'{{es,map_popup}}', '"Indefinido"')
WHERE resourceinstanceid = '{str_ficha_id}'
AND descriptors->'es'->>'map_popup' LIKE '<%>'
RETURNING descriptors->'es'->>'map_popup';
"""
with connection.cursor() as cursor:
cursor.execute(qry)
nvo_descriptor = cursor.fetchone()
if nvo_descriptor is not None:
return nvo_descriptor
else:
return ('Sin actualizar',)
except Exception as e:
self.logger.error(e,"\t*** Falla al actualizar descriptor")
def extraer_depto(str_val_depto): # retorna el departamento como str_concept_value
try:
div_admin = '4f6caa05-23e3-42a4-8322-98b914fb3ebf' # (División administrativa)
region = '2eb6660c-1754-42cf-959d-0edaf7779dae' # Region de Bolivia
qry = f"""
SELECT val.valueid::text, val.value
FROM values val
JOIN values vval
ON val.conceptid = vval.conceptid
JOIN relations rel
ON val.conceptid = rel.conceptidto
WHERE val.valuetype = 'prefLabel'
AND vval.valuetype = 'hiddenLabel'
AND (
rel.conceptidfrom = '{div_admin}'
OR rel.conceptidfrom = '{region}')
AND vval.value = '{str_val_depto}';
"""
with connection.cursor() as cursor:
cursor.execute(qry)
departamento = cursor.fetchone()
if departamento is not None:
return departamento
else:
return ()
except Exception as e:
self.logger.error(e,"\t*** Falla al extraer departamento")
def establecer_grupos(string_id_recurso, int_id_grp_eta, sufijo_eta_model, esRecurso = True):
int_id_todas_las_etas = 10000
int_id_eta_model = int_id_grp_eta + sufijo_eta_model
try:
qry = f"""
INSERT INTO guardian_groupobjectpermission (object_pk, content_type_id, group_id, permission_id)
VALUES
('{string_id_recurso}',33,2,130),
('{string_id_recurso}',33,2,131),
('{string_id_recurso}',33,{int_id_todas_las_etas},132),
('{string_id_recurso}',33,{int_id_grp_eta},132),
('{string_id_recurso}',33,{int_id_eta_model},132)
ON CONFLICT (group_id, permission_id, object_pk) DO NOTHING;
"""
with connection.cursor() as cursor:
cursor.execute(qry)
except Exception as e:
self.logger.error(e,"\t*** Falla al establecer grupos")
def es_ambito(string_id_concepto_value):
qry = f"""
SELECT CASE WHEN EXISTS (
SELECT *
FROM relations r RIGHT JOIN values v ON r.conceptidfrom = v.conceptid
WHERE r.relationtype = 'narrower' AND v.valueid = '{string_id_concepto_value}'
)
THEN CAST(1 AS BIT)
ELSE CAST(0 AS BIT) END;
"""
with connection.cursor() as cursor:
cursor.execute(qry)
esAmbito = cursor.fetchone()[0]
return esAmbito
def extraer_concepto_valor(string_id_concepto_value, str_tipo_valor): # return string numeric code
qry = f"""
SELECT val.value FROM values val INNER JOIN values v ON v.conceptid = val.conceptid
WHERE val.valuetype='{str_tipo_valor}' AND v.valueid='{string_id_concepto_value}'
"""
with connection.cursor() as cursor:
cursor.execute(qry)
valorConcepto = cursor.fetchone()
if valorConcepto is not None:
return valorConcepto[0]
else:
return '00'
def extraer_datos_recurso(string_id_recurso): # Return ['graph_id', principaluser_id]
if string_id_recurso is not None:
qry = f"""
SELECT graphid::text, principaluser_id FROM resource_instances
WHERE resourceinstanceid ='{string_id_recurso}'
"""
cursor = connection.cursor()
cursor.execute(qry)
datos_recurso = list(cursor.fetchone())
cursor.close()
else:
datos_recurso = []
return datos_recurso
def extraer_etaId(usuario_id): # Return 'recurso_eta_id'
eta_grp_ndeId = '263826f2-f4d9-11ef-822e-0242ac120006'
qry = f"""
SELECT t.resourceinstanceid AS etaid, ug.eta_grpid AS etagrp FROM tiles t
JOIN (
SELECT user_id, ((MAX(group_id) / 100) * 100) AS eta_grpid
FROM auth_user_groups WHERE group_id > 9999 GROUP BY user_id) ug
ON t.tiledata['{eta_grp_ndeId}']::int = ug.eta_grpid
WHERE ug.user_id = {usuario_id} AND t.nodegroupid = '{eta_grp_ndeId}';
"""
with connection.cursor() as cursor:
cursor.execute(qry)
etaNdeId = cursor.fetchone()
if etaNdeId is not None :
return etaNdeId
else:
return ['deb0ae78-56f1-4bcf-abb2-d02ffcdd33c1', 10100] # Actor Min.Culturas
def insertar_actor(rsi_id, eta_tileid, eta_actorid, eta_ndeid, GraphId, eta_ndegrpid):
ontProp = 'ac41d9be-79db-4256-b368-2f4559cfbe55'
actorGraphId = '72420da0-db59-4048-aea2-03b0bc5cdcdf'
try:
qry = f"""
INSERT INTO resource_x_resource (
resourcexid, relationshiptype,
resourceinstanceidfrom,
resourceinstanceidto,
modified, created, inverserelationshiptype,
tileid,
nodeid,
resourceinstancefrom_graphid,
resourceinstanceto_graphid
)
VALUES (
gen_random_uuid(), '{ontProp}',
'{rsi_id}',
'{eta_actorid}',
now(), now(), '{ontProp}',
'{eta_tileid}',
'{eta_ndeid}',
'{GraphId}',
'{actorGraphId}'
) returning resourcexid
"""
with connection.cursor() as cursor:
cursor.execute(qry)
nvo_resxid = cursor.fetchone()[0]
qry2 = f"""
UPDATE tiles SET tiledata = jsonb_set(
tiledata,
'{{{eta_ndeid}}}', '[{{
"resourceId": "{eta_actorid}",
"ontologyProperty": "{ontProp}",
"resourceXresourceId": "{nvo_resxid}",
"inverseOntologyProperty": "{ontProp}"
}}]'
)
WHERE nodegroupid = '{eta_ndegrpid}'
AND resourceinstanceid = '{rsi_id}'
returning tiledata
"""
cursor.execute(qry2)
cursor.close()
except Exception as e:
self.logger.error(e,"\t*** Falla al insertar actor")
def actualizar_codigo(MetaCodigo):
#calculo de la secuencia
nvo_codigo = [MetaCodigo['tipoM'],MetaCodigo['codU'], MetaCodigo['codAm'], MetaCodigo['codsA'], MetaCodigo['num']]
#languages = models.Language.objects.all()
#default_language = languages.get(code=settings.LANGUAGE_CODE)
#return {default_language.code: {"value": '-'.join(nvo_codigo), "direction": default_language.default_direction}}
return '-'.join(nvo_codigo)
def extraer_num_siguiente(strCod_ndeId, strCod_ndegrpId, MetaCodigo): # Return int
str_cod_num = '-'.join([MetaCodigo['tipoM'],MetaCodigo['codU'], MetaCodigo['codAm'], MetaCodigo['codsA']])
qry = f"""
SELECT MAX(regexp_match(tiledata->>'{strCod_ndeId}','-(\d+)$')) AS max_cod_num
FROM tiles WHERE nodegroupid = '{strCod_ndegrpId}'
AND tiledata->>'{strCod_ndeId}' LIKE '{str_cod_num}%';
"""
with connection.cursor() as cursor:
cursor.execute(qry)
ultimo_num = cursor.fetchone()
if ultimo_num is not None and ultimo_num[0] is not None:
print(f"Siguiente número del código: {ultimo_num}")
return str(int(ultimo_num[0][0]) + 1)
else:
return str(1)
def extraer_metavalor(metavalor_ngId, metavalor_nId, resource_id, metavalor={}): # Retorna o actualiza metavalor dict
metaValor_tiles = Tile.objects.filter(nodegroup_id = str(metavalor_ngId), resourceinstance_id = resource_id)
metavalor_actual = {}
if len(metaValor_tiles) > 0 :
metaValor_tile = metaValor_tiles[0]
if metavalor_nId in metaValor_tile.data and metaValor_tile.data[metavalor_nId] is not None:
metavalor_actual = eval(metaValor_tile.data[metavalor_nId])
elif len(metaValor_tiles) == 0 and len(metavalor) > 0:
metavalor_actual.update(metavalor)
metaValor_tile = Tile.get_blank_tile_from_nodegroup_id(metavalor_ngId, resource_id, None)
metaValor_tile.data[metavalor_nId] = str(metavalor_actual)
metaValor_tile.save()
return metavalor_actual
def extraer_codigo(codigo_ngId, codigo_nId, resource_id): # Return str
cod_num = ""
codigo_tiles = Tile.objects.filter(nodegroup_id = str(codigo_ngId), resourceinstance_id = resource_id)
if len(codigo_tiles) > 0 :
for codi in codigo_tiles :
# Si el código es del tipo principal
codigo_tile = codi if 'c07032eb-b50a-4048-b18c-bde5e8e6ad6a' in codi.data.values() else codigo_tiles[0]
if codigo_nId in codigo_tile.data and codigo_tile.data[codigo_nId] is not None:
#cod_num = str(codigo_tile.data[codigo_nId]['es']['value'])
cod_num = str(codigo_tile.data[codigo_nId])
cod_num = re.sub(r'^(.+)-','',cod_num)
if cod_num != '' : cod_num = cod_num.rjust(6,'0')
return cod_num
try:
# procedimiento para definir los metadatos
metaCodlugar = {'tipoM': 'PM', 'codU': '000000', 'num': '000000'}
metaCodIden = {'codAm': '00', 'codsA': '00'}
metaGrupo = {'grp_eta': 10100, 'sufijo_grp_model': 0}
datos_recurso = extraer_datos_recurso(tile.resourceinstance_id) # ['graph_id', principaluser_id]
etaActor_id_grp = extraer_etaId(datos_recurso[1])
etaActor_id = etaActor_id_grp[0]
metaGrupo.update({'grp_eta': etaActor_id_grp[1]})
es_patrimonio = True
match datos_recurso[0]:
case '72420da0-db59-4048-aea2-03b0bc5cdcdf' : # Actores
es_patrimonio = False
metaGrupo.update({'sufijo_grp_model': 17})
case '6afc4e9e-727e-4152-8084-309267752650' : # Medios
es_patrimonio = False
metaGrupo.update({'sufijo_grp_model': 16})
case "2bf240e6-718e-44f0-8073-4962d84931fe" : # Si el modelo es del tipo actividades
metaGrupo.update({'sufijo_grp_model': 15})
case '76331d8a-4d85-11ee-a708-ffcd06c08a5d': # Si el modelo es del tipo arqueologico
metaGrupo.update({'sufijo_grp_model': 14})
case '9b0e22c7-5c56-4be6-aa89-ae2632ba28e8': # Si el modelo es del tipo inmaterial
metaCodlugar.update({'tipoM': 'PI'})
metaGrupo.update({'sufijo_grp_model': 13})
case '10b03fec-4d85-11ee-a708-ffcd06c08a5d': # Si el modelo es del tipo bienes muebles
metaGrupo.update({'sufijo_grp_model': 12})
case '8ea3bf67-89ce-474c-b3cb-6dd5ee3d6198': # Si el modelo es del tipo bienes inmuebles
metaCodlugar.update({'num': 'D00M00L00'})
metaGrupo.update({'sufijo_grp_model': 11})
metaIden_ndegrp = self.config['meta_identificacion']['ndegrpId']
metaIden_nde = self.config['meta_identificacion']['ndeId']
lugarAmplio_nde = self.config['lugar_amplio']['ndeId']
metaIden = {}
tup_depto = ()
if es_patrimonio:
# procedimiento para fijar el el actor administrativo
eta_ndegrpId = self.config['eta']['ndegrpId']
eta_ndeId = self.config['eta']['ndeId']
etaTile = Tile.objects.filter(nodegroup_id=str(eta_ndegrpId), resourceinstance_id=tile.resourceinstance_id)
if len(etaTile) == 0:
eta_tile = Tile.get_blank_tile_from_nodegroup_id(eta_ndegrpId, tile.resourceinstance_id, None)
eta_tile.save()
insertar_actor(tile.resourceinstance_id, str(eta_tile.tileid), etaActor_id, eta_ndeId, datos_recurso[0], eta_ndegrpId)
metaLugar_ndegrp = self.config['meta_ubicacion']['ndegrpId']
metaLugar_nde = self.config['meta_ubicacion']['ndeId']
codigo_ndegrp = self.config['codigo']['ndegrpId']
codigo_nde = self.config['codigo']['ndeId']
codigo_num = metaCodlugar['num']
# Código de lugar a meta lugar
if str(tile.nodegroup_id) == self.config['lugar']['ndegrpId']:
lugar_value = tile.data[self.config["lugar"]["ndeId"]]
if lugar_value is not None :
lugar_cod_value = extraer_concepto_valor(lugar_value, "hiddenLabel")
if len(lugar_cod_value) > 2 :
lugar_cod_value = lugar_cod_value.ljust(5,'0')
lugar_cod_value = lugar_cod_value.rjust(6,'0')
else:
lugar_cod_value = lugar_cod_value.ljust(6,'0')
metaCodlugar.update(codU = lugar_cod_value)
# establecer lugar extendido
tup_depto = extraer_depto(lugar_cod_value[0:2])
print("\t*** tup_depto: ",tup_depto)
else:
tup_depto = extraer_depto('00')
if len(tup_depto) > 0 and tup_depto[0] is not None: tile.data[self.config['lugar_amplio']['ndeId']] = [str(tup_depto[0])]
# extraer/crear número del código
num_anterior = extraer_codigo(codigo_ndegrp, codigo_nde, tile.resourceinstance_id)
if num_anterior != '' : metaCodlugar.update({'num': num_anterior})
tile.data[metaLugar_nde] = str(metaCodlugar)
# Ámbito a meta identificacion y código si es el caso
if str(tile.nodegroup_id) == metaIden_ndegrp and codigo_ndegrp == metaIden_ndegrp :
metalugar = extraer_metavalor(metaLugar_ndegrp, metaLugar_nde, tile.resourceinstance_id)
if 'tipoM' in metalugar: metaCodlugar.update({'tipoM': metalugar['tipoM']})
if 'codU' in metalugar: metaCodlugar.update({'codU': metalugar['codU']})
if 'num' in metalugar: metaCodlugar.update({'num': metalugar['num']})
alcance_cod_values = tile.data[self.config["alcance"]["ndeId"]]
if alcance_cod_values is not None:
match len(alcance_cod_values):
case 0:
metaCodIden = {'codAm': '00', 'codsA': '00'}
case 1:
metaCodIden = {'codAm': extraer_concepto_valor(alcance_cod_values[0], "hiddenLabel").rjust(2,'0'), 'codsA': '00'}
if es_ambito(alcance_cod_values[0]) == '1':
metaCodIden = {'codAm': extraer_concepto_valor(alcance_cod_values[0], "hiddenLabel").rjust(2,'0'), 'codsA': '00'}
else:
metaCodIden = {'codAm': '00', 'codsA': extraer_concepto_valor(alcance_cod_values[0], "hiddenLabel").rjust(2,'0')}
case _ if len(alcance_cod_values) > 1: # Falta completar validación de ámbito y subambito
if es_ambito(alcance_cod_values[0]) == '1':
metaCodIden = {'codAm': extraer_concepto_valor(alcance_cod_values[0], "hiddenLabel").rjust(2,'0'), 'codsA':
extraer_concepto_valor(alcance_cod_values[1], "hiddenLabel").rjust(2,'0')}
else:
metaCodIden = {'codAm': extraer_concepto_valor(alcance_cod_values[1], "hiddenLabel").rjust(2,'0'), 'codsA':
extraer_concepto_valor(alcance_cod_values[0], "hiddenLabel").rjust(2,'0')}
metacodigo = metaCodlugar | metaCodIden
metaIden.update(metaCodIden)
metaIden.update(metaGrupo)
# Procesado del número del código
if codigo_nde in tile.data and tile.data[codigo_nde] is not None :
codigo_num = str(tile.data[codigo_nde])
codigo_num = re.sub(r'^(.+)-','',codigo_num)
if "?" in re.sub(r'^(.+)-','',codigo_num) or codigo_num == '000000':
metacodigo.update(num = extraer_num_siguiente(codigo_nde, codigo_ndegrp, metacodigo).rjust(6,'0'))
else:
metacodigo.update(num = codigo_num.rjust(6,'0'))
tile.data[self.config['codigo']['ndeId']] = actualizar_codigo(metacodigo)
# insertar permisos de ficha
metaiden = extraer_metavalor(metaIden_ndegrp, metaIden_nde, tile.resourceinstance_id)
if "grp_eta" not in metaiden :
establecer_grupos(tile.resourceinstance_id, metaGrupo['grp_eta'], metaGrupo['sufijo_grp_model'])
print("\t*** metacodigo : ", metacodigo)
metaIden.update(metaGrupo)
tile.data[metaIden_nde] = str(metaIden)
if str(tile.nodegroup_id) == self.config['nombre']['ndegrpId'] and not es_patrimonio:
print("\t*** insertar permisos de ficha complementaria")
metaIden = extraer_metavalor(metaIden_ndegrp, metaIden_nde, tile.resourceinstance_id)
if "grp_eta" not in metaIden :
establecer_grupos(tile.resourceinstance_id, metaGrupo['grp_eta'], metaGrupo['sufijo_grp_model'])
metaIden = extraer_metavalor(metaIden_ndegrp, metaIden_nde, tile.resourceinstance_id, metaGrupo)
print("\t*** metaIden : ", metaIden,"\n")
print("\t*** Nvo. map_popup: ",actualizar_descriptors(tile.resourceinstance_id)[0])
return
except Exception as e:
self.logger.error(e,"\t*** Falla al fijar los datos iniciales")
return
def save(self, tile, request, context=None):
self.logger = logging.getLogger(__name__)
print("\n\t*** llamado a la función datos_iniciales.save ***")
self.componer_codigo(tile=tile, request=request, is_function_save_method=True)
return
# occurrs after Tile.save
def post_save(self, *args, **kwargs):
raise NotImplementedError
def delete(self, tile, request):
raise NotImplementedError
def on_import(self, tile):
self.componer_codigo(tile=tile, request=None, is_function_save_method=False)
return
# saves changes to the function itself
def after_function_save(self, tile, request):
raise NotImplementedError