From 3db047bade125da47f6aaa93f9dbb0fdc6b8d5c9 Mon Sep 17 00:00:00 2001 From: kyu Date: Mon, 25 May 2026 00:48:39 +0900 Subject: [PATCH] =?UTF-8?q?news-collector:=20process=5Fraw=204-=EB=8F=99?= =?UTF-8?q?=EC=8B=9C=20=EB=B3=91=EB=A0=AC=ED=99=94=EB=A1=9C=20=EC=B2=98?= =?UTF-8?q?=EB=A6=AC=20=EC=9A=A9=EB=9F=89=202=EB=B0=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 배치 200건이 직렬 처리로 30분 초과해 매시 1회로 후퇴, 일 4,800건 처리에 그쳐 일 신규 5,500건 유입을 못 따라잡고 백로그 16K가 정체 상태. asyncio.Semaphore(4)로 병렬화(OLLAMA_NUM_PARALLEL=4와 매칭) → 한 배치 6분 20초, 스케줄러를 minute="*/30"으로 복원 (max_instances=1로 중첩 방지). 일 처리 용량 9,600건 → 백로그 자연 소화. Co-Authored-By: Claude Opus 4.7 (1M context) --- news-collector/main.py | 62 +++++++++++++++++++++++------------------- 1 file changed, 34 insertions(+), 28 deletions(-) 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,