201 lines
13 KiB
Python
Executable File
201 lines
13 KiB
Python
Executable File
import os, re, json, random, hashlib, logging, threading
|
|
from datetime import datetime, date
|
|
from decimal import Decimal, InvalidOperation
|
|
from zoneinfo import ZoneInfo
|
|
import psycopg
|
|
from psycopg.rows import dict_row
|
|
import requests
|
|
from fastapi import FastAPI, BackgroundTasks
|
|
from apscheduler.schedulers.background import BackgroundScheduler
|
|
|
|
logging.basicConfig(level=os.getenv('LOG_LEVEL', 'INFO'))
|
|
log = logging.getLogger('mealplanner-importer')
|
|
DB = dict(host=os.getenv('DB_HOST','db'), port=int(os.getenv('DB_PORT','5432')), dbname=os.getenv('DB_NAME','mealplanner'), user=os.getenv('DB_USER','mealplanner'), password=os.environ['DB_PASSWORD'], connect_timeout=8)
|
|
POSTAL_CODE = re.sub(r'\s+','',os.getenv('FLIPP_POSTAL_CODE','J7R6S4')).upper()
|
|
LOCALE = os.getenv('FLIPP_LOCALE','fr')
|
|
FLIPP_FLYERS_URL = 'https://flyers-ng.flippback.com/api/flipp/data'
|
|
FLIPP_ITEMS_URL = 'https://flyers-ng.flippback.com/api/flipp/flyers/{}/flyer_items'
|
|
FLIPP_HOME = 'https://www.flipp.com/'
|
|
app = FastAPI(title='MealPlanner Flipp Importer')
|
|
scheduler = BackgroundScheduler(timezone=ZoneInfo('America/Toronto'))
|
|
session = requests.Session()
|
|
session.headers.update({'User-Agent':'Mozilla/5.0 (compatible; MealPlanner-personal/1.0)','Accept':'application/json'})
|
|
run_lock = threading.Lock()
|
|
|
|
def db():
|
|
return psycopg.connect(**DB, row_factory=dict_row)
|
|
|
|
def ensure_tables():
|
|
# Tables Flipp distinctes : les anciennes tables PDF et les autres données restent intactes.
|
|
with db() as conn:
|
|
conn.execute("CREATE TABLE IF NOT EXISTS flipp_flyers (id SERIAL PRIMARY KEY, store_id INTEGER NOT NULL REFERENCES stores(id) ON DELETE CASCADE, external_flyer_id BIGINT NOT NULL, title TEXT NOT NULL, merchant TEXT NOT NULL, postal_code TEXT NOT NULL, valid_from DATE, valid_to DATE, thumbnail_url TEXT, categories JSONB, imported_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), UNIQUE(store_id, external_flyer_id))")
|
|
conn.execute("CREATE TABLE IF NOT EXISTS flipp_offers (id SERIAL PRIMARY KEY, flyer_id INTEGER NOT NULL REFERENCES flipp_flyers(id) ON DELETE CASCADE, store_id INTEGER NOT NULL REFERENCES stores(id) ON DELETE CASCADE, external_item_id TEXT, item_name TEXT NOT NULL, brand TEXT, price NUMERIC(10,2) NOT NULL, price_text TEXT, valid_from DATE, valid_to DATE, image_url TEXT, offer_key TEXT NOT NULL UNIQUE, imported_at TIMESTAMPTZ NOT NULL DEFAULT NOW())")
|
|
conn.execute("ALTER TABLE flipp_offers ADD COLUMN IF NOT EXISTS price_unit TEXT NOT NULL DEFAULT 'unknown'")
|
|
conn.execute("ALTER TABLE flipp_offers ADD COLUMN IF NOT EXISTS product_cut TEXT")
|
|
conn.execute('CREATE INDEX IF NOT EXISTS flipp_offers_valid_to_idx ON flipp_offers(valid_to)')
|
|
conn.execute('CREATE INDEX IF NOT EXISTS flipp_flyers_valid_to_idx ON flipp_flyers(valid_to)')
|
|
conn.execute("CREATE TABLE IF NOT EXISTS flyer_sources (id SERIAL PRIMARY KEY, store_id INTEGER NOT NULL REFERENCES stores(id) ON DELETE CASCADE, source_url TEXT NOT NULL, active BOOLEAN NOT NULL DEFAULT TRUE, last_checked_at TIMESTAMPTZ, last_success_at TIMESTAMPTZ, last_error TEXT, UNIQUE(store_id, source_url))")
|
|
conn.execute("CREATE TABLE IF NOT EXISTS flyer_import_runs (id SERIAL PRIMARY KEY, store_id INTEGER REFERENCES stores(id) ON DELETE SET NULL, source_url TEXT NOT NULL, started_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), finished_at TIMESTAMPTZ, status TEXT NOT NULL DEFAULT 'running', message TEXT)")
|
|
|
|
def sid():
|
|
return ''.join(str(random.randint(0,9)) for _ in range(16))
|
|
|
|
def parsedate(value):
|
|
if not value: return None
|
|
try: return datetime.fromisoformat(str(value).replace('Z','+00:00')).date()
|
|
except (ValueError,TypeError):
|
|
try: return date.fromisoformat(str(value)[:10])
|
|
except (ValueError,TypeError): return None
|
|
|
|
def parsedprice(value):
|
|
if value is None or str(value).strip()=='': return None
|
|
try:
|
|
if isinstance(value,(int,float,Decimal)): return Decimal(str(value))
|
|
nums=re.findall(r'\d+(?:[.,]\d{1,2})?',str(value))
|
|
return Decimal(nums[-1].replace(',','.')) if nums else None
|
|
except InvalidOperation: return None
|
|
|
|
|
|
def pricebasis(item, price_text):
|
|
"""Interprète les unités explicites de Flipp; ne devine pas le prix au poids."""
|
|
candidates = [
|
|
item.get('unit'), item.get('price_unit'), item.get('measurement'),
|
|
item.get('size'), item.get('description'), item.get('display_name'),
|
|
price_text
|
|
]
|
|
raw = ' '.join(str(x) for x in candidates if x is not None).lower()
|
|
raw = raw.replace(' ', '').replace(',', '.')
|
|
if any(x in raw for x in ['/lb', 'perlb', 'lb.', '/livre', 'parlivre', 'pound', 'lb']):
|
|
return 'lb'
|
|
if any(x in raw for x in ['/kg', 'perkg', '/kilo', 'parkg', 'kg']):
|
|
return 'kg'
|
|
if any(x in raw for x in ['/100g', 'per100g', '/100 g']):
|
|
return '100g'
|
|
if any(x in raw for x in ['/each', 'each', 'chacun', '/unité', '/unite']):
|
|
return 'each'
|
|
# Si Flipp n'indique pas l'unité, on conserve « unité » et on ne calcule
|
|
# jamais un prix au poids supposé.
|
|
return 'unknown'
|
|
|
|
def product_cut(name):
|
|
n = str(name or '').lower()
|
|
if any(x in n for x in ['thigh', 'cuisse', 'haut de cuisse', 'pilons', 'drumstick']):
|
|
return 'cuisse'
|
|
if any(x in n for x in ['breast', 'poitrine', 'suprême', 'supreme']):
|
|
return 'poitrine'
|
|
if any(x in n for x in ['whole chicken', 'poulet entier']):
|
|
return 'entier'
|
|
return None
|
|
|
|
def getstores():
|
|
with db() as conn:
|
|
rows=conn.execute("SELECT id,name FROM stores WHERE name IN ('IGA','Metro') AND enabled=TRUE").fetchall()
|
|
return {x['name']:x['id'] for x in rows}
|
|
|
|
def sourceid(store_id):
|
|
with db() as conn:
|
|
row=conn.execute('SELECT id FROM flyer_sources WHERE store_id=%s AND source_url=%s',(store_id,FLIPP_HOME)).fetchone()
|
|
if row: return row['id']
|
|
return conn.execute('INSERT INTO flyer_sources(store_id,source_url) VALUES(%s,%s) RETURNING id',(store_id,FLIPP_HOME)).fetchone()['id']
|
|
|
|
def fetchflyers():
|
|
session_id=sid()
|
|
r=session.get(FLIPP_FLYERS_URL,params={'locale':LOCALE,'postal_code':POSTAL_CODE,'sid':session_id},timeout=30)
|
|
r.raise_for_status()
|
|
data=r.json(); flyers=data.get('flyers',[]) if isinstance(data,dict) else []
|
|
if not isinstance(flyers,list): raise RuntimeError('Réponse Flipp invalide.')
|
|
return session_id,flyers
|
|
|
|
def storefor(flyer):
|
|
merchant=str(flyer.get('merchant','')).strip().casefold()
|
|
return 'IGA' if merchant=='iga' else ('Metro' if merchant=='metro' else None)
|
|
|
|
def isgrocery(flyer):
|
|
cats=flyer.get('categories',[])
|
|
if isinstance(cats,str): cats=[x.strip() for x in cats.split(',')]
|
|
return 'Groceries' in cats
|
|
|
|
def saveflyer(store_id,flyer):
|
|
cats=flyer.get('categories',[])
|
|
if isinstance(cats,str): cats=[x.strip() for x in cats.split(',')]
|
|
title=str(flyer.get('name') or f"Circulaire {flyer.get('merchant','')}").strip()
|
|
thumb=flyer.get('thumbnail_url') or flyer.get('premium_thumbnail_url') or flyer.get('mobile_thumbnail_url')
|
|
with db() as conn:
|
|
conn.execute("INSERT INTO flipp_flyers (store_id,external_flyer_id,title,merchant,postal_code,valid_from,valid_to,thumbnail_url,categories) VALUES(%s,%s,%s,%s,%s,%s,%s,%s,%s::jsonb) ON CONFLICT(store_id,external_flyer_id) DO UPDATE SET title=EXCLUDED.title,valid_from=EXCLUDED.valid_from,valid_to=EXCLUDED.valid_to,thumbnail_url=EXCLUDED.thumbnail_url,categories=EXCLUDED.categories",(store_id,int(flyer['id']),title,flyer.get('merchant'),POSTAL_CODE,parsedate(flyer.get('valid_from')),parsedate(flyer.get('valid_to')),thumb,json.dumps(cats,ensure_ascii=False)))
|
|
return conn.execute('SELECT id FROM flipp_flyers WHERE store_id=%s AND external_flyer_id=%s',(store_id,int(flyer['id']))).fetchone()['id']
|
|
|
|
def saveitems(store_id,flyer_id,flyer,items):
|
|
kept=skipped=0
|
|
with db() as conn:
|
|
for item in items:
|
|
if not isinstance(item,dict): skipped+=1; continue
|
|
name=' '.join(str(item.get('name') or '').split()).strip()
|
|
price_text=str(item.get('price') or '').strip(); price=parsedprice(item.get('price'))
|
|
if not name or not price_text or price is None: skipped+=1; continue
|
|
itemid=item.get('id') or item.get('item_id')
|
|
start=parsedate(item.get('valid_from')) or parsedate(flyer.get('valid_from'))
|
|
end=parsedate(item.get('valid_to')) or parsedate(flyer.get('valid_to'))
|
|
brand=item.get('brand') or item.get('manufacturer')
|
|
unit=pricebasis(item, price_text)
|
|
cut=product_cut(name)
|
|
img=item.get('clean_image_url') or item.get('image_url') or item.get('thumbnail_url') or item.get('image')
|
|
if not isinstance(img,str): img=None
|
|
identity='|'.join([str(flyer.get('id')),str(itemid or ''),name.casefold(),price_text,str(start or '')])
|
|
key=hashlib.sha256(identity.encode('utf-8')).hexdigest()
|
|
conn.execute("INSERT INTO flipp_offers (flyer_id,store_id,external_item_id,item_name,brand,price,price_text,valid_from,valid_to,image_url,offer_key,price_unit,product_cut) VALUES(%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s) ON CONFLICT(offer_key) DO UPDATE SET price=EXCLUDED.price,price_text=EXCLUDED.price_text,item_name=EXCLUDED.item_name,brand=EXCLUDED.brand,valid_from=EXCLUDED.valid_from,valid_to=EXCLUDED.valid_to,image_url=EXCLUDED.image_url,flyer_id=EXCLUDED.flyer_id,price_unit=EXCLUDED.price_unit,product_cut=EXCLUDED.product_cut",(flyer_id,store_id,str(itemid) if itemid is not None else None,name,brand,price,price_text,start,end,img,key,unit,cut))
|
|
kept+=1
|
|
return kept,skipped
|
|
|
|
def run_imports():
|
|
if not run_lock.acquire(blocking=False): log.info('Import Flipp déjà en cours; exécution ignorée.'); return
|
|
try:
|
|
ensure_tables(); stores=getstores()
|
|
if not stores: raise RuntimeError('Magasins IGA et Metro absents ou désactivés.')
|
|
session_id,allflyers=fetchflyers(); selected={'IGA':[],'Metro':[]}
|
|
for f in allflyers:
|
|
target=storefor(f)
|
|
if target and isgrocery(f) and f.get('id'): selected[target].append(f)
|
|
for name in ('IGA','Metro'):
|
|
if name not in stores: log.warning('%s absent ou désactivé.',name); continue
|
|
store_id=stores[name]; src=sourceid(store_id)
|
|
with db() as conn:
|
|
run_id=conn.execute("INSERT INTO flyer_import_runs(store_id,source_url,status,message) VALUES(%s,%s,'running','Importation Flipp en cours') RETURNING id",(store_id,FLIPP_HOME)).fetchone()['id']
|
|
conn.execute('UPDATE flyer_sources SET last_checked_at=NOW(),last_error=NULL WHERE id=%s',(src,))
|
|
total=skipped=processed=0; errors=[]
|
|
for flyer in selected[name]:
|
|
try:
|
|
record=saveflyer(store_id,flyer)
|
|
r=session.get(FLIPP_ITEMS_URL.format(flyer['id']),params={'locale':LOCALE,'sid':session_id},timeout=30)
|
|
r.raise_for_status(); items=r.json()
|
|
if not isinstance(items,list): raise RuntimeError('Format articles inattendu.')
|
|
good,bad=saveitems(store_id,record,flyer,items); total+=good; skipped+=bad; processed+=1
|
|
log.info('%s — %s (ID %s): %s offres conservées, %s ignorées.',name,flyer.get('name'),flyer.get('id'),good,bad)
|
|
except Exception as exc:
|
|
errors.append(f"{flyer.get('name') or flyer.get('id')}: {exc}"); log.exception('Erreur circulaire %s pour %s',flyer.get('id'),name)
|
|
msg=f'{processed}/{len(selected[name])} circulaires traitées; {total} offres conservées; {skipped} entrées sans nom/prix ignorées.'
|
|
if errors: msg+=' Erreurs: '+' | '.join(errors[:3])
|
|
with db() as conn:
|
|
conn.execute('UPDATE flyer_sources SET last_success_at=NOW(),last_error=%s WHERE id=%s',('; '.join(errors)[:1000] if errors else None,src))
|
|
conn.execute('UPDATE flyer_import_runs SET finished_at=NOW(),status=%s,message=%s WHERE id=%s',('imported' if processed else 'error',msg,run_id))
|
|
log.info('%s — %s',name,msg)
|
|
except Exception: log.exception('Importation Flipp échouée')
|
|
finally: run_lock.release()
|
|
|
|
@app.on_event('startup')
|
|
def startup():
|
|
ensure_tables()
|
|
scheduler.add_job(run_imports,'cron',hour=6,minute=0,id='daily-flipp',replace_existing=True)
|
|
scheduler.start()
|
|
if os.getenv('RUN_ON_START','true').lower()=='true': scheduler.add_job(run_imports,'date',run_date=datetime.now(ZoneInfo('America/Toronto')),id='initial-flipp',replace_existing=True)
|
|
|
|
@app.on_event('shutdown')
|
|
def shutdown(): scheduler.shutdown(wait=False)
|
|
|
|
@app.get('/health')
|
|
def health(): return {'status':'ok','service':'flipp-importer','postal_code':POSTAL_CODE}
|
|
|
|
@app.post('/run')
|
|
def run_now(background_tasks:BackgroundTasks):
|
|
background_tasks.add_task(run_imports)
|
|
return {'status':'queued','message':f'Importation régionale IGA et Metro pour {POSTAL_CODE} lancée.'}
|