445 lines
16 KiB
Python
445 lines
16 KiB
Python
#!/usr/bin/env python3
|
|
"""Prepare a deterministic, read-only legacy import batch from the latest workbook.
|
|
|
|
This script never connects to PostgreSQL. It keeps source Use/Balance and raw
|
|
room-type text, filters rows without Confirmation, and writes auditable JSON
|
|
artifacts for the later database importer.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import re
|
|
import sys
|
|
from collections import Counter, defaultdict
|
|
from datetime import date, datetime
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from openpyxl import load_workbook
|
|
|
|
|
|
PROJECT_ROOT = Path(__file__).resolve().parents[2]
|
|
DEFAULT_WORKBOOK = Path(
|
|
"/Users/chillishark/Desktop/Condo公寓/Confirmation Report 2026 สิทธิ์ 15 วันคอนโดเท่านั้น(2).xlsx"
|
|
)
|
|
DEFAULT_OUTPUT = PROJECT_ROOT / ".planning" / "data_import_audit" / "import_batch_2026_07_31"
|
|
MASTER_SHEET = "Total2024-2026"
|
|
ROOM_TYPES = {
|
|
"RM1",
|
|
"RM2",
|
|
"RM3",
|
|
"RM4",
|
|
"UG1",
|
|
"UG2",
|
|
"SU1",
|
|
"SU2",
|
|
"SU6",
|
|
"SU3",
|
|
"AC2",
|
|
}
|
|
|
|
|
|
def scalar(value: Any) -> Any:
|
|
if isinstance(value, datetime):
|
|
return value.date().isoformat()
|
|
if isinstance(value, date):
|
|
return value.isoformat()
|
|
if isinstance(value, float) and value.is_integer():
|
|
return int(value)
|
|
return value
|
|
|
|
|
|
def text(value: Any) -> str:
|
|
value = scalar(value)
|
|
return "" if value is None else str(value).strip()
|
|
|
|
|
|
def number(value: Any) -> int | None:
|
|
value = scalar(value)
|
|
if isinstance(value, int):
|
|
return value
|
|
if isinstance(value, str) and re.fullmatch(r"-?\d+", value.strip()):
|
|
return int(value.strip())
|
|
return None
|
|
|
|
|
|
def parsed_date(value: Any) -> date | None:
|
|
value = scalar(value)
|
|
if isinstance(value, date):
|
|
return value
|
|
if isinstance(value, str):
|
|
for fmt in ("%Y-%m-%d", "%d-%b-%y", "%d-%b-%Y"):
|
|
try:
|
|
return datetime.strptime(value.strip(), fmt).date()
|
|
except ValueError:
|
|
pass
|
|
return None
|
|
|
|
|
|
def normalize_header(value: Any) -> str:
|
|
return re.sub(r"[^a-z0-9]+", "", text(value).lower())
|
|
|
|
|
|
def normalize_room_type(value: Any) -> str:
|
|
return re.sub(r"\r?\n", "<br>", text(value))
|
|
|
|
|
|
def room_numbers(value: Any) -> list[str]:
|
|
return list(dict.fromkeys(re.findall(r"\b\d{4}\b", text(value))))
|
|
|
|
|
|
def canonical_room_type(raw: str) -> str | None:
|
|
"""Only exact standard codes are canonicalized; legacy text is not guessed."""
|
|
return raw if raw in ROOM_TYPES else None
|
|
|
|
|
|
def write_json(path: Path, value: Any) -> None:
|
|
path.write_text(
|
|
json.dumps(value, ensure_ascii=False, indent=2, sort_keys=True) + "\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
|
|
def prepare(workbook_path: Path, output_dir: Path) -> dict[str, Any]:
|
|
workbook_bytes = workbook_path.read_bytes()
|
|
source_sha256 = hashlib.sha256(workbook_bytes).hexdigest()
|
|
workbook = load_workbook(workbook_path, data_only=True, read_only=False)
|
|
if MASTER_SHEET not in workbook.sheetnames:
|
|
raise ValueError(f"Missing master sheet: {MASTER_SHEET}")
|
|
|
|
master = workbook[MASTER_SHEET]
|
|
header_map: dict[str, int] = {}
|
|
for column in range(1, master.max_column + 1):
|
|
header = normalize_header(master.cell(2, column).value)
|
|
if header:
|
|
header_map.setdefault(header, column)
|
|
|
|
def column(header: str, fallback: int) -> int:
|
|
return header_map.get(normalize_header(header), fallback)
|
|
|
|
columns = {
|
|
"no": column("No", 1),
|
|
"transfer_date": column("Transfer Date", 2),
|
|
"name": column("Name", 3),
|
|
"room_no": column("Room No.", 4),
|
|
"room_type": column("Room Type", 5),
|
|
"unit_no": column("Unit No.", 6),
|
|
"member_no": column("Member No.", 7),
|
|
"remaining": column("Remaining stay privileges", 8),
|
|
}
|
|
|
|
owners: list[dict[str, Any]] = []
|
|
owner_by_room: dict[str, dict[str, Any]] = {}
|
|
owner_issues: list[dict[str, Any]] = []
|
|
for row in range(4, master.max_row + 1):
|
|
room_no = text(master.cell(row, columns["room_no"]).value)
|
|
if not re.fullmatch(r"\d{4}", room_no):
|
|
continue
|
|
owner = {
|
|
"account_no": number(master.cell(row, columns["no"]).value),
|
|
"transfer_date": scalar(master.cell(row, columns["transfer_date"]).value),
|
|
"name": text(master.cell(row, columns["name"]).value),
|
|
"room_no": room_no,
|
|
"purchased_room_type_code": text(master.cell(row, columns["room_type"]).value),
|
|
"unit_no": text(master.cell(row, columns["unit_no"]).value),
|
|
"member_no": text(master.cell(row, columns["member_no"]).value),
|
|
"reported_remaining": number(master.cell(row, columns["remaining"]).value),
|
|
"source_sheet": MASTER_SHEET,
|
|
"source_row": row,
|
|
}
|
|
missing = [
|
|
key
|
|
for key in (
|
|
"account_no",
|
|
"transfer_date",
|
|
"name",
|
|
"room_no",
|
|
"purchased_room_type_code",
|
|
"unit_no",
|
|
"member_no",
|
|
"reported_remaining",
|
|
)
|
|
if owner[key] in (None, "")
|
|
]
|
|
if missing:
|
|
owner_issues.append({"source_row": row, "room_no": room_no, "missing": missing})
|
|
if owner["purchased_room_type_code"] not in ROOM_TYPES:
|
|
owner_issues.append(
|
|
{
|
|
"source_row": row,
|
|
"room_no": room_no,
|
|
"invalid_purchased_room_type": owner["purchased_room_type_code"],
|
|
}
|
|
)
|
|
if owner["reported_remaining"] is not None and not 0 <= owner["reported_remaining"] <= 15:
|
|
owner_issues.append(
|
|
{
|
|
"source_row": row,
|
|
"room_no": room_no,
|
|
"remaining_out_of_range": owner["reported_remaining"],
|
|
}
|
|
)
|
|
owners.append(owner)
|
|
owner_by_room[room_no] = owner
|
|
|
|
usage: list[dict[str, Any]] = []
|
|
rejected: list[dict[str, Any]] = []
|
|
sheet_diagnostics: list[dict[str, Any]] = []
|
|
arithmetic_issues: list[dict[str, Any]] = []
|
|
chain_issues: list[dict[str, Any]] = []
|
|
required_issues: list[dict[str, Any]] = []
|
|
date_issues: list[dict[str, Any]] = []
|
|
identity_issues: list[dict[str, Any]] = []
|
|
previous_by_sheet: dict[str, dict[str, Any]] = {}
|
|
duplicate_confirmation_locations: dict[str, list[dict[str, Any]]] = defaultdict(list)
|
|
|
|
expected_headers = [
|
|
"No.",
|
|
"Confirmation No.",
|
|
"Check -IN",
|
|
"Check-OUT",
|
|
"Night",
|
|
"Room",
|
|
"Total",
|
|
"Use",
|
|
"Balance",
|
|
"Room Type",
|
|
"Remark",
|
|
]
|
|
|
|
for title in workbook.sheetnames:
|
|
if title == MASTER_SHEET:
|
|
continue
|
|
sheet = workbook[title]
|
|
title_rooms = room_numbers(title)
|
|
if not title_rooms:
|
|
sheet_diagnostics.append({"sheet": title, "kind": "template", "usage_count": 0})
|
|
continue
|
|
header_values = [text(sheet.cell(3, column).value).rstrip(" ") for column in range(1, 12)]
|
|
normalized_headers = [header.replace("Check-OUT ", "Check-OUT").strip() for header in header_values]
|
|
header_issue = normalized_headers != expected_headers
|
|
header_rooms = room_numbers(sheet.cell(2, 4).value)
|
|
if set(title_rooms) != set(header_rooms):
|
|
identity_issues.append(
|
|
{
|
|
"sheet": title,
|
|
"title_rooms": title_rooms,
|
|
"header_rooms": header_rooms,
|
|
"resolution": "sheet title is authoritative",
|
|
}
|
|
)
|
|
accepted_count = 0
|
|
sheet_rows: list[dict[str, Any]] = []
|
|
for row in range(4, sheet.max_row + 1):
|
|
raw = {
|
|
"no": number(sheet.cell(row, 1).value),
|
|
"confirmation_no": text(sheet.cell(row, 2).value),
|
|
"check_in": scalar(sheet.cell(row, 3).value),
|
|
"check_out": scalar(sheet.cell(row, 4).value),
|
|
"night": number(sheet.cell(row, 5).value),
|
|
"room": number(sheet.cell(row, 6).value),
|
|
"total": number(sheet.cell(row, 7).value),
|
|
"use": number(sheet.cell(row, 8).value),
|
|
"balance": number(sheet.cell(row, 9).value),
|
|
"raw_used_room_type": normalize_room_type(sheet.cell(row, 10).value),
|
|
"remark": text(sheet.cell(row, 11).value),
|
|
}
|
|
if not raw["confirmation_no"]:
|
|
populated_columns = [
|
|
column
|
|
for column in (3, 4, 6, 10, 11)
|
|
if text(sheet.cell(row, column).value)
|
|
]
|
|
if populated_columns:
|
|
rejected.append(
|
|
{
|
|
"reason_code": "missing_confirmation",
|
|
"reason": "Confirmation No. is empty; clear/exclude this row",
|
|
"source_sheet": title,
|
|
"source_row": row,
|
|
"populated_columns": populated_columns,
|
|
**raw,
|
|
}
|
|
)
|
|
continue
|
|
|
|
missing = [
|
|
key
|
|
for key in (
|
|
"confirmation_no",
|
|
"check_in",
|
|
"check_out",
|
|
"night",
|
|
"room",
|
|
"total",
|
|
"use",
|
|
"balance",
|
|
"raw_used_room_type",
|
|
"remark",
|
|
)
|
|
if raw[key] in (None, "")
|
|
]
|
|
if missing:
|
|
required_issues.append({"source_sheet": title, "source_row": row, "missing": missing})
|
|
|
|
check_in = parsed_date(raw["check_in"])
|
|
check_out = parsed_date(raw["check_out"])
|
|
if (
|
|
check_in is None
|
|
or check_out is None
|
|
or check_out <= check_in
|
|
or raw["night"] != (check_out - check_in).days
|
|
):
|
|
date_issues.append({"source_sheet": title, "source_row": row, **raw})
|
|
if raw["total"] is None or raw["use"] is None or raw["balance"] is None:
|
|
arithmetic_issues.append({"source_sheet": title, "source_row": row, **raw})
|
|
elif raw["total"] - raw["use"] != raw["balance"]:
|
|
arithmetic_issues.append({"source_sheet": title, "source_row": row, **raw})
|
|
previous = previous_by_sheet.get(title)
|
|
if previous is not None and raw["total"] != previous["balance"]:
|
|
chain_issues.append(
|
|
{
|
|
"source_sheet": title,
|
|
"source_row": row,
|
|
"expected_total": previous["balance"],
|
|
"actual_total": raw["total"],
|
|
}
|
|
)
|
|
previous_by_sheet[title] = raw
|
|
period_year = check_in.year if check_in else None
|
|
record = {
|
|
"owner_room_no": title_rooms[0] if len(title_rooms) == 1 else None,
|
|
"confirmation_no": raw["confirmation_no"],
|
|
"check_in": raw["check_in"],
|
|
"check_out": raw["check_out"],
|
|
"night": raw["night"],
|
|
"room": raw["room"],
|
|
"total": raw["total"],
|
|
"use": raw["use"],
|
|
"balance": raw["balance"],
|
|
"raw_used_room_type": raw["raw_used_room_type"],
|
|
"canonical_used_room_type_code": canonical_room_type(raw["raw_used_room_type"]),
|
|
"remark": raw["remark"],
|
|
"source_sheet": title,
|
|
"source_row": row,
|
|
"source_sequence": raw["no"],
|
|
"period_year": period_year,
|
|
"applied_multiplier": None,
|
|
"rule_version": "legacy-source",
|
|
}
|
|
usage.append(record)
|
|
sheet_rows.append(record)
|
|
accepted_count += 1
|
|
duplicate_confirmation_locations[raw["confirmation_no"]].append(
|
|
{"source_sheet": title, "source_row": row}
|
|
)
|
|
sheet_diagnostics.append(
|
|
{
|
|
"sheet": title,
|
|
"kind": "room_usage",
|
|
"title_rooms": title_rooms,
|
|
"header_rooms": header_rooms,
|
|
"header_issue": header_issue,
|
|
"usage_count": accepted_count,
|
|
"rejected_count": sum(
|
|
1 for row in rejected if row["source_sheet"] == title
|
|
),
|
|
"final_balance": sheet_rows[-1]["balance"] if sheet_rows else None,
|
|
}
|
|
)
|
|
|
|
duplicate_confirmations = {
|
|
confirmation: locations
|
|
for confirmation, locations in duplicate_confirmation_locations.items()
|
|
if len(locations) > 1
|
|
}
|
|
master_remaining_sum = sum(owner["reported_remaining"] or 0 for owner in owners)
|
|
usage_sum = sum(record["use"] or 0 for record in usage)
|
|
empty_usage_sheets = [
|
|
item["sheet"]
|
|
for item in sheet_diagnostics
|
|
if item["kind"] == "room_usage" and item["usage_count"] == 0
|
|
]
|
|
noncanonical = [
|
|
record
|
|
for record in usage
|
|
if record["canonical_used_room_type_code"] is None
|
|
]
|
|
manifest = {
|
|
"mode": "local-import-preflight",
|
|
"source_workbook": str(workbook_path),
|
|
"source_sha256": source_sha256,
|
|
"master_sheet": MASTER_SHEET,
|
|
"rules": {
|
|
"latest_workbook_only": True,
|
|
"missing_confirmation": "clear_or_exclude",
|
|
"usage_owner": "sheet_title",
|
|
"owner_metadata": "master_sheet",
|
|
"history_values": "preserve_source_use_balance_total_room",
|
|
"legacy_applied_multiplier": None,
|
|
"legacy_rule_version": "legacy-source",
|
|
"new_usage_room_type": "standard_code_only",
|
|
"database_write_performed": False,
|
|
},
|
|
"counts": {
|
|
"owner_count": len(owners),
|
|
"unique_owner_room_count": len(owner_by_room),
|
|
"accepted_usage_count": len(usage),
|
|
"rejected_row_count": len(rejected),
|
|
"rejected_missing_confirmation_count": sum(
|
|
item["reason_code"] == "missing_confirmation" for item in rejected
|
|
),
|
|
"usage_use_sum": usage_sum,
|
|
"master_remaining_sum": master_remaining_sum,
|
|
"implied_used": len(owners) * 15 - master_remaining_sum,
|
|
"canonical_used_room_type_count": len(usage) - len(noncanonical),
|
|
"noncanonical_used_room_type_count": len(noncanonical),
|
|
"duplicate_confirmation_group_count": len(duplicate_confirmations),
|
|
},
|
|
"checks": {
|
|
"owner_issues": owner_issues,
|
|
"required_usage_issues": required_issues,
|
|
"date_night_issues": date_issues,
|
|
"arithmetic_issues": arithmetic_issues,
|
|
"balance_chain_issues": chain_issues,
|
|
"room_gt_one_count": sum((record["room"] or 0) > 1 for record in usage),
|
|
"missing_owner_rooms": sorted(
|
|
{
|
|
record["owner_room_no"]
|
|
for record in usage
|
|
if record["owner_room_no"] not in owner_by_room
|
|
}
|
|
),
|
|
"empty_usage_sheets": empty_usage_sheets,
|
|
},
|
|
"duplicate_confirmations": duplicate_confirmations,
|
|
"year_counts": dict(sorted(Counter(record["period_year"] for record in usage).items())),
|
|
"raw_used_room_type_counts": dict(
|
|
sorted(Counter(record["raw_used_room_type"] for record in usage).items())
|
|
),
|
|
}
|
|
|
|
output_dir.mkdir(parents=True, exist_ok=True)
|
|
write_json(output_dir / "owners.json", owners)
|
|
write_json(output_dir / "usage_legacy.json", usage)
|
|
write_json(output_dir / "rejected_rows.json", rejected)
|
|
write_json(output_dir / "sheet_diagnostics.json", sheet_diagnostics)
|
|
write_json(output_dir / "identity_issues.json", identity_issues)
|
|
write_json(output_dir / "manifest.json", manifest)
|
|
return manifest
|
|
|
|
|
|
def main() -> None:
|
|
workbook_path = Path(sys.argv[1]).expanduser() if len(sys.argv) > 1 else DEFAULT_WORKBOOK
|
|
output_dir = Path(sys.argv[2]).expanduser() if len(sys.argv) > 2 else DEFAULT_OUTPUT
|
|
if not workbook_path.is_file():
|
|
raise SystemExit(f"Workbook not found: {workbook_path}")
|
|
manifest = prepare(workbook_path, output_dir)
|
|
print(json.dumps(manifest, ensure_ascii=False, indent=2, sort_keys=True))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|