dtf-system/app/tiny_probe.py

"""Supervised checks against the client's real Tiny, run by hand from a
container console (docker exec, or Portainer > Containers > worker > Console).
Tiny has no sandbox, so this is how the adapter is proven on the real account.

    python -m app.tiny_probe produtos [termo]   list active products (read-only)
    python -m app.tiny_probe formas-envio [termo]  list formas de envio, to find pickup (read-only)
    python -m app.tiny_probe conferir           connection, the four product ids and pickup (read-only)
    python -m app.tiny_probe pedido --cnpj ... --email ... --celular ... [--modo file]
                                                show the test order; --confirmar creates it

The test order goes through TinyOrders.deliver, exactly as the worker sends a
paid order. The duplicate guard is then proven read-only first (the order must
be found by its purchase-order number) and only then by a second delivery,
which must return the existing order. If the search cannot find the order, the
second delivery is not attempted: it would create a duplicate. Cancel the test
order in Olist afterwards.
"""
import argparse
import json
import sys
import time
from datetime import datetime

from .core.pricing import TIERS
from .core.secrets import load as load_secret_files
from . import tiny

FIND_ATTEMPTS = 4
FIND_WAIT_SECONDS = 5


def test_order(cnpj, email, celular, mode, now=None):
    number = 'TESTE-' + (now or datetime.now(tiny.BRASILIA)).strftime('%Y%m%d%H%M')
    unit = TIERS[mode][0][1]
    return {'order_id': 'teste-integracao', 'number': number, 'event': 'payment_approved',
            'order': {'customer': {'cnpj': cnpj, 'mail': email, 'zap': celular},
                      'items': [{'mode': mode, 'grade': 100, 'billed_metres': '1', 'unit_cents': unit}],
                      'freight': {'service': 'pickup', 'total_cents': 0},
                      'destination': None, 'total_cents': unit}}


def produtos(orders, termo, out):
    params = {'situacao': 'A', 'limit': 100}
    if termo:
        params['nome'] = termo
    found = orders.request('GET', '/produtos', params=params).get('itens') or []
    for item in found:
        preco = (item.get('precos') or {}).get('preco')
        out(f"{item['id']}\t{item.get('sku') or '-'}\t{item.get('descricao')}\tR$ {preco}")
    out(f'{len(found)} produto(s) ativo(s)' + (f' com "{termo}"' if termo else '') + '.')


def formas_envio(orders, termo, out):
    params = {'limit': 100}
    if termo:
        params['nome'] = termo
    found = orders.request('GET', '/formas-envio', params=params).get('itens') or []
    for item in found:
        out(f"{item.get('id')}\t{item.get('tipo')}\t{item.get('nome')}")
    out(f'{len(found)} forma(s) de envio' + (f' com "{termo}"' if termo else '') +
        '. A retirada é a do tipo 6 (Customizado) que a equipe usa para retirar pessoalmente.')


def conferir(orders, out):
    ok = True
    for mode, setting in tiny.PRODUCT_SETTINGS.items():
        product, result = tiny.configured_product(orders, mode)
        ok &= result == 'ok'
        described = f"{product.get('sku') or '-'} · {product.get('descricao')}" if product else ''
        out(f'{setting} ({tiny.PRODUCTS[mode]}): {result} {described}'.rstrip())
    method, result = tiny.configured_pickup(orders)
    ok &= result == 'ok'
    out(f"{tiny.PICKUP_SETTING} (retirada): {result} {method.get('nome') if method else ''}".rstrip())
    out(f"{tiny.STATUS_UPDATES_SETTING}: {'ligado' if tiny.status_updates() else 'desligado'}")
    for name, result in tiny.check(orders=orders).items():
        if name in ('pedidos', 'contatos'):
            ok &= result == 'ok'
            out(f'{name}: {result}')
    out('Tudo conferido.' if ok else 'Há pendências acima.')
    return ok


def pedido(orders, payload, confirm, out, wait=time.sleep):
    cnpj = payload['order']['customer']['cnpj']
    contacts = orders.request('GET', '/contatos', params={'cpfCnpj': cnpj, 'limit': 5}).get('itens') or []
    known = [c for c in contacts if ''.join(ch for ch in str(c.get('cpfCnpj') or '') if ch.isdigit()) == cnpj]
    out(f"Contato {cnpj}: " + (f"já existe no Tiny (id {known[0]['id']})" if known
                               else 'não existe; será criado com o e-mail como nome'))
    out('Pedido que será enviado:')
    out(json.dumps(tiny.order_payload(payload, known[0]['id'] if known else 0), ensure_ascii=False, indent=2))
    if not confirm:
        out('Nada foi criado. Repita com --confirmar para criar este pedido no Tiny.')
        return None
    first = orders.deliver('teste-integracao', payload)
    out(f"1º envio: {first['status']} · Tiny id {first['tiny_id']} · nº {first['tiny_number']}")
    for attempt in range(FIND_ATTEMPTS):
        found = orders.find(payload)
        if found:
            break
        if attempt < FIND_ATTEMPTS - 1:
            wait(FIND_WAIT_SECONDS)
    else:
        out(f'FALHA: a busca por {tiny.purchase_order(payload["number"])} não encontrou o pedido. '
            'Um reenvio criaria um duplicado; o 2º envio não foi feito. Cancele o pedido no Olist.')
        return False
    second = orders.deliver('teste-integracao', payload)
    out(f"2º envio: {second['status']} · Tiny id {second['tiny_id']}")
    passed = second['status'] == 'already-created' and second['tiny_id'] == first['tiny_id']
    out(('OK: o reenvio encontrou o mesmo pedido.' if passed else 'FALHA: o reenvio não devolveu o mesmo pedido.')
        + f" Cancele o pedido nº {first['tiny_number']} no Olist.")
    return passed


def main(argv=None, orders=None, out=print):
    parser = argparse.ArgumentParser(prog='python -m app.tiny_probe')
    commands = parser.add_subparsers(dest='command', required=True)
    listing = commands.add_parser('produtos')
    listing.add_argument('termo', nargs='?', default='')
    methods = commands.add_parser('formas-envio')
    methods.add_argument('termo', nargs='?', default='')
    commands.add_parser('conferir')
    order = commands.add_parser('pedido')
    order.add_argument('--cnpj', required=True)
    order.add_argument('--email', required=True)
    order.add_argument('--celular', required=True)
    order.add_argument('--modo', choices=sorted(tiny.PRODUCT_SETTINGS), default='file')
    order.add_argument('--confirmar', action='store_true')
    args = parser.parse_args(argv)
    if orders is None:
        load_secret_files()
        orders = tiny.TinyOrders(auth=tiny.TinyAuth())
    try:
        if args.command == 'produtos':
            produtos(orders, args.termo, out)
            return 0
        if args.command == 'formas-envio':
            formas_envio(orders, args.termo, out)
            return 0
        if args.command == 'conferir':
            return 0 if conferir(orders, out) else 1
        cnpj = ''.join(ch for ch in args.cnpj if ch.isdigit())
        if len(cnpj) != 14:
            out('Informe um CNPJ com 14 dígitos.')
            return 2
        payload = test_order(cnpj, args.email, args.celular, args.modo)
        return 0 if pedido(orders, payload, args.confirmar, out) is not False else 1
    except tiny.TinyError as exc:
        out(f'Tiny: {exc}')
        return 1


if __name__ == '__main__':
    sys.exit(main())