# 表 2 加购行：领取、成功盖章、失败改状态。成功不插新行。
#
# 领取时盖「加购中」。进程中途退出时行停在这一戳，超过租约（update_time 30 分钟）才能再领。
# 未过期的「加购中」需改回「系统管理员」再跑。

from __future__ import annotations

from contextlib import contextmanager
from dataclasses import dataclass
from datetime import datetime, timedelta
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

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
LEASE_MINUTES = 30

_CLAIM_SQL = """
UPDATE rpa_cart_settlement_goods_data
SET update_user_name = :stamp,
    update_time = SYSDATE
WHERE fid IN ({placeholders})
"""

_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
"""

_RELEASE_SQL = """
UPDATE rpa_cart_settlement_goods_data
SET update_user_name = :stamp,
    fail_reason = :fail_reason,
    update_time = SYSDATE
WHERE fid = :fid
  AND update_user_name = :claim_stamp
"""


# 表 2 一行待加购或处理中的结算货品。
@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


# 把失败原因截到 FAIL_REASON_MAX 字，超长末尾加省略号。
def clip_reason(text: str) -> str:
    text = (text or "").strip()
    if len(text) <= FAIL_REASON_MAX:
        return text
    return text[: FAIL_REASON_MAX - 1] + "…"


# 判断内存行是否可领：系统管理员，或加购中且 update_time 已过租约。
def _is_pending(row: dict[str, Any], now: datetime) -> bool:
    stamp = row.get("update_user_name") or STAMP_SELECTION
    if (
        int(row.get("operate_type") or 0) != OPERATE_TYPE_ADD_CART
        or int(row.get("status") or 0) != STATUS_OK
        or int(row.get("push_status") or 0) != PUSH_NOT_SENT
    ):
        return False
    if stamp == STAMP_SELECTION:
        return True
    if stamp == STAMP_CLAIMING:
        claimed_at = row.get("update_time")
        if claimed_at is None:
            return False
        return now - claimed_at >= timedelta(minutes=LEASE_MINUTES)
    return False


# 把字典收成 CartRow。缺数量当 0。
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),
    )


# 回滚当前事务。回滚本身失败则忽略。
def _rollback(conn) -> None:
    try:
        conn.rollback()
    except Exception:
        pass


# 拼领取 SELECT：系统管理员或过期加购中，末尾 FOR UPDATE SKIP LOCKED。
def _pending_select_sql(
    *, limit: int | None, platform_types: list[int] | None
) -> tuple[str, dict[str, Any]]:
    params: dict[str, Any] = {
        "selection_stamp": STAMP_SELECTION,
        "claim_stamp": STAMP_CLAIMING,
        "lease_minutes": LEASE_MINUTES,
    }
    filters = [
        "operate_type = 1",
        "status = 1",
        "push_status = 0",
        """(
            NVL(update_user_name, :selection_stamp) = :selection_stamp
            OR (
                update_user_name = :claim_stamp
                AND update_time < SYSDATE - NUMTODSINTERVAL(:lease_minutes, 'MINUTE')
            )
        )""",
    ]
    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 + "\nFOR UPDATE SKIP LOCKED", params


# 加购班次的内存仓库，测规则和班次编排，不测 SQL。
class MemoryCartRepo:
    # 用给定的表 2 行初始化。同一 fid 后写覆盖先写。
    def __init__(self, rows: list[dict[str, Any]]):
        self.rows: dict[int, dict[str, Any]] = {int(row["fid"]): dict(row) for row in rows}
        self.shift_errors: list[ShiftError] = []

    # 领取待加购任务，把戳从系统管理员或过期加购中改成加购中。
    #
    # 未过租约的加购中不领。limit 为空表示不限制条数。now 只给单测钉住时钟。
    def claim_pending(
        self,
        limit: int | None = None,
        platform_types: list[int] | None = None,
        *,
        now: datetime | None = None,
    ) -> list[CartRow]:
        clock = now or datetime.now()
        pending = sorted(
            (row for row in self.rows.values() if _is_pending(row, clock)),
            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
            row["update_time"] = clock
            claimed.append(_row_from_dict(row))
        return claimed

    # 成功盖加购班，清失败原因，不改 status。
    def mark_success(self, fid: int) -> None:
        row = self.rows[fid]
        row["update_user_name"] = STAMP_DONE
        row["fail_reason"] = None

    # 失败改 status=2，写下截断后的原因，戳改成加购班。
    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

    def mark_release(self, fid: int, reason: str) -> None:
        """登录或查询失败放回待领。戳改回系统管理员，status 保持 1。"""
        row = self.rows[fid]
        row["update_user_name"] = STAMP_SELECTION
        row["fail_reason"] = clip_reason(reason)

    # 记下未捕获异常原文。不改表 2。
    def save_shift_errors(self, items: list[ShiftError]) -> None:
        self.shift_errors.extend(items)


# 加购班次连 Oracle 的仓库，管表 2 领取和盖章。
#
# 连接从 rpa_oracle 连接池取。领取戳是加购中。
# 进程退出后行停在加购中，超过租约才能再领。
class OracleCartRepo:
    # 挂上连接池。方法里按次 acquire，班次 session 内复用一条。
    def __init__(self, pool):
        self._pool = pool
        self._held = None

    # 按环境变量建班次池。缺 HOST / 用户 / 密码抛「数据库不可用」。
    @classmethod
    def from_env(cls) -> "OracleCartRepo":
        return cls(pool=job_pool())

    # 未捕获异常原文进共用审计表。独立事务，失败只记日志。
    def save_shift_errors(self, items: list[ShiftError]) -> None:
        write_shift_errors(self._pool, items)

    # 班次内借一条连接，领取和盖章共用。每笔盖章单独提交。
    @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, platform_types: list[int] | None = None
    ) -> list[CartRow]:
        """领取待加购任务，把表 2 的戳从系统管理员改成加购中。

        使用 FOR UPDATE SKIP LOCKED，两个班次不会领到同一行。
        更新按每段 500 个 fid 执行，同一事务，任一段失败整笔回滚。
        进程退出后行停在加购中，超过租约（update_time）才能再领。
        limit 仍只截这一班领多少；为空表示不限制。没有可领的行时返回空列表。
        """
        sql, params = _pending_select_sql(limit=limit, platform_types=platform_types)
        try:
            with self._acquire() as conn:
                try:
                    with conn.cursor() as cur:
                        cur.execute(sql, params)
                        fetched = [_row_from_cursor(cur, rec) for rec in cur.fetchall()]
                        if fetched:
                            update_by_fid_chunks(
                                cur,
                                _CLAIM_SQL,
                                [row.fid for row in fetched],
                                {"stamp": STAMP_CLAIMING},
                            )
                            for row in fetched:
                                row.update_user_name = STAMP_CLAIMING
                    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

    # 成功盖加购班。只改仍停在加购中的那一行。
    def mark_success(self, fid: int) -> None:
        self._execute(
            _SUCCESS_SQL,
            {"fid": int(fid), "stamp": STAMP_DONE, "claim_stamp": STAMP_CLAIMING},
        )

    # 失败改 status=2 并写原因。只改仍停在加购中的那一行。
    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 mark_release(self, fid: int, reason: str) -> None:
        """登录或查询失败放回待领。只改仍停在加购中的那一行，status 不动。"""
        self._execute(
            _RELEASE_SQL,
            {
                "fid": int(fid),
                "fail_reason": clip_reason(reason),
                "stamp": STAMP_SELECTION,
                "claim_stamp": STAMP_CLAIMING,
            },
        )

    # 在当前连接上执行一条 UPDATE 并提交。库故障对外「数据库不可用」。
    def _execute(self, sql: str, params: dict[str, Any]) -> None:
        try:
            with self._acquire() as conn:
                try:
                    with conn.cursor() as cur:
                        cur.execute(sql, params)
                    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


# 把查询结果行收成 CartRow。
def _row_from_cursor(cur, rec) -> CartRow:
    cols = [item[0].lower() for item in cur.description]
    return _row_from_dict(dict(zip(cols, rec)))