如何在Altair中记录用户点击的Tooltip对应ID至日志或DataFrame？

一、基于现有Altair+Streamlit的实现方案 #

要获取点击国家的ID并记录，需调整Altair选择器配置，并通过Streamlit回调函数捕获选择事件，具体步骤如下：

1. 修改Altair选择器，绑定ID字段
   明确让selection_multi基于id字段选择，确保选中值为国家ID：

click = alt.selection_multi(fields=['id'], empty="none")

2. 添加点击事件回调函数
   定义回调函数捕获选中的ID，映射为国家名称后保存到日志文件或DataFrame：

def handle_click(selected):
    if not selected or 'id' not in selected:
        return
    
    # 获取选中的国家ID列表
    selected_ids = selected['id']
    # 匹配对应的国家名称
    selected_countries = country_names[country_names['id'].isin(selected_ids)]
    
    # 初始化会话状态存储点击记录
    if 'click_log_df' not in st.session_state:
        st.session_state.click_log_df = pd.DataFrame(columns=['timestamp', 'country_id', 'country_name'])
    
    # 生成新记录
    new_records = pd.DataFrame({
        'timestamp': [pd.Timestamp.now()] * len(selected_ids),
        'country_id': selected_ids,
        'country_name': selected_countries['name'].tolist()
    })
    
    # 更新记录DataFrame
    st.session_state.click_log_df = pd.concat([st.session_state.click_log_df, new_records], ignore_index=True)
    
    # 写入日志文件（可选）
    with open('country_clicks.log', 'a', encoding='utf-8') as f:
        for _, row in new_records.iterrows():
            f.write(f"{row['timestamp']} | ID:{row['country_id']} | 国家:{row['country_name']}\n")

3. 绑定回调到Altair图表
   调用st.altair_chart时传入on_select参数：

st.altair_chart(
    background.interactive(),
    use_container_width=False,
    on_select=handle_click
)

4. 完整修改后的代码

import pandas as pd
import altair as alt
import streamlit as st
from vega_datasets import data


@st.cache
def get_iso_names(url: str) -> pd.DataFrame:
    return pd.read_csv(url)


# Data generators for the background
sphere = alt.sphere()
graticule = alt.graticule()

# Source of land data
source = alt.topo_feature(data.world_110m.url, "countries")
iso_name_url = "https://raw.githubusercontent.com/stefangabos/world_countries/master/data/countries/en/world.csv"
country_names = get_iso_names(iso_name_url)

# 修改选择器，绑定id字段
click = alt.selection_multi(fields=['id'], empty="none")

# 点击事件回调函数
def handle_click(selected):
    if not selected or 'id' not in selected:
        return
    
    selected_ids = selected['id']
    selected_countries = country_names[country_names['id'].isin(selected_ids)]
    
    if 'click_log_df' not in st.session_state:
        st.session_state.click_log_df = pd.DataFrame(columns=['timestamp', 'country_id', 'country_name'])
    
    new_records = pd.DataFrame({
        'timestamp': [pd.Timestamp.now()] * len(selected_ids),
        'country_id': selected_ids,
        'country_name': selected_countries['name'].tolist()
    })
    
    st.session_state.click_log_df = pd.concat([st.session_state.click_log_df, new_records], ignore_index=True)
    
    # 写入日志文件
    with open('country_clicks.log', 'a', encoding='utf-8') as f:
        for _, row in new_records.iterrows():
            f.write(f"{row['timestamp']} | ID:{row['country_id']} | 国家:{row['country_name']}\n")

# Layering and configuring the components
background = (
    alt.layer(
        alt.Chart(sphere).mark_geoshape(fill="lightblue"),
        alt.Chart(graticule).mark_geoshape(stroke="white", strokeWidth=0.5),
        alt.Chart(source)
        .mark_geoshape(
            stroke="black",
            strokeWidth=0.15,
        )
        .encode(
            tooltip=[
                alt.Tooltip("name:N", title="Country"),
                # 可选：在tooltip显示ID方便调试
                # alt.Tooltip("id:Q", title="Country ID"),
            ],
            color=alt.condition(click, alt.value("firebrick"), alt.value("white")),
        )
        .transform_lookup(
            lookup="id",
            from_=alt.LookupData(country_names, "id", ["name"]),
        ),
    )
    .add_selection(click)
    .project("naturalEarth1")
    .properties(width=1200, height=800)
)


st.altair_chart(
    background.interactive(),
    use_container_width=False,
    on_select=handle_click
)

# 可选：在页面显示已记录的点击
if 'click_log_df' in st.session_state:
    st.subheader("已点击国家记录")
    st.dataframe(st.session_state.click_log_df.drop_duplicates(subset=['country_id']))

二、Plotly+Streamlit替代方案 #

如果觉得Altair的交互捕获不够直观，Plotly与Streamlit的交互绑定更直接，代码示例如下：

import pandas as pd
import plotly.express as px
import streamlit as st
from vega_datasets import data

@st.cache
def get_iso_names(url: str) -> pd.DataFrame:
    return pd.read_csv(url)

# 加载国家数据
iso_name_url = "https://raw.githubusercontent.com/stefangabos/world_countries/master/data/countries/en/world.csv"
country_names = get_iso_names(iso_name_url)
world_topo = data.world_110m.url

# 绘制地图
fig = px.choropleth_mapbox(
    country_names,
    geojson=world_topo,
    locations='id',
    featureidkey="properties.id",
    color='name',
    mapbox_style="carto-positron",
    zoom=1,
    center={"lat": 0, "lon": 0},
    hover_data=['id', 'name'],
    color_discrete_sequence=['white'],
    opacity=0.8
)

fig.update_traces(marker_line_width=0.15, marker_line_color='black')
fig.update_layout(margin={"r":0,"t":0,"l":0,"b":0}, width=1200, height=800)

# 点击事件处理函数
def handle_map_click(trace, points, state):
    if not points.point_inds:
        return
    
    # 获取点击的国家数据
    selected_row = country_names.iloc[points.point_inds[0]]
    selected_id = selected_row['id']
    selected_name = selected_row['name']
    
    # 初始化记录存储
    if 'click_log_df' not in st.session_state:
        st.session_state.click_log_df = pd.DataFrame(columns=['timestamp', 'country_id', 'country_name'])
    
    # 添加新记录
    new_record = pd.DataFrame({
        'timestamp': [pd.Timestamp.now()],
        'country_id': [selected_id],
        'country_name': [selected_name]
    })
    st.session_state.click_log_df = pd.concat([st.session_state.click_log_df, new_record], ignore_index=True)
    
    # 写入日志
    with open('country_clicks.log', 'a', encoding='utf-8') as f:
        f.write(f"{pd.Timestamp.now()} | ID:{selected_id} | 国家:{selected_name}\n")

# 绑定点击事件
fig.data[0].on_click(handle_map_click)

# 渲染图表
st.plotly_chart(fig, use_container_width=False)

# 显示记录
if 'click_log_df' in st.session_state:
    st.subheader("已点击国家记录")
    st.dataframe(st.session_state.click_log_df.drop_duplicates(subset=['country_id']))

内容的提问来源于stack exchange，提问作者a-doering