firefrost-services/services/modpack-version-checker/create_all_files.sh

#!/bin/bash
# Auto-generated script to create all modpack-version-checker files

BASE_DIR="/home/claude/firefrost-operations-manual/docs/tasks/modpack-version-checker/code/modpack-version-checker"
cd "$BASE_DIR"

# Already created: src/modpack_checker/__init__.py, src/modpack_checker/config.py

# Create database.py
cat > src/modpack_checker/database.py << 'EOF'
"""SQLite database layer using SQLAlchemy 2.0 ORM."""

from __future__ import annotations

from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
from typing import List, Optional

from sqlalchemy import Boolean, DateTime, ForeignKey, Integer, String, create_engine, select
from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column, relationship


class _Base(DeclarativeBase):
    pass


class _ModpackRow(_Base):
    __tablename__ = "modpacks"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    curseforge_id: Mapped[int] = mapped_column(Integer, unique=True, nullable=False)
    name: Mapped[str] = mapped_column(String(255), nullable=False)
    current_version: Mapped[Optional[str]] = mapped_column(String(255), nullable=True)
    last_checked: Mapped[Optional[datetime]] = mapped_column(DateTime, nullable=True)
    notification_enabled: Mapped[bool] = mapped_column(Boolean, default=True, nullable=False)

    history: Mapped[List["_CheckHistoryRow"]] = relationship(
        back_populates="modpack", cascade="all, delete-orphan"
    )


class _CheckHistoryRow(_Base):
    __tablename__ = "check_history"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    modpack_id: Mapped[int] = mapped_column(ForeignKey("modpacks.id"), nullable=False)
    checked_at: Mapped[datetime] = mapped_column(DateTime, nullable=False)
    version_found: Mapped[Optional[str]] = mapped_column(String(255), nullable=True)
    notification_sent: Mapped[bool] = mapped_column(Boolean, default=False, nullable=False)

    modpack: Mapped["_ModpackRow"] = relationship(back_populates="history")


@dataclass
class Modpack:
    id: int
    curseforge_id: int
    name: str
    current_version: Optional[str]
    last_checked: Optional[datetime]
    notification_enabled: bool

    @classmethod
    def _from_row(cls, row: _ModpackRow) -> "Modpack":
        return cls(
            id=row.id,
            curseforge_id=row.curseforge_id,
            name=row.name,
            current_version=row.current_version,
            last_checked=row.last_checked,
            notification_enabled=row.notification_enabled,
        )


@dataclass
class CheckHistory:
    id: int
    modpack_id: int
    checked_at: datetime
    version_found: Optional[str]
    notification_sent: bool

    @classmethod
    def _from_row(cls, row: _CheckHistoryRow) -> "CheckHistory":
        return cls(
            id=row.id,
            modpack_id=row.modpack_id,
            checked_at=row.checked_at,
            version_found=row.version_found,
            notification_sent=row.notification_sent,
        )


class Database:
    def __init__(self, db_path: str) -> None:
        Path(db_path).parent.mkdir(parents=True, exist_ok=True)
        self.engine = create_engine(f"sqlite:///{db_path}", echo=False)
        _Base.metadata.create_all(self.engine)

    def add_modpack(self, curseforge_id: int, name: str) -> Modpack:
        with Session(self.engine) as session:
            existing = session.scalar(
                select(_ModpackRow).where(_ModpackRow.curseforge_id == curseforge_id)
            )
            if existing is not None:
                raise ValueError(f"Modpack ID {curseforge_id} is already being tracked.")
            row = _ModpackRow(curseforge_id=curseforge_id, name=name, notification_enabled=True)
            session.add(row)
            session.commit()
            return Modpack._from_row(row)

    def remove_modpack(self, curseforge_id: int) -> bool:
        with Session(self.engine) as session:
            row = session.scalar(
                select(_ModpackRow).where(_ModpackRow.curseforge_id == curseforge_id)
            )
            if row is None:
                return False
            session.delete(row)
            session.commit()
            return True

    def update_version(self, curseforge_id: int, version: str, notification_sent: bool = False) -> None:
        with Session(self.engine) as session:
            row = session.scalar(
                select(_ModpackRow).where(_ModpackRow.curseforge_id == curseforge_id)
            )
            if row is None:
                raise ValueError(f"Modpack {curseforge_id} not found in database.")
            row.current_version = version
            row.last_checked = datetime.utcnow()
            session.add(
                _CheckHistoryRow(
                    modpack_id=row.id,
                    checked_at=datetime.utcnow(),
                    version_found=version,
                    notification_sent=notification_sent,
                )
            )
            session.commit()

    def toggle_notifications(self, curseforge_id: int, enabled: bool) -> bool:
        with Session(self.engine) as session:
            row = session.scalar(
                select(_ModpackRow).where(_ModpackRow.curseforge_id == curseforge_id)
            )
            if row is None:
                return False
            row.notification_enabled = enabled
            session.commit()
            return True

    def get_modpack(self, curseforge_id: int) -> Optional[Modpack]:
        with Session(self.engine) as session:
            row = session.scalar(
                select(_ModpackRow).where(_ModpackRow.curseforge_id == curseforge_id)
            )
            return Modpack._from_row(row) if row else None

    def get_all_modpacks(self) -> List[Modpack]:
        with Session(self.engine) as session:
            rows = session.scalars(select(_ModpackRow)).all()
            return [Modpack._from_row(r) for r in rows]

    def get_check_history(self, curseforge_id: int, limit: int = 10) -> List[CheckHistory]:
        with Session(self.engine) as session:
            modpack_row = session.scalar(
                select(_ModpackRow).where(_ModpackRow.curseforge_id == curseforge_id)
            )
            if modpack_row is None:
                return []
            rows = session.scalars(
                select(_CheckHistoryRow)
                .where(_CheckHistoryRow.modpack_id == modpack_row.id)
                .order_by(_CheckHistoryRow.checked_at.desc())
                .limit(limit)
            ).all()
            return [CheckHistory._from_row(r) for r in rows]
EOF

echo "Created database.py"