"""表 2 加购行:领取、成功盖章、失败改状态。成功不插新行。""" from __future__ import annotations from collections.abc import Callable from dataclasses import dataclass from typing import Any from db import RepoUnavailable, connect_from_env, oracle_dsn OPERATE_TYPE_ADD_CART = 1 STATUS_OK = 1 STATUS_FAIL = 2 PUSH_NOT_SENT = 0 STAMP_SELECTION = "系统管理员" STAMP_CLAIMING = "加购中" STAMP_DONE = "加购班" FAIL_REASON_MAX = 500 _CLAIM_SQL = """ UPDATE rpa_cart_settlement_goods_data SET update_user_name = :stamp, update_time = SYSDATE WHERE fid = :fid AND operate_type = 1 AND status = 1 AND push_status = 0 AND NVL(update_user_name, :selection_stamp) = :selection_stamp """ _SUCCESS_SQL = """ UPDATE rpa_cart_settlement_goods_data SET update_user_name = :stamp, fail_reason = NULL, update_time = SYSDATE WHERE fid = :fid AND update_user_name = :claim_stamp """ _FAIL_SQL = """ UPDATE rpa_cart_settlement_goods_data SET status = :status, fail_reason = :fail_reason, update_user_name = :stamp, update_time = SYSDATE WHERE fid = :fid AND update_user_name = :claim_stamp """ @dataclass class CartRow: fid: int oms_id: int platform_type: int supplier_name: str bar_code: str purchase_qty: int status: int fail_reason: str | None push_status: int update_user_name: str operate_type: int = OPERATE_TYPE_ADD_CART def clip_reason(text: str) -> str: text = (text or "").strip() if len(text) <= FAIL_REASON_MAX: return text return text[: FAIL_REASON_MAX - 1] + "…" def _is_pending(row: dict[str, Any]) -> bool: stamp = row.get("update_user_name") or STAMP_SELECTION return ( int(row.get("operate_type") or 0) == OPERATE_TYPE_ADD_CART and int(row.get("status") or 0) == STATUS_OK and int(row.get("push_status") or 0) == PUSH_NOT_SENT and stamp == STAMP_SELECTION ) def _row_from_dict(data: dict[str, Any]) -> CartRow: qty = data.get("purchase_qty") return CartRow( fid=int(data["fid"]), oms_id=int(data["oms_id"]), platform_type=int(data["platform_type"]), supplier_name=str(data.get("supplier_name") or ""), bar_code=str(data.get("bar_code") or ""), purchase_qty=int(qty) if qty is not None else 0, status=int(data.get("status") or 0), fail_reason=data.get("fail_reason"), push_status=int(data.get("push_status") or 0), update_user_name=str(data.get("update_user_name") or ""), operate_type=int(data.get("operate_type") or OPERATE_TYPE_ADD_CART), ) class MemoryCartRepo: def __init__(self, rows: list[dict[str, Any]]): self.rows: dict[int, dict[str, Any]] = {int(row["fid"]): dict(row) for row in rows} def claim_pending( self, limit: int | None = None, platform_types: list[int] | None = None ) -> list[CartRow]: pending = sorted( (row for row in self.rows.values() if _is_pending(row)), key=lambda row: (int(row["oms_id"]), int(row["fid"])), ) if platform_types is not None: allowed = set(platform_types) pending = [row for row in pending if int(row["platform_type"]) in allowed] if limit is not None: pending = pending[:limit] claimed: list[CartRow] = [] for row in pending: row["update_user_name"] = STAMP_CLAIMING claimed.append(_row_from_dict(row)) return claimed def mark_success(self, fid: int) -> None: row = self.rows[fid] row["update_user_name"] = STAMP_DONE row["fail_reason"] = None def mark_fail(self, fid: int, reason: str) -> None: row = self.rows[fid] row["status"] = STATUS_FAIL row["fail_reason"] = clip_reason(reason) row["update_user_name"] = STAMP_DONE class OracleCartRepo: def __init__(self, connect: Callable): self._connect = connect @classmethod def from_env(cls) -> "OracleCartRepo": oracle_dsn() return cls(connect=connect_from_env) def claim_pending( self, limit: int | None = None, platform_types: list[int] | None = None ) -> list[CartRow]: conn = self._connect() try: with conn.cursor() as cur: sql, params = _pending_sql(limit=limit, platform_types=platform_types) cur.execute(sql, params) fetched = [_row_from_cursor(cur, rec) for rec in cur.fetchall()] claimed: list[CartRow] = [] for row in fetched: cur.execute( _CLAIM_SQL, { "fid": row.fid, "stamp": STAMP_CLAIMING, "selection_stamp": STAMP_SELECTION, }, ) if cur.rowcount == 1: row.update_user_name = STAMP_CLAIMING claimed.append(row) conn.commit() return claimed except Exception as exc: try: conn.rollback() except Exception: pass raise RepoUnavailable(str(exc)) from exc finally: conn.close() def mark_success(self, fid: int) -> None: self._execute( _SUCCESS_SQL, {"fid": int(fid), "stamp": STAMP_DONE, "claim_stamp": STAMP_CLAIMING}, ) def mark_fail(self, fid: int, reason: str) -> None: self._execute( _FAIL_SQL, { "fid": int(fid), "status": STATUS_FAIL, "fail_reason": clip_reason(reason), "stamp": STAMP_DONE, "claim_stamp": STAMP_CLAIMING, }, ) def _execute(self, sql: str, params: dict[str, Any]) -> None: conn = self._connect() try: with conn.cursor() as cur: cur.execute(sql, params) conn.commit() except Exception as exc: try: conn.rollback() except Exception: pass raise RepoUnavailable(str(exc)) from exc finally: conn.close() def _row_from_cursor(cur, rec) -> CartRow: cols = [item[0].lower() for item in cur.description] return _row_from_dict(dict(zip(cols, rec))) def _pending_sql( *, limit: int | None, platform_types: list[int] | None ) -> tuple[str, dict[str, Any]]: params: dict[str, Any] = {"selection_stamp": STAMP_SELECTION} filters = [ "operate_type = 1", "status = 1", "push_status = 0", "NVL(update_user_name, :selection_stamp) = :selection_stamp", ] if platform_types is not None: if not platform_types: filters.append("1 = 0") else: keys: list[str] = [] for index, platform in enumerate(platform_types): key = f"p{index}" keys.append(f":{key}") params[key] = int(platform) filters.append(f"platform_type IN ({', '.join(keys)})") sql = f""" SELECT fid, oms_id, platform_type, supplier_name, bar_code, purchase_qty, status, fail_reason, push_status, update_user_name, operate_type FROM rpa_cart_settlement_goods_data WHERE {' AND '.join(filters)} ORDER BY oms_id, fid """ if limit is not None: params["limit"] = int(limit) sql = f"SELECT * FROM (\n{sql}\n) WHERE ROWNUM <= :limit" return sql, params