442 lines
15 KiB
Python
442 lines
15 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Magic: The Gathering Rules Updater
|
|
|
|
This module handles:
|
|
1. Downloading/updating the rules repository from GitHub
|
|
2. Scanning and validating the rules format
|
|
3. Updating the hardcoded keyword database
|
|
4. Scheduling weekly checks
|
|
|
|
Usage:
|
|
from mtg_rules_engine.updater import RulesUpdater
|
|
|
|
updater = RulesUpdater()
|
|
|
|
# Check for updates
|
|
updates = updater.check_for_updates()
|
|
if updates['has_updates']:
|
|
updater.apply_updates()
|
|
|
|
# Run on startup
|
|
updater.initialize()
|
|
"""
|
|
|
|
import os
|
|
import re
|
|
import json
|
|
import subprocess
|
|
import logging
|
|
from datetime import datetime, timedelta
|
|
from typing import Dict, List, Optional, Tuple, Any
|
|
from pathlib import Path
|
|
|
|
# Configure logging
|
|
logging.basicConfig(level=logging.INFO)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class RulesUpdater:
|
|
"""
|
|
Handles downloading, validating, and updating the MTG rules database.
|
|
|
|
This updater:
|
|
- Clones/pulls the rules repository from GitHub
|
|
- Validates the rules format and completeness
|
|
- Updates the hardcoded keyword database
|
|
- Tracks the current rules version
|
|
"""
|
|
|
|
# GitHub repository URL
|
|
REPO_URL = "https://github.com/chaoticgoodcomputing/mtg-rules.git"
|
|
|
|
# Local rules directory
|
|
RULES_DIR = Path("/home/user/wall-o/mtg-rules")
|
|
|
|
# Engine directory
|
|
ENGINE_DIR = Path("/home/user/wall-o/mtg_rules_engine")
|
|
|
|
# Keywords file
|
|
KEYWORDS_FILE = ENGINE_DIR / "keywords.py"
|
|
|
|
# State file for tracking updates
|
|
STATE_FILE = ENGINE_DIR / "updater_state.json"
|
|
|
|
def __init__(self):
|
|
"""Initialize the rules updater."""
|
|
self._state = self._load_state()
|
|
self._last_check = None
|
|
self._last_update = None
|
|
|
|
def _load_state(self) -> Dict[str, Any]:
|
|
"""Load the updater state from disk."""
|
|
if self.STATE_FILE.exists():
|
|
with open(self.STATE_FILE, 'r') as f:
|
|
return json.load(f)
|
|
return {
|
|
"last_check": None,
|
|
"last_update": None,
|
|
"current_version": None,
|
|
"rules_count": 0,
|
|
"last_scan_status": None,
|
|
}
|
|
|
|
def _save_state(self):
|
|
"""Save the updater state to disk."""
|
|
self.STATE_FILE.parent.mkdir(parents=True, exist_ok=True)
|
|
with open(self.STATE_FILE, 'w') as f:
|
|
json.dump(self._state, f, indent=2)
|
|
|
|
def initialize(self):
|
|
"""
|
|
Initialize the updater: download rules and run initial scan.
|
|
|
|
This should be called on engine startup.
|
|
"""
|
|
logger.info("Initializing rules updater...")
|
|
|
|
# Step 1: Ensure rules directory exists
|
|
if not self.RULES_DIR.exists():
|
|
logger.info("Cloning rules repository...")
|
|
self._clone_repo()
|
|
else:
|
|
logger.info("Pulling latest rules...")
|
|
self._pull_repo()
|
|
|
|
# Step 2: Scan and validate rules
|
|
logger.info("Scanning rules...")
|
|
scan_result = self.scan_rules()
|
|
|
|
if scan_result['status'] == 'error':
|
|
logger.error(f"Rules scan failed: {scan_result['message']}")
|
|
raise RuntimeError(f"Rules scan failed: {scan_result['message']}")
|
|
|
|
# Step 3: Update keyword database if needed
|
|
if scan_result['needs_update']:
|
|
logger.info("Updating keyword database...")
|
|
self._update_keywords(scan_result)
|
|
|
|
# Step 4: Update state
|
|
self._state['last_check'] = datetime.now().isoformat()
|
|
self._state['last_update'] = datetime.now().isoformat()
|
|
self._state['current_version'] = scan_result['version']
|
|
self._state['rules_count'] = scan_result['rules_count']
|
|
self._state['last_scan_status'] = 'success'
|
|
self._save_state()
|
|
|
|
logger.info(f"Initialization complete. Version: {scan_result['version']}")
|
|
|
|
def check_for_updates(self) -> Dict[str, Any]:
|
|
"""
|
|
Check if there are updates available from the rules repository.
|
|
|
|
Returns:
|
|
Dictionary with update information:
|
|
- has_updates: bool
|
|
- current_version: str
|
|
- latest_version: str
|
|
- changes: list of change descriptions
|
|
"""
|
|
logger.info("Checking for rules updates...")
|
|
|
|
# Get current version
|
|
current_version = self._get_current_version()
|
|
|
|
# Get latest version from remote
|
|
latest_version = self._get_latest_version()
|
|
|
|
has_updates = current_version != latest_version
|
|
|
|
return {
|
|
'has_updates': has_updates,
|
|
'current_version': current_version,
|
|
'latest_version': latest_version,
|
|
'changes': [] if not has_updates else ['New rules version available'],
|
|
}
|
|
|
|
def apply_updates(self):
|
|
"""Apply any available updates to the rules repository."""
|
|
logger.info("Applying rules updates...")
|
|
|
|
# Pull latest changes
|
|
self._pull_repo()
|
|
|
|
# Scan and validate
|
|
scan_result = self.scan_rules()
|
|
|
|
if scan_result['status'] == 'error':
|
|
logger.error(f"Rules scan failed after update: {scan_result['message']}")
|
|
raise RuntimeError(f"Rules scan failed: {scan_result['message']}")
|
|
|
|
# Update keyword database
|
|
if scan_result['needs_update']:
|
|
self._update_keywords(scan_result)
|
|
|
|
# Update state
|
|
self._state['last_update'] = datetime.now().isoformat()
|
|
self._state['current_version'] = scan_result['version']
|
|
self._state['rules_count'] = scan_result['rules_count']
|
|
self._save_state()
|
|
|
|
logger.info(f"Updates applied. New version: {scan_result['version']}")
|
|
|
|
def scan_rules(self) -> Dict[str, Any]:
|
|
"""
|
|
Scan the rules repository and validate format.
|
|
|
|
Returns:
|
|
Dictionary with scan results:
|
|
- status: 'success' or 'error'
|
|
- message: description of result
|
|
- version: current rules version
|
|
- rules_count: number of rules found
|
|
- needs_update: whether keyword database needs updating
|
|
- errors: list of any errors found
|
|
"""
|
|
errors = []
|
|
rules_count = 0
|
|
version = None
|
|
|
|
# Check if rules directory exists
|
|
if not self.RULES_DIR.exists():
|
|
return {
|
|
'status': 'error',
|
|
'message': 'Rules directory not found',
|
|
'version': None,
|
|
'rules_count': 0,
|
|
'needs_update': False,
|
|
'errors': ['Rules directory not found'],
|
|
}
|
|
|
|
# Get version from VERSION file
|
|
version_file = self.RULES_DIR / "VERSION"
|
|
if version_file.exists():
|
|
with open(version_file, 'r') as f:
|
|
version = f.read().strip()
|
|
else:
|
|
errors.append("VERSION file not found")
|
|
|
|
# Scan all markdown files in rules directory
|
|
rules_dir = self.RULES_DIR / "rules"
|
|
if rules_dir.exists():
|
|
for md_file in rules_dir.rglob("*.md"):
|
|
rules_count += 1
|
|
|
|
# Validate file format
|
|
file_errors = self._validate_rule_file(md_file)
|
|
errors.extend(file_errors)
|
|
|
|
# Check for required files
|
|
required_files = ["INTRO.md", "TABLE_OF_CONTENTS.md", "GLOSSARY.md", "CREDITS.md"]
|
|
for req_file in required_files:
|
|
if not (rules_dir / req_file).exists():
|
|
errors.append(f"Required file missing: {req_file}")
|
|
|
|
# Check for rules subdirectories
|
|
rules_subdirs = rules_dir / "rules"
|
|
if rules_subdirs.exists():
|
|
for subdir in rules_subdirs.iterdir():
|
|
if subdir.is_dir():
|
|
# Check that subdirectory has markdown files
|
|
md_files = list(subdir.glob("*.md"))
|
|
if not md_files:
|
|
errors.append(f"Rules subdirectory has no markdown files: {subdir.name}")
|
|
|
|
needs_update = len(errors) > 0 or rules_count != self._state.get('rules_count', 0)
|
|
|
|
return {
|
|
'status': 'error' if errors else 'success',
|
|
'message': '; '.join(errors) if errors else 'All rules validated successfully',
|
|
'version': version,
|
|
'rules_count': rules_count,
|
|
'needs_update': needs_update,
|
|
'errors': errors,
|
|
}
|
|
|
|
def _validate_rule_file(self, file_path: Path) -> List[str]:
|
|
"""
|
|
Validate a single rule file's format.
|
|
|
|
Args:
|
|
file_path: Path to the markdown file
|
|
|
|
Returns:
|
|
List of validation errors (empty if valid)
|
|
"""
|
|
errors = []
|
|
|
|
try:
|
|
with open(file_path, 'r', encoding='utf-8') as f:
|
|
content = f.read()
|
|
except Exception as e:
|
|
return [f"Cannot read file {file_path}: {str(e)}"]
|
|
|
|
# Check for empty files
|
|
if not content.strip():
|
|
errors.append(f"Empty file: {file_path.relative_to(self.RULES_DIR)}")
|
|
return errors
|
|
|
|
# Check for rule number format in filename (e.g., "100-general.md")
|
|
if file_path.parent.name == "rules":
|
|
# This is a main rule section
|
|
filename = file_path.stem
|
|
if not re.match(r'^\d+', filename):
|
|
errors.append(f"Rule file doesn't start with number: {file_path.relative_to(self.RULES_DIR)}")
|
|
|
|
return errors
|
|
|
|
def _update_keywords(self, scan_result: Dict[str, Any]):
|
|
"""
|
|
Update the hardcoded keyword database based on scanned rules.
|
|
|
|
Args:
|
|
scan_result: Results from scan_rules()
|
|
"""
|
|
logger.info("Extracting keywords from rules...")
|
|
|
|
# This would extract keywords from the rules markdown files
|
|
# and update the keywords.py file
|
|
# For now, we'll just log that an update is needed
|
|
|
|
logger.info("Keyword database update would be performed here")
|
|
logger.info(f"Rules version: {scan_result['version']}")
|
|
logger.info(f"Rules count: {scan_result['rules_count']}")
|
|
|
|
def _clone_repo(self):
|
|
"""Clone the rules repository."""
|
|
try:
|
|
subprocess.run(
|
|
["git", "clone", self.REPO_URL, str(self.RULES_DIR)],
|
|
check=True,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
logger.info("Repository cloned successfully")
|
|
except subprocess.CalledProcessError as e:
|
|
logger.error(f"Failed to clone repository: {e.stderr}")
|
|
raise
|
|
|
|
def _pull_repo(self):
|
|
"""Pull latest changes from the repository."""
|
|
try:
|
|
subprocess.run(
|
|
["git", "-C", str(self.RULES_DIR), "pull"],
|
|
check=True,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
logger.info("Repository pulled successfully")
|
|
except subprocess.CalledProcessError as e:
|
|
logger.error(f"Failed to pull repository: {e.stderr}")
|
|
raise
|
|
|
|
def _get_current_version(self) -> str:
|
|
"""Get the current rules version."""
|
|
version_file = self.RULES_DIR / "VERSION"
|
|
if version_file.exists():
|
|
with open(version_file, 'r') as f:
|
|
return f.read().strip()
|
|
return "unknown"
|
|
|
|
def _get_latest_version(self) -> str:
|
|
"""Get the latest rules version from the remote repository."""
|
|
try:
|
|
result = subprocess.run(
|
|
["git", "-C", str(self.RULES_DIR), "ls-remote", "origin", "HEAD"],
|
|
check=True,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
# Parse the output to get the latest commit hash
|
|
lines = result.stdout.strip().split('\n')
|
|
if lines:
|
|
return lines[0].split()[0]
|
|
except subprocess.CalledProcessError:
|
|
pass
|
|
|
|
return self._get_current_version()
|
|
|
|
def schedule_weekly_check(self):
|
|
"""
|
|
Schedule a weekly check for rules updates.
|
|
|
|
This creates a cron job that runs every Monday at 9 AM.
|
|
"""
|
|
cron_expression = "0 9 * * 1" # Every Monday at 9 AM
|
|
|
|
# Create a script that runs the updater
|
|
script_path = self.ENGINE_DIR / "weekly_update.sh"
|
|
script_content = f"""#!/bin/bash
|
|
cd {self.ENGINE_DIR.parent}
|
|
python -m mtg_rules_engine.updater --weekly
|
|
"""
|
|
|
|
with open(script_path, 'w') as f:
|
|
f.write(script_content)
|
|
|
|
os.chmod(script_path, 0o755)
|
|
|
|
# Add to crontab
|
|
cron_job = f"{cron_expression} {script_path}\n"
|
|
existing_cron = subprocess.run(
|
|
["crontab", "-l"],
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
|
|
if existing_cron.returncode == 0:
|
|
new_cron = existing_cron.stdout + cron_job
|
|
else:
|
|
new_cron = cron_job
|
|
|
|
subprocess.run(
|
|
["crontab", "-"],
|
|
input=new_cron,
|
|
text=True,
|
|
)
|
|
|
|
logger.info(f"Weekly update scheduled at {cron_expression}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
import argparse
|
|
|
|
parser = argparse.ArgumentParser(description="MTG Rules Updater")
|
|
parser.add_argument("--check", action="store_true", help="Check for updates")
|
|
parser.add_argument("--apply", action="store_true", help="Apply updates")
|
|
parser.add_argument("--scan", action="store_true", help="Scan rules")
|
|
parser.add_argument("--weekly", action="store_true", help="Run weekly check")
|
|
|
|
args = parser.parse_args()
|
|
|
|
updater = RulesUpdater()
|
|
|
|
if args.check:
|
|
updates = updater.check_for_updates()
|
|
print(f"Has updates: {updates['has_updates']}")
|
|
print(f"Current version: {updates['current_version']}")
|
|
print(f"Latest version: {updates['latest_version']}")
|
|
|
|
elif args.apply:
|
|
updater.apply_updates()
|
|
|
|
elif args.scan:
|
|
result = updater.scan_rules()
|
|
print(f"Status: {result['status']}")
|
|
print(f"Message: {result['message']}")
|
|
print(f"Version: {result['version']}")
|
|
print(f"Rules count: {result['rules_count']}")
|
|
if result['errors']:
|
|
print(f"Errors: {result['errors']}")
|
|
|
|
elif args.weekly:
|
|
updates = updater.check_for_updates()
|
|
if updates['has_updates']:
|
|
updater.apply_updates()
|
|
else:
|
|
print("No updates available")
|
|
|
|
else:
|
|
# Default: initialize
|
|
updater.initialize()
|