#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ keinen build script（v2フォーム＋v1フォームを統合） GoogleフォームのCSV回答 + ダウンロード済み画像 から、公開ページが読む `data/records.json` と 公開用 `data/keinen-vX.Y.csv`、および `images/NNN.webp` を生成する。 設計方針（ビルドブリーフより）: - 個人情報列（ワーカー名 / Googleアカウント名 / メールアドレス / タイムスタンプ / 同意欄）は 出力に一切含めない。 - 画像は長辺1200pxのwebpに変換。元画像はリポジトリに入れない（form_*/ は .gitignore）。 - 自由記述は絶対に編集しない（原文のまま）。 - データが増えたら本スクリプトを再実行するだけで更新される。 統合仕様: - v2フォーム（form_260819）: 現在の設問。schema_version="v2"。 - v1フォーム（form_v1（260819））: 新旧ペア構造。**古い方を1レコード**として扱い schema_version="v1"。 - 使用年数はフリーテキストなので正規化してから格納 - 「新しい方と比べて何が変わったか」は change にマップ（v2には該当設問がないため、 現状 change を持つのは v1 の2件だけ。これは正しい状態） - 新品側の触感ラベルは texture_words_new として保持（カードには表示しない） - 画像は【古いもの】接写のみ採用 使い方: python3 scripts/build.py python3 scripts/build.py --v2 form_260819 --v1 "form_v1（260819）" 前提: - Pillow (webp対応): pip install pillow - scripts/image_map.json に「Drive file id → 元画像ファイル名」の対応があること。 """ import argparse import csv import glob import json import os import re import sys from datetime import date try: from PIL import Image, ImageOps except ImportError: sys.exit("Pillow が見つかりません。 `pip install pillow` を実行してください。") # ---- 設定 ------------------------------------------------------------- VERSION = "0.2" LICENSE = "CC BY 4.0" LONG_EDGE = 1200 WEBP_QUALITY = 82 ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) DATA_DIR = os.path.join(ROOT, "data") IMAGES_DIR = os.path.join(ROOT, "images") IMAGE_MAP_PATH = os.path.join(ROOT, "scripts", "image_map.json") CONSENT_OK = "同意する" NA_PREFIX = "該当なし" DRIVE_ID_RE = re.compile(r"[?&]id=([A-Za-z0-9_-]+)") ZEN2HAN = str.maketrans("０１２３４５６７８９", "0123456789") # 使用年数の正規化先（フォームの実際の選択肢に合わせる） YEARS_CANON = { "半年未満", "半年〜1年くらい", "1〜3年くらい", "3〜5年くらい", "5〜10年くらい", "10年以上", "わからない・人からもらった", } # v2 CSV の列インデックス V2 = dict(TIMESTAMP=0, WORKER=1, CATEGORY=2, IMAGE_URL=3, YEARS=4, FREQUENCY=5, DESCRIPTION=6, TEXTURE=7, LOCAL_VARIATION=8, CONSENT=9) # v1 CSV の列インデックス V1 = dict(TIMESTAMP=0, CATEGORY=1, WASH=2, CARE=3, IMG_NEW_WIDE=4, IMG_OLD_WIDE=5, IMG_NEW_CLOSE=6, IMG_OLD_CLOSE=7, BUY_NEW=8, BUY_OLD=9, FREQUENCY=10, DESCRIPTION=11, TEXTURE_NEW=12, TEXTURE_OLD=13, CHANGE=14, AUDIO=15, CONSENT=16, WORKER=17) # ---- ユーティリティ --------------------------------------------------- def extract_drive_id(url): m = DRIVE_ID_RE.search(url or "") if not m: m = re.search(r"/d/([A-Za-z0-9_-]+)", url or "") return m.group(1) if m else "" def split_words(cell): if not cell: return [] return [p.strip() for p in re.split(r"[,、]", cell) if p.strip()] def normalize_years(s): """フリーテキストの使用年数を YEARS_CANON のいずれかに正規化する。""" s = (s or "").strip() if not s: return "" if s in YEARS_CANON: return s if "わからない" in s or "もらった" in s: return "わからない・人からもらった" if "10年以上" in s: return "10年以上" t = s.translate(ZEN2HAN) nums = [int(x) for x in re.findall(r"\d+", t)] if nums: n = max(nums) if n >= 10: return "10年以上" if n >= 5: return "5〜10年くらい" if n >= 3: return "3〜5年くらい" if n >= 1: return "1〜3年くらい" if "半年" in s: return "半年未満" return s # 最後の手段: 原文のまま（重み0で最下部に落ちる） def find_source_image(root, filename): """root配下（サブフォルダ含む）から basename が一致する画像を探す。""" for base, _dirs, files in os.walk(root): if filename in files: return os.path.join(base, filename) return None def account_name_from(filename): """元ファイル名の ' - アカウント名.ext' 部分を取り出す（PII自己チェック用）。""" m = re.search(r" - (.+)\.[^.]+$", filename) return m.group(1).strip() if m else "" def convert_image(src_path, dst_path): with Image.open(src_path) as im: im = ImageOps.exif_transpose(im) im = im.convert("RGB") w, h = im.size scale = LONG_EDGE / max(w, h) if scale < 1: im = im.resize((round(w * scale), round(h * scale)), Image.LANCZOS) im.save(dst_path, "WEBP", quality=WEBP_QUALITY, method=6) # ---- 各フォームのステージング（id採番・画像変換は後でまとめて行う）---- def stage_v2(source_dir, image_map, warnings): csvs = sorted(glob.glob(os.path.join(source_dir, "*.csv"))) if not csvs: sys.exit(f"v2 CSVが {source_dir} に見つかりません。") with open(csvs[0], encoding="utf-8") as f: rows = list(csv.reader(f))[1:] print(f"v2 CSV: {os.path.relpath(csvs[0], ROOT)} ({len(rows)}行)") staged = [] for i, row in enumerate(rows, 1): if len(row) <= V2["CONSENT"] or row[V2["CONSENT"]].strip() != CONSENT_OK: warnings.append(f"v2 行{i}: 同意なし/列不足でスキップ") continue did = extract_drive_id(row[V2["IMAGE_URL"]]) fn = image_map.get(did) if not fn: warnings.append(f"v2 行{i}: image_mapに id {did} なし → スキップ") continue src = find_source_image(source_dir, fn) if not src: warnings.append(f"v2 行{i}: 画像 {fn} が見つからず → スキップ") continue rec = { "schema_version": "v2", "category": row[V2["CATEGORY"]].strip(), "years": normalize_years(row[V2["YEARS"]]), "frequency": row[V2["FREQUENCY"]].strip(), "texture_words": split_words(row[V2["TEXTURE"]]), "description": row[V2["DESCRIPTION"]].strip(), } lv = row[V2["LOCAL_VARIATION"]].strip() if lv: rec["local_variation"] = lv rec["_src"] = src rec["_pii"] = [row[V2["WORKER"]].strip(), account_name_from(fn)] staged.append(rec) return staged def stage_v1(v1_dir, image_map, warnings): csvs = sorted(glob.glob(os.path.join(v1_dir, "*.csv"))) if not csvs: warnings.append(f"v1 CSVが {v1_dir} に見つからない → v1は統合しない") return [] with open(csvs[0], encoding="utf-8") as f: rows = list(csv.reader(f))[1:] print(f"v1 CSV: {os.path.relpath(csvs[0], ROOT)} ({len(rows)}行)") staged = [] for i, row in enumerate(rows, 1): if len(row) <= V1["WORKER"] or row[V1["CONSENT"]].strip() != CONSENT_OK: warnings.append(f"v1 行{i}: 同意なし/列不足でスキップ") continue did = extract_drive_id(row[V1["IMG_OLD_CLOSE"]]) # 古いもの接写のみ採用 fn = image_map.get(did) if not fn: warnings.append(f"v1 行{i}: image_mapに id {did}（古接写）なし → スキップ") continue src = find_source_image(v1_dir, fn) if not src: warnings.append(f"v1 行{i}: 画像 {fn} が見つからず → スキップ") continue rec = { "schema_version": "v1", "category": row[V1["CATEGORY"]].strip(), "years": normalize_years(row[V1["BUY_OLD"]]), "frequency": row[V1["FREQUENCY"]].strip(), "texture_words": split_words(row[V1["TEXTURE_OLD"]]), "texture_words_new": split_words(row[V1["TEXTURE_NEW"]]), # 新品側（非表示） "description": row[V1["DESCRIPTION"]].strip(), } change = row[V1["CHANGE"]].strip() if change: rec["change"] = change # v1固有: 洗濯 or 手入れ の頻度（該当する方だけ） wash, care = row[V1["WASH"]].strip(), row[V1["CARE"]].strip() if wash and not wash.startswith(NA_PREFIX): rec["maintenance"] = wash rec["maintenance_kind"] = "洗濯・使用頻度" elif care and not care.startswith(NA_PREFIX): rec["maintenance"] = care rec["maintenance_kind"] = "手入れ" rec["_src"] = src rec["_pii"] = [row[V1["WORKER"]].strip(), account_name_from(fn)] staged.append(rec) return staged # ---- メイン ----------------------------------------------------------- # 出力するキーの順序（存在するものだけ書き出す） FIELD_ORDER = ["id", "schema_version", "category", "years", "frequency", "image", "texture_words", "texture_words_new", "description", "change", "local_variation", "maintenance", "maintenance_kind"] def main(): ap = argparse.ArgumentParser() ap.add_argument("--v2", default="form_260819") ap.add_argument("--v1", default="form_v1（260819）") args = ap.parse_args() def resolve(p): return p if os.path.isabs(p) else os.path.join(ROOT, p) v2_dir, v1_dir = resolve(args.v2), resolve(args.v1) with open(IMAGE_MAP_PATH, encoding="utf-8") as f: image_map = {k: v for k, v in json.load(f).items() if not k.startswith("_")} os.makedirs(DATA_DIR, exist_ok=True) os.makedirs(IMAGES_DIR, exist_ok=True) warnings = [] staged = stage_v2(v2_dir, image_map, warnings) + stage_v1(v1_dir, image_map, warnings) # 既存webpを掃除（件数が減ったときの取り残し防止） for old in glob.glob(os.path.join(IMAGES_DIR, "*.webp")): os.remove(old) forbidden = set() records = [] for i, rec in enumerate(staged, 1): rid = f"{i:03d}" rel_image = f"images/{rid}.webp" convert_image(rec.pop("_src"), os.path.join(ROOT, rel_image)) for bad in rec.pop("_pii"): if bad: forbidden.add(bad) rec["id"] = rid rec["image"] = rel_image records.append({k: rec[k] for k in FIELD_ORDER if k in rec}) payload = { "version": VERSION, "updated": date.today().isoformat(), "license": LICENSE, "count": len(records), "records": records, } records_json_path = os.path.join(DATA_DIR, "records.json") with open(records_json_path, "w", encoding="utf-8") as f: json.dump(payload, f, ensure_ascii=False, indent=2) csv_out_path = os.path.join(DATA_DIR, f"keinen-v{VERSION}.csv") cols = ["id", "schema_version", "category", "years", "frequency", "image", "texture_words", "texture_words_new", "change", "description", "local_variation", "maintenance", "maintenance_kind"] with open(csv_out_path, "w", encoding="utf-8", newline="") as f: w = csv.writer(f) w.writerow(cols) for r in records: w.writerow([ r.get(c) if not isinstance(r.get(c), list) else " / ".join(r.get(c, [])) for c in cols ]) # 旧バージョンのCSVが残っていたら消す for old in glob.glob(os.path.join(DATA_DIR, "keinen-v*.csv")): if os.path.basename(old) != f"keinen-v{VERSION}.csv": os.remove(old) problems = check_pii(records_json_path, csv_out_path, forbidden) n_v1 = sum(1 for r in records if r["schema_version"] == "v1") n_v2 = sum(1 for r in records if r["schema_version"] == "v2") print(f"\n生成: data/records.json (全{len(records)}件 / v2={n_v2}・v1={n_v1})") print(f"生成: data/keinen-v{VERSION}.csv") print(f"画像: images/001.webp 〜 {len(records):03d}.webp") if warnings: print("\n[注意]") for w in warnings: print(" -", w) if problems: print("\n\033[31m[PIIチェック失敗]\033[0m") for p in problems: print(" -", p) sys.exit(1) print("\n[PIIチェック] OK（'@' / ワーカー名 / アカウント名 は生成物に含まれていません）") print("※ 画像そのものへの写り込みは人間が目視確認してください。") def check_pii(records_json_path, csv_out_path, forbidden): problems = [] for path in (records_json_path, csv_out_path): with open(path, encoding="utf-8") as f: text = f.read() if "@" in text: problems.append(f"{os.path.basename(path)}: '@' を含む") for bad in forbidden: if bad and bad in text: problems.append(f"{os.path.basename(path)}: 個人名らしき文字列『{bad}』を含む") return problems if __name__ == "__main__": main()