{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Score Caldova feedback with Fabric AI Functions\n",
    "\n",
    "This notebook loads synthetic Caldova product and feedback data, adds product context to each comment, applies `ai.analyze_sentiment`, validates the labels, and writes managed Delta tables for a Fabric data agent or ontology."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import synapse.ml.spark.aifunc as aifunc\n",
    "from pyspark.sql import functions as F\n",
    "from pyspark.sql.types import DateType, IntegerType, StringType, StructField, StructType\n",
    "\n",
    "DATA_ROOT = \"/lakehouse/default/Files/caldova-sentiment\"\n",
    "SUPPORTED_SENTIMENTS = [\"positive\", \"negative\", \"neutral\", \"mixed\"]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "product_schema = StructType([\n",
    "    StructField(\"ProductID\", StringType(), False),\n",
    "    StructField(\"ProductName\", StringType(), False),\n",
    "    StructField(\"ProductDescription\", StringType(), False),\n",
    "])\n",
    "\n",
    "feedback_schema = StructType([\n",
    "    StructField(\"FeedbackID\", IntegerType(), False),\n",
    "    StructField(\"ProductID\", StringType(), False),\n",
    "    StructField(\"FeedbackDate\", DateType(), False),\n",
    "    StructField(\"Channel\", StringType(), False),\n",
    "    StructField(\"Region\", StringType(), False),\n",
    "    StructField(\"FeedbackText\", StringType(), False),\n",
    "])\n",
    "\n",
    "product = (spark.read\n",
    "           .option(\"header\", True)\n",
    "           .schema(product_schema)\n",
    "           .csv(f\"{DATA_ROOT}/products.csv\"))\n",
    "\n",
    "feedback = (spark.read\n",
    "            .option(\"header\", True)\n",
    "            .option(\"dateFormat\", \"yyyy-MM-dd\")\n",
    "            .schema(feedback_schema)\n",
    "            .csv(f\"{DATA_ROOT}/customer_feedback.csv\"))\n",
    "\n",
    "if product.count() != 3:\n",
    "    raise RuntimeError(\"Expected 3 product rows.\")\n",
    "if feedback.count() != 16:\n",
    "    raise RuntimeError(\"Expected 16 feedback rows.\")\n",
    "if product.select(\"ProductID\").distinct().count() != 3:\n",
    "    raise RuntimeError(\"ProductID values must be unique.\")\n",
    "if feedback.select(\"FeedbackID\").distinct().count() != 16:\n",
    "    raise RuntimeError(\"FeedbackID values must be unique.\")\n",
    "\n",
    "orphaned_feedback = feedback.join(product, \"ProductID\", \"left_anti\").count()\n",
    "if orphaned_feedback:\n",
    "    raise RuntimeError(f\"Found {orphaned_feedback} feedback rows without a product.\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "analysis_input = (feedback\n",
    "                  .join(product, \"ProductID\", \"inner\")\n",
    "                  .withColumn(\n",
    "                      \"AnalysisText\",\n",
    "                      F.concat_ws(\n",
    "                          \" \",\n",
    "                          F.concat(F.lit(\"Product name: \"), F.col(\"ProductName\"), F.lit(\".\")),\n",
    "                          F.concat(F.lit(\"Product context: \"), F.col(\"ProductDescription\")),\n",
    "                          F.concat(F.lit(\"Customer feedback: \"), F.col(\"FeedbackText\")),\n",
    "                      ),\n",
    "                  ))\n",
    "\n",
    "# This code uses AI. Always review output for mistakes.\n",
    "scored = analysis_input.ai.analyze_sentiment(\n",
    "    input_col=\"AnalysisText\",\n",
    "    output_col=\"Sentiment\",\n",
    "    error_col=\"AIError\",\n",
    ").withColumn(\"Sentiment\", F.lower(F.col(\"Sentiment\")))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "failed_rows = scored.filter(\n",
    "    F.col(\"Sentiment\").isNull()\n",
    "    | (~F.col(\"Sentiment\").isin(SUPPORTED_SENTIMENTS))\n",
    "    | F.col(\"AIError\").isNotNull()\n",
    ")\n",
    "\n",
    "if failed_rows.count():\n",
    "    display(failed_rows.select(\"FeedbackID\", \"Sentiment\", \"AIError\"))\n",
    "    raise RuntimeError(\"One or more feedback records were not scored.\")\n",
    "\n",
    "product.write.format(\"delta\").mode(\"overwrite\").saveAsTable(\"product\")\n",
    "\n",
    "output_columns = [\n",
    "    \"FeedbackID\", \"ProductID\", \"ProductName\", \"ProductDescription\",\n",
    "    \"FeedbackDate\", \"Channel\", \"Region\", \"FeedbackText\",\n",
    "    \"AnalysisText\", \"Sentiment\", \"AIError\",\n",
    "]\n",
    "(scored.select(*output_columns)\n",
    " .write\n",
    " .format(\"delta\")\n",
    " .mode(\"overwrite\")\n",
    " .option(\"overwriteSchema\", \"true\")\n",
    " .saveAsTable(\"customer_feedback_sentiment\"))\n",
    "\n",
    "print(\"Validation passed: 3 products and 16 feedback records; all records have supported sentiment labels.\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "display(\n",
    "    scored.groupBy(\"Sentiment\")\n",
    "          .count()\n",
    "          .withColumnRenamed(\"count\", \"FeedbackCount\")\n",
    "          .orderBy(\"Sentiment\")\n",
    ")\n",
    "\n",
    "display(\n",
    "    scored.select(\n",
    "        \"FeedbackID\", \"ProductName\", \"Region\", \"FeedbackText\", \"Sentiment\"\n",
    "    ).orderBy(\"FeedbackID\")\n",
    ")"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Synapse PySpark",
   "language": "Python",
   "name": "synapse_pyspark"
  },
  "language_info": {
   "name": "python"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}