File added for the new name
This commit is contained in:
@@ -0,0 +1,306 @@
|
||||
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()
|
||||
Reference in New Issue
Block a user