""" AI Service for RateRight Provides machine learning capabilities for smart matching, recommendations, and predictions """ import json import re import numpy as np from datetime import datetime, timedelta from typing import Dict, List, Optional, Tuple import logging from sqlalchemy import func, and_, or_ from app.models import db from app.models.user import User from app.models.job import Job from app.models.contract import Contract from app.models.rating import Rating from app.models.message import Message from app.models.category import Category logger = logging.getLogger(__name__) class AIService: """Main AI service for intelligent features""" # Skills database for Australian construction industry TRADE_SKILLS = { 'carpentry': ['framing', 'decking', 'roofing', 'formwork', 'finishing', 'cabinet making'], 'plumbing': ['pipe fitting', 'drainage', 'gas fitting', 'hot water', 'sewerage', 'backflow'], 'electrical': ['wiring', 'switchboard', 'lighting', 'power points', 'solar', 'data cabling'], 'bricklaying': ['block laying', 'pointing', 'rendering', 'stone masonry', 'retaining walls'], 'concreting': ['formwork', 'steel fixing', 'pumping', 'finishing', 'polishing', 'cutting'], 'painting': ['interior', 'exterior', 'spray painting', 'wallpapering', 'surface prep'], 'tiling': ['floor tiling', 'wall tiling', 'waterproofing', 'grouting', 'mosaic'], 'plastering': ['dry wall', 'cornice', 'rendering', 'texture coating', 'repairs'], 'landscaping': ['paving', 'retaining walls', 'irrigation', 'planting', 'excavation'], 'roofing': ['metal roofing', 'tile roofing', 'guttering', 'fascia', 'roof plumbing'] } # Price ranges per trade (AUD per hour) PRICE_RANGES = { 'carpentry': {'min': 45, 'avg': 65, 'max': 95}, 'plumbing': {'min': 80, 'avg': 110, 'max': 150}, 'electrical': {'min': 85, 'avg': 115, 'max': 160}, 'bricklaying': {'min': 50, 'avg': 70, 'max': 100}, 'concreting': {'min': 45, 'avg': 60, 'max': 85}, 'painting': {'min': 40, 'avg': 55, 'max': 75}, 'tiling': {'min': 45, 'avg': 65, 'max': 90}, 'plastering': {'min': 50, 'avg': 70, 'max': 95}, 'landscaping': {'min': 40, 'avg': 55, 'max': 75}, 'roofing': {'min': 55, 'avg': 75, 'max': 105} } # Fraud indicators FRAUD_PATTERNS = [ 'too good to be true pricing', 'rush payment requests', 'avoiding written contracts', 'no ABN or expired insurance', 'multiple accounts same details', 'suspicious review patterns', 'location mismatches', 'copy-pasted descriptions' ] @classmethod def smart_job_matching(cls, worker_id: int, limit: int = 10) -> List[Dict]: """ AI-powered job matching based on skills, location, availability, and past performance """ try: worker = User.query.get(worker_id) if not worker or worker.role != 'worker': return [] # Extract worker's skills worker_skills = cls.extract_skills_from_text( f"{worker.primary_trade} {worker.bio if hasattr(worker, 'bio') else ''}" ) # Get all open jobs open_jobs = Job.query.filter_by(status='open').all() matches = [] for job in open_jobs: score = 0 # Skill matching (40% weight) job_skills = cls.extract_skills_from_text(job.description) skill_match = len(set(worker_skills) & set(job_skills)) score += (skill_match / max(len(job_skills), 1)) * 40 # Location proximity (20% weight) if cls._calculate_location_match(worker.location, job.location): score += 20 # Price range matching (20% weight) if job.budget: trade_price = cls.PRICE_RANGES.get(worker.primary_trade, {}) if trade_price.get('avg', 0) <= job.budget: score += 20 # Rating bonus (10% weight) if worker.average_rating > 4.0: score += 10 # Availability match (10% weight) if cls._check_availability_match(worker_id, job.start_date): score += 10 matches.append({ 'job_id': job.id, 'title': job.title, 'description': job.description[:200], 'budget': job.budget, 'location': job.location, 'match_score': score, 'match_reasons': cls._generate_match_reasons(score, skill_match, worker.average_rating) }) # Sort by match score and return top matches matches.sort(key=lambda x: x['match_score'], reverse=True) return matches[:limit] except Exception as e: logger.error(f"Error in smart job matching: {str(e)}") return [] @classmethod def price_recommendation(cls, job_details: Dict) -> Dict: """ AI-powered price recommendation based on job type, scope, and market rates """ try: trade = job_details.get('trade', 'general') description = job_details.get('description', '') location = job_details.get('location', '') urgency = job_details.get('urgency', 'normal') # Base price from trade base_price = cls.PRICE_RANGES.get(trade, {'avg': 60}) recommended_hourly = base_price['avg'] # Adjust for complexity complexity_keywords = ['complex', 'difficult', 'specialist', 'custom', 'heritage'] if any(keyword in description.lower() for keyword in complexity_keywords): recommended_hourly *= 1.2 # Adjust for urgency if urgency == 'urgent': recommended_hourly *= 1.3 elif urgency == 'emergency': recommended_hourly *= 1.5 # Adjust for location (metropolitan vs regional) metro_areas = ['sydney', 'melbourne', 'brisbane', 'perth', 'adelaide'] if any(city in location.lower() for city in metro_areas): recommended_hourly *= 1.1 # Estimate hours based on job scope estimated_hours = cls._estimate_job_hours(description, trade) # Calculate total estimate total_estimate = recommended_hourly * estimated_hours # Get market comparison similar_jobs = cls._get_similar_job_prices(trade, location) return { 'recommended_hourly': round(recommended_hourly, 2), 'estimated_hours': estimated_hours, 'total_estimate': round(total_estimate, 2), 'price_range': { 'low': round(total_estimate * 0.8, 2), 'high': round(total_estimate * 1.2, 2) }, 'market_comparison': similar_jobs, 'factors': { 'base_rate': base_price['avg'], 'complexity_adjustment': 'Applied' if 'complex' in description.lower() else 'None', 'urgency_adjustment': urgency, 'location_adjustment': 'Metropolitan' if any(city in location.lower() for city in metro_areas) else 'Regional' }, 'confidence': 'high' if similar_jobs else 'medium' } except Exception as e: logger.error(f"Error in price recommendation: {str(e)}") return {'error': 'Unable to generate price recommendation'} @classmethod def fraud_detection(cls, user_id: int, context: Dict = None) -> Dict: """ AI-powered fraud detection system """ try: user = User.query.get(user_id) if not user: return {'risk_level': 'unknown', 'score': 0} risk_score = 0 risk_factors = [] # Check ABN validity if not user.abn_number or len(user.abn_number) != 11: risk_score += 20 risk_factors.append('Invalid or missing ABN') # Check insurance status if not user.public_liability_insurance: risk_score += 15 risk_factors.append('No public liability insurance') if user.insurance_expiry_date and user.insurance_expiry_date < datetime.now().date(): risk_score += 25 risk_factors.append('Expired insurance') # Check account age account_age = (datetime.now() - user.date_created).days if account_age < 7: risk_score += 10 risk_factors.append('New account (less than 7 days)') # Check for duplicate accounts duplicates = User.query.filter( or_( User.phone_number == user.phone_number, User.abn_number == user.abn_number ), User.id != user.id ).count() if duplicates > 0: risk_score += 30 risk_factors.append('Potential duplicate accounts') # Check review patterns if user.average_rating > 0: reviews = Rating.query.filter_by(rated_user_id=user_id).all() if cls._detect_suspicious_reviews(reviews): risk_score += 20 risk_factors.append('Suspicious review patterns') # Check messaging patterns if context and context.get('check_messages'): suspicious_messages = cls._check_message_patterns(user_id) if suspicious_messages: risk_score += 15 risk_factors.append('Suspicious messaging patterns') # Determine risk level if risk_score >= 60: risk_level = 'high' elif risk_score >= 30: risk_level = 'medium' else: risk_level = 'low' return { 'risk_level': risk_level, 'risk_score': risk_score, 'risk_factors': risk_factors, 'recommendations': cls._get_fraud_recommendations(risk_level, risk_factors) } except Exception as e: logger.error(f"Error in fraud detection: {str(e)}") return {'risk_level': 'unknown', 'score': 0, 'error': str(e)} @classmethod def quality_prediction(cls, worker_id: int, job_details: Dict) -> Dict: """ Predict quality of work based on historical performance """ try: worker = User.query.get(worker_id) if not worker: return {'predicted_quality': 'unknown'} # Base quality from average rating base_quality = worker.average_rating or 3.0 # Analyze past contracts past_contracts = Contract.query.filter_by(worker_id=worker_id, status='completed').all() if past_contracts: # Calculate on-time completion rate on_time_rate = sum(1 for c in past_contracts if c.completed_on_time) / len(past_contracts) # Calculate dispute rate dispute_rate = sum(1 for c in past_contracts if c.had_dispute) / len(past_contracts) # Adjust quality prediction quality_score = base_quality quality_score += on_time_rate * 0.5 # Bonus for on-time completion quality_score -= dispute_rate * 1.0 # Penalty for disputes # Factor in experience if len(past_contracts) > 20: quality_score += 0.3 # Experience bonus else: quality_score = base_quality # Ensure score is within bounds quality_score = max(1.0, min(5.0, quality_score)) # Generate prediction if quality_score >= 4.5: prediction = 'excellent' confidence = 'high' elif quality_score >= 4.0: prediction = 'very_good' confidence = 'high' elif quality_score >= 3.5: prediction = 'good' confidence = 'medium' elif quality_score >= 3.0: prediction = 'satisfactory' confidence = 'medium' else: prediction = 'needs_improvement' confidence = 'low' return { 'predicted_quality': prediction, 'quality_score': round(quality_score, 2), 'confidence': confidence, 'factors': { 'average_rating': round(base_quality, 2), 'completed_jobs': len(past_contracts), 'on_time_rate': round(on_time_rate * 100, 1) if past_contracts else None, 'dispute_rate': round(dispute_rate * 100, 1) if past_contracts else None }, 'recommendation': cls._get_quality_recommendation(prediction) } except Exception as e: logger.error(f"Error in quality prediction: {str(e)}") return {'predicted_quality': 'unknown', 'error': str(e)} @classmethod def demand_forecasting(cls, trade: str, location: str, period_days: int = 30) -> Dict: """ Forecast demand for specific trades in locations """ try: # Get historical job data end_date = datetime.now() start_date = end_date - timedelta(days=90) # 3 months of data historical_jobs = Job.query.filter( Job.created_at >= start_date, Job.created_at <= end_date, Job.category == trade ).all() # Calculate weekly averages weeks_data = {} for job in historical_jobs: week = job.created_at.isocalendar()[1] if week not in weeks_data: weeks_data[week] = 0 weeks_data[week] += 1 avg_weekly_demand = sum(weeks_data.values()) / max(len(weeks_data), 1) # Seasonal adjustments current_month = datetime.now().month seasonal_factor = cls._get_seasonal_factor(trade, current_month) # Calculate forecast predicted_demand = avg_weekly_demand * (period_days / 7) * seasonal_factor # Determine trend recent_weeks = sorted(weeks_data.keys())[-4:] if len(weeks_data) >= 4 else sorted(weeks_data.keys()) if recent_weeks: recent_avg = sum(weeks_data[w] for w in recent_weeks) / len(recent_weeks) trend = 'increasing' if recent_avg > avg_weekly_demand else 'decreasing' else: trend = 'stable' # Get supply metrics active_workers = User.query.filter_by( role='worker', primary_trade=trade, is_active=True ).count() supply_demand_ratio = active_workers / max(predicted_demand, 1) market_condition = 'high_demand' if supply_demand_ratio < 0.5 else 'balanced' if supply_demand_ratio < 2 else 'oversupplied' return { 'trade': trade, 'location': location, 'forecast_period': f"{period_days} days", 'predicted_jobs': round(predicted_demand), 'current_trend': trend, 'market_condition': market_condition, 'metrics': { 'historical_weekly_avg': round(avg_weekly_demand, 1), 'seasonal_factor': seasonal_factor, 'active_workers': active_workers, 'supply_demand_ratio': round(supply_demand_ratio, 2) }, 'recommendations': cls._get_demand_recommendations(market_condition, trend) } except Exception as e: logger.error(f"Error in demand forecasting: {str(e)}") return {'error': 'Unable to generate forecast'} @classmethod def chatbot_response(cls, user_input: str, context: Dict = None) -> Dict: """ AI-powered chatbot for customer support """ input_lower = user_input.lower() # Intent classification intent = cls._classify_intent(input_lower) responses = { 'greeting': { 'text': "G'day! Welcome to RateRight. How can I help you find the right tradie today?", 'suggestions': ['Find a worker', 'Post a job', 'Check my messages', 'View my dashboard'] }, 'find_worker': { 'text': "I can help you find skilled workers. What type of trade are you looking for?", 'suggestions': ['Carpenter', 'Plumber', 'Electrician', 'Painter', 'All trades'] }, 'post_job': { 'text': "Let's get your job posted! You can post a job by clicking 'Post Job' in the menu. I'll need details about the work, location, and your budget.", 'suggestions': ['Go to post job', 'See pricing guide', 'Browse workers first'] }, 'pricing': { 'text': f"Here are typical hourly rates in Australia:\n• Carpenter: $45-95/hr\n• Plumber: $80-150/hr\n• Electrician: $85-160/hr\n• Painter: $40-75/hr\nWould you like a specific quote?", 'suggestions': ['Get quote', 'See all trades', 'Post job'] }, 'safety': { 'text': "Safety is our priority! All workers on RateRight must have:\n✓ Valid ABN\n✓ Public liability insurance\n✓ White Card (for construction sites)\n✓ Verified credentials", 'suggestions': ['Verify a worker', 'Report an issue', 'Safety guidelines'] }, 'payment': { 'text': "RateRight uses secure milestone payments. You only pay when work stages are completed and approved. We hold funds in escrow for your protection.", 'suggestions': ['How payments work', 'Payment issues', 'Refund policy'] }, 'help': { 'text': "I'm here to help! What would you like to know about?", 'suggestions': ['Finding workers', 'Posting jobs', 'Payments', 'Safety', 'My account'] }, 'unknown': { 'text': "I'm not quite sure about that. Let me connect you with our support team or try asking about:", 'suggestions': ['Finding workers', 'Posting jobs', 'Pricing', 'Contact support'] } } response = responses.get(intent, responses['unknown']) # Add context-aware information if context: if context.get('user_role') == 'worker': response['suggestions'].append('Find jobs') elif context.get('user_role') == 'contractor': response['suggestions'].append('Post a job') return { 'intent': intent, 'response': response['text'], 'suggestions': response['suggestions'], 'confidence': 0.8 if intent != 'unknown' else 0.3 } @classmethod def auto_categorize_job(cls, job_description: str) -> Dict: """ Automatically categorize jobs based on description """ description_lower = job_description.lower() # Define category keywords category_keywords = { 'carpentry': ['carpenter', 'wood', 'timber', 'deck', 'frame', 'cabinet'], 'plumbing': ['plumber', 'pipe', 'water', 'drainage', 'tap', 'toilet', 'sink'], 'electrical': ['electrician', 'wiring', 'power', 'lights', 'switch', 'outlet'], 'painting': ['paint', 'painter', 'walls', 'ceiling', 'exterior', 'interior'], 'tiling': ['tiles', 'tiler', 'bathroom', 'floor', 'grout'], 'concreting': ['concrete', 'slab', 'driveway', 'foundation', 'pour'], 'landscaping': ['garden', 'lawn', 'landscap', 'plants', 'irrigation'], 'roofing': ['roof', 'gutter', 'tiles', 'metal roof', 'fascia'], 'bricklaying': ['brick', 'block', 'masonry', 'mortar', 'wall'], 'plastering': ['plaster', 'render', 'gyprock', 'ceiling', 'cornice'] } # Score each category category_scores = {} for category, keywords in category_keywords.items(): score = sum(1 for keyword in keywords if keyword in description_lower) if score > 0: category_scores[category] = score if category_scores: # Get top category primary_category = max(category_scores, key=category_scores.get) confidence = min(category_scores[primary_category] / 3, 1.0) # Normalize confidence # Get secondary categories secondary_categories = [cat for cat, score in category_scores.items() if cat != primary_category and score > 0] else: primary_category = 'general' secondary_categories = [] confidence = 0.3 # Extract specific requirements requirements = cls._extract_requirements(job_description) return { 'primary_category': primary_category, 'secondary_categories': secondary_categories[:2], # Max 2 secondary 'confidence': round(confidence, 2), 'extracted_requirements': requirements, 'suggested_skills': cls.TRADE_SKILLS.get(primary_category, []) } @classmethod def sentiment_analysis(cls, review_text: str) -> Dict: """ Analyze sentiment of reviews """ text_lower = review_text.lower() # Positive indicators positive_words = [ 'excellent', 'great', 'amazing', 'fantastic', 'outstanding', 'professional', 'reliable', 'punctual', 'quality', 'recommend', 'happy', 'pleased', 'satisfied', 'perfect', 'best', 'efficient', 'friendly', 'honest', 'trustworthy', 'skilled' ] # Negative indicators negative_words = [ 'poor', 'bad', 'terrible', 'awful', 'horrible', 'unprofessional', 'late', 'messy', 'incomplete', 'overcharged', 'disappointed', 'upset', 'angry', 'worst', 'avoid', 'shoddy', 'careless', 'rude', 'dishonest', 'incompetent' ] # Neutral/mixed indicators neutral_words = ['okay', 'fine', 'average', 'acceptable', 'adequate'] # Count occurrences positive_count = sum(1 for word in positive_words if word in text_lower) negative_count = sum(1 for word in negative_words if word in text_lower) neutral_count = sum(1 for word in neutral_words if word in text_lower) # Calculate sentiment score (-1 to 1) total_indicators = positive_count + negative_count + neutral_count if total_indicators > 0: sentiment_score = (positive_count - negative_count) / total_indicators else: sentiment_score = 0 # Determine sentiment category if sentiment_score > 0.3: sentiment = 'positive' elif sentiment_score < -0.3: sentiment = 'negative' else: sentiment = 'neutral' # Check for specific issues mentioned issues = [] issue_keywords = { 'timing': ['late', 'delay', 'overdue', 'schedule'], 'quality': ['poor quality', 'shoddy', 'redo', 'fix', 'mistake'], 'communication': ['no response', 'ignored', 'hard to reach', 'communication'], 'pricing': ['overcharged', 'expensive', 'hidden costs', 'extra charges'], 'cleanliness': ['messy', 'dirty', 'clean up', 'debris'] } for issue_type, keywords in issue_keywords.items(): if any(keyword in text_lower for keyword in keywords): issues.append(issue_type) # Extract key phrases key_phrases = cls._extract_key_phrases(review_text) return { 'sentiment': sentiment, 'sentiment_score': round(sentiment_score, 2), 'confidence': min(total_indicators / 5, 1.0), # Normalize confidence 'positive_indicators': positive_count, 'negative_indicators': negative_count, 'issues_mentioned': issues, 'key_phrases': key_phrases, 'recommendation': cls._get_sentiment_recommendation(sentiment, issues) } @classmethod def extract_skills_from_text(cls, text: str) -> List[str]: """ Extract skills from job descriptions or profiles """ text_lower = text.lower() extracted_skills = [] # Check against all trade skills for trade, skills in cls.TRADE_SKILLS.items(): # Check trade name if trade in text_lower: extracted_skills.append(trade) # Check specific skills for skill in skills: if skill in text_lower: extracted_skills.append(skill) # Check for certifications and qualifications certifications = [ 'white card', 'ewp', 'forklift', 'crane', 'scaffolding', 'asbestos', 'confined space', 'working at heights', 'first aid', 'cert iii', 'cert iv', 'diploma', 'license' ] for cert in certifications: if cert in text_lower: extracted_skills.append(cert) # Check for tools and equipment tools = [ 'excavator', 'bobcat', 'power tools', 'hand tools', 'welding', 'cutting', 'drilling', 'measuring' ] for tool in tools: if tool in text_lower: extracted_skills.append(f"{tool} operation") # Remove duplicates and return return list(set(extracted_skills)) # Helper methods @classmethod def _calculate_location_match(cls, worker_location: str, job_location: str) -> bool: """Check if locations are in same area""" if not worker_location or not job_location: return False # Simple suburb/city matching worker_parts = worker_location.lower().split(',') job_parts = job_location.lower().split(',') # Check if any part matches for wp in worker_parts: for jp in job_parts: if wp.strip() in jp.strip() or jp.strip() in wp.strip(): return True return False @classmethod def _check_availability_match(cls, worker_id: int, job_date) -> bool: """Check if worker is available for job date""" # Simplified check - in production would check actual availability calendar return True @classmethod def _generate_match_reasons(cls, score: float, skill_match: int, rating: float) -> List[str]: """Generate human-readable match reasons""" reasons = [] if score >= 80: reasons.append("Excellent match for your skills") elif score >= 60: reasons.append("Good match for your profile") if skill_match > 2: reasons.append(f"{skill_match} matching skills") if rating > 4.0: reasons.append("Your high rating gives you an advantage") return reasons @classmethod def _estimate_job_hours(cls, description: str, trade: str) -> int: """Estimate hours needed for job""" # Simple estimation based on keywords small_keywords = ['small', 'minor', 'quick', 'simple'] medium_keywords = ['standard', 'typical', 'regular'] large_keywords = ['large', 'major', 'extensive', 'complete', 'full'] description_lower = description.lower() if any(keyword in description_lower for keyword in large_keywords): return 24 # 3 days elif any(keyword in description_lower for keyword in small_keywords): return 4 # Half day else: return 8 # Standard day @classmethod def _get_similar_job_prices(cls, trade: str, location: str) -> Dict: """Get prices of similar recent jobs""" # In production, would query actual completed jobs return { 'average': cls.PRICE_RANGES.get(trade, {}).get('avg', 60) * 8, 'sample_size': 15 } @classmethod def _detect_suspicious_reviews(cls, reviews: List) -> bool: """Detect suspicious review patterns""" if len(reviews) < 3: return False # Check for reviews all on same day review_dates = [r.created_at.date() for r in reviews] if len(set(review_dates)) == 1 and len(reviews) > 2: return True # Check for identical ratings ratings = [r.rating for r in reviews] if len(set(ratings)) == 1 and len(reviews) > 5: return True return False @classmethod def _check_message_patterns(cls, user_id: int) -> bool: """Check for suspicious messaging patterns""" try: recent_messages = Message.query.filter_by(sender_id=user_id).limit(20).all() if not recent_messages: return False # Check for spam-like behavior message_texts = [m.content for m in recent_messages] unique_messages = set(message_texts) # If sending same message repeatedly if len(unique_messages) < len(message_texts) / 3: return True # Check for suspicious keywords suspicious_keywords = ['urgent payment', 'pay now', 'transfer immediately', 'avoid fees', 'limited time', 'act now'] for msg in message_texts: if any(keyword in msg.lower() for keyword in suspicious_keywords): return True return False except: return False @classmethod def _get_fraud_recommendations(cls, risk_level: str, risk_factors: List[str]) -> List[str]: """Get recommendations based on fraud risk""" recommendations = [] if risk_level == 'high': recommendations.append("Verify identity and credentials before proceeding") recommendations.append("Request additional documentation") recommendations.append("Consider using escrow payment") elif risk_level == 'medium': recommendations.append("Proceed with caution") recommendations.append("Verify ABN and insurance") if 'Invalid or missing ABN' in risk_factors: recommendations.append("Request valid ABN before contracting") if 'Expired insurance' in risk_factors: recommendations.append("Request current insurance certificate") return recommendations @classmethod def _get_quality_recommendation(cls, prediction: str) -> str: """Get recommendation based on quality prediction""" recommendations = { 'excellent': "Highly recommended - expect outstanding results", 'very_good': "Strong choice - reliable quality expected", 'good': "Good option - solid track record", 'satisfactory': "Acceptable choice - monitor progress closely", 'needs_improvement': "Consider other options or request references" } return recommendations.get(prediction, "Unable to provide recommendation") @classmethod def _get_seasonal_factor(cls, trade: str, month: int) -> float: """Get seasonal adjustment factor for demand""" # Australian seasons (Southern Hemisphere) # Summer: Dec-Feb, Autumn: Mar-May, Winter: Jun-Aug, Spring: Sep-Nov seasonal_factors = { 'roofing': { 12: 0.7, 1: 0.7, 2: 0.8, # Summer - less roofing 3: 1.2, 4: 1.2, 5: 1.1, # Autumn - peak 6: 0.9, 7: 0.8, 8: 0.9, # Winter - moderate 9: 1.1, 10: 1.2, 11: 1.1 # Spring - busy }, 'landscaping': { 12: 0.8, 1: 0.7, 2: 0.8, # Summer - hot 3: 1.2, 4: 1.3, 5: 1.1, # Autumn - peak 6: 0.6, 7: 0.5, 8: 0.6, # Winter - slow 9: 1.3, 10: 1.4, 11: 1.3 # Spring - very busy }, 'painting': { 12: 1.1, 1: 1.0, 2: 1.1, # Summer - good weather 3: 1.2, 4: 1.2, 5: 1.0, # Autumn - ideal 6: 0.7, 7: 0.6, 8: 0.7, # Winter - weather issues 9: 1.1, 10: 1.2, 11: 1.2 # Spring - busy } } # Get factor for specific trade or use default trade_factors = seasonal_factors.get(trade, {}) return trade_factors.get(month, 1.0) @classmethod def _get_demand_recommendations(cls, market_condition: str, trend: str) -> List[str]: """Get recommendations based on demand forecast""" recommendations = [] if market_condition == 'high_demand': recommendations.append("Good time to increase rates") recommendations.append("Consider hiring additional workers") recommendations.append("Fast response times will win more jobs") elif market_condition == 'oversupplied': recommendations.append("Focus on competitive pricing") recommendations.append("Emphasize quality and differentiation") recommendations.append("Consider expanding service area") if trend == 'increasing': recommendations.append("Prepare for busier period ahead") elif trend == 'decreasing': recommendations.append("Consider diversifying services") return recommendations @classmethod def _classify_intent(cls, text: str) -> str: """Classify user intent from input text""" text = text.lower() # Intent patterns intents = { 'greeting': ['hello', 'hi', 'hey', "g'day", 'good morning', 'good afternoon'], 'find_worker': ['find', 'need', 'looking for', 'worker', 'tradie', 'contractor'], 'post_job': ['post', 'create', 'add', 'job', 'work', 'project'], 'pricing': ['price', 'cost', 'rate', 'charge', 'fee', 'quote', 'expensive'], 'safety': ['safety', 'insurance', 'white card', 'whs', 'compliance', 'liability'], 'payment': ['payment', 'pay', 'invoice', 'milestone', 'escrow', 'refund'], 'help': ['help', 'support', 'assist', 'problem', 'issue', 'question'] } for intent, keywords in intents.items(): if any(keyword in text for keyword in keywords): return intent return 'unknown' @classmethod def _extract_requirements(cls, description: str) -> List[str]: """Extract specific requirements from job description""" requirements = [] description_lower = description.lower() # Time requirements if 'urgent' in description_lower or 'asap' in description_lower: requirements.append('Urgent timeline') if 'weekend' in description_lower: requirements.append('Weekend work') if 'after hours' in description_lower or 'evening' in description_lower: requirements.append('After hours work') # Certification requirements if 'licensed' in description_lower or 'licence' in description_lower: requirements.append('License required') if 'white card' in description_lower: requirements.append('White Card required') if 'height' in description_lower: requirements.append('Working at heights') # Scope indicators if 'commercial' in description_lower: requirements.append('Commercial project') if 'residential' in description_lower: requirements.append('Residential project') if 'industrial' in description_lower: requirements.append('Industrial project') return requirements @classmethod def _extract_key_phrases(cls, text: str) -> List[str]: """Extract key phrases from review text""" key_phrases = [] # Common review phrases phrase_patterns = [ 'on time', 'on budget', 'great work', 'poor quality', 'highly recommend', 'would not recommend', 'very professional', 'excellent service', 'terrible experience', 'good value', 'overpriced', 'attention to detail', 'left a mess', 'cleaned up', 'friendly service', 'rude behavior' ] text_lower = text.lower() for phrase in phrase_patterns: if phrase in text_lower: key_phrases.append(phrase) return key_phrases[:5] # Return top 5 phrases @classmethod def _get_sentiment_recommendation(cls, sentiment: str, issues: List[str]) -> str: """Get recommendation based on sentiment analysis""" if sentiment == 'positive': return "Feature this review prominently" elif sentiment == 'negative': if 'communication' in issues: return "Reach out to resolve communication issues" elif 'quality' in issues: return "Offer to address quality concerns" else: return "Contact customer for service recovery" else: return "Monitor for additional feedback"