gustavosantos/GTI_GERENCIADOR_FISCAL
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
8d6122c096026ecfb9b22fbd5beb1bb8eb53e296
GTI_GERENCIADOR_FISCAL/central/app/models.py
T
gtiandClaude Sonnet 5 8d6122c096 feat: Agent distribution with secure enrollment from the control panel 
One universal Windows Agent package per version (agent_releases: version/
platform/architecture/sha256/status DRAFT|PUBLISHED|REVOKED; the binary
lives on a local release-storage volume, never in PostgreSQL). An operator
creates a pending installation (agent_enrollments) from Agents / Servidores
-> + Nova instalação, then either downloads it directly or generates a
one-time HTTPS link (short TTL, single use). Either path streams the
package plus a freshly minted enrollment.json whose token is never written
to storage in plaintext, only its SHA-256 hash. On the client host,
gti-agent.ps1 install-auto verifies the bundle's SHA-256, refuses a non-
HTTPS Hub, enrolls via POST /v1/agents/enroll (deleting the enrollment
token immediately after, success or failure), writes an ACL-restricted
.env with the newly issued permanent Agent token, and registers the
Windows Service without starting it (Firebird stays unconfigured on
purpose). Idempotent, replay-proof, rate-limited per IP, fully audited,
never logs a secret.

Migration 0022_agent_releases_enrollment (verified fresh-DB ==
upgrade-from-0021 by pg_dump diff). GTI_FISCAL_VERSION/AGENT_VERSION
bumped to 1.1.0 as the single source of truth across code, packages, the
release manifest and every published AgentRelease row.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2026-09-23 13:19:31 -03:00

706 lines
43 KiB
Python
Raw Blame History

"""PostgreSQL records owned by the Fiscal Hub."""
from datetime import datetime
from sqlalchemy import Boolean, CheckConstraint, DateTime, ForeignKey, Integer, JSON, String, Text, UniqueConstraint, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
class Base(DeclarativeBase):
pass
class Dataset(Base):
__tablename__ = "datasets"
__table_args__ = (UniqueConstraint("provider", "sha256", name="uq_dataset_provider_hash"),)
id: Mapped[int] = mapped_column(primary_key=True)
kind: Mapped[str] = mapped_column(String(32), index=True)
provider: Mapped[str] = mapped_column(String(128))
sha256: Mapped[str] = mapped_column(String(64))
raw_sha256: Mapped[str | None] = mapped_column(String(64), nullable=True)
version_label: Mapped[str] = mapped_column(String(256))
act: Mapped[str] = mapped_column(Text)
source_url: Mapped[str] = mapped_column(Text)
downloaded_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
status: Mapped[str] = mapped_column(String(32), index=True)
record_count: Mapped[int] = mapped_column(Integer, default=0)
class NcmEntry(Base):
__tablename__ = "ncm_entries"
__table_args__ = (UniqueConstraint("dataset_id", "code", "valid_from", "valid_to", name="uq_ncm_versioned_entry"),)
id: Mapped[int] = mapped_column(primary_key=True)
dataset_id: Mapped[int] = mapped_column(ForeignKey("datasets.id"), index=True)
code: Mapped[str] = mapped_column(String(16), index=True)
description: Mapped[str] = mapped_column(Text)
valid_from: Mapped[str] = mapped_column(String(32))
valid_to: Mapped[str] = mapped_column(String(32))
class LegacyNcmEntry(Base):
__tablename__ = "legacy_ncm_entries"
code: Mapped[str] = mapped_column(String(16), primary_key=True)
source: Mapped[str] = mapped_column(Text)
source_date: Mapped[str] = mapped_column(String(32))
reason: Mapped[str] = mapped_column(Text)
successor_code: Mapped[str | None] = mapped_column(String(16), nullable=True)
class CestEntry(Base):
__tablename__ = "cest_entries"
__table_args__ = (UniqueConstraint("dataset_id", "code", name="uq_cest_versioned_entry"),)
id: Mapped[int] = mapped_column(primary_key=True)
dataset_id: Mapped[int] = mapped_column(ForeignKey("datasets.id"), index=True)
code: Mapped[str] = mapped_column(String(16), index=True)
description: Mapped[str] = mapped_column(Text)
annex: Mapped[str] = mapped_column(String(8))
class CestNcmEntry(Base):
__tablename__ = "cest_ncm_entries"
__table_args__ = (UniqueConstraint("cest_entry_id", "ncm_code", name="uq_cest_ncm_link"),)
id: Mapped[int] = mapped_column(primary_key=True)
cest_entry_id: Mapped[int] = mapped_column(ForeignKey("cest_entries.id"), index=True)
ncm_code: Mapped[str] = mapped_column(String(16), index=True)
class TipiEntry(Base):
__tablename__ = "tipi_entries"
id: Mapped[int] = mapped_column(primary_key=True)
dataset_id: Mapped[int] = mapped_column(ForeignKey("datasets.id"), index=True)
ncm: Mapped[str] = mapped_column(String(8), index=True)
ex: Mapped[str | None] = mapped_column(String(32), nullable=True)
description: Mapped[str] = mapped_column(Text)
ipi_rate: Mapped[float | None] = mapped_column(nullable=True)
ipi_rate_text: Mapped[str | None] = mapped_column(String(32), nullable=True)
source_row: Mapped[int] = mapped_column(Integer)
metadata_json: Mapped[dict] = mapped_column(JSON, default=dict)
class BaStMvaEntry(Base):
__tablename__ = "ba_st_mva_entries"
id: Mapped[int] = mapped_column(primary_key=True)
dataset_id: Mapped[int] = mapped_column(ForeignKey("datasets.id"), index=True)
item: Mapped[str] = mapped_column(String(32))
segment: Mapped[str | None] = mapped_column(Text, nullable=True)
cest: Mapped[str | None] = mapped_column(String(16), index=True, nullable=True)
description: Mapped[str | None] = mapped_column(Text, nullable=True)
mva_original: Mapped[str | None] = mapped_column(String(32), nullable=True)
mva_adjusted: Mapped[str | None] = mapped_column(String(32), nullable=True)
agreement_info: Mapped[str | None] = mapped_column(Text, nullable=True)
source_page: Mapped[int] = mapped_column(Integer)
metadata_json: Mapped[dict] = mapped_column(JSON, default=dict)
class BaStMvaNcmSh(Base):
__tablename__ = "ba_st_mva_ncm_sh"
__table_args__ = (UniqueConstraint("entry_id", "ncm_sh", name="uq_ba_st_mva_entry_ncm_sh"),)
id: Mapped[int] = mapped_column(primary_key=True)
entry_id: Mapped[int] = mapped_column(ForeignKey("ba_st_mva_entries.id"), index=True)
ncm_sh: Mapped[str] = mapped_column(String(32), index=True)
metadata_json: Mapped[dict] = mapped_column(JSON, default=dict)
class FiscalCodeEntry(Base):
"""Small national fiscal code domains sourced from the official NF-e MOC
manual: ORIGEM_MERCADORIA, CST_ICMS, CSOSN, CST_PIS, CST_COFINS."""
__tablename__ = "fiscal_code_entries"
__table_args__ = (UniqueConstraint("dataset_id", "domain", "code", name="uq_fiscal_code_entry"),)
id: Mapped[int] = mapped_column(primary_key=True)
dataset_id: Mapped[int] = mapped_column(ForeignKey("datasets.id"), index=True)
domain: Mapped[str] = mapped_column(String(32), index=True)
code: Mapped[str] = mapped_column(String(8), index=True)
description: Mapped[str] = mapped_column(Text)
source_section: Mapped[str] = mapped_column(String(96))
metadata_json: Mapped[dict] = mapped_column(JSON, default=dict)
class CfopEntry(Base):
"""CFOP (Código Fiscal de Operações e Prestações) — Portal Nacional NF-e."""
__tablename__ = "cfop_entries"
__table_args__ = (UniqueConstraint("dataset_id", "code", name="uq_cfop_entry"),)
id: Mapped[int] = mapped_column(primary_key=True)
dataset_id: Mapped[int] = mapped_column(ForeignKey("datasets.id"), index=True)
code: Mapped[str] = mapped_column(String(8), index=True)
title: Mapped[str] = mapped_column(Text)
description: Mapped[str | None] = mapped_column(Text, nullable=True)
effective_from: Mapped[str | None] = mapped_column(String(32), nullable=True)
effective_to: Mapped[str | None] = mapped_column(String(32), nullable=True)
ind_nfe: Mapped[str | None] = mapped_column(String(4), nullable=True)
ind_comunica: Mapped[str | None] = mapped_column(String(4), nullable=True)
ind_transp: Mapped[str | None] = mapped_column(String(4), nullable=True)
ind_devol: Mapped[str | None] = mapped_column(String(4), nullable=True)
ind_retor: Mapped[str | None] = mapped_column(String(4), nullable=True)
ind_anula: Mapped[str | None] = mapped_column(String(4), nullable=True)
ind_remes: Mapped[str | None] = mapped_column(String(4), nullable=True)
ind_comb: Mapped[str | None] = mapped_column(String(4), nullable=True)
ind_exc_ibscbs: Mapped[str | None] = mapped_column(String(4), nullable=True)
class FcpRate(Base):
"""FCP (Fundo de Combate à Pobreza) alíquotas por UF — Portal Nacional NF-e."""
__tablename__ = "fcp_rates"
__table_args__ = (UniqueConstraint("dataset_id", "uf", name="uq_fcp_rate_uf"),)
id: Mapped[int] = mapped_column(primary_key=True)
dataset_id: Mapped[int] = mapped_column(ForeignKey("datasets.id"), index=True)
uf_code: Mapped[str] = mapped_column(String(4))
uf: Mapped[str] = mapped_column(String(4), index=True)
uf_name: Mapped[str] = mapped_column(String(64))
aliq_1: Mapped[str | None] = mapped_column(String(32), nullable=True)
aliq_2: Mapped[str | None] = mapped_column(String(32), nullable=True)
aliq_3: Mapped[str | None] = mapped_column(String(32), nullable=True)
observacao: Mapped[str | None] = mapped_column(Text, nullable=True)
class AnpProductCode(Base):
"""cProdANP — official code list used by NF-e, published on the Portal Nacional NF-e."""
__tablename__ = "anp_product_codes"
__table_args__ = (UniqueConstraint("dataset_id", "cprod_anp", name="uq_anp_product_code"),)
id: Mapped[int] = mapped_column(primary_key=True)
dataset_id: Mapped[int] = mapped_column(ForeignKey("datasets.id"), index=True)
cprod_anp: Mapped[str] = mapped_column(String(16), index=True)
class NcmUtribEntry(Base):
"""NCM x uTrib (unidade tributável de comércio exterior) — Portal Nacional NF-e."""
__tablename__ = "ncm_utrib_entries"
__table_args__ = (UniqueConstraint("dataset_id", "ncm", name="uq_ncm_utrib_entry"),)
id: Mapped[int] = mapped_column(primary_key=True)
dataset_id: Mapped[int] = mapped_column(ForeignKey("datasets.id"), index=True)
ncm: Mapped[str] = mapped_column(String(16), index=True)
effective_from: Mapped[str | None] = mapped_column(String(32), nullable=True)
effective_to: Mapped[str | None] = mapped_column(String(32), nullable=True)
utrib_abbreviation: Mapped[str | None] = mapped_column(String(16), nullable=True)
utrib_description: Mapped[str | None] = mapped_column(Text, nullable=True)
class RtcBiofuelIndexEntry(Base):
"""RTC pBioObrigatorio (índice de biocombustível) by cProdANP — Portal Nacional NF-e."""
__tablename__ = "rtc_biofuel_index_entries"
id: Mapped[int] = mapped_column(primary_key=True)
dataset_id: Mapped[int] = mapped_column(ForeignKey("datasets.id"), index=True)
cprod_anp: Mapped[str | None] = mapped_column(String(16), index=True, nullable=True)
description: Mapped[str | None] = mapped_column(Text, nullable=True)
p_bio_percent: Mapped[float | None] = mapped_column(nullable=True)
effective_from: Mapped[str | None] = mapped_column(String(32), nullable=True)
effective_to: Mapped[str | None] = mapped_column(String(32), nullable=True)
is_historical: Mapped[bool] = mapped_column(Boolean, default=False)
class RtcCredPresEntry(Base):
"""RTC cCredPres (crédito presumido IBS/CBS), LC 214/2025 — Portal Nacional NF-e.
Keeps every official field distinctly so future automatic determination
doesn't require re-parsing the source."""
__tablename__ = "rtc_cred_pres_entries"
__table_args__ = (UniqueConstraint("dataset_id", "code", name="uq_rtc_cred_pres_code"),)
id: Mapped[int] = mapped_column(primary_key=True)
dataset_id: Mapped[int] = mapped_column(ForeignKey("datasets.id"), index=True)
code: Mapped[str] = mapped_column(String(8), index=True)
description: Mapped[str] = mapped_column(Text)
legal_basis: Mapped[str | None] = mapped_column(Text, nullable=True)
apropria_via_nf: Mapped[bool | None] = mapped_column(Boolean, nullable=True)
apropria_via_evento: Mapped[bool | None] = mapped_column(Boolean, nullable=True)
ind_deduz_cred_pres: Mapped[str | None] = mapped_column(String(8), nullable=True)
ind_g_cbs_cred_pres: Mapped[str | None] = mapped_column(String(8), nullable=True)
ind_g_ibs_cred_pres: Mapped[str | None] = mapped_column(String(8), nullable=True)
aliquota_cbs_text: Mapped[str | None] = mapped_column(Text, nullable=True)
aliquota_ibs_text: Mapped[str | None] = mapped_column(Text, nullable=True)
p_aliq_cred_pres_cbs: Mapped[str | None] = mapped_column(Text, nullable=True)
p_aliq_cred_pres_ibs: Mapped[str | None] = mapped_column(Text, nullable=True)
p_red_transicao_ibs: Mapped[str | None] = mapped_column(Text, nullable=True)
# Free text in some rows (e.g. "cClass de alíquota zero"), not always the numeric code.
cclass_nota_referenciada: Mapped[str | None] = mapped_column(Text, nullable=True)
d_ini_vig_cbs: Mapped[str | None] = mapped_column(String(32), nullable=True)
d_fim_vig_cbs: Mapped[str | None] = mapped_column(String(32), nullable=True)
d_ini_vig_ibs: Mapped[str | None] = mapped_column(String(32), nullable=True)
d_fim_vig_ibs: Mapped[str | None] = mapped_column(String(32), nullable=True)
class Tenant(Base):
__tablename__ = "tenants"
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column(String(128), unique=True)
class Agent(Base):
__tablename__ = "agents"
id: Mapped[int] = mapped_column(primary_key=True)
tenant_id: Mapped[int] = mapped_column(ForeignKey("tenants.id"), index=True) # company_id
name: Mapped[str] = mapped_column(String(128))
token_hash: Mapped[str] = mapped_column(String(64), unique=True)
enabled: Mapped[bool] = mapped_column(Boolean, default=True, server_default="true")
last_seen_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
last_status: Mapped[str | None] = mapped_column(String(32), nullable=True)
schema_fingerprint: Mapped[str | None] = mapped_column(String(64), nullable=True)
agent_version: Mapped[str | None] = mapped_column(String(32), nullable=True)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now())
class Installation(Base):
__tablename__ = "installations"
__table_args__ = (UniqueConstraint("installation_id", name="uq_installation_identity"), UniqueConstraint("agent_id", name="uq_installation_agent"))
id: Mapped[int] = mapped_column(primary_key=True)
agent_id: Mapped[int] = mapped_column(ForeignKey("agents.id"), index=True)
tenant_id: Mapped[int] = mapped_column(ForeignKey("tenants.id"), index=True)
installation_id: Mapped[str] = mapped_column(String(36), index=True)
display_name: Mapped[str] = mapped_column(String(160), default="", server_default="")
customer_code: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True)
notes: Mapped[str | None] = mapped_column(Text, nullable=True)
hostname: Mapped[str | None] = mapped_column(String(255), nullable=True, index=True)
windows_version: Mapped[str | None] = mapped_column(String(128), nullable=True)
machine_fingerprint: Mapped[str | None] = mapped_column(String(128), nullable=True)
company_code: Mapped[str | None] = mapped_column(String(64), nullable=True)
company_name: Mapped[str | None] = mapped_column(String(160), nullable=True)
company_cnpj_masked: Mapped[str | None] = mapped_column(String(24), nullable=True)
company_uf: Mapped[str | None] = mapped_column(String(2), nullable=True)
company_crt: Mapped[str | None] = mapped_column(String(16), nullable=True)
firebird_version: Mapped[str | None] = mapped_column(String(64), nullable=True)
firebird_ods: Mapped[str | None] = mapped_column(String(32), nullable=True)
database_identifier: Mapped[str | None] = mapped_column(String(255), nullable=True)
capabilities: Mapped[dict] = mapped_column(JSON, default=dict, server_default="{}")
fiscal_engine_version: Mapped[str | None] = mapped_column(String(64), nullable=True)
dataset_versions: Mapped[dict] = mapped_column(JSON, default=dict, server_default="{}")
started_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
last_seen_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True, index=True)
last_status: Mapped[str | None] = mapped_column(String(32), nullable=True)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now())
class AgentSnapshot(Base):
__tablename__ = "agent_snapshots"
id: Mapped[int] = mapped_column(primary_key=True)
installation_id: Mapped[int] = mapped_column(ForeignKey("installations.id"), unique=True, index=True)
total_products: Mapped[int] = mapped_column(Integer, default=0, server_default="0")
fiscally_complete: Mapped[int] = mapped_column(Integer, default=0, server_default="0")
safe_blocked: Mapped[int] = mapped_column(Integer, default=0, server_default="0")
failed_products: Mapped[int] = mapped_column(Integer, default=0, server_default="0")
coverage_percent: Mapped[float] = mapped_column(default=0.0, server_default="0")
blockers: Mapped[dict] = mapped_column(JSON, default=dict, server_default="{}")
last_job_status: Mapped[str | None] = mapped_column(String(16), nullable=True)
last_job_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
last_successful_job_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
agent_health: Mapped[str] = mapped_column(String(16), default="UNKNOWN", server_default="UNKNOWN", index=True)
health_reason: Mapped[str | None] = mapped_column(Text, nullable=True)
version_drift: Mapped[str] = mapped_column(String(16), default="UNKNOWN", server_default="UNKNOWN", index=True)
updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now(), index=True)
class AgentEvent(Base):
__tablename__ = "agent_events"
id: Mapped[int] = mapped_column(primary_key=True)
tenant_id: Mapped[int] = mapped_column(ForeignKey("tenants.id"), index=True)
installation_id: Mapped[int] = mapped_column(ForeignKey("installations.id"), index=True)
agent_id: Mapped[int] = mapped_column(ForeignKey("agents.id"), index=True)
severity: Mapped[str] = mapped_column(String(16), index=True)
event_type: Mapped[str] = mapped_column(String(48), index=True)
message: Mapped[str] = mapped_column(Text)
structured_context: Mapped[dict] = mapped_column(JSON, default=dict)
job_id: Mapped[int | None] = mapped_column(ForeignKey("jobs.id"), nullable=True, index=True)
product_code: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True)
blocker_id: Mapped[int | None] = mapped_column(ForeignKey("blocked_products.id"), nullable=True, index=True)
is_audit: Mapped[bool] = mapped_column(Boolean, default=False, server_default="false", index=True)
occurred_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now(), index=True)
class AdminUser(Base):
"""Human principal of the control panel.
LOCAL: break-glass account with a password (create_admin.py).
GTI_GERENCIADOR: collaborator provisioned by SSO (central/app/sso.py) — no
password at all; identified by the panel's immutable subject.
"""
__tablename__ = "admin_users"
__table_args__ = (
UniqueConstraint("external_subject", name="uq_admin_users_external_subject"),
CheckConstraint(
"(auth_source = 'LOCAL' AND password_hash IS NOT NULL AND external_subject IS NULL)"
" OR (auth_source = 'GTI_GERENCIADOR' AND password_hash IS NULL AND external_subject IS NOT NULL)",
name="ck_admin_users_auth_source",
),
)
id: Mapped[int] = mapped_column(primary_key=True)
username: Mapped[str] = mapped_column(String(128), unique=True, index=True)
password_hash: Mapped[str | None] = mapped_column(String(255), nullable=True)
is_active: Mapped[bool] = mapped_column(Boolean, default=True, server_default="true")
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
last_login_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
auth_source: Mapped[str] = mapped_column(String(16), default="LOCAL", server_default="LOCAL")
external_subject: Mapped[str | None] = mapped_column(String(128), nullable=True)
display_name: Mapped[str | None] = mapped_column(String(160), nullable=True)
external_role: Mapped[str | None] = mapped_column(String(32), nullable=True)
updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now())
class SsoUsedAssertion(Base):
"""jti of every accepted SSO assertion, kept past its expiry to reject replays."""
__tablename__ = "sso_used_assertions"
jti: Mapped[str] = mapped_column(String(64), primary_key=True)
subject: Mapped[str] = mapped_column(String(128))
used_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
expires_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), index=True)
class WebSession(Base):
__tablename__ = "web_sessions"
id: Mapped[int] = mapped_column(primary_key=True)
admin_user_id: Mapped[int] = mapped_column(ForeignKey("admin_users.id"), index=True)
token_hash: Mapped[str] = mapped_column(String(64), unique=True, index=True)
csrf_hash: Mapped[str] = mapped_column(String(64))
expires_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), index=True)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
last_seen_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
user_agent: Mapped[str | None] = mapped_column(String(512), nullable=True)
remote_addr: Mapped[str | None] = mapped_column(String(64), nullable=True)
class AuditEvent(Base):
__tablename__ = "audit_events"
id: Mapped[int] = mapped_column(primary_key=True)
admin_user_id: Mapped[int | None] = mapped_column(ForeignKey("admin_users.id"), nullable=True, index=True)
tenant_id: Mapped[int | None] = mapped_column(ForeignKey("tenants.id"), nullable=True, index=True)
action: Mapped[str] = mapped_column(String(64), index=True)
target_type: Mapped[str] = mapped_column(String(64))
target_id: Mapped[str] = mapped_column(String(128))
context: Mapped[dict] = mapped_column(JSON, default=dict)
severity: Mapped[str] = mapped_column(String(16), default="INFO", server_default="INFO")
occurred_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now(), index=True)
class AgentRelease(Base):
"""A published build of the Windows Agent package. The universal binary/zip is
shared by every installation; storage_path points at it on the Hub's local
release storage (never inside PostgreSQL)."""
__tablename__ = "agent_releases"
__table_args__ = (UniqueConstraint("version", "platform", "architecture", name="uq_agent_release_version"),)
id: Mapped[int] = mapped_column(primary_key=True)
version: Mapped[str] = mapped_column(String(32), index=True)
platform: Mapped[str] = mapped_column(String(16), default="windows", server_default="windows")
architecture: Mapped[str] = mapped_column(String(16), default="x64", server_default="x64")
filename: Mapped[str] = mapped_column(String(255))
size_bytes: Mapped[int] = mapped_column(Integer)
sha256: Mapped[str] = mapped_column(String(64), index=True)
storage_path: Mapped[str] = mapped_column(String(512))
status: Mapped[str] = mapped_column(String(16), default="DRAFT", server_default="DRAFT", index=True) # DRAFT | PUBLISHED | REVOKED
release_notes: Mapped[str | None] = mapped_column(Text, nullable=True)
created_by: Mapped[int | None] = mapped_column(ForeignKey("admin_users.id"), nullable=True)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
published_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
class AgentEnrollment(Base):
"""A pending (or completed) Agent installation: created by an operator in the
control panel before any real Agent exists. Carries two distinct, hashed,
single-use, short-TTL secrets — a download token (for the shareable package
link) and an enrollment token (embedded in the bundle, consumed once by
POST /v1/agents/enroll). Never stores either secret in plaintext."""
__tablename__ = "agent_enrollments"
id: Mapped[int] = mapped_column(primary_key=True)
enrollment_uuid: Mapped[str] = mapped_column(String(36), unique=True, index=True)
tenant_id: Mapped[int] = mapped_column(ForeignKey("tenants.id"), index=True)
display_name: Mapped[str] = mapped_column(String(160))
expected_hostname: Mapped[str | None] = mapped_column(String(255), nullable=True)
release_id: Mapped[int] = mapped_column(ForeignKey("agent_releases.id"), index=True)
status: Mapped[str] = mapped_column(String(16), default="PENDING", server_default="PENDING", index=True)
# PENDING | LINK_CREATED | DOWNLOADED | ENROLLED | EXPIRED | REVOKED | FAILED
created_by: Mapped[int | None] = mapped_column(ForeignKey("admin_users.id"), nullable=True)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
download_token_hash: Mapped[str | None] = mapped_column(String(64), nullable=True)
download_token_expires_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
download_max_uses: Mapped[int] = mapped_column(Integer, default=1, server_default="1")
download_uses: Mapped[int] = mapped_column(Integer, default=0, server_default="0")
downloaded_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
enrollment_token_hash: Mapped[str | None] = mapped_column(String(64), nullable=True)
enrollment_token_expires_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
enrollment_used_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
machine_fingerprint_seen: Mapped[str | None] = mapped_column(String(128), nullable=True)
hostname_seen: Mapped[str | None] = mapped_column(String(255), nullable=True)
hostname_mismatch: Mapped[bool] = mapped_column(Boolean, default=False, server_default="false")
agent_id: Mapped[int | None] = mapped_column(ForeignKey("agents.id"), nullable=True, index=True)
updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now())
class Job(Base):
__tablename__ = "jobs"
id: Mapped[int] = mapped_column(primary_key=True)
agent_id: Mapped[int] = mapped_column(ForeignKey("agents.id"), index=True)
kind: Mapped[str] = mapped_column(String(32), index=True) # REPROCESS_ALL | REPROCESS_PRODUCTS
status: Mapped[str] = mapped_column(String(16), index=True, default="PENDING", server_default="PENDING")
payload: Mapped[dict] = mapped_column(JSON, default=dict)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
started_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
finished_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
attempts: Mapped[int] = mapped_column(Integer, default=0, server_default="0")
result_summary: Mapped[dict | None] = mapped_column(JSON, nullable=True)
error: Mapped[str | None] = mapped_column(Text, nullable=True)
class BlockedProduct(Base):
"""Deduplicated safe exception for one tenant/product/field/input."""
__tablename__ = "blocked_products"
__table_args__ = (UniqueConstraint("blocker_fingerprint", name="uq_blocked_product_fingerprint"),)
id: Mapped[int] = mapped_column(primary_key=True)
tenant_id: Mapped[int] = mapped_column(ForeignKey("tenants.id"), index=True)
installation_id: Mapped[int | None] = mapped_column(ForeignKey("installations.id"), nullable=True, index=True)
company: Mapped[str] = mapped_column(String(128), index=True)
product_code: Mapped[str] = mapped_column(String(64), index=True)
gtin: Mapped[str | None] = mapped_column(String(16), nullable=True, index=True)
description: Mapped[str | None] = mapped_column(Text, nullable=True)
blocker_type: Mapped[str] = mapped_column(String(48), index=True)
field: Mapped[str] = mapped_column(String(64), index=True)
reason_code: Mapped[str] = mapped_column(String(96))
evidence_summary: Mapped[dict] = mapped_column(JSON, default=dict)
input_fingerprint: Mapped[str] = mapped_column(String(64), index=True)
blocker_fingerprint: Mapped[str] = mapped_column(String(64))
first_seen_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
last_seen_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
retry_count: Mapped[int] = mapped_column(Integer, default=0, server_default="0")
next_retry_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
resolved_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
resolution_source: Mapped[str | None] = mapped_column(String(48), nullable=True)
class ClassificationRequest(Base):
__tablename__ = "classification_requests"
id: Mapped[int] = mapped_column(primary_key=True)
tenant_id: Mapped[int] = mapped_column(ForeignKey("tenants.id"), index=True)
agent_id: Mapped[int] = mapped_column(ForeignKey("agents.id"), index=True)
fingerprint: Mapped[str] = mapped_column(String(64), unique=True)
payload: Mapped[dict] = mapped_column(JSON)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
class ClassificationResult(Base):
__tablename__ = "classification_results"
id: Mapped[int] = mapped_column(primary_key=True)
request_id: Mapped[int] = mapped_column(ForeignKey("classification_requests.id"), unique=True)
result: Mapped[dict] = mapped_column(JSON)
class ClassificationCache(Base):
__tablename__ = "classification_cache"
fingerprint: Mapped[str] = mapped_column(String(64), primary_key=True)
tenant_id: Mapped[int] = mapped_column(ForeignKey("tenants.id"), index=True)
result: Mapped[dict] = mapped_column(JSON)
class RtcCstIbsCbsEntry(Base):
"""RTC CST-IBS/CBS (sheet 'CST ...' of the official cClassTrib workbook)."""
__tablename__ = "rtc_cst_ibscbs_entries"
__table_args__ = (UniqueConstraint("dataset_id", "cst", name="uq_rtc_cst_ibscbs"),)
id: Mapped[int] = mapped_column(primary_key=True)
dataset_id: Mapped[int] = mapped_column(ForeignKey("datasets.id"), index=True)
cst: Mapped[str] = mapped_column(String(8), index=True)
description: Mapped[str] = mapped_column(Text)
indicators: Mapped[dict] = mapped_column(JSON)
class RtcCClassTribEntry(Base):
"""RTC cClassTrib (sheet 'cClass ...'). Every official column is kept:
key fields typed, all ind_* flags in `indicators` keyed by official header."""
__tablename__ = "rtc_cclasstrib_entries"
__table_args__ = (UniqueConstraint("dataset_id", "cclasstrib", name="uq_rtc_cclasstrib"),)
id: Mapped[int] = mapped_column(primary_key=True)
dataset_id: Mapped[int] = mapped_column(ForeignKey("datasets.id"), index=True)
cst: Mapped[str] = mapped_column(String(8), index=True)
cst_description: Mapped[str | None] = mapped_column(Text, nullable=True)
cclasstrib: Mapped[str] = mapped_column(String(8), index=True)
name: Mapped[str | None] = mapped_column(Text, nullable=True)
description: Mapped[str | None] = mapped_column(Text, nullable=True)
lc_redacao: Mapped[str | None] = mapped_column(Text, nullable=True)
lc_214: Mapped[str | None] = mapped_column(Text, nullable=True)
regulamento_cbs: Mapped[str | None] = mapped_column(Text, nullable=True)
regulamento_ibs: Mapped[str | None] = mapped_column(Text, nullable=True)
tipo_aliquota: Mapped[str | None] = mapped_column(String(64), nullable=True)
p_red_ibs: Mapped[str | None] = mapped_column(String(32), nullable=True)
p_red_cbs: Mapped[str | None] = mapped_column(String(32), nullable=True)
d_ini_vig: Mapped[str | None] = mapped_column(String(32), nullable=True)
d_fim_vig: Mapped[str | None] = mapped_column(String(32), nullable=True)
data_atualizacao: Mapped[str | None] = mapped_column(String(32), nullable=True)
anexo: Mapped[str | None] = mapped_column(Text, nullable=True)
link: Mapped[str | None] = mapped_column(Text, nullable=True)
indicators: Mapped[dict] = mapped_column(JSON)
class RtcCbsRateEntry(Base):
"""RTC CBS rate per year (IT 2026.002). rate is NULL when the official
text has no number yet (e.g. 'aguarda legislação')."""
__tablename__ = "rtc_cbs_rate_entries"
__table_args__ = (UniqueConstraint("dataset_id", "year", name="uq_rtc_cbs_rate_year"),)
id: Mapped[int] = mapped_column(primary_key=True)
dataset_id: Mapped[int] = mapped_column(ForeignKey("datasets.id"), index=True)
year: Mapped[int] = mapped_column(Integer, index=True)
rate_text: Mapped[str] = mapped_column(Text)
rate: Mapped[float | None] = mapped_column(nullable=True)
class FiscalContextEntry(Base):
"""Operator-supplied fiscal fact (product role, origin, ST retention, treatment, ...). Hierarchy: PRODUCT > GROUP > COMPANY."""
__tablename__ = "fiscal_context_entries"
__table_args__ = (UniqueConstraint("tenant_id", "company_key", "scope", "scope_key", "key", name="uq_fiscal_context_entry"),)
id: Mapped[int] = mapped_column(primary_key=True)
tenant_id: Mapped[int] = mapped_column(ForeignKey("tenants.id"), index=True)
company_key: Mapped[str] = mapped_column(String(64), index=True)
scope: Mapped[str] = mapped_column(String(8))
scope_key: Mapped[str] = mapped_column(String(64), default="", server_default="")
key: Mapped[str] = mapped_column(String(48))
value: Mapped[str] = mapped_column(Text)
source: Mapped[str] = mapped_column(String(16), default="OPERATOR", server_default="OPERATOR")
reason: Mapped[str | None] = mapped_column(Text, nullable=True)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
# ================= AI semantic layer V1 =================
# The AI layer is NOT a fiscal authority: it only selects among official candidates and
# never writes to Firebird. Knowledge is versioned, provenance-tracked and never deleted.
class KbProduct(Base):
"""GLOBAL product knowledge: only facts inherent to the product (never tenant context)."""
__tablename__ = "kb_products"
id: Mapped[int] = mapped_column(primary_key=True)
fingerprint: Mapped[str] = mapped_column(String(64), unique=True, index=True)
gtin: Mapped[str | None] = mapped_column(String(16), unique=True, nullable=True, index=True)
identity_kind: Mapped[str] = mapped_column(String(16), default="FINGERPRINT") # GTIN | FINGERPRINT | SEMANTIC | UNKNOWN
normalized_description: Mapped[str] = mapped_column(Text, default="")
brand: Mapped[str | None] = mapped_column(Text, nullable=True)
unit: Mapped[str | None] = mapped_column(String(32), nullable=True)
package_quantity: Mapped[str | None] = mapped_column(String(32), nullable=True)
category: Mapped[str | None] = mapped_column(Text, nullable=True)
semantic_attributes: Mapped[dict] = mapped_column(JSON, default=dict)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now())
class KbKnowledge(Base):
"""Versioned knowledge rows. Unique per (product, key, value): conflicting values for the
same key are kept side by side and surfaced as KNOWLEDGE_CONFLICT, never majority-picked."""
__tablename__ = "kb_knowledge"
__table_args__ = (UniqueConstraint("kb_product_id", "knowledge_key", "value", name="uq_kb_knowledge_value"),)
id: Mapped[int] = mapped_column(primary_key=True)
kb_product_id: Mapped[int] = mapped_column(ForeignKey("kb_products.id"), index=True)
knowledge_key: Mapped[str] = mapped_column(String(48), index=True) # candidate_ncm | validated_ncm | candidate_cest | candidate_ex_tipi | ...
value: Mapped[str] = mapped_column(Text)
source: Mapped[str] = mapped_column(String(32)) # OFFICIAL_DATASET | ERP_OBSERVATION | OPERATOR | AI | AI_PLUS_OFFICIAL_VALIDATION
confidence: Mapped[float] = mapped_column(default=0.0)
evidence: Mapped[dict] = mapped_column(JSON, default=dict)
model: Mapped[str | None] = mapped_column(String(64), nullable=True)
prompt_version: Mapped[str | None] = mapped_column(String(32), nullable=True)
dataset_versions: Mapped[dict] = mapped_column(JSON, default=dict)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
validated_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
class KbObservation(Base):
"""Raw per-tenant ERP observation of a KB product (evidence aggregation only)."""
__tablename__ = "kb_observations"
id: Mapped[int] = mapped_column(primary_key=True)
kb_product_id: Mapped[int] = mapped_column(ForeignKey("kb_products.id"), index=True)
tenant_id: Mapped[int] = mapped_column(ForeignKey("tenants.id"), index=True)
company_key: Mapped[str] = mapped_column(String(64), index=True)
erp_code: Mapped[str] = mapped_column(String(32))
database_label: Mapped[str] = mapped_column(String(64))
observed_fields: Mapped[dict] = mapped_column(JSON, default=dict)
source: Mapped[str] = mapped_column(String(32), default="ERP_OBSERVATION")
observed_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
class TenantProductContext(Base):
"""TENANT-specific product context: role, ST-withheld, operation profile, operator decisions.
Never promoted to global knowledge."""
__tablename__ = "tenant_product_context"
__table_args__ = (UniqueConstraint("tenant_id", "company_key", "erp_code", name="uq_tenant_product_context"),)
id: Mapped[int] = mapped_column(primary_key=True)
tenant_id: Mapped[int] = mapped_column(ForeignKey("tenants.id"), index=True)
company_key: Mapped[str] = mapped_column(String(64), index=True)
erp_code: Mapped[str] = mapped_column(String(32))
kb_product_id: Mapped[int | None] = mapped_column(ForeignKey("kb_products.id"), nullable=True)
product_role: Mapped[str | None] = mapped_column(String(16), nullable=True) # RESALE | OWN_PRODUCTION | INPUT | SERVICE | UNKNOWN
st_withheld: Mapped[str | None] = mapped_column(String(4), nullable=True) # Y | N — never AI-decided
operation_profile: Mapped[dict] = mapped_column(JSON, default=dict)
source: Mapped[str] = mapped_column(String(32), default="ERP_OBSERVATION")
confidence: Mapped[float] = mapped_column(default=0.0)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now())
class AiWorkItem(Base):
"""Deduplicated semantic work item (one knowledge need, many pending cells)."""
__tablename__ = "ai_work_items"
__table_args__ = (UniqueConstraint("dedup_key", name="uq_ai_work_item_dedup"),)
id: Mapped[int] = mapped_column(primary_key=True)
task_type: Mapped[str] = mapped_column(String(24), index=True) # INVALID_NCM | CEST | EX_TIPI | PRODUCT_ROLE
dedup_key: Mapped[str] = mapped_column(String(160))
question_refs: Mapped[list] = mapped_column(JSON, default=list)
payload: Mapped[dict] = mapped_column(JSON, default=dict)
status: Mapped[str] = mapped_column(String(32), default="PENDING", index=True) # PENDING | KB_EXACT_HIT | KB_FINGERPRINT_HIT | HIGH_CONFIDENCE_RETRIEVAL | SINGLE_OFFICIAL_CANDIDATE | LLM_SELECTION_REQUIRED | UNRESOLVED | UNRESOLVED_NO_OFFICIAL_MAPPING | CONFLICT | DONE | SKIPPED
kb_product_id: Mapped[int | None] = mapped_column(ForeignKey("kb_products.id"), nullable=True)
cells_covered: Mapped[int] = mapped_column(Integer, default=0)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now())
class AiRun(Base):
"""One provider execution over a batch of work items, with budget-guard evidence."""
__tablename__ = "ai_runs"
id: Mapped[int] = mapped_column(primary_key=True)
provider: Mapped[str] = mapped_column(String(64))
model: Mapped[str] = mapped_column(String(64))
prompt_version: Mapped[str] = mapped_column(String(32))
work_item_ids: Mapped[list] = mapped_column(JSON, default=list)
input_tokens: Mapped[int] = mapped_column(Integer, default=0)
output_tokens: Mapped[int] = mapped_column(Integer, default=0)
cost_usd: Mapped[float] = mapped_column(default=0.0)
status: Mapped[str] = mapped_column(String(16), default="RUN")
results: Mapped[list] = mapped_column(JSON, default=list)
budget_guard: Mapped[dict] = mapped_column(JSON, default=dict)
started_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
finished_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
class AiEvalCache(Base):
"""Persistent LLM result cache: identical (input_fingerprint, candidate_set_hash,
prompt_version, model, dataset_versions) re-execution makes ZERO new calls."""
__tablename__ = "ai_eval_cache"
__table_args__ = (UniqueConstraint("input_fingerprint", "candidate_set_hash", "prompt_version", "model", "dataset_version_ncm", name="uq_ai_eval_cache"),)
id: Mapped[int] = mapped_column(primary_key=True)
input_fingerprint: Mapped[str] = mapped_column(String(64), index=True)
candidate_set_hash: Mapped[str] = mapped_column(String(64))
prompt_version: Mapped[str] = mapped_column(String(32))
model: Mapped[str] = mapped_column(String(64))
dataset_version_ncm: Mapped[str] = mapped_column(String(64))
phase: Mapped[str] = mapped_column(String(16))
answer: Mapped[dict] = mapped_column(JSON)
gold_ncm: Mapped[str | None] = mapped_column(String(8), nullable=True)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
class AiUsageLedger(Base):
"""Central persistent usage ledger reserved before any future provider call."""
__tablename__ = "ai_usage_ledger"
id: Mapped[int] = mapped_column(primary_key=True)
provider: Mapped[str] = mapped_column(String(64))
model: Mapped[str] = mapped_column(String(64))
run_id: Mapped[str] = mapped_column(String(128), index=True)
tenant: Mapped[str] = mapped_column(String(128), index=True)
items: Mapped[int] = mapped_column(Integer, default=0)
input_tokens: Mapped[int] = mapped_column(Integer, default=0)
output_tokens: Mapped[int] = mapped_column(Integer, default=0)
cost: Mapped[float] = mapped_column(default=0.0)
timestamp: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now(), index=True)
class ExternalProductCache(Base):
"""Versioned GTIN cache; legal/provider flags remain explicit metadata."""
__tablename__ = "external_product_cache"
__table_args__ = (UniqueConstraint("gtin", "provider", "provider_record_version", name="uq_external_product_cache_record"),)
id: Mapped[int] = mapped_column(primary_key=True)
gtin: Mapped[str] = mapped_column(String(16), index=True)
provider: Mapped[str] = mapped_column(String(64))
provider_record_version: Mapped[str] = mapped_column(String(64))
retrieved_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
raw_hash: Mapped[str] = mapped_column(String(64))
normalized_data: Mapped[dict] = mapped_column(JSON, default=dict)
cache_allowed: Mapped[bool] = mapped_column(Boolean, default=False)
redistribution_allowed: Mapped[bool] = mapped_column(Boolean, default=False)
Reference in New Issue View Git Blame Copy Permalink