feat: add Condo owner desk frontend and backend
This commit is contained in:
Wyndham ARR
444
backend/scripts/prepare-confirmation-import.py
Normal file
444
backend/scripts/prepare-confirmation-import.py
Normal file
@@ -0,0 +1,444 @@
|
||||
#!/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()
|
||||
Reference in New Issue
Block a user