backend/create_blog_db_from_csv.py

import argparse
import csv
import json
import sqlite3
from pathlib import Path


BASE_DIR = Path(__file__).resolve().parent
DEFAULT_CSV_PATH = BASE_DIR / "blog_posts_rows.csv"
DEFAULT_DB_PATH = BASE_DIR / "blog.db"


def parse_bool(value: str) -> int:
    return 1 if str(value).strip().lower() in {"1", "true", "t", "yes", "y"} else 0


def normalize_tags(value: str) -> str:
    if not value:
        return "[]"

    try:
        parsed = json.loads(value)
    except json.JSONDecodeError:
        parsed = [tag.strip() for tag in value.split(",") if tag.strip()]

    if not isinstance(parsed, list):
        return "[]"

    return json.dumps([str(tag).strip() for tag in parsed if str(tag).strip()])


def parse_optional_int(value: str):
    cleaned = str(value or "").strip()
    return int(cleaned) if cleaned else None


def create_schema(conn: sqlite3.Connection) -> None:
    conn.executescript(
        """
        DROP TABLE IF EXISTS blog_post_images;
        DROP TABLE IF EXISTS images;
        DROP TABLE IF EXISTS blog_posts;

        CREATE TABLE blog_posts (
            id INTEGER PRIMARY KEY,
            title TEXT NOT NULL,
            content TEXT NOT NULL,
            author TEXT DEFAULT 'Admin',
            created_at TEXT DEFAULT CURRENT_TIMESTAMP,
            published INTEGER DEFAULT 1,
            tags TEXT DEFAULT '[]',
            featured_image_id INTEGER,
            category TEXT
        );

        CREATE TABLE images (
            id INTEGER PRIMARY KEY,
            filename TEXT NOT NULL,
            original_filename TEXT NOT NULL,
            file_path TEXT NOT NULL,
            file_size INTEGER,
            mime_type TEXT,
            alt_text TEXT DEFAULT '',
            caption TEXT DEFAULT '',
            width INTEGER,
            height INTEGER,
            created_at TEXT DEFAULT CURRENT_TIMESTAMP
        );

        CREATE TABLE blog_post_images (
            id INTEGER PRIMARY KEY,
            blog_post_id INTEGER,
            image_id INTEGER,
            image_type TEXT DEFAULT 'post_content',
            image_order INTEGER DEFAULT 0,
            position_in_content INTEGER,
            FOREIGN KEY (blog_post_id) REFERENCES blog_posts (id),
            FOREIGN KEY (image_id) REFERENCES images (id)
        );

        CREATE INDEX idx_blog_posts_published_created
            ON blog_posts (published, created_at DESC);
        CREATE INDEX idx_blog_posts_category
            ON blog_posts (category);
        """
    )


def import_posts(csv_path: Path, db_path: Path) -> int:
    db_path.parent.mkdir(parents=True, exist_ok=True)

    with sqlite3.connect(db_path) as conn:
        create_schema(conn)

        with csv_path.open(newline="", encoding="utf-8") as csv_file:
            reader = csv.DictReader(csv_file)
            rows = []
            for row in reader:
                title = (row.get("title") or "").strip()
                content = (row.get("content") or "").strip()
                if not title or not content:
                    continue

                rows.append(
                    (
                        parse_optional_int(row.get("id", "")),
                        title,
                        content,
                        (row.get("author") or "Admin").strip() or "Admin",
                        (row.get("created_at") or "").strip() or None,
                        parse_bool(row.get("published", "true")),
                        normalize_tags(row.get("tags", "")),
                        parse_optional_int(row.get("featured_image_id", "")),
                        (row.get("category") or "").strip() or None,
                    )
                )

        conn.executemany(
            """
            INSERT INTO blog_posts (
                id, title, content, author, created_at, published,
                tags, featured_image_id, category
            )
            VALUES (?, ?, ?, ?, COALESCE(?, CURRENT_TIMESTAMP), ?, ?, ?, ?)
            """,
            rows,
        )
        conn.commit()

    return len(rows)


def main() -> None:
    parser = argparse.ArgumentParser(description="Create the local blog SQLite DB from CSV export.")
    parser.add_argument("--csv", type=Path, default=DEFAULT_CSV_PATH, help="Path to blog_posts_rows.csv")
    parser.add_argument("--db", type=Path, default=DEFAULT_DB_PATH, help="Path to output SQLite database")
    args = parser.parse_args()

    count = import_posts(args.csv, args.db)
    print(f"Imported {count} blog posts into {args.db}")


if __name__ == "__main__":
    main()