dev #18

Merged
sbcrumb merged 28 commits from dev into main 2025-09-27 17:52:46 -04:00
23 changed files with 5255 additions and 3 deletions
+161
View File
@@ -0,0 +1,161 @@
name: Local Docker Build (Dev)
on:
push:
branches: [ dev ]
pull_request:
branches: [ dev ]
jobs:
build-dev:
runs-on: host
steps:
- name: Checkout code (DEV)
run: |
echo "Current workspace: $(pwd)"
# Clean workspace and temp directory first
rm -rf * .git* 2>/dev/null || true
rm -rf /tmp/repo 2>/dev/null || true
# Clone the repository since Gitea runner doesn't auto-checkout
echo "Cloning repository..."
git clone --branch dev http://${{ secrets.username }}:${{ secrets.token }}@192.168.253.221:3000/sbcrumb/nfoguard.git /tmp/repo
cp -r /tmp/repo/* .
cp -r /tmp/repo/.* . 2>/dev/null || true
rm -rf /tmp/repo
echo "Repository cloned successfully"
ls -la
echo "Checking Dockerfile:"
head -30 Dockerfile
- name: Check Docker availability (DEV)
run: |
echo "Checking Docker installation..."
which docker || (echo "Docker not found in PATH" && exit 1)
docker --version || (echo "Docker not available" && exit 1)
docker info || (echo "Docker daemon not running" && exit 1)
- name: Configure Docker for local registry (DEV)
run: |
echo "Configuring Docker client for insecure local registry..."
LOCAL_REGISTRY="192.168.253.221:3000"
export DOCKER_CONFIG=/tmp/docker-config-dev
mkdir -p $DOCKER_CONFIG
# Create Docker client config for insecure registry
cat > $DOCKER_CONFIG/config.json << EOF
{
"auths": {},
"HttpHeaders": {
"User-Agent": "Docker-Client/20.10.0 (linux)"
},
"credsStore": "",
"credHelpers": {},
"experimental": "disabled"
}
EOF
echo "Docker client config created for DEV"
echo "Testing registry connectivity..."
curl -s "http://$LOCAL_REGISTRY/v2/" && echo "✅ Registry accessible via HTTP" || echo "❌ Registry not accessible"
- name: Build Docker image with caching (DEV)
run: |
echo "Building DEV Docker image with layer caching..."
LOCAL_REGISTRY="192.168.253.221:3000"
CACHE_IMAGE="$LOCAL_REGISTRY/sbcrumb/nfoguard:dev-buildcache"
# Clear any existing Docker configs to prevent permission issues
unset DOCKER_CONFIG
export HOME=/tmp/docker-home-dev
mkdir -p $HOME
# Enable Docker BuildKit for better caching
export DOCKER_BUILDKIT=1
# Check if DEV cache image exists and pull it
echo "Checking for existing DEV cache image..."
CACHE_ARGS=""
if curl -s "http://$LOCAL_REGISTRY/v2/sbcrumb/nfoguard/manifests/dev-buildcache" > /dev/null 2>&1; then
echo "✅ DEV cache manifest found, pulling image..."
if docker pull "$CACHE_IMAGE" 2>/dev/null; then
echo "✅ DEV cache image pulled successfully"
CACHE_ARGS="--cache-from=$CACHE_IMAGE"
else
echo "⚠️ DEV cache manifest exists but pull failed"
fi
else
echo "️ No DEV cache image found (first build or cache expired)"
fi
# Build DEV image with cache (if available)
echo "Building DEV image with layer caching..."
VERSION=$(cat VERSION 2>/dev/null || echo "unknown")
echo "Building version: $VERSION"
docker build \
$CACHE_ARGS \
--build-arg GIT_BRANCH=dev \
--build-arg BUILD_SOURCE=gitea \
--tag nfoguard:dev \
--tag nfoguard:${VERSION}-dev-gitea \
--tag nfoguard:dev-${{ github.sha }} \
--tag "$CACHE_IMAGE" \
.
echo "DEV Docker image built and tagged successfully"
- name: Login and Push to Gitea registry (DEV tags)
continue-on-error: true
run: |
echo "Using LOCAL IP ONLY for DEV push!"
export DOCKER_CONFIG=/tmp/docker-config-dev
LOCAL_REGISTRY="192.168.253.221:3000"
echo "Registry: $LOCAL_REGISTRY (DEV TAGS)"
# Try login to local registry
if echo "${{ secrets.token }}" | docker --config $DOCKER_CONFIG login "$LOCAL_REGISTRY" -u "${{ secrets.username }}" --password-stdin 2>/dev/null; then
echo "✅ DEV Login succeeded"
# Tag for local registry
VERSION=$(cat VERSION 2>/dev/null || echo "unknown")
docker tag nfoguard:dev "$LOCAL_REGISTRY/sbcrumb/nfoguard:dev"
docker tag nfoguard:${VERSION}-dev-gitea "$LOCAL_REGISTRY/sbcrumb/nfoguard:${VERSION}-dev-gitea"
docker tag nfoguard:dev "$LOCAL_REGISTRY/sbcrumb/nfoguard:dev-${{ github.sha }}"
# Push DEV images
echo "=== Pushing DEV cache image ==="
timeout 60 docker --config $DOCKER_CONFIG push "$LOCAL_REGISTRY/sbcrumb/nfoguard:dev-buildcache" && echo "✅ DEV cache pushed" || echo "⚠️ DEV cache push failed"
echo "=== Pushing DEV latest tag ==="
if timeout 120 docker --config $DOCKER_CONFIG push "$LOCAL_REGISTRY/sbcrumb/nfoguard:dev"; then
echo "✅ DEV tag pushed successfully"
echo "=== Pushing version-dev-gitea tag ==="
if timeout 60 docker --config $DOCKER_CONFIG push "$LOCAL_REGISTRY/sbcrumb/nfoguard:${VERSION}-dev-gitea"; then
echo "✅ Version-dev-gitea tag pushed successfully: ${VERSION}-dev-gitea"
fi
if timeout 60 docker --config $DOCKER_CONFIG push "$LOCAL_REGISTRY/sbcrumb/nfoguard:dev-${{ github.sha }}"; then
echo "✅ DEV SHA tag pushed successfully"
fi
fi
else
echo "❌ DEV local registry login failed - continuing to Docker Hub"
fi
echo ""
echo "🎉 DEV build completed successfully!"
echo ""
VERSION=$(cat VERSION 2>/dev/null || echo "unknown")
echo "📦 Available DEV images:"
echo " Local only: nfoguard:dev"
echo " Version tag: nfoguard:${VERSION}-dev-gitea"
echo " Local SHA: nfoguard:dev-${{ github.sha }}"
echo ""
echo "🐳 Pull with: docker pull 192.168.253.221:3000/sbcrumb/nfoguard:${VERSION}-dev-gitea"
+222
View File
@@ -0,0 +1,222 @@
name: Local Docker Build (Main)
on:
push:
branches: [ main ]
pull_request:
branches: [ main ]
jobs:
build:
runs-on: host
steps:
- name: Checkout code
run: |
echo "Current workspace: $(pwd)"
# Clean workspace and temp directory first
rm -rf * .git* 2>/dev/null || true
rm -rf /tmp/repo 2>/dev/null || true
# Clone the repository since Gitea runner doesn't auto-checkout
echo "Cloning repository..."
git clone --branch main http://${{ secrets.username }}:${{ secrets.token }}@192.168.253.221:3000/sbcrumb/nfoguard.git /tmp/repo
cp -r /tmp/repo/* .
cp -r /tmp/repo/.* . 2>/dev/null || true
rm -rf /tmp/repo
echo "Repository cloned successfully"
ls -la
echo "Checking Dockerfile:"
head -30 Dockerfile
- name: Check Docker availability
run: |
echo "Checking Docker installation..."
which docker || (echo "Docker not found in PATH" && exit 1)
docker --version || (echo "Docker not available" && exit 1)
docker info || (echo "Docker daemon not running" && exit 1)
- name: Configure Docker for local registry
run: |
echo "Configuring Docker client for insecure local registry..."
LOCAL_REGISTRY="192.168.253.221:3000"
# We can't use sudo in the container, so we'll configure the client differently
export DOCKER_CONFIG=/tmp/docker-config
mkdir -p $DOCKER_CONFIG
# Create Docker client config for insecure registry
cat > $DOCKER_CONFIG/config.json << EOF
{
"auths": {},
"HttpHeaders": {
"User-Agent": "Docker-Client/20.10.0 (linux)"
},
"credsStore": "",
"credHelpers": {},
"experimental": "disabled"
}
EOF
echo "Docker client config created"
echo "Note: Since we can't modify daemon config in container, we'll use --insecure-registry flag"
echo "Testing registry connectivity..."
curl -s "http://$LOCAL_REGISTRY/v2/" && echo "✅ Registry accessible via HTTP" || echo "❌ Registry not accessible"
- name: Build Docker image with caching
run: |
echo "Building Docker image with layer caching..."
LOCAL_REGISTRY="192.168.253.221:3000"
CACHE_IMAGE="$LOCAL_REGISTRY/sbcrumb/nfoguard:buildcache"
# Clear any existing Docker configs to prevent permission issues
unset DOCKER_CONFIG
export HOME=/tmp/docker-home
mkdir -p $HOME
# Enable Docker BuildKit for better caching
export DOCKER_BUILDKIT=1
# Check if cache image exists and pull it
echo "Checking for existing cache image..."
CACHE_ARGS=""
if curl -s "http://$LOCAL_REGISTRY/v2/sbcrumb/nfoguard/manifests/buildcache" > /dev/null 2>&1; then
echo "✅ Cache manifest found, pulling image..."
if docker pull "$CACHE_IMAGE" 2>/dev/null; then
echo "✅ Cache image pulled successfully"
CACHE_ARGS="--cache-from=$CACHE_IMAGE"
else
echo "⚠️ Cache manifest exists but pull failed"
fi
else
echo "️ No cache image found (first build or cache expired)"
fi
# Build with cache (if available)
echo "Building with layer caching..."
VERSION=$(cat VERSION 2>/dev/null || echo "unknown")
echo "Building version: $VERSION"
docker build \
$CACHE_ARGS \
--build-arg GIT_BRANCH=main \
--build-arg BUILD_SOURCE=gitea \
--tag nfoguard:latest \
--tag nfoguard:${VERSION}-gitea \
--tag nfoguard:${{ github.sha }} \
--tag "$CACHE_IMAGE" \
.
echo "Docker image built and tagged successfully"
# Show layer info for debugging (with error handling)
echo "=== Image layer history ==="
if docker history nfoguard:latest --format "table {{.CreatedBy}}\t{{.Size}}" 2>/dev/null; then
echo "✅ Image history displayed successfully"
else
echo "⚠️ Could not display image history (permissions issue, but build succeeded)"
fi
- name: Login and Push to Gitea registry (local IP only)
continue-on-error: true # Don't fail the build if local registry fails
run: |
echo "Using LOCAL IP ONLY - no Cloudflare tunnel!"
export DOCKER_CONFIG=/tmp/docker-config
# Force local IP - never use domain
LOCAL_REGISTRY="192.168.253.221:3000"
echo "Registry: $LOCAL_REGISTRY (LOCAL IP ONLY)"
# Check image size
IMAGE_SIZE=$(docker images nfoguard:latest --format "table {{.Size}}" | tail -n1)
echo "Image size: $IMAGE_SIZE"
echo "Testing local registry connectivity..."
curl -s "http://$LOCAL_REGISTRY/v2/" && echo "✅ Local registry accessible via HTTP" || echo "❌ Local registry not accessible"
# Try multiple approaches to handle the HTTP issue
echo "Attempting login to LOCAL registry..."
# Approach 1: Try with explicit HTTP in auth
if echo "${{ secrets.token }}" | docker --config $DOCKER_CONFIG login "$LOCAL_REGISTRY" -u "${{ secrets.username }}" --password-stdin 2>/dev/null; then
echo "✅ Login succeeded with direct method"
# Tag for local registry (cache image already tagged in build step)
VERSION=$(cat VERSION 2>/dev/null || echo "unknown")
docker tag nfoguard:latest "$LOCAL_REGISTRY/sbcrumb/nfoguard:latest"
docker tag nfoguard:${VERSION}-gitea "$LOCAL_REGISTRY/sbcrumb/nfoguard:${VERSION}-gitea"
docker tag nfoguard:latest "$LOCAL_REGISTRY/sbcrumb/nfoguard:${{ github.sha }}"
echo "Pushing to LOCAL registry (gigabit speed!)..."
# Push with reasonable timeout for local network
echo "=== Pushing cache image (for faster future builds) ==="
timeout 60 docker --config $DOCKER_CONFIG push "$LOCAL_REGISTRY/sbcrumb/nfoguard:buildcache" && echo "✅ Cache image pushed" || echo "⚠️ Cache push failed"
echo "=== Pushing latest tag to LOCAL IP ==="
if timeout 120 docker --config $DOCKER_CONFIG push "$LOCAL_REGISTRY/sbcrumb/nfoguard:latest"; then
echo "✅ Latest tag pushed successfully to LOCAL registry"
echo "=== Pushing version-gitea tag ==="
if timeout 60 docker --config $DOCKER_CONFIG push "$LOCAL_REGISTRY/sbcrumb/nfoguard:${VERSION}-gitea"; then
echo "✅ Version-gitea tag pushed successfully: ${VERSION}-gitea"
fi
echo "=== Pushing SHA tag to LOCAL IP ==="
if timeout 60 docker --config $DOCKER_CONFIG push "$LOCAL_REGISTRY/sbcrumb/nfoguard:${{ github.sha }}"; then
echo "✅ SHA tag pushed successfully to LOCAL registry"
else
echo "⚠️ SHA tag push failed but latest succeeded"
fi
else
echo "❌ Push to LOCAL registry failed"
echo "This is likely because Docker daemon needs insecure-registry configuration"
fi
else
echo "❌ Local registry login failed"
echo "This is expected - Docker daemon needs insecure-registry configuration"
echo ""
echo "🔧 To fix, run on the host machine:"
echo "sudo mkdir -p /etc/docker"
echo 'echo '"'"'{"insecure-registries":["'$LOCAL_REGISTRY'"]}'"'"' | sudo tee /etc/docker/daemon.json'
echo "sudo systemctl restart docker"
echo ""
echo "⏭️ Continuing to Docker Hub push..."
fi
echo "🎉 Local registry step completed!"
echo ""
echo "🎉 Local build completed successfully!"
echo ""
VERSION=$(cat VERSION 2>/dev/null || echo "unknown")
echo "📦 Available images:"
echo " Local Docker: nfoguard:latest"
echo " Version tag: nfoguard:${VERSION}-gitea"
echo " Local SHA: nfoguard:${{ github.sha }}"
echo " Local Gitea: 192.168.253.221:3000/sbcrumb/nfoguard:latest"
echo ""
echo "🐳 Pull with: docker pull 192.168.253.221:3000/sbcrumb/nfoguard:${VERSION}-gitea"
deploy:
needs: build
runs-on: host
if: github.ref == 'refs/heads/main'
steps:
- name: Deploy application
run: |
echo "Deploying application..."
# Add your deployment commands here
# Examples:
# docker-compose pull
# docker-compose up -d
# or
# docker stop nfoguard || true
# docker run -d --name nfoguard -p 8080:8080 nfoguard:latest
echo "Deployment completed"
+1 -1
View File
@@ -68,7 +68,7 @@ sync-to-github.sh
.github/
# Ignore Gitea workflows when pushing to GitHub
.gitea/
#.gitea/
# Local development documentation (Gitea only, not for GitHub)
.local/
+1 -1
View File
@@ -1 +1 @@
2.1.5-preserve-long-nfo-episodes-only
2.0.0
View File
+53
View File
@@ -0,0 +1,53 @@
"""
Pydantic models for NFOGuard API
"""
from pydantic import BaseModel
from typing import Optional, Dict, Any, List
class SonarrWebhook(BaseModel):
"""Sonarr webhook payload model"""
eventType: str
series: Optional[Dict[str, Any]] = None
episodes: Optional[list] = []
episodeFile: Optional[Dict[str, Any]] = None
isUpgrade: Optional[bool] = False
class Config:
extra = "allow"
class RadarrWebhook(BaseModel):
"""Radarr webhook payload model"""
eventType: str
movie: Optional[Dict[str, Any]] = None
movieFile: Optional[Dict[str, Any]] = None
isUpgrade: Optional[bool] = False
deletedFiles: Optional[list] = []
remoteMovie: Optional[Dict[str, Any]] = None
renamedMovieFiles: Optional[List[Dict[str, Any]]] = None
class Config:
extra = "allow"
class HealthResponse(BaseModel):
"""Health check response model"""
status: str
version: str
uptime: str
database_status: str
radarr_database: Optional[Dict[str, Any]] = None
class TVSeasonRequest(BaseModel):
"""TV season processing request model"""
series_path: str
season_name: str
class TVEpisodeRequest(BaseModel):
"""TV episode processing request model"""
series_path: str
season_name: str
episode_name: str
+870
View File
@@ -0,0 +1,870 @@
"""
FastAPI routes for NFOGuard - extracted from main nfoguard.py for modular architecture
"""
import os
import json
from pathlib import Path
from datetime import datetime, timezone
from fastapi import HTTPException, BackgroundTasks, Request
from typing import Optional
# Import models
from api.models import SonarrWebhook, RadarrWebhook, HealthResponse, TVSeasonRequest, TVEpisodeRequest
# ---------------------------
# Helper Functions
# ---------------------------
async def _read_payload(request: Request) -> dict:
"""Read webhook payload from request"""
content_type = (request.headers.get("content-type") or "").lower()
try:
if "application/json" in content_type:
return await request.json()
form = await request.form()
if "payload" in form:
return json.loads(form["payload"])
return dict(form)
except Exception as e:
print(f"ERROR: Failed to read webhook payload: {e}") # Using print since _log is not available
return {}
# ---------------------------
# Route Handlers
# ---------------------------
async def sonarr_webhook(request: Request, background_tasks: BackgroundTasks, dependencies: dict):
"""Handle Sonarr webhooks"""
tv_processor = dependencies["tv_processor"]
batcher = dependencies["batcher"]
config = dependencies["config"]
try:
payload = await _read_payload(request)
if not payload:
raise HTTPException(status_code=422, detail="Empty Sonarr payload")
webhook = SonarrWebhook(**payload)
print(f"INFO: Received Sonarr webhook: {webhook.eventType}")
if webhook.eventType not in ["Download", "Upgrade", "Rename"]:
return {"status": "ignored", "reason": f"Event type {webhook.eventType} not processed"}
if not webhook.series:
return {"status": "ignored", "reason": "No series data"}
series_info = webhook.series
series_title = series_info.get("title", "")
imdb_id = series_info.get("imdbId", "").replace("tt", "").strip()
if imdb_id:
imdb_id = f"tt{imdb_id}"
sonarr_path = series_info.get("path", "")
if not imdb_id:
print(f"ERROR: No IMDb ID for series: {series_title}")
return {"status": "error", "reason": "No IMDb ID"}
# Find series path
series_path = tv_processor.find_series_path(series_title, imdb_id, sonarr_path)
if not series_path:
print(f"ERROR: Could not find series directory: {series_title} ({imdb_id})")
return {"status": "error", "reason": "Series directory not found"}
# Add to batch queue with TV-prefixed key to avoid movie conflicts
tv_batch_key = f"tv:{imdb_id}"
webhook_dict = {
'path': str(series_path),
'series_info': series_info,
'event_type': webhook.eventType,
'episodes': webhook.episodes or [], # Include episode data for targeted processing
'processing_mode': config.tv_webhook_processing_mode
}
batcher.add_webhook(tv_batch_key, webhook_dict, 'tv')
return {"status": "accepted", "message": f"Sonarr webhook queued for {tv_batch_key}"}
except Exception as e:
print(f"ERROR: Sonarr webhook error: {e}")
raise HTTPException(status_code=422, detail=f"Invalid webhook: {e}")
async def radarr_webhook(request: Request, background_tasks: BackgroundTasks, dependencies: dict):
"""Handle Radarr webhooks"""
path_mapper = dependencies["path_mapper"]
batcher = dependencies["batcher"]
try:
payload = await _read_payload(request)
print(f"INFO: Received Radarr webhook: {payload.get('eventType', 'Unknown')}")
print(f"DEBUG: Full Radarr webhook payload: {payload}")
# Filter supported event types (same as Sonarr: Download, Upgrade, Rename)
event_type = payload.get('eventType', '')
if event_type not in ["Download", "Upgrade", "Rename"]:
return {"status": "ignored", "reason": f"Event type {event_type} not processed"}
# Extract movie info
movie_data = payload.get("movie", {})
if not movie_data:
print("WARNING: No movie data in Radarr webhook")
return {"status": "error", "message": "No movie data"}
# Get IMDb ID for batching key
imdb_id = movie_data.get("imdbId", "").lower()
if not imdb_id:
print("WARNING: No IMDb ID in Radarr webhook movie data")
return {"status": "error", "message": "No IMDb ID"}
# Get movie path and map it
movie_path = movie_data.get("folderPath") or movie_data.get("path", "")
if not movie_path:
print("ERROR: No movie path in Radarr webhook")
return {"status": "error", "message": "No movie path provided"}
# Map the path to container path
container_path = path_mapper.radarr_path_to_container_path(movie_path)
print(f"DEBUG: Mapped Radarr path {movie_path} -> {container_path}")
# CRITICAL: Verify the mapped path actually exists
if not Path(container_path).exists():
print(f"ERROR: RADARR WEBHOOK REJECTED: Mapped path does not exist: {container_path}")
print(f"ERROR: This prevents processing wrong movies due to path mapping issues")
return {"status": "error", "message": f"Mapped movie path does not exist: {container_path}"}
# Verify the path contains the expected IMDb ID
if imdb_id not in container_path.lower():
print(f"WARNING: IMDb ID {imdb_id} not found in container path {container_path}")
# Create movie-specific webhook data with proper path validation
movie_webhook_data = {
'path': container_path, # Use verified container path
'movie_info': movie_data,
'event_type': payload.get('eventType'),
'original_payload': payload
}
# Add to batch queue with movie-prefixed key to avoid TV conflicts
movie_batch_key = f"movie:{imdb_id}"
print(f"DEBUG: Adding Radarr webhook to batch: key={movie_batch_key}, movie_title={movie_data.get('title', 'Unknown')}")
batcher.add_webhook(movie_batch_key, movie_webhook_data, "movie")
return {"status": "success", "message": f"Radarr webhook queued for {movie_batch_key}"}
except Exception as e:
print(f"ERROR: Radarr webhook error: {e}")
return {"status": "error", "message": str(e)}
async def health(dependencies: dict) -> HealthResponse:
"""Health check endpoint with Radarr database status"""
db = dependencies["db"]
movie_processor = dependencies["movie_processor"]
start_time = dependencies["start_time"]
version = dependencies["version"]
uptime = datetime.now(timezone.utc) - start_time
# Check NFOGuard database
try:
with db.get_connection() as conn:
conn.execute("SELECT 1").fetchone()
db_status = "healthy"
except Exception as e:
db_status = f"error: {e}"
# Check Radarr database if available
radarr_db_health = None
overall_status = "healthy" if db_status == "healthy" else "degraded"
# Get Radarr client with database access from movie processor
try:
if hasattr(movie_processor, 'radarr') and movie_processor.radarr:
radarr_client = movie_processor.radarr
if hasattr(radarr_client, 'db_client') and radarr_client.db_client:
try:
radarr_db_health = radarr_client.db_client.health_check()
if radarr_db_health["status"] != "healthy":
overall_status = "degraded"
except Exception as e:
radarr_db_health = {
"status": "error",
"error": str(e),
"tested_at": datetime.now(timezone.utc).isoformat(timespec="seconds")
}
overall_status = "degraded"
except Exception as e:
# If movie processor isn't available, skip database health check
print(f"DEBUG: Skipping Radarr database health check: {e}")
return HealthResponse(
status=overall_status,
version=version,
uptime=str(uptime),
database_status=db_status,
radarr_database=radarr_db_health
)
async def get_stats(dependencies: dict):
"""Get database statistics"""
db = dependencies["db"]
try:
return db.get_stats()
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
async def batch_status(dependencies: dict):
"""Get batch queue status"""
batcher = dependencies["batcher"]
return batcher.get_status()
async def debug_movie_import_date(imdb_id: str, dependencies: dict):
"""Debug endpoint to analyze movie import date detection"""
movie_processor = dependencies["movie_processor"]
try:
if not imdb_id.startswith("tt"):
imdb_id = f"tt{imdb_id}"
print(f"INFO: === DEBUG MOVIE IMPORT DATE: {imdb_id} ===")
if not (os.environ.get("RADARR_URL") and os.environ.get("RADARR_API_KEY")):
return {
"error": "Radarr not configured",
"imdb_id": imdb_id,
"radarr_configured": False
}
# Create Radarr client
from clients.radarr_client import RadarrClient
radarr_client = RadarrClient(
os.environ.get("RADARR_URL"),
os.environ.get("RADARR_API_KEY")
)
# Look up movie
movie_obj = radarr_client.movie_by_imdb(imdb_id)
if not movie_obj:
return {
"error": f"Movie not found in Radarr for IMDb ID {imdb_id}",
"imdb_id": imdb_id,
"radarr_configured": True,
"movie_found": False
}
movie_id = movie_obj.get("id")
movie_title = movie_obj.get("title")
print(f"INFO: Found movie: {movie_title} (Radarr ID: {movie_id})")
# Test the FULL movie processing pipeline (not just database lookup)
print(f"INFO: === TESTING FULL MOVIE PROCESSING PIPELINE ===")
# Create a dummy path for testing the decision logic
dummy_path = Path("/tmp/test")
try:
# Use the global movie processor instance to test full decision logic
if movie_processor:
# First check external clients configuration
print(f"INFO: === CHECKING EXTERNAL CLIENTS CONFIG ===")
try:
tmdb_key = os.environ.get("TMDB_API_KEY", "")
print(f"INFO: TMDB API Key configured: {'✅ YES' if tmdb_key else '❌ NO'}")
if tmdb_key:
print(f"INFO: TMDB API Key length: {len(tmdb_key)} chars")
# Check if external clients exist
external_clients_available = hasattr(movie_processor, 'external_clients') and movie_processor.external_clients
print(f"INFO: External clients initialized: {'✅ YES' if external_clients_available else '❌ NO'}")
except Exception as e:
print(f"ERROR: Error checking external clients config: {e}")
# Test the full decision logic (including TMDB fallback)
final_date, final_source, released = movie_processor._decide_movie_dates(
imdb_id, dummy_path, should_query=True, existing=None
)
print(f"INFO: === FULL PIPELINE RESULT ===")
print(f"INFO: Final date: {final_date}")
print(f"INFO: Final source: {final_source}")
print(f"INFO: Released (theater): {released}")
return {
"imdb_id": imdb_id,
"radarr_configured": True,
"movie_found": True,
"movie_title": movie_title,
"movie_id": movie_id,
"full_pipeline_test": {
"final_date": final_date,
"final_source": final_source,
"theater_release": released,
"decision_logic": "✅ TESTED FULL PIPELINE INCLUDING TMDB FALLBACK"
},
"database_only_test": {
"radarr_db_result": radarr_client.get_movie_import_date(movie_id, fallback_to_file_date=True),
"note": "This is just the database part - fallback happens in full pipeline"
},
"debug_info": {
"radarr_url": os.environ.get("RADARR_URL"),
"movie_digital_release": movie_obj.get("digitalRelease"),
"movie_in_cinemas": movie_obj.get("inCinemas"),
"movie_physical_release": movie_obj.get("physicalRelease"),
"movie_folder_path": movie_obj.get("folderPath")
}
}
else:
print("ERROR: Movie processor not available - testing database only")
# Fallback to database-only testing
import_date, source = radarr_client.get_movie_import_date(movie_id, fallback_to_file_date=True)
return {
"error": "Movie processor not available - only database test performed",
"imdb_id": imdb_id,
"radarr_configured": True,
"movie_found": True,
"movie_title": movie_title,
"movie_id": movie_id,
"detected_import_date": import_date,
"import_source": source,
"debug_info": {
"note": "FULL PIPELINE TEST FAILED - movie processor not initialized"
}
}
except Exception as pipeline_error:
print(f"ERROR: Full pipeline test failed: {pipeline_error}")
# Fallback to database-only testing
import_date, source = radarr_client.get_movie_import_date(movie_id, fallback_to_file_date=True)
return {
"pipeline_error": str(pipeline_error),
"imdb_id": imdb_id,
"radarr_configured": True,
"movie_found": True,
"movie_title": movie_title,
"movie_id": movie_id,
"detected_import_date": import_date,
"import_source": source,
"debug_info": {
"note": "FULL PIPELINE TEST FAILED - showing database-only result"
}
}
except Exception as e:
print(f"ERROR: Debug endpoint error for {imdb_id}: {e}")
return {
"error": str(e),
"imdb_id": imdb_id,
"success": False
}
async def debug_movie_history(imdb_id: str, dependencies: dict):
"""Detailed history analysis for a movie"""
movie_processor = dependencies["movie_processor"]
try:
if not imdb_id.startswith("tt"):
imdb_id = f"tt{imdb_id}"
print(f"INFO: === DETAILED HISTORY ANALYSIS: {imdb_id} ===")
# This would need the rest of the implementation from the original function
# For now, returning a placeholder
return {
"imdb_id": imdb_id,
"message": "History analysis endpoint - implementation needed"
}
except Exception as e:
print(f"ERROR: Debug history endpoint error for {imdb_id}: {e}")
return {
"error": str(e),
"imdb_id": imdb_id,
"success": False
}
async def manual_scan(background_tasks: BackgroundTasks, path: Optional[str] = None, scan_type: str = "both", dependencies: dict = None):
"""Manual scan endpoint"""
config = dependencies["config"]
nfo_manager = dependencies["nfo_manager"]
tv_processor = dependencies["tv_processor"]
movie_processor = dependencies["movie_processor"]
if scan_type not in ["both", "tv", "movies"]:
raise HTTPException(status_code=400, detail="scan_type must be 'both', 'tv', or 'movies'")
async def run_scan():
paths_to_scan = []
if path:
paths_to_scan = [Path(path)]
else:
if scan_type in ["both", "tv"]:
paths_to_scan.extend(config.tv_paths)
if scan_type in ["both", "movies"]:
paths_to_scan.extend(config.movie_paths)
for scan_path in paths_to_scan:
if not scan_path.exists():
continue
if scan_type in ["both", "tv"] and (scan_path in config.tv_paths or path):
# Handle specific season/episode path
if path and scan_path.name.lower().startswith('season'):
# Single season processing
series_path = scan_path.parent
if nfo_manager.parse_imdb_from_path(series_path):
print(f"INFO: Processing single season: {scan_path}")
try:
tv_processor.process_season(series_path, scan_path)
except Exception as e:
print(f"ERROR: Failed processing season {scan_path}: {e}")
elif path and scan_path.is_file() and scan_path.suffix.lower() in ('.mkv', '.mp4', '.avi'):
# Single episode processing
season_path = scan_path.parent
series_path = season_path.parent
if nfo_manager.parse_imdb_from_path(series_path):
print(f"INFO: Processing single episode: {scan_path}")
try:
tv_processor.process_episode_file(series_path, season_path, scan_path)
except Exception as e:
print(f"ERROR: Failed processing episode {scan_path}: {e}")
else:
# Check if this path itself is a series (has IMDb ID in the directory name)
if nfo_manager.parse_imdb_from_path(scan_path):
try:
tv_processor.process_series(scan_path)
except Exception as e:
print(f"ERROR: Failed processing TV series {scan_path}: {e}")
else:
# Full series processing - scan subdirectories
import re
for item in scan_path.iterdir():
if (item.is_dir() and
not item.name.lower().startswith('season') and
not re.match(r'^season\s+\d+$', item.name, re.IGNORECASE) and
nfo_manager.parse_imdb_from_path(item)):
try:
tv_processor.process_series(item)
except Exception as e:
print(f"ERROR: Failed processing TV series {item}: {e}")
if scan_type in ["both", "movies"] and scan_path in config.movie_paths:
print(f"INFO: Scanning movies in: {scan_path}")
movie_count = 0
for item in scan_path.iterdir():
if item.is_dir() and nfo_manager.find_movie_imdb_id(item):
movie_count += 1
print(f"INFO: Processing movie: {item.name}")
try:
movie_processor.process_movie(item)
except Exception as e:
print(f"ERROR: Failed processing movie {item}: {e}")
print(f"INFO: Completed movie scan: {movie_count} movies processed in {scan_path}")
background_tasks.add_task(run_scan)
return {"status": "started", "message": f"Manual {scan_type} scan started"}
async def scan_tv_season(background_tasks: BackgroundTasks, request: TVSeasonRequest, dependencies: dict):
"""Scan a specific TV season - URL-safe endpoint"""
nfo_manager = dependencies["nfo_manager"]
tv_processor = dependencies["tv_processor"]
try:
series_dir = Path(request.series_path)
season_dir = series_dir / request.season_name
if not series_dir.exists():
raise HTTPException(status_code=404, detail=f"Series path not found: {request.series_path}")
if not season_dir.exists():
raise HTTPException(status_code=404, detail=f"Season path not found: {season_dir}")
imdb_id = nfo_manager.parse_imdb_from_path(series_dir)
if not imdb_id:
raise HTTPException(status_code=400, detail="No IMDb ID found in series path")
async def process_season():
print(f"INFO: Processing TV season: {season_dir}")
try:
tv_processor.process_season(series_dir, season_dir)
except Exception as e:
print(f"ERROR: Failed processing season {season_dir}: {e}")
background_tasks.add_task(process_season)
return {"status": "started", "message": f"Season scan started for {request.season_name}"}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
async def scan_tv_episode(background_tasks: BackgroundTasks, request: TVEpisodeRequest, dependencies: dict):
"""Scan a specific TV episode - URL-safe endpoint"""
nfo_manager = dependencies["nfo_manager"]
tv_processor = dependencies["tv_processor"]
try:
series_dir = Path(request.series_path)
season_dir = series_dir / request.season_name
episode_file = season_dir / request.episode_name
if not series_dir.exists():
raise HTTPException(status_code=404, detail=f"Series path not found: {request.series_path}")
if not episode_file.exists():
raise HTTPException(status_code=404, detail=f"Episode file not found: {episode_file}")
imdb_id = nfo_manager.parse_imdb_from_path(series_dir)
if not imdb_id:
raise HTTPException(status_code=400, detail="No IMDb ID found in series path")
async def process_episode():
print(f"INFO: Processing TV episode: {episode_file}")
try:
tv_processor.process_episode_file(series_dir, season_dir, episode_file)
except Exception as e:
print(f"ERROR: Failed processing episode {episode_file}: {e}")
background_tasks.add_task(process_episode)
return {"status": "started", "message": f"Episode scan started for {request.episode_name}"}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
async def test_bulk_update(dependencies: dict):
"""Test bulk update functionality without modifying data"""
try:
from clients.radarr_db_client import RadarrDbClient
# Test Radarr database
radarr_db = RadarrDbClient.from_env()
if not radarr_db:
return {"status": "error", "message": "Radarr database connection failed"}
# Test query execution
query = 'SELECT COUNT(*) FROM "Movies" m JOIN "MovieMetadata" mm ON m."MovieMetadataId" = mm."Id" WHERE mm."ImdbId" IS NOT NULL'
with radarr_db._get_connection() as conn:
cursor = conn.cursor()
cursor.execute(query)
movie_count = cursor.fetchone()[0]
return {
"status": "success",
"message": "Bulk update test passed",
"movies_with_imdb": movie_count,
"database_type": radarr_db.db_type
}
except Exception as e:
return {"status": "error", "message": f"Bulk update test failed: {e}"}
async def test_movie_scan(dependencies: dict):
"""Test movie directory scanning logic"""
config = dependencies["config"]
nfo_manager = dependencies["nfo_manager"]
try:
results = []
for path in config.movie_paths:
path_result = {
"path": str(path),
"exists": path.exists(),
"movies_found": 0
}
if path.exists():
for item in path.iterdir():
if item.is_dir() and nfo_manager.find_movie_imdb_id(item):
path_result["movies_found"] += 1
results.append(path_result)
total_movies = sum(r["movies_found"] for r in results)
return {
"status": "success",
"message": f"Movie scan test found {total_movies} movies",
"path_results": results
}
except Exception as e:
return {"status": "error", "message": f"Movie scan test failed: {e}"}
async def trigger_bulk_update(background_tasks: BackgroundTasks, dependencies: dict):
"""Trigger bulk update of all movies"""
async def run_bulk_update():
try:
from bulk_update_movies import bulk_update_all_movies
success = bulk_update_all_movies()
print(f"INFO: Bulk update completed: {'success' if success else 'failed'}")
except Exception as e:
print(f"ERROR: Bulk update error: {e}")
background_tasks.add_task(run_bulk_update)
return {"status": "started", "message": "Bulk update started"}
async def debug_movie_priority_logic(imdb_id: str, dependencies: dict):
"""Debug endpoint showing how MOVIE_PRIORITY affects date selection"""
config = dependencies["config"]
movie_processor = dependencies["movie_processor"]
try:
if not imdb_id.startswith("tt"):
imdb_id = f"tt{imdb_id}"
result = {
"imdb_id": imdb_id,
"movie_priority": config.movie_priority,
"release_date_priority": config.release_date_priority,
"priority_explanation": "",
"date_sources": {},
"selected_date": None,
"selected_source": None
}
# Get Radarr import date
if movie_processor.radarr.api_key:
radarr_movie = movie_processor.radarr.movie_by_imdb(imdb_id)
if radarr_movie:
movie_id = radarr_movie.get("id")
if movie_id:
import_date, import_source = movie_processor.radarr.get_movie_import_date(movie_id)
if import_date:
result["date_sources"]["radarr_import"] = {
"date": import_date,
"source": import_source
}
# Get digital release dates with detailed logging
digital_date, digital_source = movie_processor._get_digital_release_date(imdb_id)
if digital_date:
result["date_sources"]["digital_release"] = {
"date": digital_date,
"source": digital_source
}
else:
# Add debug info about why digital date wasn't found
candidates = movie_processor.external_clients.get_digital_release_candidates(imdb_id)
result["date_sources"]["digital_release_debug"] = {
"candidates_found": len(candidates),
"candidates": candidates[:3] if candidates else [], # Show first 3
"reason": digital_source if digital_source else "no_digital_dates_found"
}
# Show priority logic
if config.movie_priority == "import_then_digital":
priority_list = "".join(config.release_date_priority)
result["priority_explanation"] = f"1st: Radarr import history, 2nd: Release dates ({priority_list}), 3rd: file mtime. Note: If import is only file date, prefer reasonable release dates."
radarr_import = result["date_sources"].get("radarr_import")
digital_release = result["date_sources"].get("digital_release")
# Check for file date fallback logic
if radarr_import and radarr_import["source"] == "radarr:db.file.dateAdded" and digital_release:
# Test the smart logic
would_prefer_digital = movie_processor._should_prefer_release_over_file_date(
digital_release["date"],
digital_release["source"],
None, # We don't have theatrical date in this debug context
imdb_id
)
result["file_date_detected"] = True
result["would_prefer_digital"] = would_prefer_digital
if would_prefer_digital:
result["selected_date"] = digital_release["date"]
result["selected_source"] = digital_release["source"] + " (preferred over file date)"
else:
result["selected_date"] = radarr_import["date"]
result["selected_source"] = radarr_import["source"] + " (digital too old)"
elif radarr_import and radarr_import["source"] != "radarr:db.file.dateAdded":
result["selected_date"] = radarr_import["date"]
result["selected_source"] = radarr_import["source"]
elif digital_release:
result["selected_date"] = digital_release["date"]
result["selected_source"] = digital_release["source"]
else: # digital_then_import
result["priority_explanation"] = "1st: TMDB/OMDb digital release, 2nd: Radarr import history, 3rd: file mtime"
if result["date_sources"].get("digital_release"):
result["selected_date"] = result["date_sources"]["digital_release"]["date"]
result["selected_source"] = result["date_sources"]["digital_release"]["source"]
elif result["date_sources"].get("radarr_import"):
result["selected_date"] = result["date_sources"]["radarr_import"]["date"]
result["selected_source"] = result["date_sources"]["radarr_import"]["source"]
# Show external API status
result["external_apis"] = {
"tmdb_enabled": movie_processor.external_clients.tmdb.enabled,
"omdb_enabled": movie_processor.external_clients.omdb.enabled,
"jellyseerr_enabled": movie_processor.external_clients.jellyseerr.enabled
}
return result
except Exception as e:
return {"error": str(e), "imdb_id": imdb_id}
async def debug_tmdb_lookup(imdb_id: str, dependencies: dict):
"""Debug TMDB API lookup for a specific movie"""
movie_processor = dependencies["movie_processor"]
try:
if not imdb_id.startswith("tt"):
imdb_id = f"tt{imdb_id}"
result = {
"imdb_id": imdb_id,
"tmdb_api_enabled": movie_processor.external_clients.tmdb.enabled,
"tmdb_api_key_configured": bool(movie_processor.external_clients.tmdb.api_key),
"steps": {}
}
if not movie_processor.external_clients.tmdb.enabled:
result["error"] = "TMDB API not enabled - check TMDB_API_KEY environment variable"
return result
# Step 1: Find movie by IMDb ID
print(f"INFO: TMDB Debug: Looking up {imdb_id}")
tmdb_movie = movie_processor.external_clients.tmdb.find_by_imdb(imdb_id)
result["steps"]["1_find_by_imdb"] = {
"found": bool(tmdb_movie),
"tmdb_movie": tmdb_movie if tmdb_movie else None
}
if not tmdb_movie:
result["error"] = f"Movie {imdb_id} not found in TMDB"
return result
tmdb_id = tmdb_movie.get("id")
result["tmdb_id"] = tmdb_id
# Step 2: Get release dates
if tmdb_id:
print(f"INFO: TMDB Debug: Getting release dates for TMDB ID {tmdb_id}")
release_dates_result = movie_processor.external_clients.tmdb._get(f"/movie/{tmdb_id}/release_dates")
result["steps"]["2_release_dates"] = {
"raw_response": release_dates_result,
"has_results": bool(release_dates_result and release_dates_result.get("results"))
}
# Step 3: Look for US digital releases
if release_dates_result and release_dates_result.get("results"):
us_releases = []
for country_data in release_dates_result["results"]:
if country_data.get("iso_3166_1") == "US":
us_releases = country_data.get("release_dates", [])
break
result["steps"]["3_us_releases"] = {
"found_us_data": bool(us_releases),
"us_releases": us_releases
}
# Step 4: Look for digital releases (type 4)
digital_releases = [r for r in us_releases if r.get("type") == 4]
result["steps"]["4_digital_releases"] = {
"digital_count": len(digital_releases),
"digital_releases": digital_releases
}
# Step 5: Test the full digital release function
digital_date = movie_processor.external_clients.tmdb.get_digital_release_date(imdb_id)
result["steps"]["5_final_result"] = {
"digital_date": digital_date,
"success": bool(digital_date)
}
return result
except Exception as e:
return {"error": str(e), "imdb_id": imdb_id, "traceback": str(e)}
# ---------------------------
# Route Registration
# ---------------------------
def register_routes(app, dependencies: dict):
"""
Register all routes with the FastAPI app
Args:
app: FastAPI application instance
dependencies: Dictionary containing:
- db: NFOGuardDatabase instance
- nfo_manager: NFOManager instance
- path_mapper: PathMapper instance
- tv_processor: TVProcessor instance
- movie_processor: MovieProcessor instance
- batcher: WebhookBatcher instance
- start_time: Application start time
- config: NFOGuardConfig instance
- version: Application version string
"""
@app.post("/webhook/sonarr")
async def _sonarr_webhook(request: Request, background_tasks: BackgroundTasks):
return await sonarr_webhook(request, background_tasks, dependencies)
@app.post("/webhook/radarr")
async def _radarr_webhook(request: Request, background_tasks: BackgroundTasks):
return await radarr_webhook(request, background_tasks, dependencies)
@app.get("/health")
async def _health() -> HealthResponse:
return await health(dependencies)
@app.get("/stats")
async def _get_stats():
return await get_stats(dependencies)
@app.get("/batch/status")
async def _batch_status():
return await batch_status(dependencies)
@app.get("/debug/movie/{imdb_id}")
async def _debug_movie_import_date(imdb_id: str):
return await debug_movie_import_date(imdb_id, dependencies)
@app.get("/debug/movie/{imdb_id}/history")
async def _debug_movie_history(imdb_id: str):
return await debug_movie_history(imdb_id, dependencies)
@app.post("/manual/scan")
async def _manual_scan(background_tasks: BackgroundTasks, path: Optional[str] = None, scan_type: str = "both"):
return await manual_scan(background_tasks, path, scan_type, dependencies)
@app.post("/tv/scan-season")
async def _scan_tv_season(background_tasks: BackgroundTasks, request: TVSeasonRequest):
return await scan_tv_season(background_tasks, request, dependencies)
@app.post("/tv/scan-episode")
async def _scan_tv_episode(background_tasks: BackgroundTasks, request: TVEpisodeRequest):
return await scan_tv_episode(background_tasks, request, dependencies)
@app.post("/test/bulk-update")
async def _test_bulk_update():
return await test_bulk_update(dependencies)
@app.post("/test/movie-scan")
async def _test_movie_scan():
return await test_movie_scan(dependencies)
@app.post("/bulk/update")
async def _trigger_bulk_update(background_tasks: BackgroundTasks):
return await trigger_bulk_update(background_tasks, dependencies)
@app.get("/debug/movie/{imdb_id}/priority")
async def _debug_movie_priority_logic(imdb_id: str):
return await debug_movie_priority_logic(imdb_id, dependencies)
@app.get("/debug/tmdb/{imdb_id}")
async def _debug_tmdb_lookup(imdb_id: str):
return await debug_tmdb_lookup(imdb_id, dependencies)
View File
+65
View File
@@ -0,0 +1,65 @@
"""
NFOGuard Configuration Module
Handles all configuration loading and validation
"""
import os
from pathlib import Path
from typing import List
def _bool_env(name: str, default: bool) -> bool:
"""Convert environment variable to boolean"""
v = os.environ.get(name)
if v is None:
return default
return v.lower() in ("1", "true", "yes", "y", "on")
class NFOGuardConfig:
"""Configuration class for NFOGuard"""
def __init__(self):
# Paths - No hardcoded defaults, must be configured via environment
tv_paths_env = os.environ.get("TV_PATHS", "")
movie_paths_env = os.environ.get("MOVIE_PATHS", "")
if not tv_paths_env:
raise ValueError("TV_PATHS environment variable is required but not set")
if not movie_paths_env:
raise ValueError("MOVIE_PATHS environment variable is required but not set")
self.tv_paths = [Path(p.strip()) for p in tv_paths_env.split(",") if p.strip()]
self.movie_paths = [Path(p.strip()) for p in movie_paths_env.split(",") if p.strip()]
# Core settings
self.manage_nfo = _bool_env("MANAGE_NFO", True)
self.fix_dir_mtimes = _bool_env("FIX_DIR_MTIMES", True)
self.lock_metadata = _bool_env("LOCK_METADATA", True)
self.debug = _bool_env("DEBUG", False)
self.manager_brand = os.environ.get("MANAGER_BRAND", "NFOGuard")
# Batching
self.batch_delay = float(os.environ.get("BATCH_DELAY", "5.0"))
self.max_concurrent = int(os.environ.get("MAX_CONCURRENT_SERIES", "3"))
# Database
self.db_path = Path(os.environ.get("DB_PATH", "/app/data/media_dates.db"))
# Movie processing
self.movie_priority = os.environ.get("MOVIE_PRIORITY", "import_then_digital").lower()
self.prefer_release_dates_over_file_dates = _bool_env("PREFER_RELEASE_DATES_OVER_FILE_DATES", True)
self.allow_file_date_fallback = _bool_env("ALLOW_FILE_DATE_FALLBACK", False)
self.release_date_priority = [p.strip() for p in os.environ.get("RELEASE_DATE_PRIORITY", "digital,physical,theatrical").split(",")]
self.enable_smart_date_validation = _bool_env("ENABLE_SMART_DATE_VALIDATION", True)
self.max_release_date_gap_years = int(os.environ.get("MAX_RELEASE_DATE_GAP_YEARS", "10"))
self.movie_poll_mode = os.environ.get("MOVIE_POLL_MODE", "always").lower()
self.movie_update_mode = os.environ.get("MOVIE_DATE_UPDATE_MODE", "backfill_only").lower()
# TV processing
self.tv_season_dir_format = os.environ.get("TV_SEASON_DIR_FORMAT", "Season {season:02d}")
self.tv_season_dir_pattern = os.environ.get("TV_SEASON_DIR_PATTERN", "season ").lower()
self.tv_webhook_processing_mode = os.environ.get("TV_WEBHOOK_PROCESSING_MODE", "targeted").lower()
# Global config instance
config = NFOGuardConfig()
+423
View File
@@ -0,0 +1,423 @@
"""
Async NFO Manager for NFOGuard
High-performance async NFO file operations with concurrent processing
"""
import asyncio
from pathlib import Path
from typing import Optional, List, Dict, Any
from datetime import datetime
import xml.etree.ElementTree as ET
from utils.logging import _log
from utils.async_file_utils import (
async_read_nfo_file,
async_write_nfo_file,
async_file_exists,
async_set_file_mtime,
async_batch_nfo_operations,
async_concurrent_episode_processing
)
from utils.nfo_patterns import (
create_basic_nfo_structure,
extract_imdb_from_nfo_content,
extract_dates_from_nfo,
extract_imdb_id_from_text
)
from utils.validation import validate_date_string
class AsyncNFOManager:
"""Async NFO file manager with concurrent processing capabilities"""
def __init__(self, manager_brand: str = "NFOGuard", debug: bool = False):
self.manager_brand = manager_brand
self.debug = debug
async def async_parse_imdb_from_path(self, path: Path) -> Optional[str]:
"""
Async extract IMDb ID from directory path or filename
Args:
path: Path to examine
Returns:
IMDb ID if found, None otherwise
"""
# Use the sync version since it's just string processing
return extract_imdb_id_from_text(str(path))
async def async_parse_imdb_from_nfo(self, nfo_path: Path) -> Optional[str]:
"""
Async extract IMDb ID from NFO file content
Args:
nfo_path: Path to NFO file
Returns:
IMDb ID if found, None otherwise
"""
root = await async_read_nfo_file(nfo_path)
if root is None:
return None
return extract_imdb_from_nfo_content(root)
async def async_find_movie_imdb_id(self, movie_dir: Path) -> Optional[str]:
"""
Async find IMDb ID from directory name, filenames, or NFO file
Args:
movie_dir: Path to movie directory
Returns:
IMDb ID if found, None otherwise
"""
if self.debug:
_log("DEBUG", f"Async searching for IMDb ID in: {movie_dir.name}")
# First try directory name
imdb_id = await self.async_parse_imdb_from_path(movie_dir)
if imdb_id:
if self.debug:
_log("DEBUG", f"Found IMDb ID in directory name: {imdb_id}")
return imdb_id
# Try all files in the directory concurrently
if await async_file_exists(movie_dir):
try:
from utils.async_file_utils import aiofiles
entries = await aiofiles.os.listdir(movie_dir)
# Create tasks to check all files
async def check_file(filename: str) -> Optional[str]:
file_path = movie_dir / filename
if await aiofiles.os.path.isfile(file_path):
return await self.async_parse_imdb_from_path(file_path)
return None
# Check all files concurrently
results = await asyncio.gather(
*[check_file(filename) for filename in entries],
return_exceptions=True
)
# Find first valid IMDb ID
for result in results:
if isinstance(result, str) and result:
if self.debug:
_log("DEBUG", f"Found IMDb ID in filename: {result}")
return result
except Exception as e:
_log("WARNING", f"Failed to scan directory {movie_dir}: {e}")
# Finally, try NFO file content
nfo_path = movie_dir / "movie.nfo"
imdb_id = await self.async_parse_imdb_from_nfo(nfo_path)
if imdb_id:
if self.debug:
_log("DEBUG", f"Found IMDb ID in NFO file: {imdb_id}")
return imdb_id
if self.debug:
_log("DEBUG", f"No IMDb ID found for: {movie_dir.name}")
return None
async def async_create_movie_nfo(
self,
movie_dir: Path,
imdb_id: str,
dateadded: str,
premiered: Optional[str] = None,
lock_metadata: bool = True
) -> bool:
"""
Async create movie NFO file
Args:
movie_dir: Path to movie directory
imdb_id: IMDb ID
dateadded: Date added
premiered: Optional premiere date
lock_metadata: Whether to lock metadata
Returns:
True if successful, False otherwise
"""
try:
nfo_path = movie_dir / "movie.nfo"
# Prepare dates
dates = {"dateadded": dateadded}
if premiered and validate_date_string(premiered):
dates["premiered"] = premiered
# Create NFO structure
root = create_basic_nfo_structure(
media_type="movie",
title=movie_dir.name,
imdb_id=imdb_id,
dates=dates,
additional_fields={"source": self.manager_brand}
)
# Write NFO file asynchronously
success = await async_write_nfo_file(nfo_path, root, lock_metadata)
if success and self.debug:
_log("DEBUG", f"Created movie NFO: {nfo_path}")
return success
except Exception as e:
_log("ERROR", f"Failed to create movie NFO for {movie_dir}: {e}")
return False
async def async_create_episode_nfo(
self,
season_dir: Path,
season: int,
episode: int,
aired: Optional[str] = None,
dateadded: Optional[str] = None,
source: str = "unknown",
lock_metadata: bool = True
) -> bool:
"""
Async create episode NFO file
Args:
season_dir: Path to season directory
season: Season number
episode: Episode number
aired: Optional air date
dateadded: Optional date added
source: Source of the data
lock_metadata: Whether to lock metadata
Returns:
True if successful, False otherwise
"""
try:
nfo_filename = f"S{season:02d}E{episode:02d}.nfo"
nfo_path = season_dir / nfo_filename
# Prepare dates
dates = {}
if aired and validate_date_string(aired):
dates["aired"] = aired
if dateadded and validate_date_string(dateadded):
dates["dateadded"] = dateadded
# Create NFO structure
root = create_basic_nfo_structure(
media_type="episodedetails",
title=f"S{season:02d}E{episode:02d}",
dates=dates,
additional_fields={
"season": str(season),
"episode": str(episode),
"source": source
}
)
# Write NFO file asynchronously
success = await async_write_nfo_file(nfo_path, root, lock_metadata)
if success and self.debug:
_log("DEBUG", f"Created episode NFO: {nfo_path}")
return success
except Exception as e:
_log("ERROR", f"Failed to create episode NFO S{season:02d}E{episode:02d}: {e}")
return False
async def async_batch_create_episode_nfos(
self,
episode_data_list: List[Dict[str, Any]],
max_concurrent: int = 5
) -> List[bool]:
"""
Batch create multiple episode NFOs concurrently
Args:
episode_data_list: List of episode data dictionaries
max_concurrent: Maximum concurrent NFO operations
Returns:
List of success/failure results
"""
async def _create_episode_nfo(episode_data: Dict[str, Any]) -> bool:
return await self.async_create_episode_nfo(
season_dir=episode_data.get('season_dir'),
season=episode_data.get('season'),
episode=episode_data.get('episode'),
aired=episode_data.get('aired'),
dateadded=episode_data.get('dateadded'),
source=episode_data.get('source', 'unknown'),
lock_metadata=episode_data.get('lock_metadata', True)
)
return await async_concurrent_episode_processing(
episode_data_list,
_create_episode_nfo,
max_concurrent
)
async def async_set_file_mtime(self, file_path: Path, date_str: str) -> bool:
"""
Async set file modification time from date string
Args:
file_path: Path to file
date_str: Date string in ISO format
Returns:
True if successful, False otherwise
"""
try:
if not validate_date_string(date_str):
_log("WARNING", f"Invalid date format for mtime: {date_str}")
return False
# Parse date string to timestamp
dt = datetime.fromisoformat(date_str.replace('Z', '+00:00'))
timestamp = dt.timestamp()
# Set file mtime asynchronously
success = await async_set_file_mtime(file_path, timestamp)
if success and self.debug:
_log("DEBUG", f"Set mtime for {file_path}: {date_str}")
return success
except Exception as e:
_log("ERROR", f"Failed to set mtime for {file_path}: {e}")
return False
async def async_batch_set_file_mtimes(
self,
file_mtime_pairs: List[Tuple[Path, str]],
max_concurrent: int = 10
) -> List[bool]:
"""
Batch set file modification times concurrently
Args:
file_mtime_pairs: List of (file_path, date_str) tuples
max_concurrent: Maximum concurrent operations
Returns:
List of success/failure results
"""
async def _set_single_mtime(file_path: Path, date_str: str) -> bool:
return await self.async_set_file_mtime(file_path, date_str)
# Create tasks for all mtime operations
semaphore = asyncio.Semaphore(max_concurrent)
async def _set_mtime_with_semaphore(file_path: Path, date_str: str) -> bool:
async with semaphore:
return await _set_single_mtime(file_path, date_str)
tasks = [
_set_mtime_with_semaphore(file_path, date_str)
for file_path, date_str in file_mtime_pairs
]
results = await asyncio.gather(*tasks, return_exceptions=True)
return [result if isinstance(result, bool) else False for result in results]
async def async_validate_nfo_integrity(
self,
nfo_paths: List[Path],
max_concurrent: int = 10
) -> Dict[str, Any]:
"""
Async validate integrity of multiple NFO files
Args:
nfo_paths: List of NFO file paths to validate
max_concurrent: Maximum concurrent validations
Returns:
Dictionary with validation results and statistics
"""
results = {
'total_files': len(nfo_paths),
'valid_files': 0,
'invalid_files': 0,
'missing_files': 0,
'validation_errors': [],
'file_results': {}
}
async def _validate_single_nfo(nfo_path: Path) -> Dict[str, Any]:
file_result = {
'path': str(nfo_path),
'exists': False,
'valid_xml': False,
'has_imdb_id': False,
'has_dates': False,
'error': None
}
try:
# Check if file exists
if not await async_file_exists(nfo_path):
file_result['error'] = 'File does not exist'
return file_result
file_result['exists'] = True
# Try to parse NFO
root = await async_read_nfo_file(nfo_path)
if root is None:
file_result['error'] = 'Failed to parse XML'
return file_result
file_result['valid_xml'] = True
# Check for IMDb ID
imdb_id = extract_imdb_from_nfo_content(root)
file_result['has_imdb_id'] = bool(imdb_id)
# Check for dates
dates = extract_dates_from_nfo(root)
file_result['has_dates'] = any(dates.values())
except Exception as e:
file_result['error'] = str(e)
return file_result
# Validate all files concurrently
semaphore = asyncio.Semaphore(max_concurrent)
async def _validate_with_semaphore(nfo_path: Path) -> Dict[str, Any]:
async with semaphore:
return await _validate_single_nfo(nfo_path)
tasks = [_validate_with_semaphore(nfo_path) for nfo_path in nfo_paths]
file_results = await asyncio.gather(*tasks, return_exceptions=True)
# Process results
for i, result in enumerate(file_results):
if isinstance(result, dict):
path_str = str(nfo_paths[i])
results['file_results'][path_str] = result
if not result['exists']:
results['missing_files'] += 1
elif result['valid_xml']:
results['valid_files'] += 1
else:
results['invalid_files'] += 1
if result.get('error'):
results['validation_errors'].append(f"{path_str}: {result['error']}")
return results
+160
View File
@@ -0,0 +1,160 @@
#!/usr/bin/env python3
"""
NFOGuard - Automated NFO file management for Radarr and Sonarr
Modular architecture with webhook processing and intelligent date handling
"""
import os
import sys
import signal
from pathlib import Path
from datetime import datetime, timezone
import uvicorn
from fastapi import FastAPI
# Import configuration first
from config.settings import config
from utils.logging import _log
# Import core components
from core.database import NFOGuardDatabase
from core.nfo_manager import NFOManager
from core.path_mapper import PathMapper
# Import clients
from clients.external_clients import ExternalClientManager
# Import processors
from processors.tv_processor import TVProcessor
from processors.movie_processor import MovieProcessor
# Import webhook handling
from webhooks.webhook_batcher import WebhookBatcher
# Import API routes
from api.routes import register_routes
def get_version() -> str:
"""Get application version"""
try:
version = (Path(__file__).parent / "VERSION").read_text().strip()
except:
version = "0.1.0"
# Check if running from dev branch (detect at runtime)
try:
# Try to read git branch from .git/HEAD
git_head_path = Path(__file__).parent / ".git" / "HEAD"
if git_head_path.exists():
head_content = git_head_path.read_text().strip()
if "ref: refs/heads/dev" in head_content:
version = f"{version}-dev"
elif head_content.startswith("ref: refs/heads/"):
# Extract branch name for other branches
branch = head_content.split("refs/heads/")[-1]
if branch != "main":
version = f"{version}-{branch}"
except Exception:
# If git detection fails, that's fine - use base version
pass
# Check for build source (only add -gitea for local Gitea builds)
build_source = os.environ.get("BUILD_SOURCE", "")
if build_source == "gitea":
if "gitea" not in version: # Don't double-add gitea suffix
version = f"{version}-gitea"
return version
def create_app() -> FastAPI:
"""Create and configure the FastAPI application"""
version = get_version()
app = FastAPI(
title="NFOGuard",
description="Webhook server for preserving media import dates",
version=version
)
return app
def initialize_components():
"""Initialize all application components"""
start_time = datetime.now(timezone.utc)
# Initialize core components
db = NFOGuardDatabase(config.db_path)
nfo_manager = NFOManager(config.manager_brand, config.debug)
path_mapper = PathMapper(config)
# Initialize processors
tv_processor = TVProcessor(db, nfo_manager, path_mapper)
movie_processor = MovieProcessor(db, nfo_manager, path_mapper)
# Initialize webhook batcher
batcher = WebhookBatcher()
batcher.set_processors(tv_processor, movie_processor)
return {
"db": db,
"nfo_manager": nfo_manager,
"path_mapper": path_mapper,
"tv_processor": tv_processor,
"movie_processor": movie_processor,
"batcher": batcher,
"start_time": start_time,
"config": config,
"version": get_version()
}
def signal_handler(signum, frame):
"""Handle shutdown signals gracefully"""
_log("INFO", f"Received signal {signum}, shutting down gracefully...")
sys.exit(0)
def main():
"""Main application entry point"""
# Register signal handlers for graceful shutdown
signal.signal(signal.SIGTERM, signal_handler)
signal.signal(signal.SIGINT, signal_handler)
version = get_version()
_log("INFO", "Starting NFOGuard")
_log("INFO", f"Version: {version}")
_log("INFO", f"TV paths: {[str(p) for p in config.tv_paths]}")
_log("INFO", f"Movie paths: {[str(p) for p in config.movie_paths]}")
_log("INFO", f"Database: {config.db_path}")
_log("INFO", f"Config: manage_nfo={config.manage_nfo}, fix_mtimes={config.fix_dir_mtimes}")
_log("INFO", f"Movie priority: {config.movie_priority}")
# Create FastAPI app
app = create_app()
# Initialize components
dependencies = initialize_components()
# Register routes
register_routes(app, dependencies)
try:
uvicorn.run(
app,
host="0.0.0.0",
port=int(os.environ.get("PORT", "8080")),
reload=False
)
except KeyboardInterrupt:
_log("INFO", "NFOGuard stopped by user")
except Exception as e:
_log("ERROR", f"NFOGuard crashed: {e}")
sys.exit(1)
if __name__ == "__main__":
main()
+538
View File
@@ -0,0 +1,538 @@
"""
Movie Processor for NFOGuard
Handles movie processing and metadata management
"""
import os
import re
import xml.etree.ElementTree as ET
from pathlib import Path
from typing import Optional, Dict, List, Tuple
from datetime import datetime, timezone
from zoneinfo import ZoneInfo
from core.database import NFOGuardDatabase
from core.nfo_manager import NFOManager
from core.path_mapper import PathMapper
from clients.radarr_client import RadarrClient
from clients.external_clients import ExternalClientManager
from config.settings import config
from utils.logging import _log
from utils.file_utils import find_media_path_by_imdb_and_title
def _get_local_timezone():
"""Get the local timezone, respecting TZ environment variable"""
tz_name = os.environ.get('TZ', 'UTC')
try:
# Try zoneinfo first (Python 3.9+)
return ZoneInfo(tz_name)
except ImportError:
# Fallback for older Python versions
try:
import pytz
return pytz.timezone(tz_name)
except:
# Final fallback to UTC
return timezone.utc
except:
# Final fallback to UTC
return timezone.utc
def convert_utc_to_local(utc_iso_string: str) -> str:
"""Convert UTC ISO timestamp to local timezone timestamp"""
if not utc_iso_string:
return utc_iso_string
try:
# Parse UTC timestamp
if utc_iso_string.endswith('Z'):
dt_utc = datetime.fromisoformat(utc_iso_string.replace('Z', '+00:00'))
elif '+00:00' in utc_iso_string:
dt_utc = datetime.fromisoformat(utc_iso_string)
else:
# Assume UTC if no timezone info
dt_utc = datetime.fromisoformat(utc_iso_string).replace(tzinfo=timezone.utc)
# Convert to local timezone
local_tz = _get_local_timezone()
dt_local = dt_utc.astimezone(local_tz)
return dt_local.isoformat(timespec='seconds')
except Exception:
# If conversion fails, return original
return utc_iso_string
class MovieProcessor:
"""Handles movie processing"""
def __init__(self, db: NFOGuardDatabase, nfo_manager: NFOManager, path_mapper: PathMapper):
self.db = db
self.nfo_manager = nfo_manager
self.path_mapper = path_mapper
self.radarr = RadarrClient(
os.environ.get("RADARR_URL", ""),
os.environ.get("RADARR_API_KEY", "")
)
self.external_clients = ExternalClientManager()
def find_movie_path(self, movie_title: str, imdb_id: str, radarr_path: str = None) -> Optional[Path]:
"""Find movie directory path using unified file utilities"""
return find_media_path_by_imdb_and_title(
title=movie_title,
imdb_id=imdb_id,
search_paths=config.movie_paths,
webhook_path=radarr_path,
path_mapper=self.path_mapper
)
def process_movie(self, movie_path: Path, webhook_mode: bool = False) -> None:
"""Process a movie directory"""
imdb_id = self.nfo_manager.find_movie_imdb_id(movie_path)
if not imdb_id:
_log("ERROR", f"No IMDb ID found in movie directory, filenames, or NFO file: {movie_path}")
return
# Handle TMDB ID fallback case
is_tmdb_fallback = imdb_id.startswith("tmdb-")
if is_tmdb_fallback:
_log("INFO", f"Processing movie: {movie_path.name} (TMDB: {imdb_id})")
else:
_log("INFO", f"Processing movie: {movie_path.name} (IMDb: {imdb_id})")
# Update database
self.db.upsert_movie(imdb_id, str(movie_path))
# Check for video files
video_exts = (".mkv", ".mp4", ".avi", ".mov", ".m4v")
has_video = any(f.is_file() and f.suffix.lower() in video_exts for f in movie_path.iterdir())
if not has_video:
_log("WARNING", f"No video files found in: {movie_path}")
self.db.upsert_movie_dates(imdb_id, None, None, None, False)
return
# TIER 1: Check if NFO file already has NFOGuard data (fastest - no DB or API calls)
nfo_path = movie_path / "movie.nfo"
nfo_data = self.nfo_manager.extract_nfoguard_dates_from_nfo(nfo_path)
if nfo_data:
_log("INFO", f"🚀 Using existing NFOGuard data from NFO file: {nfo_data['dateadded']} (source: {nfo_data['source']})")
dateadded = nfo_data["dateadded"]
source = nfo_data["source"]
released = nfo_data.get("released")
# Update file mtimes if enabled (NFO is already correct)
if config.fix_dir_mtimes and dateadded:
self.nfo_manager.update_movie_files_mtime(movie_path, dateadded)
_log("INFO", f"Completed processing movie: {movie_path.name} (source: {source}) [nfo-only]")
return
# TIER 1.5: Special handling for TMDB-only movies - extract dates from existing NFO
if is_tmdb_fallback:
tmdb_nfo_data = self._extract_dates_from_tmdb_nfo(nfo_path)
if tmdb_nfo_data:
_log("INFO", f"🎬 Using TMDB data from existing NFO file: {tmdb_nfo_data['dateadded']} (source: {tmdb_nfo_data['source']})")
dateadded = tmdb_nfo_data["dateadded"]
source = tmdb_nfo_data["source"]
released = tmdb_nfo_data.get("released")
# Create NFO with NFOGuard fields added
if config.manage_nfo:
self.nfo_manager.create_movie_nfo(
movie_path, imdb_id, dateadded, released, source, config.lock_metadata
)
# Update file mtimes if enabled
if config.fix_dir_mtimes and dateadded:
self.nfo_manager.update_movie_files_mtime(movie_path, dateadded)
# Save to database
self.db.upsert_movie_dates(imdb_id, released, dateadded, source, True)
_log("INFO", f"Completed processing movie: {movie_path.name} (source: {source}) [tmdb-nfo]")
return
# TIER 2: Check database for existing data
existing = self.db.get_movie_dates(imdb_id)
_log("DEBUG", f"Database lookup for {imdb_id}: {existing}")
# If we have complete data in database, use it and skip expensive API calls
if existing and existing.get("dateadded") and existing.get("source") != "no_valid_date_source":
_log("INFO", f"✅ Using complete database data for {imdb_id}: {existing['dateadded']} (source: {existing['source']})")
# Still create NFO and update files but skip API queries
dateadded, source, released = existing["dateadded"], existing["source"], existing.get("released")
# Create NFO with existing data
if config.manage_nfo:
self.nfo_manager.create_movie_nfo(
movie_path, imdb_id, dateadded, released, source, config.lock_metadata
)
# Update file mtimes if enabled
if config.fix_dir_mtimes and dateadded:
self.nfo_manager.update_movie_files_mtime(movie_path, dateadded)
_log("INFO", f"Completed processing movie: {movie_path.name} (source: {source}) [database-only]")
return
# Handle webhook mode - prioritize database, then use proper date logic
if webhook_mode:
_log("DEBUG", f"Webhook mode: existing={bool(existing)}, has_dateadded={bool(existing and existing.get('dateadded')) if existing else 'N/A'}")
if existing and existing.get("dateadded"):
_log("INFO", f"Webhook processing - using existing database entry: {existing['dateadded']} (source: {existing.get('source', 'unknown')})")
dateadded, source, released = existing["dateadded"], existing["source"], existing.get("released")
else:
if existing:
_log("INFO", f"Webhook processing - database entry exists but no dateadded field: {existing}")
else:
_log("INFO", f"Webhook processing - no database entry found for {imdb_id}")
_log("INFO", f"Using full date decision logic")
# Use same logic as manual scan to check Radarr import dates, release dates, etc.
should_query = True # Always query for webhooks when no database entry exists
dateadded, source, released = self._decide_movie_dates(imdb_id, movie_path, should_query, existing)
# Only if ALL date sources fail, fall back to current timestamp
if dateadded is None:
local_tz = _get_local_timezone()
current_time = datetime.now(local_tz).isoformat(timespec="seconds")
_log("INFO", f"Webhook processing - all date sources failed, using current timestamp as last resort: {current_time}")
dateadded, source = current_time, "webhook:fallback_timestamp"
else:
# Manual scan mode - determine if we should query APIs
should_query = (
config.movie_poll_mode == "always" or
(config.movie_poll_mode == "if_missing" and not existing) or
(config.movie_poll_mode == "if_missing" and existing and existing.get("source") == "file:mtime") or
(config.movie_poll_mode == "if_missing" and existing and not existing.get("dateadded"))
)
_log("DEBUG", f"Movie {imdb_id}: should_query={should_query}, poll_mode={config.movie_poll_mode}, existing={bool(existing)}, has_dateadded={bool(existing and existing.get('dateadded')) if existing else False}")
# Use existing movie date decision logic
dateadded, source, released = self._decide_movie_dates(imdb_id, movie_path, should_query, existing)
# If we don't have an import/download date but we have a release date, use it as dateadded
# This ensures we save digital release dates, theatrical dates, etc. to the database
final_dateadded = dateadded
final_source = source
if dateadded is None and released is not None:
final_dateadded = released
final_source = f"{source}_as_dateadded" if source else "release_date_fallback"
_log("INFO", f"Using release date as dateadded: {final_dateadded} (source: {final_source})")
# Create NFO regardless of date availability (preserves existing metadata)
if config.manage_nfo:
self.nfo_manager.create_movie_nfo(
movie_path, imdb_id, final_dateadded, released, final_source, config.lock_metadata
)
# Skip remaining processing if no valid date found and file dates disabled
if final_dateadded is None:
_log("WARNING", f"Movie {movie_path.name} - no valid date source available, but NFO was still processed")
self.db.upsert_movie_dates(imdb_id, released, None, source, True)
return
# Update dateadded and source for the rest of processing
dateadded = final_dateadded
source = final_source
_log("DEBUG", f"Movie {movie_path.name} proceeding to save: dateadded={dateadded}, source={source}")
# Update file mtimes (only if we have a valid date)
if config.fix_dir_mtimes and dateadded and dateadded != "MANUAL_REVIEW_NEEDED":
self.nfo_manager.update_movie_files_mtime(movie_path, dateadded)
_log("DEBUG", f"Movie processing reached file mtime section: fix_dir_mtimes={config.fix_dir_mtimes}, dateadded={dateadded}")
# Save to database
_log("DEBUG", f"About to save to database: imdb_id={imdb_id}, dateadded={dateadded}")
try:
self.db.upsert_movie_dates(imdb_id, released, dateadded, source, True)
_log("DEBUG", f"Database save completed for {imdb_id}")
except Exception as e:
_log("ERROR", f"Database save failed for {imdb_id}: {e}")
raise
_log("INFO", f"Completed processing movie: {movie_path.name} (source: {source})")
def _extract_dates_from_tmdb_nfo(self, nfo_path: Path) -> Optional[Dict[str, str]]:
"""Extract date information from TMDB-based NFO file"""
if not nfo_path.exists():
return None
try:
root = self.nfo_manager._parse_nfo_with_tolerance(nfo_path)
# Look for premiered date (from TMDB)
premiered_elem = root.find('.//premiered')
if premiered_elem is not None and premiered_elem.text:
premiered_date = premiered_elem.text.strip()
print(f"✅ Found TMDB premiered date: {premiered_date}")
return {
"dateadded": premiered_date,
"source": "tmdb:premiered_from_nfo",
"released": premiered_date
}
except (ET.ParseError, Exception) as e:
print(f"⚠️ Error parsing TMDB NFO for dates: {e}")
return None
def _decide_movie_dates(self, imdb_id: str, movie_path: Path, should_query: bool, existing: Optional[Dict]) -> Tuple[str, str, Optional[str]]:
"""Decide movie dates based on configuration and available data"""
_log("DEBUG", f"_decide_movie_dates for {imdb_id}: should_query={should_query}, existing={existing}")
if not should_query and existing:
_log("DEBUG", f"Using existing data without querying: dateadded={existing.get('dateadded')}, source={existing.get('source')}")
return existing["dateadded"], existing["source"], existing.get("released")
# Query Radarr for movie info
radarr_movie = None
if should_query and self.radarr.api_key:
radarr_movie = self.radarr.movie_by_imdb(imdb_id)
released = None
if radarr_movie:
released = self._parse_date_to_iso(radarr_movie.get("inCinemas"))
# Try import history first if configured
if config.movie_priority == "import_then_digital":
import_date, import_source = None, None
if radarr_movie:
movie_id = radarr_movie.get("id")
if movie_id:
import_date, import_source = self.radarr.get_movie_import_date(movie_id, fallback_to_file_date=config.allow_file_date_fallback)
_log("INFO", f"Movie {imdb_id}: Radarr import result: date={import_date}, source={import_source}")
# Check for special case: rename-first scenario (should prefer release dates)
if import_source == "radarr:db.prefer_release_dates":
_log("INFO", f"🎯 Movie {imdb_id} has rename-first history - skipping import, preferring release dates")
# Fall through to release date logic below
# Check if we got a real import date or just file date fallback
elif import_date and import_source != "radarr:db.file.dateAdded":
# Convert import date to local timezone for NFO files
local_import_date = convert_utc_to_local(import_date)
_log("INFO", f"✅ Movie {imdb_id}: Using import date {local_import_date} from {import_source}")
return local_import_date, import_source, released
# Get digital release date for comparison/fallback
_log("INFO", f"🔍 Movie {imdb_id}: Trying digital release date fallback...")
digital_date, digital_source = self._get_digital_release_date(imdb_id)
_log("INFO", f"Movie {imdb_id}: Digital release result: date={digital_date}, source={digital_source}")
# If we only have file date and release date exists, prefer it if reasonable and enabled
if import_date and import_source == "radarr:db.file.dateAdded" and digital_date and config.prefer_release_dates_over_file_dates:
# Compare dates - prefer release date if it's reasonable
if self._should_prefer_release_over_file_date(digital_date, digital_source, released, imdb_id):
_log("INFO", f"✅ Movie {imdb_id}: Preferring digital release date {digital_date} over file date")
return digital_date, digital_source, released
else:
# Convert file date to local timezone for NFO files
local_file_date = convert_utc_to_local(import_date)
_log("INFO", f"✅ Movie {imdb_id}: Keeping file date {local_file_date} - digital date not reasonable")
return local_file_date, import_source, released
# Use whichever we have
if import_date:
# Convert import date to local timezone for NFO files
local_import_date = convert_utc_to_local(import_date)
_log("INFO", f"✅ Movie {imdb_id}: Using import date {local_import_date} from {import_source}")
return local_import_date, import_source, released
elif digital_date:
_log("INFO", f"✅ Movie {imdb_id}: Using digital release date {digital_date} from {digital_source}")
return digital_date, digital_source, released
else:
_log("WARNING", f"⚠️ Movie {imdb_id}: No import date OR digital release date found - trying additional fallbacks")
# Try Radarr's own NFO premiered date as fallback
radarr_premiered = self._get_radarr_nfo_premiered_date(movie_path)
if radarr_premiered:
_log("INFO", f"✅ Movie {imdb_id}: Using Radarr NFO premiered date {radarr_premiered}")
return radarr_premiered, "radarr:nfo.premiered", released
else: # digital_then_import
# Try digital release first
digital_date, digital_source = self._get_digital_release_date(imdb_id)
if digital_date:
return digital_date, digital_source, released
# Fall back to import history
if radarr_movie:
movie_id = radarr_movie.get("id")
if movie_id:
import_date, import_source = self.radarr.get_movie_import_date(movie_id, fallback_to_file_date=config.allow_file_date_fallback)
if import_date:
# Convert import date to local timezone for NFO files
local_import_date = convert_utc_to_local(import_date)
return local_import_date, import_source, released
# Last resort: file mtime (if allowed)
if config.allow_file_date_fallback:
return self._get_file_mtime_date(movie_path)
else:
_log("INFO", f"No valid dates found for {imdb_id} and file date fallback disabled - skipping NFO creation")
# Log to failed movies debug file for troubleshooting
self._log_failed_movie(movie_path, imdb_id, "No import date, no release date, file date fallback disabled")
return None, "no_valid_date_source", None
def _get_digital_release_date(self, imdb_id: str) -> Tuple[Optional[str], str]:
"""Get release date from external sources using configured priority"""
_log("INFO", f"🔍 Calling external clients for {imdb_id}")
_log("INFO", f"Release date priority: {config.release_date_priority}")
_log("INFO", f"Smart validation enabled: {config.enable_smart_date_validation}")
try:
release_result = self.external_clients.get_release_date_by_priority(
imdb_id,
config.release_date_priority,
enable_smart_validation=config.enable_smart_date_validation
)
_log("INFO", f"External clients result for {imdb_id}: {release_result}")
if release_result:
_log("INFO", f"✅ Got release date: {release_result[0]} from {release_result[1]}")
return release_result[0], release_result[1]
else:
_log("WARNING", f"❌ No release date found from external clients for {imdb_id}")
return None, "release:none"
except Exception as e:
_log("ERROR", f"External clients error for {imdb_id}: {e}")
return None, f"release:error:{str(e)}"
def _get_radarr_nfo_premiered_date(self, movie_path: Path) -> Optional[str]:
"""Extract premiered date from Radarr's existing movie.nfo file"""
try:
nfo_path = movie_path / "movie.nfo"
if not nfo_path.exists():
_log("DEBUG", f"No existing NFO file found at {nfo_path}")
return None
nfo_content = nfo_path.read_text(encoding='utf-8')
# Look for <premiered>YYYY-MM-DD</premiered>
match = re.search(r'<premiered>(\d{4}-\d{2}-\d{2})</premiered>', nfo_content)
if match:
premiered_date = match.group(1)
# Convert to ISO format with timezone
iso_date = f"{premiered_date}T00:00:00+00:00"
_log("INFO", f"✅ Found Radarr NFO premiered date: {premiered_date}")
return iso_date
else:
_log("DEBUG", f"No <premiered> tag found in existing NFO")
return None
except Exception as e:
_log("ERROR", f"Error reading Radarr NFO file: {e}")
return None
def _log_failed_movie(self, movie_path: Path, imdb_id: str, reason: str, available_countries: List[str] = None):
"""Log movies that failed to get valid dates to a debug file"""
try:
log_dir = Path("logs")
log_dir.mkdir(exist_ok=True)
failed_log_path = log_dir / "failed_movies.log"
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
log_entry = f"[{timestamp}] {movie_path.name} | IMDb: {imdb_id} | Reason: {reason}"
if available_countries:
log_entry += f" | Available Countries: {', '.join(available_countries)}"
log_entry += "\n"
with open(failed_log_path, "a", encoding="utf-8") as f:
f.write(log_entry)
_log("INFO", f"📝 Logged failed movie to {failed_log_path}: {movie_path.name}")
except Exception as e:
_log("ERROR", f"Failed to write to failed movies log: {e}")
def _get_file_mtime_date(self, movie_path: Path) -> Tuple[str, str, Optional[str]]:
"""Get date from file modification time as last resort"""
video_exts = (".mkv", ".mp4", ".avi", ".mov", ".m4v")
newest_mtime = None
for file_path in movie_path.iterdir():
if file_path.is_file() and file_path.suffix.lower() in video_exts:
try:
mtime = file_path.stat().st_mtime
if newest_mtime is None or mtime > newest_mtime:
newest_mtime = mtime
except Exception:
continue
if newest_mtime:
try:
# Use local timezone for file modification times
local_tz = _get_local_timezone()
iso_date = datetime.fromtimestamp(newest_mtime, tz=local_tz).isoformat(timespec="seconds")
return iso_date, "file:mtime", None
except Exception:
pass
return "MANUAL_REVIEW_NEEDED", "manual_review_required", None
def _should_prefer_release_over_file_date(self, release_date: str, release_source: str, theatrical_release: Optional[str], imdb_id: str) -> bool:
"""
Decide if release date should be preferred over file date
Logic:
- For theatrical dates: Always prefer over file dates (they're authoritative)
- For physical dates: Usually prefer over file dates
- For digital dates: Prefer if reasonable (not decades before theatrical)
"""
try:
release_dt = datetime.fromisoformat(release_date.replace("Z", "+00:00"))
# Always prefer theatrical and physical releases over file dates
if any(release_type in release_source for release_type in ["theatrical", "physical"]):
_log("INFO", f"Release date {release_date} ({release_source}) for {imdb_id}, preferring over file date")
return True
# If we have theatrical release date, compare digital against it
if theatrical_release:
theatrical_dt = datetime.fromisoformat(theatrical_release.replace("Z", "+00:00"))
year_diff = release_dt.year - theatrical_dt.year
# If digital is more than 10 years before theatrical, it's probably wrong
if year_diff < -10:
_log("INFO", f"Release date {release_date} is {abs(year_diff)} years before theatrical {theatrical_release} for {imdb_id}, using file date instead")
return False
# If digital is within reasonable range (theatrical to +20 years), use it
if -2 <= year_diff <= 20:
_log("INFO", f"Release date {release_date} is reasonable for {imdb_id} (theatrical: {theatrical_release}), preferring over file date")
return True
# If no theatrical date, use digital if it's not absurdly old
if release_dt.year >= 1990: # Reasonable minimum for digital releases
_log("INFO", f"Release date {release_date} seems reasonable for {imdb_id}, preferring over file date")
return True
_log("INFO", f"Release date {release_date} seems too old for {imdb_id}, using file date instead")
return False
except Exception as e:
_log("WARNING", f"Error comparing dates for {imdb_id}: {e}")
return False
def _parse_date_to_iso(self, date_str: str) -> Optional[str]:
"""Parse date string to ISO format"""
if not date_str:
return None
try:
if len(date_str) == 10 and date_str[4] == "-":
dt = datetime.fromisoformat(date_str).replace(tzinfo=timezone.utc)
else:
dt = datetime.fromisoformat(date_str.replace("Z", "+00:00")).astimezone(timezone.utc)
return dt.isoformat(timespec="seconds")
except Exception:
return None
+513
View File
@@ -0,0 +1,513 @@
"""
TV Series Processor for NFOGuard
Handles TV series processing and episode management with async I/O support
"""
import os
import re
import asyncio
from pathlib import Path
from typing import Optional, Dict, List, Set, Tuple, Any
from datetime import datetime
from core.database import NFOGuardDatabase
from core.nfo_manager import NFOManager
from core.async_nfo_manager import AsyncNFOManager
from core.path_mapper import PathMapper
from clients.sonarr_client import SonarrClient
from clients.external_clients import ExternalClientManager
from config.settings import config
from utils.logging import _log
from utils.file_utils import (
find_media_path_by_imdb_and_title,
find_episodes_on_disk,
extract_title_from_directory_name
)
from utils.async_file_utils import (
async_find_episodes_on_disk,
async_concurrent_episode_processing
)
class TVProcessor:
"""Handles TV series processing"""
def __init__(self, db: NFOGuardDatabase, nfo_manager: NFOManager, path_mapper: PathMapper):
self.db = db
self.nfo_manager = nfo_manager
self.async_nfo_manager = AsyncNFOManager(config.manager_brand, config.debug)
self.path_mapper = path_mapper
self.sonarr = SonarrClient(
os.environ.get("SONARR_URL", ""),
os.environ.get("SONARR_API_KEY", "")
)
self.external_clients = ExternalClientManager()
def find_series_path(self, series_title: str, imdb_id: str, sonarr_path: str = None) -> Optional[Path]:
"""Find series directory path using unified file utilities"""
return find_media_path_by_imdb_and_title(
title=series_title,
imdb_id=imdb_id,
search_paths=config.tv_paths,
webhook_path=sonarr_path,
path_mapper=self.path_mapper
)
def process_series(self, series_path: Path) -> None:
"""Process a TV series directory"""
imdb_id = self.nfo_manager.parse_imdb_from_path(series_path)
if not imdb_id:
_log("ERROR", f"No IMDb ID found in series path: {series_path}")
return
_log("INFO", f"Processing TV series: {series_path.name}")
# Update database
self.db.upsert_series(imdb_id, str(series_path))
# Find video files
disk_episodes = find_episodes_on_disk(series_path)
_log("INFO", f"Found {len(disk_episodes)} episodes on disk")
# Get episode dates
episode_dates = self._gather_episode_dates(series_path, imdb_id, disk_episodes)
# Process episodes
for (season, episode), (aired, dateadded, source) in episode_dates.items():
if (season, episode) in disk_episodes:
# Create NFO
if config.manage_nfo:
season_dir = series_path / config.tv_season_dir_format.format(season=season)
self.nfo_manager.create_episode_nfo(
season_dir,
season, episode, aired, dateadded, source, config.lock_metadata
)
# Update file mtimes
if config.fix_dir_mtimes and dateadded:
video_files = disk_episodes[(season, episode)]
for video_file in video_files:
self.nfo_manager.set_file_mtime(video_file, dateadded)
# Save to database
self.db.upsert_episode_date(imdb_id, season, episode, aired, dateadded, source, True)
# Skip season.nfo and tvshow.nfo creation - focus only on episode NFOs
pass
_log("INFO", f"Completed processing TV series: {series_path.name}")
def _extract_series_title_from_path(self, series_path: Path) -> Optional[str]:
"""Extract series title from directory path using unified file utilities"""
return extract_title_from_directory_name(series_path.name)
def _gather_episode_dates(self, series_path: Path, imdb_id: str, disk_episodes: Dict[Tuple[int, int], List[Path]]) -> Dict[Tuple[int, int], Tuple[Optional[str], Optional[str], str]]:
"""Gather episode air dates and date added information"""
episode_dates = {}
# Get data from Sonarr first if available
sonarr_episodes = self._get_sonarr_episodes(imdb_id)
# Get data from external sources for missing information
for (season, episode) in disk_episodes:
aired = None
dateadded = None
source = "unknown"
# Try Sonarr first
if (season, episode) in sonarr_episodes:
sonarr_data = sonarr_episodes[(season, episode)]
aired = sonarr_data.get('airDate')
dateadded = sonarr_data.get('dateAdded')
if dateadded:
source = "sonarr"
# Fallback to external sources if needed
if not aired:
external_aired = self.external_clients.get_episode_air_date(imdb_id, season, episode)
if external_aired:
aired = external_aired
if not dateadded:
source = "external"
# Use air date as fallback for dateadded if available
if not dateadded and aired:
dateadded = aired
source = f"{source}_fallback" if source != "unknown" else "aired_fallback"
episode_dates[(season, episode)] = (aired, dateadded, source)
return episode_dates
def _get_sonarr_episodes(self, imdb_id: str) -> Dict[Tuple[int, int], Dict[str, Any]]:
"""Get episode information from Sonarr"""
try:
series_data = self.sonarr.get_series_by_imdb(imdb_id)
if not series_data:
return {}
series_id = series_data.get('id')
if not series_id:
return {}
episodes = self.sonarr.get_episodes(series_id)
episode_map = {}
for episode in episodes:
season = episode.get('seasonNumber', 0)
episode_num = episode.get('episodeNumber', 0)
if season > 0 and episode_num > 0:
episode_map[(season, episode_num)] = {
'airDate': episode.get('airDate'),
'dateAdded': episode.get('episodeFile', {}).get('dateAdded') if episode.get('hasFile') else None
}
return episode_map
except Exception as e:
_log("ERROR", f"Failed to get Sonarr episodes for {imdb_id}: {e}")
return {}
def process_season(self, series_path: str, season_name: str) -> Dict[str, Any]:
"""Process a specific season"""
series_path_obj = Path(series_path)
if not series_path_obj.exists():
raise FileNotFoundError(f"Series path not found: {series_path}")
season_path = series_path_obj / season_name
if not season_path.exists():
raise FileNotFoundError(f"Season directory not found: {season_path}")
# Extract season number from directory name
season_match = re.search(r'(\d+)', season_name)
if not season_match:
raise ValueError(f"Could not extract season number from: {season_name}")
season_num = int(season_match.group(1))
# Get series IMDb ID
imdb_id = self.nfo_manager.parse_imdb_from_path(series_path_obj)
if not imdb_id:
raise ValueError(f"No IMDb ID found in series path: {series_path}")
_log("INFO", f"Processing season {season_num} of series: {series_path_obj.name}")
# Find episodes in this season
disk_episodes = find_episodes_on_disk(series_path_obj)
season_episodes = {k: v for k, v in disk_episodes.items() if k[0] == season_num}
if not season_episodes:
return {"status": "no_episodes", "season": season_num, "episodes_found": 0}
# Get episode dates
episode_dates = self._gather_episode_dates(series_path_obj, imdb_id, season_episodes)
# Process episodes
processed_count = 0
for (season, episode), (aired, dateadded, source) in episode_dates.items():
if (season, episode) in season_episodes:
# Create NFO
if config.manage_nfo:
self.nfo_manager.create_episode_nfo(
season_path,
season, episode, aired, dateadded, source, config.lock_metadata
)
# Update file mtimes
if config.fix_dir_mtimes and dateadded:
video_files = season_episodes[(season, episode)]
for video_file in video_files:
self.nfo_manager.set_file_mtime(video_file, dateadded)
# Save to database
self.db.upsert_episode_date(imdb_id, season, episode, aired, dateadded, source, True)
processed_count += 1
_log("INFO", f"Processed {processed_count} episodes in season {season_num}")
return {
"status": "success",
"season": season_num,
"episodes_found": len(season_episodes),
"episodes_processed": processed_count
}
def process_episode(self, series_path: str, season_name: str, episode_name: str) -> Dict[str, Any]:
"""Process a specific episode"""
series_path_obj = Path(series_path)
if not series_path_obj.exists():
raise FileNotFoundError(f"Series path not found: {series_path}")
season_path = series_path_obj / season_name
if not season_path.exists():
raise FileNotFoundError(f"Season directory not found: {season_path}")
episode_path = season_path / episode_name
if not episode_path.exists():
raise FileNotFoundError(f"Episode file not found: {episode_path}")
# Extract season and episode numbers
season_match = re.search(r'(\d+)', season_name)
episode_match = re.search(r'[sS](\d+)[eE](\d+)|(\d+)x(\d+)', episode_name)
if not season_match:
raise ValueError(f"Could not extract season number from: {season_name}")
if not episode_match:
raise ValueError(f"Could not extract episode number from: {episode_name}")
season_num = int(season_match.group(1))
if episode_match.group(1) and episode_match.group(2): # SxxExx format
episode_num = int(episode_match.group(2))
elif episode_match.group(3) and episode_match.group(4): # NxNN format
episode_num = int(episode_match.group(4))
else:
raise ValueError(f"Could not parse episode number from: {episode_name}")
# Get series IMDb ID
imdb_id = self.nfo_manager.parse_imdb_from_path(series_path_obj)
if not imdb_id:
raise ValueError(f"No IMDb ID found in series path: {series_path}")
_log("INFO", f"Processing episode S{season_num:02d}E{episode_num:02d} of series: {series_path_obj.name}")
# Get episode data
disk_episodes = {(season_num, episode_num): [episode_path]}
episode_dates = self._gather_episode_dates(series_path_obj, imdb_id, disk_episodes)
if (season_num, episode_num) not in episode_dates:
return {"status": "no_data", "season": season_num, "episode": episode_num}
aired, dateadded, source = episode_dates[(season_num, episode_num)]
# Create NFO
if config.manage_nfo:
self.nfo_manager.create_episode_nfo(
season_path,
season_num, episode_num, aired, dateadded, source, config.lock_metadata
)
# Update file mtime
if config.fix_dir_mtimes and dateadded:
self.nfo_manager.set_file_mtime(episode_path, dateadded)
# Save to database
self.db.upsert_episode_date(imdb_id, season_num, episode_num, aired, dateadded, source, True)
_log("INFO", f"Processed episode S{season_num:02d}E{episode_num:02d}")
return {
"status": "success",
"season": season_num,
"episode": episode_num,
"aired": aired,
"dateadded": dateadded,
"source": source
}
# ===== ASYNC METHODS =====
async def async_process_series(self, series_path: Path) -> Dict[str, Any]:
"""
Async process a TV series directory with concurrent episode processing
Args:
series_path: Path to series directory
Returns:
Dictionary with processing results
"""
imdb_id = self.nfo_manager.parse_imdb_from_path(series_path)
if not imdb_id:
return {"status": "error", "reason": f"No IMDb ID found in series path: {series_path}"}
_log("INFO", f"Async processing TV series: {series_path.name}")
# Update database
self.db.upsert_series(imdb_id, str(series_path))
# Find video files asynchronously
disk_episodes = await async_find_episodes_on_disk(series_path)
_log("INFO", f"Found {len(disk_episodes)} episodes on disk")
# Get episode dates (sync for now, could be made async later)
episode_dates = self._gather_episode_dates(series_path, imdb_id, disk_episodes)
# Prepare episode data for concurrent processing
episode_data_list = []
mtime_operations = []
for (season, episode), (aired, dateadded, source) in episode_dates.items():
if (season, episode) in disk_episodes:
season_dir = series_path / config.tv_season_dir_format.format(season=season)
# Prepare NFO creation data
if config.manage_nfo:
episode_data_list.append({
'season_dir': season_dir,
'season': season,
'episode': episode,
'aired': aired,
'dateadded': dateadded,
'source': source,
'lock_metadata': config.lock_metadata
})
# Prepare mtime operations
if config.fix_dir_mtimes and dateadded:
video_files = disk_episodes[(season, episode)]
for video_file in video_files:
mtime_operations.append((video_file, dateadded))
# Save to database
self.db.upsert_episode_date(imdb_id, season, episode, aired, dateadded, source, True)
# Process NFOs and mtimes concurrently
results = {}
if episode_data_list:
_log("INFO", f"Creating {len(episode_data_list)} episode NFOs concurrently")
nfo_results = await self.async_nfo_manager.async_batch_create_episode_nfos(
episode_data_list,
max_concurrent=config.max_concurrent
)
results['nfo_created'] = sum(nfo_results)
results['nfo_failed'] = len(nfo_results) - sum(nfo_results)
if mtime_operations:
_log("INFO", f"Setting mtimes for {len(mtime_operations)} files concurrently")
mtime_results = await self.async_nfo_manager.async_batch_set_file_mtimes(
mtime_operations,
max_concurrent=10
)
results['mtime_updated'] = sum(mtime_results)
results['mtime_failed'] = len(mtime_results) - sum(mtime_results)
_log("INFO", f"Completed async processing TV series: {series_path.name}")
return {
"status": "success",
"imdb_id": imdb_id,
"episodes_found": len(disk_episodes),
"episodes_processed": len(episode_data_list),
"results": results
}
async def async_process_multiple_series(
self,
series_paths: List[Path],
max_concurrent: int = 2
) -> List[Dict[str, Any]]:
"""
Process multiple TV series concurrently
Args:
series_paths: List of series directory paths
max_concurrent: Maximum concurrent series processing
Returns:
List of processing results for each series
"""
semaphore = asyncio.Semaphore(max_concurrent)
async def _process_series_with_semaphore(series_path: Path) -> Dict[str, Any]:
async with semaphore:
try:
return await self.async_process_series(series_path)
except Exception as e:
_log("ERROR", f"Failed to process series {series_path}: {e}")
return {"status": "error", "path": str(series_path), "reason": str(e)}
_log("INFO", f"Processing {len(series_paths)} series with max {max_concurrent} concurrent")
tasks = [_process_series_with_semaphore(path) for path in series_paths]
results = await asyncio.gather(*tasks, return_exceptions=True)
# Filter out exceptions and convert to proper results
processed_results = []
for i, result in enumerate(results):
if isinstance(result, Exception):
processed_results.append({
"status": "error",
"path": str(series_paths[i]),
"reason": str(result)
})
else:
processed_results.append(result)
return processed_results
async def async_batch_episode_processing(
self,
episodes_data: List[Dict[str, Any]],
max_concurrent: int = 5
) -> Dict[str, Any]:
"""
Process episodes from webhook data concurrently
Args:
episodes_data: List of episode data from webhooks
max_concurrent: Maximum concurrent episode processing
Returns:
Processing results summary
"""
async def _process_single_episode(episode_data: Dict[str, Any]) -> Dict[str, Any]:
try:
# Extract episode information
series_path = Path(episode_data.get('series_path'))
season = episode_data.get('season')
episode = episode_data.get('episode')
aired = episode_data.get('aired')
dateadded = episode_data.get('dateadded')
# Get IMDb ID
imdb_id = self.nfo_manager.parse_imdb_from_path(series_path)
if not imdb_id:
return {"status": "error", "reason": "No IMDb ID found"}
# Create NFO if needed
nfo_success = True
if config.manage_nfo:
season_dir = series_path / config.tv_season_dir_format.format(season=season)
nfo_success = await self.async_nfo_manager.async_create_episode_nfo(
season_dir, season, episode, aired, dateadded, "webhook", config.lock_metadata
)
# Update database
self.db.upsert_episode_date(imdb_id, season, episode, aired, dateadded, "webhook", True)
return {
"status": "success",
"season": season,
"episode": episode,
"nfo_created": nfo_success
}
except Exception as e:
return {
"status": "error",
"episode_data": episode_data,
"reason": str(e)
}
_log("INFO", f"Processing {len(episodes_data)} episodes concurrently")
results = await async_concurrent_episode_processing(
episodes_data,
_process_single_episode,
max_concurrent
)
# Summarize results
successful = sum(1 for r in results if r and r.get("status") == "success")
failed = len(results) - successful
return {
"total_episodes": len(episodes_data),
"successful": successful,
"failed": failed,
"results": results
}
View File
+468
View File
@@ -0,0 +1,468 @@
"""
Async file utilities for NFOGuard
High-performance async file operations with concurrent processing
"""
import asyncio
import aiofiles
import aiofiles.os
from pathlib import Path
from typing import List, Dict, Optional, Tuple, Any, Set
import xml.etree.ElementTree as ET
from datetime import datetime
from utils.logging import _log
from utils.exceptions import FileOperationError, NFOCreationError
from utils.file_utils import VIDEO_EXTENSIONS, extract_episode_info, extract_imdb_id_from_path
from utils.nfo_patterns import parse_nfo_with_tolerance, write_nfo_file
async def async_read_text_file(file_path: Path, encoding: str = 'utf-8') -> Optional[str]:
"""
Async read text file with error handling
Args:
file_path: Path to file to read
encoding: Text encoding (default: utf-8)
Returns:
File content as string or None if error
"""
try:
async with aiofiles.open(file_path, 'r', encoding=encoding, errors='ignore') as f:
return await f.read()
except Exception as e:
_log("WARNING", f"Failed to read file {file_path}: {e}")
return None
async def async_write_text_file(file_path: Path, content: str, encoding: str = 'utf-8') -> bool:
"""
Async write text file with error handling
Args:
file_path: Path to file to write
content: Content to write
encoding: Text encoding (default: utf-8)
Returns:
True if successful, False otherwise
"""
try:
# Ensure parent directory exists
await aiofiles.os.makedirs(file_path.parent, exist_ok=True)
async with aiofiles.open(file_path, 'w', encoding=encoding) as f:
await f.write(content)
return True
except Exception as e:
_log("ERROR", f"Failed to write file {file_path}: {e}")
return False
async def async_file_exists(file_path: Path) -> bool:
"""
Async check if file exists
Args:
file_path: Path to check
Returns:
True if file exists, False otherwise
"""
try:
return await aiofiles.os.path.exists(file_path)
except Exception:
return False
async def async_get_file_mtime(file_path: Path) -> Optional[float]:
"""
Async get file modification time
Args:
file_path: Path to file
Returns:
Modification time as timestamp or None if error
"""
try:
stat_result = await aiofiles.os.stat(file_path)
return stat_result.st_mtime
except Exception:
return None
async def async_set_file_mtime(file_path: Path, mtime: float) -> bool:
"""
Async set file modification time
Args:
file_path: Path to file
mtime: New modification time as timestamp
Returns:
True if successful, False otherwise
"""
try:
await aiofiles.os.utime(file_path, (mtime, mtime))
return True
except Exception as e:
_log("WARNING", f"Failed to set mtime for {file_path}: {e}")
return False
async def async_find_video_files(directory: Path, recursive: bool = True) -> List[Path]:
"""
Async find all video files in a directory
Args:
directory: Directory to search
recursive: Whether to search recursively
Returns:
List of video file paths
"""
if not await async_file_exists(directory):
return []
video_files = []
try:
if recursive:
# Use os.walk equivalent for async
async def _walk_directory(path: Path):
try:
entries = await aiofiles.os.listdir(path)
for entry in entries:
entry_path = path / entry
if await aiofiles.os.path.isfile(entry_path):
if entry_path.suffix.lower() in VIDEO_EXTENSIONS:
video_files.append(entry_path)
elif await aiofiles.os.path.isdir(entry_path):
await _walk_directory(entry_path)
except Exception as e:
_log("WARNING", f"Failed to scan directory {path}: {e}")
await _walk_directory(directory)
else:
# Non-recursive scan
try:
entries = await aiofiles.os.listdir(directory)
for entry in entries:
entry_path = directory / entry
if await aiofiles.os.path.isfile(entry_path):
if entry_path.suffix.lower() in VIDEO_EXTENSIONS:
video_files.append(entry_path)
except Exception as e:
_log("WARNING", f"Failed to scan directory {directory}: {e}")
except Exception as e:
_log("ERROR", f"Failed to find video files in {directory}: {e}")
return video_files
async def async_find_episodes_on_disk(series_path: Path) -> Dict[Tuple[int, int], List[Path]]:
"""
Async find all episodes on disk with concurrent processing
Args:
series_path: Path to series directory
Returns:
Dictionary mapping (season, episode) tuples to lists of video files
"""
episodes = {}
if not await async_file_exists(series_path):
return episodes
# Get all video files concurrently
video_files = await async_find_video_files(series_path, recursive=True)
# Process files to extract episode information
for video_file in video_files:
episode_info = extract_episode_info(video_file.name)
if episode_info:
season, episode = episode_info["season"], episode_info["episode"]
key = (season, episode)
if key not in episodes:
episodes[key] = []
episodes[key].append(video_file)
return episodes
async def async_read_nfo_file(nfo_path: Path) -> Optional[ET.Element]:
"""
Async read and parse NFO file
Args:
nfo_path: Path to NFO file
Returns:
XML root element if successful, None otherwise
"""
if not await async_file_exists(nfo_path):
return None
try:
content = await async_read_text_file(nfo_path)
if not content:
return None
# Parse XML content
try:
root = ET.fromstring(content)
return root
except ET.ParseError:
# Try with tolerance (sync operation for now)
return parse_nfo_with_tolerance(nfo_path)
except Exception as e:
_log("ERROR", f"Failed to read NFO file {nfo_path}: {e}")
return None
async def async_write_nfo_file(
nfo_path: Path,
root: ET.Element,
lock_metadata: bool = True
) -> bool:
"""
Async write NFO XML content to file
Args:
nfo_path: Path where to write the NFO file
root: XML root element to write
lock_metadata: Whether to add file locking attributes
Returns:
True if successful, False otherwise
"""
try:
# Ensure parent directory exists
await aiofiles.os.makedirs(nfo_path.parent, exist_ok=True)
# Add file locking if requested
if lock_metadata:
root.set('nfoguard_managed', 'true')
root.set('last_updated', datetime.now().isoformat())
# Create tree and format
tree = ET.ElementTree(root)
ET.indent(tree, space=" ", level=0) # Pretty formatting
# Convert to string
xml_str = ET.tostring(root, encoding='unicode', xml_declaration=False)
xml_content = f'<?xml version="1.0" encoding="utf-8"?>\n{xml_str}'
# Write asynchronously
success = await async_write_text_file(nfo_path, xml_content)
if success:
_log("DEBUG", f"Successfully wrote NFO file: {nfo_path}")
return success
except Exception as e:
_log("ERROR", f"Failed to write NFO file {nfo_path}: {e}")
return False
async def async_batch_process_files(
file_paths: List[Path],
process_func,
max_concurrent: int = 10,
progress_callback: Optional[callable] = None
) -> List[Any]:
"""
Process multiple files concurrently with controlled concurrency
Args:
file_paths: List of file paths to process
process_func: Async function to process each file
max_concurrent: Maximum number of concurrent operations
progress_callback: Optional callback for progress updates
Returns:
List of results from processing each file
"""
semaphore = asyncio.Semaphore(max_concurrent)
results = []
async def _process_with_semaphore(file_path: Path, index: int) -> Any:
async with semaphore:
try:
result = await process_func(file_path)
if progress_callback:
progress_callback(index + 1, len(file_paths), file_path)
return result
except Exception as e:
_log("ERROR", f"Failed to process {file_path}: {e}")
return None
# Create tasks for all files
tasks = [
_process_with_semaphore(file_path, i)
for i, file_path in enumerate(file_paths)
]
# Execute all tasks concurrently with controlled concurrency
results = await asyncio.gather(*tasks, return_exceptions=True)
return results
async def async_batch_nfo_operations(
nfo_operations: List[Dict[str, Any]],
max_concurrent: int = 5
) -> List[bool]:
"""
Batch NFO operations (read/write) with controlled concurrency
Args:
nfo_operations: List of operation dictionaries with 'type', 'path', and other params
max_concurrent: Maximum number of concurrent operations
Returns:
List of success/failure results
"""
async def _execute_nfo_operation(operation: Dict[str, Any]) -> bool:
try:
op_type = operation.get('type')
path = operation.get('path')
if op_type == 'read':
result = await async_read_nfo_file(path)
return result is not None
elif op_type == 'write':
root = operation.get('root')
lock_metadata = operation.get('lock_metadata', True)
return await async_write_nfo_file(path, root, lock_metadata)
else:
_log("ERROR", f"Unknown NFO operation type: {op_type}")
return False
except Exception as e:
_log("ERROR", f"Failed to execute NFO operation: {e}")
return False
return await async_batch_process_files(
[op.get('path') for op in nfo_operations],
lambda path: _execute_nfo_operation(next(op for op in nfo_operations if op.get('path') == path)),
max_concurrent
)
async def async_concurrent_episode_processing(
episodes_data: List[Dict[str, Any]],
process_episode_func,
max_concurrent: int = 3
) -> List[Any]:
"""
Process multiple episodes concurrently
Args:
episodes_data: List of episode data dictionaries
process_episode_func: Async function to process each episode
max_concurrent: Maximum number of concurrent episode processes
Returns:
List of processing results
"""
semaphore = asyncio.Semaphore(max_concurrent)
async def _process_episode_with_semaphore(episode_data: Dict[str, Any]) -> Any:
async with semaphore:
try:
return await process_episode_func(episode_data)
except Exception as e:
_log("ERROR", f"Failed to process episode {episode_data}: {e}")
return None
# Create tasks for all episodes
tasks = [
_process_episode_with_semaphore(episode_data)
for episode_data in episodes_data
]
# Execute all tasks concurrently
results = await asyncio.gather(*tasks, return_exceptions=True)
return results
async def async_directory_scan_with_stats(
directories: List[Path],
file_extensions: Optional[Set[str]] = None
) -> Dict[str, Any]:
"""
Async scan multiple directories and gather statistics
Args:
directories: List of directories to scan
file_extensions: Optional set of file extensions to filter by
Returns:
Dictionary with scan statistics and file lists
"""
if file_extensions is None:
file_extensions = VIDEO_EXTENSIONS
stats = {
'total_files': 0,
'total_directories': len(directories),
'files_by_directory': {},
'scan_errors': [],
'total_size_bytes': 0
}
async def _scan_single_directory(directory: Path) -> Dict[str, Any]:
dir_stats = {
'path': str(directory),
'files': [],
'file_count': 0,
'size_bytes': 0,
'error': None
}
try:
if not await async_file_exists(directory):
dir_stats['error'] = 'Directory does not exist'
return dir_stats
files = await async_find_video_files(directory, recursive=True)
dir_stats['files'] = [str(f) for f in files]
dir_stats['file_count'] = len(files)
# Calculate total size
for file_path in files:
try:
stat_result = await aiofiles.os.stat(file_path)
dir_stats['size_bytes'] += stat_result.st_size
except Exception:
pass # Skip files we can't stat
except Exception as e:
dir_stats['error'] = str(e)
stats['scan_errors'].append(f"{directory}: {e}")
return dir_stats
# Scan all directories concurrently
directory_results = await asyncio.gather(
*[_scan_single_directory(directory) for directory in directories],
return_exceptions=True
)
# Aggregate results
for result in directory_results:
if isinstance(result, dict) and not result.get('error'):
stats['files_by_directory'][result['path']] = result
stats['total_files'] += result['file_count']
stats['total_size_bytes'] += result['size_bytes']
return stats
+284
View File
@@ -0,0 +1,284 @@
"""
Error handling utilities for NFOGuard
Provides structured error handling, retry mechanisms, and error reporting
"""
import time
import functools
from typing import Callable, Optional, Type, Union, List, Any
from pathlib import Path
from utils.logging import _log
from utils.exceptions import (
NFOGuardException,
RetryableError,
NetworkRetryableError,
TemporaryFileError,
ExternalAPIError,
FileOperationError
)
def with_error_handling(
operation_name: str,
log_errors: bool = True,
reraise: bool = True,
fallback_value: Any = None
):
"""
Decorator for standardized error handling
Args:
operation_name: Name of the operation for logging
log_errors: Whether to log errors automatically
reraise: Whether to reraise exceptions after logging
fallback_value: Value to return if error occurs and reraise=False
"""
def decorator(func: Callable) -> Callable:
@functools.wraps(func)
def wrapper(*args, **kwargs):
try:
return func(*args, **kwargs)
except NFOGuardException as e:
if log_errors:
_log("ERROR", f"{operation_name} failed: {e.message}")
if e.details:
_log("DEBUG", f"{operation_name} error details: {e.details}")
if reraise:
raise
return fallback_value
except Exception as e:
if log_errors:
_log("ERROR", f"{operation_name} failed with unexpected error: {e}")
if reraise:
# Wrap unexpected errors in our custom exception
raise NFOGuardException(
f"{operation_name} failed: {str(e)}",
{"original_error": str(e), "error_type": type(e).__name__}
)
return fallback_value
return wrapper
return decorator
def with_retry(
max_attempts: int = 3,
delay: float = 1.0,
backoff_factor: float = 2.0,
retry_on: Union[Type[Exception], List[Type[Exception]]] = None
):
"""
Decorator for retry logic on retryable errors
Args:
max_attempts: Maximum number of retry attempts
delay: Initial delay between retries in seconds
backoff_factor: Factor to multiply delay by after each attempt
retry_on: Exception types to retry on (defaults to RetryableError)
"""
if retry_on is None:
retry_on = [RetryableError]
elif not isinstance(retry_on, list):
retry_on = [retry_on]
def decorator(func: Callable) -> Callable:
@functools.wraps(func)
def wrapper(*args, **kwargs):
current_delay = delay
last_exception = None
for attempt in range(max_attempts):
try:
return func(*args, **kwargs)
except Exception as e:
last_exception = e
# Check if this is a retryable error
should_retry = any(isinstance(e, exc_type) for exc_type in retry_on)
if not should_retry or attempt == max_attempts - 1:
# Don't retry or max attempts reached
raise
# Use custom retry delay if available
retry_delay = current_delay
if isinstance(e, RetryableError) and e.retry_after:
retry_delay = e.retry_after
_log("WARNING", f"Attempt {attempt + 1}/{max_attempts} failed: {e}. Retrying in {retry_delay}s...")
time.sleep(retry_delay)
current_delay *= backoff_factor
# This should never be reached, but just in case
raise last_exception
return wrapper
return decorator
def safe_file_operation(
operation: str,
file_path: Union[str, Path],
operation_func: Callable,
*args,
**kwargs
) -> Any:
"""
Safely perform file operations with error handling
Args:
operation: Description of the operation
file_path: Path to the file being operated on
operation_func: Function to execute
*args, **kwargs: Arguments to pass to operation_func
Returns:
Result of operation_func or None if error
Raises:
FileOperationError: If file operation fails
"""
try:
return operation_func(*args, **kwargs)
except PermissionError as e:
raise FileOperationError(operation, str(file_path), f"Permission denied: {e}")
except FileNotFoundError as e:
raise FileOperationError(operation, str(file_path), f"File not found: {e}")
except OSError as e:
# Check if this might be a temporary error
if e.errno in [28, 122]: # No space left, quota exceeded
raise TemporaryFileError(str(file_path), operation, f"Disk space issue: {e}")
raise FileOperationError(operation, str(file_path), f"OS error: {e}")
except Exception as e:
raise FileOperationError(operation, str(file_path), f"Unexpected error: {e}")
def safe_api_call(
api_name: str,
operation: str,
api_func: Callable,
*args,
**kwargs
) -> Any:
"""
Safely perform API calls with error handling
Args:
api_name: Name of the API (e.g., "Sonarr", "TMDB")
operation: Description of the operation
api_func: Function to execute
*args, **kwargs: Arguments to pass to api_func
Returns:
Result of api_func
Raises:
ExternalAPIError: If API call fails
NetworkRetryableError: If network error that can be retried
"""
try:
return api_func(*args, **kwargs)
except ConnectionError as e:
raise NetworkRetryableError(f"{api_name} API", f"Connection error: {e}")
except TimeoutError as e:
raise NetworkRetryableError(f"{api_name} API", f"Timeout error: {e}")
except Exception as e:
# Check if it's an HTTP error with status code
status_code = getattr(e, 'status_code', None) or getattr(e, 'response', {}).get('status_code')
response_text = getattr(e, 'text', None) or str(e)
# Retry on certain HTTP status codes
if status_code in [429, 502, 503, 504]: # Rate limit, bad gateway, service unavailable, gateway timeout
raise NetworkRetryableError(f"{api_name} API", f"HTTP {status_code}: {response_text}")
raise ExternalAPIError(api_name, operation, status_code, response_text)
def log_structured_error(error: NFOGuardException, context: Optional[str] = None) -> None:
"""
Log structured error information
Args:
error: NFOGuardException to log
context: Additional context for the error
"""
error_dict = error.to_dict()
if context:
error_dict['context'] = context
_log("ERROR", f"Structured error: {error.message}")
_log("DEBUG", f"Error details: {error_dict}")
def create_error_response(error: NFOGuardException, include_details: bool = False) -> dict:
"""
Create standardized error response for API endpoints
Args:
error: NFOGuardException to convert
include_details: Whether to include detailed error information
Returns:
Dictionary suitable for JSON response
"""
response = {
"status": "error",
"error_type": error.__class__.__name__,
"message": error.message
}
if include_details and error.details:
response["details"] = error.details
return response
def validate_required_config(config_dict: dict, required_keys: List[str]) -> None:
"""
Validate that required configuration keys are present and not empty
Args:
config_dict: Configuration dictionary to validate
required_keys: List of required configuration keys
Raises:
ConfigurationError: If required configuration is missing or invalid
"""
from utils.exceptions import ConfigurationError
missing_keys = []
empty_keys = []
for key in required_keys:
if key not in config_dict:
missing_keys.append(key)
elif not config_dict[key]:
empty_keys.append(key)
if missing_keys:
raise ConfigurationError(
"missing_required_config",
f"Missing required configuration keys: {missing_keys}",
{"missing_keys": missing_keys}
)
if empty_keys:
raise ConfigurationError(
"empty_required_config",
f"Required configuration keys are empty: {empty_keys}",
{"empty_keys": empty_keys}
)
class ErrorContext:
"""Context manager for adding context to errors"""
def __init__(self, context: str):
self.context = context
def __enter__(self):
return self
def __exit__(self, exc_type, exc_val, exc_tb):
if exc_type and issubclass(exc_type, NFOGuardException):
exc_val.details = exc_val.details or {}
exc_val.details['error_context'] = self.context
return False # Don't suppress the exception
+172
View File
@@ -0,0 +1,172 @@
"""
Custom exceptions for NFOGuard
Provides structured error handling and better error reporting
"""
from typing import Optional, Dict, Any
class NFOGuardException(Exception):
"""Base exception for all NFOGuard errors"""
def __init__(self, message: str, details: Optional[Dict[str, Any]] = None):
super().__init__(message)
self.message = message
self.details = details or {}
def to_dict(self) -> Dict[str, Any]:
"""Convert exception to dictionary for structured logging"""
return {
"error_type": self.__class__.__name__,
"message": self.message,
"details": self.details
}
class MediaPathNotFoundError(NFOGuardException):
"""Raised when media directory cannot be found"""
def __init__(self, media_type: str, title: str, imdb_id: Optional[str] = None, search_paths: Optional[list] = None):
details = {
"media_type": media_type,
"title": title,
"imdb_id": imdb_id,
"search_paths": [str(p) for p in (search_paths or [])]
}
message = f"{media_type.title()} directory not found: {title}"
if imdb_id:
message += f" (IMDb: {imdb_id})"
super().__init__(message, details)
class IMDbIDNotFoundError(NFOGuardException):
"""Raised when IMDb ID cannot be extracted from path or files"""
def __init__(self, path: str, media_type: str = "media"):
details = {
"path": path,
"media_type": media_type
}
message = f"No IMDb ID found for {media_type}: {path}"
super().__init__(message, details)
class WebhookProcessingError(NFOGuardException):
"""Raised when webhook processing fails"""
def __init__(self, webhook_type: str, reason: str, payload: Optional[Dict] = None):
details = {
"webhook_type": webhook_type,
"reason": reason,
"payload": payload
}
message = f"{webhook_type} webhook processing failed: {reason}"
super().__init__(message, details)
class ExternalAPIError(NFOGuardException):
"""Raised when external API calls fail"""
def __init__(self, api_name: str, operation: str, status_code: Optional[int] = None, response: Optional[str] = None):
details = {
"api_name": api_name,
"operation": operation,
"status_code": status_code,
"response": response
}
message = f"{api_name} API error during {operation}"
if status_code:
message += f" (HTTP {status_code})"
super().__init__(message, details)
class DatabaseError(NFOGuardException):
"""Raised when database operations fail"""
def __init__(self, operation: str, table: Optional[str] = None, original_error: Optional[Exception] = None):
details = {
"operation": operation,
"table": table,
"original_error": str(original_error) if original_error else None
}
message = f"Database error during {operation}"
if table:
message += f" on table {table}"
super().__init__(message, details)
class NFOCreationError(NFOGuardException):
"""Raised when NFO file creation fails"""
def __init__(self, nfo_path: str, reason: str, media_type: str = "media"):
details = {
"nfo_path": nfo_path,
"reason": reason,
"media_type": media_type
}
message = f"Failed to create {media_type} NFO file: {reason}"
super().__init__(message, details)
class ConfigurationError(NFOGuardException):
"""Raised when configuration is invalid or missing"""
def __init__(self, setting: str, reason: str, current_value: Optional[Any] = None):
details = {
"setting": setting,
"reason": reason,
"current_value": current_value
}
message = f"Configuration error for {setting}: {reason}"
super().__init__(message, details)
class FileOperationError(NFOGuardException):
"""Raised when file operations fail"""
def __init__(self, operation: str, file_path: str, reason: str):
details = {
"operation": operation,
"file_path": file_path,
"reason": reason
}
message = f"File {operation} failed for {file_path}: {reason}"
super().__init__(message, details)
class DateProcessingError(NFOGuardException):
"""Raised when date processing or parsing fails"""
def __init__(self, date_value: str, operation: str, media_type: str = "media"):
details = {
"date_value": date_value,
"operation": operation,
"media_type": media_type
}
message = f"Date processing error during {operation} for {media_type}: {date_value}"
super().__init__(message, details)
class RetryableError(NFOGuardException):
"""Base class for errors that can be retried"""
def __init__(self, message: str, details: Optional[Dict[str, Any]] = None, retry_after: Optional[int] = None):
super().__init__(message, details)
self.retry_after = retry_after # Seconds to wait before retry
class NetworkRetryableError(RetryableError):
"""Network errors that can be retried"""
def __init__(self, url: str, reason: str, retry_after: Optional[int] = 30):
details = {"url": url, "reason": reason}
message = f"Network error for {url}: {reason}"
super().__init__(message, details, retry_after)
class TemporaryFileError(RetryableError):
"""Temporary file system errors that can be retried"""
def __init__(self, file_path: str, operation: str, reason: str, retry_after: Optional[int] = 5):
details = {"file_path": file_path, "operation": operation, "reason": reason}
message = f"Temporary file error during {operation} for {file_path}: {reason}"
super().__init__(message, details, retry_after)
+276
View File
@@ -0,0 +1,276 @@
"""
File utility functions for NFOGuard
Common file operations to eliminate code duplication
"""
import glob
import re
from pathlib import Path
from typing import Optional, List, Dict, Tuple, Union
from utils.logging import _log
# Video file extensions used throughout the application
VIDEO_EXTENSIONS = {'.mkv', '.mp4', '.avi', '.m4v', '.mov', '.ts'}
# Episode pattern for TV series files
EPISODE_PATTERN = re.compile(
r'.*[sS](\d{1,2})[eE](\d{1,3}).*|.*(\d{1,2})x(\d{1,3}).*'
)
def find_media_path_by_imdb_and_title(
title: str,
imdb_id: str,
search_paths: List[Path],
webhook_path: Optional[str] = None,
path_mapper = None
) -> Optional[Path]:
"""
Unified media path finder for both TV series and movies
Args:
title: Media title to search for
imdb_id: IMDb ID to search for
search_paths: List of paths to search in (tv_paths or movie_paths)
webhook_path: Optional webhook path to try first
path_mapper: Optional path mapper for webhook path conversion
Returns:
Path to media directory if found, None otherwise
"""
# Try webhook path first if provided
if webhook_path and path_mapper:
try:
if hasattr(path_mapper, 'sonarr_path_to_container_path'):
container_path = path_mapper.sonarr_path_to_container_path(webhook_path)
elif hasattr(path_mapper, 'radarr_path_to_container_path'):
container_path = path_mapper.radarr_path_to_container_path(webhook_path)
else:
container_path = webhook_path
path_obj = Path(container_path)
if path_obj.exists():
return path_obj
except Exception as e:
_log("WARNING", f"Failed to process webhook path {webhook_path}: {e}")
# Search by IMDb ID or title in configured paths
for media_path in search_paths:
if not media_path.exists():
continue
# Search by IMDb ID first (more reliable)
if imdb_id:
pattern = str(media_path / f"*[imdb-{imdb_id}]*")
matches = glob.glob(pattern)
if matches:
return Path(matches[0])
# Search by title as fallback
if title:
title_clean = clean_title_for_search(title)
for item in media_path.iterdir():
if item.is_dir() and "[imdb-" in item.name.lower():
item_clean = clean_title_for_search(item.name)
if title_clean in item_clean:
return item
return None
def clean_title_for_search(title: str) -> str:
"""
Clean title for fuzzy matching
Args:
title: Raw title string
Returns:
Cleaned title for comparison
"""
return title.lower().replace(" ", "").replace("-", "").replace(".", "")
def find_video_files(directory: Path, recursive: bool = True) -> List[Path]:
"""
Find all video files in a directory
Args:
directory: Directory to search
recursive: Whether to search recursively
Returns:
List of video file paths
"""
if not directory.exists():
return []
video_files = []
if recursive:
for item in directory.rglob('*'):
if item.is_file() and item.suffix.lower() in VIDEO_EXTENSIONS:
video_files.append(item)
else:
for item in directory.iterdir():
if item.is_file() and item.suffix.lower() in VIDEO_EXTENSIONS:
video_files.append(item)
return video_files
def extract_episode_info(filename: str) -> Optional[Tuple[int, int]]:
"""
Extract season and episode numbers from filename
Args:
filename: Video filename to parse
Returns:
Tuple of (season, episode) if found, None otherwise
"""
match = EPISODE_PATTERN.match(filename)
if not match:
return None
if match.group(1) and match.group(2): # SxxExx format
season = int(match.group(1))
episode = int(match.group(2))
elif match.group(3) and match.group(4): # NxNN format
season = int(match.group(3))
episode = int(match.group(4))
else:
return None
return (season, episode)
def find_episodes_on_disk(series_path: Path) -> Dict[Tuple[int, int], List[Path]]:
"""
Find all episodes on disk and return mapping of (season, episode) -> [video_files]
Args:
series_path: Path to series directory
Returns:
Dictionary mapping (season, episode) tuples to lists of video files
"""
episodes = {}
if not series_path.exists():
return episodes
for video_file in find_video_files(series_path, recursive=True):
episode_info = extract_episode_info(video_file.name)
if episode_info:
season, episode = episode_info
key = (season, episode)
if key not in episodes:
episodes[key] = []
episodes[key].append(video_file)
return episodes
def extract_title_from_directory_name(directory_name: str) -> Optional[str]:
"""
Extract clean title from directory name, removing year and IMDb ID
Args:
directory_name: Directory name to parse
Returns:
Cleaned title or None if no title found
"""
name = directory_name
# Remove IMDb ID part: [imdb-ttXXXXXX] or [ttXXXXXX]
name = re.sub(r'\s*\[imdb-?tt\d+\]', '', name, flags=re.IGNORECASE)
name = re.sub(r'\s*\[tt\d+\]', '', name, flags=re.IGNORECASE)
# Remove year in parentheses: (YYYY)
name = re.sub(r'\s*\(\d{4}\)', '', name)
# Clean up extra spaces
name = ' '.join(name.split())
return name.strip() if name.strip() else None
def extract_imdb_id_from_path(path: Union[str, Path]) -> Optional[str]:
"""
Extract IMDb ID from directory or file path
Args:
path: Path to examine (string or Path object)
Returns:
IMDb ID if found, None otherwise
"""
path_str = str(path)
# Look for [imdb-ttXXXXXX] or [ttXXXXXX] patterns
patterns = [
r'\[imdb-(tt\d+)\]', # [imdb-tt1234567]
r'\[(tt\d+)\]', # [tt1234567]
r'imdb[_-]?(tt\d+)', # imdb_tt1234567 or imdb-tt1234567
r'(tt\d{7,})', # standalone tt1234567 (7+ digits)
]
for pattern in patterns:
match = re.search(pattern, path_str, re.IGNORECASE)
if match:
imdb_id = match.group(1)
# Ensure it starts with 'tt'
if not imdb_id.startswith('tt'):
imdb_id = f'tt{imdb_id}'
return imdb_id
return None
def is_video_file(file_path: Path) -> bool:
"""
Check if a file is a video file based on extension
Args:
file_path: Path to check
Returns:
True if it's a video file, False otherwise
"""
return file_path.suffix.lower() in VIDEO_EXTENSIONS
def safe_directory_scan(directory: Path, pattern: str = "*") -> List[Path]:
"""
Safely scan directory with error handling
Args:
directory: Directory to scan
pattern: Glob pattern to match
Returns:
List of matching paths, empty list if scan fails
"""
try:
if not directory.exists():
return []
return list(directory.glob(pattern))
except (PermissionError, OSError) as e:
_log("WARNING", f"Failed to scan directory {directory}: {e}")
return []
def normalize_path_separators(path: str) -> str:
"""
Normalize path separators for cross-platform compatibility
Args:
path: Path string to normalize
Returns:
Normalized path string
"""
return str(Path(path).as_posix())
+170
View File
@@ -0,0 +1,170 @@
"""
Logging utilities for NFOGuard
"""
import os
import re
import logging
import logging.handlers
from pathlib import Path
from datetime import datetime, timezone
from zoneinfo import ZoneInfo
class TimezoneAwareFormatter(logging.Formatter):
"""Formatter that respects the container timezone"""
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.timezone = self._get_local_timezone()
def _get_local_timezone(self):
"""Get the local timezone, respecting TZ environment variable"""
tz_name = os.environ.get('TZ', 'UTC')
try:
# Try zoneinfo first (Python 3.9+)
return ZoneInfo(tz_name)
except ImportError:
# Fallback for older Python versions
try:
import pytz
return pytz.timezone(tz_name)
except:
# Final fallback to UTC
return timezone.utc
except:
# If zone name is invalid, fallback to UTC
return timezone.utc
def formatTime(self, record, datefmt=None):
dt = datetime.fromtimestamp(record.created, tz=self.timezone)
if datefmt:
return dt.strftime(datefmt)
return dt.isoformat(timespec='seconds')
def _setup_file_logging():
"""Setup file logging for NFOGuard"""
log_dir = Path(os.environ.get("LOG_DIR", "/app/data/logs"))
log_dir.mkdir(parents=True, exist_ok=True)
logger = logging.getLogger("NFOGuard")
logger.setLevel(logging.DEBUG)
file_handler = logging.handlers.RotatingFileHandler(
log_dir / "nfoguard.log", maxBytes=50*1024*1024, backupCount=3
)
formatter = TimezoneAwareFormatter(
'[%(asctime)s] %(levelname)s: %(message)s'
)
file_handler.setFormatter(formatter)
logger.addHandler(file_handler)
return logger
def _mask_sensitive_data(msg: str) -> str:
"""Mask API keys and other sensitive data in log messages"""
# List of patterns to mask
sensitive_patterns = [
(r'api_key=([a-zA-Z0-9_\-]+)', r'api_key=***masked***'),
(r'password=([^\s&]+)', r'password=***masked***'),
(r'token=([a-zA-Z0-9_\-]+)', r'token=***masked***'),
(r'key=([a-zA-Z0-9_\-]{8,})', r'key=***masked***'), # Keys longer than 8 chars
(r'([a-zA-Z0-9]{32,})', lambda m: m.group(1)[:8] + '***masked***' if len(m.group(1)) > 16 else m.group(1)) # Long strings likely to be keys
]
masked_msg = msg
for pattern, replacement in sensitive_patterns:
if isinstance(replacement, str):
masked_msg = re.sub(pattern, replacement, masked_msg, flags=re.IGNORECASE)
else:
masked_msg = re.sub(pattern, replacement, masked_msg, flags=re.IGNORECASE)
return masked_msg
def _get_local_timezone():
"""Get the local timezone, respecting TZ environment variable"""
tz_name = os.environ.get('TZ', 'UTC')
try:
# Try zoneinfo first (Python 3.9+)
return ZoneInfo(tz_name)
except ImportError:
# Fallback for older Python versions
try:
import pytz
return pytz.timezone(tz_name)
except:
# Final fallback to UTC
return timezone.utc
except:
# If zone name is invalid, fallback to UTC
return timezone.utc
def _log(level: str, msg: str):
"""Enhanced logging that writes to both console and file with sensitive data masking"""
masked_msg = _mask_sensitive_data(msg)
tz = _get_local_timezone()
print(f"[{datetime.now(tz).isoformat(timespec='seconds')}] {level}: {masked_msg}")
try:
file_logger = _setup_file_logging()
getattr(file_logger, level.lower(), file_logger.info)(masked_msg)
except Exception as e:
print(f"File logging error: {e}")
def convert_utc_to_local(utc_iso_string: str) -> str:
"""Convert UTC ISO timestamp to local timezone timestamp"""
if not utc_iso_string:
return utc_iso_string
try:
# Parse UTC timestamp
if utc_iso_string.endswith('Z'):
dt_utc = datetime.fromisoformat(utc_iso_string.replace('Z', '+00:00'))
elif '+00:00' in utc_iso_string:
dt_utc = datetime.fromisoformat(utc_iso_string)
else:
# Assume UTC if no timezone info
dt_utc = datetime.fromisoformat(utc_iso_string).replace(tzinfo=timezone.utc)
# Convert to local timezone
local_tz = _get_local_timezone()
dt_local = dt_utc.astimezone(local_tz)
return dt_local.isoformat(timespec='seconds')
except Exception:
# If conversion fails, return original
return utc_iso_string
def _load_environment_files():
"""Load environment variables from .env and optionally .env.secrets"""
from pathlib import Path
# Try to load from python-dotenv if available
try:
from dotenv import load_dotenv
# Load main .env file
env_file = Path(".env")
if env_file.exists():
load_dotenv(env_file)
_log("INFO", f"Loaded environment from {env_file}")
# Load secrets file if it exists
secrets_file = Path(".env.secrets")
if secrets_file.exists():
load_dotenv(secrets_file)
_log("INFO", f"Loaded secrets from {secrets_file}")
except ImportError:
_log("WARNING", "python-dotenv not available - environment files not loaded")
# Initialize logging and load environment files
_setup_file_logging()
_load_environment_files()
+375
View File
@@ -0,0 +1,375 @@
"""
NFO parsing patterns and utilities for NFOGuard
Consolidates common NFO parsing logic and patterns
"""
import re
import xml.etree.ElementTree as ET
from pathlib import Path
from typing import Optional, Dict, Any, List
from datetime import datetime
from utils.logging import _log
from utils.exceptions import NFOCreationError, FileOperationError
from utils.validation import validate_imdb_id, validate_date_string
# Common IMDb ID patterns used across the application
IMDB_PATTERNS = [
r'\[imdb-?(tt\d+)\]', # [imdb-tt1234567] or [imdb-1234567]
r'\[(tt\d+)\]', # [tt1234567]
r'\{imdb-?(tt\d+)\}', # {imdb-tt1234567} or {imdb-1234567}
r'\(imdb-?(tt\d+)\)', # (imdb-tt1234567) or (imdb-1234567)
r'[-_\s](tt\d+)$', # tt1234567 at end of string
r'imdb[_-]?(tt\d+)', # imdb_tt1234567 or imdb-tt1234567
]
# Episode filename patterns
EPISODE_PATTERNS = [
r'.*[sS](\d{1,2})[eE](\d{1,3}).*', # S01E01
r'.*(\d{1,2})x(\d{1,3}).*', # 1x01
r'.*[sS](\d{1,2})\.?[eE](\d{1,3}).*', # S01.E01
r'.*Season[_\s]?(\d{1,2})[_\s]?Episode[_\s]?(\d{1,3}).*', # Season 1 Episode 1
]
# Common NFO XML namespaces
NFO_NAMESPACES = {
'kodi': 'http://kodi.tv/moviedb',
'tmdb': 'https://www.themoviedb.org',
'imdb': 'https://www.imdb.com'
}
def extract_imdb_id_from_text(text: str) -> Optional[str]:
"""
Extract IMDb ID from text using consolidated patterns
Args:
text: Text to search for IMDb ID
Returns:
IMDb ID if found, None otherwise
"""
if not text:
return None
text_lower = text.lower()
for pattern in IMDB_PATTERNS:
match = re.search(pattern, text_lower)
if match:
imdb_id = match.group(1)
# Ensure it starts with 'tt'
if not imdb_id.startswith('tt'):
imdb_id = f'tt{imdb_id}'
if validate_imdb_id(imdb_id):
return imdb_id
return None
def extract_episode_info_from_filename(filename: str) -> Optional[Dict[str, int]]:
"""
Extract season and episode information from filename
Args:
filename: Filename to parse
Returns:
Dictionary with 'season' and 'episode' keys if found, None otherwise
"""
for pattern in EPISODE_PATTERNS:
match = re.search(pattern, filename, re.IGNORECASE)
if match:
try:
season = int(match.group(1))
episode = int(match.group(2))
# Validate reasonable ranges
if 0 <= season <= 99 and 1 <= episode <= 999:
return {"season": season, "episode": episode}
except (ValueError, IndexError):
continue
return None
def parse_nfo_with_tolerance(nfo_path: Path) -> Optional[ET.Element]:
"""
Parse NFO file with error tolerance
Args:
nfo_path: Path to NFO file
Returns:
XML root element if successful, None otherwise
"""
if not nfo_path.exists():
return None
try:
# Try normal parsing first
tree = ET.parse(nfo_path)
return tree.getroot()
except ET.ParseError as e:
_log("WARNING", f"NFO parse error for {nfo_path}: {e}. Trying with tolerance...")
try:
# Try reading and cleaning the content
content = nfo_path.read_text(encoding='utf-8', errors='ignore')
# Basic cleanup for common issues
content = content.replace('&', '&amp;') # Fix unescaped ampersands
content = re.sub(r'<(\w+)([^>]*?)(?<!/)>', r'<\1\2/>', content) # Fix unclosed tags
root = ET.fromstring(content)
return root
except Exception as e:
_log("ERROR", f"Failed to parse NFO file {nfo_path} even with tolerance: {e}")
return None
def extract_text_from_nfo_element(root: ET.Element, xpath: str, namespaces: Optional[Dict] = None) -> Optional[str]:
"""
Extract text content from NFO element using XPath
Args:
root: XML root element
xpath: XPath expression
namespaces: Optional namespace dictionary
Returns:
Text content if found, None otherwise
"""
try:
if namespaces:
elements = root.findall(xpath, namespaces)
else:
elements = root.findall(xpath)
if elements and elements[0].text:
return elements[0].text.strip()
except Exception as e:
_log("DEBUG", f"Failed to extract text from XPath {xpath}: {e}")
return None
def extract_imdb_from_nfo_content(root: ET.Element) -> Optional[str]:
"""
Extract IMDb ID from NFO XML content
Args:
root: XML root element
Returns:
IMDb ID if found, None otherwise
"""
# Common XPath patterns for IMDb ID
imdb_xpaths = [
'.//imdb',
'.//imdbid',
'.//id[@type="imdb"]',
'.//uniqueid[@type="imdb"]',
'.//uniqueid[@default="true"]',
'.//id',
'.//uniqueid'
]
for xpath in imdb_xpaths:
imdb_text = extract_text_from_nfo_element(root, xpath)
if imdb_text:
imdb_id = extract_imdb_id_from_text(imdb_text)
if imdb_id:
return imdb_id
# Check in plot/overview text as fallback
plot_xpaths = ['.//plot', './/overview', './/summary']
for xpath in plot_xpaths:
plot_text = extract_text_from_nfo_element(root, xpath)
if plot_text:
imdb_id = extract_imdb_id_from_text(plot_text)
if imdb_id:
return imdb_id
return None
def extract_dates_from_nfo(root: ET.Element) -> Dict[str, Optional[str]]:
"""
Extract various date fields from NFO content
Args:
root: XML root element
Returns:
Dictionary with date fields (premiered, aired, dateadded, etc.)
"""
date_fields = {
'premiered': ['.//premiered', './/releasedate', './/year'],
'aired': ['.//aired', './/firstaired'],
'dateadded': ['.//dateadded', './/added'],
'lastplayed': ['.//lastplayed'],
'filelastmodified': ['.//filelastmodified']
}
result = {}
for field_name, xpaths in date_fields.items():
for xpath in xpaths:
date_text = extract_text_from_nfo_element(root, xpath)
if date_text and validate_date_string(date_text):
result[field_name] = date_text
break
else:
result[field_name] = None
return result
def create_basic_nfo_structure(
media_type: str,
title: str,
imdb_id: Optional[str] = None,
dates: Optional[Dict[str, str]] = None,
additional_fields: Optional[Dict[str, str]] = None
) -> ET.Element:
"""
Create basic NFO XML structure
Args:
media_type: Type of media ('movie', 'tvshow', 'episode')
title: Media title
imdb_id: Optional IMDb ID
dates: Optional dictionary of date fields
additional_fields: Optional additional fields to include
Returns:
XML root element
"""
root = ET.Element(media_type)
# Add title
title_elem = ET.SubElement(root, 'title')
title_elem.text = title
# Add IMDb ID if provided
if imdb_id and validate_imdb_id(imdb_id):
imdb_elem = ET.SubElement(root, 'imdb')
imdb_elem.text = imdb_id
# Also add as uniqueid
uniqueid_elem = ET.SubElement(root, 'uniqueid', type='imdb', default='true')
uniqueid_elem.text = imdb_id
# Add dates if provided
if dates:
for field_name, date_value in dates.items():
if date_value and validate_date_string(date_value):
date_elem = ET.SubElement(root, field_name)
date_elem.text = date_value
# Add additional fields
if additional_fields:
for field_name, field_value in additional_fields.items():
if field_value:
field_elem = ET.SubElement(root, field_name)
field_elem.text = str(field_value)
return root
def write_nfo_file(
nfo_path: Path,
root: ET.Element,
lock_metadata: bool = True
) -> None:
"""
Write NFO XML content to file
Args:
nfo_path: Path where to write the NFO file
root: XML root element to write
lock_metadata: Whether to add file locking attributes
Raises:
NFOCreationError: If writing fails
"""
try:
# Ensure parent directory exists
nfo_path.parent.mkdir(parents=True, exist_ok=True)
# Add file locking if requested
if lock_metadata:
root.set('nfoguard_managed', 'true')
root.set('last_updated', datetime.now().isoformat())
# Create tree and write
tree = ET.ElementTree(root)
ET.indent(tree, space=" ", level=0) # Pretty formatting
# Write with proper XML declaration
with open(nfo_path, 'wb') as f:
tree.write(f, encoding='utf-8', xml_declaration=True)
_log("DEBUG", f"Successfully wrote NFO file: {nfo_path}")
except (OSError, ET.ParseError) as e:
raise NFOCreationError(
str(nfo_path),
f"Failed to write NFO file: {e}",
"unknown"
)
def is_nfo_managed_by_nfoguard(nfo_path: Path) -> bool:
"""
Check if NFO file is managed by NFOGuard
Args:
nfo_path: Path to NFO file
Returns:
True if managed by NFOGuard, False otherwise
"""
root = parse_nfo_with_tolerance(nfo_path)
if root is None:
return False
return root.get('nfoguard_managed') == 'true'
def extract_title_from_directory_name(directory_name: str) -> Optional[str]:
"""
Extract clean title from directory name
Args:
directory_name: Directory name to parse
Returns:
Cleaned title or None if no title found
"""
name = directory_name
# Remove IMDb ID patterns
for pattern in IMDB_PATTERNS:
name = re.sub(pattern, '', name, flags=re.IGNORECASE)
# Remove year in parentheses: (YYYY)
name = re.sub(r'\s*\(\d{4}\)', '', name)
# Remove common release info patterns
release_patterns = [
r'\s*\[.*?\]', # [1080p], [BluRay], etc.
r'\s*\{.*?\}', # {edition info}
r'\s*\(.*?\)', # (additional info)
]
for pattern in release_patterns:
name = re.sub(pattern, '', name, flags=re.IGNORECASE)
# Clean up extra spaces and special characters
name = re.sub(r'[._-]+', ' ', name) # Replace dots, underscores, dashes with spaces
name = ' '.join(name.split()) # Normalize whitespace
return name.strip() if name.strip() else None
+372
View File
@@ -0,0 +1,372 @@
"""
Validation utilities for NFOGuard
Provides runtime validation and type checking for critical paths
"""
import re
from pathlib import Path
from typing import Any, Dict, List, Optional, Union, Callable, TypeVar, Type
from datetime import datetime
from utils.exceptions import ConfigurationError, NFOGuardException
T = TypeVar('T')
def validate_imdb_id(imdb_id: str) -> bool:
"""
Validate IMDb ID format
Args:
imdb_id: IMDb ID to validate
Returns:
True if valid, False otherwise
"""
if not imdb_id or not isinstance(imdb_id, str):
return False
# Must start with 'tt' followed by 7+ digits
return bool(re.match(r'^tt\d{7,}$', imdb_id))
def validate_tmdb_id(tmdb_id: str) -> bool:
"""
Validate TMDB ID format
Args:
tmdb_id: TMDB ID to validate
Returns:
True if valid, False otherwise
"""
if not tmdb_id or not isinstance(tmdb_id, str):
return False
# Can be numeric or have tmdb- prefix
if tmdb_id.startswith('tmdb-'):
return tmdb_id[5:].isdigit()
return tmdb_id.isdigit()
def validate_season_episode(season: int, episode: int) -> bool:
"""
Validate season and episode numbers
Args:
season: Season number
episode: Episode number
Returns:
True if valid, False otherwise
"""
return (
isinstance(season, int) and season >= 0 and
isinstance(episode, int) and episode >= 1
)
def validate_date_string(date_str: str) -> bool:
"""
Validate date string format (ISO format)
Args:
date_str: Date string to validate
Returns:
True if valid ISO date, False otherwise
"""
if not date_str or not isinstance(date_str, str):
return False
try:
datetime.fromisoformat(date_str.replace('Z', '+00:00'))
return True
except ValueError:
return False
def validate_path_exists(path: Union[str, Path]) -> bool:
"""
Validate that a path exists
Args:
path: Path to validate
Returns:
True if path exists, False otherwise
"""
try:
return Path(path).exists()
except (OSError, ValueError):
return False
def validate_video_file(file_path: Union[str, Path]) -> bool:
"""
Validate that a file is a video file
Args:
file_path: Path to validate
Returns:
True if valid video file, False otherwise
"""
try:
path = Path(file_path)
video_extensions = {'.mkv', '.mp4', '.avi', '.m4v', '.mov', '.ts'}
return path.is_file() and path.suffix.lower() in video_extensions
except (OSError, ValueError):
return False
def validate_webhook_payload(payload: Dict[str, Any], required_fields: List[str]) -> List[str]:
"""
Validate webhook payload has required fields
Args:
payload: Webhook payload to validate
required_fields: List of required field names
Returns:
List of missing field names (empty if all present)
"""
missing_fields = []
for field in required_fields:
if field not in payload or payload[field] is None:
missing_fields.append(field)
return missing_fields
def validate_config_paths(paths: List[Union[str, Path]], path_type: str) -> None:
"""
Validate configuration paths exist and are directories
Args:
paths: List of paths to validate
path_type: Type of paths for error messages (e.g., "TV", "Movie")
Raises:
ConfigurationError: If any paths are invalid
"""
invalid_paths = []
for path in paths:
try:
path_obj = Path(path)
if not path_obj.exists():
invalid_paths.append(f"{path} (does not exist)")
elif not path_obj.is_dir():
invalid_paths.append(f"{path} (not a directory)")
except (OSError, ValueError) as e:
invalid_paths.append(f"{path} (invalid: {e})")
if invalid_paths:
raise ConfigurationError(
f"{path_type.lower()}_paths",
f"Invalid {path_type} paths found",
{"invalid_paths": invalid_paths}
)
def require_type(value: Any, expected_type: Type[T], name: str) -> T:
"""
Require value to be of specific type
Args:
value: Value to check
expected_type: Expected type
name: Name of the value for error messages
Returns:
The value if it matches the type
Raises:
TypeError: If value is not of expected type
"""
if not isinstance(value, expected_type):
raise TypeError(
f"{name} must be {expected_type.__name__}, got {type(value).__name__}"
)
return value
def require_non_empty(value: Optional[str], name: str) -> str:
"""
Require string value to be non-empty
Args:
value: String value to check
name: Name of the value for error messages
Returns:
The value if it's non-empty
Raises:
ValueError: If value is None or empty
"""
if not value:
raise ValueError(f"{name} cannot be None or empty")
return value
def validate_and_clean_imdb_id(imdb_id: Optional[str]) -> Optional[str]:
"""
Validate and clean IMDb ID
Args:
imdb_id: IMDb ID to validate and clean
Returns:
Cleaned IMDb ID or None if invalid
"""
if not imdb_id:
return None
# Clean the ID
cleaned = imdb_id.strip().lower()
# Remove common prefixes
if cleaned.startswith('imdb-'):
cleaned = cleaned[5:]
elif cleaned.startswith('imdb_'):
cleaned = cleaned[5:]
# Ensure it starts with 'tt'
if not cleaned.startswith('tt'):
cleaned = f'tt{cleaned}'
# Validate format
if validate_imdb_id(cleaned):
return cleaned
return None
def create_validator(validation_func: Callable[[Any], bool], error_message: str) -> Callable:
"""
Create a validator decorator
Args:
validation_func: Function that returns True if value is valid
error_message: Error message to raise if validation fails
Returns:
Decorator function
"""
def decorator(func: Callable) -> Callable:
def wrapper(*args, **kwargs):
# Apply validation to first argument
if args and not validation_func(args[0]):
raise ValueError(error_message.format(args[0]))
return func(*args, **kwargs)
return wrapper
return decorator
# Common validators
validate_imdb_required = create_validator(
lambda x: validate_imdb_id(x),
"Invalid IMDb ID format: {}"
)
validate_path_required = create_validator(
lambda x: validate_path_exists(x),
"Path does not exist: {}"
)
validate_date_required = create_validator(
lambda x: validate_date_string(x),
"Invalid date format: {}"
)
class ValidationError(NFOGuardException):
"""Raised when validation fails"""
def __init__(self, field_name: str, value: Any, reason: str):
details = {
"field_name": field_name,
"value": str(value),
"reason": reason
}
message = f"Validation failed for {field_name}: {reason}"
super().__init__(message, details)
def validate_episode_file_pattern(filename: str) -> Optional[Dict[str, int]]:
"""
Validate and extract episode information from filename
Args:
filename: Filename to validate
Returns:
Dictionary with season and episode if valid, None otherwise
"""
# Episode patterns
patterns = [
r'[sS](\d{1,2})[eE](\d{1,3})', # S01E01
r'(\d{1,2})x(\d{1,3})', # 1x01
r'[sS](\d{1,2})\.?[eE](\d{1,3})', # S01.E01
]
for pattern in patterns:
match = re.search(pattern, filename)
if match:
season = int(match.group(1))
episode = int(match.group(2))
if validate_season_episode(season, episode):
return {"season": season, "episode": episode}
return None
def sanitize_filename(filename: str) -> str:
"""
Sanitize filename by removing invalid characters
Args:
filename: Filename to sanitize
Returns:
Sanitized filename
"""
# Remove invalid characters for most filesystems
invalid_chars = r'<>:"/\|?*'
for char in invalid_chars:
filename = filename.replace(char, '_')
# Remove leading/trailing dots and spaces
filename = filename.strip('. ')
# Ensure it's not empty
if not filename:
filename = 'unnamed'
return filename
def validate_url_format(url: str) -> bool:
"""
Validate URL format
Args:
url: URL to validate
Returns:
True if valid URL format, False otherwise
"""
if not url or not isinstance(url, str):
return False
# Basic URL validation
url_pattern = re.compile(
r'^https?://' # http:// or https://
r'(?:(?:[A-Z0-9](?:[A-Z0-9-]{0,61}[A-Z0-9])?\.)+[A-Z]{2,6}\.?|' # domain...
r'localhost|' # localhost...
r'\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3})' # ...or ip
r'(?::\d+)?' # optional port
r'(?:/?|[/?]\S+)$', re.IGNORECASE)
return bool(url_pattern.match(url))
View File
+130
View File
@@ -0,0 +1,130 @@
"""
Webhook Batching System for NFOGuard
Handles batching and processing of webhook events to avoid processing storms
"""
import threading
from pathlib import Path
from typing import Dict, Set
from concurrent.futures import ThreadPoolExecutor
from config.settings import config
from utils.logging import _log
class WebhookBatcher:
"""Batches webhook events to avoid processing storms"""
def __init__(self):
self.pending: Dict[str, Dict] = {}
self.timers: Dict[str, threading.Timer] = {}
self.processing: Set[str] = set()
self.lock = threading.Lock()
self.executor = ThreadPoolExecutor(max_workers=config.max_concurrent)
# Will be set by the application when processors are available
self.tv_processor = None
self.movie_processor = None
def set_processors(self, tv_processor, movie_processor):
"""Set the processor instances"""
self.tv_processor = tv_processor
self.movie_processor = movie_processor
def add_webhook(self, key: str, webhook_data: Dict, media_type: str):
"""Add webhook to batch queue"""
with self.lock:
if key in self.timers:
self.timers[key].cancel()
webhook_data['media_type'] = media_type
self.pending[key] = webhook_data
_log("INFO", f"Batched {media_type} webhook for {key}")
_log("DEBUG", f"Batch added - key: {key}, media_type: {media_type}, timer scheduled for {config.batch_delay}s")
timer = threading.Timer(config.batch_delay, self._process_item, args=[key])
self.timers[key] = timer
timer.start()
def _process_item(self, key: str):
"""Process a batched item"""
with self.lock:
if key in self.processing or key not in self.pending:
return
self.processing.add(key)
webhook_data = self.pending.pop(key)
self.timers.pop(key, None)
try:
self.executor.submit(self._process_sync, key, webhook_data)
except Exception as e:
_log("ERROR", f"Error submitting processing for {key}: {e}")
with self.lock:
self.processing.discard(key)
def _process_sync(self, key: str, webhook_data: Dict):
"""Synchronous processing of webhook data with validation"""
try:
media_type = webhook_data.get('media_type')
path_str = webhook_data.get('path')
_log("DEBUG", f"Processing batch item: key={key}, media_type={media_type}, path={path_str}")
if not path_str:
_log("ERROR", f"No path found for {media_type} {key}")
return
path_obj = Path(path_str)
if not path_obj.exists():
_log("ERROR", f"BATCH PROCESSING FAILED: Path does not exist: {path_obj}")
_log("ERROR", f"This indicates a path mapping issue - webhook rejected to prevent wrong processing")
return
# CRITICAL: Validate that the path contains the expected IMDb ID for movies
if media_type == 'movie':
expected_imdb = key.replace('movie:', '') if key.startswith('movie:') else key
if expected_imdb not in path_str.lower():
_log("ERROR", f"BATCH VALIDATION FAILED: Expected IMDb {expected_imdb} not found in path {path_str}")
_log("ERROR", f"This prevents processing wrong movies due to batch corruption")
return
_log("DEBUG", f"Batch validation passed: IMDb {expected_imdb} found in path")
# Process based on media type
if media_type == 'tv':
if not self.tv_processor:
_log("ERROR", "TV processor not available")
return
# Check processing mode for TV webhooks
processing_mode = webhook_data.get('processing_mode', config.tv_webhook_processing_mode)
episodes_data = webhook_data.get('episodes', [])
if processing_mode == 'targeted' and episodes_data:
_log("INFO", f"Using targeted episode processing for {len(episodes_data)} episodes")
self.tv_processor.process_webhook_episodes(path_obj, episodes_data)
else:
_log("INFO", f"Using series processing mode (fallback or configured)")
self.tv_processor.process_series(path_obj)
elif media_type == 'movie':
if not self.movie_processor:
_log("ERROR", "Movie processor not available")
return
self.movie_processor.process_movie(path_obj, webhook_mode=True)
else:
_log("ERROR", f"Unknown media type: {media_type}")
except Exception as e:
_log("ERROR", f"Error processing {media_type} {key}: {e}")
finally:
with self.lock:
self.processing.discard(key)
def get_status(self) -> Dict:
"""Get batch queue status"""
with self.lock:
return {
"pending_items": list(self.pending.keys()),
"processing_items": list(self.processing),
"pending_count": len(self.pending),
"processing_count": len(self.processing)
}