Crawl4AI Python 教程：限定采集范围、正文校验与来源追踪

python3.11 -m venv .venv
source .venv/bin/activate
python -m pip install 'crawl4ai==0.9.5' 'playwright==1.64.1' 'pyOpenSSL==25.3.0' 'cryptography==46.0.3'
python -m pip check

from http.server import BaseHTTPRequestHandler,ThreadingHTTPServer

ARTICLE='''<h1>测试产品退款规则</h1><p>REFUND-731：购买后七天内可以申请退款。金额以原始订单记录为准。</p><p>请保留订单编号与申请日期，先核对来源版本，再交由人工确认。</p>'''
class Handler(BaseHTTPRequestHandler):
 def do_GET(self):
  if self.path=='/redirect':
   self.send_response(302);self.send_header('Location','/outside');self.end_headers();return
  status=404 if self.path=='/missing' else 200
  content=ARTICLE if self.path=='/article' else '<h1>其他页面</h1><p>OUTSIDE-731</p>'
  if self.path=='/dynamic':content='<h1>动态正文</h1><div id="result">等待正文</div>'
  if self.path=='/empty':content=''
  html='<html><head><meta charset="utf-8"><title>本地测试页</title></head><body><nav>NAV-NOISE</nav><article>'+content+'</article><footer>FOOTER-NOISE</footer>'
  if self.path=='/dynamic':html+="<script>setTimeout(()=>document.querySelector('#result').innerHTML='<p>DYNAMIC-731：动态内容已加载，请保留采集时间与来源。</p>',150)</script>"
  html+='</body></html>';data=html.encode()
  self.send_response(status);self.send_header('Content-Type','text/html; charset=utf-8');self.send_header('Content-Length',str(len(data)));self.end_headers();self.wfile.write(data)
 def log_message(self,*args):pass

if __name__=='__main__':
 server=ThreadingHTTPServer(('127.0.0.1',8765),Handler)
 print('Fixture server: http://127.0.0.1:8765',flush=True)
 try:server.serve_forever()
 finally:server.server_close()

python fixture_server.py

import os
os.environ.setdefault('CRAWL4_AI_BASE_DIRECTORY','/private/tmp/crawl4ai-demo-runtime')
import asyncio,hashlib,json
from datetime import datetime,timezone
from importlib.metadata import version
from pathlib import Path
from urllib.parse import urlsplit
from crawl4ai import AsyncWebCrawler,BrowserConfig,CrawlerRunConfig,CacheMode

BASE='http://127.0.0.1:8765'
PATHS=['/article','/dynamic','/missing','/empty','/redirect']

def approved(url):
 p=urlsplit(url)
 if p.scheme!='http' or p.hostname!='127.0.0.1' or p.port!=8765:
  raise ValueError('demo origin not approved')
 if p.username or p.password or p.query or p.fragment or p.path not in PATHS:
  raise ValueError('demo URL not approved')
 return url

async def main():
 records=[]
 async with AsyncWebCrawler(config=BrowserConfig(headless=True,chrome_channel='chrome',ignore_https_errors=False,verbose=False)) as crawler:
  for path in PATHS:
   url=approved(BASE+path)
   row=dict(source_url=url,collected_at=datetime.now(timezone.utc).isoformat(),status='failed')
   try:
    cfg=CrawlerRunConfig(cache_mode=CacheMode.BYPASS,css_selector='article',
      excluded_tags=['nav','footer'],word_count_threshold=0,page_timeout=10000,
      wait_for="css:#result p" if path=='/dynamic' else None,verbose=False)
    result=await asyncio.wait_for(crawler.arun(url=url,config=cfg),timeout=15)
    row.update(http_status=result.status_code,final_url=result.redirected_url or result.url)
    if row['final_url']!=url:raise ValueError('redirect needs review')
    if not result.success or result.status_code!=200:raise ValueError('HTTP or crawl failed')
    markdown=result.markdown.raw_markdown if result.markdown else ''
    if not markdown.strip():raise ValueError('empty article markdown')
    row.update(status='ok',markdown=markdown,
      markdown_sha256=hashlib.sha256(markdown.encode()).hexdigest())
   except Exception as exc:row.update(error_type=type(exc).__name__,error=str(exc)[:400])
   records.append(row)
 out=dict(versions={n:version(n) for n in ['crawl4ai','playwright']},
   records=records,summary=dict(attempted=len(records),ok=sum(x['status']=='ok' for x in records)))
 Path('crawl-output.json').write_text(json.dumps(out,ensure_ascii=False,indent=2))
 print('Report saved to crawl-output.json; inspect every record.')

if __name__=='__main__':asyncio.run(main())

python crawl_local.py

import hashlib
import json
from pathlib import Path

out = json.loads(Path('crawl-output.json').read_text())
rows = {r['source_url'].split('/')[-1]: r for r in out['records']}
assert out['summary'] == {'attempted': 5, 'ok': 2}
for key, marker in [('article', 'REFUND-731'), ('dynamic', 'DYNAMIC-731')]:
    row = rows[key]
    assert row['status'] == 'ok' and marker in row['markdown']
    assert 'NAV-NOISE' not in row['markdown']
    assert 'FOOTER-NOISE' not in row['markdown']
    assert row['markdown_sha256'] == hashlib.sha256(row['markdown'].encode()).hexdigest()
for key in ['missing', 'empty', 'redirect']:
    assert rows[key]['status'] == 'failed' and rows[key].get('error')
assert all(r.get('collected_at') and r.get('final_url') for r in rows.values())
print('Report checked; failures remain visible and no automatic indexing is performed.')

python check_report.py