New FastAPI python page for easily human reading of the json objects we are capturing. Addition of a template parser that can be used to interpret and send/push jobMaterials to the server. Adding the docs dir just so that it is all together.
This commit is contained in:
@@ -0,0 +1,196 @@
|
||||
import argparse
|
||||
import json
|
||||
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"
|
||||
|
||||
|
||||
def clean_text(value: Any) -> str:
|
||||
if value is None:
|
||||
return ""
|
||||
return str(value).replace("\r\n", "\n").replace("\r", "\n").strip()
|
||||
|
||||
|
||||
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]] = []
|
||||
|
||||
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
|
||||
|
||||
desired_job_materials: List[Dict[str, Any]] = []
|
||||
next_sort = 100
|
||||
|
||||
if 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 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,
|
||||
"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")
|
||||
args = parser.parse_args()
|
||||
|
||||
payload = load_input_file(args.input)
|
||||
parsed = parse_quote_template_form_response(payload)
|
||||
|
||||
if args.pretty:
|
||||
print(json.dumps(parsed, indent=2, ensure_ascii=False))
|
||||
else:
|
||||
print(json.dumps(parsed, ensure_ascii=False))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user