Compare commits
38 Commits
508f8b984f
...
main
| Author | SHA1 | Date | |
|---|---|---|---|
| a9561d299a | |||
| f204930d0e | |||
| 9d874486ae | |||
| 835de8359e | |||
| 5386355c7f | |||
| 532dad0ce5 | |||
| b078ecca06 | |||
| 5cf1958333 | |||
| bff20cbd72 | |||
| d1482056c0 | |||
| 458ff6f89f | |||
| 1b8619f95e | |||
| 9ee21faba6 | |||
| f697200d13 | |||
| 4d5a88145d | |||
| d69441779b | |||
| 83ea8b2ac5 | |||
| 022dbc35c5 | |||
| 4919087148 | |||
| e1828a8c80 | |||
| 674209a994 | |||
| 2157e8a015 | |||
|
|
be238e8e48 | ||
|
|
81acf82312 | ||
|
|
a974acc623 | ||
| 74cdc8693e | |||
|
|
f5d76bd92c | ||
|
|
22a6a56a71 | ||
|
|
32db283e16 | ||
| de36aacfd1 | |||
|
|
29a3e91a01 | ||
|
|
3e97d453b9 | ||
| 2959ae5ab4 | |||
| 3faa808b9d | |||
| 0a88ed2683 | |||
| 5ac871dcb5 | |||
|
|
7b695b1fda | ||
|
|
0b9a543ae2 |
7
Api/bill_example.py
Normal file
7
Api/bill_example.py
Normal file
@@ -0,0 +1,7 @@
|
|||||||
|
from formats import BillPrinter
|
||||||
|
from tools import load_json
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
data = load_json('./test/fixtures/bill.json')
|
||||||
|
printer = BillPrinter("")
|
||||||
|
printer.print_bill(data, "")
|
||||||
7
Api/customer_order_example.py
Normal file
7
Api/customer_order_example.py
Normal file
@@ -0,0 +1,7 @@
|
|||||||
|
from formats import CustomerOrderPrinter
|
||||||
|
from tools import load_json
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
data = load_json('./test/fixtures/customer_order.json')
|
||||||
|
printer = CustomerOrderPrinter("")
|
||||||
|
printer.print_order(data, "")
|
||||||
256
Api/formats.py
Normal file
256
Api/formats.py
Normal file
@@ -0,0 +1,256 @@
|
|||||||
|
#!/usr/bin/env python3
|
||||||
|
import sys
|
||||||
|
import tempfile
|
||||||
|
from datetime import datetime
|
||||||
|
from typing import Dict, Any, Optional
|
||||||
|
|
||||||
|
import pytz
|
||||||
|
from qr_generator import QRCodeGenerator
|
||||||
|
from printer_factory import PrinterFactory
|
||||||
|
|
||||||
|
|
||||||
|
class TicketPrinter:
|
||||||
|
"""Clase principal para manejar la impresión de tickets"""
|
||||||
|
|
||||||
|
def __init__(self, address: Optional[str] = None):
|
||||||
|
self.address = address
|
||||||
|
self.printer = self._get_printer()
|
||||||
|
|
||||||
|
def _get_printer(self):
|
||||||
|
"""Obtiene la instancia de impresora según la configuración"""
|
||||||
|
if self.address:
|
||||||
|
printer = PrinterFactory(
|
||||||
|
type_="network", host=self.address)._get_printer()
|
||||||
|
printer.open()
|
||||||
|
return printer
|
||||||
|
else:
|
||||||
|
return PrinterFactory(type_="dummy")._get_printer()
|
||||||
|
|
||||||
|
def _close_printer(self):
|
||||||
|
"""Cierra la conexión con la impresora si es necesaria"""
|
||||||
|
if self.address and hasattr(self.printer, 'close'):
|
||||||
|
self.printer.close()
|
||||||
|
|
||||||
|
def _get_current_time_bogota(self) -> str:
|
||||||
|
"""Obtiene la hora actual en formato America/Bogota"""
|
||||||
|
format_ = "%Y-%m-%d %H:%M:%S"
|
||||||
|
bogota_tz = pytz.timezone('America/Bogota')
|
||||||
|
return datetime.now(bogota_tz).strftime(format_)
|
||||||
|
|
||||||
|
|
||||||
|
class BillPrinter(TicketPrinter):
|
||||||
|
"""Manejador específico para facturas"""
|
||||||
|
|
||||||
|
def print_bill(self, data: Dict[str, Any], waiter: Optional[str] = None):
|
||||||
|
"""Imprime una factura"""
|
||||||
|
try:
|
||||||
|
self._print_bill_content(data, waiter)
|
||||||
|
if self.address:
|
||||||
|
self.printer.cut()
|
||||||
|
else:
|
||||||
|
self._print_dummy_output()
|
||||||
|
finally:
|
||||||
|
self._close_printer()
|
||||||
|
|
||||||
|
def _print_bill_content(self, data: Dict[str, Any], waiter: Optional[str]):
|
||||||
|
"""Genera el contenido de la factura"""
|
||||||
|
self._print_header(data)
|
||||||
|
self._print_invoice_info(data)
|
||||||
|
self._print_customer_info(data)
|
||||||
|
self._print_items(data)
|
||||||
|
self._print_totals(data)
|
||||||
|
self._print_payments(data)
|
||||||
|
self._print_qr_code(data)
|
||||||
|
self._print_footer(waiter)
|
||||||
|
|
||||||
|
def _print_header(self, data: Dict[str, Any]):
|
||||||
|
"""Imprime el encabezado de la factura"""
|
||||||
|
self.printer.set(align='center', bold=False, height=1, width=1)
|
||||||
|
self.printer.text(f"{data['shop_name']}\n")
|
||||||
|
self.printer.text(f"{data['shop_nit']}\n")
|
||||||
|
self.printer.text(f"{data['shop_address']}\n")
|
||||||
|
self._print_separator()
|
||||||
|
|
||||||
|
def _print_invoice_info(self, data: Dict[str, Any]):
|
||||||
|
"""Imprime información de facturación"""
|
||||||
|
self.printer.set(align='left', bold=False, height=1, width=1)
|
||||||
|
self.printer.textln(data['state'])
|
||||||
|
|
||||||
|
if data.get('invoice') and data['invoice'].get('resolution'):
|
||||||
|
resolution = data['invoice']['resolution']
|
||||||
|
resolution_number = resolution['resolution_number']
|
||||||
|
self.printer.textln(
|
||||||
|
f"Resolucion de Facturacion # {resolution_number}\n"
|
||||||
|
f"Valida desde {resolution['valid_date_time_from']} "
|
||||||
|
f"hasta {resolution['valid_date_time_to']}"
|
||||||
|
)
|
||||||
|
self.printer.ln()
|
||||||
|
self.printer.textln(
|
||||||
|
f"Factura #: {data['invoice']['invoice_number']}")
|
||||||
|
|
||||||
|
def _print_customer_info(self, data: Dict[str, Any]):
|
||||||
|
"""Imprime información del cliente"""
|
||||||
|
self.printer.text(f"Cliente: {data['party']}\n")
|
||||||
|
self.printer.text(f"CC/NIT: {data['tax_identifier_code']}\n")
|
||||||
|
self.printer.text(f"Direccion: {data['address']}\n")
|
||||||
|
self.printer.text(f"MESA: {data['table']}\n")
|
||||||
|
self._print_separator()
|
||||||
|
self.printer.ln()
|
||||||
|
|
||||||
|
def _print_items(self, data: Dict[str, Any]):
|
||||||
|
"""Imprime los items de la factura"""
|
||||||
|
for line in data["lines"]:
|
||||||
|
if line['type'] != 'title':
|
||||||
|
self.printer.text(line['product'])
|
||||||
|
self.printer.ln()
|
||||||
|
self.printer.text(
|
||||||
|
f"{line['quantity']} ${line['unit_price']}\n")
|
||||||
|
|
||||||
|
def _print_totals(self, data: Dict[str, Any]):
|
||||||
|
"""Imprime los totales"""
|
||||||
|
self.printer.set(align='right', bold=False, height=1, width=1)
|
||||||
|
self._print_separator()
|
||||||
|
self.printer.text(f"Descuento Realizado: {data['total_discount']}\n")
|
||||||
|
self.printer.text(f"Propina: {data['total_tip']}\n")
|
||||||
|
self.printer.text(f"Total (sin impuestos): {data['untaxed_amount']}\n")
|
||||||
|
self.printer.text(f"Impuestos (INC): {data['tax_amount']}\n")
|
||||||
|
self.printer.text(f"Total: {data['total']}\n")
|
||||||
|
self.printer.ln()
|
||||||
|
|
||||||
|
def _print_payments(self, data: Dict[str, Any]):
|
||||||
|
"""Imprime información de pagos"""
|
||||||
|
if 'payments' in data:
|
||||||
|
self.printer.textln("Pagos: ")
|
||||||
|
self._print_separator()
|
||||||
|
for payment in data['payments']:
|
||||||
|
self.printer.textln(
|
||||||
|
f"{payment['statement']} ${payment['amount']}")
|
||||||
|
|
||||||
|
def _print_qr_code(self, data: Dict[str, Any]):
|
||||||
|
"""Imprime el código QR si está disponible"""
|
||||||
|
if data.get("fe_cufe"):
|
||||||
|
self.printer.set(align='center', bold=False, height=1, width=1)
|
||||||
|
self._print_separator()
|
||||||
|
|
||||||
|
qr = QRCodeGenerator(data["fe_cufe"]).generate_qr()
|
||||||
|
with tempfile.NamedTemporaryFile(
|
||||||
|
delete=True, suffix='.png', dir='/tmp') as temp_file:
|
||||||
|
temp_filename = temp_file.name
|
||||||
|
qr.save(temp_filename)
|
||||||
|
self.printer.image(temp_filename)
|
||||||
|
|
||||||
|
self.printer.set(align='left', bold=False, height=1, width=1)
|
||||||
|
self.printer.textln(f"CUFE: {data['cufe']}")
|
||||||
|
|
||||||
|
def _print_footer(self, waiter: Optional[str]):
|
||||||
|
"""Imprime el pie de página"""
|
||||||
|
self.printer.set(align='center', bold=False, height=1, width=1)
|
||||||
|
self._print_separator()
|
||||||
|
self.printer.text("Sigue nuestras redes sociales\n")
|
||||||
|
self.printer.text("@bicipizza\n")
|
||||||
|
self.printer.text("Recuerde que la propina es voluntaria.\n")
|
||||||
|
self.printer.text("Gracias por visitarnos, vuelva pronto.\n")
|
||||||
|
self.printer.text("SOFTWARE POTENCIADO POR ONECLUSTER.ORG.\n")
|
||||||
|
self.printer.text(f"{self._get_current_time_bogota()}\n")
|
||||||
|
|
||||||
|
if waiter:
|
||||||
|
self.printer.text("Atendido Por: \n")
|
||||||
|
self.printer.text(f"{waiter}\n")
|
||||||
|
|
||||||
|
def _print_separator(self):
|
||||||
|
"""Imprime una línea separadora"""
|
||||||
|
self.printer.textln('================================================')
|
||||||
|
|
||||||
|
def _print_dummy_output(self):
|
||||||
|
"""Muestra la salida cuando no hay impresora real"""
|
||||||
|
ticket_content = self.printer.output
|
||||||
|
sys.stdout.write(ticket_content.decode('utf-8', errors='ignore'))
|
||||||
|
|
||||||
|
|
||||||
|
class CustomerOrderPrinter(TicketPrinter):
|
||||||
|
"""Manejador específico para órdenes de cliente"""
|
||||||
|
|
||||||
|
def print_order(self, data: Dict[str, Any], waiter: Optional[str] = None):
|
||||||
|
"""Imprime una orden de cliente"""
|
||||||
|
try:
|
||||||
|
self._print_order_content(data, waiter)
|
||||||
|
if self.address:
|
||||||
|
self.printer.cut()
|
||||||
|
else:
|
||||||
|
self._print_dummy_output()
|
||||||
|
finally:
|
||||||
|
self._close_printer()
|
||||||
|
|
||||||
|
def _print_order_content(
|
||||||
|
self, data: Dict[str, Any], waiter: Optional[str]):
|
||||||
|
"""Genera el contenido de la orden"""
|
||||||
|
self._print_header(waiter)
|
||||||
|
self._print_table_info(data)
|
||||||
|
self._print_order_items(data)
|
||||||
|
|
||||||
|
def _print_header(self, waiter: Optional[str]):
|
||||||
|
"""Imprime el encabezado de la orden"""
|
||||||
|
self.printer.set(align='center', bold=False, height=1, width=1)
|
||||||
|
self.printer.text(f"{self._get_current_time_bogota()}\n")
|
||||||
|
|
||||||
|
if waiter:
|
||||||
|
self.printer.text("Pedido Por: \n")
|
||||||
|
self.printer.text(f"{waiter}\n")
|
||||||
|
|
||||||
|
def _print_table_info(self, data: Dict[str, Any]):
|
||||||
|
"""Imprime información de la mesa"""
|
||||||
|
self.printer.set(
|
||||||
|
align='center', bold=False, height=2, width=2, custom_size=True)
|
||||||
|
self.printer.text('========================\n')
|
||||||
|
self.printer.text(f"MESA: {data['table']}\n")
|
||||||
|
self.printer.text('========================\n')
|
||||||
|
|
||||||
|
def _print_order_items(self, data: Dict[str, Any]):
|
||||||
|
"""Imprime los items de la orden"""
|
||||||
|
combination_pizza = False
|
||||||
|
pizza_count = 0
|
||||||
|
|
||||||
|
for line in data["lines"]:
|
||||||
|
if line['type'] != 'title':
|
||||||
|
self._set_item_format(combination_pizza, pizza_count)
|
||||||
|
|
||||||
|
text = f"{line['product']} {line['quantity']}\n"
|
||||||
|
self.printer.text(text)
|
||||||
|
|
||||||
|
if line.get('description'):
|
||||||
|
self.printer.text(f"{line['description']}\n")
|
||||||
|
|
||||||
|
if combination_pizza:
|
||||||
|
pizza_count += 1
|
||||||
|
if pizza_count >= 2:
|
||||||
|
self.printer.ln()
|
||||||
|
pizza_count = 0
|
||||||
|
combination_pizza = False
|
||||||
|
else:
|
||||||
|
self._print_combined_pizza_header()
|
||||||
|
combination_pizza = True
|
||||||
|
pizza_count = 0
|
||||||
|
|
||||||
|
def _set_item_format(self, is_combined: bool, pizza_count: int):
|
||||||
|
"""Configura el formato según el tipo de item"""
|
||||||
|
if is_combined and pizza_count < 2:
|
||||||
|
self.printer.set(
|
||||||
|
align='center',
|
||||||
|
bold=False,
|
||||||
|
height=2,
|
||||||
|
width=2,
|
||||||
|
custom_size=True)
|
||||||
|
else:
|
||||||
|
self.printer.set(
|
||||||
|
align='left', bold=False, height=2, width=2, custom_size=True)
|
||||||
|
|
||||||
|
def _print_combined_pizza_header(self):
|
||||||
|
"""Imprime el encabezado para pizza combinada"""
|
||||||
|
self.printer.set(
|
||||||
|
align='left', bold=True, height=2, width=2, custom_size=True)
|
||||||
|
self.printer.text("\nPIZZA COMBINADA\n")
|
||||||
|
|
||||||
|
def _print_dummy_output(self):
|
||||||
|
"""Muestra la salida cuando no hay impresora real"""
|
||||||
|
ticket_content = self.printer.output
|
||||||
|
sys.stdout.write(ticket_content.decode('utf-8', errors='ignore'))
|
||||||
121
Api/main.py
Normal file
121
Api/main.py
Normal file
@@ -0,0 +1,121 @@
|
|||||||
|
import json
|
||||||
|
import logging
|
||||||
|
from typing import Dict, Any, Tuple
|
||||||
|
from fastapi import FastAPI, HTTPException
|
||||||
|
from pydantic import BaseModel
|
||||||
|
from formats import BillPrinter, CustomerOrderPrinter
|
||||||
|
|
||||||
|
# Configurar logging
|
||||||
|
logging.basicConfig(level=logging.INFO)
|
||||||
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
app = FastAPI(
|
||||||
|
title="Print Server FastAPI",
|
||||||
|
description="Server that receive request for printing",
|
||||||
|
version="0.0.1"
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
class PrintRequest(BaseModel):
|
||||||
|
content: str
|
||||||
|
ip_printer: str
|
||||||
|
user_name: str
|
||||||
|
|
||||||
|
|
||||||
|
class PrintResponse(BaseModel):
|
||||||
|
message: str
|
||||||
|
success: bool
|
||||||
|
print_type: str
|
||||||
|
|
||||||
|
|
||||||
|
def _parse_request_data(info: PrintRequest) -> Tuple[Dict[str, Any], str, str]:
|
||||||
|
"""Parse and validate request data with better error handling"""
|
||||||
|
try:
|
||||||
|
content_clean = info.content.replace("'", "\"")
|
||||||
|
data = json.loads(content_clean)
|
||||||
|
|
||||||
|
logger.info(
|
||||||
|
f"Print request from {info.user_name} to {info.ip_printer}")
|
||||||
|
return data, info.ip_printer, info.user_name
|
||||||
|
|
||||||
|
except json.JSONDecodeError as e:
|
||||||
|
logger.error(f"JSON decode error: {str(e)}")
|
||||||
|
raise HTTPException(status_code=400, detail="Invalid JSON format")
|
||||||
|
except ValueError as e:
|
||||||
|
logger.error(f"Validation error: {str(e)}")
|
||||||
|
raise HTTPException(status_code=400, detail=str(e))
|
||||||
|
except Exception as e:
|
||||||
|
logger.error(f"Unexpected error parsing request: {str(e)}")
|
||||||
|
raise HTTPException(status_code=400, detail="Error processing request")
|
||||||
|
|
||||||
|
|
||||||
|
@app.post("/print_bill", response_model=PrintResponse)
|
||||||
|
def print_ticket_bill(info: PrintRequest) -> PrintResponse:
|
||||||
|
"""Print bill ticket"""
|
||||||
|
try:
|
||||||
|
data, address, waiter = _parse_request_data(info)
|
||||||
|
printer = BillPrinter(address)
|
||||||
|
printer.print_bill(data, waiter)
|
||||||
|
|
||||||
|
logger.info(f"Bill printed successfully for {waiter}")
|
||||||
|
return PrintResponse(
|
||||||
|
message="✅ Impresión de cuenta realizada exitosamente!",
|
||||||
|
success=True,
|
||||||
|
print_type="bill"
|
||||||
|
)
|
||||||
|
except Exception as e:
|
||||||
|
logger.error(f"Error printing bill: {str(e)}")
|
||||||
|
raise HTTPException(status_code=500, detail="Error printing bill")
|
||||||
|
|
||||||
|
|
||||||
|
@app.post("/order_kitchen", response_model=PrintResponse)
|
||||||
|
def print_ticket_kitchen(info: PrintRequest) -> PrintResponse:
|
||||||
|
"""Print kitchen order"""
|
||||||
|
try:
|
||||||
|
data, address, waiter = _parse_request_data(info)
|
||||||
|
printer = CustomerOrderPrinter(address)
|
||||||
|
printer.print_order(data, waiter)
|
||||||
|
|
||||||
|
logger.info(f"Kitchen order printed successfully for {waiter}")
|
||||||
|
return PrintResponse(
|
||||||
|
message="✅ Pedido de cocina impreso exitosamente!",
|
||||||
|
success=True,
|
||||||
|
print_type="kitchen"
|
||||||
|
)
|
||||||
|
except Exception as e:
|
||||||
|
logger.error(f"Error printing kitchen order: {str(e)}")
|
||||||
|
raise HTTPException(
|
||||||
|
status_code=500, detail="Error printing kitchen order")
|
||||||
|
|
||||||
|
|
||||||
|
@app.post("/order_bar", response_model=PrintResponse)
|
||||||
|
def print_ticket_bar(info: PrintRequest) -> PrintResponse:
|
||||||
|
"""Print bar order"""
|
||||||
|
try:
|
||||||
|
data, address, waiter = _parse_request_data(info)
|
||||||
|
printer = CustomerOrderPrinter(address)
|
||||||
|
printer.print_order(data, waiter)
|
||||||
|
|
||||||
|
logger.info(f"Bar order printed successfully for {waiter}")
|
||||||
|
return PrintResponse(
|
||||||
|
message="✅ Pedido de barra impreso exitosamente!",
|
||||||
|
success=True,
|
||||||
|
print_type="bar"
|
||||||
|
)
|
||||||
|
except Exception as e:
|
||||||
|
logger.error(f"Error printing bar order: {str(e)}")
|
||||||
|
raise HTTPException(status_code=500, detail="Error printing bar order")
|
||||||
|
|
||||||
|
|
||||||
|
@app.get("/health")
|
||||||
|
def health_check() -> Dict[str, str]:
|
||||||
|
return {
|
||||||
|
"status": "healthy",
|
||||||
|
"service": "print_server",
|
||||||
|
"version": "0.0.1"
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
@app.get("/")
|
||||||
|
def root() -> Dict[str, str]:
|
||||||
|
return {"message": "Print Server API is running"}
|
||||||
17
Api/printer_factory.py
Normal file
17
Api/printer_factory.py
Normal file
@@ -0,0 +1,17 @@
|
|||||||
|
from escpos.printer import Dummy, Network
|
||||||
|
|
||||||
|
|
||||||
|
class PrinterFactory:
|
||||||
|
|
||||||
|
def __init__(self, type_="dummy", host="0.0.0.0", port=9100):
|
||||||
|
self.type_ = type_
|
||||||
|
self.host = host
|
||||||
|
self.port = port
|
||||||
|
|
||||||
|
def _get_printer(self):
|
||||||
|
if self.type_ == "dummy":
|
||||||
|
return Dummy()
|
||||||
|
elif self.type_ == "network":
|
||||||
|
return Network(self.host, self.port)
|
||||||
|
else:
|
||||||
|
raise ValueError(f"Tipo de impresora no soportado: {self.type_}")
|
||||||
23
Api/qr_generator.py
Normal file
23
Api/qr_generator.py
Normal file
@@ -0,0 +1,23 @@
|
|||||||
|
#!/usr/bin/env python3
|
||||||
|
import qrcode
|
||||||
|
|
||||||
|
|
||||||
|
class QRCodeGenerator:
|
||||||
|
"""Qr Generato"""
|
||||||
|
|
||||||
|
def __init__(self, url):
|
||||||
|
self.url = url
|
||||||
|
|
||||||
|
def generate_qr(self):
|
||||||
|
"""Genera un código QR a partir de la URL"""
|
||||||
|
qr = qrcode.QRCode(
|
||||||
|
version=1,
|
||||||
|
error_correction=qrcode.constants.ERROR_CORRECT_L,
|
||||||
|
box_size=6,
|
||||||
|
border=4,
|
||||||
|
)
|
||||||
|
qr.add_data(self.url)
|
||||||
|
qr.make(fit=True)
|
||||||
|
img = qr.make_image(fill_color="black", back_color="white")
|
||||||
|
|
||||||
|
return img
|
||||||
0
Api/test/__init__.py
Normal file
0
Api/test/__init__.py
Normal file
0
Api/test/fixtures/__init__.py
vendored
Normal file
0
Api/test/fixtures/__init__.py
vendored
Normal file
35
Api/test/fixtures/bill.json
vendored
Normal file
35
Api/test/fixtures/bill.json
vendored
Normal file
@@ -0,0 +1,35 @@
|
|||||||
|
{"shop_name": "SE",
|
||||||
|
"shop_nit": "902929200",
|
||||||
|
"shop_address": "Calle Arriba 1234",
|
||||||
|
"fe_cufe": "https://catalogo-vpfe.dian.gov.co/document/searchqr?documentkey=936ef10e23f9fbae2d2c70869050b907b7af55a7d2c80a9f2e906450a8f98b20a88f7e4219fd0d02962f6e639b29ba20",
|
||||||
|
"cufe": "936ef10e23f9fbae2d2c70869050b907b7af55a7d2c80a9f2e906450a8f98b20a88f7e4219fd0d02962f6e639b29ba20",
|
||||||
|
"invoice": {
|
||||||
|
"invoice_number": "FPES2068",
|
||||||
|
"resolution": {
|
||||||
|
"resolution_number": "18764072418755",
|
||||||
|
"resolution_prefix": "FPES",
|
||||||
|
"valid_date_time_from": "2024-06-06",
|
||||||
|
"valid_date_time_to": "2026-06-06",
|
||||||
|
"from_number": 1, "to_number": 30000}
|
||||||
|
}, "party": "00-Consumidor Final",
|
||||||
|
"tax_identifier_type": "NIT",
|
||||||
|
"tax_identifier_code": "222222222",
|
||||||
|
"address": "Dg 74A #C-2-56", "city": "Medellín",
|
||||||
|
"zone": "CHIMENEA (CH)",
|
||||||
|
"table": "CH1",
|
||||||
|
"lines": [
|
||||||
|
{"type": "line", "product": "Club negra", "quantity": 1.0, "uom": "u", "unit_price": "9800.00", "taxes": "8.00%"},
|
||||||
|
{"type": "line", "product": "Club roja", "quantity": 1.0, "uom": "u", "unit_price": "9800.00", "taxes": "8.00%"},
|
||||||
|
{"type": "line", "product": "Aguila ligth", "quantity": 1.0, "uom": "u", "unit_price": "8600.00", "taxes": "8.00%"},
|
||||||
|
{"type": "line", "product": "Propinas", "quantity": 1.0, "uom": "u", "unit_price": "27333.00", "taxes": null}],
|
||||||
|
"total_discount": "0.00",
|
||||||
|
"untaxed_amount": "300666.30",
|
||||||
|
"tax_amount": "21866.67",
|
||||||
|
"total_tip": "10000",
|
||||||
|
"total": "322532.97",
|
||||||
|
"state": "CUENTA FINAL",
|
||||||
|
"payments": [
|
||||||
|
{"statement": "Transferencia TPV", "amount": "222532.00"},
|
||||||
|
{"statement": "Efectivo TPV", "amount": "100000.97"}
|
||||||
|
]
|
||||||
|
}
|
||||||
24
Api/test/fixtures/customer_order.json
vendored
Normal file
24
Api/test/fixtures/customer_order.json
vendored
Normal file
@@ -0,0 +1,24 @@
|
|||||||
|
{
|
||||||
|
"party": "00-Consumidor Final",
|
||||||
|
"tax_identifier_type": "NIT",
|
||||||
|
"tax_identifier_code": "222222222",
|
||||||
|
"address": "Dg 74A #C-2-56",
|
||||||
|
"city": "Medellín",
|
||||||
|
"zone": "CHIMENEA (CH)",
|
||||||
|
"table": "CH1",
|
||||||
|
"lines": [
|
||||||
|
{"type": "line", "product": "Club negra", "description": "", "quantity": 1.0, "uom": "Unidad"},
|
||||||
|
{"type": "line", "product": "Club roja", "description": "", "quantity": 1.0, "uom": "Unidad"},
|
||||||
|
{"type": "line", "product": "Aguila ligth", "description": "", "quantity": 1.0, "uom": "Unidad"},
|
||||||
|
{"type": "line", "product": "Té hatsu", "description": "", "quantity": 1.0, "uom": "Unidad"},
|
||||||
|
{"type": "line", "product": "Aromatica frutos rojos", "description": "", "quantity": 1.0, "uom": "Unidad"},
|
||||||
|
{"type": "title", "product": null, "description": null, "quantity": null, "uom": null},
|
||||||
|
{"type": "line", "product": "Santiago Botero", "description": "", "quantity": 0.5, "uom": "Media Unidad"},
|
||||||
|
{"type": "line", "product": "Villa Clara", "description": "", "quantity": 0.5, "uom": "Media Unidad"},
|
||||||
|
{"type": "title", "product": null, "description": null, "quantity": null, "uom": null},
|
||||||
|
{"type": "line", "product": "Granma", "description": "", "quantity": 1.0, "uom": "Media Unidad"},
|
||||||
|
{"type": "title", "product": null, "description": null, "quantity": null, "uom": null},
|
||||||
|
{"type": "line", "product": "Camilo Cienfuegos", "description": "", "quantity": 0.5, "uom": "Media Unidad"},
|
||||||
|
{"type": "line", "product": "Propinas", "description": null, "quantity": 1.0, "uom": "Unidad"}],
|
||||||
|
"deleted_lines": []
|
||||||
|
}
|
||||||
13
Api/test/fixtures/customer_order_deleted_lines.json
vendored
Normal file
13
Api/test/fixtures/customer_order_deleted_lines.json
vendored
Normal file
@@ -0,0 +1,13 @@
|
|||||||
|
{
|
||||||
|
"party": "00-Consumidor Final",
|
||||||
|
"tax_identifier_type": "NIT",
|
||||||
|
"tax_identifier_code": "222222222",
|
||||||
|
"address": "Dg 74A #C-2-56",
|
||||||
|
"city": "Medellín",
|
||||||
|
"zone": "SALON",
|
||||||
|
"table": "SL5",
|
||||||
|
"lines": [],
|
||||||
|
"deleted_lines": [
|
||||||
|
{"product": "Playa Girón", "quantity": -1.0, "unit": "Media Unidad"},
|
||||||
|
{"product": "Ciclobi", "quantity": -1.0, "unit": "Media Unidad"}
|
||||||
|
]}
|
||||||
10
Api/test/test_generate_qr.py
Normal file
10
Api/test/test_generate_qr.py
Normal file
@@ -0,0 +1,10 @@
|
|||||||
|
#!/usr/bin/env python3
|
||||||
|
from qr_generator import QRCodeGenerator
|
||||||
|
|
||||||
|
|
||||||
|
def test_generate_qr():
|
||||||
|
url = "https://www.gnu.org/"
|
||||||
|
qr_generator = QRCodeGenerator(url)
|
||||||
|
filename = qr_generator.generate_qr()
|
||||||
|
|
||||||
|
print(filename)
|
||||||
101
Api/test/test_main.py
Normal file
101
Api/test/test_main.py
Normal file
@@ -0,0 +1,101 @@
|
|||||||
|
#!/usr/bin/env python3
|
||||||
|
from fastapi.testclient import TestClient
|
||||||
|
from main import app
|
||||||
|
from tools import load_json
|
||||||
|
import json
|
||||||
|
|
||||||
|
|
||||||
|
client = TestClient(app)
|
||||||
|
|
||||||
|
|
||||||
|
def test_print_bill():
|
||||||
|
test_info = {
|
||||||
|
"content": str(
|
||||||
|
json.dumps(
|
||||||
|
load_json('test/fixtures/bill.json')
|
||||||
|
)),
|
||||||
|
"ip_printer": "",
|
||||||
|
"user_name": "Juan"
|
||||||
|
}
|
||||||
|
|
||||||
|
response = client.post("/print_bill", json=test_info)
|
||||||
|
assert response.status_code == 200
|
||||||
|
|
||||||
|
response_data = response.json()
|
||||||
|
expected_data = {
|
||||||
|
"message": "✅ Impresión de cuenta realizada exitosamente!",
|
||||||
|
"success": True,
|
||||||
|
"print_type": "bill"
|
||||||
|
}
|
||||||
|
assert response_data == expected_data
|
||||||
|
|
||||||
|
|
||||||
|
def test_print_customer_order():
|
||||||
|
test_info = {
|
||||||
|
"content": str(
|
||||||
|
json.dumps(
|
||||||
|
load_json('test/fixtures/customer_order.json')
|
||||||
|
)),
|
||||||
|
"ip_printer": "",
|
||||||
|
"user_name": "Juan"
|
||||||
|
}
|
||||||
|
|
||||||
|
response = client.post("/order_kitchen", json=test_info)
|
||||||
|
|
||||||
|
assert response.status_code == 200
|
||||||
|
|
||||||
|
response_data = response.json()
|
||||||
|
expected_data = {
|
||||||
|
"message": "✅ Pedido de cocina impreso exitosamente!",
|
||||||
|
"success": True,
|
||||||
|
"print_type": "kitchen"
|
||||||
|
}
|
||||||
|
assert response_data == expected_data
|
||||||
|
|
||||||
|
|
||||||
|
def test_print_customer_order_deleted_lines():
|
||||||
|
test_info = {
|
||||||
|
"content": str(
|
||||||
|
json.dumps(
|
||||||
|
load_json(
|
||||||
|
'test/fixtures/customer_order_deleted_lines.json')
|
||||||
|
)),
|
||||||
|
"ip_printer": "",
|
||||||
|
"user_name": "Juan"
|
||||||
|
}
|
||||||
|
|
||||||
|
response = client.post("/order_kitchen", json=test_info)
|
||||||
|
|
||||||
|
assert response.status_code == 200
|
||||||
|
|
||||||
|
response_data = response.json()
|
||||||
|
expected_data = {
|
||||||
|
"message": "✅ Pedido de cocina impreso exitosamente!",
|
||||||
|
"success": True,
|
||||||
|
"print_type": "kitchen"
|
||||||
|
}
|
||||||
|
|
||||||
|
assert response_data == expected_data
|
||||||
|
|
||||||
|
|
||||||
|
def test_print_bar_order():
|
||||||
|
test_info = {
|
||||||
|
"content": str(
|
||||||
|
json.dumps(
|
||||||
|
load_json('test/fixtures/customer_order.json')
|
||||||
|
)),
|
||||||
|
"ip_printer": "",
|
||||||
|
"user_name": "Juan"
|
||||||
|
}
|
||||||
|
|
||||||
|
response = client.post("/order_bar", json=test_info)
|
||||||
|
|
||||||
|
assert response.status_code == 200
|
||||||
|
|
||||||
|
response_data = response.json()
|
||||||
|
expected_data = {
|
||||||
|
"message": "✅ Pedido de barra impreso exitosamente!",
|
||||||
|
"success": True,
|
||||||
|
"print_type": "bar"
|
||||||
|
}
|
||||||
|
assert response_data == expected_data
|
||||||
41
Api/test/test_printerfactory.py
Normal file
41
Api/test/test_printerfactory.py
Normal file
@@ -0,0 +1,41 @@
|
|||||||
|
import unittest
|
||||||
|
from unittest.mock import patch, MagicMock
|
||||||
|
from escpos.printer import Dummy
|
||||||
|
from printer_factory import PrinterFactory
|
||||||
|
|
||||||
|
|
||||||
|
class TestPrinterFactory(unittest.TestCase):
|
||||||
|
|
||||||
|
def setUp(self):
|
||||||
|
self.factory_dummy = PrinterFactory(type_="dummy")
|
||||||
|
self.factory_network = PrinterFactory(
|
||||||
|
type_="network", host="192.168.1.100")
|
||||||
|
|
||||||
|
def test_create_dummy_printer(self):
|
||||||
|
"""Test creación de impresora dummy"""
|
||||||
|
|
||||||
|
printer = self.factory_dummy._get_printer()
|
||||||
|
self.assertIsInstance(printer, Dummy)
|
||||||
|
|
||||||
|
@patch('printer_factory.Network')
|
||||||
|
def test_create_network_printer(self, mock_network):
|
||||||
|
"""Test creación de impresora de red con mock"""
|
||||||
|
|
||||||
|
mock_printer_instance = MagicMock()
|
||||||
|
mock_network.return_value = mock_printer_instance
|
||||||
|
|
||||||
|
printer = self.factory_network._get_printer()
|
||||||
|
|
||||||
|
mock_network.assert_called_once_with("192.168.1.100", 9100)
|
||||||
|
|
||||||
|
self.assertEqual(printer, mock_printer_instance)
|
||||||
|
|
||||||
|
def test_create_unknown_printer_type(self):
|
||||||
|
"""Test para tipo de impresora desconocido"""
|
||||||
|
|
||||||
|
with self.assertRaises(ValueError):
|
||||||
|
PrinterFactory(type_="invalid_type")._get_printer()
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == '__main__':
|
||||||
|
unittest.main()
|
||||||
7
Api/tools.py
Normal file
7
Api/tools.py
Normal file
@@ -0,0 +1,7 @@
|
|||||||
|
#!/usr/bin/env python3
|
||||||
|
import json
|
||||||
|
|
||||||
|
|
||||||
|
def load_json(file_path):
|
||||||
|
with open(file_path, 'r') as filejson:
|
||||||
|
return json.load(filejson)
|
||||||
18
Dockerfile
Normal file
18
Dockerfile
Normal file
@@ -0,0 +1,18 @@
|
|||||||
|
# Usa una imagen base de Python
|
||||||
|
FROM python:3.11-slim
|
||||||
|
|
||||||
|
# Establece el directorio de trabajo
|
||||||
|
WORKDIR /app
|
||||||
|
|
||||||
|
# Copia el archivo de requerimientos y lo instala
|
||||||
|
COPY ./requirements.txt /tmp/
|
||||||
|
RUN pip install --no-cache-dir -r /tmp/requirements.txt
|
||||||
|
|
||||||
|
# Copia el código fuente
|
||||||
|
COPY . .
|
||||||
|
|
||||||
|
# Expone el puerto que usará FastAPI
|
||||||
|
EXPOSE 8000
|
||||||
|
|
||||||
|
# Comando de arranque
|
||||||
|
CMD ["uvicorn", "Api.main:app", "--host", "0.0.0.0", "--port", "8000"]
|
||||||
62
Rakefile
Normal file
62
Rakefile
Normal file
@@ -0,0 +1,62 @@
|
|||||||
|
require 'bundler/setup'
|
||||||
|
$:.unshift File.expand_path('../lib', __FILE__)
|
||||||
|
|
||||||
|
DOCKER_COMPOSE='docker-compose.yml'
|
||||||
|
|
||||||
|
desc 'entorno vivo'
|
||||||
|
namespace :live do
|
||||||
|
task :up do
|
||||||
|
compose('up', '--build', '-d', compose: DOCKER_COMPOSE)
|
||||||
|
end
|
||||||
|
|
||||||
|
desc 'monitorear salida'
|
||||||
|
task :tail do
|
||||||
|
compose('logs', '-f', 'escpos', compose: DOCKER_COMPOSE)
|
||||||
|
end
|
||||||
|
|
||||||
|
desc 'monitorear salida'
|
||||||
|
task :tail_end do
|
||||||
|
compose('logs', '-f', '-n 50', 'escpos', compose: DOCKER_COMPOSE)
|
||||||
|
end
|
||||||
|
|
||||||
|
desc 'detener entorno'
|
||||||
|
task :down do
|
||||||
|
compose('down', compose: DOCKER_COMPOSE)
|
||||||
|
end
|
||||||
|
|
||||||
|
desc 'detener entorno'
|
||||||
|
task :stop do
|
||||||
|
compose('stop', compose: DOCKER_COMPOSE)
|
||||||
|
end
|
||||||
|
|
||||||
|
desc 'eliminar entorno'
|
||||||
|
task :del do
|
||||||
|
compose('down', '-v', '--rmi', 'all', compose: DOCKER_COMPOSE)
|
||||||
|
end
|
||||||
|
|
||||||
|
desc 'reiniciar entorno'
|
||||||
|
task :restart do
|
||||||
|
compose('restart', compose: DOCKER_COMPOSE)
|
||||||
|
end
|
||||||
|
|
||||||
|
desc 'detener entorno'
|
||||||
|
task :stop do
|
||||||
|
compose('stop', compose: DOCKER_COMPOSE)
|
||||||
|
end
|
||||||
|
|
||||||
|
desc 'terminal'
|
||||||
|
task :sh do
|
||||||
|
compose('exec', 'escpos', 'bash')
|
||||||
|
end
|
||||||
|
|
||||||
|
end
|
||||||
|
|
||||||
|
desc 'iterar'
|
||||||
|
task :tdd do
|
||||||
|
compose('exec', 'escpos', "bash -c 'cd Api && flake8 *'")
|
||||||
|
compose('exec', 'escpos', "bash -c 'cd Api && pytest -vvv'")
|
||||||
|
end
|
||||||
|
|
||||||
|
def compose(*arg, compose: DOCKER_COMPOSE)
|
||||||
|
sh "docker compose -f #{compose} #{arg.join(' ')}"
|
||||||
|
end
|
||||||
95
api.py
95
api.py
@@ -1,95 +0,0 @@
|
|||||||
from fastapi import FastAPI, Response
|
|
||||||
from escpos.printer import Dummy, Network
|
|
||||||
import sys
|
|
||||||
import json
|
|
||||||
|
|
||||||
from pydantic import BaseModel
|
|
||||||
from time import sleep
|
|
||||||
from datetime import datetime
|
|
||||||
|
|
||||||
app = FastAPI(
|
|
||||||
title="Print Server FastAPI",
|
|
||||||
description="Server that receive request for printing",
|
|
||||||
version="0.0.1"
|
|
||||||
)
|
|
||||||
|
|
||||||
class Info(BaseModel):
|
|
||||||
content : str
|
|
||||||
ip_printer : str
|
|
||||||
|
|
||||||
|
|
||||||
def imprimir_ticket_de_prueba(data, address):
|
|
||||||
d = data
|
|
||||||
# Crea una instancia de la impresora ficticia
|
|
||||||
#printer = Network(str(address))
|
|
||||||
#printer.open()
|
|
||||||
printer = Dummy()
|
|
||||||
|
|
||||||
# Imprime el encabezado
|
|
||||||
printer.set(align='center', bold=False, height=1, width=1)
|
|
||||||
printer.text("Pedido Por: " + str(d['user'])+ '\n')
|
|
||||||
format_date_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
|
|
||||||
printer.text(str(format_date_time)+'\n')
|
|
||||||
printer.set(align='center', bold=False, height=2, width=2, custom_size=True)
|
|
||||||
printer.text('========================\n')
|
|
||||||
text = 'MESA: ' + str(d['table'] + "\n")
|
|
||||||
printer.text(text)
|
|
||||||
printer.text('========================\n')
|
|
||||||
|
|
||||||
printer.set(align='left', bold=False, height=6, width=6)
|
|
||||||
pizza = None
|
|
||||||
for line in d["lines"]:
|
|
||||||
if line['type'] != 'title':
|
|
||||||
if combination_pizza and pizza < 2:
|
|
||||||
printer.set(align='center', bold=False, height=2, width=2, custom_size=True)
|
|
||||||
pizza += 1
|
|
||||||
elif pizza >=2:
|
|
||||||
combination_pizza = False
|
|
||||||
printer.set(align='left', bold=False, height=2, width=2, custom_size=True)
|
|
||||||
else:
|
|
||||||
printer.set(align='left', bold=False, height=2, width=2, custom_size=True)
|
|
||||||
|
|
||||||
|
|
||||||
text = line['product'] +" "+str(line['quantity'])+"\n"
|
|
||||||
printer.text(text)
|
|
||||||
if pizza == 2:
|
|
||||||
printer.text('\n')
|
|
||||||
pizza = 0
|
|
||||||
combination_pizza = False
|
|
||||||
else:
|
|
||||||
printer.set(align='left', bold=True, height=2, width=2, custom_size=True)
|
|
||||||
printer.text("\nPIZZA COMBINADA\n")
|
|
||||||
combination_pizza = True
|
|
||||||
pizza = 0
|
|
||||||
# Corta el papel (solo para impresoras que soportan esta función)
|
|
||||||
printer.cut()
|
|
||||||
printer.close()
|
|
||||||
# Obtiene el contenido del ticket de prueba
|
|
||||||
ticket_contenido = printer.output
|
|
||||||
|
|
||||||
# Imprime el contenido en la consola
|
|
||||||
sys.stdout.write(ticket_contenido.decode('utf-8'))
|
|
||||||
|
|
||||||
@app.post("/order_kitchen")
|
|
||||||
def print_ticket_file_kitchen(info : Info):
|
|
||||||
info = dict(info)
|
|
||||||
data = info["content"]
|
|
||||||
address = info["ip_printer"]
|
|
||||||
data = json.loads(data.replace("'", "\""))
|
|
||||||
imprimir_ticket_de_prueba(data, address)
|
|
||||||
|
|
||||||
message = "!Impresión Realizada!"
|
|
||||||
|
|
||||||
return Response(content=message, status_code=200)
|
|
||||||
|
|
||||||
@app.post("/order_bar")
|
|
||||||
def print_ticket_file_bar(info : Info):
|
|
||||||
info = dict(info)
|
|
||||||
data = info["content"]
|
|
||||||
data = json.loads(data.replace("'", "\""))
|
|
||||||
address = info["ip_printer"]
|
|
||||||
imprimir_ticket_de_prueba(data, address)
|
|
||||||
|
|
||||||
message = "!Impresión Realizada!"
|
|
||||||
|
|
||||||
return Response(content=message, status_code=200)
|
|
||||||
9
docker-compose.yml
Normal file
9
docker-compose.yml
Normal file
@@ -0,0 +1,9 @@
|
|||||||
|
services:
|
||||||
|
escpos:
|
||||||
|
build: .
|
||||||
|
container_name: escpos
|
||||||
|
command: uvicorn Api.main:app --host 0.0.0.0 --port 8000 --reload
|
||||||
|
volumes:
|
||||||
|
- .:/app
|
||||||
|
ports:
|
||||||
|
- "8050:8000"
|
||||||
8
requirements.txt
Normal file
8
requirements.txt
Normal file
@@ -0,0 +1,8 @@
|
|||||||
|
fastapi
|
||||||
|
uvicorn[standard]
|
||||||
|
httpx
|
||||||
|
pytest
|
||||||
|
escpos
|
||||||
|
qrcode
|
||||||
|
flake8
|
||||||
|
pytz
|
||||||
Reference in New Issue
Block a user