菜单
Performance optimization patterns for Python scraper pipelines — connection pooling, memory efficiency, async I/O, and batch operations for high-throughput data ingestion.
用 Codex 或 Claude 帮你安装 复制这段 Prompt，粘贴到 Codex、Claude 或其他助手里，让它检查 Skill 页面并帮你完成安装。
基于 SOC 职业分类
Use this skill when expanding the scraper network to new Florida counties. Contains the 67-county expansion roadmap, prioritization framework, site reconnaissance procedure, and lead analysis strategy. The goal is total Florida coverage for maximum bail bond lead generation.
Use this skill when creating a new county arrest scraper, modifying an existing solver, or adding a new county to the scraping pipeline. Contains the canonical architecture, naming conventions, and integration checklist.
Non-negotiable rules, safe refactoring patterns, shared code modification guidelines, secrets management, config hierarchy, and schema change process for the swfl-arrest-scrapers repo. Read this before modifying any core infrastructure.
Testing patterns, fixture guidelines, parser test templates, smoke tests, and red-green TDD methodology for the scraper pipeline. Use when writing tests for new counties, verifying core/ changes, or debugging test failures.
Google Sheets: Read and write spreadsheets via the Sheets API v4.
Systematically strengthen scraper pipelines against network failures, anti-bot measures, data edge cases, and real-world brittleness that breaks idealized scraper designs.
| name | python-performance-optimization |
| description | Performance optimization patterns for Python scraper pipelines — connection pooling, memory efficiency, async I/O, and batch operations for high-throughput data ingestion. |
# Reuse sessions across requests — don't create new connections per page
session = requests.Session()
session.headers.update(HEADERS)
# Session persists TCP connections, cookies, and auth
# Process records in streams, not full lists
def process_records(records):
for record in records:
yield normalize(record) # Generator, not list comprehension
# Use __slots__ for high-volume record objects
class ArrestRecord:
__slots__ = ['name', 'booking_number', 'dob', 'charges', 'bond']
# ❌ Bad: One API call per record
for record in records:
sheet.append_row(record)
# ✅ Good: Batch all records in single API call
sheet.batch_update([{
'range': f'{tab}!A2',
'values': [record_to_row(r) for r in records]
}])
# O(1) lookup instead of O(n) scan
existing_ids = set(sheet.col_values(booking_number_col))
new_records = [r for r in records if r['Booking_Number'] not in existing_ids]
from concurrent.futures import ThreadPoolExecutor
def scrape_detail_page(url):
return session.get(url, timeout=30).text
with ThreadPoolExecutor(max_workers=3) as executor:
results = list(executor.map(scrape_detail_page, detail_urls))
# Time the entire pipeline
time python counties/lee/runner.py --days-back 1
# Profile bottlenecks
python -m cProfile -s cumulative counties/lee/runner.py