Spaces:

gui-sparim
/

base_conhecimento

Sleeping

base_conhecimento / scripts /import_mqc_examples.py

Guilherme Silberfarb Costa

Simplifica acervo e revisão posterior com autoria documental e leitura limpa

ec5864a 10 days ago

12.6 kB

"""Import the captured MESA MQC tables through CARLA's catalog service.
Run explicitly with --author <existing username>. Existing entries are preserved.
"""

import argparse
import json
import sys
from datetime import datetime
from html import escape
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / "backend"))


def table(headers, rows):
    head = "".join(f"<th>{escape(str(value))}</th>" for value in headers)
    body = "".join(
        "<tr>" + "".join(f"<td>{escape(str(value))}</td>" for value in row) + "</tr>"
        for row in rows
    )
    return f"<table><thead><tr>{head}</tr></thead><tbody>{body}</tbody></table>"


def number(value, decimals=0):
    return (
        f"{value:,.{decimals}f}".replace(",", "_").replace(".", ",").replace("_", ".")
    )


def examples(tables):
    return [
        {
            "code": "MESA-MQC-RH",
            "title": "Ross-Heidecke — estados de conservação e coeficientes",
            "summary": "Os nove estados de conservação da tabela R&H do MQC, com coeficientes de Heidecke e descrições usadas pela MESA.",
            "subjects": ["MQC", "Depreciação", "Benfeitorias"],
            "body": "<h2>Estados de conservação</h2><p>Na aba R&H, a MESA apresenta os coeficientes em percentual. Esta tabela mantém essa apresentação.</p>"
            + table(
                ["Estado", "Coeficiente de Heidecke (%)", "Descrição"],
                [
                    [
                        row["estado"],
                        number(row["coeficiente"] * 100, 2) + "%",
                        row.get("descricao", "—"),
                    ]
                    for row in tables["estadosConservacao"]
                ],
            ),
        },
        {
            "code": "MESA-MQC-VIDA-UTIL",
            "title": "Vida útil — referências por tipo de construção",
            "summary": "Tabela de vida útil da MESA para 14 tipos de construção, incluindo apartamentos, casas, lojas, armazéns e galpões.",
            "subjects": ["MQC", "Depreciação", "Benfeitorias"],
            "body": "<h2>Vida útil por tipo construtivo</h2>"
            + table(
                ["Tipo de construção", "Vida útil (anos)"],
                [[row["tipo"], row["vidaUtil"]] for row in tables["vidaUtilTipos"]],
            ),
        },
        {
            "code": "MESA-MQC-AREA-EQUIVALENTE",
            "title": "Áreas equivalentes — coeficientes de referência do MQC",
            "summary": "Quinze coeficientes e intervalos para áreas equivalentes, como garagens, varandas, terraços e áreas sem acabamento. O MQC cita o item 5 da NBR 12.721/2006.",
            "subjects": ["MQC", "Benfeitorias", "Normas técnicas"],
            "body": "<h2>Coeficientes para áreas equivalentes</h2><p>A nota exibida na MESA é: “Critérios para determinação e cálculo de áreas — item 5 da NBR 12.721/2006.” Esta entrada reproduz a tabela auxiliar do MQC e não contém o texto integral da norma.</p>"
            + table(
                ["Item", "Área", "Coeficiente médio"],
                [
                    [row["item"], row["descricao"], row["coeficiente"]]
                    for row in tables["areaEquivalenteCoeficientes"]
                ],
            ),
        },
        {
            "code": "MESA-MQC-BDI",
            "title": "BDI — adicionais e faixas por tipo de construção",
            "summary": "As duas tabelas de BDI do MQC: seis componentes adicionais e 17 referências por tipo construtivo, preservando as faixas exibidas pela MESA.",
            "subjects": ["MQC", "Benfeitorias", "Fundamentação"],
            "body": "<h2>Adicionais de BDI</h2>"
            + table(
                [
                    "Item",
                    "Descrição",
                    "Percentual 1",
                    "Percentual 2",
                    "Percentual 3",
                    "Escolha",
                ],
                [
                    [row["item"], row["descricao"]]
                    + [
                        number(row[key] * 100, 1) + "%" if row[key] is not None else "—"
                        for key in ["perc1", "perc2", "perc3", "escolha"]
                    ]
                    for row in tables["bdiAdicionais"]
                ],
            )
            + "<h2>BDI por tipo de construção</h2>"
            + table(
                ["Tipo de construção", "BDI"],
                [[row["tipo"], row["bdi"]] for row in tables["bdiTiposConstrucao"]],
            ),
        },
        {
            "code": "MESA-MQC-VALOR-RESIDUAL",
            "title": "Valor residual — referências por classe, tipo e padrão",
            "summary": "Tabela com 29 referências de valor residual e vida referencial para construções residenciais e comerciais, conforme a aba Valor Residual do MQC.",
            "subjects": ["MQC", "Depreciação", "Benfeitorias"],
            "body": "<h2>Vida referencial e valor residual</h2><p>A coluna de vida referencial desta tabela é preservada separadamente da tabela Vida útil do MQC.</p>"
            + table(
                [
                    "Classe",
                    "Tipo",
                    "Padrão",
                    "Vida referencial (anos)",
                    "Valor residual (%)",
                ],
                [
                    [
                        row["classe"],
                        row["tipo"],
                        row["padrao"],
                        row["vidaReferencialIr"],
                        str(row["valorResidual"]) + "%",
                    ]
                    for row in tables["valorResidualReferencias"]
                ],
            ),
        },
        {
            "code": "MESA-MQC-SIGLAS-CUB",
            "title": "CUB — siglas e características dos projetos padrão",
            "summary": "Dez siglas da tabela SIGLAS do MQC, com descrição, dormitórios, área real e área equivalente: R1-B, R1-N, R1-A, RP1Q, PIS, PP4-B, PP4-N, R8-N, CAL8-N e GI.",
            "subjects": ["MQC", "Benfeitorias"],
            "body": "<h2>Projetos padrão do CUB</h2>"
            + table(
                [
                    "Sigla",
                    "Descrição",
                    "Dormitórios",
                    "Área real (m²)",
                    "Área equivalente (m²)",
                ],
                [
                    [
                        row["sigla"],
                        row["descricao"],
                        row["dormitorios"],
                        number(row["areaReal"], 2),
                        number(row["areaEquivalente"], 2),
                    ]
                    for row in tables["siglasCub"]
                ],
            ),
        },
    ]


def origin(capture):
    captured = datetime.fromisoformat(capture["captured_at"])
    return (
        "<h2>Origem</h2><p>Importado das tabelas auxiliares da aba MQC da MESA em "
        + captured.strftime("%d/%m/%Y")
        + ". Os valores e descrições registram o conteúdo disponível na MESA nessa data.</p>"
    )


def example_content(record, capture, subject_ids):
    return {
        "title": record["title"],
        "type": "reference",
        "summary": record["summary"],
        "content_html": record["body"] + origin(capture),
        "subject_ids": subject_ids,
    }


def simplify_snapshot(snapshot, capture):
    # Only captured examples are passed here; never discard later attachments.
    if any(snapshot.get(key) for key in ("file_ids", "links", "relations")):
        raise ValueError("Esta referência recebeu anexos ou conexões; revise antes de simplificar.")
    content = {key: snapshot[key] for key in ("title", "type", "summary", "content_html", "subject_ids")}
    if "Importado das tabelas auxiliares da aba MQC da MESA em " not in content["content_html"]:
        content["content_html"] += origin(capture)
    return content


def known_examples(db):
    from sqlalchemy import select
    from app.models import Audit, Revision, Entry
    # Old immutable revisions retain the import code even after current fields change.
    known = {
        rev.snapshot["metadata"]["code"]: rev.entry_id
        for rev in db.scalars(select(Revision).order_by(Revision.number))
        if rev.snapshot.get("metadata", {}).get("code", "").startswith("MESA-MQC-")
    }
    for event in db.scalars(select(Audit).where(Audit.action == "mqc_examples_imported")):
        known.update(event.details.get("references", {}))
    capture = json.loads((ROOT / "scripts/data/mqc_reference_examples.json").read_text())
    titles = {record["title"]: record["code"] for record in examples(capture["tables"])}
    for entry in db.scalars(select(Entry)):
        revision = db.scalar(
            select(Revision).where(Revision.entry_id == entry.id, Revision.number == entry.latest)
        )
        if revision and revision.snapshot["title"] in titles:
            known.setdefault(titles[revision.snapshot["title"]], entry.id)
    return known


def run(author, simplify_existing=False):
    from sqlalchemy import select
    from app import catalog, schemas
    from app.database import SessionLocal
    from app.models import Entry, Subject, User
    from app.security import audit

    capture = json.loads((ROOT / "scripts/data/mqc_reference_examples.json").read_text())
    records = examples(capture["tables"])
    with SessionLocal.begin() as db:
        user = db.scalar(select(User).where(User.username == author, User.active == True))
        if not user or not catalog.curator(user):
            raise ValueError("Informe um usuário existente e ativo com perfil de administrador ou curador.")
        subjects = {
            s.name: s.id for s in db.scalars(
                select(Subject).where(Subject.active == True, Subject.approved == True)
            )
        }
        missing = {name for record in records for name in record["subjects"] if name not in subjects}
        if missing:
            raise ValueError(f"Assuntos aprovados ausentes: {', '.join(sorted(missing))}")
        existing = known_examples(db)
        imported, updated, skipped, references = [], [], [], {}
        for record in records:
            code = record["code"]
            entry = db.get(Entry, existing[code]) if code in existing else None
            if entry:
                references[code] = entry.id
                snapshot = catalog.revision(db, entry, entry.latest).snapshot
                extras = any(snapshot.get(key) for key in ("source_author", "source_name", "usage", "metadata")) or snapshot.get("source_state", "unchecked") != "unchecked"
                if not simplify_existing or not extras:
                    skipped.append(entry.id)
                    continue
                content = simplify_snapshot(snapshot, capture)
                reason = "Adequação do exemplo aos cinco campos do cadastro simplificado."
            else:
                content = example_content(record, capture, [subjects[name] for name in record["subjects"]])
                reason = "Importação das referências do MQC da MESA, por solicitação de Guilherme."
            payload = schemas.SaveInput(
                content=schemas.Content(**content),
                expected_version=entry.lock_version if entry else 0,
                reason=reason,
                action="draft" if entry else "publish",
            )
            result = catalog.save(db, user, payload, entry.id if entry else None)
            references[code] = result["id"]
            (updated if entry else imported).append(result["id"])
        if imported or updated:
            audit(
                db, user, "mqc_examples_imported", entry_ids=imported,
                updated_entry_ids=updated, references=references,
                source_file=capture["source_file"], source_sha256=capture["source_sha256"],
                captured_at=capture["captured_at"],
            )
        result = {"created": imported, "updated": updated, "preserved": skipped}
        print(json.dumps(result, ensure_ascii=False))
        return result


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--author", required=True, help="Usuário existente responsável pelo cadastro.")
    parser.add_argument("--simplify-existing", action="store_true", help="Revisar exemplos antigos preservando tabelas, autoria e histórico.")
    args = parser.parse_args()
    run(args.author, args.simplify_existing)

	"""Import the captured MESA MQC tables through CARLA's catalog service.

	Run explicitly with --author <existing username>. Existing entries are preserved.
	"""

	import argparse
	import json
	import sys
	from datetime import datetime
	from html import escape
	from pathlib import Path

	ROOT = Path(__file__).resolve().parents[1]
	sys.path.insert(0, str(ROOT / "backend"))


	def table(headers, rows):
	head = "".join(f"<th>{escape(str(value))}</th>" for value in headers)
	body = "".join(
	"<tr>" + "".join(f"<td>{escape(str(value))}</td>" for value in row) + "</tr>"
	for row in rows
	)
	return f"<table><thead><tr>{head}</tr></thead><tbody>{body}</tbody></table>"


	def number(value, decimals=0):
	return (
	f"{value:,.{decimals}f}".replace(",", "_").replace(".", ",").replace("_", ".")
	)


	def examples(tables):
	return [
	{
	"code": "MESA-MQC-RH",
	"title": "Ross-Heidecke — estados de conservação e coeficientes",
	"summary": "Os nove estados de conservação da tabela R&H do MQC, com coeficientes de Heidecke e descrições usadas pela MESA.",
	"subjects": ["MQC", "Depreciação", "Benfeitorias"],
	"body": "<h2>Estados de conservação</h2><p>Na aba R&H, a MESA apresenta os coeficientes em percentual. Esta tabela mantém essa apresentação.</p>"
	+ table(
	["Estado", "Coeficiente de Heidecke (%)", "Descrição"],
	[
	[
	row["estado"],
	number(row["coeficiente"] * 100, 2) + "%",
	row.get("descricao", "—"),
	]
	for row in tables["estadosConservacao"]
	],
	),
	},
	{
	"code": "MESA-MQC-VIDA-UTIL",
	"title": "Vida útil — referências por tipo de construção",
	"summary": "Tabela de vida útil da MESA para 14 tipos de construção, incluindo apartamentos, casas, lojas, armazéns e galpões.",
	"subjects": ["MQC", "Depreciação", "Benfeitorias"],
	"body": "<h2>Vida útil por tipo construtivo</h2>"
	+ table(
	["Tipo de construção", "Vida útil (anos)"],
	[[row["tipo"], row["vidaUtil"]] for row in tables["vidaUtilTipos"]],
	),
	},
	{
	"code": "MESA-MQC-AREA-EQUIVALENTE",
	"title": "Áreas equivalentes — coeficientes de referência do MQC",
	"summary": "Quinze coeficientes e intervalos para áreas equivalentes, como garagens, varandas, terraços e áreas sem acabamento. O MQC cita o item 5 da NBR 12.721/2006.",
	"subjects": ["MQC", "Benfeitorias", "Normas técnicas"],
	"body": "<h2>Coeficientes para áreas equivalentes</h2><p>A nota exibida na MESA é: “Critérios para determinação e cálculo de áreas — item 5 da NBR 12.721/2006.” Esta entrada reproduz a tabela auxiliar do MQC e não contém o texto integral da norma.</p>"
	+ table(
	["Item", "Área", "Coeficiente médio"],
	[
	[row["item"], row["descricao"], row["coeficiente"]]
	for row in tables["areaEquivalenteCoeficientes"]
	],
	),
	},
	{
	"code": "MESA-MQC-BDI",
	"title": "BDI — adicionais e faixas por tipo de construção",
	"summary": "As duas tabelas de BDI do MQC: seis componentes adicionais e 17 referências por tipo construtivo, preservando as faixas exibidas pela MESA.",
	"subjects": ["MQC", "Benfeitorias", "Fundamentação"],
	"body": "<h2>Adicionais de BDI</h2>"
	+ table(
	[
	"Item",
	"Descrição",
	"Percentual 1",
	"Percentual 2",
	"Percentual 3",
	"Escolha",
	],
	[
	[row["item"], row["descricao"]]
	+ [
	number(row[key] * 100, 1) + "%" if row[key] is not None else "—"
	for key in ["perc1", "perc2", "perc3", "escolha"]
	]
	for row in tables["bdiAdicionais"]
	],
	)
	+ "<h2>BDI por tipo de construção</h2>"
	+ table(
	["Tipo de construção", "BDI"],
	[[row["tipo"], row["bdi"]] for row in tables["bdiTiposConstrucao"]],
	),
	},
	{
	"code": "MESA-MQC-VALOR-RESIDUAL",
	"title": "Valor residual — referências por classe, tipo e padrão",
	"summary": "Tabela com 29 referências de valor residual e vida referencial para construções residenciais e comerciais, conforme a aba Valor Residual do MQC.",
	"subjects": ["MQC", "Depreciação", "Benfeitorias"],
	"body": "<h2>Vida referencial e valor residual</h2><p>A coluna de vida referencial desta tabela é preservada separadamente da tabela Vida útil do MQC.</p>"
	+ table(
	[
	"Classe",
	"Tipo",
	"Padrão",
	"Vida referencial (anos)",
	"Valor residual (%)",
	],
	[
	[
	row["classe"],
	row["tipo"],
	row["padrao"],
	row["vidaReferencialIr"],
	str(row["valorResidual"]) + "%",
	]
	for row in tables["valorResidualReferencias"]
	],
	),
	},
	{
	"code": "MESA-MQC-SIGLAS-CUB",
	"title": "CUB — siglas e características dos projetos padrão",
	"summary": "Dez siglas da tabela SIGLAS do MQC, com descrição, dormitórios, área real e área equivalente: R1-B, R1-N, R1-A, RP1Q, PIS, PP4-B, PP4-N, R8-N, CAL8-N e GI.",
	"subjects": ["MQC", "Benfeitorias"],
	"body": "<h2>Projetos padrão do CUB</h2>"
	+ table(
	[
	"Sigla",
	"Descrição",
	"Dormitórios",
	"Área real (m²)",
	"Área equivalente (m²)",
	],
	[
	[
	row["sigla"],
	row["descricao"],
	row["dormitorios"],
	number(row["areaReal"], 2),
	number(row["areaEquivalente"], 2),
	]
	for row in tables["siglasCub"]
	],
	),
	},
	]


	def origin(capture):
	captured = datetime.fromisoformat(capture["captured_at"])
	return (
	"<h2>Origem</h2><p>Importado das tabelas auxiliares da aba MQC da MESA em "
	+ captured.strftime("%d/%m/%Y")
	+ ". Os valores e descrições registram o conteúdo disponível na MESA nessa data.</p>"
	)


	def example_content(record, capture, subject_ids):
	return {
	"title": record["title"],
	"type": "reference",
	"summary": record["summary"],
	"content_html": record["body"] + origin(capture),
	"subject_ids": subject_ids,
	}


	def simplify_snapshot(snapshot, capture):
	# Only captured examples are passed here; never discard later attachments.
	if any(snapshot.get(key) for key in ("file_ids", "links", "relations")):
	raise ValueError("Esta referência recebeu anexos ou conexões; revise antes de simplificar.")
	content = {key: snapshot[key] for key in ("title", "type", "summary", "content_html", "subject_ids")}
	if "Importado das tabelas auxiliares da aba MQC da MESA em " not in content["content_html"]:
	content["content_html"] += origin(capture)
	return content


	def known_examples(db):
	from sqlalchemy import select
	from app.models import Audit, Revision, Entry
	# Old immutable revisions retain the import code even after current fields change.
	known = {
	rev.snapshot["metadata"]["code"]: rev.entry_id
	for rev in db.scalars(select(Revision).order_by(Revision.number))
	if rev.snapshot.get("metadata", {}).get("code", "").startswith("MESA-MQC-")
	}
	for event in db.scalars(select(Audit).where(Audit.action == "mqc_examples_imported")):
	known.update(event.details.get("references", {}))
	capture = json.loads((ROOT / "scripts/data/mqc_reference_examples.json").read_text())
	titles = {record["title"]: record["code"] for record in examples(capture["tables"])}
	for entry in db.scalars(select(Entry)):
	revision = db.scalar(
	select(Revision).where(Revision.entry_id == entry.id, Revision.number == entry.latest)
	)
	if revision and revision.snapshot["title"] in titles:
	known.setdefault(titles[revision.snapshot["title"]], entry.id)
	return known


	def run(author, simplify_existing=False):
	from sqlalchemy import select
	from app import catalog, schemas
	from app.database import SessionLocal
	from app.models import Entry, Subject, User
	from app.security import audit

	capture = json.loads((ROOT / "scripts/data/mqc_reference_examples.json").read_text())
	records = examples(capture["tables"])
	with SessionLocal.begin() as db:
	user = db.scalar(select(User).where(User.username == author, User.active == True))
	if not user or not catalog.curator(user):
	raise ValueError("Informe um usuário existente e ativo com perfil de administrador ou curador.")
	subjects = {
	s.name: s.id for s in db.scalars(
	select(Subject).where(Subject.active == True, Subject.approved == True)
	)
	}
	missing = {name for record in records for name in record["subjects"] if name not in subjects}
	if missing:
	raise ValueError(f"Assuntos aprovados ausentes: {', '.join(sorted(missing))}")
	existing = known_examples(db)
	imported, updated, skipped, references = [], [], [], {}
	for record in records:
	code = record["code"]
	entry = db.get(Entry, existing[code]) if code in existing else None
	if entry:
	references[code] = entry.id
	snapshot = catalog.revision(db, entry, entry.latest).snapshot
	extras = any(snapshot.get(key) for key in ("source_author", "source_name", "usage", "metadata")) or snapshot.get("source_state", "unchecked") != "unchecked"
	if not simplify_existing or not extras:
	skipped.append(entry.id)
	continue
	content = simplify_snapshot(snapshot, capture)
	reason = "Adequação do exemplo aos cinco campos do cadastro simplificado."
	else:
	content = example_content(record, capture, [subjects[name] for name in record["subjects"]])
	reason = "Importação das referências do MQC da MESA, por solicitação de Guilherme."
	payload = schemas.SaveInput(
	content=schemas.Content(**content),
	expected_version=entry.lock_version if entry else 0,
	reason=reason,
	action="draft" if entry else "publish",
	)
	result = catalog.save(db, user, payload, entry.id if entry else None)
	references[code] = result["id"]
	(updated if entry else imported).append(result["id"])
	if imported or updated:
	audit(
	db, user, "mqc_examples_imported", entry_ids=imported,
	updated_entry_ids=updated, references=references,
	source_file=capture["source_file"], source_sha256=capture["source_sha256"],
	captured_at=capture["captured_at"],
	)
	result = {"created": imported, "updated": updated, "preserved": skipped}
	print(json.dumps(result, ensure_ascii=False))
	return result


	if __name__ == "__main__":
	parser = argparse.ArgumentParser(description=__doc__)
	parser.add_argument("--author", required=True, help="Usuário existente responsável pelo cadastro.")
	parser.add_argument("--simplify-existing", action="store_true", help="Revisar exemplos antigos preservando tabelas, autoria e histórico.")
	args = parser.parse_args()
	run(args.author, args.simplify_existing)