114 lines
5.1 KiB
Python
114 lines
5.1 KiB
Python
"""Copy saved Ocado product photos into Grocy's native product pictures."""
|
|
import base64
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
import tomllib
|
|
from urllib.parse import quote, urlsplit
|
|
|
|
import click
|
|
import requests
|
|
|
|
from .grocy import Grocy, imported_products
|
|
from .paths import config_file, archive_directory, journal_file
|
|
from .history import private_json
|
|
from .product_metadata import read_metadata
|
|
|
|
|
|
def image_type(data):
|
|
if data.startswith(b'\xff\xd8\xff'):
|
|
return 'jpg'
|
|
if data.startswith(b'\x89PNG\r\n\x1a\n'):
|
|
return 'png'
|
|
if data.startswith(b'RIFF') and data[8:12] == b'WEBP':
|
|
return 'webp'
|
|
raise ValueError('Image response is not JPEG, PNG or WebP')
|
|
|
|
|
|
def file_path(filename):
|
|
return '/files/productpictures/' + quote(base64.b64encode(filename.encode()).decode(), safe='')
|
|
|
|
|
|
def add_picture(client, downloader, product):
|
|
if product.get('picture_file_name'):
|
|
return 'existing_picture'
|
|
metadata = read_metadata(product.get('description'))
|
|
url = metadata.get('image_url')
|
|
if not url:
|
|
return 'no_image_url'
|
|
parsed = urlsplit(url)
|
|
if parsed.scheme != 'https' or parsed.hostname not in {'www.ocado.com', 'ocado.com'}:
|
|
raise ValueError('Expected an HTTPS Ocado image URL')
|
|
# Separate session: never send the Grocy API key to the image host.
|
|
response = downloader.get(url, timeout=30, allow_redirects=False)
|
|
response.raise_for_status()
|
|
if response.status_code != 200:
|
|
raise ValueError('Image URL did not return HTTP 200')
|
|
data = response.content
|
|
extension = image_type(data)
|
|
if len(data) > 10_000_000:
|
|
raise ValueError('Image exceeds 10 MB')
|
|
filename = f"ocado-{int(product['id'])}-{hashlib.sha256(data).hexdigest()[:16]}.{extension}"
|
|
current = client.get(f"/objects/products/{product['id']}")
|
|
if not current:
|
|
return 'deleted_product'
|
|
if current.get('picture_file_name'):
|
|
return 'existing_picture'
|
|
if read_metadata(current.get('description')).get('Ocado product ID') != metadata.get('Ocado product ID'):
|
|
raise ValueError('Product identity changed')
|
|
path = file_path(filename)
|
|
uploaded = client.session.put(client.url + path, data=data,
|
|
headers={'Content-Type':'application/octet-stream'},
|
|
timeout=client.timeout, allow_redirects=False)
|
|
uploaded.raise_for_status()
|
|
if uploaded.status_code not in (200, 201, 204):
|
|
raise ValueError('Unexpected upload response')
|
|
# Check the stored file before setting the product's picture reference.
|
|
stored = client.session.get(client.url + path, timeout=client.timeout, allow_redirects=False)
|
|
stored.raise_for_status()
|
|
if stored.status_code != 200 or stored.content != data:
|
|
raise ValueError('Uploaded image verification failed')
|
|
client.request('PUT', f"/objects/products/{product['id']}", {'picture_file_name':filename})
|
|
return 'updated'
|
|
|
|
|
|
@click.command()
|
|
@click.option('--config', type=click.Path(path_type=Path), default=config_file, show_default='~/.config/ocado-grocy/config.toml')
|
|
@click.option('--report', type=click.Path(path_type=Path), default=lambda: archive_directory() / 'images/report.json', show_default='archive/images/report.json')
|
|
@click.option('--apply', is_flag=True, help='Upload and assign missing product pictures.')
|
|
@click.option('--order-id', 'order_ids', multiple=True, help='Limit to products imported from these orders.')
|
|
def main(config, report, apply, order_ids):
|
|
"""Add Ocado images to journal-linked products; keep existing pictures."""
|
|
settings = tomllib.loads(config.read_text())
|
|
client = Grocy(**settings['grocy'])
|
|
state = journal_file(settings, config)
|
|
products = imported_products(client, state, order_ids)
|
|
return run_images(client, products, report, apply=apply)
|
|
|
|
|
|
def run_images(client, products, report, *, apply=True, raise_errors=True):
|
|
downloader = requests.Session()
|
|
downloader.headers['User-Agent'] = 'ocado-grocy/0.1'
|
|
results = []
|
|
private_json(report, {'apply':apply, 'products':results})
|
|
for product in products:
|
|
row = {'product_id':product['id'], 'name':product['name']}
|
|
try:
|
|
if apply:
|
|
row['status'] = add_picture(client, downloader, product)
|
|
else:
|
|
row['status'] = 'existing_picture' if product.get('picture_file_name') else 'ready' if read_metadata(product.get('description')).get('image_url') else 'no_image_url'
|
|
except (requests.RequestException, ValueError) as exc:
|
|
row.update(status='error', error=str(exc))
|
|
results.append(row)
|
|
private_json(report, {'apply':apply, 'products':results})
|
|
click.echo(f"{product['id']} {row['status']}: {product['name']}")
|
|
counts = {s:sum(r['status']==s for r in results) for s in sorted({r['status'] for r in results})}
|
|
click.echo(json.dumps(counts))
|
|
if counts.get('error') and raise_errors:
|
|
raise click.ClickException(f"{counts['error']} image updates failed; see {report}")
|
|
return {'products':results, 'counts':counts}
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|