基于Streamlit开发的OCR可视化校验工具,提供现代化的Web界面和丰富的交互体验,让OCR结果校验变得直观高效。特别新增了强大的表格数据分析功能。
# 安装Streamlit和相关依赖
pip install streamlit plotly pandas pillow numpy opencv-python openpyxl
# 方法1: 完整功能版本
python -m streamlit run streamlit_ocr_validator.py
# 方法2: 使用启动脚本
python run_streamlit_validator.py
# 方法3: 开发模式(自动重载)
streamlit run streamlit_ocr_validator.py --server.runOnSave true
浏览器会自动打开 http://localhost:8501,如果没有自动打开,请手动访问该地址。
┌─────────────────────────────────────────────────────────────────────┐
│ 🔍 OCR可视化校验工具 │
├─────────────────────────────────────────────────────────────────────┤
│ 📊 总文本块: 13 🔗 可点击: 9 ❌ 标记错误: 2 ✅ 准确率: 85.7% │
├─────────────────────────┬───────────────────────────────────────────┤
│ 📄 OCR识别内容 │ 🖼️ 原图标注 │
│ │ │
│ 🔍 搜索框 │ [显示选中位置的红框标注] │
│ 📍 选择文本下拉框 │ [图片缩放和详细信息] │
│ │ │
│ 📝 MD内容预览 │ 📍 选中文本详情 │
│ [4种渲染模式选择] │ - 文本内容: xxx │
│ ○ HTML渲染 │ - 位置: [x1,y1,x2,y2] │
│ ● Markdown渲染 │ - 宽度: xxx px │
│ ○ DataFrame表格 ⭐ │ - 高度: xxx px │
│ ○ 原始文本 │ │
│ │ │
│ 🎯 可点击文本列表 │ │
│ [📍 文本1] [❌] [✅] │ │
│ [📍 文本2] [❌] [✅] │ │
└─────────────────────────┴───────────────────────────────────────────┘
┌─────────────────────┐
│ 📁 文件选择 │
│ [选择OCR结果文件] │
│ [🔄 加载文件] │
│ │
│ 🎛️ 控制面板 │
│ [🧹 清除选择] │
│ [❌ 清除错误标记] │
│ │
│ 📊 表格快捷操作 ⭐ │
│ [🔍 快速预览表格] │
│ [📥 一键导出所有表格] │
│ │
│ 🔧 调试信息 │
│ [调试信息开关] │
└─────────────────────┘
选择文件
浏览统计信息
交互校验
搜索过滤
表格数据分析 ⭐ 新增
错误标记管理
streamlit_ocr_validator.py)# 表格检测和转换
if '<table>' in display_content.lower():
st.session_state.validator.display_html_table_as_dataframe(display_content)
else:
st.info("当前内容中没有检测到HTML表格")
基础操作
数据过滤
数据排序
统计分析
数据导出
# 在streamlit界面中
def display_html_table_as_dataframe(self, html_content: str, enable_editing: bool = False):
"""将HTML表格解析为DataFrame显示"""
import pandas as pd
from io import StringIO, BytesIO
try:
# 使用pandas直接读取HTML表格
tables = pd.read_html(StringIO(html_content))
if tables:
for i, table in enumerate(tables):
st.subheader(f"📊 表格 {i+1}")
# 创建表格操作按钮
col1, col2, col3, col4 = st.columns(4)
with col1:
show_info = st.checkbox(f"显示表格信息", key=f"info_{i}")
with col2:
show_stats = st.checkbox(f"显示统计信息", key=f"stats_{i}")
with col3:
enable_filter = st.checkbox(f"启用过滤", key=f"filter_{i}")
with col4:
enable_sort = st.checkbox(f"启用排序", key=f"sort_{i}")
# 显示表格
st.dataframe(table, use_container_width=True)
import plotly.express as px
import streamlit as st
def create_table_visualization(df):
"""创建表格数据可视化"""
if not df.empty:
numeric_cols = df.select_dtypes(include=[np.number]).columns
if len(numeric_cols) > 0:
# 创建统计图表
fig = px.bar(
x=df.index,
y=df[numeric_cols[0]],
title=f"{numeric_cols[0]} 分布"
)
st.plotly_chart(fig, use_container_width=True)
# 创建散点图
if len(numeric_cols) > 1:
fig_scatter = px.scatter(
df,
x=numeric_cols[0],
y=numeric_cols[1],
title=f"{numeric_cols[0]} vs {numeric_cols[1]}"
)
st.plotly_chart(fig_scatter, use_container_width=True)
# 在主应用中使用
if st.checkbox("显示数据可视化"):
create_table_visualization(filtered_table)
def advanced_table_editor(df):
"""高级表格编辑器"""
st.subheader("🔧 高级编辑")
# 数据编辑
edited_df = st.data_editor(
df,
use_container_width=True,
num_rows="dynamic", # 允许添加删除行
key="advanced_editor"
)
# 数据验证
if not edited_df.equals(df):
st.success("✏️ 数据已修改")
# 显示变更统计
changes = len(edited_df) - len(df)
st.info(f"行数变化: {changes:+d}")
# 导出修改后的数据
if st.button("💾 保存修改"):
csv_data = edited_df.to_csv(index=False)
st.download_button(
"下载修改后的数据",
csv_data,
"modified_table.csv",
"text/csv"
)
return edited_df
def batch_table_processing():
"""批量表格处理功能"""
st.subheader("📦 批量表格处理")
uploaded_files = st.file_uploader(
"上传多个包含表格的文件",
type=['md', 'html'],
accept_multiple_files=True
)
if uploaded_files and st.button("开始批量处理"):
all_tables = []
progress_bar = st.progress(0)
for i, file in enumerate(uploaded_files):
content = file.read().decode('utf-8')
if '<table' in content.lower():
tables = pd.read_html(StringIO(content))
for j, table in enumerate(tables):
table['source_file'] = file.name
table['table_index'] = j
all_tables.append(table)
progress_bar.progress((i + 1) / len(uploaded_files))
if all_tables:
st.success(f"✅ 共处理 {len(all_tables)} 个表格")
# 合并所有表格
if st.checkbox("合并所有表格"):
try:
merged_df = pd.concat(all_tables, ignore_index=True)
st.dataframe(merged_df)
# 导出合并结果
csv_data = merged_df.to_csv(index=False)
st.download_button(
"下载合并表格",
csv_data,
"merged_tables.csv",
"text/csv"
)
except Exception as e:
st.error(f"合并失败: {e}")
def table_quality_check(df):
"""表格数据质量检查"""
st.subheader("🔍 数据质量检查")
# 基础统计
col1, col2, col3 = st.columns(3)
with col1:
st.metric("总行数", len(df))
with col2:
st.metric("总列数", len(df.columns))
with col3:
null_percent = (df.isnull().sum().sum() / (len(df) * len(df.columns))) * 100
st.metric("缺失值比例", f"{null_percent:.1f}%")
# 详细质量报告
quality_issues = []
# 检查空值
null_cols = df.columns[df.isnull().any()].tolist()
if null_cols:
quality_issues.append(f"发现 {len(null_cols)} 列存在空值: {', '.join(null_cols)}")
# 检查重复行
duplicate_rows = df.duplicated().sum()
if duplicate_rows > 0:
quality_issues.append(f"发现 {duplicate_rows} 行重复数据")
# 检查数据类型一致性
for col in df.columns:
if df[col].dtype == 'object':
# 检查是否应该是数值类型
numeric_like = df[col].str.replace(',', '').str.replace('$', '')
try:
pd.to_numeric(numeric_like, errors='raise')
quality_issues.append(f"列 '{col}' 可能应该是数值类型")
except:
pass
if quality_issues:
st.warning("⚠️ 发现数据质量问题:")
for issue in quality_issues:
st.write(f"- {issue}")
else:
st.success("✅ 数据质量良好")
@st.cache_data
def load_and_process_ocr_data(file_path: str):
"""缓存OCR数据加载和处理"""
with open(file_path, 'r') as f:
ocr_data = json.load(f)
# 处理数据
processed_data = process_ocr_data(ocr_data)
return processed_data
@st.cache_resource
def load_image(image_path: str):
"""缓存图片加载"""
return Image.open(image_path)
@st.cache_data
def parse_html_tables(html_content: str):
"""缓存表格解析结果"""
try:
tables = pd.read_html(StringIO(html_content))
return tables
except:
return []
def handle_large_tables():
"""处理大型表格"""
if 'page_size' not in st.session_state:
st.session_state.page_size = 100
# 分页显示表格
if not df.empty:
total_rows = len(df)
pages = (total_rows - 1) // st.session_state.page_size + 1
col1, col2, col3 = st.columns([1, 2, 1])
with col2:
current_page = st.slider("页数", 1, pages, 1)
# 显示当前页数据
start_idx = (current_page - 1) * st.session_state.page_size
end_idx = min(start_idx + st.session_state.page_size, total_rows)
current_df = df.iloc[start_idx:end_idx]
st.dataframe(current_df, use_container_width=True)
st.info(f"显示第 {start_idx+1}-{end_idx} 行,共 {total_rows} 行")
def optimize_dataframe_memory(df):
"""优化DataFrame内存使用"""
initial_memory = df.memory_usage(deep=True).sum()
# 优化数值类型
for col in df.select_dtypes(include=['int']).columns:
df[col] = pd.to_numeric(df[col], downcast='integer')
for col in df.select_dtypes(include=['float']).columns:
df[col] = pd.to_numeric(df[col], downcast='float')
# 优化字符串类型
for col in df.select_dtypes(include=['object']).columns:
if df[col].nunique() < len(df) * 0.5: # 如果唯一值少于50%,转换为category
df[col] = df[col].astype('category')
final_memory = df.memory_usage(deep=True).sum()
reduction = (initial_memory - final_memory) / initial_memory * 100
st.info(f"内存优化:减少 {reduction:.1f}% ({initial_memory/1024/1024:.1f}MB → {final_memory/1024/1024:.1f}MB)")
return df
# 开发模式运行(自动重载)
streamlit run streamlit_ocr_validator.py --server.runOnSave true
# 指定端口运行
streamlit run streamlit_ocr_validator.py --server.port 8502
# 指定主机运行(局域网访问)
streamlit run streamlit_ocr_validator.py --server.address 0.0.0.0
FROM python:3.9-slim
WORKDIR /app
# 安装系统依赖
RUN apt-update && apt-get install -y \
libgl1-mesa-glx \
libglib2.0-0 \
libsm6 \
libxext6 \
libxrender-dev \
libgomp1 \
&& rm -rf /var/lib/apt/lists/*
# 安装Python依赖
COPY requirements.txt .
RUN pip install -r requirements.txt
COPY . .
EXPOSE 8501
CMD ["streamlit", "run", "streamlit_ocr_validator.py", "--server.address=0.0.0.0"]
st.spinner()显示加载状态st.error()友好地显示错误信息st.progress()显示处理进度@st.cache_data提升性能st.columns()合理分布内容Streamlit版本的OCR校验工具经过升级后提供了更加强大的功能:
✅ 基础功能:实时交互、动态更新、错误管理
✅ 表格分析:HTML表格转DataFrame、多种操作、导出功能 ⭐
✅ 数据处理:过滤、排序、统计分析、可视化 ⭐
✅ 批量操作:多文件处理、批量导出、合并功能 ⭐
✅ 质量检查:数据质量分析、问题检测、优化建议 ⭐
✅ 扩展性:易于添加新功能和自定义组件
✅ 用户体验:现代化界面、响应式设计、直观操作
新增的表格分析功能使其不仅能够校验OCR结果,更能深入分析表格数据,成为一个完整的OCR数据处理工作台!
🌟 特别推荐:使用DataFrame表格模式分析财务报表等结构化数据,体验完整的数据处理工作流程。