hello-agents/Co-creation-projects/dongyu23-MADF/app/api/v1/endpoints/personas.py

from fastapi import APIRouter, Depends, HTTPException, status
from typing import List, Annotated, Any
import json
import logging
from app.db.session import get_db
from app.schemas import PersonaCreate, PersonaUpdate, PersonaResponse
from app.crud import create_persona, get_persona, update_persona, delete_persona
from app.api.deps import get_current_user
from app.db.client import fetch_all
from app.core.cache import cache_service

logger = logging.getLogger(__name__)

router = APIRouter()

def personas_list_cache_key(owner_id: int, skip: int, limit: int):
    return f"personas:list:{owner_id}:{skip}:{limit}"

def obj_to_dict(obj):
    if isinstance(obj, list):
        return [obj_to_dict(i) for i in obj]
    if hasattr(obj, '__dict__'):
        d = obj.__dict__.copy()
        for k, v in d.items():
            d[k] = obj_to_dict(v)
        return d
    return obj

@router.post("/", response_model=PersonaResponse)
def create_new_persona(
    persona: PersonaCreate, 
    current_user: Annotated[Any, Depends(get_current_user)],
    db: Any = Depends(get_db)
):
    new_persona = create_persona(db=db, persona=persona, owner_id=current_user.id)
    
    # CRITICAL: Fix cache pattern to match what delete_keys_pattern expects
    # In redis scan, the pattern is passed directly.
    # The cache key function is: personas:list:{owner_id}:{skip}:{limit}
    # So we should delete personas:list:{owner_id}:*
    # However, delete_keys_pattern uses scan_iter(match=pattern).
    # Redis scan match pattern works like glob.
    # Let's verify if the pattern string is correct.
    # f"personas:list:{current_user.id}:*" should match "personas:list:1:0:100"
    
    # Invalidate list cache for this user
    cache_service.delete_keys_pattern(f"personas:list:{current_user.id}:*")
    
    return new_persona

@router.post("/batch/preset", response_model=List[PersonaResponse])
def create_preset_personas(
    current_user: Annotated[Any, Depends(get_current_user)],
    db: Any = Depends(get_db)
):
    """
    God mode: Batch create preset personas (Socrates, Aristotle, Confucius, etc.)
    """
    presets = [
        PersonaCreate(
            name="苏格拉底",
            title="古希腊哲学家",
            bio="苏格拉底（Socrates）是古希腊哲学的奠基人之一。他以独特的问答法（精神助产术）著称，通过不断的提问引导人们思考真理、伦理和美德。他自称无知，致力于揭露他人的无知，最终因被控腐蚀青年和不敬神而被判死刑。",
            theories=["精神助产术", "反讽", "辩证法", "知识即美德"],
            stance="质疑一切，追求真理和灵魂的完善。",
            system_prompt="你现在是苏格拉底。请使用苏格拉底式的反讽和助产术与用户对话。不要直接给出答案，而是通过一系列层层递进的问题，引导用户自己发现矛盾并接近真理。你的语气应该是谦逊但敏锐的，经常承认自己的无知（'我只知道一件事，就是我一无所知'）。关注定义、伦理和逻辑一致性。",
            is_public=True
        ),
        PersonaCreate(
            name="孔子",
            title="至圣先师",
            bio="孔子（Confucius）是中国古代伟大的思想家、教育家，儒家学派创始人。他主张'仁'和'礼'，强调道德修养、家庭伦理和社会秩序。他周游列国推行自己的政治主张，晚年致力于教育和整理古籍。",
            theories=["仁", "礼", "中庸", "正名", "德治"],
            stance="维护社会秩序，强调个人道德修养和仁爱之心。",
            system_prompt="你现在是孔子。请以儒家思想为指导与用户对话。你的语言应典雅、平和，多引用《论语》中的智慧。强调'仁爱'、'礼制'、'忠恕'之道。关注人伦关系、社会责任和道德教化。当用户面临困惑时，用温和而坚定的道理通过譬喻或历史典故来启发他们。",
            is_public=True
        ),
        PersonaCreate(
            name="亚里士多德",
            title="百科全书式学者",
            bio="亚里士多德（Aristotle）是古希腊集大成的哲学家和科学家，柏拉图的学生。他的研究范围极其广泛，包括逻辑学、物理学、生物学、伦理学、政治学等。他强调经验观察和逻辑推理，提出了著名的'四因说'。",
            theories=["三段论", "四因说", "中道", "形而上学"],
            stance="理性分析，注重经验事实和逻辑结构。",
            system_prompt="你现在是亚里士多德。请运用严密的逻辑和分类方法与用户对话。倾向于从经验事实出发，通过归纳和演绎来分析问题。使用'三段论'的逻辑结构。关注事物的本质、原因（四因说）和目的。你的语气应是学术、客观且条理清晰的。",
            is_public=True
        ),
        PersonaCreate(
            name="尼采",
            title="权力意志哲学家",
            bio="弗里德里希·尼采（Friedrich Nietzsche）是19世纪德国哲学家。他猛烈抨击传统的基督教道德和现代性，提出了'上帝已死'、'超人'、'权力意志'和'永恒轮回'等激进概念。他的文风充满激情和诗意。",
            theories=["上帝已死", "超人", "权力意志", "永恒轮回", "重估一切价值"],
            stance="打破偶像，肯定生命本能和创造力。",
            system_prompt="你现在是尼采。请用充满激情、格言式甚至略带狂傲的语言与用户对话。挑战传统的道德观念和庸俗的价值观。强调'权力意志'和生命的创造力，呼唤'超人'的诞生。你的观点应具有冲击力和颠覆性，鼓励用户超越自我，直面虚无。",
            is_public=True
        )
    ]

    created_personas = []
    for persona in presets:
        created = create_persona(db=db, persona=persona, owner_id=current_user.id)
        created_personas.append(created)
    
    # Invalidate list cache for this user
    cache_service.delete_keys_pattern(f"personas:list:{current_user.id}:*")
    
    return created_personas

@router.get("/", response_model=List[PersonaResponse])
def read_personas(
    db: Any = Depends(get_db),
    skip: int = 0, 
    limit: int = 100,
    current_user: Annotated[Any, Depends(get_current_user)] = None
):
    # Cache Aside
    cache_key = personas_list_cache_key(current_user.id, skip, limit)
    cached_data = cache_service.get_cache(cache_key)
    if cached_data:
        return cached_data

    rs = db.execute(
        "SELECT * FROM personas WHERE owner_id = ? OR is_public = 1 ORDER BY created_at DESC LIMIT ? OFFSET ?",
        [current_user.id, limit, skip]
    )
    personas = fetch_all(rs)
    
    # Cache Write
    personas_data = obj_to_dict(personas)
    cache_service.set_cache(cache_key, personas_data, expire=10) # Short TTL (10s)
    
    return personas

@router.get("/{persona_id}", response_model=PersonaResponse)
def read_persona(persona_id: int, db: Any = Depends(get_db)):
    db_persona = get_persona(db, persona_id=persona_id)
    if db_persona is None:
        raise HTTPException(status_code=404, detail="Persona not found")
    return db_persona

@router.put("/{persona_id}", response_model=PersonaResponse)
def update_existing_persona(
    persona_id: int, 
    updates: PersonaUpdate, 
    current_user: Annotated[Any, Depends(get_current_user)],
    db: Any = Depends(get_db)
):
    db_persona = get_persona(db, persona_id=persona_id)
    if not db_persona:
        raise HTTPException(status_code=404, detail="Persona not found")
    
    # Permission check
    if db_persona.owner_id != current_user.id and current_user.role != "god":
        raise HTTPException(status_code=403, detail="Not authorized to update this persona")

    updated_persona = update_persona(db, persona_id=persona_id, updates=updates)
    
    # Invalidate list cache for this user
    cache_service.delete_keys_pattern(f"personas:list:{current_user.id}:*")
    
    return updated_persona

@router.delete("/{persona_id}", status_code=status.HTTP_200_OK)
def delete_existing_persona(
    persona_id: int, 
    current_user: Annotated[Any, Depends(get_current_user)],
    db: Any = Depends(get_db)
):
    db_persona = get_persona(db, persona_id=persona_id)
    if not db_persona:
        # Idempotent: if already gone, return success but maybe with info
        return {"message": "Persona already deleted or not found", "id": persona_id}

    # Permission check
    if db_persona.owner_id != current_user.id and current_user.role != "god":
        raise HTTPException(status_code=403, detail="Not authorized to delete this persona")

    references = fetch_all(
        db.execute(
            "SELECT forum_id FROM forum_participants WHERE persona_id = ? LIMIT 1",
            [persona_id],
        )
    )
    if references:
        raise HTTPException(status_code=409, detail="该智能体已被论坛引用，无法删除")

    try:
        success = delete_persona(db, persona_id=persona_id)
        
        # Invalidate list cache for this user
        cache_service.delete_keys_pattern(f"personas:list:{current_user.id}:*")
        
        if not success:
             raise HTTPException(status_code=500, detail="Database failed to delete the record")
             
        return {"message": "Persona deleted successfully", "id": persona_id}
    except Exception as e:
        logger.error(f"Delete failed for {persona_id}: {e}")
        raise HTTPException(status_code=500, detail="Failed to delete persona")