Spaces:
Running
Running
| # -*- coding: utf-8 -*- | |
| import streamlit as st | |
| from docx import Document | |
| from config import APP_CONFIG | |
| try: | |
| import pandas as pd | |
| PANDAS_AVAILABLE = True | |
| except ImportError: | |
| PANDAS_AVAILABLE = False | |
| try: | |
| import fitz # PyMuPDF 的导入名称是 fitz | |
| PYMUPDF_AVAILABLE = True | |
| except ImportError: | |
| PYMUPDF_AVAILABLE = False | |
| def extract_text_from_files(uploaded_files, max_chars=None): | |
| if max_chars is None: | |
| max_chars = APP_CONFIG.get("max_extract_chars", 150000) | |
| extracted_text = "" | |
| for file in uploaded_files: | |
| try: | |
| file.seek(0) | |
| if file.name.lower().endswith('.pdf') and PYMUPDF_AVAILABLE: | |
| # 从内存中读取文件字节流 | |
| pdf_bytes = file.read() | |
| # 使用上下文管理器确保内存被及时释放 | |
| with fitz.open(stream=pdf_bytes, filetype="pdf") as pdf_doc: | |
| for page in pdf_doc: | |
| text = page.get_text() | |
| if text: | |
| extracted_text += text + "\n" | |
| if len(extracted_text) > max_chars: | |
| break | |
| elif file.name.lower().endswith(('.docx', '.doc')): | |
| doc = Document(file) | |
| for para in doc.paragraphs: | |
| extracted_text += para.text + "\n" | |
| if len(extracted_text) > max_chars: | |
| break | |
| else: | |
| content = file.getvalue().decode('utf-8', errors='ignore') | |
| extracted_text += content + "\n" | |
| if len(extracted_text) > max_chars: | |
| st.warning(f"⚠️ 文件 {file.name} 已安全截断至 {max_chars} 字符。") | |
| break | |
| except Exception as e: | |
| st.warning(f"⚠️ 解析失败: {e}") | |
| finally: | |
| file.seek(0) | |
| return extracted_text[:max_chars] | |
| def extract_data_from_files(uploaded_file, max_chars=20000): | |
| if not uploaded_file: | |
| return "" | |
| try: | |
| uploaded_file.seek(0) | |
| if PANDAS_AVAILABLE: | |
| df = pd.read_csv(uploaded_file) if uploaded_file.name.lower().endswith('.csv') else pd.read_excel(uploaded_file) | |
| if len(df) > 100: | |
| st.warning(f"⚠️ 数据表超长，安全截取前100行。") | |
| df = df.head(100) | |
| return df.to_markdown(index=False) | |
| else: | |
| return uploaded_file.getvalue().decode('utf-8', errors='ignore')[:max_chars] | |
| except Exception as e: | |
| st.warning(f"⚠️ 解析失败: {e}") | |
| return "" | |
| finally: | |
| uploaded_file.seek(0) |