import gradio as gr import json import requests import os import pandas as pd import folium from folium.plugins import MeasureControl, Fullscreen, MarkerCluster from geopy.geocoders import Nominatim from geopy.exc import GeocoderTimedOut, GeocoderServiceError import time import random from typing import List, Tuple, Optional import io import tempfile import warnings warnings.filterwarnings("ignore") # Map Tile Providers with reliable sources MAP_TILES = { "GreenMap": { "url": "https://server.arcgisonline.com/ArcGIS/rest/services/World_Imagery/MapServer/tile/{z}/{y}/{x}", "attr": "Esri" } } # NuExtract API configuration API_URL = "https://api-inference.huggingface.co/models/numind/NuExtract-1.5" headers = {"Authorization": f"Bearer {os.environ.get('HF_TOKEN', '')}"} class SafeGeocoder: def __init__(self): user_agent = f"location_mapper_v1_{random.randint(1000, 9999)}" self.geolocator = Nominatim(user_agent=user_agent, timeout=10) self.cache = {} self.last_request = 0 def _respect_rate_limit(self): current_time = time.time() elapsed = current_time - self.last_request if elapsed < 1.0: time.sleep(1.0 - elapsed) self.last_request = time.time() def get_coords(self, location: str): if not location or pd.isna(location): return None location = str(location).strip() if location in self.cache: return self.cache[location] try: self._respect_rate_limit() result = self.geolocator.geocode(location) if result: coords = (result.latitude, result.longitude) self.cache[location] = coords return coords self.cache[location] = None return None except Exception as e: print(f"Geocoding error for '{location}': {e}") self.cache[location] = None return None # Function to just load the model def load_model(): try: # Send a minimal request just to trigger model loading payload = { "inputs": "<|input|>\n### Template:\n{\"test\": \"\"}\n### Text:\ntest\n\n<|output|>", "parameters": { "max_new_tokens": 10, "do_sample": False } } response = requests.post(API_URL, headers=headers, json=payload) if response.status_code == 503: response_json = response.json() if "error" in response_json and "loading" in response_json["error"]: estimated_time = response_json.get("estimated_time", "unknown") return f"⏳ Modell lädt... (ca. {int(float(estimated_time)) if isinstance(estimated_time, (int, float, str)) else 'unbekannt'} Sekunden)" if response.status_code != 200: return f"❌ API Fehler: {response.status_code}" return "✅ Modell erfolgreich geladen! Sie können jetzt mit der Extraktion beginnen." except Exception as e: return f"❌ Fehler: {str(e)}" def extract_info(template, text): try: prompt = f"<|input|>\n### Template:\n{template}\n### Text:\n{text}\n\n<|output|>" payload = { "inputs": prompt, "parameters": { "max_new_tokens": 1000, "do_sample": False } } response = requests.post(API_URL, headers=headers, json=payload) if response.status_code == 503: response_json = response.json() if "error" in response_json and "loading" in response_json["error"]: estimated_time = response_json.get("estimated_time", "unknown") return f"⏳ Modell lädt... (ca. {int(float(estimated_time)) if isinstance(estimated_time, (int, float, str)) else 'unbekannt'} Sekunden)", "Bitte versuchen Sie es in einigen Minuten erneut oder nutzen Sie den 'Modell laden' Button" if response.status_code != 200: return f"❌ API Fehler: {response.status_code}", response.text result = response.json() if isinstance(result, list) and len(result) > 0: result_text = result[0].get("generated_text", "") else: result_text = str(result) if "<|output|>" in result_text: json_text = result_text.split("<|output|>")[1].strip() else: json_text = result_text try: extracted = json.loads(json_text) formatted = json.dumps(extracted, indent=2) except json.JSONDecodeError: return "❌ JSON Parsing Fehler", json_text return "✅ Erfolgreich extrahiert", formatted except Exception as e: return f"❌ Fehler: {str(e)}", "{}" def create_map(df, location_col): m = folium.Map( location=[20, 0], zoom_start=2, control_scale=True ) folium.TileLayer( tiles=MAP_TILES["GreenMap"]["url"], attr=MAP_TILES["GreenMap"]["attr"], name="GreenMap", overlay=False, control=False ).add_to(m) Fullscreen().add_to(m) MeasureControl(position='topright', primary_length_unit='kilometers').add_to(m) geocoder = SafeGeocoder() coords = [] marker_cluster = MarkerCluster(name="Locations").add_to(m) processed_count = 0 for idx, row in df.iterrows(): if pd.isna(row[location_col]): continue location = str(row[location_col]).strip() additional_info = "" for col in df.columns: if col != location_col and not pd.isna(row[col]): additional_info += f"
{col}: {row[col]}" try: locations = [loc.strip() for loc in location.split(',') if loc.strip()] if not locations: locations = [location] except: locations = [location] for loc in locations: point = geocoder.get_coords(loc) if point: popup_content = f"""

{loc}

{additional_info}
""" folium.Marker( location=point, popup=folium.Popup(popup_content, max_width=300), tooltip=loc, icon=folium.Icon(color="blue", icon="info-sign") ).add_to(marker_cluster) coords.append(point) processed_count += 1 if coords: m.fit_bounds(coords) custom_css = """ """ m.get_root().header.add_child(folium.Element(custom_css)) return m._repr_html_(), processed_count def process_excel(file, places_column): if file is None: return None, "No file uploaded", None try: if hasattr(file, 'name'): df = pd.read_excel(file.name) elif isinstance(file, bytes): df = pd.read_excel(io.BytesIO(file)) else: df = pd.read_excel(file) print(f"Spalten in der Excel-Tabelle: {list(df.columns)}") if places_column not in df.columns: return None, f"Spalte '{places_column}' wurde in der Excel-Datei nicht gefunden. Verfügbare Spalten: {', '.join(df.columns)}", None map_html, processed_count = create_map(df, places_column) with tempfile.NamedTemporaryFile(suffix=".xlsx", delete=False) as tmp: processed_path = tmp.name df.to_excel(processed_path, index=False) total_locations = df[places_column].count() success_rate = (processed_count / total_locations * 100) if total_locations > 0 else 0 stats = f"Gefunden: {processed_count} von {total_locations} Orten ({success_rate:.1f}%)" return map_html, stats, processed_path except Exception as e: import traceback trace = traceback.format_exc() print(f"Error processing file: {e}\n{trace}") return None, f"Fehler bei der Verarbeitung der Datei: {str(e)}", None custom_css = """ """ with gr.Blocks(css=custom_css, title="Daten Strukturieren und Analysieren") as demo: gr.HTML("""

Strukturierung und Visualisierung von historischen Daten

In dieser Unterrichtseinheit befassen wir uns mit der Strukturierung unstrukturierter historischer Texte und der Visualisierung von extrahierten Daten auf Karten. Die systematische Strukturierung von Daten wird mit einem für Informationsextrahierung trainiertem Sprachmodell durchgeführt, das auf der Question-Answering-Methode basiert. Diese Methode erlaubt es, Informationen mit Hilfe einer Frage zu extrahieren, wie etwa „Wo fand das Erdbeben statt"? Dies ermöglicht die Extrahierung des Ortes, an dem ein Erdbeben stattfand, auch wenn im Text selbst noch andere genannt werden.

Nene Erdbeben in Japan. London, 15. Jan. (Drahtber.) Reuter meldet aus Osaka: Die telephonische Verbindung zwischen Osaka und Tokio ist heute um 5.45 Uhr durch ein Erdbeben unterbrochen worden. Die Straßenbahnen in Tokio liegen still. Der Eisenbahnverkehr Tokio — Osaka ist unterbrochen. Die kaiserliche Familie ist in Sicherheit. In Suvamo, einer Borstadt Tokios, sind Brände ausgebrochen. Ein Eisenbahnzug stürzte in den Bajubawo, einem Fluß zwischen Gotemba und Tokio. Sechs Züge wurden umgeworfen.
Ort des Erdbebens: Japan
Andere Orte: London, Osaka, Tokio, Suvamo, Gotemba.

Methodik: Vom unstrukturierten Text zur strukturierten Information

Die grundlegende Herausforderung bei der Arbeit mit historischen Quellen ist, dass relevante Informationen in langen Fließtexten eingebettet sind und manuell mühsam extrahiert werden müssen. Dieser Ansatz automatisiert diesen Prozess.

Wie funktioniert die Informationsextraktion?

  1. Template-Definition: Sie definieren ein JSON-Template mit den Informationstypen, die Sie extrahieren möchten: 
    {"earthquake location": "", "dateline location": ""}
  2. Question-Answering-Methode: Das Sprachmodell interpretiert jedes leere Feld als implizite Frage:
    • "earthquake location": "" → "Wo ist das Erdbeben passiert?"
    • "dateline location": "" → "Von wo wird berichtet?"
  3. Sprachmodell-Verarbeitung: Das NuExtract-1.5 Modell (ein Sequence-to-Sequence Transformer) analysiert den Text vollständig und identifiziert die relevanten Informationen für jedes Template-Feld.
  4. Strukturierte Ausgabe: Das Modell füllt das Template mit den extrahierten Informationen: 
    {"earthquake location": "Japan, Yokohama", "dateline location": "Tokio"}

Technische Funktionsweise des Sprachmodells

Intern erfolgt die Verarbeitung in mehreren Schritten:

  1. Tokenisierung: Der Text wird in bearbeitbare Einheiten zerlegt.
  2. Kontextuelle Analyse: Der Transformer-Mechanismus ermöglicht die Analyse von Beziehungen zwischen allen Textteilen gleichzeitig.
  3. Selektive Aufmerksamkeit: Das Modell fokussiert sich auf Textpassagen, die Antworten auf die impliziten Fragen enthalten könnten.
  4. Generierung: Die erkannten Informationen werden in das vorgegebene Template eingefügt.

Die Kartierungsfunktion

Nach der Extraktion der Ortsangaben ermöglicht unsere Anwendung die automatische Visualisierung dieser Daten auf einer interaktiven Karte:

  1. Geokodierung: Die extrahierten Ortsnamen werden mittels eines geografischen Dienstes in geografische Koordinaten (Längen- und Breitengrade) umgewandelt.
  2. Kartenerstellung: Die Koordinaten werden auf einer interaktiven Karte platziert, wobei jeder Ort durch einen Marker dargestellt wird.
  3. Kontextinformationen: Beim Klick auf einen Marker werden zusätzliche Informationen aus dem Originaltext angezeigt.
  4. Räumliche Analyse: Die Karte ermöglicht die visuelle Analyse der räumlichen Verteilung historischer Ereignisse.

Dieser kombinierte Ansatz aus Textextraktion und geografischer Visualisierung eröffnet neue Möglichkeiten für die räumliche Analyse historischer Quellen und erlaubt es, geografische Muster zu erkennen, die in den reinen Textdaten nicht unmittelbar sichtbar wären.

Diese Methode ermöglicht die effiziente Extraktion und Visualisierung historischer Daten aus unstrukturierten Quellen.

""") with gr.Tabs() as tabs: with gr.TabItem("🔍 Text Extrahierung"): gr.HTML("""

Extrahieren Sie strukturierte Daten aus unstrukturiertem Text

Verwenden Sie das Sprachmodell NuExtract-1.5 um automatisch Informationen zu extrahieren.

""") # Add model loading button and status at the top with gr.Row(): load_btn = gr.Button("Modell laden (1. Schritt)", variant="secondary") model_status = gr.Textbox( label="Modell Status", value="Modell nicht geladen. Bitte zuerst das Modell laden, um Verzögerungen zu vermeiden.", elem_classes="model-status" ) # Connect loading button load_btn.click( fn=load_model, inputs=[], outputs=[model_status] ) gr.HTML("""
""") with gr.Row(): with gr.Column(): template = gr.Textbox( label="JSON Template", value='{"earthquake location": "", "dateline location": ""}', lines=5 ) text = gr.Textbox( label="Hier unstrukturierten Text einfügen", value="Nene Erdbeben in Japan. London, 15. Jan. (Drahtber.) Reuter meldet aus Osaka: Die telephonische Verbindung zwischen Osaka und Tokio ist heute um 5.45 Uhr durch ein Erdbeben unterbrochen worden. Die Straßenbahnen in Tokio liegen still. Der Eisenbahnverkehr Tokio — Osaka ist unterbrochen. Die kaiserliche Familie ist in Sicherheit. In Suvamo, einer Borstadt Tokios, sind Brände ausgebrochen. Ein Eisenbahnzug stürzte in den Bajubawo, einem Fluß zwischen Gotemba und Tokio. Sechs Züge wurden umgeworfen. Nenqork, 15. Jan. (Drahtber.) Aus Tokio wird berichtet, daß in Uokohama bei dem Erdbeben sechs Personen getötet und 22 verletzt wurden. In Tokio wurden vier Personen getötet und 20 verletzt. In Nokohama wurden 800 Häuser zerstört.", lines=8 ) extract_btn = gr.Button("Extrahieren Sie Informationen (2. Schritt)", 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] ) with gr.TabItem("📍 Mapping von strukturierten Daten"): gr.HTML("""

Visualisieren Sie Daten auf Karten

Laden Sie eine Excel-Tabelle hoch und erstelle eine interaktive Karte.

""") with gr.Row(): with gr.Column(): excel_file = gr.File( label="Upload Excel File", file_types=[".xlsx", ".xls"], elem_classes="file-upload-box" ) places_column = gr.Textbox( label="Name der Tabellenspalte mit Ortsname", value="earthquake_location", placeholder="Füge den Namen der Spalte mit den Orten ein" ) process_btn = gr.Button("Erstellen Sie die Karte", variant="primary") with gr.Column(): map_output = gr.HTML( label="Interaktive Karte", value="""

Your map will appear here after processing

""", elem_id="map-container" ) stats_output = gr.Textbox( label="Status", lines=2, elem_classes="stats-box" ) processed_file = gr.File( label="Bearbeitete Daten herunterladen", visible=True, interactive=False ) def process_and_map(file, column): if file is None: return None, "Hier bitte die Excel-Tabelle hochladen", None try: map_html, stats, processed_path = process_excel(file, column) if map_html and processed_path: responsive_html = f"""
{map_html}
""" return responsive_html, stats, processed_path else: return None, stats, None except Exception as e: import traceback trace = traceback.format_exc() print(f"Error in process_and_map: {e}\n{trace}") return None, f"Error: {str(e)}", None process_btn.click( fn=process_and_map, inputs=[excel_file, places_column], outputs=[map_output, stats_output, processed_file] ) gr.HTML("""

Made with ❤ for historical research

""") if __name__ == "__main__": demo.launch()