542 lines
18 KiB
Python
542 lines
18 KiB
Python
# 选品班次的领取和落表。
|
||
#
|
||
# 从表 1 领取 deal_flag=0 的行改成 3，选品成功后写表 2、表 1 改成 1。
|
||
# 进程中途退出时行停在 3，超过租约（deal_time）才能再领。
|
||
# 表 2 的更新人由落库行带入。凑齐后的加车在班次里做，不在这个仓库里。
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
from dataclasses import dataclass
|
||
from datetime import datetime, timedelta
|
||
from typing import Any
|
||
|
||
from rpa_oracle import RepoUnavailable, job_pool, update_by_fid_chunks
|
||
|
||
DEAL_USER_NAME = "rpa-select"
|
||
|
||
logger = logging.getLogger("rpa_selection")
|
||
LEASE_MINUTES = 30
|
||
_QUERY_ERROR_MAX = 4000
|
||
|
||
_SELECT_BODY = """
|
||
SELECT fid, oms_id, bar_code, sale_price, plan_purchase_qty,
|
||
NVL(approval_no, '') AS approval_no,
|
||
NVL(goods_name, '') AS goods_name,
|
||
NVL(goods_spec, '') AS goods_spec,
|
||
NVL(manufacturer, '') AS manufacturer,
|
||
NVL(org_code, '') AS org_code,
|
||
deal_flag
|
||
FROM rpa_sales_goods_mapping_data
|
||
WHERE deal_flag = 0
|
||
OR (deal_flag = 3 AND deal_time < SYSDATE - NUMTODSINTERVAL(:lease_minutes, 'MINUTE'))
|
||
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_sales_goods_mapping_data
|
||
SET deal_flag = 3,
|
||
deal_time = SYSDATE,
|
||
update_time = SYSDATE,
|
||
deal_user_name = :deal_user_name
|
||
WHERE fid IN ({placeholders})
|
||
"""
|
||
|
||
# 选品无登录用户：创建人固定 0 / 系统管理员。更新人由调用方传入。
|
||
# 凑齐后马上加车的行传入「加购中」，避免补偿班在加车完成前领走。fid 由触发器生成，RETURNING 取回。
|
||
_INSERT_CART_SQL = """
|
||
INSERT INTO rpa_cart_settlement_goods_data (
|
||
operate_type, oms_id, platform_type, supplier_name, bar_code,
|
||
purchase_price, purchase_qty, purchase_total, status, fail_reason,
|
||
push_status, org_code, multi_platform_flag,
|
||
create_time, create_user_id, create_user_name,
|
||
update_time, update_user_id, update_user_name
|
||
) VALUES (
|
||
:operate_type, :oms_id, :platform_type, :supplier_name, :bar_code,
|
||
:purchase_price, :purchase_qty, :purchase_total, :status, :fail_reason,
|
||
:push_status, :org_code, :multi_platform_flag,
|
||
SYSDATE, 0, '系统管理员',
|
||
SYSDATE, 0, :update_user_name
|
||
)
|
||
RETURNING fid INTO :out_fid
|
||
"""
|
||
|
||
STAMP_SELECTION = "系统管理员"
|
||
|
||
_COMPLETE_SQL = """
|
||
UPDATE rpa_sales_goods_mapping_data
|
||
SET deal_flag = :deal_flag,
|
||
deal_result = :deal_result,
|
||
deal_time = SYSDATE,
|
||
update_time = SYSDATE,
|
||
deal_user_name = :deal_user_name
|
||
WHERE fid = :fid
|
||
AND deal_flag = 3
|
||
"""
|
||
|
||
_OWN_SQL = """
|
||
SELECT deal_flag
|
||
FROM rpa_sales_goods_mapping_data
|
||
WHERE fid = :fid
|
||
"""
|
||
|
||
_CART_EXISTS_SQL = """
|
||
SELECT COUNT(*)
|
||
FROM rpa_cart_settlement_goods_data
|
||
WHERE oms_id = :oms_id
|
||
"""
|
||
|
||
_TOUCH_SQL = """
|
||
UPDATE rpa_sales_goods_mapping_data
|
||
SET deal_time = SYSDATE,
|
||
update_time = SYSDATE
|
||
WHERE fid = :fid
|
||
AND deal_flag = 3
|
||
"""
|
||
|
||
_INSERT_QUERY_AUDIT_SQL = """
|
||
INSERT INTO rpa_selection_query_audit (
|
||
oms_id, mapping_fid, mall_name, bar_code, error_text
|
||
) VALUES (
|
||
:oms_id, :mapping_fid, :mall_name, :bar_code, :error_text
|
||
)
|
||
"""
|
||
|
||
_CART_BIND_KEYS = (
|
||
"operate_type",
|
||
"oms_id",
|
||
"platform_type",
|
||
"supplier_name",
|
||
"bar_code",
|
||
"purchase_price",
|
||
"purchase_qty",
|
||
"purchase_total",
|
||
"status",
|
||
"fail_reason",
|
||
"push_status",
|
||
"org_code",
|
||
"multi_platform_flag",
|
||
)
|
||
|
||
|
||
# 一次平台查询失败要写入审计表的原文。
|
||
@dataclass
|
||
class QueryAudit:
|
||
oms_id: int
|
||
mapping_fid: int
|
||
mall_name: str
|
||
bar_code: str
|
||
error_text: str
|
||
|
||
|
||
# 一单要落的表 2 行，以及表 1 按 fid 的完成更新。
|
||
@dataclass
|
||
class FinishWrite:
|
||
oms_id: int
|
||
fid: int
|
||
deal_flag: int
|
||
deal_result: str
|
||
cart_rows: list[dict[str, Any]]
|
||
|
||
|
||
# 表 1 一行待选品或处理中的映射。
|
||
@dataclass
|
||
class MappingRow:
|
||
fid: int
|
||
oms_id: int
|
||
bar_code: str
|
||
sale_price: float | None
|
||
plan_purchase_qty: float | None
|
||
approval_no: str
|
||
goods_name: str
|
||
goods_spec: str
|
||
manufacturer: str
|
||
org_code: str
|
||
deal_flag: int
|
||
deal_result: str = ""
|
||
deal_time: datetime | None = None
|
||
|
||
|
||
# 判断内存行是否可领：0，或 3 且 deal_time 已过租约。
|
||
def _is_claimable(row: MappingRow, now: datetime) -> bool:
|
||
if row.deal_flag == 0:
|
||
return True
|
||
if row.deal_flag == 3 and row.deal_time is not None:
|
||
return now - row.deal_time >= timedelta(minutes=LEASE_MINUTES)
|
||
return False
|
||
|
||
|
||
# 异常原文截到审计列 VARCHAR2(4000)。
|
||
def _clip_error_text(text: str) -> str:
|
||
if len(text) <= _QUERY_ERROR_MAX:
|
||
return text
|
||
return text[:_QUERY_ERROR_MAX]
|
||
|
||
|
||
# 回滚当前事务。回滚本身失败则忽略。
|
||
def _rollback(conn) -> None:
|
||
try:
|
||
conn.rollback()
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
# 选品班次的内存仓库，测规则和班次编排，不测 SQL。
|
||
class MemorySelectRepo:
|
||
# 用给定的表 1 行初始化。同一 oms_id 后写覆盖先写。
|
||
def __init__(self, rows: list[MappingRow]):
|
||
self.mapping: dict[int, MappingRow] = {row.oms_id: row for row in rows}
|
||
self.cart: list[dict[str, Any]] = []
|
||
self.query_audits: list[QueryAudit] = []
|
||
self._cart_fid = 0
|
||
|
||
# 领取待选品任务，把 deal_flag 从 0 或过期 3 改成 3。
|
||
#
|
||
# 未过租约的 3 不领。limit 为空表示不限制条数。now 只给单测钉住时钟。
|
||
def claim_pending(
|
||
self, limit: int | None = None, *, now: datetime | None = None
|
||
) -> list[MappingRow]:
|
||
clock = now or datetime.now()
|
||
pending = sorted(
|
||
(row for row in self.mapping.values() if _is_claimable(row, clock)),
|
||
key=lambda row: row.fid,
|
||
)
|
||
if limit is not None:
|
||
pending = pending[:limit]
|
||
for row in pending:
|
||
row.deal_flag = 3
|
||
row.deal_time = clock
|
||
return pending
|
||
|
||
# 把一单选品结果写入内存表。班次请用 save_results。
|
||
def save_result(
|
||
self,
|
||
oms_id: int,
|
||
deal_flag: int,
|
||
deal_result: str,
|
||
cart_rows: list[dict[str, Any]],
|
||
*,
|
||
fid: int,
|
||
) -> None:
|
||
self.save_results(
|
||
[
|
||
FinishWrite(
|
||
oms_id=oms_id,
|
||
fid=fid,
|
||
deal_flag=deal_flag,
|
||
deal_result=deal_result,
|
||
cart_rows=cart_rows,
|
||
)
|
||
]
|
||
)
|
||
|
||
def touch_claim(self, fid: int, *, now: datetime | None = None) -> None:
|
||
"""搜索过程中刷新这一行的 deal_time。只动仍是 deal_flag=3 的行。"""
|
||
row = self._row_by_fid(fid)
|
||
if row.deal_flag != 3:
|
||
return
|
||
row.deal_time = now or datetime.now()
|
||
|
||
def save_results(self, items: list[FinishWrite]) -> list[list[int]]:
|
||
"""把选品结果写入内存表 2，并按 fid 改表 1。
|
||
|
||
落库前再读：仍是 deal_flag=3，并且表 2 还没有该 oms_id，才插入。
|
||
已经不是本班领的，整单跳过。已有加购行时只改表 1，不插第二套。
|
||
返回值与 items 对齐：每一单新插入的表 2 fid，顺序与 cart_rows 一致。未插入则为空。
|
||
"""
|
||
inserted: list[list[int]] = []
|
||
for item in items:
|
||
row = self._row_by_fid(item.fid)
|
||
if row.oms_id != item.oms_id:
|
||
raise KeyError(item.oms_id)
|
||
if row.deal_flag != 3:
|
||
inserted.append([])
|
||
continue
|
||
exists = any(int(cart.get("oms_id") or 0) == int(item.oms_id) for cart in self.cart)
|
||
fids: list[int] = []
|
||
if item.cart_rows and not exists:
|
||
for cart_row in item.cart_rows:
|
||
stored = dict(cart_row)
|
||
stored["fid"] = self._alloc_cart_fid()
|
||
stored.setdefault("update_user_name", STAMP_SELECTION)
|
||
self.cart.append(stored)
|
||
fids.append(int(stored["fid"]))
|
||
row.deal_flag = item.deal_flag
|
||
row.deal_result = item.deal_result
|
||
inserted.append(fids)
|
||
return inserted
|
||
|
||
# 给新插入的内存表 2 行配一个 fid。已有行的 fid 不重用。
|
||
def _alloc_cart_fid(self) -> int:
|
||
used = [int(row["fid"]) for row in self.cart if row.get("fid") is not None]
|
||
self._cart_fid = max([self._cart_fid, *used, 0]) + 1
|
||
return self._cart_fid
|
||
|
||
# 记下查询失败原文。不改表 1 / 表 2。
|
||
def save_query_audits(self, items: list[QueryAudit]) -> None:
|
||
self.query_audits.extend(_clipped_audit(item) for item in items)
|
||
|
||
# 按主键找表 1 行。找不到则 KeyError。
|
||
def _row_by_fid(self, fid: int) -> MappingRow:
|
||
for row in self.mapping.values():
|
||
if row.fid == fid:
|
||
return row
|
||
raise KeyError(fid)
|
||
|
||
|
||
# 选品班次连 Oracle 的仓库，管表 1 领取和表 2 写入。
|
||
#
|
||
# 连接从 rpa_oracle 连接池取。领取戳是 rpa-select。
|
||
# 进程退出后行停在 3，超过租约才能再领。
|
||
class OracleSelectRepo:
|
||
# 挂上连接池。方法里按次 acquire，不要在仓库外再 connect。
|
||
def __init__(self, pool):
|
||
self._pool = pool
|
||
|
||
# 按环境变量建班次池。缺 HOST / 用户 / 密码抛「数据库不可用」。
|
||
@classmethod
|
||
def from_env(cls) -> "OracleSelectRepo":
|
||
return cls(pool=job_pool())
|
||
|
||
def claim_pending(self, limit: int | None = None) -> list[MappingRow]:
|
||
"""领取待选品任务，把表 1 的 deal_flag 从 0 改成 3。
|
||
|
||
使用 FOR UPDATE SKIP LOCKED，两个班次不会领到同一行。
|
||
更新按每段 500 个 fid 执行，同一事务，任一段失败整笔回滚。
|
||
进程退出后行停在 3，超过租约（deal_time）才能再领。
|
||
limit 仍只截这一班领多少；为空表示不限制。没有可领的行时返回空列表。
|
||
"""
|
||
binds: dict = {"lease_minutes": LEASE_MINUTES}
|
||
try:
|
||
with self._pool.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)},
|
||
)
|
||
fetched = [
|
||
_mapping_from_cursor(cur, rec) for rec in cur.fetchall()
|
||
]
|
||
if fetched:
|
||
update_by_fid_chunks(
|
||
cur,
|
||
_CLAIM_SQL,
|
||
[row.fid for row in fetched],
|
||
{"deal_user_name": DEAL_USER_NAME},
|
||
)
|
||
for row in fetched:
|
||
row.deal_flag = 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
|
||
|
||
# 把一单选品结果落库。班次请用 save_results。
|
||
def save_result(
|
||
self,
|
||
oms_id: int,
|
||
deal_flag: int,
|
||
deal_result: str,
|
||
cart_rows: list[dict[str, Any]],
|
||
*,
|
||
fid: int,
|
||
) -> None:
|
||
self.save_results(
|
||
[
|
||
FinishWrite(
|
||
oms_id=oms_id,
|
||
fid=fid,
|
||
deal_flag=deal_flag,
|
||
deal_result=deal_result,
|
||
cart_rows=cart_rows,
|
||
)
|
||
]
|
||
)
|
||
|
||
def touch_claim(self, fid: int) -> None:
|
||
"""搜索过程中刷新 deal_time。只改仍是 deal_flag=3 的行，并马上提交。"""
|
||
try:
|
||
with self._pool.acquire() as conn:
|
||
try:
|
||
with conn.cursor() as cur:
|
||
cur.execute(_TOUCH_SQL, {"fid": int(fid)})
|
||
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
|
||
|
||
def save_results(self, items: list[FinishWrite]) -> list[list[int]]:
|
||
"""把这一调用里的表 2 行和表 1 完成更新落在同一个事务。
|
||
|
||
每一单先再读：仍是 deal_flag=3，并且表 2 还没有该 oms_id，才插入。
|
||
已有加购行时只改表 1。空列表不借连接。任一行失败整笔回滚。
|
||
库故障对外「数据库不可用」，不含 ORA 原文。
|
||
返回值与 items 对齐：每一单新插入的表 2 fid，顺序与 cart_rows 一致。未插入则为空。
|
||
"""
|
||
if not items:
|
||
return []
|
||
inserted: list[list[int]] = [[] for _ in items]
|
||
try:
|
||
with self._pool.acquire() as conn:
|
||
try:
|
||
with conn.cursor() as cur:
|
||
complete_binds: list[dict] = []
|
||
for index, item in enumerate(items):
|
||
if not _claim_still_open(cur, item.fid):
|
||
continue
|
||
if item.cart_rows and not _cart_already_written(cur, item.oms_id):
|
||
inserted[index] = [
|
||
_insert_cart_row(cur, cart_row) for cart_row in item.cart_rows
|
||
]
|
||
complete_binds.append(
|
||
{
|
||
"deal_flag": item.deal_flag,
|
||
"deal_result": item.deal_result,
|
||
"deal_user_name": DEAL_USER_NAME,
|
||
"fid": item.fid,
|
||
}
|
||
)
|
||
if complete_binds:
|
||
cur.executemany(_COMPLETE_SQL, complete_binds)
|
||
conn.commit()
|
||
return inserted
|
||
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
|
||
|
||
# 把查询失败原文写入审计表。独立事务，失败只记日志。
|
||
def save_query_audits(self, items: list[QueryAudit]) -> None:
|
||
if not items:
|
||
return
|
||
binds = [_audit_bind(item) for item in items]
|
||
try:
|
||
with self._pool.acquire() as conn:
|
||
try:
|
||
with conn.cursor() as cur:
|
||
cur.executemany(_INSERT_QUERY_AUDIT_SQL, binds)
|
||
conn.commit()
|
||
except Exception:
|
||
_rollback(conn)
|
||
logger.exception("选品查询审计写入失败")
|
||
except Exception:
|
||
logger.exception("选品查询审计写入失败")
|
||
|
||
|
||
def _claim_still_open(cur, fid: int) -> bool:
|
||
"""再读表 1。deal_flag 仍是 3 才算还是本班领的。"""
|
||
cur.execute(_OWN_SQL, {"fid": int(fid)})
|
||
rows = cur.fetchall()
|
||
if not rows:
|
||
return False
|
||
return int(rows[0][0]) == 3
|
||
|
||
|
||
def _cart_already_written(cur, oms_id: int) -> bool:
|
||
"""表 2 已有该 oms_id 时，重跑不能再插一套加购行。"""
|
||
cur.execute(_CART_EXISTS_SQL, {"oms_id": int(oms_id)})
|
||
rows = cur.fetchall()
|
||
if not rows:
|
||
return False
|
||
return int(rows[0][0]) > 0
|
||
|
||
|
||
# 插入一行表 2 并取回触发器生成的 fid。缺更新人时写成系统管理员。
|
||
def _insert_cart_row(cur, cart_row: dict[str, Any]) -> int:
|
||
params = {key: cart_row.get(key) for key in _CART_BIND_KEYS}
|
||
params["update_user_name"] = cart_row.get("update_user_name") or STAMP_SELECTION
|
||
out = cur.var(int)
|
||
params["out_fid"] = out
|
||
cur.execute(_INSERT_CART_SQL, params)
|
||
return _fid_from_var(out)
|
||
|
||
|
||
# RETURNING 的值可能是单元素列表，也可能是标量。空值当落库失败。
|
||
def _fid_from_var(var) -> int:
|
||
value = var.getvalue()
|
||
if isinstance(value, list):
|
||
value = value[0] if value else None
|
||
if value is None:
|
||
raise RepoUnavailable("数据库不可用")
|
||
return int(value)
|
||
|
||
|
||
# 返回截断原文后的审计行。
|
||
def _clipped_audit(item: QueryAudit) -> QueryAudit:
|
||
return QueryAudit(
|
||
oms_id=item.oms_id,
|
||
mapping_fid=item.mapping_fid,
|
||
mall_name=item.mall_name,
|
||
bar_code=item.bar_code,
|
||
error_text=_clip_error_text(item.error_text),
|
||
)
|
||
|
||
|
||
# 收成审计插入 bind。不绑 fid，留给触发器。
|
||
def _audit_bind(item: QueryAudit) -> dict[str, Any]:
|
||
clipped = _clipped_audit(item)
|
||
return {
|
||
"oms_id": clipped.oms_id,
|
||
"mapping_fid": clipped.mapping_fid,
|
||
"mall_name": clipped.mall_name,
|
||
"bar_code": clipped.bar_code,
|
||
"error_text": clipped.error_text,
|
||
}
|
||
|
||
|
||
# 把查询结果行收成 MappingRow。缺省字符串当空，数字缺省当 None。
|
||
def _mapping_from_cursor(cur, rec) -> MappingRow:
|
||
cols = [item[0].lower() for item in cur.description]
|
||
data = dict(zip(cols, rec))
|
||
return MappingRow(
|
||
fid=int(data["fid"]),
|
||
oms_id=int(data["oms_id"]),
|
||
bar_code=str(data.get("bar_code") or ""),
|
||
sale_price=_float_or_none(data.get("sale_price")),
|
||
plan_purchase_qty=_float_or_none(data.get("plan_purchase_qty")),
|
||
approval_no=str(data.get("approval_no") or ""),
|
||
goods_name=str(data.get("goods_name") or ""),
|
||
goods_spec=str(data.get("goods_spec") or ""),
|
||
manufacturer=str(data.get("manufacturer") or ""),
|
||
org_code=str(data.get("org_code") or ""),
|
||
deal_flag=int(data.get("deal_flag") or 0),
|
||
)
|
||
|
||
|
||
# 把库里的数字收成 float；空值保持 None。
|
||
def _float_or_none(value: object) -> float | None:
|
||
if value is None or value == "":
|
||
return None
|
||
return float(value)
|