203 lines
7.0 KiB
Python
203 lines
7.0 KiB
Python
import pymupdf
|
|
import os
|
|
import PIL
|
|
import openfoodfacts
|
|
import psycopg2
|
|
import datetime
|
|
|
|
|
|
from application.receipts import receipts_database
|
|
from application import database_payloads
|
|
from application.items.items_processes import postNewBlankItem
|
|
import config
|
|
|
|
def create_pdf_preview(pdf_path, output_path, size=(600, 400)):
|
|
pdf = pymupdf.open(pdf_path)
|
|
page = pdf[0]
|
|
file_name = os.path.basename(pdf_path).replace('.pdf', "")
|
|
pix = page.get_pixmap()
|
|
img = PIL.Image.frombytes("RGB", (pix.width, pix.height), pix.samples)
|
|
output_path = output_path + file_name + '.jpg'
|
|
img.thumbnail(size)
|
|
img.save(output_path)
|
|
return file_name + '.jpg'
|
|
|
|
def linkItem(site, user_id, data, conn=None):
|
|
""" this is a higher level function used to process a new item into the system,
|
|
link it to another item, and update the receipt_item to the new linked item data.
|
|
|
|
Args:
|
|
site (_type_): _description_
|
|
user_id (_type_): _description_
|
|
data (_type_): {'receipt_item_id', 'linked_list_id', 'conv_factor'}
|
|
conn (_type_, optional): Passed Connector. Defaults to None.
|
|
|
|
Returns:
|
|
_type_: _description_
|
|
"""
|
|
self_conn = False
|
|
if not conn:
|
|
database_config = config.config()
|
|
conn = psycopg2.connect(**database_config)
|
|
conn.autocommit = False
|
|
self_conn = True
|
|
|
|
# Select receipt item
|
|
receipt_item = receipts_database.selectReceiptItemsTuple(site, (data['receipt_item_id'],), conn=conn)
|
|
# select linked item
|
|
linked_list = receipts_database.getItemAllByID(site, (data['linked_list_id'],), conn=conn)
|
|
|
|
if receipt_item['type'] == 'api':
|
|
new_item_data = {
|
|
'barcode': receipt_item['barcode'],
|
|
'name': receipt_item['name'],
|
|
'subtype': 'FOOD'
|
|
}
|
|
postNewBlankItem(site, user_id, new_item_data, conn=conn)
|
|
|
|
name = receipt_item['name']
|
|
if receipt_item['name'] == "unknown":
|
|
name = linked_list['item_name']
|
|
if receipt_item['type'] == "new sku":
|
|
new_item_data = {
|
|
'barcode': receipt_item['barcode'],
|
|
'name': name,
|
|
'subtype': 'FOOD'
|
|
}
|
|
postNewBlankItem(site, user_id, new_item_data, conn=conn)
|
|
|
|
new_item = receipts_database.getItemAllByBarcode(site, (receipt_item['barcode'], ), conn=conn)
|
|
new_item = receipts_database.updateItemsTuple(site, {'id': new_item['id'], 'update':{'row_type': 'link'}}, conn=conn)
|
|
|
|
|
|
item_link = database_payloads.ItemLinkPayload(
|
|
new_item['barcode'],
|
|
linked_list['id'],
|
|
new_item,
|
|
data['conv_factor']
|
|
)
|
|
|
|
receipts_database.insertItemLinksTuple(site, item_link.payload(), conn=conn)
|
|
|
|
payload = {
|
|
'id': receipt_item['id'],
|
|
'update': {
|
|
'barcode': linked_list['barcode'],
|
|
'name': linked_list['item_name'],
|
|
'uom': linked_list['item_info']['uom']['id'],
|
|
'qty': float(receipt_item['qty']*data['conv_factor']),
|
|
'type': 'sku'
|
|
}
|
|
}
|
|
|
|
receipts_database.updateReceiptItemsTuple(site, payload, conn=conn)
|
|
|
|
if self_conn:
|
|
conn.commit()
|
|
conn.close()
|
|
return False
|
|
|
|
return conn
|
|
|
|
def postLine(site, user_id, data, conn=None):
|
|
self_conn = False
|
|
if not conn:
|
|
database_config = config.config()
|
|
conn = psycopg2.connect(**database_config)
|
|
conn.autocommit = False
|
|
self_conn = True
|
|
transaction_time = datetime.datetime.now()
|
|
receipt_item = receipts_database.selectReceiptItemsTuple(site, (data['line_id'],), conn=conn)
|
|
receipt = receipts_database.getReceiptByID(site, (receipt_item['receipt_id'], ), conn=conn)
|
|
conv_factor = 1.0
|
|
if receipt_item['data']['expires'] is not False:
|
|
expiration = datetime.datetime.strptime(receipt_item['data']['expires'], "%Y-%m-%d")
|
|
else:
|
|
expiration = None
|
|
|
|
if receipt_item['type'] == 'sku':
|
|
linked_item = receipts_database.getLinkedItemByBarcode(site, (receipt_item['barcode'], ), conn=conn)
|
|
if len(linked_item) > 1:
|
|
conv_factor = linked_item['conv_factor']
|
|
receipt_item['data']['linked_child'] = linked_item['barcode']
|
|
|
|
if receipt_item['type'] == 'api':
|
|
new_item_data = {
|
|
'barcode': receipt_item['barcode'],
|
|
'name': receipt_item['name'],
|
|
'subtype': 'FOOD'
|
|
}
|
|
postNewBlankItem(site, user_id, new_item_data, conn=conn)
|
|
|
|
if receipt_item['type'] == "new sku":
|
|
new_item_data = {
|
|
'barcode': receipt_item['barcode'],
|
|
'name': receipt_item['name'],
|
|
'subtype': 'FOOD'
|
|
}
|
|
postNewBlankItem(site, user_id, new_item_data, conn=conn)
|
|
|
|
item = receipts_database.getItemAllByBarcode(site, (receipt_item['barcode'], ), conn=conn)
|
|
|
|
location = receipts_database.selectItemLocationsTuple(site, (item['id'], item['logistics_info']['primary_location']['id']), conn=conn)
|
|
cost_layers: list = location['cost_layers']
|
|
|
|
receipt_item['data']['location'] = item['logistics_info']['primary_location']['uuid']
|
|
|
|
transaction = database_payloads.TransactionPayload(
|
|
timestamp=transaction_time,
|
|
logistics_info_id=item['logistics_info_id'],
|
|
barcode=item['barcode'],
|
|
name=item['item_name'],
|
|
transaction_type="Adjust In",
|
|
quantity=(float(receipt_item['qty'])*conv_factor),
|
|
description=f"{receipt['receipt_id']}",
|
|
user_id=user_id,
|
|
data=receipt_item['data']
|
|
)
|
|
|
|
cost_layer = database_payloads.CostLayerPayload(
|
|
aquisition_date=transaction_time,
|
|
quantity=float(receipt_item['qty']),
|
|
cost=float(receipt_item['data']['cost']),
|
|
currency_type="USD",
|
|
vendor=receipt['vendor_id'],
|
|
expires=expiration
|
|
)
|
|
|
|
cost_layer = receipts_database.insertCostLayersTuple(site, cost_layer.payload(), conn=conn)
|
|
cost_layers.append(cost_layer['id'])
|
|
|
|
quantity_on_hand = float(location['quantity_on_hand']) + float(receipt_item['qty'])
|
|
|
|
updated_item_location_payload = (cost_layers, quantity_on_hand, item['id'], item['logistics_info']['primary_location']['id'])
|
|
receipts_database.updateItemLocation(site, updated_item_location_payload, conn=conn)
|
|
|
|
|
|
site_location = receipts_database.selectLocationsTuple(site, (location['location_id'], ), conn=conn)
|
|
|
|
receipt_item['data']['location'] = site_location['uuid']
|
|
receipts_database.insertTransactionsTuple(site, transaction.payload(), conn=conn)
|
|
|
|
receipts_database.updateReceiptItemsTuple(site, {'id': receipt_item['id'], 'update': {'status': "Resolved"}}, conn=conn)
|
|
|
|
|
|
|
|
if self_conn:
|
|
conn.commit()
|
|
conn.close()
|
|
return False
|
|
|
|
return conn
|
|
|
|
# OPEN FOOD FACTS API INTEGRATION
|
|
open_food_api = openfoodfacts.API(user_agent="MyAwesomeApp/1.0")
|
|
open_food_enabled = True
|
|
|
|
def get_open_facts(barcode):
|
|
if open_food_enabled:
|
|
barcode: str = barcode.replace('%', "")
|
|
data = open_food_api.product.get(barcode)
|
|
if data != None:
|
|
return True, data
|
|
return False, {} |