351 lines
12 KiB
Python
351 lines
12 KiB
Python
# 表 3 领取、表 4 写入。
|
||
#
|
||
# 从表 3 领取 process_status=0 的行改成 3，合单成功后写表 4、表 3 改成 1。
|
||
# 进程中途退出时行停在 3，超过租约（process_time 30 分钟）才能再领。
|
||
# 跳过协议重录行。不推接口 4。
|
||
|
||
from __future__ import annotations
|
||
|
||
from contextlib import contextmanager
|
||
from dataclasses import dataclass
|
||
from datetime import date, datetime, timedelta
|
||
from decimal import Decimal
|
||
from typing import Any
|
||
|
||
from rpa_oracle import RepoUnavailable, job_pool, update_by_fid_chunks
|
||
from rpa_oracle.audit import ShiftError, save_shift_errors as write_shift_errors
|
||
|
||
LEASE_MINUTES = 30
|
||
|
||
_SELECT_BODY = """
|
||
SELECT
|
||
fid, oms_id, supplier_id, supplier_name,
|
||
contract_sign_date, contract_start_date, contract_end_date,
|
||
payment_term, goods_id, goods_name, reg_no, spec, unit,
|
||
qty, unit_price, purchaser_id, purchaser_name, planner_id, planner_name,
|
||
contract_entry_status, contract_no, process_status, org_code,
|
||
goods_mdm_code, supplier_mdm_code,
|
||
NVL(process_result, '') AS process_result
|
||
FROM rpa_purchase_contract_data
|
||
WHERE (
|
||
process_status = 0
|
||
OR (process_status = 3 AND process_time < SYSDATE - NUMTODSINTERVAL(:lease_minutes, 'MINUTE'))
|
||
)
|
||
AND NOT (NVL(contract_entry_status, -1) = 0
|
||
AND contract_no IS NOT NULL
|
||
AND LENGTH(TRIM(contract_no)) > 0)
|
||
ORDER BY fid
|
||
"""
|
||
|
||
_SELECT_PENDING = _SELECT_BODY + "\nFOR UPDATE SKIP LOCKED"
|
||
|
||
_SELECT_PENDING_LIMIT = f"""
|
||
SELECT * FROM (
|
||
{_SELECT_BODY}
|
||
) WHERE ROWNUM <= :limit
|
||
FOR UPDATE SKIP LOCKED
|
||
"""
|
||
|
||
_CLAIM_SQL = """
|
||
UPDATE rpa_purchase_contract_data
|
||
SET process_status = 3,
|
||
process_time = SYSDATE,
|
||
update_time = SYSDATE
|
||
WHERE fid IN ({placeholders})
|
||
"""
|
||
|
||
_INSERT_TABLE4_SQL = """
|
||
INSERT INTO rpa_contract_agreement_result (
|
||
oms_id, action_code, contract_status, contract_fail_reason, contract_no,
|
||
agreement_status, agreement_fail_reason, agreement_id, agreement_dtl_id,
|
||
push_status, process_status, create_time, update_time,
|
||
create_user_id, create_user_name, update_user_id, update_user_name, org_code
|
||
) VALUES (
|
||
:oms_id, :action_code, :contract_status, :contract_fail_reason, :contract_no,
|
||
:agreement_status, :agreement_fail_reason, :agreement_id, :agreement_dtl_id,
|
||
0, :process_status, SYSTIMESTAMP, SYSTIMESTAMP,
|
||
0, '系统管理员', 0, '系统管理员', :org_code
|
||
)
|
||
"""
|
||
|
||
_COMPLETE_SQL = """
|
||
UPDATE rpa_purchase_contract_data
|
||
SET process_status = :process_status,
|
||
process_result = :process_result,
|
||
process_time = SYSTIMESTAMP,
|
||
update_time = SYSTIMESTAMP
|
||
WHERE fid = :fid AND process_status = 3
|
||
"""
|
||
|
||
_TABLE4_BIND_KEYS = (
|
||
"oms_id",
|
||
"action_code",
|
||
"contract_status",
|
||
"contract_fail_reason",
|
||
"contract_no",
|
||
"agreement_status",
|
||
"agreement_fail_reason",
|
||
"agreement_id",
|
||
"agreement_dtl_id",
|
||
"process_status",
|
||
"org_code",
|
||
)
|
||
|
||
|
||
# 表 3 一行待录或处理中的采购合同。
|
||
@dataclass
|
||
class PurchaseContractRow:
|
||
fid: int
|
||
oms_id: int
|
||
supplier_id: str
|
||
supplier_name: str
|
||
contract_sign_date: date | None
|
||
contract_start_date: date | None
|
||
contract_end_date: date | None
|
||
payment_term: int
|
||
goods_id: str
|
||
goods_name: str
|
||
reg_no: str
|
||
spec: str
|
||
unit: str
|
||
qty: Decimal | float | int
|
||
unit_price: Decimal | float | int
|
||
purchaser_id: str
|
||
purchaser_name: str
|
||
planner_id: str
|
||
planner_name: str
|
||
contract_entry_status: int | None
|
||
contract_no: str | None
|
||
process_status: int
|
||
org_code: str | None
|
||
goods_mdm_code: str | None
|
||
supplier_mdm_code: str | None
|
||
process_result: str = ""
|
||
process_time: datetime | None = None
|
||
|
||
|
||
# 判断是否只录协议行：contract_entry_status=0 且已有合同号。
|
||
def _is_protocol_rerun(row: PurchaseContractRow) -> bool:
|
||
if row.contract_entry_status is not None and int(row.contract_entry_status) == 0:
|
||
return bool((row.contract_no or "").strip())
|
||
return False
|
||
|
||
|
||
# 判断内存行是否可领：0，或 3 且 process_time 已过租约。协议重录行永不领。
|
||
def _is_claimable(row: PurchaseContractRow, now: datetime) -> bool:
|
||
if _is_protocol_rerun(row):
|
||
return False
|
||
if int(row.process_status or 0) == 0:
|
||
return True
|
||
if int(row.process_status or 0) == 3 and row.process_time is not None:
|
||
return now - row.process_time >= timedelta(minutes=LEASE_MINUTES)
|
||
return False
|
||
|
||
|
||
# 回滚当前事务。回滚本身失败则忽略。
|
||
def _rollback(conn) -> None:
|
||
try:
|
||
conn.rollback()
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
# 把查询结果行收成 PurchaseContractRow。
|
||
def _row_from_record(cols: list[str], rec) -> PurchaseContractRow:
|
||
data = dict(zip(cols, rec))
|
||
return PurchaseContractRow(
|
||
fid=int(data["fid"]),
|
||
oms_id=int(data["oms_id"]),
|
||
supplier_id=data["supplier_id"],
|
||
supplier_name=data["supplier_name"],
|
||
contract_sign_date=data["contract_sign_date"],
|
||
contract_start_date=data["contract_start_date"],
|
||
contract_end_date=data["contract_end_date"],
|
||
payment_term=int(data["payment_term"]),
|
||
goods_id=data["goods_id"],
|
||
goods_name=data["goods_name"],
|
||
reg_no=data["reg_no"],
|
||
spec=data["spec"],
|
||
unit=data["unit"],
|
||
qty=data["qty"],
|
||
unit_price=data["unit_price"],
|
||
purchaser_id=data["purchaser_id"],
|
||
purchaser_name=data["purchaser_name"],
|
||
planner_id=data["planner_id"],
|
||
planner_name=data["planner_name"],
|
||
contract_entry_status=data["contract_entry_status"],
|
||
contract_no=data["contract_no"],
|
||
process_status=int(data["process_status"]),
|
||
org_code=data["org_code"],
|
||
goods_mdm_code=data["goods_mdm_code"],
|
||
supplier_mdm_code=data["supplier_mdm_code"],
|
||
process_result=data.get("process_result") or "",
|
||
)
|
||
|
||
|
||
# 合同班次的内存仓库，测合单和班次编排，不测 SQL。
|
||
class MemoryContractRepo:
|
||
# 用给定的表 3 行初始化。同一 fid 后写覆盖先写。
|
||
def __init__(self, rows: list[PurchaseContractRow]):
|
||
self.table3 = {row.fid: row for row in rows}
|
||
self.table4: list[dict] = []
|
||
self.shift_errors: list[ShiftError] = []
|
||
self._next_fid = 1
|
||
|
||
# 领取待录合同任务，把 process_status 从 0 或过期 3 改成 3。
|
||
#
|
||
# 未过租约的 3 不领。协议重录行不领。limit 为空表示不限制条数。now 只给单测钉住时钟。
|
||
def claim_pending(
|
||
self, limit: int | None = None, *, now: datetime | None = None
|
||
) -> list[PurchaseContractRow]:
|
||
clock = now or datetime.now()
|
||
pending = sorted(
|
||
(row for row in self.table3.values() if _is_claimable(row, clock)),
|
||
key=lambda row: row.fid,
|
||
)
|
||
if limit is not None:
|
||
pending = pending[:limit]
|
||
for row in pending:
|
||
row.process_status = 3
|
||
row.process_time = clock
|
||
return pending
|
||
|
||
# 把合单结果写入内存表 4，并改表 3 的 process_status。
|
||
def save_group(self, payload: Any) -> None:
|
||
for row in payload.table4_rows:
|
||
item = dict(row)
|
||
item["fid"] = self._next_fid
|
||
self._next_fid += 1
|
||
self.table4.append(item)
|
||
for fid in payload.fids:
|
||
target = self.table3[fid]
|
||
target.process_status = payload.process_status
|
||
target.process_result = payload.process_result
|
||
|
||
# 记下未捕获异常原文。不改表 3 / 表 4。
|
||
def save_shift_errors(self, items: list[ShiftError]) -> None:
|
||
self.shift_errors.extend(items)
|
||
|
||
|
||
# 合同班次连 Oracle 的仓库，管表 3 领取和表 4 写入。
|
||
#
|
||
# 连接从 rpa_oracle 连接池取。领取戳是 process_status=3。
|
||
# 进程退出后行停在 3，超过租约才能再领。
|
||
class OracleContractRepo:
|
||
# 挂上连接池。方法里按次 acquire，班次 session 内复用一条。
|
||
def __init__(self, pool):
|
||
self._pool = pool
|
||
self._held = None
|
||
|
||
# 按环境变量建班次池。缺 HOST / 用户 / 密码抛「数据库不可用」。
|
||
@classmethod
|
||
def from_env(cls) -> "OracleContractRepo":
|
||
return cls(pool=job_pool())
|
||
|
||
# 未捕获异常原文进共用审计表。独立事务，失败只记日志。
|
||
def save_shift_errors(self, items: list[ShiftError]) -> None:
|
||
write_shift_errors(self._pool, items)
|
||
|
||
# 班次内借一条连接，领取和落表共用。每笔 save_group 单独提交。
|
||
@contextmanager
|
||
def session(self):
|
||
if self._held is not None:
|
||
yield
|
||
return
|
||
with self._pool.acquire() as conn:
|
||
self._held = conn
|
||
try:
|
||
yield
|
||
finally:
|
||
self._held = None
|
||
|
||
# 有 session 时交出已借连接，否则从池里新借一条。
|
||
def _acquire(self):
|
||
if self._held is not None:
|
||
|
||
# 交出班次已借的那条连接，退出时不归还。
|
||
@contextmanager
|
||
def _held():
|
||
yield self._held
|
||
|
||
return _held()
|
||
return self._pool.acquire()
|
||
|
||
def claim_pending(self, limit: int | None = None) -> list[PurchaseContractRow]:
|
||
"""领取待录合同任务，把表 3 的 process_status 从 0 改成 3。
|
||
|
||
使用 FOR UPDATE SKIP LOCKED，两个班次不会领到同一行。
|
||
更新按每段 500 个 fid 执行，同一事务，任一段失败整笔回滚。
|
||
进程退出后行停在 3，超过租约（process_time）才能再领。
|
||
limit 仍只截这一班领多少；为空表示不限制。没有可领的行时返回空列表。
|
||
"""
|
||
binds: dict = {"lease_minutes": LEASE_MINUTES}
|
||
try:
|
||
with self._acquire() as conn:
|
||
try:
|
||
with conn.cursor() as cur:
|
||
if limit is None:
|
||
cur.execute(_SELECT_PENDING, binds)
|
||
else:
|
||
cur.execute(
|
||
_SELECT_PENDING_LIMIT,
|
||
{**binds, "limit": int(limit)},
|
||
)
|
||
cols = [item[0].lower() for item in cur.description]
|
||
fetched = [_row_from_record(cols, rec) for rec in cur.fetchall()]
|
||
if fetched:
|
||
update_by_fid_chunks(
|
||
cur,
|
||
_CLAIM_SQL,
|
||
[row.fid for row in fetched],
|
||
)
|
||
for row in fetched:
|
||
row.process_status = 3
|
||
conn.commit()
|
||
return fetched
|
||
except Exception as exc:
|
||
_rollback(conn)
|
||
if isinstance(exc, RepoUnavailable):
|
||
raise
|
||
raise RepoUnavailable("数据库不可用") from exc
|
||
except RepoUnavailable:
|
||
raise
|
||
except Exception as exc:
|
||
raise RepoUnavailable("数据库不可用") from exc
|
||
|
||
# 把合单结果写入表 4，并把表 3 的 process_status 改成 1 或 2。
|
||
#
|
||
# 表 4 插入和表 3 更新各一次 executemany，同一连接同一事务。
|
||
# 某一侧没有行就跳过那一次调用。库故障对外「数据库不可用」，不含 ORA 原文。
|
||
def save_group(self, payload: Any) -> None:
|
||
table4_binds = [
|
||
{key: row.get(key) for key in _TABLE4_BIND_KEYS}
|
||
for row in payload.table4_rows
|
||
]
|
||
complete_binds = [
|
||
{
|
||
"process_status": payload.process_status,
|
||
"process_result": payload.process_result,
|
||
"fid": fid,
|
||
}
|
||
for fid in payload.fids
|
||
]
|
||
try:
|
||
with self._acquire() as conn:
|
||
try:
|
||
with conn.cursor() as cur:
|
||
if table4_binds:
|
||
cur.executemany(_INSERT_TABLE4_SQL, table4_binds)
|
||
if complete_binds:
|
||
cur.executemany(_COMPLETE_SQL, complete_binds)
|
||
conn.commit()
|
||
except Exception as exc:
|
||
_rollback(conn)
|
||
if isinstance(exc, RepoUnavailable):
|
||
raise
|
||
raise RepoUnavailable("数据库不可用") from exc
|
||
except RepoUnavailable:
|
||
raise
|
||
except Exception as exc:
|
||
raise RepoUnavailable("数据库不可用") from exc
|