teste_docs / document /evolutivo.py
gui-sparim's picture
Organiza justificativa e composição do BDI por construção
46d350b verified
Raw History Blame Contribute Delete
44.5 kB
"""EVO narrative: explain saved evidence in the body, with annexes as support."""
from collections import OrderedDict
from decimal import Decimal, ROUND_HALF_UP
import json
import re
from docx.enum.text import WD_ALIGN_PARAGRAPH
from docx.oxml import OxmlElement
from docx.oxml.ns import qn
from .formatters import (add_body_text, add_simple_table, add_subsection_title,
add_subsubsection_title, add_subsubsubsection_title, add_subsubsubsubsection_title)
from .avaliacoes_importadas import currency
from .mesa_content import content_for
from .evo_adjustments import REDUCERS, applied_reducers
from .fundamentacao import number, global_foundation
from .footnotes import add_footnote, add_first_footnote, order_footnotes
from .equations import cost_equations, depreciation_equations
from .formatters.table import set_table_font_size
from .formatters.table_fit import fit_table_to_page
from .anexos.metodologia import (_gerar_variaveis, _adicionar_pesquisa,
_adicionar_fundamentos_comparativos, _adicionar_tabela_equacao)
from models.imported_context import context_key, snapshot_model, matching_local
from models.construction_identity import construction_identity
def fmt(value, digits=2):
value = number(value)
if value is None:
return 'Não informado'
text = f'{value.quantize(Decimal(1).scaleb(-digits), rounding=ROUND_HALF_UP):,.{digits}f}'
return text.replace(',', '#').replace('.', ',').replace('#', '.')
def pct(value, fraction=False):
value = number(value)
return 'Não informado' if value is None else fmt(value * (100 if fraction else 1)) + '%'
def ref(row, show_base=True):
value = f"IPTU {row['ano']}"
if show_base:
value += f" / base {row['ano_base']}"
return value + (f" — {row['titulo_origem']} [{row['id']}]" if row.get('referencia_duplicada') else '')
def source(row):
return row.get('evolutivo') or {}
def terrain(row):
return source(row).get('terreno') or {}
def result(row):
return source(row).get('resultadoComposicao') or {}
def complement(row, key, fallback='Não informado no arquivo; complemento técnico pendente.'):
return (row.get('complementos') or {}).get(key) or fallback
def table(doc, rows, widths=None, compact=False, keep_together=False, font_size=None, auto_fit=True):
section = doc.sections[-1]
annual = any(re.search(r'\bIPTU\s+\d{4}\b', str(value)) for row in rows for value in row)
if doc.paragraphs:
previous = doc.paragraphs[-1]
if previous.text.startswith(('Construção ', 'Composição das áreas', 'IPTU ', 'Memória integral', 'Memória das etapas')):
previous.paragraph_format.keep_with_next = True
obj = add_simple_table(doc, rows, largura_colunas=widths,
largura_total=int(section.page_width-section.left_margin-section.right_margin))
for row_index,row in enumerate(obj.rows):
row._tr.get_or_add_trPr().append(OxmlElement('w:cantSplit'))
for index, cell in enumerate(row.cells):
props = cell._tc.get_or_add_tcPr()
for no_wrap in list(props.findall(qn('w:noWrap'))): props.remove(no_wrap)
margins = OxmlElement('w:tcMar')
padding = 25 if compact or annual else 65
for side, size in [('top',padding),('bottom',padding),('left',80),('right',80)]:
node = OxmlElement('w:'+side); node.set(qn('w:w'), str(size)); node.set(qn('w:type'),'dxa'); margins.append(node)
props.append(margins)
for para in cell.paragraphs:
para.paragraph_format.line_spacing = 1.05
para.paragraph_format.keep_with_next = keep_together and row_index<len(obj.rows)-1
para.alignment = WD_ALIGN_PARAGRAPH.LEFT if index == 0 else WD_ALIGN_PARAGRAPH.CENTER
borders = OxmlElement('w:tblBorders')
for side in ['top','left','bottom','right','insideH','insideV']:
node = OxmlElement('w:'+side); node.set(qn('w:val'),'single'); node.set(qn('w:sz'),'4'); node.set(qn('w:color'),'D9D9D9'); borders.append(node)
obj._tbl.tblPr.append(borders)
if annual and auto_fit:
fit_table_to_page(doc, obj)
elif font_size is not None:
set_table_font_size(obj, font_size)
return obj
def heading(doc, num, label, level):
counter = {2:num.subsecao,3:num.subsubsecao,4:num.subsubsubsecao,5:num.subsubsubsubsecao}[level]
builder = {2:add_subsection_title,3:add_subsubsection_title,4:add_subsubsubsection_title,5:add_subsubsubsubsection_title}[level]
value = counter()
paragraph = builder(doc,value,label)
paragraph.paragraph_format.keep_with_next = True
return value
def equation(doc, text):
# Cost/depreciation formulas remain separate from the statistical model table.
para = add_body_text(doc,text)
if para is not None:
para.alignment = WD_ALIGN_PARAGRAPH.CENTER
para.paragraph_format.keep_together = True
def compared(doc, rows, metrics, notes=None, show_base=False):
"""Keep all exercises in one table and fit after values and notes exist."""
if not rows:
return
labels = [ref(row, show_base=show_base) for row in rows]
if len(set(labels)) != len(labels):
labels = [f"{label}\n{row['titulo_origem']} [{row['id']}]" for label,row in zip(labels,rows)]
obj = table(doc, [['Parâmetro / valor']+labels] + [[label]+[getter(row) for row in rows] for label,getter in metrics],
[2.4]+[1.45]*len(rows), auto_fit=False)
used_notes = []
for index, (label, _) in enumerate(metrics, 1):
if notes and label in notes:
add_footnote(doc, obj.rows[index].cells[0].paragraphs[0], notes[label])
used_notes.append(notes[label])
fit_table_to_page(doc, obj, notes=used_notes)
return obj
def model_analysis(doc, model, tipo_y):
"""Resumo estatístico no padrão do laudo, sem inferir diagnósticos ausentes."""
r2 = number(model.r2) if model else None
text = ''
if r2 is not None and 0 <= r2 <= 1:
value = 'valor total em reais' if tipo_y == 'total' else 'valor unitário em reais por metro quadrado'
correlation = r2.sqrt()
interpretation = (', indicando a existência de um alto grau de correlação entre as variáveis consideradas.'
if correlation >= Decimal('.9') else '.')
text = (f'O coeficiente de correlação linear do modelo é de {fmt(correlation,4)}{interpretation} '
f'O modelo de regressão obtido explica aproximadamente {fmt(r2*100,0)}% da variabilidade do {value}, '
'sendo o restante da variabilidade atribuída a outras variáveis não consideradas ou a oscilações de caráter aleatório. ')
text += ('As especificações e os resultados detalhados encontram-se no Anexo I - Banco de Dados, '
'no Anexo II - Planilha de Cálculo e Resultados Estatísticos e no Anexo III - Gráficos.')
add_body_text(doc, text)
def grouped_notes(doc, rows, getter, prefix, always_refs=False):
groups = OrderedDict()
for row in rows:
text = getter(row)
if text: groups.setdefault(str(text), []).append(ref(row))
for text, references in groups.items():
scope = f" ({'; '.join(references)})" if always_refs or len(references)!=len(rows) else ""
add_body_text(doc, f"{prefix}{scope}: {text}")
def definitions_for(doc, rows, keys):
"""Different historical packages retain their own wording and applicability."""
for key in keys:
variants = OrderedDict()
for row in rows:
text = content_for(source(row)).get('notas',{}).get(key)
if text: variants.setdefault(text,[]).append(ref(row))
for text,references in variants.items():
add_body_text(doc,text + (f" Aplicável a {'; '.join(references)}." if len(variants)>1 else ''))
def gerar_corpo_evolutivo(doc, imported, num, caracteristicas_refs=None):
rows = imported['avaliacoes']
add_body_text(doc, 'O valor do imóvel é obtido pelo método evolutivo, mediante a composição do valor do terreno, estimado pelo método comparativo de dados de mercado, com o custo de reedição das construções, pelo método da quantificação de custo. A composição considera os ajustes registrados e, ao final, o fator de comercialização (Fc).')
add_body_text(doc, 'A demonstração a seguir transcreve os parâmetros e resultados salvos na avaliação, mantendo suas referências temporais. O Anexo IV reúne a memória detalhada. O valor calculado permanece distinto de eventual valor final adotado pelo avaliador.')
refs = {}
refs['terreno'] = heading(doc,num,'AVALIAÇÃO DO TERRENO',2)
heading(doc,num,'Aplicação Método Comparativo',3)
models = OrderedDict()
for row in rows: models.setdefault(context_key(row),[]).append(row)
groups = [(snapshot_model(group[0]), group) for group in models.values()]
for model, group in groups:
if model:
if len(groups) > 1:
add_body_text(doc, f"Pesquisa utilizada nas avaliações: {'; '.join(ref(row) for row in group)}.")
_adicionar_pesquisa(doc, model)
if any(model for model, _ in groups):
_adicionar_fundamentos_comparativos(doc)
for model, group in groups:
if model:
if len(groups) > 1:
add_body_text(doc, f"As variáveis e a equação a seguir se aplicam às avaliações: {'; '.join(ref(row) for row in group)}.")
_gerar_variaveis(doc, model, num, nivel=4, tipo_y=(terrain(group[0]).get('avaliacao') or {}).get('tipo_y','unitario'))
heading(doc,num,'Equação Modelo Estatístico',4)
add_body_text(doc, 'O modelo de regressão que melhor se ajustou aos dados da amostra é transcrito a seguir:')
doc.paragraphs[-1].paragraph_format.keep_with_next = True
_adicionar_tabela_equacao(doc, model)
local, _ = matching_local(group[0])
model_analysis(doc, local or model, (terrain(group[0]).get('avaliacao') or {}).get('tipo_y'))
else:
add_body_text(doc,'O arquivo não preserva o contexto estatístico necessário para transcrever a equação, a amostra e o cadastro de variáveis com segurança. Permanecem documentados os parâmetros e resultados importados; a versão correspondente do modelo deve ser complementada.')
heading(doc,num,'Cálculo do valor do terreno',3)
heading(doc,num,'Parâmetros e valor estimado pelo modelo',4)
keys = list(dict.fromkeys(key for row in rows for key in (terrain(row).get('parametrosAplicados') or row['avaliacao'].get('valores_x') or {})))
metrics = [(str(key),lambda row,k=key: str((terrain(row).get('parametrosAplicados') or row['avaliacao'].get('valores_x') or {}).get(k,'Não informado'))) for key in keys]
compared(doc,rows,metrics + [
('Estimado Unitário', lambda row: currency(row['avaliacao'].get('estimado_unitario'))),
('CA −15%', lambda row: currency(row['avaliacao'].get('ca_inf'))),
('CA +15%', lambda row: currency(row['avaliacao'].get('ca_sup'))),
('IC 80% Inf.', lambda row: currency(row['avaliacao'].get('ic_inf'))),
('IC 80% Sup.', lambda row: currency(row['avaliacao'].get('ic_sup'))),
], notes={
'Estimado Unitário': 'Estimativa do valor do terreno por metro quadrado (R$/m²), antes dos ajustes. Quando o modelo estima o valor total, corresponde à conversão desse valor pela área utilizada no modelo.',
'CA −15%': 'CA: campo de arbítrio. CA −15% e CA +15% são os limites inferior e superior, correspondentes a 85% e 115% da estimativa central. Não representam descontos automaticamente aplicados. Os limites são expressos na unidade da variável dependente do modelo: R$/m² no modelo unitário e R$ no modelo total.',
'IC 80% Inf.': 'IC 80%: intervalo de confiança de 80% da estimativa do modelo. Inf. e Sup. indicam seus limites inferior e superior. Expressam a incerteza estatística e não se confundem com o campo de arbítrio. Os limites são expressos em R$/m² no modelo unitário e R$ no modelo total.',
})
add_body_text(doc,'A referência anual das avaliações é distinta do ano-base omitido na codificação das variáveis dicotômicas do modelo.')
heading(doc,num,'Ajustes aplicados e valor territorial adotado',4)
# Load after shared formatters are available to the narrative helper.
from .territorial_narrative import gerar_ajustes_territoriais, justificar_area
gerar_ajustes_territoriais(doc,rows,num,caracteristicas_refs or [])
heading(doc,num,'Síntese do valor territorial adotado',5)
add_body_text(doc,'Concluída a análise dos ajustes, o quadro a seguir reúne a formação do valor territorial em cada referência: parte-se do valor unitário do modelo, consideram-se os fatores aplicados e a escolha no campo de arbítrio e multiplica-se o valor unitário adotado pela área de aplicação.').paragraph_format.keep_with_next = True
metrics = [('VU antes dos ajustes (R$/m²)',lambda row:currency(terrain(row).get('valorUnitarioOriginal',row['avaliacao'].get('estimado_unitario'))))]
metrics += reducer_metrics(rows,True)
if any(applied_reducers(row,True) for row in rows):
metrics.append(('VU após redutores territoriais (R$/m²)',lambda row:currency(result(row).get('valorUnitarioAposRedutoresTerritoriais'))))
metrics += [
('Fator do campo de arbítrio',lambda row:fmt(result(row).get('fatorCaFinal'),4)),
('VU territorial adotado (R$/m²)',lambda row:currency(row.get('valor_unitario'))),
('Área utilizada no modelo (m²)',lambda row:fmt(terrain(row).get('areaModelo',row['avaliacao'].get('valor_area')))),
('Área de aplicação do VU (m²)',lambda row:fmt(row.get('area'))),
('Valor do terreno (R$)',lambda row:currency(row.get('valor_terreno'))),
]
compared(doc,rows,metrics,show_base=False)
justificar_area(doc,rows)
refs['construcoes'] = heading(doc,num,'AVALIAÇÃO DAS CONSTRUÇÕES',2)
heading(doc,num,'Aplicação do método da quantificação de custo',3)
add_body_text(doc,'O custo de reedição é documentado a partir da área de cálculo, do custo unitário básico (CUB) histórico, das adequações adotadas, do BDI e da depreciação física com valor residual. Os valores e coeficientes são os registrados para cada sequência construtiva.')
cost_equations(doc)
content = content_for(source(rows[0]))
heading(doc,num,'Características das construções e áreas',4)
for note in content.get('notas_area',[]): add_body_text(doc,note)
groups = construction_groups(rows)
if not groups: add_body_text(doc,'Não há sequências construtivas detalhadas preservadas; o total importado permanece identificado na composição.')
construction_characteristics(doc,groups,num=num)
heading(doc,num,'CUB histórico, referência temporal e adequações',4)
construction_cub_notes(doc,groups)
heading(doc,num,'BDI e justificativa',4)
definitions_for(doc,rows,['bdi'])
add_body_text(doc,'O BDI informado em percentual é convertido no multiplicador 1 + BDI/100. Os percentuais de referência orientam a análise; não constituem, por si, uma composição de BDI efetivamente adotada.')
construction_bdi_notes(doc,groups,num=num)
heading(doc,num,'Depreciação e conservação',4)
add_body_text(doc,'A depreciação física considera a vida útil consumida e o estado de conservação, pelo método de Ross-Heidecke. O valor residual integra o fator remanescente aplicado ao custo de reedição.').paragraph_format.keep_with_next = True
depreciation_equations(doc)
add_body_text(doc,'Os resultados apresentados são os registrados na avaliação, sem recomposição a partir dos coeficientes arredondados para exibição.')
heading(doc,num,'Cálculo por construção e exercício',4)
for entries in groups.values():
heading(doc,num,construction_name(entries[0][1],entries[0][2]),5)
by_id = {row['id']: c for row,c,i in entries}
manual_notes = OrderedDict()
for row,c,i in entries:
note = construction_depreciation_notes(row,c).get('manual')
if note: manual_notes.setdefault(note,[]).append(ref(row,show_base=False))
manual_text = ' '.join(f"{'; '.join(references)}: {text}" for text,references in manual_notes.items())
# One comparative calculation table per effective construction, with all
# annual inputs together; characteristics above are grouped by changes.
compared(doc,[row for row,c,i in entries],[
('Data do CUB',lambda row:cub_date(row)),
('Projeto CUB',lambda row:'Composto' if by_id[row['id']].get('cubCompostoAtivo') else str(by_id[row['id']].get('cubTipo') or 'Não informado').split(' (',1)[0]),
('CUB (R$/m²)',lambda row:currency(by_id[row['id']].get('valorCub'))),
('Adequação CUB (%)',lambda row:pct(by_id[row['id']].get('percentualCubCalculado',by_id[row['id']].get('percentualCub')))),
('CUNIT (R$/m²)',lambda row:currency(by_id[row['id']].get('custoUnitario'))),
('BDI (%)',lambda row:pct(by_id[row['id']].get('bdi'))),
('Fator BDI',lambda row:fmt(1+number(by_id[row['id']]['bdi'])/100,4) if number(by_id[row['id']].get('bdi')) is not None else 'Não informado'),
('Vida útil consumida (%)',lambda row:pct(by_id[row['id']].get('percentualVidaUtil'))),
('Estado de conservação',lambda row:conservation_code(by_id[row['id']].get('estadoConservacao'))),
('Coeficiente H',lambda row:fmt(by_id[row['id']].get('coefHeidecke'),4)),
('Depreciação K',lambda row:fmt(by_id[row['id']].get('depreciacao'),4)),
('Kd',lambda row:fmt(by_id[row['id']].get('kd'),3)),
('Valor da construção (R$)',lambda row:currency(by_id[row['id']].get('vcalcConstSeq'))),
],notes={'Vida útil consumida (%)':manual_text} if manual_text else None)
refs['composicao'] = heading(doc,num,'AVALIAÇÃO DO IMÓVEL OBJETO',2)
add_body_text(doc,'Na ausência de outros ajustes, a composição corresponde a VI = (VT + VC) × Fc, em que VT é o valor territorial e VC é o custo de reedição das construções. Havendo ajustes, a sequência efetivamente salva, apresentada abaixo, prevalece sobre essa expressão simplificada.').paragraph_format.keep_with_next = True
gerar_ajustes(doc,rows,territorial=False)
metrics = [
('Valor do terreno',lambda row:currency(row['valor_terreno'])),
('Valor das construções',lambda row:currency(row['valor_construcao'])),
]
if any(applied_reducers(row,False) for row in rows):
metrics += [('Base antes dos redutores do imóvel',lambda row:currency(result(row).get('valorAposConstrucao')))]
metrics += reducer_metrics(rows,False)
metrics += [('Base após redutores / antes do Fc',lambda row:currency(result(row).get('valorAposRedutoresImovel')))]
metrics += [
('Fc aplicado',lambda row:fmt(row.get('fator_comercializacao'),4)),
('Valor total calculado',lambda row:currency(result(row).get('valorTotalCalculado',result(row).get('valorTotal')))),
]
compared(doc,rows,metrics)
grouped_notes(doc,rows,lambda row:complement(row,'justificativa_fc'),'Justificativa do fator de comercialização')
grouped_notes(doc,rows,lambda row:complement(row,'ajustes_nao_aplicados',''),'Ajustes expressamente não aplicados e justificativa')
for row in rows:
for warning in row.get('revisao_documental',{}).get('conferencias',[]):
add_body_text(doc,f"Conferência — {ref(row)}: {warning}")
return refs
def construction_groups(rows):
groups = OrderedDict()
for row in rows:
for index,c in enumerate(row.get('construcoes') or []):
key = construction_identity(source(row),c,index)
groups.setdefault(key,[]).append((row,c,index))
return groups
def construction_name(c,index):
return f"Construção {c.get('seq') or index+1}" + (f" — {c['nome']}" if c.get('nome') else '')
def cub_date(row):
construction = source(row).get('construcao') or {}
value = (construction.get('avaliacao') or {}).get('cubData') or construction.get('cubData')
if not value: return 'Não informada'
text = str(value)
return f'{text[5:7]}/{text[:4]}' if len(text) >= 7 and text[4] == '-' else text
def entry_scope(entries, groups, show_base=True):
"""Compact applicability while keeping changes and duplicate years explicit."""
selected = {(row['id'],i) for row,c,i in entries}
labels = []
for full in groups.values():
subset = [(row,c,i) for row,c,i in full if (row['id'],i) in selected]
if not subset: continue
label = construction_name(full[0][1],full[0][2])
if len(subset) != len(full): label += ' (' + '; '.join(ref(row, show_base=show_base) for row,c,i in subset) + ')'
labels.append(label)
return '; '.join(labels)
def construction_notes(doc, notes, groups):
for text,entries in notes.items():
add_body_text(doc,f"{text} Aplicável a {entry_scope(entries,groups)}.")
def construction_characteristics(doc, groups, num=None):
records = [['Construção / referência','Tipo / ano','Área real / de cálculo (m²)','Vida útil / residual R','Estado de conservação','H']]
record_notes = []
for entries in groups.values():
variants = OrderedDict()
for row,c,i in entries:
values = (f"{c.get('tipoConstrucao') or 'Não informado'} / {c.get('anoConstrucao') or 'Não informado'}",
f"{fmt(c.get('areaTotal'))} / {fmt(c.get('areaCalculada'))}",
f"{fmt(c.get('vidaUtil'),0)} anos / {pct(c.get('valorResidual'),True)}",
conservation_code(c.get('estadoConservacao')),fmt(c.get('coefHeidecke'),4))
notes = construction_depreciation_notes(row,c)
details = (notes.get('life',''),notes.get('state',''))
variants.setdefault((values,details),[]).append((row,c,i))
for (values,details),subset in variants.items():
records.append([entry_scope(subset,groups,show_base=False)]+list(values))
record_notes.append((len(records)-1,details))
if len(records)>1:
obj = table(doc,records,[1.8,1.6,1.2,1.1,.9,.65],auto_fit=False)
seen = set()
for index,details in record_notes:
for column,text in zip((3,4),details):
if text and text not in seen:
add_footnote(doc,obj.cell(index,column).paragraphs[0],text)
seen.add(text)
fit_table_to_page(doc,obj,notes=seen)
area_records = [['Construção / referência','Tipo de área','Área real (m²)','Coeficiente','Parcela equivalente (m²)']]
for entries in groups.values():
areas = OrderedDict()
for row,c,i in entries:
details = [{key:a.get(key) for key in ('descricao','area','coeficiente')} for a in c.get('areasEquivalentes') or []]
# A single area at coefficient 1 contributes no additional explanation.
if not details or (len(details)==1 and number(details[0].get('coeficiente'))==1
and number(c.get('areaTotal'))==number(c.get('areaCalculada'))==number(details[0].get('area'))): continue
areas.setdefault(json.dumps(details,sort_keys=True,ensure_ascii=False),[]).append((row,c,i))
for identity,subset in areas.items():
label = entry_scope(subset,groups,show_base=False)
area_records.extend([
[label,a.get('descricao') or 'Não informado',fmt(a.get('area')),fmt(a.get('coeficiente'),4),
fmt(number(a['area'])*number(a['coeficiente'])) if number(a.get('area')) is not None and number(a.get('coeficiente')) is not None else 'Não informado'] for a in json.loads(identity)])
if len(area_records)>1:
if num is not None:
heading(doc,num,'Composição das áreas',5)
else:
add_body_text(doc,'Composição das áreas').paragraph_format.keep_with_next = True
obj = table(doc,area_records,[2.1,2.8,1,1,1.4],auto_fit=False)
fit_table_to_page(doc,obj)
def construction_cub_notes(doc, groups):
entries_all = [entry for entries in groups.values() for entry in entries]
if not entries_all:
return
rows = list({row['id']:row for row,c,i in entries_all}.values())
labels = [ref(row,show_base=False) for row in rows]
records = [['Construção','Projeto CUB'] + labels]
composition_notes, siglas = [], OrderedDict()
for entries in groups.values():
by_id = {row['id']:c for row,c,i in entries}
variants = OrderedDict()
for row,c,i in entries:
if c.get('cubCompostoAtivo'):
composition = c.get('cubComposicaoCalculada') or c.get('cubComposicao') or [
{'cubTipo':c.get('cubTipo'),'peso':c.get('cubPesoPrincipalCalculado')},
{'cubTipo':c.get('cubTipoSecundario'),'peso':c.get('cubPesoSecundarioCalculado')}]
parts = [f"{item.get('cubTipo') or 'Projeto não informado'}: {pct(item['peso']) if number(item.get('peso')) is not None else 'peso não informado'}"
for item in composition if isinstance(item,dict)]
description = '; '.join(parts) or 'Composição não preservada no arquivo.'
project = 'CUB composto'
else:
project = str(c.get('cubTipo') or 'Não informado').split(' (',1)[0] + ' (100%)'
description = project
short = str(c.get('cubTipo') or '').split(' (',1)[0]
catalog = content_for(source(row))['referencias'].get('siglasCub',[])
found = next((item for item in catalog if item.get('sigla')==short),{})
if found:
siglas.setdefault(short,found['descricao'])
variants.setdefault((project,description),[]).append((row,c,i))
project_labels = list(dict.fromkeys(key[0] for key in variants))
label = project_labels[0] if len(project_labels)==1 else 'Varia por exercício'
records.append([construction_name(entries[0][1],entries[0][2]),label] + [currency(by_id[row['id']].get('valorCub')) if row['id'] in by_id else '—' for row in rows])
if any(key[0]=='CUB composto' for key in variants) or len(variants)>1:
details = []
for (project,description),subset in variants.items():
scope = '; '.join(ref(row,show_base=False) for row,c,i in subset) if len(variants)>1 else 'Todos os exercícios apresentados'
details.append(f'{scope}: {description}.')
composition_notes.append((len(records)-1,' '.join(details),any(key[0]=='CUB composto' for key in variants)))
obj = table(doc,records,[2.5,1.5]+[1.15]*len(rows),auto_fit=False)
for key,pattern in [('cubGroup',r'\bCUB\b'),('cubType',r'Projeto CUB'),
('cubValue',r'CUB histórico'),('cubPercent',r'%\s+(?:do\s+)?CUB')]:
variants = OrderedDict()
for row,c,i in entries_all:
text = content_for(source(row)).get('notas',{}).get(key)
if text:
variants.setdefault(text,[]).append(ref(row,show_base=False))
if variants:
text = ' '.join(text + (f" Aplicável a {'; '.join(dict.fromkeys(scope))}." if len(variants)>1 else '') for text,scope in variants.items())
if add_first_footnote(doc,pattern,text) is None:
add_footnote(doc,obj.cell(0,1).paragraphs[0],text)
used_notes = []
explained_composite = False
for index,text,has_composite in composition_notes:
if has_composite and not explained_composite:
text = 'O CUB composto corresponde à média ponderada dos projetos: CUB = Σ(CUBᵢ × pesoᵢ) / Σ(pesoᵢ). ' + text
explained_composite = True
add_footnote(doc,obj.cell(index,1).paragraphs[0],text)
used_notes.append(text)
for short,description in siglas.items():
add_first_footnote(doc,r'\b'+re.escape(short)+r'\b',f'{short}: {description}.')
order_footnotes(doc)
fit_table_to_page(doc,obj,notes=used_notes)
def construction_bdi_notes(doc, groups, num=None):
for entries in groups.values():
name = construction_name(entries[0][1],entries[0][2])
if num is not None:
heading(doc,num,name,5)
else:
add_subsubsubsubsection_title(doc,'',name)
reasons,references = OrderedDict(),OrderedDict()
components,by_id,variants = OrderedDict(),{},set()
for row,c,i in entries:
key = str(c.get('id') or c.get('seq') or i+1)
note = (row.get('complementos') or {}).get('bdi',{}).get(key)
if note and note.strip(): reasons.setdefault(note.strip(),[]).append(row)
reference = next((item for item in content_for(source(row))['referencias'].get('bdiTiposConstrucao',[]) if item.get('tipo')==c.get('tipoConstrucao')),None)
if reference:
text = f"Referência técnica para {reference['tipo']}: BDI {reference['bdi']}. O percentual adotado é o registrado nesta tabela."
references.setdefault(text,[]).append(row)
values,counts = OrderedDict(),{}
composition = c.get('bdiComposicaoAdotada') or []
for item in composition if isinstance(composition,list) else []:
if not isinstance(item,dict): continue
label = str(item.get('descricao') or 'Componente não informado')
counts[label] = counts.get(label,0)+1
identity = (label,counts[label])
components.setdefault(identity,label)
values[identity] = number(item.get('percentual'))
total = sum(values.values(),Decimal(0)) if values and all(v is not None for v in values.values()) else None
adopted = number(c.get('bdi'))
by_id[row['id']] = {'components':values,'sum':total,'adopted':adopted}
variants.add((tuple(values.items()),adopted))
for text,applicable in reasons.items():
scope = ' ('+'; '.join(ref(row,show_base=False) for row in applicable)+')' if len(applicable)!=len(entries) else ''
add_body_text(doc,f'Justificativa do BDI{scope}: {text}').paragraph_format.keep_with_next = True
metrics = [(label,lambda row,k=key: pct(by_id[row['id']]['components'][k]) if k in by_id[row['id']]['components'] else '—') for key,label in components.items()]
differs = any(value['sum'] is not None and (value['adopted'] is None or
value['sum'].quantize(Decimal('.01')) != value['adopted'].quantize(Decimal('.01'))) for value in by_id.values())
if differs:
metrics.append(('Soma dos componentes',lambda row:pct(by_id[row['id']]['sum'])))
metrics.append(('Total do BDI',lambda row:pct(by_id[row['id']]['adopted'])))
used_notes = []
for text,applicable in references.items():
scope = '; '.join(ref(row,show_base=False) for row in applicable)+': ' if len(applicable)!=len(entries) else ''
used_notes.append(scope+text)
for row,c,i in entries:
value = by_id[row['id']]
text = ''
if not value['components']:
text = 'O arquivo registra o percentual global do BDI sem discriminar sua composição.'
elif value['sum'] is None:
text = 'A composição registrada contém percentuais não informados; o total do BDI é o percentual global adotado no arquivo.'
elif value['adopted'] is None or value['sum'].quantize(Decimal('.01')) != value['adopted'].quantize(Decimal('.01')):
text = 'A soma dos componentes difere do percentual global registrado. O total do BDI preserva o percentual adotado no arquivo.'
if text:
scope = ref(row,show_base=False)+': ' if len(variants)>1 else ''
if scope+text not in used_notes: used_notes.append(scope+text)
if len(variants)>1:
obj = compared(doc,[row for row,c,i in entries],metrics)
else:
row = entries[0][0]
obj = table(doc,[['Componente de BDI adotado','Percentual']]+[[label,getter(row)] for label,getter in metrics],[4.5,1.5],auto_fit=False)
for cell in obj.rows[-1].cells:
for paragraph in cell.paragraphs:
for run in paragraph.runs: run.bold = True
if used_notes:
add_footnote(doc,obj.rows[-1].cells[0].paragraphs[0],' '.join(used_notes))
fit_table_to_page(doc,obj,notes=used_notes)
def conservation_code(value):
text = str(value or '').strip()
match = re.match(r'^([A-I])(?:\s*[-–—]|$)',text,re.I)
return match[1].upper() if match else text or 'Não informado'
def construction_depreciation_notes(row,c):
from backend.app.evo_documentation import public_reason
notes = {}
saved = ((source(row).get('construcao') or {}).get('avaliacao') or {}).get('referenciasCalculo') or {}
fallback = content_for(source(row))['referencias']
life = next((item for item in saved.get('vidaUtilTipos',fallback.get('vidaUtilTipos',[])) if item.get('tipo')==c.get('tipoConstrucao')),None)
if life:
notes['life'] = f"Referência de vida útil para {life['tipo']}: {life.get('vidaUtil','não informada')} anos. A vida consumida considera a idade no ano de referência, limitada a 100%, ou o ajuste manual registrado. R é o valor residual, expresso em percentual no quadro."
raw = str(c.get('estadoConservacao') or '').strip()
state = next((item for item in saved.get('estadosConservacao',fallback.get('estadosConservacao',[])) if str(item.get('estado','')).strip().casefold()==raw.casefold()),None)
if raw:
notes['state'] = f"{raw}: {state['descricao']}" if state and state.get('descricao') else f"Estado de conservação registrado: {raw}. A descrição técnica desta classe não foi preservada no arquivo."
notes['state'] += ' H é o coeficiente de conservação adotado no cálculo de Ross-Heidecke; o valor registrado consta no quadro.'
if c.get('percentualVidaUtilManualAtivo') or c.get('percentualVidaUtilAjusteManual'):
reason = public_reason(c,'percentualVidaUtilJustificativa')
notes['manual'] = f"Ajuste manual de vida consumida: padrão {pct(c.get('percentualVidaUtilPadrao'))}; adotado {pct(c.get('percentualVidaUtil'))}. " + (reason or 'A justificativa pública não está disponível no arquivo.')
return notes
def reducer_metrics(rows,territorial):
keys = OrderedDict((key,label) for row in rows for key,label,rate in applied_reducers(row,territorial))
def display(row,key):
active = {k:rate for k,label,rate in applied_reducers(row,territorial)}
return pct(active[key],True) if key in active else '—'
return [(label+' — redução efetiva',lambda row,k=key:display(row,k)) for key,label in keys.items()]
def condition_description(row,territorial):
active = {key:rate for key,label,rate in applied_reducers(row,territorial)}
if not active: return ''
config = ((terrain(row).get('avaliacao') or {}).get('ajuste_territorial') or {}).get('configuracao') or {}
saved = source(row).get('catalogoAjustes') or {}
lines = []
if 'redutor88I' in active:
catalog = saved.get('art_88_i') or {}
minimum = number(catalog.get('limite_minimo_area'))
if minimum is not None:
lines.append(f"Inciso I: participam as condições com área atingida superior a {pct(minimum,True)}; cada contribuição corresponde à área relativa multiplicada pela redução de referência.")
for item in catalog.get('condicoes') or []:
area = number(((config.get('art_88_i') or {}).get('condicoes',{}).get(item.get('codigo')) or {}).get('percentual_area'))
rate = number(item.get('redutor',item.get('percentual_reducao')))
if area is not None and area>minimum and rate is not None and rate>0:
lines.append(f"{item.get('rotulo') or item['codigo']}: área {pct(area,True)} × redução de referência {pct(rate,True)} = contribuição {pct(area*rate,True)}.")
for key,group in [('redutor6061','art_60_61'),('redutor88II','art_88_ii')]:
if key not in active: continue
selected = (config.get(group) or {}).get('selecionados') or []
choices = [item for item in (saved.get(group) or {}).get('condicoes',[]) if item.get('codigo') in selected]
if key=='redutor6061':
choices = [item for item in choices if number(item.get('redutor'))==active[key]]
lines.append('Arts. 60 e 61: aplica-se somente a maior redução selecionada.')
if choices:
from .territorial_narrative import legal_reference, ARTICLE_88_DECREE
if key == 'redutor88II':
lines.append('Condições de incidência: '+ '; '.join(
f"{item.get('rotulo') or item['codigo']} ({legal_reference(key, [item['codigo']])})" for item in choices)+'.')
lines.append(f'Os dispositivos referidos integram o {ARTICLE_88_DECREE}.')
else:
lines.append('Condição de incidência: '+ '; '.join(item.get('rotulo') or item['codigo'] for item in choices)+'.')
if 'redutorTopografia' in active:
for part,rate_key in [('relevo','redutorRelevo'),('nivel_rua','redutorNivel')]:
code = (config.get('topografia') or {}).get(part)
item = next((item for item in (saved.get('topografia') or {}).get(part,[]) if item.get('codigo')==code),{})
rate = number(result(row).get(rate_key)) if rate_key in result(row) else number(item.get('redutor'))
if item and rate is not None and rate>0: lines.append(f"{item.get('rotulo') or code}: {pct(rate,True)}.")
if 'redutorAgua' in active:
code = (config.get('agua') or {}).get('situacao')
item = next((item for item in (saved.get('agua') or {}).get('condicoes',[]) if item.get('codigo')==code),{})
if item: lines.append((item.get('rotulo') or code)+'.')
if not saved: lines.append('Catálogo histórico de ajustes ausente; a redução efetiva foi transcrita do resultado salvo, sem atribuir condições presumidas.')
return ' '.join(lines)
def gerar_ajustes(doc,rows,territorial):
active_rows = [row for row in rows if applied_reducers(row,territorial)]
descriptions = OrderedDict()
for row in active_rows:
terms = content_for(source(row)).get('redutores_termos',{})
for key,label,rate in applied_reducers(row,territorial):
desc = terms.get(REDUCERS.get(key,('',''))[1])
if desc: descriptions.setdefault(label+': '+desc+'.',[]).append(row)
for text,applicable in descriptions.items():
add_body_text(doc,text + (f" Aplicável a {'; '.join(ref(row) for row in applicable)}." if len(applicable)!=len(rows) else ''))
if territorial and active_rows:
add_body_text(doc,'Os fatores dos redutores territoriais efetivamente aplicados incidem de forma multiplicativa, antes da escolha no campo de arbítrio. O quadro apresenta os percentuais de redução e os resultados salvos.')
grouped_notes(doc,rows,lambda row:condition_description(row,territorial),'Condições dos redutores aplicados')
if territorial:
apan_rows = [row for row in rows if (number(result(row).get('redutorApan')) or 0)>0
and number(result(row).get('fatorCaFinal')) not in (None,1)]
if apan_rows:
add_body_text(doc,'A referência de APAN participa da decisão no campo de arbítrio; não constitui desconto territorial autônomo e não é aplicada novamente.')
grouped_notes(doc,apan_rows,lambda row:f"Área atingida: {pct(((terrain(row).get('avaliacao') or {}).get('ajuste_territorial') or {}).get('configuracao',{}).get('apan',{}).get('percentual_area'),True)}; referência de APAN: {pct(result(row).get('redutorApan'),True)}; fator final adotado: {fmt(result(row).get('fatorCaFinal'),4)}.",'APAN considerada no campo de arbítrio',always_refs=True)
else:
limited = [row for row in rows if result(row).get('limiteArt88Aplicado')]
if limited:
add_body_text(doc,'O limite conjunto dos incisos I e II do art. 88, previsto em seu § 3º, foi efetivamente acionado nas referências abaixo. O percentual do inciso II exibido na composição já considera o limite; os valores de verificação são transcritos da avaliação.')
def cap_note(row):
catalog = source(row).get('catalogoAjustes') or {}
limit = number(catalog.get('limite_reducao_art_88',(catalog.get('art_88_ii') or {}).get('limite_combinado_art_88')))
if limit is None or not 0 <= limit <= 1:
return 'Percentual do limite não preservado no catálogo histórico; mantidos os valores de verificação salvos.'
return f'Limite conjunto de redução: {pct(limit,True)}; piso correspondente: {pct(1-limit,True)} do valor de referência.'
grouped_notes(doc,limited,cap_note,'Limite do art. 88',always_refs=True)
compared(doc,limited,[
('Referência do art. 88',lambda row:currency(result(row).get('valorReferenciaArt88'))),
('Resultado após inciso I',lambda row:currency(result(row).get('valorAposIncisoIArt88'))),
('Resultado bruto dos incisos',lambda row:currency(result(row).get('valorAposArt88Bruto'))),
('Resultado após limite do art. 88',lambda row:currency(result(row).get('valorAposArt88Aplicado'))),
])
def gerar_especificacao_evolutivo(doc,imported,refs,num=None):
rows = imported['avaliacoes']
foundations = {row['id']:global_foundation(source(row)) for row in rows}
if doc.paragraphs: doc.paragraphs[-1].paragraph_format.keep_with_next = True
table(doc,[['IPTU','Terreno','Construções','Global','Precisão / amplitude do terreno']]+[
[ref(row,show_base=False),foundations[row['id']]['terrain']['grade'],foundations[row['id']]['construction']['grade'],foundations[row['id']]['grade'],str(row['avaliacao'].get('precisao') or 'Indisponível') + (' / ' + pct(row['avaliacao']['amplitude']) if row['avaliacao'].get('amplitude') is not None else '')] for row in rows],[1.4,1,1,1,1.4],font_size=7)
add_body_text(doc,f"A fundamentação do terreno corresponde à demonstração do item {refs['terreno']}; a das construções, ao item {refs['construcoes']}; e a composição global, ao item {refs['composicao']}. A precisão e a amplitude relativa do intervalo de confiança dizem respeito à estimativa territorial por regressão.")
content = content_for(source(rows[0]))
for section,key in [('terreno','terrain'),('construcoes','construction'),('global',None)]:
norm = content['normas'][section]
if num is not None:
heading(doc,num,norm['title'],2)
else:
add_subsubsection_title(doc,'',norm['title']).paragraph_format.keep_with_next = True
add_body_text(doc,norm['source'] + ' ' + norm['framing']).paragraph_format.keep_with_next = True
first = foundations[rows[0]['id']][key] if key else foundations[rows[0]['id']]
compared(doc,rows,[(item['label'],lambda row,i=index,k=key: (foundations[row['id']][k] if k else foundations[row['id']])['items'][i]['grade'] + ' / ' + str((foundations[row['id']][k] if k else foundations[row['id']])['items'][i]['points'] if (foundations[row['id']][k] if k else foundations[row['id']])['items'][i]['points'] is not None else '—') + ' ponto(s)') for index,item in enumerate(first['items'])])
grouped_notes(doc,rows,lambda row,k=key: ' '.join(f"{item['label']}: {item['detail']}" for item in (foundations[row['id']][k] if k else foundations[row['id']])['items']),'Evidências e critérios')
for row in rows:
f = foundations[row['id']]['terrain']
if f.get('missingRequirements'): add_body_text(doc,f"{ref(row)} — o Grau III do terreno depende da declaração dos requisitos adicionais; enquadramento pendente.")