update

app.py

@@ -0,0 +1,145 @@
+import logging
+import os
+from datetime import datetime, timedelta
+import gradio as gr
+import json
+from log_reader import RemoteLogReader
+
+# logging.basicConfig(level=logging.INFO)
+# log = logging.getLogger(__name__)
+
+def get_file_data(content: str) -> tuple[str, bool]:
+    """Read file content and return IP and vote condition status"""
+    try:
+        lines = [line.strip() for line in content.split('\n') if line.strip()]
+        if not lines:
+            return None, False
+        
+        # Get IP from first line
+        try:
+            first_line_data = json.loads(lines[0])
+            ip = first_line_data.get('ip')
+        except json.JSONDecodeError:
+            ip = None
+        
+        # Check vote conditions from last line
+        try:
+            last_line_data = json.loads(lines[-1])
+            feedback = last_line_data.get('feedback')
+            vote_conditions_met = (last_line_data.get('type') == 'vote' and 
+                                isinstance(feedback, dict) and 
+                                len(feedback) == 6)
+        except json.JSONDecodeError:
+            vote_conditions_met = False
+            
+        return ip, vote_conditions_met
+    except Exception as e:
+        logging.error(f"Error processing file content: {e}")
+        return None, False
+
+def search_battle_anony(date_str: str, search_query: str) -> list[list]:
+    """Search battle_anony conversations for a specific date and query."""
+    results = []
+    
+    try:
+        # Initialize RemoteLogReader
+        reader = RemoteLogReader()
+        
+        # Get conversation logs for battle_anony mode
+        conv_logs = reader.get_conv_logs(date_str)
+        battle_anony_logs = conv_logs.get('battle_anony', {})
+        
+        # Process each conversation
+        for conv_id, logs in battle_anony_logs.items():
+            found_query = False
+            ip = None
+            
+            # Convert messages to file content format for validation check
+            content = '\n'.join(json.dumps(msg) for msg in logs)
+            ip, is_valid = get_file_data(content)
+            
+            if not ip:
+                continue
+                
+            # Search through messages for the query
+            for log in logs:
+                if "state" in log and "messages" in log["state"]:
+                    messages = log["state"]["messages"]
+                    
+                    # Search through each message
+                    for _, message in messages:
+                        if search_query in message:
+                            found_query = True
+                            break
+                
+                if found_query:
+                    break
+            
+            if found_query:
+                # Convert to list format instead of dict
+                results.append([
+                    ip,
+                    conv_id,
+                    'Yes' if is_valid else 'No',
+                    'Yes'
+                ])
+    
+    except Exception as e:
+        logging.error(f"Error searching logs for date {date_str}: {e}")
+    
+    return results
+
+def create_ui():
+    def process_search(date, query):
+        if not query.strip():
+            return []
+            
+        # Convert timestamp to datetime object
+        date_obj = datetime.fromtimestamp(date)
+        # Convert date to required format (YYYY_MM_DD)
+        date_str = date_obj.strftime("%Y_%m_%d")
+        results = search_battle_anony(date_str, query)
+        
+        if not results:
+            # Return empty list with correct structure for DataFrame
+            return []
+        return results
+    
+    with gr.Blocks(title="Battle Search") as app:
+        gr.Markdown("# Battle Search")
+        gr.Markdown("Search for specific content in battle_anony conversations by date and view IP addresses with validation status.")
+        
+        with gr.Row():
+            date_input = gr.DateTime(
+                label="Select Date",
+                include_time=False,
+            )
+            
+            query_input = gr.Textbox(
+                label="Search Query",
+                placeholder="Enter text to search for in conversations...",
+                lines=1
+            )
+        
+        with gr.Row():
+            search_button = gr.Button("Search")
+        
+        with gr.Row():
+            table_output = gr.DataFrame(
+                headers=['IP', 'Conversation ID', 'Is Valid', 'Contains Query'],
+                label="Results Table",
+                interactive=False,
+                value=[]  # Initialize with empty list
+            )
+        
+        search_button.click(
+            fn=process_search,
+            inputs=[date_input, query_input],
+            outputs=[table_output]
+        )
+    
+    return app
+
+if __name__ == "__main__":
+    app = create_ui()
+    app.launch()

log_reader.py

@@ -0,0 +1,118 @@
+'''
+Facade for reading logs on remote storage.
+'''
+
+from collections import defaultdict
+import json
+import os
+from typing import Any
+from azure.storage.fileshare import ShareServiceClient
+
+
+class RemoteLogReader:
+    '''
+    remote log reader
+    '''
+
+    LOG_CONNECTION_STRING = os.getenv("AZURE_STORAGE_CONNECTION_STRING") or ""
+    LOG_SHARE_NAME = "swearenalogsfileshare"
+
+    IMAGE_DIR_NAME = "serve_images"
+    '''
+    Directory for storing user uploaded images.
+    '''
+    CONV_LOG_DIR_NAME = "conv_logs"
+    '''
+    Directory for conversation logs.
+    '''
+    SANDBOX_LOG_DIR_NAME = "sandbox_logs"
+    '''
+    Directory for sandbox logs.
+    '''
+
+    CHAT_MODES = ["battle_anony", "battle_named", "direct"]
+
+    def __init__(
+        self,
+        connection_string: str = LOG_CONNECTION_STRING,
+        share_name: str = LOG_SHARE_NAME,
+    ):
+        if not connection_string:
+            raise ValueError("Connection string is required.")
+        if not share_name:
+            raise ValueError("Share name is required.")
+
+        self.share_service = ShareServiceClient.from_connection_string(
+            conn_str=connection_string)
+        self.share_client = self.share_service.get_share_client(share=share_name)
+
+    def is_conv_log(self, file_name: str) -> bool:
+        return file_name.startswith("conv-log") and file_name.endswith(".json")
+
+    def get_conv_id_from_name(self, file_name: str) -> str:
+        return file_name.split("-")[2].strip('.json')
+
+    def is_sandbox_log(self, file_name: str) -> bool:
+        return file_name.startswith("sandbox-log") and file_name.endswith(".json")
+
+    def get_file_content(self, file_path: str) -> bytes:
+        file_client = self.share_client.get_file_client(file_path)
+        file_content = file_client.download_file().readall()
+        return file_content
+
+    def get_conv_logs(self, date: str) -> dict[str, defaultdict[str, list[Any]]]:
+        '''
+        Return conversation logs based on the date.
+        Returns a dict:
+            mode -> conv_id -> list of logs.
+        '''
+        conv_logs = {
+            mode: defaultdict(list) for mode in self.CHAT_MODES
+        }
+        for mode in self.CHAT_MODES:
+            conv_log_dir = f"{date}/{self.CONV_LOG_DIR_NAME}/{mode}/"
+            # check if the directory exists
+            if not self.share_client.get_directory_client(conv_log_dir).exists():
+                continue
+            for file in self.share_client.list_directories_and_files(conv_log_dir):
+                if not self.is_conv_log(file.name):
+                    continue
+                conv_id = self.get_conv_id_from_name(file.name)
+                file_content = self.get_file_content(
+                    conv_log_dir + file.name).decode("utf-8").strip(' \n')
+                for line in file_content.split('\n'):
+                    if line:
+                        conv_logs[mode][conv_id].append(json.loads(line))
+        return conv_logs
+
+    def get_sandbox_logs(self, date: str) -> list[str]:
+        '''
+        Return sandbox logs based on the date.
+        '''
+        sandbox_logs = []
+        sandbox_log_dir = f"{date}/{self.SANDBOX_LOG_DIR_NAME}/"
+        for file in self.share_client.list_directories_and_files(sandbox_log_dir):
+            if self.is_sandbox_log(file.name):
+                file_content = self.get_file_content(
+                    sandbox_log_dir + file.name).decode("utf-8").strip(' \n')
+                sandbox_logs.append(json.loads(file_content))
+        return sandbox_logs
+
+    def get_image(self, image_id: str) -> bytes:
+        '''
+        Return image data based on the image id.
+        '''
+        image_path = f"{self.IMAGE_DIR_NAME}/{image_id}.png"
+        return self.get_file_content(image_path)
+
+
+if __name__ == "__main__":
+    # Example usages
+    log_reader = RemoteLogReader()
+    date = "2025_02_20"
+    conv_logs = log_reader.get_conv_logs(date)
+    sandbox_logs = log_reader.get_sandbox_logs(date)
+    image_data = log_reader.get_image("051fdac24285ff6e219a9ba06d1ac843")
+    print(conv_logs)
+    print(sandbox_logs)
+    print(image_data)

requirements.txt

@@ -0,0 +1,3 @@
+gradio>=4.0.0
+pandas>=2.0.0
+azure-storage-file-share