数据分析
未发现用户侧风险
电子表格验证执行
在 Python 脚本执行电子表格任务之前进行必要的数据验证和源可访问性检查
文件预览
2 个文件
SKILL.md
19.7 KB · 可预览
---
name: spreadsheet-validated-exec
description: Execute Python scripts for spreadsheets with prerequisite data validation and source accessibility checks
---
# Validated Python Execution for Spreadsheet Tasks
## Overview
This skill extends direct Python execution for spreadsheet operations by adding a **mandatory data validation phase** before any processing begins. This prevents wasted iterations on inaccessible data sources and provides clear error documentation when data cannot be accessed.
## When to Use This Skill
Use validated direct `run_shell` with Python scripts for spreadsheet operations when:
- Reading or writing complex Excel files with multiple sheets
- Data sources have already been validated as accessible
- You have fallback sources identified in case of access failures
- Applying formulas, formatting, or data transformations
- Working with `openpyxl`, `pandas`, or similar libraries
- The operation involves multiple steps that could exceed agent step limits
- You need precise control over error handling and debugging
- Complex scripts benefit from file-based execution for better reliability
## Why Validated Direct Execution?
Beyond standard shell_agent limitations, **unvalidated data access** causes:
- Wasted iterations attempting to process non-existent data
- Unclear error messages when source files are inaccessible
- No graceful degradation when primary sources fail
- Missing documentation of why data operations couldn't complete
Direct `run_shell` with Python validation is more reliable because it:
- Executes validation in a single step with no iteration limits
- Provides clearer, immediate error messages for access failures
- Handles complex operations without step constraints
- Gives full control over library imports and execution flow
- Writing scripts to `.py` files first avoids shell_agent parsing issues with heredocs
- Documents all access failures for troubleshooting and reporting
## Phase 0: Data Source Validation (REQUIRED)
**Before writing any spreadsheet processing code**, verify your data sources are accessible:
### Step 0.1: Verify Data Availability
```python
import os
import requests
from pathlib import Path
# For local files
def verify_local_file(file_path):
path = Path(file_path)
if not path.exists():
raise FileNotFoundError(f"Data file not found: {file_path}")
if not path.is_file():
raise ValueError(f"Path is not a file: {file_path}")
if path.stat().st_size == 0:
raise ValueError(f"Data file is empty: {file_path}")
print(f"✓ Local file verified: {file_path} ({path.stat().st_size} bytes)")
return True
# For remote URLs
def verify_remote_url(url, timeout=30):
try:
# Try HEAD request first (lighter than GET)
response = requests.head(url, timeout=timeout, allow_redirects=True)
if response.status_code == 405: # HEAD not allowed, try GET
response = requests.get(url, timeout=timeout, stream=True)
response.raise_for_status()
# Check content type if available
content_type = response.headers.get('Content-Type', '')
if 'text/html' in content_type and 'excel' not in url.lower():
print(f"⚠ Warning: URL may return HTML, not data file")
print(f"✓ Remote URL verified: {url} (Status: {response.status_code})")
return True
except requests.exceptions.SSLError as e:
print(f"✗ SSL Error: {str(e)}")
return False
except requests.exceptions.ConnectionError as e:
print(f"✗ Connection Error: {str(e)}")
return False
except requests.exceptions.Timeout as e:
print(f"✗ Timeout Error: {str(e)}")
return False
except Exception as e:
print(f"✗ Verification Failed: {str(e)}")
return False
```
### Step 0.2: Handle Inaccessible Sources
**If primary data source is unavailable:**
1. **Check alternative locations:**
- Is data available from a backup URL?
- Is there a cached/local copy available?
- Can data be obtained from an API instead of web scraping?
- Is there an alternative data provider?
2. **Document the failure:**
```python
def log_access_failure(source, error_type, timestamp=None):
from datetime import datetime
if not timestamp:
timestamp = datetime.now().isoformat()
error_log = {
'timestamp': timestamp,
'source': source,
'error_type': error_type,
'alternatives_attempted': [],
'resolution': 'pending'
}
# Save to error log file
import json
with open('data_access_errors.json', 'a') as f:
f.write(json.dumps(error_log) + '\n')
return error_log
```
3. **Implement fallback strategy:**
```python
def get_data_with_fallback(primary_source, fallback_sources):
sources = [primary_source] + fallback_sources
for i, source in enumerate(sources):
print(f"Attempting source {i+1}/{len(sources)}: {source}")
if source.startswith('http'):
if verify_remote_url(source):
return download_data(source)
else:
if verify_local_file(source):
return load_local_data(source)
print(f"Source {source} unavailable, trying next...")
raise Exception(f"All {len(sources)} data sources unavailable")
```
### Step 0.3: Pre-Execution Checklist
Before proceeding to spreadsheet operations, confirm:
- [ ] Primary data source is accessible (file exists / URL responds)
- [ ] Data file is not empty (has content to process)
- [ ] Required permissions are in place (file not locked, API keys valid)
- [ ] Alternative sources identified if primary fails
- [ ] Error logging mechanism is configured
**If any check fails:** Do NOT proceed to spreadsheet processing. Report the blocking issue and either:
- Resolve the data access problem first
- Switch to an alternative data source
- Abort the task with clear error documentation
## How to Use (With Validation)
### Complete Workflow Template
**Recommended Pattern: Validate → Process → Report**
```bash
# Step 1: Write validation + processing script to file
cat > validated_spreadsheet_process.py << 'EOF'
import sys
import os
from pathlib import Path
import pandas as pd
from openpyxl import load_workbook
# === PHASE 0: VALIDATION ===
def validate_sources():
sources_to_check = [
('input_data.xlsx', 'local'),
# ('https://backup-source.com/data.xlsx', 'remote')
]
validated_sources = []
for source, source_type in sources_to_check:
try:
if source_type == 'local':
if not Path(source).exists():
print(f"ERROR: Local file not found: {source}")
continue
if Path(source).stat().st_size == 0:
print(f"ERROR: Local file is empty: {source}")
continue
validated_sources.append(source)
print(f"✓ Validated: {source}")
except Exception as e:
print(f"ERROR validating {source}: {e}")
if not validated_sources:
print("FATAL: No valid data sources available. Aborting.")
sys.exit(1)
return validated_sources
# === PHASE 1: PROCESSING ===
def process_data(source_file):
# Your spreadsheet operations here
pass
# === PHASE 2: REPORTING ===
def generate_report(success, details):
report = {
'status': 'success' if success else 'failed',
'details': details
}
print(f"Report: {report}")
return report
if __name__ == '__main__':
try:
# Validate first
sources = validate_sources()
# Process validated sources
for source in sources:
process_data(source)
generate_report(True, f"Processed {len(sources)} sources")
except Exception as e:
generate_report(False, str(e))
sys.exit(1)
EOF
# Step 2: Execute the validated script
python3 validated_spreadsheet_process.py
```
### Pattern 1: Write Script to File First (Complex Operations)
For complex multi-line scripts with multiple data sources:
```bash
# Step 1: Write the Python script to a file
cat > process_spreadsheet.py << 'EOF'
import openpyxl
from openpyxl import Workbook
from pathlib import Path
import sys
# Validate first
input_file = Path('file.xlsx')
if not input_file.exists():
print(f"ERROR: Input file not found: {input_file}")
sys.exit(1)
if input_file.stat().st_size == 0:
print(f"ERROR: Input file is empty: {input_file}")
sys.exit(1)
# Your spreadsheet code here
wb = openpyxl.load_workbook('file.xlsx')
# ... operations ...
wb.save('output.xlsx')
print('Success')
EOF
# Step 2: Execute the script
python3 process_spreadsheet.py
```
### Pattern 2: Inline Heredoc (Simple Scripts with Validation)
For short scripts when NOT using shell_agent, include validation inline:
```bash
# Include quick validation before processing
python3 << 'EOF'
import os
import sys
# Quick validation
input_file = 'data.xlsx'
if not os.path.exists(input_file):
print(f"ERROR: {input_file} not found")
sys.exit(1)
if os.path.getsize(input_file) == 0:
print(f"ERROR: {input_file} is empty")
sys.exit(1)
# Continue with processing...
EOF
```
## Examples
### Example 1: Validated Read and Transform
**With prerequisite validation:**
```python
import pandas as pd
import sys
from pathlib import Path
# Validate input file exists and has content
input_path = Path('input.xlsx')
if not input_path.exists():
print(f"ERROR: Input file not found: {input_path}")
sys.exit(1)
if input_path.stat().st_size == 0:
print(f"ERROR: Input file is empty: {input_path}")
sys.exit(1)
# Now safe to process
df = pd.read_excel('input.xlsx', sheet_name='Revenue')
# Apply transformations
df['Net_Revenue'] = df['Gross_Revenue'] * (1 - df['Tax_Rate'])
# Save results
df.to_excel('output.xlsx', index=False, sheet_name='Processed')
```
### Example 2: Multi-Sheet with Source Validation
```python
from openpyxl import load_workbook
import sys
from pathlib import Path
# Validate workbook accessibility
wb_path = Path('tour_data.xlsx')
if not wb_path.exists():
print(f"ERROR: Workbook not found: {wb_path.absolute()}")
sys.exit(1)
try:
wb = load_workbook('tour_data.xlsx')
except Exception as e:
print(f"ERROR: Cannot open workbook: {e}")
print("Possible causes: file locked, corrupted, or wrong format")
sys.exit(1)
# Iterate through sheets
for sheet_name in wb.sheetnames:
ws = wb[sheet_name]
# Apply formatting or calculations
for row in ws.iter_rows(min_row=2, max_col=5):
# Process cells
pass
wb.save('tour_data_processed.xlsx')
```
### Example 3: Formatting with Pre-Checks
```python
from openpyxl import load_workbook
from openpyxl.styles import Font, PatternFill, Alignment
from pathlib import Path
import sys
# Pre-check file
report_path = Path('report.xlsx')
if not report_path.exists():
print(f"ERROR: Report file not found: {report_path}")
sys.exit(1)
wb = load_workbook('report.xlsx')
ws = wb.active
# Apply header styling
header_fill = PatternFill(start_color='4472C4', fill_type='solid')
header_font = Font(bold=True, color='FFFFFF')
for cell in ws[1]:
cell.fill = header_fill
cell.font = header_font
cell.alignment = Alignment(horizontal='center')
wb.save('report_formatted.xlsx')
```
### Example 4: Comprehensive Error Handling with Logging
```python
import sys
import json
from datetime import datetime
from pathlib import Path
from openpyxl import load_workbook
def log_error(source, error_msg, error_type):
"""Log errors for later analysis and reporting"""
error_record = {
'timestamp': datetime.now().isoformat(),
'source': source,
'error_type': error_type,
'message': error_msg
}
with open('processing_errors.log', 'a') as f:
f.write(json.dumps(error_record) + '\n')
return error_record
try:
# Validate file before opening
if not Path('data.xlsx').exists():
log_error('data.xlsx', 'File not found', 'ValidationError')
raise FileNotFoundError('data.xlsx')
wb = load_workbook('data.xlsx')
ws = wb.active
# Your operations here
value = ws['A1'].value
wb.save('output.xlsx')
print(f"Success: Processed {ws.max_row} rows")
except FileNotFoundError as e:
log_error('data.xlsx', str(e), 'FileNotFoundError')
print(f"ERROR: {str(e)}", file=sys.stderr)
print("ACTION: Verify file path and permissions")
sys.exit(1)
except PermissionError as e:
log_error('data.xlsx', str(e), 'PermissionError')
print(f"ERROR: Permission denied - file may be open in another application")
print("ACTION: Close the file in other applications and retry")
sys.exit(1)
except Exception as e:
log_error('data.xlsx', str(e), type(e).__name__)
print(f"ERROR: {str(e)}", file=sys.stderr)
sys.exit(1)
```
## Best Practices (Enhanced)
1. **Always validate before processing**: Check data source availability before any spreadsheet operations
2. **Log all access failures**: Document when and why data sources become unavailable
3. **Identify fallbacks upfront**: Know your alternative sources before starting
4. **Fail fast on validation errors**: Don't waste iterations on impossible operations
5. **Prefer file-based execution** for complex scripts: write to `.py` file first, then execute via `run_shell`
6. **Import only needed libraries** to reduce execution time
7. **Print clear success/error messages** for debugging
8. **Save intermediate results** for complex multi-step transformations
9. **Test with small data** before scaling to large spreadsheets
10. **Use pandas for data manipulation** and openpyxl for formatting when both are needed
11. **Clean up temporary script files** after execution if they won't be reused
## Data Access Failure Protocols
### When Primary Source Fails
1. **Immediate actions:**
- Log the failure with timestamp, source, and error type
- Check if error is temporary (timeout) or permanent (404, file doesn't exist)
- Attempt alternative sources if available
2. **For temporary failures (timeouts, 5xx errors):**
```python
def retry_with_backoff(operation, max_retries=3, base_delay=1):
import time
for attempt in range(max_retries):
try:
return operation()
except (TimeoutError, ConnectionError) as e:
if attempt == max_retries - 1:
raise
delay = base_delay * (2 ** attempt)
print(f"Retry {attempt + 1}/{max_retries} after {delay}s delay")
time.sleep(delay)
```
3. **For permanent failures (file not found, 404):**
- Switch to identified fallback source immediately
- Document the primary source failure
- Continue with fallback if available, otherwise abort cleanly
### Error Reporting Template
```python
def report_data_access_issue(issue_type, source, details, alternatives_tried=None):
"""Standardize error reporting for data access failures"""
from datetime import datetime
report = f"""
=== DATA ACCESS FAILURE REPORT ===
Type: {issue_type}
Source: {source}
Time: {datetime.now().isoformat()}
Details: {details}
Alternatives Attempted: {alternatives_tried or 'None'}
Recommended Action: {get_recommended_action(issue_type)}
==================================
"""
print(report)
return report
def get_recommended_action(issue_type):
actions = {
'FileNotFound': 'Verify file path, check if file was moved or deleted',
'ConnectionError': 'Check network connectivity, try alternative source',
'Timeout': 'Increase timeout, check source server status',
'PermissionError': 'Check file permissions, ensure file not open elsewhere',
'SSLError': 'Verify SSL certificate, try HTTP if appropriate',
'default': 'Review error details, check source availability'
}
return actions.get(issue_type, actions['default'])
```
## When NOT to Use This Skill
- Simple single-cell reads/writes (validation overhead not warranted)
- Operations that require interactive user input
- Tasks where you need the agent to iteratively refine the approach
- When you have no way to validate data sources (skip to error handling only)
## Common Libraries
| Library | Best For | Also Use For |
|---------|----------|--------------|
| `requests` | Web requests | Source URL validation, health checks |
| `openpyxl` | Reading/writing .xlsx files, formatting, formulas | - |
| `pandas` | Data manipulation, analysis, merging datasets | Chunked reading for large files |
| `xlrd` | Reading older .xls files (read-only) | - |
| `xlsxwriter` | Creating new .xlsx files with advanced formatting | - |
## Troubleshooting (Enhanced)
**Issue**: Data source unavailable / file not found
- **Solution**: Run validation check first before processing. If validation fails:
1. Verify the path/URL is correct
2. Check if file was moved, renamed, or deleted
3. Try alternative sources if available
4. Document the failure and abort cleanly rather than proceeding
**Issue**: Source was accessible yesterday but not today
- **Solution**: Implement fallback sources and log the change. Consider:
- Setting up monitoring for critical data sources
- Caching important data locally when possible
- Having contact information for data source maintainers
**Issue**: Validation passes but processing fails
- **Solution**: Validation checks accessibility, not content validity. Add:
- Schema validation (check expected columns exist)
- Content validation (check data types, ranges)
- Sample verification (read first few rows to confirm structure)
**Issue**: Heredoc syntax fails with 'unknown error' when using shell_agent
- **Solution**: Write the Python script to a `.py` file first with full validation logic, then execute with `python3 script.py`. This is significantly more reliable than inline heredoc when shell_agent is the executor.
**Issue**: FileNotFoundError (after validation passed)
- **Solution**: File may have been moved/deleted between validation and processing, or path is relative and working directory changed. Use absolute paths.
**Issue**: PermissionError
- **Solution**: Ensure the file is not open in another application. On Linux/Mac, check with `lsof | grep filename`
**Issue**: ConnectionError / Timeout on remote sources
- **Solution**: Check network connectivity, try with increased timeout, use retry logic with exponential backoff, or switch to cached/local copy if available
**Issue**: SSL Certificate errors
- **Solution**: Verify the certificate is valid. As last resort for trusted internal sources only: `requests.get(url, verify=False)` - but log this as a security concern
**Issue**: All alternative sources fail
- **Solution**: Abort with comprehensive error report documenting:
- All sources attempted
- Error types for each
- Timestamp of failures
- Recommended next steps for human operator
**Issue**: MemoryError on large files
- **Solution**: Process data in chunks using pandas `chunksize` parameter. Validate chunk size during initial validation phase.
**Issue**: Formatting not applying
- **Solution**: Ensure you're modifying cell styles before saving, and use `.copy()` for style objects. Verify workbook is not in read-only mode.
SKILL.md
元数据
| name | spreadsheet-validated-exec |
|---|---|
| description | 在 Python 脚本执行电子表格任务之前进行必要的数据验证和源可访问性检查 |
电子表格任务的验证式 Python 执行
概述
该技能通过在处理开始前添加一个必要的数据验证阶段,扩展了针对电子表格操作的直接 Python 执行。这可以防止在不可访问的数据源上浪费迭代,并在数据无法访问时提供清晰的错误记录。
何时使用本技能
在以下情况下,对电子表格操作使用带验证的直接 run_shell 和 Python 脚本:
- 读取或写入包含多个工作表的复杂 Excel 文件
- 数据源已经验证为可访问
- 已识别好回退源,以防访问失败
- 应用公式、格式或数据转换
- 使用
openpyxl、pandas或类似库 - 操作涉及多个步骤,可能超出代理步骤限制
- 需要精确控制错误处理和调试
- 复杂脚本受益于基于文件的执行以获得更好的可靠性
为什么选择验证式直接执行?
除了标准的 shell_agent 限制之外,未经验证的数据访问会导致:
- 浪费尝试处理不存在数据的迭代
- 源文件不可访问时,错误消息不清晰
- 主源失败时无法优雅降级
- 缺少为什么数据操作无法完成的文档记录
带有 Python 验证的直接 run_shell 更可靠,因为它:
- 在单步中执行验证,没有迭代限制
- 为访问失败提供更清晰、即时的错误消息
- 处理复杂操作不受步骤限制
- 完全控制库导入和执行流程
- 先将脚本写入
.py文件可以避免 shell_agent 在 heredoc 方面的解析问题 - 记录所有访问失败以便故障排除和报告
第零阶段:数据源验证(必需)
在编写任何电子表格处理代码之前,先验证你的数据源是否可访问:
步骤 0.1:验证数据可用性
python
import os
import requests
from pathlib import Path
# For local files
def verify_local_file(file_path):
path = Path(file_path)
if not path.exists():
raise FileNotFoundError(f"Data file not found: {file_path}")
if not path.is_file():
raise ValueError(f"Path is not a file: {file_path}")
if path.stat().st_size == 0:
raise ValueError(f"Data file is empty: {file_path}")
print(f"✓ Local file verified: {file_path} ({path.stat().st_size} bytes)")
return True
# For remote URLs
def verify_remote_url(url, timeout=30):
try:
# Try HEAD request first (lighter than GET)
response = requests.head(url, timeout=timeout, allow_redirects=True)
if response.status_code == 405: # HEAD not allowed, try GET
response = requests.get(url, timeout=timeout, stream=True)
response.raise_for_status()
# Check content type if available
content_type = response.headers.get('Content-Type', '')
if 'text/html' in content_type and 'excel' not in url.lower():
print(f"⚠ Warning: URL may return HTML, not data file")
print(f"✓ Remote URL verified: {url} (Status: {response.status_code})")
return True
except requests.exceptions.SSLError as e:
print(f"✗ SSL Error: {str(e)}")
return False
except requests.exceptions.ConnectionError as e:
print(f"✗ Connection Error: {str(e)}")
return False
except requests.exceptions.Timeout as e:
print(f"✗ Timeout Error: {str(e)}")
return False
except Exception as e:
print(f"✗ Verification Failed: {str(e)}")
return False步骤 0.2:处理不可访问的数据源
如果主数据源不可用:
-
检查替代位置:
- 数据是否可从备份 URL 获取?
- 是否有可用的缓存/本地副本?
- 是否可以通过 API 而不是网络爬虫获取数据?
- 是否有其他数据提供者?
-
记录失败:
pythondef log_access_failure(source, error_type, timestamp=None): from datetime import datetime if not timestamp: timestamp = datetime.now().isoformat() error_log = { 'timestamp': timestamp, 'source': source, 'error_type': error_type, 'alternatives_attempted': [], 'resolution': 'pending' } # Save to error log file import json with open('data_access_errors.json', 'a') as f: f.write(json.dumps(error_log) + '\n') return error_log -
实现回退策略:
pythondef get_data_with_fallback(primary_source, fallback_sources): sources = [primary_source] + fallback_sources for i, source in enumerate(sources): print(f"Attempting source {i+1}/{len(sources)}: {source}") if source.startswith('http'): if verify_remote_url(source): return download_data(source) else: if verify_local_file(source): return load_local_data(source) print(f"Source {source} unavailable, trying next...") raise Exception(f"All {len(sources)} data sources unavailable")
步骤 0.3:执行前检查清单
在进入电子表格操作之前,确认如下:
- 主数据源可访问(文件存在/URL 有响应)
- 数据文件非空(有内容可处理)
- 所需权限就绪(文件未锁定、API 密钥有效)
- 如果主源失败,已确定备用源
- 错误日志记录机制已配置
如果任一检查未通过: 不要进入电子表格处理。报告阻塞问题,并选择:
- 先解决数据访问问题
- 切换至备用数据源
- 中止任务并提供清晰的错误文档
如何使用(带验证)
完整工作流模板
推荐模式:验证 → 处理 → 报告
bash
# Step 1: Write validation + processing script to file
cat > validated_spreadsheet_process.py << 'EOF'
import sys
import os
from pathlib import Path
import pandas as pd
from openpyxl import load_workbook
# === PHASE 0: VALIDATION ===
def validate_sources():
sources_to_check = [
('input_data.xlsx', 'local'),
# ('https://backup-source.com/data.xlsx', 'remote')
]
validated_sources = []
for source, source_type in sources_to_check:
try:
if source_type == 'local':
if not Path(source).exists():
print(f"ERROR: Local file not found: {source}")
continue
if Path(source).stat().st_size == 0:
print(f"ERROR: Local file is empty: {source}")
continue
validated_sources.append(source)
print(f"✓ Validated: {source}")
except Exception as e:
print(f"ERROR validating {source}: {e}")
if not validated_sources:
print("FATAL: No valid data sources available. Aborting.")
sys.exit(1)
return validated_sources
# === PHASE 1: PROCESSING ===
def process_data(source_file):
# Your spreadsheet operations here
pass
# === PHASE 2: REPORTING ===
def generate_report(success, details):
report = {
'status': 'success' if success else 'failed',
'details': details
}
print(f"Report: {report}")
return report
if __name__ == '__main__':
try:
# Validate first
sources = validate_sources()
# Process validated sources
for source in sources:
process_data(source)
generate_report(True, f"Processed {len(sources)} sources")
except Exception as e:
generate_report(False, str(e))
sys.exit(1)
EOF
# Step 2: Execute the validated script
python3 validated_spreadsheet_process.py模式 1:先将脚本写入文件(复杂操作)
适用于包含多个数据源的复杂多行脚本:
bash
# Step 1: Write the Python script to a file
cat > process_spreadsheet.py << 'EOF'
import openpyxl
from openpyxl import Workbook
from pathlib import Path
import sys
# Validate first
input_file = Path('file.xlsx')
if not input_file.exists():
print(f"ERROR: Input file not found: {input_file}")
sys.exit(1)
if input_file.stat().st_size == 0:
print(f"ERROR: Input file is empty: {input_file}")
sys.exit(1)
# Your spreadsheet code here
wb = openpyxl.load_workbook('file.xlsx')
# ... operations ...
wb.save('output.xlsx')
print('Success')
EOF
# Step 2: Execute the script
python3 process_spreadsheet.py模式 2:内联 Heredoc(包含验证的简单脚本)
当不使用 shell_agent 时,对于短脚本,可内联包含验证:
bash
# Include quick validation before processing
python3 << 'EOF'
import os
import sys
# Quick validation
input_file = 'data.xlsx'
if not os.path.exists(input_file):
print(f"ERROR: {input_file} not found")
sys.exit(1)
if os.path.getsize(input_file) == 0:
print(f"ERROR: {input_file} is empty")
sys.exit(1)
# Continue with processing...
EOF示例
示例 1:带验证的读取与转换
包含前置验证:
python
import pandas as pd
import sys
from pathlib import Path
# Validate input file exists and has content
input_path = Path('input.xlsx')
if not input_path.exists():
print(f"ERROR: Input file not found: {input_path}")
sys.exit(1)
if input_path.stat().st_size == 0:
print(f"ERROR: Input file is empty: {input_path}")
sys.exit(1)
# Now safe to process
df = pd.read_excel('input.xlsx', sheet_name='Revenue')
# Apply transformations
df['Net_Revenue'] = df['Gross_Revenue'] * (1 - df['Tax_Rate'])
# Save results
df.to_excel('output.xlsx', index=False, sheet_name='Processed')示例 2:含数据源验证的多工作表处理
python
from openpyxl import load_workbook
import sys
from pathlib import Path
# Validate workbook accessibility
wb_path = Path('tour_data.xlsx')
if not wb_path.exists():
print(f"ERROR: Workbook not found: {wb_path.absolute()}")
sys.exit(1)
try:
wb = load_workbook('tour_data.xlsx')
except Exception as e:
print(f"ERROR: Cannot open workbook: {e}")
print("Possible causes: file locked, corrupted, or wrong format")
sys.exit(1)
# Iterate through sheets
for sheet_name in wb.sheetnames:
ws = wb[sheet_name]
# Apply formatting or calculations
for row in ws.iter_rows(min_row=2, max_col=5):
# Process cells
pass
wb.save('tour_data_processed.xlsx')示例 3:带预检查的格式化
python
from openpyxl import load_workbook
from openpyxl.styles import Font, PatternFill, Alignment
from pathlib import Path
import sys
# Pre-check file
report_path = Path('report.xlsx')
if not report_path.exists():
print(f"ERROR: Report file not found: {report_path}")
sys.exit(1)
wb = load_workbook('report.xlsx')
ws = wb.active
# Apply header styling
header_fill = PatternFill(start_color='4472C4', fill_type='solid')
header_font = Font(bold=True, color='FFFFFF')
for cell in ws[1]:
cell.fill = header_fill
cell.font = header_font
cell.alignment = Alignment(horizontal='center')
wb.save('report_formatted.xlsx')示例 4:带日志记录的全面错误处理
python
import sys
import json
from datetime import datetime
from pathlib import Path
from openpyxl import load_workbook
def log_error(source, error_msg, error_type):
"""Log errors for later analysis and reporting"""
error_record = {
'timestamp': datetime.now().isoformat(),
'source': source,
'error_type': error_type,
'message': error_msg
}
with open('processing_errors.log', 'a') as f:
f.write(json.dumps(error_record) + '\n')
return error_record
try:
# Validate file before opening
if not Path('data.xlsx').exists():
log_error('data.xlsx', 'File not found', 'ValidationError')
raise FileNotFoundError('data.xlsx')
wb = load_workbook('data.xlsx')
ws = wb.active
# Your operations here
value = ws['A1'].value
wb.save('output.xlsx')
print(f"Success: Processed {ws.max_row} rows")
except FileNotFoundError as e:
log_error('data.xlsx', str(e), 'FileNotFoundError')
print(f"ERROR: {str(e)}", file=sys.stderr)
print("ACTION: Verify file path and permissions")
sys.exit(1)
except PermissionError as e:
log_error('data.xlsx', str(e), 'PermissionError')
print(f"ERROR: Permission denied - file may be open in another application")
print("ACTION: Close the file in other applications and retry")
sys.exit(1)
except Exception as e:
log_error('data.xlsx', str(e), type(e).__name__)
print(f"ERROR: {str(e)}", file=sys.stderr)
sys.exit(1)最佳实践(增强)
- 总是先验证再处理:在任何电子表格操作前检查数据源可用性
- 记录所有访问失败:文档化数据源何时以及为何变得不可用
- 提前确定回退方案:在开始前了解你的备用数据源
- 验证错误时快速失败:不要在不可能的操作上浪费迭代
- 对于复杂脚本优先使用基于文件的执行:先写入
.py文件,再通过run_shell执行 - 只导入需要的库以减少执行时间
- 打印清晰的 成功/错误 消息以便调试
- 对复杂的多步转换保存中间结果
- 在扩展到大型电子表格前先用小数据测试
- 当需要两者时,使用 pandas 进行数据操作,openpyxl 进行格式化
- 执行后如果没有重用需求,清理临时脚本文件
数据访问失败协议
当主源失败时
-
立即行动:
- 记录失败的时间戳、数据源和错误类型
- 检查错误是暂时的(超时)还是永久的(404,文件不存在)
- 如果有备用源,尝试备用源
-
对于暂时性失败(超时,5xx 错误):
pythondef retry_with_backoff(operation, max_retries=3, base_delay=1): import time for attempt in range(max_retries): try: return operation() except (TimeoutError, ConnectionError) as e: if attempt == max_retries - 1: raise delay = base_delay * (2 ** attempt) print(f"Retry {attempt + 1}/{max_retries} after {delay}s delay") time.sleep(delay) -
对于永久性失败(文件未找到,404):
- 立即切换至已确定的备用源
- 记录主源失败
- 如果有备用源则继续,否则干净地中止
错误报告模板
python
def report_data_access_issue(issue_type, source, details, alternatives_tried=None):
"""Standardize error reporting for data access failures"""
from datetime import datetime
report = f"""
=== DATA ACCESS FAILURE REPORT ===
Type: {issue_type}
Source: {source}
Time: {datetime.now().isoformat()}
Details: {details}
Alternatives Attempted: {alternatives_tried or 'None'}
Recommended Action: {get_recommended_action(issue_type)}
==================================
"""
print(report)
return report
def get_recommended_action(issue_type):
actions = {
'FileNotFound': 'Verify file path, check if file was moved or deleted',
'ConnectionError': 'Check network connectivity, try alternative source',
'Timeout': 'Increase timeout, check source server status',
'PermissionError': 'Check file permissions, ensure file not open elsewhere',
'SSLError': 'Verify SSL certificate, try HTTP if appropriate',
'default': 'Review error details, check source availability'
}
return actions.get(issue_type, actions['default'])何时不使用本技能
- 简单的单单元格读取/写入(验证开销不划算)
- 需要用户交互输入的操作
- 需要代理迭代优化方案的任务
- 当无法验证数据源时(仅跳转到错误处理)
常用库
| 库 | 最适合 | 也用于 |
|---|---|---|
requests | 网络请求 | 源 URL 验证,健康检查 |
openpyxl | 读取/写入 .xlsx 文件,格式化,公式 | - |
pandas | 数据操作,分析,合并数据集 | 对大文件分块读取 |
xlrd | 读取旧版 .xls 文件(只读) | - |
xlsxwriter | 创建带高级格式的新 .xlsx 文件 | - |
故障排除(增强)
问题:数据源不可用/文件未找到
- 解决方案:处理前先运行验证检查。如果验证失败:
- 检查路径/URL 是否正确
- 检查文件是否被移动、重命名或删除
- 如果有备用源,尝试备用源
- 记录失败并干净地中止,而不是继续尝试
问题:数据源昨天还能访问,今天不行
- 解决方案:实现备用源并记录变化。考虑:
- 为关键数据源设置监控
- 在可能时将重要数据缓存到本地
- 掌握数据源维护者的联系信息
问题:验证通过但处理失败
- 解决方案:验证只检查可访问性,不检查内容有效性。添加:
- 模式验证(检查预期列是否存在)
- 内容验证(检查数据类型、范围)
- 样例验证(读取前几行以确认结构)
问题:当使用 shell_agent 时 heredoc 语法出现 'unknown error'
- 解决方案:先将带完整验证逻辑的 Python 脚本写入
.py文件,然后用python3 script.py执行。当 shell_agent 是执行器时,这比内联 heredoc 可靠得多。
问题:FileNotFoundError(验证通过后)
- 解决方案:文件可能在验证和处理之间被移动/删除,或者路径是相对路径而工作目录发生了变化。使用绝对路径。
问题:PermissionError
- 解决方案:确保文件未在其他应用程序中打开。在 Linux/Mac 上,使用
lsof | grep filename检查。
问题:远程源上的 ConnectionError / Timeout
- 解决方案:检查网络连接,尝试增加超时,使用指数退避的重试逻辑,或切换至缓存/本地副本(如果有)。
问题:SSL 证书错误
- 解决方案:验证证书是否有效。仅作为可信内部源的最后手段:
requests.get(url, verify=False)—— 但将此记录为安全顾虑
问题:所有备用源都失败
- 解决方案:中止并生成完整的错误报告,记录:
- 所有已尝试的数据源
- 每个源的错误类型
- 故障的时间戳
- 给操作员的推荐后续步骤
问题:大文件 MemoryError
- 解决方案:使用 pandas 的
chunksize参数分块处理数据。在初始验证阶段验证块大小。
问题:格式未应用
- 解决方案:确保在保存前修改了单元格样式,并对样式对象使用
.copy()。验证工作簿不是只读模式。