588 lines
21 KiB
Python
588 lines
21 KiB
Python
# 表 4 起草行领取、表 3 只录协议行领取、表 4 写入。
|
||
#
|
||
# 表 4 领取时 process_status 从 0 改成 3；表 3 只录协议行同样 0 改成 3。
|
||
# 进程中途退出时行停在 3，超过租约（表 4 update_time / 表 3 process_time 30 分钟）才能再领。
|
||
# 未过期把表 4 改回 0、表 3 改回 0 再跑。人名固定系统管理员。不推接口 4。
|
||
# 同一合同号整组领取，limit 不把一组切开。
|
||
|
||
from __future__ import annotations
|
||
|
||
from contextlib import contextmanager
|
||
from dataclasses import dataclass, field
|
||
from datetime import datetime, timedelta
|
||
|
||
from rpa_oracle import RepoUnavailable, job_pool, update_by_fid_chunks
|
||
from rpa_oracle.audit import ShiftError, save_shift_errors as write_shift_errors
|
||
from persist import USER_NAME, GroupSave
|
||
|
||
ACTION_CONTRACT = 1
|
||
CONTRACT_DRAFT = 1
|
||
PROCESS_PENDING = 0
|
||
PROCESS_CLAIMING = 3
|
||
ENTRY_PROTOCOL_ONLY = 0
|
||
LEASE_MINUTES = 30
|
||
|
||
_SELECT_DRAFTS_BODY = """
|
||
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,
|
||
process_status
|
||
FROM rpa_contract_agreement_result src
|
||
WHERE action_code = 1
|
||
AND contract_status = 1
|
||
AND contract_no IS NOT NULL
|
||
AND LENGTH(TRIM(contract_no)) > 0
|
||
AND (
|
||
process_status = 0
|
||
OR (
|
||
process_status = 3
|
||
AND update_time < SYSDATE - NUMTODSINTERVAL(:lease_minutes, 'MINUTE')
|
||
)
|
||
)
|
||
AND NOT EXISTS (
|
||
SELECT 1
|
||
FROM rpa_contract_agreement_result held
|
||
WHERE held.action_code = 1
|
||
AND TRIM(held.contract_no) = TRIM(src.contract_no)
|
||
AND held.process_status = 3
|
||
AND (
|
||
held.update_time IS NULL
|
||
OR held.update_time >= SYSDATE - NUMTODSINTERVAL(:lease_minutes, 'MINUTE')
|
||
)
|
||
)
|
||
ORDER BY fid
|
||
"""
|
||
|
||
_SELECT_DRAFTS = _SELECT_DRAFTS_BODY + "\nFOR UPDATE SKIP LOCKED"
|
||
|
||
# 起草行按合同号在领取后截组，SQL 不按行 ROWNUM 切开。
|
||
_SELECT_DRAFTS_LIMIT = _SELECT_DRAFTS
|
||
|
||
_CLAIM_DRAFT_SQL = """
|
||
UPDATE rpa_contract_agreement_result
|
||
SET process_status = 3,
|
||
update_user_id = 0,
|
||
update_user_name = '系统管理员',
|
||
update_time = SYSTIMESTAMP
|
||
WHERE fid IN ({placeholders})
|
||
"""
|
||
|
||
_SELECT_PROTOCOLS_BODY = """
|
||
SELECT
|
||
fid, oms_id, contract_no, contract_entry_status, process_status, org_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 NVL(contract_entry_status, -1) = 0
|
||
AND contract_no IS NOT NULL
|
||
AND LENGTH(TRIM(contract_no)) > 0
|
||
ORDER BY fid
|
||
"""
|
||
|
||
_SELECT_PROTOCOLS = _SELECT_PROTOCOLS_BODY + "\nFOR UPDATE SKIP LOCKED"
|
||
|
||
_SELECT_PROTOCOLS_LIMIT = f"""
|
||
SELECT * FROM (
|
||
{_SELECT_PROTOCOLS_BODY}
|
||
) WHERE ROWNUM <= :limit
|
||
FOR UPDATE SKIP LOCKED
|
||
"""
|
||
|
||
_CLAIM_PROTOCOL_SQL = """
|
||
UPDATE rpa_purchase_contract_data
|
||
SET process_status = 3,
|
||
process_time = SYSTIMESTAMP,
|
||
update_time = SYSTIMESTAMP
|
||
WHERE fid IN ({placeholders})
|
||
"""
|
||
|
||
_UPDATE_DRAFT_SQL = """
|
||
UPDATE rpa_contract_agreement_result
|
||
SET contract_status = :contract_status,
|
||
process_status = :process_status,
|
||
agreement_fail_reason = :agreement_fail_reason,
|
||
update_user_id = 0,
|
||
update_user_name = '系统管理员',
|
||
update_time = SYSTIMESTAMP
|
||
WHERE fid = :fid AND process_status = 3
|
||
"""
|
||
|
||
_RELEASE_DRAFT_SQL = """
|
||
UPDATE rpa_contract_agreement_result
|
||
SET process_status = 0,
|
||
update_user_id = 0,
|
||
update_user_name = '系统管理员',
|
||
update_time = SYSTIMESTAMP
|
||
WHERE fid = :fid AND process_status = 3
|
||
"""
|
||
|
||
_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_PROTOCOL_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",
|
||
)
|
||
|
||
_SELECT_CONTRACT_GOODS = """
|
||
SELECT
|
||
fid, oms_id, supplier_id, supplier_name, supplier_mdm_code,
|
||
planner_id, planner_name, purchaser_name, org_code,
|
||
goods_id, qty, unit_price, goods_source_id, purchase_tax_rate,
|
||
contract_start_date, contract_end_date, payment_term
|
||
FROM rpa_purchase_contract_data
|
||
WHERE TRIM(contract_no) = :contract_no
|
||
OR oms_id IN (
|
||
SELECT oms_id FROM rpa_contract_agreement_result
|
||
WHERE TRIM(contract_no) = :contract_no
|
||
)
|
||
ORDER BY fid
|
||
"""
|
||
|
||
|
||
# 一次领取拿到的表 4 起草行和表 3 只录协议行。
|
||
@dataclass
|
||
class Claimed:
|
||
drafts: list[dict] = field(default_factory=list)
|
||
protocols: list[dict] = field(default_factory=list)
|
||
|
||
|
||
# 合同号去掉空白后非空才算有号。
|
||
def _has_contract_no(row: dict) -> bool:
|
||
return bool((row.get("contract_no") or "").strip())
|
||
|
||
|
||
# 判断是否待查号起草行：action_code=1、contract_status=1、process_status=0、已有合同号。
|
||
def is_draft(row: dict) -> bool:
|
||
if int(row.get("action_code") or 0) != ACTION_CONTRACT:
|
||
return False
|
||
if int(row.get("contract_status") or 0) != CONTRACT_DRAFT:
|
||
return False
|
||
if int(row.get("process_status") or 0) != PROCESS_PENDING:
|
||
return False
|
||
return _has_contract_no(row)
|
||
|
||
|
||
# 判断是否待领的只录协议行：process_status=0、contract_entry_status=0、已有合同号。
|
||
def is_protocol_only(row: dict) -> bool:
|
||
if int(row.get("process_status") or 0) != PROCESS_PENDING:
|
||
return False
|
||
entry = row.get("contract_entry_status")
|
||
if entry is None or int(entry) != ENTRY_PROTOCOL_ONLY:
|
||
return False
|
||
return _has_contract_no(row)
|
||
|
||
|
||
# 合同号去掉空白，作为整组领取的键。
|
||
def _contract_key(row: dict) -> str:
|
||
return (row.get("contract_no") or "").strip()
|
||
|
||
|
||
# 领取时间已满租约才算过期。没有时间不算过期。
|
||
def _lease_expired(claimed_at: datetime | None, now: datetime) -> bool:
|
||
if claimed_at is None:
|
||
return False
|
||
return now - claimed_at >= timedelta(minutes=LEASE_MINUTES)
|
||
|
||
|
||
# 判断同合同号是否仍被未过租约的 process_status=3 占着。
|
||
def _is_fresh_hold(row: dict, now: datetime) -> bool:
|
||
if int(row.get("action_code") or 0) != ACTION_CONTRACT:
|
||
return False
|
||
if int(row.get("contract_status") or 0) != CONTRACT_DRAFT:
|
||
return False
|
||
if not _has_contract_no(row):
|
||
return False
|
||
if int(row.get("process_status") or 0) != PROCESS_CLAIMING:
|
||
return False
|
||
return not _lease_expired(row.get("update_time"), now)
|
||
|
||
|
||
# 判断内存表 4 行是否可领：process_status=0，或 3 且 update_time 已过租约。
|
||
def _is_claimable_draft(row: dict, now: datetime) -> bool:
|
||
if int(row.get("action_code") or 0) != ACTION_CONTRACT:
|
||
return False
|
||
if int(row.get("contract_status") or 0) != CONTRACT_DRAFT:
|
||
return False
|
||
if not _has_contract_no(row):
|
||
return False
|
||
status = int(row.get("process_status") or 0)
|
||
if status == PROCESS_PENDING:
|
||
return True
|
||
if status == PROCESS_CLAIMING:
|
||
return _lease_expired(row.get("update_time"), now)
|
||
return False
|
||
|
||
|
||
# 按合同号成组截到 limit。第一组超限也整组拿走，之后装不下的整组留下。
|
||
def _take_draft_groups(rows: list[dict], limit: int | None) -> list[dict]:
|
||
groups: dict[str, list[dict]] = {}
|
||
order: list[str] = []
|
||
for row in sorted(rows, key=lambda item: int(item["fid"])):
|
||
key = _contract_key(row)
|
||
if key not in groups:
|
||
order.append(key)
|
||
groups[key] = []
|
||
groups[key].append(row)
|
||
chosen: list[dict] = []
|
||
for key in order:
|
||
group = groups[key]
|
||
if limit is not None and chosen and len(chosen) + len(group) > limit:
|
||
break
|
||
chosen.extend(group)
|
||
return chosen
|
||
|
||
|
||
# 判断内存表 3 行是否可领：只录协议且状态 0，或 3 且 process_time 已过租约。
|
||
def _is_claimable_protocol(row: dict, now: datetime) -> bool:
|
||
entry = row.get("contract_entry_status")
|
||
if entry is None or int(entry) != ENTRY_PROTOCOL_ONLY:
|
||
return False
|
||
if not _has_contract_no(row):
|
||
return False
|
||
status = int(row.get("process_status") or 0)
|
||
if status == PROCESS_PENDING:
|
||
return True
|
||
if status == PROCESS_CLAIMING:
|
||
claimed_at = row.get("process_time")
|
||
if claimed_at is None:
|
||
return False
|
||
return now - claimed_at >= timedelta(minutes=LEASE_MINUTES)
|
||
return False
|
||
|
||
|
||
# 回滚当前事务。回滚本身失败则忽略。
|
||
def _rollback(conn) -> None:
|
||
try:
|
||
conn.rollback()
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
# 把当前游标结果收成小写列名的字典列表。
|
||
def _records(cur) -> list[dict]:
|
||
cols = [item[0].lower() for item in cur.description]
|
||
return [dict(zip(cols, rec)) for rec in cur.fetchall()]
|
||
|
||
|
||
# 库故障对外「数据库不可用」，已是该异常则原样抛出。
|
||
def _wrap_unavailable(exc: BaseException) -> RepoUnavailable:
|
||
if isinstance(exc, RepoUnavailable):
|
||
return exc
|
||
return RepoUnavailable("数据库不可用")
|
||
|
||
|
||
# 返利班次的内存仓库，测领取、合组和落表，不测 SQL。
|
||
class MemoryRebateRepo:
|
||
# 用给定的表 4 / 表 3 行初始化。同一 fid 后写覆盖先写。
|
||
def __init__(
|
||
self,
|
||
table4: list[dict] | None = None,
|
||
table3: list[dict] | None = None,
|
||
):
|
||
self.table4: dict[int, dict] = {
|
||
int(row["fid"]): dict(row) for row in (table4 or [])
|
||
}
|
||
self.table3: dict[int, dict] = {
|
||
int(row["fid"]): dict(row) for row in (table3 or [])
|
||
}
|
||
self._next_fid = (max(self.table4) + 1) if self.table4 else 1
|
||
self.shift_errors: list[ShiftError] = []
|
||
|
||
# 领取待查号任务，表 4 process_status 改成 3，表 3 只录协议行改成 3。
|
||
#
|
||
# 先领起草行，同一合同号整组进出，limit 有余再领协议行。
|
||
# 组内还有未过租约的 3 则整组不领。now 只给单测钉住时钟。
|
||
def claim_pending(
|
||
self, limit: int | None = None, *, now: datetime | None = None
|
||
) -> Claimed:
|
||
clock = now or datetime.now()
|
||
held = {
|
||
_contract_key(row)
|
||
for row in self.table4.values()
|
||
if _is_fresh_hold(row, clock)
|
||
}
|
||
claimable = [
|
||
row
|
||
for row in self.table4.values()
|
||
if _is_claimable_draft(row, clock) and _contract_key(row) not in held
|
||
]
|
||
drafts = _take_draft_groups(claimable, limit)
|
||
protocols = sorted(
|
||
(row for row in self.table3.values() if _is_claimable_protocol(row, clock)),
|
||
key=lambda row: int(row["fid"]),
|
||
)
|
||
if limit is not None:
|
||
remaining = limit - len(drafts)
|
||
if remaining <= 0:
|
||
protocols = []
|
||
else:
|
||
protocols = protocols[:remaining]
|
||
for row in drafts:
|
||
row["process_status"] = PROCESS_CLAIMING
|
||
row["update_user_id"] = 0
|
||
row["update_user_name"] = USER_NAME
|
||
row["update_time"] = clock
|
||
for row in protocols:
|
||
row["process_status"] = PROCESS_CLAIMING
|
||
row["process_time"] = clock
|
||
return Claimed(drafts=list(drafts), protocols=list(protocols))
|
||
|
||
# 把一组查号/录协议结果写入内存表 4，并改表 3 的 process_status。
|
||
def save_group(self, payload: GroupSave) -> None:
|
||
for item in payload.table4_updates:
|
||
target = self.table4[int(item["fid"])]
|
||
target["contract_status"] = item["contract_status"]
|
||
target["process_status"] = item["process_status"]
|
||
if "agreement_fail_reason" in item:
|
||
target["agreement_fail_reason"] = item["agreement_fail_reason"]
|
||
target["update_user_id"] = 0
|
||
target["update_user_name"] = USER_NAME
|
||
for fid in payload.table4_release_fids:
|
||
target = self.table4[int(fid)]
|
||
target["process_status"] = PROCESS_PENDING
|
||
target["update_user_id"] = 0
|
||
target["update_user_name"] = USER_NAME
|
||
for row in payload.table4_inserts:
|
||
item = dict(row)
|
||
item["fid"] = self._next_fid
|
||
self._next_fid += 1
|
||
self.table4[item["fid"]] = item
|
||
for item in payload.table3_updates:
|
||
target = self.table3[int(item["fid"])]
|
||
target["process_status"] = item["process_status"]
|
||
target["process_result"] = item["process_result"]
|
||
|
||
# 记下未捕获异常原文。不改表 3 / 表 4。
|
||
def save_shift_errors(self, items: list[ShiftError]) -> None:
|
||
self.shift_errors.extend(items)
|
||
|
||
# 按合同号从表 3 取货品行；表 4 同号的 oms_id 也会对上。
|
||
def load_contract_goods(self, contract_no: str) -> list[dict]:
|
||
number = (contract_no or "").strip()
|
||
oms_ids = {
|
||
int(row["oms_id"])
|
||
for row in self.table4.values()
|
||
if (row.get("contract_no") or "").strip() == number and row.get("oms_id") is not None
|
||
}
|
||
return [
|
||
dict(row)
|
||
for row in self.table3.values()
|
||
if (row.get("contract_no") or "").strip() == number
|
||
or int(row.get("oms_id") or 0) in oms_ids
|
||
]
|
||
|
||
|
||
# 返利班次连 Oracle 的仓库，管表 4 / 表 3 领取和表 4 写入。
|
||
#
|
||
# 连接从 rpa_oracle 连接池取。领取戳是表 4 / 表 3 的 process_status=3。
|
||
# 进程退出后行停在 3，超过租约才能再领。
|
||
class OracleRebateRepo:
|
||
# 挂上连接池。方法里按次 acquire，班次 session 内复用一条。
|
||
def __init__(self, pool):
|
||
self._pool = pool
|
||
self._held = None
|
||
|
||
# 按环境变量建班次池。缺 HOST / 用户 / 密码抛「数据库不可用」。
|
||
@classmethod
|
||
def from_env(cls) -> "OracleRebateRepo":
|
||
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()
|
||
|
||
# 领取待查号任务，表 4 process_status 改成 3，表 3 只录协议行改成 3。
|
||
#
|
||
# 使用 FOR UPDATE SKIP LOCKED，两个班次不会领到同一行。
|
||
# 进程退出后行停在 3，超过租约才能再领。
|
||
# 先领起草行，同一合同号整组进出，limit 有余再领协议行。没有可领的行时返回空 Claimed。
|
||
def claim_pending(self, limit: int | None = None) -> Claimed:
|
||
draft_binds: dict = {"lease_minutes": LEASE_MINUTES}
|
||
protocol_binds: dict = {"lease_minutes": LEASE_MINUTES}
|
||
try:
|
||
with self._acquire() as conn:
|
||
try:
|
||
with conn.cursor() as cur:
|
||
drafts = self._claim_drafts(cur, limit, draft_binds)
|
||
remaining = None if limit is None else max(0, int(limit) - len(drafts))
|
||
protocols: list[dict] = []
|
||
if remaining is None or remaining > 0:
|
||
protocols = self._claim_protocols(cur, remaining, protocol_binds)
|
||
conn.commit()
|
||
return Claimed(drafts=drafts, protocols=protocols)
|
||
except Exception as exc:
|
||
_rollback(conn)
|
||
raise _wrap_unavailable(exc) from exc
|
||
except RepoUnavailable:
|
||
raise
|
||
except Exception as exc:
|
||
raise RepoUnavailable("数据库不可用") from exc
|
||
|
||
# 按合同号从表 3 取货品行，供总单/细单页面填字段。
|
||
#
|
||
# 表 4 同号的 oms_id 也会对上。空合同号返回空列表。库故障对外「数据库不可用」。
|
||
def load_contract_goods(self, contract_no: str) -> list[dict]:
|
||
number = (contract_no or "").strip()
|
||
if not number:
|
||
return []
|
||
try:
|
||
with self._acquire() as conn:
|
||
try:
|
||
with conn.cursor() as cur:
|
||
cur.execute(_SELECT_CONTRACT_GOODS, {"contract_no": number})
|
||
return _records(cur)
|
||
except Exception as exc:
|
||
raise _wrap_unavailable(exc) from exc
|
||
except RepoUnavailable:
|
||
raise
|
||
except Exception as exc:
|
||
raise RepoUnavailable("数据库不可用") from exc
|
||
|
||
# 把一组查号/录协议结果写入表 4，并改表 3 的 process_status。
|
||
#
|
||
# 起草更新、释放、插入、表 3 回写各一次 executemany，同一连接同一事务。
|
||
# 某一侧没有行就跳过那一次调用。库故障对外「数据库不可用」，不含 ORA 原文。
|
||
def save_group(self, payload: GroupSave) -> None:
|
||
draft_binds = [
|
||
{
|
||
"fid": int(item["fid"]),
|
||
"contract_status": item["contract_status"],
|
||
"process_status": item["process_status"],
|
||
"agreement_fail_reason": item.get("agreement_fail_reason"),
|
||
}
|
||
for item in payload.table4_updates
|
||
]
|
||
release_binds = [{"fid": int(fid)} for fid in payload.table4_release_fids]
|
||
insert_binds = [
|
||
{key: row.get(key) for key in _TABLE4_BIND_KEYS}
|
||
for row in payload.table4_inserts
|
||
]
|
||
protocol_binds = [
|
||
{
|
||
"fid": int(item["fid"]),
|
||
"process_status": item["process_status"],
|
||
"process_result": item["process_result"],
|
||
}
|
||
for item in payload.table3_updates
|
||
]
|
||
try:
|
||
with self._acquire() as conn:
|
||
try:
|
||
with conn.cursor() as cur:
|
||
if draft_binds:
|
||
cur.executemany(_UPDATE_DRAFT_SQL, draft_binds)
|
||
if release_binds:
|
||
cur.executemany(_RELEASE_DRAFT_SQL, release_binds)
|
||
if insert_binds:
|
||
cur.executemany(_INSERT_TABLE4_SQL, insert_binds)
|
||
if protocol_binds:
|
||
cur.executemany(_COMPLETE_PROTOCOL_SQL, protocol_binds)
|
||
conn.commit()
|
||
except Exception as exc:
|
||
_rollback(conn)
|
||
raise _wrap_unavailable(exc) from exc
|
||
except RepoUnavailable:
|
||
raise
|
||
except Exception as exc:
|
||
raise RepoUnavailable("数据库不可用") from exc
|
||
|
||
def _claim_drafts(self, cur, limit: int | None, binds: dict) -> list[dict]:
|
||
"""SKIP LOCKED 选出表 4 起草行，按合同号整组改成 process_status=3。
|
||
|
||
第一组超 limit 仍整组领走。更新按每段 500 个 fid 执行，不在这里提交。
|
||
"""
|
||
cur.execute(_SELECT_DRAFTS, binds)
|
||
fetched = _take_draft_groups(_records(cur), limit)
|
||
if not fetched:
|
||
return []
|
||
update_by_fid_chunks(
|
||
cur,
|
||
_CLAIM_DRAFT_SQL,
|
||
[int(row["fid"]) for row in fetched],
|
||
)
|
||
for row in fetched:
|
||
row["process_status"] = PROCESS_CLAIMING
|
||
row["update_user_id"] = 0
|
||
row["update_user_name"] = USER_NAME
|
||
return fetched
|
||
|
||
def _claim_protocols(self, cur, limit: int | None, binds: dict) -> list[dict]:
|
||
"""SKIP LOCKED 选出表 3 只录协议行并改成 process_status=3。
|
||
|
||
更新按每段 500 个 fid 执行，不在这里提交。
|
||
"""
|
||
if limit is None:
|
||
cur.execute(_SELECT_PROTOCOLS, binds)
|
||
else:
|
||
cur.execute(_SELECT_PROTOCOLS_LIMIT, {**binds, "limit": int(limit)})
|
||
fetched = _records(cur)
|
||
if not fetched:
|
||
return []
|
||
update_by_fid_chunks(
|
||
cur,
|
||
_CLAIM_PROTOCOL_SQL,
|
||
[int(row["fid"]) for row in fetched],
|
||
)
|
||
for row in fetched:
|
||
row["process_status"] = PROCESS_CLAIMING
|
||
return fetched
|