adam/proposal-system
1
0
Fork 0
mirror of https://github.com/Sea-Haven-Industries/proposal-system.git synced 2026-09-30 06:33:13 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Apply ruff formatting to all Lambda Python files

Browse source
This commit is contained in:
Adam Moussa 2026-05-17 13:04:06 -04:00
parent 456f0d4252
commit 539878e84e
5 changed files with 390 additions and 243 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

10
lambdas/library-ingest/app.py
View file

@ -132,10 +132,12 @@ def format_proposal_document(proposal: dict, line_items: list[dict]) -> str:
lines.append(f"| {i} | {desc} | {qty} | {unit} | {up_str} | {tp_str} |")
lines.extend([
"",
f"**Total: ${total:,.2f}**",
])
lines.extend(
[
"",
f"**Total: ${total:,.2f}**",
]
)
return "\n".join(lines)

134
lambdas/library-ingest/batch_ingest.py
View file

@ -23,7 +23,14 @@ import pdfplumber
try:
from rich.console import Console
from rich.progress import Progress, SpinnerColumn, TextColumn, BarColumn, TaskProgressColumn
from rich.progress import (
Progress,
SpinnerColumn,
TextColumn,
BarColumn,
TaskProgressColumn,
)
console = Console()
HAS_RICH = True
except ImportError:
@ -39,12 +46,20 @@ SERVICE_CATEGORIES = ["HVAC", "Plumbing", "Electrical", "General", "Renovation"]
def main():
parser = argparse.ArgumentParser(description="Batch ingest historical proposals into Bedrock KB")
parser.add_argument("--input-dir", required=True, help="Directory containing historical proposal PDFs")
parser = argparse.ArgumentParser(
description="Batch ingest historical proposals into Bedrock KB"
)
parser.add_argument(
"--input-dir",
required=True,
help="Directory containing historical proposal PDFs",
)
parser.add_argument("--bucket", required=True, help="S3 library bucket name")
parser.add_argument("--kb-id", required=True, help="Bedrock Knowledge Base ID")
parser.add_argument("--ds-id", required=True, help="Bedrock KB Data Source ID")
parser.add_argument("--dry-run", action="store_true", help="Parse and extract only, don't upload")
parser.add_argument(
"--dry-run", action="store_true", help="Parse and extract only, don't upload"
)
args = parser.parse_args()
input_dir = Path(args.input_dir)
@ -73,7 +88,11 @@ def main():
for pdf_path in pdf_files:
status = process_pdf(pdf_path, args.bucket, args.dry_run)
results[status] += 1
progress.update(task, advance=1, description=f"[{'green' if status == 'success' else 'red'}]{pdf_path.name}")
progress.update(
task,
advance=1,
description=f"[{'green' if status == 'success' else 'red'}]{pdf_path.name}",
)
else:
for i, pdf_path in enumerate(pdf_files, 1):
print(f"[{i}/{len(pdf_files)}] Processing {pdf_path.name}...", end=" ")
@ -81,12 +100,16 @@ def main():
results[status] += 1
print(f"[{status.upper()}]")
print(f"\nResults: {results['success']} succeeded, {results['failed']} failed, {results['skipped']} skipped")
print(
f"\nResults: {results['success']} succeeded, {results['failed']} failed, {results['skipped']} skipped"
)
if not args.dry_run and results["success"] > 0:
print("\nTriggering Knowledge Base sync...")
trigger_kb_sync(args.kb_id, args.ds_id)
print("Done. KB ingestion job started — check AWS console for completion status.")
print(
"Done. KB ingestion job started — check AWS console for completion status."
)
def process_pdf(pdf_path: Path, bucket: str, dry_run: bool) -> str:
@ -163,23 +186,25 @@ def extract_with_claude(pdf_path: Path) -> dict:
modelId=MODEL_ID,
contentType="application/json",
accept="application/json",
body=json.dumps({
"anthropic_version": "bedrock-2023-05-31",
"max_tokens": 4096,
"messages": [{
"role": "user",
"content": [
body=json.dumps(
{
"anthropic_version": "bedrock-2023-05-31",
"max_tokens": 4096,
"messages": [
{
"type": "document",
"source": {
"type": "base64",
"media_type": "application/pdf",
"data": pdf_b64,
},
},
{
"type": "text",
"text": f"""Extract structured data from this proposal PDF.
"role": "user",
"content": [
{
"type": "document",
"source": {
"type": "base64",
"media_type": "application/pdf",
"data": pdf_b64,
},
},
{
"type": "text",
"text": f"""Extract structured data from this proposal PDF.
Return a JSON object with:
- customerName: the customer/client name
- serviceCategory: one of {SERVICE_CATEGORIES}
@ -190,11 +215,13 @@ Return a JSON object with:
- totalAmount: the grand total
Respond ONLY with the JSON object.""",
},
},
],
}
],
}],
"temperature": 0.1,
}),
"temperature": 0.1,
}
),
)
response_body = json.loads(response["body"].read())
@ -219,10 +246,16 @@ def parse_tables(tables: list) -> list[dict]:
header = [str(cell).lower().strip() if cell else "" for cell in table[0]]
desc_col = _find_column(header, ["description", "item", "service", "work", "scope"])
desc_col = _find_column(
header, ["description", "item", "service", "work", "scope"]
)
qty_col = _find_column(header, ["qty", "quantity", "count"])
price_col = _find_column(header, ["unit price", "rate", "price/unit", "unit cost"])
total_col = _find_column(header, ["total", "amount", "ext", "extended", "line total"])
price_col = _find_column(
header, ["unit price", "rate", "price/unit", "unit cost"]
)
total_col = _find_column(
header, ["total", "amount", "ext", "extended", "line total"]
)
if desc_col is None:
continue
@ -232,24 +265,43 @@ def parse_tables(tables: list) -> list[dict]:
continue
description = str(row[desc_col]).strip() if row[desc_col] else ""
if not description or description.lower() in ("", "total", "subtotal", "grand total"):
if not description or description.lower() in (
"",
"total",
"subtotal",
"grand total",
):
continue
quantity = _parse_number(row[qty_col]) if qty_col is not None and qty_col < len(row) else None
unit_price = _parse_number(row[price_col]) if price_col is not None and price_col < len(row) else None
total = _parse_number(row[total_col]) if total_col is not None and total_col < len(row) else None
quantity = (
_parse_number(row[qty_col])
if qty_col is not None and qty_col < len(row)
else None
)
unit_price = (
_parse_number(row[price_col])
if price_col is not None and price_col < len(row)
else None
)
total = (
_parse_number(row[total_col])
if total_col is not None and total_col < len(row)
else None
)
if total is None and quantity and unit_price:
total = quantity * unit_price
if description and (total or unit_price):
line_items.append({
"description": description,
"quantity": quantity or 1,
"unit": "each",
"unitPrice": unit_price,
"totalPrice": total or 0,
})
line_items.append(
{
"description": description,
"quantity": quantity or 1,
"unit": "each",
"unitPrice": unit_price,
"totalPrice": total or 0,
}
)
return line_items

102
lambdas/pdf-extract/app.py
View file

@ -138,10 +138,16 @@ def parse_tables(tables: list) -> list[dict]:
header = [str(cell).lower().strip() if cell else "" for cell in table[0]]
desc_col = find_column(header, ["description", "item", "service", "work", "scope"])
desc_col = find_column(
header, ["description", "item", "service", "work", "scope"]
)
qty_col = find_column(header, ["qty", "quantity", "count"])
price_col = find_column(header, ["unit price", "rate", "price/unit", "unit cost"])
total_col = find_column(header, ["total", "amount", "ext", "extended", "line total"])
price_col = find_column(
header, ["unit price", "rate", "price/unit", "unit cost"]
)
total_col = find_column(
header, ["total", "amount", "ext", "extended", "line total"]
)
if desc_col is None:
continue
@ -151,23 +157,42 @@ def parse_tables(tables: list) -> list[dict]:
continue
description = str(row[desc_col]).strip() if row[desc_col] else ""
if not description or description.lower() in ("", "total", "subtotal", "grand total"):
if not description or description.lower() in (
"",
"total",
"subtotal",
"grand total",
):
continue
quantity = parse_number(row[qty_col]) if qty_col is not None and qty_col < len(row) else None
unit_price = parse_number(row[price_col]) if price_col is not None and price_col < len(row) else None
total = parse_number(row[total_col]) if total_col is not None and total_col < len(row) else None
quantity = (
parse_number(row[qty_col])
if qty_col is not None and qty_col < len(row)
else None
)
unit_price = (
parse_number(row[price_col])
if price_col is not None and price_col < len(row)
else None
)
total = (
parse_number(row[total_col])
if total_col is not None and total_col < len(row)
else None
)
if total is None and quantity and unit_price:
total = quantity * unit_price
if description and (total or unit_price):
line_items.append({
"description": description,
"quantity": quantity,
"unitPrice": unit_price,
"total": total,
})
line_items.append(
{
"description": description,
"quantity": quantity,
"unitPrice": unit_price,
"total": total,
}
)
return line_items
@ -203,34 +228,38 @@ def extract_with_claude_multimodal(pdf_path: str) -> dict:
modelId=MODEL_ID,
contentType="application/json",
accept="application/json",
body=json.dumps({
"anthropic_version": "bedrock-2023-05-31",
"max_tokens": 4096,
"messages": [{
"role": "user",
"content": [
body=json.dumps(
{
"anthropic_version": "bedrock-2023-05-31",
"max_tokens": 4096,
"messages": [
{
"type": "document",
"source": {
"type": "base64",
"media_type": "application/pdf",
"data": pdf_b64,
},
},
{
"type": "text",
"text": """Extract all line items from this vendor proposal PDF.
"role": "user",
"content": [
{
"type": "document",
"source": {
"type": "base64",
"media_type": "application/pdf",
"data": pdf_b64,
},
},
{
"type": "text",
"text": """Extract all line items from this vendor proposal PDF.
Return a JSON object with these fields:
- vendorName: the vendor/company name
- lineItems: array of objects with: description, quantity (number or null), unitPrice (number or null), total (number or null)
- totalVendorCost: the grand total amount
Respond ONLY with the JSON object, no additional text.""",
},
},
],
}
],
}],
"temperature": 0.1,
}),
"temperature": 0.1,
}
),
)
response_body = json.loads(response["body"].read())
@ -251,7 +280,12 @@ Respond ONLY with the JSON object, no additional text.""",
except Exception as e:
print(f"Claude multimodal extraction failed: {e}")
return {"vendorName": "", "lineItems": [], "rawText": "", "totalVendorCost": 0.0}
return {
"vendorName": "",
"lineItems": [],
"rawText": "",
"totalVendorCost": 0.0,
}
def save_extraction(vendor_proposal_id: str, extracted: dict):

295
lambdas/pdf-generate/app.py
View file

@ -160,93 +160,113 @@ def build_pdf(proposal: dict, line_items: list[dict]) -> bytes:
def _get_styles():
styles = getSampleStyleSheet()
styles.add(ParagraphStyle(
"CompanyName",
parent=styles["Heading1"],
fontSize=18,
leading=22,
textColor=colors.HexColor("#1a237e"),
spaceAfter=2,
))
styles.add(
ParagraphStyle(
"CompanyName",
parent=styles["Heading1"],
fontSize=18,
leading=22,
textColor=colors.HexColor("#1a237e"),
spaceAfter=2,
)
)
styles.add(ParagraphStyle(
"CompanyInfo",
parent=styles["Normal"],
fontSize=9,
leading=12,
textColor=colors.HexColor("#555555"),
))
styles.add(
ParagraphStyle(
"CompanyInfo",
parent=styles["Normal"],
fontSize=9,
leading=12,
textColor=colors.HexColor("#555555"),
)
)
styles.add(ParagraphStyle(
"ProposalTitle",
parent=styles["Heading2"],
fontSize=14,
leading=18,
textColor=colors.HexColor("#1a237e"),
spaceBefore=6,
spaceAfter=12,
))
styles.add(
ParagraphStyle(
"ProposalTitle",
parent=styles["Heading2"],
fontSize=14,
leading=18,
textColor=colors.HexColor("#1a237e"),
spaceBefore=6,
spaceAfter=12,
)
)
styles.add(ParagraphStyle(
"SectionHeader",
parent=styles["Heading3"],
fontSize=11,
leading=14,
textColor=colors.HexColor("#1a237e"),
spaceBefore=8,
spaceAfter=6,
borderWidth=0,
))
styles.add(
ParagraphStyle(
"SectionHeader",
parent=styles["Heading3"],
fontSize=11,
leading=14,
textColor=colors.HexColor("#1a237e"),
spaceBefore=8,
spaceAfter=6,
borderWidth=0,
)
)
styles.add(ParagraphStyle(
"MetaLabel",
parent=styles["Normal"],
fontSize=9,
leading=12,
textColor=colors.HexColor("#666666"),
))
styles.add(
ParagraphStyle(
"MetaLabel",
parent=styles["Normal"],
fontSize=9,
leading=12,
textColor=colors.HexColor("#666666"),
)
)
styles.add(ParagraphStyle(
"MetaValue",
parent=styles["Normal"],
fontSize=10,
leading=13,
fontName="Helvetica-Bold",
))
styles.add(
ParagraphStyle(
"MetaValue",
parent=styles["Normal"],
fontSize=10,
leading=13,
fontName="Helvetica-Bold",
)
)
styles.add(ParagraphStyle(
"ScopeText",
parent=styles["Normal"],
fontSize=10,
leading=14,
spaceBefore=4,
))
styles.add(
ParagraphStyle(
"ScopeText",
parent=styles["Normal"],
fontSize=10,
leading=14,
spaceBefore=4,
)
)
styles.add(ParagraphStyle(
"TermsText",
parent=styles["Normal"],
fontSize=8,
leading=11,
textColor=colors.HexColor("#555555"),
))
styles.add(
ParagraphStyle(
"TermsText",
parent=styles["Normal"],
fontSize=8,
leading=11,
textColor=colors.HexColor("#555555"),
)
)
styles.add(ParagraphStyle(
"TotalLabel",
parent=styles["Normal"],
fontSize=11,
leading=14,
fontName="Helvetica-Bold",
alignment=TA_RIGHT,
))
styles.add(
ParagraphStyle(
"TotalLabel",
parent=styles["Normal"],
fontSize=11,
leading=14,
fontName="Helvetica-Bold",
alignment=TA_RIGHT,
)
)
styles.add(ParagraphStyle(
"FooterText",
parent=styles["Normal"],
fontSize=8,
leading=10,
textColor=colors.HexColor("#888888"),
alignment=TA_CENTER,
))
styles.add(
ParagraphStyle(
"FooterText",
parent=styles["Normal"],
fontSize=8,
leading=10,
textColor=colors.HexColor("#888888"),
alignment=TA_CENTER,
)
)
return styles
@ -267,12 +287,16 @@ def _build_header(proposal: dict, styles) -> list:
]
header_table = Table(header_data, colWidths=[3.5 * inch, 3.5 * inch])
header_table.setStyle(TableStyle([
("VALIGN", (0, 0), (-1, -1), "TOP"),
("ALIGN", (1, 0), (1, -1), "RIGHT"),
("LINEBELOW", (0, -1), (-1, -1), 1.5, colors.HexColor("#1a237e")),
("BOTTOMPADDING", (0, -1), (-1, -1), 8),
]))
header_table.setStyle(
TableStyle(
[
("VALIGN", (0, 0), (-1, -1), "TOP"),
("ALIGN", (1, 0), (1, -1), "RIGHT"),
("LINEBELOW", (0, -1), (-1, -1), 1.5, colors.HexColor("#1a237e")),
("BOTTOMPADDING", (0, -1), (-1, -1), 8),
]
)
)
return [header_table]
@ -322,11 +346,15 @@ def _build_metadata(proposal: dict, styles) -> list:
]
meta_table = Table(meta_data, colWidths=[3.5 * inch, 3.5 * inch])
meta_table.setStyle(TableStyle([
("VALIGN", (0, 0), (-1, -1), "TOP"),
("TOPPADDING", (0, 0), (-1, -1), 2),
("BOTTOMPADDING", (0, 0), (-1, -1), 2),
]))
meta_table.setStyle(
TableStyle(
[
("VALIGN", (0, 0), (-1, -1), "TOP"),
("TOPPADDING", (0, 0), (-1, -1), 2),
("BOTTOMPADDING", (0, 0), (-1, -1), 2),
]
)
)
return [meta_table]
@ -381,36 +409,49 @@ def _build_line_items_table(line_items: list[dict], styles) -> list:
]
table_data.append(row)
col_widths = [0.35 * inch, 3.15 * inch, 0.55 * inch, 0.7 * inch, 1.0 * inch, 1.0 * inch]
col_widths = [
0.35 * inch,
3.15 * inch,
0.55 * inch,
0.7 * inch,
1.0 * inch,
1.0 * inch,
]
table = Table(table_data, colWidths=col_widths, repeatRows=1)
table.setStyle(TableStyle([
# Header row
("BACKGROUND", (0, 0), (-1, 0), colors.HexColor("#1a237e")),
("TEXTCOLOR", (0, 0), (-1, 0), colors.white),
("FONTNAME", (0, 0), (-1, 0), "Helvetica-Bold"),
("FONTSIZE", (0, 0), (-1, 0), 9),
("BOTTOMPADDING", (0, 0), (-1, 0), 6),
("TOPPADDING", (0, 0), (-1, 0), 6),
# Data rows
("FONTSIZE", (0, 1), (-1, -1), 9),
("TOPPADDING", (0, 1), (-1, -1), 4),
("BOTTOMPADDING", (0, 1), (-1, -1), 4),
("VALIGN", (0, 0), (-1, -1), "MIDDLE"),
# Alignment
("ALIGN", (0, 0), (0, -1), "CENTER"),
("ALIGN", (2, 0), (2, -1), "CENTER"),
("ALIGN", (3, 0), (3, -1), "CENTER"),
("ALIGN", (4, 0), (4, -1), "RIGHT"),
("ALIGN", (5, 0), (5, -1), "RIGHT"),
# Grid
("LINEBELOW", (0, 0), (-1, 0), 1, colors.HexColor("#1a237e")),
("LINEBELOW", (0, 1), (-1, -2), 0.5, colors.HexColor("#e0e0e0")),
("LINEBELOW", (0, -1), (-1, -1), 1, colors.HexColor("#1a237e")),
# Alternating row colors
*[("BACKGROUND", (0, i), (-1, i), colors.HexColor("#f5f5f5"))
for i in range(2, len(table_data), 2)],
]))
table.setStyle(
TableStyle(
[
# Header row
("BACKGROUND", (0, 0), (-1, 0), colors.HexColor("#1a237e")),
("TEXTCOLOR", (0, 0), (-1, 0), colors.white),
("FONTNAME", (0, 0), (-1, 0), "Helvetica-Bold"),
("FONTSIZE", (0, 0), (-1, 0), 9),
("BOTTOMPADDING", (0, 0), (-1, 0), 6),
("TOPPADDING", (0, 0), (-1, 0), 6),
# Data rows
("FONTSIZE", (0, 1), (-1, -1), 9),
("TOPPADDING", (0, 1), (-1, -1), 4),
("BOTTOMPADDING", (0, 1), (-1, -1), 4),
("VALIGN", (0, 0), (-1, -1), "MIDDLE"),
# Alignment
("ALIGN", (0, 0), (0, -1), "CENTER"),
("ALIGN", (2, 0), (2, -1), "CENTER"),
("ALIGN", (3, 0), (3, -1), "CENTER"),
("ALIGN", (4, 0), (4, -1), "RIGHT"),
("ALIGN", (5, 0), (5, -1), "RIGHT"),
# Grid
("LINEBELOW", (0, 0), (-1, 0), 1, colors.HexColor("#1a237e")),
("LINEBELOW", (0, 1), (-1, -2), 0.5, colors.HexColor("#e0e0e0")),
("LINEBELOW", (0, -1), (-1, -1), 1, colors.HexColor("#1a237e")),
# Alternating row colors
*[
("BACKGROUND", (0, i), (-1, i), colors.HexColor("#f5f5f5"))
for i in range(2, len(table_data), 2)
],
]
)
)
elements.append(table)
elements.append(Spacer(1, 0.15 * inch))
@ -420,14 +461,18 @@ def _build_line_items_table(line_items: list[dict], styles) -> list:
["", "", "", "", "TOTAL:", f"${subtotal:,.2f}"],
]
total_table = Table(total_data, colWidths=col_widths)
total_table.setStyle(TableStyle([
("FONTNAME", (0, 0), (-1, -1), "Helvetica-Bold"),
("FONTSIZE", (0, 0), (-1, -1), 11),
("ALIGN", (4, 0), (4, 0), "RIGHT"),
("ALIGN", (5, 0), (5, 0), "RIGHT"),
("TOPPADDING", (0, 0), (-1, -1), 4),
("LINEABOVE", (4, 0), (5, 0), 1.5, colors.HexColor("#1a237e")),
]))
total_table.setStyle(
TableStyle(
[
("FONTNAME", (0, 0), (-1, -1), "Helvetica-Bold"),
("FONTSIZE", (0, 0), (-1, -1), 11),
("ALIGN", (4, 0), (4, 0), "RIGHT"),
("ALIGN", (5, 0), (5, 0), "RIGHT"),
("TOPPADDING", (0, 0), (-1, -1), 4),
("LINEABOVE", (4, 0), (5, 0), 1.5, colors.HexColor("#1a237e")),
]
)
)
elements.append(total_table)
return elements

92
lambdas/suggestions/app.py
View file

@ -100,9 +100,11 @@ def retrieve_similar(scope: str, category: str) -> list[dict]:
return []
try:
filter_config = {
"equals": {"key": "service_category", "value": category}
} if category else None
filter_config = (
{"equals": {"key": "service_category", "value": category}}
if category
else None
)
params = {
"knowledgeBaseId": KNOWLEDGE_BASE_ID,
@ -115,7 +117,9 @@ def retrieve_similar(scope: str, category: str) -> list[dict]:
}
if filter_config:
params["retrievalConfiguration"]["vectorSearchConfiguration"]["filter"] = filter_config
params["retrievalConfiguration"]["vectorSearchConfiguration"]["filter"] = (
filter_config
)
response = bedrock_agent.retrieve(**params)
@ -126,12 +130,14 @@ def retrieve_similar(scope: str, category: str) -> list[dict]:
metadata = result.get("metadata", {})
source_uri = result.get("location", {}).get("s3Location", {}).get("uri", "")
results.append({
"content": content,
"score": score,
"metadata": metadata,
"sourceUri": source_uri,
})
results.append(
{
"content": content,
"score": score,
"metadata": metadata,
"sourceUri": source_uri,
}
)
return results
@ -150,7 +156,9 @@ def generate_line_items(
if similar_proposals:
context_block = "Here are similar historical proposals and their line items for reference:\n\n"
for i, sp in enumerate(similar_proposals[:5], 1):
context_block += f"--- Similar Proposal {i} (relevance: {sp['score']:.2f}) ---\n"
context_block += (
f"--- Similar Proposal {i} (relevance: {sp['score']:.2f}) ---\n"
)
context_block += sp["content"] + "\n\n"
prompt = f"""You are a construction/facilities proposal estimator for Sea Haven Industries.
@ -180,12 +188,14 @@ Respond ONLY with the JSON array, no additional text."""
modelId=MODEL_ID,
contentType="application/json",
accept="application/json",
body=json.dumps({
"anthropic_version": "bedrock-2023-05-31",
"max_tokens": 4096,
"messages": [{"role": "user", "content": prompt}],
"temperature": 0.3,
}),
body=json.dumps(
{
"anthropic_version": "bedrock-2023-05-31",
"max_tokens": 4096,
"messages": [{"role": "user", "content": prompt}],
"temperature": 0.3,
}
),
)
response_body = json.loads(response["body"].read())
@ -213,17 +223,19 @@ def post_line_items(proposal_id: str, items: list[dict], existing_items: list[di
# Preserve non-AI items (Manual, Vendor, Historical)
preserved = [li for li in existing_items if li.get("source") != "AI"]
for i, li in enumerate(preserved):
line_items_payload.append({
"id": li.get("id"),
"description": li["description"],
"quantity": float(li.get("quantity", 1)),
"unit": li.get("unit", "each"),
"unitPrice": li.get("unitPrice"),
"totalPrice": float(li.get("totalPrice", 0)),
"pricingMode": li.get("pricingMode", "TotalPrice"),
"sortOrder": i + 1,
"source": li.get("source", "Manual"),
})
line_items_payload.append(
{
"id": li.get("id"),
"description": li["description"],
"quantity": float(li.get("quantity", 1)),
"unit": li.get("unit", "each"),
"unitPrice": li.get("unitPrice"),
"totalPrice": float(li.get("totalPrice", 0)),
"pricingMode": li.get("pricingMode", "TotalPrice"),
"sortOrder": i + 1,
"source": li.get("source", "Manual"),
}
)
# Add new AI-generated items after preserved ones
offset = len(line_items_payload)
@ -232,17 +244,19 @@ def post_line_items(proposal_id: str, items: list[dict], existing_items: list[di
if pricing_mode not in ("UnitPrice", "TotalPrice", "Both"):
pricing_mode = "UnitPrice" if item.get("unitPrice") else "TotalPrice"
line_items_payload.append({
"id": None,
"description": item["description"],
"quantity": float(item.get("quantity", 1)),
"unit": item.get("unit", "each"),
"unitPrice": item.get("unitPrice"),
"totalPrice": float(item.get("totalPrice", 0)),
"pricingMode": pricing_mode,
"sortOrder": offset + i + 1,
"source": "AI",
})
line_items_payload.append(
{
"id": None,
"description": item["description"],
"quantity": float(item.get("quantity", 1)),
"unit": item.get("unit", "each"),
"unitPrice": item.get("unitPrice"),
"totalPrice": float(item.get("totalPrice", 0)),
"pricingMode": pricing_mode,
"sortOrder": offset + i + 1,
"source": "AI",
}
)
try:
resp = httpx.put(