Update npc_core.py

npc_core.py

@@ -2,6 +2,7 @@
 All NPC-related logic extracted from app.py for modularity.
 """
 import aiosqlite, asyncio, os, random, logging, json, re
+from contextlib import asynccontextmanager
 from datetime import datetime, timezone, timedelta, date
 from typing import List, Dict, Optional, Tuple
 from groq import Groq
@@ -13,6 +14,36 @@ from npc_trading import (
 )
 logger = logging.getLogger(__name__)
 KST = timezone(timedelta(hours=9))
+
+# ══════════════════════════════════════════════════
+# Global DB Connection Pool — database locked 방지
+# ══════════════════════════════════════════════════
+_db_semaphore = asyncio.Semaphore(5)       # 최대 5개 동시 DB 연결
+_db_write_lock = asyncio.Semaphore(2)      # 최대 2개 동시 쓰기
+
+@asynccontextmanager
+async def get_db(db_path, write=False, timeout=30.0):
+    """Semaphore-controlled DB connection — prevents 'database is locked'"""
+    sem = _db_write_lock if write else _db_semaphore
+    async with sem:
+        async with aiosqlite.connect(db_path, timeout=timeout) as db:
+            await db.execute("PRAGMA busy_timeout=30000")
+            await db.execute("PRAGMA journal_mode=WAL")
+            yield db
+            if write:
+                try:
+                    await db.commit()
+                except Exception:
+                    pass
+
+@asynccontextmanager
+async def get_db_read(db_path, timeout=15.0):
+    """Read-only shortcut — higher concurrency"""
+    async with _db_semaphore:
+        async with aiosqlite.connect(db_path, timeout=timeout) as db:
+            await db.execute("PRAGMA busy_timeout=15000")
+            await db.execute("PRAGMA journal_mode=WAL")
+            yield db
 def get_current_time_kst():
     return datetime.now(KST)
 def get_current_date_kst():
@@ -267,17 +298,16 @@ async def calculate_like_economics(post_likes: int, liker_total_likes_given: int
 # ──────────────────────────────────────────────
 # DB Helpers
 # ──────────────────────────────────────────────
-async def db_write_with_retry(db, sql, params=None, max_retries=5):
+async def db_write_with_retry(db, sql, params=None, max_retries=7):
     for attempt in range(max_retries):
         try:
-            if params:
-                cursor = await db.execute(sql, params)
-            else:
-                cursor = await db.execute(sql)
+            cursor = await db.execute(sql, params) if params else await db.execute(sql)
             return cursor
         except Exception as e:
             if "locked" in str(e).lower() and attempt < max_retries - 1:
-                await asyncio.sleep(random.uniform(0.5, 2.0) * (attempt + 1))
+                wait = random.uniform(0.3, 1.5) * (2 ** attempt)  # exponential backoff
+                logger.debug(f"DB locked (attempt {attempt+1}/{max_retries}), retry in {wait:.1f}s")
+                await asyncio.sleep(wait)
                 continue
             raise
 async def deduct_gpu(db, identifier: str, amount: float, tx_type: str, is_npc: bool):
@@ -422,8 +452,7 @@ class NPCMemorySystem:
     def __init__(self, db_path: str):
         self.db_path = db_path
     async def store_post_memory(self, agent_id: str, post_id: int, title: str, content: str, board_key: str):
-        async with aiosqlite.connect(self.db_path, timeout=60.0) as db:
-            await db.execute("PRAGMA busy_timeout=60000")
+        async with get_db(self.db_path) as db:
             topic = self._extract_topic(content, board_key)
             await db.execute("""INSERT INTO npc_memory (agent_id, memory_type, content, metadata) VALUES (?, 'post', ?, ?)""",
                 (agent_id, f"{title}\n{content[:500]}", json.dumps({
@@ -431,8 +460,7 @@ class NPCMemorySystem:
                     'initial_time': datetime.now().isoformat()})))
             await db.commit()
     async def update_feedback(self, agent_id: str, post_id: int):
-        async with aiosqlite.connect(self.db_path, timeout=60.0) as db:
-            await db.execute("PRAGMA busy_timeout=60000")
+        async with get_db(self.db_path) as db:
             cursor = await db.execute("SELECT likes_count, dislikes_count, comment_count FROM posts WHERE id=? AND author_agent_id=?", (post_id, agent_id))
             stats = await cursor.fetchone()
             if stats:
@@ -442,8 +470,7 @@ class NPCMemorySystem:
                     (success_score, likes, dislikes, comments, success_score, agent_id, post_id))
                 await db.commit()
     async def learn_from_feedback(self, agent_id: str):
-        async with aiosqlite.connect(self.db_path, timeout=60.0) as db:
-            await db.execute("PRAGMA busy_timeout=60000")
+        async with get_db(self.db_path) as db:
             cursor = await db.execute("SELECT content, metadata FROM npc_memory WHERE agent_id = ? AND memory_type = 'post' AND importance_score > 0.7 ORDER BY importance_score DESC LIMIT 10", (agent_id,))
             successful_posts = await cursor.fetchall()
             if successful_posts:
@@ -468,13 +495,11 @@ class NPCMemorySystem:
         return {'preferred_topics': list(set(topics))[:5], 'successful_styles': list(set(styles)),
                 'high_engagement_keywords': list(set(keywords))[:20]}
     async def get_memory_context(self, agent_id: str, limit: int = 5) -> List:
-        async with aiosqlite.connect(self.db_path, timeout=60.0) as db:
-            await db.execute("PRAGMA busy_timeout=60000")
+        async with get_db(self.db_path) as db:
             cursor = await db.execute("SELECT memory_type, content, metadata, importance_score FROM npc_memory WHERE agent_id = ? ORDER BY importance_score DESC, created_at DESC LIMIT ?", (agent_id, limit))
             return await cursor.fetchall()
     async def get_learned_patterns(self, agent_id: str) -> Optional[Dict]:
-        async with aiosqlite.connect(self.db_path, timeout=60.0) as db:
-            await db.execute("PRAGMA busy_timeout=60000")
+        async with get_db(self.db_path) as db:
             cursor = await db.execute("SELECT content FROM npc_memory WHERE agent_id = ? AND memory_type = 'learned_pattern' ORDER BY created_at DESC LIMIT 1", (agent_id,))
             result = await cursor.fetchone()
             if result:
@@ -499,8 +524,7 @@ class NPCLearningScheduler:
         self.db_path = db_path; self.memory_system = NPCMemorySystem(db_path)
     async def daily_learning_cycle(self):
         logger.info("🧠 Daily learning cycle starting")
-        async with aiosqlite.connect(self.db_path, timeout=60.0) as db:
-            await db.execute("PRAGMA busy_timeout=60000")
+        async with get_db(self.db_path) as db:
             cursor = await db.execute("SELECT agent_id FROM npc_agents WHERE is_active=1")
             agents = await cursor.fetchall()
             learned_count = 0
@@ -516,15 +540,13 @@ class NPCLearningScheduler:
             logger.info(f"✅ Daily learning complete ({learned_count} agents)")
     async def _update_yesterday_feedback(self, agent_id: str):
         yesterday = (get_current_date_kst() - timedelta(days=1)).isoformat()
-        async with aiosqlite.connect(self.db_path, timeout=60.0) as db:
-            await db.execute("PRAGMA busy_timeout=60000")
+        async with get_db(self.db_path) as db:
             cursor = await db.execute("SELECT id FROM posts WHERE author_agent_id = ? AND DATE(created_at) = ?", (agent_id, yesterday))
             posts = await cursor.fetchall()
             for (post_id,) in posts:
                 await self.memory_system.update_feedback(agent_id, post_id)
     async def _cleanup_old_memories(self, agent_id: str):
-        async with aiosqlite.connect(self.db_path, timeout=60.0) as db:
-            await db.execute("PRAGMA busy_timeout=60000")
+        async with get_db(self.db_path) as db:
             await db.execute("DELETE FROM npc_memory WHERE agent_id = ? AND id NOT IN (SELECT id FROM npc_memory WHERE agent_id = ? ORDER BY importance_score DESC, created_at DESC LIMIT 100)", (agent_id, agent_id))
             await db.commit()
 # ──────────────────────────────────────────────
@@ -1060,8 +1082,7 @@ class NPCGenerator:
     def __init__(self, db_path: str):
         self.db_path = db_path
     async def generate_npc_population(self, target_count: int = TARGET_NPC_COUNT):
-        async with aiosqlite.connect(self.db_path, timeout=60.0) as db:
-            await db.execute("PRAGMA busy_timeout=60000")
+        async with get_db(self.db_path) as db:
             cursor = await db.execute("SELECT COUNT(*) FROM npc_agents")
             current = (await cursor.fetchone())[0]
             if current >= target_count:
@@ -1285,8 +1306,7 @@ def build_chat_msg(msg_type, username, identity, style, ticker, price, chg, is_b
         return random.choice(msgs), target[0]
     return f"Interesting day in the markets. Watching {ticker} closely {emoji}", reply_to
 async def generate_npc_chat_messages(db_path: str):
-    async with aiosqlite.connect(db_path, timeout=30.0) as db:
-        await db.execute("PRAGMA busy_timeout=30000")
+    async with get_db(db_path) as db:
         cursor = await db.execute("""
             SELECT agent_id, username, ai_identity, mbti, gpu_dollars
             FROM npc_agents WHERE is_active=1 ORDER BY RANDOM() LIMIT ?
@@ -1534,8 +1554,7 @@ def build_research_report(agent_id, username, identity, mbti, ticker, company,
 async def generate_npc_research_reports(db_path: str, ai_client=None):
     """Top 30 NPC가 랜덤 티커에 대해 심층 리서치 생산"""
     try:
-        async with aiosqlite.connect(db_path, timeout=30.0) as db:
-            await db.execute("PRAGMA busy_timeout=30000")
+        async with get_db(db_path) as db:
             cursor = await db.execute("""
                 SELECT a.agent_id, a.username, a.ai_identity, a.mbti,
                     COUNT(*) as total,
@@ -1599,8 +1618,7 @@ async def generate_npc_research_reports(db_path: str, ai_client=None):
 async def npc_auto_purchase_research(db_path: str):
     """NPC가 자동으로 고품질 리서치 구매"""
     try:
-        async with aiosqlite.connect(db_path, timeout=30.0) as db:
-            await db.execute("PRAGMA busy_timeout=30000")
+        async with get_db(db_path) as db:
             cursor = await db.execute("""
                 SELECT id, ticker, author_agent_id, quality_grade, gpu_price
                 FROM npc_research_reports
@@ -1633,8 +1651,7 @@ async def npc_auto_purchase_research(db_path: str):
 async def bootstrap_research_reports(db_path: str):
     """★ 서버 부팅 시 Research Reports가 0건이면 즉시 3건 생성 + 나머지 백그라운드"""
     try:
-        async with aiosqlite.connect(db_path, timeout=30.0) as db:
-            await db.execute("PRAGMA busy_timeout=30000")
+        async with get_db(db_path) as db:
             cursor = await db.execute("SELECT COUNT(*) FROM npc_research_reports")
             existing_count = (await cursor.fetchone())[0]
             if existing_count >= 3:
@@ -1701,8 +1718,7 @@ async def background_research_fill(db_path: str, tickers, npcs, start_npc_idx):
             total_trades, wins, total_pnl = npc[5] or 0, npc[6] or 0, npc[7] or 0
             win_rate = round(wins / total_trades * 100) if total_trades > 0 else random.randint(45, 65)
             if total_trades == 0: total_trades = random.randint(5, 15)
-            async with aiosqlite.connect(db_path, timeout=30.0) as db:
-                await db.execute("PRAGMA busy_timeout=30000")
+            async with get_db(db_path) as db:
                 cursor = await db.execute("SELECT COUNT(*) FROM npc_research_reports WHERE ticker=?", (ticker,))
                 if (await cursor.fetchone())[0] > 0: continue
                 cursor = await db.execute(
@@ -1743,8 +1759,7 @@ async def generate_npc_comment_replies(db_path: str, groq_api_key: str,
             if fc['sources']:
                 fact_context = ' | '.join(fc['sources'][:2])
         post_author_identity = ''
-        async with aiosqlite.connect(db_path, timeout=30.0) as db:
-            await db.execute("PRAGMA busy_timeout=30000")
+        async with get_db(db_path) as db:
             pac = await db.execute("SELECT a.ai_identity FROM posts p JOIN npc_agents a ON p.author_agent_id=a.agent_id WHERE p.id=?", (post_id,))
             par = await pac.fetchone()
             if par: post_author_identity = par[0] or ''
@@ -1789,8 +1804,7 @@ async def generate_npc_chat_replies_to_user(db_path: str, groq_api_key: str,
     """유저 메시지에 대해 NPC들이 캐릭터에 맞는 반응 생성"""
     try:
         ai = GroqAIClient(groq_api_key)
-        async with aiosqlite.connect(db_path, timeout=30.0) as db:
-            await db.execute("PRAGMA busy_timeout=30000")
+        async with get_db(db_path) as db:
             for npc in npcs:
                 agent_id, npc_username, identity, mbti = npc
                 try:
@@ -1812,4 +1826,4 @@ Reply ONLY with the message text, nothing else."""
                 except Exception as e:
                     logger.warning(f"NPC reply error ({npc_username}): {e}")
     except Exception as e:
-        logger.error(f"NPC reply generation error: {e}")
+        logger.error(f"NPC reply generation error: {e}")