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 = "" author_name = "" 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 == "Name": author_name = 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, "author_name": author_name, "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()