463 lines
16 KiB
Python
463 lines
16 KiB
Python
"""信号相关 API 路由"""
|
||
|
||
import json
|
||
from typing import List, Optional
|
||
|
||
from fastapi import APIRouter, Depends, HTTPException, BackgroundTasks, Query
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
from sqlalchemy.future import select
|
||
|
||
from ..database import get_db
|
||
from ..dependencies.tenant import get_tenant_id
|
||
from ..models import Signal, SignalFeedback, Project, ProjectSignalLink
|
||
from ..schemas import (
|
||
SignalCreate,
|
||
SignalPriority,
|
||
classify_signal_priority,
|
||
SignalFeedbackCreate,
|
||
SignalFeedbackResponse,
|
||
)
|
||
from ..services.fetcher import fetch_hn_stories
|
||
from ..services.analyzer import analyze_impact
|
||
from ..services.llm import llm_service
|
||
from ..services.plane import plane_service
|
||
from ..services.lark import LarkService
|
||
|
||
router = APIRouter(prefix="/signals", tags=["signals"])
|
||
lark_service = LarkService()
|
||
|
||
|
||
@router.get("/")
|
||
async def list_signals(
|
||
roles: Optional[str] = Query(None, description="逗号分隔的角色列表，如 'frontend,backend'"),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
"""获取信号列表，支持按角色过滤"""
|
||
result = await db.execute(select(Signal).order_by(Signal.impact_score.desc()))
|
||
signals = result.scalars().all()
|
||
|
||
if roles:
|
||
role_list = [r.strip() for r in roles.split(",")]
|
||
filtered_signals = []
|
||
for signal in signals:
|
||
tasks = []
|
||
if signal.tasks_generated:
|
||
try:
|
||
tasks = json.loads(signal.tasks_generated)
|
||
except Exception:
|
||
tasks = []
|
||
|
||
filtered_tasks = [t for t in tasks if t.get("type") in role_list]
|
||
|
||
if filtered_tasks:
|
||
signal_dict = {
|
||
"id": str(signal.id),
|
||
"source": signal.source,
|
||
"content": signal.content,
|
||
"impact_score": signal.impact_score,
|
||
"priority": signal.priority,
|
||
"processed": signal.processed,
|
||
"strategic_analysis": signal.strategic_analysis,
|
||
"tasks_generated": signal.tasks_generated,
|
||
"tasks_filtered": filtered_tasks,
|
||
"created_at": signal.created_at.isoformat() if signal.created_at else None,
|
||
}
|
||
filtered_signals.append(signal_dict)
|
||
|
||
return filtered_signals
|
||
|
||
return signals
|
||
|
||
|
||
@router.post("/", response_model=dict)
|
||
async def create_signal(signal: SignalCreate, db: AsyncSession = Depends(get_db)):
|
||
new_signal = Signal(
|
||
source=signal.source,
|
||
content=signal.content,
|
||
impact_score=signal.impact_score,
|
||
)
|
||
db.add(new_signal)
|
||
await db.commit()
|
||
await db.refresh(new_signal)
|
||
return {"id": str(new_signal.id), "status": "received"}
|
||
|
||
|
||
@router.post("/ingest", response_model=dict)
|
||
async def ingest_signal(
|
||
signal: SignalCreate,
|
||
background_tasks: BackgroundTasks,
|
||
db: AsyncSession = Depends(get_db),
|
||
tenant_id: int = Depends(get_tenant_id),
|
||
):
|
||
"""接入外部信号，按优先级处理"""
|
||
initial_priority = signal.priority or classify_signal_priority(signal.impact_score)
|
||
|
||
new_signal = Signal(
|
||
source=signal.source,
|
||
content=signal.content,
|
||
impact_score=signal.impact_score,
|
||
priority=initial_priority.value,
|
||
)
|
||
db.add(new_signal)
|
||
await db.commit()
|
||
await db.refresh(new_signal)
|
||
|
||
if initial_priority == SignalPriority.LOW:
|
||
return {
|
||
"id": str(new_signal.id),
|
||
"priority": initial_priority.value,
|
||
"status": "recorded",
|
||
"message": "Low priority signal recorded without processing",
|
||
}
|
||
|
||
async def process_signal_internal():
|
||
try:
|
||
relevant_context = await llm_service.find_relevant_context(
|
||
signal_content=signal.content,
|
||
db_session=db,
|
||
top_k=3,
|
||
min_similarity=0.3,
|
||
tenant_id=tenant_id,
|
||
)
|
||
|
||
strategy_result = await llm_service.evaluate_with_voting_async(
|
||
signal_content=signal.content,
|
||
relevant_context=relevant_context,
|
||
impact_score=signal.impact_score,
|
||
db=db,
|
||
tenant_id=tenant_id,
|
||
)
|
||
except RuntimeError:
|
||
if initial_priority == SignalPriority.CRITICAL:
|
||
raise HTTPException(status_code=403, detail="预算超限，AI 调用被阻止（hard limit）")
|
||
return {
|
||
"id": str(new_signal.id),
|
||
"priority": initial_priority.value,
|
||
"status": "blocked",
|
||
"processing_mode": "background",
|
||
"message": "预算超限，AI 调用被阻止（hard limit）",
|
||
}
|
||
|
||
is_strategic = strategy_result.get("is_strategic", False)
|
||
decision_proposal = strategy_result.get("decision_proposal", {})
|
||
tasks = strategy_result.get("tasks", [])
|
||
confidence = strategy_result.get("confidence", 0.0)
|
||
|
||
final_priority = classify_signal_priority(signal.impact_score, confidence)
|
||
new_signal.priority = final_priority.value
|
||
|
||
sync_result = None
|
||
notification_sent = False
|
||
|
||
if final_priority in [SignalPriority.CRITICAL, SignalPriority.HIGH]:
|
||
if is_strategic and tasks:
|
||
sync_result = await plane_service.sync_tasks(
|
||
tasks, is_urgent=(final_priority == SignalPriority.CRITICAL)
|
||
)
|
||
default_chat_id = "oc_1234567890"
|
||
await lark_service.send_strategic_alert(
|
||
default_chat_id, signal.content, decision_proposal, tasks
|
||
)
|
||
notification_sent = True
|
||
new_signal.impact_score = 1.0
|
||
|
||
analysis_data = {
|
||
"decision_proposal": decision_proposal,
|
||
"competitiveness_radar": strategy_result.get("competitiveness_radar"),
|
||
"detected_tech_stacks": strategy_result.get("detected_tech_stacks", []),
|
||
"related_knowledge_nodes": strategy_result.get("related_knowledge_nodes", []),
|
||
"rag_context_used": len(relevant_context) if relevant_context else 0,
|
||
"rag_context_sources": [ctx["id"] for ctx in (relevant_context or [])],
|
||
}
|
||
|
||
if "_voting" in strategy_result:
|
||
voting_info = strategy_result["_voting"]
|
||
analysis_data["voting"] = {
|
||
"enabled": voting_info.get("enabled"),
|
||
"models_used": voting_info.get("models_used"),
|
||
"strategic_votes": voting_info.get("strategic_votes"),
|
||
"total_votes": voting_info.get("total_votes"),
|
||
"unanimous": voting_info.get("unanimous"),
|
||
}
|
||
|
||
new_signal.strategic_analysis = json.dumps(analysis_data)
|
||
new_signal.tasks_generated = json.dumps(tasks)
|
||
new_signal.processed = True
|
||
|
||
await db.commit()
|
||
await db.refresh(new_signal)
|
||
|
||
return {
|
||
"id": str(new_signal.id),
|
||
"priority": final_priority.value,
|
||
"is_strategic": is_strategic,
|
||
"confidence": confidence,
|
||
"decision_proposal": decision_proposal,
|
||
"rag_context_used": len(relevant_context) if relevant_context else 0,
|
||
"voting": strategy_result.get("_voting"),
|
||
"tasks_generated": len(tasks),
|
||
"sync_result": sync_result,
|
||
"notification_sent": notification_sent,
|
||
}
|
||
|
||
if initial_priority == SignalPriority.CRITICAL:
|
||
result = await process_signal_internal()
|
||
result["processing_mode"] = "synchronous"
|
||
return result
|
||
else:
|
||
background_tasks.add_task(process_signal_internal)
|
||
return {
|
||
"id": str(new_signal.id),
|
||
"priority": initial_priority.value,
|
||
"status": "queued",
|
||
"processing_mode": "background",
|
||
"message": f"{initial_priority.value.upper()} priority signal queued for processing",
|
||
}
|
||
|
||
|
||
@router.post("/fetch", response_model=dict)
|
||
async def trigger_signal_fetch(
|
||
background_tasks: BackgroundTasks,
|
||
db: AsyncSession = Depends(get_db),
|
||
tenant_id: int = Depends(get_tenant_id),
|
||
):
|
||
"""手动触发抓取并分析信号"""
|
||
|
||
def _stop_ollama():
|
||
import subprocess
|
||
import signal
|
||
import os
|
||
|
||
killed = False
|
||
try:
|
||
result = subprocess.run(
|
||
["pgrep", "-f", "ollama"],
|
||
capture_output=True,
|
||
text=True,
|
||
)
|
||
if result.returncode == 0:
|
||
for pid in result.stdout.strip().split("\n"):
|
||
if pid:
|
||
try:
|
||
os.kill(int(pid), signal.SIGTERM)
|
||
killed = True
|
||
except ProcessLookupError:
|
||
pass
|
||
except Exception as e:
|
||
print(f"[AI] pgrep failed: {e}")
|
||
|
||
if not killed:
|
||
try:
|
||
result = subprocess.run(
|
||
["pkill", "-9", "-f", "ollama"],
|
||
capture_output=True,
|
||
timeout=5,
|
||
)
|
||
killed = result.returncode == 0
|
||
except Exception as e:
|
||
print(f"[AI] pkill failed: {e}")
|
||
|
||
if killed:
|
||
print("[AI] Ollama stopped - analysis complete")
|
||
|
||
raw_stories = await fetch_hn_stories(limit=30)
|
||
new_signals_count = 0
|
||
processed_count = 0
|
||
|
||
for story in raw_stories:
|
||
result = await db.execute(select(Signal).where(Signal.content == story["content"]))
|
||
if result.scalars().first():
|
||
continue
|
||
|
||
impact = analyze_impact(story["content"])
|
||
|
||
if impact > 0.4:
|
||
initial_priority = classify_signal_priority(impact)
|
||
|
||
new_signal = Signal(
|
||
source=story["source"],
|
||
content=story["content"],
|
||
impact_score=impact,
|
||
priority=initial_priority.value,
|
||
)
|
||
db.add(new_signal)
|
||
await db.commit()
|
||
await db.refresh(new_signal)
|
||
new_signals_count += 1
|
||
|
||
if initial_priority == SignalPriority.LOW:
|
||
continue
|
||
|
||
async def process_fetched():
|
||
nonlocal processed_count
|
||
try:
|
||
relevant_context = await llm_service.find_relevant_context(
|
||
signal_content=story["content"],
|
||
db_session=db,
|
||
top_k=3,
|
||
min_similarity=0.3,
|
||
tenant_id=tenant_id,
|
||
)
|
||
strategy_result = await llm_service.evaluate_with_voting_async(
|
||
signal_content=story["content"],
|
||
relevant_context=relevant_context,
|
||
impact_score=impact,
|
||
db=db,
|
||
tenant_id=tenant_id,
|
||
)
|
||
except RuntimeError:
|
||
return
|
||
is_strategic = strategy_result.get("is_strategic", False)
|
||
decision_proposal = strategy_result.get("decision_proposal", {})
|
||
confidence = strategy_result.get("confidence", 0.0)
|
||
final_priority = classify_signal_priority(impact, confidence)
|
||
new_signal.priority = final_priority.value
|
||
|
||
analysis_data = {
|
||
"decision_proposal": decision_proposal,
|
||
"competitiveness_radar": strategy_result.get("competitiveness_radar"),
|
||
"detected_tech_stacks": strategy_result.get("detected_tech_stacks", []),
|
||
"related_knowledge_nodes": strategy_result.get(
|
||
"related_knowledge_nodes", []
|
||
),
|
||
"rag_context_used": len(relevant_context) if relevant_context else 0,
|
||
"rag_context_sources": [ctx["id"] for ctx in (relevant_context or [])],
|
||
"confidence": confidence,
|
||
"is_strategic": is_strategic,
|
||
}
|
||
if "_voting" in strategy_result:
|
||
vi = strategy_result["_voting"]
|
||
analysis_data["voting"] = {
|
||
"enabled": vi.get("enabled"),
|
||
"models_used": vi.get("models_used"),
|
||
"strategic_votes": vi.get("strategic_votes"),
|
||
"total_votes": vi.get("total_votes"),
|
||
"unanimous": vi.get("unanimous"),
|
||
}
|
||
new_signal.strategic_analysis = json.dumps(analysis_data)
|
||
new_signal.tasks_generated = json.dumps(strategy_result.get("tasks", []))
|
||
new_signal.processed = True
|
||
await db.commit()
|
||
processed_count += 1
|
||
|
||
if initial_priority in [SignalPriority.MEDIUM, SignalPriority.HIGH]:
|
||
background_tasks.add_task(process_fetched)
|
||
else:
|
||
await process_fetched()
|
||
|
||
_stop_ollama()
|
||
|
||
return {
|
||
"status": "success",
|
||
"new_signals_count": new_signals_count,
|
||
"processed_count": processed_count,
|
||
"message": f"Fetched {new_signals_count} signals, {processed_count} analyzed",
|
||
}
|
||
|
||
|
||
@router.get("/{signal_id}/projects")
|
||
async def get_signal_projects(
|
||
signal_id: str,
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
"""获取信号关联的项目"""
|
||
try:
|
||
sid = int(signal_id)
|
||
except (ValueError, TypeError):
|
||
raise HTTPException(status_code=400, detail="Invalid signal_id")
|
||
|
||
signal_result = await db.execute(select(Signal).where(Signal.id == sid))
|
||
if not signal_result.scalar_one_or_none():
|
||
raise HTTPException(status_code=404, detail="Signal not found")
|
||
|
||
result = await db.execute(
|
||
select(Project, ProjectSignalLink)
|
||
.join(ProjectSignalLink, Project.id == ProjectSignalLink.project_id)
|
||
.where(ProjectSignalLink.signal_id == sid)
|
||
)
|
||
rows = result.all()
|
||
|
||
return [
|
||
{
|
||
"id": str(p.id),
|
||
"title": p.title,
|
||
"status": p.status,
|
||
"assigned_role": p.assigned_role,
|
||
"link_type": link.link_type,
|
||
"relevance_score": link.relevance_score,
|
||
}
|
||
for p, link in rows
|
||
]
|
||
|
||
|
||
@router.post("/{signal_id}/feedback", response_model=SignalFeedbackResponse)
|
||
async def submit_signal_feedback(
|
||
signal_id: int,
|
||
feedback: SignalFeedbackCreate,
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
"""提交信号反馈"""
|
||
result = await db.execute(select(Signal).where(Signal.id == signal_id))
|
||
signal = result.scalar_one_or_none()
|
||
if not signal:
|
||
raise HTTPException(status_code=404, detail="Signal not found")
|
||
|
||
ai_predicted_strategic = False
|
||
ai_confidence = 0.0
|
||
if signal.strategic_analysis:
|
||
try:
|
||
analysis = json.loads(signal.strategic_analysis)
|
||
ai_predicted_strategic = analysis.get("is_strategic", signal.impact_score >= 0.9)
|
||
ai_confidence = analysis.get("confidence", 0.0)
|
||
except Exception:
|
||
ai_predicted_strategic = signal.impact_score >= 0.9
|
||
|
||
new_feedback = SignalFeedback(
|
||
signal_id=signal_id,
|
||
was_strategic=feedback.was_strategic,
|
||
reason=feedback.reason,
|
||
user_id=feedback.user_id,
|
||
ai_predicted_strategic=ai_predicted_strategic,
|
||
ai_confidence=ai_confidence,
|
||
)
|
||
db.add(new_feedback)
|
||
await db.commit()
|
||
await db.refresh(new_feedback)
|
||
|
||
is_correct = ai_predicted_strategic == feedback.was_strategic
|
||
|
||
return SignalFeedbackResponse(
|
||
id=new_feedback.id,
|
||
signal_id=new_feedback.signal_id,
|
||
was_strategic=new_feedback.was_strategic,
|
||
ai_predicted_strategic=new_feedback.ai_predicted_strategic,
|
||
ai_confidence=new_feedback.ai_confidence,
|
||
is_correct=is_correct,
|
||
)
|
||
|
||
|
||
@router.get("/{signal_id}/feedbacks", response_model=List[SignalFeedbackResponse])
|
||
async def get_signal_feedbacks(
|
||
signal_id: int,
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
"""获取某信号的全部反馈"""
|
||
result = await db.execute(
|
||
select(SignalFeedback)
|
||
.where(SignalFeedback.signal_id == signal_id)
|
||
.order_by(SignalFeedback.created_at.desc())
|
||
)
|
||
feedbacks = result.scalars().all()
|
||
|
||
return [
|
||
SignalFeedbackResponse(
|
||
id=f.id,
|
||
signal_id=f.signal_id,
|
||
was_strategic=f.was_strategic,
|
||
ai_predicted_strategic=f.ai_predicted_strategic,
|
||
ai_confidence=f.ai_confidence or 0.0,
|
||
is_correct=f.ai_predicted_strategic == f.was_strategic,
|
||
)
|
||
for f in feedbacks
|
||
]
|