news-collector: process_raw 4-동시 병렬화로 처리 용량 2배

배치 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) <noreply@anthropic.com>

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:
+        async def _one(r):
-            item = {
+            nonlocal ok
-                "title": r["title"], "url": r["url"],
+            async with sem:
-                "source": r["source"] or "네이버금융",
+                item = {
-                "published_at": r["published_at_text"] or "",
+                    "title": r["title"], "url": r["url"],
-                "content": "",
+                    "source": r["source"] or "네이버금융",
-                "stock_code": r["stock_code"],
+                    "published_at": r["published_at_text"] or "",
-            }
+                    "content": "",
-            try:
+                    "stock_code": r["stock_code"],
-                if await is_dup(item["title"], item["url"]):
+                }
-                    stats.duplicates += 1
+                try:
-                else:
+                    if await is_dup(item["title"], item["url"]):
-                    stats.collected += 1
+                        stats.duplicates += 1
-                    res = await pipeline(item, c)
+                    else:
-                    if res == "ok": ok += 1; stats.processed += 1
+                        stats.collected += 1
-                    elif res == "noise": stats.noise += 1
+                        res = await pipeline(item, c)
-                    elif res == "off_topic": stats.off_topic += 1
+                        if res == "ok": ok += 1; stats.processed += 1
-                async with pg_pool.acquire() as conn:
+                        elif res == "noise": stats.noise += 1
-                    await conn.execute(
+                        elif res == "off_topic": stats.off_topic += 1
-                        "UPDATE news_raw SET processed=true, processed_at=NOW() WHERE id=$1",
+                    async with pg_pool.acquire() as conn:
-                        r["id"])
+                        await conn.execute(
-            except Exception as e:
+                            "UPDATE news_raw SET processed=true, processed_at=NOW() WHERE id=$1",
-                stats.errors += 1
+                            r["id"])
-                logger.warning("process_raw.err", id=r["id"], error=str(e))
+                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건 처리 (백로그 소화용)
+    # raw 뉴스 분석: 30분마다 batch 200건 처리 (병렬화 4-동시로 30분 내 완주)
-    scheduler.add_job(job_process_raw,"cron",hour="*",minute="*/30",
+    scheduler.add_job(job_process_raw,"cron",minute="*/30",
-                      id="process_raw",replace_existing=True,kwargs={"batch_size":200})
+                      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,