▮ REFACTORED PATCH
# Fix 1: count skipped items at filter time
async def _process_items(self):
done = self.checkpoint.completed | self.checkpoint.failed
to_process = []
for item in self.items:
if item in done: self.results['skipped'] += 1 # count here
else: to_process.append(item)
tasks = [asyncio.create_task(self._process_item(i)) for i in to_process]
for coro in asyncio.as_completed(tasks): await coro
self._update_summary()
save_checkpoint(self.checkpoint, self.checkpoint_path)
print(json.dumps({'succeeded': self.results['succeeded'], 'failed': self.results['failed'],
'skipped': self.results['skipped'], 'total': len(self.items),
'elapsed_ms': int((time.time()-self.start_time)*1000)}))
# Fix 2: bounded submission + checkpoint flush on shutdown
pending = [i for i in self.items if i not in done]
sem = self.semaphore
async def run_one(item):
async with sem:
if not self.running: return
await self._process_item(item)
bg = [asyncio.create_task(run_one(i)) for i in pending]
try:
for t in asyncio.as_completed(bg): await t
finally:
save_checkpoint(self.checkpoint, self.checkpoint_path) # flush on any exit