Added the creation of jobmaterials script and the creation script ofr live updates to the ServiceM8 server (default is -dry-run). Altered inspector.py to now let us view jobmaterials DB.

servicem8-quote-template-parser.py

@@ -1,5 +1,8 @@
 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
 
@@ -10,6 +13,18 @@ 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:
@@ -18,6 +33,81 @@ def clean_text(value: Any) -> str:
     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)
@@ -57,6 +147,7 @@ def parse_quote_template_field_rows(field_rows: List[Dict[str, Any]]) -> Dict[st
     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"))
@@ -92,10 +183,23 @@ def parse_quote_template_field_rows(field_rows: List[Dict[str, Any]]) -> Dict[st
             )
             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
 
-    if include_items:
+    combined_include_items = include_items + extra_include_items
+
+    if combined_include_items:
         desired_job_materials.append(
             build_job_material_line(
                 kind="include_header",
@@ -107,7 +211,7 @@ def parse_quote_template_field_rows(field_rows: List[Dict[str, Any]]) -> Dict[st
             )
         )
         next_sort += 10
-        for item in include_items:
+        for item in combined_include_items:
             desired_job_materials.append(
                 build_job_material_line(
                     kind="include_item",
@@ -150,6 +254,7 @@ def parse_quote_template_field_rows(field_rows: List[Dict[str, Any]]) -> Dict[st
     return {
         "description": description,
         "include_items": include_items,
+        "extra_include_items": extra_include_items,
         "exclude_items": exclude_items,
         "desired_job_materials": desired_job_materials,
     }
@@ -181,10 +286,15 @@ 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))