#!/usr/bin/env python3 """ Proposal Approval Workflow for Genesis/AgileAdapt ================================================= HITL (Human-In-The-Loop) workflow specifically designed for proposal management. Extends core.hitl_approval with proposal-specific features: - Auto-submit proposals to HITL queue - View, edit, approve, reject, or request revision - PDF generation on approval - Rejection logging for training data - Slack notifications for new submissions - Revision tracking with full history VERIFICATION_STAMP Story: STORY-3.1 Verified By: CLAUDE Verified At: 2026-01-24 Tests: tests/test_proposal_approval.py Coverage: Full - black box and white box tests """ import json import os import sys from dataclasses import asdict, dataclass, field from datetime import datetime from enum import Enum from io import BytesIO from pathlib import Path from typing import Any, Callable, Dict, List, Optional, Tuple, Union import hashlib import logging # Genesis system paths sys.path.insert(0, '/mnt/e/genesis-system') sys.path.insert(0, '/mnt/e/genesis-system/data/genesis-memory') # Database imports - MANDATORY: Use Elestio PostgreSQL (NO SQLite) try: from elestio_config import PostgresConfig import psycopg2 from psycopg2.extras import Json, RealDictCursor HAS_DB = True except ImportError: HAS_DB = False print("Warning: PostgreSQL not available, using file-based queue") # PDF generation try: from reportlab.lib import colors from reportlab.lib.pagesizes import A4 from reportlab.lib.styles import ParagraphStyle, getSampleStyleSheet from reportlab.lib.units import mm from reportlab.platypus import ( Paragraph, SimpleDocTemplate, Spacer, Table, TableStyle, PageBreak ) HAS_REPORTLAB = True except ImportError: HAS_REPORTLAB = False print("Warning: reportlab not available, PDF generation disabled") # Slack notifications try: import requests HAS_REQUESTS = True except ImportError: HAS_REQUESTS = False print("Warning: requests not available, Slack notifications disabled") # Import from HITL base from core.hitl_approval import ( ApprovalStatus, OutputType, ApprovalRequest, HITLApprovalQueue, get_hitl_queue, ) # Import from template engine from core.proposals.template_engine import ( ProposalTemplateEngine, ProposalType, PricingTier, BusinessContext, ProposalResult, ) # Configure logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) class RevisionStatus(Enum): """Status of a revision request.""" PENDING = "pending" ADDRESSED = "addressed" SUPERSEDED = "superseded" @dataclass class ProposalRevision: """A revision request/response for a proposal.""" revision_id: str request_id: str revision_number: int original_content: str revised_content: Optional[str] = None feedback: str = "" status: RevisionStatus = RevisionStatus.PENDING created_at: str = field(default_factory=lambda: datetime.now().isoformat()) addressed_at: Optional[str] = None def to_dict(self) -> Dict[str, Any]: """Convert to dictionary for storage.""" d = asdict(self) d['status'] = self.status.value return d @classmethod def from_dict(cls, d: Dict[str, Any]) -> 'ProposalRevision': """Create from dictionary.""" d['status'] = RevisionStatus(d['status']) return cls(**d) @dataclass class ProposalSubmission: """A proposal submitted for approval with full metadata.""" request_id: str proposal_type: ProposalType business_context: BusinessContext proposal_content: str monthly_price: int buildout_price: int total_first_month: int status: ApprovalStatus = ApprovalStatus.PENDING revisions: List[ProposalRevision] = field(default_factory=list) pdf_path: Optional[str] = None created_at: str = field(default_factory=lambda: datetime.now().isoformat()) approved_at: Optional[str] = None reviewer_notes: Optional[str] = None rejection_reason: Optional[str] = None def to_dict(self) -> Dict[str, Any]: """Convert to dictionary for storage.""" d = { 'request_id': self.request_id, 'proposal_type': self.proposal_type.value, 'business_context': asdict(self.business_context), 'proposal_content': self.proposal_content, 'monthly_price': self.monthly_price, 'buildout_price': self.buildout_price, 'total_first_month': self.total_first_month, 'status': self.status.value, 'revisions': [r.to_dict() for r in self.revisions], 'pdf_path': self.pdf_path, 'created_at': self.created_at, 'approved_at': self.approved_at, 'reviewer_notes': self.reviewer_notes, 'rejection_reason': self.rejection_reason, } return d @classmethod def from_dict(cls, d: Dict[str, Any]) -> 'ProposalSubmission': """Create from dictionary.""" d['proposal_type'] = ProposalType(d['proposal_type']) d['business_context'] = BusinessContext(**d['business_context']) d['status'] = ApprovalStatus(d['status']) d['revisions'] = [ProposalRevision.from_dict(r) for r in d.get('revisions', [])] return cls(**d) class ProposalApprovalWorkflow: """ Proposal-specific approval workflow extending HITL base. Key Features: - Auto-submit proposals to HITL queue - Edit capability before approval - PDF generation on approval - Revision tracking with history - Slack notifications - Rejection logging for training data """ PROPOSALS_DIR = Path("/mnt/e/genesis-system/data/proposals") PDF_OUTPUT_DIR = Path("/mnt/e/genesis-system/data/proposals/pdfs") REJECTION_LOG = Path("/mnt/e/genesis-system/data/proposals/rejections.jsonl") def __init__( self, slack_webhook_url: Optional[str] = None, auto_notify: bool = True ): """ Initialize proposal approval workflow. Args: slack_webhook_url: Slack webhook for notifications (or use env var) auto_notify: Whether to auto-send Slack notifications """ self.slack_webhook_url = slack_webhook_url or os.getenv("SLACK_WEBHOOK_URL") self.auto_notify = auto_notify self.hitl_queue = get_hitl_queue() self.template_engine = ProposalTemplateEngine() # Ensure directories exist self.PROPOSALS_DIR.mkdir(parents=True, exist_ok=True) self.PDF_OUTPUT_DIR.mkdir(parents=True, exist_ok=True) # Initialize database schema if HAS_DB: self._init_database() def _init_database(self): """Initialize PostgreSQL tables for proposal revisions.""" try: conn = psycopg2.connect(**PostgresConfig.get_connection_params()) cur = conn.cursor() # Create proposal_submissions table cur.execute(""" CREATE TABLE IF NOT EXISTS proposal_submissions ( request_id TEXT PRIMARY KEY, proposal_type TEXT NOT NULL, business_context JSONB NOT NULL, proposal_content TEXT NOT NULL, monthly_price INTEGER NOT NULL, buildout_price INTEGER NOT NULL, total_first_month INTEGER NOT NULL, status TEXT DEFAULT 'pending', pdf_path TEXT, created_at TIMESTAMP DEFAULT NOW(), approved_at TIMESTAMP, reviewer_notes TEXT, rejection_reason TEXT ) """) # Create proposal_revisions table cur.execute(""" CREATE TABLE IF NOT EXISTS proposal_revisions ( revision_id TEXT PRIMARY KEY, request_id TEXT REFERENCES proposal_submissions(request_id) ON DELETE CASCADE, revision_number INTEGER NOT NULL, original_content TEXT NOT NULL, revised_content TEXT, feedback TEXT, status TEXT DEFAULT 'pending', created_at TIMESTAMP DEFAULT NOW(), addressed_at TIMESTAMP ) """) # Create indices cur.execute(""" CREATE INDEX IF NOT EXISTS idx_proposal_status ON proposal_submissions(status) """) cur.execute(""" CREATE INDEX IF NOT EXISTS idx_revision_request ON proposal_revisions(request_id) """) conn.commit() cur.close() conn.close() logger.info("Database schema initialized successfully") except Exception as e: logger.error(f"Database init error: {e}") def _generate_request_id(self, business_name: str) -> str: """Generate unique proposal request ID.""" timestamp = datetime.now().strftime("%Y%m%d%H%M%S") name_hash = hashlib.md5(business_name.encode()).hexdigest()[:6] return f"PROP-{timestamp}-{name_hash.upper()}" def _generate_revision_id(self, request_id: str, revision_number: int) -> str: """Generate revision ID.""" return f"{request_id}-REV{revision_number:02d}" def submit_proposal( self, proposal_result: ProposalResult, priority: str = "normal" ) -> ProposalSubmission: """ Submit a generated proposal for approval. Auto-submits to HITL queue and sends Slack notification. Args: proposal_result: Result from template engine priority: Approval urgency (low, normal, high, urgent) Returns: ProposalSubmission with pending status """ context = proposal_result.business_context request_id = self._generate_request_id(context.business_name) # Create submission record submission = ProposalSubmission( request_id=request_id, proposal_type=proposal_result.proposal_type, business_context=context, proposal_content=proposal_result.content, monthly_price=proposal_result.monthly_price, buildout_price=proposal_result.buildout_price, total_first_month=proposal_result.total_first_month, ) # Submit to HITL queue hitl_request = self.hitl_queue.submit_for_approval( output_type=OutputType.PROPOSAL, title=f"Proposal: {context.business_name} - {proposal_result.proposal_type.value}", content=proposal_result.content, target_audience=f"{context.contact_name} at {context.business_name}", context={ "business_name": context.business_name, "contact_name": context.contact_name, "industry": context.industry, "proposal_type": proposal_result.proposal_type.value, "monthly_price": proposal_result.monthly_price, "buildout_price": proposal_result.buildout_price, "proposal_request_id": request_id, }, priority=priority ) # Save to database if HAS_DB: self._save_submission_to_db(submission) # Save to file for backup self._save_submission_to_file(submission) # Send Slack notification if self.auto_notify: self.notify_new_proposal( proposal_title=f"{context.business_name} - {proposal_result.proposal_type.value}", request_id=request_id, monthly_price=proposal_result.monthly_price, buildout_price=proposal_result.buildout_price, ) logger.info(f"Proposal submitted for approval: {request_id}") return submission def _save_submission_to_db(self, submission: ProposalSubmission): """Save submission to PostgreSQL.""" try: conn = psycopg2.connect(**PostgresConfig.get_connection_params()) cur = conn.cursor() cur.execute(""" INSERT INTO proposal_submissions (request_id, proposal_type, business_context, proposal_content, monthly_price, buildout_price, total_first_month, status, pdf_path, created_at, approved_at, reviewer_notes, rejection_reason) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) ON CONFLICT (request_id) DO UPDATE SET proposal_content = EXCLUDED.proposal_content, status = EXCLUDED.status, pdf_path = EXCLUDED.pdf_path, approved_at = EXCLUDED.approved_at, reviewer_notes = EXCLUDED.reviewer_notes, rejection_reason = EXCLUDED.rejection_reason """, ( submission.request_id, submission.proposal_type.value, Json(asdict(submission.business_context)), submission.proposal_content, submission.monthly_price, submission.buildout_price, submission.total_first_month, submission.status.value, submission.pdf_path, submission.created_at, submission.approved_at, submission.reviewer_notes, submission.rejection_reason, )) conn.commit() cur.close() conn.close() except Exception as e: logger.error(f"Database save error: {e}") def _save_submission_to_file(self, submission: ProposalSubmission): """Save submission to file for backup/visibility.""" status_dir = self.PROPOSALS_DIR / submission.status.value status_dir.mkdir(exist_ok=True) file_path = status_dir / f"{submission.request_id}.json" with open(file_path, 'w') as f: json.dump(submission.to_dict(), f, indent=2) def get_submission(self, request_id: str) -> Optional[ProposalSubmission]: """ Get a proposal submission by ID. Args: request_id: Proposal request ID Returns: ProposalSubmission if found, None otherwise """ # Try database first if HAS_DB: try: conn = psycopg2.connect(**PostgresConfig.get_connection_params()) cur = conn.cursor(cursor_factory=RealDictCursor) cur.execute(""" SELECT * FROM proposal_submissions WHERE request_id = %s """, (request_id,)) row = cur.fetchone() if row: # Get revisions cur.execute(""" SELECT * FROM proposal_revisions WHERE request_id = %s ORDER BY revision_number """, (request_id,)) revisions = cur.fetchall() cur.close() conn.close() return ProposalSubmission( request_id=row['request_id'], proposal_type=ProposalType(row['proposal_type']), business_context=BusinessContext(**row['business_context']), proposal_content=row['proposal_content'], monthly_price=row['monthly_price'], buildout_price=row['buildout_price'], total_first_month=row['total_first_month'], status=ApprovalStatus(row['status']), revisions=[ProposalRevision.from_dict(dict(r)) for r in revisions], pdf_path=row['pdf_path'], created_at=str(row['created_at']) if row['created_at'] else None, approved_at=str(row['approved_at']) if row['approved_at'] else None, reviewer_notes=row['reviewer_notes'], rejection_reason=row['rejection_reason'], ) cur.close() conn.close() except Exception as e: logger.error(f"Database read error: {e}") # Fall back to file for status in ApprovalStatus: status_dir = self.PROPOSALS_DIR / status.value file_path = status_dir / f"{request_id}.json" if file_path.exists(): with open(file_path, 'r') as f: data = json.load(f) return ProposalSubmission.from_dict(data) return None def edit_proposal( self, request_id: str, new_content: str, editor_notes: Optional[str] = None ) -> bool: """ Edit a proposal's content before approval. Args: request_id: Proposal to edit new_content: Updated content editor_notes: Notes about the edit Returns: True if edited successfully """ submission = self.get_submission(request_id) if not submission: logger.error(f"Submission not found: {request_id}") return False if submission.status != ApprovalStatus.PENDING: logger.error(f"Cannot edit non-pending submission: {request_id}") return False # Track the edit as a revision revision_number = len(submission.revisions) + 1 revision = ProposalRevision( revision_id=self._generate_revision_id(request_id, revision_number), request_id=request_id, revision_number=revision_number, original_content=submission.proposal_content, revised_content=new_content, feedback=editor_notes or "Manual edit before approval", status=RevisionStatus.ADDRESSED, addressed_at=datetime.now().isoformat(), ) # Update submission submission.proposal_content = new_content submission.revisions.append(revision) # Save changes if HAS_DB: self._save_submission_to_db(submission) self._save_revision_to_db(revision) self._save_submission_to_file(submission) logger.info(f"Proposal edited: {request_id}") return True def _save_revision_to_db(self, revision: ProposalRevision): """Save revision to PostgreSQL.""" try: conn = psycopg2.connect(**PostgresConfig.get_connection_params()) cur = conn.cursor() cur.execute(""" INSERT INTO proposal_revisions (revision_id, request_id, revision_number, original_content, revised_content, feedback, status, created_at, addressed_at) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s) ON CONFLICT (revision_id) DO UPDATE SET revised_content = EXCLUDED.revised_content, status = EXCLUDED.status, addressed_at = EXCLUDED.addressed_at """, ( revision.revision_id, revision.request_id, revision.revision_number, revision.original_content, revision.revised_content, revision.feedback, revision.status.value, revision.created_at, revision.addressed_at, )) conn.commit() cur.close() conn.close() except Exception as e: logger.error(f"Revision save error: {e}") def approve_proposal( self, request_id: str, notes: Optional[str] = None, generate_pdf: bool = True ) -> Tuple[bool, Optional[str]]: """ Approve a proposal and optionally generate PDF. Args: request_id: Proposal to approve notes: Approval notes generate_pdf: Whether to generate final PDF Returns: Tuple of (success, pdf_path or None) """ submission = self.get_submission(request_id) if not submission: logger.error(f"Submission not found: {request_id}") return False, None if submission.status not in [ApprovalStatus.PENDING, ApprovalStatus.REVISION_REQUESTED]: logger.error(f"Cannot approve submission with status: {submission.status}") return False, None # Update status submission.status = ApprovalStatus.APPROVED submission.approved_at = datetime.now().isoformat() submission.reviewer_notes = notes # Generate PDF pdf_path = None if generate_pdf and HAS_REPORTLAB: pdf_path = self.generate_final_pdf(submission) submission.pdf_path = pdf_path # Update HITL queue self.hitl_queue.approve(request_id, notes) # Save changes if HAS_DB: self._save_submission_to_db(submission) self._save_submission_to_file(submission) # Send approval notification if self.auto_notify: self._notify_approval(submission) logger.info(f"Proposal approved: {request_id}") return True, pdf_path def reject_proposal( self, request_id: str, reason: str ) -> bool: """ Reject a proposal and log for training data. Args: request_id: Proposal to reject reason: Rejection reason (for training data) Returns: True if rejected successfully """ submission = self.get_submission(request_id) if not submission: logger.error(f"Submission not found: {request_id}") return False if submission.status != ApprovalStatus.PENDING: logger.error(f"Cannot reject non-pending submission: {request_id}") return False # Update status submission.status = ApprovalStatus.REJECTED submission.rejection_reason = reason # Update HITL queue self.hitl_queue.reject(request_id, reason) # Log rejection for training self._log_rejection(submission, reason) # Save changes if HAS_DB: self._save_submission_to_db(submission) self._save_submission_to_file(submission) logger.info(f"Proposal rejected: {request_id}") return True def request_revision( self, request_id: str, feedback: str ) -> Optional[ProposalRevision]: """ Request revision of a proposal. Args: request_id: Proposal needing revision feedback: What needs to change Returns: ProposalRevision record if successful """ submission = self.get_submission(request_id) if not submission: logger.error(f"Submission not found: {request_id}") return None if submission.status not in [ApprovalStatus.PENDING, ApprovalStatus.REVISION_REQUESTED]: logger.error(f"Cannot request revision for status: {submission.status}") return None # Create revision record revision_number = len(submission.revisions) + 1 revision = ProposalRevision( revision_id=self._generate_revision_id(request_id, revision_number), request_id=request_id, revision_number=revision_number, original_content=submission.proposal_content, feedback=feedback, ) # Update submission submission.status = ApprovalStatus.REVISION_REQUESTED submission.revisions.append(revision) # Update HITL queue self.hitl_queue.request_revision(request_id, feedback) # Save changes if HAS_DB: self._save_submission_to_db(submission) self._save_revision_to_db(revision) self._save_submission_to_file(submission) # Notify about revision request if self.auto_notify: self._notify_revision_requested(submission, feedback) logger.info(f"Revision requested for: {request_id}") return revision def address_revision( self, request_id: str, revised_content: str ) -> bool: """ Address a revision request with new content. Args: request_id: Proposal being revised revised_content: New content addressing feedback Returns: True if revision addressed successfully """ submission = self.get_submission(request_id) if not submission: logger.error(f"Submission not found: {request_id}") return False if submission.status != ApprovalStatus.REVISION_REQUESTED: logger.error(f"No pending revision for: {request_id}") return False # Find pending revision pending_revision = None for revision in submission.revisions: if revision.status == RevisionStatus.PENDING: pending_revision = revision break if not pending_revision: logger.error(f"No pending revision found for: {request_id}") return False # Update revision pending_revision.revised_content = revised_content pending_revision.status = RevisionStatus.ADDRESSED pending_revision.addressed_at = datetime.now().isoformat() # Update submission submission.proposal_content = revised_content submission.status = ApprovalStatus.PENDING # Back to pending for re-review # Save changes if HAS_DB: self._save_submission_to_db(submission) self._save_revision_to_db(pending_revision) self._save_submission_to_file(submission) # Notify about revision addressed if self.auto_notify: self._notify_revision_addressed(submission) logger.info(f"Revision addressed for: {request_id}") return True def _log_rejection(self, submission: ProposalSubmission, reason: str): """Log rejection for training data.""" rejection_record = { "request_id": submission.request_id, "proposal_type": submission.proposal_type.value, "business_name": submission.business_context.business_name, "industry": submission.business_context.industry, "rejection_reason": reason, "proposal_content_preview": submission.proposal_content[:500], "rejected_at": datetime.now().isoformat(), } with open(self.REJECTION_LOG, 'a') as f: f.write(json.dumps(rejection_record) + "\n") logger.info(f"Rejection logged for training: {submission.request_id}") def generate_final_pdf(self, submission: ProposalSubmission) -> Optional[str]: """ Generate final PDF for approved proposal. Args: submission: Approved proposal submission Returns: Path to generated PDF or None """ if not HAS_REPORTLAB: logger.warning("reportlab not available, skipping PDF generation") return None context = submission.business_context safe_name = "".join(c for c in context.business_name if c.isalnum() or c in " -_") filename = f"{submission.request_id}_{safe_name}.pdf" pdf_path = self.PDF_OUTPUT_DIR / filename try: doc = SimpleDocTemplate( str(pdf_path), pagesize=A4, rightMargin=20*mm, leftMargin=20*mm, topMargin=20*mm, bottomMargin=20*mm, ) styles = getSampleStyleSheet() # Custom styles title_style = ParagraphStyle( 'ProposalTitle', parent=styles['Heading1'], fontSize=24, spaceAfter=12*mm, textColor=colors.HexColor('#1a365d'), ) heading_style = ParagraphStyle( 'ProposalHeading', parent=styles['Heading2'], fontSize=16, spaceBefore=8*mm, spaceAfter=4*mm, textColor=colors.HexColor('#2d4a6f'), ) body_style = ParagraphStyle( 'ProposalBody', parent=styles['Normal'], fontSize=11, leading=16, spaceAfter=4*mm, ) # Build document content story = [] # Title story.append(Paragraph( f"Business Proposal for {context.business_name}", title_style )) # Contact info story.append(Paragraph( f"Prepared for: {context.contact_name}", body_style )) story.append(Paragraph( f"Industry: {context.industry}", body_style )) story.append(Paragraph( f"Date: {datetime.now().strftime('%d %B %Y')}", body_style )) story.append(Spacer(1, 10*mm)) # Proposal content (convert markdown to paragraphs) content_lines = submission.proposal_content.split('\n') for line in content_lines: line = line.strip() if not line: story.append(Spacer(1, 4*mm)) elif line.startswith('# '): story.append(Paragraph(line[2:], title_style)) elif line.startswith('## '): story.append(Paragraph(line[3:], heading_style)) elif line.startswith('### '): story.append(Paragraph(line[4:], heading_style)) elif line.startswith('- '): # Bullet point bullet_text = f"• {line[2:]}" story.append(Paragraph(bullet_text, body_style)) else: # Clean up markdown formatting for PDF clean_line = line.replace('**', '').replace('*', '') if clean_line: story.append(Paragraph(clean_line, body_style)) # Pricing summary table story.append(Spacer(1, 10*mm)) story.append(Paragraph("Investment Summary", heading_style)) pricing_data = [ ["Item", "Amount (AUD)"], ["One-Time Setup", f"${submission.buildout_price:,}"], ["Monthly Subscription", f"${submission.monthly_price:,}/month"], ["First Month Total", f"${submission.total_first_month:,}"], ] pricing_table = Table(pricing_data, colWidths=[100*mm, 50*mm]) pricing_table.setStyle(TableStyle([ ('BACKGROUND', (0, 0), (-1, 0), colors.HexColor('#1a365d')), ('TEXTCOLOR', (0, 0), (-1, 0), colors.whitesmoke), ('ALIGN', (0, 0), (-1, -1), 'CENTER'), ('FONTNAME', (0, 0), (-1, 0), 'Helvetica-Bold'), ('FONTSIZE', (0, 0), (-1, 0), 12), ('BOTTOMPADDING', (0, 0), (-1, 0), 12), ('BACKGROUND', (0, 1), (-1, -1), colors.HexColor('#f7fafc')), ('GRID', (0, 0), (-1, -1), 1, colors.HexColor('#e2e8f0')), ('FONTNAME', (0, -1), (-1, -1), 'Helvetica-Bold'), ])) story.append(pricing_table) # Footer story.append(Spacer(1, 15*mm)) story.append(Paragraph( f"Proposal ID: {submission.request_id}", ParagraphStyle('Footer', fontSize=9, textColor=colors.gray) )) story.append(Paragraph( f"Generated by AgileAdapt on {datetime.now().strftime('%d %B %Y')}", ParagraphStyle('Footer', fontSize=9, textColor=colors.gray) )) # Build PDF doc.build(story) logger.info(f"PDF generated: {pdf_path}") return str(pdf_path) except Exception as e: logger.error(f"PDF generation error: {e}") return None # ========================================================================== # SLACK NOTIFICATION METHODS # ========================================================================== def notify_new_proposal( self, proposal_title: str, request_id: str, monthly_price: int, buildout_price: int, ) -> bool: """ Send Slack notification for new proposal submission. Args: proposal_title: Title of the proposal request_id: Proposal request ID monthly_price: Monthly subscription price buildout_price: One-time setup price Returns: True if notification sent successfully """ if not self.slack_webhook_url or not HAS_REQUESTS: logger.warning("Slack webhook not configured or requests not available") return False message = { "text": f"New proposal awaiting approval: {proposal_title}", "blocks": [ { "type": "header", "text": { "type": "plain_text", "text": "New Proposal Submitted", } }, { "type": "section", "fields": [ { "type": "mrkdwn", "text": f"*Proposal:*\n{proposal_title}" }, { "type": "mrkdwn", "text": f"*ID:*\n`{request_id}`" }, { "type": "mrkdwn", "text": f"*Monthly:*\n${monthly_price:,}" }, { "type": "mrkdwn", "text": f"*Buildout:*\n${buildout_price:,}" }, ] }, { "type": "context", "elements": [ { "type": "mrkdwn", "text": f"Submitted at {datetime.now().strftime('%Y-%m-%d %H:%M')}" } ] } ] } return self._send_slack_message(message) def _notify_approval(self, submission: ProposalSubmission) -> bool: """Send notification for approved proposal.""" if not self.slack_webhook_url or not HAS_REQUESTS: return False message = { "text": f"Proposal approved: {submission.business_context.business_name}", "blocks": [ { "type": "header", "text": { "type": "plain_text", "text": "Proposal Approved", } }, { "type": "section", "text": { "type": "mrkdwn", "text": f"*{submission.business_context.business_name}* proposal has been approved.\n\nID: `{submission.request_id}`" } }, ] } if submission.pdf_path: message["blocks"].append({ "type": "context", "elements": [{ "type": "mrkdwn", "text": f"PDF generated: {submission.pdf_path}" }] }) return self._send_slack_message(message) def _notify_revision_requested( self, submission: ProposalSubmission, feedback: str ) -> bool: """Send notification for revision request.""" if not self.slack_webhook_url or not HAS_REQUESTS: return False message = { "text": f"Revision requested: {submission.business_context.business_name}", "blocks": [ { "type": "header", "text": { "type": "plain_text", "text": "Revision Requested", } }, { "type": "section", "text": { "type": "mrkdwn", "text": f"*{submission.business_context.business_name}* proposal needs revision.\n\nID: `{submission.request_id}`" } }, { "type": "section", "text": { "type": "mrkdwn", "text": f"*Feedback:*\n{feedback}" } }, ] } return self._send_slack_message(message) def _notify_revision_addressed(self, submission: ProposalSubmission) -> bool: """Send notification when revision is addressed.""" if not self.slack_webhook_url or not HAS_REQUESTS: return False message = { "text": f"Revision addressed: {submission.business_context.business_name}", "blocks": [ { "type": "header", "text": { "type": "plain_text", "text": "Revision Addressed", } }, { "type": "section", "text": { "type": "mrkdwn", "text": f"*{submission.business_context.business_name}* proposal revision has been addressed and is ready for re-review.\n\nID: `{submission.request_id}`" } }, ] } return self._send_slack_message(message) def _send_slack_message(self, message: Dict[str, Any]) -> bool: """Send message to Slack webhook.""" try: response = requests.post( self.slack_webhook_url, json=message, timeout=10 ) if response.status_code == 200: logger.info("Slack notification sent") return True else: logger.error(f"Slack notification failed: {response.status_code}") return False except Exception as e: logger.error(f"Slack notification error: {e}") return False # ========================================================================== # QUEUE MANAGEMENT METHODS # ========================================================================== def get_pending_proposals(self) -> List[ProposalSubmission]: """Get all pending proposal submissions.""" submissions = [] if HAS_DB: try: conn = psycopg2.connect(**PostgresConfig.get_connection_params()) cur = conn.cursor(cursor_factory=RealDictCursor) cur.execute(""" SELECT * FROM proposal_submissions WHERE status IN ('pending', 'revision_requested') ORDER BY created_at """) for row in cur.fetchall(): submissions.append(ProposalSubmission( request_id=row['request_id'], proposal_type=ProposalType(row['proposal_type']), business_context=BusinessContext(**row['business_context']), proposal_content=row['proposal_content'], monthly_price=row['monthly_price'], buildout_price=row['buildout_price'], total_first_month=row['total_first_month'], status=ApprovalStatus(row['status']), pdf_path=row['pdf_path'], created_at=str(row['created_at']) if row['created_at'] else None, )) cur.close() conn.close() return submissions except Exception as e: logger.error(f"Database read error: {e}") # Fall back to file system for status in [ApprovalStatus.PENDING, ApprovalStatus.REVISION_REQUESTED]: status_dir = self.PROPOSALS_DIR / status.value if status_dir.exists(): for file_path in status_dir.glob("PROP-*.json"): try: with open(file_path, 'r') as f: data = json.load(f) submissions.append(ProposalSubmission.from_dict(data)) except Exception as e: logger.error(f"Error loading {file_path}: {e}") return submissions def get_queue_summary(self) -> Dict[str, Any]: """Get summary of proposal queue.""" summary = { "pending": 0, "approved": 0, "rejected": 0, "revision_requested": 0, "total_value_pending": 0, "total_value_approved": 0, } if HAS_DB: try: conn = psycopg2.connect(**PostgresConfig.get_connection_params()) cur = conn.cursor(cursor_factory=RealDictCursor) cur.execute(""" SELECT status, COUNT(*) as count, COALESCE(SUM(total_first_month), 0) as total_value FROM proposal_submissions GROUP BY status """) for row in cur.fetchall(): status = row['status'] summary[status] = row['count'] if status == 'pending': summary['total_value_pending'] = row['total_value'] elif status == 'approved': summary['total_value_approved'] = row['total_value'] cur.close() conn.close() except Exception as e: logger.error(f"Database read error: {e}") return summary # ========================================================================== # CONVENIENCE FUNCTIONS # ========================================================================== _workflow_instance: Optional[ProposalApprovalWorkflow] = None def get_proposal_workflow() -> ProposalApprovalWorkflow: """Get singleton proposal workflow instance.""" global _workflow_instance if _workflow_instance is None: _workflow_instance = ProposalApprovalWorkflow() return _workflow_instance def submit_proposal_for_approval( proposal_result: ProposalResult, priority: str = "normal" ) -> ProposalSubmission: """ Quick function to submit a proposal for approval. Example: from core.proposals.approval_workflow import submit_proposal_for_approval from core.proposals.template_engine import ProposalTemplateEngine, BusinessContext engine = ProposalTemplateEngine() context = BusinessContext( business_name="Smith's Plumbing", contact_name="John Smith", industry="Trades" ) result = engine.generate_proposal(context) submission = submit_proposal_for_approval(result, priority="high") """ workflow = get_proposal_workflow() return workflow.submit_proposal(proposal_result, priority) def approve_proposal( request_id: str, notes: Optional[str] = None ) -> Tuple[bool, Optional[str]]: """Quick function to approve a proposal.""" workflow = get_proposal_workflow() return workflow.approve_proposal(request_id, notes) def reject_proposal(request_id: str, reason: str) -> bool: """Quick function to reject a proposal.""" workflow = get_proposal_workflow() return workflow.reject_proposal(request_id, reason) def request_proposal_revision(request_id: str, feedback: str) -> Optional[ProposalRevision]: """Quick function to request proposal revision.""" workflow = get_proposal_workflow() return workflow.request_revision(request_id, feedback) # ========================================================================== # CLI INTERFACE # ========================================================================== def review_proposals(): """Interactive CLI for reviewing proposal queue.""" workflow = get_proposal_workflow() print("\n" + "="*60) print("PROPOSAL APPROVAL QUEUE") print("="*60) summary = workflow.get_queue_summary() print(f"\nQueue Summary:") print(f" Pending: {summary['pending']}") print(f" Approved: {summary['approved']}") print(f" Rejected: {summary['rejected']}") print(f" Revision Requested: {summary['revision_requested']}") print(f" Pending Value: ${summary['total_value_pending']:,}") print(f" Approved Value: ${summary['total_value_approved']:,}") pending = workflow.get_pending_proposals() if not pending: print("\nNo pending proposals.") return print(f"\n{len(pending)} pending proposal(s):\n") for i, submission in enumerate(pending, 1): ctx = submission.business_context print(f"[{i}] {submission.request_id}") print(f" Business: {ctx.business_name}") print(f" Contact: {ctx.contact_name}") print(f" Type: {submission.proposal_type.value}") print(f" Monthly: ${submission.monthly_price:,}") print(f" Buildout: ${submission.buildout_price:,}") print(f" Total: ${submission.total_first_month:,}") print(f" Status: {submission.status.value}") print(f" Revisions: {len(submission.revisions)}") print("-" * 40) print(f" Content Preview:") preview = submission.proposal_content[:300] + "..." if len(submission.proposal_content) > 300 else submission.proposal_content for line in preview.split('\n')[:5]: print(f" | {line}") print("-" * 40) print() if __name__ == "__main__": review_proposals()