leihlokal_product_descriptions_fix.py

# before: https://anchr.io/i/1NQ3x.png
# after: https://anchr.io/i/iMkiK.png

# script is interactive
# it will first fetch all broken products and then ask you for confirmation before actually updating them

import re
import sys
from typing import Tuple, Set, List, Any, Dict
import requests

api_url = 'https://buergerstiftung-karlsruhe.de/wp-json/wc/v3'
api_key = ''
api_secret = ''

s = requests.Session()
s.auth = (api_key, api_secret)

def find_products_batch(page: int = 1, limit: int = 50) -> Tuple[List[Dict[str, Any]], bool]:
    products = s.get(f'{api_url}/products', params=dict(page=page, per_page=limit)).json()
    matching_products = []

    for p in products:
        try:
            sku = p['sku']
            plain_description = p['yoast_head_json']['og_description']
        except KeyError:
            continue

        match = re.search(f'^(Art\.Nr\.: {sku})\d', plain_description)
        if match is not None:
            matching_products.append(p)

    return matching_products, len(products) > 0

def find_products_all() -> List[Dict[str, Any]]:
    matching_products = []

    for i in range(1, int(1e10)):
        print(f'Fetching page {i} ...')
        result, has_more = find_products_batch(i, 50)
        if not has_more:
            break
        matching_products += result

    return matching_products

def fix_product(p: Dict[str, Any]):
    r = s.put(f'{api_url}/products/{p["id"]}', json=dict(
        short_description=p['short_description']  # when returned from api, short_description already contains properly formatted html, including line breaks, etc.
    ))
    r.raise_for_status()


if __name__ == '__main__':
    products = find_products_all()

    print(f'Got {len(products)} broken products. Fix them one by one (type "y")?')
    answer = input()

    if answer != 'y':
        print('Products to fix (by SKU):')
        print('---')
        print(list(map(lambda p: int(p['sku']), products)))
        print()
        print('Products to fix (by ID):')
        print('---')
        print(list(map(lambda p: p['id'], products)))
        sys.exit(0)

    for i, p in enumerate(products):
        print(f'Updating product ({i+1} of {len(products)}) {p["id"]} (SKU {p["sku"]}) ...')
        try:
            fix_product(p)
        except Exception as e:
            print('Failed.')
            print(e)