Update app.py

app.py

@@ -2,22 +2,153 @@ import gradio as gr
 import json
 import requests
 import os
+import pandas as pd
+import folium
+from geopy.geocoders import Nominatim
+from geopy.exc import GeocoderTimedOut, GeocoderServiceError
+import time
+import random
+from typing import List, Tuple, Optional
+import tempfile
+import io
 
-# Use the Hugging Face Inference API instead of loading the model
+# NuExtract API configuration
 API_URL = "https://api-inference.huggingface.co/models/numind/NuExtract-1.5"
 headers = {"Authorization": f"Bearer {os.environ.get('HF_TOKEN', '')}"}
 
-def test_function(template, text):
-    print(f"Test function called with template: {template[:30]} and text: {text[:30]}")
-    return "Button clicked successfully", "Function was called"
+# Geocoding Service
+class GeocodingService:
+    def __init__(self, user_agent: str = None, timeout: int = 10, rate_limit: float = 1.1):
+        if user_agent is None:
+            user_agent = f"python_geocoding_script_{random.randint(1000, 9999)}"
 
+        self.geolocator = Nominatim(
+            user_agent=user_agent,
+            timeout=timeout
+        )
+        self.rate_limit = rate_limit
+        self.last_request = 0
+
+    def _rate_limit_wait(self):
+        current_time = time.time()
+        time_since_last = current_time - self.last_request
+        if time_since_last < self.rate_limit:
+            time.sleep(self.rate_limit - time_since_last)
+        self.last_request = time.time()
+
+    def geocode_location(self, location: str, max_retries: int = 3) -> Optional[Tuple[float, float]]:
+        for attempt in range(max_retries):
+            try:
+                self._rate_limit_wait()
+                location_data = self.geolocator.geocode(location)
+                if location_data:
+                    return (location_data.latitude, location_data.longitude)
+                return None
+            except (GeocoderTimedOut, GeocoderServiceError) as e:
+                if attempt == max_retries - 1:
+                    print(f"Failed to geocode '{location}' after {max_retries} attempts: {e}")
+                    return None
+                time.sleep(2 ** attempt)  # Exponential backoff
+            except Exception as e:
+                print(f"Error geocoding '{location}': {e}")
+                return None
+        return None
+
+    def process_locations(self, locations: str) -> List[Optional[Tuple[float, float]]]:
+        if pd.isna(locations) or not locations:
+            return []
+
+        location_list = [loc.strip() for loc in locations.split(',')]
+        return [self.geocode_location(loc) for loc in location_list]
+
+# Mapping Functions
+def create_location_map(df: pd.DataFrame,
+                       coordinates_col: str = 'coordinates',
+                       places_col: str = 'places',
+                       title_col: Optional[str] = None) -> folium.Map:
+    # Initialize the map
+    m = folium.Map(location=[0, 0], zoom_start=2)
+    all_coords = []
+
+    # Process each row in the DataFrame
+    for idx, row in df.iterrows():
+        coordinates = row[coordinates_col]
+        places = row[places_col].split(',') if pd.notna(row[places_col]) else []
+        title = row[title_col] if title_col and pd.notna(row[title_col]) else None
+
+        # Skip if no coordinates
+        if not coordinates:
+            continue
+
+        # Add individual markers for each location
+        for i, (coord, place) in enumerate(zip(coordinates, places)):
+            if coord is not None:  # Skip None coordinates
+                lat, lon = coord
+                place_name = place.strip()
+
+                # Create popup content
+                popup_content = f"<b>{place_name}</b>"
+                if title:
+                    popup_content += f"<br>{title}"
+
+                # Add marker to the map
+                folium.Marker(
+                    location=[lat, lon],
+                    popup=folium.Popup(popup_content, max_width=300),
+                    tooltip=place_name,
+                ).add_to(m)
+
+                all_coords.append([lat, lon])
+
+    # If we have coordinates, fit the map bounds to include all points
+    if all_coords:
+        m.fit_bounds(all_coords)
+
+    return m
+
+# Processing Functions
+def process_excel(file, places_column):
+    # Read the Excel file
+    df = pd.read_excel(io.BytesIO(file))
+    
+    if places_column not in df.columns:
+        return None, f"Column '{places_column}' not found in the Excel file. Available columns: {', '.join(df.columns)}"
+    
+    # Initialize the geocoding service
+    geocoder = GeocodingService(user_agent="gradio_map_visualization_app")
+    
+    # Process locations and add coordinates
+    df['coordinates'] = df[places_column].apply(geocoder.process_locations)
+    
+    # Create the map
+    map_obj = create_location_map(df, coordinates_col='coordinates', places_col=places_column)
+    
+    # Save the map to a temporary HTML file
+    temp_map_path = "temp_map.html"
+    map_obj.save(temp_map_path)
+    
+    # Save the processed DataFrame to Excel
+    processed_file_path = "processed_data.xlsx"
+    df.to_excel(processed_file_path, index=False)
+    
+    # Statistics
+    total_locations = len(df)
+    successful_geocodes = df['coordinates'].apply(lambda x: len([c for c in x if c is not None])).sum()
+    failed_geocodes = df['coordinates'].apply(lambda x: len([c for c in x if c is None])).sum()
+    
+    stats = f"Total locations: {total_locations}\n"
+    stats += f"Successfully geocoded: {successful_geocodes}\n"
+    stats += f"Failed to geocode: {failed_geocodes}"
+    
+    return temp_map_path, stats, processed_file_path
+
+# NuExtract Functions
 def extract_info(template, text):
     try:
         # Format prompt according to NuExtract-1.5 requirements
         prompt = f"<|input|>\n### Template:\n{template}\n### Text:\n{text}\n\n<|output|>"
-        print(f"Processing with prompt: {prompt[:100]}...")
         
-        # Call API instead of using local model
+        # Call API
         payload = {
             "inputs": prompt,
             "parameters": {
@@ -26,11 +157,9 @@ def extract_info(template, text):
             }
         }
         
-        print("Calling API...")
         response = requests.post(API_URL, headers=headers, json=payload)
         
         if response.status_code != 200:
-            print(f"API error: {response.status_code}, {response.text}")
             return f"❌ API Error: {response.status_code}", response.text
         
         # Process result
@@ -49,56 +178,88 @@ def extract_info(template, text):
             json_text = result_text
         
         # Try to parse as JSON
-        print("Parsing JSON...")
         try:
             extracted = json.loads(json_text)
             formatted = json.dumps(extracted, indent=2)
         except json.JSONDecodeError:
-            print(f"JSON parsing failed. Raw output: {json_text[:100]}...")
             return "❌ JSON parsing error", json_text
             
         return "✅ Success", formatted
     except Exception as e:
-        print(f"Error in extraction: {str(e)}")
         return f"❌ Error: {str(e)}", "{}"
 
-# Create a simple interface
+# Create the Gradio interface
 with gr.Blocks() as demo:
-    gr.Markdown("# NuExtract-1.5 Extraction Tool")
+    gr.Markdown("# Historical Data Analysis Tools")
     
-    with gr.Row():
-        with gr.Column():
-            template = gr.Textbox(
-                label="JSON Template", 
-                value='{"name": "", "email": ""}',
-                lines=5
+    with gr.Tabs():
+        with gr.TabItem("Text Extraction"):
+            gr.Markdown("## NuExtract-1.5 Structured Data Extraction")
+            
+            with gr.Row():
+                with gr.Column():
+                    template = gr.Textbox(
+                        label="JSON Template", 
+                        value='{"name": "", "email": ""}',
+                        lines=5
+                    )
+                    text = gr.Textbox(
+                        label="Text to Extract From",
+                        value="Contact: John Smith ([email protected])",
+                        lines=8
+                    )
+                    extract_btn = gr.Button("Extract Information", variant="primary")
+                
+                with gr.Column():
+                    status = gr.Textbox(label="Status")
+                    output = gr.Textbox(label="Output", lines=10)
+            
+            extract_btn.click(
+                fn=extract_info,
+                inputs=[template, text],
+                outputs=[status, output]
             )
-            text = gr.Textbox(
-                label="Text to Extract From",
-                value="Contact: John Smith ([email protected])",
-                lines=8
+        
+        with gr.TabItem("Geocoding & Mapping"):
+            gr.Markdown("## Location Mapping Tool")
+            
+            with gr.Row():
+                with gr.Column():
+                    excel_file = gr.File(label="Upload Excel File")
+                    places_column = gr.Textbox(label="Places Column Name", value="places")
+                    process_btn = gr.Button("Process and Map", variant="primary")
+                
+                with gr.Column():
+                    map_output = gr.HTML(label="Map Visualization")
+                    stats_output = gr.Textbox(label="Statistics", lines=3)
+                    download_btn = gr.Button("Download Processed Data")
+                    processed_file = gr.File(label="Processed Data", visible=False)
+            
+            def process_and_map(file, column):
+                if file is None:
+                    return None, "Please upload an Excel file", None
+                
+                map_path, stats, processed_path = process_excel(file, column)
+                
+                if map_path:
+                    with open(map_path, "r") as f:
+                        map_html = f.read()
+                    
+                    return map_html, stats, processed_path
+                else:
+                    return None, stats, None
+            
+            process_btn.click(
+                fn=process_and_map,
+                inputs=[excel_file, places_column],
+                outputs=[map_output, stats_output, processed_file]
             )
             
-            # Two buttons for testing
-            test_btn = gr.Button("Test Click")
-            extract_btn = gr.Button("Extract Information", variant="primary")
-        
-        with gr.Column():
-            status = gr.Textbox(label="Status")
-            output = gr.Textbox(label="Output", lines=10)
-    
-    # Connect both buttons
-    test_btn.click(
-        fn=test_function,
-        inputs=[template, text],
-        outputs=[status, output]
-    )
-    
-    extract_btn.click(
-        fn=extract_info,
-        inputs=[template, text],
-        outputs=[status, output]
-    )
+            download_btn.click(
+                fn=lambda x: x,
+                inputs=[processed_file],
+                outputs=[processed_file]
+            )
 
 if __name__ == "__main__":
     demo.launch()