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