蚂蚁 AI Coding 8.20 真题解析：征信对账系统账单解析引擎

from pathlib import Path
import pandas as pd


def inspect_workbooks(root: str) -> None:
    for path in sorted(Path(root).glob("*.xls*")):
        print(f"\n=== {path.name} ===")
        book = pd.ExcelFile(path)
        print("sheets:", book.sheet_names)

        for sheet in book.sheet_names:
            raw = pd.read_excel(path, sheet_name=sheet, header=None, dtype=object)
            print(f"\n-- {sheet}: shape={raw.shape} --")
            print(raw.head(12).to_string(index=False, header=False))


if __name__ == "__main__":
    inspect_workbooks(".")

if "机构A" in filename:
    df = pd.read_excel(path, sheet_name="账单明细", skiprows=2)
elif "机构B" in filename:
    df = pd.read_excel(path, sheet_name="Sheet1", skiprows=5)

from dataclasses import asdict, dataclass
from decimal import Decimal
from typing import Optional


@dataclass
class BillItem:
    institution_name: str
    institution_code: Optional[str]
    billing_period: Optional[str]
    product_name: str
    quantity: Optional[int]
    unit_price: Optional[Decimal]
    amount: Decimal
    source_file: str
    source_sheet: str
    source_row: int

    def to_dict(self) -> dict:
        data = asdict(self)
        data["unit_price"] = (
            str(self.unit_price) if self.unit_price is not None else None
        )
        data["amount"] = str(self.amount)
        return data

HEADER_ALIASES = {
    "institution_name": {"机构", "机构名称", "合作机构", "数据源"},
    "institution_code": {"机构编码", "机构代码", "渠道编码"},
    "product_name": {"产品", "产品名称", "业务类型", "服务名称"},
    "quantity": {"数量", "调用量", "查询次数", "计费数量"},
    "unit_price": {"单价", "计费单价", "含税单价"},
    "amount": {"金额", "费用", "应付金额", "合计金额"},
}


def normalize_header(value: object) -> str:
    if value is None:
        return ""
    return "".join(str(value).strip().lower().split())


def find_header_row(raw, scan_rows: int = 20) -> int | None:
    aliases = {
        normalize_header(alias)
        for names in HEADER_ALIASES.values()
        for alias in names
    }
    best_row = None
    best_score = 0

    for row_index in range(min(scan_rows, len(raw))):
        values = {normalize_header(v) for v in raw.iloc[row_index].tolist()}
        score = len(values & aliases)
        if score > best_score:
            best_row = row_index
            best_score = score

    return best_row if best_score >= 2 else None

def build_column_mapping(columns) -> dict[str, str]:
    mapping = {}
    for column in columns:
        normalized = normalize_header(column)
        for target, aliases in HEADER_ALIASES.items():
            normalized_aliases = {normalize_header(x) for x in aliases}
            if normalized in normalized_aliases:
                mapping[column] = target
                break
    return mapping

INSTITUTION_ALIASES = {
    "机构A": {"机构A", "A机构", "机构A有限公司", "CODE001"},
    "机构B": {"机构B", "B机构", "机构B有限公司", "CODE002"},
}


def normalize_institution(value: str) -> str:
    compact = "".join(str(value).split())
    for canonical, aliases in INSTITUTION_ALIASES.items():
        if compact in {"".join(x.split()) for x in aliases}:
            return canonical
    return compact

SUMMARY_WORDS = {
    "合计", "总计", "小计", "本页合计", "费用合计", "备注", "说明"
}


def is_detail_row(row: dict) -> bool:
    text = " ".join(
        str(value).strip()
        for value in row.values()
        if value is not None
    )

    if not text:
        return False
    if any(word in text for word in SUMMARY_WORDS):
        return False

    product = row.get("product_name")
    amount = row.get("amount")
    quantity = row.get("quantity")

    # 至少出现产品，或出现可计费的数量/金额信息。
    return bool(product) and (amount is not None or quantity is not None)

import re
from decimal import Decimal, InvalidOperation


def parse_decimal(value) -> Decimal | None:
    if value is None:
        return None

    text = str(value).strip()
    if not text or text.lower() in {"nan", "none", "-", "--"}:
        return None

    multiplier = Decimal("1")
    if text.endswith("万"):
        multiplier = Decimal("10000")
        text = text[:-1]

    text = re.sub(r"[￥¥$,，\s]", "", text)
    text = text.replace("元", "")

    try:
        return Decimal(text) * multiplier
    except InvalidOperation:
        return None

from decimal import Decimal


def amount_is_consistent(quantity, unit_price, amount) -> bool:
    if quantity is None or unit_price is None or amount is None:
        return True
    expected = Decimal(quantity) * unit_price
    return abs(expected - amount) <= Decimal("0.01")

from pathlib import Path
import json
import pandas as pd


def parse_sheet(path: Path, sheet_name: str) -> list[dict]:
    raw = pd.read_excel(
        path,
        sheet_name=sheet_name,
        header=None,
        dtype=object,
    )
    header_row = find_header_row(raw)
    if header_row is None:
        return []

    headers = raw.iloc[header_row].tolist()
    table = raw.iloc[header_row + 1:].copy()
    table.columns = headers
    table = table.where(pd.notna(table), None)

    mapping = build_column_mapping(table.columns)
    table = table.rename(columns=mapping)

    records = []
    for offset, series in table.iterrows():
        row = series.to_dict()
        if not is_detail_row(row):
            continue

        # 这里继续执行机构、产品、账期和金额归一化，
        # 再映射成 README 要求的最终字段。
        row["source_file"] = path.name
        row["source_sheet"] = sheet_name
        row["source_row"] = int(offset) + 1
        records.append(row)
    return records


def parse_directory(root: str) -> list[dict]:
    records = []
    for path in sorted(Path(root).glob("*.xls*")):
        book = pd.ExcelFile(path)
        for sheet_name in book.sheet_names:
            records.extend(parse_sheet(path, sheet_name))
    return records


def main() -> None:
    records = parse_directory(".")
    print(json.dumps(records, ensure_ascii=False, allow_nan=False))


if __name__ == "__main__":
    main()

from decimal import Decimal


def test_parse_decimal():
    assert parse_decimal("1,234.50") == Decimal("1234.50")
    assert parse_decimal("￥88.00元") == Decimal("88.00")
    assert parse_decimal("1.2万") == Decimal("12000.0")
    assert parse_decimal("-") is None