feat(family): deploy Single-Writer DuckDB backend, Mila parent cockpit & feeding analytics, and Moritz macro tracker
This commit is contained in:
Moritz Graf
@@ -0,0 +1,326 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import datetime
|
||||
import logging
|
||||
import os
|
||||
import subprocess
|
||||
import threading
|
||||
import urllib.parse
|
||||
from pathlib import Path
|
||||
from typing import Callable, Optional
|
||||
|
||||
from app.config import (
|
||||
AUTO_GIT_SYNC,
|
||||
GIT_AUTHOR_EMAIL,
|
||||
GIT_AUTHOR_NAME,
|
||||
GIT_BRANCH,
|
||||
GIT_PASSWORD,
|
||||
GIT_REPO_URL,
|
||||
GIT_SYNC_INTERVAL_SECONDS,
|
||||
GIT_USERNAME,
|
||||
WIKI_DIR,
|
||||
)
|
||||
|
||||
logger = logging.getLogger("duckdb_backend.git_sync")
|
||||
|
||||
|
||||
class GitSyncManager:
|
||||
def __init__(self, wiki_dir: Optional[Path] = None, on_data_pulled: Optional[Callable[[], None]] = None):
|
||||
self.wiki_dir = wiki_dir or WIKI_DIR
|
||||
self.on_data_pulled = on_data_pulled
|
||||
self._lock = threading.Lock()
|
||||
self.status = "ok"
|
||||
self.is_conflict = False
|
||||
self.last_sync_time: Optional[str] = None
|
||||
self.last_commit_hash: Optional[str] = None
|
||||
self.last_error: Optional[str] = None
|
||||
self._bg_task: Optional[asyncio.Task] = None
|
||||
self._stop_event = asyncio.Event()
|
||||
|
||||
def _configure_git_client(self):
|
||||
"""Configures git client with safe.directory and author identity."""
|
||||
try:
|
||||
subprocess.run(["git", "config", "--global", "http.sslVerify", "false"], check=False)
|
||||
subprocess.run(["git", "config", "--global", "--add", "safe.directory", "*"], check=False)
|
||||
if (self.wiki_dir / ".git").is_dir():
|
||||
subprocess.run(["git", "-C", str(self.wiki_dir), "config", "user.name", GIT_AUTHOR_NAME], check=False)
|
||||
subprocess.run(["git", "-C", str(self.wiki_dir), "config", "user.email", GIT_AUTHOR_EMAIL], check=False)
|
||||
except Exception as e:
|
||||
logger.warning(f"Failed to configure git client options: {e}")
|
||||
|
||||
def get_auth_repo_url(self) -> str:
|
||||
if GIT_USERNAME and GIT_PASSWORD and "://" in GIT_REPO_URL:
|
||||
proto, rest = GIT_REPO_URL.split("://", 1)
|
||||
# Remove existing credentials in URL if any
|
||||
if "@" in rest:
|
||||
rest = rest.split("@", 1)[1]
|
||||
user_enc = urllib.parse.quote(GIT_USERNAME, safe="")
|
||||
pass_enc = urllib.parse.quote(GIT_PASSWORD, safe="")
|
||||
return f"{proto}://{user_enc}:{pass_enc}@{rest}"
|
||||
return GIT_REPO_URL
|
||||
|
||||
def ensure_repo(self) -> bool:
|
||||
"""Ensures repository exists and is initialized."""
|
||||
with self._lock:
|
||||
self._configure_git_client()
|
||||
if (self.wiki_dir / ".git").is_dir():
|
||||
try:
|
||||
res = subprocess.run(
|
||||
["git", "-C", str(self.wiki_dir), "rev-parse", "--short", "HEAD"],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False,
|
||||
)
|
||||
if res.returncode == 0:
|
||||
self.last_commit_hash = res.stdout.strip()
|
||||
return True
|
||||
except Exception as e:
|
||||
logger.warning(f"Failed to check git rev: {e}")
|
||||
return True
|
||||
|
||||
if not GIT_REPO_URL:
|
||||
logger.info(f"Wiki directory {self.wiki_dir} is not a git repo, but no GIT_REPO_URL configured.")
|
||||
return False
|
||||
|
||||
logger.info(f"Initializing repository {GIT_REPO_URL} in {self.wiki_dir}...")
|
||||
self.wiki_dir.mkdir(parents=True, exist_ok=True)
|
||||
auth_url = self.get_auth_repo_url()
|
||||
try:
|
||||
# Use git init + remote + fetch + checkout to safely support non-empty directories (e.g. lost+found on PVC)
|
||||
subprocess.run(["git", "-C", str(self.wiki_dir), "init"], capture_output=True, check=False)
|
||||
self._configure_git_client()
|
||||
subprocess.run(
|
||||
["git", "-C", str(self.wiki_dir), "remote", "add", "origin", auth_url],
|
||||
capture_output=True,
|
||||
check=False,
|
||||
)
|
||||
subprocess.run(
|
||||
["git", "-C", str(self.wiki_dir), "remote", "set-url", "origin", auth_url],
|
||||
capture_output=True,
|
||||
check=False,
|
||||
)
|
||||
res = subprocess.run(
|
||||
["git", "-C", str(self.wiki_dir), "fetch", "origin", GIT_BRANCH],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False,
|
||||
timeout=60,
|
||||
)
|
||||
if res.returncode == 0:
|
||||
if (self.wiki_dir / ".git" / "shallow").exists():
|
||||
subprocess.run(
|
||||
["git", "-C", str(self.wiki_dir), "fetch", "--unshallow", "origin", GIT_BRANCH],
|
||||
capture_output=True,
|
||||
check=False,
|
||||
)
|
||||
checkout_res = subprocess.run(
|
||||
["git", "-C", str(self.wiki_dir), "checkout", "-f", "-B", GIT_BRANCH, f"origin/{GIT_BRANCH}"],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False,
|
||||
)
|
||||
if checkout_res.returncode == 0:
|
||||
logger.info("Successfully initialized and checked out wiki repository.")
|
||||
self.status = "ok"
|
||||
self.is_conflict = False
|
||||
self.last_sync_time = datetime.datetime.now(datetime.timezone.utc).isoformat()
|
||||
return True
|
||||
else:
|
||||
self.status = "error"
|
||||
self.last_error = f"Checkout failed: {checkout_res.stderr.strip()}"
|
||||
logger.error(self.last_error)
|
||||
return False
|
||||
else:
|
||||
self.status = "error"
|
||||
self.last_error = f"Fetch failed: {res.stderr.strip()}"
|
||||
logger.error(self.last_error)
|
||||
return False
|
||||
except Exception as e:
|
||||
self.status = "error"
|
||||
self.last_error = f"Clone exception: {e}"
|
||||
logger.error(self.last_error)
|
||||
return False
|
||||
|
||||
def pull_rebase(self) -> bool:
|
||||
"""Pulls latest changes via rebase. Handles conflicts cleanly."""
|
||||
if not (self.wiki_dir / ".git").is_dir():
|
||||
return False
|
||||
|
||||
with self._lock:
|
||||
try:
|
||||
# Update remote url with credentials if provided
|
||||
auth_url = self.get_auth_repo_url()
|
||||
subprocess.run(
|
||||
["git", "-C", str(self.wiki_dir), "remote", "set-url", "origin", auth_url],
|
||||
capture_output=True,
|
||||
check=False,
|
||||
)
|
||||
|
||||
# Fetch and rebase with autostash
|
||||
res = subprocess.run(
|
||||
["git", "-C", str(self.wiki_dir), "pull", "--rebase", "--autostash", "origin", GIT_BRANCH],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False,
|
||||
timeout=30,
|
||||
)
|
||||
if res.returncode == 0:
|
||||
self.status = "ok"
|
||||
self.is_conflict = False
|
||||
self.last_error = None
|
||||
self.last_sync_time = datetime.datetime.now(datetime.timezone.utc).isoformat()
|
||||
# Update commit hash
|
||||
rev = subprocess.run(
|
||||
["git", "-C", str(self.wiki_dir), "rev-parse", "--short", "HEAD"],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False,
|
||||
)
|
||||
if rev.returncode == 0:
|
||||
self.last_commit_hash = rev.stdout.strip()
|
||||
|
||||
if self.on_data_pulled:
|
||||
try:
|
||||
self.on_data_pulled()
|
||||
except Exception as ex:
|
||||
logger.error(f"Error in on_data_pulled callback: {ex}")
|
||||
return True
|
||||
else:
|
||||
# Check if conflict occurred
|
||||
stderr = res.stderr or res.stdout
|
||||
logger.warning(f"Git pull --rebase failed: {stderr}")
|
||||
# Abort any in-progress rebase to keep working tree stable
|
||||
subprocess.run(
|
||||
["git", "-C", str(self.wiki_dir), "rebase", "--abort"],
|
||||
capture_output=True,
|
||||
check=False,
|
||||
)
|
||||
self.is_conflict = True
|
||||
self.status = "conflict"
|
||||
self.last_error = f"Rebase conflict: {stderr.strip()}"
|
||||
return False
|
||||
except Exception as e:
|
||||
self.status = "error"
|
||||
self.last_error = f"Pull error: {e}"
|
||||
logger.error(self.last_error)
|
||||
return False
|
||||
|
||||
def commit_and_push(self, commit_msg: str) -> bool:
|
||||
"""Stages data/, commits with descriptive message, rebases and pushes."""
|
||||
if not (self.wiki_dir / ".git").is_dir():
|
||||
return False
|
||||
|
||||
with self._lock:
|
||||
try:
|
||||
self._configure_git_client()
|
||||
data_rel = "data"
|
||||
subprocess.run(
|
||||
["git", "-C", str(self.wiki_dir), "add", data_rel],
|
||||
capture_output=True,
|
||||
check=False,
|
||||
)
|
||||
commit_res = subprocess.run(
|
||||
["git", "-C", str(self.wiki_dir), "commit", "-m", f"{commit_msg} [skip ci]"],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False,
|
||||
)
|
||||
if commit_res.returncode != 0:
|
||||
# Nothing to commit or already committed
|
||||
logger.debug(f"Commit returncode {commit_res.returncode}: {commit_res.stdout.strip()}")
|
||||
return True
|
||||
|
||||
# Pull rebase with autostash before push
|
||||
auth_url = self.get_auth_repo_url()
|
||||
subprocess.run(
|
||||
["git", "-C", str(self.wiki_dir), "remote", "set-url", "origin", auth_url],
|
||||
capture_output=True,
|
||||
check=False,
|
||||
)
|
||||
|
||||
rebase_res = subprocess.run(
|
||||
["git", "-C", str(self.wiki_dir), "pull", "--rebase", "--autostash", "origin", GIT_BRANCH],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False,
|
||||
timeout=30,
|
||||
)
|
||||
if rebase_res.returncode != 0:
|
||||
stderr = rebase_res.stderr or rebase_res.stdout
|
||||
logger.warning(f"Rebase failed during commit_and_push: {stderr}")
|
||||
subprocess.run(
|
||||
["git", "-C", str(self.wiki_dir), "rebase", "--abort"],
|
||||
capture_output=True,
|
||||
check=False,
|
||||
)
|
||||
self.is_conflict = True
|
||||
self.status = "conflict"
|
||||
self.last_error = f"Rebase conflict on push: {stderr.strip()}"
|
||||
return False
|
||||
|
||||
# Push to remote
|
||||
push_res = subprocess.run(
|
||||
["git", "-C", str(self.wiki_dir), "push", "origin", GIT_BRANCH],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False,
|
||||
timeout=30,
|
||||
)
|
||||
if push_res.returncode == 0:
|
||||
self.status = "ok"
|
||||
self.is_conflict = False
|
||||
self.last_error = None
|
||||
self.last_sync_time = datetime.datetime.now(datetime.timezone.utc).isoformat()
|
||||
rev = subprocess.run(
|
||||
["git", "-C", str(self.wiki_dir), "rev-parse", "--short", "HEAD"],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False,
|
||||
)
|
||||
if rev.returncode == 0:
|
||||
self.last_commit_hash = rev.stdout.strip()
|
||||
return True
|
||||
else:
|
||||
self.status = "error"
|
||||
self.last_error = f"Push error: {push_res.stderr.strip()}"
|
||||
logger.error(self.last_error)
|
||||
return False
|
||||
except Exception as e:
|
||||
self.status = "error"
|
||||
self.last_error = f"Commit/push error: {e}"
|
||||
logger.error(self.last_error)
|
||||
return False
|
||||
|
||||
async def start_periodic_sync(self):
|
||||
"""Starts background periodic git pull loop."""
|
||||
if not AUTO_GIT_SYNC:
|
||||
logger.info("Automatic git sync is disabled by configuration.")
|
||||
return
|
||||
|
||||
self._stop_event.clear()
|
||||
|
||||
async def _loop():
|
||||
logger.info(f"Starting periodic git sync every {GIT_SYNC_INTERVAL_SECONDS}s")
|
||||
while not self._stop_event.is_set():
|
||||
try:
|
||||
await asyncio.to_thread(self.pull_rebase)
|
||||
except Exception as ex:
|
||||
logger.error(f"Error during periodic git pull: {ex}")
|
||||
try:
|
||||
await asyncio.wait_for(self._stop_event.wait(), timeout=GIT_SYNC_INTERVAL_SECONDS)
|
||||
except asyncio.TimeoutError:
|
||||
pass
|
||||
|
||||
self._bg_task = asyncio.create_task(_loop())
|
||||
|
||||
async def stop_periodic_sync(self):
|
||||
"""Stops background sync loop."""
|
||||
self._stop_event.set()
|
||||
if self._bg_task:
|
||||
self._bg_task.cancel()
|
||||
try:
|
||||
await self._bg_task
|
||||
except asyncio.CancelledError:
|
||||
pass
|
||||
self._bg_task = None
|
||||
Reference in New Issue
Block a user