pandas 完整实践：销售质量校验与客户月报 · 输入与成功标准至原子发布与失败边界

前置：连接与分组。目标是把销售事实表与客户地区维度合并，在上海时区按月统计订单数与总分。所有金额和日期规则都在进入聚合前验证；发现坏数据就不发布整份报告。示例未执行，以下输出是预期结果，静态检查不等于运行正确。

pandas 完整实践：销售质量校验与客户月报

输入与成功标准

pandas销售月报输入是什么：sales.csv每行是一笔销售，customers.csv每行是一位客户，销售主键唯一而客户维度键唯一。成功标准是所有销售都匹配客户，输出五个按月份和地区排序的分组，总计7500分，订单数合计5。001的前导零保留，空地区归入unknown，零金额订单仍计入订单数。

列规则、每表行数上限、金额范围和本地时间窗口见下方同源README。样例覆盖UTC月末跨月和三种地区；它不是随机数据，因此预期结果能手算核对。输出只包含month、region、order_count、total_cents，不包含用户明细。先读数据模型，避免把输入字符串转换和业务合法性混成一个步骤。

处理流程与代码

pandas月报流水线怎么组织：先严格解析CSV结构，再验证ID、地区、金额和时间，随后执行多对一连接与命名聚合，最后只发布一份CSV。read_table负责结构，build_report负责领域规则，publish负责文件替换；main把可预期输入和I/O错误转成非零退出。按边界拆函数是为了单独断言数据和输出行为，不引入插件、配置中心或服务。

全部工程文件均以同源引用展示，修改任一源码会同步影响本页，避免复制片段与实际文件分叉。工程没有导入时执行CLI的副作用。DataFrame均在函数内从字符串行构造，调用者不需要理解隐藏的全局选项。

report.py：完整实现

python

"""Local sales quality gate and monthly report. Teaching example: 未执行."""
import argparse
import csv
import os
from pathlib import Path
import re
import sys
import tempfile

import pandas as pd

MAX_ROWS = 100_000
MAX_CENTS = 10**12
INT64_MAX = 2**63 - 1
SALES = ['sale_id', 'customer_id', 'sold_at', 'amount_cents']
CUSTOMERS = ['customer_id', 'region']
IDENTIFIER = re.compile(r'[A-Za-z0-9_-]{1,32}')
TIMESTAMP = re.compile(r'2025-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:Z|[+-]\d{2}:\d{2})')


def read_table(path, columns):
    # A strict CSV shape gate prevents silent row truncation/index inference.
    with Path(path).open(encoding='utf-8-sig', newline='') as stream:
        reader = csv.reader(stream, strict=True)
        if next(reader, None) != columns:
            raise ValueError(f'{Path(path).name}: exact header required')
        rows = []
        for row in reader:
            if len(row) != len(columns):
                raise ValueError(f'{Path(path).name}: wrong field count')
            if len(rows) >= MAX_ROWS:
                raise ValueError('row limit exceeded')
            rows.append(row)
    return pd.DataFrame(rows, columns=columns, dtype='string')


def require_ids(frame, column, unique=False):
    if not frame[column].str.fullmatch(IDENTIFIER).all():
        raise ValueError(f'invalid {column}')
    if unique and frame[column].duplicated().any():
        raise ValueError(f'duplicate {column}')


def build_report(sales_path, customers_path):
    sales = read_table(sales_path, SALES)
    customers = read_table(customers_path, CUSTOMERS)
    if sales.empty or customers.empty:
        raise ValueError('empty input table')
    require_ids(sales, 'sale_id', unique=True)
    require_ids(sales, 'customer_id')
    require_ids(customers, 'customer_id', unique=True)
    if not customers['region'].isin(['east', 'west', '']).all():
        raise ValueError('invalid region')
    # Only an empty region is nullable. No implicit NA/NULL string conversion.
    customers['region'] = customers['region'].replace('', pd.NA).fillna('unknown')
    amount = sales['amount_cents']
    if not amount.str.fullmatch(r'0|[1-9][0-9]{0,12}').all():
        raise ValueError('invalid amount_cents')
    integers = [int(value) for value in amount]
    if any(value > MAX_CENTS for value in integers):
        raise ValueError('amount_cents exceeds limit')
    if sum(integers) > INT64_MAX:
        raise ValueError('total exceeds int64')
    sales['amount_cents'] = pd.Series(integers, dtype='int64')
    if not sales['sold_at'].str.fullmatch(TIMESTAMP).all():
        raise ValueError('sold_at requires 2025 ISO timestamp with offset')
    stamps = pd.to_datetime(sales['sold_at'], format='ISO8601', errors='raise', utc=True)
    local = stamps.dt.tz_convert('Asia/Shanghai')
    if not ((local >= pd.Timestamp('2025-01-01', tz='Asia/Shanghai')) &
            (local < pd.Timestamp('2026-01-01', tz='Asia/Shanghai'))).all():
        raise ValueError('sold_at outside local 2025 window')
    sales['month'] = local.dt.strftime('%Y-%m')
    joined = sales.merge(customers, on='customer_id', how='left',
                         validate='many_to_one', indicator=True)
    if not joined['_merge'].eq('both').all():
        raise ValueError('unknown customer_id')
    report = (joined.groupby(['month', 'region'], as_index=False, sort=True,
                             dropna=False, observed=True)
              .agg(order_count=('sale_id', 'size'), total_cents=('amount_cents', 'sum')))
    return report[['month', 'region', 'order_count', 'total_cents']]


def publish(report, target):
    target = Path(target)
    temporary = None
    try:
        # Same-directory temporary file keeps os.replace on one filesystem.
        with tempfile.NamedTemporaryFile(mode='w', encoding='utf-8', newline='',
                                         dir=target.parent, prefix='.sales-',
                                         suffix='.tmp', delete=False) as stream:
            temporary = Path(stream.name)
            report.to_csv(stream, index=False, lineterminator='\n')
            stream.flush()
            os.fsync(stream.fileno())
        os.replace(temporary, target)
    finally:
        if temporary is not None:
            temporary.unlink(missing_ok=True)


def main(argv=None):
    parser = argparse.ArgumentParser(description='Validate sales and publish one monthly CSV')
    parser.add_argument('sales', type=Path)
    parser.add_argument('customers', type=Path)
    parser.add_argument('output', type=Path)
    args = parser.parse_args(argv)
    try:
        if args.output.resolve() in {args.sales.resolve(), args.customers.resolve()}:
            raise ValueError('output must differ from inputs')
        result = build_report(args.sales, args.customers)
        publish(result, args.output)
    except (OSError, UnicodeError, csv.Error, ValueError, OverflowError) as exc:
        print(f'error: {exc}', file=sys.stderr)
        return 1
    print(f'written {len(result)} groups')
    return 0


if __name__ == '__main__':
    raise SystemExit(main())

直接依赖与环境约束

text

pandas==2.3.3
tzdata==2025.2

toml

[project]
name = "p2-pandas-sales-notes"
version = "0.1.0"
requires-python = ">=3.12,<3.13"
dependencies = ["pandas==2.3.3", "tzdata==2025.2"]

有效输入与预期输出

text

sale_id,customer_id,sold_at,amount_cents
s01,001,2025-01-31T15:59:59Z,1000
s02,001,2025-01-31T16:00:00Z,2000
s03,002,2025-02-15T12:00:00+08:00,3000
s04,003,2025-02-16T12:00:00+08:00,0
s05,002,2025-02-28T16:00:00Z,1500

text

customer_id,region
001,east
002,west
003,

text

month,region,order_count,total_cents
2025-01,east,1,1000
2025-02,east,1,2000
2025-02,unknown,1,0
2025-02,west,1,3000
2025-03,west,1,1500

执行说明与预期结果

pandas月报命令怎样核对：在examples/python/p2-pandas目录使用准备好的CPython 3.12虚拟环境，运行report.py并传入销售、客户和输出三个路径，预期退出0且stdout为written 5 groups。比较输出文件与expected/monthly.csv时核对列顺序、LF换行及index=False，不能只看屏幕上打印的DataFrame摘要。

PowerShell与Bash对应命令及依赖安装说明见同源README，均未在本次执行。已有输出会被成功结果替换，应选择教学目录内专用输出。输出父目录须已存在，输入路径按当前工作目录解析。不要把脚本命名为pandas.py，否则会遮蔽第三方模块。

README：准备、运行、错误与限制

# pandas 销售质量与月报

示例未执行。仅做源码 AST、CSV/TOML/JSON 结构与文档模板静态检查；未安装依赖，未导入 pandas，未运行测试。目标 CPython 3.12、pandas 2.3.3；tzdata 为 Windows 提供 IANA 时区。requirements 与 pyproject 是同一组直接依赖声明，不是完整传递依赖锁文件。

在此目录准备环境后，可手动使用 PowerShell `python -m venv .venv`、`.\.venv\Scripts\python.exe -m pip install -r requirements.txt`；Bash 对应 `python3 -m venv .venv`、`.venv/bin/python -m pip install -r requirements.txt`。这些命令是读者操作说明，本次未执行。

Windows 用 `.\.venv\Scripts\python.exe report.py input/sales.csv input/customers.csv monthly.csv`；Linux/macOS 用 `.venv/bin/python report.py input/sales.csv input/customers.csv monthly.csv`。预期输出 `written 5 groups`、退出0，monthly.csv 与 expected/monthly.csv 按字节相同。字段顺序固定，LF换行，无 DataFrame 索引列，整数分无小数点。两平台测试命令分别为 `.\.venv\Scripts\python.exe -m unittest -v` 和 `.venv/bin/python -m unittest -v`，预期5个测试方法通过（含子场景），实际未执行。

输入必须为 UTF-8/BOM 可选、准确表头、不含空记录或多余列，每表1至100000行。sale_id 唯一，customer_id 可重复出现在销售表但客户表唯一；ID为1–32个ASCII字母数字下划线连字符，保留001前导零。仅region空字符串允许缺失，填为unknown；其他region仅east/west。金额单位分，ASCII非负无前导零整数（0除外），每条不超过10^12。先用 Python int 求和并检查 int64 上界，再转 int64，避免聚合溢出；既定行数与单行上界还使总额最多10^17。

时间必须完整秒、带Z或数值偏移，文本年份2025；转UTC后再转Asia/Shanghai，以当地2025年左闭右开窗口验收。2025-01-31T16:00:00Z属于二月；2025-12-31T16:00:00Z超出窗口。客户必须匹配，未知客户立即失败；按当地月份与region聚合，按这两列排序。

invalid/ 依次提供重复客户、未知客户、小数金额、无时区时间、空ID、跨年和重复销售；替换相应输入时预期退出1、stderr以error开头、已有输出不变。缺少参数由argparse退出2。测试还覆盖坏表头、多列、空数据、非法日期、超额、负数、输出等于输入以及替换失败时临时文件清理。

先验证全部数据、在内存生成结果，再在输出目录写临时文件、flush/fsync后os.replace。目标父目录须已存在；不会创建目录。正常校验失败不会产生新输出；替换前I/O失败保留旧文件并清理临时文件。单文件替换不提供多文件事务、掉电后目录持久化或并发写隔离；进程被强杀可能留下临时文件，读者清理前应确认没有活跃写者。Windows文件锁可能令替换失败。拒绝相同解析路径，但不防硬链接别名或恶意并发改路径，输入目录应可信且在运行期间稳定。

全部数据留在内存，不承诺大数据流处理，也不输出个人身份信息或访问网络。金额不支持退款、汇率与税务，报表不是会计账本。

原子发布与失败边界

pandas校验失败如何保留旧报表：全部数据验证完成后才创建同目录临时文件，写入、flush和fsync成功再用os.replace替换目标。输入错误不会创建部分报表，替换前写入失败由finally清理临时文件，旧输出保持原样。这里的原子性只涵盖单文件目录项替换，不代表多文件事务或崩溃恢复系统。

程序不创建父目录；Windows目标被其他程序锁住时替换可能失败。进程被强杀可能遗留临时文件，且未fsync父目录，不能承诺掉电后目录项持久化。并发写者采用最后成功替换者结果，没有锁；输入目录需可信且稳定，路径相等检查不防恶意硬链接和竞态。目标是清楚交代边界，而不是为教学报表建立复杂事务层。

文件替换与跨文件系统边界参考Python中文os.replace文档。

pandas 完整实践：销售质量校验与客户月报 · 输入与成功标准至原子发布与失败边界 ​