diff --git a/news-collector/main.py b/news-collector/main.py index 2bacf72..6a55e78 100644 --- a/news-collector/main.py +++ b/news-collector/main.py @@ -656,7 +656,8 @@ async def job_historical_raw(count: int = 0, max_pages: int = 100): async def job_process_raw(batch_size: int = 200): - """news_raw의 미처리 행을 batch_size만큼 EXAONE 분석 → news_analysis로 이동""" + """news_raw의 미처리 행을 batch_size만큼 EXAONE 분석 → news_analysis로 이동. + OLLAMA_NUM_PARALLEL=4와 매칭되는 4-동시 병렬 처리.""" async with pg_pool.acquire() as conn: rows = await conn.fetch(""" SELECT id, stock_code, title, url, source, published_at_text @@ -666,31 +667,35 @@ async def job_process_raw(batch_size: int = 200): if not rows: logger.info("process_raw.empty"); return 0 ok = 0 + sem = asyncio.Semaphore(4) async with httpx.AsyncClient() as c: - for r in rows: - item = { - "title": r["title"], "url": r["url"], - "source": r["source"] or "네이버금융", - "published_at": r["published_at_text"] or "", - "content": "", - "stock_code": r["stock_code"], - } - try: - if await is_dup(item["title"], item["url"]): - stats.duplicates += 1 - else: - stats.collected += 1 - res = await pipeline(item, c) - if res == "ok": ok += 1; stats.processed += 1 - elif res == "noise": stats.noise += 1 - elif res == "off_topic": stats.off_topic += 1 - async with pg_pool.acquire() as conn: - await conn.execute( - "UPDATE news_raw SET processed=true, processed_at=NOW() WHERE id=$1", - r["id"]) - except Exception as e: - stats.errors += 1 - logger.warning("process_raw.err", id=r["id"], error=str(e)) + async def _one(r): + nonlocal ok + async with sem: + item = { + "title": r["title"], "url": r["url"], + "source": r["source"] or "네이버금융", + "published_at": r["published_at_text"] or "", + "content": "", + "stock_code": r["stock_code"], + } + try: + if await is_dup(item["title"], item["url"]): + stats.duplicates += 1 + else: + stats.collected += 1 + res = await pipeline(item, c) + if res == "ok": ok += 1; stats.processed += 1 + elif res == "noise": stats.noise += 1 + elif res == "off_topic": stats.off_topic += 1 + async with pg_pool.acquire() as conn: + await conn.execute( + "UPDATE news_raw SET processed=true, processed_at=NOW() WHERE id=$1", + r["id"]) + except Exception as e: + stats.errors += 1 + logger.warning("process_raw.err", id=r["id"], error=str(e)) + await asyncio.gather(*(_one(r) for r in rows)) logger.info("process_raw.batch_done", batch=len(rows), processed_ok=ok) return ok @@ -818,9 +823,10 @@ async def startup(): scheduler.add_job(job_rss,"cron",day_of_week="sat,sun",hour="8-22",minute="*/15",id="rss_weekend",replace_existing=True) # 평일 마켓 (네이버 금융): 9-17시 10분마다 scheduler.add_job(job_market,"cron",day_of_week="mon-fri",hour="9-17",minute="*/10",id="market",replace_existing=True) - # raw 뉴스 분석: 24시간 30분마다 batch 200건 처리 (백로그 소화용) - scheduler.add_job(job_process_raw,"cron",hour="*",minute="*/30", - id="process_raw",replace_existing=True,kwargs={"batch_size":200}) + # raw 뉴스 분석: 30분마다 batch 200건 처리 (병렬화 4-동시로 30분 내 완주) + scheduler.add_job(job_process_raw,"cron",minute="*/30", + id="process_raw",replace_existing=True, + max_instances=1, kwargs={"batch_size":200}) # 매주 일요일 새벽 2시 raw 백필 (전체 활성종목) scheduler.add_job(job_historical_raw,"cron",day_of_week="sun",hour="2",minute="0", id="historical_raw_weekly",replace_existing=True,