shanxia2.0/backend/app/services/bx/reimburse_import.py

from __future__ import annotations

import base64
from datetime import date, datetime, timedelta
from decimal import Decimal, InvalidOperation
from io import BytesIO
from typing import Any, Dict, List, Optional, Tuple

from openpyxl import Workbook, load_workbook
from openpyxl.styles import PatternFill
from openpyxl.utils import get_column_letter
from openpyxl.worksheet.worksheet import Worksheet
from sqlalchemy.orm import Session

from app.core.bx_auth import BxUserContext
from app.models.bx_models import BxExpenseItem, BxWorkflowDef
from app.models.models import SysUser
from app.services.bx.expense_item import ExpenseItemService
from app.services.bx.reimburse import ReimburseService
from app.utils.bx_codegen import generate_apply_code, parse_optional_date, parse_reimburse_date
from app.utils.excel_list_validation import add_list_validation, write_option_sheet

EXAMPLE_REMARK_PREFIX = "【示例】"
PATH_SEP = " / "
MAX_IMPORT_ROWS = 500

IMPORT_SHEET = "报销导入"
GUIDE_SHEET = "填写说明"
EXPENSE_OPT_SHEET = "_选项_费用项目"
APPLICANT_OPT_SHEET = "_选项_申请人"

HEADERS = [
    "分组号",
    "申请人",
    "报销日期",
    "报销标题",
    "费用项目",
    "金额（元）",
    "报销事由",
    "发生时间",
]

_GUIDE_LINES = [
    "1. 请使用本系统「下载模板」获取最新文件；费用项/人员变更后请重新下载。",
    "2. 推荐 Excel 或 WPS 桌面版；在线表格可能丢失下拉。",
    "3. 一行一条费用明细；相同「分组号」合成一张报销单。",
    "4. 分组号可空：空则该行单独成单。",
    "5. 同一分组内：申请人、报销日期、报销标题必须每行都填且完全一致（标题可都留空以自动生成）。",
    "6. 费用项目、申请人请用下拉选择，勿手改名称。费用项目下拉与系统费用树一致：仅启用项，不含已删除/已停用，也不含父级已删的孤儿节点。",
    "7. 灰色示例行仅供参考，导入时自动跳过（报销事由以【示例】开头），请从示例下方空行开始填写。",
    "8. 导入结果为「待提交」草稿；发票请导入后在系统内关联。",
    "9. 请勿修改或删除隐藏的 _选项_* 工作表。",
    "10. Excel 行号从 1 起计（第 1 行为表头）；错误清单中的行号与打开文件所见一致。",
]

_EXAMPLE_FILL = PatternFill(start_color="F1F5F9", end_color="F1F5F9", fill_type="solid")


def build_applicant_label(nick_name: str, user_no: str, user_id: str) -> str:
    name = (nick_name or user_no or user_id or "").strip() or user_id
    suffix = (user_no or "").strip() or str(user_id)[-6:]
    return f"{name}（{suffix}）"


def _ids_visible_in_web_tree(nodes: List[Dict[str, Any]]) -> set:
    """与 ExpenseItemService._build_tree 一致：父级不存在的孤儿节点不会出现在 Web 树中。"""
    by_id = {str(n["id"]): n for n in nodes}
    visible: set = set()

    def walk(nid: str) -> None:
        if nid in visible:
            return
        if nid not in by_id:
            return
        visible.add(nid)
        for cid, c in by_id.items():
            if str(c.get("parent_id") or "0") == nid:
                walk(cid)

    for n in nodes:
        pid = str(n.get("parent_id") or "0")
        if not pid or pid == "0":
            walk(str(n["id"]))
    return visible


def build_expense_path_labels(nodes: List[Dict[str, Any]]) -> Dict[str, str]:
    """返回 {id: 展示文案}。

    与 Web 费用树对齐：仅启用项，且排除父级已删导致的孤儿节点。
    冲突时追加 （编码:xxx）。
    """
    by_id = {str(n["id"]): n for n in nodes}
    visible = _ids_visible_in_web_tree(nodes)
    enabled = [
        n
        for n in nodes
        if int(n.get("status") or 0) == 1 and str(n["id"]) in visible
    ]

    def path_of(nid: str) -> str:
        parts: List[str] = []
        cur = by_id.get(nid)
        guard = 0
        while cur and guard < 20:
            parts.append(str(cur.get("item_name") or "").strip())
            pid = str(cur.get("parent_id") or "0")
            if not pid or pid == "0":
                break
            cur = by_id.get(pid)
            guard += 1
        return PATH_SEP.join(reversed([p for p in parts if p]))

    raw: Dict[str, str] = {str(n["id"]): path_of(str(n["id"])) for n in enabled}
    path_to_ids: Dict[str, List[str]] = {}
    for iid, p in raw.items():
        path_to_ids.setdefault(p, []).append(iid)
    out: Dict[str, str] = {}
    for p, ids in path_to_ids.items():
        if len(ids) == 1:
            out[ids[0]] = p
            continue
        for iid in ids:
            code = (by_id[iid].get("item_code") or "").strip() or iid[-6:]
            out[iid] = f"{p}（编码:{code}）"
    return out


def is_example_row(row: Dict[str, Any]) -> bool:
    remark = str(row.get("报销事由") or "").strip()
    return remark.startswith(EXAMPLE_REMARK_PREFIX)


def partition_import_groups(rows: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
    ordered: List[Dict[str, Any]] = []
    index: Dict[str, int] = {}
    for row in rows:
        g = str(row.get("分组号") or "").strip()
        if not g:
            ordered.append({"group_no": "", "rows": [row]})
            continue
        if g not in index:
            index[g] = len(ordered)
            ordered.append({"group_no": g, "rows": [row]})
        else:
            ordered[index[g]]["rows"].append(row)
    return ordered


def validate_group_headers(rows: List[Dict[str, Any]]) -> Optional[str]:
    if not rows:
        return "分组无有效行"
    apps = {str(r.get("申请人") or "").strip() for r in rows}
    dates = {str(r.get("报销日期") or "").strip() for r in rows}
    titles = [str(r.get("报销标题") or "").strip() for r in rows]
    if len(apps) != 1 or not next(iter(apps)):
        return "同组申请人不一致或为空"
    if len(dates) != 1 or not next(iter(dates)):
        return "同组报销日期不一致或为空"
    empty_n = sum(1 for t in titles if not t)
    if 0 < empty_n < len(titles):
        return "同组报销标题不能部分为空"
    if empty_n == 0 and len(set(titles)) != 1:
        return "同组报销标题不一致"
    return None


def cell_str(v: Any) -> str:
    if v is None:
        return ""
    if isinstance(v, datetime):
        return v.date().isoformat()
    if isinstance(v, date):
        return v.isoformat()
    # Excel 常把纯数字单元格读成 float（如 29.0），需归一成 "29"
    if isinstance(v, float) and v.is_integer():
        return str(int(v))
    if isinstance(v, int) and not isinstance(v, bool):
        return str(v)
    return str(v).strip()


def parse_amount(v: Any) -> Tuple[Optional[Decimal], Optional[str]]:
    if v is None or cell_str(v) == "":
        return None, "金额不能为空"
    try:
        amt = Decimal(str(v).strip())
    except (InvalidOperation, ValueError):
        return None, "金额格式无效"
    if amt <= 0:
        return None, "金额必须大于 0"
    return amt, None


def load_expense_nodes_flat(db: Session, tenant_no: str) -> List[Dict[str, Any]]:
    rows = (
        db.query(BxExpenseItem)
        .filter(BxExpenseItem.del_flag == 0, BxExpenseItem.tenant_no == tenant_no)
        .order_by(BxExpenseItem.sort_order.asc(), BxExpenseItem.create_time.desc())
        .all()
    )
    return [
        {
            "id": r.id,
            "parent_id": r.parent_id or "0",
            "item_name": r.item_name,
            "status": r.status,
            "item_code": r.item_code,
        }
        for r in rows
    ]


def load_applicant_options(
    db: Session, *, user: BxUserContext, tenant_no: str
) -> List[Dict[str, Any]]:
    q = db.query(SysUser).filter(
        SysUser.del_flag == 0,
        SysUser.status == "1",
        SysUser.tenant_no == tenant_no,
    )
    if not ReimburseService.can_change_applicant(db, user):
        q = q.filter(SysUser.id == user.id)
    users = q.order_by(SysUser.create_time.desc()).all()
    return [
        {"id": u.id, "nick_name": u.nick_name or "", "user_no": u.user_no or ""}
        for u in users
    ]


def build_import_template_bytes(
    *,
    expense_nodes: List[Dict[str, Any]],
    applicants: List[Dict[str, Any]],
    current_applicant_label: str,
    today: date,
) -> bytes:
    wb = Workbook()

    guide = wb.active
    guide.title = GUIDE_SHEET
    for line in _GUIDE_LINES:
        guide.append([line])
    if not expense_nodes or not build_expense_path_labels(expense_nodes):
        guide.append(["注意：当前租户暂无启用费用项目，请先在系统中配置后再填写费用项目列。"])

    ws = wb.create_sheet(IMPORT_SHEET, 0)
    ws.append(list(HEADERS))

    expense_labels = build_expense_path_labels(expense_nodes)
    expense_items = sorted(expense_labels.items(), key=lambda x: x[1])
    exp_ids = [iid for iid, _ in expense_items]
    exp_labs = [lab for _, lab in expense_items]
    write_option_sheet(wb, EXPENSE_OPT_SHEET, exp_labs, exp_ids, hidden=True)

    app_labs: List[str] = []
    app_ids: List[str] = []
    for a in applicants:
        lab = build_applicant_label(
            str(a.get("nick_name") or ""),
            str(a.get("user_no") or ""),
            str(a.get("id") or ""),
        )
        app_labs.append(lab)
        app_ids.append(str(a.get("id") or ""))
    write_option_sheet(wb, APPLICANT_OPT_SHEET, app_labs, app_ids, hidden=True)

    fee1 = exp_labs[0] if exp_labs else ""
    fee2 = exp_labs[1] if len(exp_labs) > 1 else fee1
    yesterday = (today - timedelta(days=1)).isoformat()
    today_s = today.isoformat()

    examples = [
        ["G1", current_applicant_label, today_s, "", fee1, 50.00, f"{EXAMPLE_REMARK_PREFIX}地铁通勤，请删除本行", ""],
        ["G1", current_applicant_label, today_s, "", fee2, 300.00, f"{EXAMPLE_REMARK_PREFIX}住宿，请删除本行", yesterday],
        ["", current_applicant_label, today_s, "", fee1, 20.00, f"{EXAMPLE_REMARK_PREFIX}空分组号单独成单", ""],
    ]
    for row in examples:
        ws.append(row)
        for col in range(1, len(HEADERS) + 1):
            ws.cell(row=ws.max_row, column=col).fill = _EXAMPLE_FILL

    data_end = 1 + MAX_IMPORT_ROWS
    if app_labs:
        add_list_validation(
            ws,
            option_sheet=APPLICANT_OPT_SHEET,
            option_rows=len(app_labs),
            target_col=2,
            start_row=2,
            end_row=data_end,
        )
    if exp_labs:
        add_list_validation(
            ws,
            option_sheet=EXPENSE_OPT_SHEET,
            option_rows=len(exp_labs),
            target_col=5,
            start_row=2,
            end_row=data_end,
        )

    buf = BytesIO()
    wb.save(buf)
    return buf.getvalue()


RESULT_STATUS_COL = "导入结果"
RESULT_MSG_COL = "失败原因"


def _read_sheet_header(ws: Worksheet) -> Tuple[List[str], Dict[str, int]]:
    header_cells = [cell_str(c.value) for c in next(ws.iter_rows(min_row=1, max_row=1))]
    missing = [h for h in HEADERS if h not in header_cells]
    if missing:
        raise ValueError(f"表头缺少列：{', '.join(missing)}")
    col_index = {name: header_cells.index(name) for name in HEADERS}
    return header_cells, col_index


def scan_sheet_rows(ws: Worksheet) -> List[Dict[str, Any]]:
    """扫描非空行。含示例行（_kind=example）与可导入行（_kind=data）。"""
    _header_cells, col_index = _read_sheet_header(ws)
    rows: List[Dict[str, Any]] = []
    for excel_row, cells in enumerate(ws.iter_rows(min_row=2, values_only=True), start=2):
        values = list(cells) if cells else []
        row: Dict[str, Any] = {"_row": excel_row}
        empty = True
        for name, idx in col_index.items():
            raw = values[idx] if idx < len(values) else None
            if name in ("报销日期", "发生时间"):
                row[name] = cell_str(raw)
            elif name == "金额（元）":
                row[name] = raw
            else:
                row[name] = cell_str(raw)
            if cell_str(raw) != "" and raw is not None:
                empty = False
        if empty:
            continue
        row["_kind"] = "example" if is_example_row(row) else "data"
        rows.append(row)
    return rows


def iter_data_rows(ws: Worksheet) -> List[Dict[str, Any]]:
    """仅返回可导入数据行（跳过示例），兼容旧测试。"""
    return [r for r in scan_sheet_rows(ws) if r.get("_kind") == "data"]


def build_annotated_result_bytes(
    file_bytes: bytes,
    outcomes: Dict[int, Dict[str, str]],
) -> bytes:
    """在原表「报销导入」末尾追加「导入结果」「失败原因」列。"""
    wb = load_workbook(BytesIO(file_bytes))
    if IMPORT_SHEET not in wb.sheetnames:
        raise ValueError(f"缺少工作表「{IMPORT_SHEET}」")
    ws = wb[IMPORT_SHEET]
    header_cells = [cell_str(c.value) for c in next(ws.iter_rows(min_row=1, max_row=1))]
    # 若已存在结果列则复用，否则追加
    if RESULT_STATUS_COL in header_cells:
        status_col = header_cells.index(RESULT_STATUS_COL) + 1
    else:
        status_col = len(header_cells) + 1
        ws.cell(row=1, column=status_col, value=RESULT_STATUS_COL)
        header_cells.append(RESULT_STATUS_COL)
    if RESULT_MSG_COL in header_cells:
        msg_col = header_cells.index(RESULT_MSG_COL) + 1
    else:
        msg_col = len(header_cells) + 1
        ws.cell(row=1, column=msg_col, value=RESULT_MSG_COL)

    max_row = max(ws.max_row or 1, max(outcomes.keys(), default=1))
    for r in range(2, max_row + 1):
        oc = outcomes.get(r)
        if not oc:
            continue
        ws.cell(row=r, column=status_col, value=oc.get("status") or "")
        ws.cell(row=r, column=msg_col, value=oc.get("message") or "")

    # 略加宽结果列
    ws.column_dimensions[get_column_letter(status_col)].width = 12
    ws.column_dimensions[get_column_letter(msg_col)].width = 36

    buf = BytesIO()
    wb.save(buf)
    return buf.getvalue()


def _mark_rows(
    outcomes: Dict[int, Dict[str, str]],
    row_nums: List[int],
    *,
    status: str,
    message: str = "",
    group_no: str = "",
) -> None:
    for rn in row_nums:
        outcomes[rn] = {
            "status": status,
            "message": message,
            "groupNo": group_no,
            "excelRow": str(rn),
        }


def import_reimburse_workbook(
    db: Session,
    *,
    file_bytes: bytes,
    user: BxUserContext,
) -> Dict[str, Any]:
    try:
        wb = load_workbook(BytesIO(file_bytes), data_only=True)
    except Exception as exc:
        raise ValueError("无法解析 Excel 文件，请上传有效的 .xlsx") from exc

    if IMPORT_SHEET not in wb.sheetnames:
        raise ValueError(f"缺少工作表「{IMPORT_SHEET}」")

    scanned = scan_sheet_rows(wb[IMPORT_SHEET])
    outcomes: Dict[int, Dict[str, str]] = {}
    for r in scanned:
        if r.get("_kind") == "example":
            _mark_rows(
                outcomes,
                [int(r["_row"])],
                status="跳过",
                message="示例行，未导入",
                group_no=cell_str(r.get("分组号")),
            )

    rows = [r for r in scanned if r.get("_kind") == "data"]
    if len(rows) > MAX_IMPORT_ROWS:
        raise ValueError(f"有效数据超过 {MAX_IMPORT_ROWS} 行，请拆分后导入")

    tenant_no = ReimburseService._tenant_no(db, user)
    expense_nodes = load_expense_nodes_flat(db, tenant_no)
    expense_label_to_id = {lab: iid for iid, lab in build_expense_path_labels(expense_nodes).items()}

    applicants = load_applicant_options(db, user=user, tenant_no=tenant_no)
    applicant_label_to_id = {
        build_applicant_label(a["nick_name"], a["user_no"], a["id"]): a["id"] for a in applicants
    }

    wf = (
        db.query(BxWorkflowDef)
        .filter(
            BxWorkflowDef.biz_type == "reimburse",
            BxWorkflowDef.tenant_no == tenant_no,
            BxWorkflowDef.del_flag == 0,
        )
        .first()
    )

    created: List[Dict[str, Any]] = []
    errors: List[Dict[str, Any]] = []
    success_rows = 0
    fail_rows = 0

    def fail_group(group_no: str, grow: List[Dict[str, Any]], message: str) -> None:
        nonlocal fail_rows
        row_nums = [int(r["_row"]) for r in grow]
        errors.append({"groupNo": group_no, "rows": row_nums, "message": message})
        _mark_rows(outcomes, row_nums, status="失败", message=message, group_no=group_no)
        fail_rows += len(grow)

    for group in partition_import_groups(rows):
        group_no = group["group_no"]
        grow: List[Dict[str, Any]] = group["rows"]
        row_nums = [int(r["_row"]) for r in grow]

        header_err = validate_group_headers(grow)
        if header_err:
            fail_group(group_no, grow, header_err)
            continue

        applicant_label = cell_str(grow[0].get("申请人"))
        applicant_id = applicant_label_to_id.get(applicant_label)
        if not applicant_id:
            fail_group(group_no, grow, f"申请人「{applicant_label}」不存在或不在可选范围")
            continue

        try:
            resolved_id, resolved_name = ReimburseService.resolve_applicant(
                db, user, applicant_id, tenant_no=tenant_no
            )
        except Exception as exc:
            detail = getattr(exc, "detail", None) or str(exc)
            fail_group(group_no, grow, str(detail))
            continue

        date_text = cell_str(grow[0].get("报销日期"))
        try:
            reimburse_date = parse_reimburse_date(date_text)
        except ValueError:
            fail_group(group_no, grow, "报销日期格式无效，请使用 YYYY-MM-DD")
            continue

        title = cell_str(grow[0].get("报销标题"))
        details_data: List[dict] = []
        row_fail: Optional[str] = None
        for r in grow:
            expense_label = cell_str(r.get("费用项目"))
            expense_id = expense_label_to_id.get(expense_label)
            if not expense_id:
                row_fail = f"费用项目「{expense_label}」不存在或已停用"
                break
            item = ExpenseItemService.get_by_id(
                db, expense_id, tenant_no, user=user, skip_scope=True
            )
            if not item or int(item.status or 0) != 1:
                row_fail = f"费用项目「{expense_label}」不存在或已停用"
                break
            amount, amount_err = parse_amount(r.get("金额（元）"))
            if amount_err:
                row_fail = amount_err
                break
            remark = cell_str(r.get("报销事由"))
            if not remark:
                row_fail = "报销事由不能为空"
                break
            occur_raw = r.get("发生时间")
            try:
                occur_date = parse_optional_date(occur_raw if occur_raw not in ("", None) else None)
            except ValueError:
                row_fail = "发生时间格式无效，请使用 YYYY-MM-DD"
                break
            details_data.append(
                {
                    "expense_id": expense_id,
                    "expense_name": item.item_name,
                    "amount": amount,
                    "remark": remark,
                    "occur_date": occur_date,
                }
            )

        if row_fail:
            fail_group(group_no, grow, row_fail)
            continue

        data: Dict[str, Any] = {
            "apply_code": generate_apply_code(db, "BX", tenant_no),
            "status": "draft",
            "remark": "",
            "reimburse_date": reimburse_date,
        }
        if title:
            data["title"] = title
        if wf:
            data["wf_id"] = wf.id

        try:
            apply_obj = ReimburseService.create_with_details(
                db,
                data,
                details_data,
                user=user,
                applicant_id=resolved_id,
                applicant_name=resolved_name,
            )
            db.commit()
            created.append(
                {
                    "groupNo": group_no,
                    "applyId": apply_obj.id,
                    "applyCode": apply_obj.apply_code,
                }
            )
            ok_msg = f"成功，单号 {apply_obj.apply_code}"
            _mark_rows(outcomes, row_nums, status="成功", message=ok_msg, group_no=group_no)
            success_rows += len(grow)
        except Exception as exc:
            db.rollback()
            fail_group(group_no, grow, f"写入失败：{exc}")

    tip = ""
    if not rows and scanned:
        tip = "没有可导入的有效数据。模板中的【示例】行已自动跳过，请填写正式数据（报销事由不要带【示例】）后再导入。"
    elif not rows and not scanned:
        tip = "表格中没有可识别的数据行，请检查是否填写了「报销导入」工作表。"

    row_results = [
        {
            "excelRow": int(k),
            "groupNo": v.get("groupNo") or "",
            "status": v.get("status") or "",
            "message": v.get("message") or "",
        }
        for k, v in sorted(outcomes.items(), key=lambda x: x[0])
    ]

    result_file_base64 = ""
    try:
        annotated = build_annotated_result_bytes(file_bytes, outcomes)
        result_file_base64 = base64.b64encode(annotated).decode("ascii")
    except Exception:
        result_file_base64 = ""

    return {
        "successGroups": len(created),
        "successRows": success_rows,
        "failGroups": len(errors),
        "failRows": fail_rows,
        "skippedRows": sum(1 for r in row_results if r["status"] == "跳过"),
        "created": created,
        "errors": errors,
        "rowResults": row_results,
        "resultFileBase64": result_file_base64,
        "resultFileName": "报销申请导入结果.xlsx",
        "tip": tip,
    }