1771 lines
71 KiB
Python
1771 lines
71 KiB
Python
"""Project Knowledge Graph service.
|
||
|
||
第一阶段目标：
|
||
1. 构建统一实体/关系/证据写入能力
|
||
2. 支持结构化数据（project/task/signal/user）同步
|
||
3. 支持代码目录扫描快照
|
||
4. 提供图谱概览统计
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import hashlib
|
||
from datetime import datetime, timedelta, timezone
|
||
from pathlib import Path
|
||
from typing import Any, Optional
|
||
|
||
from sqlalchemy import and_, func, select
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from ..models import (
|
||
IngestEvent,
|
||
IngestOffset,
|
||
KGEvidence,
|
||
KGEntity,
|
||
KGRelation,
|
||
KGSyncRun,
|
||
Project,
|
||
FeishuItem,
|
||
Signal,
|
||
SymbolDefinition,
|
||
SymbolReference,
|
||
Task,
|
||
User,
|
||
)
|
||
from ..utils.timezone import now_cn_naive
|
||
from .context_cache import context_cache_service
|
||
from .symbol_index_service import SymbolIndexService
|
||
|
||
|
||
def _utc_now_naive() -> datetime:
|
||
return now_cn_naive()
|
||
|
||
|
||
class ProjectKnowledgeGraphService:
|
||
def __init__(self, db: AsyncSession, *, tenant_id: int) -> None:
|
||
self.db = db
|
||
self.tenant_id = int(tenant_id)
|
||
|
||
async def upsert_entity(
|
||
self,
|
||
*,
|
||
entity_type: str,
|
||
entity_key: str,
|
||
display_name: str,
|
||
payload: Optional[dict[str, Any]] = None,
|
||
source_type: str = "system",
|
||
source_ref: Optional[str] = None,
|
||
) -> KGEntity:
|
||
row = (
|
||
await self.db.execute(
|
||
select(KGEntity).where(
|
||
and_(
|
||
KGEntity.tenant_id == self.tenant_id,
|
||
KGEntity.entity_type == str(entity_type).strip()[:40],
|
||
KGEntity.entity_key == str(entity_key).strip()[:255],
|
||
)
|
||
)
|
||
)
|
||
).scalar_one_or_none()
|
||
|
||
if row is None:
|
||
row = KGEntity(
|
||
tenant_id=self.tenant_id,
|
||
entity_type=str(entity_type).strip()[:40],
|
||
entity_key=str(entity_key).strip()[:255],
|
||
display_name=str(display_name or entity_key).strip()[:255],
|
||
payload=payload or {},
|
||
source_type=str(source_type or "system").strip()[:32],
|
||
source_ref=(str(source_ref).strip()[:255] if source_ref else None),
|
||
last_seen_at=_utc_now_naive(),
|
||
)
|
||
self.db.add(row)
|
||
await self.db.flush()
|
||
return row
|
||
|
||
row.display_name = str(display_name or row.display_name).strip()[:255]
|
||
row.payload = payload or {}
|
||
row.source_type = str(source_type or row.source_type or "system").strip()[:32]
|
||
row.source_ref = str(source_ref).strip()[:255] if source_ref else row.source_ref
|
||
row.last_seen_at = _utc_now_naive()
|
||
await self.db.flush()
|
||
return row
|
||
|
||
async def upsert_relation(
|
||
self,
|
||
*,
|
||
relation_type: str,
|
||
subject_entity_id: int,
|
||
object_entity_id: int,
|
||
weight: float = 1.0,
|
||
payload: Optional[dict[str, Any]] = None,
|
||
source_type: str = "system",
|
||
source_ref: Optional[str] = None,
|
||
) -> KGRelation:
|
||
row = (
|
||
await self.db.execute(
|
||
select(KGRelation).where(
|
||
and_(
|
||
KGRelation.tenant_id == self.tenant_id,
|
||
KGRelation.relation_type == str(relation_type).strip()[:40],
|
||
KGRelation.subject_entity_id == int(subject_entity_id),
|
||
KGRelation.object_entity_id == int(object_entity_id),
|
||
)
|
||
)
|
||
)
|
||
).scalar_one_or_none()
|
||
|
||
if row is None:
|
||
row = KGRelation(
|
||
tenant_id=self.tenant_id,
|
||
relation_type=str(relation_type).strip()[:40],
|
||
subject_entity_id=int(subject_entity_id),
|
||
object_entity_id=int(object_entity_id),
|
||
weight=float(weight),
|
||
payload=payload or {},
|
||
source_type=str(source_type or "system").strip()[:32],
|
||
source_ref=(str(source_ref).strip()[:255] if source_ref else None),
|
||
)
|
||
self.db.add(row)
|
||
await self.db.flush()
|
||
return row
|
||
|
||
row.weight = float(weight)
|
||
row.payload = payload or {}
|
||
row.source_type = str(source_type or row.source_type or "system").strip()[:32]
|
||
row.source_ref = str(source_ref).strip()[:255] if source_ref else row.source_ref
|
||
await self.db.flush()
|
||
return row
|
||
|
||
async def add_evidence(
|
||
self,
|
||
*,
|
||
evidence_type: str,
|
||
content_excerpt: str,
|
||
entity_id: Optional[int] = None,
|
||
relation_id: Optional[int] = None,
|
||
source_type: str = "system",
|
||
source_ref: Optional[str] = None,
|
||
metadata: Optional[dict[str, Any]] = None,
|
||
) -> KGEvidence:
|
||
clean = str(content_excerpt or "").strip()
|
||
digest = hashlib.md5(clean.encode("utf-8")).hexdigest()
|
||
|
||
row = KGEvidence(
|
||
tenant_id=self.tenant_id,
|
||
entity_id=int(entity_id) if entity_id is not None else None,
|
||
relation_id=int(relation_id) if relation_id is not None else None,
|
||
evidence_type=str(evidence_type or "snapshot").strip()[:40],
|
||
content_hash=digest,
|
||
content_excerpt=clean[:4000] if clean else "(empty)",
|
||
source_type=str(source_type or "system").strip()[:32],
|
||
source_ref=(str(source_ref).strip()[:255] if source_ref else None),
|
||
meta_payload=metadata or {},
|
||
occurred_at=_utc_now_naive(),
|
||
)
|
||
self.db.add(row)
|
||
await self.db.flush()
|
||
return row
|
||
|
||
async def enqueue_ingest_event(
|
||
self,
|
||
*,
|
||
channel: str,
|
||
event_key: str,
|
||
payload: Optional[dict[str, Any]] = None,
|
||
status: str = "pending",
|
||
) -> IngestEvent:
|
||
row = (
|
||
await self.db.execute(
|
||
select(IngestEvent).where(
|
||
and_(
|
||
IngestEvent.tenant_id == self.tenant_id,
|
||
IngestEvent.channel == str(channel).strip()[:32],
|
||
IngestEvent.event_key == str(event_key).strip()[:128],
|
||
)
|
||
)
|
||
)
|
||
).scalar_one_or_none()
|
||
if row is None:
|
||
row = IngestEvent(
|
||
tenant_id=self.tenant_id,
|
||
channel=str(channel).strip()[:32],
|
||
event_key=str(event_key).strip()[:128],
|
||
status=str(status or "pending").strip()[:20],
|
||
payload=payload or {},
|
||
)
|
||
self.db.add(row)
|
||
else:
|
||
row.status = str(status or row.status or "pending").strip()[:20]
|
||
row.payload = payload or row.payload or {}
|
||
row.error_message = None
|
||
await self.db.flush()
|
||
return row
|
||
|
||
async def set_ingest_offset(
|
||
self,
|
||
*,
|
||
channel: str,
|
||
offset_key: str,
|
||
offset_value: str,
|
||
payload: Optional[dict[str, Any]] = None,
|
||
) -> IngestOffset:
|
||
row = (
|
||
await self.db.execute(
|
||
select(IngestOffset).where(
|
||
and_(
|
||
IngestOffset.tenant_id == self.tenant_id,
|
||
IngestOffset.channel == str(channel).strip()[:32],
|
||
IngestOffset.offset_key == str(offset_key).strip()[:128],
|
||
)
|
||
)
|
||
)
|
||
).scalar_one_or_none()
|
||
|
||
if row is None:
|
||
row = IngestOffset(
|
||
tenant_id=self.tenant_id,
|
||
channel=str(channel).strip()[:32],
|
||
offset_key=str(offset_key).strip()[:128],
|
||
offset_value=str(offset_value or "").strip()[:255],
|
||
payload=payload or {},
|
||
)
|
||
self.db.add(row)
|
||
else:
|
||
row.offset_value = str(offset_value or row.offset_value).strip()[:255]
|
||
row.payload = payload or row.payload or {}
|
||
await self.db.flush()
|
||
return row
|
||
|
||
async def get_ingest_offset(self, *, channel: str, offset_key: str) -> Optional[str]:
|
||
row = (
|
||
await self.db.execute(
|
||
select(IngestOffset).where(
|
||
and_(
|
||
IngestOffset.tenant_id == self.tenant_id,
|
||
IngestOffset.channel == str(channel).strip()[:32],
|
||
IngestOffset.offset_key == str(offset_key).strip()[:128],
|
||
)
|
||
)
|
||
)
|
||
).scalar_one_or_none()
|
||
if row is None:
|
||
return None
|
||
return str(row.offset_value or "").strip() or None
|
||
|
||
@staticmethod
|
||
def _parse_iso_dt(value: Optional[str]) -> Optional[datetime]:
|
||
raw = str(value or "").strip()
|
||
if not raw:
|
||
return None
|
||
try:
|
||
dt = datetime.fromisoformat(raw)
|
||
return dt.replace(tzinfo=None) if dt.tzinfo else dt
|
||
except Exception:
|
||
return None
|
||
|
||
async def _open_sync_run(
|
||
self,
|
||
*,
|
||
run_type: str,
|
||
trigger_source: str,
|
||
input_payload: Optional[dict[str, Any]] = None,
|
||
) -> KGSyncRun:
|
||
now = _utc_now_naive()
|
||
row = KGSyncRun(
|
||
tenant_id=self.tenant_id,
|
||
run_type=str(run_type or "incremental").strip()[:40],
|
||
trigger_source=str(trigger_source or "system").strip()[:40],
|
||
status="running",
|
||
started_at=now,
|
||
input_payload=input_payload or {},
|
||
result_payload={},
|
||
)
|
||
self.db.add(row)
|
||
await self.db.flush()
|
||
await self.db.commit()
|
||
await self.db.refresh(row)
|
||
return row
|
||
|
||
async def _close_sync_run(
|
||
self,
|
||
*,
|
||
row: Optional[KGSyncRun],
|
||
status: str,
|
||
result_payload: Optional[dict[str, Any]] = None,
|
||
error_message: Optional[str] = None,
|
||
started_at_hint: Optional[datetime] = None,
|
||
) -> None:
|
||
if row is None:
|
||
return
|
||
now = _utc_now_naive()
|
||
started = started_at_hint or row.started_at or now
|
||
row.status = str(status or "unknown").strip()[:20]
|
||
row.ended_at = now
|
||
row.duration_ms = max(int((now - started).total_seconds() * 1000), 0)
|
||
row.result_payload = result_payload or {}
|
||
row.error_message = (str(error_message).strip()[:2000] if error_message else None)
|
||
await self.db.commit()
|
||
|
||
async def list_sync_runs(
|
||
self,
|
||
*,
|
||
limit: int = 50,
|
||
run_type: Optional[str] = None,
|
||
status: Optional[str] = None,
|
||
) -> dict[str, Any]:
|
||
max_rows = max(1, min(int(limit), 200))
|
||
stmt = select(KGSyncRun).where(KGSyncRun.tenant_id == self.tenant_id)
|
||
if run_type:
|
||
stmt = stmt.where(KGSyncRun.run_type == str(run_type).strip()[:40])
|
||
if status:
|
||
stmt = stmt.where(KGSyncRun.status == str(status).strip()[:20])
|
||
rows = (
|
||
await self.db.execute(stmt.order_by(KGSyncRun.started_at.desc(), KGSyncRun.id.desc()).limit(max_rows))
|
||
).scalars().all()
|
||
|
||
items = []
|
||
for row in rows:
|
||
items.append(
|
||
{
|
||
"id": int(row.id),
|
||
"run_type": str(row.run_type or ""),
|
||
"trigger_source": str(row.trigger_source or ""),
|
||
"status": str(row.status or ""),
|
||
"started_at": row.started_at.isoformat() if row.started_at else None,
|
||
"ended_at": row.ended_at.isoformat() if row.ended_at else None,
|
||
"duration_ms": int(row.duration_ms or 0),
|
||
"input_payload": row.input_payload or {},
|
||
"result_payload": row.result_payload or {},
|
||
"error_message": str(row.error_message or ""),
|
||
}
|
||
)
|
||
return {
|
||
"tenant_id": self.tenant_id,
|
||
"count": len(items),
|
||
"items": items,
|
||
}
|
||
|
||
async def get_sync_slo(self, *, lookback_hours: int = 24) -> dict[str, Any]:
|
||
hours = max(1, min(int(lookback_hours), 24 * 30))
|
||
now = _utc_now_naive()
|
||
since = now - timedelta(hours=hours)
|
||
|
||
runs = (
|
||
await self.db.execute(
|
||
select(KGSyncRun).where(
|
||
and_(
|
||
KGSyncRun.tenant_id == self.tenant_id,
|
||
KGSyncRun.started_at >= since,
|
||
KGSyncRun.run_type == "incremental",
|
||
)
|
||
)
|
||
)
|
||
).scalars().all()
|
||
|
||
total = len(runs)
|
||
success = sum(1 for x in runs if str(x.status or "") == "success")
|
||
partial = sum(1 for x in runs if str(x.status or "") == "partial")
|
||
failed = sum(1 for x in runs if str(x.status or "") == "failed")
|
||
avg_duration_ms = (
|
||
int(sum(int(x.duration_ms or 0) for x in runs) / total)
|
||
if total > 0
|
||
else 0
|
||
)
|
||
|
||
latest_success = (
|
||
await self.db.execute(
|
||
select(KGSyncRun)
|
||
.where(
|
||
and_(
|
||
KGSyncRun.tenant_id == self.tenant_id,
|
||
KGSyncRun.run_type == "incremental",
|
||
KGSyncRun.status == "success",
|
||
)
|
||
)
|
||
.order_by(KGSyncRun.started_at.desc(), KGSyncRun.id.desc())
|
||
.limit(1)
|
||
)
|
||
).scalar_one_or_none()
|
||
|
||
dead_letter = (
|
||
await self.db.execute(
|
||
select(func.count(IngestEvent.id)).where(
|
||
and_(
|
||
IngestEvent.tenant_id == self.tenant_id,
|
||
IngestEvent.status == "dead_letter",
|
||
)
|
||
)
|
||
)
|
||
).scalar_one()
|
||
pending_events = (
|
||
await self.db.execute(
|
||
select(func.count(IngestEvent.id)).where(
|
||
and_(
|
||
IngestEvent.tenant_id == self.tenant_id,
|
||
IngestEvent.status.in_(["pending", "retry"]),
|
||
)
|
||
)
|
||
)
|
||
).scalar_one()
|
||
|
||
async def _freshness_minutes(channel: str, offset_key: str) -> Optional[int]:
|
||
raw = await self.get_ingest_offset(channel=channel, offset_key=offset_key)
|
||
dt = self._parse_iso_dt(raw)
|
||
if dt is None:
|
||
return None
|
||
return max(int((now - dt).total_seconds() // 60), 0)
|
||
|
||
return {
|
||
"tenant_id": self.tenant_id,
|
||
"lookback_hours": hours,
|
||
"window_started_at": since.isoformat(),
|
||
"window_ended_at": now.isoformat(),
|
||
"runs_total": total,
|
||
"runs_success": success,
|
||
"runs_partial": partial,
|
||
"runs_failed": failed,
|
||
"success_rate": round((success / total), 4) if total > 0 else 0.0,
|
||
"avg_duration_ms": avg_duration_ms,
|
||
"latest_success_at": latest_success.started_at.isoformat() if latest_success and latest_success.started_at else None,
|
||
"dead_letter_events": int(dead_letter or 0),
|
||
"pending_events": int(pending_events or 0),
|
||
"freshness_minutes": {
|
||
"db_structured": await _freshness_minutes("db_delta", "last_structured_sync_at"),
|
||
"feishu_delta": await _freshness_minutes("feishu_delta", "last_feishu_item_updated_at"),
|
||
"code_scan": await _freshness_minutes("code_scan", "last_code_scan_at"),
|
||
"symbol_scan": await _freshness_minutes("symbol_scan", "last_symbol_scan_at"),
|
||
"feature_catalog": await _freshness_minutes("feature_catalog", "last_feature_catalog_sync_at"),
|
||
},
|
||
}
|
||
|
||
async def sync_structured_delta(self, *, limit: int = 1000) -> dict[str, Any]:
|
||
max_rows = max(1, min(int(limit), 5000))
|
||
last_raw = await self.get_ingest_offset(channel="db_delta", offset_key="last_structured_sync_at")
|
||
last_dt = self._parse_iso_dt(last_raw) or (datetime(1970, 1, 1))
|
||
counters = {
|
||
"projects": 0,
|
||
"tasks": 0,
|
||
"signals": 0,
|
||
"users": 0,
|
||
"relations": 0,
|
||
"evidences": 0,
|
||
"from_offset": last_dt.isoformat(),
|
||
}
|
||
newest_seen = last_dt
|
||
|
||
project_rows = (
|
||
await self.db.execute(
|
||
select(Project)
|
||
.where(
|
||
and_(
|
||
Project.tenant_id == self.tenant_id,
|
||
Project.updated_at >= last_dt,
|
||
)
|
||
)
|
||
.order_by(Project.updated_at.asc(), Project.id.asc())
|
||
.limit(max_rows)
|
||
)
|
||
).scalars().all()
|
||
project_map: dict[int, int] = {}
|
||
for p in project_rows:
|
||
entity = await self.upsert_entity(
|
||
entity_type="project",
|
||
entity_key=f"project:{int(p.id)}",
|
||
display_name=str(p.title or p.name or f"Project#{int(p.id)}"),
|
||
payload={
|
||
"project_id": int(p.id),
|
||
"status": str(p.status or "unknown"),
|
||
"owner_id": int(p.owner_id) if p.owner_id is not None else None,
|
||
},
|
||
source_type="db",
|
||
source_ref=f"projects:{int(p.id)}",
|
||
)
|
||
project_map[int(p.id)] = int(entity.id)
|
||
counters["projects"] += 1
|
||
if p.updated_at and p.updated_at > newest_seen:
|
||
newest_seen = p.updated_at
|
||
|
||
user_rows = (
|
||
await self.db.execute(
|
||
select(User)
|
||
.where(
|
||
and_(
|
||
User.tenant_id == self.tenant_id,
|
||
User.updated_at >= last_dt,
|
||
)
|
||
)
|
||
.order_by(User.updated_at.asc(), User.id.asc())
|
||
.limit(max_rows)
|
||
)
|
||
).scalars().all()
|
||
user_map: dict[int, int] = {}
|
||
for u in user_rows:
|
||
entity = await self.upsert_entity(
|
||
entity_type="user",
|
||
entity_key=f"user:{int(u.id)}",
|
||
display_name=str(u.name or f"User#{int(u.id)}"),
|
||
payload={
|
||
"user_id": int(u.id),
|
||
"role": str(u.role or "unknown"),
|
||
"is_active": bool(u.is_active),
|
||
},
|
||
source_type="db",
|
||
source_ref=f"users:{int(u.id)}",
|
||
)
|
||
user_map[int(u.id)] = int(entity.id)
|
||
counters["users"] += 1
|
||
if u.updated_at and u.updated_at > newest_seen:
|
||
newest_seen = u.updated_at
|
||
|
||
task_rows = (
|
||
await self.db.execute(
|
||
select(Task)
|
||
.where(
|
||
and_(
|
||
Task.tenant_id == self.tenant_id,
|
||
Task.updated_at >= last_dt,
|
||
)
|
||
)
|
||
.order_by(Task.updated_at.asc(), Task.id.asc())
|
||
.limit(max_rows)
|
||
)
|
||
).scalars().all()
|
||
for t in task_rows:
|
||
task_entity = await self.upsert_entity(
|
||
entity_type="task",
|
||
entity_key=f"task:{int(t.id)}",
|
||
display_name=str(t.title or f"Task#{int(t.id)}"),
|
||
payload={
|
||
"task_id": int(t.id),
|
||
"project_id": int(t.project_id) if t.project_id is not None else None,
|
||
"assignee_id": int(t.assignee_id) if t.assignee_id is not None else None,
|
||
"status": str(t.status or "unknown"),
|
||
},
|
||
source_type="db",
|
||
source_ref=f"tasks:{int(t.id)}",
|
||
)
|
||
counters["tasks"] += 1
|
||
if t.updated_at and t.updated_at > newest_seen:
|
||
newest_seen = t.updated_at
|
||
|
||
if t.project_id is not None:
|
||
target_pid = int(t.project_id)
|
||
if target_pid not in project_map:
|
||
p = (
|
||
await self.db.execute(
|
||
select(Project).where(
|
||
and_(
|
||
Project.tenant_id == self.tenant_id,
|
||
Project.id == target_pid,
|
||
)
|
||
)
|
||
)
|
||
).scalar_one_or_none()
|
||
if p is not None:
|
||
p_entity = await self.upsert_entity(
|
||
entity_type="project",
|
||
entity_key=f"project:{int(p.id)}",
|
||
display_name=str(p.title or p.name or f"Project#{int(p.id)}"),
|
||
payload={
|
||
"project_id": int(p.id),
|
||
"status": str(p.status or "unknown"),
|
||
"owner_id": int(p.owner_id) if p.owner_id is not None else None,
|
||
},
|
||
source_type="db",
|
||
source_ref=f"projects:{int(p.id)}",
|
||
)
|
||
project_map[int(p.id)] = int(p_entity.id)
|
||
if target_pid in project_map:
|
||
rel = await self.upsert_relation(
|
||
relation_type="belongs_to",
|
||
subject_entity_id=int(task_entity.id),
|
||
object_entity_id=int(project_map[target_pid]),
|
||
source_type="db",
|
||
source_ref=f"tasks:{int(t.id)}",
|
||
)
|
||
counters["relations"] += 1
|
||
await self.add_evidence(
|
||
evidence_type="db_delta",
|
||
content_excerpt=f"task#{int(t.id)} belongs_to project#{target_pid}",
|
||
relation_id=int(rel.id),
|
||
source_type="db",
|
||
source_ref=f"tasks:{int(t.id)}",
|
||
metadata={"table": "tasks"},
|
||
)
|
||
counters["evidences"] += 1
|
||
|
||
if t.assignee_id is not None:
|
||
uid = int(t.assignee_id)
|
||
if uid not in user_map:
|
||
u = (
|
||
await self.db.execute(
|
||
select(User).where(and_(User.tenant_id == self.tenant_id, User.id == uid))
|
||
)
|
||
).scalar_one_or_none()
|
||
if u is not None:
|
||
u_entity = await self.upsert_entity(
|
||
entity_type="user",
|
||
entity_key=f"user:{uid}",
|
||
display_name=str(u.name or f"User#{uid}"),
|
||
payload={
|
||
"user_id": uid,
|
||
"role": str(u.role or "unknown"),
|
||
"is_active": bool(u.is_active),
|
||
},
|
||
source_type="db",
|
||
source_ref=f"users:{uid}",
|
||
)
|
||
user_map[uid] = int(u_entity.id)
|
||
if uid in user_map:
|
||
rel = await self.upsert_relation(
|
||
relation_type="assigned_to",
|
||
subject_entity_id=int(task_entity.id),
|
||
object_entity_id=int(user_map[uid]),
|
||
source_type="db",
|
||
source_ref=f"tasks:{int(t.id)}",
|
||
)
|
||
counters["relations"] += 1
|
||
await self.add_evidence(
|
||
evidence_type="db_delta",
|
||
content_excerpt=f"task#{int(t.id)} assigned_to user#{uid}",
|
||
relation_id=int(rel.id),
|
||
source_type="db",
|
||
source_ref=f"tasks:{int(t.id)}",
|
||
metadata={"table": "tasks"},
|
||
)
|
||
counters["evidences"] += 1
|
||
|
||
signal_rows = (
|
||
await self.db.execute(
|
||
select(Signal)
|
||
.where(
|
||
and_(
|
||
Signal.tenant_id == self.tenant_id,
|
||
Signal.updated_at >= last_dt,
|
||
)
|
||
)
|
||
.order_by(Signal.updated_at.asc(), Signal.id.asc())
|
||
.limit(max_rows)
|
||
)
|
||
).scalars().all()
|
||
for s in signal_rows:
|
||
signal_entity = await self.upsert_entity(
|
||
entity_type="signal",
|
||
entity_key=f"signal:{int(s.id)}",
|
||
display_name=str(s.title or s.content or f"Signal#{int(s.id)}")[:255],
|
||
payload={
|
||
"signal_id": int(s.id),
|
||
"priority": str(s.priority or ""),
|
||
"status": str(s.status or ""),
|
||
"matched_project_id": (
|
||
int(s.matched_project_id) if getattr(s, "matched_project_id", None) is not None else None
|
||
),
|
||
},
|
||
source_type="db",
|
||
source_ref=f"signals:{int(s.id)}",
|
||
)
|
||
counters["signals"] += 1
|
||
if s.updated_at and s.updated_at > newest_seen:
|
||
newest_seen = s.updated_at
|
||
|
||
pid = int(s.matched_project_id) if getattr(s, "matched_project_id", None) is not None else None
|
||
if pid is not None:
|
||
if pid not in project_map:
|
||
p = (
|
||
await self.db.execute(
|
||
select(Project).where(and_(Project.tenant_id == self.tenant_id, Project.id == pid))
|
||
)
|
||
).scalar_one_or_none()
|
||
if p is not None:
|
||
p_entity = await self.upsert_entity(
|
||
entity_type="project",
|
||
entity_key=f"project:{pid}",
|
||
display_name=str(p.title or p.name or f"Project#{pid}"),
|
||
payload={
|
||
"project_id": pid,
|
||
"status": str(p.status or "unknown"),
|
||
"owner_id": int(p.owner_id) if p.owner_id is not None else None,
|
||
},
|
||
source_type="db",
|
||
source_ref=f"projects:{pid}",
|
||
)
|
||
project_map[pid] = int(p_entity.id)
|
||
if pid in project_map:
|
||
rel = await self.upsert_relation(
|
||
relation_type="matches_project",
|
||
subject_entity_id=int(signal_entity.id),
|
||
object_entity_id=int(project_map[pid]),
|
||
source_type="db",
|
||
source_ref=f"signals:{int(s.id)}",
|
||
)
|
||
counters["relations"] += 1
|
||
await self.add_evidence(
|
||
evidence_type="db_delta",
|
||
content_excerpt=f"signal#{int(s.id)} matches project#{pid}",
|
||
relation_id=int(rel.id),
|
||
source_type="db",
|
||
source_ref=f"signals:{int(s.id)}",
|
||
metadata={"table": "signals"},
|
||
)
|
||
counters["evidences"] += 1
|
||
|
||
await self.enqueue_ingest_event(
|
||
channel="db_delta",
|
||
event_key=f"db_delta:{_utc_now_naive().strftime('%Y%m%d%H%M%S')}",
|
||
payload=counters,
|
||
status="done",
|
||
)
|
||
await self.set_ingest_offset(
|
||
channel="db_delta",
|
||
offset_key="last_structured_sync_at",
|
||
offset_value=newest_seen.isoformat(),
|
||
payload={"counters": counters},
|
||
)
|
||
await self.db.commit()
|
||
counters["to_offset"] = newest_seen.isoformat()
|
||
return counters
|
||
|
||
async def enqueue_feishu_delta_events(self, *, limit: int = 500) -> dict[str, Any]:
|
||
max_rows = max(1, min(int(limit), 2000))
|
||
last_raw = await self.get_ingest_offset(channel="feishu_delta", offset_key="last_feishu_item_updated_at")
|
||
last_dt = self._parse_iso_dt(last_raw) or (datetime(1970, 1, 1))
|
||
|
||
rows = (
|
||
await self.db.execute(
|
||
select(FeishuItem)
|
||
.where(
|
||
and_(
|
||
FeishuItem.tenant_id == self.tenant_id,
|
||
FeishuItem.updated_at >= last_dt,
|
||
)
|
||
)
|
||
.order_by(FeishuItem.updated_at.asc(), FeishuItem.id.asc())
|
||
.limit(max_rows)
|
||
)
|
||
).scalars().all()
|
||
enqueued = 0
|
||
newest_seen = last_dt
|
||
for item in rows:
|
||
event_key = str(item.feishu_message_id or f"feishu_item:{int(item.id)}")
|
||
await self.enqueue_ingest_event(
|
||
channel="feishu_delta",
|
||
event_key=event_key[:128],
|
||
payload={
|
||
"feishu_item_id": int(item.id),
|
||
"matched_project_id": int(item.matched_project_id) if item.matched_project_id is not None else None,
|
||
"level": str(item.level or ""),
|
||
"route_status": str(item.route_status or ""),
|
||
},
|
||
status="pending",
|
||
)
|
||
enqueued += 1
|
||
if item.updated_at and item.updated_at > newest_seen:
|
||
newest_seen = item.updated_at
|
||
|
||
await self.set_ingest_offset(
|
||
channel="feishu_delta",
|
||
offset_key="last_feishu_item_updated_at",
|
||
offset_value=newest_seen.isoformat(),
|
||
payload={"batch_size": len(rows)},
|
||
)
|
||
await self.db.commit()
|
||
return {
|
||
"enqueued": enqueued,
|
||
"from_offset": last_dt.isoformat(),
|
||
"to_offset": newest_seen.isoformat(),
|
||
"candidate_count": len(rows),
|
||
}
|
||
|
||
async def process_pending_events(self, *, max_batch: int = 200, max_retry: int = 3) -> dict[str, Any]:
|
||
batch = max(1, min(int(max_batch), 1000))
|
||
retries = max(0, min(int(max_retry), 10))
|
||
|
||
rows = (
|
||
await self.db.execute(
|
||
select(IngestEvent)
|
||
.where(
|
||
and_(
|
||
IngestEvent.tenant_id == self.tenant_id,
|
||
IngestEvent.status.in_(["pending", "retry"]),
|
||
)
|
||
)
|
||
.order_by(IngestEvent.created_at.asc(), IngestEvent.id.asc())
|
||
.limit(batch)
|
||
)
|
||
).scalars().all()
|
||
|
||
stats = {"processed": 0, "done": 0, "retry": 0, "dead_letter": 0}
|
||
for ev in rows:
|
||
stats["processed"] += 1
|
||
try:
|
||
if str(ev.channel) == "feishu_delta":
|
||
await self._process_feishu_event(ev)
|
||
ev.status = "done"
|
||
ev.error_message = None
|
||
ev.processed_at = _utc_now_naive()
|
||
stats["done"] += 1
|
||
except Exception as exc:
|
||
ev.retry_count = int(ev.retry_count or 0) + 1
|
||
ev.error_message = str(exc)[:1000]
|
||
if ev.retry_count > retries:
|
||
ev.status = "dead_letter"
|
||
ev.processed_at = _utc_now_naive()
|
||
stats["dead_letter"] += 1
|
||
else:
|
||
ev.status = "retry"
|
||
stats["retry"] += 1
|
||
|
||
await self.db.commit()
|
||
return stats
|
||
|
||
async def _process_feishu_event(self, event: IngestEvent) -> None:
|
||
payload = event.payload if isinstance(event.payload, dict) else {}
|
||
item_id = payload.get("feishu_item_id")
|
||
if item_id is None:
|
||
return
|
||
item = (
|
||
await self.db.execute(
|
||
select(FeishuItem).where(
|
||
and_(
|
||
FeishuItem.tenant_id == self.tenant_id,
|
||
FeishuItem.id == int(item_id),
|
||
)
|
||
)
|
||
)
|
||
).scalar_one_or_none()
|
||
if item is None:
|
||
return
|
||
|
||
signal_entity = await self.upsert_entity(
|
||
entity_type="signal",
|
||
entity_key=f"feishu_item:{int(item.id)}",
|
||
display_name=str(item.raw_message or f"FeishuItem#{int(item.id)}")[:255],
|
||
payload={
|
||
"feishu_item_id": int(item.id),
|
||
"level": str(item.level or ""),
|
||
"route_status": str(item.route_status or ""),
|
||
"matched_project_id": int(item.matched_project_id) if item.matched_project_id is not None else None,
|
||
},
|
||
source_type="feishu",
|
||
source_ref=str(item.feishu_message_id or f"item:{int(item.id)}")[:255],
|
||
)
|
||
await self.add_evidence(
|
||
evidence_type="feishu_delta",
|
||
content_excerpt=str(item.raw_message or "")[:1000] or f"feishu_item#{int(item.id)}",
|
||
entity_id=int(signal_entity.id),
|
||
source_type="feishu",
|
||
source_ref=str(item.feishu_message_id or f"item:{int(item.id)}")[:255],
|
||
metadata={"feishu_item_id": int(item.id)},
|
||
)
|
||
|
||
if item.matched_project_id is not None:
|
||
project_entity = await self.upsert_entity(
|
||
entity_type="project",
|
||
entity_key=f"project:{int(item.matched_project_id)}",
|
||
display_name=f"Project#{int(item.matched_project_id)}",
|
||
payload={"project_id": int(item.matched_project_id)},
|
||
source_type="db",
|
||
source_ref=f"projects:{int(item.matched_project_id)}",
|
||
)
|
||
rel = await self.upsert_relation(
|
||
relation_type="matches_project",
|
||
subject_entity_id=int(signal_entity.id),
|
||
object_entity_id=int(project_entity.id),
|
||
source_type="feishu",
|
||
source_ref=str(item.feishu_message_id or f"item:{int(item.id)}")[:255],
|
||
)
|
||
await self.add_evidence(
|
||
evidence_type="feishu_delta",
|
||
content_excerpt=f"feishu_item#{int(item.id)} matches project#{int(item.matched_project_id)}",
|
||
relation_id=int(rel.id),
|
||
source_type="feishu",
|
||
source_ref=str(item.feishu_message_id or f"item:{int(item.id)}")[:255],
|
||
metadata={"feishu_item_id": int(item.id)},
|
||
)
|
||
|
||
async def sync_codebase_if_due(
|
||
self,
|
||
*,
|
||
base_dir: Optional[str] = None,
|
||
interval_hours: int = 6,
|
||
) -> dict[str, Any]:
|
||
hours = max(1, int(interval_hours))
|
||
offset_raw = await self.get_ingest_offset(channel="code_scan", offset_key="last_code_scan_at")
|
||
last_scan_at = self._parse_iso_dt(offset_raw)
|
||
now = _utc_now_naive()
|
||
if last_scan_at is not None and now - last_scan_at < timedelta(hours=hours):
|
||
return {
|
||
"executed": False,
|
||
"reason": "not_due",
|
||
"last_scan_at": last_scan_at.isoformat(),
|
||
"next_due_at": (last_scan_at + timedelta(hours=hours)).isoformat(),
|
||
}
|
||
payload = await self.sync_codebase_snapshot(base_dir=base_dir)
|
||
return {
|
||
"executed": True,
|
||
"reason": "due",
|
||
"payload": payload,
|
||
}
|
||
|
||
async def sync_symbol_index_if_due(
|
||
self,
|
||
*,
|
||
base_dir: Optional[str] = None,
|
||
interval_hours: int = 6,
|
||
max_files: int = 500,
|
||
) -> dict[str, Any]:
|
||
hours = max(1, int(interval_hours))
|
||
offset_raw = await self.get_ingest_offset(channel="symbol_scan", offset_key="last_symbol_scan_at")
|
||
last_scan_at = self._parse_iso_dt(offset_raw)
|
||
now = _utc_now_naive()
|
||
if last_scan_at is not None and now - last_scan_at < timedelta(hours=hours):
|
||
return {
|
||
"executed": False,
|
||
"reason": "not_due",
|
||
"last_scan_at": last_scan_at.isoformat(),
|
||
"next_due_at": (last_scan_at + timedelta(hours=hours)).isoformat(),
|
||
}
|
||
|
||
symbol_service = SymbolIndexService(self.db, tenant_id=self.tenant_id)
|
||
payload = await symbol_service.sync_codebase_symbols(
|
||
base_dir=base_dir,
|
||
max_files=max_files,
|
||
force_reindex=False,
|
||
)
|
||
symbol_graph = await self._sync_symbol_entities(limit=800)
|
||
payload["kg_symbol_sync"] = symbol_graph
|
||
await self.enqueue_ingest_event(
|
||
channel="symbol_scan",
|
||
event_key=f"symbol_scan:{_utc_now_naive().strftime('%Y%m%d%H%M%S')}",
|
||
payload=payload,
|
||
status="done",
|
||
)
|
||
await self.set_ingest_offset(
|
||
channel="symbol_scan",
|
||
offset_key="last_symbol_scan_at",
|
||
offset_value=_utc_now_naive().isoformat(),
|
||
payload=payload,
|
||
)
|
||
await self.db.commit()
|
||
return {
|
||
"executed": True,
|
||
"reason": "due",
|
||
"payload": payload,
|
||
}
|
||
|
||
async def _sync_symbol_entities(self, *, limit: int = 800) -> dict[str, Any]:
|
||
max_rows = max(10, min(int(limit), 3000))
|
||
defs = (
|
||
await self.db.execute(
|
||
select(SymbolDefinition)
|
||
.where(SymbolDefinition.tenant_id == self.tenant_id)
|
||
.order_by(SymbolDefinition.updated_at.desc(), SymbolDefinition.id.desc())
|
||
.limit(max_rows)
|
||
)
|
||
).scalars().all()
|
||
if not defs:
|
||
return {"symbols": 0, "modules": 0, "relations": 0, "evidences": 0}
|
||
|
||
counters = {"symbols": 0, "modules": 0, "relations": 0, "evidences": 0}
|
||
module_entity_map: dict[str, int] = {}
|
||
symbol_entity_map: dict[int, int] = {}
|
||
name_entity_map: dict[str, int] = {}
|
||
qn_entity_map: dict[str, int] = {}
|
||
|
||
for row in defs:
|
||
qn = str(row.qualified_name or row.symbol_name or f"symbol#{int(row.id)}")
|
||
symbol_entity = await self.upsert_entity(
|
||
entity_type="symbol",
|
||
entity_key=f"symbol:{int(row.id)}",
|
||
display_name=qn[:255],
|
||
payload={
|
||
"symbol_id": int(row.id),
|
||
"symbol_name": str(row.symbol_name or ""),
|
||
"qualified_name": str(row.qualified_name or ""),
|
||
"symbol_type": str(row.symbol_type or ""),
|
||
"language": str(row.language or ""),
|
||
"file_path": str(row.file_path or ""),
|
||
"start_line": int(row.start_line) if row.start_line is not None else None,
|
||
"end_line": int(row.end_line) if row.end_line is not None else None,
|
||
},
|
||
source_type="symbol_index",
|
||
source_ref=str(row.file_path or "")[:255],
|
||
)
|
||
counters["symbols"] += 1
|
||
symbol_entity_map[int(row.id)] = int(symbol_entity.id)
|
||
if str(row.symbol_name or "").strip():
|
||
name_entity_map[str(row.symbol_name).strip()] = int(symbol_entity.id)
|
||
if str(row.qualified_name or "").strip():
|
||
qn_entity_map[str(row.qualified_name).strip()] = int(symbol_entity.id)
|
||
|
||
file_key = str(row.file_path or "").strip()
|
||
if not file_key:
|
||
continue
|
||
module_entity_id = module_entity_map.get(file_key)
|
||
if module_entity_id is None:
|
||
module_entity = await self.upsert_entity(
|
||
entity_type="module",
|
||
entity_key=f"module:{file_key}",
|
||
display_name=file_key[:255],
|
||
payload={"file_path": file_key},
|
||
source_type="symbol_index",
|
||
source_ref=file_key[:255],
|
||
)
|
||
module_entity_id = int(module_entity.id)
|
||
module_entity_map[file_key] = module_entity_id
|
||
counters["modules"] += 1
|
||
rel = await self.upsert_relation(
|
||
relation_type="defined_in",
|
||
subject_entity_id=int(symbol_entity.id),
|
||
object_entity_id=int(module_entity_id),
|
||
source_type="symbol_index",
|
||
source_ref=file_key[:255],
|
||
payload={
|
||
"symbol_id": int(row.id),
|
||
"symbol_type": str(row.symbol_type or ""),
|
||
"language": str(row.language or ""),
|
||
},
|
||
)
|
||
counters["relations"] += 1
|
||
await self.add_evidence(
|
||
evidence_type="symbol_index",
|
||
content_excerpt=(
|
||
f"{str(row.symbol_name or '')} defined_in {file_key}:"
|
||
f"{int(row.start_line) if row.start_line is not None else 0}"
|
||
),
|
||
relation_id=int(rel.id),
|
||
source_type="symbol_index",
|
||
source_ref=file_key[:255],
|
||
metadata={"symbol_id": int(row.id), "file_path": file_key},
|
||
)
|
||
counters["evidences"] += 1
|
||
|
||
refs = (
|
||
await self.db.execute(
|
||
select(SymbolReference)
|
||
.where(SymbolReference.tenant_id == self.tenant_id)
|
||
.order_by(SymbolReference.updated_at.desc(), SymbolReference.id.desc())
|
||
.limit(max_rows * 3)
|
||
)
|
||
).scalars().all()
|
||
for row in refs:
|
||
source_entity_id = None
|
||
if row.source_symbol_id is not None:
|
||
source_entity_id = symbol_entity_map.get(int(row.source_symbol_id))
|
||
if source_entity_id is None:
|
||
continue
|
||
target_entity_id = None
|
||
target_qn = str(row.target_qualified_name or "").strip()
|
||
target_name = str(row.target_symbol_name or "").strip()
|
||
if target_qn:
|
||
target_entity_id = qn_entity_map.get(target_qn)
|
||
if target_entity_id is None and target_name:
|
||
target_entity_id = name_entity_map.get(target_name)
|
||
if target_entity_id is None:
|
||
continue
|
||
rel_type = str(row.relation_type or "references").strip()
|
||
rel = await self.upsert_relation(
|
||
relation_type=rel_type[:40],
|
||
subject_entity_id=int(source_entity_id),
|
||
object_entity_id=int(target_entity_id),
|
||
source_type="symbol_index",
|
||
source_ref=str(row.file_path or "")[:255],
|
||
payload={
|
||
"symbol_reference_id": int(row.id),
|
||
"target_symbol_name": target_name,
|
||
"target_qualified_name": target_qn,
|
||
},
|
||
)
|
||
counters["relations"] += 1
|
||
await self.add_evidence(
|
||
evidence_type="symbol_reference",
|
||
content_excerpt=f"{int(source_entity_id)} {rel_type} {target_name or target_qn}",
|
||
relation_id=int(rel.id),
|
||
source_type="symbol_index",
|
||
source_ref=str(row.file_path or "")[:255],
|
||
metadata={
|
||
"symbol_reference_id": int(row.id),
|
||
"target_symbol_name": target_name,
|
||
"target_qualified_name": target_qn,
|
||
},
|
||
)
|
||
counters["evidences"] += 1
|
||
return counters
|
||
|
||
async def sync_feature_catalog_if_due(
|
||
self,
|
||
*,
|
||
interval_hours: int = 6,
|
||
) -> dict[str, Any]:
|
||
hours = max(1, int(interval_hours))
|
||
offset_raw = await self.get_ingest_offset(channel="feature_catalog", offset_key="last_feature_catalog_sync_at")
|
||
last_sync_at = self._parse_iso_dt(offset_raw)
|
||
now = _utc_now_naive()
|
||
if last_sync_at is not None and now - last_sync_at < timedelta(hours=hours):
|
||
return {
|
||
"executed": False,
|
||
"reason": "not_due",
|
||
"last_sync_at": last_sync_at.isoformat(),
|
||
"next_due_at": (last_sync_at + timedelta(hours=hours)).isoformat(),
|
||
}
|
||
|
||
payload = await self._sync_feature_catalog_entities(limit=300)
|
||
await self.enqueue_ingest_event(
|
||
channel="feature_catalog",
|
||
event_key=f"feature_catalog:{_utc_now_naive().strftime('%Y%m%d%H%M%S')}",
|
||
payload=payload,
|
||
status="done",
|
||
)
|
||
await self.set_ingest_offset(
|
||
channel="feature_catalog",
|
||
offset_key="last_feature_catalog_sync_at",
|
||
offset_value=_utc_now_naive().isoformat(),
|
||
payload=payload,
|
||
)
|
||
await self.db.commit()
|
||
return {
|
||
"executed": True,
|
||
"reason": "due",
|
||
"payload": payload,
|
||
}
|
||
|
||
async def sync_feature_catalog_now(self) -> dict[str, Any]:
|
||
payload = await self._sync_feature_catalog_entities(limit=300)
|
||
await self.enqueue_ingest_event(
|
||
channel="feature_catalog",
|
||
event_key=f"feature_catalog:{_utc_now_naive().strftime('%Y%m%d%H%M%S')}",
|
||
payload=payload,
|
||
status="done",
|
||
)
|
||
await self.set_ingest_offset(
|
||
channel="feature_catalog",
|
||
offset_key="last_feature_catalog_sync_at",
|
||
offset_value=_utc_now_naive().isoformat(),
|
||
payload=payload,
|
||
)
|
||
await self.db.commit()
|
||
return payload
|
||
|
||
async def _sync_feature_catalog_entities(self, *, limit: int = 300) -> dict[str, Any]:
|
||
snapshot = await context_cache_service.get_system_snapshot(
|
||
db=self.db,
|
||
tenant_id=self.tenant_id,
|
||
)
|
||
platform = snapshot.get("platform_profile") if isinstance(snapshot.get("platform_profile"), dict) else {}
|
||
catalog = platform.get("feature_catalog") if isinstance(platform.get("feature_catalog"), list) else []
|
||
rows = [row for row in catalog if isinstance(row, dict)][: max(1, min(int(limit), 1000))]
|
||
if not rows:
|
||
return {"features": 0, "routes": 0, "pages": 0, "apis": 0, "relations": 0, "evidences": 0}
|
||
|
||
counters = {"features": 0, "routes": 0, "pages": 0, "apis": 0, "relations": 0, "evidences": 0}
|
||
route_entity_map: dict[str, int] = {}
|
||
page_entity_map: dict[str, int] = {}
|
||
api_entity_map: dict[str, int] = {}
|
||
|
||
for row in rows:
|
||
name = str(row.get("name") or "").strip() or "unknown_feature"
|
||
route = str(row.get("route") or "").strip()
|
||
summary = str(row.get("summary") or "").strip()
|
||
page_file = str(row.get("page_file") or "").strip()
|
||
page_component = str(row.get("page_component") or "").strip()
|
||
apis = row.get("apis") if isinstance(row.get("apis"), list) else []
|
||
aliases = row.get("aliases") if isinstance(row.get("aliases"), list) else []
|
||
|
||
feature_slug = (route or name).strip().lower()[:200]
|
||
feature_entity = await self.upsert_entity(
|
||
entity_type="feature",
|
||
entity_key=f"feature:{feature_slug}",
|
||
display_name=name[:255],
|
||
payload={
|
||
"name": name,
|
||
"route": route,
|
||
"summary": summary,
|
||
"page_file": page_file,
|
||
"page_component": page_component,
|
||
"apis": [str(x) for x in apis[:20] if str(x).strip()],
|
||
"aliases": [str(x) for x in aliases[:20] if str(x).strip()],
|
||
},
|
||
source_type="feature_catalog",
|
||
source_ref=(route or page_file or name)[:255],
|
||
)
|
||
counters["features"] += 1
|
||
await self.add_evidence(
|
||
evidence_type="feature_catalog",
|
||
content_excerpt=f"{name} | route={route or 'n/a'} | page={page_file or 'n/a'}",
|
||
entity_id=int(feature_entity.id),
|
||
source_type="feature_catalog",
|
||
source_ref=(route or page_file or name)[:255],
|
||
metadata={"feature_name": name, "route": route, "page_file": page_file},
|
||
)
|
||
counters["evidences"] += 1
|
||
|
||
route_entity_id: Optional[int] = None
|
||
if route:
|
||
if route not in route_entity_map:
|
||
route_entity = await self.upsert_entity(
|
||
entity_type="route",
|
||
entity_key=f"route:{route[:200]}",
|
||
display_name=route[:255],
|
||
payload={"route": route},
|
||
source_type="feature_catalog",
|
||
source_ref=route[:255],
|
||
)
|
||
route_entity_map[route] = int(route_entity.id)
|
||
counters["routes"] += 1
|
||
route_entity_id = route_entity_map[route]
|
||
rel = await self.upsert_relation(
|
||
relation_type="feature_has_route",
|
||
subject_entity_id=int(feature_entity.id),
|
||
object_entity_id=int(route_entity_id),
|
||
source_type="feature_catalog",
|
||
source_ref=route[:255],
|
||
)
|
||
counters["relations"] += 1
|
||
await self.add_evidence(
|
||
evidence_type="feature_catalog",
|
||
content_excerpt=f"{name} has route {route}",
|
||
relation_id=int(rel.id),
|
||
source_type="feature_catalog",
|
||
source_ref=route[:255],
|
||
metadata={"feature_name": name, "route": route},
|
||
)
|
||
counters["evidences"] += 1
|
||
|
||
page_entity_id: Optional[int] = None
|
||
if page_file:
|
||
if page_file not in page_entity_map:
|
||
page_entity = await self.upsert_entity(
|
||
entity_type="page",
|
||
entity_key=f"page:{page_file[:220]}",
|
||
display_name=(page_component or page_file)[:255],
|
||
payload={"page_file": page_file, "page_component": page_component},
|
||
source_type="feature_catalog",
|
||
source_ref=page_file[:255],
|
||
)
|
||
page_entity_map[page_file] = int(page_entity.id)
|
||
counters["pages"] += 1
|
||
page_entity_id = page_entity_map[page_file]
|
||
rel = await self.upsert_relation(
|
||
relation_type="feature_rendered_by",
|
||
subject_entity_id=int(feature_entity.id),
|
||
object_entity_id=int(page_entity_id),
|
||
source_type="feature_catalog",
|
||
source_ref=page_file[:255],
|
||
)
|
||
counters["relations"] += 1
|
||
await self.add_evidence(
|
||
evidence_type="feature_catalog",
|
||
content_excerpt=f"{name} rendered_by {page_file}",
|
||
relation_id=int(rel.id),
|
||
source_type="feature_catalog",
|
||
source_ref=page_file[:255],
|
||
metadata={"feature_name": name, "page_file": page_file},
|
||
)
|
||
counters["evidences"] += 1
|
||
|
||
if route_entity_id is not None and page_entity_id is not None:
|
||
rel = await self.upsert_relation(
|
||
relation_type="route_to_page",
|
||
subject_entity_id=int(route_entity_id),
|
||
object_entity_id=int(page_entity_id),
|
||
source_type="feature_catalog",
|
||
source_ref=(route or page_file)[:255],
|
||
)
|
||
counters["relations"] += 1
|
||
await self.add_evidence(
|
||
evidence_type="feature_catalog",
|
||
content_excerpt=f"{route} maps_to {page_file}",
|
||
relation_id=int(rel.id),
|
||
source_type="feature_catalog",
|
||
source_ref=(route or page_file)[:255],
|
||
metadata={"route": route, "page_file": page_file},
|
||
)
|
||
counters["evidences"] += 1
|
||
|
||
for raw_api in apis[:20]:
|
||
api_path = str(raw_api or "").strip()
|
||
if not api_path:
|
||
continue
|
||
if api_path not in api_entity_map:
|
||
api_entity = await self.upsert_entity(
|
||
entity_type="api_endpoint",
|
||
entity_key=f"api:{api_path[:220]}",
|
||
display_name=api_path[:255],
|
||
payload={"path": api_path},
|
||
source_type="feature_catalog",
|
||
source_ref=api_path[:255],
|
||
)
|
||
api_entity_map[api_path] = int(api_entity.id)
|
||
counters["apis"] += 1
|
||
rel = await self.upsert_relation(
|
||
relation_type="feature_calls_api",
|
||
subject_entity_id=int(feature_entity.id),
|
||
object_entity_id=int(api_entity_map[api_path]),
|
||
source_type="feature_catalog",
|
||
source_ref=api_path[:255],
|
||
)
|
||
counters["relations"] += 1
|
||
await self.add_evidence(
|
||
evidence_type="feature_catalog",
|
||
content_excerpt=f"{name} calls {api_path}",
|
||
relation_id=int(rel.id),
|
||
source_type="feature_catalog",
|
||
source_ref=api_path[:255],
|
||
metadata={"feature_name": name, "api_path": api_path},
|
||
)
|
||
counters["evidences"] += 1
|
||
|
||
return counters
|
||
|
||
async def run_incremental_cycle(
|
||
self,
|
||
*,
|
||
delta_limit: int = 1000,
|
||
event_batch: int = 200,
|
||
code_scan_hours: int = 6,
|
||
feature_scan_hours: int = 6,
|
||
code_scan_base_dir: Optional[str] = None,
|
||
run_source: str = "system",
|
||
) -> dict[str, Any]:
|
||
started = _utc_now_naive()
|
||
sync_run = await self._open_sync_run(
|
||
run_type="incremental",
|
||
trigger_source=run_source,
|
||
input_payload={
|
||
"delta_limit": int(delta_limit),
|
||
"event_batch": int(event_batch),
|
||
"code_scan_hours": int(code_scan_hours),
|
||
"feature_scan_hours": int(feature_scan_hours),
|
||
"code_scan_base_dir": str(code_scan_base_dir or ""),
|
||
},
|
||
)
|
||
run_id = int(sync_run.id)
|
||
run_started_at = sync_run.started_at or started
|
||
stage_results: dict[str, Any] = {}
|
||
stage_errors: dict[str, str] = {}
|
||
success_count = 0
|
||
|
||
try:
|
||
try:
|
||
stage_results["structured_delta"] = await self.sync_structured_delta(limit=delta_limit)
|
||
success_count += 1
|
||
except Exception as exc:
|
||
await self.db.rollback()
|
||
stage_errors["structured_delta"] = str(exc)[:500]
|
||
stage_results["structured_delta"] = {"error": stage_errors["structured_delta"]}
|
||
|
||
try:
|
||
stage_results["feishu_enqueue"] = await self.enqueue_feishu_delta_events(limit=min(delta_limit, 1000))
|
||
success_count += 1
|
||
except Exception as exc:
|
||
await self.db.rollback()
|
||
stage_errors["feishu_enqueue"] = str(exc)[:500]
|
||
stage_results["feishu_enqueue"] = {"error": stage_errors["feishu_enqueue"]}
|
||
|
||
try:
|
||
stage_results["event_processing"] = await self.process_pending_events(max_batch=event_batch)
|
||
success_count += 1
|
||
except Exception as exc:
|
||
await self.db.rollback()
|
||
stage_errors["event_processing"] = str(exc)[:500]
|
||
stage_results["event_processing"] = {"error": stage_errors["event_processing"]}
|
||
|
||
try:
|
||
stage_results["code_scan"] = await self.sync_codebase_if_due(
|
||
base_dir=code_scan_base_dir,
|
||
interval_hours=code_scan_hours,
|
||
)
|
||
success_count += 1
|
||
except Exception as exc:
|
||
await self.db.rollback()
|
||
stage_errors["code_scan"] = str(exc)[:500]
|
||
stage_results["code_scan"] = {"error": stage_errors["code_scan"]}
|
||
|
||
try:
|
||
stage_results["symbol_index"] = await self.sync_symbol_index_if_due(
|
||
base_dir=code_scan_base_dir,
|
||
interval_hours=code_scan_hours,
|
||
)
|
||
success_count += 1
|
||
except Exception as exc:
|
||
await self.db.rollback()
|
||
stage_errors["symbol_index"] = str(exc)[:500]
|
||
stage_results["symbol_index"] = {"error": stage_errors["symbol_index"]}
|
||
|
||
try:
|
||
stage_results["feature_catalog"] = await self.sync_feature_catalog_if_due(
|
||
interval_hours=feature_scan_hours,
|
||
)
|
||
success_count += 1
|
||
except Exception as exc:
|
||
await self.db.rollback()
|
||
stage_errors["feature_catalog"] = str(exc)[:500]
|
||
stage_results["feature_catalog"] = {"error": stage_errors["feature_catalog"]}
|
||
|
||
ended = _utc_now_naive()
|
||
if success_count == 6:
|
||
final_status = "success"
|
||
elif success_count == 0:
|
||
final_status = "failed"
|
||
else:
|
||
final_status = "partial"
|
||
|
||
payload = {
|
||
"tenant_id": self.tenant_id,
|
||
"started_at": started.isoformat(),
|
||
"ended_at": ended.isoformat(),
|
||
"run_id": run_id,
|
||
"run_status": final_status,
|
||
"success_stage_count": success_count,
|
||
"failed_stage_count": len(stage_errors),
|
||
"stage_errors": stage_errors,
|
||
**stage_results,
|
||
}
|
||
await self._close_sync_run(
|
||
row=sync_run,
|
||
status=final_status,
|
||
result_payload=payload,
|
||
error_message=("; ".join(f"{k}:{v}" for k, v in stage_errors.items()) if stage_errors else None),
|
||
started_at_hint=run_started_at,
|
||
)
|
||
return payload
|
||
except Exception as exc:
|
||
await self.db.rollback()
|
||
ended = _utc_now_naive()
|
||
payload = {
|
||
"tenant_id": self.tenant_id,
|
||
"started_at": started.isoformat(),
|
||
"ended_at": ended.isoformat(),
|
||
"run_id": run_id,
|
||
"run_status": "failed",
|
||
"success_stage_count": success_count,
|
||
"failed_stage_count": max(1, len(stage_errors)),
|
||
"stage_errors": stage_errors,
|
||
**stage_results,
|
||
}
|
||
await self._close_sync_run(
|
||
row=sync_run,
|
||
status="failed",
|
||
result_payload=payload,
|
||
error_message=str(exc)[:1000],
|
||
started_at_hint=run_started_at,
|
||
)
|
||
return payload
|
||
|
||
async def sync_structured_entities(self, *, limit: int = 1000) -> dict[str, Any]:
|
||
max_rows = max(1, min(int(limit), 5000))
|
||
counters = {
|
||
"projects": 0,
|
||
"tasks": 0,
|
||
"signals": 0,
|
||
"users": 0,
|
||
"relations": 0,
|
||
"evidences": 0,
|
||
}
|
||
|
||
project_map: dict[int, int] = {}
|
||
projects = (
|
||
await self.db.execute(
|
||
select(Project)
|
||
.where(Project.tenant_id == self.tenant_id)
|
||
.order_by(Project.updated_at.desc(), Project.id.desc())
|
||
.limit(max_rows)
|
||
)
|
||
).scalars().all()
|
||
for p in projects:
|
||
entity = await self.upsert_entity(
|
||
entity_type="project",
|
||
entity_key=f"project:{int(p.id)}",
|
||
display_name=str(p.title or p.name or f"Project#{int(p.id)}"),
|
||
payload={
|
||
"project_id": int(p.id),
|
||
"status": str(p.status or "unknown"),
|
||
"owner_id": int(p.owner_id) if p.owner_id is not None else None,
|
||
},
|
||
source_type="db",
|
||
source_ref=f"projects:{int(p.id)}",
|
||
)
|
||
project_map[int(p.id)] = int(entity.id)
|
||
counters["projects"] += 1
|
||
|
||
user_map: dict[int, int] = {}
|
||
users = (
|
||
await self.db.execute(
|
||
select(User)
|
||
.where(User.tenant_id == self.tenant_id)
|
||
.order_by(User.updated_at.desc(), User.id.desc())
|
||
.limit(max_rows)
|
||
)
|
||
).scalars().all()
|
||
for u in users:
|
||
entity = await self.upsert_entity(
|
||
entity_type="user",
|
||
entity_key=f"user:{int(u.id)}",
|
||
display_name=str(u.name or f"User#{int(u.id)}"),
|
||
payload={
|
||
"user_id": int(u.id),
|
||
"role": str(u.role or "unknown"),
|
||
"is_active": bool(u.is_active),
|
||
},
|
||
source_type="db",
|
||
source_ref=f"users:{int(u.id)}",
|
||
)
|
||
user_map[int(u.id)] = int(entity.id)
|
||
counters["users"] += 1
|
||
|
||
tasks = (
|
||
await self.db.execute(
|
||
select(Task)
|
||
.where(Task.tenant_id == self.tenant_id)
|
||
.order_by(Task.updated_at.desc(), Task.id.desc())
|
||
.limit(max_rows)
|
||
)
|
||
).scalars().all()
|
||
for t in tasks:
|
||
task_entity = await self.upsert_entity(
|
||
entity_type="task",
|
||
entity_key=f"task:{int(t.id)}",
|
||
display_name=str(t.title or f"Task#{int(t.id)}"),
|
||
payload={
|
||
"task_id": int(t.id),
|
||
"project_id": int(t.project_id) if t.project_id is not None else None,
|
||
"assignee_id": int(t.assignee_id) if t.assignee_id is not None else None,
|
||
"status": str(t.status or "unknown"),
|
||
},
|
||
source_type="db",
|
||
source_ref=f"tasks:{int(t.id)}",
|
||
)
|
||
counters["tasks"] += 1
|
||
|
||
if t.project_id is not None and int(t.project_id) in project_map:
|
||
rel = await self.upsert_relation(
|
||
relation_type="belongs_to",
|
||
subject_entity_id=int(task_entity.id),
|
||
object_entity_id=int(project_map[int(t.project_id)]),
|
||
source_type="db",
|
||
source_ref=f"tasks:{int(t.id)}",
|
||
)
|
||
counters["relations"] += 1
|
||
await self.add_evidence(
|
||
evidence_type="db_row",
|
||
content_excerpt=f"task#{int(t.id)} belongs_to project#{int(t.project_id)}",
|
||
relation_id=int(rel.id),
|
||
source_type="db",
|
||
source_ref=f"tasks:{int(t.id)}",
|
||
metadata={"table": "tasks"},
|
||
)
|
||
counters["evidences"] += 1
|
||
|
||
if t.assignee_id is not None and int(t.assignee_id) in user_map:
|
||
rel = await self.upsert_relation(
|
||
relation_type="assigned_to",
|
||
subject_entity_id=int(task_entity.id),
|
||
object_entity_id=int(user_map[int(t.assignee_id)]),
|
||
source_type="db",
|
||
source_ref=f"tasks:{int(t.id)}",
|
||
)
|
||
counters["relations"] += 1
|
||
await self.add_evidence(
|
||
evidence_type="db_row",
|
||
content_excerpt=f"task#{int(t.id)} assigned_to user#{int(t.assignee_id)}",
|
||
relation_id=int(rel.id),
|
||
source_type="db",
|
||
source_ref=f"tasks:{int(t.id)}",
|
||
metadata={"table": "tasks"},
|
||
)
|
||
counters["evidences"] += 1
|
||
|
||
signals = (
|
||
await self.db.execute(
|
||
select(Signal)
|
||
.where(Signal.tenant_id == self.tenant_id)
|
||
.order_by(Signal.updated_at.desc(), Signal.id.desc())
|
||
.limit(max_rows)
|
||
)
|
||
).scalars().all()
|
||
for s in signals:
|
||
signal_entity = await self.upsert_entity(
|
||
entity_type="signal",
|
||
entity_key=f"signal:{int(s.id)}",
|
||
display_name=str(s.title or s.content or f"Signal#{int(s.id)}")[:255],
|
||
payload={
|
||
"signal_id": int(s.id),
|
||
"priority": str(s.priority or ""),
|
||
"status": str(s.status or ""),
|
||
"matched_project_id": (
|
||
int(s.matched_project_id) if getattr(s, "matched_project_id", None) is not None else None
|
||
),
|
||
},
|
||
source_type="db",
|
||
source_ref=f"signals:{int(s.id)}",
|
||
)
|
||
counters["signals"] += 1
|
||
|
||
target_pid = int(s.matched_project_id) if getattr(s, "matched_project_id", None) is not None else None
|
||
if target_pid is not None and target_pid in project_map:
|
||
rel = await self.upsert_relation(
|
||
relation_type="matches_project",
|
||
subject_entity_id=int(signal_entity.id),
|
||
object_entity_id=int(project_map[target_pid]),
|
||
source_type="db",
|
||
source_ref=f"signals:{int(s.id)}",
|
||
)
|
||
counters["relations"] += 1
|
||
await self.add_evidence(
|
||
evidence_type="db_row",
|
||
content_excerpt=f"signal#{int(s.id)} matches project#{target_pid}",
|
||
relation_id=int(rel.id),
|
||
source_type="db",
|
||
source_ref=f"signals:{int(s.id)}",
|
||
metadata={"table": "signals"},
|
||
)
|
||
counters["evidences"] += 1
|
||
|
||
await self.enqueue_ingest_event(
|
||
channel="db_delta",
|
||
event_key=f"structured_sync:{_utc_now_naive().strftime('%Y%m%d%H%M%S')}",
|
||
payload={"limit": max_rows, "counters": counters},
|
||
status="done",
|
||
)
|
||
await self.set_ingest_offset(
|
||
channel="db_delta",
|
||
offset_key="last_structured_sync_at",
|
||
offset_value=_utc_now_naive().isoformat(),
|
||
payload={"counters": counters},
|
||
)
|
||
|
||
await self.db.commit()
|
||
return counters
|
||
|
||
async def sync_codebase_snapshot(self, *, base_dir: Optional[str] = None) -> dict[str, Any]:
|
||
root = Path(base_dir or "/app/backend").resolve()
|
||
if not root.exists() or not root.is_dir():
|
||
raise ValueError(f"codebase path not found: {root}")
|
||
|
||
py_files = [p for p in root.rglob("*.py") if p.is_file()]
|
||
route_markers = ("@router.get(", "@router.post(", "@router.put(", "@router.patch(", "@router.delete(")
|
||
route_count = 0
|
||
service_file_count = 0
|
||
model_file_count = 0
|
||
|
||
for py_file in py_files:
|
||
rel = str(py_file.relative_to(root))
|
||
if rel.startswith("app/services/"):
|
||
service_file_count += 1
|
||
if rel.startswith("app/models/"):
|
||
model_file_count += 1
|
||
try:
|
||
text = py_file.read_text(encoding="utf-8", errors="ignore")
|
||
except Exception:
|
||
continue
|
||
for marker in route_markers:
|
||
route_count += text.count(marker)
|
||
|
||
payload = {
|
||
"root": str(root),
|
||
"python_files": len(py_files),
|
||
"service_files": service_file_count,
|
||
"model_files": model_file_count,
|
||
"api_route_markers": route_count,
|
||
"scanned_at": _utc_now_naive().isoformat(),
|
||
}
|
||
entity = await self.upsert_entity(
|
||
entity_type="service",
|
||
entity_key="service:backend_codebase",
|
||
display_name="Backend Codebase Snapshot",
|
||
payload=payload,
|
||
source_type="code_scan",
|
||
source_ref="filesystem",
|
||
)
|
||
await self.add_evidence(
|
||
evidence_type="code_scan",
|
||
content_excerpt=f"python_files={len(py_files)}, routes={route_count}",
|
||
entity_id=int(entity.id),
|
||
source_type="code_scan",
|
||
source_ref=str(root),
|
||
metadata=payload,
|
||
)
|
||
await self.enqueue_ingest_event(
|
||
channel="code_scan",
|
||
event_key=f"code_scan:{_utc_now_naive().strftime('%Y%m%d%H%M%S')}",
|
||
payload=payload,
|
||
status="done",
|
||
)
|
||
await self.set_ingest_offset(
|
||
channel="code_scan",
|
||
offset_key="last_code_scan_at",
|
||
offset_value=_utc_now_naive().isoformat(),
|
||
payload=payload,
|
||
)
|
||
await self.db.commit()
|
||
return payload
|
||
|
||
async def get_overview(self) -> dict[str, Any]:
|
||
entity_total = (
|
||
await self.db.execute(select(func.count(KGEntity.id)).where(KGEntity.tenant_id == self.tenant_id))
|
||
).scalar_one()
|
||
relation_total = (
|
||
await self.db.execute(select(func.count(KGRelation.id)).where(KGRelation.tenant_id == self.tenant_id))
|
||
).scalar_one()
|
||
evidence_total = (
|
||
await self.db.execute(select(func.count(KGEvidence.id)).where(KGEvidence.tenant_id == self.tenant_id))
|
||
).scalar_one()
|
||
pending_events = (
|
||
await self.db.execute(
|
||
select(func.count(IngestEvent.id)).where(
|
||
and_(
|
||
IngestEvent.tenant_id == self.tenant_id,
|
||
IngestEvent.status.in_(["pending", "retry"]),
|
||
)
|
||
)
|
||
)
|
||
).scalar_one()
|
||
latest_run = (
|
||
await self.db.execute(
|
||
select(KGSyncRun)
|
||
.where(KGSyncRun.tenant_id == self.tenant_id)
|
||
.order_by(KGSyncRun.started_at.desc(), KGSyncRun.id.desc())
|
||
.limit(1)
|
||
)
|
||
).scalar_one_or_none()
|
||
|
||
entity_groups = (
|
||
await self.db.execute(
|
||
select(KGEntity.entity_type, func.count(KGEntity.id))
|
||
.where(KGEntity.tenant_id == self.tenant_id)
|
||
.group_by(KGEntity.entity_type)
|
||
)
|
||
).all()
|
||
relation_groups = (
|
||
await self.db.execute(
|
||
select(KGRelation.relation_type, func.count(KGRelation.id))
|
||
.where(KGRelation.tenant_id == self.tenant_id)
|
||
.group_by(KGRelation.relation_type)
|
||
)
|
||
).all()
|
||
|
||
return {
|
||
"tenant_id": self.tenant_id,
|
||
"entity_total": int(entity_total or 0),
|
||
"relation_total": int(relation_total or 0),
|
||
"evidence_total": int(evidence_total or 0),
|
||
"pending_events": int(pending_events or 0),
|
||
"entities_by_type": {str(k): int(v or 0) for k, v in entity_groups},
|
||
"relations_by_type": {str(k): int(v or 0) for k, v in relation_groups},
|
||
"latest_sync_run": (
|
||
{
|
||
"id": int(latest_run.id),
|
||
"run_type": str(latest_run.run_type or ""),
|
||
"trigger_source": str(latest_run.trigger_source or ""),
|
||
"status": str(latest_run.status or ""),
|
||
"started_at": latest_run.started_at.isoformat() if latest_run.started_at else None,
|
||
"ended_at": latest_run.ended_at.isoformat() if latest_run.ended_at else None,
|
||
"duration_ms": int(latest_run.duration_ms or 0),
|
||
}
|
||
if latest_run is not None
|
||
else None
|
||
),
|
||
"snapshot_at": _utc_now_naive().isoformat(),
|
||
}
|