Al-HUWAITI Shell
Al-huwaiti


Server : Apache
System : Linux dedi-14684855.grupobig.com 5.14.0-611.49.1.el9_7.x86_64 #1 SMP PREEMPT_DYNAMIC Tue Apr 21 16:39:08 EDT 2026 x86_64
User : grupo692 ( 1004)
PHP Version : 8.2.31
Disable Function : NONE
Directory :  /opt/dash_backend_new/app/

Upload File :
current_dir [ Writeable ] document_root [ Writeable ]

 

Current File : //opt/dash_backend_new/app/sandbox_data.py
# app/sandbox_data.py
"""
Modo SANDBOX (empresa "test"):

- Nunca conecta no Oracle.
- Gera dados fictícios com a MESMA "cara" dos endpoints reais:
  - /vendas/unidades
  - /vendas/grupos
  - /dashboard/resumo
  - /unidades/{id}/projecao

Regras:
- Determinístico (bom para QA): para o mesmo período/unidades, retorna sempre os mesmos números.
- Coerente: ticket médio = total/qtd; YoY baseado em fator; metas relacionadas a vendas.

⚠️ Segurança:
- Não coloque credenciais reais aqui.
- Tudo é controlado por variáveis de ambiente.
"""

from __future__ import annotations

import os
import hashlib
from datetime import date, timedelta
from typing import Any, Dict, List, Optional, Tuple


def _env(name: str, default: str = "") -> str:
    v = os.getenv(name)
    return default if v is None else v


def is_sandbox_empresa(empresa: str) -> bool:
    return (empresa or "").strip().lower() == _env("SANDBOX_EMPRESA", "test").strip().lower()


def _parse_int_list(csv: str) -> List[int]:
    if not csv:
        return []
    out: List[int] = []
    for p in csv.split(","):
        p = p.strip()
        if not p:
            continue
        try:
            out.append(int(p))
        except ValueError:
            continue
    return out


def sandbox_unidades() -> List[int]:
    return _parse_int_list(_env("SANDBOX_UNIDADES", "2,3,4,9"))


def _parse_unidade_nomes(raw: str) -> Dict[int, str]:
    """
    Formato:
      "1=Loja 01,2=Loja 02,3=Loja 03"
    """
    out: Dict[int, str] = {}
    if not raw:
        return out
    parts = [p.strip() for p in raw.split(",") if p.strip()]
    for part in parts:
        if "=" not in part:
            continue
        k, v = part.split("=", 1)
        k = k.strip()
        v = v.strip()
        if not k or not v:
            continue
        try:
            out[int(k)] = v
        except ValueError:
            continue
    return out


def sandbox_unidade_nome_map() -> Dict[int, str]:
    return _parse_unidade_nomes(_env("SANDBOX_UNIDADE_NOMES", ""))


def sandbox_login_ok(usuario: int, senha: str) -> Tuple[bool, Dict[str, Any]]:
    """
    Valida login sandbox via env vars:
      SANDBOX_PASSWORD
      SANDBOX_ALLOWED_USERS (csv)
      SANDBOX_NOME
      SANDBOX_ADMIN
      SANDBOX_UNIDADES
    """
    senha_ok = senha == _env("SANDBOX_PASSWORD", "test")
    allowed = _parse_int_list(_env("SANDBOX_ALLOWED_USERS", "1"))
    user_ok = (usuario in allowed) if allowed else True

    if not (senha_ok and user_ok):
        return False, {}

    unidades = sandbox_unidades()
    nome = _env("SANDBOX_NOME", "Usuário Sandbox")
    admin = _env("SANDBOX_ADMIN", "N")

    return True, {
        "usuario_id": usuario,
        "nome": nome,
        "admin": admin,
        # no PROD este campo costuma ser string "1, 2, 3"
        "unidades": ", ".join(str(u) for u in unidades),
        "unidades_list": unidades,
    }


# -----------------------------------------------------------------------------
# Geradores determinísticos
# -----------------------------------------------------------------------------
def _seed(*parts: Any) -> int:
    """
    Cria seed determinística baseada em args (período/unidade etc).
    """
    h = hashlib.sha256("|".join(map(str, parts)).encode("utf-8")).hexdigest()
    return int(h[:12], 16)


def _days_inclusive(start: date, end: date) -> int:
    if end < start:
        return 0
    return (end - start).days + 1


def _clamp(v: float, lo: float, hi: float) -> float:
    return max(lo, min(hi, v))


def vendas_por_unidade_fake(start_date: date, end_date: date, unidades: Optional[List[int]] = None) -> List[Dict[str, Any]]:
    """
    Retorna lista com campos:
      cod_und, unidade, qntd_pedidos, valor_total, mta_meta, mta_big_meta,
      tkt_medio, mta_meta_tkt_medio, valor_total_ano_anterior, crescimento_pct
    """
    unidades = unidades or sandbox_unidades()
    nomes = sandbox_unidade_nome_map()

    dias = max(1, _days_inclusive(start_date, end_date))
    base_period = _seed("unidades", start_date.isoformat(), end_date.isoformat())

    items: List[Dict[str, Any]] = []
    for u in unidades:
        s = _seed(base_period, u)

        # qtd pedidos cresce com dias e com unidade
        qtd = int(20 + (s % 50) + dias * (1 + (u % 3)))
        qtd = max(0, qtd)

        # ticket médio: 60..220
        tkt = 60.0 + ((s // 7) % 160)
        # total
        total = round(qtd * tkt, 2)

        # metas: relacionadas ao total do período (para ficar "real")
        meta_mes = round(total * (1.10 + ((s % 20) / 100.0)), 2)  # 110%..129%
        big_meta = round(meta_mes * (1.05 + (((s // 11) % 10) / 100.0)), 2)

        # meta de ticket: 70..200
        meta_tkt = round(70.0 + ((s // 13) % 130), 2)

        # YoY: ano anterior como 85%..115% do atual
        yoy_factor = 0.85 + (((s // 17) % 31) / 100.0)
        total_yoy = round(total * yoy_factor, 2)
        if total_yoy > 0:
            crescimento = round(((total - total_yoy) / total_yoy) * 100.0, 2)
        else:
            crescimento = 0.0

        items.append(
            {
                "cod_und": str(u),
                "unidade": nomes.get(u, f"UNIDADE {u}"),
                "qntd_pedidos": int(qtd),
                "valor_total": float(total),
                "mta_meta": float(meta_mes),
                "mta_big_meta": float(big_meta),
                "tkt_medio": float(round((total / qtd), 2)) if qtd > 0 else 0.0,
                "mta_meta_tkt_medio": float(meta_tkt),
                "valor_total_ano_anterior": float(total_yoy),
                "crescimento_pct": float(crescimento),
            }
        )

    # mesma ordenação do SQL (valor_total desc)
    items.sort(key=lambda x: float(x.get("valor_total", 0.0)), reverse=True)
    return items


def vendas_por_grupo_fake(start_date: date, end_date: date) -> List[Dict[str, Any]]:
    """
    Retorna lista com campos:
      grupo, valor
    """
    s = _seed("grupos", start_date.isoformat(), end_date.isoformat())
    # "cara" parecida: strings curtas tipo "001 - ALIM"
    grupos = [
        ("001 - ALIMENTOS", 1.0),
        ("002 - BAZAR", 0.72),
        ("003 - BEBIDAS", 0.95),
        ("004 - LIMPEZA", 0.64),
        ("005 - HIGIENE", 0.58),
        ("006 - PADARIA", 0.48),
        ("007 - ACOUGUE", 0.55),
    ]

    # total do período base (coerente com vendas por unidade)
    dias = max(1, _days_inclusive(start_date, end_date))
    total_base = 50000 + (s % 90000) + dias * 1200

    # distribui pelos pesos
    weights = []
    for i, (_, w) in enumerate(grupos):
        wobble = 0.85 + (((s // (19 + i)) % 31) / 100.0)  # 0.85..1.15
        weights.append(w * wobble)
    sw = sum(weights) or 1.0

    out: List[Dict[str, Any]] = []
    for (name, _), w in zip(grupos, weights):
        val = round((total_base * (w / sw)), 2)
        out.append({"grupo": name[:15], "valor": float(val)})

    # ordena como SQL (VALOR desc)
    out.sort(key=lambda x: float(x["valor"]), reverse=True)
    return out


def dashboard_resumo_fake(start_date: date, end_date: date, unidades: Optional[List[int]] = None) -> Dict[str, Any]:
    """
    Retorna campos do DashboardResumoResponse.
    Mantém coerência com /vendas/unidades.
    """
    unidades = unidades or sandbox_unidades()
    items = vendas_por_unidade_fake(start_date, end_date, unidades)

    total_vendas = round(sum(float(i["valor_total"]) for i in items), 2)
    qtd_vendas = int(sum(int(i["qntd_pedidos"]) for i in items))
    ticket_medio = round((total_vendas / qtd_vendas), 2) if qtd_vendas > 0 else 0.0

    # meta do mês (soma das metas por unidade)
    meta_mes = round(sum(float(i["mta_meta"]) for i in items), 2)
    percentual_meta = round((total_vendas / meta_mes) * 100.0, 2) if meta_mes > 0 else 0.0

    # projeção: usa mês corrente do start_date (como no PROD)
    projecao_start = date(start_date.year, start_date.month, 1)
    # projeção_end = último dia do mês
    # cálculo simples sem calendar: pega 1º do mês seguinte -1
    if start_date.month == 12:
        next_month = date(start_date.year + 1, 1, 1)
    else:
        next_month = date(start_date.year, start_date.month + 1, 1)
    projecao_end = next_month - timedelta(days=1)

    # valor_liq do mês até end_date (clamp)
    dias_mes = max(1, (projecao_end - projecao_start).days + 1)
    dias_passados = max(1, (min(end_date, projecao_end) - projecao_start).days + 1)
    projecao_valor_liq = total_vendas
    # projeta linearmente
    projecao_total = round((projecao_valor_liq / dias_passados) * dias_mes, 2)
    projecao_total_pct = round((projecao_total / meta_mes) * 100.0, 2) if meta_mes > 0 else 0.0

    # denom meta/peso (mantemos campos do PROD)
    projecao_denom_meta_peso = float(meta_mes) if meta_mes > 0 else 0.0

    # YoY agregado
    total_vendas_yoy = round(sum(float(i.get("valor_total_ano_anterior") or 0.0) for i in items), 2)
    # qty yoy aproximada pelo fator
    s = _seed("dashyoy", start_date.isoformat(), end_date.isoformat())
    qty_factor = 0.90 + ((s % 21) / 100.0)  # 0.90..1.10
    qtd_vendas_yoy = int(max(0, round(qtd_vendas * qty_factor)))
    ticket_medio_yoy = round((total_vendas_yoy / qtd_vendas_yoy), 2) if qtd_vendas_yoy > 0 else 0.0

    crescimento_yoy_pct = round(((total_vendas - total_vendas_yoy) / total_vendas_yoy) * 100.0, 2) if total_vendas_yoy > 0 else 0.0
    qtd_vendas_yoy_pct = round(((qtd_vendas - qtd_vendas_yoy) / qtd_vendas_yoy) * 100.0, 2) if qtd_vendas_yoy > 0 else 0.0
    ticket_medio_yoy_pct = round(((ticket_medio - ticket_medio_yoy) / ticket_medio_yoy) * 100.0, 2) if ticket_medio_yoy > 0 else 0.0

    # período YoY (mesmas datas - 1 ano, com ajuste simples para 29/02)
    def shift_year(d: date, years: int) -> date:
        try:
            return d.replace(year=d.year + years)
        except ValueError:
            # 29/02 -> 28/02
            return d.replace(month=2, day=28, year=d.year + years)

    start_date_yoy = shift_year(start_date, -1)
    end_date_yoy = shift_year(end_date, -1)

    return {
        "total_vendas": float(total_vendas),
        "qtd_vendas": int(qtd_vendas),
        "ticket_medio": float(ticket_medio),
        "percentual_meta": float(percentual_meta),
        "meta_mes": float(meta_mes),
        "projecao_total": float(projecao_total),
        "projecao_total_pct": float(projecao_total_pct),
        "projecao_start": projecao_start,
        "projecao_end": projecao_end,
        "projecao_valor_liq": float(projecao_valor_liq),
        "projecao_denom_meta_peso": float(projecao_denom_meta_peso),
        "total_vendas_yoy": float(total_vendas_yoy),
        "qtd_vendas_yoy": int(qtd_vendas_yoy),
        "ticket_medio_yoy": float(ticket_medio_yoy),
        "crescimento_yoy_pct": float(crescimento_yoy_pct),
        "qtd_vendas_yoy_pct": float(qtd_vendas_yoy_pct),
        "ticket_medio_yoy_pct": float(ticket_medio_yoy_pct),
        "start_date": start_date,
        "end_date": end_date,
        "start_date_yoy": start_date_yoy,
        "end_date_yoy": end_date_yoy,
    }


def projecao_unidade_fake(unidade_id: int, start_date: date, month_start: date, month_end: date) -> Dict[str, Any]:
    """
    Retorna campos do ProjecaoResponse.
    month_start/end já calculados pela regra real (mantemos a regra no endpoint).
    """
    s = _seed("proj", unidade_id, start_date.isoformat(), month_start.isoformat(), month_end.isoformat())
    # valor_liq do mês até month_end
    dias = max(1, (month_end - month_start).days + 1)
    qtd_base = 40000 + (s % 60000)
    valor_liq = round((qtd_base / 30.0) * dias, 2)

    # meta e peso coerentes
    meta = round(valor_liq * (1.15 + ((s % 21) / 100.0)), 2)
    peso = round(0.85 + (((s // 9) % 31) / 100.0), 2)  # 0.85..1.16

    denom = meta * peso
    if denom <= 0:
        proj = 0.0
        proj_pct = 0.0
    else:
        proj = round(valor_liq / denom, 4)
        proj_pct = round(proj * 100.0, 2)

    return {
        "unidade_id": int(unidade_id),
        "month_start": month_start,
        "month_end": month_end,
        "valor_liq": float(valor_liq),
        "meta": float(meta),
        "peso": float(peso),
        "projecao": float(proj),
        "projecao_pct": float(proj_pct),
    }

Al-HUWAITI Shell