diff --git a/Dockerfile b/Dockerfile
new file mode 100644
index 0000000..29cd36e
--- /dev/null
+++ b/Dockerfile
@@ -0,0 +1,18 @@
+FROM python:3.11-slim
+
+RUN apt-get update && apt-get install -y --no-install-recommends gnupg \
+ && rm -rf /var/lib/apt/lists/*
+
+WORKDIR /app
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+COPY backup_bc.py web.py ./
+COPY templates/ templates/
+
+VOLUME ["/data"]
+ENV BCBAK_DATA=/data
+ENV PORT=1890
+EXPOSE 1890
+
+CMD ["python", "web.py"]
diff --git a/backup_bc.py b/backup_bc.py
new file mode 100644
index 0000000..1e1c5a8
--- /dev/null
+++ b/backup_bc.py
@@ -0,0 +1,674 @@
+#!/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
diff --git a/docker-compose.yml b/docker-compose.yml
new file mode 100644
index 0000000..ab6f7e4
--- /dev/null
+++ b/docker-compose.yml
@@ -0,0 +1,16 @@
+services:
+ bcbak:
+ build: .
+ container_name: bcbak
+ restart: unless-stopped
+ ports:
+ - "1890:1890"
+ volumes:
+ - bcbak_data:/data
+ environment:
+ - BCBAK_DATA=/data
+ - PORT=1890
+ - SECRET_KEY=${SECRET_KEY:-change-me-in-production}
+
+volumes:
+ bcbak_data:
diff --git a/requirements.txt b/requirements.txt
new file mode 100644
index 0000000..d8a1d2a
--- /dev/null
+++ b/requirements.txt
@@ -0,0 +1,4 @@
+boto3>=1.34
+requests>=2.31
+flask>=3.0
+APScheduler>=3.10
diff --git a/templates/base.html b/templates/base.html
new file mode 100644
index 0000000..0e83692
--- /dev/null
+++ b/templates/base.html
@@ -0,0 +1,278 @@
+
+
+
+
+
+ BackupBC{% block title %}{% endblock %}
+
+
+
+
+
+
+
+
+
+ {% with messages = get_flashed_messages(with_categories=true) %}
+ {% if messages %}
+ {% for category, message in messages %}
+
+
+ {{ message }}
+
+
+ {% endfor %}
+ {% endif %}
+ {% endwith %}
+
+ {% block content %}{% endblock %}
+
+
+
+
+
+
+
+
+
+
+
Authenticating with Azure AD…
+
+
+
+
+
+
+
+
+
+
+{% block scripts %}{% endblock %}
+
+
diff --git a/templates/dashboard.html b/templates/dashboard.html
new file mode 100644
index 0000000..1296391
--- /dev/null
+++ b/templates/dashboard.html
@@ -0,0 +1,163 @@
+{% extends "base.html" %}
+{% block title %} — Dashboard{% endblock %}
+
+{% block content %}
+
+
+
+
+
+
+
+
+
+
+
+
Total Backups
+
{{ total }}
+
+
+
+
+
+
+
+
+
+
+
+
+
Last Success
+
{{ last_ok['started_at'] | fmtdt if last_ok else '—' }}
+ {% if last_ok %}
+
{{ last_ok['total_records'] or 0 | int }} records
+ {% endif %}
+
+
+
+
+
+
+
+
+
+
+
+
+
Last Full Backup
+
{{ last_full['started_at'] | fmtdt if last_full else '—' }}
+ {% if last_full %}
+
{{ '%.1f' | format(last_full['archive_size_mb'] or 0) }} MB
+ {% endif %}
+
+
+
+
+
+
+
+
+
+
+
+
+
Next Incremental
+
{{ next_incr or 'Not scheduled' }}
+
Full: {{ next_full_run or 'Not scheduled' }}
+
+
+
+
+
+
+
+
+
+
+
+
+
+ #
+ Started
+ Type
+ Trigger
+ Companies
+ Records
+ Size
+ Status
+
+
+
+ {% for r in recent %}
+
+ {{ r['id'] }}
+ {{ r['started_at'] | fmtdt }}
+
+ {% if r['backup_type'] == 'full' %}
+ Full
+ {% else %}
+ Incr
+ {% endif %}
+
+
+
+ {{ r['triggered_by'] }}
+
+
+
+ {% if r['companies'] %}
+ {{ r['companies'] | replace('["','') | replace('"]','') | replace('","',' · ') }}
+ {% else %}—{% endif %}
+
+ {{ r['total_records'] or '—' }}
+ {% if r['archive_size_mb'] %}{{ '%.1f' | format(r['archive_size_mb']) }} MB{% else %}—{% endif %}
+
+ {% if r['status'] == 'success' %}
+ OK
+ {% elif r['status'] == 'failed' %}
+ Failed
+ {% else %}
+ Running
+ {% endif %}
+
+
+ {% else %}
+
+
+
+ No backups yet — click Full Backup to start.
+
+
+ {% endfor %}
+
+
+
+
+{% endblock %}
diff --git a/templates/history.html b/templates/history.html
new file mode 100644
index 0000000..dbc3110
--- /dev/null
+++ b/templates/history.html
@@ -0,0 +1,131 @@
+{% extends "base.html" %}
+{% block title %} — History{% endblock %}
+
+{% block content %}
+
+
+
+
+
+
+
+ #
+ Started
+ Completed
+ Duration
+ Type
+ Trigger
+ Companies
+ Entities OK
+ Records
+ Size
+ Status
+ Archive
+
+
+
+
+ {% for r in runs %}
+
+ {{ r['id'] }}
+ {{ r['started_at'] | fmtdt }}
+ {{ r['completed_at'] | fmtdt }}
+ {{ r['started_at'] | duration(r['completed_at']) }}
+
+ {% if r['backup_type'] == 'full' %}
+ Full
+ {% else %}
+ Incr
+ {% endif %}
+
+
+
+ {{ r['triggered_by'] }}
+
+
+
+ {% if r['companies'] %}
+ {{ r['companies'] | replace('["','') | replace('"]','') | replace('","',' · ') }}
+ {% else %}—{% endif %}
+
+ {{ r['entities_ok'] or '—' }}
+ {{ r['total_records'] or '—' }}
+ {% if r['archive_size_mb'] %}{{ '%.1f' | format(r['archive_size_mb']) }} MB{% else %}—{% endif %}
+
+ {% if r['status'] == 'success' %}
+ Success
+ {% elif r['status'] == 'failed' %}
+
+ Failed
+
+ {% else %}
+ Running
+ {% endif %}
+
+
+ {% if r['s3_key'] %}
+ {{ r['s3_key'].split('/')[-1] }}
+ {% else %}—{% endif %}
+
+
+ {% if r['s3_key'] and r['status'] == 'success' %}
+
+ Restore
+
+ {% endif %}
+
+
+ {% if r['status'] == 'failed' and r['error_message'] %}
+
+
+ {{ r['error_message'] }}
+
+
+ {% endif %}
+ {% else %}
+
+
+ No backup history yet
+
+
+ {% endfor %}
+
+
+
+
+ {% if pages > 1 %}
+
+ {% endif %}
+
+{% endblock %}
diff --git a/templates/restore_list.html b/templates/restore_list.html
new file mode 100644
index 0000000..5e725a5
--- /dev/null
+++ b/templates/restore_list.html
@@ -0,0 +1,356 @@
+{% extends "base.html" %}
+{% block title %} — Restore{% endblock %}
+
+{% block content %}
+
+
+
+
+
+
+
+
+
+ Archive
+ Last Modified
+ Size
+
+
+
+
+ {% for b in remote_backups %}
+
+
+ {% if b.get('backup_type') == 'incremental' %}
+ Incr
+ {% else %}
+ Full
+ {% endif %}
+ {{ b['name'] }}
+
+ {{ b['last_modified'][:16].replace('T',' ') }}
+ {{ b['size_mb'] }} MB
+
+
+ Restore
+
+
+
+ {% else %}
+
+
+
+ No backups found — check your S3 settings or run a backup first.
+
+
+ {% endfor %}
+
+
+
+
+
+
+
+
+
+
+
+
+ #
+ Started
+ Duration
+ Archive
+ Files
+ Status
+
+
+
+
+ {% for r in restores %}
+
+ {{ r['id'] }}
+ {{ r['started_at'] | fmtdt }}
+ {{ r['started_at'] | duration(r['completed_at']) }}
+
+ {{ r['s3_key'].split('/')[-1] }}
+
+ {{ r['file_count'] or '—' }}
+
+ {% if r['status'] == 'success' %}
+ Ready
+ {% elif r['status'] == 'failed' %}
+
+ Failed
+
+ {% else %}
+ Running
+ {% endif %}
+
+
+ {% if r['status'] == 'success' %}
+
+ Browse
+
+
+ ZIP
+
+ {% elif r['status'] == 'running' %}
+
+ View
+
+ {% endif %}
+
+
+
+
+
+ {% else %}
+
+ No restores yet
+
+ {% endfor %}
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+ Full
+
+
+
+
+
+
+
+
+ Chain Restore
+
+
+
+ Downloads the parent full backup + all incrementals up to this point,
+ merged into a complete database snapshot.
+
+
+
+ Partial restore — only entities changed in this specific incremental will be
+ extracted. Enable Chain Restore above for a complete snapshot.
+
+
+
+
+
+ Entity filter (optional)
+
+
+
Comma-separated entity names to extract.
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
Downloading from S3…
+
+
+
+
+
+
+
+{% endblock %}
+
+{% block scripts %}
+
+{% endblock %}
diff --git a/templates/restore_view.html b/templates/restore_view.html
new file mode 100644
index 0000000..4c611e4
--- /dev/null
+++ b/templates/restore_view.html
@@ -0,0 +1,165 @@
+{% extends "base.html" %}
+{% block title %} — Restore #{{ restore['id'] }}{% endblock %}
+
+{% block content %}
+
+
+{% if streaming_id %}
+
+
+
+
+
+
Restore in progress…
+
+
+
+
+
+{% elif restore['status'] == 'failed' %}
+
+
+ Restore failed: {{ restore['error_message'] or 'Unknown error' }}
+
+
+{% else %}
+
+{% set rmode = restore_info.get('restore_mode', 'full') %}
+{% if rmode == 'chain' %}
+
+
+ Chain Restore — {{ restore_info.get('chain_length', 1) }} archives merged
+ (1 full + {{ restore_info.get('chain_length', 1) - 1 }} incremental).
+ {{ restore['file_count'] }} entity files — complete snapshot.
+
+{% elif rmode == 'incremental' %}
+
+
+ Partial Restore — this is a single incremental archive.
+ Only {{ restore['file_count'] }} entity files with changes in that specific run are present.
+ Use Chain Restore from the Restore page for a complete snapshot.
+
+{% else %}
+
+
+ Full Snapshot — {{ restore['file_count'] }} entity files extracted.
+
+{% endif %}
+
+{% if by_company %}
+
+
+
+
+
+
+ {% for company in by_company | sort %}
+ {% set entities = by_company[company] %}
+
+
+
+
+
+
+
+
+ Entity
+ Format
+ Size
+
+
+
+
+ {% for f in entities %}
+
+ {{ f['entity'] }}
+ {{ f['ext'] }}
+ {{ f['size_kb'] }} KB
+
+
+ Download
+
+
+
+ {% endfor %}
+
+
+
+
+
+
+ {% endfor %}
+
+{% else %}
+No entity files found in this restore.
+{% endif %}
+{% endif %}
+{% endblock %}
+
+{% block scripts %}
+{% if streaming_id %}
+
+{% endif %}
+
+{% if by_company %}
+
+{% endif %}
+{% endblock %}
diff --git a/templates/settings.html b/templates/settings.html
new file mode 100644
index 0000000..eba22dc
--- /dev/null
+++ b/templates/settings.html
@@ -0,0 +1,259 @@
+{% extends "base.html" %}
+{% block title %} — Settings{% endblock %}
+
+{% block content %}
+
+
+
+{% endblock %}
+
+{% block scripts %}
+
+{% endblock %}
diff --git a/web.py b/web.py
new file mode 100644
index 0000000..200c9ce
--- /dev/null
+++ b/web.py
@@ -0,0 +1,717 @@
+#!/usr/bin/env python3
+"""BackupBC Web Interface — Flask app with APScheduler and live SSE log streaming."""
+
+import datetime
+import io
+import json
+import logging
+import os
+import queue
+import shutil
+import sqlite3
+import threading
+import zipfile
+from pathlib import Path
+
+from apscheduler.schedulers.background import BackgroundScheduler
+from apscheduler.triggers.cron import CronTrigger
+from flask import (Flask, Response, abort, flash, jsonify, redirect,
+ render_template, request, send_file, url_for)
+
+from backup_bc import chain_restore, list_remote_backups, restore_backup, run_backup
+
+# ---------------------------------------------------------------------------
+# Paths
+# ---------------------------------------------------------------------------
+
+DATA_DIR = Path(os.environ.get("BCBAK_DATA", Path(__file__).parent / "data"))
+DATA_DIR.mkdir(parents=True, exist_ok=True)
+
+DB_PATH = DATA_DIR / "bcbak.db"
+LOG_DIR = DATA_DIR / "logs"
+RESTORE_DIR = DATA_DIR / "restores"
+
+LOG_DIR.mkdir(parents=True, exist_ok=True)
+RESTORE_DIR.mkdir(parents=True, exist_ok=True)
+
+# ---------------------------------------------------------------------------
+# App
+# ---------------------------------------------------------------------------
+
+app = Flask(__name__)
+app.secret_key = os.environ.get("SECRET_KEY", os.urandom(24).hex())
+
+logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)-8s %(message)s")
+
+scheduler = BackgroundScheduler(timezone="UTC")
+
+_active: dict[int, dict] = {}
+_active_lock = threading.Lock()
+
+_active_restores: dict[int, dict] = {}
+_restore_lock = threading.Lock()
+
+# ---------------------------------------------------------------------------
+# Database
+# ---------------------------------------------------------------------------
+
+def get_db() -> sqlite3.Connection:
+ conn = sqlite3.connect(DB_PATH, check_same_thread=False)
+ conn.row_factory = sqlite3.Row
+ conn.execute("PRAGMA journal_mode=WAL")
+ return conn
+
+
+def init_db() -> None:
+ with get_db() as conn:
+ conn.executescript("""
+ CREATE TABLE IF NOT EXISTS config (
+ key TEXT PRIMARY KEY,
+ value TEXT
+ );
+ CREATE TABLE IF NOT EXISTS runs (
+ id INTEGER PRIMARY KEY AUTOINCREMENT,
+ started_at TEXT NOT NULL,
+ completed_at TEXT,
+ status TEXT NOT NULL DEFAULT 'running',
+ backup_type TEXT NOT NULL DEFAULT 'full',
+ triggered_by TEXT NOT NULL DEFAULT 'scheduled',
+ companies TEXT,
+ entities_ok INTEGER DEFAULT 0,
+ entities_error INTEGER DEFAULT 0,
+ total_records INTEGER DEFAULT 0,
+ archive_size_mb REAL,
+ s3_key TEXT,
+ parent_full_s3_key TEXT,
+ run_start_time TEXT,
+ error_message TEXT
+ );
+ CREATE TABLE IF NOT EXISTS restores (
+ id INTEGER PRIMARY KEY AUTOINCREMENT,
+ started_at TEXT NOT NULL,
+ completed_at TEXT,
+ status TEXT NOT NULL DEFAULT 'running',
+ s3_key TEXT NOT NULL,
+ output_dir TEXT,
+ file_count INTEGER DEFAULT 0,
+ is_chain INTEGER DEFAULT 0,
+ chain_length INTEGER DEFAULT 1,
+ error_message TEXT
+ );
+ """)
+ # Migrations for any missing columns
+ runs_cols = {row[1] for row in conn.execute("PRAGMA table_info(runs)").fetchall()}
+ restore_cols = {row[1] for row in conn.execute("PRAGMA table_info(restores)").fetchall()}
+ for col, defn in [
+ ("parent_full_s3_key", "TEXT"),
+ ("run_start_time", "TEXT"),
+ ("total_records", "INTEGER DEFAULT 0"),
+ ]:
+ if col not in runs_cols:
+ conn.execute(f"ALTER TABLE runs ADD COLUMN {col} {defn}")
+ for col, defn in [
+ ("is_chain", "INTEGER DEFAULT 0"),
+ ("chain_length", "INTEGER DEFAULT 1"),
+ ]:
+ if col not in restore_cols:
+ conn.execute(f"ALTER TABLE restores ADD COLUMN {col} {defn}")
+
+
+def cfg_get(key: str, default: str = "") -> str:
+ with get_db() as conn:
+ row = conn.execute("SELECT value FROM config WHERE key=?", (key,)).fetchone()
+ return row["value"] if row else default
+
+
+def cfg_all() -> dict:
+ with get_db() as conn:
+ rows = conn.execute("SELECT key, value FROM config").fetchall()
+ return {r["key"]: r["value"] for r in rows}
+
+
+def cfg_set(data: dict) -> None:
+ with get_db() as conn:
+ for k, v in data.items():
+ conn.execute("INSERT OR REPLACE INTO config(key,value) VALUES(?,?)", (k, str(v)))
+
+
+# ---------------------------------------------------------------------------
+# Config builder
+# ---------------------------------------------------------------------------
+
+def _config_for_backup() -> dict:
+ c = cfg_all()
+ return {
+ "azure_tenant_id": c.get("azure_tenant_id", ""),
+ "azure_client_id": c.get("azure_client_id", ""),
+ "azure_client_secret": c.get("azure_client_secret", ""),
+ "bc_environment_name": c.get("bc_environment_name", "Production"),
+ "bc_company_name": c.get("bc_company_name", ""),
+ "s3_endpoint": c.get("s3_endpoint", ""),
+ "s3_access_key": c.get("s3_access_key", ""),
+ "s3_secret_key": c.get("s3_secret_key", ""),
+ "s3_bucket": c.get("s3_bucket", "bcbak"),
+ "s3_region": c.get("s3_region", "us-east-1"),
+ "encryption_passphrase": c.get("encryption_passphrase", ""),
+ "log_dir": str(LOG_DIR),
+ "last_full_s3_key": c.get("last_full_s3_key", ""),
+ }
+
+
+# ---------------------------------------------------------------------------
+# Backup runner
+# ---------------------------------------------------------------------------
+
+def _launch_backup(triggered_by: str = "scheduled", force_full: bool = False) -> int:
+ with _active_lock:
+ if any(i["status"] == "running" for i in _active.values()):
+ raise RuntimeError("A backup is already in progress")
+
+ backup_type = "full" if force_full else "incremental"
+ with get_db() as conn:
+ cur = conn.execute(
+ "INSERT INTO runs(started_at,status,backup_type,triggered_by) VALUES(?,?,?,?)",
+ (datetime.datetime.now().isoformat(timespec="seconds"), "running",
+ backup_type, triggered_by),
+ )
+ run_id = cur.lastrowid
+
+ log_q: queue.Queue = queue.Queue()
+ with _active_lock:
+ _active[run_id] = {"queue": log_q, "status": "running"}
+
+ def worker() -> None:
+ try:
+ last_run_time = None
+ if not force_full:
+ last_run_time = cfg_get("last_incr_run_time") or None
+
+ result = run_backup(
+ _config_for_backup(),
+ log_callback=lambda m: log_q.put(m),
+ force_full=force_full,
+ last_run_time=last_run_time,
+ )
+ status = result.get("status", "failed")
+ if status == "skipped":
+ status = "success" # skipped = no changes = not a failure
+
+ if status == "success" and result.get("s3_key"):
+ if force_full:
+ cfg_set({"last_full_s3_key": result["s3_key"]})
+ if result.get("run_start_time"):
+ cfg_set({"last_incr_run_time": result["run_start_time"]})
+
+ with get_db() as conn:
+ conn.execute(
+ """UPDATE runs
+ SET status=?, completed_at=?, companies=?,
+ entities_ok=?, entities_error=?, total_records=?,
+ archive_size_mb=?, s3_key=?, parent_full_s3_key=?,
+ run_start_time=?, error_message=?
+ WHERE id=?""",
+ (status,
+ datetime.datetime.now().isoformat(timespec="seconds"),
+ json.dumps(result.get("companies", [])),
+ result.get("entities_ok", 0),
+ result.get("entities_error", 0),
+ result.get("total_records", 0),
+ result.get("archive_size_mb"),
+ result.get("s3_key"),
+ result.get("parent_full_s3_key"),
+ result.get("run_start_time"),
+ result.get("error"),
+ run_id),
+ )
+ with _active_lock:
+ _active[run_id]["status"] = status
+ except Exception as exc:
+ app.logger.error(f"Backup worker error: {exc}", exc_info=True)
+ with get_db() as conn:
+ conn.execute(
+ "UPDATE runs SET status='failed', completed_at=?, error_message=? WHERE id=?",
+ (datetime.datetime.now().isoformat(timespec="seconds"), str(exc), run_id),
+ )
+ log_q.put(f"CRITICAL: {exc}")
+ with _active_lock:
+ _active[run_id]["status"] = "failed"
+ finally:
+ log_q.put(None)
+
+ t = threading.Thread(target=worker, daemon=True, name=f"backup-{run_id}")
+ t.start()
+ with _active_lock:
+ _active[run_id]["thread"] = t
+ return run_id
+
+
+# ---------------------------------------------------------------------------
+# Restore runner
+# ---------------------------------------------------------------------------
+
+def _get_restore_chain(target_s3_key: str) -> list[str] | None:
+ """Return ordered [full_key, incr_1, ..., target_key] or None if unavailable."""
+ with get_db() as conn:
+ target = conn.execute(
+ "SELECT * FROM runs WHERE s3_key=? AND status='success'",
+ (target_s3_key,),
+ ).fetchone()
+ if not target or target["backup_type"] == "full":
+ return None
+ parent_key = target["parent_full_s3_key"]
+ if not parent_key:
+ return None
+ full_run = conn.execute(
+ "SELECT started_at FROM runs WHERE s3_key=? AND status='success'",
+ (parent_key,),
+ ).fetchone()
+ if not full_run:
+ return None
+ incrementals = conn.execute(
+ """SELECT s3_key FROM runs
+ WHERE backup_type='incremental' AND parent_full_s3_key=?
+ AND started_at >= ? AND started_at <= ? AND status='success'
+ ORDER BY started_at ASC""",
+ (parent_key, full_run["started_at"], target["started_at"]),
+ ).fetchall()
+ return [parent_key] + [r["s3_key"] for r in incrementals]
+
+
+def _launch_restore(s3_key: str, entity_filter: str | None = None, chain: bool = False) -> int:
+ chain_keys: list[str] | None = None
+ if chain:
+ chain_keys = _get_restore_chain(s3_key)
+ if not chain_keys:
+ chain = False
+
+ is_chain = chain and chain_keys is not None
+ chain_len = len(chain_keys) if chain_keys else 1
+
+ with get_db() as conn:
+ cur = conn.execute(
+ "INSERT INTO restores(started_at,status,s3_key,is_chain,chain_length) VALUES(?,?,?,?,?)",
+ (datetime.datetime.now().isoformat(timespec="seconds"), "running",
+ s3_key, int(is_chain), chain_len),
+ )
+ restore_id = cur.lastrowid
+
+ out_dir = RESTORE_DIR / str(restore_id)
+ log_q: queue.Queue = queue.Queue()
+ with _restore_lock:
+ _active_restores[restore_id] = {"queue": log_q, "status": "running"}
+
+ def worker() -> None:
+ try:
+ cfg = _config_for_backup()
+ if is_chain:
+ result = chain_restore(chain_keys, cfg, out_dir,
+ log_callback=lambda m: log_q.put(m),
+ entity_filter=entity_filter)
+ else:
+ result = restore_backup(s3_key, cfg, out_dir,
+ log_callback=lambda m: log_q.put(m),
+ entity_filter=entity_filter)
+ status = result["status"]
+ with get_db() as conn:
+ conn.execute(
+ """UPDATE restores
+ SET status=?,completed_at=?,output_dir=?,file_count=?,error_message=?
+ WHERE id=?""",
+ (status,
+ datetime.datetime.now().isoformat(timespec="seconds"),
+ str(out_dir),
+ result.get("file_count", 0),
+ result.get("error"),
+ restore_id),
+ )
+ with _restore_lock:
+ _active_restores[restore_id]["status"] = status
+ except Exception as exc:
+ app.logger.error(f"Restore worker error: {exc}", exc_info=True)
+ with get_db() as conn:
+ conn.execute(
+ "UPDATE restores SET status='failed',completed_at=?,error_message=? WHERE id=?",
+ (datetime.datetime.now().isoformat(timespec="seconds"), str(exc), restore_id),
+ )
+ log_q.put(f"CRITICAL: {exc}")
+ with _restore_lock:
+ _active_restores[restore_id]["status"] = "failed"
+ finally:
+ log_q.put(None)
+
+ t = threading.Thread(target=worker, daemon=True, name=f"restore-{restore_id}")
+ t.start()
+ with _restore_lock:
+ _active_restores[restore_id]["thread"] = t
+ return restore_id
+
+
+# ---------------------------------------------------------------------------
+# SSE helper
+# ---------------------------------------------------------------------------
+
+def _sse_stream(run_id: int, active_dict: dict, lock: threading.Lock):
+ with lock:
+ info = active_dict.get(run_id)
+ if info is None:
+ yield f"data: {json.dumps({'done': True, 'status': 'unknown'})}\n\n"
+ return
+ log_q = info["queue"]
+ while True:
+ try:
+ msg = log_q.get(timeout=25)
+ if msg is None:
+ with lock:
+ status = active_dict[run_id]["status"]
+ yield f"data: {json.dumps({'done': True, 'status': status})}\n\n"
+ return
+ yield f"data: {json.dumps({'log': msg})}\n\n"
+ except Exception:
+ yield f"data: {json.dumps({'keepalive': True})}\n\n"
+
+
+# ---------------------------------------------------------------------------
+# Scheduler
+# ---------------------------------------------------------------------------
+
+def _cron_trigger(expr: str) -> CronTrigger:
+ parts = expr.strip().split()
+ return CronTrigger(
+ minute=parts[0], hour=parts[1], day=parts[2],
+ month=parts[3], day_of_week=parts[4],
+ )
+
+
+def _apply_schedule() -> None:
+ scheduler.remove_all_jobs()
+ if cfg_get("schedule_incr_enabled", "false").lower() == "true":
+ cron = cfg_get("schedule_incr_cron", "0 * * * *").strip()
+ try:
+ scheduler.add_job(_launch_backup, _cron_trigger(cron), id="incr_backup",
+ kwargs={"triggered_by": "scheduled", "force_full": False},
+ misfire_grace_time=3600)
+ app.logger.info(f"Incremental backup scheduled: {cron}")
+ except Exception as exc:
+ app.logger.error(f"Invalid incremental cron '{cron}': {exc}")
+ if cfg_get("schedule_full_enabled", "false").lower() == "true":
+ cron = cfg_get("schedule_full_cron", "0 1 * * 0").strip()
+ try:
+ scheduler.add_job(_launch_backup, _cron_trigger(cron), id="full_backup",
+ kwargs={"triggered_by": "scheduled", "force_full": True},
+ misfire_grace_time=3600)
+ app.logger.info(f"Full backup scheduled: {cron}")
+ except Exception as exc:
+ app.logger.error(f"Invalid full backup cron '{cron}': {exc}")
+
+
+# ---------------------------------------------------------------------------
+# Template filters
+# ---------------------------------------------------------------------------
+
+@app.template_filter("duration")
+def duration_filter(started: str, completed: str | None) -> str:
+ if not started or not completed:
+ return "—"
+ try:
+ secs = int((datetime.datetime.fromisoformat(completed) -
+ datetime.datetime.fromisoformat(started)).total_seconds())
+ return f"{secs // 60}m {secs % 60}s" if secs >= 60 else f"{secs}s"
+ except Exception:
+ return "—"
+
+
+@app.template_filter("fmtdt")
+def fmtdt_filter(dt_str: str | None) -> str:
+ return dt_str[:16].replace("T", " ") if dt_str else "—"
+
+
+# ---------------------------------------------------------------------------
+# Routes — pages
+# ---------------------------------------------------------------------------
+
+@app.route("/")
+def dashboard():
+ with get_db() as conn:
+ recent = conn.execute("SELECT * FROM runs ORDER BY started_at DESC LIMIT 5").fetchall()
+ total = conn.execute("SELECT COUNT(*) FROM runs").fetchone()[0]
+ last_ok = conn.execute("SELECT * FROM runs WHERE status='success' ORDER BY started_at DESC LIMIT 1").fetchone()
+ last_full = conn.execute("SELECT * FROM runs WHERE status='success' AND backup_type='full' ORDER BY started_at DESC LIMIT 1").fetchone()
+ last_failed = conn.execute("SELECT * FROM runs WHERE status='failed' ORDER BY started_at DESC LIMIT 1").fetchone()
+
+ with _active_lock:
+ running_id = next((rid for rid, i in _active.items() if i["status"] == "running"), None)
+
+ job_incr = scheduler.get_job("incr_backup")
+ job_full = scheduler.get_job("full_backup")
+ next_incr = job_incr.next_run_time.strftime("%Y-%m-%d %H:%M UTC") if job_incr and job_incr.next_run_time else None
+ next_full_run = job_full.next_run_time.strftime("%Y-%m-%d %H:%M UTC") if job_full and job_full.next_run_time else None
+
+ return render_template("dashboard.html", recent=recent, total=total,
+ last_ok=last_ok, last_full=last_full, last_failed=last_failed,
+ running_id=running_id, next_incr=next_incr, next_full_run=next_full_run)
+
+
+@app.route("/history")
+def history():
+ page = max(1, request.args.get("page", 1, type=int))
+ per_page = 25
+ offset = (page - 1) * per_page
+ with get_db() as conn:
+ runs = conn.execute("SELECT * FROM runs ORDER BY started_at DESC LIMIT ? OFFSET ?",
+ (per_page, offset)).fetchall()
+ total = conn.execute("SELECT COUNT(*) FROM runs").fetchone()[0]
+ pages = max(1, (total + per_page - 1) // per_page)
+ return render_template("history.html", runs=runs, page=page, pages=pages, total=total)
+
+
+@app.route("/restore")
+def restore_list():
+ with get_db() as conn:
+ restores = conn.execute("SELECT * FROM restores ORDER BY started_at DESC").fetchall()
+
+ remote_backups = []
+ try:
+ remote_backups = list_remote_backups(_config_for_backup())
+ with get_db() as conn:
+ for b in remote_backups:
+ run = conn.execute(
+ "SELECT backup_type, parent_full_s3_key, started_at FROM runs WHERE s3_key=? AND status='success'",
+ (b["key"],),
+ ).fetchone()
+ if run:
+ b["backup_type"] = run["backup_type"]
+ parent = run["parent_full_s3_key"]
+ if run["backup_type"] == "incremental" and parent:
+ full_run = conn.execute(
+ "SELECT started_at FROM runs WHERE s3_key=? AND status='success'",
+ (parent,),
+ ).fetchone()
+ if full_run:
+ cnt = conn.execute(
+ """SELECT COUNT(*) FROM runs
+ WHERE backup_type='incremental' AND parent_full_s3_key=?
+ AND started_at>=? AND started_at<=? AND status='success'""",
+ (parent, full_run["started_at"], run["started_at"]),
+ ).fetchone()[0]
+ b["chain_available"] = True
+ b["chain_length"] = cnt + 1
+ else:
+ b["chain_available"] = False
+ else:
+ b["chain_available"] = False
+ except Exception as exc:
+ app.logger.warning(f"Could not list remote backups: {exc}")
+
+ return render_template("restore_list.html", restores=restores, remote_backups=remote_backups)
+
+
+@app.route("/restore/")
+def restore_view(restore_id: int):
+ with get_db() as conn:
+ restore = conn.execute("SELECT * FROM restores WHERE id=?", (restore_id,)).fetchone()
+ if not restore:
+ abort(404)
+
+ by_company: dict[str, list] = {}
+ restore_info: dict = {}
+ SKIP = {"manifest.json", "chain_manifest.json", "_company.json"}
+
+ if restore["status"] == "success" and restore["output_dir"]:
+ output_dir = Path(restore["output_dir"])
+
+ # Read manifest
+ chain_manifest = output_dir / "chain" / "chain_manifest.json"
+ if chain_manifest.exists():
+ try:
+ restore_info = json.loads(chain_manifest.read_text())
+ restore_info["restore_mode"] = "chain"
+ except Exception:
+ pass
+ else:
+ for mf in output_dir.rglob("manifest.json"):
+ try:
+ restore_info = json.loads(mf.read_text())
+ restore_info["restore_mode"] = restore_info.get("backup_type", "full")
+ except Exception:
+ pass
+ break
+
+ for f in sorted(list(output_dir.rglob("*.json")) + list(output_dir.rglob("*.jsonl"))):
+ if f.name in SKIP:
+ continue
+ rel = f.relative_to(output_dir)
+ parts = rel.parts
+ # chain: chain/{company}/{entity}.json → parts[1]=company
+ # regular: {archive_name}/{company}/{entity}.json → parts[1]=company
+ company = parts[1] if len(parts) >= 3 else "root"
+ by_company.setdefault(company, []).append({
+ "entity": f.stem,
+ "ext": f.suffix,
+ "path": str(rel),
+ "size_kb": round(f.stat().st_size / 1024, 1),
+ })
+
+ with _restore_lock:
+ streaming_id = (restore_id
+ if restore_id in _active_restores
+ and _active_restores[restore_id]["status"] == "running"
+ else None)
+
+ return render_template("restore_view.html", restore=restore, restore_info=restore_info,
+ by_company=by_company, streaming_id=streaming_id)
+
+
+@app.route("/settings", methods=["GET", "POST"])
+def settings():
+ if request.method == "POST":
+ data = dict(request.form)
+ for checkbox in ("schedule_incr_enabled", "schedule_full_enabled"):
+ data[checkbox] = "true" if checkbox in request.form else "false"
+ cfg_set(data)
+ _apply_schedule()
+ flash("Settings saved.", "success")
+ return redirect(url_for("settings"))
+ return render_template("settings.html", cfg=cfg_all())
+
+
+# ---------------------------------------------------------------------------
+# Routes — API (backup)
+# ---------------------------------------------------------------------------
+
+@app.route("/api/run", methods=["POST"])
+def api_run():
+ data = request.get_json(silent=True) or {}
+ force_full = bool(data.get("force_full", True))
+ try:
+ run_id = _launch_backup(triggered_by="manual", force_full=force_full)
+ return jsonify({"run_id": run_id, "backup_type": "full" if force_full else "incremental"})
+ except RuntimeError as exc:
+ return jsonify({"error": str(exc)}), 409
+ except Exception as exc:
+ return jsonify({"error": str(exc)}), 500
+
+
+@app.route("/api/run//stream")
+def api_stream(run_id: int):
+ return Response(
+ _sse_stream(run_id, _active, _active_lock),
+ mimetype="text/event-stream",
+ headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
+ )
+
+
+@app.route("/api/status")
+def api_status():
+ with _active_lock:
+ running_id = next((rid for rid, i in _active.items() if i["status"] == "running"), None)
+ return jsonify({"running": running_id is not None, "run_id": running_id})
+
+
+# ---------------------------------------------------------------------------
+# Routes — API (restore)
+# ---------------------------------------------------------------------------
+
+@app.route("/api/restore/chain-info")
+def api_restore_chain_info():
+ s3_key = request.args.get("s3_key", "").strip()
+ if not s3_key:
+ return jsonify({"error": "s3_key required"}), 400
+ chain_keys = _get_restore_chain(s3_key)
+ return jsonify({
+ "chain_available": chain_keys is not None,
+ "chain_length": len(chain_keys) if chain_keys else 1,
+ "chain_keys": chain_keys or [],
+ })
+
+
+@app.route("/api/restore", methods=["POST"])
+def api_restore():
+ data = request.get_json(silent=True) or {}
+ s3_key = data.get("s3_key", "").strip()
+ if not s3_key:
+ return jsonify({"error": "s3_key is required"}), 400
+ entity_filter = data.get("entities", "").strip() or None
+ chain = bool(data.get("chain", False))
+ try:
+ restore_id = _launch_restore(s3_key, entity_filter=entity_filter, chain=chain)
+ return jsonify({"restore_id": restore_id})
+ except Exception as exc:
+ return jsonify({"error": str(exc)}), 500
+
+
+@app.route("/api/restore//stream")
+def api_restore_stream(restore_id: int):
+ return Response(
+ _sse_stream(restore_id, _active_restores, _restore_lock),
+ mimetype="text/event-stream",
+ headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
+ )
+
+
+@app.route("/restore//download")
+def restore_download_file(restore_id: int):
+ with get_db() as conn:
+ restore = conn.execute("SELECT * FROM restores WHERE id=?", (restore_id,)).fetchone()
+ if not restore or not restore["output_dir"]:
+ abort(404)
+ rel_path = request.args.get("path", "")
+ if not rel_path:
+ abort(400)
+ output_dir = Path(restore["output_dir"]).resolve()
+ target = (output_dir / rel_path).resolve()
+ if not str(target).startswith(str(output_dir)):
+ abort(403)
+ if not target.exists():
+ abort(404)
+ return send_file(target, as_attachment=True, download_name=target.name)
+
+
+@app.route("/restore//download-zip")
+def restore_download_zip(restore_id: int):
+ with get_db() as conn:
+ restore = conn.execute("SELECT * FROM restores WHERE id=?", (restore_id,)).fetchone()
+ if not restore or restore["status"] != "success" or not restore["output_dir"]:
+ abort(404)
+ output_dir = Path(restore["output_dir"])
+ archive_name = Path(restore["s3_key"]).name.replace(".tar.gz.gpg", "") + "_restored.zip"
+ buf = io.BytesIO()
+ with zipfile.ZipFile(buf, "w", zipfile.ZIP_DEFLATED, compresslevel=6) as zf:
+ for f in sorted(output_dir.rglob("*.json")):
+ zf.write(f, arcname=str(f.relative_to(output_dir)))
+ for f in sorted(output_dir.rglob("*.jsonl")):
+ zf.write(f, arcname=str(f.relative_to(output_dir)))
+ buf.seek(0)
+ return send_file(buf, mimetype="application/zip",
+ as_attachment=True, download_name=archive_name)
+
+
+@app.route("/api/restore/", methods=["DELETE"])
+def api_restore_delete(restore_id: int):
+ with get_db() as conn:
+ restore = conn.execute("SELECT * FROM restores WHERE id=?", (restore_id,)).fetchone()
+ if not restore:
+ abort(404)
+ with _restore_lock:
+ if restore_id in _active_restores and _active_restores[restore_id]["status"] == "running":
+ return jsonify({"error": "Restore is still running"}), 409
+ if restore["output_dir"]:
+ out = Path(restore["output_dir"])
+ if out.exists():
+ shutil.rmtree(out, ignore_errors=True)
+ with get_db() as conn:
+ conn.execute("DELETE FROM restores WHERE id=?", (restore_id,))
+ return jsonify({"ok": True})
+
+
+# ---------------------------------------------------------------------------
+# Bootstrap
+# ---------------------------------------------------------------------------
+
+if __name__ == "__main__":
+ init_db()
+ _apply_schedule()
+ scheduler.start()
+
+ port = int(os.environ.get("PORT", 1890))
+ app.logger.info(f"BackupBC web UI starting on port {port}")
+ app.run(host="0.0.0.0", port=port, debug=False, threaded=True)