mirror of
https://github.com/arkorty/B.Tech-Project-III.git
synced 2026-04-19 12:41:48 +00:00
514 lines
21 KiB
Python
514 lines
21 KiB
Python
"""
|
||
pdf_generator.py — negoT8 Deal Agreement PDF
|
||
|
||
Generates a printable/shareable PDF for resolved freelance or marketplace deals.
|
||
Uses fpdf2 (pure-Python, zero system deps).
|
||
|
||
Usage:
|
||
from tools.pdf_generator import generate_deal_pdf
|
||
|
||
pdf_path = await generate_deal_pdf(
|
||
negotiation_id = "aa271ee7",
|
||
feature_type = "freelance", # "freelance" | "marketplace"
|
||
final_proposal = {...}, # final_proposal dict from resolution
|
||
user_a = {"id": 123, "name": "Alice", "username": "alice"},
|
||
user_b = {"id": 456, "name": "Bob", "username": "bob"},
|
||
rounds_taken = 4,
|
||
sat_a = 82.0,
|
||
sat_b = 78.0,
|
||
blockchain_proof = {...} | None,
|
||
)
|
||
# Returns an absolute path to /tmp/negot8_deal_<neg_id>.pdf
|
||
# Caller is responsible for deleting the file after sending.
|
||
"""
|
||
|
||
import asyncio
|
||
import os
|
||
import textwrap
|
||
from datetime import datetime
|
||
from typing import Optional
|
||
|
||
# ── fpdf2 ────────────────────────────────────────────────────────────────────
|
||
try:
|
||
from fpdf import FPDF
|
||
except ImportError as e:
|
||
raise ImportError(
|
||
"fpdf2 is required for PDF generation.\n"
|
||
"Install it with: pip install fpdf2"
|
||
) from e
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# Helpers
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
_FEATURE_LABELS = {
|
||
"freelance": "Freelance Project Agreement",
|
||
"marketplace": "Buy / Sell Deal Agreement",
|
||
}
|
||
|
||
_SECTION_FILL = (230, 240, 255) # light blue
|
||
_HEADER_FILL = (30, 60, 120) # dark navy
|
||
_LINE_COLOR = (180, 180, 200)
|
||
_TEXT_DARK = (20, 20, 40)
|
||
_TEXT_MUTED = (90, 90, 110)
|
||
_GREEN = (20, 130, 60)
|
||
_RED = (180, 30, 30)
|
||
|
||
|
||
def _safe(val) -> str:
|
||
"""Convert any value to a clean Latin-1 safe string (fpdf2 default encoding)."""
|
||
if val is None:
|
||
return "—"
|
||
s = str(val).strip()
|
||
# Replace common Unicode dashes / bullets that Latin-1 can't handle
|
||
replacements = {
|
||
"\u2013": "-", "\u2014": "-", "\u2022": "*",
|
||
"\u20b9": "Rs.", "\u2192": "->", "\u2714": "[x]",
|
||
"\u2713": "[x]", "\u00d7": "x",
|
||
}
|
||
for ch, rep in replacements.items():
|
||
s = s.replace(ch, rep)
|
||
return s.encode("latin-1", errors="replace").decode("latin-1")
|
||
|
||
|
||
def _wrap(text: str, width: int = 90) -> list[str]:
|
||
"""Wrap long text into lines of at most `width` characters."""
|
||
if not text:
|
||
return ["—"]
|
||
return textwrap.wrap(_safe(text), width) or [_safe(text)]
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# PDF builder
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
class DealPDF(FPDF):
|
||
"""Custom FPDF subclass with header/footer branding."""
|
||
|
||
def header(self):
|
||
# Navy banner
|
||
self.set_fill_color(*_HEADER_FILL)
|
||
self.rect(0, 0, 210, 22, "F")
|
||
|
||
self.set_font("Helvetica", "B", 15)
|
||
self.set_text_color(255, 255, 255)
|
||
self.set_xy(10, 4)
|
||
self.cell(0, 8, "negoT8", ln=False)
|
||
|
||
self.set_font("Helvetica", "", 9)
|
||
self.set_text_color(200, 210, 240)
|
||
self.set_xy(10, 13)
|
||
self.cell(0, 5, "AI-Negotiated Deal Agreement | Blockchain-Verified", ln=True)
|
||
|
||
self.set_draw_color(*_LINE_COLOR)
|
||
self.set_line_width(0.3)
|
||
self.ln(4)
|
||
|
||
def footer(self):
|
||
self.set_y(-14)
|
||
self.set_font("Helvetica", "I", 8)
|
||
self.set_text_color(*_TEXT_MUTED)
|
||
self.cell(
|
||
0, 5,
|
||
f"negoT8 | Generated {datetime.utcnow().strftime('%d %b %Y %H:%M')} UTC "
|
||
f"| Page {self.page_no()}",
|
||
align="C",
|
||
)
|
||
|
||
# ── Section title ─────────────────────────────────────────────────────────
|
||
def section_title(self, title: str):
|
||
self.ln(3)
|
||
self.set_fill_color(*_SECTION_FILL)
|
||
self.set_text_color(*_TEXT_DARK)
|
||
self.set_font("Helvetica", "B", 10)
|
||
self.set_draw_color(*_LINE_COLOR)
|
||
self.set_line_width(0.3)
|
||
self.cell(0, 7, f" {_safe(title)}", border="B", ln=True, fill=True)
|
||
self.ln(1)
|
||
|
||
# ── Key-value row ─────────────────────────────────────────────────────────
|
||
def kv_row(self, key: str, value: str, bold_value: bool = False):
|
||
self.set_font("Helvetica", "B", 9)
|
||
self.set_text_color(*_TEXT_MUTED)
|
||
self.cell(52, 6, _safe(key), ln=False)
|
||
|
||
self.set_font("Helvetica", "B" if bold_value else "", 9)
|
||
self.set_text_color(*_TEXT_DARK)
|
||
|
||
# Multi-line safe: wrap long values
|
||
lines = _wrap(value, 70)
|
||
self.cell(0, 6, lines[0], ln=True)
|
||
for extra in lines[1:]:
|
||
self.cell(52, 5, "", ln=False)
|
||
self.set_font("Helvetica", "", 9)
|
||
self.cell(0, 5, extra, ln=True)
|
||
|
||
# ── Bullet item ───────────────────────────────────────────────────────────
|
||
def bullet(self, text: str):
|
||
self.set_font("Helvetica", "", 9)
|
||
self.set_text_color(*_TEXT_DARK)
|
||
lines = _wrap(text, 85)
|
||
first = True
|
||
for line in lines:
|
||
prefix = " * " if first else " "
|
||
self.cell(0, 5, f"{prefix}{line}", ln=True)
|
||
first = False
|
||
|
||
# ── Thin horizontal rule ──────────────────────────────────────────────────
|
||
def hr(self):
|
||
self.set_draw_color(*_LINE_COLOR)
|
||
self.set_line_width(0.2)
|
||
self.line(10, self.get_y(), 200, self.get_y())
|
||
self.ln(2)
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# Term extractors (feature-specific)
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
def _extract_freelance_terms(final_proposal: dict, preferences_a: dict, preferences_b: dict) -> list[tuple]:
|
||
"""Return a list of (label, value) tuples for agreed freelance terms."""
|
||
details = final_proposal.get("details", {})
|
||
terms = []
|
||
|
||
budget = (
|
||
details.get("budget") or details.get("agreed_budget")
|
||
or details.get("price") or details.get("total_amount")
|
||
or details.get("agreed_price")
|
||
)
|
||
if budget:
|
||
terms.append(("Agreed Budget", f"Rs. {budget}"))
|
||
|
||
timeline = details.get("timeline") or details.get("duration") or details.get("deadline")
|
||
if timeline:
|
||
terms.append(("Timeline / Deadline", str(timeline)))
|
||
|
||
scope = details.get("scope") or details.get("deliverables") or []
|
||
if isinstance(scope, list) and scope:
|
||
terms.append(("Scope / Deliverables", " | ".join(str(s) for s in scope[:6])))
|
||
elif isinstance(scope, str) and scope:
|
||
terms.append(("Scope / Deliverables", scope))
|
||
|
||
payment = (
|
||
details.get("payment_schedule") or details.get("payments")
|
||
or details.get("payment_terms")
|
||
)
|
||
if payment:
|
||
terms.append(("Payment Schedule", str(payment)))
|
||
|
||
upfront = details.get("upfront") or details.get("milestone_1") or details.get("advance")
|
||
if upfront:
|
||
terms.append(("Upfront / First Milestone", f"Rs. {upfront}"))
|
||
|
||
ip = details.get("ip_ownership") or details.get("intellectual_property")
|
||
if ip:
|
||
terms.append(("IP Ownership", str(ip)))
|
||
|
||
# Fall back to raw preferences if details are sparse
|
||
if not terms:
|
||
raw_a = preferences_a.get("raw_details", {})
|
||
raw_b = preferences_b.get("raw_details", {})
|
||
for label, keys in [
|
||
("Project / Skill", ["skill", "expertise", "project_type", "tech_stack"]),
|
||
("Rate", ["rate", "hourly_rate"]),
|
||
("Hours", ["hours", "estimated_hours"]),
|
||
("Client Budget", ["budget", "max_budget"]),
|
||
]:
|
||
val = next((raw_a.get(k) or raw_b.get(k) for k in keys if raw_a.get(k) or raw_b.get(k)), None)
|
||
if val:
|
||
terms.append((label, str(val)))
|
||
|
||
# Summary as final catch-all
|
||
summary = final_proposal.get("summary", "")
|
||
if summary and summary != "Agreement reached":
|
||
terms.append(("Summary", summary))
|
||
|
||
return terms
|
||
|
||
|
||
def _extract_marketplace_terms(final_proposal: dict, preferences_a: dict, preferences_b: dict) -> list[tuple]:
|
||
"""Return a list of (label, value) tuples for agreed buy/sell terms."""
|
||
details = final_proposal.get("details", {})
|
||
raw_a = preferences_a.get("raw_details", {})
|
||
raw_b = preferences_b.get("raw_details", {})
|
||
terms = []
|
||
|
||
item = raw_a.get("item") or raw_b.get("item") or details.get("item") or "Item"
|
||
terms.append(("Item", str(item)))
|
||
|
||
price = (
|
||
details.get("agreed_price") or details.get("price")
|
||
or details.get("final_price") or details.get("amount")
|
||
)
|
||
if price:
|
||
terms.append(("Agreed Price", f"Rs. {price}"))
|
||
|
||
delivery = details.get("delivery") or details.get("handover") or details.get("pickup")
|
||
if delivery:
|
||
terms.append(("Delivery / Handover", str(delivery)))
|
||
|
||
condition = details.get("condition") or raw_a.get("condition") or raw_b.get("condition")
|
||
if condition:
|
||
terms.append(("Item Condition", str(condition)))
|
||
|
||
market_ref = details.get("market_price") or details.get("market_reference")
|
||
if market_ref:
|
||
terms.append(("Market Reference Price", f"Rs. {market_ref}"))
|
||
|
||
summary = final_proposal.get("summary", "")
|
||
if summary and summary != "Agreement reached":
|
||
terms.append(("Summary", summary))
|
||
|
||
return terms
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# Public API
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
async def generate_deal_pdf(
|
||
negotiation_id: str,
|
||
feature_type: str,
|
||
final_proposal: dict,
|
||
user_a: dict,
|
||
user_b: dict,
|
||
rounds_taken: int,
|
||
sat_a: float,
|
||
sat_b: float,
|
||
preferences_a: Optional[dict] = None,
|
||
preferences_b: Optional[dict] = None,
|
||
blockchain_proof: Optional[dict] = None,
|
||
) -> str:
|
||
"""
|
||
Build a Deal Agreement PDF and save it to /tmp.
|
||
|
||
Parameters
|
||
----------
|
||
negotiation_id : short negotiation ID (e.g. "aa271ee7")
|
||
feature_type : "freelance" or "marketplace"
|
||
final_proposal : the final_proposal dict from the resolution payload
|
||
user_a / user_b : dicts with keys: id, name, username
|
||
rounds_taken : number of negotiation rounds
|
||
sat_a / sat_b : final satisfaction scores (0–100)
|
||
preferences_a/b : raw preference dicts (used for term extraction fallbacks)
|
||
blockchain_proof: optional dict from register_agreement_on_chain
|
||
|
||
Returns
|
||
-------
|
||
Absolute path to the generated PDF file.
|
||
"""
|
||
# Run the synchronous PDF build in a thread executor so we don't block the event loop
|
||
loop = asyncio.get_event_loop()
|
||
path = await loop.run_in_executor(
|
||
None,
|
||
_build_pdf,
|
||
negotiation_id, feature_type, final_proposal,
|
||
user_a, user_b, rounds_taken, sat_a, sat_b,
|
||
preferences_a or {}, preferences_b or {},
|
||
blockchain_proof,
|
||
)
|
||
return path
|
||
|
||
|
||
def _build_pdf(
|
||
negotiation_id: str,
|
||
feature_type: str,
|
||
final_proposal: dict,
|
||
user_a: dict,
|
||
user_b: dict,
|
||
rounds_taken: int,
|
||
sat_a: float,
|
||
sat_b: float,
|
||
preferences_a: dict,
|
||
preferences_b: dict,
|
||
blockchain_proof: Optional[dict],
|
||
) -> str:
|
||
"""Synchronous PDF build — called in a thread via run_in_executor."""
|
||
|
||
doc_label = _FEATURE_LABELS.get(feature_type, "Deal Agreement")
|
||
date_str = datetime.utcnow().strftime("%d %B %Y")
|
||
neg_short = negotiation_id[:8].upper()
|
||
|
||
pdf = DealPDF(orientation="P", unit="mm", format="A4")
|
||
pdf.set_margins(10, 28, 10)
|
||
pdf.set_auto_page_break(auto=True, margin=18)
|
||
pdf.add_page()
|
||
|
||
# ── Title block ───────────────────────────────────────────────────────────
|
||
pdf.set_font("Helvetica", "B", 17)
|
||
pdf.set_text_color(*_TEXT_DARK)
|
||
pdf.cell(0, 10, _safe(doc_label), ln=True, align="C")
|
||
|
||
pdf.set_font("Helvetica", "", 9)
|
||
pdf.set_text_color(*_TEXT_MUTED)
|
||
pdf.cell(0, 5, "Generated by negoT8 AI Agents | Hackathon Edition", ln=True, align="C")
|
||
pdf.ln(2)
|
||
pdf.hr()
|
||
|
||
# ── Agreement meta ────────────────────────────────────────────────────────
|
||
pdf.section_title("Agreement Details")
|
||
pdf.kv_row("Agreement ID", neg_short)
|
||
pdf.kv_row("Document Type", doc_label)
|
||
pdf.kv_row("Date Issued", date_str)
|
||
pdf.kv_row("Status", "EXECUTED — Mutually Accepted", bold_value=True)
|
||
pdf.ln(2)
|
||
|
||
# ── Parties ───────────────────────────────────────────────────────────────
|
||
pdf.section_title("Contracting Parties")
|
||
|
||
def _party_name(u: dict) -> str:
|
||
name = u.get("name") or u.get("display_name") or ""
|
||
uname = u.get("username") or ""
|
||
uid = u.get("id") or u.get("telegram_id") or ""
|
||
parts = []
|
||
if name:
|
||
parts.append(name)
|
||
if uname:
|
||
parts.append(f"@{uname}")
|
||
if uid:
|
||
parts.append(f"(ID: {uid})")
|
||
return " ".join(parts) if parts else "Unknown"
|
||
|
||
pdf.kv_row("Party A", _party_name(user_a))
|
||
pdf.kv_row("Party B", _party_name(user_b))
|
||
pdf.ln(2)
|
||
|
||
# ── Agreed terms ──────────────────────────────────────────────────────────
|
||
pdf.section_title("Agreed Terms")
|
||
|
||
if feature_type == "freelance":
|
||
terms = _extract_freelance_terms(final_proposal, preferences_a, preferences_b)
|
||
elif feature_type == "marketplace":
|
||
terms = _extract_marketplace_terms(final_proposal, preferences_a, preferences_b)
|
||
else:
|
||
terms = []
|
||
summary = final_proposal.get("summary", "")
|
||
if summary:
|
||
terms.append(("Summary", summary))
|
||
for k, v in (final_proposal.get("details") or {}).items():
|
||
if v:
|
||
terms.append((k.replace("_", " ").title(), str(v)))
|
||
|
||
if terms:
|
||
for label, value in terms:
|
||
pdf.kv_row(label, value)
|
||
else:
|
||
pdf.set_font("Helvetica", "I", 9)
|
||
pdf.set_text_color(*_TEXT_MUTED)
|
||
pdf.cell(0, 6, " See negotiation summary for full details.", ln=True)
|
||
pdf.ln(2)
|
||
|
||
# ── Negotiation stats ─────────────────────────────────────────────────────
|
||
pdf.section_title("Negotiation Statistics")
|
||
pdf.kv_row("Rounds Taken", str(rounds_taken))
|
||
pdf.kv_row("Party A Satisfaction", f"{sat_a:.0f}%")
|
||
pdf.kv_row("Party B Satisfaction", f"{sat_b:.0f}%")
|
||
fairness = 100 - abs(sat_a - sat_b)
|
||
pdf.kv_row("Fairness Score", f"{fairness:.0f}%", bold_value=True)
|
||
pdf.ln(2)
|
||
|
||
# ── Blockchain proof ──────────────────────────────────────────────────────
|
||
pdf.section_title("Blockchain Proof of Agreement")
|
||
|
||
if blockchain_proof and blockchain_proof.get("tx_hash"):
|
||
tx = blockchain_proof.get("tx_hash", "")
|
||
blk = blockchain_proof.get("block_number", "")
|
||
ahash = blockchain_proof.get("agreement_hash", "")
|
||
url = blockchain_proof.get("explorer_url", "")
|
||
mock = blockchain_proof.get("mock", False)
|
||
|
||
pdf.kv_row("Network", "Polygon Amoy Testnet")
|
||
pdf.kv_row("TX Hash", tx[:42] + "..." if len(tx) > 42 else tx)
|
||
if blk:
|
||
pdf.kv_row("Block Number", str(blk))
|
||
if ahash:
|
||
pdf.kv_row("Agreement Hash", str(ahash)[:42] + "...")
|
||
if url and not mock:
|
||
pdf.kv_row("Explorer URL", url)
|
||
|
||
if mock:
|
||
pdf.set_font("Helvetica", "I", 8)
|
||
pdf.set_text_color(*_TEXT_MUTED)
|
||
pdf.cell(0, 5, " * Blockchain entry recorded (testnet / mock mode)", ln=True)
|
||
else:
|
||
pdf.set_font("Helvetica", "I", 8)
|
||
pdf.set_text_color(*_GREEN)
|
||
pdf.cell(0, 5, " * Permanently and immutably recorded on the Polygon blockchain.", ln=True)
|
||
else:
|
||
pdf.set_font("Helvetica", "I", 9)
|
||
pdf.set_text_color(*_TEXT_MUTED)
|
||
pdf.cell(0, 6, f" Negotiation ID: {_safe(negotiation_id)}", ln=True)
|
||
pdf.cell(0, 6, " Blockchain proof will be recorded on deal finalisation.", ln=True)
|
||
|
||
pdf.ln(2)
|
||
|
||
# ── Terms & Disclaimer ────────────────────────────────────────────────────
|
||
pdf.section_title("Terms & Disclaimer")
|
||
disclaimer_lines = [
|
||
"1. This document was generated automatically by negoT8 AI agents acting on",
|
||
" behalf of the parties named above.",
|
||
"2. Both parties accepted the agreed terms via the negoT8 Telegram Bot on",
|
||
f" {date_str}.",
|
||
"3. The blockchain hash above independently verifies that this agreement existed",
|
||
" at the recorded block height.",
|
||
"4. This document is provided for reference and record-keeping purposes.",
|
||
" For legally binding contracts, please consult a qualified legal professional.",
|
||
"5. negoT8 and its AI agents are not parties to this agreement and bear no",
|
||
" liability for non-performance by either party.",
|
||
]
|
||
pdf.set_font("Helvetica", "", 8)
|
||
pdf.set_text_color(*_TEXT_MUTED)
|
||
for line in disclaimer_lines:
|
||
pdf.cell(0, 4.5, _safe(line), ln=True)
|
||
|
||
pdf.ln(4)
|
||
|
||
# ── Signature placeholders ────────────────────────────────────────────────
|
||
pdf.hr()
|
||
pdf.set_font("Helvetica", "B", 9)
|
||
pdf.set_text_color(*_TEXT_DARK)
|
||
pdf.ln(2)
|
||
|
||
# Two columns: Party A left, Party B right
|
||
y_sig = pdf.get_y()
|
||
col_w = 88
|
||
|
||
pdf.set_xy(10, y_sig)
|
||
pdf.cell(col_w, 5, "Party A", ln=False)
|
||
pdf.set_x(112)
|
||
pdf.cell(col_w, 5, "Party B", ln=True)
|
||
|
||
# Name lines
|
||
pdf.set_font("Helvetica", "", 9)
|
||
pdf.set_text_color(*_TEXT_MUTED)
|
||
a_display = user_a.get("name") or user_a.get("display_name") or f"@{user_a.get('username','')}"
|
||
b_display = user_b.get("name") or user_b.get("display_name") or f"@{user_b.get('username','')}"
|
||
pdf.set_x(10)
|
||
pdf.cell(col_w, 5, _safe(a_display), ln=False)
|
||
pdf.set_x(112)
|
||
pdf.cell(col_w, 5, _safe(b_display), ln=True)
|
||
|
||
pdf.ln(6)
|
||
# Draw signature lines
|
||
sig_y = pdf.get_y()
|
||
pdf.set_draw_color(*_TEXT_DARK)
|
||
pdf.set_line_width(0.4)
|
||
pdf.line(10, sig_y, 98, sig_y) # Party A
|
||
pdf.line(112, sig_y, 200, sig_y) # Party B
|
||
|
||
pdf.ln(3)
|
||
pdf.set_font("Helvetica", "I", 8)
|
||
pdf.set_text_color(*_TEXT_MUTED)
|
||
pdf.set_x(10)
|
||
pdf.cell(col_w, 4, "Accepted via negoT8", ln=False)
|
||
pdf.set_x(112)
|
||
pdf.cell(col_w, 4, "Accepted via negoT8", ln=True)
|
||
|
||
# ── Save ──────────────────────────────────────────────────────────────────
|
||
out_path = os.path.join(
|
||
"/tmp", f"negot8_deal_{negotiation_id}.pdf"
|
||
)
|
||
pdf.output(out_path)
|
||
print(f"[PDF] Deal agreement saved → {out_path}")
|
||
return out_path
|