import logging import uuid from datetime import datetime, timedelta, timezone from sqlalchemy import select, func from sqlalchemy.ext.asyncio import AsyncSession from app.models.monitoring import MonitoringRecord, ContentBaseline from app.models.query import Query from app.models.citation_record import CitationRecord from app.models.brand import Brand logger = logging.getLogger(__name__) class MonitorService: async def create_monitoring_record( self, db: AsyncSession, brand_id: uuid.UUID, content_id: str | None = None, query_keywords: str | None = None, platform: str | None = None, check_interval_hours: int = 24, ) -> MonitoringRecord: now = datetime.now(timezone.utc) stmt = select(Brand).where(Brand.id == brand_id) result = await db.execute(stmt) brand = result.scalar_one_or_none() brand_name = brand.name if brand else "" baseline_data = await self._get_current_metrics(db, brand_id, query_keywords, platform) record = MonitoringRecord( brand_id=brand_id, content_id=content_id, query_keywords=query_keywords, platform=platform, baseline_citation_count=baseline_data.get("citation_count", 0), baseline_sentiment=baseline_data.get("positive_ratio"), baseline_rank=baseline_data.get("avg_rank"), current_citation_count=baseline_data.get("citation_count", 0), current_sentiment=baseline_data.get("positive_ratio"), current_rank=baseline_data.get("avg_rank"), change_type="neutral", change_details=None, check_interval_hours=check_interval_hours, last_checked_at=now, next_check_at=now + timedelta(hours=check_interval_hours), status="active", ) db.add(record) await db.flush() await self._create_baseline_snapshot( db=db, record_id=record.id, brand_name=brand_name, query_keywords=query_keywords, platform=platform, metrics=baseline_data, ) await db.commit() await db.refresh(record) return record async def _create_baseline_snapshot( self, db: AsyncSession, record_id: uuid.UUID, brand_name: str, query_keywords: str | None, platform: str | None, metrics: dict, ) -> ContentBaseline: baseline = ContentBaseline( monitoring_record_id=record_id, brand_name=brand_name, keyword=query_keywords or "", platform=platform or "", citation_count=metrics.get("citation_count", 0), sentiment_score=metrics.get("positive_ratio"), rank_position=metrics.get("avg_rank"), snapshot_data=metrics, ) db.add(baseline) await db.flush() return baseline async def get_brand_monitoring( self, db: AsyncSession, brand_id: uuid.UUID, skip: int = 0, limit: int = 20, ) -> tuple[list[MonitoringRecord], int]: count_stmt = select(func.count()).select_from(MonitoringRecord).where( MonitoringRecord.brand_id == brand_id, ) count_result = await db.execute(count_stmt) total = count_result.scalar_one() stmt = ( select(MonitoringRecord) .where(MonitoringRecord.brand_id == brand_id) .order_by(MonitoringRecord.created_at.desc()) .offset(skip) .limit(limit) ) result = await db.execute(stmt) records = list(result.scalars().all()) return records, total async def check_and_compare( self, db: AsyncSession, record_id: uuid.UUID, ) -> MonitoringRecord | None: stmt = select(MonitoringRecord).where(MonitoringRecord.id == record_id) result = await db.execute(stmt) record = result.scalar_one_or_none() if not record: return None current_data = await self._get_current_metrics( db, record.brand_id, record.query_keywords, record.platform, ) record.current_citation_count = current_data.get("citation_count", 0) record.current_sentiment = current_data.get("positive_ratio") record.current_rank = current_data.get("avg_rank") change_type = self.determine_change_type( baseline_citation=record.baseline_citation_count, current_citation=record.current_citation_count, baseline_sentiment=record.baseline_sentiment, current_sentiment=record.current_sentiment, baseline_rank=record.baseline_rank, current_rank=record.current_rank, ) record.change_type = change_type change_details = self._build_change_details(record, current_data) record.change_details = change_details now = datetime.now(timezone.utc) record.last_checked_at = now record.next_check_at = now + timedelta(hours=record.check_interval_hours) await db.commit() await db.refresh(record) return record def determine_change_type( self, baseline_citation: int, current_citation: int, baseline_sentiment: float | None = None, current_sentiment: float | None = None, baseline_rank: int | None = None, current_rank: int | None = None, ) -> str: positive_signals = 0 negative_signals = 0 if current_citation > baseline_citation: positive_signals += 1 elif current_citation < baseline_citation: negative_signals += 1 if baseline_sentiment is not None and current_sentiment is not None: if current_sentiment > baseline_sentiment: positive_signals += 1 elif current_sentiment < baseline_sentiment: negative_signals += 1 if baseline_rank is not None and current_rank is not None: if current_rank < baseline_rank: positive_signals += 1 elif current_rank > baseline_rank: negative_signals += 1 if positive_signals > negative_signals: return "positive" elif negative_signals > positive_signals: return "negative" return "neutral" def _build_change_details(self, record: MonitoringRecord, current_data: dict) -> dict: details = { "citation_change": { "baseline": record.baseline_citation_count, "current": record.current_citation_count, "delta": record.current_citation_count - record.baseline_citation_count, }, } if record.baseline_sentiment is not None and record.current_sentiment is not None: details["sentiment_change"] = { "baseline": record.baseline_sentiment, "current": record.current_sentiment, "delta": round(record.current_sentiment - record.baseline_sentiment, 4), } if record.baseline_rank is not None and record.current_rank is not None: details["rank_change"] = { "baseline": record.baseline_rank, "current": record.current_rank, "delta": record.current_rank - record.baseline_rank, } details["platform_data"] = current_data.get("platform_data", {}) details["checked_at"] = datetime.now(timezone.utc).isoformat() return details async def generate_change_report( self, db: AsyncSession, record_id: uuid.UUID, ) -> dict | None: stmt = select(MonitoringRecord).where(MonitoringRecord.id == record_id) result = await db.execute(stmt) record = result.scalar_one_or_none() if not record: return None recommendations = self._generate_recommendations(record) baseline = { "citation_count": record.baseline_citation_count, "sentiment": record.baseline_sentiment, "rank": record.baseline_rank, } current = { "citation_count": record.current_citation_count, "sentiment": record.current_sentiment, "rank": record.current_rank, } return { "monitoring_record_id": str(record.id), "brand_id": str(record.brand_id), "change_type": record.change_type, "change_details": record.change_details, "baseline": baseline, "current": current, "recommendations": recommendations, } def _generate_recommendations(self, record: MonitoringRecord) -> list[str]: recommendations = [] if record.change_type == "negative": if record.current_citation_count < record.baseline_citation_count: recommendations.append("引用量下降，建议增加高质量内容发布频率，提升品牌在AI搜索引擎中的曝光") if record.current_sentiment is not None and record.baseline_sentiment is not None: if record.current_sentiment < record.baseline_sentiment: recommendations.append("情感倾向下降，建议关注负面评价并优化品牌形象内容") if record.current_rank is not None and record.baseline_rank is not None: if record.current_rank > record.baseline_rank: recommendations.append("排名下降，建议优化GEO策略，提升内容在AI搜索中的引用优先级") elif record.change_type == "positive": if record.current_citation_count > record.baseline_citation_count: recommendations.append("引用量上升，建议继续保持当前内容策略") if record.current_sentiment is not None and record.baseline_sentiment is not None: if record.current_sentiment > record.baseline_sentiment: recommendations.append("情感倾向改善，当前品牌内容策略效果良好") else: recommendations.append("各项指标保持稳定，建议持续监测") return recommendations async def _get_current_metrics( self, db: AsyncSession, brand_id: uuid.UUID, query_keywords: str | None = None, platform: str | None = None, ) -> dict: stmt = select(Brand).where(Brand.id == brand_id) result = await db.execute(stmt) brand = result.scalar_one_or_none() if not brand: return { "citation_count": 0, "positive_ratio": 0.0, "avg_rank": 0, "platform_data": {}, } conditions = [Query.target_brand == brand.name] if query_keywords: conditions.append(Query.keyword.contains(query_keywords)) queries_stmt = select(Query).where(*conditions) queries_result = await db.execute(queries_stmt) queries = list(queries_result.scalars().all()) if not queries: return { "citation_count": 0, "positive_ratio": 0.0, "avg_rank": 0, "platform_data": {}, } query_ids = [q.id for q in queries] citation_conditions = [CitationRecord.query_id.in_(query_ids)] if platform: citation_conditions.append(CitationRecord.platform == platform) citations_stmt = select(CitationRecord).where(*citation_conditions) citations_result = await db.execute(citations_stmt) all_citations = list(citations_result.scalars().all()) brand_citations = [c for c in all_citations if c.cited] citation_count = len(brand_citations) sentiment_counts = {"positive": 0, "neutral": 0, "negative": 0} for citation in brand_citations: if citation.sentiment and citation.sentiment in sentiment_counts: sentiment_counts[citation.sentiment] += 1 else: sentiment_counts["neutral"] += 1 total_with_sentiment = sum(sentiment_counts.values()) positive_ratio = ( sentiment_counts["positive"] / total_with_sentiment if total_with_sentiment > 0 else 0.0 ) positions = [c.citation_position for c in brand_citations if c.citation_position is not None] avg_rank = int(sum(positions) / len(positions)) if positions else 0 platform_data = {} for citation in all_citations: p = citation.platform if p not in platform_data: platform_data[p] = {"total": 0, "cited": 0} platform_data[p]["total"] += 1 if citation.cited: platform_data[p]["cited"] += 1 platform_scores = {} for p, data in platform_data.items(): platform_scores[p] = round( (data["cited"] / data["total"] * 100) if data["total"] > 0 else 0.0, 2 ) return { "citation_count": citation_count, "positive_ratio": round(positive_ratio, 4), "avg_rank": avg_rank, "platform_data": platform_scores, }