cosmo/backend/app/services/star_system_service.py

"""
StarSystem Service
恒星系统服务层
"""
from typing import List, Optional
from sqlalchemy import select, func, update, delete, or_
from sqlalchemy.ext.asyncio import AsyncSession
from app.models.db.star_system import StarSystem
from app.models.db.celestial_body import CelestialBody


class StarSystemService:
    """恒星系统服务"""

    @staticmethod
    async def get_all(
        db: AsyncSession,
        skip: int = 0,
        limit: int = 100,
        exclude_solar: bool = False,
        search: Optional[str] = None
    ) -> List[StarSystem]:
        """
        获取所有恒星系统

        Args:
            db: 数据库会话
            skip: 跳过记录数
            limit: 返回记录数
            exclude_solar: 是否排除太阳系
            search: 搜索关键词（匹配名称）
        """
        query = select(StarSystem).order_by(StarSystem.distance_pc.asc().nulls_first())

        # 排除太阳系
        if exclude_solar:
            query = query.where(StarSystem.id != 1)

        # 搜索
        if search:
            search_pattern = f"%{search}%"
            query = query.where(
                or_(
                    StarSystem.name.ilike(search_pattern),
                    StarSystem.name_zh.ilike(search_pattern),
                    StarSystem.host_star_name.ilike(search_pattern)
                )
            )

        query = query.offset(skip).limit(limit)
        result = await db.execute(query)
        return list(result.scalars().all())

    @staticmethod
    async def get_by_id(db: AsyncSession, system_id: int) -> Optional[StarSystem]:
        """根据ID获取恒星系统"""
        result = await db.execute(
            select(StarSystem).where(StarSystem.id == system_id)
        )
        return result.scalar_one_or_none()

    @staticmethod
    async def get_by_name(db: AsyncSession, name: str) -> Optional[StarSystem]:
        """根据名称获取恒星系统"""
        result = await db.execute(
            select(StarSystem).where(StarSystem.name == name)
        )
        return result.scalar_one_or_none()

    @staticmethod
    async def create(db: AsyncSession, system_data: dict) -> StarSystem:
        """创建恒星系统"""
        system = StarSystem(**system_data)
        db.add(system)
        await db.commit()
        await db.refresh(system)
        return system

    @staticmethod
    async def update(db: AsyncSession, system_id: int, system_data: dict) -> Optional[StarSystem]:
        """更新恒星系统"""
        result = await db.execute(
            select(StarSystem).where(StarSystem.id == system_id)
        )
        system = result.scalar_one_or_none()

        if not system:
            return None

        for key, value in system_data.items():
            if hasattr(system, key):
                setattr(system, key, value)

        await db.commit()
        await db.refresh(system)
        return system

    @staticmethod
    async def delete_system(db: AsyncSession, system_id: int) -> bool:
        """
        删除恒星系统（级联删除所有关联天体）
        不允许删除太阳系（id=1）
        """
        if system_id == 1:
            raise ValueError("不能删除太阳系")

        result = await db.execute(
            delete(StarSystem).where(StarSystem.id == system_id)
        )
        await db.commit()
        return result.rowcount > 0

    @staticmethod
    async def get_with_bodies(db: AsyncSession, system_id: int) -> Optional[dict]:
        """
        获取恒星系统及其所有天体

        Returns:
            包含 system 和 bodies 的字典
        """
        # 获取恒星系统
        system_result = await db.execute(
            select(StarSystem).where(StarSystem.id == system_id)
        )
        system = system_result.scalar_one_or_none()

        if not system:
            return None

        # 获取关联的天体（仅返回活跃状态的天体）
        bodies_result = await db.execute(
            select(CelestialBody)
            .where(CelestialBody.system_id == system_id)
            .where(CelestialBody.is_active == True)
            .order_by(CelestialBody.type, CelestialBody.name)
        )
        bodies = list(bodies_result.scalars().all())

        return {
            "system": system,
            "bodies": bodies,
            "body_count": len(bodies)
        }

    @staticmethod
    async def update_planet_count(db: AsyncSession, system_id: int) -> None:
        """更新恒星系统的行星数量统计"""
        result = await db.execute(
            select(func.count(CelestialBody.id))
            .where(CelestialBody.system_id == system_id)
            .where(CelestialBody.type != 'star')  # 排除恒星本身
        )
        count = result.scalar()

        await db.execute(
            update(StarSystem)
            .where(StarSystem.id == system_id)
            .values(planet_count=count)
        )
        await db.commit()

    @staticmethod
    async def get_statistics(db: AsyncSession) -> dict:
        """获取恒星系统统计信息"""
        # 总恒星系统数
        total_systems_result = await db.execute(select(func.count(StarSystem.id)))
        total_systems = total_systems_result.scalar()

        # 系外恒星系统数
        exo_systems_result = await db.execute(
            select(func.count(StarSystem.id)).where(StarSystem.id != 1)
        )
        exo_systems = exo_systems_result.scalar()

        # 总行星数
        total_planets_result = await db.execute(
            select(func.count(CelestialBody.id))
            .where(CelestialBody.type == 'planet')
        )
        total_planets = total_planets_result.scalar()

        # 系外行星数
        exo_planets_result = await db.execute(
            select(func.count(CelestialBody.id))
            .where(CelestialBody.type == 'planet')
            .where(CelestialBody.system_id > 1)
        )
        exo_planets = exo_planets_result.scalar()

        # 距离最近的10个恒星系统
        nearest_systems_result = await db.execute(
            select(StarSystem.name, StarSystem.name_zh, StarSystem.distance_ly, StarSystem.planet_count)
            .where(StarSystem.id != 1)
            .order_by(StarSystem.distance_pc.asc())
            .limit(10)
        )
        nearest_systems = [
            {
                "name": name,
                "name_zh": name_zh,
                "distance_ly": distance_ly,
                "planet_count": planet_count
            }
            for name, name_zh, distance_ly, planet_count in nearest_systems_result
        ]

        return {
            "total_systems": total_systems,
            "exo_systems": exo_systems,
            "total_planets": total_planets,
            "exo_planets": exo_planets,
            "solar_system_planets": total_planets - exo_planets,
            "nearest_systems": nearest_systems
        }


# 创建服务实例
star_system_service = StarSystemService()