Appearance
File Lineage — Mirakl / PIM / Offers
Companion to data-lineage.md (which traces domains — description, ID matching, stock — with prose-first citations). This document instead catalogs every file each task reads and writes, C4-context style: ecom_products_all (jobs-alg-all.py) and ecom_w1_full (jobs-inf-w1-full.py) as the two DAGs, one file-flow table each, plus a consolidated cross-DAG handoff table. Drill into a specific field's construction only where data-lineage.md already does so (§1–§6).
Grounded in direct reads of jobs-alg-all.py, jobs-inf-w1-full.py, bloomreach/xmlbuilder/all_products.py, bloomreach/xmlbuilder/supply_sync.py, and jobs-shw-gm.py in /Users/pmccluskey/work/pim/airflow-pim-dags. jobs-inf-w1-full-pre.py/jobs-inf-w1-delta.py/jobs-inf-w1-delta-pre.py follow the same file-flow skeleton as jobs-inf-w1-full.py — see "Diffs" below for where they genuinely differ.
Scope note
This covers ecom_products_all and ecom_w1_full (representative of the four W1 DAGs) only — the same two-DAG slice data-lineage.md documents. jobs-shw-gm.py (a per-BU Shopware DAG) appears only as a confirmed consumer of W1's output, not fully catalogued itself. Per-BU Algolia DAGs (jobs-alg-{gm,ec,pp,bk}-inr.py) were checked only for the specific {bu}_product.json question below, not catalogued in full.
ecom_products_all (jobs-alg-all.py)
| Task ID | python_callable | Input file(s) | Output file(s) | Notes |
|---|---|---|---|---|
| Prepare-Environment | (bash) | — | temp/import/intermediate/staging dirs (cleared) | |
| Download-PIM | (bash) | S3 ALL_S3_PIM2_DIRECTORY | import/all.zip | |
| Unpack-PIM | (bash) | import/all.zip | unpacked PIM CSVs in import/ | |
| Download-Schema | (bash) | S3 schema/{target}/ | intermediate/{schema_template_file} | |
| Download-BK-ERP-Data | s3_download_by_sts | S3 BK_S3_ERP_SRC (STS role) | import/AVAILABILITY_BK.CSV | |
| Download-FC-ERP-Data | s3_download_by_sts | S3 BK_S3_ERP_SRC (STS role) | import/AVAILABILITY_FC.CSV | |
| Download-BK-Regions-Map | (bash) | S3 BK_S3_SRC/erp-data | import/daily_branch_data_bk.csv | |
| Normalize-BK-Regions-Map | branchNormalizer.normalize_mincron | import/daily_branch_data_bk.csv | intermediate/branch_data_BK.csv | also caches to S3 |
| Download-CE-ERP-Data | s3_download | S3 CE_S3_SRC/erp-data | import/AVAILABILITY_CE.csv | |
| Download-EC-ERP-Data | s3_download_by_sts | S3 EC_S3_ERP_SRC (STS role) | import/AVAILABILITY_EC_NEW.csv | |
| Get-GM/ACME/HandC-Regions-Map | jane.get_regions_{gm,ac,hc} | Jane/Conduit API | import/{GM,AC,HC}_branches.csv | HTTP pull, not file transform |
| Download-GM-ERP-Data | s3_download | S3 GM_S3_MINCRON_SRC/mincron | import/mincron.zip | |
| Unpack-GM-ERP-Data | (bash) | import/mincron.zip | import/AlgoliaExport.json | |
| Split-GM-ERP-Data | (bash) | import/AlgoliaExport.json | product_family_export.json, product_headers.json, product_location_info.json | |
| Convert-GM-Availability | availabilityNormalizer.convert_mincron_json | import/product_headers.json | import/erp_pim_data_GM.csv, AVAILABILITY_GM.csv | |
| Split-GM-Availability | availabilityNormalizer.split_gm_availability | AVAILABILITY_GM.csv + branch maps | AVAILABILITY_AC.csv, AVAILABILITY_HC.csv | |
| Download-PP-ERP-Data | s3_download | S3 PP_AVAILABILITY_S3_SRC | import/AVAILABILITY_PP.csv | |
| Convert-CE-Availability | availabilityNormalizer.convert_ce_data | AVAILABILITY_CE.csv | same file, normalized | |
| Download-TE-ERP-Data → Convert-TE-Availability | availabilityNormalizer.convert_te_data | import/TECExport.csv → AVAILABILITY_TE.csv | erp_pim_data_TE.csv | |
| Download/Convert-Neuco-* | availabilityNormalizer.convert_neuco_data | AVAILABILITY_NEUCO.csv, NEUCO-items.csv | attr_data_NEUCO.csv, erp_pim_data_NC.csv | gated by Airflow Variable |
| Download-Rheem-Data → Convert-Rheem-Availability | availabilityNormalizer.convert_rheem_data | Rheem FTP CSV + endless-aisle file | attr_data_RHEEM.csv, erp_pim_data_RM.csv | gated by Airflow Variable |
| Download-{BK,CE,EC,GM,HM,PP}-PIM-Data | (bash, optional) | S3 {BU}_S3_SRC/erp-data/{target} | erp_pim_data_{BU}.csv, catalog_code_data_{BU}.csv, uom_data_{BU}.csv | |
| File-Cleanup → Re-Encode | (bash) | import/* | intermediate/* (UTF-8) | import → intermediate boundary |
| Handle-Attributes | all_etl.merge_attrs | intermediate/attributes_data.csv | processed attributes | |
| Handle-Structures | all_etl.process_structure | intermediate/structures_data.csv | processed structures | |
| Merge-Baker / Split-Homans / Drop-Obsolete-Columns | availabilityNormalizer.{merge_baker,split_homans,normalize_cols} | per-BU availability CSVs | normalized per-BU availability CSVs | |
| Merge-Norm-Availability | availabilityNormalizer.merge | all intermediate/AVAILABILITY_*.csv | intermediate/availability.csv | NC/RM inclusion gated by Variables |
| Merge-ERP-Data | all_etl.merge_erp | erp_pim_data_{BU}.csv × all BUs, items file | intermediate/erp_data.csv | |
| Merge-Catalog-Codes / Merge-UOM-Data | all_etl.merge_erp | catalog_code_data_{BU}.csv / uom_data_{BU}.csv | catalog_code_data.csv / uom_data.csv | |
| Translate | all_etl.process | all merged intermediate CSVs + availability.csv + erp_data.csv | staging/all_products.json | main transform — builds the availability field via _availability_dict(), see data-lineage.md §3 |
| Count-Check / Schema-Check | count_utils.simple_json_count / json_utils.schema_check | staging/all_products.json | pass/fail signal | glob correct in this file (no backtick bug) |
| Transfer | json_utils.push_index_simple | staging/all_products.json | Algolia index (ALL_ALGOLIA_INDEX) | |
| Upload-W1 | (bash) | staging/all_products.json | S3 W1_S3_ARCHIVE/w1_json_feeds/{target}/all_products.json | cross-DAG handoff point — see table below |
| Archive | (bash) | staging/all_products.json, import dir | S3 ALL_S3_ARCHIVE |
ecom_w1_full (jobs-inf-w1-full.py)
| Task ID | python_callable | Input file(s) | Output file(s) | Notes |
|---|---|---|---|---|
| Prepare-Environment | (bash) | — | temp/import/intermediate/staging dirs (cleared) | |
| Download-Sales → Unpack → Convert | (bash) | S3 SS_S3_SRC/pending | import/ss_sales_qtyshipped.csv | ASCII-normalized |
| Download-ALL-PIM / Download-W1-PIM | (bash) | S3 ALL_S3_PIM2_DIRECTORY / W1_S3_PIM2_DIRECTORY | unpacked PIM/W1 CSVs in import/ | |
| Download-Schema | (bash, optional) | S3 W1/schema/{target} | intermediate/{schema_template_file}, {supply_sync_schema_template_file} | |
| Check-CSV-Headers / Check-Sales-Hash | bash_utils.validate_header / file_utils.hashes_check | sales file vs headers/hash cache | pass/fail signal, SNS alert on failure | |
| Merge-Full-PID | (bash) | per-BU dirs under intermediate/ | full_pid.csv | |
| Download-BU-Feeds | (bash) | S3 W1_S3_ARCHIVE/w1_json_feeds/{target}/*_product*.json | intermediate/*_product*.json | cross-DAG handoff — this glob downloads all_products.json itself; see correction below |
| File-Cleanup → Re-Encode | (bash) | import/* | intermediate/* (UTF-8) | |
| Get-PIM-List | (bash) | intermediate/itemdataflex.csv | pim_list.csv | |
| Handle-Attributes / Structures / Categories / Item-Flex | all_etl.{merge_attrs,process_structure,process_category,process_flex} | respective intermediate/*_data.csv | processed attribute/structure/category/flex data | see data-lineage.md §1 |
| Merge-ERP-Data / Catalog-Codes / Descriptions / Status / Title / UOM / URL-Key | all_etl.merge_bu_erp (7 calls, one per field) | per-BU field data, items file | erp_data.csv, catalog_code_data.csv, description_data.csv, status_data.csv, title_data.csv, uom_data.csv, url_keys_data.csv | see data-lineage.md §1/§2 |
| Merge-BU-Attrs | all_etl.merge_bu_attrs | per-BU attr exports, items file | bu_attr_data.csv | feeds Shopware + Supply-Sync steps |
| Translate | mirakl_etl.process | all merged intermediate CSVs | staging/w1_products.json | no availability field — see data-lineage.md §3 |
| Count-Check / Schema-Check | count_utils.simple_json_count / json_utils.schema_check | staging/w1_products.json | pass/fail signal | BUG: Schema-Check's glob is "w1_products.\json"(stray backtick) — matches nothing; isolated to this file,delta.py/delta-pre.py` use the correct pattern |
| Upload-W1-Feed | (bash) | staging/w1_products.json | S3 W1_S3_ARCHIVE/W1/{target}.w1_products*.json | cross-DAG handoff — consumed by jobs-shw-gm.py's Download-W1-Feed |
| Shopware-Products / BU-Data / Association / Media / Dimensions / Descriptions / Titles / Taxonomy / Attribution | shopware_etl.shopware_*_w1 | w1_products.json, items.csv, bu_attr_data.csv, mirakl_pid_2.csv, category/attribute files | tmp/shopware_*.csv (10 files) | |
| Pack-Shopware / Upload-Shopware | (bash) | 10 tmp/shopware_*.csv | tmp/shopware.tar.gz → S3 SHOPWARE_S3_ARCHIVE/shopware_import_files | |
| Supply-Sync-Algolia | supply_sync_etl.supply_sync_algolia_process | items.csv, all_products.json (the file Download-BU-Feeds just downloaded — confirmed same file, see correction below), bu_attr_data.csv, mirakl_pid_2.csv, shopware_descriptions.csv, shopware_titles.csv, shopware_item_taxonomy.csv, taxonomy_paths.csv, mirakl_offers.csv | tmp/supply_sync.json | pass-through of all_products.json's availability field — see data-lineage.md §3 |
| Supply-Sync-Schema-Check / Transfer / Upload-Supply-Sync | json_utils.{schema_check,push_index_simple} / bash | tmp/supply_sync.json | Algolia index (SUPPLY_SYNC_ALGOLIA_INDEX); S3 SHOPWARE_S3_ARCHIVE/supply_sync_feed | |
| Full-Add-Products / Mirakl-Get-Products / Mirakl-Split-Products | mirakl_etl.{filter_products,get_products,split_products} | full_pid.csv, w1_products.json, Mirakl API | insert_pid.csv, update_pid.csv, delete_pid.csv | |
| Mirakl-Insert/Update/Delete-Products | mirakl_etl.mirakl_products_process | w1_products.json + respective pid list | mirakl_{insert,update,delete}_products.json | |
| Mirakl-Get-Taxonomy / Mirakl-Taxonomy / Push / Report / Errors | mirakl_etl.{get_taxonomy,mirakl_taxonomy_process,push_taxonomy,report_taxonomy,err_taxonomy} | category.csv, Mirakl API | mirakl_taxonomy.csv, taxonomy_paths.csv (feeds Supply-Sync-Algolia) | Push gated by MIRAKL_PUSH_DISABLED |
| Mirakl-Taxonomy-Delete-Push | (dummy) | — | — | real logic commented out — see anomaly below |
| Mirakl-Get-Offers / Aggregate-Offers | mirakl_etl.{get_offers,aggr_offers} | Mirakl API | mirakl_offers.csv (feeds Supply-Sync-Algolia) | |
| Mirakl-Operator-Insert/Update-Push / Report / Errors | mirakl_etl.{push_operator_products,report_operator_products,err_operator_products} | mirakl_{insert,update}_products.json | tracking IDs, error CSVs | Push gated by MIRAKL_PUSH_DISABLED |
| Mirakl-Operator-Delete-Push | (dummy) | — | — | real logic commented out — see anomaly below |
| Mirakl-Get-Products-2 | mirakl_etl.get_products | Mirakl API | mirakl_pid_2.csv | re-fetch feeding all Shopware-* and Supply-Sync tasks |
| Mirakl-Get-Deleted-Products | mirakl_etl.get_deleted_products | Mirakl API | mirakl_deleted_products.json | |
| Handle-Uncategorized / Inactive / Inactive-CEUS | mirakl_etl.handle_{uncategorized,inactive,inactive_ceus} | w1_products.json | report CSVs → S3 W1/reports/full | |
| Pack-Report / Report-Notify | bash / upload_and_message | tmp report files | report.tar.gz; SNS notification; S3 W1/reports/full | |
| Archive | (bash) | w1_products.json, supply_sync.json, import dir | S3 W1_S3_ARCHIVE |
Diffs: full-pre / delta / delta-pre vs. full
full-pre: same task IDs and file-flow skeleton; differs only in environment wiring (FULL_PREbase path, PREPROD Mirakl/Algolia endpoints,W1_PRE/supply_sync_feed_preS3 paths).Mirakl-Taxonomy-Delete-Push/Mirakl-Operator-Delete-Pushare livePythonOperators here (not dummies) — a functional divergence fromfull, not just naming.delta: structural differences — adds a per-BU incremental download loop (Download-Delta-{BU}/Unpack-{BU}/Re-Encode-{BU}, pulling froms3://{BU}_S3_BLOWER_SRC/{BU}/deltas/), usesdiff_pid.csv/Merge-Diff-PIDinstead offull_pid.csv/Merge-Full-PID, and dropsCount-Checkand the entire delete-handling chain (Mirakl-Delete-Products,Mirakl-Taxonomy-Delete-*,Mirakl-Get-Deleted-Products) entirely — confirmed absent from the task list, not just dummied out.delta-pre: same PREPROD/W1_PRE/DELTA_PREpattern asfull-prevsfull, applied to thedeltaskeleton.- The
Schema-Checkglob bug is confirmed isolated tojobs-inf-w1-full.py— bothdelta.pyanddelta-pre.pyuse the correct"w1_products.json"pattern with no stray backtick.
Cross-DAG file handoffs
| File | Produced by (DAG:Task) | Consumed by (DAG:Task) | S3 path | Confidence |
|---|---|---|---|---|
all_products.json | ecom_products_all:Upload-W1 | ecom_w1_full:Download-BU-Feeds (glob match) → Supply-Sync-Algolia | W1_S3_ARCHIVE/w1_json_feeds/{target}/all_products.json | Confirmed — Download-BU-Feeds's lpath=INTERMEDIATE_PATH places the file exactly where Supply-Sync-Algolia's bare-filename arg ('all_products.json') resolves; same file, not a separate copy. |
w1_products.json | ecom_w1_full:Translate → Upload-W1-Feed | jobs-shw-gm.py:Download-W1-Feed (pulls most recent, not necessarily same-run) | W1_S3_ARCHIVE/W1/{target}.w1_products*.json | Confirmed for the handoff mechanism; not confirmed whether jobs-shw-gm.py always gets the same-run file vs. a stale prior run. |
shopware.tar.gz | ecom_w1_full:Pack-Shopware → Upload-Shopware | Shopware (external, pulls per README) | SHOPWARE_S3_ARCHIVE/shopware_import_files | Confirmed upload mechanism; Shopware-side pull not traced in this repo. |
supply_sync.json | ecom_w1_full:Supply-Sync-Algolia → Transfer/Upload-Supply-Sync | Algolia (SUPPLY_SYNC_ALGOLIA_INDEX); S3 archive | SHOPWARE_S3_ARCHIVE/supply_sync_feed | Confirmed. |
{bu}_product.json (e.g. GM_product.json) | Confirmed real and live — see note below | merge_bu_erp/merge_bu_attrs (data-lineage.md §1/§2) | w1_json_feeds/production/{BU}_product.json | Confirmed via live task log — producer DAG/task not re-identified after an earlier false trace, but not pursued further per scope decision (see note). |
Note: {bu}_product.json is confirmed real and live — an earlier pass this session was wrong
A Download-BU-Feeds task log from a production ecom_w1_delta run (s3://wwi.ops.production.airflow.logs/logs/dag_id=ecom_w1_delta/run_id=manual__2026-08-20T02:08:58.941190+00:00/task_id=Download-BU-Feeds/attempt=1.log) shows the task's aws s3 sync s3://watsco-aux/w1_json_feeds/production ... downloading all of the following in the same run: all_products.json and 18 per-BU files — BK, CE, CECA, CECPD, CEFL, CEMA, CEMX, CENE, CEPR, CESC, CESE, CEST, CIAC, EC, GM, HM, PP, TW (each as {BU}_product.json). This also confirms W1_S3_ARCHIVE resolves to the watsco-aux bucket.
An earlier pass this session concluded {bu}_product.json had no live producer, based on checking only jobs-alg-all.py (uploads only all_products.json), jobs-alg-{gm,ec,pp,bk}-inr.py (found w1_json_feeds references only inside dead/commented code), and all jobs-shw-*.py files (zero references) — and incorrectly generalized "not found in the files I checked" to "doesn't exist." The live log evidence overrides that conclusion. The actual producer DAG/task was not re-identified — with 98 jobs-*.py files in the repo and the confirmed candidates all ruled out, finding the real producer would require a broader systematic search that wasn't pursued, since it doesn't change anything about what W1 actually consumes (which is what this document tracks). If the producer matters for a future task, start from the per-BU jobs-alg-{bu}.py (non--inr) files, not the ones already ruled out.
Other anomalies found while building this catalog
Mirakl-Taxonomy-Delete-Push/Mirakl-Operator-Delete-Pushare dummy operators inecom_w1_full(real logic present but commented out directly below each). Their downstream*-Delete-Reporttasks read tracking-ID files (mirakl_taxonomy_delete_tracking_ids.csv,mirakl_operator_delete_tracking_ids.csv) that are never written as a result — these report tasks likely run against stale or empty files every execution. Not confirmed whether this is an intentionally disabled feature or a bug.Schema-Check's glob bug (jobs-inf-w1-full.pyonly): the pattern string is"w1_products.`json"— a stray backtick beforejson— which won't match the real filew1_products.json.json_utils.schema_checksilently receives an empty file list for this task. Confirmed correct injobs-inf-w1-delta.py/jobs-inf-w1-delta-pre.py.
Appendix: S3 buckets
The code never hardcodes bucket names — every S3 reference is built from environ['{VAR}'] (bucket) + a literal path fragment (see the tables above). These are genuine OS/MWAA environment variables, not Airflow Variables — the daily Airflow Variables snapshot at s3://watsco-aux/airflow_vars/production/{date}.json (848 keys, checked against 2026-08-19.json) does not contain any of the *_S3_SRC/*_S3_ARCHIVE/*_S3_ARGS names, so bucket names for most of these variables could not be resolved this session without direct MWAA environment-config access (not available under the enterprise-ro/enterprise-power profiles used — mwaa:ListEnvironments was denied under -ro, and further escalation wasn't attempted without asking first).
Confirmed bucket:
| Bucket | Confirmed via | Contents observed |
|---|---|---|
watsco-aux | Direct s3api list-objects-v2 (enterprise-ro); a full archived tarball download/extraction earlier this session; and a live Download-BU-Feeds task log (ecom_w1_delta, wwi.ops.production.airflow.logs bucket, account 184348296362, via ecom-logs-ops-power) showing the exact resolved command aws s3 sync s3://watsco-aux/w1_json_feeds/production ... | watsco-aux = W1_S3_ARCHIVE, confirmed directly (not inferred from path-pattern matching, per the resolved bash command). Top-level prefixes: W1/, w1_json_feeds/, archive/, airflow_vars/ (daily Airflow Variables JSON snapshots), reports/, cache/, algolia_config/, hydros/, hydros_archive/, xref//XREF/, BRANCH_GEO/, BLOWER_TRIG_LOCKS/, and per-BU prefixes (BK/, CE/, CENE/, EC/, GM/, HM/, REFONLY/, csd/, _TMP/). |
Task logs live in a separate bucket/account — wwi.ops.production.airflow.logs, in AWS account 184348296362 (not the 784596786944 account enterprise-ro/enterprise-power resolve to). Use the ecom-logs-ops-ro/ecom-logs-ops-power aws-vault profiles for that bucket; ro was denied GetObject by IAM, power worked. Listing (ListObjectsV2) worked under enterprise-ro/enterprise-power for watsco-aux directly, but reading individual objects there required enterprise-power (enterprise-ro was denied GetObject specifically, not just HeadObject).
Unconfirmed — environment variable name and code-referenced path only (bucket name not resolved; do not assume these all point to the same bucket, or to watsco-aux):
| Variable | Path fragment used in code | Referenced by | Purpose (from code context) |
|---|---|---|---|
ALL_S3_SRC | (root) | jobs-alg-all.py:Download-PIM | Source for ALL/FULL PIM2 export zip |
ALL_S3_ARCHIVE | (root) | jobs-alg-all.py:Archive | Archive destination for staging/all_products.json + import dir |
AUX_S3_SRC | — | referenced but not traced to a specific task this session | Unclear — name suggests overlap with watsco-aux, not confirmed same bucket |
BK_S3_SRC, BK_S3_ERP_SRC, BK_S3_ERP_PATH | erp-data, STS-assumed-role path | jobs-alg-all.py:Download-BK-Regions-Map, Download-BK-ERP-Data | BK's raw ERP/branch-map source (STS role — separate account/bucket from the default profile) |
CE_S3_SRC | erp-data | jobs-alg-all.py:Download-CE-ERP-Data | CE's raw ERP availability source |
CEHO_S3_SRC | — | referenced, not traced to a specific task this session | CEHO (Homans) equivalent of the above pattern |
EC_S3_SRC, EC_S3_ERP_SRC, EC_S3_ERP_PATH | STS-assumed-role path | jobs-alg-all.py:Download-EC-ERP-Data | EC's raw ERP source (STS role) |
GM_S3_SRC, GM_S3_MINCRON_SRC | mincron | jobs-alg-all.py:Download-GM-ERP-Data | GM's Mincron ERP export source |
NEUCO_S3_SRC, NEUCO_S3_PIM_SRC | — | jobs-alg-all.py (NeuCo tasks, gated by Airflow Variable) | NeuCo source — see opportunities.md, not reachable from W1 |
PP_S3_SRC, PP_AVAILABILITY_S3_SRC | — | jobs-alg-all.py:Download-PP-ERP-Data | PP's raw ERP availability source |
SHOPWARE_S3_ARCHIVE | shopware_import_files, supply_sync_feed | ecom_w1_full:Upload-Shopware, Upload-Supply-Sync | Shopware import package + Supply-Sync Algolia feed archive destination |
SS_S3_SRC | pending | ecom_w1_full:Download-Sales | Sales-quantity-shipped source feed |
TE_S3_ERP_SRC | — | jobs-alg-all.py (TE availability conversion) | TE's raw ERP source |
*_S3_ARGS (one per BU + ALL/W1/SHOPWARE/SS) | — | passed as s3args to the shared template_download/template_upload bash templates | CLI flags (e.g. --recursive, profile/region overrides) for the corresponding _SRC/_ARCHIVE bucket's transfer — not a bucket identifier itself |
To fully resolve this table: either (a) get MWAA environment-config read access (airflow:GetEnvironment or equivalent) and inspect the environment's variable definitions directly, or (b) pull a BashOperator task log for one of the Download-*/Upload-* tasks in the table above — Airflow logs the fully-rendered bash command, which would show the resolved bucket name in the aws s3 cp/sync invocation.