import uuid from fastapi import APIRouter, Depends, HTTPException, Query, status from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.api.deps import get_current_user from app.database import get_db from app.models.user import User from app.models.brand import Brand from app.models.monitoring import MonitoringRecord from app.schemas.monitoring import ( MonitoringRecordCreate, MonitoringRecordResponse, MonitoringRecordList, MonitoringChangeReport, MonitoringStatusUpdate, ) from app.services.monitoring.monitor_service import MonitorService router = APIRouter() def _to_uuid(value: str | uuid.UUID) -> uuid.UUID: if isinstance(value, uuid.UUID): return value return uuid.UUID(str(value)) async def _get_brand_with_access( brand_id: uuid.UUID, db: AsyncSession, current_user: User, ) -> Brand: stmt = select(Brand).where( Brand.id == brand_id, Brand.user_id == _to_uuid(current_user.id), ) result = await db.execute(stmt) brand = result.scalar_one_or_none() if not brand: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="品牌不存在", ) return brand def _record_to_response(record: MonitoringRecord) -> MonitoringRecordResponse: return MonitoringRecordResponse( id=record.id, brand_id=record.brand_id, content_id=record.content_id, query_keywords=record.query_keywords, platform=record.platform, baseline_citation_count=record.baseline_citation_count, baseline_sentiment=record.baseline_sentiment, baseline_rank=record.baseline_rank, current_citation_count=record.current_citation_count, current_sentiment=record.current_sentiment, current_rank=record.current_rank, change_type=record.change_type, change_details=record.change_details, check_interval_hours=record.check_interval_hours, last_checked_at=record.last_checked_at, next_check_at=record.next_check_at, status=record.status, created_at=record.created_at, updated_at=record.updated_at, ) @router.post("/tasks", response_model=MonitoringRecordResponse) async def create_monitoring_task( request: MonitoringRecordCreate, current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): await _get_brand_with_access(request.brand_id, db, current_user) service = MonitorService() record = await service.create_monitoring_record( db=db, brand_id=request.brand_id, content_id=request.content_id, query_keywords=request.query_keywords, platform=request.platform, check_interval_hours=request.check_interval_hours, ) return _record_to_response(record) @router.get("/brand/{brand_id}", response_model=MonitoringRecordList) async def get_brand_monitoring( brand_id: uuid.UUID, skip: int = Query(0, ge=0), limit: int = Query(20, ge=1, le=100), current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): await _get_brand_with_access(brand_id, db, current_user) service = MonitorService() records, total = await service.get_brand_monitoring( db=db, brand_id=brand_id, skip=skip, limit=limit, ) return MonitoringRecordList( records=[_record_to_response(r) for r in records], total=total, ) @router.get("/{record_id}/report", response_model=MonitoringChangeReport) async def get_change_report( record_id: uuid.UUID, current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): stmt = select(MonitoringRecord).where(MonitoringRecord.id == record_id) result = await db.execute(stmt) record = result.scalar_one_or_none() if not record: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="监测记录不存在", ) await _get_brand_with_access(record.brand_id, db, current_user) service = MonitorService() report = await service.generate_change_report(db, record_id) if not report: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="变化报告不存在", ) return MonitoringChangeReport( monitoring_record_id=uuid.UUID(report["monitoring_record_id"]), brand_id=uuid.UUID(report["brand_id"]), change_type=report["change_type"], change_details=report["change_details"], baseline=report["baseline"], current=report["current"], recommendations=report["recommendations"], ) @router.put("/{record_id}/status", response_model=MonitoringRecordResponse) async def update_monitoring_status( record_id: uuid.UUID, status_update: MonitoringStatusUpdate, current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): valid_statuses = {"active", "paused", "completed"} if status_update.status not in valid_statuses: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=f"无效的状态值,支持: {', '.join(valid_statuses)}", ) stmt = select(MonitoringRecord).where(MonitoringRecord.id == record_id) result = await db.execute(stmt) record = result.scalar_one_or_none() if not record: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="监测记录不存在", ) await _get_brand_with_access(record.brand_id, db, current_user) record.status = status_update.status await db.commit() await db.refresh(record) return _record_to_response(record) @router.post("/{record_id}/check", response_model=MonitoringRecordResponse) async def trigger_manual_check( record_id: uuid.UUID, current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): stmt = select(MonitoringRecord).where(MonitoringRecord.id == record_id) result = await db.execute(stmt) record = result.scalar_one_or_none() if not record: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="监测记录不存在", ) await _get_brand_with_access(record.brand_id, db, current_user) service = MonitorService() updated_record = await service.check_and_compare(db, record_id) if not updated_record: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="检测执行失败", ) return _record_to_response(updated_record)