307 lines
10 KiB
Python
307 lines
10 KiB
Python
import argparse
|
|
import json
|
|
import sqlite3
|
|
from contextlib import closing
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
QUOTE_TEMPLATE_FORM_UUID = "3621b6be-1d19-4756-9ab4-9d5e4120f6d9"
|
|
QUOTE_INCLUDE_HEADER_FIELD_UUID = "b752046f-2543-40a8-82bf-241f035a6c3d"
|
|
QUOTE_INCLUDE_HEADER_MATERIAL_UUID = "1924893b-917f-474a-adaa-2093bd622d4b"
|
|
QUOTE_INCLUDE_ITEM_MATERIAL_UUID = "8c00ca29-2178-403e-be76-241cfaddeedb"
|
|
QUOTE_EXCLUDE_HEADER_FIELD_UUID = "5e9aeda9-2c59-43db-ba64-241f0b7812bd"
|
|
QUOTE_EXCLUDE_HEADER_MATERIAL_UUID = "4947bfd7-4875-48f7-9caf-2093b9751b9b"
|
|
QUOTE_EXCLUDE_ITEM_MATERIAL_UUID = "8c00ca29-2178-403e-be76-241cfaddeedb"
|
|
STATE_DB_PATH = Path(__file__).with_name("servicem8_quote_materials_state.db")
|
|
EXTRA_INCLUDE_FIELDS = [
|
|
"Number of trades needed",
|
|
"Labour Hours Required",
|
|
"Materials",
|
|
"Excavation",
|
|
"Number of hours required",
|
|
"Scaffolding?",
|
|
"Scaffolding requirements",
|
|
"Equipment Hire?",
|
|
"Equipment required",
|
|
]
|
|
|
|
|
|
def clean_text(value: Any) -> str:
|
|
if value is None:
|
|
return ""
|
|
return str(value).replace("\r\n", "\n").replace("\r", "\n").strip()
|
|
|
|
|
|
def get_state_conn(db_path: Path = STATE_DB_PATH):
|
|
conn = sqlite3.connect(db_path)
|
|
conn.row_factory = sqlite3.Row
|
|
return conn
|
|
|
|
|
|
def init_state_db(db_path: Path = STATE_DB_PATH):
|
|
with closing(get_state_conn(db_path)) as conn:
|
|
conn.execute(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS generated_job_materials (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
job_uuid TEXT NOT NULL,
|
|
form_response_uuid TEXT,
|
|
job_material_uuid TEXT,
|
|
kind TEXT NOT NULL,
|
|
source_field_uuid TEXT,
|
|
source_question TEXT,
|
|
source_text TEXT,
|
|
created_at TEXT NOT NULL,
|
|
updated_at TEXT NOT NULL
|
|
)
|
|
"""
|
|
)
|
|
conn.execute(
|
|
"CREATE INDEX IF NOT EXISTS idx_generated_job_materials_job_uuid ON generated_job_materials(job_uuid)"
|
|
)
|
|
conn.execute(
|
|
"CREATE INDEX IF NOT EXISTS idx_generated_job_materials_form_response_uuid ON generated_job_materials(form_response_uuid)"
|
|
)
|
|
conn.commit()
|
|
|
|
|
|
def record_generated_job_material(
|
|
*,
|
|
job_uuid: str,
|
|
form_response_uuid: str,
|
|
job_material_uuid: str,
|
|
kind: str,
|
|
source_field_uuid: str,
|
|
source_question: str,
|
|
source_text: str,
|
|
db_path: Path = STATE_DB_PATH,
|
|
):
|
|
now = datetime.now(timezone.utc).isoformat()
|
|
with closing(get_state_conn(db_path)) as conn:
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO generated_job_materials (
|
|
job_uuid,
|
|
form_response_uuid,
|
|
job_material_uuid,
|
|
kind,
|
|
source_field_uuid,
|
|
source_question,
|
|
source_text,
|
|
created_at,
|
|
updated_at
|
|
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
""",
|
|
(
|
|
job_uuid,
|
|
form_response_uuid,
|
|
job_material_uuid,
|
|
kind,
|
|
source_field_uuid,
|
|
source_question,
|
|
source_text,
|
|
now,
|
|
now,
|
|
),
|
|
)
|
|
conn.commit()
|
|
|
|
|
|
def parse_field_data(field_data: Any) -> List[Dict[str, Any]]:
|
|
if isinstance(field_data, str):
|
|
return json.loads(field_data)
|
|
if isinstance(field_data, list):
|
|
return field_data
|
|
raise ValueError("field_data must be a JSON string or list")
|
|
|
|
|
|
def sort_key(row: Dict[str, Any]):
|
|
return int(row.get("SortOrder", 9999) or 9999)
|
|
|
|
|
|
def build_job_material_line(
|
|
kind: str,
|
|
name: str,
|
|
material_uuid: str,
|
|
sort_order: int,
|
|
source_question: Optional[str] = None,
|
|
source_field_uuid: Optional[str] = None,
|
|
) -> Dict[str, Any]:
|
|
return {
|
|
"kind": kind,
|
|
"name": clean_text(name),
|
|
"material_uuid": material_uuid,
|
|
"quantity": "1",
|
|
"price": "0",
|
|
"displayed_amount": "0",
|
|
"displayed_amount_is_tax_inclusive": "0",
|
|
"sort_order": str(sort_order),
|
|
"source_question": source_question or "",
|
|
"source_field_uuid": source_field_uuid or "",
|
|
}
|
|
|
|
|
|
def parse_quote_template_field_rows(field_rows: List[Dict[str, Any]]) -> Dict[str, Any]:
|
|
ordered = sorted(field_rows, key=sort_key)
|
|
description = ""
|
|
include_items: List[Dict[str, Any]] = []
|
|
exclude_items: List[Dict[str, Any]] = []
|
|
extra_include_items: List[Dict[str, Any]] = []
|
|
|
|
for row in ordered:
|
|
question = clean_text(row.get("Question"))
|
|
response = clean_text(row.get("Response"))
|
|
field_uuid = clean_text(row.get("UUID"))
|
|
|
|
if not question:
|
|
continue
|
|
|
|
if question == "Description of Works to be Quoted":
|
|
description = response
|
|
continue
|
|
|
|
if question.startswith("Item ") and response:
|
|
include_items.append(
|
|
{
|
|
"question": question,
|
|
"response": response,
|
|
"field_uuid": field_uuid,
|
|
"sort_order": int(row.get("SortOrder", 0) or 0),
|
|
}
|
|
)
|
|
continue
|
|
|
|
if question.startswith("Works excluded ") and response:
|
|
exclude_items.append(
|
|
{
|
|
"question": question,
|
|
"response": response,
|
|
"field_uuid": field_uuid,
|
|
"sort_order": int(row.get("SortOrder", 0) or 0),
|
|
}
|
|
)
|
|
continue
|
|
|
|
if question in EXTRA_INCLUDE_FIELDS and response:
|
|
extra_include_items.append(
|
|
{
|
|
"question": question,
|
|
"response": f"{question}: {response}",
|
|
"field_uuid": field_uuid,
|
|
"sort_order": int(row.get("SortOrder", 0) or 0),
|
|
}
|
|
)
|
|
continue
|
|
|
|
desired_job_materials: List[Dict[str, Any]] = []
|
|
next_sort = 100
|
|
|
|
combined_include_items = include_items + extra_include_items
|
|
|
|
if combined_include_items:
|
|
desired_job_materials.append(
|
|
build_job_material_line(
|
|
kind="include_header",
|
|
name="QUOTE INCLUDES",
|
|
material_uuid=QUOTE_INCLUDE_HEADER_MATERIAL_UUID,
|
|
sort_order=next_sort,
|
|
source_question="QUOTE INCLUDES",
|
|
source_field_uuid=QUOTE_INCLUDE_HEADER_FIELD_UUID,
|
|
)
|
|
)
|
|
next_sort += 10
|
|
for item in combined_include_items:
|
|
desired_job_materials.append(
|
|
build_job_material_line(
|
|
kind="include_item",
|
|
name=item["response"],
|
|
material_uuid=QUOTE_INCLUDE_ITEM_MATERIAL_UUID,
|
|
sort_order=next_sort,
|
|
source_question=item["question"],
|
|
source_field_uuid=item["field_uuid"],
|
|
)
|
|
)
|
|
next_sort += 10
|
|
|
|
if exclude_items:
|
|
if desired_job_materials:
|
|
next_sort = max(next_sort, 200)
|
|
desired_job_materials.append(
|
|
build_job_material_line(
|
|
kind="exclude_header",
|
|
name="QUOTE EXCLUDES",
|
|
material_uuid=QUOTE_EXCLUDE_HEADER_MATERIAL_UUID,
|
|
sort_order=next_sort,
|
|
source_question="QUOTE EXCLUDES",
|
|
source_field_uuid=QUOTE_EXCLUDE_HEADER_FIELD_UUID,
|
|
)
|
|
)
|
|
next_sort += 10
|
|
for item in exclude_items:
|
|
desired_job_materials.append(
|
|
build_job_material_line(
|
|
kind="exclude_item",
|
|
name=item["response"],
|
|
material_uuid=QUOTE_EXCLUDE_ITEM_MATERIAL_UUID,
|
|
sort_order=next_sort,
|
|
source_question=item["question"],
|
|
source_field_uuid=item["field_uuid"],
|
|
)
|
|
)
|
|
next_sort += 10
|
|
|
|
return {
|
|
"description": description,
|
|
"include_items": include_items,
|
|
"extra_include_items": extra_include_items,
|
|
"exclude_items": exclude_items,
|
|
"desired_job_materials": desired_job_materials,
|
|
}
|
|
|
|
|
|
def parse_quote_template_form_response(payload: Dict[str, Any]) -> Dict[str, Any]:
|
|
data = payload.get("data", {}) if isinstance(payload, dict) else {}
|
|
form_uuid = clean_text(data.get("form_uuid"))
|
|
if form_uuid and form_uuid != QUOTE_TEMPLATE_FORM_UUID:
|
|
raise ValueError(f"Not a Quote Template form response: {form_uuid}")
|
|
|
|
field_rows = parse_field_data(data.get("field_data", []))
|
|
parsed = parse_quote_template_field_rows(field_rows)
|
|
parsed["form_uuid"] = form_uuid
|
|
parsed["form_response_uuid"] = clean_text(data.get("uuid"))
|
|
parsed["job_uuid"] = clean_text(data.get("regarding_object_uuid"))
|
|
return parsed
|
|
|
|
|
|
def load_input_file(path: str) -> Dict[str, Any]:
|
|
raw = Path(path).read_text()
|
|
payload = json.loads(raw)
|
|
if isinstance(payload, dict) and "field_data" in payload and "data" not in payload:
|
|
return {"data": payload}
|
|
return payload
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description="Parse ServiceM8 Quote Template form responses into desired Job Material rows")
|
|
parser.add_argument("input", help="Path to JSON file containing full form response payload or a data object with field_data")
|
|
parser.add_argument("--pretty", action="store_true", help="Pretty-print output JSON")
|
|
parser.add_argument("--init-db", action="store_true", help="Initialize the local generated-job-materials state DB")
|
|
args = parser.parse_args()
|
|
|
|
if args.init_db:
|
|
init_state_db()
|
|
|
|
payload = load_input_file(args.input)
|
|
parsed = parse_quote_template_form_response(payload)
|
|
parsed["state_db_path"] = str(STATE_DB_PATH)
|
|
|
|
if args.pretty:
|
|
print(json.dumps(parsed, indent=2, ensure_ascii=False))
|
|
else:
|
|
print(json.dumps(parsed, ensure_ascii=False))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|