"""mão de obra agrupada em Depto-Cargo: cargo, grupo_mao_de_obra, alocação e rateio por grupo Revision ID: f2a4c6e8b0d1 Revises: e8d3a5b7c1f9 Create Date: 2026-10-30 12:00:00.000000 - `cargo` (catálogo a partir dos textos de cargo dos colaboradores) e `grupo_mao_de_obra` (Centro de Custo × cargo × classificação); `funcionario.cargo_id` e `funcionario.grupo_id`. - `roteiro_mao_de_obra.funcionario_id` → `grupo_id`: cada linha passa a ser a do grupo do colaborador (linhas do mesmo produto, atividade e direcionador que caem no mesmo grupo somam as horas); o JSON de `historico_versao` é reescrito igual. - `cenario_funcionario_desligado` → `cenario_grupo_desligado` (o grupo fica desligado se todos os membros estavam). Só roda nos schemas de Planejamento (ver f1a9c3d7e284). O downgrade é aproximado: cada linha volta para o primeiro colaborador do grupo (linhas de grupo sem colaboradores são descartadas) e o rateio desliga todos os membros do grupo desligado. """ from typing import Sequence, Union import sqlalchemy as sa from alembic import op # revision identifiers, used by Alembic. revision: str = 'f2a4c6e8b0d1' down_revision: Union[str, None] = 'e8d3a5b7c1f9' branch_labels: Union[str, Sequence[str], None] = None depends_on: Union[str, Sequence[str], None] = None SUFIXO = {'fixo': 'F', 'variavel': 'V'} def _no_schema_comum() -> bool: contexto = op.get_context() return contexto.dialect.name == 'postgresql' and contexto.version_table_schema is None def upgrade() -> None: if _no_schema_comum(): return liga = op.get_bind() op.create_table( 'cargo', sa.Column('id', sa.Integer(), primary_key=True), sa.Column('codigo', sa.String(20), nullable=False, unique=True), sa.Column('nome', sa.String(200), nullable=False, unique=True), sa.Column('digitado_por', sa.String(200), nullable=True), sa.Column('digitado_em', sa.DateTime(), nullable=True), ) op.create_table( 'grupo_mao_de_obra', sa.Column('id', sa.Integer(), primary_key=True), sa.Column('codigo', sa.String(60), nullable=False, unique=True), sa.Column('centro_de_custo_id', sa.Integer(), sa.ForeignKey('centro_de_custo.id'), nullable=False), sa.Column('cargo_id', sa.Integer(), sa.ForeignKey('cargo.id'), nullable=False), sa.Column('classificacao', sa.String(20), nullable=False), sa.Column('digitado_por', sa.String(200), nullable=True), sa.Column('digitado_em', sa.DateTime(), nullable=True), sa.CheckConstraint("classificacao in ('fixo', 'variavel')", name='ck_grupo_mao_de_obra_classificacao'), sa.UniqueConstraint('centro_de_custo_id', 'cargo_id', 'classificacao', name='uq_grupo_mao_de_obra'), ) with op.batch_alter_table('funcionario') as lote: lote.add_column(sa.Column('cargo_id', sa.Integer(), nullable=True)) lote.add_column(sa.Column('grupo_id', sa.Integer(), nullable=True)) lote.create_foreign_key('fk_funcionario_cargo', 'cargo', ['cargo_id'], ['id']) lote.create_foreign_key('fk_funcionario_grupo', 'grupo_mao_de_obra', ['grupo_id'], ['id']) # --- cargos e grupos a partir dos colaboradores pessoas = liga.execute(sa.text('select id, centro_de_custo_id, cargo, classificacao from funcionario order by id')).fetchall() cargos: dict[str, int] = {} for _, _, cargo, _ in pessoas: nome = ' '.join((cargo or '').split()) or '(sem cargo)' if nome.casefold() not in {n.casefold() for n in cargos}: liga.execute(sa.text('insert into cargo (codigo, nome) values (:c, :n)'), {'c': f'CG{len(cargos) + 1:02d}', 'n': nome}) cargos[nome] = liga.execute(sa.text('select id from cargo where nome = :n'), {'n': nome}).scalar() por_nome = {n.casefold(): i for n, i in cargos.items()} codigos_de_cargo = dict(liga.execute(sa.text('select id, codigo from cargo')).fetchall()) centros = dict(liga.execute(sa.text('select id, codigo from centro_de_custo')).fetchall()) grupos: dict[tuple, int] = {} grupo_da_pessoa: dict[int, int] = {} for pessoa_id, centro_id, cargo, classificacao in pessoas: cargo_id = por_nome[(' '.join((cargo or '').split()) or '(sem cargo)').casefold()] chave = (centro_id, cargo_id, classificacao) if chave not in grupos: codigo = f'{centros[centro_id]}-{codigos_de_cargo[cargo_id]}-{SUFIXO[classificacao]}' liga.execute( sa.text('insert into grupo_mao_de_obra (codigo, centro_de_custo_id, cargo_id, classificacao) values (:c, :ce, :ca, :cl)'), {'c': codigo, 'ce': centro_id, 'ca': cargo_id, 'cl': classificacao}, ) grupos[chave] = liga.execute(sa.text('select id from grupo_mao_de_obra where codigo = :c'), {'c': codigo}).scalar() grupo_da_pessoa[pessoa_id] = grupos[chave] liga.execute(sa.text('update funcionario set cargo_id = :ca, grupo_id = :g where id = :i'), {'ca': cargo_id, 'g': grupos[chave], 'i': pessoa_id}) # --- alocação: funcionario_id -> grupo_id with op.batch_alter_table('roteiro_mao_de_obra') as lote: lote.add_column(sa.Column('grupo_id', sa.Integer(), nullable=True)) linhas = liga.execute(sa.text( 'select id, produto_id, atividade_id, direcionador_id, funcionario_id, tempo_horas from roteiro_mao_de_obra order by id' )).fetchall() mantidas: dict[tuple, int] = {} for linha_id, produto_id, atividade_id, direcionador_id, funcionario_id, horas in linhas: grupo_id = grupo_da_pessoa[funcionario_id] chave = (produto_id, atividade_id, direcionador_id, grupo_id) if chave in mantidas: # outra linha do mesmo grupo: soma as horas e descarta esta liga.execute(sa.text('update roteiro_mao_de_obra set tempo_horas = tempo_horas + :h where id = :i'), {'h': horas, 'i': mantidas[chave]}) liga.execute(sa.text('delete from roteiro_mao_de_obra where id = :i'), {'i': linha_id}) else: mantidas[chave] = linha_id liga.execute(sa.text('update roteiro_mao_de_obra set grupo_id = :g where id = :i'), {'g': grupo_id, 'i': linha_id}) with op.batch_alter_table('roteiro_mao_de_obra') as lote: lote.drop_column('funcionario_id') lote.alter_column('grupo_id', existing_type=sa.Integer(), nullable=False) lote.create_foreign_key('fk_roteiro_mao_de_obra_grupo', 'grupo_mao_de_obra', ['grupo_id'], ['id']) # --- histórico das linhas de mão de obra (valores em JSON) historico = sa.table('historico_versao', sa.column('id', sa.Integer), sa.column('entidade', sa.String), sa.column('valores', sa.JSON)) for linha_id, valores in liga.execute(sa.select(historico.c.id, historico.c.valores).where(historico.c.entidade == 'roteiro_mao_de_obra')).fetchall(): novos = dict(valores or {}) if 'funcionario_id' in novos: novos['grupo_id'] = grupo_da_pessoa.get(novos.pop('funcionario_id')) liga.execute(historico.update().where(historico.c.id == linha_id).values(valores=novos)) # --- rateio por Centro de Resultado op.create_table( 'cenario_grupo_desligado', sa.Column('cenario_id', sa.Integer(), sa.ForeignKey('cenario_precificacao.id'), primary_key=True), sa.Column('grupo_id', sa.Integer(), sa.ForeignKey('grupo_mao_de_obra.id'), primary_key=True), ) membros: dict[int, set[int]] = {} for pessoa_id, grupo_id in grupo_da_pessoa.items(): membros.setdefault(grupo_id, set()).add(pessoa_id) desligados: dict[int, set[int]] = {} for cenario_id, pessoa_id in liga.execute(sa.text('select cenario_id, funcionario_id from cenario_funcionario_desligado')).fetchall(): desligados.setdefault(cenario_id, set()).add(pessoa_id) for cenario_id, pessoas_off in desligados.items(): for grupo_id, ids in membros.items(): if ids <= pessoas_off: liga.execute(sa.text('insert into cenario_grupo_desligado (cenario_id, grupo_id) values (:c, :g)'), {'c': cenario_id, 'g': grupo_id}) op.drop_table('cenario_funcionario_desligado') def downgrade() -> None: if _no_schema_comum(): return liga = op.get_bind() primeiro = dict(liga.execute(sa.text('select grupo_id, min(id) from funcionario where grupo_id is not null group by grupo_id')).fetchall()) op.create_table( 'cenario_funcionario_desligado', sa.Column('cenario_id', sa.Integer(), sa.ForeignKey('cenario_precificacao.id'), primary_key=True), sa.Column('funcionario_id', sa.Integer(), sa.ForeignKey('funcionario.id'), primary_key=True), ) for cenario_id, grupo_id in liga.execute(sa.text('select cenario_id, grupo_id from cenario_grupo_desligado')).fetchall(): for (pessoa_id,) in liga.execute(sa.text('select id from funcionario where grupo_id = :g'), {'g': grupo_id}).fetchall(): liga.execute(sa.text('insert into cenario_funcionario_desligado (cenario_id, funcionario_id) values (:c, :f)'), {'c': cenario_id, 'f': pessoa_id}) op.drop_table('cenario_grupo_desligado') with op.batch_alter_table('roteiro_mao_de_obra') as lote: lote.add_column(sa.Column('funcionario_id', sa.Integer(), nullable=True)) historico = sa.table('historico_versao', sa.column('id', sa.Integer), sa.column('entidade', sa.String), sa.column('valores', sa.JSON)) for linha_id, valores in liga.execute(sa.select(historico.c.id, historico.c.valores).where(historico.c.entidade == 'roteiro_mao_de_obra')).fetchall(): novos = dict(valores or {}) if 'grupo_id' in novos: novos['funcionario_id'] = primeiro.get(novos.pop('grupo_id')) liga.execute(historico.update().where(historico.c.id == linha_id).values(valores=novos)) for linha_id, grupo_id in liga.execute(sa.text('select id, grupo_id from roteiro_mao_de_obra')).fetchall(): if grupo_id in primeiro: liga.execute(sa.text('update roteiro_mao_de_obra set funcionario_id = :f where id = :i'), {'f': primeiro[grupo_id], 'i': linha_id}) else: liga.execute(sa.text('delete from roteiro_mao_de_obra where id = :i'), {'i': linha_id}) with op.batch_alter_table('roteiro_mao_de_obra') as lote: lote.drop_constraint('fk_roteiro_mao_de_obra_grupo', type_='foreignkey') lote.drop_column('grupo_id') lote.alter_column('funcionario_id', existing_type=sa.Integer(), nullable=False) lote.create_foreign_key('fk_roteiro_mao_de_obra_funcionario', 'funcionario', ['funcionario_id'], ['id']) with op.batch_alter_table('funcionario') as lote: lote.drop_constraint('fk_funcionario_grupo', type_='foreignkey') lote.drop_constraint('fk_funcionario_cargo', type_='foreignkey') lote.drop_column('grupo_id') lote.drop_column('cargo_id') op.drop_table('grupo_mao_de_obra') op.drop_table('cargo')