{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "[![Open In Colab](https://colab.research.google.com/assets/colab-badge.svg)](https://colab.research.google.com/github/Hawksight-AI/semantica/blob/main/cookbook/introduction/02_Data_Ingestion.ipynb)\n",
    "\n",
    "# Data Ingestion - Comprehensive Guide\n",
    "\n",
    "## Overview\n",
    "\n",
    "This notebook provides a comprehensive guide to Semantica's data ingestion capabilities. It covers all submodules, classes, and helper functions available in the `semantica.ingest` module.\n",
    "\n",
    "**Documentation**: [Ingest API Reference](https://semantica.readthedocs.io/reference/ingest/)\n",
    "\n",
    "### Table of Contents\n",
    "\n",
    "1.  **Unified Ingestion**: `ingest` function\n",
    "2.  **File Ingestion**: `FileIngestor`, `FileTypeDetector`, `CloudStorageIngestor`\n",
    "3.  **Web Ingestion**: `WebIngestor`, `ContentExtractor`, `SitemapCrawler`, `RobotsChecker`\n",
    "4.  **Feed Ingestion**: `FeedIngestor`, `FeedMonitor`\n",
    "5.  **Stream Ingestion**: `StreamIngestor`, `StreamMonitor`\n",
    "6.  **Repository Ingestion**: `RepoIngestor`, `CodeExtractor`, `GitAnalyzer`\n",
    "7.  **Email Ingestion**: `EmailIngestor`, `AttachmentProcessor`\n",
    "8.  **Database Ingestion**: `DBIngestor`, `DatabaseConnector`\n",
    "9.  **MCP Ingestion**: `MCPIngestor`\n",
    "10. **Configuration**: `IngestConfig`\n",
    "\n",
    "## Installation\n",
    "\n",
    "Install Semantica with all dependencies:\n",
    "\n",
    "```bash\n",
    "pip install semantica[all]\n",
    "```\n",
    "\n",
    "---\n",
    "\n",
    "## 1. Unified Ingestion\n",
    "\n",
    "The `ingest` function is the main entry point for quick data loading. It automatically detects the source type.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!pip install semantica\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 2. File Ingestion\n",
    "\n",
    "Detailed control over file processing using `FileIngestor` and helper classes.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "import tempfile\n",
    "from semantica.ingest import FileIngestor, FileTypeDetector, CloudStorageIngestor\n",
    "\n",
    "# Ensure dependencies from previous cells are available\n",
    "if 'temp_dir' not in locals():\n",
    "    temp_dir = tempfile.mkdtemp()\n",
    "    print(f\"Created temporary directory: {temp_dir}\")\n",
    "\n",
    "if 'sample_file' not in locals():\n",
    "    sample_file = os.path.join(temp_dir, \"sample_large.txt\")\n",
    "\n",
    "if not os.path.exists(sample_file):\n",
    "    # Create a sample file with a lot of info\n",
    "    with open(sample_file, 'w') as f:\n",
    "        f.write(\"# Semantica Data Ingestion Guide\\n\\n\")\n",
    "        f.write(\"Semantica is a powerful framework for semantic data processing.\\n\")\n",
    "        # ... (more content) ...\n",
    "    print(f\"Created sample file: {sample_file}\")\n",
    "\n",
    "# --- FileTypeDetector ---\n",
    "detector = FileTypeDetector()\n",
    "detected_type = detector.detect_type(sample_file)\n",
    "print(f\"Detected Type: {detected_type}\")\n",
    "# ..."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 3. Web Ingestion\n",
    "\n",
    "Scraping and crawling with `WebIngestor`, `ContentExtractor`, and `SitemapCrawler`.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import requests\n",
    "from semantica.ingest import WebIngestor, ContentExtractor, SitemapCrawler, RobotsChecker\n",
    "\n",
    "# --- ContentExtractor ---\n",
    "# Demonstrating extraction from a real, content-rich web page\n",
    "extractor = ContentExtractor()\n",
    "url = \"https://en.wikipedia.org/wiki/Artificial_intelligence\"\n",
    "try:\n",
    "    # Wikipedia requires a User-Agent header\n",
    "    headers = {'User-Agent': 'Semantica/1.0 (Education/Example)'}\n",
    "    response = requests.get(url, headers=headers)\n",
    "    html_content = response.text\n",
    "    print(f\"Fetched content from {url}\")\n",
    "except Exception as e:\n",
    "    print(f\"Failed to fetch {url}: {e}\")\n",
    "    # Fallback content\n",
    "    html_content = \"<html><body><h1>Hello World</h1><p>This is a test.</p><a href='/link'>Link</a></body></html>\"\n",
    "\n",
    "text = extractor.extract_text(html_content)\n",
    "links = extractor.extract_links(html_content, base_url=url)\n",
    "print(f\"Extracted Text (excerpt): {text[:200]}...\")\n",
    "print(f\"Found {len(links)} links\")\n",
    "\n",
    "# --- RobotsChecker ---\n",
    "# Initialize with user agent\n",
    "checker = RobotsChecker(user_agent=\"SemanticaBot\")\n",
    "# Check if we can fetch a specific page (e.g. Wikipedia Special pages are often restricted)\n",
    "check_url = \"https://en.wikipedia.org/wiki/Special:Search\"\n",
    "can_fetch = checker.can_fetch(check_url)\n",
    "print(f\"Can fetch {check_url}? {can_fetch}\")\n",
    "\n",
    "# --- WebIngestor ---\n",
    "# Configure WebIngestor to be polite but allow the demo to run\n",
    "web_ingestor = WebIngestor(\n",
    "    delay=1.0,\n",
    "    user_agent=\"Semantica/1.0 (Education/Example)\",\n",
    "    respect_robots=False  # Disabled for this demo to ensure Wikipedia access\n",
    ")\n",
    "try:\n",
    "    web_content = web_ingestor.ingest_url(url)\n",
    "    print(f\"Web Content Title: {web_content.title}\")\n",
    "except Exception as e:\n",
    "    print(f\"Web ingest failed: {e}\")\n",
    "\n",
    "# --- SitemapCrawler ---\n",
    "crawler = SitemapCrawler()\n",
    "try:\n",
    "    # Using FastAPI documentation sitemap as a clean, technical example\n",
    "    sitemap_url = \"https://fastapi.tiangolo.com/sitemap.xml\"\n",
    "    urls = crawler.parse_sitemap(sitemap_url)\n",
    "    print(f\"Found {len(urls)} URLs in sitemap: {sitemap_url}\")\n",
    "except Exception as e:\n",
    "    print(f\"Sitemap crawl failed: {e}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 4. Feed Ingestion\n",
    "\n",
    "Consuming RSS/Atom feeds with `FeedIngestor` and monitoring with `FeedMonitor`.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from semantica.ingest import FeedIngestor, FeedMonitor\n",
    "import time\n",
    "\n",
    "# --- FeedIngestor ---\n",
    "feed_ingestor = FeedIngestor()\n",
    "# Using Lilian Weng's AI Blog RSS feed as a reliable source\n",
    "feed_url = \"https://lilianweng.github.io/index.xml\"\n",
    "try:\n",
    "    feed_data = feed_ingestor.ingest_feed(feed_url)\n",
    "    print(f\"Feed Title: {feed_data.title}\")\n",
    "    if feed_data.items:\n",
    "        print(f\"Latest Post: {feed_data.items[0].title}\")\n",
    "except Exception as e:\n",
    "    print(f\"Feed ingest failed: {e}\")\n",
    "\n",
    "# --- FeedMonitor ---\n",
    "def feed_callback(feed_url, new_items):\n",
    "    print(f\"Feed Updated: {feed_url} with {len(new_items)} new items\")\n",
    "\n",
    "monitor = FeedMonitor(check_interval=5)\n",
    "try:\n",
    "    monitor.add_feed(feed_url)\n",
    "    monitor.set_update_callback(feed_callback)\n",
    "    monitor.start_monitoring()\n",
    "    time.sleep(2) # Let it run briefly\n",
    "    monitor.stop_monitoring()\n",
    "except Exception as e:\n",
    "    print(f\"Feed monitor failed: {e}\")\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 5. Stream Ingestion\n",
    "\n",
    "Real-time processing with `StreamIngestor` and `StreamMonitor`.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from semantica.ingest import StreamIngestor, StreamMonitor\n",
    "\n",
    "stream_ingestor = StreamIngestor()\n",
    "\n",
    "# --- Kafka Processor ---\n",
    "# Note: This requires a running Kafka instance. We wrap it in try-except for the demo.\n",
    "kafka_config = {\"bootstrap_servers\": [\"localhost:9092\"]}\n",
    "try:\n",
    "    kafka_processor = stream_ingestor.ingest_kafka(\"my-topic\", **kafka_config)\n",
    "    print(\"Kafka processor initialized.\")\n",
    "except Exception as e:\n",
    "    print(f\"Kafka ingest skipped (requires active broker): {e}\")\n",
    "\n",
    "# --- RabbitMQ Processor ---\n",
    "# Note: This requires a running RabbitMQ instance. We wrap it in try-except for the demo.\n",
    "try:\n",
    "    rabbitmq_processor = stream_ingestor.ingest_rabbitmq(\"my-queue\", \"amqp://guest:guest@localhost:5672/\")\n",
    "    print(\"RabbitMQ processor initialized.\")\n",
    "except Exception as e:\n",
    "    print(f\"RabbitMQ ingest skipped (requires active broker): {e}\")\n",
    "\n",
    "# --- Stream Monitor ---\n",
    "monitor = stream_ingestor.monitor\n",
    "health = monitor.check_health()\n",
    "print(f\"Stream Health: {health['overall']}\")\n",
    "print(f\"Processors: {list(health['processors'].keys())}\")\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 6. Repository Ingestion\n",
    "\n",
    "Analyzing codebases with `RepoIngestor`, `CodeExtractor`, and `GitAnalyzer`.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from semantica.ingest import RepoIngestor, CodeExtractor, GitAnalyzer\n",
    "from pathlib import Path\n",
    "import os\n",
    "\n",
    "# --- CodeExtractor ---\n",
    "code_extractor = CodeExtractor()\n",
    "py_code = \"class MyClass:\\n    def my_method(self):\\n        pass\"\n",
    "# Note: Using internal method _extract_structure for demonstration on string input\n",
    "structure = code_extractor._extract_structure(py_code, language=\"python\")\n",
    "print(f\"Classes: {structure.get('classes')}\")\n",
    "print(f\"Functions: {structure.get('functions')}\")\n",
    "\n",
    "# --- RepoIngestor ---\n",
    "repo_ingestor = RepoIngestor()\n",
    "try:\n",
    "    # Ingesting a public repository (requests) for reliable demonstration\n",
    "    repo_data = repo_ingestor.ingest_repository(\"https://github.com/psf/requests.git\")\n",
    "    # Accessing repo info from the returned dictionary\n",
    "    repo_info = repo_data.get('repository_info', {})\n",
    "    print(f\"Ingested Repo URL: {repo_info.get('url')}\")\n",
    "    print(f\"Branches: {repo_info.get('branches')[:5]}...\") # Show first 5 branches\n",
    "    repo_ingestor.cleanup() # Clean up temp files\n",
    "except Exception as e:\n",
    "    print(f\"Repo ingest failed: {e}\")\n",
    "\n",
    "# --- GitAnalyzer ---\n",
    "try:\n",
    "    # Initialize analyzer\n",
    "    analyzer = GitAnalyzer()\n",
    "    \n",
    "    # Use current directory for demonstration\n",
    "    current_path = Path(\".\")\n",
    "    \n",
    "    # Metrics calculation\n",
    "    metrics = analyzer.calculate_metrics(current_path)\n",
    "    print(f\"Total Files (recursive): {metrics.get('total_files')}\")\n",
    "    print(f\"Total Lines: {metrics.get('total_lines')}\")\n",
    "except Exception as e:\n",
    "    print(f\"Git analysis failed: {e}\")\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 7. Email Ingestion\n",
    "\n",
    "Processing emails with `EmailIngestor` and `AttachmentProcessor`.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from semantica.ingest import EmailIngestor, AttachmentProcessor\n",
    "import tempfile\n",
    "import os\n",
    "\n",
    "# Create a temporary directory if not exists (though AttachmentProcessor handles its own temp dir)\n",
    "temp_dir = tempfile.gettempdir()\n",
    "\n",
    "# --- AttachmentProcessor ---\n",
    "att_processor = AttachmentProcessor()\n",
    "dummy_content = b\"PDF Content\"\n",
    "# Use the correct method 'process_attachment' instead of 'save_attachment'\n",
    "# This method saves the file and returns metadata including the saved path\n",
    "att_info = att_processor.process_attachment(dummy_content, \"doc.pdf\", \"application/pdf\")\n",
    "print(f\"Saved attachment to: {att_info.get('saved_path')}\")\n",
    "\n",
    "# --- EmailIngestor ---\n",
    "email_ingestor = EmailIngestor()\n",
    "try:\n",
    "    # Note: This will fail without real credentials, identifying it as an example\n",
    "    # We wrap it in a try-block to allow the notebook to proceed\n",
    "    email_ingestor.connect_imap(\"imap.gmail.com\", \"user\", \"pass\")\n",
    "    emails = email_ingestor.ingest_mailbox(\"INBOX\", max_emails=5)\n",
    "    print(f\"Fetched {len(emails)} emails\")\n",
    "except Exception as e:\n",
    "    print(f\"Email ingest skipped (Auth required): {e}\")\n",
    "\n",
    "# Cleanup any temp files creation by attachment processor\n",
    "att_processor.cleanup_attachments()\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 8. Database Ingestion\n",
    "\n",
    "Connecting to SQL databases with `DBIngestor` and `DatabaseConnector`.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from semantica.ingest import DBIngestor, DatabaseConnector\n",
    "import sqlite3\n",
    "import os\n",
    "import tempfile\n",
    "\n",
    "# Setup SQLite DB in temp dir\n",
    "temp_dir = tempfile.gettempdir()\n",
    "db_path = os.path.join(temp_dir, \"test.db\")\n",
    "if os.path.exists(db_path):\n",
    "    os.remove(db_path)\n",
    "\n",
    "conn = sqlite3.connect(db_path)\n",
    "conn.execute(\"CREATE TABLE items (id INT, name TEXT)\")\n",
    "conn.execute(\"INSERT INTO items VALUES (1, 'Item 1'), (2, 'Item 2')\")\n",
    "conn.commit()\n",
    "conn.close()\n",
    "\n",
    "# --- DatabaseConnector ---\n",
    "connector = DatabaseConnector()\n",
    "# Fix: Use 'connect' method, not 'create_engine'\n",
    "engine = connector.connect(f\"sqlite:///{db_path}\")\n",
    "# engine.name for sqlite is 'sqlite'\n",
    "print(f\"Connected to DB Driver: {engine.name}\")\n",
    "connector.disconnect()\n",
    "\n",
    "# --- DBIngestor ---\n",
    "db_ingestor = DBIngestor()\n",
    "# Fix: Use 'export_table' to get a single TableData object, matching the variable usage\n",
    "table_data = db_ingestor.export_table(f\"sqlite:///{db_path}\", table_name=\"items\")\n",
    "print(f\"Table: {table_data.table_name}\")\n",
    "print(f\"Rows: {table_data.row_count}\")\n",
    "print(f\"Data: {table_data.rows}\")\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 9. MCP Ingestion\n",
    "\n",
    "Integrating with Model Context Protocol servers using `MCPIngestor`.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from semantica.ingest import MCPIngestor\n",
    "import logging\n",
    "\n",
    "# --- MCPIngestor ---\n",
    "mcp_ingestor = MCPIngestor()\n",
    "\n",
    "# Public Daemon MCP Server\n",
    "# Source: https://danielmiessler.com/p/daemon-mcp-server\n",
    "mcp_server_url = \"https://mcp.daemon.danielmiessler.com\"\n",
    "\n",
    "try:\n",
    "    print(f\"Connecting to public MCP server: {mcp_server_url}...\")\n",
    "    \n",
    "    # This server supports standard JSON-RPC over HTTP\n",
    "    mcp_ingestor.connect(\"daemon_server\", url=mcp_server_url)\n",
    "\n",
    "    # 1. List Available Tools\n",
    "    print(\"\\n--- Available Tools ---\")\n",
    "    tools = mcp_ingestor.list_available_tools(\"daemon_server\")\n",
    "    for tool in tools:\n",
    "        # Print first 5 tools to avoid clutter\n",
    "        if tools.index(tool) < 5:\n",
    "             print(f\"- {tool.name}: {tool.description or 'No description'}\")\n",
    "    if len(tools) > 5:\n",
    "        print(f\"... and {len(tools) - 5} more.\")\n",
    "\n",
    "    # 2. Call Tool (get_about)\n",
    "    tool_name = \"get_about\"\n",
    "    print(f\"\\n--- Calling Tool '{tool_name}' ---\")\n",
    "    \n",
    "    result = mcp_ingestor.ingest_tool_output(\"daemon_server\", tool_name, {})\n",
    "    \n",
    "    # Parse content\n",
    "    content = result.content.get('content', [])\n",
    "    if content and isinstance(content, list):\n",
    "        for block in content:\n",
    "            if block.get('type') == 'text':\n",
    "                # Truncate if too long\n",
    "                text = block.get('text', '')\n",
    "                preview = text[:200] + \"...\" if len(text) > 200 else text\n",
    "                print(f\"Result: {preview}\")\n",
    "    else:\n",
    "        print(f\"Raw Result: {result.content}\")\n",
    "\n",
    "except Exception as e:\n",
    "    print(f\"MCP Ingestion failed: {e}\")\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 10. Configuration\n",
    "\n",
    "Managing ingestion settings with `IngestConfig`.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from semantica.ingest import IngestConfig, ingest_config\n",
    "\n",
    "# Global config\n",
    "print(f\"Default Source Type: {ingest_config.get('default_source_type')}\")\n",
    "\n",
    "# Custom config instance\n",
    "config = IngestConfig()\n",
    "config.set(\"max_file_size\", 1024 * 1024) # 1MB\n",
    "print(f\"Max File Size: {config.get('max_file_size')} bytes\")\n"
   ]
  }
 ],
 "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.11.9"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}
