feat(scanner): replace file writes with DB persistence for series
- SerieScanner.scan() now calls _persist_serie_to_db() instead of serie.save_to_file() - Added _sync_episodes_to_db() helper to handle episode CRUD during sync - EpisodeService gains delete_by_series() for targeted episode deletion - SerieList gains add_to_db() async method for DB-based series addition - test_serie_scanner_db_writes.py covers create/update/preserve/sync scenarios - DATABASE.md updated with Series Persistence Flow section Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
This commit is contained in:
@@ -10,6 +10,7 @@ Note:
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
import os
|
||||
import re
|
||||
@@ -24,7 +25,7 @@ from src.core.exceptions.Exceptions import MatchNotFoundError, NoKeyFoundExcepti
|
||||
from src.core.providers.base_provider import Loader
|
||||
|
||||
from src.server.database.connection import get_sync_session
|
||||
from src.server.database.service import AnimeSeriesService
|
||||
from src.server.database.service import AnimeSeriesService, EpisodeService
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
error_logger = logging.getLogger("error")
|
||||
@@ -208,6 +209,106 @@ class SerieScanner:
|
||||
"""Reinitialize the series dictionary (keyed by serie.key)."""
|
||||
self.keyDict: dict[str, Serie] = {}
|
||||
|
||||
async def _persist_serie_to_db(self, serie: Serie) -> None:
|
||||
"""Persist serie to database (create or update).
|
||||
|
||||
Args:
|
||||
serie: Serie domain object to persist
|
||||
"""
|
||||
try:
|
||||
from src.server.database.connection import get_async_session_factory
|
||||
|
||||
session_factory = get_async_session_factory()
|
||||
db = session_factory()
|
||||
try:
|
||||
existing = await AnimeSeriesService.get_by_key(db, serie.key)
|
||||
if existing:
|
||||
await AnimeSeriesService.update(
|
||||
db, existing.id,
|
||||
name=serie.name,
|
||||
folder=serie.folder,
|
||||
year=serie.year
|
||||
)
|
||||
await self._sync_episodes_to_db(db, existing.id, serie.episodeDict)
|
||||
else:
|
||||
anime_series = await AnimeSeriesService.create(
|
||||
db=db,
|
||||
key=serie.key,
|
||||
name=serie.name,
|
||||
site=serie.site,
|
||||
folder=serie.folder,
|
||||
year=serie.year
|
||||
)
|
||||
for season, eps in serie.episodeDict.items():
|
||||
for ep in eps:
|
||||
await EpisodeService.create(
|
||||
db=db,
|
||||
series_id=anime_series.id,
|
||||
season=season,
|
||||
episode_number=ep
|
||||
)
|
||||
await db.commit()
|
||||
logger.debug(
|
||||
"Persisted serie '%s' (key=%s) to database",
|
||||
serie.name, serie.key
|
||||
)
|
||||
except Exception as e:
|
||||
await db.rollback()
|
||||
logger.error(
|
||||
"Failed to persist serie '%s' to DB: %s",
|
||||
serie.key, e, exc_info=True
|
||||
)
|
||||
raise
|
||||
finally:
|
||||
await db.close()
|
||||
except Exception as e:
|
||||
logger.error(
|
||||
"Could not persist serie '%s' to DB (DB unavailable?): %s",
|
||||
serie.key, e
|
||||
)
|
||||
|
||||
async def _sync_episodes_to_db(
|
||||
self, db, series_id: int, episode_dict: dict[int, list[int]]
|
||||
) -> None:
|
||||
"""Sync episodes to database, preserving downloaded flags.
|
||||
|
||||
Adds missing episodes, removes episodes no longer missing,
|
||||
and preserves is_downloaded=True episodes.
|
||||
|
||||
Args:
|
||||
db: Async database session
|
||||
series_id: Database ID of the series
|
||||
episode_dict: Dict mapping season -> list of episode numbers
|
||||
"""
|
||||
existing_episodes = await EpisodeService.get_by_series(db, series_id)
|
||||
existing_map = {
|
||||
(ep.season, ep.episode_number): ep for ep in existing_episodes
|
||||
}
|
||||
new_keys = set()
|
||||
for season, eps in episode_dict.items():
|
||||
for ep_num in eps:
|
||||
new_keys.add((season, ep_num))
|
||||
for (season, ep_num), ep in existing_map.items():
|
||||
if (season, ep_num) not in new_keys:
|
||||
if ep.is_downloaded:
|
||||
logger.debug(
|
||||
"Preserving downloaded episode S%02dE%02d for series_id=%d",
|
||||
season, ep_num, series_id
|
||||
)
|
||||
else:
|
||||
await EpisodeService.delete_by_series(
|
||||
db, series_id, season, ep_num
|
||||
)
|
||||
for season, eps in episode_dict.items():
|
||||
for ep_num in eps:
|
||||
if (season, ep_num) not in existing_map:
|
||||
await EpisodeService.create(
|
||||
db=db,
|
||||
series_id=series_id,
|
||||
season=season,
|
||||
episode_number=ep_num
|
||||
)
|
||||
|
||||
def get_total_to_scan(self) -> int:
|
||||
"""Get the total number of folders to scan.
|
||||
|
||||
@@ -329,10 +430,16 @@ class SerieScanner:
|
||||
)
|
||||
serie.episodeDict = missing_episodes
|
||||
serie.folder = folder
|
||||
data_path = os.path.join(
|
||||
self.directory, folder, 'data'
|
||||
)
|
||||
serie.save_to_file(data_path)
|
||||
|
||||
# Persist to database (async)
|
||||
try:
|
||||
asyncio.run(self._persist_serie_to_db(serie))
|
||||
except Exception as e:
|
||||
logger.warning(
|
||||
"DB persistence failed for '%s', "
|
||||
"continuing without DB: %s",
|
||||
serie.key, e
|
||||
)
|
||||
|
||||
# Store by key (primary identifier), not folder
|
||||
if serie.key in self.keyDict:
|
||||
|
||||
@@ -1,15 +1,17 @@
|
||||
"""Utilities for loading and managing stored anime series metadata.
|
||||
|
||||
This module provides the SerieList class for managing collections of anime
|
||||
series metadata. It uses file-based storage only.
|
||||
series metadata. It uses file-based storage as fallback when database
|
||||
is not available.
|
||||
|
||||
Note:
|
||||
This module is part of the core domain layer and has no database
|
||||
dependencies. All database operations are handled by the service layer.
|
||||
This module is part of the core domain layer. Database operations
|
||||
are handled by the service layer via add_to_db().
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
import os
|
||||
import warnings
|
||||
@@ -106,6 +108,76 @@ class SerieList:
|
||||
|
||||
return anime_path
|
||||
|
||||
async def add_to_db(self, serie: Serie) -> bool:
|
||||
"""Persist a new series to the database.
|
||||
|
||||
Creates the filesystem folder using serie.folder, then persists
|
||||
the series metadata to the database.
|
||||
|
||||
Args:
|
||||
serie: The Serie instance to add
|
||||
|
||||
Returns:
|
||||
True if successful, False otherwise
|
||||
"""
|
||||
try:
|
||||
from src.server.database.connection import get_async_session_factory
|
||||
from src.server.database.service import AnimeSeriesService, EpisodeService
|
||||
|
||||
folder_name = serie.folder
|
||||
anime_path = os.path.join(self.directory, folder_name)
|
||||
os.makedirs(anime_path, exist_ok=True)
|
||||
|
||||
session_factory = get_async_session_factory()
|
||||
db = session_factory()
|
||||
try:
|
||||
existing = await AnimeSeriesService.get_by_key(db, serie.key)
|
||||
if existing:
|
||||
logger.debug(
|
||||
"Series '%s' (key=%s) already exists in DB, skipping",
|
||||
serie.name, serie.key
|
||||
)
|
||||
return True
|
||||
|
||||
anime_series = await AnimeSeriesService.create(
|
||||
db=db,
|
||||
key=serie.key,
|
||||
name=serie.name,
|
||||
site=serie.site,
|
||||
folder=folder_name,
|
||||
year=serie.year
|
||||
)
|
||||
for season, eps in serie.episodeDict.items():
|
||||
for ep in eps:
|
||||
await EpisodeService.create(
|
||||
db=db,
|
||||
series_id=anime_series.id,
|
||||
season=season,
|
||||
episode_number=ep
|
||||
)
|
||||
await db.commit()
|
||||
self.keyDict[serie.key] = serie
|
||||
logger.info(
|
||||
"Persisted series '%s' to database",
|
||||
serie.name
|
||||
)
|
||||
return True
|
||||
except Exception as e:
|
||||
await db.rollback()
|
||||
logger.error(
|
||||
"Failed to persist series '%s' to DB: %s",
|
||||
serie.key, e, exc_info=True
|
||||
)
|
||||
return False
|
||||
finally:
|
||||
await db.close()
|
||||
except Exception as e:
|
||||
logger.error(
|
||||
"Could not add series '%s' to DB (DB unavailable?): %s",
|
||||
serie.key, e
|
||||
)
|
||||
return False
|
||||
|
||||
def contains(self, key: str) -> bool:
|
||||
"""
|
||||
Return True when a series identified by ``key`` already exists.
|
||||
|
||||
@@ -649,11 +649,11 @@ class EpisodeService:
|
||||
@staticmethod
|
||||
async def delete(db: AsyncSession, episode_id: int) -> bool:
|
||||
"""Delete episode.
|
||||
|
||||
|
||||
Args:
|
||||
db: Database session
|
||||
episode_id: Episode primary key
|
||||
|
||||
|
||||
Returns:
|
||||
True if deleted, False if not found
|
||||
"""
|
||||
@@ -662,6 +662,33 @@ class EpisodeService:
|
||||
)
|
||||
return result.rowcount > 0
|
||||
|
||||
@staticmethod
|
||||
async def delete_by_series(
|
||||
db: AsyncSession,
|
||||
series_id: int,
|
||||
season: int,
|
||||
episode_number: int,
|
||||
) -> bool:
|
||||
"""Delete episode by series ID, season, and episode number.
|
||||
|
||||
Args:
|
||||
db: Database session
|
||||
series_id: Foreign key to AnimeSeries
|
||||
season: Season number
|
||||
episode_number: Episode number within season
|
||||
|
||||
Returns:
|
||||
True if deleted, False if not found
|
||||
"""
|
||||
result = await db.execute(
|
||||
delete(Episode).where(
|
||||
Episode.series_id == series_id,
|
||||
Episode.season == season,
|
||||
Episode.episode_number == episode_number,
|
||||
)
|
||||
)
|
||||
return result.rowcount > 0
|
||||
|
||||
@staticmethod
|
||||
async def delete_by_series_and_episode(
|
||||
db: AsyncSession,
|
||||
|
||||
Reference in New Issue
Block a user