99 lines
6.5 KiB
Python
Executable File
99 lines
6.5 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Propose and apply purchasing items from the sibling ledger receipt archive."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse, csv, datetime as dt, json, re
|
|
import xml.etree.ElementTree as ET
|
|
from pathlib import Path
|
|
from zipfile import ZipFile
|
|
import yaml
|
|
|
|
ROOT = Path(__file__).resolve().parent.parent
|
|
LEDGER = ROOT.parent / "ledger"
|
|
NS = {"m": "http://schemas.openxmlformats.org/spreadsheetml/2006/main"}
|
|
FOOD_WORDS = set("flour milk cream cheese butter oil chicken beef pork bacon sugar salt pepper rice bean beans tomato sauce juice egg eggs yeast starch vinegar mayo mayonnaise syrup chocolate cocoa bread mozzarella parmesan shortening yogurt mustard garlic onion potato broccoli sausage honey walnut walnuts nuts oats vanilla fruit vegetable vegetables mushroom mushrooms pickle tortillas rolls ham turkey canola olive dough pasta spaghetti chips".split())
|
|
|
|
def slug(value): return re.sub(r"[^a-z0-9]+", "_", value.lower()).strip("_")
|
|
def tokens(value): return set(re.findall(r"[a-z0-9]+", value.lower()))
|
|
|
|
def parse_package(name: str):
|
|
patterns = [
|
|
(r"(?:^|[, ])(\d+(?:\.\d+)?)\s*(?:fl\.?\s*oz|fluid ounces?)\b", "fluid_ounce_us"),
|
|
(r"(?:^|[, ])(\d+(?:\.\d+)?)\s*(?:lb|lbs|pounds?)\b", "pound"),
|
|
(r"(?:^|[, ])(\d+(?:\.\d+)?)\s*(?:oz|ounces?)\b", "ounce_mass"),
|
|
(r"(?:^|[, ])(\d+(?:\.\d+)?)\s*(?:gal|gallon)\b", "gallon_us"),
|
|
(r"(?:^|[, ])(\d+(?:\.\d+)?)\s*(?:count|ct)\b", "each"),
|
|
]
|
|
for pattern, unit in patterns:
|
|
match = re.search(pattern, name, re.I)
|
|
if match: return {"quantity": float(match.group(1)), "unit_id": unit}
|
|
return None
|
|
|
|
def excel_date(value):
|
|
return (dt.datetime(1899, 12, 30) + dt.timedelta(days=float(value))).date().isoformat()
|
|
|
|
def walmart_products(path: Path):
|
|
with ZipFile(path) as archive: root = ET.fromstring(archive.read("xl/worksheets/sheet2.xml"))
|
|
products = {}
|
|
for row in root.findall(".//m:row", NS)[2:]:
|
|
values=[]
|
|
for cell in row.findall("m:c", NS):
|
|
text=cell.find(".//m:t",NS); raw=cell.find("m:v",NS)
|
|
values.append(text.text if text is not None else raw.text if raw is not None else "")
|
|
if len(values)<7 or not values[5]: continue
|
|
date, _, name, quantity, price, product_id, url = values[:7]
|
|
record=products.setdefault(product_id,{"supplier_id":"walmart","supplier_sku":product_id,"name":name,"url":url,"package":parse_package(name),"prices":[]})
|
|
record["prices"].append({"amount":float(price),"currency":"USD","effective_at":excel_date(date),"source":{"source_type":"supplier","url":values[7] if len(values)>7 else url,"title":"Walmart receipt","reviewed":True}})
|
|
return list(products.values())
|
|
|
|
def sams_products(path: Path):
|
|
products={}
|
|
for row in csv.DictReader(path.open()):
|
|
sku=row["item_code"].strip(); name=row["description"].strip()
|
|
if not sku or "instant savings" in name.lower() or row.get("category")!="Expenses:Food:Groceries": continue
|
|
record=products.setdefault(sku,{"supplier_id":"sams_club","supplier_sku":sku,"name":name,"package":parse_package(name),"prices":[]})
|
|
record["prices"].append({"amount":float(row["pre_tax_amount"]),"currency":"USD","effective_at":row["date"],"source":{"source_type":"supplier","title":"Sam's Club receipt","reviewed":True}})
|
|
return list(products.values())
|
|
|
|
def score(name, ingredient):
|
|
query=tokens(ingredient["name"]); candidate=tokens(name)
|
|
aliases=[tokens(x["name"]) for x in ingredient.get("aliases",[])]
|
|
return max([len(query&candidate)/max(1,len(query)), *[len(a&candidate)/max(1,len(a)) for a in aliases]])
|
|
|
|
def propose(walmart: Path, sams: Path, output: Path):
|
|
ingredients=[yaml.safe_load(p.read_text()) for p in (ROOT/"culinary/ingredients").glob("*.yaml")]
|
|
products=[p for p in walmart_products(walmart)+sams_products(sams) if tokens(p["name"]) & FOOD_WORDS]
|
|
proposals=[]
|
|
for product in products:
|
|
ranked=sorted(((score(product["name"],i),i) for i in ingredients),key=lambda x:(-x[0],x[1]["name"]))[:5]
|
|
if ranked[0][0] < .5: continue
|
|
proposals.append({**product,"ingredient_candidates":[{"ingredient_id":i["id"],"name":i["name"],"score":round(s,3)} for s,i in ranked if s>0]})
|
|
output.parent.mkdir(parents=True,exist_ok=True)
|
|
output.write_text(yaml.safe_dump({"generated_at":dt.datetime.now(dt.timezone.utc).isoformat(),"products":proposals},sort_keys=False))
|
|
print(f"Wrote {len(proposals)} receipt product proposals to {output}")
|
|
|
|
def apply(decisions_path: Path, candidates_path: Path):
|
|
decisions=json.loads(decisions_path.read_text())["decisions"]
|
|
products=yaml.safe_load(candidates_path.read_text())["products"]
|
|
by_key={f'{p["supplier_id"]}:{p["supplier_sku"]}':p for p in products}
|
|
target=ROOT/"culinary/purchase_items"; target.mkdir(exist_ok=True)
|
|
imported=skipped=0
|
|
for key,ingredient_id in decisions.items():
|
|
if ingredient_id is None: skipped+=1; continue
|
|
product=by_key.get(key)
|
|
if not product: raise SystemExit(f"Unknown receipt product {key}")
|
|
if not product.get("package"):
|
|
print(f"Skipped {key}: package size is not explicit"); skipped+=1; continue
|
|
item_id=f'{product["supplier_id"]}_{slug(product["supplier_sku"])}'
|
|
item={"schema_version":2,"id":item_id,"ingredient_id":ingredient_id,"name":product["name"],"supplier_id":product["supplier_id"],"supplier_sku":product["supplier_sku"],"status":"active","package":product["package"],"prices":product["prices"],"nutrition_mapping_ids":[],"allergen_mapping_ids":[]}
|
|
(target/f"{item_id}.yaml").write_text(yaml.safe_dump(item,sort_keys=False)); imported+=1
|
|
print(f"Imported {imported} purchasing items; skipped {skipped}")
|
|
|
|
def main():
|
|
parser=argparse.ArgumentParser(description=__doc__); sub=parser.add_subparsers(dest="command",required=True)
|
|
p=sub.add_parser("propose"); p.add_argument("--walmart",type=Path,default=LEDGER/"documents/_inbox/walmart-receipts.xlsx"); p.add_argument("--sams",type=Path,default=LEDGER/"documents/merchants/sams-club/receipts/through-2026-07-29_sams-club_card-8766_receipt-items.csv"); p.add_argument("--output",type=Path,default=ROOT/"generated/receipt-product-candidates.yaml")
|
|
a=sub.add_parser("apply"); a.add_argument("decisions",type=Path); a.add_argument("--candidates",type=Path,default=ROOT/"generated/receipt-product-candidates.yaml")
|
|
args=parser.parse_args(); propose(args.walmart,args.sams,args.output) if args.command=="propose" else apply(args.decisions,args.candidates)
|
|
if __name__=="__main__": main()
|