424 lines
13 KiB
Python
424 lines
13 KiB
Python
"""
|
||
Proactive Blocker Detection Service - 主动式阻塞检测服务
|
||
|
||
核心功能：
|
||
1. 扫描代码注释中的阻塞关键词
|
||
2. 自动识别常见的开发阻塞模式
|
||
3. AI 生成解决方案建议
|
||
4. 追踪阻塞解决进度
|
||
|
||
支持的阻塞模式：
|
||
- "等待 X 接口完成"
|
||
- "这个 API 达不到效果"
|
||
- "由于 Y 的限制，暂时跳过"
|
||
- "TODO: 等待 Z 确定"
|
||
"""
|
||
import asyncio
|
||
import json
|
||
import re
|
||
from datetime import datetime, timezone, timedelta
|
||
from typing import Any, Dict, List, Optional
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
from sqlalchemy.future import select
|
||
from sqlalchemy import text
|
||
|
||
from ..models import (
|
||
User, Project, Task, Signal,
|
||
CognitiveConflict, TruthNode
|
||
)
|
||
from .llm import LLMService
|
||
|
||
|
||
# 阻塞模式正则表达式
|
||
BLOCKER_PATTERNS = [
|
||
r'等待.*接口',
|
||
r'等待.*API',
|
||
r'等待.*确定',
|
||
r'等待.*完成',
|
||
r'blocked by',
|
||
r'blocked due to',
|
||
r'can\'t proceed until',
|
||
r'暂时跳过',
|
||
r'TODO.*等待',
|
||
r'FIXME.*等待',
|
||
r'HACK.*临时',
|
||
r'达不到效果',
|
||
r'无法.*因为',
|
||
r'pending.*approval',
|
||
r'waiting for',
|
||
r'blocked on',
|
||
]
|
||
|
||
|
||
class BlockerDetectionService:
|
||
"""主动式阻塞检测服务"""
|
||
|
||
def __init__(self, db: AsyncSession):
|
||
self.db = db
|
||
self.llm = LLMService()
|
||
|
||
async def scan_code_comment(
|
||
self,
|
||
code_content: str,
|
||
file_path: str,
|
||
line_number: int,
|
||
project_id: int,
|
||
user_id: int,
|
||
tenant_id: int = 1
|
||
) -> Optional[Dict[str, Any]]:
|
||
"""
|
||
扫描代码注释，检测阻塞
|
||
|
||
Args:
|
||
code_content: 代码内容
|
||
file_path: 文件路径
|
||
line_number: 行号
|
||
project_id: 项目ID
|
||
user_id: 用户ID
|
||
tenant_id: 租户ID
|
||
|
||
Returns:
|
||
如果检测到阻塞，返回阻塞信息；否则返回 None
|
||
"""
|
||
# 检查是否匹配阻塞模式
|
||
detected_pattern = None
|
||
for pattern in BLOCKER_PATTERNS:
|
||
if re.search(pattern, code_content, re.IGNORECASE):
|
||
detected_pattern = pattern
|
||
break
|
||
|
||
if not detected_pattern:
|
||
return None
|
||
|
||
# AI 分析阻塞严重程度和生成解决方案
|
||
analysis = await self._analyze_blocker_with_ai(
|
||
code_content=code_content,
|
||
file_path=file_path,
|
||
pattern=detected_pattern,
|
||
tenant_id=tenant_id,
|
||
)
|
||
|
||
# 创建阻塞记录（复用 CognitiveConflict 模型）
|
||
from sqlalchemy import text
|
||
next_conflict_id = (await self.db.execute(text("SELECT COALESCE(MAX(id),0)+1 FROM cognitive_conflicts"))).scalar() or 1
|
||
|
||
blocker = CognitiveConflict(
|
||
id=next_conflict_id,
|
||
truth_node_id=None, # 阻塞可能不关联特定宪章
|
||
conflict_type='code_blocker',
|
||
description=f"代码阻塞检测：\n\n文件：{file_path}:{line_number}\n代码：{code_content[:200]}\n\nAI 分析：{analysis.get('problem_analysis', '')}",
|
||
involved_user_ids=json.dumps([user_id]),
|
||
status='detected',
|
||
tenant_id=tenant_id
|
||
)
|
||
self.db.add(blocker)
|
||
await self.db.commit()
|
||
|
||
# 自动生成任务建议
|
||
suggested_tasks = await self._generate_blocker_resolution_tasks(
|
||
blocker_id=blocker.id,
|
||
analysis=analysis,
|
||
project_id=project_id,
|
||
tenant_id=tenant_id
|
||
)
|
||
|
||
return {
|
||
'blocker_id': blocker.id,
|
||
'severity': analysis.get('severity', 'medium'),
|
||
'problem_analysis': analysis.get('problem_analysis', ''),
|
||
'suggested_solutions': analysis.get('suggested_solutions', []),
|
||
'suggested_tasks': suggested_tasks
|
||
}
|
||
|
||
async def detect_task_blockers(
|
||
self,
|
||
tenant_id: int = 1
|
||
) -> List[Dict[str, Any]]:
|
||
"""
|
||
检测所有任务中的阻塞
|
||
|
||
扫描所有进行中的任务，检测是否存在阻塞
|
||
"""
|
||
from sqlalchemy import text
|
||
|
||
# 获取所有进行中的任务
|
||
result = await self.db.execute(
|
||
text("""
|
||
SELECT t.id, t.title, t.description, t.status, t.project_id
|
||
FROM tasks t
|
||
WHERE t.status IN ('pending', 'in_progress')
|
||
AND t.tenant_id = :tenant_id
|
||
AND (t.updated_at < datetime('now', '-3 days')
|
||
OR t.updated_at < datetime('now', '-3 days', 'localtime'))
|
||
"""),
|
||
{"tenant_id": tenant_id}
|
||
)
|
||
|
||
tasks = result.fetchall()
|
||
|
||
detected_blockers = []
|
||
|
||
for task in tasks:
|
||
task_id, title, description, status, project_id = task
|
||
|
||
# AI 分析任务是否可能被阻塞
|
||
analysis = await self._analyze_task_for_blockers(
|
||
task_id=task_id,
|
||
title=title,
|
||
description=description,
|
||
tenant_id=tenant_id,
|
||
)
|
||
|
||
if analysis.get('is_blocked', False):
|
||
# 创建阻塞记录
|
||
next_conflict_id = (await self.db.execute(text("SELECT COALESCE(MAX(id),0)+1 FROM cognitive_conflicts"))).scalar() or 1
|
||
|
||
blocker = CognitiveConflict(
|
||
id=next_conflict_id,
|
||
truth_node_id=None,
|
||
conflict_type='task_blocker',
|
||
description=f"任务阻塞检测：\n\n任务：{title}\nID: {task_id}\n\nAI 分析：{analysis.get('reason', '')}",
|
||
involved_user_ids=json.dumps([]),
|
||
status='detected',
|
||
tenant_id=tenant_id
|
||
)
|
||
self.db.add(blocker)
|
||
await self.db.commit()
|
||
|
||
detected_blockers.append({
|
||
'blocker_id': blocker.id,
|
||
'task_id': task_id,
|
||
'task_title': title,
|
||
'severity': analysis.get('severity', 'medium'),
|
||
'reason': analysis.get('reason', ''),
|
||
'suggested_actions': analysis.get('suggested_actions', [])
|
||
})
|
||
|
||
return detected_blockers
|
||
|
||
async def resolve_blocker(
|
||
self,
|
||
blocker_id: int,
|
||
resolution_note: str,
|
||
tenant_id: int = 1
|
||
) -> Dict[str, Any]:
|
||
"""
|
||
标记阻塞为已解决
|
||
|
||
Args:
|
||
blocker_id: 阻塞ID
|
||
resolution_note: 解决方案说明
|
||
tenant_id: 租户ID
|
||
"""
|
||
result = await self.db.execute(
|
||
select(CognitiveConflict).where(
|
||
CognitiveConflict.id == blocker_id,
|
||
CognitiveConflict.tenant_id == tenant_id
|
||
)
|
||
)
|
||
blocker = result.scalar_one_or_none()
|
||
|
||
if not blocker:
|
||
raise ValueError(f"阻塞 {blocker_id} 不存在")
|
||
|
||
blocker.status = 'resolved'
|
||
blocker.resolved_at = datetime.now(timezone.utc)
|
||
blocker.description += f"\n\n解决方案：{resolution_note}"
|
||
|
||
await self.db.commit()
|
||
|
||
return {
|
||
'blocker_id': blocker_id,
|
||
'status': 'resolved',
|
||
'resolved_at': blocker.resolved_at.isoformat()
|
||
}
|
||
|
||
async def get_active_blockers(
|
||
self,
|
||
tenant_id: int = 1
|
||
) -> List[Dict[str, Any]]:
|
||
"""
|
||
获取所有活跃的阻塞
|
||
|
||
Returns:
|
||
阻塞列表，包含严重程度、描述等信息
|
||
"""
|
||
result = await self.db.execute(
|
||
select(CognitiveConflict).where(
|
||
CognitiveConflict.tenant_id == tenant_id,
|
||
CognitiveConflict.conflict_type.in_(['code_blocker', 'task_blocker']),
|
||
CognitiveConflict.status == 'detected'
|
||
)
|
||
)
|
||
blockers = result.scalars().all()
|
||
|
||
return [
|
||
{
|
||
'id': b.id,
|
||
'type': b.conflict_type,
|
||
'description': b.description,
|
||
'severity': self._extract_severity(b.description),
|
||
'detected_at': b.created_at.isoformat() if hasattr(b, 'created_at') else None,
|
||
'involved_users': json.loads(b.involved_user_ids) if b.involved_user_ids else []
|
||
}
|
||
for b in blockers
|
||
]
|
||
|
||
async def _analyze_blocker_with_ai(
|
||
self,
|
||
code_content: str,
|
||
file_path: str,
|
||
pattern: str,
|
||
tenant_id: int,
|
||
) -> Dict[str, Any]:
|
||
"""AI 分析代码阻塞"""
|
||
prompt = f"""分析以下代码注释中提到的阻塞问题：
|
||
|
||
文件：{file_path}
|
||
匹配到的模式：{pattern}
|
||
代码内容：
|
||
{code_content}
|
||
|
||
请分析：
|
||
1. 阻塞的根本原因是什么？
|
||
2. 严重程度（high/medium/low）？
|
||
3. 可能的解决方案有哪些？
|
||
4. 是否需要创建新任务来跟踪？
|
||
|
||
以 JSON 格式返回：
|
||
{{
|
||
"problem_analysis": "问题分析",
|
||
"severity": "high/medium/low",
|
||
"suggested_solutions": ["方案1", "方案2"],
|
||
"requires_new_task": boolean,
|
||
"suggested_task_title": "建议的任务标题"
|
||
}}"""
|
||
|
||
try:
|
||
response = await self.llm.generate(
|
||
prompt,
|
||
db=self.db,
|
||
tenant_id=tenant_id,
|
||
)
|
||
response = response.strip()
|
||
if response.startswith('```'):
|
||
response = response.split('```')[1]
|
||
if response.startswith('json'):
|
||
response = response[4:]
|
||
return json.loads(response.strip())
|
||
except:
|
||
return {
|
||
'problem_analysis': f'检测到可能的阻塞：{pattern}',
|
||
'severity': 'medium',
|
||
'suggested_solutions': ['请团队负责人评估'],
|
||
'requires_new_task': False
|
||
}
|
||
|
||
async def _analyze_task_for_blockers(
|
||
self,
|
||
task_id: int,
|
||
title: str,
|
||
description: str,
|
||
tenant_id: int,
|
||
) -> Dict[str, Any]:
|
||
"""AI 分析任务是否被阻塞"""
|
||
prompt = f"""分析以下任务是否可能被阻塞：
|
||
|
||
任务标题：{title}
|
||
任务描述：{description}
|
||
|
||
请判断：
|
||
1. 任务是否被阻塞？
|
||
2. 如果阻塞，原因是什么？
|
||
3. 严重程度（high/medium/low）？
|
||
4. 建议的下一步行动？
|
||
|
||
以 JSON 格式返回：
|
||
{{
|
||
"is_blocked": boolean,
|
||
"reason": "阻塞原因（如果阻塞）",
|
||
"severity": "high/medium/low",
|
||
"suggested_actions": ["行动1", "行动2"]
|
||
}}
|
||
|
||
如果任务没有被阻塞，is_blocked 设为 false。"""
|
||
|
||
try:
|
||
response = await self.llm.generate(
|
||
prompt,
|
||
db=self.db,
|
||
tenant_id=tenant_id,
|
||
)
|
||
response = response.strip()
|
||
if response.startswith('```'):
|
||
response = response.split('```')[1]
|
||
if response.startswith('json'):
|
||
response = response[4:]
|
||
return json.loads(response.strip())
|
||
except:
|
||
return {'is_blocked': False}
|
||
|
||
async def _generate_blocker_resolution_tasks(
|
||
self,
|
||
blocker_id: int,
|
||
analysis: Dict[str, Any],
|
||
project_id: int,
|
||
tenant_id: int
|
||
) -> List[Dict[str, Any]]:
|
||
"""生成阻塞解决任务"""
|
||
if not analysis.get('requires_new_task', False):
|
||
return []
|
||
|
||
from sqlalchemy import text
|
||
|
||
# 获取下一个任务ID
|
||
next_task_id = (await self.db.execute(text("SELECT COALESCE(MAX(id),0)+1 FROM tasks"))).scalar() or 1
|
||
|
||
task_title = analysis.get('suggested_task_title', f'解决阻塞 #{blocker_id}')
|
||
task_description = f"""自动生成的阻塞解决任务
|
||
|
||
阻塞原因：
|
||
{analysis.get('problem_analysis', '')}
|
||
|
||
建议的解决方案：
|
||
{chr(10).join(analysis.get('suggested_solutions', []))}"""
|
||
|
||
task = Task(
|
||
id=next_task_id,
|
||
truth_node_id=None, # 不关联特定宪章
|
||
project_id=project_id,
|
||
assignee_id=None,
|
||
title=task_title,
|
||
description=task_description,
|
||
required_skill_tag_ids=[],
|
||
status='pending',
|
||
priority='HIGH' if analysis.get('severity') == 'high' else 'MEDIUM',
|
||
estimated_hours=4,
|
||
actual_hours=0,
|
||
quality_score=0,
|
||
tenant_id=tenant_id
|
||
)
|
||
self.db.add(task)
|
||
await self.db.commit()
|
||
|
||
return [{
|
||
'task_id': task.id,
|
||
'title': task.title,
|
||
'priority': task.priority
|
||
}]
|
||
|
||
def _extract_severity(self, description: str) -> str:
|
||
"""从描述中提取严重程度"""
|
||
desc_lower = description.lower()
|
||
if 'high' in desc_lower or '严重' in desc_lower or 'critical' in desc_lower:
|
||
return 'high'
|
||
elif 'low' in desc_lower or '轻微' in desc_lower:
|
||
return 'low'
|
||
return 'medium'
|
||
|
||
|
||
# 全局服务实例
|
||
def get_blocker_detection_service(db: AsyncSession) -> BlockerDetectionService:
|
||
"""获取阻塞检测服务实例"""
|
||
return BlockerDetectionService(db)
|