← 返回 Semantica 专题首页 🌱 SEMANTICA · COOKBOOK · 入门系列

数据摄取 - 全面指南

Semantica 官方 Cookbook 中文翻译 · 第 2 / 37 篇

📦 semantica 🕸️ 知识图谱 🔎 GraphRAG

1概览

本 notebook 全面介绍 Semantica 的数据摄取能力。它涵盖了 semantica.ingest 模块中所有可用的子模块、类和辅助函数。

文档：Ingest API 参考

目录

  1. 统一摄取：ingest 函数
  2. 文件摄取：FileIngestor、FileTypeDetector、CloudStorageIngestor
  3. 网页摄取：WebIngestor、ContentExtractor、SitemapCrawler、RobotsChecker
  4. 订阅源摄取：FeedIngestor、FeedMonitor
  5. 流摄取：StreamIngestor、StreamMonitor
  6. 代码仓库摄取：RepoIngestor、CodeExtractor、GitAnalyzer
  7. 邮件摄取：EmailIngestor、AttachmentProcessor
  8. 数据库摄取：DBIngestor、DatabaseConnector
  9. MCP 摄取：MCPIngestor
  10. 配置：IngestConfig

2安装

安装 Semantica 及其所有依赖：

# 安装 Semantica 及其所有可选依赖
pip install semantica[all]

31. 统一摄取

ingest 函数是快速加载数据的主要入口。它会自动检测数据源类型。

# 安装 semantica 包
!pip install semantica

42. 文件摄取

使用 FileIngestor 和辅助类对文件处理进行精细控制。

import os
import tempfile
from semantica.ingest import FileIngestor, FileTypeDetector, CloudStorageIngestor

# 确保之前单元格中的依赖可用
if 'temp_dir' not in locals():
    temp_dir = tempfile.mkdtemp()
    print(f"Created temporary directory: {temp_dir}")

if 'sample_file' not in locals():
    sample_file = os.path.join(temp_dir, "sample_large.txt")

if not os.path.exists(sample_file):
    # 创建一个包含大量信息的示例文件
    with open(sample_file, 'w') as f:
        f.write("# Semantica Data Ingestion Guide\n\n")
        f.write("Semantica is a powerful framework for semantic data processing.\n")
        # ...（更多内容）...
    print(f"Created sample file: {sample_file}")

# --- FileTypeDetector ---
detector = FileTypeDetector()
detected_type = detector.detect_type(sample_file)
print(f"Detected Type: {detected_type}")
# ...

53. 网页摄取

使用 WebIngestor、ContentExtractor 和 SitemapCrawler 进行抓取和爬取。

import requests
from semantica.ingest import WebIngestor, ContentExtractor, SitemapCrawler, RobotsChecker

# --- ContentExtractor ---
# 演示从真实、内容丰富的网页中抽取
extractor = ContentExtractor()
url = "https://en.wikipedia.org/wiki/Artificial_intelligence"
try:
    # Wikipedia 需要 User-Agent 请求头
    headers = {'User-Agent': 'Semantica/1.0 (Education/Example)'}
    response = requests.get(url, headers=headers)
    html_content = response.text
    print(f"Fetched content from {url}")
except Exception as e:
    print(f"Failed to fetch {url}: {e}")
    # 回退内容
    html_content = "<html><body><h1>Hello World</h1><p>This is a test.</p><a href='/link'>Link</a></body></html>"

text = extractor.extract_text(html_content)
links = extractor.extract_links(html_content, base_url=url)
print(f"Extracted Text (excerpt): {text[:200]}...")
print(f"Found {len(links)} links")

# --- RobotsChecker ---
# 使用用户代理初始化
checker = RobotsChecker(user_agent="SemanticaBot")
# 检查是否可以抓取特定页面（例如 Wikipedia 的特殊页面通常受限）
check_url = "https://en.wikipedia.org/wiki/Special:Search"
can_fetch = checker.can_fetch(check_url)
print(f"Can fetch {check_url}? {can_fetch}")

# --- WebIngestor ---
# 配置 WebIngestor 使其礼貌但允许演示运行
web_ingestor = WebIngestor(
    delay=1.0,
    user_agent="Semantica/1.0 (Education/Example)",
    respect_robots=False  # 为本次演示禁用，以确保能访问 Wikipedia
)
try:
    web_content = web_ingestor.ingest_url(url)
    print(f"Web Content Title: {web_content.title}")
except Exception as e:
    print(f"Web ingest failed: {e}")

# --- SitemapCrawler ---
crawler = SitemapCrawler()
try:
    # 使用 FastAPI 文档的 sitemap 作为干净的技术示例
    sitemap_url = "https://fastapi.tiangolo.com/sitemap.xml"
    urls = crawler.parse_sitemap(sitemap_url)
    print(f"Found {len(urls)} URLs in sitemap: {sitemap_url}")
except Exception as e:
    print(f"Sitemap crawl failed: {e}")

64. 订阅源摄取

使用 FeedIngestor 消费 RSS/Atom 订阅源，并使用 FeedMonitor 进行监控。

from semantica.ingest import FeedIngestor, FeedMonitor
import time

# --- FeedIngestor ---
feed_ingestor = FeedIngestor()
# 使用 Lilian Weng 的 AI 博客 RSS 订阅源作为可靠来源
feed_url = "https://lilianweng.github.io/index.xml"
try:
    feed_data = feed_ingestor.ingest_feed(feed_url)
    print(f"Feed Title: {feed_data.title}")
    if feed_data.items:
        print(f"Latest Post: {feed_data.items[0].title}")
except Exception as e:
    print(f"Feed ingest failed: {e}")

# --- FeedMonitor ---
def feed_callback(feed_url, new_items):
    print(f"Feed Updated: {feed_url} with {len(new_items)} new items")

monitor = FeedMonitor(check_interval=5)
try:
    monitor.add_feed(feed_url)
    monitor.set_update_callback(feed_callback)
    monitor.start_monitoring()
    time.sleep(2) # 让它短暂运行一下
    monitor.stop_monitoring()
except Exception as e:
    print(f"Feed monitor failed: {e}")

75. 流摄取

使用 StreamIngestor 和 StreamMonitor 进行实时处理。

from semantica.ingest import StreamIngestor, StreamMonitor

stream_ingestor = StreamIngestor()

# --- Kafka Processor ---
# 注意：这需要一个正在运行的 Kafka 实例。为演示起见，我们将其包裹在 try-except 中。
kafka_config = {"bootstrap_servers": ["localhost:9092"]}
try:
    kafka_processor = stream_ingestor.ingest_kafka("my-topic", **kafka_config)
    print("Kafka processor initialized.")
except Exception as e:
    print(f"Kafka ingest skipped (requires active broker): {e}")

# --- RabbitMQ Processor ---
# 注意：这需要一个正在运行的 RabbitMQ 实例。为演示起见，我们将其包裹在 try-except 中。
try:
    rabbitmq_processor = stream_ingestor.ingest_rabbitmq("my-queue", "amqp://guest:guest@localhost:5672/")
    print("RabbitMQ processor initialized.")
except Exception as e:
    print(f"RabbitMQ ingest skipped (requires active broker): {e}")

# --- Stream Monitor ---
monitor = stream_ingestor.monitor
health = monitor.check_health()
print(f"Stream Health: {health['overall']}")
print(f"Processors: {list(health['processors'].keys())}")

86. 代码仓库摄取

使用 RepoIngestor、CodeExtractor 和 GitAnalyzer 分析代码库。

from semantica.ingest import RepoIngestor, CodeExtractor, GitAnalyzer
from pathlib import Path
import os

# --- CodeExtractor ---
code_extractor = CodeExtractor()
py_code = "class MyClass:\n    def my_method(self):\n        pass"
# 注意：使用内部方法 _extract_structure 对字符串输入进行演示
structure = code_extractor._extract_structure(py_code, language="python")
print(f"Classes: {structure.get('classes')}")
print(f"Functions: {structure.get('functions')}")

# --- RepoIngestor ---
repo_ingestor = RepoIngestor()
try:
    # 摄取一个公共仓库（requests）以进行可靠演示
    repo_data = repo_ingestor.ingest_repository("https://github.com/psf/requests.git")
    # 从返回的字典中访问仓库信息
    repo_info = repo_data.get('repository_info', {})
    print(f"Ingested Repo URL: {repo_info.get('url')}")
    print(f"Branches: {repo_info.get('branches')[:5]}...") # 显示前 5 个分支
    repo_ingestor.cleanup() # 清理临时文件
except Exception as e:
    print(f"Repo ingest failed: {e}")

# --- GitAnalyzer ---
try:
    # 初始化分析器
    analyzer = GitAnalyzer()

    # 使用当前目录进行演示
    current_path = Path(".")

    # 指标计算
    metrics = analyzer.calculate_metrics(current_path)
    print(f"Total Files (recursive): {metrics.get('total_files')}")
    print(f"Total Lines: {metrics.get('total_lines')}")
except Exception as e:
    print(f"Git analysis failed: {e}")

97. 邮件摄取

使用 EmailIngestor 和 AttachmentProcessor 处理邮件。

from semantica.ingest import EmailIngestor, AttachmentProcessor
import tempfile
import os

# 如果不存在则创建临时目录（尽管 AttachmentProcessor 会处理自己的临时目录）
temp_dir = tempfile.gettempdir()

# --- AttachmentProcessor ---
att_processor = AttachmentProcessor()
dummy_content = b"PDF Content"
# 使用正确的方法 'process_attachment' 而不是 'save_attachment'
# 此方法保存文件并返回包含保存路径的元数据
att_info = att_processor.process_attachment(dummy_content, "doc.pdf", "application/pdf")
print(f"Saved attachment to: {att_info.get('saved_path')}")

# --- EmailIngestor ---
email_ingestor = EmailIngestor()
try:
    # 注意：没有真实凭据时这会失败，表明它只是一个示例
    # 我们将其包裹在 try 块中，以便 notebook 能继续运行
    email_ingestor.connect_imap("imap.gmail.com", "user", "pass")
    emails = email_ingestor.ingest_mailbox("INBOX", max_emails=5)
    print(f"Fetched {len(emails)} emails")
except Exception as e:
    print(f"Email ingest skipped (Auth required): {e}")

# 清理附件处理器创建的临时文件
att_processor.cleanup_attachments()

108. 数据库摄取

使用 DBIngestor 和 DatabaseConnector 连接 SQL 数据库。

from semantica.ingest import DBIngestor, DatabaseConnector
import sqlite3
import os
import tempfile

# 在临时目录中设置 SQLite 数据库
temp_dir = tempfile.gettempdir()
db_path = os.path.join(temp_dir, "test.db")
if os.path.exists(db_path):
    os.remove(db_path)

conn = sqlite3.connect(db_path)
conn.execute("CREATE TABLE items (id INT, name TEXT)")
conn.execute("INSERT INTO items VALUES (1, 'Item 1'), (2, 'Item 2')")
conn.commit()
conn.close()

# --- DatabaseConnector ---
connector = DatabaseConnector()
# 修正：使用 'connect' 方法，而不是 'create_engine'
engine = connector.connect(f"sqlite:///{db_path}")
# sqlite 的 engine.name 是 'sqlite'
print(f"Connected to DB Driver: {engine.name}")
connector.disconnect()

# --- DBIngestor ---
db_ingestor = DBIngestor()
# 修正：使用 'export_table' 获取单个 TableData 对象，以匹配变量的使用
table_data = db_ingestor.export_table(f"sqlite:///{db_path}", table_name="items")
print(f"Table: {table_data.table_name}")
print(f"Rows: {table_data.row_count}")
print(f"Data: {table_data.rows}")

119. MCP 摄取

使用 MCPIngestor 与 Model Context Protocol 服务器集成。

from semantica.ingest import MCPIngestor
import logging

# --- MCPIngestor ---
mcp_ingestor = MCPIngestor()

# 公共 Daemon MCP 服务器
# 来源：https://danielmiessler.com/p/daemon-mcp-server
mcp_server_url = "https://mcp.daemon.danielmiessler.com"

try:
    print(f"Connecting to public MCP server: {mcp_server_url}...")

    # 该服务器支持基于 HTTP 的标准 JSON-RPC
    mcp_ingestor.connect("daemon_server", url=mcp_server_url)

    # 1. 列出可用工具
    print("\n--- Available Tools ---")
    tools = mcp_ingestor.list_available_tools("daemon_server")
    for tool in tools:
        # 只打印前 5 个工具以避免杂乱
        if tools.index(tool) < 5:
             print(f"- {tool.name}: {tool.description or 'No description'}")
    if len(tools) > 5:
        print(f"... and {len(tools) - 5} more.")

    # 2. 调用工具（get_about）
    tool_name = "get_about"
    print(f"\n--- Calling Tool '{tool_name}' ---")

    result = mcp_ingestor.ingest_tool_output("daemon_server", tool_name, {})

    # 解析内容
    content = result.content.get('content', [])
    if content and isinstance(content, list):
        for block in content:
            if block.get('type') == 'text':
                # 如果太长则截断
                text = block.get('text', '')
                preview = text[:200] + "..." if len(text) > 200 else text
                print(f"Result: {preview}")
    else:
        print(f"Raw Result: {result.content}")

except Exception as e:
    print(f"MCP Ingestion failed: {e}")

1210. 配置

使用 IngestConfig 管理摄取设置。

from semantica.ingest import IngestConfig, ingest_config

# 全局配置
print(f"Default Source Type: {ingest_config.get('default_source_type')}")

# 自定义配置实例
config = IngestConfig()
config.set("max_file_size", 1024 * 1024) # 1MB
print(f"Max File Size: {config.get('max_file_size')} bytes")