"""Gradio event handlers for MorphSQL.""" from __future__ import annotations import json import re import tempfile import zipfile from pathlib import Path import plotly.graph_objects as go from morphsql.assistant.copilot import MigrationCopilot from morphsql.dbt_generator.decomposer import decompose_to_dbt, format_dbt_project, is_dbt_target from morphsql.intelligence.lineage_viz import lineage_to_plotly from morphsql.intelligence.rationalization import generate_rationalization from morphsql.intelligence.runbook import generate_executive_summary, generate_runbook from morphsql.lineage.builder import build_lineage_graph from morphsql.models import Dialect, MigrationObject, MigrationReport, ObjectType from morphsql.pipeline import MigrationPipeline from morphsql.translator.engine import translate_sql from morphsql.translator.pandas_codegen import is_pandas_target from morphsql.translator.pyspark_codegen import is_pyspark_target from demo.theme import C_MUTED, C_PANEL, C_TEXT EXAMPLES_DIR = Path(__file__).parent.parent / "examples" / "vertica_legacy" _copilot = MigrationCopilot() def _sanitize_project(name: str) -> str: return re.sub(r"[^a-zA-Z0-9_]", "_", name.lower()).strip("_") or "migration_project" def _apply_dark_layout(fig: go.Figure, height: int = 260) -> go.Figure: fig.update_layout( template="plotly_dark", height=height, margin=dict(t=40, b=40, l=30, r=20), paper_bgcolor=C_PANEL, plot_bgcolor=C_PANEL, font={"color": C_TEXT, "family": "Inter, system-ui, sans-serif"}, xaxis={"gridcolor": "#334155", "zerolinecolor": "#334155", "color": C_TEXT}, yaxis={"gridcolor": "#334155", "zerolinecolor": "#334155", "color": C_TEXT}, ) return fig def _risk_gauge(score: int) -> go.Figure: score = max(0, min(100, int(score or 0))) color = "#22c55e" if score < 30 else "#eab308" if score < 60 else "#ef4444" label = "Low" if score < 30 else "Medium" if score < 60 else "High" fig = go.Figure( go.Indicator( mode="gauge+number", value=score, number={"suffix": " / 100", "font": {"color": C_TEXT, "size": 28}}, title={"text": f"Portfolio risk · {label}", "font": {"color": C_MUTED, "size": 13}}, gauge={ "axis": {"range": [0, 100], "tickcolor": C_MUTED, "tickfont": {"color": C_MUTED}}, "bar": {"color": color, "thickness": 0.8}, "bgcolor": "#0f172a", "bordercolor": "#334155", "steps": [ {"range": [0, 30], "color": "#14532d"}, {"range": [30, 60], "color": "#713f12"}, {"range": [60, 100], "color": "#7f1d1d"}, ], }, ) ) return _apply_dark_layout(fig, height=280) def _distribution_chart(report: MigrationReport) -> go.Figure: d = report.dashboard fig = go.Figure( go.Bar( x=["Auto-migrate", "Review", "Redesign", "Retire"], y=[ d.auto_migratable, d.requires_review, d.requires_redesign, d.recommended_retirement, ], marker_color=["#22c55e", "#eab308", "#ef4444", "#64748b"], text=[ d.auto_migratable, d.requires_review, d.requires_redesign, d.recommended_retirement, ], textposition="outside", textfont={"color": C_TEXT, "size": 14}, ) ) fig.update_layout(title={"text": "Object distribution", "font": {"color": C_MUTED, "size": 13}}) return _apply_dark_layout(fig, height=280) def _distribution_chart_empty() -> go.Figure: fig = go.Figure( go.Bar( x=["Auto-migrate", "Review", "Redesign", "Retire"], y=[0, 0, 0, 0], marker_color=["#22c55e", "#eab308", "#ef4444", "#64748b"], text=[0, 0, 0, 0], textposition="outside", textfont={"color": C_TEXT}, ) ) fig.update_layout( title={"text": "Object distribution (run scan)", "font": {"color": C_MUTED, "size": 13}}, yaxis={"range": [0, 5]}, ) return _apply_dark_layout(fig, height=280) def figure_to_html(fig: go.Figure) -> str: """Reliable Plotly HTML embed for Gradio (avoids blank Plot widgets).""" return fig.to_html( include_plotlyjs="cdn", full_html=False, config={"displayModeBar": False, "responsive": True}, ) def empty_lineage_plot() -> go.Figure: return lineage_to_plotly(build_lineage_graph([], Dialect.VERTICA)) def _resolve_repo_path(upload_file, use_sample: bool) -> Path | None: if use_sample: return EXAMPLES_DIR if EXAMPLES_DIR.exists() else None if not upload_file: return None # Gradio File may return str path, Path, or tempfile-like object if isinstance(upload_file, (list, tuple)) and upload_file: upload_file = upload_file[0] path = Path(str(upload_file.name if hasattr(upload_file, "name") else upload_file)) if not path.exists(): return None if path.suffix.lower() == ".zip": tmp = Path(tempfile.mkdtemp(prefix="morphsql_")) with zipfile.ZipFile(path) as zf: zf.extractall(tmp) return tmp if path.is_dir(): return path return None def _report_to_dict(report: MigrationReport | None) -> dict | None: if report is None: return None return report.model_dump(mode="json") def _dict_to_report(data: dict | MigrationReport | None) -> MigrationReport | None: if data is None: return None if isinstance(data, MigrationReport): return data if isinstance(data, dict): try: return MigrationReport.model_validate(data) except Exception: return None return None def run_migration_workbench( upload_file, use_sample: bool, source: str, target: str, ) -> tuple: """Full migration intelligence pipeline for workbench tab.""" empty_msg = ( "Enable **Use sample repository** or upload a zip file, " "then click **Run migration intelligence**." ) empty_risk = _risk_gauge(0) empty_dist = _distribution_chart_empty() empty_lineage = empty_lineage_plot() empty = ( empty_msg, "No objects yet.", "No rationalization yet.", "No runbook yet.", "No dbt preview yet.", "No validation results yet.", "Portfolio risk: —", empty_risk, empty_dist, empty_lineage, json.dumps({"status": "waiting_for_input"}, indent=2), None, ) try: repo_path = _resolve_repo_path(upload_file, bool(use_sample)) except Exception as exc: return ( f"Failed to open repository: {exc}", *empty[1:], ) if repo_path is None: return empty try: source_d = Dialect(source) wants_dbt = is_dbt_target(target) if wants_dbt: target_d = Dialect.SNOWFLAKE elif is_pandas_target(target): target_d = Dialect.PANDAS elif is_pyspark_target(target): target_d = Dialect.PYSPARK else: target_d = Dialect(target) pipeline = MigrationPipeline(source=source_d, target=target_d) report = pipeline.analyze(str(repo_path)) report = pipeline.convert(report) report = pipeline.validate(report) if wants_dbt: report.target_dialect = Dialect.DBT_SNOWFLAKE exec_summary = generate_executive_summary(report) runbook = generate_runbook(report) rationalization = generate_rationalization(report) dbt_preview = "No suitable object for dbt decomposition." candidates = sorted(report.objects, key=lambda o: o.complexity_score, reverse=True) dbt_parts = ["### dbt project preview (Snowflake)\n"] shown = 0 for obj in candidates: if obj.object_type.value not in ("stored_procedure", "sql_script", "view"): continue files = decompose_to_dbt(obj, source_d, project_name=_sanitize_project(obj.name)) dbt_parts.append(f"#### `{obj.name}` → {len(files)} files\n") # Show model SQL files first model_files = [p for p in sorted(files) if p.startswith("models/") and p.endswith(".sql")] for rel in model_files[:5]: dbt_parts.append(f"**{rel}**\n```sql\n{files[rel][:1200]}\n```\n") shown += 1 if shown >= 2: break if shown: dbt_preview = "\n".join(dbt_parts) elif not wants_dbt: dbt_preview = ( "Target is not **dbt-snowflake**. " "Select **dbt-snowflake** as the target to generate staging / intermediate / mart models, " "or open Architecture (dbt) after a dbt-snowflake run." ) val_lines = ["### Validation results\n"] passed = sum(1 for r in report.validation_results if r.passed) total = len(report.validation_results) val_lines.append(f"**{passed}/{total}** checks passed\n") for r in report.validation_results[:20]: status = "PASS" if r.passed else "FAIL" val_lines.append(f"- **[{status}]** {r.object_name} · {r.check_name}") if r.root_cause: val_lines.append(f" - {r.root_cause}") if r.recommendation and not r.passed: val_lines.append(f" - Fix: {r.recommendation}") validation_md = "\n".join(val_lines) if total else "No validation checks generated." obj_lines = [ "### Discovered objects\n", "| Object | Type | Complexity | Risk | Confidence | Category |", "|--------|------|------------|------|------------|----------|", ] for obj in report.objects: obj_lines.append( f"| {obj.name} | {obj.object_type.value} | {obj.complexity_score} " f"| {obj.risk_level.value} | {obj.conversion_confidence:.0f}% " f"| {obj.migration_category.value.replace('_', ' ')} |" ) if not report.objects: obj_lines.append("| — | — | — | — | — | No objects found |") metrics_md = ( f"**Portfolio risk:** {int(report.dashboard.migration_risk_score)} / 100 · " f"**Objects:** {report.dashboard.total_objects} · " f"**Auto-migrate:** {report.dashboard.auto_migratable} · " f"**Review:** {report.dashboard.requires_review} · " f"**Redesign:** {report.dashboard.requires_redesign} · " f"**Retire:** {report.dashboard.recommended_retirement}" ) graph = build_lineage_graph(report.objects, source_d) lineage_fig = lineage_to_plotly(graph, report.objects) risk_fig = _risk_gauge(int(report.dashboard.migration_risk_score)) dist_fig = _distribution_chart(report) export = json.dumps( { "repository": str(repo_path), "source": source, "target": target, "dashboard": report.dashboard.model_dump(mode="json"), "objects": [ { "name": o.name, "type": o.object_type.value, "complexity": o.complexity_score, "risk": o.risk_level.value, "confidence": o.conversion_confidence, "category": o.migration_category.value, } for o in report.objects ], }, indent=2, ) return ( exec_summary + "\n\n" + metrics_md, "\n".join(obj_lines), rationalization, runbook, dbt_preview, validation_md, metrics_md, risk_fig, dist_fig, lineage_fig, export, _report_to_dict(report), ) except Exception as exc: return ( f"**Workbench error:** {type(exc).__name__}: {exc}", *empty[1:], ) def analyze_sql_object(sql: str, source: str, target: str) -> tuple[str, go.Figure, str, str, str]: """Assess + convert a single SQL object. Returns: analysis_md, risk_fig, badge, converted_sql_or_dbt, notes_md """ if not (sql or "").strip(): return ( "Paste SQL above, then click **Assess & Convert**.", _risk_gauge(0), "—", "", "Waiting for SQL input.", ) try: from morphsql.parser.sql_parser import count_sql_complexity, detect_unsupported_features from morphsql.risk.scorer import extract_business_rules, score_object from morphsql.validation.reconciliation import generate_incremental_strategy source_d = Dialect(source) wants_dbt = is_dbt_target(target) wants_pandas = is_pandas_target(target) wants_pyspark = is_pyspark_target(target) if wants_dbt: target_d = Dialect.SNOWFLAKE elif wants_pandas: target_d = Dialect.PANDAS elif wants_pyspark: target_d = Dialect.PYSPARK else: target_d = Dialect(target) # Infer object type for better dbt decomposition obj_type = ObjectType.SQL_SCRIPT if re.search(r"\bCREATE\s+(?:OR\s+REPLACE\s+)?PROCEDURE\b", sql, re.I): obj_type = ObjectType.STORED_PROCEDURE elif re.search(r"\bCREATE\s+(?:OR\s+REPLACE\s+)?VIEW\b", sql, re.I): obj_type = ObjectType.VIEW score_target = ( Dialect.SNOWFLAKE if (wants_pandas or wants_pyspark) else target_d ) obj = MigrationObject(name="input_object", object_type=obj_type, source_sql=sql) obj = score_object(obj, source_d, score_target) complexity = count_sql_complexity(sql, source_d) unsupported = detect_unsupported_features(sql, source_d, score_target) rules = extract_business_rules(sql) incremental = generate_incremental_strategy(sql) converted, conf, auto, review = translate_sql(sql, source_d, target_d) obj.target_sql = converted obj.conversion_confidence = conf obj.auto_converted = auto obj.requires_review = review output_sql = converted dbt_note = "" if wants_dbt: files = decompose_to_dbt(obj, source_d, project_name="input_object") output_sql = format_dbt_project(files) dbt_note = ( f"**dbt project generated:** {len(files)} files " f"({len(obj.dbt_models)} models) — staging / intermediate / marts\n\n" ) lines = [ "### Object assessment", "", "| Field | Value |", "|-------|-------|", f"| Complexity | {obj.complexity_score} / 100 |", f"| Risk | {obj.risk_level.value.title()} |", f"| Category | {obj.migration_category.value.replace('_', ' ').title()} |", f"| Conversion confidence | {conf:.0f}% |", f"| Output | {'dbt models (Snowflake)' if wants_dbt else target} |", f"| Lines | {complexity.get('lines', 0)} |", f"| CTEs | {complexity.get('ctes', 0)} |", f"| Joins | {complexity.get('joins', 0)} |", f"| Temp tables | {complexity.get('temp_tables', 0)} |", f"| Window functions | {complexity.get('window_functions', 0)} |", "", ] if wants_dbt and obj.dbt_models: lines += ["**dbt models**", ""] + [f"- `{m}`" for m in obj.dbt_models] + [""] if obj.risk_factors: lines += ["**Risk factors**", ""] + [f"- {rf.description}" for rf in obj.risk_factors] + [""] if unsupported: lines += ["**Unsupported syntax**", ""] + [f"- {u}" for u in unsupported] + [""] if rules: lines += ["**Business rules**", ""] + [f"- {r}" for r in rules[:5]] + [""] if incremental: lines += ["**Incremental pattern**", ""] for k, v in incremental.items(): lines.append(f"- {k.replace('_', ' ').title()}: {v}") notes = [ f"**Route:** {source} → {target}", f"**Confidence:** {conf:.0f}%", "", ] if dbt_note: notes.append(dbt_note) if auto: notes.append("**SQL transformations**") notes.extend(f"- {a}" for a in auto[:10]) notes.append("") if review: notes.append("**Manual review**") notes.extend(f"- {r}" for r in review[:10]) badge = f"{obj.complexity_score} / 100 · {obj.risk_level.value.title()} · {conf:.0f}% conf" return ( "\n".join(lines), _risk_gauge(obj.complexity_score), badge, output_sql, "\n".join(notes), ) except Exception as exc: return ( f"**Assessment error:** {type(exc).__name__}: {exc}", _risk_gauge(0), "Error", "", str(exc), ) def get_sample_workbench() -> tuple: """Precompute sample workbench outputs so UI is never blank on load.""" return run_migration_workbench(None, True, "vertica", "snowflake") def copilot_chat( message: str, history: list | None, report_data: dict | MigrationReport | None, sql: str, source: str, target: str, ) -> tuple[list, str]: history = list(history or []) if not (message or "").strip(): return history, "" report = _dict_to_report(report_data) normalized: list[dict] = [] for turn in history: if isinstance(turn, dict) and "role" in turn and "content" in turn: normalized.append({"role": turn["role"], "content": str(turn["content"])}) elif isinstance(turn, (list, tuple)) and len(turn) == 2: if turn[0]: normalized.append({"role": "user", "content": str(turn[0])}) if turn[1]: normalized.append({"role": "assistant", "content": str(turn[1])}) reply = _copilot.respond(message.strip(), normalized, report, sql or "", source, target) normalized.append({"role": "user", "content": message.strip()}) normalized.append({"role": "assistant", "content": reply}) return normalized, "" def report_to_context(report_data: dict | MigrationReport | None) -> str: report = _dict_to_report(report_data) if report is None: return ( "No repository scan loaded yet. " "Open Migration Workbench, run a scan, then return here — " "the copilot will use that context." ) ctx = _copilot.build_context(report) # Avoid markdown code fences that can go white-on-white in Gradio themes safe = ctx[:2500].replace("`", "'") return f"### Active scan context\n\n{safe}" HERO_EXAMPLE = ( "SELECT customer_id, COALESCE(order_amount, 0) AS order_amount, " "COALESCE(discount, 0) AS discount FROM staging.orders " "WHERE order_date >= CURRENT_DATE - 30" ) FEATURE_SQL_PATH = Path(__file__).parent.parent / "examples" / "ml_features" / "churn_feature_sql.sql" # Human-readable labels (defined before convert helpers) # Snowflake first — most common warehouse for DS/AI users SOURCE_LABELS = { "snowflake": "Snowflake", "bigquery": "BigQuery", "redshift": "Redshift", "oracle": "Oracle", "vertica": "Vertica", } TARGET_LABELS = { "pandas": "Python (pandas)", "pyspark": "Python (PySpark)", "snowflake": "Snowflake SQL", "bigquery": "BigQuery SQL", "dbt-snowflake": "dbt project (Snowflake)", } SOURCE_DROPDOWN = [(label, key) for key, label in SOURCE_LABELS.items()] TARGET_DROPDOWN = [ ("Python (pandas) — recommended for notebooks", "pandas"), ("Python (PySpark) — Spark DataFrame API", "pyspark"), ("Snowflake SQL", "snowflake"), ("BigQuery SQL", "bigquery"), ("dbt project (Snowflake)", "dbt-snowflake"), ] def run_hero_agent(sql: str, source: str, target: str) -> tuple[str, str, str, str]: """Convert SQL and return (notes_md, output, status, share_md).""" from morphsql.risk.scorer import score_object sql = sql or HERO_EXAMPLE source_d = Dialect(source) wants_dbt = is_dbt_target(target) wants_pandas = is_pandas_target(target) wants_pyspark = is_pyspark_target(target) if wants_dbt: target_d = Dialect.SNOWFLAKE elif wants_pandas: target_d = Dialect.PANDAS elif wants_pyspark: target_d = Dialect.PYSPARK else: target_d = Dialect(target) converted, conf, auto, review = translate_sql(sql, source_d, target_d) score_target = ( Dialect.SNOWFLAKE if (wants_pandas or wants_pyspark) else target_d ) obj = MigrationObject( name="hero_query", object_type=ObjectType.SQL_SCRIPT, source_sql=sql, target_sql=converted, ) obj = score_object(obj, source_d, score_target) output = converted if wants_dbt: files = decompose_to_dbt(obj, source_d, project_name="morphsql_dbt") output = format_dbt_project(files, max_files=12) source_label = SOURCE_LABELS.get(source, source) target_label = TARGET_LABELS.get(target, target) if wants_pandas: kind = "Python (pandas)" next_steps = [ "1. Review the generated Python on the right.", "2. Check the **sample preview** (synthetic tables).", "3. Download the `.py` file or open the notebook starter cell.", "4. Point `tables['…']` at your real DataFrames (`read_parquet` / `read_sql`).", ] elif wants_pyspark: kind = "Python (PySpark)" next_steps = [ "1. Review the generated PySpark code on the right.", "2. Check the **sample preview** (same query logic on synthetic tables).", "3. Download the `.py` file or open the Spark notebook starter.", "4. Provide `tables['…']` or replace with `spark.table` / `spark.read`.", ] elif wants_dbt: kind = "dbt project" next_steps = [ "1. Copy the generated dbt files.", "2. Check the **sample preview** for query-shape sanity.", "3. Drop models into a dbt project and run `dbt run`.", "4. Review models marked for manual checks.", ] else: kind = f"{target_label} SQL" next_steps = [ "1. Copy the converted SQL on the right.", "2. Check the **sample preview** (same query logic on synthetic tables).", "3. Run the SQL on the target warehouse.", "4. Check any items under Needs review.", ] changes: list[str] = [] for a in auto[:8]: if any( k in a for k in ( "→", "Removed", "Converted", "JOIN", "WHERE", "GROUP", "SELECT", "pandas", "Dialect", "FROM dual", ) ): changes.append(f"- {a}") notes = [ f"### {source_label} → {target_label}", "", f"**{conf:.0f}%** confidence · output: **{kind}**", "", "**Next steps**", *next_steps, "", ] if changes: notes.append("**What changed**") notes.extend(changes) notes.append("") if review: notes.append("**Needs review**") notes.extend(f"- {r}" for r in review[:5]) notes.append("") status = f"{conf:.0f}% · {kind}" space_url = "https://huggingface.co/spaces/dgvj-work/morphsql" share = ( f"MorphSQL converted **{source_label} → {target_label}** " f"({conf:.0f}% confidence).\n\n" f"[Open Space]({space_url}) · " f"[GitHub](https://github.com/dgvj-work/morphsql)" ) return "\n".join(notes), output, status, share def _infer_columns_from_sql(sql: str) -> list[str]: cols: list[str] = [] # AS aliases cols.extend(re.findall(r"\bAS\s+([A-Za-z_][\w]*)", sql or "", flags=re.I)) # bare identifiers after SELECT / commas (best-effort) cols.extend(re.findall(r"(?:SELECT|,)\s*(?:[\w.]+\.)?([A-Za-z_][\w]*)\b", sql or "", flags=re.I)) # common warehouse columns used in demos cols.extend( [ "customer_id", "user_id", "order_amount", "discount", "order_date", "amount", "name", "id", "a", "b", "x", "dt", "dept", "event_value", "event_ts", "event_type", "value", "score", ] ) # de-dupe preserving order, drop SQL keywords skip = { "select", "from", "where", "group", "by", "order", "limit", "as", "and", "or", "on", "join", "left", "right", "inner", "outer", "case", "when", "then", "else", "end", "null", "not", "is", "in", "distinct", "count", "sum", "avg", "min", "max", "coalesce", "nvl", "zeroifnull", "ifnull", "current_date", "sysdate", "getdate", } out: list[str] = [] for c in cols: if c.lower() in skip: continue if c not in out: out.append(c) return out[:24] def build_sample_tables(code: str, sql: str = "") -> dict: """Build tiny demo DataFrames so generated pandas can execute in the Space.""" import pandas as pd keys = list(dict.fromkeys(re.findall(r"tables\[['\"]([^'\"]+)['\"]\]", code or ""))) if not keys and sql: # Fallback: FROM / JOIN identifiers in the source SQL keys = list( dict.fromkeys( re.findall( r"(?:FROM|JOIN)\s+([A-Za-z_][\w]*(?:\.[A-Za-z_][\w]*)?)", sql, flags=re.I, ) ) ) keys = [k for k in keys if k.lower() not in {"dual", "select"}] cols = _infer_columns_from_sql(sql) # Include columns referenced in generated code (e.g. procedure params used as cols) for c in re.findall(r"\[['\"]([^'\"]+)['\"]\]", code or ""): if "." in c or c in keys: continue # table keys like staging.orders if c not in cols: cols.append(c) if not cols: cols = ["id", "value"] n = 5 today = pd.Timestamp.today().normalize() frames: dict = {} for key in keys: data: dict = {} for i, col in enumerate(cols): cl = col.lower() if any(t in cl for t in ("date", "ts", "time", "dt")): data[col] = [today - pd.Timedelta(days=j) for j in range(n)] elif any(t in cl for t in ("id", "count", "nunique", "flag", "label")): data[col] = list(range(1, n + 1)) elif any(t in cl for t in ("amount", "value", "score", "avg", "sum", "discount")): data[col] = [None if j == 0 else float(10 * (j + 1)) for j in range(n)] elif any(t in cl for t in ("name", "type", "dept")): data[col] = [f"item_{j}" for j in range(n)] else: data[col] = [j + 1 for j in range(n)] frames[key] = pd.DataFrame(data) return frames def _preview_code_for_target(output: str, target: str, sql: str, source: str) -> tuple[str, str]: """ Return (python_code, via_label) used to produce a sample DataFrame preview. Pandas targets execute the generated code. All other targets reuse the same source SQL → pandas path so preview works without Spark / a warehouse. """ from morphsql.translator.pandas_codegen import sql_to_pandas code = output or "" if is_pandas_target(target) and "import pandas" in code: return code, "generated pandas" if not (sql or "").strip(): return "", "" try: source_d = Dialect(source) except ValueError: source_d = Dialect.SNOWFLAKE preview_code, _conf, _auto, _review = sql_to_pandas(sql, source_d) label = TARGET_LABELS.get(target, target) return preview_code, f"pandas runtime · same query logic (output above is {label})" def run_sample_preview( output: str, target: str, sql: str = "", source: str = "snowflake", ) -> tuple: """ Execute query logic against synthetic tables for a sample DataFrame preview. Works for every Convert target: pandas runs the generated code; PySpark / SQL / dbt reuse a pandas translation of the source SQL so the Space can show rows without Spark or a warehouse connection. Returns (dataframe_or_none, note_md). """ import pandas as pd preview_code, via = _preview_code_for_target(output, target, sql, source) if not preview_code or "import pandas" not in preview_code: return None, "_Could not build a sample preview for this input._" tables = build_sample_tables(preview_code, sql) if not tables and "tables[" in preview_code: return None, "_Could not infer input tables for a live preview._" if not tables: tables = {} ns: dict = {"pd": pd, "np": __import__("numpy"), "tables": tables} try: exec(preview_code, ns, ns) # noqa: S102 — intentional demo sandbox for generated code except Exception as exc: return None, f"_Preview could not run automatically:_ `{exc}`" result = ns.get("result") if not isinstance(result, pd.DataFrame): # Procedures may leave the last SELECT under result_N for key, val in reversed(list(ns.items())): if key.startswith("result") and isinstance(val, pd.DataFrame): result = val break if not isinstance(result, pd.DataFrame): return None, "_Preview ran, but no `result` DataFrame was produced._" note = ( f"**Sample preview** · {via} on {len(tables)} synthetic table(s) " f"→ `{result.shape[0]}` rows × `{result.shape[1]}` cols. " "Replace inputs with your real data when you run the converted output." ) return result.head(20), note def write_output_download( output: str, target: str, stem: str | None = None, ) -> str: """Write converted output to a downloadable temp file; return path.""" safe = re.sub(r"[^\w.\-]+", "_", (stem or "morphsql").strip())[:80] or "morphsql" if is_pandas_target(target): suffix, kind = ".py", "pandas" elif is_pyspark_target(target): suffix, kind = ".py", "pyspark" elif is_dbt_target(target): suffix, kind = ".txt", "dbt" else: suffix, kind = ".sql", str(target).replace("-", "_") path = Path(tempfile.gettempdir()) / f"{safe}_{kind}{suffix}" # Avoid collisions when converting multiple files in one session if path.exists(): path = Path(tempfile.gettempdir()) / f"{safe}_{kind}_{abs(hash(output)) % 100_000}{suffix}" path.write_text(output or "", encoding="utf-8") return str(path) def _resolve_upload_path(upload_file) -> Path | None: if not upload_file: return None if isinstance(upload_file, (list, tuple)) and upload_file: upload_file = upload_file[0] path = Path(str(getattr(upload_file, "name", upload_file))) return path if path.exists() else None def _collect_sql_files(upload_file) -> list[Path]: """Return SQL/text files from an upload (.sql/.txt or .zip of those).""" path = _resolve_upload_path(upload_file) if path is None: return [] suffix = path.suffix.lower() if suffix in {".sql", ".txt", ".ddl", ".prc", ".pkb", ".pks"}: return [path] if suffix == ".zip": tmp = Path(tempfile.mkdtemp(prefix="morphsql_upload_")) with zipfile.ZipFile(path) as zf: zf.extractall(tmp) files = sorted( p for p in tmp.rglob("*") if p.is_file() and p.suffix.lower() in {".sql", ".txt", ".ddl", ".prc"} ) return files # Unknown extension — try reading as text SQL return [path] def load_sql_from_upload(upload_file, current_sql: str = "") -> str: """Load uploaded SQL into the input box (first file if a zip). Keeps current text if empty.""" files = _collect_sql_files(upload_file) if not files: return current_sql or "" try: return files[0].read_text(encoding="utf-8", errors="replace") except OSError: return current_sql or "" def convert_upload_for_ui(upload_file, sql: str, source: str, target: str): """ Convert pasted SQL or an uploaded .sql/.txt/.zip. Single file → normal Convert payload with a named download. Zip with multiple SQL files → convert each and return a zip download. Returns: (sql_in, notes, output, status, share, preview, download, notebook, api) """ files = _collect_sql_files(upload_file) if not files: text = (sql or "").strip() if not text: empty = convert_for_ui("", source, target) return ("", *empty) notes, output, status, share, preview, download, nb, api = convert_for_ui( text, source, target ) return text, notes, output, status, share, preview, download, nb, api if len(files) == 1: text = files[0].read_text(encoding="utf-8", errors="replace") notes, output, status, share, preview, _old_dl, nb, api = convert_for_ui( text, source, target ) download = write_output_download(output, target, stem=files[0].stem) note = ( f"\n**Upload** · converted `{files[0].name}` → download " f"`{Path(download).name}`.\n" ) return text, notes + note, output, status, share, preview, download, nb, api # Multi-file zip → convert each and package out_dir = Path(tempfile.mkdtemp(prefix="morphsql_batch_")) converted_paths: list[Path] = [] previews = [] combined_notes = [ f"### Batch upload · {len(files)} SQL file(s) → **{TARGET_LABELS.get(target, target)}**", "", ] first_output = "" first_sql = "" statuses: list[str] = [] for fp in files: try: text = fp.read_text(encoding="utf-8", errors="replace") except OSError as exc: combined_notes.append(f"- `{fp.name}`: read error ({exc})") continue if not first_sql: first_sql = text notes, output, status, _share, preview, _dl, _nb, _api = convert_for_ui( text, source, target ) if not first_output: first_output = output if preview is not None: previews.append(preview) if is_pandas_target(target): dest = out_dir / f"{fp.stem}_pandas.py" elif is_pyspark_target(target): dest = out_dir / f"{fp.stem}_pyspark.py" elif is_dbt_target(target): dest = out_dir / f"{fp.stem}_dbt.txt" else: dest = out_dir / f"{fp.stem}_{target.replace('-', '_')}.sql" dest.write_text(output or "", encoding="utf-8") converted_paths.append(dest) statuses.append(status) combined_notes.append(f"- `{fp.name}` → `{dest.name}` · {status}") zip_path = Path(tempfile.gettempdir()) / f"morphsql_batch_{target.replace('-', '_')}.zip" with zipfile.ZipFile(zip_path, "w", compression=zipfile.ZIP_DEFLATED) as zf: for p in converted_paths: zf.write(p, arcname=p.name) preview = previews[0] if previews else None status = statuses[0] if statuses else "Batch convert" share = ( f"Converted **{len(converted_paths)}** file(s) to `{target}`. " f"Download the zip and drop files into your notebook / warehouse / dbt project.\n\n" f"[Space](https://huggingface.co/spaces/dgvj-work/morphsql) · " f"[GitHub](https://github.com/dgvj-work/morphsql)" ) nb = notebook_cell(first_output, target) api = hf_pipeline_snippet(first_sql or HERO_EXAMPLE, source, target) notes = "\n".join(combined_notes) + ( f"\n\n**Download** the zip (`{zip_path.name}`) for all converted files.\n" ) return ( first_sql, notes, first_output or "(see zip for all converted files)", status, share, preview, str(zip_path), nb, api, ) def notebook_cell(output: str, target: str) -> str: """Short copy-paste cell for Jupyter / Colab / Databricks.""" if is_pyspark_target(target): return ( "# MorphSQL → PySpark (paste into a Databricks / Spark notebook)\n" "from pyspark.sql import SparkSession, functions as F, Window\n\n" "# spark = SparkSession.builder.getOrCreate()\n" "# 1) Provide input frames (or replace with spark.table / spark.read)\n" "# tables = {\n" "# 'staging.orders': spark.table('staging.orders'),\n" "# }\n\n" "# 2) Paste the generated MorphSQL code below (or %run the downloaded .py)\n" "# 3) Use `result` as your Spark DataFrame\n" ) if not is_pandas_target(target): return ( "# Output is SQL/dbt — paste into your warehouse client or dbt project.\n" "# Tip: switch Convert to → Python (pandas) or Python (PySpark).\n" ) return ( "# MorphSQL → pandas (paste into a notebook cell)\n" "import pandas as pd\n" "import numpy as np\n\n" "# 1) Load your real data (parquet / sql / HF datasets)\n" "# tables = {\n" "# 'staging.orders': pd.read_parquet('orders.parquet'),\n" "# }\n\n" "# 2) Paste the generated MorphSQL code below (or %run the downloaded .py)\n" "# 3) Use `result` as features for sklearn / XGBoost / embeddings pipelines\n" ) def hf_pipeline_snippet(sql: str, source: str, target: str) -> str: """Hugging Face-style API snippet for AI / ML users.""" sql_short = (sql or "").strip().replace("\\", "\\\\").replace('"""', "'''") if len(sql_short) > 180: sql_short = sql_short[:177] + "..." return ( "from morphsql.ai import pipeline\n\n" "# Same API style as transformers.pipeline\n" 'pipe = pipeline("sql-migration")\n' "out = pipe(\n" f' """{sql_short}""",\n' f' source="{source}",\n' f' target="{target}",\n' ")\n" 'print(out["converted_sql"][:500]) # pandas code or SQL\n' 'print(out.get("risk")) # optional risk head\n' ) def convert_for_ui(sql: str, source: str, target: str): """Full Convert-tab payload for the Space UI.""" notes, output, status, share = run_hero_agent(sql, source, target) preview, preview_note = run_sample_preview( output, target, sql=sql or "", source=source or "snowflake" ) download = write_output_download(output, target) nb = notebook_cell(output, target) api = hf_pipeline_snippet(sql or HERO_EXAMPLE, source, target) if preview_note: notes = notes + "\n" + preview_note + "\n" return notes, output, status, share, preview, download, nb, api PLAYGROUND_EXAMPLE_LABELS = [ "DS: Snowflake orders → pandas (fillna / filters)", "DS: Snowflake orders → PySpark (F.coalesce / filter)", "AI: Feature aggregates → pandas (training features)", "DS: Vertica ZEROIFNULL → pandas", "DS: Oracle dual constants → pandas", "DS: Redshift window slice → pandas", "DS: BigQuery null handling → pandas", "Warehouse: Vertica → Snowflake SQL", "Warehouse: Vertica procedure → dbt", ] PLAYGROUND_EXAMPLES = [ [ "SELECT customer_id, COALESCE(order_amount, 0) AS order_amount, COALESCE(discount, 0) AS discount FROM staging.orders WHERE order_date >= CURRENT_DATE - 30", "snowflake", "pandas", ], [ "SELECT customer_id, COALESCE(order_amount, 0) AS order_amount, COALESCE(discount, 0) AS discount FROM staging.orders WHERE order_date >= CURRENT_DATE - 30", "snowflake", "pyspark", ], [ "SELECT user_id, COUNT(*) AS event_count_90d, SUM(ZEROIFNULL(event_value)) AS value_sum_90d, AVG(ZEROIFNULL(event_value)) AS value_avg_90d FROM staging.product_events WHERE event_ts >= CURRENT_DATE - 90 GROUP BY user_id", "vertica", "pandas", ], [ "SELECT customer_id, ZEROIFNULL(order_amount) AS order_amount, NVL(discount, 0) AS discount FROM staging.orders WHERE order_date >= CURRENT_DATE - 30", "vertica", "pandas", ], [ "SELECT NVL(amount, 0) AS amount, SYSDATE AS ts FROM dual", "oracle", "pandas", ], [ "SELECT GETDATE() AS ts, name FROM users WHERE id > 1 LIMIT 5", "redshift", "pandas", ], [ "SELECT IFNULL(a, 0) AS a, b FROM t WHERE a IS NOT NULL", "bigquery", "pandas", ], [ "SELECT customer_id, ZEROIFNULL(order_amount) AS order_amount FROM staging.orders", "vertica", "snowflake", ], [ "CREATE OR REPLACE PROCEDURE p(load_date DATE) AS $$ BEGIN CREATE LOCAL TEMP TABLE tmp ON COMMIT PRESERVE ROWS AS SELECT customer_id, ZEROIFNULL(amount) AS amount FROM staging.orders WHERE order_date = load_date; INSERT INTO analytics.daily SELECT * FROM tmp; END; $$;", "vertica", "dbt-snowflake", ], ] def load_playground_example(index: int) -> tuple[str, str, str]: """Return (sql, source, target) for a preset example index.""" idx = max(0, min(int(index), len(PLAYGROUND_EXAMPLES) - 1)) sql, source, target = PLAYGROUND_EXAMPLES[idx] return sql, source, target def load_and_convert_example(index: int) -> tuple[str, str, str, str, str, str, str]: """Load a preset and immediately convert (legacy 7-tuple).""" sql, source, target = load_playground_example(index) explain, output, badge, share = run_hero_agent(sql, source, target) return sql, source, target, explain, output, badge, share def on_example_selected(label: str | None): """Dropdown helper used by the Convert tab (extended UI payload).""" if not label or label not in PLAYGROUND_EXAMPLE_LABELS: label = PLAYGROUND_EXAMPLE_LABELS[0] sql, source, target = load_playground_example(PLAYGROUND_EXAMPLE_LABELS.index(label)) notes, output, status, share, preview, download, nb, api = convert_for_ui( sql, source, target ) return sql, source, target, notes, output, status, share, preview, download, nb, api AGENT_PROMPTS = [ ( "Convert this SQL to pandas", PLAYGROUND_EXAMPLES[0][0], "snowflake", "pandas", ), ( "Convert this SQL to PySpark", PLAYGROUND_EXAMPLES[1][0], "snowflake", "pyspark", ), ( "Convert feature SQL to pandas", PLAYGROUND_EXAMPLES[2][0], "vertica", "pandas", ), ( "Emit a dbt project from this procedure", PLAYGROUND_EXAMPLES[8][0], "vertica", "dbt-snowflake", ), ] def load_agent_example(index: int) -> tuple[str, str, str, str]: idx = max(0, min(int(index), len(AGENT_PROMPTS) - 1)) return AGENT_PROMPTS[idx] def run_eval_suite(limit: int, category: str) -> tuple[str, str, dict]: """Run eval suite and return markdown summary, detail table, metrics dict.""" from morphsql.eval.metrics import run_eval from morphsql.eval.pairs import ensure_pairs_file ensure_pairs_file() cats = None if category in ("all", "", None) else [category] limit_i = int(limit) if limit else 50 results, summary = run_eval(limit=limit_i, categories=cats) lines = [ "### Eval suite results", "", f"| Metric | Score |", f"|--------|-------|", f"| Pairs | {summary['n_pairs']} |", f"| Exact match | {100 * summary['exact_match']:.1f}% |", f"| Token F1 | {100 * summary['token_f1']:.1f}% |", f"| Fuzzy (Dice) | {100 * summary['fuzzy']:.1f}% |", f"| Pass rate | {100 * summary['pass_rate']:.1f}% |", "", "#### By category", "", "| Category | N | Exact | Token F1 | Pass |", "|----------|---|-------|----------|------|", ] for cat, stats in summary.get("by_category", {}).items(): lines.append( f"| {cat} | {stats['n']} | {100 * stats['exact_match']:.0f}% " f"| {100 * stats['token_f1']:.0f}% | {100 * stats['pass_rate']:.0f}% |" ) detail = [ "### Sample predictions", "", "| ID | Pass | F1 | Exact |", "|----|------|----|-------|", ] for r in results[:25]: detail.append( f"| {r.pair_id} | {'yes' if r.passed else 'no'} " f"| {100 * r.token_f1:.0f}% | {100 * r.exact_match:.0f}% |" ) return "\n".join(lines), "\n".join(detail), summary def submit_eval_score(name: str, summary: dict | None) -> str: from morphsql.eval.leaderboard import format_leaderboard_md, submit_score if not summary or not summary.get("n_pairs"): return format_leaderboard_md() + "\n\n_Run the eval suite before submitting._" board = submit_score( name=name or "anonymous", exact_match=summary.get("exact_match", 0), token_f1=summary.get("token_f1", 0), fuzzy=summary.get("fuzzy", 0), pass_rate=summary.get("pass_rate", 0), n_pairs=summary.get("n_pairs", 0), notes="MorphSQL hybrid translator", ) return format_leaderboard_md(board) def run_behavior_rag(query: str, source: str, target: str) -> str: from morphsql.intelligence.rag import get_rag return get_rag().answer(query or "NULL empty string timezone", source, target) def run_feature_migration(target: str) -> tuple[str, str]: """Convert ML feature SQL and optionally emit dbt feature mart.""" sql = FEATURE_SQL_PATH.read_text(encoding="utf-8") if FEATURE_SQL_PATH.exists() else HERO_EXAMPLE wants_dbt = is_dbt_target(target) or target == "dbt-snowflake" target_d = Dialect.SNOWFLAKE converted, conf, auto, review = translate_sql(sql, Dialect.VERTICA, target_d) obj = MigrationObject( name="churn_features", object_type=ObjectType.SQL_SCRIPT, source_sql=sql, target_sql=converted, ) if wants_dbt: files = decompose_to_dbt(obj, Dialect.VERTICA, project_name="ml_feature_mart") out = format_dbt_project(files, max_files=14) else: out = converted md = [ "### ML / DS feature SQL migration", "", "Legacy Vertica feature engineering SQL → Snowflake" + (" dbt feature mart" if wants_dbt else ""), "", f"**Confidence:** {conf:.0f}%", "", "This path is for **data scientists / ML engineers** migrating training-feature SQL " "into warehouse-native, versioned dbt models.", "", "**Transforms**", ] md.extend(f"- {a}" for a in auto[:10]) if review: md.append("") md.append("**Review**") md.extend(f"- {r}" for r in review[:6]) return "\n".join(md), out def analyze_sql_object_ui(sql: str, source: str, target: str) -> tuple: """Gradio-friendly object assess (Plotly → HTML).""" analysis, risk_fig, badge, output, notes = analyze_sql_object(sql, source, target) return analysis, figure_to_html(risk_fig), badge, output, notes def run_workbench_ui(upload_file, use_sample: bool, source: str, target: str) -> tuple: """Gradio-friendly workbench (Plotly figs → HTML).""" ( summary, objects, rationalization, runbook, dbt_preview, validation, metrics, risk_fig, dist_fig, lineage_fig, export, report_data, ) = run_migration_workbench(upload_file, use_sample, source, target) return ( summary, objects, rationalization, runbook, dbt_preview, validation, metrics, figure_to_html(risk_fig) if isinstance(risk_fig, go.Figure) else str(risk_fig), figure_to_html(dist_fig) if isinstance(dist_fig, go.Figure) else str(dist_fig), figure_to_html(lineage_fig) if isinstance(lineage_fig, go.Figure) else str(lineage_fig), export, report_data, ) def get_leaderboard_md() -> str: from morphsql.eval.leaderboard import format_leaderboard_md return format_leaderboard_md()