AutoRpa/rpa_oms/app/outbound/oracle_repo.py

# 接口 2 / 4：读 push_status=0，按组在 SQL 侧截断。
#
# 连接由调用方从 rpa_oracle 池借入，仓库不 connect、不 close。
# 不向订单中心发 HTTP。mark_pushed 备好回写，本段不接推送。

from __future__ import annotations

from rpa_oracle.errors import RepoUnavailable

from ..inbound.oracle_repo import in_binds

_LIST_UNPUSHED_SQL = """
WITH src AS (
    SELECT
        fid, operate_type, oms_id, platform_type, supplier_name, bar_code,
        purchase_price, purchase_qty, purchase_total, status, fail_reason,
        pay_order_no, pay_order_dtl_id, org_code, push_status
    FROM rpa_cart_settlement_goods_data
    WHERE push_status = 0
),
ranked AS (
    SELECT
        src.*,
        COUNT(*) OVER (PARTITION BY oms_id) AS grp_size
    FROM src
),
grp AS (
    SELECT oms_id, MIN(grp_size) AS grp_size
    FROM ranked
    GROUP BY oms_id
),
run AS (
    SELECT
        oms_id,
        grp_size,
        SUM(grp_size) OVER (ORDER BY oms_id) AS running_end,
        SUM(grp_size) OVER (ORDER BY oms_id) - grp_size AS running_start
    FROM grp
)
SELECT
    r.fid, r.operate_type, r.oms_id, r.platform_type, r.supplier_name, r.bar_code,
    r.purchase_price, r.purchase_qty, r.purchase_total, r.status, r.fail_reason,
    r.pay_order_no, r.pay_order_dtl_id, r.org_code, r.push_status
FROM ranked r
JOIN run ON run.oms_id = r.oms_id
WHERE run.running_start = 0 OR run.running_end <= :limit
ORDER BY r.oms_id, r.fid
"""

_LIST_AGREEMENT_SQL = """
WITH src AS (
    SELECT
        fid, oms_id, action_code, contract_status, contract_fail_reason,
        contract_no, agreement_status, agreement_fail_reason,
        agreement_id, agreement_dtl_id, org_code, push_status,
        CASE
            WHEN TRIM(contract_no) IS NULL THEN 'F:' || TO_CHAR(fid)
            ELSE 'N:' || TRIM(contract_no)
        END AS grp_key
    FROM rpa_contract_agreement_result
    WHERE push_status = 0
),
ranked AS (
    SELECT
        src.*,
        COUNT(*) OVER (PARTITION BY grp_key) AS grp_size
    FROM src
),
grp AS (
    SELECT grp_key, MIN(grp_size) AS grp_size, MIN(fid) AS grp_first_fid
    FROM ranked
    GROUP BY grp_key
),
run AS (
    SELECT
        grp_key,
        grp_size,
        SUM(grp_size) OVER (ORDER BY grp_first_fid, grp_key) AS running_end,
        SUM(grp_size) OVER (ORDER BY grp_first_fid, grp_key) - grp_size AS running_start
    FROM grp
)
SELECT
    r.fid, r.oms_id, r.action_code, r.contract_status, r.contract_fail_reason,
    r.contract_no, r.agreement_status, r.agreement_fail_reason,
    r.agreement_id, r.agreement_dtl_id, r.org_code, r.push_status
FROM ranked r
JOIN run ON run.grp_key = r.grp_key
WHERE run.running_start = 0 OR run.running_end <= :limit
ORDER BY r.fid
"""

_CART_TABLE = "rpa_cart_settlement_goods_data"
_AGREEMENT_TABLE = "rpa_contract_agreement_result"


# 接口 2 连 Oracle 的仓库，管表 2 未推取数和回写。
class OracleCartSettlementRepo:
    # 挂上调用方借来的连接。不要在仓库里再 connect 或 close。
    def __init__(self, conn):
        self._conn = conn

    # 捞 push_status=0，同一 oms_id 不拆组，在 SQL 侧按 limit 截断。
    def list_unpushed(self, limit: int = 500) -> list[dict]:
        return _fetch(self._conn, _LIST_UNPUSHED_SQL, {"limit": limit})

    # 一次 UPDATE 回写表 2 的 push_status。空列表不访问库。
    def mark_pushed(self, fids: list[int], status: int, reason: str | None) -> None:
        _mark(self._conn, _CART_TABLE, fids, status, reason, "SYSDATE")


# 接口 4 连 Oracle 的仓库，管表 4 未推取数和回写。
class OracleContractAgreementRepo:
    # 挂上调用方借来的连接。不要在仓库里再 connect 或 close。
    def __init__(self, conn):
        self._conn = conn

    # 捞 push_status=0，同一非空 contract_no 不拆组，在 SQL 侧按 limit 截断。
    def list_unpushed(self, limit: int = 500) -> list[dict]:
        return _fetch(self._conn, _LIST_AGREEMENT_SQL, {"limit": limit})

    # 一次 UPDATE 回写表 4 的 push_status。空列表不访问库。
    def mark_pushed(self, fids: list[int], status: int, reason: str | None) -> None:
        _mark(self._conn, _AGREEMENT_TABLE, fids, status, reason, "SYSTIMESTAMP")


# 执行查询并把列名转小写。失败抛「数据库不可用」，不含 ORA。
def _fetch(conn, sql: str, binds: dict) -> list[dict]:
    try:
        with conn.cursor() as cur:
            cur.execute(sql, binds)
            cols = [item[0].lower() for item in cur.description]
            return [dict(zip(cols, row)) for row in cur.fetchall()]
    except Exception as exc:
        raise RepoUnavailable("数据库不可用") from exc


# 一次 UPDATE ... WHERE fid IN (...)。失败回滚并抛「数据库不可用」。
def _mark(
    conn,
    table: str,
    fids: list[int],
    status: int,
    reason: str | None,
    time_expr: str,
) -> None:
    if not fids:
        return
    placeholders, binds = in_binds(fids)
    binds["status"] = status
    binds["reason"] = reason
    sql = (
        f"UPDATE {table} "
        f"SET push_status = :status, push_result = :reason, "
        f"push_time = {time_expr}, update_time = {time_expr} "
        f"WHERE fid IN ({placeholders})"
    )
    try:
        with conn.cursor() as cur:
            cur.execute(sql, binds)
        conn.commit()
    except Exception as exc:
        _rollback(conn)
        raise RepoUnavailable("数据库不可用") from exc


# 回滚本连接已写入的行；回滚失败忽略。
def _rollback(conn) -> None:
    try:
        conn.rollback()
    except Exception:
        pass