_tmp_cap_yield.py 2.0 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849
  1. # -*- coding: utf-8 -*-
  2. """两次验证码之间的爬取量: 用交叉验证日志(带时间戳)分桶统计"""
  3. import json
  4. import re
  5. from collections import Counter
  6. from datetime import datetime
  7. from pathlib import Path
  8. recs = []
  9. for ln in Path('logs/cross_validate.jsonl').read_text(encoding='utf-8').splitlines():
  10. if not ln.strip():
  11. continue
  12. r = json.loads(ln)
  13. m = re.search(r'screenshots[/\\]+([^/\\]+)[/\\]step3', r.get('image', ''))
  14. r['dev'] = m.group(1) if m else '?'
  15. r['dt'] = datetime.strptime(r['ts'], '%Y-%m-%d %H:%M:%S')
  16. recs.append(r)
  17. print(f'交叉日志共 {len(recs)} 条')
  18. print('按设备分布:', dict(Counter(r['dev'] for r in recs)))
  19. def parse_day(day):
  20. out = {}
  21. for f in Path('logs').glob('captcha_log_*.txt'):
  22. dev = f.stem.replace('captcha_log_', '')
  23. ts = [datetime.strptime(m.group(1), '%Y-%m-%d %H:%M:%S')
  24. for ln in f.read_text(encoding='utf-8', errors='ignore').splitlines()
  25. if ln.startswith(day) for m in [re.match(r'(\S+ \S+)', ln)] if m]
  26. if ts:
  27. out[dev] = sorted(ts)
  28. return out
  29. for day in ['2026-09-15', '2026-09-14']:
  30. caps = parse_day(day)
  31. day_recs = [r for r in recs if r['dt'].strftime('%Y-%m-%d') == day]
  32. print(f'\n===== {day} 两次验证码之间的爬取量 =====')
  33. for dev, ts in sorted(caps.items(), key=lambda x: -len(x[1])):
  34. dev_recs = [r for r in day_recs if r['dev'] == dev]
  35. print(f'{dev} ({len(ts)}次验证码, 交叉解析{len(dev_recs)}屏):')
  36. for i in range(len(ts)):
  37. t0 = ts[i]
  38. t1 = ts[i + 1] if i + 1 < len(ts) else ts[i].replace(hour=23, minute=59)
  39. inwin = [r for r in dev_recs if t0 <= r['dt'] < t1]
  40. rows = sum(r.get('final', 0) for r in inwin)
  41. print(f' 第{i+1}次{t0.strftime("%H:%M")} → {"第" + str(i + 2) + "次" + t1.strftime("%H:%M") if i + 1 < len(ts) else "结束"}: 解析{len(inwin)}屏, 识别{rows}行')
  42. if not ts:
  43. print(' (无验证码)')