Skip to content

Commit 127a4bc

Browse files
jiwenxu025-boopjiwenxu025-boop
committed
feat: add SQLite runtime data migration foundation
1 parent ebb4805 commit 127a4bc

2 files changed

Lines changed: 399 additions & 0 deletions

File tree

‎scripts/bazi_engine/runtime_store.py‎

Lines changed: 327 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,327 @@
1+
"""SQLite runtime-data store and idempotent legacy JSON/JSONL importer."""
2+
import hashlib
3+
import json
4+
import sqlite3
5+
from dataclasses import dataclass
6+
from pathlib import Path
7+
8+
9+
@dataclass(frozen=True)
10+
class MigrationReport:
11+
activation_codes: int = 0
12+
free_usage: int = 0
13+
fusion_generations: int = 0
14+
fusion_feedback: int = 0
15+
family_feedback: int = 0
16+
skipped_records: int = 0
17+
18+
def total_imported(self) -> int:
19+
return sum((
20+
self.activation_codes,
21+
self.free_usage,
22+
self.fusion_generations,
23+
self.fusion_feedback,
24+
self.family_feedback,
25+
))
26+
27+
28+
class RuntimeStore:
29+
"""Owns durable runtime state without touching calibration data."""
30+
31+
def __init__(self, path: Path):
32+
self.path = path
33+
34+
def connect(self) -> sqlite3.Connection:
35+
self.path.parent.mkdir(parents=True, exist_ok=True)
36+
connection = sqlite3.connect(self.path, timeout=5.0)
37+
connection.row_factory = sqlite3.Row
38+
connection.execute("PRAGMA foreign_keys = ON")
39+
connection.execute("PRAGMA busy_timeout = 5000")
40+
return connection
41+
42+
def initialize(self) -> None:
43+
with self.connect() as connection:
44+
connection.execute("PRAGMA journal_mode = WAL")
45+
connection.execute("PRAGMA synchronous = NORMAL")
46+
connection.executescript(
47+
"""
48+
CREATE TABLE IF NOT EXISTS schema_meta (
49+
key TEXT PRIMARY KEY,
50+
value TEXT NOT NULL
51+
);
52+
CREATE TABLE IF NOT EXISTS activation_codes (
53+
code TEXT PRIMARY KEY,
54+
remaining INTEGER NOT NULL CHECK (remaining >= 0),
55+
note TEXT NOT NULL DEFAULT '',
56+
updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
57+
);
58+
CREATE TABLE IF NOT EXISTS free_usage (
59+
client_hash TEXT PRIMARY KEY,
60+
usage_date TEXT NOT NULL,
61+
count INTEGER NOT NULL CHECK (count >= 0)
62+
);
63+
CREATE TABLE IF NOT EXISTS fusion_generations (
64+
generation_id TEXT PRIMARY KEY,
65+
timestamp TEXT NOT NULL,
66+
generation_type TEXT NOT NULL,
67+
outcome TEXT NOT NULL,
68+
duration_ms INTEGER,
69+
prompt_version TEXT,
70+
model TEXT,
71+
temperature REAL,
72+
repaired INTEGER NOT NULL DEFAULT 0,
73+
error_class TEXT,
74+
synthetic INTEGER NOT NULL DEFAULT 0
75+
);
76+
CREATE TABLE IF NOT EXISTS fusion_feedback (
77+
legacy_hash TEXT PRIMARY KEY,
78+
generation_id TEXT,
79+
timestamp TEXT NOT NULL,
80+
rating TEXT NOT NULL,
81+
inaccurate_section TEXT NOT NULL DEFAULT '',
82+
report_hash TEXT,
83+
report_length INTEGER,
84+
prompt_version TEXT,
85+
model TEXT,
86+
temperature REAL,
87+
repaired INTEGER NOT NULL DEFAULT 0,
88+
synthetic INTEGER NOT NULL DEFAULT 0
89+
);
90+
CREATE UNIQUE INDEX IF NOT EXISTS fusion_feedback_generation_id
91+
ON fusion_feedback(generation_id) WHERE generation_id IS NOT NULL;
92+
CREATE TABLE IF NOT EXISTS family_feedback (
93+
legacy_hash TEXT PRIMARY KEY,
94+
timestamp TEXT NOT NULL,
95+
engine_level TEXT NOT NULL DEFAULT '',
96+
user_level TEXT NOT NULL,
97+
discrepancy INTEGER NOT NULL DEFAULT 0,
98+
discrepancy_detail TEXT,
99+
synthetic INTEGER NOT NULL DEFAULT 0
100+
);
101+
CREATE TABLE IF NOT EXISTS legacy_imports (
102+
content_hash TEXT PRIMARY KEY,
103+
source TEXT NOT NULL,
104+
imported_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
105+
);
106+
INSERT OR IGNORE INTO schema_meta(key, value) VALUES ('version', '1');
107+
"""
108+
)
109+
110+
def summary_counts(self) -> dict[str, int]:
111+
self.initialize()
112+
tables = (
113+
"activation_codes",
114+
"free_usage",
115+
"fusion_generations",
116+
"fusion_feedback",
117+
"family_feedback",
118+
)
119+
with self.connect() as connection:
120+
return {
121+
table: connection.execute(f"SELECT COUNT(*) FROM {table}").fetchone()[0]
122+
for table in tables
123+
}
124+
125+
def import_legacy(
126+
self,
127+
*,
128+
activation_file: Path | None = None,
129+
free_usage_file: Path | None = None,
130+
feedback_dir: Path | None = None,
131+
generation_dir: Path | None = None,
132+
) -> MigrationReport:
133+
"""Import append-only runtime files. Re-running never duplicates JSONL rows."""
134+
self.initialize()
135+
counts = {
136+
"activation_codes": 0,
137+
"free_usage": 0,
138+
"fusion_generations": 0,
139+
"fusion_feedback": 0,
140+
"family_feedback": 0,
141+
"skipped_records": 0,
142+
}
143+
with self.connect() as connection:
144+
if activation_file is not None:
145+
counts["activation_codes"] += self._import_activation_codes(connection, activation_file)
146+
if free_usage_file is not None:
147+
counts["free_usage"] += self._import_free_usage(connection, free_usage_file)
148+
if feedback_dir is not None:
149+
self._import_feedback_dir(connection, feedback_dir, counts)
150+
if generation_dir is not None:
151+
self._import_generation_dir(connection, generation_dir, counts)
152+
return MigrationReport(**counts)
153+
154+
@staticmethod
155+
def _read_json(path: Path) -> dict:
156+
if not path.exists():
157+
return {}
158+
try:
159+
data = json.loads(path.read_text(encoding="utf-8"))
160+
except (OSError, json.JSONDecodeError):
161+
return {}
162+
return data if isinstance(data, dict) else {}
163+
164+
def _import_activation_codes(self, connection: sqlite3.Connection, path: Path) -> int:
165+
count = 0
166+
for code, entry in self._read_json(path).items():
167+
if not isinstance(entry, dict):
168+
continue
169+
remaining = entry.get("剩余", 0)
170+
if not isinstance(remaining, int) or remaining < 0:
171+
continue
172+
connection.execute(
173+
"""
174+
INSERT INTO activation_codes(code, remaining, note)
175+
VALUES (?, ?, ?)
176+
ON CONFLICT(code) DO UPDATE SET
177+
remaining = excluded.remaining,
178+
note = excluded.note,
179+
updated_at = CURRENT_TIMESTAMP
180+
""",
181+
(str(code).upper(), remaining, str(entry.get("备注", ""))),
182+
)
183+
count += 1
184+
return count
185+
186+
def _import_free_usage(self, connection: sqlite3.Connection, path: Path) -> int:
187+
count = 0
188+
for client_hash, entry in self._read_json(path).items():
189+
if not isinstance(entry, dict):
190+
continue
191+
usage_date, usage_count = entry.get("date"), entry.get("count")
192+
if not isinstance(usage_date, str) or not isinstance(usage_count, int) or usage_count < 0:
193+
continue
194+
connection.execute(
195+
"""
196+
INSERT INTO free_usage(client_hash, usage_date, count)
197+
VALUES (?, ?, ?)
198+
ON CONFLICT(client_hash) DO UPDATE SET
199+
usage_date = excluded.usage_date,
200+
count = excluded.count
201+
""",
202+
(str(client_hash), usage_date, usage_count),
203+
)
204+
count += 1
205+
return count
206+
207+
def _import_feedback_dir(self, connection: sqlite3.Connection, directory: Path, counts: dict[str, int]) -> None:
208+
for path in sorted(directory.glob("fusion_feedback_*.jsonl")):
209+
self._import_jsonl(connection, path, "fusion_feedback", counts)
210+
for path in sorted(directory.glob("feedback_*.jsonl")):
211+
self._import_jsonl(connection, path, "family_feedback", counts)
212+
213+
def _import_generation_dir(self, connection: sqlite3.Connection, directory: Path, counts: dict[str, int]) -> None:
214+
for path in sorted(directory.glob("fusion_generation_*.jsonl")):
215+
self._import_jsonl(connection, path, "fusion_generations", counts)
216+
217+
def _import_jsonl(
218+
self,
219+
connection: sqlite3.Connection,
220+
path: Path,
221+
destination: str,
222+
counts: dict[str, int],
223+
) -> None:
224+
if not path.exists():
225+
return
226+
try:
227+
lines = path.read_text(encoding="utf-8").splitlines()
228+
except OSError:
229+
return
230+
for line in lines:
231+
try:
232+
record = json.loads(line)
233+
except json.JSONDecodeError:
234+
counts["skipped_records"] += 1
235+
continue
236+
if not isinstance(record, dict):
237+
counts["skipped_records"] += 1
238+
continue
239+
content_hash = self._legacy_hash(destination, record)
240+
inserted = connection.execute(
241+
"INSERT OR IGNORE INTO legacy_imports(content_hash, source) VALUES (?, ?)",
242+
(content_hash, str(path)),
243+
).rowcount
244+
if not inserted:
245+
continue
246+
if self._insert_record(connection, destination, content_hash, record):
247+
counts[destination] += 1
248+
else:
249+
counts["skipped_records"] += 1
250+
251+
@staticmethod
252+
def _legacy_hash(destination: str, record: dict) -> str:
253+
serialized = json.dumps(record, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
254+
return hashlib.sha256(f"{destination}:{serialized}".encode()).hexdigest()
255+
256+
@staticmethod
257+
def _number(value) -> float | None:
258+
try:
259+
return float(value) if value is not None else None
260+
except (TypeError, ValueError):
261+
return None
262+
263+
def _insert_record(
264+
self,
265+
connection: sqlite3.Connection,
266+
destination: str,
267+
content_hash: str,
268+
record: dict,
269+
) -> bool:
270+
if destination == "fusion_generations":
271+
generation_id = record.get("generation_id")
272+
timestamp = record.get("timestamp")
273+
if not isinstance(generation_id, str) or not isinstance(timestamp, str):
274+
return False
275+
result = connection.execute(
276+
"""
277+
INSERT OR IGNORE INTO fusion_generations(
278+
generation_id, timestamp, generation_type, outcome, duration_ms,
279+
prompt_version, model, temperature, repaired, error_class, synthetic
280+
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
281+
""",
282+
(
283+
generation_id, timestamp, str(record.get("generation_type", "fusion")),
284+
str(record.get("outcome", "unknown")), record.get("duration_ms"),
285+
record.get("prompt_version"), record.get("model"), self._number(record.get("temperature")),
286+
int(bool(record.get("repaired"))), record.get("error_class"), int(bool(record.get("synthetic"))),
287+
),
288+
)
289+
return result.rowcount == 1
290+
if destination == "fusion_feedback":
291+
timestamp, rating = record.get("timestamp"), record.get("rating")
292+
if not isinstance(timestamp, str) or not isinstance(rating, str):
293+
return False
294+
result = connection.execute(
295+
"""
296+
INSERT OR IGNORE INTO fusion_feedback(
297+
legacy_hash, generation_id, timestamp, rating, inaccurate_section,
298+
report_hash, report_length, prompt_version, model, temperature, repaired, synthetic
299+
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
300+
""",
301+
(
302+
content_hash, record.get("generation_id"), timestamp, rating,
303+
str(record.get("inaccurate_section", "")), record.get("report_hash"),
304+
record.get("report_length"), record.get("prompt_version"), record.get("model"),
305+
self._number(record.get("temperature")), int(bool(record.get("repaired"))),
306+
int(bool(record.get("synthetic"))),
307+
),
308+
)
309+
return result.rowcount == 1
310+
if destination == "family_feedback":
311+
timestamp, user_level = record.get("timestamp"), record.get("user_level")
312+
if not isinstance(timestamp, str) or not isinstance(user_level, str):
313+
return False
314+
result = connection.execute(
315+
"""
316+
INSERT OR IGNORE INTO family_feedback(
317+
legacy_hash, timestamp, engine_level, user_level, discrepancy, discrepancy_detail, synthetic
318+
) VALUES (?, ?, ?, ?, ?, ?, ?)
319+
""",
320+
(
321+
content_hash, timestamp, str(record.get("engine_level", "")), user_level,
322+
int(bool(record.get("discrepancy"))), record.get("discrepancy_detail"),
323+
int(bool(record.get("synthetic"))),
324+
),
325+
)
326+
return result.rowcount == 1
327+
return False

0 commit comments

Comments
 (0)