# 选品班次的领取和落表。
#
# 从表 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)