main.py

1import argparse2import sys3from catalog import fetch4from catalog import s35from catalog.correct import correct_products6from catalog.helpers import all_ids, live_ids, load_catalog, load_corrections7from catalog.helpers import load_product, sort_catalog8from catalog.parse import parse_products9from catalog.sort import sort_products1011AOP = "All-Over Print"12CUT_SEW = "CUT-SEW"1314def normalize(name: str) -> str:15    return name.replace("’", "'")1617def is_aop(item: dict) -> bool:18    return item["type"] == CUT_SEW and AOP in normalize(item["name"])1920# GATES2122def gate_rows(rows: list[dict], catalog: list[dict]) -> list[str]:23    live = {item["id"]: item for item in rows}24    corrections = load_corrections()["variants"]25    failures = []26    for row in catalog:27        id = row["id"]28        item = live.get(id)29        if item is None:30            failures.append(f"{id} {row['title']}: absent from the live catalog")31            continue32        if item["is_discontinued"]:33            failures.append(f"{id} {row['title']}: is_discontinued")34        techniques = [t["key"] for t in item["techniques"]]35        if techniques != ["cut-sew"]:36            failures.append(f"{id} {row['title']}: techniques {techniques}")37        if len(item["colors"]) != 1 and str(id) not in corrections:38            names = [c["name"] for c in item["colors"]]39            print(f"Colours: {id} {row['title']} - {names}")40    return failures4142def gate_products(ids: list[int]) -> list[str]:43    failures = []44    for id in ids:45        product = load_product(id)46        colors = {v.color for v in product.variants if not v.is_ignored}47        if len(colors) != 1:48            failures.append(f"{id} {product.title}: {len(colors)} live colours {sorted(colors)}")49        if not [p for p in product.placements if not p.is_ignored]:50            failures.append(f"{id} {product.title}: no live placement")51        if not [m for m in product.mockups if not m.is_ignored]:52            failures.append(f"{id} {product.title}: no live mockup")53    return failures5455def fail(failures: list[str]) -> None:56    if not failures:57        return58    print(f"Failed {len(failures)} checks")59    for line in failures:60        print(f"- {line}")61    sys.exit(1)6263# VERBS6465def build(args) -> None:66    catalog = load_catalog()67    ids = all_ids() if args.all else live_ids()68    if args.id:69        ids = args.id70    print(f"Building {len(ids)} of {len(catalog)} catalog rows")71    rows = fetch.load_raw_catalog(args.refetch)72    fail(gate_rows(rows, catalog))73    print(f"Gate: {len(catalog)} rows are live, cut-sew and current")74    sort_catalog()75    fetch.fetch_products(ids, args.refetch)76    parse_products(ids)77    sort_products(ids)78    correct_products(ids)79    fail(gate_products(ids))80    print(f"Gate: {len(ids)} products carry one colour, a placement and a mockup")81    if args.local:82        print("Local only, nothing uploaded")83        return84    s3.upload_products(ids)8586def new(args) -> None:87    rows = fetch.load_raw_catalog(args.refetch)88    known = set(all_ids())89    for item in sorted(rows, key=lambda i: i["id"]):90        if is_aop(item) and item["id"] not in known:91            print(f"{item['id']} - {normalize(item['name'])}")9293def gate(args) -> None:94    rows = fetch.load_raw_catalog(args.refetch)95    catalog = load_catalog()96    fail(gate_rows(rows, catalog))97    print(f"Gate: {len(catalog)} rows are live, cut-sew and current")9899def show(args) -> None:100    for row in load_catalog():101        mark = "LIVE" if row.get("live") else "    "102        print(f"{mark} {row['id']:>5} {row['category']:<12} {row['title']}")103104def clean(args) -> None:105    s3.delete_products()106107# CLI108109def parser() -> argparse.ArgumentParser:110    p = argparse.ArgumentParser(prog="catalog.main")111    subs = p.add_subparsers(dest="verb", required=True)112113    b = subs.add_parser("build")114    b.add_argument("--all", action="store_true")115    b.add_argument("--id", type=int, nargs="+")116    b.add_argument("--refetch", action="store_true")117    b.add_argument("--local", action="store_true")118    b.add_argument("--yes", action="store_true")119    b.set_defaults(run=build, gated=True)120121    n = subs.add_parser("new")122    n.add_argument("--refetch", action="store_true")123    n.set_defaults(run=new, gated=False)124125    g = subs.add_parser("gate")126    g.add_argument("--refetch", action="store_true")127    g.set_defaults(run=gate, gated=False)128129    s = subs.add_parser("show")130    s.set_defaults(run=show, gated=False)131132    c = subs.add_parser("clean")133    c.add_argument("--yes", action="store_true")134    c.set_defaults(run=clean, gated=True)135136    return p137138def main() -> None:139    args = parser().parse_args()140    if args.gated and not args.yes:141        print(f"PLAN {args.verb}. Add --yes to run it")142        sys.exit(1)143    args.run(args)144145if __name__ == "__main__":146    main()