118 lines
3.6 KiB
Python
118 lines
3.6 KiB
Python
from dataclasses import dataclass
|
|
from decimal import Decimal, ROUND_UP
|
|
from typing import Dict, Optional, List
|
|
import logging
|
|
import scrapers
|
|
import shlex
|
|
|
|
profit_margin = Decimal('1.3')
|
|
|
|
|
|
@dataclass
|
|
class Product:
|
|
aliases: List[str]
|
|
price: Decimal
|
|
description: str
|
|
metadata: Dict[str, Optional[str]]
|
|
|
|
@staticmethod
|
|
def from_line(line: str) -> "Product":
|
|
if not line.strip():
|
|
raise Exception('line is empty')
|
|
if line.startswith('#'):
|
|
raise Exception('line is a comment')
|
|
|
|
fields = shlex.split(line)
|
|
aliases = fields[0].split(',')
|
|
price = Decimal(fields[1])
|
|
description = fields[2]
|
|
# TODO: support addons
|
|
|
|
metadata = {}
|
|
for f in fields:
|
|
if f.startswith('#'):
|
|
s = f.lstrip('#').split('=')
|
|
(k, v) = (s[0], None) if len(s) == 1 else s
|
|
metadata[k] = v
|
|
|
|
return Product(
|
|
aliases=aliases,
|
|
price=price,
|
|
description=description,
|
|
metadata=metadata,
|
|
)
|
|
|
|
def format_line(self):
|
|
aliases = ','.join(self.aliases)
|
|
price = f'{self.price:.2f}'
|
|
description = f'"{self.description}"'
|
|
metadata = ' '.join(sorted(f'#{k}' if v is None else f'#{k}={v}' for (k, v) in self.metadata.items()))
|
|
|
|
accum = ''
|
|
for col, txt in [
|
|
(30, aliases),
|
|
(37, price),
|
|
(98, description),
|
|
(0, metadata),
|
|
]:
|
|
accum += txt + ' ' + ' '*max(0, col-len(accum)-len(txt))
|
|
return accum.rstrip()
|
|
|
|
|
|
class NoAutoUpdate(Exception):
|
|
def __init__(self):
|
|
super().__init__('no auto update directive')
|
|
|
|
|
|
def find_product_details(product: Product):
|
|
if 'ah' in product.metadata:
|
|
return scrapers.ah_get_by_gtin(product.aliases[0])
|
|
if 'sligro' in product.metadata:
|
|
return scrapers.sligro_get_by_gtin(product.aliases[0])
|
|
raise NoAutoUpdate()
|
|
|
|
|
|
def update_product_pricings(src):
|
|
lines_out = []
|
|
for line in src.split('\n'):
|
|
try:
|
|
product = Product.from_line(line)
|
|
except Exception:
|
|
lines_out.append(line)
|
|
continue
|
|
|
|
try:
|
|
prod_info = find_product_details(product)
|
|
except NoAutoUpdate:
|
|
logging.debug('no auto update: "%s"', product.description)
|
|
lines_out.append(line)
|
|
continue
|
|
except scrapers.ProductNotFoundError:
|
|
logging.warn('not found "%s"', product.description)
|
|
product.metadata['err'] = 'not_found'
|
|
lines_out.append(product.format_line())
|
|
continue
|
|
except Exception as err:
|
|
logging.error('did not update "%s": %s', product.description, err)
|
|
lines_out.append(line)
|
|
continue
|
|
|
|
aliases = sorted(set(product.aliases) - {prod_info.gtin})
|
|
product.aliases = [prod_info.gtin, *aliases]
|
|
|
|
# Apply profit margin and divide by the number of units per sold packaging.
|
|
unit_price = prod_info.price * profit_margin / prod_info.units
|
|
# Round up to 5ct.
|
|
previous_price = product.price
|
|
product.price = (unit_price * 20).quantize(Decimal('1'), rounding=ROUND_UP) / 20
|
|
|
|
if 'err' in product.metadata:
|
|
del product.metadata['err']
|
|
|
|
lines_out.append(product.format_line())
|
|
|
|
logging.debug(f'Found "{prod_info.name}", buy €{prod_info.price/prod_info.units:.2f}, sell €{product.price:.2f}')
|
|
if product.price != previous_price:
|
|
logging.info(f'Adjusted "{prod_info.name}", €{previous_price:.2f} -> €{product.price:.2f}')
|
|
|
|
return '\n'.join(lines_out)
|