update: 1.新增职业管理模块和角色职业关联 2.章节分析自动更新角色职业状态 3.优化章节生成的角色信息构建 4.批量生成强制开启同步分析 5.章节内容批量生成增加系统提示

This commit is contained in:
xiamuceer
2025-12-22 19:53:31 +08:00
parent 6886d903fe
commit b2dec41464
25 changed files with 4635 additions and 89 deletions
+909
View File
@@ -0,0 +1,909 @@
"""职业管理API"""
from fastapi import APIRouter, Depends, HTTPException, Request
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, func, and_
import json
from typing import AsyncGenerator
from app.database import get_db
from app.utils.sse_response import SSEResponse, create_sse_response
from app.models.career import Career, CharacterCareer
from app.models.character import Character
from app.models.project import Project
from app.schemas.career import (
CareerCreate,
CareerUpdate,
CareerResponse,
CareerListResponse,
CareerGenerateRequest,
CharacterCareerResponse,
CharacterCareerDetail,
SetMainCareerRequest,
AddSubCareerRequest,
UpdateCareerStageRequest,
CareerStage
)
from app.services.ai_service import AIService
from app.logger import get_logger
from app.api.settings import get_user_ai_service
router = APIRouter(prefix="/careers", tags=["职业管理"])
logger = get_logger(__name__)
async def verify_project_access(project_id: str, user_id: str, db: AsyncSession) -> Project:
"""验证用户是否有权访问指定项目"""
if not user_id:
raise HTTPException(status_code=401, detail="未登录")
result = await db.execute(
select(Project).where(
Project.id == project_id,
Project.user_id == user_id
)
)
project = result.scalar_one_or_none()
if not project:
logger.warning(f"项目访问被拒绝: project_id={project_id}, user_id={user_id}")
raise HTTPException(status_code=404, detail="项目不存在或无权访问")
return project
@router.get("", response_model=CareerListResponse, summary="获取职业列表")
async def get_careers(
project_id: str,
request: Request,
db: AsyncSession = Depends(get_db)
):
"""获取指定项目的所有职业"""
user_id = getattr(request.state, 'user_id', None)
await verify_project_access(project_id, user_id, db)
# 获取总数
count_result = await db.execute(
select(func.count(Career.id)).where(Career.project_id == project_id)
)
total = count_result.scalar_one()
# 获取职业列表
result = await db.execute(
select(Career)
.where(Career.project_id == project_id)
.order_by(Career.type, Career.created_at.desc())
)
careers = result.scalars().all()
# 分类返回
main_careers = []
sub_careers = []
for career in careers:
# 解析JSON字段
stages = json.loads(career.stages) if career.stages else []
attribute_bonuses = json.loads(career.attribute_bonuses) if career.attribute_bonuses else None
career_dict = {
"id": career.id,
"project_id": career.project_id,
"name": career.name,
"type": career.type,
"description": career.description,
"category": career.category,
"stages": stages,
"max_stage": career.max_stage,
"requirements": career.requirements,
"special_abilities": career.special_abilities,
"worldview_rules": career.worldview_rules,
"attribute_bonuses": attribute_bonuses,
"source": career.source,
"created_at": career.created_at,
"updated_at": career.updated_at
}
if career.type == "main":
main_careers.append(career_dict)
else:
sub_careers.append(career_dict)
return CareerListResponse(
total=total,
main_careers=main_careers,
sub_careers=sub_careers
)
@router.post("", response_model=CareerResponse, summary="创建职业")
async def create_career(
career_data: CareerCreate,
request: Request,
db: AsyncSession = Depends(get_db)
):
"""手动创建职业"""
user_id = getattr(request.state, 'user_id', None)
await verify_project_access(career_data.project_id, user_id, db)
try:
# 转换stages为JSON字符串
stages_json = json.dumps([stage.model_dump() for stage in career_data.stages], ensure_ascii=False)
attribute_bonuses_json = json.dumps(career_data.attribute_bonuses, ensure_ascii=False) if career_data.attribute_bonuses else None
# 创建职业
career = Career(
project_id=career_data.project_id,
name=career_data.name,
type=career_data.type,
description=career_data.description,
category=career_data.category,
stages=stages_json,
max_stage=career_data.max_stage,
requirements=career_data.requirements,
special_abilities=career_data.special_abilities,
worldview_rules=career_data.worldview_rules,
attribute_bonuses=attribute_bonuses_json,
source=career_data.source
)
db.add(career)
await db.commit()
await db.refresh(career)
logger.info(f"✅ 创建职业成功:{career.name} (ID: {career.id}, 类型: {career.type})")
return CareerResponse(
id=career.id,
project_id=career.project_id,
name=career.name,
type=career.type,
description=career.description,
category=career.category,
stages=career_data.stages,
max_stage=career.max_stage,
requirements=career.requirements,
special_abilities=career.special_abilities,
worldview_rules=career.worldview_rules,
attribute_bonuses=career_data.attribute_bonuses,
source=career.source,
created_at=career.created_at,
updated_at=career.updated_at
)
except Exception as e:
logger.error(f"创建职业失败: {str(e)}")
raise HTTPException(status_code=500, detail=f"创建职业失败: {str(e)}")
@router.get("/generate-system", summary="AI生成新职业(增量式,流式)")
async def generate_career_system(
project_id: str,
main_career_count: int = 3,
sub_career_count: int = 6,
enable_mcp: bool = False,
http_request: Request = None,
db: AsyncSession = Depends(get_db),
user_ai_service: AIService = Depends(get_user_ai_service)
):
"""
使用AI生成新职业(增量式,基于已有职业补充,支持SSE流式进度显示)
通过Server-Sent Events返回实时进度信息
"""
async def generate() -> AsyncGenerator[str, None]:
try:
# 验证用户权限和项目是否存在
user_id = getattr(http_request.state, 'user_id', None)
project = await verify_project_access(project_id, user_id, db)
yield await SSEResponse.send_progress("开始生成新职业...", 0)
# 获取已有职业列表
yield await SSEResponse.send_progress("分析已有职业...", 5)
existing_careers_result = await db.execute(
select(Career).where(Career.project_id == project_id)
)
existing_careers = existing_careers_result.scalars().all()
# 构建已有职业摘要
existing_main_careers = []
existing_sub_careers = []
for career in existing_careers:
career_summary = f"- {career.name}{career.category or '未分类'}{career.max_stage}阶)"
if career.description:
career_summary += f": {career.description[:50]}"
if career.type == "main":
existing_main_careers.append(career_summary)
else:
existing_sub_careers.append(career_summary)
existing_careers_text = ""
if existing_main_careers:
existing_careers_text += f"\n已有主职业({len(existing_main_careers)}个):\n" + "\n".join(existing_main_careers)
if existing_sub_careers:
existing_careers_text += f"\n\n已有副职业({len(existing_sub_careers)}个):\n" + "\n".join(existing_sub_careers)
if not existing_careers_text:
existing_careers_text = "\n当前还没有任何职业,这是第一次创建职业体系。"
# 构建项目上下文
yield await SSEResponse.send_progress("分析项目世界观...", 15)
project_context = f"""
项目信息:
- 书名:{project.title}
- 类型:{project.genre or '未设定'}
- 主题:{project.theme or '未设定'}
- 时间背景:{project.world_time_period or '未设定'}
- 地理位置:{project.world_location or '未设定'}
- 氛围基调:{project.world_atmosphere or '未设定'}
- 世界规则:{project.world_rules or '未设定'}
"""
user_requirements = f"""
已有职业情况:{existing_careers_text}
生成要求(增量式):
- 本次新增主职业:{main_career_count}
- 本次新增副职业:{sub_career_count}
- ⚠️ 重要:请生成与已有职业**不重复**的新职业,形成互补体系
- 新职业应填补已有职业体系的空缺,丰富职业多样性
- 主职业必须严格符合世界观规则,体现核心能力体系
- 副职业可以更加自由灵活,包含生产、辅助、特殊类型
"""
yield await SSEResponse.send_progress("构建AI提示词...", 20)
# 构建提示词
prompt = f"""{project_context}
{user_requirements}
请为这个小说项目生成新的补充职业(增量式)。要求:
1. **仔细分析已有职业**,避免生成重复或相似的职业
2. **填补职业体系的空缺**,让职业体系更加完善和多样化
3. 如果已有职业较少,可以生成核心基础职业
4. 如果已有职业较多,可以生成特色化、专精化的职业
返回JSON格式,结构如下:
{{
"main_careers": [
{{
"name": "职业名称",
"description": "职业描述",
"category": "职业分类(如:战斗系、法术系等)",
"stages": [
{{"level": 1, "name": "阶段名称", "description": "阶段描述"}},
{{"level": 2, "name": "阶段名称", "description": "阶段描述"}},
...
],
"max_stage": 10,
"requirements": "职业要求",
"special_abilities": "特殊能力",
"worldview_rules": "世界观规则关联",
"attribute_bonuses": {{"strength": "+10%", "intelligence": "+5%"}}
}}
],
"sub_careers": [
{{
"name": "副职业名称",
"description": "职业描述",
"category": "生产系/辅助系/特殊系",
"stages": [...],
"max_stage": 5,
"requirements": "职业要求",
"special_abilities": "特殊能力"
}}
]
}}
注意事项:
1. **避免重复**:生成的职业名称和定位不能与已有职业重复
2. **互补性**:新职业应与已有职业形成互补,丰富职业体系
3. 主职业的阶段设定要详细,体现明确的成长路径
4. 阶段名称要符合世界观特色
5. 副职业可以相对简化,但要有独特性
6. 所有职业都要符合项目的整体世界观设定
7. 只返回纯JSON,不要添加任何解释文字
"""
yield await SSEResponse.send_progress("调用AI生成新职业...", 30)
logger.info(f"🎯 开始为项目 {project_id} 生成新职业(增量式,已有{len(existing_careers)}个职业)")
try:
# 调用AI生成
result = await user_ai_service.generate_text(prompt=prompt)
ai_response = result.get('content', '') if isinstance(result, dict) else result
except Exception as ai_error:
logger.error(f"❌ AI服务调用异常:{str(ai_error)}")
yield await SSEResponse.send_error(f"AI服务调用失败:{str(ai_error)}")
return
if not ai_response or not ai_response.strip():
yield await SSEResponse.send_error("AI服务返回空响应")
return
yield await SSEResponse.send_progress("解析AI响应...", 50)
# 清洗并解析JSON
try:
cleaned_response = user_ai_service._clean_json_response(ai_response)
career_data = json.loads(cleaned_response)
logger.info(f"✅ 职业体系JSON解析成功")
except json.JSONDecodeError as e:
logger.error(f"❌ 职业体系JSON解析失败: {e}")
logger.error(f" 原始响应预览: {ai_response[:200]}")
yield await SSEResponse.send_error(f"AI返回的内容无法解析为JSON{str(e)}")
return
yield await SSEResponse.send_progress("保存主职业...", 60)
# 保存主职业
main_careers_created = []
for idx, career_info in enumerate(career_data.get("main_careers", [])):
try:
stages_json = json.dumps(career_info.get("stages", []), ensure_ascii=False)
attribute_bonuses = career_info.get("attribute_bonuses")
attribute_bonuses_json = json.dumps(attribute_bonuses, ensure_ascii=False) if attribute_bonuses else None
career = Career(
project_id=project_id,
name=career_info.get("name", f"未命名主职业{idx+1}"),
type="main",
description=career_info.get("description"),
category=career_info.get("category"),
stages=stages_json,
max_stage=career_info.get("max_stage", 10),
requirements=career_info.get("requirements"),
special_abilities=career_info.get("special_abilities"),
worldview_rules=career_info.get("worldview_rules"),
attribute_bonuses=attribute_bonuses_json,
source="ai"
)
db.add(career)
await db.flush()
main_careers_created.append(career.name)
logger.info(f" ✅ 创建主职业:{career.name}")
except Exception as e:
logger.error(f" ❌ 创建主职业失败:{str(e)}")
continue
yield await SSEResponse.send_progress("保存副职业...", 80)
# 保存副职业
sub_careers_created = []
for idx, career_info in enumerate(career_data.get("sub_careers", [])):
try:
stages_json = json.dumps(career_info.get("stages", []), ensure_ascii=False)
attribute_bonuses = career_info.get("attribute_bonuses")
attribute_bonuses_json = json.dumps(attribute_bonuses, ensure_ascii=False) if attribute_bonuses else None
career = Career(
project_id=project_id,
name=career_info.get("name", f"未命名副职业{idx+1}"),
type="sub",
description=career_info.get("description"),
category=career_info.get("category"),
stages=stages_json,
max_stage=career_info.get("max_stage", 5),
requirements=career_info.get("requirements"),
special_abilities=career_info.get("special_abilities"),
worldview_rules=career_info.get("worldview_rules"),
attribute_bonuses=attribute_bonuses_json,
source="ai"
)
db.add(career)
await db.flush()
sub_careers_created.append(career.name)
logger.info(f" ✅ 创建副职业:{career.name}")
except Exception as e:
logger.error(f" ❌ 创建副职业失败:{str(e)}")
continue
await db.commit()
total_main = len(existing_main_careers) + len(main_careers_created)
total_sub = len(existing_sub_careers) + len(sub_careers_created)
logger.info(f"🎉 新职业生成完成:新增主职业{len(main_careers_created)}个,新增副职业{len(sub_careers_created)}")
logger.info(f" 职业体系总数:主职业{total_main}个,副职业{total_sub}")
yield await SSEResponse.send_progress(f"新职业生成完成!(主职业{total_main}个,副职业{total_sub}个)", 100, "success")
# 发送结果数据
yield await SSEResponse.send_result({
"main_careers_count": len(main_careers_created),
"sub_careers_count": len(sub_careers_created),
"main_careers": main_careers_created,
"sub_careers": sub_careers_created
})
yield await SSEResponse.send_done()
except HTTPException as he:
logger.error(f"HTTP异常: {he.detail}")
yield await SSEResponse.send_error(he.detail, he.status_code)
except Exception as e:
logger.error(f"生成职业体系失败: {str(e)}")
yield await SSEResponse.send_error(f"生成新职业失败: {str(e)}")
return create_sse_response(generate())
@router.put("/{career_id}", response_model=CareerResponse, summary="更新职业")
async def update_career(
career_id: str,
career_update: CareerUpdate,
request: Request,
db: AsyncSession = Depends(get_db)
):
"""更新职业信息"""
result = await db.execute(
select(Career).where(Career.id == career_id)
)
career = result.scalar_one_or_none()
if not career:
raise HTTPException(status_code=404, detail="职业不存在")
# 验证用户权限
user_id = getattr(request.state, 'user_id', None)
await verify_project_access(career.project_id, user_id, db)
# 更新字段
update_data = career_update.model_dump(exclude_unset=True)
for field, value in update_data.items():
if field == "stages" and value is not None:
# 转换为JSON字符串
setattr(career, field, json.dumps([stage.model_dump() for stage in value], ensure_ascii=False))
elif field == "attribute_bonuses" and value is not None:
# 转换为JSON字符串
setattr(career, field, json.dumps(value, ensure_ascii=False))
else:
setattr(career, field, value)
await db.commit()
await db.refresh(career)
logger.info(f"✅ 更新职业成功:{career.name} (ID: {career_id})")
# 解析JSON返回
stages = json.loads(career.stages) if career.stages else []
attribute_bonuses = json.loads(career.attribute_bonuses) if career.attribute_bonuses else None
return CareerResponse(
id=career.id,
project_id=career.project_id,
name=career.name,
type=career.type,
description=career.description,
category=career.category,
stages=stages,
max_stage=career.max_stage,
requirements=career.requirements,
special_abilities=career.special_abilities,
worldview_rules=career.worldview_rules,
attribute_bonuses=attribute_bonuses,
source=career.source,
created_at=career.created_at,
updated_at=career.updated_at
)
@router.delete("/{career_id}", summary="删除职业")
async def delete_career(
career_id: str,
request: Request,
db: AsyncSession = Depends(get_db)
):
"""删除职业"""
result = await db.execute(
select(Career).where(Career.id == career_id)
)
career = result.scalar_one_or_none()
if not career:
raise HTTPException(status_code=404, detail="职业不存在")
# 验证用户权限
user_id = getattr(request.state, 'user_id', None)
await verify_project_access(career.project_id, user_id, db)
# 检查是否有角色使用该职业
char_career_result = await db.execute(
select(func.count(CharacterCareer.id)).where(CharacterCareer.career_id == career_id)
)
usage_count = char_career_result.scalar_one()
if usage_count > 0:
raise HTTPException(
status_code=400,
detail=f"该职业被{usage_count}个角色使用,无法删除。请先移除角色的职业关联。"
)
await db.delete(career)
await db.commit()
logger.info(f"✅ 删除职业成功:{career.name} (ID: {career_id})")
return {"message": "职业删除成功"}
@router.get("/{career_id}", response_model=CareerResponse, summary="获取职业详情")
async def get_career(
career_id: str,
request: Request,
db: AsyncSession = Depends(get_db)
):
"""根据ID获取职业详情"""
result = await db.execute(
select(Career).where(Career.id == career_id)
)
career = result.scalar_one_or_none()
if not career:
raise HTTPException(status_code=404, detail="职业不存在")
# 验证用户权限
user_id = getattr(request.state, 'user_id', None)
await verify_project_access(career.project_id, user_id, db)
# 解析JSON字段
stages = json.loads(career.stages) if career.stages else []
attribute_bonuses = json.loads(career.attribute_bonuses) if career.attribute_bonuses else None
return CareerResponse(
id=career.id,
project_id=career.project_id,
name=career.name,
type=career.type,
description=career.description,
category=career.category,
stages=stages,
max_stage=career.max_stage,
requirements=career.requirements,
special_abilities=career.special_abilities,
worldview_rules=career.worldview_rules,
attribute_bonuses=attribute_bonuses,
source=career.source,
created_at=career.created_at,
updated_at=career.updated_at
)
# ===== 角色职业关联API =====
@router.get("/character/{character_id}/careers", response_model=CharacterCareerResponse, summary="获取角色的职业信息")
async def get_character_careers(
character_id: str,
request: Request,
db: AsyncSession = Depends(get_db)
):
"""获取角色的所有职业信息(主职业和副职业)"""
# 验证角色存在
char_result = await db.execute(
select(Character).where(Character.id == character_id)
)
character = char_result.scalar_one_or_none()
if not character:
raise HTTPException(status_code=404, detail="角色不存在")
# 验证用户权限
user_id = getattr(request.state, 'user_id', None)
await verify_project_access(character.project_id, user_id, db)
# 获取角色的所有职业关联
result = await db.execute(
select(CharacterCareer, Career)
.join(Career, CharacterCareer.career_id == Career.id)
.where(CharacterCareer.character_id == character_id)
.order_by(CharacterCareer.career_type.desc()) # main排在前
)
career_relations = result.all()
main_career = None
sub_careers = []
for char_career, career in career_relations:
# 解析职业的阶段信息
stages = json.loads(career.stages) if career.stages else []
# 找到当前阶段信息
stage_name = "未知阶段"
stage_description = None
for stage in stages:
if stage.get("level") == char_career.current_stage:
stage_name = stage.get("name", f"{char_career.current_stage}阶段")
stage_description = stage.get("description")
break
career_detail = CharacterCareerDetail(
id=char_career.id,
character_id=char_career.character_id,
career_id=char_career.career_id,
career_name=career.name,
career_type=char_career.career_type,
current_stage=char_career.current_stage,
stage_name=stage_name,
stage_description=stage_description,
stage_progress=char_career.stage_progress,
max_stage=career.max_stage,
started_at=char_career.started_at,
reached_current_stage_at=char_career.reached_current_stage_at,
notes=char_career.notes,
created_at=char_career.created_at,
updated_at=char_career.updated_at
)
if char_career.career_type == "main":
main_career = career_detail
else:
sub_careers.append(career_detail)
return CharacterCareerResponse(
main_career=main_career,
sub_careers=sub_careers
)
@router.post("/character/{character_id}/careers/main", summary="设置角色主职业")
async def set_main_career(
character_id: str,
career_request: SetMainCareerRequest,
request: Request,
db: AsyncSession = Depends(get_db)
):
"""设置或更换角色的主职业"""
# 验证角色存在
char_result = await db.execute(
select(Character).where(Character.id == character_id)
)
character = char_result.scalar_one_or_none()
if not character:
raise HTTPException(status_code=404, detail="角色不存在")
# 验证用户权限
user_id = getattr(request.state, 'user_id', None)
await verify_project_access(character.project_id, user_id, db)
# 验证职业存在且为主职业类型
career_result = await db.execute(
select(Career).where(
Career.id == career_request.career_id,
Career.project_id == character.project_id
)
)
career = career_result.scalar_one_or_none()
if not career:
raise HTTPException(status_code=404, detail="职业不存在")
if career.type != "main":
raise HTTPException(status_code=400, detail="该职业不是主职业类型,无法设置为主职业")
# 验证阶段有效性
if career_request.current_stage > career.max_stage:
raise HTTPException(
status_code=400,
detail=f"阶段超出范围,该职业最大阶段为{career.max_stage}"
)
# 检查是否已有主职业
existing_main = await db.execute(
select(CharacterCareer).where(
CharacterCareer.character_id == character_id,
CharacterCareer.career_type == "main"
)
)
current_main = existing_main.scalar_one_or_none()
if current_main:
# 删除旧的主职业
await db.delete(current_main)
logger.info(f" 移除旧主职业关联: {current_main.career_id}")
# 创建新的主职业关联
char_career = CharacterCareer(
character_id=character_id,
career_id=career_request.career_id,
career_type="main",
current_stage=career_request.current_stage,
stage_progress=0,
started_at=career_request.started_at,
reached_current_stage_at=career_request.started_at
)
db.add(char_career)
await db.commit()
logger.info(f"✅ 设置主职业成功:角色{character.name} -> {career.name}(第{career_request.current_stage}阶段)")
return {"message": "主职业设置成功", "career_name": career.name}
@router.post("/character/{character_id}/careers/sub", summary="添加角色副职业")
async def add_sub_career(
character_id: str,
career_request: AddSubCareerRequest,
request: Request,
db: AsyncSession = Depends(get_db)
):
"""为角色添加副职业"""
# 验证角色存在
char_result = await db.execute(
select(Character).where(Character.id == character_id)
)
character = char_result.scalar_one_or_none()
if not character:
raise HTTPException(status_code=404, detail="角色不存在")
# 验证用户权限
user_id = getattr(request.state, 'user_id', None)
await verify_project_access(character.project_id, user_id, db)
# 验证职业存在且为副职业类型
career_result = await db.execute(
select(Career).where(
Career.id == career_request.career_id,
Career.project_id == character.project_id
)
)
career = career_result.scalar_one_or_none()
if not career:
raise HTTPException(status_code=404, detail="职业不存在")
if career.type != "sub":
raise HTTPException(status_code=400, detail="该职业不是副职业类型,无法添加为副职业")
# 验证阶段有效性
if career_request.current_stage > career.max_stage:
raise HTTPException(
status_code=400,
detail=f"阶段超出范围,该职业最大阶段为{career.max_stage}"
)
# 检查是否已存在
existing_check = await db.execute(
select(CharacterCareer).where(
CharacterCareer.character_id == character_id,
CharacterCareer.career_id == career_request.career_id
)
)
if existing_check.scalar_one_or_none():
raise HTTPException(status_code=400, detail="该角色已拥有此副职业")
# 检查副职业数量限制(可选,这里设置为最多5个)
sub_count_result = await db.execute(
select(func.count(CharacterCareer.id)).where(
CharacterCareer.character_id == character_id,
CharacterCareer.career_type == "sub"
)
)
sub_count = sub_count_result.scalar_one()
if sub_count >= 5:
raise HTTPException(status_code=400, detail="副职业数量已达上限(最多5个)")
# 创建副职业关联
char_career = CharacterCareer(
character_id=character_id,
career_id=career_request.career_id,
career_type="sub",
current_stage=career_request.current_stage,
stage_progress=0,
started_at=career_request.started_at,
reached_current_stage_at=career_request.started_at
)
db.add(char_career)
await db.commit()
logger.info(f"✅ 添加副职业成功:角色{character.name} -> {career.name}(第{career_request.current_stage}阶段)")
return {"message": "副职业添加成功", "career_name": career.name}
@router.put("/character/{character_id}/careers/{career_id}/stage", summary="更新职业阶段")
async def update_career_stage(
character_id: str,
career_id: str,
stage_request: UpdateCareerStageRequest,
request: Request,
db: AsyncSession = Depends(get_db)
):
"""更新角色在某个职业的阶段"""
# 验证角色职业关联存在
result = await db.execute(
select(CharacterCareer, Career, Character)
.join(Career, CharacterCareer.career_id == Career.id)
.join(Character, CharacterCareer.character_id == Character.id)
.where(
CharacterCareer.character_id == character_id,
CharacterCareer.career_id == career_id
)
)
relation_data = result.one_or_none()
if not relation_data:
raise HTTPException(status_code=404, detail="角色职业关联不存在")
char_career, career, character = relation_data
# 验证用户权限
user_id = getattr(request.state, 'user_id', None)
await verify_project_access(character.project_id, user_id, db)
# 验证新阶段有效性
if stage_request.current_stage > career.max_stage:
raise HTTPException(
status_code=400,
detail=f"阶段超出范围,该职业最大阶段为{career.max_stage}"
)
# 验证阶段递增规则(不能倒退,除非降级)
if stage_request.current_stage < char_career.current_stage:
logger.warning(f"⚠️ 角色{character.name}的职业{career.name}阶段降低:{char_career.current_stage} -> {stage_request.current_stage}")
# 更新阶段信息
char_career.current_stage = stage_request.current_stage
char_career.stage_progress = stage_request.stage_progress
if stage_request.reached_current_stage_at:
char_career.reached_current_stage_at = stage_request.reached_current_stage_at
if stage_request.notes is not None:
char_career.notes = stage_request.notes
await db.commit()
logger.info(f"✅ 更新职业阶段成功:{character.name}{career.name} -> 第{stage_request.current_stage}阶段")
return {
"message": "职业阶段更新成功",
"career_name": career.name,
"new_stage": stage_request.current_stage
}
@router.delete("/character/{character_id}/careers/{career_id}", summary="删除角色副职业")
async def remove_sub_career(
character_id: str,
career_id: str,
request: Request,
db: AsyncSession = Depends(get_db)
):
"""删除角色的副职业"""
# 验证角色职业关联存在
result = await db.execute(
select(CharacterCareer, Character)
.join(Character, CharacterCareer.character_id == Character.id)
.where(
CharacterCareer.character_id == character_id,
CharacterCareer.career_id == career_id
)
)
relation_data = result.one_or_none()
if not relation_data:
raise HTTPException(status_code=404, detail="角色职业关联不存在")
char_career, character = relation_data
# 验证用户权限
user_id = getattr(request.state, 'user_id', None)
await verify_project_access(character.project_id, user_id, db)
# 不允许删除主职业
if char_career.career_type == "main":
raise HTTPException(status_code=400, detail="无法删除主职业,只能更换")
await db.delete(char_career)
await db.commit()
logger.info(f"✅ 删除副职业成功:角色{character.name}移除职业{career_id}")
return {"message": "副职业删除成功"}
+341 -39
View File
@@ -14,6 +14,7 @@ from app.models.chapter import Chapter
from app.models.project import Project
from app.models.outline import Outline
from app.models.character import Character
from app.models.career import Career, CharacterCareer
from app.models.generation_history import GenerationHistory
from app.models.writing_style import WritingStyle
from app.models.analysis_task import AnalysisTask
@@ -665,6 +666,114 @@ async def build_smart_chapter_context(
return context_parts
async def build_characters_info_with_careers(
db: AsyncSession,
project_id: str,
characters: list[Character],
filter_character_names: Optional[list[str]] = None
) -> str:
"""
构建包含职业信息的角色上下文
Args:
db: 数据库会话
project_id: 项目ID
characters: 角色列表
filter_character_names: 可选,筛选特定角色名称列表(用于1-1模式的structure.characters或1-n模式的expansion_plan.character_focus
Returns:
格式化的角色信息字符串,包含职业信息
"""
if not characters:
return '暂无角色信息'
# 如果提供了筛选名单,只保留匹配的角色
if filter_character_names:
filtered_characters = [c for c in characters if c.name in filter_character_names]
if not filtered_characters:
logger.warning(f"筛选后无匹配角色,使用全部角色。筛选名单: {filter_character_names}")
filtered_characters = characters
else:
logger.info(f"根据筛选名单保留 {len(filtered_characters)}/{len(characters)} 个角色: {[c.name for c in filtered_characters]}")
characters = filtered_characters
# 获取所有职业信息(一次性查询,提高效率)
careers_result = await db.execute(
select(Career).where(Career.project_id == project_id)
)
careers_map = {c.id: c for c in careers_result.scalars().all()}
# 获取所有角色的职业关联(一次性查询)
character_ids = [c.id for c in characters]
if not character_ids:
return '暂无角色信息'
character_careers_result = await db.execute(
select(CharacterCareer).where(CharacterCareer.character_id.in_(character_ids))
)
character_careers = character_careers_result.scalars().all()
# 构建角色ID到职业信息的映射
char_career_map = {}
for cc in character_careers:
if cc.character_id not in char_career_map:
char_career_map[cc.character_id] = {'main': None, 'sub': []}
career = careers_map.get(cc.career_id)
if not career:
continue
career_info = {
'name': career.name,
'stage': cc.current_stage,
'max_stage': career.max_stage,
'stage_progress': cc.stage_progress
}
if cc.career_type == 'main':
char_career_map[cc.character_id]['main'] = career_info
else:
char_career_map[cc.character_id]['sub'].append(career_info)
# 构建角色信息字符串
characters_info_parts = []
for c in characters:
# 基本信息
entity_type = '组织' if c.is_organization else '角色'
base_info = f"- {c.name}({entity_type}, {c.role_type})"
# 职业信息
career_info_str = ""
if c.id in char_career_map:
career_data = char_career_map[c.id]
# 主职业
if career_data['main']:
main = career_data['main']
stage_desc = f"{main['stage']}/{main['max_stage']}"
career_info_str += f" | 主职业: {main['name']}({stage_desc})"
# 副职业
if career_data['sub']:
sub_list = []
for sub in career_data['sub']:
stage_desc = f"{sub['stage']}/{sub['max_stage']}"
sub_list.append(f"{sub['name']}({stage_desc})")
career_info_str += f" | 副职业: {', '.join(sub_list)}"
# 性格描述
personality_str = ""
if c.personality:
personality_preview = c.personality[:100] if len(c.personality) > 100 else c.personality
personality_str = f": {personality_preview}"
# 组合完整信息
full_info = base_info + career_info_str + personality_str
characters_info_parts.append(full_info)
return "\n".join(characters_info_parts)
@router.get("/{chapter_id}/can-generate", summary="检查章节是否可以生成")
async def check_can_generate(
chapter_id: str,
@@ -716,7 +825,7 @@ async def analyze_chapter_background(
project_id: str,
task_id: str,
ai_service: AIService
):
) -> bool:
"""
后台异步分析章节(支持并发,使用锁保护数据库写入)
@@ -726,6 +835,9 @@ async def analyze_chapter_background(
project_id: 项目ID
task_id: 任务ID
ai_service: AI服务实例
Returns:
bool: True表示分析成功,False表示分析失败
"""
db_session = None
write_lock = await get_db_write_lock(user_id)
@@ -942,6 +1054,37 @@ async def analyze_chapter_background(
)
logger.info(f"✅ 添加{added_count}条记忆到向量库")
# 💼 更新角色职业(根据分析结果)
if analysis_result.get('character_states'):
try:
from app.services.career_update_service import CareerUpdateService
logger.info(f"💼 开始根据分析结果更新角色职业...")
career_update_result = await CareerUpdateService.update_careers_from_analysis(
db=db_session,
project_id=project_id,
character_states=analysis_result.get('character_states', []),
chapter_id=chapter_id,
chapter_number=chapter.chapter_number
)
if career_update_result['updated_count'] > 0:
logger.info(
f"✅ 更新了 {career_update_result['updated_count']} 个角色的职业信息: "
f"{', '.join(career_update_result['updated_characters'])}"
)
if career_update_result['changes']:
for change in career_update_result['changes']:
logger.info(f" - {change}")
else:
logger.info("️ 本章节无角色职业变化")
except Exception as career_error:
# 职业更新失败不应影响整个分析流程
logger.error(f"⚠️ 更新角色职业失败: {str(career_error)}", exc_info=True)
else:
logger.debug("📋 分析结果中无角色状态信息,跳过职业更新")
# 最终更新任务状态(写操作,需要锁)- 增加重试机制
update_success = False
for retry in range(3):
@@ -965,6 +1108,9 @@ async def analyze_chapter_background(
if not update_success:
logger.warning(f"⚠️ 章节分析完成但状态更新失败: {chapter_id}")
# 返回成功状态
return True
except Exception as e:
logger.error(f"❌ 后台分析异常: {str(e)}", exc_info=True)
# 确保任务状态被更新为failed(写操作,需要锁)
@@ -995,6 +1141,10 @@ async def analyze_chapter_background(
await asyncio.sleep(0.1) # 短暂等待后重试
else:
logger.error(f"❌ 任务状态更新失败,已达到最大重试次数: {task_id}")
# 返回失败状态
return False
finally:
if db_session:
await db_session.close()
@@ -1108,15 +1258,41 @@ async def generate_chapter_content_stream(
for o in all_outlines
])
# 获取角色信息
# 获取角色信息(包含职业信息)
characters_result = await db_session.execute(
select(Character).where(Character.project_id == current_chapter.project_id)
)
characters = characters_result.scalars().all()
characters_info = "\n".join([
f"- {c.name}({'组织' if c.is_organization else '角色'}, {c.role_type}): {c.personality[:100] if c.personality else ''}"
for c in characters
])
# 📝 根据大纲模式智能筛选相关角色
filter_character_names = None
if outline_mode == 'one-to-one':
# 1-1模式:从outline.structure中提取characters字段
if outline and outline.structure:
try:
structure = json.loads(outline.structure)
filter_character_names = structure.get('characters', [])
if filter_character_names:
logger.info(f"📋 1-1模式:从structure提取角色列表 {filter_character_names}")
except json.JSONDecodeError:
logger.warning(f"⚠️ outline.structure解析失败,使用全部角色")
else:
# 1-n模式:从chapter.expansion_plan中提取character_focus字段
if current_chapter.expansion_plan:
try:
plan = json.loads(current_chapter.expansion_plan)
filter_character_names = plan.get('character_focus', [])
if filter_character_names:
logger.info(f"📋 1-n模式:从expansion_plan提取角色焦点 {filter_character_names}")
except json.JSONDecodeError:
logger.warning(f"⚠️ expansion_plan解析失败,使用全部角色")
characters_info = await build_characters_info_with_careers(
db=db_session,
project_id=current_chapter.project_id,
characters=characters,
filter_character_names=filter_character_names
)
# 获取写作风格
style_content = ""
@@ -2325,28 +2501,83 @@ async def execute_batch_generation_in_order(
if task.enable_analysis:
logger.info(f"🔍 开始同步分析章节: 第{chapter.chapter_number}")
async with write_lock:
analysis_task = AnalysisTask(
chapter_id=chapter_id,
user_id=user_id,
project_id=task.project_id,
status='pending',
progress=0
)
db_session.add(analysis_task)
await db_session.commit()
await db_session.refresh(analysis_task)
# 分析重试机制(最多3次)
analysis_retry_count = 0
analysis_success = False
last_analysis_error = None
# 同步执行分析(等待完成)
await analyze_chapter_background(
chapter_id=chapter_id,
user_id=user_id,
project_id=task.project_id,
task_id=analysis_task.id,
ai_service=ai_service
)
logger.info(f"✅ 章节分析完成: 第{chapter.chapter_number}")
while analysis_retry_count < 3 and not analysis_success:
try:
if analysis_retry_count > 0:
logger.info(f"🔄 重试分析章节 (第{analysis_retry_count}次): 第{chapter.chapter_number}")
async with write_lock:
analysis_task = AnalysisTask(
chapter_id=chapter_id,
user_id=user_id,
project_id=task.project_id,
status='pending',
progress=0
)
db_session.add(analysis_task)
await db_session.commit()
await db_session.refresh(analysis_task)
# 同步执行分析,直接使用返回值判断成功/失败
analysis_result = await analyze_chapter_background(
chapter_id=chapter_id,
user_id=user_id,
project_id=task.project_id,
task_id=analysis_task.id,
ai_service=ai_service
)
# 直接根据返回值判断
if not analysis_result:
last_analysis_error = "分析函数返回失败"
logger.error(f"❌ 章节分析失败: 第{chapter.chapter_number}")
raise Exception(f"章节分析失败")
# 分析成功
analysis_success = True
logger.info(f"✅ 章节分析成功: 第{chapter.chapter_number}")
except Exception as analysis_error:
last_analysis_error = str(analysis_error)
analysis_retry_count += 1
if analysis_retry_count < 3:
# 还有重试机会,等待后重试
wait_time = min(2 ** analysis_retry_count, 10)
logger.warning(f"⏳ 分析失败,等待 {wait_time} 秒后重试...")
await asyncio.sleep(wait_time)
else:
# 达到最大重试次数,必须终止整个批量任务
logger.error(f"❌ 章节分析失败,已达最大重试次数(3次): 第{chapter.chapter_number}")
# 记录失败信息
failed_info = {
'chapter_id': chapter_id,
'chapter_number': chapter.chapter_number,
'title': chapter.title,
'error': f"分析失败(重试3次): {last_analysis_error}",
'retry_count': 3
}
async with write_lock:
if task.failed_chapters is None:
task.failed_chapters = []
task.failed_chapters.append(failed_info)
# 标记任务失败并终止
task.status = 'failed'
task.error_message = f"{chapter.chapter_number}章分析失败(重试3次): {last_analysis_error}"[:500]
task.completed_at = datetime.now()
task.current_retry_count = 0
await db_session.commit()
logger.error(f"🛑 批量生成中断: 第{chapter.chapter_number}章分析失败")
return # 立即终止整个批量生成任务
# 标记成功
chapter_success = True
@@ -2361,7 +2592,8 @@ async def execute_batch_generation_in_order(
except Exception as e:
last_error = str(e)
logger.error(f"❌ 章节生成失败: {chapter.chapter_number if chapter else '?'}, 错误: {last_error}")
error_msg = f"{chapter.chapter_number if chapter else '?'}出错: {last_error}"
logger.error(f"{error_msg}")
retry_count += 1
@@ -2394,7 +2626,13 @@ async def execute_batch_generation_in_order(
task.current_retry_count = 0
await db_session.commit()
logger.error(f"🛑 批量生成终止于第{chapter.chapter_number}")
# ⚠️ 如果启用了同步分析,任何错误都应该中断任务
# 因为章节生成或分析失败会影响后续章节的职业更新和剧情连贯性
if task.enable_analysis:
logger.error(f"🛑 批量生成中断: 因启用同步分析,任何错误都会中断任务以确保职业信息和剧情连贯性")
else:
logger.error(f"🛑 批量生成终止于第{chapter.chapter_number}")
return
# 全部完成
@@ -2469,15 +2707,41 @@ async def generate_single_chapter_for_batch(
for o in all_outlines
])
# 获取角色信息
# 获取角色信息(包含职业信息)
characters_result = await db_session.execute(
select(Character).where(Character.project_id == chapter.project_id)
)
characters = characters_result.scalars().all()
characters_info = "\n".join([
f"- {c.name}({'组织' if c.is_organization else '角色'}, {c.role_type}): {c.personality[:100] if c.personality else ''}"
for c in characters
])
# 📝 根据大纲模式智能筛选相关角色(批量生成)
filter_character_names = None
if outline_mode == 'one-to-one':
# 1-1模式:从outline.structure中提取characters字段
if outline and outline.structure:
try:
structure = json.loads(outline.structure)
filter_character_names = structure.get('characters', [])
if filter_character_names:
logger.info(f"📋 批量生成 - 1-1模式:从structure提取角色列表 {filter_character_names}")
except json.JSONDecodeError:
logger.warning(f"⚠️ 批量生成 - outline.structure解析失败,使用全部角色")
else:
# 1-n模式:从chapter.expansion_plan中提取character_focus字段
if chapter.expansion_plan:
try:
plan = json.loads(chapter.expansion_plan)
filter_character_names = plan.get('character_focus', [])
if filter_character_names:
logger.info(f"📋 批量生成 - 1-n模式:从expansion_plan提取角色焦点 {filter_character_names}")
except json.JSONDecodeError:
logger.warning(f"⚠️ 批量生成 - expansion_plan解析失败,使用全部角色")
characters_info = await build_characters_info_with_careers(
db=db_session,
project_id=chapter.project_id,
characters=characters,
filter_character_names=filter_character_names
)
# 获取写作风格
style_content = ""
@@ -2721,12 +2985,53 @@ async def regenerate_chapter_stream(
)
project = project_result.scalar_one_or_none()
# 获取角色信息
# 获取角色信息(包含职业信息)
characters_result = await temp_db.execute(
select(Character).where(Character.project_id == chapter.project_id)
)
characters = characters_result.scalars().all()
# 📝 根据大纲模式智能筛选相关角色(重新生成)
outline_mode_result = await temp_db.execute(
select(Project.outline_mode).where(Project.id == chapter.project_id)
)
outline_mode = outline_mode_result.scalar_one_or_none() or 'one-to-many'
filter_character_names = None
if outline_mode == 'one-to-one':
# 1-1模式:从outline.structure中提取characters字段
outline_result_temp = await temp_db.execute(
select(Outline.structure)
.where(Outline.project_id == chapter.project_id)
.where(Outline.order_index == chapter.chapter_number)
)
outline_structure = outline_result_temp.scalar_one_or_none()
if outline_structure:
try:
structure = json.loads(outline_structure)
filter_character_names = structure.get('characters', [])
if filter_character_names:
logger.info(f"📋 重新生成 - 1-1模式:从structure提取角色列表 {filter_character_names}")
except json.JSONDecodeError:
logger.warning(f"⚠️ 重新生成 - outline.structure解析失败,使用全部角色")
else:
# 1-n模式:从chapter.expansion_plan中提取character_focus字段
if chapter.expansion_plan:
try:
plan = json.loads(chapter.expansion_plan)
filter_character_names = plan.get('character_focus', [])
if filter_character_names:
logger.info(f"📋 重新生成 - 1-n模式:从expansion_plan提取角色焦点 {filter_character_names}")
except json.JSONDecodeError:
logger.warning(f"⚠️ 重新生成 - expansion_plan解析失败,使用全部角色")
characters_info_with_careers = await build_characters_info_with_careers(
db=temp_db,
project_id=chapter.project_id,
characters=characters,
filter_character_names=filter_character_names
)
# 获取章节大纲
outline_result = await temp_db.execute(
select(Outline)
@@ -2779,10 +3084,7 @@ async def regenerate_chapter_stream(
'time_period': project.world_time_period if project else '未设定',
'location': project.world_location if project else '未设定',
'atmosphere': project.world_atmosphere if project else '未设定',
'characters_info': "\n".join([
f"- {c.name}({'组织' if c.is_organization else '角色'}, {c.role_type}): {c.personality[:100] if c.personality else ''}"
for c in characters
]) if characters else '暂无角色信息',
'characters_info': characters_info_with_careers,
'chapter_outline': outline.content if outline else chapter.summary or '暂无大纲',
'previous_context': '' # 可以后续扩展添加前置章节上下文
}
+440 -7
View File
@@ -85,7 +85,7 @@ async def get_characters(
)
characters = result.scalars().all()
# 为组织类型的角色填充Organization表的额外字段
# 为组织类型的角色填充Organization表的额外字段,并添加职业信息
enriched_characters = []
for char in characters:
char_dict = {
@@ -110,7 +110,10 @@ async def get_characters(
"power_level": None,
"location": None,
"motto": None,
"color": None
"color": None,
"main_career_id": char.main_career_id,
"main_career_stage": char.main_career_stage,
"sub_careers": json.loads(char.sub_careers) if char.sub_careers else None
}
if char.is_organization:
@@ -156,7 +159,7 @@ async def get_project_characters(
)
characters = result.scalars().all()
# 为组织类型的角色填充Organization表的额外字段
# 为组织类型的角色填充Organization表的额外字段,并添加职业信息
enriched_characters = []
for char in characters:
char_dict = {
@@ -181,7 +184,10 @@ async def get_project_characters(
"power_level": None,
"location": None,
"motto": None,
"color": None
"color": None,
"main_career_id": char.main_career_id,
"main_career_stage": char.main_career_stage,
"sub_careers": json.loads(char.sub_careers) if char.sub_careers else None
}
if char.is_organization:
@@ -232,6 +238,8 @@ async def update_character(
db: AsyncSession = Depends(get_db)
):
"""更新角色信息"""
from app.models.career import CharacterCareer, Career
result = await db.execute(
select(Character).where(Character.id == character_id)
)
@@ -260,6 +268,139 @@ async def update_character(
if 'color' in update_data:
org_fields['color'] = update_data.pop('color')
# 处理主职业和副职业更新
main_career_id = update_data.pop('main_career_id', None)
main_career_stage = update_data.pop('main_career_stage', None)
sub_careers_json = update_data.pop('sub_careers', None)
if main_career_id is not None:
# 验证职业存在
if main_career_id: # 不为空
career_result = await db.execute(
select(Career).where(
Career.id == main_career_id,
Career.project_id == character.project_id,
Career.type == 'main'
)
)
career = career_result.scalar_one_or_none()
if not career:
raise HTTPException(status_code=400, detail="主职业不存在或类型错误")
# 验证阶段有效性
if main_career_stage and main_career_stage > career.max_stage:
raise HTTPException(status_code=400, detail=f"阶段超出范围,该职业最大阶段为{career.max_stage}")
# 更新或创建CharacterCareer关联
char_career_result = await db.execute(
select(CharacterCareer).where(
CharacterCareer.character_id == character_id,
CharacterCareer.career_type == 'main'
)
)
char_career = char_career_result.scalar_one_or_none()
if char_career:
# 更新现有关联
char_career.career_id = main_career_id
if main_career_stage:
char_career.current_stage = main_career_stage
logger.info(f"更新主职业关联:{character.name} -> {career.name}")
else:
# 创建新关联
char_career = CharacterCareer(
character_id=character_id,
career_id=main_career_id,
career_type='main',
current_stage=main_career_stage or 1,
stage_progress=0
)
db.add(char_career)
logger.info(f"创建主职业关联:{character.name} -> {career.name}")
# 更新Character表的冗余字段
character.main_career_id = main_career_id
character.main_career_stage = main_career_stage or char_career.current_stage
else:
# 清空主职业
char_career_result = await db.execute(
select(CharacterCareer).where(
CharacterCareer.character_id == character_id,
CharacterCareer.career_type == 'main'
)
)
char_career = char_career_result.scalar_one_or_none()
if char_career:
await db.delete(char_career)
logger.info(f"移除主职业关联:{character.name}")
character.main_career_id = None
character.main_career_stage = None
elif main_career_stage is not None and character.main_career_id:
# 只更新阶段
char_career_result = await db.execute(
select(CharacterCareer).where(
CharacterCareer.character_id == character_id,
CharacterCareer.career_type == 'main'
)
)
char_career = char_career_result.scalar_one_or_none()
if char_career:
char_career.current_stage = main_career_stage
character.main_career_stage = main_career_stage
logger.info(f"更新主职业阶段:{character.name} -> 阶段{main_career_stage}")
# 处理副职业更新
if sub_careers_json is not None:
# 解析副职业JSON
try:
sub_careers_data = json.loads(sub_careers_json) if isinstance(sub_careers_json, str) else sub_careers_json
except:
sub_careers_data = []
# 删除现有的所有副职业关联
existing_subs = await db.execute(
select(CharacterCareer).where(
CharacterCareer.character_id == character_id,
CharacterCareer.career_type == 'sub'
)
)
for sub_career in existing_subs.scalars():
await db.delete(sub_career)
# 创建新的副职业关联
for sub_data in sub_careers_data[:2]: # 最多2个副职业
career_id = sub_data.get('career_id')
if not career_id:
continue
# 验证副职业存在
career_result = await db.execute(
select(Career).where(
Career.id == career_id,
Career.project_id == character.project_id,
Career.type == 'sub'
)
)
career = career_result.scalar_one_or_none()
if career:
# 创建副职业关联
char_career = CharacterCareer(
character_id=character_id,
career_id=career_id,
career_type='sub',
current_stage=sub_data.get('stage', 1),
stage_progress=0
)
db.add(char_career)
logger.info(f"添加副职业关联:{character.name} -> {career.name}")
# 更新Character表的sub_careers冗余字段
character.sub_careers = sub_careers_json if isinstance(sub_careers_json, str) else json.dumps(sub_careers_data, ensure_ascii=False)
logger.info(f"更新副职业信息:{character.name}")
# 更新 Character 表字段
for field, value in update_data.items():
setattr(character, field, value)
@@ -290,7 +431,51 @@ async def update_character(
await db.refresh(character)
logger.info(f"更新角色/组织成功:{character.name} (ID: {character_id})")
return character
# 构建响应,确保sub_careers是list类型
response_data = {
"id": character.id,
"project_id": character.project_id,
"name": character.name,
"age": character.age,
"gender": character.gender,
"is_organization": character.is_organization,
"role_type": character.role_type,
"personality": character.personality,
"background": character.background,
"appearance": character.appearance,
"relationships": character.relationships,
"organization_type": character.organization_type,
"organization_purpose": character.organization_purpose,
"organization_members": character.organization_members,
"traits": character.traits,
"avatar_url": character.avatar_url,
"created_at": character.created_at,
"updated_at": character.updated_at,
"main_career_id": character.main_career_id,
"main_career_stage": character.main_career_stage,
"sub_careers": json.loads(character.sub_careers) if character.sub_careers else None,
"power_level": None,
"location": None,
"motto": None,
"color": None
}
# 如果是组织,添加组织额外字段
if character.is_organization:
org_result = await db.execute(
select(Organization).where(Organization.character_id == character_id)
)
org = org_result.scalar_one_or_none()
if org:
response_data.update({
"power_level": org.power_level,
"location": org.location,
"motto": org.motto,
"color": org.color
})
return response_data
@router.delete("/{character_id}", summary="删除角色")
@@ -330,7 +515,10 @@ async def create_character(
- 可以创建普通角色(is_organization=False
- 也可以创建组织(is_organization=True
- 如果创建组织且提供了组织额外字段,会自动创建Organization详情记录
- 支持设置主职业和副职业
"""
from app.models.career import CharacterCareer, Career
# 验证用户权限
user_id = getattr(request.state, 'user_id', None)
await verify_project_access(character_data.project_id, user_id, db)
@@ -352,13 +540,78 @@ async def create_character(
organization_purpose=character_data.organization_purpose,
organization_members=character_data.organization_members,
traits=character_data.traits,
avatar_url=character_data.avatar_url
avatar_url=character_data.avatar_url,
main_career_id=character_data.main_career_id,
main_career_stage=character_data.main_career_stage,
sub_careers=character_data.sub_careers
)
db.add(character)
await db.flush() # 获取character.id
logger.info(f"✅ 手动创建角色成功:{character.name} (ID: {character.id}, 是否组织: {character.is_organization})")
# 处理主职业关联
if character_data.main_career_id and not character.is_organization:
# 验证职业存在
career_result = await db.execute(
select(Career).where(
Career.id == character_data.main_career_id,
Career.project_id == character_data.project_id,
Career.type == 'main'
)
)
career = career_result.scalar_one_or_none()
if career:
# 创建主职业关联
char_career = CharacterCareer(
character_id=character.id,
career_id=character_data.main_career_id,
career_type='main',
current_stage=character_data.main_career_stage or 1,
stage_progress=0
)
db.add(char_career)
logger.info(f"✅ 创建主职业关联:{character.name} -> {career.name}")
else:
logger.warning(f"⚠️ 主职业ID不存在或类型错误: {character_data.main_career_id}")
# 处理副职业关联
if character_data.sub_careers and not character.is_organization:
try:
sub_careers_data = json.loads(character_data.sub_careers) if isinstance(character_data.sub_careers, str) else character_data.sub_careers
for sub_data in sub_careers_data[:2]: # 最多2个副职业
career_id = sub_data.get('career_id')
if not career_id:
continue
# 验证副职业存在
career_result = await db.execute(
select(Career).where(
Career.id == career_id,
Career.project_id == character_data.project_id,
Career.type == 'sub'
)
)
career = career_result.scalar_one_or_none()
if career:
# 创建副职业关联
char_career = CharacterCareer(
character_id=character.id,
career_id=career_id,
career_type='sub',
current_stage=sub_data.get('stage', 1),
stage_progress=0
)
db.add(char_career)
logger.info(f"✅ 创建副职业关联:{character.name} -> {career.name}")
else:
logger.warning(f"⚠️ 副职业ID不存在或类型错误: {career_id}")
except Exception as e:
logger.warning(f"⚠️ 解析副职业数据失败: {e}")
# 如果是组织,且提供了组织额外字段,自动创建Organization详情记录
if character.is_organization and (
character_data.power_level is not None or
@@ -438,6 +691,50 @@ async def generate_character_stream(
if organization_list:
existing_chars_info += "\n\n已有组织:\n" + "\n".join(organization_list)
# 🎯 获取项目职业列表
from app.models.career import Career
careers_result = await db.execute(
select(Career)
.where(Career.project_id == request.project_id)
.order_by(Career.type, Career.name)
)
careers = careers_result.scalars().all()
# 构建职业信息摘要
careers_info = ""
if careers:
main_careers = [c for c in careers if c.type == 'main']
sub_careers = [c for c in careers if c.type == 'sub']
if main_careers:
careers_info += "\n\n可用主职业列表(请在career_info中填写职业名称,系统会自动匹配ID):\n"
for career in main_careers:
# 解析阶段信息
import json as json_lib
try:
stages = json_lib.loads(career.stages) if career.stages else []
stage_names = [s.get('name', f'阶段{s.get("level")}') for s in stages[:3]] # 只显示前3个阶段
stage_info = "".join(stage_names)
if len(stages) > 3:
stage_info += " → ..."
except:
stage_info = f"{career.max_stage}个阶段"
careers_info += f"- 名称: {career.name}"
if career.description:
careers_info += f", 描述: {career.description[:50]}"
careers_info += f", 阶段: {stage_info}\n"
if sub_careers:
careers_info += "\n可用副职业列表(请在career_info中填写职业名称,系统会自动匹配ID):\n"
for career in sub_careers[:5]: # 最多显示5个副职业
careers_info += f"- 名称: {career.name}"
if career.description:
careers_info += f", 描述: {career.description[:50]}"
careers_info += "\n"
else:
careers_info = "\n\n⚠️ 项目中暂无职业设定"
# 构建项目上下文
project_context = f"""
项目信息:
@@ -449,6 +746,7 @@ async def generate_character_stream(
- 氛围基调:{project.world_atmosphere or '未设定'}
- 世界规则:{project.world_rules or '未设定'}
{existing_chars_info}
{careers_info}
"""
user_input = f"""
@@ -544,6 +842,62 @@ async def generate_character_stream(
traits_json = json.dumps(character_data.get("traits", []), ensure_ascii=False) if character_data.get("traits") else None
is_organization = character_data.get("is_organization", False)
# 提取职业信息(支持通过名称匹配)
career_info = character_data.get("career_info", {})
raw_main_career_name = career_info.get("main_career_name") if career_info else None
main_career_stage = career_info.get("main_career_stage", 1) if career_info else None
raw_sub_careers_data = career_info.get("sub_careers", []) if career_info else []
# 调试日志:输出职业信息
logger.info(f"🔍 提取职业信息 - career_info: {career_info}")
logger.info(f"🔍 raw_main_career_name: {raw_main_career_name}, main_career_stage: {main_career_stage}")
logger.info(f"🔍 raw_sub_careers_data类型: {type(raw_sub_careers_data)}, 内容: {raw_sub_careers_data}")
# 🔧 通过职业名称匹配数据库中的职业ID
from app.models.career import Career
main_career_id = None
sub_careers_data = []
# 匹配主职业名称
if raw_main_career_name and not is_organization:
career_check = await db.execute(
select(Career).where(
Career.name == raw_main_career_name,
Career.project_id == request.project_id,
Career.type == 'main'
)
)
matched_career = career_check.scalar_one_or_none()
if matched_career:
main_career_id = matched_career.id
logger.info(f"✅ 主职业名称匹配成功: {raw_main_career_name} -> ID: {main_career_id}")
else:
logger.warning(f"⚠️ AI返回的主职业名称未找到: {raw_main_career_name}")
# 匹配副职业名称
if raw_sub_careers_data and not is_organization and isinstance(raw_sub_careers_data, list):
for sub_data in raw_sub_careers_data[:2]:
if isinstance(sub_data, dict):
career_name = sub_data.get('career_name')
if career_name:
career_check = await db.execute(
select(Career).where(
Career.name == career_name,
Career.project_id == request.project_id,
Career.type == 'sub'
)
)
matched_career = career_check.scalar_one_or_none()
if matched_career:
# 转换为包含ID的格式
sub_careers_data.append({
'career_id': matched_career.id,
'stage': sub_data.get('stage', 1)
})
logger.info(f"✅ 副职业名称匹配成功: {career_name} -> ID: {matched_career.id}")
else:
logger.warning(f"⚠️ AI返回的副职业名称未找到: {career_name}")
# 创建角色
character = Character(
project_id=request.project_id,
@@ -559,13 +913,92 @@ async def generate_character_stream(
organization_type=character_data.get("organization_type") if is_organization else None,
organization_purpose=character_data.get("organization_purpose") if is_organization else None,
organization_members=json.dumps(character_data.get("organization_members", []), ensure_ascii=False) if is_organization else None,
traits=traits_json
traits=traits_json,
main_career_id=main_career_id,
main_career_stage=main_career_stage if main_career_id else None,
sub_careers=json.dumps(sub_careers_data, ensure_ascii=False) if sub_careers_data else None
)
db.add(character)
await db.flush()
logger.info(f"✅ 角色创建成功:{character.name} (ID: {character.id})")
# 处理主职业关联
if main_career_id and not is_organization:
from app.models.career import CharacterCareer, Career
career_result = await db.execute(
select(Career).where(
Career.id == main_career_id,
Career.project_id == request.project_id,
Career.type == 'main'
)
)
career = career_result.scalar_one_or_none()
if career:
char_career = CharacterCareer(
character_id=character.id,
career_id=main_career_id,
career_type='main',
current_stage=main_career_stage,
stage_progress=0
)
db.add(char_career)
logger.info(f"✅ AI生成角色-创建主职业关联:{character.name} -> {career.name}")
else:
logger.warning(f"⚠️ AI返回的主职业ID不存在: {main_career_id}")
# 处理副职业关联
if sub_careers_data and not is_organization:
from app.models.career import CharacterCareer, Career
logger.info(f"🔍 开始处理副职业关联,数据: {sub_careers_data}")
# 确保sub_careers_data是列表
if not isinstance(sub_careers_data, list):
logger.warning(f"⚠️ sub_careers_data不是列表类型: {type(sub_careers_data)}")
sub_careers_data = []
for idx, sub_data in enumerate(sub_careers_data[:2]): # 最多2个副职业
logger.info(f"🔍 处理第{idx+1}个副职业,数据: {sub_data}, 类型: {type(sub_data)}")
# 兼容不同的数据格式
if isinstance(sub_data, dict):
career_id = sub_data.get('career_id')
stage = sub_data.get('stage', 1)
else:
logger.warning(f"⚠️ 副职业数据格式错误,应为dict: {sub_data}")
continue
if not career_id:
logger.warning(f"⚠️ 副职业数据缺少career_id字段")
continue
logger.info(f"🔍 查询副职业: career_id={career_id}, project_id={request.project_id}")
career_result = await db.execute(
select(Career).where(
Career.id == career_id,
Career.project_id == request.project_id,
Career.type == 'sub'
)
)
career = career_result.scalar_one_or_none()
if career:
char_career = CharacterCareer(
character_id=character.id,
career_id=career_id,
career_type='sub',
current_stage=stage,
stage_progress=0
)
db.add(char_career)
logger.info(f"✅ AI生成角色-创建副职业关联:{character.name} -> {career.name} (阶段{stage})")
else:
logger.warning(f"⚠️ AI返回的副职业ID不存在: {career_id} (项目ID: {request.project_id})")
# 如果是组织,创建Organization详情
if is_organization:
yield await SSEResponse.send_progress("创建组织详情...", 85)
+245 -7
View File
@@ -11,6 +11,7 @@ from app.models.project import Project
from app.models.character import Character
from app.models.outline import Outline
from app.models.chapter import Chapter
from app.models.career import Career, CharacterCareer
from app.models.relationship import CharacterRelationship, Organization, OrganizationMember, RelationshipType
from app.models.writing_style import WritingStyle
from app.models.project_default_style import ProjectDefaultStyle
@@ -239,6 +240,121 @@ async def world_building_generator(
project.wizard_step = 1
await db.commit()
# ===== 自动生成职业体系 =====
yield await SSEResponse.send_progress("🎯 开始生成职业体系框架...", 75)
logger.info(f"🎯 世界观已完成,开始为项目 {project.id} 自动生成职业体系")
try:
# 获取职业生成提示词模板(支持用户自定义)
template = await PromptService.get_template("CAREER_SYSTEM_GENERATION", user_id, db)
career_prompt = PromptService.format_prompt(
template,
title=project.title,
genre=genre or '未设定',
theme=theme or '未设定',
time_period=world_data.get('time_period', '未设定'),
location=world_data.get('location', '未设定'),
atmosphere=world_data.get('atmosphere', '未设定'),
rules=world_data.get('rules', '未设定')
)
yield await SSEResponse.send_progress("正在生成职业体系...", 78)
# 调用AI生成职业
result = await user_ai_service.generate_text(prompt=career_prompt)
career_response = result.get('content', '') if isinstance(result, dict) else result
if not career_response or not career_response.strip():
logger.warning("⚠️ AI返回空职业体系,跳过职业生成")
yield await SSEResponse.send_progress("职业体系生成跳过(AI返回为空)", 85)
else:
yield await SSEResponse.send_progress("解析职业体系数据...", 82)
# 清洗并解析JSON
try:
cleaned_response = user_ai_service._clean_json_response(career_response)
career_data = json.loads(cleaned_response)
logger.info(f"✅ 职业体系JSON解析成功")
# 保存主职业
main_careers_created = []
for idx, career_info in enumerate(career_data.get("main_careers", [])):
try:
stages_json = json.dumps(career_info.get("stages", []), ensure_ascii=False)
attribute_bonuses = career_info.get("attribute_bonuses")
attribute_bonuses_json = json.dumps(attribute_bonuses, ensure_ascii=False) if attribute_bonuses else None
career = Career(
project_id=project.id,
name=career_info.get("name", f"未命名主职业{idx+1}"),
type="main",
description=career_info.get("description"),
category=career_info.get("category"),
stages=stages_json,
max_stage=career_info.get("max_stage", 10),
requirements=career_info.get("requirements"),
special_abilities=career_info.get("special_abilities"),
worldview_rules=career_info.get("worldview_rules"),
attribute_bonuses=attribute_bonuses_json,
source="ai"
)
db.add(career)
await db.flush()
main_careers_created.append(career.name)
logger.info(f" ✅ 创建主职业:{career.name}")
except Exception as e:
logger.error(f" ❌ 创建主职业失败:{str(e)}")
continue
# 保存副职业
sub_careers_created = []
for idx, career_info in enumerate(career_data.get("sub_careers", [])):
try:
stages_json = json.dumps(career_info.get("stages", []), ensure_ascii=False)
attribute_bonuses = career_info.get("attribute_bonuses")
attribute_bonuses_json = json.dumps(attribute_bonuses, ensure_ascii=False) if attribute_bonuses else None
career = Career(
project_id=project.id,
name=career_info.get("name", f"未命名副职业{idx+1}"),
type="sub",
description=career_info.get("description"),
category=career_info.get("category"),
stages=stages_json,
max_stage=career_info.get("max_stage", 5),
requirements=career_info.get("requirements"),
special_abilities=career_info.get("special_abilities"),
worldview_rules=career_info.get("worldview_rules"),
attribute_bonuses=attribute_bonuses_json,
source="ai"
)
db.add(career)
await db.flush()
sub_careers_created.append(career.name)
logger.info(f" ✅ 创建副职业:{career.name}")
except Exception as e:
logger.error(f" ❌ 创建副职业失败:{str(e)}")
continue
await db.commit()
logger.info(f"🎉 职业体系生成完成:主职业{len(main_careers_created)}个,副职业{len(sub_careers_created)}")
yield await SSEResponse.send_progress(
f"✅ 职业体系生成完成(主{len(main_careers_created)}+副{len(sub_careers_created)}",
90
)
except json.JSONDecodeError as e:
logger.error(f"❌ 职业体系JSON解析失败: {e}")
yield await SSEResponse.send_progress("⚠️ 职业体系解析失败,已跳过", 85)
except Exception as e:
logger.error(f"❌ 职业体系保存失败: {e}")
yield await SSEResponse.send_progress("⚠️ 职业体系保存失败,已跳过", 85)
except Exception as e:
logger.error(f"❌ 职业体系生成异常: {e}")
yield await SSEResponse.send_progress("⚠️ 职业体系生成失败,已跳过(不影响项目创建)", 85)
db_committed = True
# 发送最终结果
@@ -381,6 +497,40 @@ async def characters_generator(
logger.warning(f"MCP工具调用失败(降级处理): {e}")
yield await SSEResponse.send_progress("⚠️ MCP工具暂时不可用,使用基础模式", 12)
# 获取项目的职业列表,用于角色职业分配
yield await SSEResponse.send_progress("加载职业体系...", 13)
career_result = await db.execute(
select(Career).where(Career.project_id == project_id).order_by(Career.type, Career.id)
)
careers = career_result.scalars().all()
main_careers = [c for c in careers if c.type == "main"]
sub_careers = [c for c in careers if c.type == "sub"]
# 构建职业上下文
careers_context = ""
if main_careers or sub_careers:
careers_context = "\n\n【职业体系】\n"
if main_careers:
careers_context += "主职业:\n"
for career in main_careers:
careers_context += f"- {career.name}: {career.description or '暂无描述'}\n"
if sub_careers:
careers_context += "\n副职业:\n"
for career in sub_careers:
careers_context += f"- {career.name}: {career.description or '暂无描述'}\n"
careers_context += "\n请为每个角色分配职业:\n"
careers_context += "- 每个角色必须有1个主职业(从上述主职业中选择)\n"
careers_context += "- 每个角色可以有0-2个副职业(从上述副职业中选择,可选)\n"
careers_context += "- 主职业初始阶段建议为1-3\n"
careers_context += "- 副职业初始阶段建议为1-2\n"
careers_context += "- 请在返回的JSON中包含 career_assignment 字段:\n"
careers_context += ' {"main_career": "职业名称", "main_stage": 2, "sub_careers": [{"career": "副职业名称", "stage": 1}]}\n'
logger.info(f"✅ 加载了{len(main_careers)}个主职业和{len(sub_careers)}个副职业")
else:
logger.warning("⚠️ 项目没有职业体系,跳过职业分配")
# 优化的分批策略:每批生成3个,平衡效率和成功率
BATCH_SIZE = 3 # 每批生成3个角色
MAX_RETRIES = 3 # 每批最多重试3次
@@ -445,7 +595,7 @@ async def characters_generator(
rules=world_context.get("rules", ""),
theme=theme or project.theme or "",
genre=genre or project.genre or "",
requirements=batch_requirements
requirements=batch_requirements + careers_context # 添加职业上下文
)
# 如果有MCP参考资料,增强提示词
@@ -626,14 +776,102 @@ async def characters_generator(
await db.flush() # 获取所有角色的ID
# 第二阶段:为角色分配职业并创建CharacterCareer关联
if main_careers or sub_careers:
yield await SSEResponse.send_progress("分配角色职业...", 86)
careers_assigned = 0
# 构建职业名称到对象的映射
career_name_to_obj = {c.name: c for c in careers}
for character, char_data in created_characters:
# 跳过组织
if character.is_organization:
continue
try:
career_assignment = char_data.get("career_assignment", {})
# 分配主职业
main_career_name = career_assignment.get("main_career")
main_career_stage = career_assignment.get("main_stage", 1)
if main_career_name and main_career_name in career_name_to_obj:
main_career = career_name_to_obj[main_career_name]
# 创建CharacterCareer关联
char_career = CharacterCareer(
character_id=character.id,
career_id=main_career.id,
career_type="main",
current_stage=min(main_career_stage, main_career.max_stage),
stage_progress=0
)
db.add(char_career)
# 更新Character冗余字段
character.main_career_id = main_career.id
character.main_career_stage = char_career.current_stage
careers_assigned += 1
logger.info(f" ✅ 分配主职业:{character.name} -> {main_career.name} (阶段{char_career.current_stage})")
else:
if main_career_name:
logger.warning(f" ⚠️ 主职业不存在:{character.name} -> {main_career_name}")
# 分配副职业
sub_career_assignments = career_assignment.get("sub_careers", [])
sub_career_list = []
for sub_assign in sub_career_assignments[:2]: # 最多2个副职业
sub_career_name = sub_assign.get("career")
sub_career_stage = sub_assign.get("stage", 1)
if sub_career_name and sub_career_name in career_name_to_obj:
sub_career = career_name_to_obj[sub_career_name]
# 创建CharacterCareer关联
char_career = CharacterCareer(
character_id=character.id,
career_id=sub_career.id,
career_type="sub",
current_stage=min(sub_career_stage, sub_career.max_stage),
stage_progress=0
)
db.add(char_career)
# 添加到副职业列表
sub_career_list.append({
"career_id": sub_career.id,
"stage": char_career.current_stage
})
careers_assigned += 1
logger.info(f" ✅ 分配副职业:{character.name} -> {sub_career.name} (阶段{char_career.current_stage})")
else:
if sub_career_name:
logger.warning(f" ⚠️ 副职业不存在:{character.name} -> {sub_career_name}")
# 更新Character冗余字段
if sub_career_list:
character.sub_careers = json.dumps(sub_career_list, ensure_ascii=False)
except Exception as e:
logger.warning(f" ❌ 分配职业失败:{character.name} - {str(e)}")
continue
await db.flush()
logger.info(f"💼 职业分配完成:共分配{careers_assigned}个职业")
yield await SSEResponse.send_progress(f"已分配{careers_assigned}个职业", 87)
# 刷新并建立名称映射
for character, _ in created_characters:
await db.refresh(character)
character_name_to_obj[character.name] = character
logger.info(f"向导创建角色:{character.name} (ID: {character.id}, 是否组织: {character.is_organization})")
# 为is_organization=True的角色创建Organization记录
yield await SSEResponse.send_progress("创建组织记录...", 87)
# 第三阶段:为is_organization=True的角色创建Organization记录
yield await SSEResponse.send_progress("创建组织记录...", 88)
organization_name_to_obj = {} # 组织名称到Organization对象的映射
for character, char_data in created_characters:
@@ -669,8 +907,8 @@ async def characters_generator(
for character, _ in created_characters:
await db.refresh(character)
# 第阶段:创建角色间的关系
yield await SSEResponse.send_progress("创建角色关系...", 90)
# 第阶段:创建角色间的关系
yield await SSEResponse.send_progress("创建角色关系...", 91)
relationships_created = 0
for character, char_data in created_characters:
@@ -737,8 +975,8 @@ async def characters_generator(
logger.warning(f" ❌ 向导创建关系失败:{character.name} - {str(e)}")
continue
# 第阶段:创建组织成员关系
yield await SSEResponse.send_progress("创建组织成员关系...", 93)
# 第阶段:创建组织成员关系
yield await SSEResponse.send_progress("创建组织成员关系...", 94)
members_created = 0
for character, char_data in created_characters:
+2 -1
View File
@@ -143,7 +143,7 @@ from app.api import (
wizard_stream, relationships, organizations,
auth, users, settings, writing_styles, memories,
mcp_plugins, admin, inspiration, prompt_templates,
changelog
changelog, careers
)
app.include_router(auth.router, prefix="/api")
@@ -156,6 +156,7 @@ app.include_router(wizard_stream.router, prefix="/api")
app.include_router(inspiration.router, prefix="/api")
app.include_router(outlines.router, prefix="/api")
app.include_router(characters.router, prefix="/api")
app.include_router(careers.router, prefix="/api") # 职业管理API
app.include_router(chapters.router, prefix="/api")
app.include_router(relationships.router, prefix="/api")
app.include_router(organizations.router, prefix="/api")
+77
View File
@@ -0,0 +1,77 @@
"""职业数据模型"""
from sqlalchemy import Column, String, Text, DateTime, Integer, ForeignKey, Index
from sqlalchemy.sql import func
from app.database import Base
import uuid
class Career(Base):
"""职业表"""
__tablename__ = "careers"
id = Column(String(36), primary_key=True, default=lambda: str(uuid.uuid4()))
project_id = Column(String(36), ForeignKey("projects.id", ondelete="CASCADE"), nullable=False)
# 基本信息
name = Column(String(100), nullable=False, comment="职业名称")
type = Column(String(20), nullable=False, comment="职业类型: main(主职业)/sub(副职业)")
description = Column(Text, comment="职业描述")
category = Column(String(50), comment="职业分类(如:战斗系、生产系、辅助系)")
# 阶段设定
stages = Column(Text, nullable=False, comment="职业阶段列表(JSON): [{level:1, name:'', description:''}, ...]")
max_stage = Column(Integer, nullable=False, default=10, comment="最大阶段数")
# 职业特性
requirements = Column(Text, comment="职业要求/限制")
special_abilities = Column(Text, comment="特殊能力描述")
worldview_rules = Column(Text, comment="世界观规则关联")
# 职业属性加成(可选,JSON格式)
attribute_bonuses = Column(Text, comment="属性加成(JSON): {strength: '+10%', intelligence: '+5%'}")
# 元数据
source = Column(String(20), default='ai', comment="来源: ai/manual")
created_at = Column(DateTime, server_default=func.now(), comment="创建时间")
updated_at = Column(DateTime, server_default=func.now(), onupdate=func.now(), comment="更新时间")
__table_args__ = (
Index('idx_project_id', 'project_id'),
Index('idx_type', 'type'),
)
def __repr__(self):
return f"<Career(id={self.id}, name={self.name}, type={self.type})>"
class CharacterCareer(Base):
"""角色职业关联表"""
__tablename__ = "character_careers"
id = Column(String(36), primary_key=True, default=lambda: str(uuid.uuid4()))
character_id = Column(String(36), ForeignKey("characters.id", ondelete="CASCADE"), nullable=False)
career_id = Column(String(36), ForeignKey("careers.id", ondelete="CASCADE"), nullable=False)
career_type = Column(String(20), nullable=False, comment="main(主职业)/sub(副职业)")
# 阶段进度
current_stage = Column(Integer, nullable=False, default=1, comment="当前阶段(对应职业中的数值)")
stage_progress = Column(Integer, default=0, comment="阶段内进度(0-100")
# 时间记录
started_at = Column(String(100), comment="开始修炼时间(小说时间线)")
reached_current_stage_at = Column(String(100), comment="到达当前阶段时间")
# 备注
notes = Column(Text, comment="备注(如:修炼心得、特殊事件)")
created_at = Column(DateTime, server_default=func.now(), comment="创建时间")
updated_at = Column(DateTime, server_default=func.now(), onupdate=func.now(), comment="更新时间")
__table_args__ = (
Index('idx_character_id', 'character_id'),
Index('idx_career_type', 'career_type'),
Index('idx_character_career', 'character_id', 'career_id', unique=True),
)
def __repr__(self):
return f"<CharacterCareer(character_id={self.character_id}, career_id={self.career_id}, type={self.career_type})>"
+6 -1
View File
@@ -1,5 +1,5 @@
"""角色数据模型"""
from sqlalchemy import Column, String, Text, DateTime, ForeignKey, Boolean
from sqlalchemy import Column, String, Text, DateTime, ForeignKey, Boolean, Integer
from sqlalchemy.sql import func
from app.database import Base
import uuid
@@ -32,6 +32,11 @@ class Character(Base):
organization_purpose = Column(String(500), comment="组织目的")
organization_members = Column(Text, comment="组织成员(JSON)")
# 职业相关字段(冗余字段,用于提升查询性能)
main_career_id = Column(String(36), ForeignKey("careers.id", ondelete="SET NULL"), comment="主职业ID")
main_career_stage = Column(Integer, comment="主职业当前阶段")
sub_careers = Column(Text, comment="副职业列表(JSON): [{\"career_id\": \"xxx\", \"stage\": 3}, ...]")
# 其他
avatar_url = Column(String(500), comment="头像URL")
traits = Column(Text, comment="特征标签(JSON)")
+155
View File
@@ -0,0 +1,155 @@
"""职业相关的Pydantic模型"""
from pydantic import BaseModel, Field
from typing import Optional, List, Dict, Any
from datetime import datetime
class CareerStage(BaseModel):
"""职业阶段模型"""
level: int = Field(..., description="阶段等级")
name: str = Field(..., description="阶段名称")
description: Optional[str] = Field(None, description="阶段描述")
class CareerBase(BaseModel):
"""职业基础模型"""
name: str = Field(..., description="职业名称")
type: str = Field(..., description="职业类型: main(主职业)/sub(副职业)")
description: Optional[str] = Field(None, description="职业描述")
category: Optional[str] = Field(None, description="职业分类")
stages: List[CareerStage] = Field(..., description="职业阶段列表")
max_stage: int = Field(10, description="最大阶段数")
requirements: Optional[str] = Field(None, description="职业要求/限制")
special_abilities: Optional[str] = Field(None, description="特殊能力描述")
worldview_rules: Optional[str] = Field(None, description="世界观规则关联")
attribute_bonuses: Optional[Dict[str, str]] = Field(None, description="属性加成")
class CareerCreate(CareerBase):
"""创建职业的请求模型"""
project_id: str = Field(..., description="项目ID")
source: str = Field("manual", description="来源: ai/manual")
class CareerUpdate(BaseModel):
"""更新职业的请求模型"""
name: Optional[str] = None
type: Optional[str] = None
description: Optional[str] = None
category: Optional[str] = None
stages: Optional[List[CareerStage]] = None
max_stage: Optional[int] = None
requirements: Optional[str] = None
special_abilities: Optional[str] = None
worldview_rules: Optional[str] = None
attribute_bonuses: Optional[Dict[str, str]] = None
class CareerResponse(BaseModel):
"""职业响应模型"""
id: str
project_id: str
name: str
type: str
description: Optional[str] = None
category: Optional[str] = None
stages: List[CareerStage]
max_stage: int
requirements: Optional[str] = None
special_abilities: Optional[str] = None
worldview_rules: Optional[str] = None
attribute_bonuses: Optional[Dict[str, str]] = None
source: str
created_at: datetime
updated_at: datetime
class Config:
from_attributes = True
class CareerListResponse(BaseModel):
"""职业列表响应模型"""
total: int
main_careers: List[CareerResponse] = Field(default_factory=list, description="主职业列表")
sub_careers: List[CareerResponse] = Field(default_factory=list, description="副职业列表")
class CareerGenerateRequest(BaseModel):
"""AI生成职业体系的请求模型"""
project_id: str = Field(..., description="项目ID")
main_career_count: int = Field(5, description="主职业数量", ge=1, le=20)
sub_career_count: int = Field(8, description="副职业数量", ge=0, le=30)
enable_mcp: bool = Field(False, description="是否启用MCP工具增强")
# ===== 角色职业关联相关 =====
class CharacterCareerBase(BaseModel):
"""角色职业关联基础模型"""
career_id: str = Field(..., description="职业ID")
career_type: str = Field(..., description="main(主职业)/sub(副职业)")
current_stage: int = Field(1, description="当前阶段", ge=1)
stage_progress: int = Field(0, description="阶段内进度(0-100", ge=0, le=100)
started_at: Optional[str] = Field(None, description="开始修炼时间")
reached_current_stage_at: Optional[str] = Field(None, description="到达当前阶段时间")
notes: Optional[str] = Field(None, description="备注")
class CharacterCareerCreate(CharacterCareerBase):
"""创建角色职业关联的请求模型"""
character_id: str = Field(..., description="角色ID")
class CharacterCareerUpdate(BaseModel):
"""更新角色职业关联的请求模型"""
current_stage: Optional[int] = Field(None, ge=1)
stage_progress: Optional[int] = Field(None, ge=0, le=100)
reached_current_stage_at: Optional[str] = None
notes: Optional[str] = None
class CharacterCareerDetail(BaseModel):
"""角色职业详情模型(包含职业信息)"""
id: str
character_id: str
career_id: str
career_name: str = Field(..., description="职业名称")
career_type: str
current_stage: int
stage_name: str = Field(..., description="当前阶段名称")
stage_description: Optional[str] = Field(None, description="当前阶段描述")
stage_progress: int
max_stage: int = Field(..., description="该职业的最大阶段")
started_at: Optional[str] = None
reached_current_stage_at: Optional[str] = None
notes: Optional[str] = None
created_at: datetime
updated_at: datetime
class CharacterCareerResponse(BaseModel):
"""角色职业响应模型"""
main_career: Optional[CharacterCareerDetail] = Field(None, description="主职业")
sub_careers: List[CharacterCareerDetail] = Field(default_factory=list, description="副职业列表")
class SetMainCareerRequest(BaseModel):
"""设置主职业请求模型"""
career_id: str = Field(..., description="职业ID")
current_stage: int = Field(1, description="当前阶段", ge=1)
started_at: Optional[str] = Field(None, description="开始修炼时间")
class AddSubCareerRequest(BaseModel):
"""添加副职业请求模型"""
career_id: str = Field(..., description="职业ID")
current_stage: int = Field(1, description="当前阶段", ge=1)
started_at: Optional[str] = Field(None, description="开始修炼时间")
class UpdateCareerStageRequest(BaseModel):
"""更新职业阶段请求模型"""
current_stage: int = Field(..., description="新的阶段", ge=1)
stage_progress: int = Field(0, description="阶段进度", ge=0, le=100)
reached_current_stage_at: Optional[str] = Field(None, description="到达时间")
notes: Optional[str] = Field(None, description="备注")
+15
View File
@@ -44,6 +44,11 @@ class CharacterCreate(BaseModel):
location: Optional[str] = Field(None, description="组织所在地")
motto: Optional[str] = Field(None, description="组织格言/口号")
color: Optional[str] = Field(None, description="组织代表颜色")
# 职业字段
main_career_id: Optional[str] = Field(None, description="主职业ID")
main_career_stage: Optional[int] = Field(None, description="主职业阶段")
sub_careers: Optional[str] = Field(None, description="副职业列表JSON字符串")
class CharacterUpdate(BaseModel):
@@ -67,6 +72,11 @@ class CharacterUpdate(BaseModel):
location: Optional[str] = Field(None, description="组织所在地")
motto: Optional[str] = Field(None, description="组织格言/口号")
color: Optional[str] = Field(None, description="组织代表颜色")
# 职业字段(会同步到CharacterCareer表)
main_career_id: Optional[str] = Field(None, description="主职业ID")
main_career_stage: Optional[int] = Field(None, description="主职业阶段")
sub_careers: Optional[str] = Field(None, description="副职业列表JSON字符串")
class CharacterResponse(CharacterBase):
@@ -83,6 +93,11 @@ class CharacterResponse(CharacterBase):
motto: Optional[str] = Field(None, description="组织格言/口号")
color: Optional[str] = Field(None, description="组织代表颜色")
# 职业信息字段
main_career_id: Optional[str] = Field(None, description="主职业ID")
main_career_stage: Optional[int] = Field(None, description="主职业阶段")
sub_careers: Optional[List[Dict[str, Any]]] = Field(None, description="副职业列表")
class Config:
from_attributes = True
+123 -2
View File
@@ -297,6 +297,39 @@ class AutoCharacterService:
) -> Dict[str, Any]:
"""生成角色详细信息"""
# 🎯 获取项目职业列表
from app.models.career import Career
careers_result = await db.execute(
select(Career)
.where(Career.project_id == project.id)
.order_by(Career.type, Career.name)
)
careers = careers_result.scalars().all()
# 构建职业信息摘要(包含最高阶段信息)
careers_info = ""
if careers:
main_careers = [c for c in careers if c.type == 'main']
sub_careers = [c for c in careers if c.type == 'sub']
if main_careers:
careers_info += "\n\n可用主职业列表(请在career_info中填写职业名称和阶段):\n"
for career in main_careers:
careers_info += f"- 名称: {career.name}, 最高阶段: {career.max_stage}"
if career.description:
careers_info += f", 描述: {career.description[:50]}"
careers_info += "\n"
if sub_careers:
careers_info += "\n可用副职业列表(请在career_info中填写职业名称和阶段):\n"
for career in sub_careers[:5]:
careers_info += f"- 名称: {career.name}, 最高阶段: {career.max_stage}"
if career.description:
careers_info += f", 描述: {career.description[:50]}"
careers_info += "\n"
careers_info += "\n⚠️ 重要提示:生成角色时,职业阶段不能超过该职业的最高阶段!\n"
# 构建角色生成提示词
template = await PromptService.get_template(
"AUTO_CHARACTER_GENERATION",
@@ -315,7 +348,7 @@ class AutoCharacterService:
location=project.world_location or "未设定",
atmosphere=project.world_atmosphere or "未设定",
rules=project.world_rules or "未设定",
existing_characters=existing_chars_summary,
existing_characters=existing_chars_summary + careers_info,
plot_context="根据剧情需要引入的新角色",
character_specification=json.dumps(spec, ensure_ascii=False, indent=2),
mcp_references="" # 暂时不使用MCP增强
@@ -367,6 +400,66 @@ class AutoCharacterService:
is_organization = character_data.get("is_organization", False)
# 提取职业信息(支持通过名称匹配)
career_info = character_data.get("career_info", {})
raw_main_career_name = career_info.get("main_career_name") if career_info else None
main_career_stage = career_info.get("main_career_stage", 1) if career_info else None
raw_sub_careers_data = career_info.get("sub_careers", []) if career_info else []
# 🔧 通过职业名称匹配数据库中的职业ID
from app.models.career import Career, CharacterCareer
main_career_id = None
sub_careers_data = []
# 匹配主职业名称
if raw_main_career_name and not is_organization:
career_check = await db.execute(
select(Career).where(
Career.name == raw_main_career_name,
Career.project_id == project_id,
Career.type == 'main'
)
)
matched_career = career_check.scalar_one_or_none()
if matched_career:
main_career_id = matched_career.id
# ✅ 验证阶段不超过最高阶段
if main_career_stage and main_career_stage > matched_career.max_stage:
logger.warning(f" ⚠️ AI返回的主职业阶段({main_career_stage})超过最高阶段({matched_career.max_stage}),自动修正为最高阶段")
main_career_stage = matched_career.max_stage
logger.info(f" ✅ 主职业名称匹配成功: {raw_main_career_name} -> ID: {main_career_id}, 阶段: {main_career_stage}/{matched_career.max_stage}")
else:
logger.warning(f" ⚠️ AI返回的主职业名称未找到: {raw_main_career_name}")
# 匹配副职业名称
if raw_sub_careers_data and not is_organization and isinstance(raw_sub_careers_data, list):
for sub_data in raw_sub_careers_data[:2]:
if isinstance(sub_data, dict):
career_name = sub_data.get('career_name')
if career_name:
career_check = await db.execute(
select(Career).where(
Career.name == career_name,
Career.project_id == project_id,
Career.type == 'sub'
)
)
matched_career = career_check.scalar_one_or_none()
if matched_career:
sub_stage = sub_data.get('stage', 1)
# ✅ 验证阶段不超过最高阶段
if sub_stage > matched_career.max_stage:
logger.warning(f" ⚠️ AI返回的副职业阶段({sub_stage})超过最高阶段({matched_career.max_stage}),自动修正为最高阶段")
sub_stage = matched_career.max_stage
sub_careers_data.append({
'career_id': matched_career.id,
'stage': sub_stage
})
logger.info(f" ✅ 副职业名称匹配成功: {career_name} -> ID: {matched_career.id}, 阶段: {sub_stage}/{matched_career.max_stage}")
else:
logger.warning(f" ⚠️ AI返回的副职业名称未找到: {career_name}")
# 创建角色
character = Character(
project_id=project_id,
@@ -381,12 +474,40 @@ class AutoCharacterService:
relationships=character_data.get("relationships_text", ""),
organization_type=character_data.get("organization_type") if is_organization else None,
organization_purpose=character_data.get("organization_purpose") if is_organization else None,
traits=json.dumps(character_data.get("traits", []), ensure_ascii=False) if character_data.get("traits") else None
traits=json.dumps(character_data.get("traits", []), ensure_ascii=False) if character_data.get("traits") else None,
main_career_id=main_career_id,
main_career_stage=main_career_stage if main_career_id else None,
sub_careers=json.dumps(sub_careers_data, ensure_ascii=False) if sub_careers_data else None
)
db.add(character)
await db.flush()
# 处理主职业关联
if main_career_id and not is_organization:
char_career = CharacterCareer(
character_id=character.id,
career_id=main_career_id,
career_type='main',
current_stage=main_career_stage,
stage_progress=0
)
db.add(char_career)
logger.info(f" ✅ 创建主职业关联: {character.name} -> {raw_main_career_name}")
# 处理副职业关联
if sub_careers_data and not is_organization:
for sub_data in sub_careers_data:
char_career = CharacterCareer(
character_id=character.id,
career_id=sub_data['career_id'],
career_type='sub',
current_stage=sub_data['stage'],
stage_progress=0
)
db.add(char_career)
logger.info(f" ✅ 创建副职业关联: {character.name}, 数量: {len(sub_careers_data)}")
# 如果是组织,创建Organization记录
if is_organization:
org = Organization(
+234
View File
@@ -0,0 +1,234 @@
"""职业生成服务"""
from typing import Dict, Any, List
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
import json
from app.models.project import Project
from app.models.career import Career
from app.services.ai_service import AIService
from app.logger import get_logger
logger = get_logger(__name__)
class CareerService:
"""职业相关业务逻辑服务"""
@staticmethod
async def get_career_generation_prompt(
project: Project,
main_career_count: int = 2,
sub_career_count: int = 6
) -> str:
"""
构建职业体系生成的提示词
Args:
project: 项目对象
main_career_count: 主职业数量
sub_career_count: 副职业数量
Returns:
完整的提示词
"""
project_context = f"""
项目信息:
- 书名:{project.title}
- 类型:{project.genre or '未设定'}
- 主题:{project.theme or '未设定'}
- 时间背景:{project.world_time_period or '未设定'}
- 地理位置:{project.world_location or '未设定'}
- 氛围基调:{project.world_atmosphere or '未设定'}
- 世界规则:{project.world_rules or '未设定'}
"""
user_requirements = f"""
生成要求:
- 主职业数量:{main_career_count}
- 副职业数量:{sub_career_count}
- 主职业必须严格符合世界观规则,体现核心能力体系
- 副职业可以更加自由灵活,包含生产、辅助、特殊类型
"""
prompt = f"""{project_context}
{user_requirements}
请为这个小说项目生成完整的职业体系。返回JSON格式,结构如下:
{{
"main_careers": [
{{
"name": "职业名称",
"description": "职业描述(100-200字)",
"category": "职业分类(如:战斗系、法术系、体修系等)",
"stages": [
{{"level": 1, "name": "阶段名称", "description": "阶段描述"}},
{{"level": 2, "name": "阶段名称", "description": "阶段描述"}},
...(共10个阶段)
],
"max_stage": 10,
"requirements": "职业要求(如:需要特定天赋、资质等)",
"special_abilities": "特殊能力描述",
"worldview_rules": "世界观规则关联(说明该职业如何融入世界观)",
"attribute_bonuses": {{"strength": "+10%", "intelligence": "+5%"}}
}}
],
"sub_careers": [
{{
"name": "副职业名称",
"description": "职业描述",
"category": "生产系/辅助系/特殊系",
"stages": [
{{"level": 1, "name": "阶段名称", "description": "阶段描述"}},
...5-8个阶段)
],
"max_stage": 5,
"requirements": "职业要求",
"special_abilities": "特殊能力"
}}
]
}}
重要注意事项:
1. 主职业的阶段设定要详细,体现明确的成长路径,阶段名称要有特色
2. 根据小说类型选择合适的职业:
- 修仙类:剑修、体修、法修、符修等,阶段如:炼气、筑基、金丹、元婴...
- 玄幻类:战士、法师、刺客等,阶段如:见习、初级、中级、高级...
- 都市异能:异能者分类,阶段如:觉醒、初阶、中阶、高阶...
- 科幻未来:基因战士、机甲师等,阶段如:E级、D级、C级、B级...
3. 副职业要有实用性和趣味性,如:炼丹师、炼器师、阵法师、驯兽师、医师等
4. 所有职业都要符合项目的整体世界观设定
5. 阶段描述要简洁明了,体现该阶段的核心特征
6. **只返回纯JSON对象,不要添加任何解释文字或markdown标记**
"""
return prompt
@staticmethod
async def parse_and_save_careers(
career_data: Dict[str, Any],
project_id: str,
db: AsyncSession
) -> Dict[str, List[str]]:
"""
解析AI返回的职业数据并保存到数据库
Args:
career_data: AI返回的职业数据(已解析为dict)
project_id: 项目ID
db: 数据库会话
Returns:
{"main_careers": [...], "sub_careers": [...]} 创建的职业名称列表
"""
result = {
"main_careers": [],
"sub_careers": []
}
# 保存主职业
for idx, career_info in enumerate(career_data.get("main_careers", [])):
try:
stages_json = json.dumps(career_info.get("stages", []), ensure_ascii=False)
attribute_bonuses = career_info.get("attribute_bonuses")
attribute_bonuses_json = json.dumps(attribute_bonuses, ensure_ascii=False) if attribute_bonuses else None
career = Career(
project_id=project_id,
name=career_info.get("name", f"未命名主职业{idx+1}"),
type="main",
description=career_info.get("description"),
category=career_info.get("category"),
stages=stages_json,
max_stage=career_info.get("max_stage", 10),
requirements=career_info.get("requirements"),
special_abilities=career_info.get("special_abilities"),
worldview_rules=career_info.get("worldview_rules"),
attribute_bonuses=attribute_bonuses_json,
source="ai"
)
db.add(career)
await db.flush()
result["main_careers"].append(career.name)
logger.info(f" ✅ 创建主职业:{career.name}")
except Exception as e:
logger.error(f" ❌ 创建主职业失败:{str(e)}")
continue
# 保存副职业
for idx, career_info in enumerate(career_data.get("sub_careers", [])):
try:
stages_json = json.dumps(career_info.get("stages", []), ensure_ascii=False)
attribute_bonuses = career_info.get("attribute_bonuses")
attribute_bonuses_json = json.dumps(attribute_bonuses, ensure_ascii=False) if attribute_bonuses else None
career = Career(
project_id=project_id,
name=career_info.get("name", f"未命名副职业{idx+1}"),
type="sub",
description=career_info.get("description"),
category=career_info.get("category"),
stages=stages_json,
max_stage=career_info.get("max_stage", 5),
requirements=career_info.get("requirements"),
special_abilities=career_info.get("special_abilities"),
worldview_rules=career_info.get("worldview_rules"),
attribute_bonuses=attribute_bonuses_json,
source="ai"
)
db.add(career)
await db.flush()
result["sub_careers"].append(career.name)
logger.info(f" ✅ 创建副职业:{career.name}")
except Exception as e:
logger.error(f" ❌ 创建副职业失败:{str(e)}")
continue
await db.commit()
return result
@staticmethod
async def get_project_careers_summary(project_id: str, db: AsyncSession) -> Dict[str, Any]:
"""
获取项目职业体系摘要
Args:
project_id: 项目ID
db: 数据库会话
Returns:
职业体系摘要信息
"""
result = await db.execute(
select(Career).where(Career.project_id == project_id)
)
careers = result.scalars().all()
main_careers = []
sub_careers = []
for career in careers:
career_info = {
"id": career.id,
"name": career.name,
"category": career.category,
"max_stage": career.max_stage
}
if career.type == "main":
main_careers.append(career_info)
else:
sub_careers.append(career_info)
return {
"main_careers": main_careers,
"sub_careers": sub_careers,
"total_count": len(careers)
}
# 创建全局服务实例
career_service = CareerService()
@@ -0,0 +1,398 @@
"""职业更新服务 - 根据章节分析自动更新角色职业信息"""
from typing import Dict, Any, List, Optional
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from app.models.character import Character
from app.models.career import Career, CharacterCareer
from app.logger import get_logger
logger = get_logger(__name__)
class CareerUpdateService:
"""职业更新服务 - 根据章节分析结果自动更新角色职业"""
@staticmethod
async def update_careers_from_analysis(
db: AsyncSession,
project_id: str,
character_states: List[Dict[str, Any]],
chapter_id: str,
chapter_number: int
) -> Dict[str, Any]:
"""
根据章节分析结果更新角色职业
Args:
db: 数据库会话
project_id: 项目ID
character_states: 角色状态变化列表(来自PlotAnalysis
chapter_id: 章节ID
chapter_number: 章节编号
Returns:
更新结果字典,包含更新数量和变更日志
"""
if not character_states:
logger.info("📋 角色状态列表为空,跳过职业更新")
return {"updated_count": 0, "changes": []}
updated_count = 0
changes_log = []
logger.info(f"🔍 开始分析第{chapter_number}章的角色职业变化...")
for char_state in character_states:
char_name = char_state.get('character_name')
career_changes = char_state.get('career_changes', {})
# 如果没有职业变化信息,跳过
if not career_changes or not isinstance(career_changes, dict):
continue
# 检查是否有实质性的职业变化
main_stage_change = career_changes.get('main_career_stage_change', 0)
sub_career_changes = career_changes.get('sub_career_changes', [])
new_careers = career_changes.get('new_careers', [])
if main_stage_change == 0 and not sub_career_changes and not new_careers:
continue
logger.info(f" 👤 检测到角色 [{char_name}] 有职业变化")
# 1. 查询角色
char_result = await db.execute(
select(Character).where(
Character.name == char_name,
Character.project_id == project_id
)
)
character = char_result.scalar_one_or_none()
if not character:
logger.warning(f" ⚠️ 角色不存在: {char_name},跳过")
continue
# 2. 更新主职业阶段
if main_stage_change != 0 and character.main_career_id:
success = await CareerUpdateService._update_main_career_stage(
db=db,
character=character,
stage_change=main_stage_change,
chapter_number=chapter_number,
career_changes=career_changes,
changes_log=changes_log
)
if success:
updated_count += 1
# 3. 更新副职业(如果有)
if sub_career_changes and isinstance(sub_career_changes, list):
for sub_change in sub_career_changes:
success = await CareerUpdateService._update_sub_career_stage(
db=db,
character=character,
project_id=project_id,
sub_change=sub_change,
chapter_number=chapter_number,
changes_log=changes_log
)
if success:
updated_count += 1
# 4. 添加新职业(如果有)
if new_careers and isinstance(new_careers, list):
for new_career_name in new_careers:
success = await CareerUpdateService._add_new_career(
db=db,
character=character,
project_id=project_id,
career_name=new_career_name,
chapter_number=chapter_number,
changes_log=changes_log
)
if success:
updated_count += 1
# 提交所有更改
if updated_count > 0:
await db.commit()
logger.info(f"✅ 职业更新完成: 共更新了 {updated_count} 个角色的职业信息")
else:
logger.info("📋 本章没有角色职业变化")
return {
"updated_count": updated_count,
"changes": changes_log
}
@staticmethod
async def _update_main_career_stage(
db: AsyncSession,
character: Character,
stage_change: int,
chapter_number: int,
career_changes: Dict[str, Any],
changes_log: List[Dict[str, Any]]
) -> bool:
"""更新主职业阶段"""
try:
# 查询主职业关联
char_career_result = await db.execute(
select(CharacterCareer).where(
CharacterCareer.character_id == character.id,
CharacterCareer.career_type == 'main'
)
)
char_career = char_career_result.scalar_one_or_none()
if not char_career:
logger.warning(f" ⚠️ {character.name} 没有主职业关联记录")
return False
# 查询职业信息
career_result = await db.execute(
select(Career).where(Career.id == char_career.career_id)
)
career = career_result.scalar_one_or_none()
if not career:
logger.warning(f" ⚠️ 职业ID {char_career.career_id} 不存在")
return False
# 计算新阶段(不超过最大阶段,不低于1)
old_stage = char_career.current_stage
new_stage = min(max(1, old_stage + stage_change), career.max_stage)
# 如果没有实际变化,跳过
if new_stage == old_stage:
logger.info(f" 📊 {character.name}{career.name} 已达到边界,无法变更")
return False
# 更新CharacterCareer表
char_career.current_stage = new_stage
# 同步更新Character表的冗余字段
character.main_career_stage = new_stage
# 记录变更日志
change_desc = f"{'晋升' if stage_change > 0 else '降级'}"
breakthrough_desc = career_changes.get('career_breakthrough', '')
changes_log.append({
'character': character.name,
'career': career.name,
'career_type': 'main',
'old_stage': old_stage,
'new_stage': new_stage,
'change': stage_change,
'chapter': chapter_number,
'description': breakthrough_desc
})
logger.info(
f"{character.name} 的主职业 [{career.name}] "
f"{old_stage}阶 → {new_stage}阶 ({change_desc})"
)
if breakthrough_desc:
logger.info(f" 突破描述: {breakthrough_desc[:50]}...")
return True
except Exception as e:
logger.error(f" ❌ 更新主职业失败: {str(e)}")
return False
@staticmethod
async def _update_sub_career_stage(
db: AsyncSession,
character: Character,
project_id: str,
sub_change: Dict[str, Any],
chapter_number: int,
changes_log: List[Dict[str, Any]]
) -> bool:
"""更新副职业阶段"""
try:
career_name = sub_change.get('career_name')
stage_change = sub_change.get('stage_change', 0)
if not career_name or stage_change == 0:
return False
# 1. 查询职业(通过名称)
career_result = await db.execute(
select(Career).where(
Career.name == career_name,
Career.project_id == project_id,
Career.type == 'sub'
)
)
career = career_result.scalar_one_or_none()
if not career:
logger.warning(f" ⚠️ 副职业 [{career_name}] 不存在")
return False
# 2. 查询角色-职业关联
char_career_result = await db.execute(
select(CharacterCareer).where(
CharacterCareer.character_id == character.id,
CharacterCareer.career_id == career.id,
CharacterCareer.career_type == 'sub'
)
)
char_career = char_career_result.scalar_one_or_none()
if not char_career:
logger.warning(f" ⚠️ {character.name} 没有 [{career_name}] 副职业")
return False
# 3. 计算新阶段
old_stage = char_career.current_stage
new_stage = min(max(1, old_stage + stage_change), career.max_stage)
if new_stage == old_stage:
return False
# 4. 更新阶段
char_career.current_stage = new_stage
# 5. 同步更新Character表的sub_careers JSON字段
import json
sub_careers = json.loads(character.sub_careers) if character.sub_careers else []
for sc in sub_careers:
if sc.get('career_id') == career.id:
sc['stage'] = new_stage
break
character.sub_careers = json.dumps(sub_careers, ensure_ascii=False)
# 6. 记录变更
changes_log.append({
'character': character.name,
'career': career.name,
'career_type': 'sub',
'old_stage': old_stage,
'new_stage': new_stage,
'change': stage_change,
'chapter': chapter_number
})
logger.info(
f"{character.name} 的副职业 [{career.name}] "
f"{old_stage}阶 → {new_stage}"
)
return True
except Exception as e:
logger.error(f" ❌ 更新副职业失败: {str(e)}")
return False
@staticmethod
async def _add_new_career(
db: AsyncSession,
character: Character,
project_id: str,
career_name: str,
chapter_number: int,
changes_log: List[Dict[str, Any]]
) -> bool:
"""为角色添加新职业"""
try:
# 1. 查询职业
career_result = await db.execute(
select(Career).where(
Career.name == career_name,
Career.project_id == project_id
)
)
career = career_result.scalar_one_or_none()
if not career:
logger.warning(f" ⚠️ 职业 [{career_name}] 不存在,无法添加")
return False
# 2. 检查是否已存在
existing_result = await db.execute(
select(CharacterCareer).where(
CharacterCareer.character_id == character.id,
CharacterCareer.career_id == career.id
)
)
if existing_result.scalar_one_or_none():
logger.info(f" 📋 {character.name} 已拥有 [{career_name}],跳过")
return False
# 3. 根据职业类型添加
if career.type == 'main':
# 检查是否已有主职业
if character.main_career_id:
logger.warning(f" ⚠️ {character.name} 已有主职业,无法添加新主职业")
return False
# 添加主职业
import uuid
new_char_career = CharacterCareer(
id=str(uuid.uuid4()),
character_id=character.id,
career_id=career.id,
career_type='main',
current_stage=1
)
db.add(new_char_career)
# 更新Character表
character.main_career_id = career.id
character.main_career_stage = 1
logger.info(f"{character.name} 获得新主职业 [{career_name}]")
else: # sub职业
# 检查副职业数量(最多2个)
sub_count_result = await db.execute(
select(CharacterCareer).where(
CharacterCareer.character_id == character.id,
CharacterCareer.career_type == 'sub'
)
)
if len(sub_count_result.scalars().all()) >= 2:
logger.warning(f" ⚠️ {character.name} 的副职业已达上限(2个)")
return False
# 添加副职业
import uuid
new_char_career = CharacterCareer(
id=str(uuid.uuid4()),
character_id=character.id,
career_id=career.id,
career_type='sub',
current_stage=1
)
db.add(new_char_career)
# 更新Character表的sub_careers JSON
import json
sub_careers = json.loads(character.sub_careers) if character.sub_careers else []
sub_careers.append({
'career_id': career.id,
'stage': 1
})
character.sub_careers = json.dumps(sub_careers, ensure_ascii=False)
logger.info(f"{character.name} 获得新副职业 [{career_name}]")
# 记录变更
changes_log.append({
'character': character.name,
'career': career.name,
'career_type': career.type,
'action': 'new',
'chapter': chapter_number
})
return True
except Exception as e:
logger.error(f" ❌ 添加新职业失败: {str(e)}")
return False
+141 -2
View File
@@ -745,6 +745,15 @@ class PromptService:
- 特殊技能或知识
- 符合世界观设定
7. **职业信息**(重要 - 如果项目上下文中包含职业列表):
- 仔细查看项目上下文中的"可用主职业""可用副职业"列表
- 主职业:必须从"可用主职业"列表中选择一个最符合角色设定的职业,填写其职业名称(name字段)
- 主职业阶段:根据职业的阶段信息和角色实力,设定合理的当前阶段(1到职业的max_stage)
- 副职业:可以从"可用副职业"列表中选择0-2个,每个包含职业名称和阶段
- 如果项目没有职业列表,则不需要填写career_info字段
- 职业选择必须与角色的背景故事、能力特点和故事定位高度契合
- ⚠️ 重要:请填写职业的名称而非ID,系统会自动匹配
**重要格式要求:**
1. 只返回纯JSON格式,不要包含任何markdown标记、代码块标记或其他说明文字
2. JSON字符串值的内容描述中严禁使用任何特殊符号(包括中文引号、英文引号、方括号、书名号等)
@@ -781,7 +790,18 @@ class PromptService:
"joined_at": "加入时间(可选)",
"status": "active"
}}
]
],
"career_info": {{
"main_career_name": "从项目上下文的可用主职业列表中复制的职业名称",
"main_career_stage": 5,
"sub_careers": [
{{
"career_name": "从项目上下文的可用副职业列表中复制的职业名称",
"stage": 3
}}
]
}}
}}
**关系类型参考(请从中选择或自定义):**
@@ -947,6 +967,12 @@ class PromptService:
- 关系变化
- 关键行动和决策
- 成长或退步
- **💼 职业变化(重要 - 新增)**:
- 如果角色在本章有职业相关的进展或突破,请详细分析
- 主职业阶段变化: 是否晋级、突破或降级(用整数表示变化量,如: +1表示晋升一阶段,-1表示退步一阶段,0表示无变化)
- 副职业变化: 是否学习新的副职业或副职业有所精进
- 职业突破描述: 具体的突破过程、原因和标志性事件
- 注意:只有当章节中明确描述了职业相关的成长、突破或变化时才填写此项
### 6. 关键情节点 (Plot Points)
列出3-5个核心情节点:
@@ -1054,6 +1080,13 @@ class PromptService:
2. keyword必须是从章节原文中逐字复制的文本,长度8-25字
3. keyword用于在前端标注文本位置,所以必须能在原文中精确找到
4. 不要使用概括性语句或改写后的文字作为keyword
5. **职业变化字段说明**:
- career_changes是可选字段,只有当章节中明确描述了职业相关变化时才填写
- main_career_stage_change: 整数,表示主职业阶段变化量(+1=晋升一阶,-1=退步一阶,0=无变化)
- sub_career_changes: 数组,包含副职业的变化,每项包含career_name(职业名称)和stage_change(阶段变化量)
- new_careers: 数组,包含新获得的职业名称(如果有)
- career_breakthrough: 字符串,描述职业突破的具体过程和标志性事件
- 如果角色没有职业变化,可以不填写career_changes字段或设为空对象
只返回JSON,不要其他说明。"""
@@ -1533,6 +1566,7 @@ class PromptService:
3. 性格、背景要有深度和独特性
4. 外貌描写要具体生动
5. 特长和能力要符合角色定位
6. **如果【已有角色】中包含职业列表,必须为角色设定职业**(参考下方职业信息要求)
**关系建立指导(非常重要):**
- 仔细审视【已有角色】列表,思考新角色与哪些现有角色有联系
@@ -1546,6 +1580,15 @@ class PromptService:
2. JSON字符串值中严禁使用特殊符号(引号、方括号、书名号等)
3. 所有专有名词直接书写,不使用任何符号包裹
【职业信息要求(重要)】
如果【已有角色】部分包含"可用主职业列表""可用副职业列表",则必须:
- 仔细查看可用的主职业和副职业列表
- 根据角色的背景、能力、故事定位,选择最合适的职业
- 主职业:从"可用主职业列表"中选择一个,填写职业名称(name字段)
- 主职业阶段:根据职业的阶段信息和角色实力,设定合理的当前阶段
- 副职业:可选择0-2个副职业,每个包含职业名称和阶段
- ⚠️ 重要:必须填写职业的名称而非ID,系统会自动匹配
请严格按照以下JSON格式返回:
{{
"name": "角色姓名",
@@ -1574,7 +1617,18 @@ class PromptService:
"rank": 5,
"loyalty": 80
}}
]
],
"career_info": {{
"main_career_name": "从可用主职业列表中选择的职业名称",
"main_career_stage": 5,
"sub_careers": [
{{
"career_name": "从可用副职业列表中选择的职业名称",
"stage": 3
}}
]
}}
}}
**关系类型参考(从中选择或自定义):**
@@ -1605,6 +1659,85 @@ class PromptService:
只返回纯JSON对象,不要有```json```这样的标记。"""
# 职业体系生成提示词
CAREER_SYSTEM_GENERATION = """你是专业的游戏/小说职业体系设计师。请根据以下世界观信息,设计一个完整且合理的职业体系。
【项目信息】
- 书名:{title}
- 类型:{genre}
- 主题:{theme}
- 时间背景:{time_period}
- 地理位置:{location}
- 氛围基调:{atmosphere}
- 世界规则:{rules}
【设计要求】
1. **主职业(main_careers**
- 根据世界观特点,决定需要多少个主职业
- 主职业是角色的核心发展方向,直接影响战斗力或核心能力
- 必须严格符合世界观规则,体现核心能力体系
- 每个主职业的阶段数量可以不同:根据职业的复杂度、重要性、修炼难度等因素,为不同职业设定不同的max_stage
2. **副职业(sub_careers**
- 根据世界需要,决定需要多少个副职业
- 副职业包含生产、辅助、特殊技能类,丰富角色的能力维度
- 每个副职业的阶段数量可以不同:简单的副职业可能只有3-5个阶段,复杂的可能有6-10个阶段
- 不要让所有副职业都是相同的阶段数
3. **阶段设计(stages**
- 每个职业的stages数组长度必须等于max_stage
- 阶段名称要符合世界观文化背景和时代特征
- 阶段描述要体现明确的能力提升和成长路径
- 重要:确保职业间的阶段数量有差异,体现职业的多样性
【JSON格式】
{{
"main_careers": [
{{
"name": "职业名称",
"description": "职业描述(100-150字),说明职业特点和定位",
"category": "职业分类(如:战斗系、法术系、体修系等)",
"stages": [
{{"level": 1, "name": "阶段1名称", "description": "阶段描述"}},
{{"level": 2, "name": "阶段2名称", "description": "阶段描述"}},
...数组长度应等于max_stage...
],
"max_stage": 根据职业复杂度自行决定的整数,
"requirements": "职业要求和前置条件",
"special_abilities": "职业特殊能力和特色",
"worldview_rules": "与世界观规则的关联",
"attribute_bonuses": {{"strength": "+10%", "intelligence": "+5%"}}
}}
],
"sub_careers": [
{{
"name": "副职业名称",
"description": "职业描述(80-120字)",
"category": "生产系/辅助系/特殊系",
"stages": [
{{"level": 1, "name": "阶段1名称", "description": "阶段描述"}},
...数组长度应等于max_stage...
],
"max_stage": 根据职业特性自行决定的整数,
"requirements": "职业要求",
"special_abilities": "特殊能力"
}}
]
}}
【重要提示】
- 职业的数量、类型完全由你根据世界观自行决定,不要受任何数字限制
- **阶段数量多样性(关键)**
- 不同职业的max_stage必须不同,不要所有职业都是相同的阶段数
- 主职业的阶段数建议范围:5-15个阶段(根据职业重要性和复杂度灵活设定)
- 副职业的阶段数建议范围:3-10个阶段(根据职业特性灵活设定)
- 例如:剑修可能有12个阶段,炼丹师可能有8个阶段,体修可能有10个阶段
- 确保职业体系与世界观高度契合,符合该世界的逻辑和文化
- 只返回纯JSON,不要添加markdown标记或其他解释文字
请让每个职业的阶段数有所不同,体现职业的独特性和多样性!"""
@staticmethod
def format_prompt(template: str, **kwargs) -> str:
"""
@@ -2043,6 +2176,12 @@ class PromptService:
"description": "根据剧情需求自动生成新角色的完整设定",
"parameters": ["title", "genre", "theme", "time_period", "location", "atmosphere", "rules",
"existing_characters", "plot_context", "character_specification", "mcp_references"]
},
"CAREER_SYSTEM_GENERATION": {
"name": "职业体系生成",
"category": "世界构建",
"description": "根据世界观自动生成完整的职业体系,包括主职业和副职业",
"parameters": ["title", "genre", "theme", "time_period", "location", "atmosphere", "rules"]
}
}