# -*- coding: utf-8 -*- """库存变更任务""" import json from .base_task import BaseTask, TaskContext from loaders.facts.inventory_change import InventoryChangeLoader from models.parsers import TypeParser class InventoryChangeTask(BaseTask): """同步库存变化记录""" def get_task_code(self) -> str: return "INVENTORY_CHANGE" def extract(self, context: TaskContext) -> dict: params = self._merge_common_params( { "siteId": context.store_id, "startTime": TypeParser.format_timestamp(context.window_start, self.tz), "endTime": TypeParser.format_timestamp(context.window_end, self.tz), } ) records, _ = self.api.get_paginated( endpoint="/GoodsStockManage/QueryGoodsOutboundReceipt", params=params, page_size=self.config.get("api.page_size", 200), data_path=("data",), list_key="queryDeliveryRecordsList", ) return {"records": records} def transform(self, extracted: dict, context: TaskContext) -> dict: parsed, skipped = [], 0 for raw in extracted.get("records", []): mapped = self._parse_change(raw, context.store_id) if mapped: parsed.append(mapped) else: skipped += 1 return { "records": parsed, "fetched": len(extracted.get("records", [])), "skipped": skipped, } def load(self, transformed: dict, context: TaskContext) -> dict: loader = InventoryChangeLoader(self.db) inserted, updated, loader_skipped = loader.upsert_changes(transformed["records"]) return { "fetched": transformed["fetched"], "inserted": inserted, "updated": updated, "skipped": transformed["skipped"] + loader_skipped, "errors": 0, } def _parse_change(self, raw: dict, store_id: int) -> dict | None: change_id = TypeParser.parse_int( raw.get("siteGoodsStockId") or raw.get("site_goods_stock_id") ) if not change_id: self.logger.warning("跳过缺少库存变动ID的记录: %s", raw) return None return { "store_id": store_id, "change_id": change_id, "site_goods_id": TypeParser.parse_int( raw.get("siteGoodsId") or raw.get("site_goods_id") ), "stock_type": raw.get("stockType") or raw.get("stock_type"), "goods_name": raw.get("goodsName"), "change_time": TypeParser.parse_timestamp( raw.get("createTime") or raw.get("create_time"), self.tz ), "start_qty": TypeParser.parse_int(raw.get("startNum")), "end_qty": TypeParser.parse_int(raw.get("endNum")), "change_qty": TypeParser.parse_int(raw.get("changeNum")), "unit": raw.get("unit"), "price": TypeParser.parse_decimal(raw.get("price")), "operator_name": raw.get("operatorName"), "remark": raw.get("remark"), "goods_category_id": TypeParser.parse_int(raw.get("goodsCategoryId")), "goods_second_category_id": TypeParser.parse_int( raw.get("goodsSecondCategoryId") ), "raw_data": json.dumps(raw, ensure_ascii=False), }