python-sandbox

Python沙盒工具使用指南 v2.5 (与后端完全匹配版)

🎯 核心能力概览

Python沙盒是一个多功能的代码执行环境，支持：

| 功能领域 | 主要用途 | 关键库 | |---------|---------|-------| | 数据分析 | 数据清洗、转换、聚合 | Pandas, Polars | | 高性能计算 | 内存SQL、表达式加速 | DuckDB, Numexpr, Bottleneck | | 可视化 | 图表生成与自动捕获 | Matplotlib, Seaborn | | 文档自动化 | Excel/Word/PDF/PPT生成 | python-docx, reportlab, openpyxl | | 机器学习 | 模型训练与评估 | scikit-learn, LightGBM | | 符号数学 | 公式证明、方程求解 | SymPy | | 科学计算 | 优化、积分、信号处理 | SciPy | | 流程图生成 | 架构图、流程图 | Graphviz, NetworkX | | 文本分析 | HTML解析、数据提取 | BeautifulSoup4, lxml | | 性能优化 | 机械硬盘优化、异步IO | aiofiles, joblib |

📁 文件处理指南 - 两种模式必须分清

模式A: 工作区文件 (/data 目录)

用途: 数据分析、处理、持久化存储 支持格式: .csv, .xlsx, .xls, .parquet, .json, .txt, .feather 访问方式: 绝对路径 /data/文件名

import pandas as pd
df = pd.read_csv('/data/sales.csv')  # ✅ 正确

模式B: 上下文文件 (Base64嵌入)

用途: 图片识别、PDF内容提取 支持格式: .png, .jpg, .jpeg, .pdf, .txt(小文件) 特点: 文件内容直接嵌入对话，不在 /data 目录

# ❌ 错误：无法从/data读取上传的图片
# img = Image.open('/data/uploaded_image.png')  # 会失败

🚀 输出规范 - 后端实际支持的格式

1. 图表输出 - 系统自动捕获

import matplotlib.pyplot as plt
plt.plot([1,2,3], [4,5,6])
plt.title('示例图表')
plt.show()  # 🎯 关键：自动捕获，无需手动处理

# 支持以下图表库的自动捕获：
# - Matplotlib (使用 plt.show() 触发)
# - Graphviz (创建 Digraph 对象自动捕获)
# - NetworkX (通过 Matplotlib 渲染)

2. 可下载文件 - 必须使用JSON格式

import base64
import json

# 生成文件内容后...
file_data = base64.b64encode(content).decode('utf-8')

# 🎯 后端实际支持的输出类型：
output = {
    "type": "excel",  # 或 "word", "pdf", "ppt"
    "title": "销售报告.xlsx",
    "data_base64": file_data  # 注意：只有image类型用"image_base64"
}

# 对于图片输出，后端自动生成：
# {
#   "type": "image",
#   "title": "图表标题",
#   "image_base64": "base64字符串"
# }

print(json.dumps(output))  # 🎯 必须用JSON格式打印

3. 文本/数据 - 直接print

print("分析结果:")
print(f"总计: {total}")
print(df.describe())  # Pandas DataFrame自动美化显示

后端实际支持的输出类型列表：

• "image" - 图表、流程图（自动捕获）
• "excel" - Excel文件
• "word" - Word文档
• "pdf" - PDF文件
• "ppt" - PowerPoint演示文稿

💾 会话持久化 - 跨代码执行的文件共享

会话机制：

• 会话ID: 每个会话有唯一ID，文件按会话隔离
• 超时时间: 24小时无活动自动清理
• 工作目录: /data 目录对应会话工作区

工作流示例：

# 第一步：处理数据并保存
import pandas as pd
df = pd.read_excel('/data/原始数据.xlsx')
processed = df.groupby('部门')['销售额'].sum()
processed.to_csv('/data/部门汇总.csv')  # ✅ 保存中间结果
print("已保存部门汇总数据")

# 第二步：读取中间结果继续分析
df_summary = pd.read_csv('/data/部门汇总.csv')
print(f"读取到 {len(df_summary)} 个部门的汇总数据")

重要提醒：

• ✅ 同一会话内文件持久化（24小时超时）
• ✅ 新会话开始时 /data 目录为空
• ✅ 建议保存中间结果避免重复计算
• ✅ 使用同一session_id可跨多次代码执行共享文件

📚 工作流参考 - 按需查阅

快速查找表：

| 任务类型 | 参考文件 | 核心库 | |---------|---------|-------| | 创建图表 | matplotlib_cookbook.md | matplotlib, seaborn | | 数据处理 | pandas_cheatsheet.md | pandas, duckdb | | 生成报告 | report_generator_workflow.md | python-docx, reportlab | | 机器学习 | ml_workflow.md | scikit-learn, lightgbm | | 符号数学 | sympy_cookbook.md | sympy | | 科学计算 | scipy_cookbook.md | scipy | | 文本解析 | text_analysis_cookbook.md | beautifulsoup4, lxml |

示例工作流：

A. 公式证明工作流

# 1. 定义符号
import sympy as sp
x, y = sp.symbols('x y')

# 2. 构建表达式
lhs = (x + y)**2
rhs = x**2 + 2*x*y + y**2

# 3. 验证恒等
difference = sp.simplify(lhs - rhs)
print(f"差值: {difference}")
print(f"是否恒等: {difference == 0}")

B. ETL数据分析工作流

# Extract
df = pd.read_csv('/data/raw.csv')

# Transform
df_clean = (df
           .dropna()
           .drop_duplicates()
           .assign(profit = lambda d: d['revenue'] - d['cost']))

# Load
df_clean.to_csv('/data/cleaned.csv', index=False)
print(df_clean.describe())

C. Graphviz流程图生成

from graphviz import Digraph

# 创建流程图
dot = Digraph(comment='工作流程', format='png')
dot.node('A', '数据采集')
dot.node('B', '数据清洗')
dot.node('C', '数据分析')
dot.node('D', '报告生成')

dot.edges(['AB', 'BC', 'CD'])
dot.attr(rankdir='LR')  # 从左到右布局

# 🎯 自动捕获：Graphviz图表会被后端自动捕获并输出为图片

⚡ 性能优化指南 (与后端完全匹配)

1. 后端资源配置

内存限制: 6GB (mem_limit: "6g")
预留内存: 4GB (mem_reservation: "4g")
Swap限制: 禁用 (memswap_limit: "0")  # 🔥 避免机械硬盘swap死机
CPU限制: 75%配额 (cpu_quota: 75_000, cpu_period: 100_000)
超时时间: 90秒
文件系统: 只读根目录，/data可写，/tmp为tmpfs
网络: 完全禁用 (network_disabled: true)

2. 大文件处理策略

分块读取 (50MB+文件)

chunks = []
for chunk in pd.read_csv('/data/large.csv', chunksize=50000):
    processed = process_chunk(chunk)  # 自定义处理函数
    chunks.append(processed)
final_df = pd.concat(chunks, ignore_index=True)

格式转换加速

# 转换CSV为Feather格式 (提速10-100倍)
import pyarrow.feather as feather
df = pd.read_csv('/data/slow.csv')
feather.write_feather(df, '/data/fast.feather')  # 保存

# 后续读取极快
df_fast = feather.read_feather('/data/fast.feather')

3. 内存外计算 (避免OOM)

DuckDB内存SQL

import duckdb

# 直接查询CSV，不加载到内存
result = duckdb.sql("""
    SELECT department, 
           AVG(salary) as avg_salary,
           COUNT(*) as count
    FROM read_csv_auto('/data/employees.csv')
    WHERE hire_date > '2024-01-01'
    GROUP BY department
    ORDER BY avg_salary DESC
    LIMIT 10
""").df()  # 最后转为DataFrame
print(result)

Numexpr表达式加速

import numexpr as ne

# 传统方式（慢）
df['result'] = df['A'] * 2 + df['B'] ** 2 - df['C'] / 3

# Numexpr方式（快3-5倍）
df['result'] = ne.evaluate(
    "A * 2 + B ** 2 - C / 3",
    local_dict={k: df[k].values for k in ['A', 'B', 'C']}
)

4. 高级优化技巧 (后端已安装支持)

异步文件操作 - aiofiles

import aiofiles
import asyncio

async def process_large_file():
    # 异步读取，不阻塞主线程（机械硬盘特别受益）
    async with aiofiles.open('/data/large_file.csv', 'r') as f:
        content = await f.read()
    
    # 处理数据...
    
    # 异步写入
    async with aiofiles.open('/data/processed.csv', 'w') as f:
        await f.write(processed_content)

# 在异步环境中调用
await process_large_file()

内存缓存与并行计算 - joblib

from joblib import Memory
import time

# 创建内存缓存（可配置到磁盘）
cachedir = '/data/cache'
memory = Memory(cachedir, verbose=0)

@memory.cache
def expensive_computation(x, y):
    """计算结果会被缓存到磁盘"""
    time.sleep(2)  # 模拟耗时计算
    return x * y + x**2

# 第一次计算慢，后续从磁盘读取快
result1 = expensive_computation(10, 20)  # 慢
result2 = expensive_computation(10, 20)  # 快（从缓存）

DuckDB替代Pandas重操作

import duckdb

# ❌ 耗内存的Pandas操作
# df = pd.read_csv('/data/large.csv')
# result = df.groupby('category').agg({'value': ['mean', 'sum', 'count']})

# ✅ 内存友好的DuckDB操作
result = duckdb.sql("""
    SELECT category, 
           AVG(value) as mean_value,
           SUM(value) as sum_value,
           COUNT(value) as count_value
    FROM read_csv('/data/large.csv')
    GROUP BY category
""").df()

📋 可用库快速参考 (与Dockerfile完全一致)

数据处理核心

import pandas as pd          # 数据分析 (v2.2.2)
import numpy as np           # 数值计算 (v1.26.4)
import duckdb                # 内存SQL (v0.10.2)
import numexpr as ne         # 表达式加速 (v2.10.0)
import bottleneck as bn      # 滚动统计加速 (v1.3.8)
import pyarrow.feather as feather  # Feather格式支持 (v14.0.2)
import polars as pl          # 高性能DataFrame (v0.20.3)

机器学习增强

from sklearn.ensemble import RandomForestClassifier  # scikit-learn v1.5.0
import lightgbm as lgb       # 梯度提升树 (v4.3.0)
import category_encoders as ce  # 分类编码 (v2.6.3)
from skopt import BayesSearchCV  # 贝叶斯优化 (v0.9.0)
import statsmodels.api as sm  # 统计模型 (v0.14.1)

可视化与图表

import matplotlib.pyplot as plt  # 基础绘图 (v3.8.4)
import seaborn as sns            # 统计可视化 (v0.13.2)
import graphviz                  # 流程图 (自动布局) - 系统安装
import networkx as nx            # 网络图

文档生成

from docx import Document        # Word文档 (v1.1.2)
from reportlab.lib.pagesizes import letter  # PDF生成 (v4.0.7)
from pptx import Presentation    # PPT演示文稿 (v0.6.23)
import openpyxl                  # Excel操作 (v3.1.2)

##