Malin
b651eca285
Python backup module (Azure AD OAuth2 → BC API v2.0 → GPG AES256 → S3), Flask web UI on port 1890 with SSE live log streaming, dual incremental + full backup schedules, chain restore merging full + incrementals, and per-entity file browser with ZIP download. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
675 lines
27 KiB
Python
675 lines
27 KiB
Python
#!/usr/bin/env python3
|
|
"""Business Central backup module — Azure AD OAuth2 → BC API v2.0 → S3/MinIO."""
|
|
|
|
import json
|
|
import logging
|
|
import os
|
|
import subprocess
|
|
import tarfile
|
|
import tempfile
|
|
import time
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
import boto3
|
|
import requests
|
|
from botocore.client import Config
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Entity definitions (mirrors bc-export.ps1)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
STANDALONE_ENTITIES = [
|
|
"accounts", "customers", "vendors", "items", "generalLedgerEntries",
|
|
"bankAccounts", "employees", "dimensions", "dimensionValues", "currencies",
|
|
"paymentTerms", "paymentMethods", "journals", "journalLines",
|
|
"countriesRegions", "companyInformation", "itemCategories",
|
|
"shipmentMethods", "taxAreas", "taxGroups", "unitsOfMeasure",
|
|
"timeRegistrationEntries", "contacts", "generalProductPostingGroups",
|
|
"inventoryPostingGroups", "itemLedgerEntries", "opportunities",
|
|
"locations", "projects", "irs1099",
|
|
]
|
|
|
|
REPORT_ENTITIES = [
|
|
"agedAccountsPayable", "agedAccountsReceivable", "balanceSheet",
|
|
"cashFlowStatement", "incomeStatement", "retainedEarningsStatement",
|
|
"trialBalance", "customerFinancialDetails", "customerSales",
|
|
"vendorPurchases",
|
|
]
|
|
|
|
DOCUMENT_ENTITIES = {
|
|
"salesInvoices": "salesInvoiceLines",
|
|
"salesOrders": "salesOrderLines",
|
|
"salesCreditMemos": "salesCreditMemoLines",
|
|
"purchaseInvoices": "purchaseInvoiceLines",
|
|
"purchaseOrders": "purchaseOrderLines",
|
|
"salesQuotes": "salesQuoteLines",
|
|
"salesShipments": "salesShipmentLines",
|
|
"purchaseReceipts": "purchaseReceiptLines",
|
|
"customerPaymentJournals": "customerPayments",
|
|
"vendorPaymentJournals": "vendorPayments",
|
|
}
|
|
|
|
_DOCUMENT_LINE_ENTITIES = set(DOCUMENT_ENTITIES.values())
|
|
|
|
_MAX_RETRIES = 3
|
|
_RETRY_BACKOFF = [5, 15]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Logging helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class _CallbackHandler(logging.Handler):
|
|
def __init__(self, cb):
|
|
super().__init__()
|
|
self._cb = cb
|
|
|
|
def emit(self, record):
|
|
try:
|
|
self._cb(self.format(record))
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def _make_logger(log_dir, name, log_callback=None):
|
|
log_dir = Path(log_dir)
|
|
log_dir.mkdir(parents=True, exist_ok=True)
|
|
logger = logging.getLogger(f"bc_{name}_{id(log_callback)}")
|
|
logger.setLevel(logging.DEBUG)
|
|
logger.propagate = False
|
|
logger.handlers.clear()
|
|
|
|
fh = logging.FileHandler(log_dir / f"{name}.log")
|
|
fh.setFormatter(logging.Formatter("%(asctime)s %(levelname)s %(message)s"))
|
|
logger.addHandler(fh)
|
|
|
|
if log_callback:
|
|
ch = _CallbackHandler(log_callback)
|
|
ch.setFormatter(logging.Formatter("%(asctime)s %(levelname)s %(message)s"))
|
|
logger.addHandler(ch)
|
|
|
|
return logger
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# BC API client
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class BCClient:
|
|
def __init__(self, tenant_id, client_id, client_secret, environment,
|
|
company_name="", timeout=120, logger=None):
|
|
self.tenant_id = tenant_id
|
|
self.client_id = client_id
|
|
self.client_secret = client_secret
|
|
self.environment = environment
|
|
self.company_name = company_name
|
|
self.timeout = timeout
|
|
self.log = logger or logging.getLogger(__name__)
|
|
self._token = None
|
|
self._token_expiry = 0.0
|
|
self._base = (
|
|
f"https://api.businesscentral.dynamics.com/v2.0"
|
|
f"/{tenant_id}/{environment}/api/v2.0"
|
|
)
|
|
|
|
def authenticate(self):
|
|
url = f"https://login.microsoftonline.com/{self.tenant_id}/oauth2/v2.0/token"
|
|
body = {
|
|
"client_id": self.client_id,
|
|
"client_secret": self.client_secret,
|
|
"scope": "https://api.businesscentral.dynamics.com/.default",
|
|
"grant_type": "client_credentials",
|
|
}
|
|
for attempt in range(1, _MAX_RETRIES + 1):
|
|
try:
|
|
resp = requests.post(url, data=body, timeout=30)
|
|
resp.raise_for_status()
|
|
data = resp.json()
|
|
self._token = data["access_token"]
|
|
self._token_expiry = time.time() + data.get("expires_in", 3600) - 300
|
|
self.log.info("Azure AD authentication successful")
|
|
return
|
|
except Exception as e:
|
|
if attempt < _MAX_RETRIES:
|
|
wait = _RETRY_BACKOFF[attempt - 1]
|
|
self.log.warning(f"Auth attempt {attempt} failed ({e}), retry in {wait}s")
|
|
time.sleep(wait)
|
|
else:
|
|
raise
|
|
|
|
def _get_token(self):
|
|
if not self._token or time.time() >= self._token_expiry:
|
|
self.authenticate()
|
|
return self._token
|
|
|
|
def _get(self, url):
|
|
for attempt in range(1, _MAX_RETRIES + 1):
|
|
try:
|
|
headers = {
|
|
"Authorization": f"Bearer {self._get_token()}",
|
|
"Accept": "application/json",
|
|
}
|
|
resp = requests.get(url, headers=headers, timeout=self.timeout)
|
|
if resp.status_code == 401 and attempt < _MAX_RETRIES:
|
|
self._token = None
|
|
continue
|
|
resp.raise_for_status()
|
|
return resp.json()
|
|
except requests.HTTPError as e:
|
|
status = e.response.status_code if e.response else 0
|
|
is_retryable = status in (429,) or status >= 500
|
|
if is_retryable and attempt < _MAX_RETRIES:
|
|
wait = 30 * attempt if status == 429 else _RETRY_BACKOFF[min(attempt - 1, len(_RETRY_BACKOFF) - 1)]
|
|
self.log.warning(f" HTTP {status} (attempt {attempt}/{_MAX_RETRIES}), retry in {wait}s")
|
|
time.sleep(wait)
|
|
else:
|
|
raise
|
|
except (requests.Timeout, requests.ConnectionError) as e:
|
|
if attempt < _MAX_RETRIES:
|
|
wait = _RETRY_BACKOFF[min(attempt - 1, len(_RETRY_BACKOFF) - 1)]
|
|
self.log.warning(f" Network error (attempt {attempt}/{_MAX_RETRIES}): {e}, retry in {wait}s")
|
|
time.sleep(wait)
|
|
else:
|
|
raise
|
|
|
|
def _paginate(self, url):
|
|
records = []
|
|
current = url
|
|
while current:
|
|
data = self._get(current)
|
|
if data.get("value"):
|
|
records.extend(data["value"])
|
|
current = data.get("@odata.nextLink")
|
|
return records
|
|
|
|
def get_companies(self):
|
|
companies = self._paginate(f"{self._base}/companies")
|
|
self.log.info(f"Found {len(companies)} company/companies")
|
|
return companies
|
|
|
|
def fetch_entity(self, company_id, entity, since=None, no_filter=False):
|
|
url = f"{self._base}/companies({company_id})/{entity}"
|
|
if since and not no_filter:
|
|
url += f"?$filter=lastModifiedDateTime gt {since}"
|
|
try:
|
|
return self._paginate(url)
|
|
except Exception as e:
|
|
self.log.warning(f" Failed to fetch {entity}: {e}")
|
|
return []
|
|
|
|
def fetch_document_with_lines(self, company_id, doc_entity, line_entity, since=None):
|
|
"""Fetch document headers + per-document line items; returns (docs, lines)."""
|
|
docs, lines = [], []
|
|
url = f"{self._base}/companies({company_id})/{doc_entity}"
|
|
if since:
|
|
url += f"?$filter=lastModifiedDateTime gt {since}"
|
|
try:
|
|
current = url
|
|
while current:
|
|
page = self._get(current)
|
|
if not page.get("value"):
|
|
break
|
|
for doc in page["value"]:
|
|
docs.append(doc)
|
|
doc_id = doc["id"]
|
|
lines_url = f"{self._base}/companies({company_id})/{doc_entity}({doc_id})/{line_entity}"
|
|
try:
|
|
doc_lines = self._paginate(lines_url)
|
|
lines.extend(doc_lines)
|
|
except Exception as e:
|
|
self.log.warning(f" Lines failed for {doc_entity}/{doc_id}: {e}")
|
|
if len(docs) % 100 == 0:
|
|
self.log.info(f" Progress: {len(docs)} docs, {len(lines)} lines")
|
|
current = page.get("@odata.nextLink")
|
|
except Exception as e:
|
|
self.log.warning(f" Failed to fetch {doc_entity}: {e}")
|
|
return docs, lines
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Encryption (GPG symmetric AES256)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _gpg_encrypt(src, dst, passphrase, log):
|
|
cmd = [
|
|
"gpg", "--batch", "--yes",
|
|
"--passphrase-fd", "0",
|
|
"--pinentry-mode", "loopback",
|
|
"--symmetric",
|
|
"--cipher-algo", "AES256",
|
|
"--compress-algo", "none",
|
|
"--output", str(dst),
|
|
str(src),
|
|
]
|
|
r = subprocess.run(cmd, input=passphrase.encode(), capture_output=True)
|
|
if r.returncode != 0:
|
|
raise RuntimeError(f"GPG encryption failed: {r.stderr.decode()}")
|
|
log.info(f"Encrypted → {Path(dst).name}")
|
|
|
|
|
|
def _gpg_decrypt(src, dst, passphrase, log):
|
|
cmd = [
|
|
"gpg", "--batch", "--yes",
|
|
"--passphrase-fd", "0",
|
|
"--pinentry-mode", "loopback",
|
|
"--decrypt",
|
|
"--output", str(dst),
|
|
str(src),
|
|
]
|
|
r = subprocess.run(cmd, input=passphrase.encode(), capture_output=True)
|
|
if r.returncode != 0:
|
|
raise RuntimeError(f"GPG decryption failed: {r.stderr.decode()}")
|
|
log.info(f"Decrypted {Path(src).name}")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# S3 helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _s3(config):
|
|
return boto3.client(
|
|
"s3",
|
|
endpoint_url=config["s3_endpoint"],
|
|
aws_access_key_id=config["s3_access_key"],
|
|
aws_secret_access_key=config["s3_secret_key"],
|
|
region_name=config.get("s3_region", "us-east-1"),
|
|
config=Config(signature_version="s3v4"),
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Backup
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def run_backup(config, log_callback=None, force_full=False, last_run_time=None):
|
|
"""
|
|
Run a BC backup. Returns dict with status, s3_key, etc.
|
|
last_run_time: ISO8601 UTC string used as the incremental cutoff.
|
|
"""
|
|
stamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
|
log_dir = Path(config.get("log_dir", "/data/logs"))
|
|
log = _make_logger(log_dir, f"backup_{stamp}", log_callback)
|
|
|
|
backup_type = "incremental" if (not force_full and last_run_time) else "full"
|
|
since = last_run_time if backup_type == "incremental" else None
|
|
run_start_time = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
|
|
|
|
if backup_type == "incremental":
|
|
log.info(f"Incremental backup — changes since {since}")
|
|
else:
|
|
log.info("Full backup")
|
|
|
|
env_name = config.get("bc_environment_name", "Production")
|
|
archive_base = f"bc_backup_{env_name}_{stamp}_{backup_type}"
|
|
|
|
result_base = {
|
|
"backup_type": backup_type,
|
|
"run_start_time": run_start_time,
|
|
"companies": [],
|
|
}
|
|
|
|
try:
|
|
client = BCClient(
|
|
tenant_id=config["azure_tenant_id"],
|
|
client_id=config["azure_client_id"],
|
|
client_secret=config["azure_client_secret"],
|
|
environment=env_name,
|
|
company_name=config.get("bc_company_name", ""),
|
|
logger=log,
|
|
)
|
|
|
|
log.info("=== Step 1: Authenticate to Azure AD ===")
|
|
client.authenticate()
|
|
|
|
log.info("=== Step 2: Fetch companies ===")
|
|
companies = client.get_companies()
|
|
if not companies:
|
|
raise RuntimeError(f"No companies found in environment '{env_name}'")
|
|
|
|
company_filter = config.get("bc_company_name", "").strip()
|
|
if company_filter:
|
|
companies = [c for c in companies
|
|
if c.get("name") == company_filter
|
|
or c.get("displayName") == company_filter]
|
|
if not companies:
|
|
raise RuntimeError(f"Company '{company_filter}' not found")
|
|
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
export_dir = Path(tmp) / archive_base
|
|
export_dir.mkdir()
|
|
|
|
total_records = 0
|
|
entities_ok = 0
|
|
entities_error = 0
|
|
company_names = []
|
|
|
|
log.info("=== Step 3: Export data ===")
|
|
for company in companies:
|
|
co_name = company.get("name", company.get("id", "unknown"))
|
|
co_id = company["id"]
|
|
company_names.append(co_name)
|
|
log.info(f"--- Company: {co_name} ---")
|
|
|
|
safe = "".join(c if c.isalnum() or c in " _-" else "_" for c in co_name)
|
|
co_dir = export_dir / safe
|
|
co_dir.mkdir()
|
|
(co_dir / "_company.json").write_text(json.dumps(company, indent=2))
|
|
|
|
for entity in STANDALONE_ENTITIES:
|
|
log.info(f" Exporting {entity}…")
|
|
records = client.fetch_entity(co_id, entity, since=since)
|
|
(co_dir / f"{entity}.json").write_text(json.dumps(records, indent=2))
|
|
log.info(f" {entity}: {len(records)} records")
|
|
total_records += len(records)
|
|
entities_ok += 1
|
|
|
|
for entity in REPORT_ENTITIES:
|
|
log.info(f" Exporting {entity} (report, always full)…")
|
|
records = client.fetch_entity(co_id, entity, no_filter=True)
|
|
(co_dir / f"{entity}.json").write_text(json.dumps(records, indent=2))
|
|
log.info(f" {entity}: {len(records)} records")
|
|
total_records += len(records)
|
|
entities_ok += 1
|
|
|
|
for doc_entity, line_entity in DOCUMENT_ENTITIES.items():
|
|
log.info(f" Exporting {doc_entity} (with {line_entity})…")
|
|
docs, lines = client.fetch_document_with_lines(co_id, doc_entity, line_entity, since=since)
|
|
(co_dir / f"{doc_entity}.jsonl").write_text(
|
|
"\n".join(json.dumps(d) for d in docs) + ("\n" if docs else "")
|
|
)
|
|
(co_dir / f"{line_entity}.jsonl").write_text(
|
|
"\n".join(json.dumps(l) for l in lines) + ("\n" if lines else "")
|
|
)
|
|
log.info(f" {doc_entity}: {len(docs)} docs, {len(lines)} lines")
|
|
total_records += len(docs) + len(lines)
|
|
entities_ok += 1
|
|
|
|
if backup_type == "incremental" and total_records == 0:
|
|
log.info("No changes detected — skipping upload")
|
|
return {**result_base, "status": "skipped", "total_records": 0,
|
|
"entities_ok": entities_ok, "entities_error": 0,
|
|
"companies": company_names, "s3_key": None}
|
|
|
|
parent_full_s3_key = config.get("last_full_s3_key") if backup_type == "incremental" else None
|
|
|
|
manifest = {
|
|
"backup_type": backup_type,
|
|
"environment": env_name,
|
|
"companies": company_names,
|
|
"since_datetime": since,
|
|
"run_start_time": run_start_time,
|
|
"total_records": total_records,
|
|
"parent_full_s3_key": parent_full_s3_key,
|
|
}
|
|
(export_dir / "manifest.json").write_text(json.dumps(manifest, indent=2))
|
|
|
|
log.info("=== Step 4: Create archive ===")
|
|
archive_path = Path(tmp) / f"{archive_base}.tar.gz"
|
|
with tarfile.open(archive_path, "w:gz") as tar:
|
|
tar.add(export_dir, arcname=archive_base)
|
|
archive_size_mb = archive_path.stat().st_size / 1024 / 1024
|
|
log.info(f"Archive: {archive_size_mb:.2f} MB")
|
|
|
|
log.info("=== Step 5: Encrypt (GPG AES256) ===")
|
|
enc_path = Path(tmp) / f"{archive_base}.tar.gz.gpg"
|
|
_gpg_encrypt(archive_path, enc_path, config["encryption_passphrase"], log)
|
|
enc_size_mb = enc_path.stat().st_size / 1024 / 1024
|
|
log.info(f"Encrypted: {enc_size_mb:.2f} MB")
|
|
|
|
log.info("=== Step 6: Upload to S3 ===")
|
|
s3_key = f"backups/{backup_type}/{archive_base}.tar.gz.gpg"
|
|
s3 = _s3(config)
|
|
s3.upload_file(
|
|
str(enc_path),
|
|
config["s3_bucket"],
|
|
s3_key,
|
|
ExtraArgs={"Metadata": {
|
|
"backup-type": backup_type,
|
|
"environment": env_name,
|
|
"timestamp": stamp,
|
|
}},
|
|
)
|
|
log.info(f"Uploaded: s3://{config['s3_bucket']}/{s3_key}")
|
|
|
|
log.info("=== Backup completed successfully ===")
|
|
return {
|
|
"status": "success",
|
|
"backup_type": backup_type,
|
|
"s3_key": s3_key,
|
|
"parent_full_s3_key": parent_full_s3_key,
|
|
"total_records": total_records,
|
|
"entities_ok": entities_ok,
|
|
"entities_error": entities_error,
|
|
"companies": company_names,
|
|
"run_start_time": run_start_time,
|
|
"archive_size_mb": enc_size_mb,
|
|
}
|
|
|
|
except Exception as e:
|
|
log.error(f"Backup failed: {e}")
|
|
return {**result_base, "status": "failed", "error": str(e),
|
|
"companies": result_base.get("companies", [])}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Restore
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _extract_archive(s3_key, config, passphrase, log):
|
|
"""Download, decrypt, and extract an archive. Returns path to extracted dir."""
|
|
tmp = tempfile.mkdtemp()
|
|
enc_path = Path(tmp) / Path(s3_key).name
|
|
log.info(f"Downloading {Path(s3_key).name}…")
|
|
_s3(config).download_file(config["s3_bucket"], s3_key, str(enc_path))
|
|
log.info(f"Downloaded {enc_path.stat().st_size / 1024 / 1024:.2f} MB")
|
|
|
|
tar_path = Path(tmp) / enc_path.name.replace(".gpg", "")
|
|
_gpg_decrypt(enc_path, tar_path, passphrase, log)
|
|
enc_path.unlink()
|
|
|
|
extract_dir = Path(tmp) / "extract"
|
|
extract_dir.mkdir()
|
|
with tarfile.open(tar_path, "r:gz") as tar:
|
|
tar.extractall(extract_dir)
|
|
tar_path.unlink()
|
|
return extract_dir, tmp
|
|
|
|
|
|
def restore_backup(s3_key, config, output_dir, log_callback=None, entity_filter=None):
|
|
"""Download, decrypt, and extract a single backup archive."""
|
|
stamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
|
log_dir = Path(config.get("log_dir", "/data/logs"))
|
|
log = _make_logger(log_dir, f"restore_{stamp}", log_callback)
|
|
output_dir = Path(output_dir)
|
|
output_dir.mkdir(parents=True, exist_ok=True)
|
|
passphrase = config["encryption_passphrase"]
|
|
|
|
try:
|
|
extract_dir, tmp = _extract_archive(s3_key, config, passphrase, log)
|
|
|
|
filters = [f.strip().lower() for f in entity_filter.split(",") if f.strip()] \
|
|
if entity_filter else None
|
|
|
|
import shutil, tempfile as _tf
|
|
log.info("Copying files…")
|
|
for f in extract_dir.rglob("*"):
|
|
if not f.is_file():
|
|
continue
|
|
if filters and f.suffix in (".json", ".jsonl"):
|
|
if not any(fl in f.stem.lower() for fl in filters):
|
|
if f.name not in ("manifest.json", "_company.json"):
|
|
continue
|
|
rel = f.relative_to(extract_dir)
|
|
dst = output_dir / rel
|
|
dst.parent.mkdir(parents=True, exist_ok=True)
|
|
shutil.copy2(f, dst)
|
|
|
|
shutil.rmtree(tmp, ignore_errors=True)
|
|
|
|
file_count = (sum(1 for _ in output_dir.rglob("*.json"))
|
|
+ sum(1 for _ in output_dir.rglob("*.jsonl")))
|
|
|
|
manifest = {}
|
|
for mf in output_dir.rglob("manifest.json"):
|
|
try:
|
|
manifest = json.loads(mf.read_text())
|
|
except Exception:
|
|
pass
|
|
break
|
|
|
|
log.info(f"Restored {file_count} files")
|
|
return {"status": "success", "file_count": file_count, "manifest": manifest}
|
|
|
|
except Exception as e:
|
|
log.error(f"Restore failed: {e}")
|
|
return {"status": "failed", "error": str(e)}
|
|
|
|
|
|
def chain_restore(s3_keys, config, output_dir, log_callback=None, entity_filter=None):
|
|
"""
|
|
Restore a chain of archives (full + incrementals) merged into a complete snapshot.
|
|
Records from later archives overwrite earlier ones (matched by 'id' or 'number').
|
|
"""
|
|
stamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
|
log_dir = Path(config.get("log_dir", "/data/logs"))
|
|
log = _make_logger(log_dir, f"chain_restore_{stamp}", log_callback)
|
|
output_dir = Path(output_dir)
|
|
chain_dir = output_dir / "chain"
|
|
chain_dir.mkdir(parents=True, exist_ok=True)
|
|
passphrase = config["encryption_passphrase"]
|
|
|
|
filters = [f.strip().lower() for f in entity_filter.split(",") if f.strip()] \
|
|
if entity_filter else None
|
|
|
|
# merged[company_safe_name][entity_stem] = {record_id: record_dict}
|
|
merged: dict[str, dict[str, dict]] = {}
|
|
last_manifest = {}
|
|
|
|
import shutil
|
|
|
|
try:
|
|
for idx, s3_key in enumerate(s3_keys):
|
|
log.info(f"[{idx+1}/{len(s3_keys)}] {Path(s3_key).name}")
|
|
extract_dir, tmp = _extract_archive(s3_key, config, passphrase, log)
|
|
|
|
for f in extract_dir.rglob("manifest.json"):
|
|
try:
|
|
last_manifest = json.loads(f.read_text())
|
|
except Exception:
|
|
pass
|
|
|
|
# Traverse company dirs inside the archive
|
|
for archive_subdir in extract_dir.iterdir():
|
|
if not archive_subdir.is_dir():
|
|
continue
|
|
for co_dir in archive_subdir.iterdir():
|
|
if not co_dir.is_dir():
|
|
continue
|
|
co_name = co_dir.name
|
|
if co_name not in merged:
|
|
merged[co_name] = {}
|
|
|
|
# Copy _company.json straight through
|
|
company_meta = co_dir / "_company.json"
|
|
if company_meta.exists():
|
|
dst = chain_dir / co_name / "_company.json"
|
|
dst.parent.mkdir(parents=True, exist_ok=True)
|
|
shutil.copy2(company_meta, dst)
|
|
|
|
for f in co_dir.iterdir():
|
|
if not f.is_file():
|
|
continue
|
|
stem = f.stem
|
|
if filters and stem.lower() not in filters:
|
|
continue
|
|
if f.name in ("_company.json",):
|
|
continue
|
|
|
|
if f.suffix == ".json":
|
|
try:
|
|
records = json.loads(f.read_text())
|
|
if not isinstance(records, list):
|
|
continue
|
|
if stem not in merged[co_name]:
|
|
merged[co_name][stem] = {}
|
|
for r in records:
|
|
rid = r.get("id") or r.get("number") or json.dumps(r)[:64]
|
|
merged[co_name][stem][rid] = r
|
|
except Exception as e:
|
|
log.warning(f" Skip {f.name}: {e}")
|
|
|
|
elif f.suffix == ".jsonl":
|
|
if stem not in merged[co_name]:
|
|
merged[co_name][stem] = {}
|
|
for line in f.read_text().splitlines():
|
|
line = line.strip()
|
|
if not line:
|
|
continue
|
|
try:
|
|
r = json.loads(line)
|
|
rid = r.get("id") or r.get("number") or line[:64]
|
|
merged[co_name][stem][rid] = r
|
|
except Exception:
|
|
pass
|
|
|
|
shutil.rmtree(tmp, ignore_errors=True)
|
|
log.info(f" Merged archive {idx+1}")
|
|
|
|
# Write merged output
|
|
log.info("Writing merged chain output…")
|
|
file_count = 0
|
|
for co_name, entities in merged.items():
|
|
co_chain_dir = chain_dir / co_name
|
|
co_chain_dir.mkdir(parents=True, exist_ok=True)
|
|
for stem, records_by_id in entities.items():
|
|
records = list(records_by_id.values())
|
|
if stem in _DOCUMENT_LINE_ENTITIES or stem in DOCUMENT_ENTITIES:
|
|
out = co_chain_dir / f"{stem}.jsonl"
|
|
out.write_text("\n".join(json.dumps(r) for r in records) + "\n")
|
|
else:
|
|
out = co_chain_dir / f"{stem}.json"
|
|
out.write_text(json.dumps(records, indent=2))
|
|
file_count += 1
|
|
|
|
chain_manifest = {
|
|
**last_manifest,
|
|
"restore_mode": "chain",
|
|
"chain_length": len(s3_keys),
|
|
"s3_keys": s3_keys,
|
|
}
|
|
(chain_dir / "chain_manifest.json").write_text(json.dumps(chain_manifest, indent=2))
|
|
|
|
log.info(f"Chain restore complete: {file_count} entity files")
|
|
return {"status": "success", "file_count": file_count, "manifest": chain_manifest}
|
|
|
|
except Exception as e:
|
|
log.error(f"Chain restore failed: {e}")
|
|
return {"status": "failed", "error": str(e)}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# List remote backups
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def list_remote_backups(config):
|
|
s3 = _s3(config)
|
|
resp = s3.list_objects_v2(Bucket=config["s3_bucket"], Prefix="backups/")
|
|
objects = resp.get("Contents", [])
|
|
result = []
|
|
for obj in objects:
|
|
key = obj["Key"]
|
|
if not key.endswith(".gpg"):
|
|
continue
|
|
name = Path(key).name
|
|
backup_type = "incremental" if "_incremental" in name else "full"
|
|
result.append({
|
|
"key": key,
|
|
"name": name,
|
|
"backup_type": backup_type,
|
|
"size_mb": round(obj["Size"] / 1024 / 1024, 2),
|
|
"last_modified": obj["LastModified"].isoformat(),
|
|
})
|
|
result.sort(key=lambda x: x["last_modified"], reverse=True)
|
|
return result
|