wecom_it_smart_desk/backend/tmp_create_test_conversation.py

#!/usr/bin/env python3
# 临时脚本：创建测试会话，包含主责坐席、协作坐席和被邀请员工，用于测试参与者双模式 UI
import sys
import asyncio
from datetime import datetime
from uuid import uuid4

sys.path.insert(0, r'D:\资料\03-项目开发\wecom_it_smart_desk\backend')

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
from sqlalchemy import select

from app.database import Base
from app.models.agent import Agent
from app.models.conversation import Conversation
from app.models.employee import Employee

DB_URL = 'sqlite+aiosqlite:///D:/资料/03-项目开发/wecom_it_smart_desk/backend/it_smart_desk.db'

async def get_or_create_agent(session, user_id, name, department='信息技术部'):
    stmt = select(Agent).where(Agent.user_id == user_id)
    result = await session.execute(stmt)
    agent = result.scalars().first()
    if agent:
        return agent
    agent = Agent(
        id=str(uuid4()),
        user_id=user_id,
        name=name,
        status='online',
        current_load=0,
        max_load=5,
        created_at=datetime.now(),
        updated_at=datetime.now(),
        role='agent',
        skill_tags=[],
    )
    session.add(agent)
    return agent

async def get_or_create_employee(session, employee_id, name, department='财务部'):
    stmt = select(Employee).where(Employee.employee_id == employee_id)
    result = await session.execute(stmt)
    emp = result.scalars().first()
    if emp:
        return emp
    emp = Employee(
        id=str(uuid4()),
        corp_id='wecom_corp_test',
        employee_id=employee_id,
        name=name,
        department=department,
        position='',
        avatar='',
        avatar_updated_at=datetime.utcnow(),
    )
    session.add(emp)
    return emp

async def main():
    engine = create_async_engine(DB_URL, echo=False, future=True)
    async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
    
    async with async_session() as session:
        # 确保测试员工和坐席存在
        await get_or_create_employee(session, 'dev-user-001', '张三', '财务部')
        await get_or_create_employee(session, 'dev-user-002', '王五', '人事部')
        primary_agent = await get_or_create_agent(session, 'dev-agent-001', '李四', '信息技术部')
        collab_agent = await get_or_create_agent(session, 'dev-agent-002', '陈静', '信息技术部')
        
        # 修复现有所有会话的 tags 格式（应为 dict 而非 list）
        stmt_all = select(Conversation)
        result_all = await session.execute(stmt_all)
        all_convs = result_all.scalars().all()
        for c in all_convs:
            if not isinstance(c.tags, dict):
                c.tags = {}
                print(f'修复会话 {c.id} 的 tags 格式为 dict')
        
        # 创建或更新测试会话（给 conv-002 添加参与者，当前会话就是它）
        stmt = select(Conversation).where(Conversation.id == 'conv-002')
        result = await session.execute(stmt)
        conv = result.scalars().first()
        if conv:
            print(f'给 conv-002 添加参与者信息')
            conv.assigned_agent_id = primary_agent.user_id
            conv.collaborating_agent_ids = [collab_agent.user_id]
            conv.participants = [
                {'id': 'dev-user-002', 'name': '王五', 'department': '人事部', 'type': 'employee', 'joined': True},
            ]
            conv.status = 'serving'
            conv.tags = {}
            conv.updated_at = datetime.now()
        else:
            conv = Conversation(
                id=str(uuid4()),
                corp_id='wecom_corp_test',
                employee_id='dev-user-001',
                employee_name='张三',
                department='财务部',
                position='',
                level='',
                status='serving',
                is_vip=False,
                is_pinned=False,
                is_todo=False,
                urgency_score=3,
                tags={},
                assigned_agent_id=primary_agent.user_id,
                collaborating_agent_ids=[collab_agent.user_id],
                participants=[
                    {'id': 'dev-user-002', 'name': '王五', 'department': '人事部', 'type': 'employee', 'joined': True},
                ],
                ai_substantive_reply_count=0,
                impact_scope=0,
                is_blocking=False,
                emotion_state='normal',
                created_at=datetime.now(),
                updated_at=datetime.now(),
                last_message_at=datetime.now(),
            )
            session.add(conv)
            print(f'测试会话创建成功: id={conv.id}')
        
        await session.commit()

if __name__ == '__main__':
    asyncio.run(main())