Ecommerce Module ZH

English version

電商模組 — beam_pipeline_toolkit/ecommerce/

一個選用、可直接改造使用的電商情境擴充模組：sessions、orders、customers,以及一列語意化的每日 metric。完全建構於 pipeline/、controlplane/、log/、safety/ 的公開 API 之上——證明通用核心不需任何修改就能支援真實領域。這裡沒有寫死任何特定公司的欄位名稱或商業規則;請自行調整,不要當作標準答案直接套用。

schemas.py

NamedTuple:SessionRecord、OrderRecord、CustomerRecord、DailyMetricRow(sessions_count、conversion_rate、order_count、average_order_value、revenue、registered_members)、InvalidRecord(刻意不攜帶原始 payload——只保留足以歸咎於某個 key 及原因代碼的資訊)。金額欄位一律為十進位字面字串,絕不使用二進位浮點數。

metrics.py

compute_daily_metrics(sessions_count, order_count, revenue, registered_members) — 純函式,Decimal 安全。average_order_value = revenue / order_count(無訂單時為 0)、conversion_rate = order_count / sessions_count(無 session 時為 0),皆以 ROUND_HALF_UP 進位。

order_policies.py

可替換的訂單有效性政策模式——不同的報表路徑對於哪些訂單該被計入可能有不同看法(取消?退貨?deleted_at 標記?)。將每條規則寫成獨立函式並明確選用,或同時執行兩個政策以精確找出差異所在。

• exclude_cancelled_and_returned(order) — 通用預設值:排除狀態為 cancelled/canceled,以及 return_status 為 returned/refunded 的訂單。
• accept_all_orders(order) — 最簡單的基準,永遠回傳 True。
• partition_by_policy(orders, policy) — 供臨時/離線使用的輔助函式,拆分為 (eligible, excluded)。

validation.py

• build_validate_fn(required_fields, date_field="metric_date") — 回傳與 pipeline.validation.Validate 相容的 validate_fn:必填欄位是否存在、ISO 日期是否有效、session 事件的 sessions_count 是否非負、order 事件的 total_price 是否可解析為十進位數。
• deduplicate_sessions(sessions, sequence_field="sequence") — 針對同一個 key 的重複紀錄,保留依 sequence 排序後的第一筆;回傳 (kept, duplicate_count)。需要分組後的視角,因此由 build_outcome_fn 呼叫,而非 validate_fn。

pipeline.py

完整的實作範例,直接組合以上所有元件與通用核心(而非透過 pipeline.main.build_pipeline,因為需要 order_policy 提供的額外分類/過濾步驟):

raw_messages
  → ParseNormalize
  → Validate (build_validate_fn)
  → TagOrigin (session / customer / order) 或 TagInvalid
  → AggregateAndEmit,以 (run_id, shop_id, metric_date) 分組
      → build_outcome_fn:
          - deduplicate_sessions
          - 依 order_policy 拆分訂單
          - compute_daily_metrics
          - 四階段遙測時間軸(SOURCE_READ、VALIDATE、FILTER、AGGREGATE)
          - invalid_summary(重複與結構性錯誤計數)

• default_schema(metrics_namespace="ecommerce_pipeline") — 此模組自己的 PipelineSchema,stage 為 {SOURCE_READ, VALIDATE, FILTER, AGGREGATE},status 為 {PASS, WARNING, COMPLETED}。
• build_outcome_fn_factory(schema, order_policy=exclude_cancelled_and_returned) — 建構上述的 build_outcome_fn。
• build_ecommerce_pipeline(raw_messages, schema=None, required_fields=DEFAULT_REQUIRED_FIELDS, order_policy=exclude_cancelled_and_returned) — 進入點。將此模組複製到自己的應用程式並替換為真實欄位名稱/規則,或若此形狀已符合需求則直接呼叫。