fix(monitoring): WeChat Work feedback recording bugs (#2108)

* fix(monitoring): fix WeChat Work feedback recording bugs

- Fix feedback events silently dropped when stream session expires:
  dispatch feedback handlers regardless of session availability
- Fix IntegrityError on repeated feedback (like→dislike) for same
  message: implement UPSERT logic in record_feedback()
- Fix cancel feedback (type=3) not removing records: add delete logic
- Fix inaccurate_reasons validation error: convert int reason codes
  to strings before creating FeedbackEvent (Pydantic expects List[str])
- Fix feedback timestamps 8 hours off in frontend: use parseUTCTimestamp
  instead of new Date() for UTC timestamp parsing
- Fix StreamSessionManager.cleanup missing _feedback_index cleanup

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* fix(monitoring): apply ruff format to wecom feedback files

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

---------

Co-authored-by: 6mvp6 <13727783693@163.com>
Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
6mvp6
2026-04-06 17:12:43 +08:00
committed by GitHub
parent 8f1317b39e
commit 83ccb33fd3
4 changed files with 109 additions and 46 deletions

View File

@@ -228,6 +228,9 @@ class StreamSessionManager:
msg_id = session.msg_id msg_id = session.msg_id
if msg_id and self._msg_index.get(msg_id) == stream_id: if msg_id and self._msg_index.get(msg_id) == stream_id:
self._msg_index.pop(msg_id, None) self._msg_index.pop(msg_id, None)
# Clean up feedback index for expired sessions
if session.feedback_id:
self._feedback_index.pop(session.feedback_id, None)
def _decrypt_file(encrypted_data: bytes, aes_key_str: str) -> bytes: def _decrypt_file(encrypted_data: bytes, aes_key_str: str) -> bytes:
@@ -903,35 +906,38 @@ class WecomBotClient:
) )
session = self.stream_sessions.get_session_by_feedback_id(feedback_id) session = self.stream_sessions.get_session_by_feedback_id(feedback_id)
if session: if session:
await self.logger.info( await self.logger.info(
f'反馈关联到会话: stream_id={session.stream_id}, msg_id={session.msg_id}, user_id={session.user_id}' f'反馈关联到会话: stream_id={session.stream_id}, msg_id={session.msg_id}, user_id={session.user_id}'
) )
for handler in self._message_handlers.get('feedback', []):
try:
await handler(
feedback_id=feedback_id,
feedback_type=feedback_type,
feedback_content=feedback_content,
inaccurate_reasons=inaccurate_reasons,
session=session,
)
except Exception:
await self.logger.error(traceback.format_exc())
if self._feedback_callback:
try:
await self._feedback_callback(
feedback_id=feedback_id,
feedback_type=feedback_type,
feedback_content=feedback_content,
inaccurate_reasons=inaccurate_reasons,
session=session,
)
except Exception:
await self.logger.error(traceback.format_exc())
else: else:
await self.logger.warning(f'未找到 feedback_id={feedback_id} 对应的会话') await self.logger.warning(f'未找到 feedback_id={feedback_id} 对应的会话,仍将记录反馈')
# Dispatch feedback event regardless of session availability
for handler in self._message_handlers.get('feedback', []):
try:
await handler(
feedback_id=feedback_id,
feedback_type=feedback_type,
feedback_content=feedback_content,
inaccurate_reasons=inaccurate_reasons,
session=session,
)
except Exception:
await self.logger.error(traceback.format_exc())
if self._feedback_callback:
try:
await self._feedback_callback(
feedback_id=feedback_id,
feedback_type=feedback_type,
feedback_content=feedback_content,
inaccurate_reasons=inaccurate_reasons,
session=session,
)
except Exception:
await self.logger.error(traceback.format_exc())
except Exception: except Exception:
await self.logger.error(traceback.format_exc()) await self.logger.error(traceback.format_exc())

View File

@@ -1224,30 +1224,83 @@ class MonitoringService:
""" """
import json import json
record_id = str(uuid.uuid4()) now = datetime.datetime.now(datetime.timezone.utc).replace(tzinfo=None)
record_data = { reasons_json = json.dumps(inaccurate_reasons, ensure_ascii=False) if inaccurate_reasons else None
'id': record_id,
'timestamp': datetime.datetime.now(datetime.timezone.utc).replace(tzinfo=None),
'feedback_id': feedback_id,
'feedback_type': feedback_type,
'feedback_content': feedback_content,
'inaccurate_reasons': json.dumps(inaccurate_reasons, ensure_ascii=False) if inaccurate_reasons else None,
'bot_id': bot_id,
'bot_name': bot_name,
'pipeline_id': pipeline_id,
'pipeline_name': pipeline_name,
'session_id': session_id,
'message_id': message_id,
'stream_id': stream_id,
'user_id': user_id,
'platform': platform,
}
await self.ap.persistence_mgr.execute_async( MonitoringFeedback = persistence_monitoring.MonitoringFeedback
sqlalchemy.insert(persistence_monitoring.MonitoringFeedback).values(record_data)
# Handle cancel feedback (type=3): delete existing record
if feedback_type == 3:
await self.ap.persistence_mgr.execute_async(
sqlalchemy.delete(MonitoringFeedback).where(MonitoringFeedback.feedback_id == feedback_id)
)
return None
# Check if record with this feedback_id already exists
existing_result = await self.ap.persistence_mgr.execute_async(
sqlalchemy.select(MonitoringFeedback).where(MonitoringFeedback.feedback_id == feedback_id)
) )
existing_row = existing_result.first()
return record_id if existing_row:
# UPDATE existing record
existing = existing_row[0] if isinstance(existing_row, tuple) else existing_row
await self.ap.persistence_mgr.execute_async(
sqlalchemy.update(MonitoringFeedback)
.where(MonitoringFeedback.feedback_id == feedback_id)
.values(
timestamp=now,
feedback_type=feedback_type,
feedback_content=feedback_content,
inaccurate_reasons=reasons_json,
bot_id=bot_id or existing.bot_id,
bot_name=bot_name or existing.bot_name,
pipeline_id=pipeline_id or existing.pipeline_id,
pipeline_name=pipeline_name or existing.pipeline_name,
session_id=session_id or existing.session_id,
message_id=message_id or existing.message_id,
stream_id=stream_id or existing.stream_id,
user_id=user_id or existing.user_id,
platform=platform or existing.platform,
)
)
return existing.id
else:
# INSERT new record with IntegrityError defense
record_id = str(uuid.uuid4())
record_data = {
'id': record_id,
'timestamp': now,
'feedback_id': feedback_id,
'feedback_type': feedback_type,
'feedback_content': feedback_content,
'inaccurate_reasons': reasons_json,
'bot_id': bot_id,
'bot_name': bot_name,
'pipeline_id': pipeline_id,
'pipeline_name': pipeline_name,
'session_id': session_id,
'message_id': message_id,
'stream_id': stream_id,
'user_id': user_id,
'platform': platform,
}
try:
await self.ap.persistence_mgr.execute_async(sqlalchemy.insert(MonitoringFeedback).values(record_data))
return record_id
except Exception:
# UNIQUE constraint conflict (concurrent feedback for same feedback_id)
await self.ap.persistence_mgr.execute_async(
sqlalchemy.update(MonitoringFeedback)
.where(MonitoringFeedback.feedback_id == feedback_id)
.values(
timestamp=now,
feedback_type=feedback_type,
feedback_content=feedback_content,
inaccurate_reasons=reasons_json,
)
)
return feedback_id
async def get_feedback_stats( async def get_feedback_stats(
self, self,

View File

@@ -328,6 +328,9 @@ class WecomBotAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter):
feedback_type = kwargs.get('feedback_type', 0) feedback_type = kwargs.get('feedback_type', 0)
feedback_content = kwargs.get('feedback_content', '') or None feedback_content = kwargs.get('feedback_content', '') or None
inaccurate_reasons = kwargs.get('inaccurate_reasons', []) or None inaccurate_reasons = kwargs.get('inaccurate_reasons', []) or None
# WeChat Work returns integer reason codes, but FeedbackEvent expects strings
if inaccurate_reasons:
inaccurate_reasons = [str(r) for r in inaccurate_reasons]
session = kwargs.get('session') session = kwargs.get('session')
session_id = None session_id = None

View File

@@ -1,6 +1,7 @@
import { useState, useEffect, useCallback, useMemo } from 'react'; import { useState, useEffect, useCallback, useMemo } from 'react';
import { httpClient } from '@/app/infra/http'; import { httpClient } from '@/app/infra/http';
import { FeedbackRecord, FeedbackStats } from '../types/monitoring'; import { FeedbackRecord, FeedbackStats } from '../types/monitoring';
import { parseUTCTimestamp } from '../utils/dateUtils';
interface UseFeedbackDataParams { interface UseFeedbackDataParams {
botIds?: string[]; botIds?: string[];
@@ -142,7 +143,7 @@ export function useFeedbackData(params: UseFeedbackDataParams = {}) {
const transformedFeedback: FeedbackRecord[] = result.feedback.map( const transformedFeedback: FeedbackRecord[] = result.feedback.map(
(item) => ({ (item) => ({
id: item.id, id: item.id,
timestamp: new Date(item.timestamp), timestamp: parseUTCTimestamp(item.timestamp),
feedbackId: item.feedback_id, feedbackId: item.feedback_id,
feedbackType: item.feedback_type === 1 ? 'like' : 'dislike', feedbackType: item.feedback_type === 1 ? 'like' : 'dislike',
feedbackContent: item.feedback_content, feedbackContent: item.feedback_content,