""" auth.py - Gerenciamento de usuários, contas e config GHL via SQLite + Flask-Login """ import os import json import secrets import sqlite3 from datetime import datetime from pathlib import Path from werkzeug.security import generate_password_hash, check_password_hash from flask_login import UserMixin # DB_PATH: por padrão o users.db ao lado deste arquivo. # Permite override via env var USERS_DB_PATH (útil pra staging/testes isolados). # Nunca mais hardcodar /opt/mia/workspace/prospeccao_ativa/users.db aqui — quebra staging. DB_PATH = Path(os.environ.get("USERS_DB_PATH") or (Path(__file__).resolve().parent / "users.db")) # Credenciais do super admin vêm do ambiente (.env). Só são usadas no seed inicial, # quando o banco ainda não tem nenhum super_admin — o login normal usa o hash no banco. SUPER_ADMIN_EMAIL = os.environ.get("SUPER_ADMIN_EMAIL", "remoraes09@gmail.com") SUPER_ADMIN_SENHA = os.environ.get("SUPER_ADMIN_SENHA", "") SUPER_ADMIN_NOME = os.environ.get("SUPER_ADMIN_NOME", "Renato Moraes") def _get_conn(): conn = sqlite3.connect(str(DB_PATH)) conn.row_factory = sqlite3.Row return conn def init_db(): conn = _get_conn() # Contas (clientes) conn.execute(""" CREATE TABLE IF NOT EXISTS contas ( id INTEGER PRIMARY KEY AUTOINCREMENT, nome TEXT NOT NULL, criado_em TEXT NOT NULL ) """) # Usuários conn.execute(""" CREATE TABLE IF NOT EXISTS users ( id INTEGER PRIMARY KEY AUTOINCREMENT, nome TEXT NOT NULL, email TEXT UNIQUE NOT NULL, password_hash TEXT NOT NULL, plano TEXT NOT NULL DEFAULT 'beta', conta_id INTEGER REFERENCES contas(id) ON DELETE SET NULL, ativo INTEGER NOT NULL DEFAULT 1, criado_em TEXT NOT NULL, ultimo_acesso TEXT ) """) # Migração: adiciona conta_id se coluna não existe ainda try: conn.execute("ALTER TABLE users ADD COLUMN conta_id INTEGER REFERENCES contas(id) ON DELETE SET NULL") conn.commit() except Exception: pass # Migração idempotente: colunas de API token para uso programático (agente externo) for ddl in ( "ALTER TABLE users ADD COLUMN api_token TEXT", "ALTER TABLE users ADD COLUMN api_allowed_ips TEXT", # JSON array de IPs, NULL = qualquer IP ): try: conn.execute(ddl) conn.commit() except Exception: pass # Config GHL por conta conn.execute(""" CREATE TABLE IF NOT EXISTS ghl_config ( conta_id INTEGER PRIMARY KEY REFERENCES contas(id) ON DELETE CASCADE, crm_type TEXT NOT NULL DEFAULT 'ghl', token TEXT NOT NULL DEFAULT '', location_id TEXT NOT NULL DEFAULT '', pipeline_id TEXT NOT NULL DEFAULT '', stage_id TEXT NOT NULL DEFAULT '', location_name TEXT NOT NULL DEFAULT '', pipeline_name TEXT NOT NULL DEFAULT '', stage_name TEXT NOT NULL DEFAULT '', atualizado_em TEXT, leads_limite INTEGER NOT NULL DEFAULT 1000, leads_usados INTEGER NOT NULL DEFAULT 0, quota_reset_em TEXT ) """) conn.commit() # Migração: adiciona campos de quota se ainda não existirem for ddl in ( "ALTER TABLE ghl_config ADD COLUMN leads_limite INTEGER NOT NULL DEFAULT 1000", "ALTER TABLE ghl_config ADD COLUMN leads_usados INTEGER NOT NULL DEFAULT 0", "ALTER TABLE ghl_config ADD COLUMN quota_reset_em TEXT", ): try: conn.execute(ddl) conn.commit() except Exception: pass # Perfis de CRM por conta (múltiplos CRMs salvos, usuário escolhe qual usar) conn.execute(""" CREATE TABLE IF NOT EXISTS crm_perfis ( id INTEGER PRIMARY KEY AUTOINCREMENT, conta_id INTEGER NOT NULL REFERENCES contas(id) ON DELETE CASCADE, nome TEXT NOT NULL, crm_type TEXT NOT NULL DEFAULT 'ghl', token TEXT NOT NULL DEFAULT '', location_id TEXT NOT NULL DEFAULT '', pipeline_id TEXT NOT NULL DEFAULT '', stage_id TEXT NOT NULL DEFAULT '', location_name TEXT NOT NULL DEFAULT '', pipeline_name TEXT NOT NULL DEFAULT '', stage_name TEXT NOT NULL DEFAULT '', connection_id TEXT NOT NULL DEFAULT '', criado_em TEXT NOT NULL ) """) conn.commit() # Migração idempotente: adiciona connection_id se não existir (bancos antigos) try: conn.execute("ALTER TABLE crm_perfis ADD COLUMN connection_id TEXT NOT NULL DEFAULT ''") conn.commit() except Exception: pass # Histórico persistente de jobs de prospecção/enriquecimento conn.execute(""" CREATE TABLE IF NOT EXISTS jobs_historico ( id TEXT PRIMARY KEY, conta_id INTEGER, user_id INTEGER, tipo TEXT DEFAULT 'prospeccao', nicho TEXT, cidade TEXT, limite INTEGER DEFAULT 0, status TEXT DEFAULT 'pendente', progresso INTEGER DEFAULT 0, total INTEGER DEFAULT 0, leads_enviados INTEGER DEFAULT 0, criado_em TEXT, concluido_em TEXT ) """) conn.commit() # Seed super_admin — só quando não existe nenhum E há senha no ambiente. # (Com o banco já provisionado, este bloco não roda; o login usa o hash existente.) cursor = conn.execute("SELECT COUNT(*) FROM users WHERE plano = 'super_admin'") if cursor.fetchone()[0] == 0: if SUPER_ADMIN_SENHA: conn.execute(""" INSERT INTO users (nome, email, password_hash, plano, ativo, criado_em) VALUES (?, ?, ?, 'super_admin', 1, ?) """, (SUPER_ADMIN_NOME, SUPER_ADMIN_EMAIL, generate_password_hash(SUPER_ADMIN_SENHA), datetime.now().isoformat())) conn.commit() print(f"[auth] Seed: super_admin criado ({SUPER_ADMIN_EMAIL})") else: print("[auth] AVISO: nenhum super_admin e SUPER_ADMIN_SENHA vazio — seed pulado.") conn.close() # ── CONTAS ────────────────────────────────────────────── def criar_conta(nome: str) -> dict: conn = _get_conn() cur = conn.execute( "INSERT INTO contas (nome, criado_em) VALUES (?, ?)", (nome.strip(), datetime.now().isoformat()) ) conn.commit() conta_id = cur.lastrowid row = conn.execute("SELECT * FROM contas WHERE id = ?", (conta_id,)).fetchone() conn.close() return dict(row) def listar_contas() -> list: conn = _get_conn() rows = conn.execute("SELECT * FROM contas ORDER BY nome").fetchall() conn.close() return [dict(r) for r in rows] def get_conta(conta_id: int) -> dict | None: conn = _get_conn() row = conn.execute("SELECT * FROM contas WHERE id = ?", (conta_id,)).fetchone() conn.close() return dict(row) if row else None def deletar_conta(conta_id: int) -> bool: conn = _get_conn() conn.execute("DELETE FROM contas WHERE id = ?", (conta_id,)) conn.commit() conn.close() return True def renomear_conta(conta_id: int, nome: str) -> bool: conn = _get_conn() conn.execute("UPDATE contas SET nome = ? WHERE id = ?", (nome.strip(), conta_id)) conn.commit() conn.close() return True # ── GHL CONFIG POR CONTA ───────────────────────────────── def get_ghl_config(conta_id: int) -> dict | None: conn = _get_conn() row = conn.execute("SELECT * FROM ghl_config WHERE conta_id = ?", (conta_id,)).fetchone() conn.close() if not row or not row["token"]: return None return dict(row) def save_ghl_config(conta_id: int, crm_type: str, token: str, location_id: str, pipeline_id: str, stage_id: str, location_name: str, pipeline_name: str, stage_name: str) -> None: conn = _get_conn() conn.execute(""" INSERT INTO ghl_config (conta_id, crm_type, token, location_id, pipeline_id, stage_id, location_name, pipeline_name, stage_name, atualizado_em) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(conta_id) DO UPDATE SET crm_type=excluded.crm_type, token=excluded.token, location_id=excluded.location_id, pipeline_id=excluded.pipeline_id, stage_id=excluded.stage_id, location_name=excluded.location_name, pipeline_name=excluded.pipeline_name, stage_name=excluded.stage_name, atualizado_em=excluded.atualizado_em """, (conta_id, crm_type, token, location_id, pipeline_id, stage_id, location_name, pipeline_name, stage_name, datetime.now().isoformat())) conn.commit() conn.close() # ── PERFIS DE CRM (múltiplos CRMs por conta) ───────────── def listar_perfis_crm(conta_id: int) -> list: """Retorna todos os perfis de CRM de uma conta, ordem alfabética por nome.""" conn = _get_conn() rows = conn.execute( "SELECT * FROM crm_perfis WHERE conta_id = ? ORDER BY nome COLLATE NOCASE", (conta_id,) ).fetchall() conn.close() return [dict(r) for r in rows] def get_perfil_crm(perfil_id: int) -> dict | None: """Retorna um perfil de CRM pelo id.""" conn = _get_conn() row = conn.execute("SELECT * FROM crm_perfis WHERE id = ?", (perfil_id,)).fetchone() conn.close() return dict(row) if row else None def salvar_perfil_crm(conta_id: int, nome: str, crm_type: str, token: str, location_id: str, pipeline_id: str, stage_id: str, location_name: str, pipeline_name: str, stage_name: str, connection_id: str = '', perfil_id: int | None = None) -> dict: """Cria (perfil_id=None) ou atualiza um perfil de CRM. Retorna o perfil salvo.""" conn = _get_conn() if perfil_id: conn.execute(""" UPDATE crm_perfis SET nome = ?, crm_type = ?, token = ?, location_id = ?, pipeline_id = ?, stage_id = ?, location_name = ?, pipeline_name = ?, stage_name = ?, connection_id = ? WHERE id = ? AND conta_id = ? """, (nome.strip(), crm_type, token, location_id, pipeline_id, stage_id, location_name, pipeline_name, stage_name, connection_id, perfil_id, conta_id)) conn.commit() new_id = perfil_id else: cur = conn.execute(""" INSERT INTO crm_perfis (conta_id, nome, crm_type, token, location_id, pipeline_id, stage_id, location_name, pipeline_name, stage_name, connection_id, criado_em) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, (conta_id, nome.strip(), crm_type, token, location_id, pipeline_id, stage_id, location_name, pipeline_name, stage_name, connection_id, datetime.now().isoformat())) conn.commit() new_id = cur.lastrowid row = conn.execute("SELECT * FROM crm_perfis WHERE id = ?", (new_id,)).fetchone() conn.close() return dict(row) if row else {} def deletar_perfil_crm(perfil_id: int) -> bool: """Deleta um perfil pelo id. Retorna True mesmo se não existir.""" conn = _get_conn() conn.execute("DELETE FROM crm_perfis WHERE id = ?", (perfil_id,)) conn.commit() conn.close() return True def contar_perfis_crm(conta_id: int) -> int: """Retorna a quantidade de perfis de CRM de uma conta.""" conn = _get_conn() row = conn.execute( "SELECT COUNT(*) as n FROM crm_perfis WHERE conta_id = ?", (conta_id,) ).fetchone() conn.close() return int(row["n"]) if row else 0 def migrar_ghl_config_para_perfis(): """Migração one-shot: para cada conta com ghl_config e sem perfis, cria um perfil. Nomes especiais: conta 1 -> "Smile", conta 3 -> "Climb Digital". Para outras contas usa location_name (ou "CRM Principal" se vazio). Idempotente: pula contas que já têm ao menos um perfil.""" nomes_especiais = {1: "Smile", 3: "Climb Digital"} conn = _get_conn() rows = conn.execute("SELECT * FROM ghl_config WHERE token != ''").fetchall() criados = 0 for cfg in rows: conta_id = cfg["conta_id"] existentes = conn.execute( "SELECT COUNT(*) as n FROM crm_perfis WHERE conta_id = ?", (conta_id,) ).fetchone() if existentes and existentes["n"] > 0: continue nome = nomes_especiais.get(conta_id) or (cfg["location_name"] or "").strip() or "CRM Principal" conn.execute(""" INSERT INTO crm_perfis (conta_id, nome, crm_type, token, location_id, pipeline_id, stage_id, location_name, pipeline_name, stage_name, connection_id, criado_em) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( conta_id, nome, cfg["crm_type"] or "ghl", cfg["token"] or "", cfg["location_id"] or "", cfg["pipeline_id"] or "", cfg["stage_id"] or "", cfg["location_name"] or "", cfg["pipeline_name"] or "", cfg["stage_name"] or "", "", # ghl_config não tem connection_id datetime.now().isoformat() )) criados += 1 conn.commit() conn.close() return criados # ── QUOTA DE LEADS POR CONTA ───────────────────────────── def _garantir_linha_quota(conn, conta_id: int): """Garante que existe uma linha em ghl_config para a conta (com defaults de quota).""" row = conn.execute( "SELECT conta_id FROM ghl_config WHERE conta_id = ?", (conta_id,) ).fetchone() if not row: conn.execute(""" INSERT INTO ghl_config (conta_id, leads_limite, leads_usados, quota_reset_em) VALUES (?, 1000, 0, ?) """, (conta_id, datetime.now().isoformat())) conn.commit() def get_quota(conta_id: int) -> dict: """Retorna {limite, usados, disponivel, reset_em} para a conta.""" conn = _get_conn() _garantir_linha_quota(conn, conta_id) row = conn.execute( "SELECT leads_limite, leads_usados, quota_reset_em FROM ghl_config WHERE conta_id = ?", (conta_id,) ).fetchone() conn.close() if not row: return {"limite": 1000, "usados": 0, "disponivel": 1000, "reset_em": None} limite = int(row["leads_limite"] or 0) usados = int(row["leads_usados"] or 0) return { "limite": limite, "usados": usados, "disponivel": max(0, limite - usados), "reset_em": row["quota_reset_em"], } def incrementar_quota(conta_id: int, quantidade: int) -> dict: """Soma 'quantidade' ao contador de leads usados da conta.""" if quantidade <= 0: return get_quota(conta_id) conn = _get_conn() _garantir_linha_quota(conn, conta_id) conn.execute( "UPDATE ghl_config SET leads_usados = leads_usados + ? WHERE conta_id = ?", (int(quantidade), conta_id) ) conn.commit() conn.close() return get_quota(conta_id) def resetar_quota_se_novo_mes(conta_id: int) -> dict: """Zera leads_usados se quota_reset_em estiver em mês anterior ao atual.""" conn = _get_conn() _garantir_linha_quota(conn, conta_id) row = conn.execute( "SELECT quota_reset_em FROM ghl_config WHERE conta_id = ?", (conta_id,) ).fetchone() agora = datetime.now() precisa_resetar = False reset_em_atual = row["quota_reset_em"] if row else None if not reset_em_atual: precisa_resetar = True else: try: ult = datetime.fromisoformat(reset_em_atual) if (ult.year, ult.month) != (agora.year, agora.month): precisa_resetar = True except Exception: precisa_resetar = True if precisa_resetar: novo_reset = datetime(agora.year, agora.month, 1).isoformat() conn.execute( "UPDATE ghl_config SET leads_usados = 0, quota_reset_em = ? WHERE conta_id = ?", (novo_reset, conta_id) ) conn.commit() conn.close() return get_quota(conta_id) # ── USUÁRIOS ───────────────────────────────────────────── class User(UserMixin): def __init__(self, row): self.id = str(row["id"]) self.nome = row["nome"] self.email = row["email"] self.password_hash = row["password_hash"] self.plano = row["plano"] self.conta_id = row["conta_id"] self.ativo = bool(row["ativo"]) self.criado_em = row["criado_em"] self.ultimo_acesso = row["ultimo_acesso"] # Campos de API token (podem não existir em bancos antigos antes da migração) try: self.api_token = row["api_token"] if "api_token" in row.keys() else None except Exception: self.api_token = None try: self.api_allowed_ips = row["api_allowed_ips"] if "api_allowed_ips" in row.keys() else None except Exception: self.api_allowed_ips = None @property def is_super_admin(self): return self.plano == "super_admin" @property def is_conta_admin(self): return self.plano in ("super_admin", "conta_admin") @staticmethod def get(user_id): conn = _get_conn() row = conn.execute("SELECT * FROM users WHERE id = ?", (user_id,)).fetchone() conn.close() return User(row) if row else None def is_active(self): return self.ativo def buscar_por_email(email: str): conn = _get_conn() row = conn.execute("SELECT * FROM users WHERE email = ?", (email.strip().lower(),)).fetchone() conn.close() return User(row) if row else None def criar_usuario(nome: str, email: str, senha: str, plano: str = "beta", conta_id: int = None) -> tuple: email = email.strip().lower() if buscar_por_email(email): return None, "E-mail já cadastrado." if not nome or not email or not senha: return None, "Nome, e-mail e senha são obrigatórios." try: conn = _get_conn() conn.execute(""" INSERT INTO users (nome, email, password_hash, plano, conta_id, ativo, criado_em) VALUES (?, ?, ?, ?, ?, 1, ?) """, (nome.strip(), email, generate_password_hash(senha), plano, conta_id, datetime.now().isoformat())) conn.commit() conn.close() return buscar_por_email(email), None except Exception as e: return None, str(e) def listar_usuarios(conta_id: int = None) -> list: conn = _get_conn() if conta_id is not None: rows = conn.execute( "SELECT * FROM users WHERE conta_id = ? ORDER BY criado_em DESC", (conta_id,) ).fetchall() else: rows = conn.execute("SELECT * FROM users ORDER BY criado_em DESC").fetchall() conn.close() return [User(r) for r in rows] def atualizar_usuario(user_id: int, **kwargs) -> bool: campos_permitidos = {"nome", "email", "plano", "ativo", "ultimo_acesso", "conta_id"} updates = {k: v for k, v in kwargs.items() if k in campos_permitidos} if not updates: return False set_clause = ", ".join(f"{k} = ?" for k in updates) values = list(updates.values()) + [user_id] conn = _get_conn() conn.execute(f"UPDATE users SET {set_clause} WHERE id = ?", values) conn.commit() conn.close() return True def deletar_usuario(user_id: int) -> bool: conn = _get_conn() conn.execute("DELETE FROM users WHERE id = ?", (user_id,)) conn.commit() conn.close() return True def verificar_senha(user: User, senha: str) -> bool: return check_password_hash(user.password_hash, senha) def registrar_acesso(user_id: int): atualizar_usuario(user_id, ultimo_acesso=datetime.now().isoformat()) # ── API TOKEN (uso programático por agentes externos) ──── def gerar_api_token(user_id: int) -> str: """Gera e salva novo token para o usuário. Retorna o token.""" token = secrets.token_hex(32) # 64 chars hex conn = _get_conn() conn.execute("UPDATE users SET api_token = ? WHERE id = ?", (token, user_id)) conn.commit() conn.close() return token def buscar_por_token(token: str): """Retorna User se token válido e ativo, None caso contrário.""" if not token: return None conn = _get_conn() try: row = conn.execute( "SELECT * FROM users WHERE api_token = ? AND ativo = 1", (token,) ).fetchone() except Exception: row = None conn.close() return User(row) if row else None def set_ip_whitelist(user_id: int, ips: list) -> None: """Define lista de IPs permitidos (vazia/None = qualquer IP).""" conn = _get_conn() conn.execute( "UPDATE users SET api_allowed_ips = ? WHERE id = ?", (json.dumps(ips) if ips else None, user_id) ) conn.commit() conn.close() def revogar_api_token(user_id: int) -> None: """Zera o api_token do usuário.""" conn = _get_conn() conn.execute("UPDATE users SET api_token = NULL WHERE id = ?", (user_id,)) conn.commit() conn.close() # ── HISTÓRICO PERSISTENTE DE JOBS ──────────────────────── def salvar_job_historico(job: dict): """Persiste metadados do job no banco. Chama ao criar e ao concluir.""" conn = _get_conn() conn.execute(""" INSERT INTO jobs_historico (id, conta_id, user_id, tipo, nicho, cidade, limite, status, progresso, total, leads_enviados, criado_em, concluido_em) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(id) DO UPDATE SET status=excluded.status, progresso=excluded.progresso, total=excluded.total, leads_enviados=excluded.leads_enviados, concluido_em=excluded.concluido_em """, ( job.get("id"), job.get("conta_id"), job.get("user_id"), job.get("tipo", "prospeccao"), job.get("nicho", ""), job.get("cidade", ""), job.get("limite", 0), job.get("status", "pendente"), job.get("progresso", 0), job.get("total", 0), job.get("leads_enviados", 0), job.get("criado_em"), job.get("concluido_em") )) conn.commit() conn.close() def listar_jobs_historico(conta_id: int = None, limit: int = 10) -> list: """Retorna os últimos jobs do histórico persistido.""" conn = _get_conn() if conta_id: rows = conn.execute( "SELECT * FROM jobs_historico WHERE conta_id=? ORDER BY criado_em DESC LIMIT ?", (conta_id, limit) ).fetchall() else: rows = conn.execute( "SELECT * FROM jobs_historico ORDER BY criado_em DESC LIMIT ?", (limit,) ).fetchall() conn.close() return [dict(r) for r in rows]