iFlow 对话记录 - {start_time.strftime('%Y-%m-%d')}

import requests import json from datetime import datetime start_time = datetime.now().replace(hour=0, minute=0, second=0) start_ns = int(start_time.timestamp()) * 1_000_000_000 url = 'http://localhost:3100/loki/api/v1/query_range' params = { 'query': '{service_name="exec"}', 'start': start_ns, 'limit': 500 } response = requests.get(url, params=params) data = response.json() messages = [] if 'data' in data and 'result' in data['data']: for stream in data['data']['result']: for timestamp, log_entry in stream['values']: try: entry = json.loads(log_entry) message_str = entry.get('message', '') message = json.loads(message_str) msg_type = message.get('type', '') timestamp_str = message.get('timestamp', '') nested_message = message.get('message', {}) content = nested_message.get('content', '') if msg_type == 'user': if isinstance(content, str) and not content.startswith('Caveat:'): messages.append({ 'type': 'user', 'content': content, 'timestamp': timestamp_str }) elif msg_type == 'assistant': if isinstance(content, list) and content: text_content = [] for item in content: if isinstance(item, dict) and item.get('type') == 'text': text_content.append(item.get('text', '')) if text_content: messages.append({ 'type': 'assistant', 'content': ' '.join(text_content), 'timestamp': timestamp_str }) except: pass # 按时间排序 messages.sort(key=lambda x: x['timestamp']) # 生成HTML html = f""" iFlow 对话记录 - {start_time.strftime('%Y-%m-%d')}

{len([m for m in messages if m['type'] == 'user'])}

用户消息

{len([m for m in messages if m['type'] == 'assistant'])}

AI回复

{len(messages)}

总计

""" for msg in messages: if msg['type'] == 'user': html += f"""

{msg['content']}

{msg['timestamp'].split('T')[1][:8]}

👤

""" else: html += f"""

🤖

{msg['content'][:500]}

{msg['timestamp'].split('T')[1][:8]}

""" html += """

""" # 保存HTML文件 output_path = '/tmp/iflow_chat.html' with open(output_path, 'w', encoding='utf-8') as f: f.write(html) print(f'✅ 已生成对话记录: {output_path}') print(f'📊 共 {len(messages)} 条消息') print(f'👤 用户: {len([m for m in messages if m["type"] == "user"])} 条') print(f'🤖 AI: {len([m for m in messages if m["type"] == "assistant"])} 条')

📅 iFlow 对话记录