Merge pull request 'Fix batch processing for large conversation files' (#48) from fix/batch-processing into main
Run Tests / Backend Tests (push) Failing after 4m18s
Run Tests / Frontend Tests (push) Failing after 3m14s
Run Tests / Test Summary (push) Failing after 18s

This commit was merged in pull request #48.
This commit is contained in:
2026-04-12 11:49:30 +08:00
2 changed files with 15 additions and 9 deletions
+9 -5
View File
@@ -52,9 +52,13 @@ async def process_file(file_path: str):
# No new lines # No new lines
return 0 return 0
# Process in batches for large files (max 200 lines at a time)
batch_size = 200
end_line = min(start_line + batch_size, total_lines)
# Parse new content # Parse new content
logger.info(f"Processing {file_path} from line {start_line} to {total_lines}") logger.info(f"Processing {file_path} from line {start_line} to {end_line} (total: {total_lines})")
conversation_data = parser.parse_conversation_file(file_path, start_line) conversation_data = parser.parse_conversation_file(file_path, start_line, end_line)
if not conversation_data: if not conversation_data:
logger.warning(f"No data extracted from {file_path}") logger.warning(f"No data extracted from {file_path}")
@@ -87,10 +91,10 @@ async def process_file(file_path: str):
for tool_call in conversation_data['tool_calls']: for tool_call in conversation_data['tool_calls']:
await db.insert_tool_call(tool_call) await db.insert_tool_call(tool_call)
# Update checkpoint # Update checkpoint (use end_line instead of total_lines for batch processing)
await db.update_checkpoint(file_path, total_lines, file_mtime) await db.update_checkpoint(file_path, end_line, file_mtime)
logger.info(f"Processed {len(conversation_data['messages'])} messages from {file_path}") logger.info(f"Processed {len(conversation_data['messages'])} messages from {file_path} (batch {start_line}-{end_line}/{total_lines})")
return len(conversation_data['messages']) return len(conversation_data['messages'])
except Exception as e: except Exception as e:
+6 -4
View File
@@ -6,14 +6,16 @@ from typing import Dict, List, Any, Optional
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
def parse_jsonl_file(file_path: str, start_line: int = 0) -> List[Dict[str, Any]]: def parse_jsonl_file(file_path: str, start_line: int = 0, end_line: int = None) -> List[Dict[str, Any]]:
"""Parse JSONL file from a specific line number""" """Parse JSONL file from a specific line number to end_line (or EOF if None)"""
messages = [] messages = []
try: try:
with open(file_path, 'r', encoding='utf-8') as f: with open(file_path, 'r', encoding='utf-8') as f:
for i, line in enumerate(f): for i, line in enumerate(f):
if i < start_line: if i < start_line:
continue continue
if end_line is not None and i >= end_line:
break
if not line.strip(): if not line.strip():
continue continue
try: try:
@@ -169,9 +171,9 @@ def calculate_conversation_stats(messages: List[Dict[str, Any]]) -> Dict[str, An
return stats return stats
def parse_conversation_file(file_path: str, start_line: int = 0) -> Dict[str, Any]: def parse_conversation_file(file_path: str, start_line: int = 0, end_line: int = None) -> Dict[str, Any]:
"""Parse a conversation file and extract all data""" """Parse a conversation file and extract all data"""
raw_messages = parse_jsonl_file(file_path, start_line) raw_messages = parse_jsonl_file(file_path, start_line, end_line)
if not raw_messages: if not raw_messages:
return None return None