420 lines
22 KiB
Python
420 lines
22 KiB
Python
"""Metadata-only catalogue synchronization; publication is the caller's job.
|
|
|
|
Only fixed-origin, allowlisted metadata GETs are possible. Image/video URLs are
|
|
opaque stored metadata and are never followed. Supply the key from the calling
|
|
process (e.g. REELBOX_UPSTREAM_KEY). Reports and errors contain no request URLs.
|
|
"""
|
|
from contextlib import closing
|
|
from datetime import datetime, timezone
|
|
import http.client
|
|
import json
|
|
import math
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import sqlite3
|
|
import ssl
|
|
import time
|
|
from urllib.error import HTTPError, URLError
|
|
from urllib.parse import quote, urlsplit
|
|
from urllib.request import HTTPRedirectHandler, Request, build_opener
|
|
|
|
|
|
UPSTREAM_BASE = 'https://ninwinios.info/api/'
|
|
MAX_RESPONSE_BYTES = 32 * 1024 * 1024
|
|
_ROUTE = re.compile(r'(?:genre/all|country/all|(?:movie|serie)/by/filtres/0/created/[0-9]{1,6}'
|
|
r'|movie/by/[1-9][0-9]{0,9}|season/by/serie/[1-9][0-9]{0,9})')
|
|
|
|
|
|
class SyncError(RuntimeError):
|
|
"""Safe, key-free error suitable for a scheduler log."""
|
|
|
|
|
|
class _NoRedirect(HTTPRedirectHandler):
|
|
def redirect_request(self, req, fp, code, msg, headers, newurl):
|
|
raise SyncError('Upstream redirect denied')
|
|
|
|
|
|
def _invalid_constant(_value):
|
|
raise ValueError('Non-finite JSON number')
|
|
|
|
|
|
class Upstream:
|
|
"""Single-worker rate-limited client. ``attempts`` is total attempts, <=10.
|
|
|
|
``opener`` is an injectable trusted urllib-compatible opener for tests.
|
|
``cache_dir`` is reserved; responses are intentionally not cached on disk,
|
|
so a subsequent run always refreshes metadata. No response/key is logged.
|
|
"""
|
|
def __init__(self, key, interval=0.25, timeout=30, attempts=3, opener=None, cache_dir=None):
|
|
if not isinstance(key, str) or not key or len(key) > 4096:
|
|
raise ValueError('Upstream key required')
|
|
if (type(attempts) is not int or not 1 <= attempts <= 10
|
|
or not isinstance(interval, (int, float)) or not math.isfinite(interval) or interval < 0
|
|
or not isinstance(timeout, (int, float)) or not math.isfinite(timeout) or timeout <= 0):
|
|
raise ValueError('Invalid upstream retry or timing configuration')
|
|
self._key = key
|
|
self.interval, self.timeout, self.attempts = interval, timeout, attempts
|
|
self._opener = opener if opener is not None else build_opener(_NoRedirect())
|
|
self.requests_count = 0
|
|
self._last_request = None
|
|
|
|
def get(self, path):
|
|
if not isinstance(path, str) or not _ROUTE.fullmatch(path):
|
|
raise ValueError('Unsupported upstream metadata route')
|
|
url = UPSTREAM_BASE + path + '/' + quote(self._key, safe='') + '/'
|
|
parsed = urlsplit(url)
|
|
if (parsed.scheme, parsed.netloc, parsed.query, parsed.fragment) != ('https', 'ninwinios.info', '', ''):
|
|
raise ValueError('Unsupported upstream origin')
|
|
request = Request(url, headers={'Accept': 'application/json', 'Accept-Encoding': 'identity'}, method='GET')
|
|
for attempt in range(self.attempts):
|
|
if self._last_request is not None:
|
|
delay = self.interval - (time.monotonic() - self._last_request)
|
|
if delay > 0:
|
|
time.sleep(delay)
|
|
self._last_request = time.monotonic()
|
|
self.requests_count += 1
|
|
try:
|
|
with self._opener.open(request, timeout=self.timeout) as response:
|
|
status = response.status
|
|
if status == 429 or 500 <= status <= 599:
|
|
raise HTTPError(url, status, 'Upstream temporarily unavailable', {}, None)
|
|
if status != 200 or response.geturl() != url:
|
|
raise SyncError('Upstream status or redirect denied')
|
|
media_type = response.headers.get('Content-Type', '').split(';')[0].strip().lower()
|
|
# The original PHP API labels valid JSON as text/html.
|
|
# Only allow this known legacy label; parsing/schema checks still apply.
|
|
if media_type and media_type not in {'application/json', 'text/html'} and not media_type.endswith('+json'):
|
|
raise SyncError('Upstream did not return JSON metadata')
|
|
declared = response.headers.get('Content-Length')
|
|
if declared is not None:
|
|
if not re.fullmatch(r'[0-9]{1,12}', declared) or int(declared) > MAX_RESPONSE_BYTES:
|
|
raise SyncError('Invalid upstream metadata length')
|
|
body = response.read(MAX_RESPONSE_BYTES + 1)
|
|
if len(body) > MAX_RESPONSE_BYTES:
|
|
raise SyncError('Upstream metadata response exceeds safety limit')
|
|
if declared is not None and len(body) != int(declared):
|
|
raise SyncError('Truncated upstream metadata response')
|
|
value = json.loads(body, parse_constant=_invalid_constant)
|
|
detail = path.startswith('movie/by/') and '/filtres/' not in path
|
|
if (detail and not isinstance(value, dict)) or (not detail and not isinstance(value, list)):
|
|
raise SyncError('Unexpected upstream metadata shape')
|
|
return value
|
|
except HTTPError as error:
|
|
retry = error.code == 429 or 500 <= error.code <= 599
|
|
error.close()
|
|
if not retry or attempt + 1 == self.attempts:
|
|
raise SyncError('Upstream HTTP request failed') from None
|
|
except (ssl.SSLError, ValueError, RecursionError):
|
|
raise SyncError('Invalid upstream TLS or JSON metadata') from None
|
|
except (URLError, OSError, http.client.HTTPException) as error:
|
|
if isinstance(getattr(error, 'reason', None), ssl.SSLError) or attempt + 1 == self.attempts:
|
|
raise SyncError('Upstream transport request failed') from None
|
|
if attempt + 1 < self.attempts:
|
|
time.sleep(min(2 ** attempt, 8))
|
|
raise SyncError('Upstream request exhausted')
|
|
|
|
|
|
try:
|
|
from .catalog_api import allowed_text
|
|
except ImportError: # Also support `PYTHONPATH=server; import catalog_sync`.
|
|
from catalog_api import allowed_text
|
|
|
|
|
|
def _encode(value):
|
|
return json.dumps(value, ensure_ascii=False, separators=(',', ':'), allow_nan=False)
|
|
|
|
|
|
def _id(value):
|
|
if type(value) is not int or not 1 <= value <= 9999999999:
|
|
raise SyncError('Malformed metadata identifier')
|
|
return value
|
|
|
|
|
|
def _objects(value):
|
|
if not isinstance(value, list) or any(not isinstance(item, dict) for item in value):
|
|
raise SyncError('Malformed metadata list')
|
|
return value
|
|
|
|
|
|
def _merge(old, incoming):
|
|
result = dict(old)
|
|
for key, value in incoming.items():
|
|
if key in {'sources', 'episodes', 'genres', 'country'}:
|
|
result[key] = _merge_sequence(old.get(key, []), value)
|
|
elif key not in result or (value is not None and value != ''):
|
|
result[key] = value
|
|
return result
|
|
|
|
|
|
def _merge_sequence(old, incoming):
|
|
"""Retain old positions; stable IDs win, missing IDs fall back to index."""
|
|
old, incoming = _objects(old), _objects(incoming)
|
|
result = list(old)
|
|
by_id = {item['id']: index for index, item in enumerate(old) if item.get('id') is not None}
|
|
seen_ids, matched = set(), set()
|
|
for index, item in enumerate(incoming):
|
|
identity = item.get('id')
|
|
if identity is not None:
|
|
_id(identity)
|
|
if identity in seen_ids:
|
|
raise SyncError('Duplicate nested metadata identifier')
|
|
seen_ids.add(identity)
|
|
target = by_id.get(identity)
|
|
if target is None and index < len(old) and old[index].get('id') is None:
|
|
target = index
|
|
else:
|
|
target = index if index < len(old) else None
|
|
if target is not None:
|
|
if target in matched:
|
|
raise SyncError('Ambiguous nested metadata identity')
|
|
matched.add(target)
|
|
result[target] = _merge(old[target], item)
|
|
else:
|
|
result.append(_merge({}, item))
|
|
return result
|
|
|
|
|
|
def _validate_record(item, kind, require_title=False):
|
|
if not isinstance(item, dict) or item.get('error') or item.get('errors'):
|
|
raise SyncError('Malformed metadata record')
|
|
if item.get('id') is not None:
|
|
_id(item['id'])
|
|
if kind in {'season', 'episode', 'source'} and item.get('id') is None:
|
|
identity = item.get('url') if kind == 'source' else item.get('title')
|
|
if not isinstance(identity, str) or not identity.strip():
|
|
raise SyncError('Nested metadata has no usable identity')
|
|
if require_title and (not isinstance(item.get('title'), str) or not item['title'].strip()):
|
|
raise SyncError('New title has no usable name')
|
|
for field in ('title', 'description', 'duration', 'classification', 'image', 'cover',
|
|
'downloadas', 'playas', 'type', 'quality', 'url'):
|
|
if field in item and item[field] is not None and not isinstance(item[field], str):
|
|
raise SyncError('Malformed text metadata field')
|
|
for field in ('year', 'imdb', 'rating'):
|
|
value = item.get(field)
|
|
if value is not None and (type(value) not in {int, float} or not math.isfinite(value)):
|
|
raise SyncError('Malformed numeric metadata field')
|
|
if 'comment' in item and item['comment'] is not None and type(item['comment']) not in {bool, int}:
|
|
raise SyncError('Malformed comment metadata flag')
|
|
for field, nested_kind in [('sources', 'source'), ('episodes', 'episode'),
|
|
('genres', 'reference'), ('country', 'reference')]:
|
|
if field in item:
|
|
for nested in _objects(item[field]):
|
|
_validate_record(nested, nested_kind)
|
|
|
|
|
|
def _reference(db, table, item):
|
|
_validate_record(item, 'reference')
|
|
_id(item.get('id'))
|
|
columns = ['id', 'title', 'image'] if table == 'countries' else ['id', 'title']
|
|
db.execute('INSERT INTO ' + table + '(' + ','.join(columns) + ') VALUES (' +
|
|
','.join('?' for _ in columns) + ') ON CONFLICT(id) DO UPDATE SET ' +
|
|
','.join(c + '=COALESCE(excluded.' + c + ',' + table + '.' + c + ')' for c in columns[1:]),
|
|
[item.get(c) for c in columns])
|
|
|
|
|
|
def _save_title(db, incoming):
|
|
title_id = _id(incoming.get('id'))
|
|
record = db.execute('SELECT raw_json FROM titles WHERE id=?', (title_id,)).fetchone()
|
|
_validate_record(incoming, 'title', require_title=not record)
|
|
old = json.loads(record[0]) if record else {}
|
|
merged = _merge(old, incoming)
|
|
raw = _encode(merged)
|
|
if record and old == merged:
|
|
return False
|
|
fields = ['id', 'type', 'title', 'description', 'year', 'imdb', 'rating', 'duration',
|
|
'classification', 'image', 'cover', 'downloadas', 'playas', 'comments_enabled', 'raw_json']
|
|
values = [merged.get(c) for c in fields]
|
|
values[-2], values[-1] = merged.get('comment'), raw
|
|
db.execute('INSERT INTO titles(' + ','.join(fields) + ') VALUES (' + ','.join('?' for _ in fields) +
|
|
') ON CONFLICT(id) DO UPDATE SET ' + ','.join(c + '=excluded.' + c for c in fields[1:]), values)
|
|
for field, table, junction, column in [('genres', 'genres', 'title_genres', 'genre_id'),
|
|
('country', 'countries', 'title_countries', 'country_id')]:
|
|
for tag in _objects(merged.get(field, [])):
|
|
_reference(db, table, tag)
|
|
db.execute('INSERT OR IGNORE INTO ' + junction + '(title_id,' + column + ') VALUES (?,?)',
|
|
(title_id, tag['id']))
|
|
for index, source in enumerate(_objects(merged.get('sources', []))):
|
|
db.execute('INSERT INTO movie_sources VALUES (?,?,?,?,?,?,?) ON CONFLICT(title_id,source_index) '
|
|
'DO UPDATE SET source_id=excluded.source_id,quality=excluded.quality,'
|
|
'media_type=excluded.media_type,url=excluded.url,raw_json=excluded.raw_json',
|
|
(title_id, index, source.get('id'), source.get('quality'), source.get('type'),
|
|
source.get('url'), _encode(source)))
|
|
return True
|
|
|
|
|
|
def _save_seasons(db, series_id, incoming):
|
|
for season in _objects(incoming):
|
|
_validate_record(season, 'season')
|
|
previous = db.execute('SELECT season_index,raw_json FROM seasons WHERE series_id=? ORDER BY season_index',
|
|
(series_id,)).fetchall()
|
|
old = [json.loads(raw) for _, raw in previous]
|
|
merged = _merge_sequence(old, incoming)
|
|
changed = False
|
|
# Existing indices are retained even if an exported snapshot has gaps.
|
|
indices = [index for index, _ in previous]
|
|
next_index = max(indices, default=-1) + 1
|
|
indices += list(range(next_index, next_index + len(merged) - len(indices)))
|
|
for position, season in enumerate(merged):
|
|
if position < len(old) and season == old[position]:
|
|
continue
|
|
changed = True
|
|
si = indices[position]
|
|
db.execute('INSERT INTO seasons VALUES (?,?,?,?,?) ON CONFLICT(series_id,season_index) DO UPDATE SET '
|
|
'id=excluded.id,title=excluded.title,raw_json=excluded.raw_json',
|
|
(series_id, si, season.get('id'), season.get('title'), _encode(season)))
|
|
for ei, episode in enumerate(_objects(season.get('episodes', []))):
|
|
raw = _encode(episode)
|
|
current = db.execute('SELECT raw_json FROM episodes WHERE series_id=? AND season_index=? AND episode_index=?',
|
|
(series_id, si, ei)).fetchone()
|
|
if current and json.loads(current[0]) == episode:
|
|
continue
|
|
db.execute('INSERT INTO episodes VALUES (?,?,?,?,?,?,?,?,?,?) '
|
|
'ON CONFLICT(series_id,season_index,episode_index) DO UPDATE SET '
|
|
'id=excluded.id,title=excluded.title,description=excluded.description,duration=excluded.duration,'
|
|
'downloadas=excluded.downloadas,playas=excluded.playas,raw_json=excluded.raw_json',
|
|
(series_id, si, ei, episode.get('id'), episode.get('title'), episode.get('description'),
|
|
episode.get('duration'), episode.get('downloadas'), episode.get('playas'), raw))
|
|
for source_index, source in enumerate(_objects(episode.get('sources', []))):
|
|
db.execute('INSERT INTO episode_sources VALUES (?,?,?,?,?,?,?,?,?) '
|
|
'ON CONFLICT(series_id,season_index,episode_index,source_index) DO UPDATE SET '
|
|
'source_id=excluded.source_id,quality=excluded.quality,media_type=excluded.media_type,'
|
|
'url=excluded.url,raw_json=excluded.raw_json',
|
|
(series_id, si, ei, source_index, source.get('id'), source.get('quality'),
|
|
source.get('type'), source.get('url'), _encode(source)))
|
|
db.execute('INSERT INTO series_fetch_status VALUES (?,?,?,?) ON CONFLICT(series_id) DO UPDATE SET '
|
|
'status=excluded.status,season_count=excluded.season_count,source_url=excluded.source_url',
|
|
(series_id, 'fetched', len(merged), UPSTREAM_BASE + f'season/by/serie/{series_id}/'))
|
|
return changed
|
|
|
|
|
|
def _counts(db):
|
|
tables = ('titles', 'genres', 'countries', 'movie_sources', 'seasons', 'episodes', 'episode_sources')
|
|
counts = {name: db.execute('SELECT COUNT(*) FROM ' + name).fetchone()[0] for name in tables}
|
|
for name, kind in [('movies', 'movie'), ('series', 'serie')]:
|
|
counts[name] = db.execute('SELECT COUNT(*) FROM titles WHERE type=?', (kind,)).fetchone()[0]
|
|
counts['sources'] = counts['movie_sources'] + counts['episode_sources']
|
|
counts['visible_titles'] = sum(1 for kind, title, description in db.execute('SELECT type,title,description FROM titles')
|
|
if kind in {'movie', 'serie'} and allowed_text(title) and allowed_text(description))
|
|
return counts
|
|
|
|
|
|
def synchronize(source_db: Path, staged_db: Path, upstream, max_pages=2000, progress=None) -> dict:
|
|
"""Copy a snapshot into an exclusively created candidate and refresh metadata.
|
|
|
|
Never publishes or mutates the source. A failed candidate is removed. The
|
|
caller owns locking, free-space checks, atomic publication and API reload.
|
|
``progress`` accepts dictionaries containing only phase/count/total.
|
|
Both catalogue types must reach an empty page within ``max_pages`` each;
|
|
all visible series are refreshed, including those absent from those pages.
|
|
The returned updated-series IDs are a sorted sample of at most 100, with
|
|
``updated_series_count`` reporting the full count. No media is fetched.
|
|
"""
|
|
if type(max_pages) is not int or not 1 <= max_pages <= 1000000:
|
|
raise ValueError('Invalid catalogue page limit')
|
|
source_db, staged_db = Path(source_db), Path(staged_db)
|
|
if source_db.resolve() == staged_db.resolve():
|
|
raise ValueError('Candidate must differ from source')
|
|
created, db = False, None
|
|
start_requests = getattr(upstream, 'requests_count', None)
|
|
|
|
def emit(phase, count, total):
|
|
if progress is not None:
|
|
progress({'phase': phase, 'count': count, 'total': total})
|
|
|
|
try:
|
|
emit('copy', 0, 1)
|
|
with closing(sqlite3.connect(source_db.resolve().as_uri() + '?mode=ro', uri=True)) as source:
|
|
fd = os.open(staged_db, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600)
|
|
created = True
|
|
os.close(fd)
|
|
db = sqlite3.connect(staged_db)
|
|
source.backup(db)
|
|
emit('copy', 1, 1)
|
|
db.execute('PRAGMA foreign_keys=ON')
|
|
db.execute('PRAGMA journal_mode=DELETE')
|
|
db.execute('BEGIN IMMEDIATE')
|
|
before = _counts(db)
|
|
metadata = {k: json.loads(v) for k, v in db.execute('SELECT * FROM metadata')}
|
|
for count, (route, table) in enumerate([('genre/all', 'genres'), ('country/all', 'countries')], 1):
|
|
for item in _objects(upstream.get(route)):
|
|
_reference(db, table, item)
|
|
emit('references', count, 2)
|
|
updated_series = set()
|
|
for kind in ('movie', 'serie'):
|
|
seen = set()
|
|
for page in range(max_pages):
|
|
items = _objects(upstream.get(f'{kind}/by/filtres/0/created/{page}'))
|
|
emit(kind + '_pages', page + 1, max_pages)
|
|
if not items:
|
|
break
|
|
for item in items:
|
|
title_id = _id(item.get('id'))
|
|
if title_id in seen:
|
|
continue
|
|
seen.add(title_id)
|
|
if item.get('type') == 'channel':
|
|
continue
|
|
if item.get('type') != kind:
|
|
raise SyncError('Unexpected catalogue title type')
|
|
if not allowed_text(item.get('title')) or not allowed_text(item.get('description')):
|
|
continue
|
|
exists = db.execute('SELECT type FROM titles WHERE id=?', (title_id,)).fetchone()
|
|
if exists and exists[0] != kind:
|
|
raise SyncError('Catalogue title type changed')
|
|
if kind == 'movie' and not exists:
|
|
detail = upstream.get(f'movie/by/{title_id}')
|
|
if not isinstance(detail, dict) or detail.get('id') != title_id or detail.get('type') != kind:
|
|
raise SyncError('Malformed movie detail')
|
|
item = _merge(item, detail)
|
|
if not allowed_text(item.get('title')) or not allowed_text(item.get('description')):
|
|
continue
|
|
if _save_title(db, item) and kind == 'serie':
|
|
updated_series.add(title_id)
|
|
else:
|
|
raise SyncError('Catalogue pagination did not reach an empty terminal page')
|
|
series = [row[0] for row in db.execute("SELECT id,title,description FROM titles WHERE type='serie'")
|
|
if allowed_text(row[1]) and allowed_text(row[2])]
|
|
for index, series_id in enumerate(series, 1):
|
|
seasons = _objects(upstream.get(f'season/by/serie/{series_id}'))
|
|
if _save_seasons(db, series_id, seasons):
|
|
updated_series.add(series_id)
|
|
emit('series', index, len(series))
|
|
after = _counts(db)
|
|
completed = datetime.now(timezone.utc).isoformat()
|
|
report = {'status': 'ok', 'counts': after,
|
|
'added': {name: after[name] - before[name] for name in
|
|
('titles', 'movies', 'series', 'episodes', 'sources')},
|
|
'updated_series_ids': sorted(updated_series)[:100], 'updated_series_count': len(updated_series),
|
|
'media_files_downloaded': False, 'synchronized_at_utc': completed,
|
|
'upstream_base': UPSTREAM_BASE}
|
|
end_requests = getattr(upstream, 'requests_count', None)
|
|
if type(start_requests) is int and type(end_requests) is int:
|
|
report['requests_count'] = end_requests - start_requests
|
|
metadata.setdefault('original_snapshot_built_at_utc', metadata.get('built_at_utc'))
|
|
metadata.update({'built_at_utc': completed, 'synchronized_at_utc': completed,
|
|
'sync_interval_seconds': 259200, 'media_files_downloaded': False,
|
|
'counts': after, 'last_sync': report, 'catalog_pagination_complete': True,
|
|
'series_episode_fetch_complete': True})
|
|
db.executemany('INSERT INTO metadata VALUES (?,?) ON CONFLICT(key) DO UPDATE SET value_json=excluded.value_json',
|
|
[(k, _encode(v)) for k, v in metadata.items()])
|
|
if db.execute('PRAGMA integrity_check').fetchall() != [('ok',)]:
|
|
raise SyncError('Candidate integrity check failed')
|
|
if db.execute('PRAGMA foreign_key_check').fetchall():
|
|
raise SyncError('Candidate foreign key check failed')
|
|
db.commit()
|
|
db.close()
|
|
db = None
|
|
emit('complete', 1, 1)
|
|
return report
|
|
except BaseException as error:
|
|
if db is not None:
|
|
db.close()
|
|
if created:
|
|
for suffix in ('', '-journal', '-wal', '-shm'):
|
|
Path(str(staged_db) + suffix).unlink(missing_ok=True)
|
|
if isinstance(error, (KeyboardInterrupt, SystemExit, FileExistsError, SyncError)):
|
|
raise
|
|
raise SyncError('Catalogue synchronization failed; candidate discarded') from None
|
|
|