"""Run SQLite's official recovery on the forensic copy, never the live database."""
import hashlib
import json
import pathlib
import re
import subprocess
import sys
import urllib.request
import zipfile
sys.stdout.reconfigure(encoding="utf-8")
root = pathlib.Path(sys.argv[1]).resolve()
tools = root / "sqlite-tools"
tools.mkdir(exist_ok=True)
page = urllib.request.urlopen("https://www.sqlite.org/download.html", timeout=30).read().decode()
product = next(line for line in page.splitlines() if line.startswith("PRODUCT,") and "/sqlite-tools-win-x64-" in line)
_, version, url, size, digest, *_ = product.split(",")
archive = tools / pathlib.Path(url).name
if not archive.exists():
with urllib.request.urlopen("https://www.sqlite.org/" + url, timeout=30) as source, archive.open("xb") as target:
while chunk := source.read(1024 * 1024):
target.write(chunk)
if hashlib.sha3_256(archive.read_bytes()).hexdigest() != digest:
raise RuntimeError("Official SQLite tool checksum mismatch")
with zipfile.ZipFile(archive) as package:
package.extract("sqlite3.exe", tools)
cli = tools / "sqlite3.exe"
source = root / "markets.db"
recovered = root / "recovered.db"
if recovered.exists():
raise RuntimeError("Choose a fresh recovered database")
with (root / "recovery-stderr.log").open("wb") as errors, (root / "import-stderr.log").open("wb") as import_errors:
producer = subprocess.Popen([str(cli), "-readonly", str(source), ".recover --ignore-freelist"], stdout=subprocess.PIPE, stderr=errors)
consumer = subprocess.Popen([str(cli), str(recovered)], stdin=producer.stdout, stdout=subprocess.DEVNULL, stderr=import_errors)
producer.stdout.close()
recover_code = producer.wait()
import_code = consumer.wait()
report = {"sqlite_version": version, "tool_sha3_256": digest, "recover_exit": recover_code, "import_exit": import_code, "bytes": recovered.stat().st_size}
check = subprocess.run([str(cli), "-readonly", "-json", str(recovered), "PRAGMA integrity_check; SELECT name FROM sqlite_schema WHERE type='table'; SELECT COUNT(*) AS stations FROM stations; SELECT COUNT(*) AS snapshots FROM snapshots; SELECT COUNT(*) AS quotes FROM quotes;"], capture_output=True, text=True, encoding="utf-8")
report["check_exit"] = check.returncode
report["check_output"] = check.stdout
report["check_error"] = check.stderr
(root / "recovery-report.json").write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8")
print(json.dumps(report, ensure_ascii=False), flush=True)
"""Run SQLite's official recovery on the forensic copy, never the live database."""
import hashlib
import json
import pathlib
import re
import subprocess
import sys
import urllib.request
import zipfile
sys.stdout.reconfigure(encoding="utf-8")
root = pathlib.Path(sys.argv[1]).resolve()
tools = root / "sqlite-tools"
tools.mkdir(exist_ok=True)
page = urllib.request.urlopen("https://www.sqlite.org/download.html", timeout=30).read().decode()
product = next(line for line in page.splitlines() if line.startswith("PRODUCT,") and "/sqlite-tools-win-x64-" in line)
_, version, url, size, digest, *_ = product.split(",")
archive = tools / pathlib.Path(url).name
if not archive.exists():
with urllib.request.urlopen("https://www.sqlite.org/" + url, timeout=30) as source, archive.open("xb") as target:
while chunk := source.read(1024 * 1024):
target.write(chunk)
if hashlib.sha3_256(archive.read_bytes()).hexdigest() != digest:
raise RuntimeError("Official SQLite tool checksum mismatch")
with zipfile.ZipFile(archive) as package:
package.extract("sqlite3.exe", tools)
cli = tools / "sqlite3.exe"
source = root / "markets.db"
recovered = root / "recovered.db"
if recovered.exists():
raise RuntimeError("Choose a fresh recovered database")
with (root / "recovery-stderr.log").open("wb") as errors, (root / "import-stderr.log").open("wb") as import_errors:
producer = subprocess.Popen([str(cli), "-readonly", str(source), ".recover --ignore-freelist"], stdout=subprocess.PIPE, stderr=errors)
consumer = subprocess.Popen([str(cli), str(recovered)], stdin=producer.stdout, stdout=subprocess.DEVNULL, stderr=import_errors)
producer.stdout.close()
recover_code = producer.wait()
import_code = consumer.wait()
report = {"sqlite_version": version, "tool_sha3_256": digest, "recover_exit": recover_code, "import_exit": import_code, "bytes": recovered.stat().st_size}
check = subprocess.run([str(cli), "-readonly", "-json", str(recovered), "PRAGMA integrity_check; SELECT name FROM sqlite_schema WHERE type='table'; SELECT COUNT(*) AS stations FROM stations; SELECT COUNT(*) AS snapshots FROM snapshots; SELECT COUNT(*) AS quotes FROM quotes;"], capture_output=True, text=True, encoding="utf-8")
report["check_exit"] = check.returncode
report["check_output"] = check.stdout
report["check_error"] = check.stderr
(root / "recovery-report.json").write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8")
print(json.dumps(report, ensure_ascii=False), flush=True)
