""" History/diff rendering for the restock_diff (price / stock availability) processor. A text diff is meaningless for restock watches - each history snapshot is just a short string like "In Stock: True - Price: 12.34". So instead of a diff, this renders the whole timeline as a simple smoothed line graph of price over time, coloured green where the item was in stock and red where it was out of stock. Conforms to processors.difference_base.DifferenceRenderer (module-level render()), and also provides an optional get_data() hook served as JSON by /diff//processor-data so the (potentially long) timeline stays out of the rendered HTML - same rationale as the preview asset endpoint. """ import re import statistics import time from flask_babel import gettext from loguru import logger # Snapshot format is written by processor.py: f"In Stock: {in_stock} - Price: {price}" _RE_PRICE = re.compile(r"Price:\s*([\d.]+)", re.IGNORECASE) _RE_INSTOCK = re.compile(r"In Stock:\s*(True|False)", re.IGNORECASE) def _parse_restock_snapshot(text): """Parse a snapshot string into (price: float|None, in_stock: bool|None).""" price = None in_stock = None if text: m = _RE_PRICE.search(text) if m: try: price = float(m.group(1)) except (TypeError, ValueError): price = None mi = _RE_INSTOCK.search(text) if mi: in_stock = mi.group(1).lower() == 'true' return price, in_stock def _currency(watch): try: return (watch.get('restock') or {}).get('currency') or '' except Exception: return '' def _build_series(watch): """Read the full history into a [{timestamp, price, in_stock}] timeline (oldest -> newest).""" series = [] for ts in list(watch.history.keys()): try: snapshot = watch.get_history_snapshot(timestamp=ts) except Exception as e: logger.error(f"Restock diff: unable to read snapshot {ts} for {watch.get('uuid')}: {e}") continue price, in_stock = _parse_restock_snapshot(snapshot) series.append({'timestamp': int(ts), 'price': price, 'in_stock': in_stock}) return series def compute_price_summary(series): """Compact price stats for the graph overlays + (future) watchlist deal-score badge. This is the single source of truth for "is the current price low/typical/high". It works off an already-built series (cheap, no extra I/O) so the same function can be reused at check time to cache the result on the watch and avoid scanning history at render time. Returns None when there are no prices. """ prices = [p['price'] for p in series if p.get('price') is not None] if not prices: return None count = len(prices) current = prices[-1] # series is oldest -> newest, so last priced value ordered = sorted(prices) mn, mx = ordered[0], ordered[-1] avg = sum(prices) / count if count >= 2: try: q = statistics.quantiles(prices, n=4, method='inclusive') # [p25, p50, p75] p25, median, p75 = q[0], q[1], q[2] except statistics.StatisticsError: median, p25, p75 = statistics.median(prices), mn, mx else: median = p25 = p75 = current # Two directional shares so the UI can phrase it naturally per status: # cheaper_than_pct = share of records MORE expensive than now (good when price is low) # pricier_than_pct = share of records CHEAPER than now (relevant when price is high) cheaper_than_pct = round(100.0 * sum(1 for x in prices if x > current) / count) pricier_than_pct = round(100.0 * sum(1 for x in prices if x < current) / count) if current <= p25: status = 'low' elif current >= p75: status = 'high' else: status = 'typical' r2 = lambda v: round(v, 2) return { 'count': count, 'min': r2(mn), 'max': r2(mx), 'avg': r2(avg), 'median': r2(median), 'p25': r2(p25), 'p75': r2(p75), 'current': r2(current), 'cheaper_than_pct': cheaper_than_pct, 'pricier_than_pct': pricier_than_pct, 'status': status, 'all_time_low': current <= mn, } def export_xlsx(watch, datastore): """Build an .xlsx of the full price/stock history. Returns (bytes, filename). xlsx (not csv) so prices stay real numbers and dates stay real dates in the user's spreadsheet, regardless of locale number formatting. Columns: Date, Stock status, Price, Currency. Served by /diff//processor-export.xlsx. """ import datetime from io import BytesIO from openpyxl import Workbook series = _build_series(watch) currency = _currency(watch) wb = Workbook() ws = wb.active ws.title = 'Price history' ws.append([gettext('Date'), gettext('Stock status'), gettext('Price'), gettext('Currency')]) for p in series: when = datetime.datetime.fromtimestamp(p['timestamp']) if p['in_stock'] is None: stock = '' else: stock = gettext('In stock') if p['in_stock'] else gettext('Out of stock') cell_date = ws.cell(row=ws.max_row + 1, column=1, value=when) cell_date.number_format = 'YYYY-MM-DD HH:MM:SS' ws.cell(row=ws.max_row, column=2, value=stock) # Real numeric price (None -> blank cell) so spreadsheet number formats apply. ws.cell(row=ws.max_row, column=3, value=p['price']) ws.cell(row=ws.max_row, column=4, value=currency) # Sensible column widths. for col, width in {'A': 20, 'B': 14, 'C': 12, 'D': 10}.items(): ws.column_dimensions[col].width = width out = BytesIO() wb.save(out) return out.getvalue(), f"price-history-{watch.get('uuid')}.xlsx" def get_data(watch, datastore, request): """JSON payload for the price/stock graph, fetched via /diff//processor-data. Keeps the full timeline out of the HTML page.""" series = _build_series(watch) priced = sum(1 for p in series if p['price'] is not None) logger.info(f"Restock diff get_data for {watch.get('uuid')}: {len(series)} snapshots, {priced} with a price") return { 'series': series, 'currency': _currency(watch), 'summary': compute_price_summary(series), } def render(watch, datastore, request, url_for, render_template, flash, redirect, extract_form=None): """Render the price/stock timeline page (shell + summary). The graph data is loaded asynchronously from get_data() so this stays light regardless of history length.""" uuid = watch.get('uuid') dates = list(watch.history.keys()) # Light render: read only the most recent snapshot for the summary badge/price. latest = None if dates: try: price, in_stock = _parse_restock_snapshot(watch.get_history_snapshot(timestamp=dates[-1])) latest = {'timestamp': int(dates[-1]), 'price': price, 'in_stock': in_stock} except Exception as e: logger.error(f"Restock diff: unable to read latest snapshot for {uuid}: {e}") # Opening the history page counts as viewing it (mirrors the text diff page). datastore.set_last_viewed(uuid, time.time()) return render_template( 'restock_diff/difference.html', uuid=uuid, watch=watch, current_diff_url=watch['url'], extra_title=f" - {watch.label} - {gettext('Price history')}", last_error=watch['last_error'], screenshot=watch.get_screenshot(), last_error_screenshot=watch.get_error_snapshot(), last_error_text=watch.get_error_text(), versions=dates, from_version=str(dates[-2]) if len(dates) >= 2 else (str(dates[-1]) if dates else ''), to_version=str(dates[-1]) if dates else '', restock_latest=latest, restock_currency=_currency(watch), has_enough_history=len(dates) >= 2, processor_data_url=url_for('ui.ui_diff.diff_history_page_processor_data', uuid=uuid), )