from features.base_feature import BaseFeature from tools.tavily_search import TavilySearchTool from tools.calculator import CalculatorTool _tavily = TavilySearchTool() _calc = CalculatorTool() class FreelanceFeature(BaseFeature): async def get_context(self, preferences_a: dict, preferences_b: dict, user_a_id: int = None, user_b_id: int = None) -> str: """ Benchmark market rates via Tavily. Pre-calculate rate × hours and detect if budget is insufficient (forcing scope reduction). """ raw_a = preferences_a.get("raw_details", {}) raw_b = preferences_b.get("raw_details", {}) # Identify freelancer vs client role_a = raw_a.get("role", preferences_a.get("goal", "")) if "client" in str(role_a).lower(): freelancer_raw, client_raw = raw_b, raw_a else: freelancer_raw, client_raw = raw_a, raw_b skill = ( freelancer_raw.get("skill") or freelancer_raw.get("expertise") or freelancer_raw.get("tech_stack") or client_raw.get("project_type") or "software development" ) rate = freelancer_raw.get("rate") or freelancer_raw.get("hourly_rate") or "" hours = freelancer_raw.get("hours") or freelancer_raw.get("estimated_hours") or "" client_budget = client_raw.get("budget") or client_raw.get("max_budget") or "" upfront_min = freelancer_raw.get("upfront_minimum") or freelancer_raw.get("upfront") or "50" scope = client_raw.get("required_features") or client_raw.get("scope") or [] # Pre-calculate rate × hours calc_text = "" if rate and hours: try: total_cost = float(str(rate).replace(",", "")) * float(str(hours).replace(",", "")) calc_text = f"Pre-calculated cost: ₹{rate}/hr × {hours} hrs = ₹{total_cost:,.0f}" if client_budget: budget_float = float(str(client_budget).replace(",", "")) if total_cost > budget_float: affordable_hours = budget_float / float(str(rate).replace(",", "")) calc_text += ( f"\n⚠️ Budget shortfall: ₹{client_budget} budget covers only " f"{affordable_hours:.1f} hrs at ₹{rate}/hr. " f"Reduce scope to fit, removing nice-to-haves first." ) else: calc_text += f"\n✅ Budget ₹{client_budget} is sufficient." except (ValueError, TypeError): calc_text = f"Rate: ₹{rate}/hr, Estimated hours: {hours}" # Market rate benchmark market_text = "" try: query = f"average freelance rate {skill} developer India 2026" result = await _tavily.execute(query) answer = result.get("answer", "") results = result.get("results", [])[:2] parts = [] if answer: parts.append(f"Market summary: {answer[:250]}") for r in results: content = r.get("content", "")[:100] title = r.get("title", "") if title: parts.append(f" • {title}: {content}") market_text = "\n".join(parts) except Exception as e: market_text = f"Market search unavailable. Use typical India rates for {skill}." lines = [ "FREELANCE NEGOTIATION DOMAIN RULES:", "• Budget is a hard constraint for the client — NEVER exceed it.", "• Freelancer's minimum rate is a hard constraint — NEVER go below it.", "• Non-negotiables (IP ownership, upfront minimum) are absolute hard constraints.", "• If budget < full scope cost: reduce scope (nice-to-haves first, then by priority).", "• Payment terms: freelancer pushes for more upfront, client for back-loaded.", "• Scope reduction must preserve the client's core 'must-have' features.", "• After agreement, include UPI ID and first milestone amount in settlement.", ] if skill: lines.append(f"\nProject skill/type: {skill}") if calc_text: lines.append(f"\n{calc_text}") if upfront_min: lines.append(f"Freelancer's minimum upfront: {upfront_min}%") if scope and isinstance(scope, list): lines.append(f"Client's required features: {', '.join(str(s) for s in scope[:5])}") if market_text: lines.append(f"\nMARKET RATE DATA (cite this):\n{market_text}") return "\n".join(lines) def format_resolution( self, resolution: dict, preferences_a: dict, preferences_b: dict ) -> str: status = resolution.get("status", "resolved") final = resolution.get("final_proposal", {}) details = final.get("details", {}) rounds = resolution.get("rounds_taken", "?") summary = resolution.get("summary", "") if status == "escalated": return ( f"⚠️ *Project Deal — Human Review Needed*\n\n" f"_{summary}_\n\n" f"Agents couldn't finalize in {rounds} round(s). " f"Please negotiate scope/budget directly." ) budget = details.get("budget") or details.get("agreed_budget") or details.get("price") or "" timeline = details.get("timeline") or details.get("duration") or "" scope = details.get("scope") or details.get("deliverables") or [] payment_schedule = details.get("payment_schedule") or details.get("payments") or "" milestone_1 = details.get("milestone_1") or details.get("upfront") or "" settlement = details.get("settlement") or {} lines = ["💼 *Project Deal Agreed!*\n"] if budget: lines.append(f"💰 *Budget:* ₹{budget}") if timeline: lines.append(f"📅 *Timeline:* {timeline}") if scope and isinstance(scope, list): lines.append(f"📋 *Scope:*") for item in scope[:5]: lines.append(f" ✓ {item}") elif scope: lines.append(f"📋 *Scope:* {scope}") if payment_schedule: lines.append(f"💳 *Payment schedule:* {payment_schedule}") elif milestone_1: lines.append(f"💳 *First milestone payment:* ₹{milestone_1}") lines.append(f"\n⏱ Agreed in {rounds} round(s)") if summary and summary != "Agreement reached": lines.append(f"_{summary}_") return "\n".join(lines)