etl_de_project_pacmann.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Proyek ETL Pipeline - DE Project Pacmann\n",
    "---"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Latar Belakang Proyek\n",
    "\n",
    "Sebagai Data Engineer di Perusahaan XYZ, saya diberi tugas untuk membuat ETL (Extract, Transform, Load) Pipeline yang robust untuk mendukung kebutuhan data dari berbagai tim dalam perusahaan. Infrastruktur data perusahaan saat ini masih dalam tahap pengembangan, sehingga diperlukan pipeline yang dapat memproses data dari berbagai sumber secara end-to-end."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Masalah dari Requirements Gathering\n",
    "\n",
    "- **Tim Sales** memiliki data penjualan barang di Database PostgreSQL, namun terdapat banyak nilai yang hilang dan format data yang tidak konsisten.\n",
    "- **Tim Product** memiliki data harga produk elektronik dalam bentuk file CSV, namun datanya berantakan dan banyak nilai yang hilang.\n",
    "- **Tim Marketing Torch** membutuhkan data produk torch untuk riset, yang saya ambil melalui web scraping dari platform Lazada dan Tokopedia."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Solusi yang Diusulkan\n",
    "\n",
    "### 1. Data dari Tim Sales\n",
    "- Data diambil langsung dari database PostgreSQL menggunakan pipeline otomatis.\n",
    "- Missing values diatasi dengan teknik interpolasi atau imputasi untuk menjaga kualitas data dan konsistensi.\n",
    "\n",
    "### 2. Data dari Tim Product\n",
    "- Data diimpor dari file CSV yang kemudian diproses melalui pipeline.\n",
    "- Selama transformasi, dilakukan:\n",
    "  - Penanganan missing values.\n",
    "  - Perbaikan format data yang tidak konsisten.\n",
    "\n",
    "### 3. Data untuk Tim Marketing Torch\n",
    "- Data produk Torch dikumpulkan melalui web scraping dari platform Lazada dan Tokopedia.\n",
    "- Proses pembersihan dan penyesuaian data dilakukan untuk memastikan data siap digunakan dalam analisis pemasaran."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "---\n",
    "# Desain ETL Pipeline"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Extraction (Extract)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Marketing Data:\n",
    "- Data diambil dari file CSV lokal `ElectronicsProductsPricingData.csv` yang ada dalam folder `source-marketing_data`.\n",
    "- Menggunakan class `ExtractMarketingData` untuk membaca dan menyimpan data ke file `extracted_marketing_data.csv`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "class ExtractMarketingData(luigi.Task):\n",
    "    def requires(self):\n",
    "        pass # Tidak ada task yang diperlukan\n",
    "\n",
    "    def output(self):\n",
    "        return luigi.LocalTarget('.../extracted_marketing_data.csv') # Menyimpan data yang diekstrak ke file CSV\n",
    "\n",
    "    def run(self):\n",
    "        # Membaca data dari file CSV\n",
    "        marketing_data = pd.read_csv('.../source-marketing_data/ElectronicsProductsPricingData.csv')\n",
    "\n",
    "        # Menyimpan data yang diekstrak ke file CSV \n",
    "        marketing_data.to_csv(self.output().path, index = False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Sales Data:\n",
    "- Data diambil dari tabel database PostgreSQL bernama `amazon_sales_data`.\n",
    "- Menggunakan class `ExtractDatabaseSalesData` untuk menjalankan query SQL dan menyimpan hasilnya ke file `extracted_sales_data.csv`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def db_source_sales_engine():\n",
    "    db_username = 'postgres' # Nama pengguna database\n",
    "    db_password = 'password123' # Kata sandi database\n",
    "    db_host = 'localhost:5433' # Host database\n",
    "    db_name = 'etl_db' # Nama database\n",
    "\n",
    "    # Membuat string koneksi untuk database PostsgreSQL\n",
    "    engine_str = f\"postgresql://{db_username}:{db_password}@{db_host}/{db_name}\"\n",
    "    engine = create_engine(engine_str) # Membuat koneksi ke database\n",
    "\n",
    "    return engine"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "class ExtractDatabaseSalesData(luigi.Task):\n",
    "    def requires(self):\n",
    "        pass # Tidak ada task yang diperlukan\n",
    "\n",
    "    def output(self):\n",
    "        return luigi.LocalTarget('.../raw-data/extracted_sales_data.csv') # Menyimpan data yang diekstrak ke file CSV\n",
    "\n",
    "    def run(self):\n",
    "        engine = db_source_sales_engine() # Menghubungkan ke database\n",
    "        query = 'SELECT * FROM amazon_sales_data' # Query untuk mengambil data dari tabel amazon_sales_data\n",
    "\n",
    "        # Mengambil data dari database menggunakan query SQL\n",
    "        db_data = pd.read_sql(query, engine)\n",
    "\n",
    "        # Menyimpan data yang diekstrak ke file CSV\n",
    "        db_data.to_csv(self.output().path, index = False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Product Data dari Tokopedia:\n",
    "- Data diambil menggunakan scraping dengan Selenium dari halaman produk Torch di Tokopedia.\n",
    "- Class `ExtractTokpedTorchData` menyimpan hasil ekstrak ke file `torch_tokped_raw.csv`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "class ExtractTokpedTorchData(luigi.Task):\n",
    "    def requires(self):\n",
    "        pass # Tidak ada task yang diperlukan\n",
    "    \n",
    "    def output(self):\n",
    "        return luigi.LocalTarget('.../raw-data/torch_tokped_raw.csv') # MTempat penyimpanan data yang diekstrak\n",
    "\n",
    "    def run(self):\n",
    "        base_url = \"https://www.tokopedia.com/torch-id/product/page/{}\" # URL dasar untuk mengambil data produk Torch dari Tokopedia\n",
    "\n",
    "        # Mengatur opsi untuk webdriver Chrome\n",
    "        options = webdriver.ChromeOptions()\n",
    "        options.add_argument('--disable-blink-features=AutomationControlled') # Menonaktifkan fitur otomatisasi\n",
    "        options.add_experimental_option('useAutomationExtension', False) # Menonaktifkan ekstensi otomatisasi\n",
    "        options.add_experimental_option(\"excludeSwitches\", [\"enable-automation\"]) # Mengecualikan switch otomatisasi\n",
    "        driver = webdriver.Chrome(options=options) # Membuat instance dari webdriver Chrome\n",
    "\n",
    "        product_data = [] # List untuk menyimpan data produk\n",
    "\n",
    "        try:\n",
    "            for page in range(1, 13): # Mengambil data dari halaman 1 hingga 12\n",
    "                url = base_url.format(page) # Membuat URL untuk halaman saat ini\n",
    "                driver.get(url) # Mengakses URL\n",
    "\n",
    "                # Menunggu hingga elemen body muncul\n",
    "                WebDriverWait(driver, 15).until(\n",
    "                    EC.presence_of_element_located((By.TAG_NAME, 'body'))\n",
    "                )\n",
    "\n",
    "                # Menggulir halaman untuk memuat lebih banyak produk\n",
    "                for _ in range(5):\n",
    "                    driver.execute_script(\"window.scrollTo(0, document.body.scrollHeight);\") # Menggulir ke bawah\n",
    "                    time.sleep(2) # Menunggu 2 detik\n",
    "                    driver.execute_script(\"window.scrollTo(0, document.body.scrollHeight / 2);\") # Menggulir ke atas\n",
    "                    time.sleep(2) # Menunggu 2 detik\n",
    "\n",
    "                # Mengambil elemen produk\n",
    "                product_containers = driver.find_elements(By.CSS_SELECTOR, \"[data-testid='divProductWrapper']\")\n",
    "\n",
    "                for container in product_containers: # Iterasi setiap elemen produk\n",
    "                    try:\n",
    "                        # Mengambil nama produk\n",
    "                        name = container.find_element(By.CSS_SELECTOR, \"[data-testid='linkProductName']\").text \n",
    "                    except:\n",
    "                        name = None # Jika gagal, set nama menjadi None\n",
    "                    \n",
    "                    # Mengambil link produk dari elemen\n",
    "                    try:\n",
    "                        # Mencari elemen link produk dan mengambil atribut 'href'\n",
    "                        link = container.find_element(By.CSS_SELECTOR, \"a.pcv3__info-content\").get_attribute('href') \n",
    "                    except:\n",
    "                        link = None # Jika gagal, set link menjadi None\n",
    "\n",
    "                    # Mengambil harga jual produk dari elemen\n",
    "                    try:\n",
    "                        # Mencari elemen harga jual produk\n",
    "                        price_sale_elem = container.find_element(By.CSS_SELECTOR, \"[data-testid='linkProductPrice']\") \n",
    "                        # Mengambil teks dari elemen harga jual produk\n",
    "                        price_sale = price_sale_elem.text if price_sale_elem else None \n",
    "                    except:\n",
    "                        price_sale = None # Jika gagal, set harga jual menjadi None\n",
    "\n",
    "                    # Mengambil harga asli produk dari elemen\n",
    "                    try:\n",
    "                        # Mencari elemen harga asli produk\n",
    "                        price_elem = container.find_element(By.CSS_SELECTOR, \"[data-testid='lblProductSlashPrice']\") \n",
    "                        price = price_elem.text if price_elem else None # Mengambil teks dari elemen harga asli produk\n",
    "                    except:\n",
    "                        price = None # Jika gagal, set harga asli menjadi None\n",
    "\n",
    "                    try:\n",
    "                        # Mencari elemen diskon produk\n",
    "                        discount_elem = container.find_element(By.CSS_SELECTOR, \"[data-testid='lblProductDiscount']\") \n",
    "                        discount = discount_elem.text if discount_elem else None # Mengambil teks dari elemen diskon produk\n",
    "                    except:\n",
    "                        discount = None # Jika gagal, set diskon menjadi None\n",
    "\n",
    "                    # Mengambil rating produk dari elemen\n",
    "                    try:\n",
    "                        # Mencari elemen rating produk\n",
    "                        rating_elem = container.find_element(By.CSS_SELECTOR, \".prd_rating-average-text\") \n",
    "                        rating = rating_elem.text if rating_elem else None # Mengambil teks dari elemen rating produk\n",
    "                    except:\n",
    "                        rating = None # Jika gagal, set rating menjadi None\n",
    "                    \n",
    "                    # Mengambil jumlah produk yang terjual dari elemen\n",
    "                    try:\n",
    "                        # Mencari elemen jumlah produk yang terjual\n",
    "                        sold_elem = container.find_element(By.CSS_SELECTOR, \".prd_label-integrity\") \n",
    "                        sold = sold_elem.text if sold_elem else None # Mengambil teks dari elemen jumlah produk yang terjual\n",
    "                    except:\n",
    "                        sold = None # Jika gagal, set jumlah produk yang terjual menjadi None\n",
    "\n",
    "                    # Mengambil link gambar produk dari elemen\n",
    "                    try:\n",
    "                        image_elem = container.find_element(By.CSS_SELECTOR, \".css-1q90pod\") # Mencari elemen gambar produk\n",
    "                        # Mengambil atribut 'src' dari elemen gambar produk\n",
    "                        image = image_elem.get_attribute('src') if image_elem else None \n",
    "                    except:\n",
    "                        image = None # Jika gagal, set link gambar menjadi None\n",
    "\n",
    "                    # Menambahkan data produk ke dalam list product_data\n",
    "                    product_data.append({\n",
    "                        'name_product': name, # Nama produk\n",
    "                        'product_link': link, # Link produk\n",
    "                        'price_sale': price_sale, # Harga jual\n",
    "                        'price_original': price, # Harga asli\n",
    "                        'discount': discount, # Diskon\n",
    "                        'sold': sold, # Jumlah produk yang terjual\n",
    "                        'rating': rating, # Rating produk\n",
    "                        'image_link': image # Link gambar produk\n",
    "                    })\n",
    "\n",
    "            # Mengonversi list product_data ke dalam DataFrame\n",
    "            torch_tokped_df = pd.DataFrame(product_data)\n",
    "\n",
    "            # Menyimpan DataFrame ke dalam file CSV\n",
    "            torch_tokped_df.to_csv(self.output().path, index=False)\n",
    "\n",
    "        except Exception as e:\n",
    "            print(f\"Terjadi kesalahan: {e}\") # Menampilkan pesan kesalahan jika terjadi kesalahan\n",
    "        \n",
    "        finally:\n",
    "            driver.quit() # Menutup browser\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Product Data dari Lazada:\n",
    "- Data diambil menggunakan scraping dengan Selenium dari halaman produk Torch di Lazada.\n",
    "- Class `ExtractLazadaTorchData` menyimpan hasil ekstrak ke file `torch_lazada_raw.csv`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Mengekstrak data produk Torch dari Lazada\n",
    "class ExtractLazadaTorchData(luigi.Task):\n",
    "    def requires(self):\n",
    "        pass # Tidak ada task yang diperlukan\n",
    "    \n",
    "    def output(self):\n",
    "        return luigi.LocalTarget('.../raw-data/torch_lazada_raw.csv') # Tempat penyimpanan data yang diekstrak\n",
    "\n",
    "    def run(self):\n",
    "        # URL dasar untuk mengambil data produk Torch dari Lazada\n",
    "        base_url = \"https://www.lazada.co.id/torch/?from=wangpu&langFlag=en&page={page}&pageTypeId=2&q=All-Products\" \n",
    "\n",
    "        options = webdriver.ChromeOptions()\n",
    "        options.add_argument('--disable-blink-features=AutomationControlled') # Menonaktifkan fitur otomatisasi\n",
    "        options.add_experimental_option('useAutomationExtension', False) # Menonaktifkan ekstensi otomatisasi\n",
    "        options.add_experimental_option(\"excludeSwitches\", [\"enable-automation\"]) # Mengecualikan switch otomatisasi\n",
    "\n",
    "        driver = webdriver.Chrome(options=options) # Membuat instance dari webdriver Chrome\n",
    "\n",
    "        product_data = [] # List untuk menyimpan data produk\n",
    "\n",
    "        try:\n",
    "            for page in range(1, 8): # Mengambil data dari halaman 1 hingga 7\n",
    "                url = base_url.format(page=page) # Membuat URL untuk halaman saat ini\n",
    "                driver.get(url) # Mengakses URL\n",
    "\n",
    "                WebDriverWait(driver, 15).until(\n",
    "                    EC.presence_of_element_located((By.TAG_NAME, 'body')) # Menunggu hingga elemen body muncul\n",
    "                )\n",
    "\n",
    "                for _ in range(5): # Menggulir halaman untuk memuat lebih banyak produk\n",
    "                    driver.execute_script(\"window.scrollTo(0, document.body.scrollHeight);\") # Menggulir ke bawah\n",
    "                    time.sleep(2) # Menunggu 2 detik\n",
    "                    driver.execute_script(\"window.scrollTo(0, document.body.scrollHeight / 2);\") # Menggulir ke atas\n",
    "                    time.sleep(2) # Menunggu 2 detik\n",
    "\n",
    "                # Mengambil elemen produk\n",
    "                product_containers = driver.find_elements(By.CSS_SELECTOR, \"[data-qa-locator='product-item']\") \n",
    "\n",
    "                for container in product_containers: # Iterasi setiap elemen produk\n",
    "                    try:\n",
    "                        name = container.find_element(By.CSS_SELECTOR, \".RfADt a\").text # Mengambil nama produk\n",
    "                    except:\n",
    "                        name = None # Jika gagal, set nama menjadi None\n",
    "\n",
    "                    try:\n",
    "                        link = container.find_element(By.CSS_SELECTOR, \".RfADt a\").get_attribute('href') # Mengambil link produk\n",
    "                    except:\n",
    "                        link = None # Jika gagal, set link menjadi None\n",
    "\n",
    "                    try:\n",
    "                        price_sale = container.find_element(By.CSS_SELECTOR, \".aBrP0 .ooOxS\").text # Mengambil harga jual produk\n",
    "                    except:\n",
    "                        price_sale = None # Jika gagal, set harga jual menjadi None\n",
    "\n",
    "                    try:\n",
    "                        price = container.find_element(By.CSS_SELECTOR, \"._1m41m del.ooOxS\").text # Mengambil harga asli produk\n",
    "                    except:\n",
    "                        price = None # Jika gagal, set harga asli menjadi None\n",
    "\n",
    "                    try:\n",
    "                        discount = container.find_element(By.CSS_SELECTOR, \".ic-dynamic-badge-text\").text # Mengambil diskon produk\n",
    "                    except:\n",
    "                        discount = None # Jika gagal, set diskon menjadi None\n",
    "\n",
    "                    try:\n",
    "                        sold_elem = container.find_element(By.CSS_SELECTOR, \"._1cEkb span\")\n",
    "                        sold = sold_elem.text.replace(' Terjual', '') # Mengambil jumlah produk yang terjual\n",
    "                    except:\n",
    "                        sold = None # Jika gagal, set jumlah produk yang terjual menjadi None\n",
    "\n",
    "                    try:\n",
    "                        rating_container = container.find_element(By.CSS_SELECTOR, \".mdmmT\") # Mengambil elemen rating produk\n",
    "                        # Menghitung jumlah bintang rating\n",
    "                        filled_stars = len(rating_container.find_elements(By.CSS_SELECTOR, \"i._9-ogB.Dy1nx\")) \n",
    "\n",
    "                        try:\n",
    "                            # Mengambil elemen jumlah orang yang memberikan rating\n",
    "                            rating_count_elem = rating_container.find_element(By.CSS_SELECTOR, \".qzqFw\")\n",
    "                            # Mengambil teks jumlah orang yang memberikan rating \n",
    "                            rating_count = rating_count_elem.text.strip('()') \n",
    "                        except:\n",
    "                            rating_count = None # Jika gagal, set jumlah orang yang memberikan rating menjadi None\n",
    "                    except:\n",
    "                        filled_stars = None # Jika gagal, set jumlah bintang rating menjadi None\n",
    "                        rating_count = None # Jika gagal, set jumlah orang yang memberikan rating menjadi None\n",
    "                    try:\n",
    "                        # Mengambil link gambar produk\n",
    "                        image = container.find_element(By.CSS_SELECTOR, \".picture-wrapper img[type='product']\").get_attribute('src') \n",
    "                    except:\n",
    "                        image = None # Jika gagal, set link gambar menjadi None\n",
    "                    \n",
    "\n",
    "                    product_data.append({\n",
    "                        'name_product': name, # Nama produk\n",
    "                        'product_link': link, # Link produk\n",
    "                        'price_sale': price_sale, # Harga jual\n",
    "                        'price_original': price, # Harga asli\n",
    "                        'discount': discount, # Diskon\n",
    "                        'sold': sold, # Jumlah produk yang terjual\n",
    "                        'rating': filled_stars, # Rating produk\n",
    "                        'rating_count': rating_count, # Jumlah orang yang memberikan rating\n",
    "                        'image_link': image # Link gambar produk\n",
    "                    })\n",
    "            torch_lazada_df = pd.DataFrame(product_data) # Mengonversi list product_data ke dalam DataFrame\n",
    "            torch_lazada_df.to_csv(self.output().path, index=False) # Menyimpan DataFrame ke dalam file CSV\n",
    "        \n",
    "        except Exception as e:\n",
    "            print(f\"Terjadi kesalahan: {e}\") # Menampilkan pesan kesalahan jika terjadi kesalahan\n",
    "        \n",
    "        finally:\n",
    "            driver.quit() # Menutup browser"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Transformation (Transform)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Sales Data:\n",
    "Transformasi dilakukan melalui class `TransformSalesData`, termasuk:\n",
    "- Pembersihan nama produk.\n",
    "- Konversi format harga (dalam rupee) dan kalkulasi diskon.\n",
    "- Penyederhanaan kategori produk menggunakan mapping tertentu.\n",
    "- Hasil disimpan ke file `sales_clean.csv`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def clean_product_name(name):\n",
    "    # Mengubah menjadi huruf kapital\n",
    "    name = name.upper()\n",
    "    # Menghapus tanda baca dan karakter khusus dari nama produk\n",
    "    name = re.sub(r'[-/\\[\\]\"\\']', ' ', name)\n",
    "    # Menghapus spasi ekstra di sekitar nama produk\n",
    "    name = re.sub(r'\\s+', ' ', name).strip()\n",
    "    return name # Mengembalikan nama produk yang telah dibersihkan\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# class untuk mentranformasi Data Sales\n",
    "class TransformSalesData(luigi.Task):\n",
    "    def requires(self):\n",
    "        # Menentukan bahwa tugas ini bergantung pada ExtractDatabaseSalesData\n",
    "        return ExtractDatabaseSalesData()\n",
    "\n",
    "    def output(self):\n",
    "        # Menentukan lokasi output untuk data yang telah dibersihkan\n",
    "        return luigi.LocalTarget(\".../transform-data/sales_clean.csv\")\n",
    "\n",
    "    def run(self):\n",
    "        # Memuat data dari file CSV yang dihasilkan oleh tugas sebelumnya\n",
    "        sales_data = pd.read_csv(self.input().path)\n",
    "        # Membersihkan data dengan menerapkan fungsi yang sesuai\n",
    "        sales_data['name'] = sales_data['name'].apply(clean_product_name) # Membersihkan nama produk\n",
    "        sales_data['name'] = sales_data['name'].str.replace('\"', '') # Menghapus tanda kutip dari nama produk\n",
    "        # Menhapus duplikat berdasarkan kolom 'name', menyimpan yang pertama\n",
    "        sales_data = sales_data.drop_duplicates(subset='name', keep='first')\n",
    "        # Menghapus baris yang memiliki NaN pada kolom 'actual_price'\n",
    "        sales_data = sales_data.dropna(subset=['actual_price'])\n",
    "        # Mengisi nilai NaN pada kolom 'discount_price' dengan nilai dari 'actual_price'\n",
    "        sales_data['discount_price'] = sales_data['discount_price'].fillna(sales_data['actual_price'])\n",
    "        # Menghapus simbol '₹' dan koma dari kolom 'discount_price' dan 'actual_price' serta mengonversinya menjadi float\n",
    "        sales_data['discount_price'] = sales_data['discount_price'].str.replace('₹', '').str.replace(',', '').astype(float)\n",
    "        sales_data['actual_price'] = sales_data['actual_price'].str.replace('₹', '').str.replace(',', '').astype(float)\n",
    "        # Menghitung diskon berdasarkan 'actual_price' dan 'discount_price'\n",
    "        sales_data['discount'] = round((sales_data['actual_price'] - sales_data['discount_price']) / sales_data['actual_price'] * 100, 0)\n",
    "        # Mengonversi diskon menjadi string dengan simbol persen\n",
    "        sales_data['discount'] = sales_data['discount'].astype(int).astype(str) + '%' \n",
    "        sales_data['discount'] = sales_data['discount'].replace('0%', 'No Discount') # Mengganti '0%' dengan 'No Discount'\n",
    "        # Mengisi nilai NaN pada kolom 'ratings' dan 'no_of_ratings'\n",
    "        sales_data['ratings'] = sales_data['ratings'].fillna('No Ratings')\n",
    "        sales_data['no_of_ratings'] = sales_data['no_of_ratings'].fillna('No Ratings')\n",
    "        # Menentukan kolom yang akan dihapus\n",
    "        columns_to_drop = ['sub_category', 'image', 'Unnamed: 0']\n",
    "        # Menghapus kolom yang tidak diperlukan\n",
    "        sales_data = sales_data.drop(columns=columns_to_drop)\n",
    "        # Mapping kategori untuk menyederhanakan kategori produk\n",
    "        category_mapping = {\n",
    "            \"women's clothing\": \"clothing\",\n",
    "            \"men's clothing\": \"clothing\",\n",
    "            \"kids' fashion\": \"kids and baby products\",\n",
    "            \"toys & baby products\": \"kids and baby products\",\n",
    "            \"home & kitchen\": \"home and kitchen\",\n",
    "            \"home, kitchen, pets\": \"home and kitchen\",\n",
    "            \"women's shoes\": \"footwear\",\n",
    "            \"men's shoes\": \"footwear\",\n",
    "            \"beauty & health\": \"beauty and health\",\n",
    "            \"grocery & gourmet foods\": \"grocery\",\n",
    "            \"tv, audio & cameras\": \"electronics\",\n",
    "            \"car & motorbike\": \"automotive\",\n",
    "            \"Testing Category\": \"testing\"\n",
    "        }\n",
    "        # Mengganti kategori utama dengan kategori yang telah dipetakan\n",
    "        sales_data['main_category'] = sales_data['main_category'].replace(category_mapping)\n",
    "        # Menentukan nama kolom baru\n",
    "        rename_columns = {\n",
    "            'name': 'name_product',\n",
    "            'main_category': 'category',\n",
    "            'actual_price': 'price_original_rupee',\n",
    "            'discount_price': 'price_sale_rupee',\n",
    "            'discount': 'discount',\n",
    "            'ratings': 'rating',\n",
    "            'no_of_ratings': 'rating_count'\n",
    "        }\n",
    "        # Mengganti nama kolom sesuai dengan mapping yang telah ditentukan\n",
    "        sales_data = sales_data.rename(columns=rename_columns)\n",
    "        # Mengonversi kolom 'price' menjadi integer\n",
    "        sales_data['price_original_rupee'] = sales_data['price_original_rupee'].astype(int)\n",
    "        sales_data['price_sale_rupee'] = sales_data['price_sale_rupee'].astype(int)\n",
    "        # Menyimpan data yang telah dibersihkan ke file CSV\n",
    "        sales_data.to_csv(\".../transform-data/sales_clean.csv\", index=False)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Marketing Data:\n",
    "Transformasi dilakukan melalui class `TransformMarketingData`, termasuk:\n",
    "- Pembersihan data tanggal, berat produk, ketersediaan, kondisi produk, dan informasi pengiriman.\n",
    "- Kalkulasi diskon berdasarkan harga maksimum dan minimum.\n",
    "- Hasil disimpan ke file `marketing_clean.csv`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Fungsi untuk mengambil tanggal terawal\n",
    "def get_earliest_date(dates):\n",
    "    # Mengonversi string tanggal yang dipisahkan koma menjadi list tanggal\n",
    "    date_list = [pd.to_datetime(date.strip(), format=\"%Y-%m-%dT%H:%M:%SZ\") for date in dates.split(\",\")]\n",
    "    # Mengambil tanggal terawal dari list tanggal\n",
    "    earliest_date = min(date_list)\n",
    "    # Mengembalikan tanggal terawal dalam format 'YYYY-MM-DD'\n",
    "    return earliest_date.strftime(\"%Y-%m-%d\")\n",
    "\n",
    "# Fungsi untuk mengonversi berat produk\n",
    "def clean_weight(weight):\n",
    "    # Menghapus spasi di awal dan akhir string dan mengubah menjadi huruf kecil\n",
    "    weight = weight.strip().lower()\n",
    "    # Mencari angka dan satuan berat (lbs, pounds, ounces, oz) dalam string\n",
    "    weight_parts = re.findall(r'(\\d+\\.?\\d*)\\s*(lbs?|pounds?|ounces?|oz)', weight)\n",
    "    total_weight = 0.0 # Inisialisasi total berat\n",
    "\n",
    "    # Menghitung total berat berdasarkan satuannya\n",
    "    for value, unit in weight_parts:\n",
    "        value = float(value)  # Mengonversi nilai berat ke tipe data float\n",
    "        if unit in ['lbs', 'lb', 'pounds']:\n",
    "            total_weight += value * 0.453592  # Mengonversi lbs ke kg\n",
    "        elif unit in ['ounces', 'ounce', 'oz']:\n",
    "            total_weight += value * 0.0283495  # Mengonversi ounces ke kg\n",
    "    # Mengembalikan total berat dalam kg hingga 2 desimal\n",
    "    return round(total_weight, 2)\n",
    "\n",
    "# Fungsi untuk membersihkan data pengiriman\n",
    "def clean_shipping(value):\n",
    "    # Memeriksa apakah nilah adalah NaN\n",
    "    if pd.isna(value):\n",
    "        return 'Undefined' # Mengembalikan 'Undefined' jika nilai adalah NaN\n",
    "    value = str(value).lower() # Mengubah nilai menjadi string dan huruf kecil \n",
    "    # Mengidentifikasi jenis pengiriman berdasarkan nilai\n",
    "    if 'free' in value:\n",
    "        return 'Free Shipping' # Mengembalikan 'Free Shipping' jika 'free' ada dalam nilai\n",
    "    elif 'expedited' in value:\n",
    "        return 'Expedited Shipping' # Mengembalikan 'Expedited Shipping' jika 'expedited' ada dalam nilai\n",
    "    elif 'standard' in value:\n",
    "        return 'Standard Shipping' # Mengembalikan 'Standard Shipping' jika 'standard' ada dalam nilai\n",
    "    elif 'freight' in value:\n",
    "        return 'Freight Shipping' # Mengembalikan 'Freight Shipping' jika 'freight' ada dalam nilai\n",
    "    elif 'usd' in value or 'cad' in value:\n",
    "        return 'Paid Shipping' # Mengembalikan 'Paid Shipping' jika 'usd' atau 'cad' ada dalam nilai\n",
    "    else:\n",
    "        return 'Undefined' # Mengembalikan 'Undefined' jika tidak ada kondisi di atas\n",
    "\n",
    "# Fungsi untuk membersihkan ketersediaan produk\n",
    "def clean_availability(value):\n",
    "    # Memeriksa apakah nilai adalah NaN\n",
    "    if pd.isna(value):\n",
    "        return 'Unavailable' # Mengembalikan 'Unavailable' jika nilai adalah NaN\n",
    "    value = str(value).lower() # Mengubah nilai menjadi string dan huruf kecil\n",
    "    # Mengidentifikasi ketersediaan produk berdasarkan nilai\n",
    "    if 'in stock' in value or 'yes' in value or 'true' in value or 'available' in value:\n",
    "        # Mengembalikan 'Available' jika 'in stock', 'yes', 'true', atau 'available' ada dalam nilai\n",
    "        return 'Available' \n",
    "    elif 'out of stock' in value or 'no' in value or 'retired' in value or 'sold' in value or 'false' in value:\n",
    "        # Mengembalikan 'Unavailable' jika 'out of stock', 'no', 'retired', 'sold', atau 'false' ada dalam nilai\n",
    "        return 'Unavailable' \n",
    "    elif 'special order' in value or 'more on the way' in value:\n",
    "        # Mengembalikan 'Pending' jika 'special order' atau 'more on the way' ada dalam nilai\n",
    "        return 'Pending' \n",
    "    else:\n",
    "        return 'Unavailable' # Mengembalikan 'Unavailable' jika tidak ada kondisi di atas\n",
    "\n",
    "# Fungsi untuk membersihkan kondisi produk\n",
    "def clean_condition(condition):\n",
    "    if pd.isna(condition):\n",
    "        return 'Undefined' # Mengembalikan 'Undefined' jika nilai adalah NaN\n",
    "    condition = str(condition).lower() # Mengubah nilai menjadi string dan huruf kecil\n",
    "    if 'new' in condition:\n",
    "        return 'New' # Mengembalikan 'New' jika 'new' ada dalam nilai\n",
    "    else:\n",
    "        return 'Used' # Mengembalikan 'Used jika kondisi tidak 'new'\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# class untuk mentranformasi Data Marketing\n",
    "class TransformMarketingData(luigi.Task):\n",
    "    def requires(self):\n",
    "        # Menentukan bahwa tugas ini bergantung pada tugas ExtractMarketingData\n",
    "        return ExtractMarketingData()\n",
    "\n",
    "    def output(self):\n",
    "        # Menentukan lokasi output untuk data yang telah dibersihkan\n",
    "        return luigi.LocalTarget(\".../transform-data/marketing_clean.csv\")\n",
    "\n",
    "    def run(self):\n",
    "        # Memuat data dari file CSV yang dihasilkan oleh tugas sebelumnya\n",
    "        marketing_data = pd.read_csv(self.input().path)\n",
    "        # Membersihkan data dengan menerapkan fungsi yang sesuai\n",
    "        # Mengambil tanggal terawal\n",
    "        marketing_data['prices.dateSeen'] = marketing_data['prices.dateSeen'].apply(get_earliest_date)\n",
    "        # Membersihkan dan mengonversi berat \n",
    "        marketing_data['weight'] = marketing_data['weight'].apply(clean_weight)\n",
    "        # Membersihkan ketersediaan \n",
    "        marketing_data['prices.availability'] = marketing_data['prices.availability'].apply(clean_availability) \n",
    "        # Membersihkan kondisi produk\n",
    "        marketing_data['prices.condition'] = marketing_data['prices.condition'].apply(clean_condition) \n",
    "        # Membersihkan informasi pengiriman\n",
    "        marketing_data['prices.shipping'] = marketing_data['prices.shipping'].apply(clean_shipping)\n",
    "        # Mengonversi kolom harga maksimum dan minimum menjadi tipe numerik\n",
    "        marketing_data['prices.amountMax'] = pd.to_numeric(marketing_data['prices.amountMax'], errors='coerce')\n",
    "        marketing_data['prices.amountMin'] = pd.to_numeric(marketing_data['prices.amountMin'], errors='coerce')\n",
    "        # Menghitung diskon berdasarkan harga maksimum dan minimum\n",
    "        marketing_data['discount'] = round((marketing_data['prices.amountMax'] - marketing_data['prices.amountMin'])\n",
    "                                            / marketing_data['prices.amountMax'] * 100, 0)\n",
    "        # Mengonversi diskon menjadi string dengan simbol persen\n",
    "        marketing_data['discount'] = marketing_data['discount'].astype(int).astype(str) + '%' \n",
    "        marketing_data['discount'] = marketing_data['discount'].replace('0%', 'No Discount') # Mengganti '0%' dengan 'No Discount'\n",
    "        # Menghapus '.com' dari nama merchant\n",
    "        marketing_data['prices.merchant'] = marketing_data['prices.merchant'].str.replace('.com', '')\n",
    "        # Membersihkan nama produk\n",
    "        marketing_data['name'] = marketing_data['name'].apply(clean_product_name)\n",
    "        marketing_data['name'] = marketing_data['name'].str.replace(' \" ', ' ') # Menghapus spasi yang tidak perlu\n",
    "        # Menentukan kolom yang dihapus\n",
    "        columns_to_drop = ['id', 'prices.currency',\n",
    "                           'prices.sourceURLs', 'prices.sourceURLs',\n",
    "                           'prices.isSale', 'prices.sourceURLs',\n",
    "                           'asins', 'brand', 'categories', 'dateAdded',\n",
    "                           'dateUpdated', 'ean', 'imageURLs', 'keys',\n",
    "                           'manufacturer', 'manufacturerNumber', 'sourceURLs',\n",
    "                           'upc', 'Unnamed: 26', 'Unnamed: 27', 'Unnamed: 28',\n",
    "                           'Unnamed: 29', 'Unnamed: 30']\n",
    "\n",
    "        # Menentukan nama kolom baru\n",
    "        rename_columns = {\n",
    "            'prices.dateSeen': 'date_seen',\n",
    "            'prices.availability': 'availability',\n",
    "            'prices.amountMax': 'price_original',\n",
    "            'prices.amountMin': 'price_sale',\n",
    "            'primaryCategories': 'category',\n",
    "            'name': 'name_product',\n",
    "            'weight': 'weight_kg',\n",
    "            'discount': 'discount',\n",
    "            'prices.shipping': 'shipping',\n",
    "            'prices.merchant': 'merchant',\n",
    "            'prices.condition': 'condition'\n",
    "        }\n",
    "        # Menghapus kolom yang tidak diperlukan\n",
    "        marketing_data = marketing_data.drop(columns=columns_to_drop)\n",
    "        # Mengganti nama kolom yang sesuai dengan mapping yang telah ditentukan\n",
    "        marketing_data = marketing_data.rename(columns=rename_columns)\n",
    "        # Mengatur urutan kolom yang akan disimpan\n",
    "        marketing_data = marketing_data[['name_product', 'category',\n",
    "                                         'price_original', 'price_sale',\n",
    "                                         'discount', 'condition', 'availability',\n",
    "                                         'date_seen', 'weight_kg', 'shipping', 'merchant']]\n",
    "                                         \n",
    "        # Menyimpan data yang telah dibersihkan ke file CSV\n",
    "        marketing_data.to_csv(\".../transform-data/marketing_clean.csv\", index=False)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Product Data Tokopedia & Lazada:\n",
    "Pembersihan dan transformasi data dilakukan melalui class `TransformTorchData`, seperti:\n",
    "- Pembersihan harga, rating, dan diskon.\n",
    "- Konversi format data untuk keseragaman.\n",
    "- Pembuangan kolom yang tidak diperlukan.\n",
    "- Hasil disimpan ke file `torch_tokped_clean.csv` dan `torch_lazada_clean.csv`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Fungsi untuk membersihkan data Tokopedia\n",
    "def clean_data_tokped(tokped):\n",
    "    # Mengisi nilai NaN pada kolom 'price_original' dengan nilai pada kolom 'price_sale'\n",
    "    tokped['price_original'] = tokped['price_original'].fillna(tokped['price_sale'])\n",
    "    # Mengisi nilai NaN pada kolom 'discount' dengan 'Tidak Ada Diskon'\n",
    "    tokped['discount'] = tokped['discount'].fillna('Tidak Ada Diskon')\n",
    "    # Mengisi nilai NaN pada kolom 'sold' dengan '0'\n",
    "    tokped['sold'] = tokped['sold'].fillna('0')\n",
    "    # Mengisi nilai NaN pada kolom 'rating' dengan 'Belum Ada Penilaian'\n",
    "    tokped['rating'] = tokped['rating'].fillna('Belum Ada Penilaian')\n",
    "    # Menghapus karakter 'Rp' dan '.' dari kolom 'price_sale' dan mengonversi ke tipe data integer\n",
    "    tokped['price_sale'] = tokped['price_sale'].str.replace('Rp', '').str.replace('.', '').astype(int)\n",
    "    # Menghapus karakter 'Rp' dan '.' dari kolom 'price_original' dan mengonversi ke tipe data integer\n",
    "    tokped['price_original'] = tokped['price_original'].str.replace('Rp', '').str.replace('.', '').astype(int)\n",
    "    # Mengambil angka dari kolom 'sold' dan mengonversi ke tipe data integer\n",
    "    tokped['sold'] = tokped['sold'].str.extract(r'(\\d+)').astype(int)\n",
    "    # Menghapus kolom 'image_link' dari DataFrame\n",
    "    drop_columns = ['image_link']\n",
    "    tokped = tokped.drop(columns=drop_columns)\n",
    "    # Menerapkan fungsi clean_product_name pada kolom 'name_product'\n",
    "    tokped['name_product'] = tokped['name_product'].apply(clean_product_name)\n",
    "    # Menghapus data duplikat berdasarkan kolom 'name_product'\n",
    "    tokped = tokped.drop_duplicates(subset='name_product', keep='first')\n",
    "\n",
    "    return tokped # Mengembalikan DataFrame yang telah dibersihkan\n",
    "\n",
    "# Fungsi untuk membersihkan data Lazada\n",
    "def clean_data_lazada(lazada):\n",
    "    # Mengisi nilai Nan pada kolom 'price_original' dengan nilai pada kolom 'price_sale'\n",
    "    lazada['price_original'] = lazada['price_original'].fillna(lazada['price_sale'])\n",
    "    # Menghapus karakter 'Rp' dan '.' dari kolom 'price_sale' dan mengonversi ke tipe data integer\n",
    "    lazada['price_sale'] = lazada['price_sale'].str.replace('Rp', '').str.replace('.', '').astype(int)\n",
    "    # Menghapus karakter 'Rp' dan '.' dari kolom 'price_original' dan mengonversi ke tipe data integer\n",
    "    lazada['price_original'] = lazada['price_original'].str.replace('Rp', '').str.replace('.', '').astype(int)\n",
    "    # Mengisi nilai Nan pada kolom 'discount' dengan 'Tidak Ada Diskon'\n",
    "    lazada['discount'] = lazada['discount'].fillna('Tidak Ada Diskon')\n",
    "    # Menghapus 'Voucher save' dari kolom 'discount'\n",
    "    lazada['discount'] = lazada['discount'].str.replace('Voucher save', '')\n",
    "    # Mengisi nilai Nan pada kolom 'rating' dengan 'Belum Ada Penilaian'\n",
    "    lazada['rating'] = lazada['rating'].fillna('Belum Ada Penilaian')\n",
    "    # Mengisi nilai Nan pada kolom 'rating_count' dengan '0' dan mengonversi ke tipe data integer\n",
    "    lazada['rating_count'] = lazada['rating_count'].fillna('0').astype(int)\n",
    "    # Mengisi nilai Nan pada kolom 'sold' dengan nilai pada kolom 'rating_count'\n",
    "    lazada['sold'] = lazada['sold'].fillna(lazada['rating_count'])\n",
    "    # Menghapus karakter ' sold' dari kolom sold\n",
    "    lazada['sold'] = lazada['sold'].str.replace(' sold', '')\n",
    "    # Mengonversi format 'K' dan titik pada kolom 'sold' menjadi angka\n",
    "    lazada['sold'] = lazada['sold'].str.replace(' K', '000').str.replace('K', '000').str.replace('.', '')\n",
    "    # Mengisi nilai NaN pada kolom 'sold' dengan nilai pada kolom 'rating_count' dan mengonversi ke tipe data integer\n",
    "    lazada['sold'] = lazada['sold'].fillna(lazada['rating_count']).astype(int)\n",
    "    # Menghapus kolom 'image_link' dari DataFrame\n",
    "    lazada = lazada.drop(columns=['image_link'])\n",
    "    # Menerapkan fungsi clean_product_name pada kolom 'name_product'\n",
    "    lazada['name_product'] = lazada['name_product'].apply(clean_product_name)\n",
    "    # Menghapus data duplikat berdasarkan kolom 'name_product'\n",
    "    lazada = lazada.drop_duplicates(subset='name_product', keep='first')\n",
    "\n",
    "    return lazada # Mengembalikan DataFrame yang telah dibersihkan\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "class TransformTorchData(luigi.Task):\n",
    "    def requires(self):\n",
    "        # Mengembalikan instance dari task ExtractTokpedTorchData dan ExtractLazadaTorchData\n",
    "        return [ExtractTokpedTorchData(), ExtractLazadaTorchData()]\n",
    "\n",
    "    def output(self):\n",
    "        # Menentukan lokasi output untuk data yang telah dibersihkan\n",
    "        return {\n",
    "            'tokped': luigi.LocalTarget('.../transform-data/torch_tokped_clean.csv'),\n",
    "            'lazada': luigi.LocalTarget('.../transform-data/torch_lazada_clean.csv')\n",
    "        }\n",
    "\n",
    "    def run(self):\n",
    "        # Memuat data dari instance task ExtractTokpedTorchData dan ExtractLazadaTorchData\n",
    "        tokped_data = pd.read_csv(self.input()[0].path)\n",
    "        lazada_data = pd.read_csv(self.input()[1].path)\n",
    "        # Membersihkan data \n",
    "        tokped_data_clean = clean_data_tokped(tokped_data)\n",
    "        lazada_data_clean = clean_data_lazada(lazada_data)\n",
    "        # Menyimpan data hasil ekstrak ke dalam format csv\n",
    "        tokped_data_clean.to_csv(self.output()['tokped'].path, index=False)\n",
    "        lazada_data_clean.to_csv(self.output()['lazada'].path, index=False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "---\n",
    "## Loading (Load)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- Data dari tahap transformasi dimuat ke database PostgreSQL bernama `de_project_pacmann`.\n",
    "- Class `LoadData` menangani proses pembuatan tabel database jika belum ada, termasuk:\n",
    "  - Tabel untuk data penjualan (`sales_clean`).\n",
    "  - Tabel untuk data pemasaran (`marketing_clean`).\n",
    "  - Tabel untuk data Tokopedia (`torch_tokped_clean`).\n",
    "  - Tabel untuk data Lazada (`torch_lazada_clean`).\n",
    "\n",
    "- Data dimasukkan ke tabel menggunakan metode `.to_sql()` dari Pandas."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Membuat koneksi ke database PostgreSQL\n",
    "def postgres_engine():\n",
    "    db_username = 'postgres' # Nama pengguna database\n",
    "    db_password = '123123' # Kata sandi database\n",
    "    db_host = 'localhost:5432' # Host database\n",
    "    db_name = 'de_project_pacmann' # Nama database\n",
    "    # Membuat string koneksi untuk database PostgreSQL\n",
    "    engine_str = f\"postgresql://{db_username}:{db_password}@{db_host}/{db_name}\"\n",
    "    engine = create_engine(engine_str) # Membuat engine database\n",
    "\n",
    "    return engine # Mengembalikan engine database\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "class LoadData(luigi.Task):\n",
    "    def requires(self):\n",
    "        # Menentukan bahwa tugas ini bergantung pada beberapa tugas transformasi\n",
    "        return [TransformSalesData(), TransformMarketingData(), TransformTorchData()]\n",
    "\n",
    "    def output(self):\n",
    "        # Menentukan lokasi output untuk data yang akan disimpan\n",
    "        return [luigi.LocalTarget('sales_db.csv'),\n",
    "                luigi.LocalTarget('marketing_db.csv'),\n",
    "                luigi.LocalTarget('torch_tokped_db.csv'),\n",
    "                luigi.LocalTarget('torch_lazada_db.csv')]\n",
    "    \n",
    "    def run(self):\n",
    "        # Menghubungkan ke database PostgreSQL\n",
    "        dw_engine = postgres_engine()\n",
    "        # Memuat data dari file  CSV yang dihasilkan oleh class sebelumnya\n",
    "        sales_data_db = pd.read_csv(self.input()[0].path)\n",
    "        marketing_data_db = pd.read_csv(self.input()[1].path)\n",
    "        torch_tokped_data_db = pd.read_csv(self.input()[2]['tokped'].path)\n",
    "        torch_laza_data_db = pd.read_csv(self.input()[2]['lazada'].path)\n",
    "        # Query untuk membuat tipe data dan tabel jika belum ada\n",
    "        create_table_query = \"\"\"\n",
    "        DO $$ BEGIN\n",
    "            IF NOT EXISTS (SELECT 1 FROM pg_type WHERE typname = 'condition_type') THEN\n",
    "                CREATE TYPE condition_type AS ENUM ('New', 'Used', 'Undefined'); -- Tipe untuk kondisi produk\n",
    "            END IF;\n",
    "            IF NOT EXISTS (SELECT 1 FROM pg_type WHERE typname = 'availability_type') THEN\n",
    "                CREATE TYPE availability_type AS ENUM ('Unavailable', 'Available', 'Pending'); -- Tipe untuk ketersediaan produk\n",
    "            END IF;\n",
    "            IF NOT EXISTS (SELECT 1 FROM pg_type WHERE typname = 'shipping_type') THEN\n",
    "                CREATE TYPE shipping_type AS ENUM ('Free Shipping', 'Expedited Shipping',\n",
    "                                                   'Standard Shipping', 'Freight Shipping',\n",
    "                                                   'Paid Shipping', 'Undefined'); -- Tipe untuk pengiriman\n",
    "            END IF;\n",
    "        END $$;\n",
    "\n",
    "        -- Membuat tabel untuk data penjualan jika belum ada\n",
    "        CREATE TABLE IF NOT EXISTS sales_clean (\n",
    "            name_product TEXT NOT NULL,\n",
    "            category VARCHAR(50) NOT NULL,\n",
    "            link TEXT NOT NULL,\n",
    "            rating TEXT NOT NULL,\n",
    "            rating_count TEXT NOT NULL,\n",
    "            price_original_rupee NUMERIC NOT NULL,\n",
    "            price_sale_rupee NUMERIC NOT NULL,\n",
    "            discount TEXT NOT NULL,\n",
    "            created_at TIMESTAMP NOT NULL DEFAULT now() -- Waktu pembuatan\n",
    "        );\n",
    "        -- Membuat tabel untuk data pemasaran jika belum ada\n",
    "        CREATE TABLE IF NOT EXISTS marketing_clean (\n",
    "            name_product TEXT NOT NULL,\n",
    "            category VARCHAR(50) NOT NULL,\n",
    "            price_original NUMERIC NOT NULL,\n",
    "            price_sale NUMERIC NOT NULL,\n",
    "            discount TEXT NOT NULL,\n",
    "            condition condition_type NOT NULL,\n",
    "            availability availability_type NOT NULL,\n",
    "            shipping shipping_type NOT NULL,\n",
    "            date_seen DATE NOT NULL,\n",
    "            weight_kg NUMERIC NOT NULL, \n",
    "            merchant VARCHAR NOT NULL,\n",
    "            created_at TIMESTAMP NOT NULL DEFAULT now()\n",
    "        );\n",
    "        -- Membuat tabel untuk data dari Tokopedia jika belum ada\n",
    "        CREATE TABLE IF NOT EXISTS torch_tokped_clean (\n",
    "            name_product TEXT NOT NULL,\n",
    "            product_link TEXT NOT NULL,\n",
    "            price_original INT NOT NULL,\n",
    "            price_sale INT NOT NULL,\n",
    "            discount TEXT NOT NULL,\n",
    "            rating TEXT NOT NULL,\n",
    "            sold INT NOT NULL,\n",
    "            created_at TIMESTAMP NOT NULL DEFAULT now()\n",
    "        );\n",
    "        -- Membuat tabel untuk data dari Lazada jika belum ada\n",
    "        CREATE TABLE IF NOT EXISTS torch_lazada_clean (\n",
    "            name_product TEXT NOT NULL,\n",
    "            product_link TEXT NOT NULL,\n",
    "            price_original INT NOT NULL,\n",
    "            price_sale INT NOT NULL,\n",
    "            discount TEXT NOT NULL,\n",
    "            sold INT NOT NULL,\n",
    "            rating TEXT NOT NULL,\n",
    "            rating_count INT NOT NULL,\n",
    "            created_at TIMESTAMP NOT NULL DEFAULT now()\n",
    "        );\n",
    "        \"\"\"\n",
    "        # Menggunakan koneksi ke database untu mengeksekusi query\n",
    "        with dw_engine.connect() as conn:\n",
    "            # Menjalankan query untuk membuat tipe data dan tabel jika belum ada\n",
    "            conn.execute(text(create_table_query))\n",
    "            conn.close() # Menutup koneksi setelah eksekusi selesai\n",
    "        # Menyimpan data penjualan ke tabel 'sales_clean' di database\n",
    "        sales_data_db.to_sql(name='sales_clean', con=dw_engine, if_exists=\"append\", index=False)\n",
    "        # Menyimpan data pemasaran e tabel 'marketing_clean' di database\n",
    "        marketing_data_db.to_sql(name='marketing_clean', con=dw_engine, if_exists=\"append\", index=False)\n",
    "        # Menyimpan data dari Tokopedia ke tabel 'torch_tokped_clean' di database\n",
    "        torch_tokped_data_db.to_sql(name='torch_tokped_clean', con=dw_engine, if_exists=\"append\", index=False)\n",
    "        # Menyimpan data dari Lazada ke tabel 'torch_lazada_clean' di databse\n",
    "        torch_laza_data_db.to_sql(name='torch_lazada_clean', con=dw_engine, if_exists=\"append\", index=False)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Penulis\n",
    "\n",
    "[Zulfa Nurfajar](https://www.linkedin.com/in/zulfanurfajar/)\n",
    "\n",
    "Jika Anda memiliki pertanyaan atau saran, silakan buka issue atau hubungi saya melalui repositori ini!"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.12.3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}