from __future__ import annotations from datetime import UTC, datetime, timedelta from .shared import * class SemanticFactAdminMixin: async def _list_project_snapshots(self, *, limit: int) -> list[dict[str, Any]]: rows = ( await self.db.execute( select(Project) .where(Project.tenant_id == self.tenant_id) .order_by(Project.updated_at.desc(), Project.id.desc()) .limit(limit) ) ).scalars().all() return [ { "entity_type": "project", "entity_id": int(row.id), "display_name": str(row.title or row.name or f"项目#{int(row.id)}").strip(), "object_status": str(row.status or "").strip(), "updated_at": row.updated_at.isoformat() if row.updated_at else None, "is_extractable": not _status_is_inactive(row.status), "summary": str(row.description or "").strip(), } for row in rows ] async def _list_task_snapshots(self, *, limit: int) -> list[dict[str, Any]]: rows = ( await self.db.execute( select(Task) .where(Task.tenant_id == self.tenant_id) .order_by(Task.updated_at.desc(), Task.id.desc()) .limit(limit) ) ).scalars().all() return [ { "entity_type": "task", "entity_id": int(row.id), "display_name": str(row.title or f"任务#{int(row.id)}").strip(), "object_status": str(row.status or "").strip(), "updated_at": row.updated_at.isoformat() if row.updated_at else None, "is_extractable": not _status_is_inactive(row.status), "summary": str(row.description or "").strip(), } for row in rows ] async def _list_signal_snapshots(self, *, limit: int) -> list[dict[str, Any]]: rows = ( await self.db.execute( select(Signal) .where(Signal.tenant_id == self.tenant_id) .order_by(Signal.updated_at.desc(), Signal.id.desc()) .limit(limit) ) ).scalars().all() return [ { "entity_type": "signal", "entity_id": int(row.id), "display_name": str(row.title or row.summary or f"信号#{int(row.id)}").strip(), "object_status": str(row.status or "").strip(), "updated_at": row.updated_at.isoformat() if row.updated_at else None, "is_extractable": not _status_is_inactive(row.status), "summary": str(row.summary or row.content or "").strip(), } for row in rows ] async def _list_document_snapshots(self, *, limit: int) -> list[dict[str, Any]]: rows = ( await self.db.execute( select(KnowledgeBaseDoc) .where(KnowledgeBaseDoc.tenant_id == self.tenant_id) .order_by(KnowledgeBaseDoc.updated_at.desc(), KnowledgeBaseDoc.id.desc()) .limit(limit) ) ).scalars().all() return [ { "entity_type": "document", "entity_id": int(row.id), "display_name": str(row.title or f"文档#{int(row.id)}").strip(), "object_status": str(row.status or "").strip(), "updated_at": row.updated_at.isoformat() if row.updated_at else None, "is_extractable": not _status_is_inactive(row.status), "summary": f"{str(row.file_type or '').strip()} / {str(row.source_type or '').strip()}", } for row in rows ] async def _list_user_snapshots(self, *, limit: int) -> list[dict[str, Any]]: rows = ( await self.db.execute( select(User) .where(User.tenant_id == self.tenant_id) .order_by(User.updated_at.desc(), User.id.desc()) .limit(limit) ) ).scalars().all() return [ { "entity_type": "user", "entity_id": int(row.id), "display_name": str(row.name or row.username or f"成员#{int(row.id)}").strip(), "object_status": str(row.status or "").strip(), "updated_at": row.updated_at.isoformat() if row.updated_at else None, "is_extractable": bool(row.is_active) and not _status_is_inactive(row.status), "summary": f"{str(row.role or '').strip()} / {str(row.username or '').strip()}", } for row in rows ] async def _list_truth_node_snapshots(self, *, entity_type: str, limit: int) -> list[dict[str, Any]]: truth_node_type = ENTITY_TYPE_TO_TRUTH_NODE_TYPE[entity_type] rows = ( await self.db.execute( select(TruthNode) .where( and_( TruthNode.tenant_id == self.tenant_id, TruthNode.node_type == truth_node_type, ) ) .order_by(TruthNode.updated_at.desc(), TruthNode.id.desc()) .limit(limit) ) ).scalars().all() return [ { "entity_type": entity_type, "entity_id": int(row.id), "display_name": str(row.title or f"{entity_type}#{int(row.id)}").strip(), "object_status": str(row.status or "").strip(), "updated_at": row.updated_at.isoformat() if row.updated_at else None, "is_extractable": not _status_is_inactive(row.status), "summary": str((row.content or {}).get("summary") or "").strip() if isinstance(row.content, dict) else "", } for row in rows ] async def list_object_snapshots( self, *, entity_types: Optional[list[str]] = None, limit_per_type: int = 200, ) -> list[dict[str, Any]]: normalized_types = [ _normalized_entity_type(item) for item in list(entity_types or sorted(SEMANTIC_FACT_SUPPORTED_ENTITY_TYPES)) if _normalized_entity_type(item) in SEMANTIC_FACT_SUPPORTED_ENTITY_TYPES ] if not normalized_types: normalized_types = sorted(SEMANTIC_FACT_SUPPORTED_ENTITY_TYPES) limit = max(1, min(int(limit_per_type), 1000)) items: list[dict[str, Any]] = [] for entity_type in normalized_types: if entity_type == "project": items.extend(await self._list_project_snapshots(limit=limit)) elif entity_type == "task": items.extend(await self._list_task_snapshots(limit=limit)) elif entity_type == "signal": items.extend(await self._list_signal_snapshots(limit=limit)) elif entity_type == "document": items.extend(await self._list_document_snapshots(limit=limit)) elif entity_type == "user": items.extend(await self._list_user_snapshots(limit=limit)) elif entity_type in {"requirement", "release", "review"}: items.extend(await self._list_truth_node_snapshots(entity_type=entity_type, limit=limit)) return items async def _load_object_metadata_maps( self, *, items: list[dict[str, Any]], ) -> tuple[dict[tuple[str, int], int], dict[tuple[str, int], list[str]], dict[tuple[str, int], dict[str, Any]], dict[tuple[str, int], dict[str, Any]]]: if not items: return {}, {}, {}, {} keys = {(str(item["entity_type"]), int(item["entity_id"])) for item in items} entity_types = sorted({entity_type for entity_type, _entity_id in keys}) fact_rows = ( await self.db.execute( select(SemanticFact).where( and_( SemanticFact.tenant_id == self.tenant_id, SemanticFact.entity_type.in_(entity_types), SemanticFact.is_active.is_(True), ) ) ) ).scalars().all() fact_counts: dict[tuple[str, int], int] = {} fact_texts: dict[tuple[str, int], list[str]] = {} for row in fact_rows: key = (str(row.entity_type), int(row.entity_id)) if key not in keys: continue fact_counts[key] = int(fact_counts.get(key) or 0) + 1 fact_texts.setdefault(key, []).append( " ".join( part for part in [ str(row.fact_key or "").strip(), str(row.fact_value_text or "").strip(), str(row.fact_value_norm or "").strip(), ] if part ) ) run_rows = ( await self.db.execute( select(SemanticFactExtractionRun) .where( and_( SemanticFactExtractionRun.tenant_id == self.tenant_id, SemanticFactExtractionRun.entity_type.in_(entity_types), ) ) .order_by(SemanticFactExtractionRun.id.desc()) ) ).scalars().all() latest_runs: dict[tuple[str, int], dict[str, Any]] = {} for row in run_rows: key = (str(row.entity_type), int(row.entity_id)) if key not in keys or key in latest_runs: continue latest_runs[key] = { "id": int(row.id), "status": str(row.status or "").strip(), "trigger_type": str(row.trigger_type or "").strip(), "facts_written": int(row.facts_written or 0), "failure_reason": str(row.failure_reason or "").strip() or None, "extract_version": str(row.extract_version or "").strip() or None, "started_at": str(row.started_at or "").strip() or None, "finished_at": str(row.finished_at or "").strip() or None, } event_rows = ( await self.db.execute( select(IngestEvent).where( and_( IngestEvent.tenant_id == self.tenant_id, IngestEvent.channel == SEMANTIC_FACT_EXTRACT_CHANNEL, ) ) ) ).scalars().all() latest_events: dict[tuple[str, int], dict[str, Any]] = {} for row in event_rows: payload = row.payload if isinstance(row.payload, dict) else {} key = ( _normalized_entity_type(payload.get("entity_type") or ""), int(payload.get("entity_id") or 0), ) if key not in keys: continue latest_events[key] = { "id": int(row.id), "status": str(row.status or "").strip(), "retry_count": int(row.retry_count or 0), "error_message": str(row.error_message or "").strip() or None, "processed_at": row.processed_at.isoformat() if row.processed_at else None, "updated_at": row.updated_at.isoformat() if row.updated_at else None, "created_at": row.created_at.isoformat() if row.created_at else None, "payload": payload, } return fact_counts, fact_texts, latest_runs, latest_events @staticmethod def _resolve_last_status(*, event: Optional[dict[str, Any]], run: Optional[dict[str, Any]]) -> str: if event and str(event.get("status") or "").strip(): return str(event.get("status") or "").strip() if run and str(run.get("status") or "").strip(): return str(run.get("status") or "").strip() return "idle" async def list_admin_objects( self, *, entity_types: Optional[list[str]] = None, q: Optional[str] = None, status: Optional[str] = None, limit_per_type: int = 200, ) -> list[dict[str, Any]]: items = await self.list_object_snapshots(entity_types=entity_types, limit_per_type=limit_per_type) fact_counts, fact_texts, latest_runs, latest_events = await self._load_object_metadata_maps(items=items) query_text = str(q or "").strip().lower() status_filter = str(status or "").strip().lower() results: list[dict[str, Any]] = [] for item in items: key = (str(item["entity_type"]), int(item["entity_id"])) last_run = latest_runs.get(key) last_event = latest_events.get(key) last_status = self._resolve_last_status(event=last_event, run=last_run) result = { **item, "facts_count": int(fact_counts.get(key) or 0), "last_extracted_at": ( str((last_run or {}).get("finished_at") or (last_run or {}).get("started_at") or "").strip() or None ), "last_status": last_status, "queue_event": last_event, "last_run": last_run, } if query_text: haystack = " ".join( [ str(item.get("display_name") or ""), str(item.get("summary") or ""), str(item.get("entity_id") or ""), " ".join(fact_texts.get(key) or []), ] ).lower() if query_text not in haystack: continue if status_filter and status_filter not in {"all", last_status.lower()}: continue results.append(result) results.sort( key=lambda row: ( _parse_iso_datetime(row.get("last_extracted_at")) or _parse_iso_datetime(row.get("updated_at")) or datetime(1970, 1, 1, tzinfo=UTC), int(row.get("entity_id") or 0), ), reverse=True, ) return results async def get_admin_detail( self, *, entity_type: str, entity_id: int, ) -> dict[str, Any]: snapshot = await self._load_entity_snapshot(entity_type=entity_type, entity_id=int(entity_id)) if snapshot is None: raise ValueError("semantic_fact_entity_not_found") facts = ( await self.db.execute( select(SemanticFact).where( and_( SemanticFact.tenant_id == self.tenant_id, SemanticFact.entity_type == _normalized_entity_type(entity_type), SemanticFact.entity_id == int(entity_id), SemanticFact.is_active.is_(True), ) ).order_by(SemanticFact.confidence.desc(), SemanticFact.updated_at.desc(), SemanticFact.id.desc()) ) ).scalars().all() runs = ( await self.db.execute( select(SemanticFactExtractionRun).where( and_( SemanticFactExtractionRun.tenant_id == self.tenant_id, SemanticFactExtractionRun.entity_type == _normalized_entity_type(entity_type), SemanticFactExtractionRun.entity_id == int(entity_id), ) ).order_by(SemanticFactExtractionRun.id.desc()).limit(10) ) ).scalars().all() event = ( await self.db.execute( select(IngestEvent).where( and_( IngestEvent.tenant_id == self.tenant_id, IngestEvent.channel == SEMANTIC_FACT_EXTRACT_CHANNEL, IngestEvent.event_key == f"{_normalized_entity_type(entity_type)}:{int(entity_id)}", ) ) ) ).scalar_one_or_none() return { "object": { "entity_type": str(snapshot["entity_type"]), "entity_id": int(snapshot["entity_id"]), "display_name": str(snapshot["display_name"]), "object_status": str(snapshot.get("object_status") or ""), "updated_at": snapshot.get("updated_at"), "is_extractable": bool(snapshot.get("is_extractable")), "source_payload": snapshot.get("source_payload") or {}, }, "facts": [ { "id": int(row.id), "fact_namespace": str(row.fact_namespace or "").strip(), "fact_key": str(row.fact_key or "").strip(), "fact_value_text": str(row.fact_value_text or "").strip(), "fact_value_norm": str(row.fact_value_norm or "").strip(), "value_type": str(row.value_type or "").strip(), "confidence": float(row.confidence or 0.0), "evidence_excerpt": str(row.evidence_excerpt or "").strip() or None, "freshness_at": str(row.freshness_at or "").strip() or None, "updated_at": row.updated_at.isoformat() if row.updated_at else None, } for row in facts ], "recent_runs": [ { "id": int(row.id), "status": str(row.status or "").strip(), "trigger_type": str(row.trigger_type or "").strip(), "facts_written": int(row.facts_written or 0), "failure_reason": str(row.failure_reason or "").strip() or None, "extract_version": str(row.extract_version or "").strip() or None, "started_at": str(row.started_at or "").strip() or None, "finished_at": str(row.finished_at or "").strip() or None, } for row in runs ], "queue_event": ( { "id": int(event.id), "status": str(event.status or "").strip(), "retry_count": int(event.retry_count or 0), "error_message": str(event.error_message or "").strip() or None, "processed_at": event.processed_at.isoformat() if event.processed_at else None, "updated_at": event.updated_at.isoformat() if event.updated_at else None, "payload": event.payload if isinstance(event.payload, dict) else {}, } if event is not None else None ), } async def enqueue_backfill( self, *, entity_types: Optional[list[str]] = None, only_missing: bool = True, limit_per_type: int = 200, ) -> dict[str, Any]: items = await self.list_object_snapshots(entity_types=entity_types, limit_per_type=limit_per_type) by_type: dict[str, int] = {} existing_pairs: set[tuple[str, int]] = set() if only_missing: fact_rows = ( await self.db.execute( select(SemanticFact.entity_type, SemanticFact.entity_id).where( and_( SemanticFact.tenant_id == self.tenant_id, SemanticFact.is_active.is_(True), ) ) ) ).all() existing_pairs = { (str(entity_type), int(entity_id)) for entity_type, entity_id in fact_rows } total = 0 for item in items: key = (str(item["entity_type"]), int(item["entity_id"])) if not bool(item.get("is_extractable")): continue if only_missing and key in existing_pairs: continue await self.enqueue_entity_event( entity_type=str(item["entity_type"]), entity_id=int(item["entity_id"]), operation="extract", trigger_type="backfill", extra_payload={ "source_updated_at": str(item.get("updated_at") or "").strip() or None, }, ) by_type[str(item["entity_type"])] = int(by_type.get(str(item["entity_type"])) or 0) + 1 total += 1 return { "enqueued": total, "only_missing": bool(only_missing), "by_type": by_type, } async def get_admin_overview(self) -> dict[str, Any]: items = await self.list_object_snapshots(limit_per_type=500) active_fact_total = int( ( await self.db.execute( select(func.count(SemanticFact.id)).where( and_( SemanticFact.tenant_id == self.tenant_id, SemanticFact.is_active.is_(True), ) ) ) ).scalar_one() or 0 ) event_rows = ( await self.db.execute( select(IngestEvent).where( and_( IngestEvent.tenant_id == self.tenant_id, IngestEvent.channel == SEMANTIC_FACT_EXTRACT_CHANNEL, ) ).order_by(IngestEvent.updated_at.desc(), IngestEvent.id.desc()) ) ).scalars().all() pending_rows = [row for row in event_rows if str(row.status or "").strip() in {"pending", "retry"}] failed_rows = [row for row in event_rows if str(row.status or "").strip() == "dead_letter"] running_count = int( ( await self.db.execute( select(func.count(SemanticFactExtractionRun.id)).where( and_( SemanticFactExtractionRun.tenant_id == self.tenant_id, SemanticFactExtractionRun.status == "running", ) ) ) ).scalar_one() or 0 ) all_runs = ( await self.db.execute( select(SemanticFactExtractionRun).where( SemanticFactExtractionRun.tenant_id == self.tenant_id ) ) ).scalars().all() cutoff = datetime.now(UTC) - timedelta(hours=24) success_24h = 0 failed_24h = 0 for row in all_runs: pivot = _parse_iso_datetime(row.finished_at or row.started_at) if pivot is None or pivot < cutoff: continue status = str(row.status or "").strip() if status == "completed": success_24h += 1 elif status == "failed": failed_24h += 1 return { "kpis": { "object_total": len(items), "active_fact_total": active_fact_total, "success_24h": success_24h, "failed_24h": failed_24h, }, "queue": { "pending": len(pending_rows), "processing": running_count, "failed": len(failed_rows), "oldest_pending_at": pending_rows[-1].created_at.isoformat() if pending_rows else None, "recent_failures": [ { "event_key": str(row.event_key or ""), "status": str(row.status or "").strip(), "error_message": str(row.error_message or "").strip() or None, "updated_at": row.updated_at.isoformat() if row.updated_at else None, } for row in failed_rows[:5] ], }, "entity_types": sorted(SEMANTIC_FACT_SUPPORTED_ENTITY_TYPES), } __all__ = ["SemanticFactAdminMixin"]