dtf-system/app/tiny.py

"""Tiny/Olist ERP (API v3): create the sales order once a payment is approved.

Written from the public API documentation and exercised only against a fake
HTTP transport. Tiny has no sandbox: the first real test creates a real order
in the client's ERP, so test with a marked order and cancel it afterwards.

Authentication is OAuth2 on Tiny's Keycloak. The client creates an
"Aplicativo" in Tiny (Configurações > Geral > Aplicativos), which yields a
client ID and secret and registers our callback URL. An operator then clicks
"Conectar Tiny" on the Kanban once; the tokens are kept in the database and
the refresh token is rotated on every refresh, under a row lock so two
workers never spend the same one.

Orders are created by contact and product id: the customer's contact is found
by CNPJ or created, and each product mode maps to a product that must already
exist in Tiny (PRODUCT_SETTINGS).

Idempotency: the outbox may deliver the same event more than once. Every order
carries numeroOrdemCompra = "DTF-<order number>", and before creating one the
customer's recent orders are searched for that number, so a second delivery
finds the first order instead of creating another.
"""
import os
import secrets
import time
from datetime import datetime, timedelta, timezone
from decimal import Decimal
from urllib.parse import urlencode

import httpx

API = 'https://api.tiny.com.br/public-api/v3'
AUTH = 'https://accounts.tiny.com.br/realms/tiny/protocol/openid-connect'
# Not TINY_PRODUCT_<MODE>: app/core/secrets.py reads any variable ending in
# _FILE as a path to a secret file, and one product mode is called "file".
PRODUCT_SETTINGS = {'file': 'TINY_PRODUCT_TEXTIL_FOLHA', 'avulsa': 'TINY_PRODUCT_TEXTIL_AVULSA',
                    'uvfile': 'TINY_PRODUCT_UV_FOLHA', 'uv': 'TINY_PRODUCT_UV_AVULSA'}
PRODUCTS = {'file': 'DTF Têxtil 57 cm · folha montada',
            'avulsa': 'DTF Têxtil 57 cm · artes avulsas',
            'uvfile': 'DTF UV 28,5 cm · folha montada',
            'uv': 'DTF UV 28,5 cm · artes avulsas'}
# How far back to look for an order a previous delivery may already have made.
SEARCH_DAYS = 7
STATE_MINUTES = 10
# Brazil has had no daylight saving since 2019; the order date is the local day.
BRASILIA = timezone(timedelta(hours=-3))


def local_today():
    return datetime.now(BRASILIA).date()


class TinyError(Exception):
    pass


class TinyNotConnected(TinyError):
    """No authorised connection yet: an operator must click "Conectar Tiny"."""


def purchase_order(number):
    return f'DTF-{number}'


def configured():
    """Whether the OAuth application is configured (orders may still be fake)."""
    return all(os.environ.get(name) for name in ('TINY_CLIENT_ID', 'TINY_CLIENT_SECRET', 'TINY_REDIRECT_URI'))


def required_settings():
    return ['TINY_CLIENT_ID', 'TINY_CLIENT_SECRET', 'TINY_REDIRECT_URI'] + list(PRODUCT_SETTINGS.values())


# OAuth -------------------------------------------------------------------

class TinyAuth:
    """The OAuth application and the stored connection."""

    def __init__(self, connect=None, transport=None, clock=time.time):
        self.client_id = os.environ.get('TINY_CLIENT_ID', '')
        self.client_secret = os.environ.get('TINY_CLIENT_SECRET', '')
        self.redirect_uri = os.environ.get('TINY_REDIRECT_URI', '')
        if connect is None:
            from .core.db import connect
        self.connect = connect
        self.http = httpx.Client(timeout=20, transport=transport)
        self.clock = clock

    def authorize_url(self, operator):
        """Start a connection. The state is single-use, short-lived, and only an
        authenticated operator can create one, which is what protects the
        callback: Tiny's redirect back is cross-site, so the operator's
        SameSite=Strict cookie does not travel with it."""
        state = secrets.token_urlsafe(32)
        with self.connect() as c:
            c.execute("DELETE FROM dtf_local.oauth_states WHERE expires_at<now()")
            c.execute('''INSERT INTO dtf_local.oauth_states(state,provider,operator,expires_at)
                VALUES(%s,'tiny',%s,now()+%s)''', (state, operator, timedelta(minutes=STATE_MINUTES)))
        return f'{AUTH}/auth?' + urlencode({'response_type': 'code', 'client_id': self.client_id,
                                            'redirect_uri': self.redirect_uri, 'scope': 'openid',
                                            'state': state})

    def complete(self, code, state):
        """Exchange the authorisation code; returns the operator who started it."""
        with self.connect() as c:
            row = c.execute('''DELETE FROM dtf_local.oauth_states WHERE state=%s AND provider='tiny'
                AND expires_at>now() RETURNING operator''', (state,)).fetchone()
            if not row:
                raise TinyError('Unknown or expired authorisation state')
            tokens = self._token({'grant_type': 'authorization_code', 'code': code,
                                  'redirect_uri': self.redirect_uri})
            self._store(c, tokens, row['operator'])
        return row['operator']

    def status(self):
        with self.connect() as c:
            row = c.execute('''SELECT connected_by,connected_at,refresh_expires_at
                FROM dtf_local.provider_tokens WHERE provider='tiny' ''').fetchone()
        if not row:
            return {'connected': False}
        expired = row['refresh_expires_at'] and row['refresh_expires_at'] <= datetime.now(timezone.utc)
        return {'connected': not expired, 'connected_by': row['connected_by'],
                'connected_at': row['connected_at'], 'expires_at': row['refresh_expires_at']}

    def access_token(self):
        """A valid access token, refreshing (and rotating) under a row lock."""
        with self.connect() as c:
            row = c.execute('''SELECT * FROM dtf_local.provider_tokens WHERE provider='tiny'
                FOR UPDATE''').fetchone()
            if not row:
                raise TinyNotConnected('Tiny is not connected; use "Conectar Tiny" on the Kanban')
            now = datetime.now(timezone.utc)
            if row['access_expires_at'] > now + timedelta(seconds=60):
                return row['access_token']
            if row['refresh_expires_at'] and row['refresh_expires_at'] <= now:
                raise TinyNotConnected('The Tiny connection expired; connect again on the Kanban')
            tokens = self._token({'grant_type': 'refresh_token', 'refresh_token': row['refresh_token']})
            self._store(c, tokens, row['connected_by'], refreshed=True)
            return tokens['access_token']

    def _token(self, form):
        response = self.http.post(f'{AUTH}/token', data={**form, 'client_id': self.client_id,
                                                         'client_secret': self.client_secret})
        if response.status_code == 400 and form['grant_type'] == 'refresh_token':
            raise TinyNotConnected('Tiny refused the stored refresh token; connect again on the Kanban')
        response.raise_for_status()
        tokens = response.json()
        if not tokens.get('access_token') or not tokens.get('refresh_token'):
            raise TinyError('Tiny token response is missing tokens')
        return tokens

    def _store(self, c, tokens, operator, refreshed=False):
        now = datetime.now(timezone.utc)
        access_expires = now + timedelta(seconds=int(tokens.get('expires_in', 300)))
        refresh_in = tokens.get('refresh_expires_in')
        refresh_expires = now + timedelta(seconds=int(refresh_in)) if refresh_in else None
        c.execute('''INSERT INTO dtf_local.provider_tokens(provider,access_token,refresh_token,
              access_expires_at,refresh_expires_at,connected_by,connected_at,updated_at)
            VALUES('tiny',%s,%s,%s,%s,%s,now(),now())
            ON CONFLICT(provider) DO UPDATE SET access_token=EXCLUDED.access_token,
              refresh_token=EXCLUDED.refresh_token, access_expires_at=EXCLUDED.access_expires_at,
              refresh_expires_at=EXCLUDED.refresh_expires_at, updated_at=now(),
              connected_by=CASE WHEN %s THEN dtf_local.provider_tokens.connected_by ELSE EXCLUDED.connected_by END,
              connected_at=CASE WHEN %s THEN dtf_local.provider_tokens.connected_at ELSE now() END''',
                  (tokens['access_token'], tokens['refresh_token'], access_expires, refresh_expires,
                   operator, refreshed, refreshed))


# Orders ------------------------------------------------------------------

def money(cents):
    return float(Decimal(cents) / 100)


def contact_payload(order):
    customer = order['customer']
    destination = order.get('destination')
    contact = {'nome': (destination or {}).get('recipient') or customer['mail'],
               'tipoPessoa': 'J', 'cpfCnpj': customer['cnpj'], 'email': customer['mail'],
               'celular': customer['zap'], 'situacao': 'A'}
    if destination:
        contact['endereco'] = address(destination)
    return contact


def address(destination):
    return {'endereco': destination['street'], 'numero': destination['number'],
            'complemento': destination.get('complement', ''), 'bairro': destination['district'],
            'municipio': destination['city'], 'cep': destination['postal_code'],
            'uf': destination['state'], 'pais': 'Brasil'}


def order_payload(payload, contact_id, today=None):
    """The v3 'pedido' for a paid order's approved snapshot."""
    order = payload['order']
    items = []
    for item in order['items']:
        setting = PRODUCT_SETTINGS[item['mode']]
        product = os.environ.get(setting, '')
        if not product.isdigit():
            raise TinyError(f'{setting} must be the Tiny product id')
        items.append({'produto': {'id': int(product)},
                      'quantidade': float(Decimal(item['billed_metres'])),
                      'valorUnitario': money(item['unit_cents']),
                      'infoAdicional': PRODUCTS[item['mode']] + f" · nota {item['grade']}"})
    freight = order['freight']
    pickup = freight.get('service') == 'pickup'
    pedido = {'data': (today or local_today()).isoformat(),
              'idContato': contact_id,
              'numeroOrdemCompra': purchase_order(payload['number']),
              'itens': items,
              'valorFrete': money(freight['total_cents']),
              'observacoes': f"Pedido DTF #{payload['number']}" + (' · retirada em Franca' if pickup else ''),
              'observacoesInternas': f"Site DTF · pago · {payload['order_id']}"}
    destination = order.get('destination')
    if destination:
        pedido['enderecoEntrega'] = {**address(destination), 'enderecoNro': destination['number'],
                                     'nomeDestinatario': destination['recipient']}
        del pedido['enderecoEntrega']['numero']
    ecommerce = os.environ.get('TINY_ECOMMERCE_ID', '')
    if ecommerce.isdigit():
        pedido['ecommerce'] = {'id': int(ecommerce), 'numeroPedidoEcommerce': purchase_order(payload['number'])}
    return pedido


class TinyOrders:
    def __init__(self, auth=None, transport=None, today=None):
        missing = [name for name in required_settings() if not os.environ.get(name)]
        if auth is None and missing:
            raise RuntimeError('Tiny needs ' + ', '.join(missing))
        self.auth = auth or TinyAuth()
        self.http = httpx.Client(base_url=API, transport=transport, timeout=30)
        self.today = today

    def request(self, method, path, **kwargs):
        headers = {'Authorization': f'Bearer {self.auth.access_token()}'}
        response = self.http.request(method, path, headers=headers, **kwargs)
        if response.status_code == 429:
            raise TinyError('Tiny rate limit reached; the outbox will retry')
        if response.status_code >= 400:
            raise TinyError(f'{method} {path}: {response.status_code} {response.text[:300]}')
        return response.json() if response.content else {}

    def contact(self, order):
        cnpj = order['customer']['cnpj']
        found = self.request('GET', '/contatos', params={'cpfCnpj': cnpj, 'limit': 5})
        for entry in found.get('itens') or []:
            if ''.join(ch for ch in str(entry.get('cpfCnpj') or '') if ch.isdigit()) == cnpj:
                return entry['id']
        return self.request('POST', '/contatos', json=contact_payload(order))['id']

    def find(self, payload):
        """An order a previous delivery already created, or None."""
        wanted = purchase_order(payload['number'])
        since = ((self.today or local_today()) - timedelta(days=SEARCH_DAYS)).isoformat()
        found = self.request('GET', '/pedidos', params={'cpfCnpj': payload['order']['customer']['cnpj'],
                                                        'dataInicial': since, 'limit': 100})
        for entry in found.get('itens') or []:
            detail = self.request('GET', f"/pedidos/{entry['id']}")
            if detail.get('numeroOrdemCompra') == wanted:
                return detail
        return None

    def deliver(self, event_key, payload):
        if payload.get('event') != 'payment_approved':
            # Production progress is not written to Tiny; only the sale is.
            return {'provider': 'tiny', 'event_key': event_key, 'status': 'not-applicable',
                    'event': payload.get('event')}
        existing = self.find(payload)
        if existing:
            return {'provider': 'tiny', 'event_key': event_key, 'status': 'already-created',
                    'tiny_id': str(existing.get('id')), 'tiny_number': str(existing.get('numeroPedido'))}
        contact_id = self.contact(payload['order'])
        created = self.request('POST', '/pedidos', json=order_payload(payload, contact_id, self.today))
        return {'provider': 'tiny', 'event_key': event_key, 'status': 'created',
                'tiny_id': str(created.get('id')), 'tiny_number': str(created.get('numeroPedido'))}