# config/verification.yaml
backup_sources:
  - name: "mysql-primary"
    type: "database"
    source:
      engine: "mysql"
      host: "db-primary"
      port: 3306
      database: "app_production"
    backup_path: "/backups/mysql/latest.sql.gz"
    verification:
      check_integrity: true
      check_constraints: true
      check_row_count: true
      restore_to: "sandbox-mysql"

  - name: "file-storage"
    type: "filesystem"
    source:
      paths:
        - "/var/www/html/uploads"
        - "/etc/nginx/sites-enabled"
    backup_path: "/backups/files/latest.tar.gz"
    verification:
      check_integrity: true
      check_permissions: true
      restore_to: "sandbox-files"

  - name: "docker-volumes"
    type: "volume"
    source:
      volumes:
        - "postgres_data"
        - "redis_data"
    backup_path: "/backups/volumes/latest.tar.gz"
    verification:
      check_integrity: true
      restore_to: "sandbox-docker"

ai_analysis:
  model: "qwen2.5:7b"
  endpoint: "http://ollama:11434"
  prompt_template: "configs/prompts/verification_analysis.txt"

notifications:
  slack_webhook: "https://hooks.slack.com/..."
  pagerduty_key: "..."
  email: ["ops@yourdomain.com"]
  fail_threshold: 2  # Alert after N consecutive failures

scheduling:
  full_verification: "0 3 * * 0"      # Full verification every Sunday 3 AM
  incremental_check: "0 3 * * 1-6"    # Light check Mon-Sat 3 AM

# docker-compose.yml
version: "3.8"

services:
  verifier:
    build: ./
    container_name: backup-verifier
    volumes:
      - ./config:/app/config:ro
      - ./reports:/app/reports
      - ./logs:/app/logs
      - /backups:/backups:ro
      - /var/run/docker.sock:/var/run/docker.sock
    environment:
      - OLLAMA_HOST=http://ollama:11434
      - SLACK_WEBHOOK=${SLACK_WEBHOOK}
    depends_on:
      - ollama
    networks:
      - verifier-net

  sandbox-mysql:
    image: mysql:8.0
    container_name: sandbox-mysql
    environment:
      MYSQL_ROOT_PASSWORD: sandbox_root_123
    volumes:
      - sandbox_mysql_data:/var/lib/mysql
    networks:
      - verifier-net
    tmpfs:
      - /var/lib/mysql

  sandbox-redis:
    image: redis:7-alpine
    container_name: sandbox-redis
    networks:
      - verifier-net
    tmpfs:
      - /data

  ollama:
    image: ollama/ollama:latest
    container_name: ollama-verifier
    volumes:
      - ollama_model:/root/.ollama
    networks:
      - verifier-net
    ports:
      - "11434:11434"

volumes:
  sandbox_mysql_data:
  ollama_model:

networks:
  verifier-net:
    driver: bridge

# scripts/extract_backup.py
#!/usr/bin/env python3
"""Extract backup files from storage for verification"""

import gzip
import hashlib
import json
import shutil
import tarfile
from datetime import datetime
from pathlib import Path

BACKUP_DIR = Path("/backups")
METADATA_FILE = BACKUP_DIR / "metadata.json"


def calculate_checksum(filepath: Path) -> dict:
    """Calculate file checksums"""
    hashes = {
        "md5": hashlib.md5(),
        "sha256": hashlib.sha256(),
    }
    size = 0
    with open(filepath, "rb") as f:
        while chunk := f.read(8192):
            for h in hashes.values():
                h.update(chunk)
            size += len(chunk)
    return {
        "size": size,
        "md5": hashes["md5"].hexdigest(),
        "sha256": hashes["sha256"].hexdigest(),
    }


def extract_mysql_backup(backup_path: Path, dest_dir: Path) -> dict:
    """Extract MySQL backup and validate structure"""
    result = {
        "type": "mysql",
        "status": "unknown",
        "tables": [],
        "row_counts": {},
        "errors": [],
    }

    dest_dir.mkdir(parents=True, exist_ok=True)

    # Decompress backup
    if backup_path.suffix == ".gz":
        with gzip.open(backup_path, "rb") as f_in:
            with open(dest_dir / "backup.sql", "wb") as f_out:
                shutil.copyfileobj(f_in, f_out)
        sql_file = dest_dir / "backup.sql"
    else:
        sql_file = backup_path

    # Parse SQL file for table info
    result["checksum"] = calculate_checksum(sql_file)
    result["file_size"] = sql_file.stat().st_size

    content = sql_file.read_text(errors="replace")

    # Check for CREATE TABLE statements
    create_tables = [
        line for line in content.split("\n")
        if line.strip().upper().startswith("CREATE TABLE")
    ]
    result["tables_found"] = len(create_tables)

    # Check INSERT statements
    insert_count = content.count("INSERT INTO")
    result["insert_statements"] = insert_count

    # Check for syntax error markers
    if "ERROR" in content.upper():
        result["errors"].append("SQL file contains ERROR markers")

    # Try connecting to sandbox to validate importability
    try:
        import subprocess
        proc = subprocess.run(
            ["mysql", "-h", "sandbox-mysql", "-u", "root",
             "-psandbox_root_123", "--one-database", "test_verify",
             "<", str(sql_file)],
            capture_output=True, text=True, timeout=300
        )
        if proc.returncode == 0:
            result["status"] = "integrity_ok"
        else:
            result["status"] = "integrity_failed"
            result["errors"].append(proc.stderr[:500])
    except Exception as e:
        result["status"] = "integrity_skipped"
        result["errors"].append(f"Connection check skipped: {e}")

    return result


def extract_file_backup(backup_path: Path, dest_dir: Path) -> dict:
    """Extract file backup and validate"""
    result = {
        "type": "filesystem",
        "status": "unknown",
        "files": [],
        "total_size": 0,
        "errors": [],
    }

    dest_dir.mkdir(parents=True, exist_ok=True)

    try:
        with tarfile.open(backup_path, "r:gz") as tar:
            # Security check: prevent path traversal
            for member in tar.getmembers():
                member_path = Path(member.name)
                if member_path.is_absolute() or ".." in member_path.parts:
                    result["errors"].append(
                        f"Security: suspicious path {member.name}"
                    )
                    continue
                tar.extract(member, dest_dir)

        # Statistics
        extracted_files = list(dest_dir.rglob("*"))
        result["files"] = [str(f.relative_to(dest_dir)) for f in extracted_files if f.is_file()]
        result["total_size"] = sum(f.stat().st_size for f in extracted_files if f.is_file())
        result["status"] = "integrity_ok"
        result["checksum"] = calculate_checksum(backup_path)
    except (tarfile.TarError, gzip.BadGzipFile) as e:
        result["status"] = "integrity_failed"
        result["errors"].append(f"Archive corruption: {e}")

    return result


def main():
    """Main entry point"""
    config_path = Path("/app/config/verification.yaml")
    output_dir = Path("/app/reports")
    output_dir.mkdir(parents=True, exist_ok=True)

    print(f"[{datetime.now()}] Starting backup extraction...")

    all_results = {}

    for source_name, source_config in config["backup_sources"].items():
        backup_path = Path(source_config["backup_path"])
        if not backup_path.exists():
            print(f"  SKIP {source_name}: backup not found at {backup_path}")
            continue

        print(f"  PROCESSING {source_name}...")
        dest_dir = Path(f"/tmp/verify-{source_name}-{datetime.now().strftime('%Y%m%d-%H%M%S')}")

        if source_config["type"] == "database":
            all_results[source_name] = extract_mysql_backup(backup_path, dest_dir)
        elif source_config["type"] == "filesystem":
            all_results[source_name] = extract_file_backup(backup_path, dest_dir)

        shutil.rmtree(dest_dir, ignore_errors=True)

    output_file = output_dir / f"extraction-{datetime.now().strftime('%Y%m%d-%H%M%S')}.json"
    with open(output_file, "w") as f:
        json.dump({
            "timestamp": datetime.now().isoformat(),
            "sources": all_results,
        }, f, indent=2, ensure_ascii=False)

    print(f"Extraction report saved to {output_file}")
    return all_results


if __name__ == "__main__":
    main()

# scripts/restore_sandbox.py
#!/usr/bin/env python3
"""Restore backups in isolated Docker containers"""

import asyncio
import json
from datetime import datetime
from pathlib import Path
import docker
import mysql.connector

client = docker.from_env()


async def restore_mysql_backup(backup_sql: Path, sandbox_name: str) -> dict:
    """Restore MySQL backup to isolated container"""
    result = {"status": "unknown", "details": []}

    try:
        try:
            container = client.containers.get(sandbox_name)
            if container.status != "running":
                container.start()
        except docker.errors.NotFound:
            container = client.containers.run(
                "mysql:8.0",
                name=sandbox_name,
                detach=True,
                environment={
                    "MYSQL_ROOT_PASSWORD": "sandbox_root_123",
                    "MYSQL_DATABASE": "verify_db",
                },
                networks=["verifier-net"],
                tmpfs={"/var/lib/mysql": "size=1g"},
            )
            for _ in range(30):
                await asyncio.sleep(1)
                if True:  # Simplified readiness check
                    break

        # Copy backup to container
        tar_bytes = backup_sql.read_bytes()
        import io
        with io.BytesIO() as tar_stream:
            import tarfile
            with tarfile.open(fileobj=tar_stream, mode="w") as tar:
                tar.add(backup_sql, arcname="backup.sql")
            tar_stream.seek(0)
            container.put_archive("/", tar_stream.read())

        # Execute restore
        proc = await container.exec_run(
            ["mysql", "-u", "root", "-psandbox_root_123",
             "verify_db", "<", "/backup.sql"],
            workdir="/tmp"
        )

        if proc.exit_code == 0:
            result["status"] = "restored"
            result["details"].append("MySQL restore successful")
        else:
            result["status"] = "restore_failed"
            result["details"].append(f"Restore error: {proc.output.decode()[:500]}")

        # Verify data
        conn = mysql.connector.connect(
            host="sandbox-mysql", user="root", password="sandbox_root_123",
            database="verify_db"
        )
        cursor = conn.cursor()
        cursor.execute("SHOW TABLES")
        tables = [t[0] for t in cursor.fetchall()]
        result["tables_restored"] = tables
        result["table_count"] = len(tables)

        for table in tables[:5]:
            try:
                cursor.execute(f"SELECT COUNT(*) FROM `{table}`")
                count = cursor.fetchone()[0]
                result[f"{table}_rows"] = count
            except Exception as e:
                result["details"].append(f"Check {table}: {e}")

        cursor.close()
        conn.close()

    except Exception as e:
        result["status"] = "error"
        result["details"].append(str(e))

    return result


async def main():
    """Main entry point"""
    report_dir = Path("/app/reports")
    timestamp = datetime.now().strftime("%Y%m%d-%H%M%S")

    print(f"[{timestamp}] Starting sandbox restoration...")

    all_results = {}

    for source_name, cfg in CONFIG["backup_sources"].items():
        print(f"  Restoring {source_name}...")
        backup_path = Path(cfg["backup_path"])

        if cfg["type"] == "database":
            all_results[source_name] = await restore_mysql_backup(
                backup_path, f"sandbox-{source_name}"
            )
        elif cfg["type"] == "filesystem":
            all_results[source_name] = await restore_file_backup(
                backup_path, f"sandbox-{source_name}"
            )

    output = report_dir / f"restoration-{timestamp}.json"
    with open(output, "w") as f:
        json.dump({"timestamp": timestamp, "results": all_results}, f, indent=2)

    print(f"Restoration report: {output}")
    return all_results


if __name__ == "__main__":
    asyncio.run(main())

# scripts/health_check.py
#!/usr/bin/env python3
"""Verify service health after recovery"""

import asyncio
import json
from datetime import datetime
from pathlib import Path
import httpx
import mysql.connector
import redis

REPORT_DIR = Path("/app/reports")


async def check_mysql_health(sandbox_name: str) -> dict:
    """Check MySQL recovery status"""
    result = {"service": "mysql", "checks": {}, "overall": "unknown"}

    try:
        conn = mysql.connector.connect(
            host="sandbox-mysql", user="root", password="sandbox_root_123",
            database="verify_db", connect_timeout=5,
        )
        cursor = conn.cursor()

        cursor.execute("SELECT VERSION()")
        result["checks"]["version"] = cursor.fetchone()[0]
        result["checks"]["connection"] = "ok"

        cursor.execute("SHOW TABLES")
        tables = cursor.fetchall()
        result["checks"]["table_count"] = len(tables)

        for table_name, in tables[:3]:
            try:
                cursor.execute(f"SELECT COUNT(*) FROM `{table_name}`")
                count = cursor.fetchone()[0]
                result["checks"][f"{table_name}_rows"] = count
                if count == 0:
                    result["checks"][f"{table_name}_warning"] = "empty table"
            except Exception as e:
                result["checks"][f"{table_name}_error"] = str(e)

        cursor.close()
        conn.close()
        result["overall"] = "healthy"

    except Exception as e:
        result["overall"] = "unhealthy"
        result["error"] = str(e)

    return result


async def check_file_integrity(sandbox_name: str) -> dict:
    """Check file recovery completeness"""
    result = {"service": "filesystem", "checks": {}, "overall": "unknown"}

    try:
        import docker
        client = docker.from_env()
        container = client.containers.get(sandbox_name)

        critical_files = [
            "/restore/etc/nginx/nginx.conf",
            "/restore/var/www/html/index.html",
        ]

        for fpath in critical_files:
            proc = await container.exec_run(["test", "-f", fpath])
            result["checks"][fpath] = "exists" if proc.exit_code == 0 else "missing"

        proc = await container.exec_run(["du", "-sh", "/restore"])
        result["checks"]["total_size"] = proc.output.decode().split()[0]

        result["overall"] = "healthy"
        container.stop()
        container.remove()

    except Exception as e:
        result["overall"] = "unhealthy"
        result["error"] = str(e)

    return result


async def main():
    """Main entry point"""
    timestamp = datetime.now().strftime("%Y%m%d-%H%M%S")
    print(f"[{timestamp}] Running health checks...")

    all_checks = {}
    all_checks["mysql"] = await check_mysql_health("sandbox-mysql")
    all_checks["filesystem"] = await check_file_integrity("sandbox-files")
    all_checks["connectivity"] = await check_service_connectivity()

    output = REPORT_DIR / f"healthcheck-{timestamp}.json"
    with open(output, "w") as f:
        json.dump({
            "timestamp": timestamp,
            "checks": all_checks,
            "summary": {
                "total": len(all_checks),
                "healthy": sum(1 for c in all_checks.values() if c["overall"] == "healthy"),
                "unhealthy": sum(1 for c in all_checks.values() if c["overall"] == "unhealthy"),
            }
        }, f, indent=2, ensure_ascii=False)

    print(f"Health check report: {output}")
    return all_checks


if __name__ == "__main__":
    asyncio.run(main())

# scripts/ai_analyze.py
#!/usr/bin/env python3
"""Use local LLM to analyze verification results and generate recommendations"""

import json
import httpx
from datetime import datetime
from pathlib import Path

OLLAMA_ENDPOINT = "http://ollama:11434/api/generate"
PROMPT_TEMPLATE = Path("/app/configs/prompts/verification_analysis.txt")


def load_prompt_template() -> str:
    return PROMPT_TEMPLATE.read_text()


def build_analysis_prompt(verification_data: dict) -> str:
    """Build analysis prompt"""
    template = load_prompt_template()
    summary = verification_data.get("summary", {})
    checks = verification_data.get("checks", {})

    details = []
    for service, result in checks.items():
        details.append(f"### {service}")
        details.append(f"Status: {result.get('overall', 'unknown')}")
        for check_name, check_result in result.get("checks", {}).items():
            details.append(f"- {check_name}: {check_result}")

    prompt = template.format(
        timestamp=datetime.now().isoformat(),
        summary=json.dumps(summary, ensure_ascii=False, indent=2),
        details="\n".join(details),
    )
    return prompt


async def analyze_with_llm(prompt: str, model: str = "qwen2.5:7b") -> dict:
    """Call LLM for analysis"""
    payload = {
        "model": model,
        "prompt": prompt,
        "stream": False,
        "options": {"temperature": 0.3, "num_predict": 2048},
    }

    async with httpx.AsyncClient(timeout=120) as client:
        resp = await client.post(OLLAMA_ENDPOINT, json=payload)
        resp.raise_for_status()
        return resp.json()


def parse_llm_response(response: dict) -> dict:
    """Parse LLM response into structured result"""
    text = response.get("response", "")

    result = {
        "risk_level": "low",
        "issues": [],
        "recommendations": [],
        "summary": text[:500],
        "full_analysis": text,
    }

    if "critical" in text.lower() or "严重" in text:
        result["risk_level"] = "critical"
    elif "high" in text.lower() or "高危" in text:
        result["risk_level"] = "high"
    elif "medium" in text.lower() or "中等" in text:
        result["risk_level"] = "medium"

    lines = text.split("\n")
    for line in lines:
        line = line.strip()
        if line.startswith("-") or line.startswith("•"):
            result["recommendations"].append(line[1:].strip())
        elif "issue" in line.lower() or "问题" in line:
            result["issues"].append(line)

    return result


async def main():
    """Main entry point"""
    report_dir = Path("/app/reports")
    timestamp = datetime.now().strftime("%Y%m%d-%H%M%S")

    health_files = sorted(report_dir.glob("healthcheck-*.json"))
    if not health_files:
        print("No health check reports found")
        return

    latest_report = json.loads(health_files[-1].read_text())
    print(f"Analyzing verification data from {latest_report.get('timestamp')}...")

    prompt = build_analysis_prompt(latest_report)
    print("Calling LLM for analysis...")
    llm_response = await analyze_with_llm(prompt)
    analysis = parse_llm_response(llm_response)

    output = report_dir / f"analysis-{timestamp}.json"
    with open(output, "w") as f:
        json.dump({
            "timestamp": timestamp,
            "model": "qwen2.5:7b",
            "risk_level": analysis["risk_level"],
            "issues": analysis["issues"],
            "recommendations": analysis["recommendations"],
            "llm_raw": llm_response.get("response", ""),
        }, f, indent=2, ensure_ascii=False)

    print(f"Analysis saved to {output}")
    print(f"Risk level: {analysis['risk_level']}")
    print(f"Issues found: {len(analysis['issues'])}")
    print(f"Recommendations: {len(analysis['recommendations'])}")


if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

# scripts/generate_report.py
#!/usr/bin/env python3
"""Generate human-readable verification reports"""

from datetime import datetime
from pathlib import Path
import json

REPORT_DIR = Path("/app/reports")


def load_reports(timestamp: str) -> dict:
    """Load all reports for this verification run"""
    reports = {}
    for pattern, key in [
        ("extraction-*.json", "extraction"),
        ("restoration-*.json", "restoration"),
        ("healthcheck-*.json", "health_check"),
        ("analysis-*.json", "analysis"),
    ]:
        files = sorted(REPORT_DIR.glob(pattern))
        if files:
            reports[key] = json.loads(files[-1].read_text())
    return reports


def generate_markdown_report(reports: dict) -> str:
    """Generate Markdown report"""
    timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    analysis = reports.get("analysis", {})
    risk_level = analysis.get("risk_level", "unknown")
    risk_colors = {"critical": "🔴", "high": "🟠", "medium": "🟡", "low": "🟢"}

    md = f"""# 📋 VPS Backup Verification Report

**Generated**: {timestamp}
**Risk Level**: {risk_colors.get(risk_level, "⚪")} {risk_level.upper()}

---

## 1. Summary

| Metric | Result |
|--------|--------|
| Backup Extraction | {'✅ Passed' if reports.get('extraction', {}).get('summary', {}).get('status') == 'integrity_ok' else '❌ Failed'} |
| Isolated Restore | {'✅ Passed' if reports.get('restoration', {}).get('summary', {}).get('status') == 'restored' else '❌ Failed'} |
| Health Check | {'✅ All healthy' if reports.get('health_check', {}).get('summary', {}).get('healthy', 0) > 0 else '⚠️ Partial issues'} |
| AI Analysis | {'✅ Low risk' if risk_level in ['low', 'medium'] else '⚠️ Attention needed'} |

---

## 2. Detailed Results

"""
    # Extraction phase
    if "extraction" in reports:
        md += "### 2.1 Backup Extraction\n\n"
        ext = reports["extraction"]
        for source, result in ext.get("sources", {}).items():
            icon = "✅" if result.get("status") == "integrity_ok" else "❌"
            md += f"- {icon} **{source}**: {result.get('status', 'unknown')}\n"
            if result.get("checksum"):
                md += f"  - Checksum: `{result['checksum'].get('sha256', 'N/A')[:16]}...`\n"
        md += "\n"

    # Restoration phase
    if "restoration" in reports:
        md += "### 2.2 Isolated Restoration\n\n"
        rest = reports["restoration"]
        for source, result in rest.get("results", {}).items():
            icon = "✅" if result.get("status") == "restored" else "❌"
            md += f"- {icon} **{source}**: {result.get('status', 'unknown')}\n"
            if result.get("tables_restored"):
                md += f"  - Tables: {', '.join(result['tables_restored'][:5])}\n"
        md += "\n"

    # Health check
    if "health_check" in reports:
        md += "### 2.3 Health Check\n\n"
        hc = reports["health_check"]
        summary = hc.get("summary", {})
        md += f"- Overall: {summary.get('healthy', 0)}/{summary.get('total', 0)} passed\n"
        for service, result in hc.get("checks", {}).items():
            icon = "✅" if result.get("overall") == "healthy" else "❌"
            md += f"- {icon} **{service}**: {result.get('overall', 'unknown')}\n"
        md += "\n"

    # AI analysis
    if "analysis" in reports:
        md += f"### 2.4 AI Analysis\n\n"
        md += f"**Risk Level**: {risk_colors.get(risk_level, '⚪')}{risk_level.upper()}\n\n"
        if analysis.get("issues"):
            md += "#### Issues Found\n\n"
            for issue in analysis["issues"][:5]:
                md += f"- {issue}\n"
            md += "\n"
        if analysis.get("recommendations"):
            md += "#### AI Recommendations\n\n"
            for rec in analysis["recommendations"][:5]:
                md += f"- {rec}\n"
            md += "\n"

    md += f"\n---\n\n*Report generated by AI Backup Verification System*\n"
    return md


def main():
    """Main entry point"""
    timestamp = datetime.now().strftime("%Y%m%d-%H%M%S")
    reports = load_reports(timestamp)
    md_content = generate_markdown_report(reports)

    md_path = REPORT_DIR / f"report-{timestamp}.md"
    md_path.write_text(md_content, encoding="utf-8")
    print(f"Report generated: {md_path}")
    print(f"Risk Level: {reports.get('analysis', {}).get('risk_level', 'unknown').upper()}")

    json_path = REPORT_DIR / f"report-{timestamp}.json"
    with open(json_path, "w") as f:
        json.dump({
            "timestamp": timestamp,
            "report_file": str(md_path),
            "risk_level": reports.get("analysis", {}).get("risk_level", "unknown"),
            "summary": reports.get("health_check", {}).get("summary", {}),
        }, f, indent=2, ensure_ascii=False)

    print(f"Summary saved: {json_path}")


if __name__ == "__main__":
    main()

# Clone project
git clone https://github.com/yourorg/vps-backup-verifier.git
cd vps-backup-verifier

# Configure environment
cp .env.example .env
# Edit .env with Slack Webhook etc.

# Pull LLM model
docker exec -it ollama-verifier ollama pull qwen2.5:7b

# Start
docker compose up -d

# First manual run
docker compose run --rm verifier python3 /app/scripts/generate_report.py

# Add to crontab
# Full verification every Sunday 3 AM
0 3 * * 0 /usr/bin/docker compose -f /opt/backup-verifier/docker-compose.yml run --rm verifier python3 /app/scripts/main.py --full

# Light check Mon-Sat 3 AM
0 3 * * 1-6 /usr/bin/docker compose -f /opt/backup-verifier/docker-compose.yml run --rm verifier python3 /app/scripts/main.py --light

# Manual trigger
curl -X POST http://localhost:8080/api/verify \
  -H "Content-Type: application/json" \
  -d '{"source": "mysql-primary", "full": true}'

# View history
curl http://localhost:8080/api/history?limit=10