From cf395091a8b8358b88cc1bb9798d2d2a3c9efd8f Mon Sep 17 00:00:00 2001 From: "Soren (Molty)" Date: Wed, 29 Apr 2026 13:14:09 +1000 Subject: [PATCH] File added for the new name --- servicem8_quote_template_parser.py | 306 +++++++++++++++++++++++++++++ 1 file changed, 306 insertions(+) create mode 100644 servicem8_quote_template_parser.py diff --git a/servicem8_quote_template_parser.py b/servicem8_quote_template_parser.py new file mode 100644 index 0000000..928016e --- /dev/null +++ b/servicem8_quote_template_parser.py @@ -0,0 +1,306 @@ +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()