用 Amazon Quick Automate 把多工作表 RFI 问卷自动整理为 CSV

s3://acme-rfi-automation/incoming/vendor-security-rfi.xlsx
s3://acme-rfi-automation/output/vendor-security-rfi.csv

读取输入的 Excel 工作簿，并检查所有非隐藏工作表。

目标：提取需要供应商回答的 RFI 问题，输出为结构化记录。

每条记录必须包含：
- source_sheet：原始工作表名称
- question_id：原始编号，没有编号时为空字符串
- section：最近且适用的章节标题
- question：去除多余换行和首尾空格后的完整问题
- response_type：只能是 text、yes_no、single_choice、multi_choice
- required：只能是 true 或 false；无法确定时使用 false
- notes：回答限制、选项或补充说明，没有时为空字符串

规则：
1. 忽略封面、填写说明、版权文本、纯标题、空行和示例答案。
2. 不要补写源文件中不存在的问题、编号或要求。
3. 合并因单元格换行而拆开的同一个问题，但不要合并两个独立问题。
4. 不同工作表出现相同编号时仍保留两条记录，并通过 source_sheet 区分。
5. 保持问题原有顺序：先按工作表顺序，再按工作表内行顺序。
6. 输出 UTF-8 CSV，列顺序严格按照上述字段顺序。

复查 Security Controls 工作表：以“Control family:”开头的行是章节标题，
不应成为问题。将其值写入后续问题的 section，直到出现下一个章节标题。
其他提取规则和输出字段保持不变。

export AWS_REGION="us-east-1"
export RFI_BUCKET="acme-rfi-automation"
export INPUT_FILE="vendor-security-rfi.xlsx"
export OUTPUT_FILE="vendor-security-rfi.csv"

aws s3 cp "$INPUT_FILE" \
  "s3://$RFI_BUCKET/incoming/$INPUT_FILE" \
  --region "$AWS_REGION"

aws s3api head-object \
  --bucket "$RFI_BUCKET" \
  --key "incoming/$INPUT_FILE"

aws s3 cp \
  "s3://$RFI_BUCKET/output/$OUTPUT_FILE" \
  "./$OUTPUT_FILE" \
  --region "$AWS_REGION"

python - "$OUTPUT_FILE" <<'PY'
import csv
import sys
from collections import Counter
from pathlib import Path

path = Path(sys.argv[1])
required_columns = [
    "source_sheet",
    "question_id",
    "section",
    "question",
    "response_type",
    "required",
    "notes",
]
allowed_types = {"text", "yes_no", "single_choice", "multi_choice"}

with path.open("r", encoding="utf-8-sig", newline="") as handle:
    reader = csv.DictReader(handle)
    if reader.fieldnames != required_columns:
        raise SystemExit(
            f"Unexpected columns: {reader.fieldnames}; expected: {required_columns}"
        )
    rows = list(reader)

errors = []
for line_number, row in enumerate(rows, start=2):
    if not row["source_sheet"].strip():
        errors.append(f"line {line_number}: source_sheet is empty")
    if not row["question"].strip():
        errors.append(f"line {line_number}: question is empty")
    if row["response_type"] not in allowed_types:
        errors.append(f"line {line_number}: invalid response_type")
    if row["required"].lower() not in {"true", "false"}:
        errors.append(f"line {line_number}: required must be true or false")

if errors:
    raise SystemExit("\n".join(errors))

counts = Counter(row["source_sheet"] for row in rows)
print(f"validated {len(rows)} questions")
for sheet, count in sorted(counts.items()):
    print(f"{sheet}: {count}")
PY