Preparing Content for Ingestion

from dataclasses import dataclass, field

@dataclass
class ContentItem:
    producer_id: str           # stable: {source}-{native-id}
    filename: str              # name for the uploaded file, e.g. "post-4812.md"
    text: str | None = None    # normalized Markdown (text you extracted)
    data: bytes | None = None  # raw file bytes (already-supported formats, passed through)
    metadata: dict = field(default_factory=dict)  # title, item_url, author, ...

interface ContentItem {
  producerId: string;        // stable: {source}-{native-id}
  filename: string;          // name for the uploaded file, e.g. "post-4812.md"
  text?: string;             // normalized Markdown (text you extracted)
  data?: Uint8Array;         // raw file bytes (already-supported formats, passed through)
  metadata: Record<string, unknown>;  // title, item_url, author, ...
}

def wordpress_adapter(site: str):
    page = 1
    while posts := get_json(f"{site}/wp-json/wp/v2/posts?page={page}"):
        for post in posts:
            yield ContentItem(
                producer_id=f"wordpress-post-{post['id']}",
                filename=f"post-{post['id']}.md",
                text=html_to_markdown(post["content"]["rendered"]),
                metadata={"item_title": post["title"]["rendered"],
                          "item_url": post["link"],
                          "publication_date": post["date"]},
            )
        page += 1

async function* wordpressAdapter(site: string): AsyncGenerator<ContentItem> {
  for (let page = 1; ; page++) {
    const posts = await getJson(`${site}/wp-json/wp/v2/posts?page=${page}`);
    if (posts.length === 0) return;
    for (const post of posts) {
      yield {
        producerId: `wordpress-post-${post.id}`,
        filename: `post-${post.id}.md`,
        text: htmlToMarkdown(post.content.rendered),
        metadata: { item_title: post.title.rendered,
                    item_url: post.link,
                    publication_date: post.date },
      };
    }
  }
}

def notion_adapter(notion, root_page_id: str):
    for page in walk_pages(notion, root_page_id):   # recursive traversal
        yield ContentItem(
            producer_id=f"notion-page-{page['id']}",
            filename=f"{page['id']}.md",
            text=blocks_to_markdown(notion.blocks.children.list(page["id"])),
            metadata={"item_title": page_title(page),
                      "item_url": page["url"]},
        )

async function* notionAdapter(notion: Client, rootPageId: string): AsyncGenerator<ContentItem> {
  for await (const page of walkPages(notion, rootPageId)) {  // recursive traversal
    const blocks = await notion.blocks.children.list({ block_id: page.id });
    yield {
      producerId: `notion-page-${page.id}`,
      filename: `${page.id}.md`,
      text: blocksToMarkdown(blocks),
      metadata: { item_title: pageTitle(page), item_url: page.url },
    };
  }
}

def kb_adapter(db):
    for row in db.execute("SELECT id, title, body, author, updated_at FROM kb_articles"):
        yield ContentItem(
            producer_id=f"kb-row-{row.id}",
            filename=f"kb-{row.id}.md",
            text=f"# {row.title}\n\n{row.body}",
            metadata={"item_title": row.title,
                      "author": [row.author]},
        )

async function* kbAdapter(db: Database): AsyncGenerator<ContentItem> {
  const rows = await db.query("SELECT id, title, body, author, updated_at FROM kb_articles");
  for (const row of rows) {
    yield {
      producerId: `kb-row-${row.id}`,
      filename: `kb-${row.id}.md`,
      text: `# ${row.title}\n\n${row.body}`,
      metadata: { item_title: row.title, author: [row.author] },
    };
  }
}

def s3_adapter(s3, bucket: str, prefix: str):
    for obj in s3.list_objects_v2(Bucket=bucket, Prefix=prefix)["Contents"]:
        if not obj["Key"].endswith((".md", ".txt", ".pdf", ".docx")):
            continue
        body = s3.get_object(Bucket=bucket, Key=obj["Key"])["Body"].read()
        yield ContentItem(
            producer_id=f"s3-{obj['Key']}",
            filename=obj["Key"].rsplit("/", 1)[-1],
            data=body,  # raw bytes uploaded as-is; Gloo detects the format
            metadata={"item_title": obj["Key"].rsplit("/", 1)[-1]},
        )

async function* s3Adapter(s3: S3Client, bucket: string, prefix: string): AsyncGenerator<ContentItem> {
  const { Contents } = await s3.send(new ListObjectsV2Command({ Bucket: bucket, Prefix: prefix }));
  for (const obj of Contents ?? []) {
    if (!/\.(md|txt|pdf|docx)$/.test(obj.Key!)) continue;
    const body = await s3.send(new GetObjectCommand({ Bucket: bucket, Key: obj.Key! }));
    yield {
      producerId: `s3-${obj.Key}`,
      filename: obj.Key!.split("/").pop()!,
      data: await body.Body!.transformToByteArray(),  // raw bytes uploaded as-is; Gloo detects the format
      metadata: { item_title: obj.Key!.split("/").pop()! },
    };
  }
}

def rss_adapter(feed_url: str, state: SyncState):
    for entry in parse_feed(feed_url).entries:
        if state.already_synced(entry.guid, entry.updated):
            continue
        yield ContentItem(
            producer_id=f"rss-{entry.guid}",
            filename=f"{slugify(entry.guid)}.md",
            text=html_to_markdown(entry.content),
            metadata={"item_title": entry.title,
                      "item_url": entry.link,
                      "publication_date": entry.published},
        )
        state.mark_synced(entry.guid, entry.updated)

async function* rssAdapter(feedUrl: string, state: SyncState): AsyncGenerator<ContentItem> {
  const feed = await parseFeed(feedUrl);
  for (const entry of feed.entries) {
    if (state.alreadySynced(entry.guid, entry.updated)) continue;
    yield {
      producerId: `rss-${entry.guid}`,
      filename: `${slugify(entry.guid)}.md`,
      text: htmlToMarkdown(entry.content),
      metadata: { item_title: entry.title,
                  item_url: entry.link,
                  publication_date: entry.published },
    };
    state.markSynced(entry.guid, entry.updated);
  }
}