yu-zhi-ran/platform/backend/api/generate.py

# 宇之然内容创作平台 - 文章生成API

from pydantic import BaseModel
from datetime import datetime
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from typing import List, Optional
import asyncio

from core.security import get_current_user, create_audit_log
from app.database import get_db
from app.models import Topic, User, GenerateTask

router = APIRouter()

class BatchGenerateRequest(BaseModel):
    topic_ids: Optional[List[int]] = None

class BatchOptimizeRequest(BaseModel):
    topic_ids: List[int]

@router.post("/run")
async def batch_generate(
    request: BatchGenerateRequest,
    current_user: User = Depends(get_current_user),
    db: Session = Depends(get_db)
):
    """批量创建选题文章"""
    
    # 如果没有指定topic_ids，则获取所有待处理的选题
    if not request.topic_ids:
        topics = db.query(Topic).filter(Topic.status == "待处理").all()
        topic_ids = [t.id for t in topics]
    else:
        topic_ids = request.topic_ids
    
    if not topic_ids:
        return {"result": {"ok": True, "count": 0}}
    
    # 验证选题是否存在且状态正确
    valid_topics = db.query(Topic).filter(
        Topic.id.in_(topic_ids),
        Topic.status == "待处理"
    ).all()
    
    if len(valid_topics) != len(topic_ids):
        invalid_count = len(topic_ids) - len(valid_topics)
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=f"有{invalid_count}个选题状态不正确或不存在"
        )
    
    # 创建生成任务记录
    tasks = []
    for topic in valid_topics:
        task = GenerateTask(
            topic_id=topic.id,
            status="pending",
            created_by=current_user.id
        )
        db.add(task)
        tasks.append(task)
    
    db.commit()
    
    # 异步执行生成任务（简化实现）
    # 实际生产环境应使用Celery等任务队列
    asyncio.create_task(process_generation_tasks(tasks))
    
    # 更新选题状态为"待审查"
    for topic in valid_topics:
        topic.generated_at = datetime.utcnow()
        topic.status = "待审查"
        topic.updated_at = datetime.utcnow()
    
    db.commit()
    
    # 记录审计日志
    create_audit_log(
        db=db,
        user_id=current_user.id,
        action="batch_generate",
        resource_type="topic",
        resource_id=None,
        details=f"处理选题数量: {len(tasks)}"
    )
    
    return {
        "result": {
            "ok": True,
            "count": len(tasks)
        }
    }

@router.post("/optimize/run")
async def batch_optimize(
    request: BatchOptimizeRequest,
    current_user: User = Depends(get_current_user),
    db: Session = Depends(get_db)
):
    """批量优化选题文章"""
    
    # 检查选题是否存在且状态正确（必须是待审查）
    valid_topics = db.query(Topic).filter(
        Topic.id.in_(request.topic_ids),
        Topic.status == "待审查"
    ).all()
    
    if len(valid_topics) != len(request.topic_ids):
        invalid_count = len(request.topic_ids) - len(valid_topics)
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=f"有{invalid_count}个选题状态不正确或不存在"
        )
    
    # 执行优化逻辑（简化实现）
    auto_passed = 0
    need_manual = 0
    
    for topic in valid_topics:
        # 这里应该调用实际的合规性检查逻辑
        # 简化实现：随机决定通过或不通过
        import random
        if random.choice([True, False]):
            topic.status = "待发布"
            auto_passed += 1
        else:
            topic.status = "待审查"
            need_manual += 1
        
        topic.updated_at = datetime.utcnow()
    
    db.commit()
    
    # 记录审计日志
    create_audit_log(
        db=db,
        user_id=current_user.id,
        action="batch_optimize",
        resource_type="topic",
        resource_id=None,
        details=f"自动通过: {auto_passed}, 需人工: {need_manual}"
    )
    
    return {
        "summary": {
            "passed_auto": auto_passed,
            "need_manual": need_manual,
            "total": len(valid_topics)
        }
    }

async def process_generation_tasks(tasks: List[GenerateTask]):
    """处理生成任务（异步函数）"""
    # 这里是生成文章的异步逻辑
    # 实际生产环境应使用Celery等专业的任务队列系统
    
    for task in tasks:
        try:
            # 模拟生成过程
            await asyncio.sleep(2)  # 模拟耗时操作
            
            # 更新任务状态为已完成
            task.status = "completed"
            task.result = {"success": True, "message": "文章生成完成"}
            
        except Exception as e:
            # 处理失败情况
            task.status = "failed"
            task.result = {"success": False, "error": str(e)}
    
    # 注意：这个函数需要访问数据库，实际实现中可能需要额外的依赖注入