hunjing/backend/app/domains/hetu/service.py

"""河图世界域应用服务：境界 CRUD、实体与关系、境界正史视图。

铁律（core-domains.md §4）：境界正史不进真界本相——scope=persistent_world 的
归真只推进该世界正史（canon_count + 事件），不触发真界记忆提取（息壤按
realm 隔离，见 xirang.on_canonized 的 scope 分流）。
"""

from __future__ import annotations

import json

from sqlalchemy import func, select

from app.core.errors import AppError
from app.core.time import utc_now
from app.db.session import AsyncSessionLocal
from app.domains.hetu.models import PersistentWorld, WorldEntity, WorldRelation
from app.domains import one_qi
from app.models.mirror_tree import Canonization


async def create_world(
    *, owner_scope: str, name: str, description: str, rules: dict, lorebook_id: str | None
) -> dict:
    async with AsyncSessionLocal() as session:
        world = PersistentWorld(
            owner_scope=owner_scope,
            name=name,
            description=description,
            rules_json=json.dumps(rules or {}, ensure_ascii=False),
            lorebook_id=lorebook_id,
        )
        session.add(world)
        await session.commit()
        session.expunge(world)
        return _world_dict(world)


async def list_worlds(owner_scope: str) -> list[dict]:
    async with AsyncSessionLocal() as session:
        rows = (
            await session.execute(
                select(PersistentWorld)
                .where(PersistentWorld.owner_scope == owner_scope, PersistentWorld.status != "archived")
                .order_by(PersistentWorld.updated_at.desc())
            )
        ).scalars().all()
        return [_world_dict(r) for r in rows]


async def get_world(world_id: str, owner_scope: str) -> dict:
    return _world_dict(await _owned(world_id, owner_scope))


async def update_world(
    world_id: str, owner_scope: str, *, name: str | None, description: str | None,
    rules: dict | None, lorebook_id: str | None,
) -> dict:
    await _owned(world_id, owner_scope)
    async with AsyncSessionLocal() as session:
        world = await session.get(PersistentWorld, world_id)
        if name is not None:
            world.name = name
        if description is not None:
            world.description = description
        if rules is not None:
            world.rules_json = json.dumps(rules, ensure_ascii=False)
        if lorebook_id is not None:
            world.lorebook_id = lorebook_id
        world.updated_at = utc_now()
        await session.commit()
        session.expunge(world)
        return _world_dict(world)


async def archive_world(world_id: str, owner_scope: str) -> dict:
    await _owned(world_id, owner_scope)
    async with AsyncSessionLocal() as session:
        world = await session.get(PersistentWorld, world_id)
        world.status = "archived"
        await session.commit()
    return {"world_id": world_id, "status": "archived"}


# ── 实体与关系（河图世界认知） ──


async def add_entity(
    world_id: str, owner_scope: str, *, kind: str, name: str, description: str, attrs: dict
) -> dict:
    await _owned(world_id, owner_scope)
    async with AsyncSessionLocal() as session:
        entity = WorldEntity(
            world_id=world_id, kind=kind, name=name, description=description,
            attrs_json=json.dumps(attrs or {}, ensure_ascii=False),
        )
        session.add(entity)
        await session.commit()
        session.expunge(entity)
        return _entity_dict(entity)


async def list_entities(world_id: str, owner_scope: str) -> list[dict]:
    await _owned(world_id, owner_scope)
    async with AsyncSessionLocal() as session:
        rows = (
            await session.execute(
                select(WorldEntity).where(WorldEntity.world_id == world_id)
                .order_by(WorldEntity.created_at.asc())
            )
        ).scalars().all()
        return [_entity_dict(r) for r in rows]


async def add_relation(
    world_id: str, owner_scope: str, *, source_entity_id: str, target_entity_id: str,
    relation: str, note: str,
) -> dict:
    await _owned(world_id, owner_scope)
    async with AsyncSessionLocal() as session:
        for eid in (source_entity_id, target_entity_id):
            entity = await session.get(WorldEntity, eid)
            if entity is None or entity.world_id != world_id:
                raise AppError("invalid_request", f"实体 {eid} 不属于该境界。")
        row = WorldRelation(
            world_id=world_id, source_entity_id=source_entity_id,
            target_entity_id=target_entity_id, relation=relation, note=note,
        )
        session.add(row)
        await session.commit()
        session.expunge(row)
        return {
            "id": row.id, "source": source_entity_id, "target": target_entity_id,
            "relation": relation, "note": note,
        }


async def list_relations(world_id: str, owner_scope: str) -> list[dict]:
    await _owned(world_id, owner_scope)
    async with AsyncSessionLocal() as session:
        rows = (
            await session.execute(
                select(WorldRelation).where(WorldRelation.world_id == world_id)
                .order_by(WorldRelation.created_at.asc())
            )
        ).scalars().all()
        return [
            {
                "id": r.id, "source": r.source_entity_id, "target": r.target_entity_id,
                "relation": r.relation, "note": r.note,
            }
            for r in rows
        ]


# ── 境界正史 ──


async def world_chronicle(world_id: str, owner_scope: str) -> dict:
    """境界正史：该 world 下 scope=persistent_world 的归真集合（只读视图）。"""
    await _owned(world_id, owner_scope)
    async with AsyncSessionLocal() as session:
        rows = (
            await session.execute(
                select(Canonization)
                .where(Canonization.world_id == world_id, Canonization.scope == "persistent_world")
                .order_by(Canonization.created_at.asc())
            )
        ).scalars().all()
        return {
            "world_id": world_id,
            "entries": [
                {
                    "canonization_id": c.id,
                    "tree_id": c.tree_id,
                    "endpoint_node_id": c.endpoint_node_id,
                    "path_nodes": json.loads(c.path_nodes_json or "[]"),
                    "created_at": c.created_at.isoformat() if c.created_at else None,
                }
                for c in rows
            ],
        }


async def on_world_canonized(world_id: str) -> None:
    """境界归真后置：推进正史计数（由镜树 canonize 的 world 分支调用）。"""
    async with AsyncSessionLocal() as session:
        world = await session.get(PersistentWorld, world_id)
        if world is None:
            return
        count = (
            await session.execute(
                select(func.count()).select_from(Canonization).where(
                    Canonization.world_id == world_id,
                    Canonization.scope == "persistent_world",
                )
            )
        ).scalar_one()
        world.canon_count = int(count)
        world.updated_at = utc_now()
        await session.commit()
    await one_qi.emit(
        "world.canonized", character_id=None,
        payload={"world_id": world_id}, source="hetu",
    )


async def _owned(world_id: str, owner_scope: str) -> PersistentWorld:
    async with AsyncSessionLocal() as session:
        world = await session.get(PersistentWorld, world_id)
        if world is None or world.owner_scope != owner_scope:
            raise AppError("not_found", "境界不存在。", status_code=404)
        session.expunge(world)
        return world


def _world_dict(w: PersistentWorld) -> dict:
    return {
        "id": w.id,
        "name": w.name,
        "description": w.description,
        "rules": json.loads(w.rules_json or "{}"),
        "lorebook_id": w.lorebook_id,
        "canon_count": w.canon_count,
        "status": w.status,
        "created_at": w.created_at.isoformat() if w.created_at else None,
        "updated_at": w.updated_at.isoformat() if w.updated_at else None,
    }


def _entity_dict(e: WorldEntity) -> dict:
    return {
        "id": e.id,
        "kind": e.kind,
        "name": e.name,
        "description": e.description,
        "attrs": json.loads(e.attrs_json or "{}"),
    }