Soren_Molty/plumbing
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

File rename for the servicem8_quote_template_parser to use underscores so that the import library works

Browse Source
This commit is contained in:
Soren (Molty)
2026-04-29 13:13:34 +10:00
parent 84b07613ee
commit 1a7f663d39
1 changed files with 0 additions and 306 deletions
Download Patch File Download Diff File Expand all files Collapse all files
servicem8-quote-template-parser.py
-306
View File
@@ -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()