"""Import one or more image manifest CSVs into master_product_image. Examples: python -m app.jobs.import_master_image_manifest --file "E:\\2026\\manifest-2026-06-14.csv" python -m app.jobs.import_master_image_manifest --file "E:\\2026\\manifest-2026-06-14.csv" --file "E:\\2026\\manifest-2026-06-16.csv" --apply """ from __future__ import annotations import argparse import json from pathlib import Path from typing import Any, Dict, Iterable, List, Optional, Sequence, Set import pandas as pd from db.source_imports import normalize_column_key def _load_name_set( values: Optional[Sequence[str]] = None, *, file_path: Optional[str] = None, ) -> Set[str]: loaded: Set[str] = set() for value in values or []: text = str(value or "").strip() if text: loaded.add(text) if file_path: path = Path(file_path) for raw_line in path.read_text(encoding="utf-8").splitlines(): text = raw_line.strip() if not text or text.startswith("#"): continue loaded.add(text) return loaded def _load_ignore_unmatched( values: Optional[Sequence[str]] = None, *, file_path: Optional[str] = None, ) -> Set[str]: return _load_name_set(values, file_path=file_path) def _filter_manifest_to_names( df: pd.DataFrame, allowed_names: Set[str], ) -> pd.DataFrame: if not allowed_names: return df name_column = None for column in df.columns: if normalize_column_key(column) in {"name", "file_name", "filename", "image_name", "image"}: name_column = column break if name_column is None: return df.iloc[0:0].copy() mask = df[name_column].map(lambda value: str(value or "").strip() in allowed_names) return df.loc[mask].copy() def _filter_unmatched( unmatched: Iterable[str], ignored: Set[str], ) -> List[str]: if not ignored: return [str(item or "").strip() for item in unmatched if str(item or "").strip()] return [ text for item in unmatched if (text := str(item or "").strip()) and text not in ignored ] def import_master_image_manifest( *, files: Sequence[str], apply: bool, source_label: str = "image_manifest", ignore_unmatched: Optional[Sequence[str]] = None, ignore_unmatched_file: Optional[str] = None, only_unmatched: Optional[Sequence[str]] = None, only_unmatched_file: Optional[str] = None, unmatched_limit: int = 50, ) -> Dict[str, Any]: from db.master_product_images import import_master_product_images_dataframe, is_images_only_csv from db.session import get_session summaries: List[Dict[str, Any]] = [] totals: Dict[str, int] = { "input_rows": 0, "images_upserted": 0, "skus_touched": 0, "skipped_rows": 0, "missing_url_rows": 0, "missing_name_rows": 0, "unmatched_rows": 0, "duplicate_rows_collapsed": 0, } unmatched: List[str] = [] ignored = _load_ignore_unmatched(ignore_unmatched, file_path=ignore_unmatched_file) only_names = _load_name_set(only_unmatched, file_path=only_unmatched_file) filtered_input_rows = 0 with get_session() as session: for file_path in files: path = Path(file_path) df = pd.read_csv(path) if not is_images_only_csv(df): raise ValueError(f"CSV does not look like an image manifest: {path}") original_input_rows = len(df) if only_names: df = _filter_manifest_to_names(df, only_names) filtered_input_rows += len(df) stats = import_master_product_images_dataframe( session, df, source_label=f"{source_label}:{path.name}", ) summaries.append( { "file": str(path), **{ **stats, "source_input_rows": original_input_rows, "sample_unmatched": _filter_unmatched(stats.get("all_unmatched") or [], ignored)[:20], }, } ) for key in totals: totals[key] += int(stats.get(key) or 0) unmatched.extend(_filter_unmatched(stats.get("all_unmatched") or [], ignored)) if apply: session.commit() else: session.rollback() return { "status": "ok", "applied": bool(apply), "file_count": len(files), "files": summaries, "totals": totals, "ignored_unmatched_count": len(ignored), "only_unmatched_count": len(only_names), "filtered_input_rows": filtered_input_rows, "sample_unmatched": unmatched[: max(0, int(unmatched_limit))], } def main(argv: Optional[Sequence[str]] = None) -> int: parser = argparse.ArgumentParser(description="Import image manifest CSV(s) into master_product_image") parser.add_argument("--file", dest="files", action="append", required=True, help="Manifest CSV path; repeatable") parser.add_argument("--apply", action="store_true", help="Commit changes (default is dry-run)") parser.add_argument("--source-label", default="image_manifest") parser.add_argument( "--ignore-unmatched", action="append", default=[], help="Known unmatched filename to suppress from sample output; repeatable", ) parser.add_argument( "--ignore-unmatched-file", help="Text file with one known unmatched filename per line to suppress from sample output", ) parser.add_argument( "--only-unmatched", action="append", default=[], help="Process only these manifest filenames; repeatable", ) parser.add_argument( "--only-unmatched-file", help="Text file with one manifest filename per line to process exclusively", ) parser.add_argument( "--unmatched-limit", type=int, default=50, help="How many unmatched filenames to print after ignore filtering (default: 50)", ) args = parser.parse_args(list(argv) if argv is not None else None) result = import_master_image_manifest( files=args.files, apply=args.apply, source_label=args.source_label, ignore_unmatched=args.ignore_unmatched, ignore_unmatched_file=args.ignore_unmatched_file, only_unmatched=args.only_unmatched, only_unmatched_file=args.only_unmatched_file, unmatched_limit=args.unmatched_limit, ) print(json.dumps(result, indent=2)) return 0 if __name__ == "__main__": raise SystemExit(main())