# /opt/docker/dev/service_finder/backend/app/services/validation_engine.py """ ValidationEngine — Gamified Provider Validation & Promotion Engine. P0 Architecture: Dynamic rule-based validation scoring for service providers. Responsibilities: 1. Fetch dynamic validation rules from gamification.validation_rules table. 2. Compute validation scores for ServiceStaging entries based on: - Admin override (score = 100) - User level from gamification.user_stats (weighted vote) - Robot base score (BOT_BASE_SCORE) 3. Log validation events in marketplace.provider_validations. 4. Auto-promote staging entries to full Organization + ServiceProfile when validation_level >= AUTO_APPROVE_THRESHOLD. Usage: engine = ValidationEngine() await engine.process_staging_score(db, staging, user_id=42) await engine.process_staging_score(db, staging, is_admin=True) """ import logging import uuid from typing import Optional from sqlalchemy import select, func from sqlalchemy.ext.asyncio import AsyncSession from app.models import ( ValidationRule, ServiceStaging, UserStats, Organization, ServiceProfile, Address, ) from app.models.identity.social import ProviderValidation from app.schemas.address import AddressIn from app.services.address_service import create_address logger = logging.getLogger("ValidationEngine") # ────────────────────────────────────────────────────────────────────────────── # Default fallback values if the validation_rules table is empty # ────────────────────────────────────────────────────────────────────────────── DEFAULT_RULES: dict[str, int] = { "BOT_BASE_SCORE": 20, "FIRST_ENTRY_SCORE": 40, "USER_CONFIRM_BASE": 20, "AUTO_APPROVE_THRESHOLD": 80, } class ValidationEngine: """ Gamified Validation Engine for service provider staging entries. Operates on ServiceStaging records, computing validation_level scores based on dynamic rules from the gamification.validation_rules table. """ # ── Rule Loading ───────────────────────────────────────────────────────── @staticmethod async def get_rules(db: AsyncSession) -> dict[str, int]: """ Fetch all active validation rules from the database. Returns a dict of rule_key → rule_value. Falls back to DEFAULT_RULES for any key not found in the table, ensuring the engine always has sensible defaults even before seeding. """ stmt = select(ValidationRule) result = await db.execute(stmt) db_rules = result.scalars().all() rules = dict(DEFAULT_RULES) # start with fallbacks for rule in db_rules: rules[rule.rule_key] = rule.rule_value return rules # ── Scoring ────────────────────────────────────────────────────────────── @staticmethod async def _get_user_weight( db: AsyncSession, user_id: int, rules: dict[str, int] ) -> int: """ Calculate a user's vote weight based on their gamification level. Uses gamification.user_stats.current_level as the multiplier. Formula: USER_CONFIRM_BASE + (current_level * 5) If no UserStats record exists, returns USER_CONFIRM_BASE as fallback. """ base = rules.get("USER_CONFIRM_BASE", DEFAULT_RULES["USER_CONFIRM_BASE"]) stmt = select(UserStats).where(UserStats.user_id == user_id) result = await db.execute(stmt) stats = result.scalar_one_or_none() if stats is None: logger.warning(f"No UserStats found for user_id={user_id}, using base weight={base}") return base level = stats.current_level or 1 weight = base + (level * 5) logger.info( f"User {user_id} weight: base={base}, level={level}, total_weight={weight}" ) return weight # ── Validation Logging ─────────────────────────────────────────────────── @staticmethod async def _log_validation( db: AsyncSession, provider_id: int, voter_user_id: int, validation_type: str, weight: int, metadata: Optional[dict] = None, ) -> Optional[ProviderValidation]: """ Log a validation event in marketplace.provider_validations. NOTE: ProviderValidation.provider_id references marketplace.service_providers.id, NOT marketplace.service_staging.id. This method should only be called AFTER promotion when a real ServiceProvider record exists. If the provider_id does not correspond to a valid service_providers row, the FK constraint will fail — in that case we log a warning and skip. Uses INSERT ... ON CONFLICT DO NOTHING semantics via a pre-check to respect the UniqueConstraint(voter_user_id, provider_id). """ # Check if this user already validated this provider stmt = select(ProviderValidation).where( ProviderValidation.voter_user_id == voter_user_id, ProviderValidation.provider_id == provider_id, ) existing = (await db.execute(stmt)).scalar_one_or_none() if existing: logger.info( f"User {voter_user_id} already validated provider {provider_id} " f"(type={existing.validation_type}, weight={existing.weight})" ) return existing record = ProviderValidation( provider_id=provider_id, voter_user_id=voter_user_id, validation_type=validation_type, weight=weight, validation_metadata=metadata or {}, ) db.add(record) await db.flush() logger.info( f"Logged validation: provider={provider_id}, user={voter_user_id}, " f"type={validation_type}, weight={weight}" ) return record # ── Promotion Logic ────────────────────────────────────────────────────── @staticmethod async def _promote_to_provider( db: AsyncSession, staging: ServiceStaging ) -> tuple[Organization, ServiceProfile]: """ Promote a ServiceStaging entry to a full Organization + ServiceProfile. Steps: 1. Create an Address from staging data (via AddressManager). 2. Create an Organization with org_type='service'. 3. Create a ServiceProfile linked to the Organization. 4. Set staging.status = 'approved'. Returns: Tuple of (Organization, ServiceProfile). """ logger.info( f"Promoting staging {staging.id} ('{staging.name}') to provider..." ) # ── 1. Create Address ──────────────────────────────────────────────── from datetime import datetime, timezone addr_in = AddressIn( city=staging.city or "", zip=staging.postal_code or "", full_address_text=staging.full_address or "", ) # NOTE: created_at has no server_default in the DB, so we set it explicitly address = Address( id=uuid.uuid4(), full_address_text=addr_in.full_address_text, created_at=datetime.now(timezone.utc), ) # Resolve postal code if zip and city provided if addr_in.zip and addr_in.city: from app.services.address_service import _resolve_postal_code postal_code_id = await _resolve_postal_code(db, addr_in.zip, addr_in.city) address.postal_code_id = postal_code_id db.add(address) await db.flush() # ── 2. Create Organization ─────────────────────────────────────────── now = datetime.now(timezone.utc) org = Organization( name=staging.name, full_name=staging.name, folder_slug=f"svc-{uuid.uuid4().hex[:12]}", org_type="service", address_id=address.id, status="active", is_active=True, is_verified=True, # NOTE: Several columns have server_default in the model but NOT # in the actual DB, so we set them explicitly: created_at=now, first_registered_at=now, current_lifecycle_started_at=now, subscription_plan="FREE", base_asset_limit=1, purchased_extra_slots=0, lifecycle_index=1, is_ownership_transferable=True, notification_settings={"notify_owner": True, "alert_days_before": [30, 15, 7, 1]}, external_integration_config={}, ) db.add(org) await db.flush() # ── 3. Create ServiceProfile ───────────────────────────────────────── from sqlalchemy import text as sa_text profile = ServiceProfile( organization_id=org.id, fingerprint=staging.fingerprint, status="active", trust_score=staging.trust_score or 20, is_verified=True, contact_phone=staging.contact_phone, contact_email=staging.contact_email, website=staging.website, # NOTE: location is NOT NULL Geometry in the DB with no default, # so we set a default point (0,0) at SRID 4326 location=func.ST_SetSRID(func.ST_MakePoint(0, 0), 4326), ) db.add(profile) await db.flush() # ── 4. Update staging record ───────────────────────────────────────── staging.status = "approved" staging.organization_id = org.id staging.service_profile_id = profile.id staging.published_at = func.now() logger.info( f"Promotion complete: staging={staging.id} → " f"org={org.id} ('{org.name}'), profile={profile.id}" ) return org, profile # ── Main Entry Point ───────────────────────────────────────────────────── async def process_staging_score( self, db: AsyncSession, staging: ServiceStaging, user_id: Optional[int] = None, is_admin: bool = False, ) -> int: """ Compute and apply a validation score for a ServiceStaging entry. Logic: - If is_admin: score = 100 (instant approval). - If user_id provided: query UserStats.current_level, compute weight = USER_CONFIRM_BASE + (level * 5). - Otherwise: use BOT_BASE_SCORE from rules. - Log the validation event in provider_validations. - Update staging.validation_level. - If validation_level >= AUTO_APPROVE_THRESHOLD, auto-promote. Args: db: Database session. staging: The ServiceStaging record to score. user_id: Optional user who triggered the validation. is_admin: If True, score = 100 (admin override). Returns: The new validation_level after applying the score. """ rules = await self.get_rules(db) auto_approve = rules.get( "AUTO_APPROVE_THRESHOLD", DEFAULT_RULES["AUTO_APPROVE_THRESHOLD"] ) # ── Compute score ──────────────────────────────────────────────────── if is_admin: score = 100 effective_user_id = 0 # system user placeholder logger.info(f"Admin override: score=100 for staging {staging.id}") elif user_id is not None: weight = await self._get_user_weight(db, user_id, rules) score = weight effective_user_id = user_id logger.info( f"User {user_id} validation: weight={weight} for staging {staging.id}" ) else: score = rules.get("BOT_BASE_SCORE", DEFAULT_RULES["BOT_BASE_SCORE"]) effective_user_id = 0 logger.info( f"Robot base score: {score} for staging {staging.id}" ) # ── Update validation_level ────────────────────────────────────────── current_level = staging.validation_level or 0 new_level = min(current_level + score, 100) # cap at 100 staging.validation_level = new_level # ── Log validation event ───────────────────────────────────────────── # NOTE: ProviderValidation.provider_id references marketplace.service_providers.id. # For staging-level validations (before promotion), we skip logging to # provider_validations since the staging entry is not a service_provider yet. # After promotion, the _promote_to_provider method can log validations. if effective_user_id > 0 and staging.organization_id is not None: # Only log if the staging has been promoted (has an org) await self._log_validation( db, provider_id=staging.id, voter_user_id=effective_user_id, validation_type="approve" if score >= 0 else "reject", weight=score, metadata={ "staging_id": staging.id, "score": score, "validation_level_before": current_level, "validation_level_after": new_level, "is_admin": is_admin, }, ) await db.flush() logger.info( f"Staging {staging.id} validation_level: {current_level} → {new_level} " f"(score={score})" ) # ── Auto-promote if threshold reached ──────────────────────────────── if new_level >= auto_approve and staging.status != "approved": logger.info( f"Auto-approve threshold reached ({new_level} >= {auto_approve}) " f"for staging {staging.id}" ) await self._promote_to_provider(db, staging) return new_level