import logging from typing import Any from aura_core.module_contract import aura_module, ModuleResult from aura_core.types import ValidatorInput logger = logging.getLogger(__name__) ANALYSIS_SCHEMA = { "domain": {"type": "string", "required": True}, "keywords": {"type": "list_of_strings", "required": True, "min_items": 1}, "interpretations": {"type": "list_of_strings", "required": False}, "domain_summary": {"type": "string", "required": False}, "niches": { "type": "list_of_dicts", "required": True, "min_items": 1, "item_schema": { "name": {"type": "string", "required": True}, "description": {"type": "string", "required": True}, "score": {"type": "number", "required": False, "default": 5}, "monetization_model": {"type": "string", "required": False, "default": "affiliate"}, "target_audience": {"type": "string", "required": False, "default": ""}, "time_to_revenue": {"type": "string", "required": False, "default": "medium"}, "valuation_band": {"type": "string", "required": False, "default": "1000-5000"}, "affiliate_programs": {"type": "list_of_strings", "required": False, "default": []}, "requires_inventory": {"type": "bool", "required": False, "default": False}, "synopsis": {"type": "string", "required": False, "default": ""}, }, }, } BRAND_SCHEMA = { "options": { "type": "list_of_dicts", "required": True, "min_items": 1, "item_schema": { "name": {"type": "string", "required": True}, "tagline": {"type": "string", "required": False, "default": ""}, }, }, "recommended": {"type": "number", "required": False, "default": 0}, "color_primary": {"type": "string", "required": False, "default": "#4F46E5"}, "color_secondary": {"type": "string", "required": False, "default": "#7C3AED"}, "color_accent": {"type": "string", "required": False, "default": "#06B6D4"}, "industry_context": {"type": "string", "required": False, "default": ""}, } FIELD_ALIASES = { "testimonials": {"author": ["name", "attribution", "person"]}, "features": {"title": ["heading", "name"], "description": ["text", "body", "detail"]}, "pricing_tiers": {"name": ["title", "tier"], "price": ["cost", "amount"]}, "team_members": {"name": ["member_name"], "role": ["title", "position", "job_title"]}, "faq_items": {"question": ["q", "title"], "answer": ["a", "response", "text"]}, "stats": {"value": ["number", "stat", "metric"], "label": ["title", "name", "description"]}, "how_it_works_steps": {"title": ["heading", "name", "step"], "description": ["text", "body", "detail"]}, "resource_items": {"title": ["name", "heading"], "description": ["text", "body", "summary"]}, "gallery_sections": {"title": ["name", "heading"]}, } SITE_COPY_LIST_FIELDS = { "features": { "required_keys": ["title", "description"], "optional_keys": ["icon"], "min_items": 0, }, "testimonials": { "required_keys": ["quote", "author"], "optional_keys": ["role", "company", "rating", "name"], "min_items": 0, }, "faq_items": { "required_keys": ["question", "answer"], "optional_keys": [], "min_items": 0, }, "stats": { "required_keys": ["value", "label"], "optional_keys": ["icon"], "min_items": 0, }, "how_it_works_steps": { "required_keys": ["title", "description"], "optional_keys": ["icon", "step_number"], "min_items": 0, }, "pricing_tiers": { "required_keys": ["name", "price"], "optional_keys": ["description", "features", "cta", "popular"], "min_items": 0, }, "team_members": { "required_keys": ["name", "role"], "optional_keys": ["bio", "image"], "min_items": 0, }, "gallery_sections": { "required_keys": ["title"], "optional_keys": ["description", "image", "items"], "min_items": 0, }, "resource_items": { "required_keys": ["title"], "optional_keys": ["description", "icon", "url"], "min_items": 0, }, } MISPLACEMENT_SIGNATURES = { "features": { "looks_like_faq": lambda item: "question" in item and "answer" in item, "target": "faq_items", }, "faq_items": { "looks_like_feature": lambda item: "icon" in item and "title" in item and "question" not in item, "target": "features", }, "testimonials": { "looks_like_team": lambda item: "role" in item and "bio" in item and "quote" not in item, "target": "team_members", }, "team_members": { "looks_like_testimonial": lambda item: "quote" in item, "target": "testimonials", }, "stats": { "looks_like_feature": lambda item: "description" in item and len(item.get("description", "")) > 80 and "value" not in item, "target": "features", }, } SITE_COPY_STRING_FIELDS = [ "headline", "subheadline", "hero_body", "cta_text", "about", "about_title", "mission_statement", "features_title", "features_subtitle", "testimonials_title", "faq_title", "problem_title", "problem_description", "solution_title", "solution_description", "pricing_title", "pricing_subtitle", "contact_title", "contact_description", "resources_title", "comparison_title", "gallery_title", "team_title", "team_subtitle", "how_it_works_title", "final_cta_title", "final_cta_body", "final_cta_button", ] class ValidationReporter: def __init__(self, context: str): self.context = context self.errors: list[dict] = [] self.warnings: list[dict] = [] self.repairs: list[dict] = [] def error(self, field: str, message: str): self.errors.append({"field": field, "message": message}) logger.error(f"[VALIDATE:{self.context}] {field}: {message}") def warn(self, field: str, message: str): self.warnings.append({"field": field, "message": message}) logger.warning(f"[VALIDATE:{self.context}] {field}: {message}") def repair(self, field: str, action: str): self.repairs.append({"field": field, "action": action}) logger.info(f"[VALIDATE:{self.context}] REPAIRED {field}: {action}") @property def is_valid(self) -> bool: return len(self.errors) == 0 @property def score(self) -> int: base = 100 base -= len(self.errors) * 15 base -= len(self.warnings) * 3 return max(0, min(100, base)) def to_dict(self) -> dict: return { "context": self.context, "valid": self.is_valid, "score": self.score, "error_count": len(self.errors), "warning_count": len(self.warnings), "repair_count": len(self.repairs), "errors": self.errors, "warnings": self.warnings, "repairs": self.repairs, } def _check_type(value: Any, expected_type: str) -> bool: if expected_type == "string": return isinstance(value, str) if expected_type == "number": return isinstance(value, (int, float)) if expected_type == "bool": return isinstance(value, bool) if expected_type == "list_of_strings": return isinstance(value, list) and all(isinstance(v, str) for v in value) if expected_type == "list_of_dicts": return isinstance(value, list) and all(isinstance(v, dict) for v in value) if expected_type == "dict": return isinstance(value, dict) return True def validate_analysis(data: Any, auto_repair: bool = True) -> tuple[dict, ValidationReporter]: report = ValidationReporter("analysis") if not isinstance(data, dict): report.error("root", f"Expected dict, got {type(data).__name__}") return data, report for field_name, spec in ANALYSIS_SCHEMA.items(): val = data.get(field_name) if val is None: if spec.get("required"): report.error(field_name, "Required field missing") if auto_repair and "default" in spec: data[field_name] = spec["default"] report.repair(field_name, f"Set default: {spec['default']}") continue if not _check_type(val, spec["type"]): report.error(field_name, f"Expected {spec['type']}, got {type(val).__name__}") if auto_repair: if spec["type"] == "string" and not isinstance(val, str): data[field_name] = str(val) report.repair(field_name, "Coerced to string") elif spec["type"] == "list_of_strings" and isinstance(val, str): data[field_name] = [val] report.repair(field_name, "Wrapped string in list") elif spec["type"] == "list_of_dicts" and isinstance(val, dict): data[field_name] = [val] report.repair(field_name, "Wrapped single dict in list") if spec["type"].startswith("list_of") and isinstance(data.get(field_name), list): min_items = spec.get("min_items", 0) if len(data[field_name]) < min_items: report.error(field_name, f"Expected at least {min_items} items, got {len(data[field_name])}") if spec["type"] == "list_of_dicts" and "item_schema" in spec: items = data.get(field_name, []) if isinstance(items, list): for i, item in enumerate(items): if not isinstance(item, dict): report.error(f"{field_name}[{i}]", f"Expected dict, got {type(item).__name__}") if auto_repair: items[i] = {list(spec["item_schema"].keys())[0]: str(item)} report.repair(f"{field_name}[{i}]", "Converted to dict") continue for item_field, item_spec in spec["item_schema"].items(): item_val = item.get(item_field) if item_val is None and item_spec.get("required"): report.warn(f"{field_name}[{i}].{item_field}", "Required sub-field missing") if auto_repair and "default" in item_spec: item[item_field] = item_spec["default"] report.repair(f"{field_name}[{i}].{item_field}", f"Set default: {item_spec['default']}") elif item_val is None and not item_spec.get("required") and "default" in item_spec: if auto_repair: item[item_field] = item_spec["default"] report.repair(f"{field_name}[{i}].{item_field}", f"Set default: {item_spec['default']}") elif item_val is not None and not _check_type(item_val, item_spec["type"]): report.warn(f"{field_name}[{i}].{item_field}", f"Expected {item_spec['type']}, got {type(item_val).__name__}") if auto_repair and item_spec["type"] == "string": item[item_field] = str(item_val) report.repair(f"{field_name}[{i}].{item_field}", "Coerced to string") elif auto_repair and item_spec["type"] == "number" and isinstance(item_val, str): try: item[item_field] = float(item_val) report.repair(f"{field_name}[{i}].{item_field}", "Coerced to number") except ValueError: item[item_field] = item_spec.get("default", 0) report.repair(f"{field_name}[{i}].{item_field}", f"Set default: {item_spec.get('default', 0)}") return data, report def validate_brand(data: Any, auto_repair: bool = True) -> tuple[dict, ValidationReporter]: report = ValidationReporter("brand") if not isinstance(data, dict): report.error("root", f"Expected dict, got {type(data).__name__}") if auto_repair: data = {"options": [{"name": "Brand", "tagline": ""}], "recommended": 0} report.repair("root", "Created minimal brand structure") return data, report for field_name, spec in BRAND_SCHEMA.items(): val = data.get(field_name) if val is None: if spec.get("required"): report.error(field_name, "Required field missing") if auto_repair and "default" in spec: data[field_name] = spec["default"] report.repair(field_name, "Set default") if spec["type"] == "list_of_dicts" and "item_schema" in spec: items = data.get(field_name, []) if isinstance(items, list): for i, item in enumerate(items): if not isinstance(item, dict): report.error(f"{field_name}[{i}]", "Expected dict in list") if auto_repair: items[i] = {"name": str(item), "tagline": ""} report.repair(f"{field_name}[{i}]", "Converted to dict") options = data.get("options", []) rec = data.get("recommended", 0) if isinstance(options, list) and isinstance(rec, (int, float)): if rec >= len(options) and options: report.warn("recommended", f"Index {rec} out of range (options length: {len(options)})") if auto_repair: data["recommended"] = 0 report.repair("recommended", "Reset to 0") return data, report def _detect_and_fix_misplacements(data: dict, report: ValidationReporter): for source_field, sigs in MISPLACEMENT_SIGNATURES.items(): source_items = data.get(source_field) if not isinstance(source_items, list) or not source_items: continue target_field = sigs["target"] check_fns = {k: v for k, v in sigs.items() if k != "target" and callable(v)} misplaced_indices = [] for i, item in enumerate(source_items): if not isinstance(item, dict): continue for check_name, check_fn in check_fns.items(): try: if check_fn(item): misplaced_indices.append(i) report.warn(f"{source_field}[{i}]", f"Looks misplaced ({check_name}), moving to {target_field}") break except Exception: pass if misplaced_indices: if target_field not in data or not isinstance(data.get(target_field), list): data[target_field] = [] for idx in sorted(misplaced_indices, reverse=True): moved_item = source_items.pop(idx) data[target_field].append(moved_item) report.repair( f"{source_field}→{target_field}", f"Moved misplaced item: {moved_item.get('title', moved_item.get('question', 'item'))[:40]}", ) def validate_site_copy(data: Any, auto_repair: bool = True) -> tuple[dict, ValidationReporter]: report = ValidationReporter("site_copy") if not isinstance(data, dict): report.error("root", f"Expected dict, got {type(data).__name__}") if auto_repair: data = {"headline": "Welcome"} report.repair("root", "Created minimal site_copy") return data, report if auto_repair: _detect_and_fix_misplacements(data, report) for field in SITE_COPY_STRING_FIELDS: val = data.get(field) if val is not None and not isinstance(val, str): report.warn(field, f"Expected string, got {type(val).__name__}") if auto_repair: if isinstance(val, dict): text = val.get(field, val.get("text", val.get("content", val.get("title", "")))) if isinstance(text, str): data[field] = text else: data[field] = str(val) else: data[field] = str(val) report.repair(field, "Coerced to string") for field_name, spec in SITE_COPY_LIST_FIELDS.items(): val = data.get(field_name) if val is None: continue if isinstance(val, dict): if field_name in val: report.warn(field_name, "Nested wrapper detected") if auto_repair: data[field_name] = val[field_name] val = data[field_name] report.repair(field_name, "Unwrapped nested structure") if not isinstance(val, list): report.error(field_name, f"Expected list, got {type(val).__name__}") if auto_repair: if isinstance(val, dict): data[field_name] = [val] else: data[field_name] = [] report.repair(field_name, "Converted to list") continue for i, item in enumerate(val): if not isinstance(item, dict): report.warn(f"{field_name}[{i}]", f"Expected dict, got {type(item).__name__}") if auto_repair: if isinstance(item, str) and spec["required_keys"]: val[i] = {spec["required_keys"][0]: item} if len(spec["required_keys"]) > 1: val[i][spec["required_keys"][1]] = "" report.repair(f"{field_name}[{i}]", "Converted string to dict") continue aliases = FIELD_ALIASES.get(field_name, {}) for req_key in spec["required_keys"]: if req_key not in item: resolved = False for alias in aliases.get(req_key, []): if alias in item and item[alias]: if auto_repair: item[req_key] = item[alias] report.repair(f"{field_name}[{i}].{req_key}", f"Resolved from alias '{alias}'") resolved = True break if not resolved: report.warn(f"{field_name}[{i}].{req_key}", "Required key missing in list item") if auto_repair: item[req_key] = "" report.repair(f"{field_name}[{i}].{req_key}", "Set empty string default") ct = data.get("comparison_table") if ct is not None: if not isinstance(ct, dict): report.warn("comparison_table", f"Expected dict, got {type(ct).__name__}") if auto_repair: data["comparison_table"] = {"headers": [], "rows": []} report.repair("comparison_table", "Reset to empty table") else: if "headers" not in ct: ct["headers"] = [] report.repair("comparison_table.headers", "Added empty headers") if "rows" not in ct: ct["rows"] = [] report.repair("comparison_table.rows", "Added empty rows") return data, report def validate_package(package_data: Any, auto_repair: bool = True) -> tuple[dict, ValidationReporter]: report = ValidationReporter("package") if not isinstance(package_data, dict): report.error("root", f"Expected dict, got {type(package_data).__name__}") if auto_repair: package_data = {} report.repair("root", "Created empty package dict") else: return package_data, report brand = package_data.get("brand", {}) brand, brand_report = validate_brand(brand, auto_repair=auto_repair) package_data["brand"] = brand report.errors.extend(brand_report.errors) report.warnings.extend(brand_report.warnings) report.repairs.extend(brand_report.repairs) site_copy = package_data.get("site_copy", {}) site_copy, copy_report = validate_site_copy(site_copy, auto_repair=auto_repair) package_data["site_copy"] = site_copy report.errors.extend(copy_report.errors) report.warnings.extend(copy_report.warnings) report.repairs.extend(copy_report.repairs) sales = package_data.get("sales_letter") if sales is not None and not isinstance(sales, str): report.warn("sales_letter", f"Expected string, got {type(sales).__name__}") if auto_repair: package_data["sales_letter"] = str(sales) report.repair("sales_letter", "Coerced to string") return package_data, report @aura_module( name="validator", version="1.0.0", description="Three-layer validation pipeline — normalizes, validates schemas for site copy and brand data, and auto-repairs missing fields with safe defaults", input_model=ValidatorInput, tags=["validator", "validation", "quality"], ) def run_validator(config: dict) -> ModuleResult: data = config.get("data", {}) auto_repair = config.get("auto_repair", True) target = config.get("target", "site_copy") if target == "analysis": result = validate_analysis(data) return ModuleResult(ok=result["valid"], output=result) elif target == "brand": validated, report = validate_brand_data(data, auto_repair=auto_repair) return ModuleResult( ok=report.is_valid, output={"data": validated, "report": report.to_dict()}, ) else: validated, report = validate_site_copy(data, auto_repair=auto_repair) return ModuleResult( ok=report.is_valid, output={"data": validated, "report": report.to_dict()}, )