yhm-shop-data/store-ops/backend/xlsx.py

# -*- coding: utf-8 -*-
"""
最小化 .xlsx 读取器（Python 标准库，无需 openpyxl）

xlsx 是一个 zip 包：
  xl/sharedStrings.xml        共享字符串表
  xl/styles.xml               样式（用于识别日期格式）
  xl/workbook.xml + .rels     工作表清单
  xl/worksheets/sheetN.xml    单元格数据

只做读取，返回 rows: [[str, ...], ...]（含表头），日期单元格转为 YYYY-MM-DD。
"""
import io
import re
import zipfile
import xml.etree.ElementTree as ET
from datetime import datetime, timedelta

NS = "{http://schemas.openxmlformats.org/spreadsheetml/2006/main}"
NS_R = "{http://schemas.openxmlformats.org/officeDocument/2006/relationships}"
NS_PKG_REL = "{http://schemas.openxmlformats.org/package/2006/relationships}"

# Excel 内置日期格式编号
BUILTIN_DATE_FMTS = {14, 15, 16, 17, 18, 19, 20, 21, 22, 27, 30, 36, 45, 46, 47}
EPOCH = datetime(1899, 12, 30)


def _col_index(ref):
    """'BC12' -> 列号 54（0-based）"""
    letters = re.match(r"([A-Z]+)", ref or "")
    if not letters:
        return None
    n = 0
    for ch in letters.group(1):
        n = n * 26 + (ord(ch) - 64)
    return n - 1


def _local(tag):
    return tag.rsplit("}", 1)[-1]


def _parse_styles(z):
    """返回 numFmtId 集合中属于日期格式的 id 集合；以及 xf 索引 -> numFmtId 映射"""
    date_fmt_ids = set(BUILTIN_DATE_FMTS)
    xf_map = {}
    if "xl/styles.xml" not in z.namelist():
        return xf_map
    root = ET.fromstring(z.read("xl/styles.xml"))
    custom = {}  # id -> 是否日期
    for nf in root.iter(NS + "numFmt"):
        try:
            fid = int(nf.get("numFmtId"))
        except (TypeError, ValueError):
            continue
        code = nf.get("formatCode") or ""
        # 排除纯数字带千分位等：含 y/m/d/h 才算日期
        if re.search(r"[ymdhs]", code, re.I) and not re.match(r"^[\d,#.\s]+$", code):
            custom[fid] = True
            date_fmt_ids.add(fid)
    cellXfs = root.find(NS + "cellXfs")
    if cellXfs is not None:
        for i, xf in enumerate(cellXfs.findall(NS + "xf")):
            try:
                xf_map[i] = int(xf.get("numFmtId") or 0)
            except ValueError:
                xf_map[i] = 0
    # 标记自定义日期格式
    for i, fid in xf_map.items():
        if fid in custom:
            xf_map[i] = -1  # 特殊标记：自定义日期
    return xf_map


def _is_date_style(xf_map, s_attr):
    if not s_attr:
        return False
    try:
        fid = xf_map.get(int(s_attr))
    except ValueError:
        return False
    return fid in BUILTIN_DATE_FMTS or fid == -1


def _serial_to_date(v):
    try:
        n = float(v)
    except ValueError:
        return None
    if n <= 0 or n > 60000:
        return None
    d = EPOCH + timedelta(days=n)
    if n % 1 == 0:
        return d.strftime("%Y-%m-%d")
    return d.strftime("%Y-%m-%d")  # 时间部分舍弃，日粒度只到天


def _shared_strings(z):
    if "xl/sharedStrings.xml" not in z.namelist():
        return []
    root = ET.fromstring(z.read("xl/sharedStrings.xml"))
    out = []
    for si in root.findall(NS + "si"):
        # 富文本 run 与纯文本 t 都归并为拼接
        out.append("".join(t.text or "" for t in si.iter(NS + "t")))
    return out


def _sheet_paths(z):
    """按 workbook.xml 里的顺序返回各 sheet 的 zip 路径"""
    names = z.namelist()
    wb = ET.fromstring(z.read("xl/workbook.xml"))
    rels = ET.fromstring(z.read("xl/_rels/workbook.xml.rels"))
    rel_map = {r.get("Id"): r.get("Target") for r in rels.findall(NS_PKG_REL + "Relationship")}
    out = []
    for sh in wb.find(NS + "sheets").findall(NS + "sheet"):
        rid = sh.get(NS_R + "id")
        target = rel_map.get(rid, "")
        target = target.lstrip("/")
        if not target.startswith("xl/"):
            target = "xl/" + target
        if target in names:
            out.append((sh.get("name") or "", target))
    return out


def _sheet_rows(z, sheet_path, sst, xf_map):
    root = ET.fromstring(z.read(sheet_path))
    rows = []
    for row in root.iter(NS + "row"):
        cells = {}
        max_i = -1
        for c in row.findall(NS + "c"):
            ref = c.get("r") or ""
            i = _col_index(ref)
            t = c.get("t") or ""
            s_attr = c.get("s")
            val = None
            if t == "inlineStr":
                is_el = c.find(NS + "is")
                if is_el is not None:
                    val = "".join(x.text or "" for x in is_el.iter(NS + "t"))
            else:
                v_el = c.find(NS + "v")
                raw = v_el.text if v_el is not None else None
                if raw is not None:
                    if t == "s":
                        try:
                            val = sst[int(raw)]
                        except (ValueError, IndexError):
                            val = raw
                    elif t == "b":
                        val = "1" if raw.strip() == "1" else "0"
                    elif t == "str":
                        val = raw
                    else:
                        # 数字或日期
                        if _is_date_style(xf_map, s_attr):
                            val = _serial_to_date(raw) or raw
                        else:
                            val = raw
            if i is None:
                i = max_i + 1
            if val is not None:
                # 4.0 -> 4，去掉多余的 .0
                if re.match(r"^-?\d+\.0+$", val.strip()):
                    val = val.strip().split(".")[0]
                cells[i] = val
                max_i = max(max_i, i)
        if cells:
            width = max(cells) + 1
            rows.append([cells.get(i, "") for i in range(width)])
    return rows


def read_bytes(raw):
    """返回 (rows, sheet_names)。取第一个非空工作表。"""
    try:
        z = zipfile.ZipFile(io.BytesIO(raw))
    except zipfile.BadZipFile:
        raise ValueError("不是有效的 xlsx 文件（可能是旧版 .xls，请另存为 .xlsx）")
    sst = _shared_strings(z)
    xf_map = _parse_styles(z)
    sheets = _sheet_paths(z)
    if not sheets:
        raise ValueError("xlsx 中没有工作表")
    for name, path in sheets:
        rows = _sheet_rows(z, path, sst, xf_map)
        if rows:
            return rows, [n for n, _ in sheets]
    return [], [n for n, _ in sheets]


def read_file(path):
    with open(path, "rb") as fp:
        return read_bytes(fp.read())