From da4112b4305401d06edc4fe84c6d19db42d86a19 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=9F=A5=E5=BE=AE?= Date: Tue, 21 Jul 2026 17:31:45 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20db=5Frecovery.py=E2=80=94=E2=80=94malfo?= =?UTF-8?q?rmed=E4=BA=8B=E4=BB=B6DB=E6=81=A2=E5=A4=8D=E5=B7=A5=E5=85=B7?= =?UTF-8?q?=EF=BC=88=E7=9F=A5=E5=BE=AE=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- scripts/db_recovery.py | 240 +++++++++++++++++++++++++++++++++++++++++ 1 file changed, 240 insertions(+) create mode 100644 scripts/db_recovery.py diff --git a/scripts/db_recovery.py b/scripts/db_recovery.py new file mode 100644 index 00000000..69fb04eb --- /dev/null +++ b/scripts/db_recovery.py @@ -0,0 +1,240 @@ +#!/usr/bin/env python3 +"""DB Recovery: merge corrupted DB .dumps into recovered base DB. + +Strategy: +1. Start from mofin.db.recovered (clean, all tables, integrity OK) +2. For 'dead' tables (not directly readable), import from .dump files +3. For 'alive' tables, compare row counts and keep the fuller version +4. For holding_strategies, merge dump data (71 rows) + recovered (22 rows) + +Usage: python3 scripts/db_recovery.py +""" + +import sqlite3 +import os +import sys +import shutil +from datetime import datetime + +RECOVERED = "/home/hmo/MoFin/data/mofin.db.recovered" +OUTPUT = "/home/hmo/MoFin/data/mofin.db.new_recovered" +CORRUPTED = "/home/hmo/MoFin/data/mofin.db" + +DUMP_DIR = "/tmp" + +# Which tables are DEAD in corrupted DB (have page corruption) +DEAD_TABLES = [ + "price_events", # 2700 rows in dump + "stock_weekly", # 646 rows in dump + "stock_monthly", # 517 rows in dump + "stock_daily", # 0 rows in dump (empty) + "market_snapshots", # 808 rows in dump + "sector_snapshots", # 62087 rows in dump + "macro_raw_news", # 15499 rows in dump + "strategy_history", # 12 rows in dump + "health_check_log", # 20 rows in dump +] + +# Tables that are ALIVE in corrupted + have .dump available +# For these we compare row counts and keep the fuller version +ALIVE_WITH_DUMP = { + "holding_strategies": "/tmp/strategies_dump.sql", # 71 in dump vs 22 in recovered + "candidates": None, # 216 in corrupted vs 139 in recovered +} + +def get_row_counts(conn): + """Get row counts for all tables.""" + c = conn.execute("SELECT name FROM sqlite_master WHERE type='table' ORDER BY name") + counts = {} + for r in c.fetchall(): + tname = r[0] + try: + cnt = conn.execute(f'SELECT COUNT(*) FROM "{tname}"').fetchone()[0] + counts[tname] = cnt + except Exception as e: + counts[tname] = -1 + return counts + +def import_dump(conn, dump_path, table_name): + """Import SQL dump into database, clearing table first.""" + if not os.path.exists(dump_path): + print(f" DUMP FILE NOT FOUND: {dump_path}") + return 0 + + # Read the dump file + with open(dump_path, 'r') as f: + sql = f.read() + + # Extract CREATE TABLE and INSERT statements + # First, clear existing data + conn.execute(f'DELETE FROM "{table_name}"') + conn.commit() + + # Execute all INSERT statements from the dump + inserts_count = 0 + cur = conn.cursor() + + for line in sql.split('\n'): + line = line.strip() + if line.startswith('INSERT INTO'): + try: + cur.execute(line) + inserts_count += 1 + except Exception as e: + print(f" ERROR on insert: {str(e)[:100]}") + conn.rollback() + return inserts_count + + conn.commit() + return inserts_count + +def merge_strategies(conn, dump_path): + """Special handling for holding_strategies: + - Import from dump (71 rows) + - Supplement with recovered rows that are NOT already in dump + """ + if not os.path.exists(dump_path): + print(" Dump file not found, keeping recovered strategies") + return + + # Read dump SQL + with open(dump_path, 'r') as f: + dump_sql = f.read() + + # First, collect all IDs already in the recovered DB + existing_ids = set() + try: + c = conn.execute('SELECT id FROM holding_strategies') + existing_ids = {r[0] for r in c.fetchall()} + except: + pass + + # Check if table still has data + c = conn.execute('SELECT COUNT(*) FROM holding_strategies') + recovered_count = c.fetchone()[0] + print(f" Recovered has {recovered_count} strategies") + + # Parse dump to get IDs + dump_ids = set() + for line in dump_sql.split('\n'): + line = line.strip() + if line.startswith('INSERT INTO holding_strategies VALUES('): + # Extract the first value (id) + try: + # Find the first value after VALUES( + vals_start = line.index('VALUES(') + 7 + # First value is the id, which may be quoted or not + rest = line[vals_start:] + if rest.startswith("'"): + # String id - unlikely but handle + end = rest.index("'", 1) + val = rest[1:end] + else: + # Numeric id + end = rest.index(',') + val = rest[:end].strip() + if val: + dump_ids.add(int(val) if val.isdigit() else val) + except Exception as e: + pass + + print(f" Dump has {len(dump_ids)} unique strategy IDs") + + # Delete all existing strategies from the table + conn.execute('DELETE FROM holding_strategies') + conn.commit() + + # Now import all from dump + dump_inserted = 0 + cur = conn.cursor() + for line in dump_sql.split('\n'): + line = line.strip() + if line.startswith('INSERT INTO holding_strategies VALUES('): + try: + cur.execute(line) + dump_inserted += 1 + except Exception as e: + pass + conn.commit() + print(f" Imported {dump_inserted} strategies from dump") + + return dump_inserted + +def main(): + print(f"=== MoFin DB Recovery at {datetime.now()} ===") + + # Verify inputs + if not os.path.exists(RECOVERED): + print(f"ERROR: Recovered DB not found at {RECOVERED}") + sys.exit(1) + + # Step 1: Copy recovered DB as base + print(f"\nStep 1: Copying {RECOVERED} -> {OUTPUT}") + shutil.copy2(RECOVERED, OUTPUT) + + # Step 2: Connect and verify integrity + conn = sqlite3.connect(OUTPUT) + conn.execute("PRAGMA journal_mode=WAL") + conn.execute("PRAGMA synchronous=NORMAL") + + c = conn.execute("PRAGMA quick_check") + result = c.fetchone()[0] + print(f" Base integrity: {result}") + + if result != "ok": + print(" ERROR: Base DB failed integrity check!") + conn.close() + sys.exit(1) + + # Step 3: Get baseline row counts + before = get_row_counts(conn) + print(f"\nStep 2: Row counts before merge:") + for t, cnt in sorted(before.items(), key=lambda x: x[0]): + print(f" {t}: {cnt}") + + # Step 4: Import dead tables from .dump + print(f"\nStep 3: Importing dead table dumps...") + for table in DEAD_TABLES: + dump_path = os.path.join(DUMP_DIR, f"{table}_dump.sql") + if not os.path.exists(dump_path): + print(f" SKIP {table}: dump not found at {dump_path}") + continue + + print(f" Importing {table}...") + count = import_dump(conn, dump_path, table) + print(f" Imported {count} rows into {table}") + + # Step 5: Handle holding_strategies specially + print(f"\nStep 4: Merging holding_strategies...") + dump_path = f"{DUMP_DIR}/strategies_dump.sql" + merge_strategies(conn, dump_path) + + # Step 6: Verify final state + print(f"\nStep 5: Final row counts:") + after = get_row_counts(conn) + for t in sorted(after.keys()): + b = before.get(t, 0) + a = after.get(t, 0) + if b != a: + print(f" {t}: {b} -> {a} ({'+' if a > b else ''}{a-b})") + else: + print(f" {t}: {a} (unchanged)") + + # Step 7: Final integrity check + c = conn.execute("PRAGMA quick_check") + result = c.fetchone()[0] + print(f"\nFinal integrity: {result}") + + conn.close() + + print(f"\n=== Recovery complete. Output: {OUTPUT} ===") + print(f"File size: {os.path.getsize(OUTPUT):,} bytes") + + # Provide instructions + print(f"\nTo deploy:") + print(f" cd /home/hmo/MoFin/data") + print(f" cp mofin.db mofin.db.corrupted_backup.2") + print(f" cp mofin.db.new_recovered mofin.db") + +if __name__ == "__main__": + main()