Upload 2 files

app.py

@@ -188,7 +188,7 @@ async def extract_tokens(pages_data, progress=gr.Progress()):
                     # Try column names first
                     is_selected = row.get('Select', row.iloc[0] if len(row) > 0 else False)
                     url = row.get('URL', row.iloc[1] if len(row) > 1 else '')
-                except:
+                except (KeyError, IndexError, TypeError):
                     # Fallback to positional
                     is_selected = row.iloc[0] if len(row) > 0 else False
                     url = row.iloc[1] if len(row) > 1 else ''
@@ -1169,88 +1169,98 @@ async def run_stage2_analysis_v2(
                 semantic_analysis = getattr(state, 'semantic_analysis', {})
                 desktop_dict = normalized_to_dict(state.desktop_normalized)
                 
-                # Run agents (with individual error handling)
-                # Brand Identifier
-                try:
-                    brand_result = await brand_agent.analyze(
-                        color_tokens=desktop_dict.get("colors", {}),
-                        semantic_analysis=semantic_analysis,
-                        log_callback=state.log,
-                    )
-                    # Log what the LLM contributed
-                    if brand_result:
-                        bp = brand_result.brand_primary or {}
-                        bs = brand_result.brand_secondary or {}
-                        state.log(f"   ├─ Brand Primary: {bp.get('color', 'N/A')} ({bp.get('confidence', 'N/A')} confidence)")
-                        state.log(f"   │  └─ Reasoning: {bp.get('reasoning', 'N/A')[:80]}")
-                        state.log(f"   ├─ Brand Secondary: {bs.get('color', 'N/A')}")
-                        state.log(f"   ├─ Palette Strategy: {brand_result.palette_strategy or 'N/A'}")
-                        state.log(f"   └─ Cohesion Score: {brand_result.cohesion_score}/10 — {brand_result.cohesion_notes[:60] if brand_result.cohesion_notes else 'N/A'}")
-                except Exception as e:
-                    state.log(f"   ⚠️ Brand Identifier failed: {str(e)[:120]}")
-                    brand_result = BrandIdentification()
+                # Run AURORA, ATLAS, SENTINEL in PARALLEL (then NEXUS sequentially)
+                async def _run_aurora():
+                    """Run AURORA (Brand Identifier) agent."""
+                    try:
+                        result = await brand_agent.analyze(
+                            color_tokens=desktop_dict.get("colors", {}),
+                            semantic_analysis=semantic_analysis,
+                            log_callback=state.log,
+                        )
+                        if result:
+                            bp = result.brand_primary or {}
+                            bs = result.brand_secondary or {}
+                            state.log(f"   ├─ Brand Primary: {bp.get('color', 'N/A')} ({bp.get('confidence', 'N/A')} confidence)")
+                            state.log(f"   │  └─ Reasoning: {bp.get('reasoning', 'N/A')[:80]}")
+                            state.log(f"   ├─ Brand Secondary: {bs.get('color', 'N/A')}")
+                            state.log(f"   ├─ Palette Strategy: {result.palette_strategy or 'N/A'}")
+                            state.log(f"   └─ Cohesion Score: {result.cohesion_score}/10 — {result.cohesion_notes[:60] if result.cohesion_notes else 'N/A'}")
+                        return result
+                    except Exception as e:
+                        state.log(f"   ⚠️ AURORA (Brand Identifier) failed: {str(e)[:120]}")
+                        return BrandIdentification()
 
-                # Benchmark Advisor
-                if benchmark_comparisons:
+                async def _run_atlas():
+                    """Run ATLAS (Benchmark Advisor) agent."""
+                    if not benchmark_comparisons:
+                        return BenchmarkAdvice()
                     try:
-                        benchmark_advice = await benchmark_agent.analyze(
+                        result = await benchmark_agent.analyze(
                             user_ratio=rule_results.typography.detected_ratio,
                             user_base=int(rule_results.typography.base_size) if rule_results.typography.sizes_px else 16,
                             user_spacing=rule_results.spacing.detected_base,
                             benchmark_comparisons=benchmark_comparisons,
                             log_callback=state.log,
                         )
-                        # Log what the LLM contributed
-                        if benchmark_advice:
-                            state.log(f"   ├─ Recommended: {benchmark_advice.recommended_benchmark_name or benchmark_advice.recommended_benchmark or 'N/A'}")
-                            state.log(f"   │  └─ Reasoning: {benchmark_advice.reasoning[:80] if benchmark_advice.reasoning else 'N/A'}")
-                            changes = benchmark_advice.alignment_changes or []
+                        if result:
+                            state.log(f"   ├─ Recommended: {result.recommended_benchmark_name or result.recommended_benchmark or 'N/A'}")
+                            state.log(f"   │  └─ Reasoning: {result.reasoning[:80] if result.reasoning else 'N/A'}")
+                            changes = result.alignment_changes or []
                             state.log(f"   ├─ Changes Needed: {len(changes)}")
                             for i, ch in enumerate(changes[:3]):
                                 if isinstance(ch, dict):
                                     state.log(f"   │  {i+1}. {ch.get('change', ch.get('what', str(ch)[:60]))}")
                                 else:
                                     state.log(f"   │  {i+1}. {str(ch)[:60]}")
-                            if benchmark_advice.pros_of_alignment:
-                                state.log(f"   └─ Pros: {', '.join(str(p)[:30] for p in benchmark_advice.pros_of_alignment[:2])}")
+                            if result.pros_of_alignment:
+                                state.log(f"   └─ Pros: {', '.join(str(p)[:30] for p in result.pros_of_alignment[:2])}")
+                        return result
                     except Exception as e:
-                        state.log(f"   ⚠️ Benchmark Advisor failed: {str(e)[:120]}")
-                        benchmark_advice = BenchmarkAdvice()
-                else:
-                    benchmark_advice = BenchmarkAdvice()
+                        state.log(f"   ⚠️ ATLAS (Benchmark Advisor) failed: {str(e)[:120]}")
+                        return BenchmarkAdvice()
 
-                # Best Practices Validator
-                try:
-                    best_practices = await best_practices_agent.analyze(
-                        rule_engine_results=rule_results,
-                        log_callback=state.log,
-                    )
-                    # Log what the LLM contributed
-                    if best_practices:
-                        state.log(f"   ├─ Overall Score: {best_practices.overall_score}/100")
-                        passing_count = len(best_practices.passing_practices) if best_practices.passing_practices else 0
-                        failing_count = len(best_practices.failing_practices) if best_practices.failing_practices else 0
-                        state.log(f"   ├─ Passing: {passing_count} | Failing: {failing_count}")
-                        # Show checks dict
-                        if best_practices.checks:
-                            for check_name, check_data in list(best_practices.checks.items())[:3]:
-                                if isinstance(check_data, dict):
-                                    status = check_data.get('status', '?')
-                                    note = check_data.get('note', '')[:50]
-                                    icon = "✅" if status == "pass" else "⚠️" if status == "warn" else "❌"
-                                    state.log(f"   │  {icon} {check_name}: {note}")
+                async def _run_sentinel():
+                    """Run SENTINEL (Best Practices Validator) agent."""
+                    try:
+                        result = await best_practices_agent.analyze(
+                            rule_engine_results=rule_results,
+                            log_callback=state.log,
+                        )
+                        if result:
+                            state.log(f"   ├─ Overall Score: {result.overall_score}/100")
+                            passing_count = len(result.passing_practices) if result.passing_practices else 0
+                            failing_count = len(result.failing_practices) if result.failing_practices else 0
+                            state.log(f"   ├─ Passing: {passing_count} | Failing: {failing_count}")
+                            if result.checks:
+                                for check_name, check_data in list(result.checks.items())[:3]:
+                                    if isinstance(check_data, dict):
+                                        status = check_data.get('status', '?')
+                                        note = check_data.get('note', '')[:50]
+                                        icon = "✅" if status == "pass" else "⚠️" if status == "warn" else "❌"
+                                        state.log(f"   │  {icon} {check_name}: {note}")
+                                    else:
+                                        state.log(f"   │  • {check_name}: {check_data}")
+                            if result.priority_fixes:
+                                top_fix = result.priority_fixes[0]
+                                if isinstance(top_fix, dict):
+                                    state.log(f"   └─ Top Fix: {top_fix.get('issue', top_fix.get('action', str(top_fix)[:60]))}")
                                 else:
-                                    state.log(f"   │  • {check_name}: {check_data}")
-                        # Show priority fixes
-                        if best_practices.priority_fixes:
-                            top_fix = best_practices.priority_fixes[0]
-                            if isinstance(top_fix, dict):
-                                state.log(f"   └─ Top Fix: {top_fix.get('issue', top_fix.get('action', str(top_fix)[:60]))}")
-                            else:
-                                state.log(f"   └─ Top Fix: {str(top_fix)[:60]}")
-                except Exception as e:
-                    state.log(f"   ⚠️ Best Practices Validator failed: {str(e)[:120]}")
-                    best_practices = BestPracticesResult(overall_score=rule_results.consistency_score)
+                                    state.log(f"   └─ Top Fix: {str(top_fix)[:60]}")
+                        return result
+                    except Exception as e:
+                        state.log(f"   ⚠️ SENTINEL (Best Practices) failed: {str(e)[:120]}")
+                        return BestPracticesResult(overall_score=rule_results.consistency_score)
+
+                # Execute AURORA + ATLAS + SENTINEL in parallel
+                import asyncio
+                state.log("")
+                state.log("   🚀 Running 3 agents in parallel: AURORA | ATLAS | SENTINEL")
+                brand_result, benchmark_advice, best_practices = await asyncio.gather(
+                    _run_aurora(),
+                    _run_atlas(),
+                    _run_sentinel(),
+                )
             else:
                 # No HF client - use defaults
                 state.log("   └─ Using default values (no LLM)")
@@ -2135,7 +2145,7 @@ def get_detected_fonts() -> dict:
         if weight:
             try:
                 weights.add(int(weight))
-            except:
+            except (ValueError, TypeError):
                 pass
     
     primary = max(fonts.items(), key=lambda x: x[1])[0] if fonts else "Unknown"
@@ -2160,7 +2170,7 @@ def get_base_font_size() -> int:
             size = float(size_str)
             if 14 <= size <= 18:
                 sizes[size] = sizes.get(size, 0) + t.frequency
-        except:
+        except (ValueError, TypeError):
             pass
     
     if sizes:
@@ -2501,7 +2511,7 @@ def format_llm_color_recommendations_table(final_recs: dict, semantic_analysis:
                 old_contrast = get_contrast_with_white(current)
                 new_contrast = get_contrast_with_white(suggested)
                 contrast_str = f"{old_contrast:.1f} → {new_contrast:.1f}"
-            except:
+            except (ValueError, TypeError, ZeroDivisionError):
                 contrast_str = "?"
             
             rows.append([
@@ -2551,7 +2561,7 @@ def format_typography_comparison_viewport(normalized_tokens, base_size: int, vie
         size_str = str(t.font_size).replace('px', '').replace('rem', '').replace('em', '')
         try:
             return float(size_str)
-        except:
+        except (ValueError, TypeError):
             return 16
     
     current_typo.sort(key=lambda t: -parse_size(t))
@@ -2694,7 +2704,7 @@ def format_radius_with_tokens() -> str:
         val = str(r.value).replace('px', '').replace('%', '')
         try:
             return float(val)
-        except:
+        except (ValueError, TypeError):
             return 999
     
     radii.sort(key=lambda r: parse_radius(r))
@@ -2790,6 +2800,21 @@ def snap_to_grid(value: float, base: int) -> int:
     return round(value / base) * base
 
 
+def reset_to_original():
+    """Reset all upgrade selections to defaults."""
+    state.selected_upgrades = {}
+    state.log("")
+    state.log("↩️ Reset all upgrade selections to original values.")
+
+    return (
+        "Scale 1.25 (Major Third) ⭐",     # type_scale_radio
+        "8px Base Grid ⭐",                  # spacing_radio
+        True,                                 # color_ramps_checkbox
+        "## ↩️ Reset Complete\n\nAll selections reverted to defaults. Review and apply again when ready.",  # apply_status
+        state.get_logs(),                     # stage2_log
+    )
+
+
 def apply_selected_upgrades(type_choice: str, spacing_choice: str, apply_ramps: bool, color_recs_table: list = None):
     """Apply selected upgrade options including LLM color recommendations."""
     if not state.upgrade_recommendations:
@@ -3126,7 +3151,7 @@ def export_tokens_json():
                                 "type": "color",
                                 "source": "upgraded" if shade != "500" else "detected",
                             }
-                except:
+                except (ValueError, KeyError, TypeError, IndexError):
                     result["colors"][clean_name] = {
                         "value": c.value,
                         "type": "color",
@@ -4312,7 +4337,7 @@ def create_ui():
         # STAGE 3: EXPORT
         # =================================================================
         
-        with gr.Accordion("📦 Stage 3: Export", open=False):
+        with gr.Accordion("📦 Stage 3: Export", open=False) as stage3_accordion:
             gr.Markdown("Export your finalized design tokens as JSON, compatible with **Figma Tokens Studio**.",
                         elem_classes=["section-desc"])
             gr.Markdown("""
@@ -4416,6 +4441,15 @@ Copy the JSON output below or save it as a `.json` file for import into Figma.
             fn=apply_selected_upgrades,
             inputs=[type_scale_radio, spacing_radio, color_ramps_checkbox, color_recommendations_table],
             outputs=[apply_status, stage2_log],
+        ).then(
+            fn=lambda: gr.update(open=True),
+            outputs=[stage3_accordion],
+        )
+
+        # Stage 2: Reset to original
+        reset_btn.click(
+            fn=reset_to_original,
+            outputs=[type_scale_radio, spacing_radio, color_ramps_checkbox, apply_status, stage2_log],
         )
         
         # Stage 1: Download JSON

requirements.txt

@@ -56,6 +56,8 @@ aiofiles>=23.0.0
 rich>=13.0.0
 tqdm>=4.66.0
 python-slugify>=8.0.0
+loguru>=0.7.0
+jsonschema>=4.20.0
 
 # -----------------------------------------------------------------------------
 # Testing (development only)