127 lines
6.5 KiB
Python
127 lines
6.5 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
将 AI 提取的结构化数据合并到 SQLite
|
||
与 panel schema 及 situationData.getSituation 对齐，支持回放
|
||
"""
|
||
import os
|
||
import sqlite3
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
from typing import Any, Dict, Optional
|
||
|
||
PROJECT_ROOT = Path(__file__).resolve().parent.parent
|
||
DB_PATH = os.environ.get("DB_PATH", str(PROJECT_ROOT / "server" / "data.db"))
|
||
|
||
|
||
def _ensure_tables(conn: sqlite3.Connection) -> None:
|
||
"""确保所需表存在（与 db.js 一致）"""
|
||
conn.execute("""
|
||
CREATE TABLE IF NOT EXISTS situation_update (
|
||
id TEXT PRIMARY KEY, timestamp TEXT NOT NULL, category TEXT NOT NULL,
|
||
summary TEXT NOT NULL, severity TEXT NOT NULL
|
||
)
|
||
""")
|
||
conn.execute("""
|
||
CREATE TABLE IF NOT EXISTS combat_losses (
|
||
side TEXT PRIMARY KEY CHECK (side IN ('us', 'iran')),
|
||
bases_destroyed INTEGER NOT NULL, bases_damaged INTEGER NOT NULL,
|
||
personnel_killed INTEGER NOT NULL, personnel_wounded INTEGER NOT NULL,
|
||
aircraft INTEGER NOT NULL, warships INTEGER NOT NULL, armor INTEGER NOT NULL, vehicles INTEGER NOT NULL
|
||
)
|
||
""")
|
||
try:
|
||
conn.execute("ALTER TABLE combat_losses ADD COLUMN civilian_killed INTEGER NOT NULL DEFAULT 0")
|
||
except sqlite3.OperationalError:
|
||
pass
|
||
try:
|
||
conn.execute("ALTER TABLE combat_losses ADD COLUMN civilian_wounded INTEGER NOT NULL DEFAULT 0")
|
||
except sqlite3.OperationalError:
|
||
pass
|
||
try:
|
||
conn.execute("ALTER TABLE combat_losses ADD COLUMN updated_at TEXT DEFAULT (datetime('now'))")
|
||
except sqlite3.OperationalError:
|
||
pass
|
||
conn.execute("CREATE TABLE IF NOT EXISTS wall_street_trend (id INTEGER PRIMARY KEY AUTOINCREMENT, time TEXT NOT NULL, value INTEGER NOT NULL)")
|
||
conn.execute("CREATE TABLE IF NOT EXISTS retaliation_current (id INTEGER PRIMARY KEY CHECK (id = 1), value INTEGER NOT NULL)")
|
||
conn.execute("CREATE TABLE IF NOT EXISTS retaliation_history (id INTEGER PRIMARY KEY AUTOINCREMENT, time TEXT NOT NULL, value INTEGER NOT NULL)")
|
||
conn.execute("CREATE TABLE IF NOT EXISTS situation (id INTEGER PRIMARY KEY CHECK (id = 1), data TEXT NOT NULL, updated_at TEXT NOT NULL)")
|
||
conn.commit()
|
||
|
||
|
||
def merge(extracted: Dict[str, Any], db_path: Optional[str] = None) -> bool:
|
||
"""将提取数据合并到 DB，返回是否有更新"""
|
||
path = db_path or DB_PATH
|
||
if not os.path.exists(path):
|
||
return False
|
||
conn = sqlite3.connect(path, timeout=10)
|
||
try:
|
||
_ensure_tables(conn)
|
||
updated = False
|
||
# situation_update
|
||
if "situation_update" in extracted:
|
||
u = extracted["situation_update"]
|
||
uid = f"ai_{hash(u.get('summary','')+u.get('timestamp','')) % 10**10}"
|
||
conn.execute(
|
||
"INSERT OR IGNORE INTO situation_update (id, timestamp, category, summary, severity) VALUES (?, ?, ?, ?, ?)",
|
||
(uid, u.get("timestamp", ""), u.get("category", "other"), u.get("summary", "")[:500], u.get("severity", "medium")),
|
||
)
|
||
if conn.total_changes > 0:
|
||
updated = True
|
||
# combat_losses：增量叠加到当前值
|
||
if "combat_losses_delta" in extracted:
|
||
for side, delta in extracted["combat_losses_delta"].items():
|
||
if side not in ("us", "iran"):
|
||
continue
|
||
try:
|
||
row = conn.execute(
|
||
"SELECT personnel_killed,personnel_wounded,civilian_killed,civilian_wounded,bases_destroyed,bases_damaged,aircraft,warships,armor,vehicles FROM combat_losses WHERE side = ?",
|
||
(side,),
|
||
).fetchone()
|
||
if not row:
|
||
continue
|
||
cur = {
|
||
"personnel_killed": row[0], "personnel_wounded": row[1], "civilian_killed": row[2] or 0,
|
||
"civilian_wounded": row[3] or 0, "bases_destroyed": row[4], "bases_damaged": row[5],
|
||
"aircraft": row[6], "warships": row[7], "armor": row[8], "vehicles": row[9],
|
||
}
|
||
pk = max(0, (cur["personnel_killed"] or 0) + delta.get("personnel_killed", 0))
|
||
pw = max(0, (cur["personnel_wounded"] or 0) + delta.get("personnel_wounded", 0))
|
||
ck = max(0, (cur["civilian_killed"] or 0) + delta.get("civilian_killed", 0))
|
||
cw = max(0, (cur["civilian_wounded"] or 0) + delta.get("civilian_wounded", 0))
|
||
bd = max(0, (cur["bases_destroyed"] or 0) + delta.get("bases_destroyed", 0))
|
||
bm = max(0, (cur["bases_damaged"] or 0) + delta.get("bases_damaged", 0))
|
||
ac = max(0, (cur["aircraft"] or 0) + delta.get("aircraft", 0))
|
||
ws = max(0, (cur["warships"] or 0) + delta.get("warships", 0))
|
||
ar = max(0, (cur["armor"] or 0) + delta.get("armor", 0))
|
||
vh = max(0, (cur["vehicles"] or 0) + delta.get("vehicles", 0))
|
||
ts = datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%S.000Z")
|
||
conn.execute(
|
||
"""UPDATE combat_losses SET personnel_killed=?, personnel_wounded=?, civilian_killed=?, civilian_wounded=?,
|
||
bases_destroyed=?, bases_damaged=?, aircraft=?, warships=?, armor=?, vehicles=?, updated_at=? WHERE side=?""",
|
||
(pk, pw, ck, cw, bd, bm, ac, ws, ar, vh, ts, side),
|
||
)
|
||
if conn.total_changes > 0:
|
||
updated = True
|
||
except Exception:
|
||
pass
|
||
# retaliation
|
||
if "retaliation" in extracted:
|
||
r = extracted["retaliation"]
|
||
conn.execute("INSERT OR REPLACE INTO retaliation_current (id, value) VALUES (1, ?)", (r["value"],))
|
||
conn.execute("INSERT INTO retaliation_history (time, value) VALUES (?, ?)", (r["time"], r["value"]))
|
||
updated = True
|
||
# wall_street_trend
|
||
if "wall_street" in extracted:
|
||
w = extracted["wall_street"]
|
||
conn.execute("INSERT INTO wall_street_trend (time, value) VALUES (?, ?)", (w["time"], w["value"]))
|
||
updated = True
|
||
if updated:
|
||
conn.execute("INSERT OR REPLACE INTO situation (id, data, updated_at) VALUES (1, '{}', ?)", (datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%S.000Z"),))
|
||
conn.commit()
|
||
return updated
|
||
except Exception as e:
|
||
conn.rollback()
|
||
raise e
|
||
finally:
|
||
conn.close()
|