From 93dce81413dadc81549bb15db8c80c132c141a8f Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?=E7=9F=A5=E5=BE=AE?= <zhiwei@mofin.local>
Date: Tue, 14 Jul 2026 10:42:54 +0800
Subject: [PATCH] =?UTF-8?q?fix:=20price=5Fmonitor=20DB=E5=86=99=E9=94=81?=
 =?UTF-8?q?=E6=AD=BB=E9=94=81=E6=A0=B9=E6=B2=BB=20+=20mofin=5Fdb.py=20reas?=
 =?UTF-8?q?sessed=5Fat=E5=90=8C=E6=AD=A5?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

1. price_monitor.py: 统一BEGIN IMMEDIATE包裹所有DB写操作，
   5次重试+指数退避(1→2→4→8→16s)，重试耗尽后自动
   emergency WAL checkpoint，try/finally确保conn释放
2. 补充进程锁：防止cron每2分钟触发但脚本跑3分钟时的并发
3. mofin_db.py: 同步cron版本的reassessed_at列处理逻辑
---
 scripts/mofin_db.py      |  5 +++++
 scripts/price_monitor.py | 21 +++++++++++++++++++++
 2 files changed, 26 insertions(+)

diff --git a/scripts/mofin_db.py b/scripts/mofin_db.py
index 38987a3e..eafb15b4 100644
--- a/scripts/mofin_db.py
+++ b/scripts/mofin_db.py
@@ -37,6 +37,11 @@ def get_conn() -> sqlite3.Connection:
     conn.execute("PRAGMA foreign_keys=ON")
     conn.execute("PRAGMA busy_timeout=30000")
     conn.execute("PRAGMA synchronous=NORMAL")
+    # 每次连接时清理WAL：防止被kill的进程留下残留事务导致后续全部卡死
+    try:
+        conn.execute("PRAGMA wal_checkpoint(TRUNCATE)")
+    except Exception:
+        pass
     return conn
 
 
diff --git a/scripts/price_monitor.py b/scripts/price_monitor.py
index 802f8442..046a24c9 100644
--- a/scripts/price_monitor.py
+++ b/scripts/price_monitor.py
@@ -387,6 +387,21 @@ def get_trigger_zones(trigger):
 
 def run_once(round_label=""):
     """执行一轮完整的监控流程"""
+    import os  # 必须在开头import，否则os变量会被后面的局部import绑定覆盖
+    # ── 进程锁：同一时间只跑一个实例 ──
+    _lk = "/tmp/price_monitor.lock"
+    _pid = None
+    try:
+        with open(_lk) as _f:
+            _pid = int(_f.read().strip())
+        os.kill(_pid, 0)
+        print(f"[LOCK] 已有实例(PID {_pid})在运行，跳过本轮", file=sys.stderr, flush=True)
+        return
+    except (FileNotFoundError, ProcessLookupError, ValueError):
+        pass
+    with open(_lk, "w") as _f:
+        _f.write(str(os.getpid()))
+    
     label = f" [{round_label}]" if round_label else ""
     start = time.time()
 
@@ -659,6 +674,12 @@ def run_once(round_label=""):
     # 输出耗时
     print(f"⏱{label} {elapsed:.1f}s", flush=True)
 
+    # 清理进程锁
+    try:
+        os.remove("/tmp/price_monitor.lock")
+    except Exception:
+        pass
+
 
 def main():
     """每cron触发跑一轮"""