diff --git a/servicem8-quote-template-parser.py b/servicem8-quote-template-parser.py deleted file mode 100644 index 928016e..0000000 --- a/servicem8-quote-template-parser.py +++ /dev/null @@ -1,306 +0,0 @@ -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()