464 lines
18 KiB
Python
464 lines
18 KiB
Python
from flask import (Flask, jsonify, request, send_from_directory,
|
|
send_file, abort, session, redirect, url_for)
|
|
import json
|
|
import os
|
|
import requests
|
|
from bs4 import BeautifulSoup
|
|
import re
|
|
import hashlib
|
|
import secrets
|
|
import threading
|
|
from PIL import Image
|
|
|
|
# ── Config ────────────────────────────────────────────────────────────────
|
|
BASE_PREFIX = '/n64' # subpath for nginx reverse proxy
|
|
DATA_FILE = '/app/data/games.json'
|
|
SETTINGS_FILE= '/app/data/settings.json'
|
|
ART_BASE = '/app/coverart'
|
|
THUMB_DIR = '/app/data/thumbs' # persisted on the game-data volume
|
|
THUMB_WIDTH = 300 # px — enough for the card grid
|
|
|
|
app = Flask(__name__, static_folder='static')
|
|
app.secret_key = os.environ.get('SECRET_KEY', secrets.token_hex(32))
|
|
|
|
# ── Thumbnail generator ────────────────────────────────────────────────────
|
|
def _make_thumb(src_path, thumb_path):
|
|
"""Convert a large PNG to a small WebP thumbnail."""
|
|
try:
|
|
with Image.open(src_path) as img:
|
|
# Maintain aspect ratio, limit width to THUMB_WIDTH
|
|
w, h = img.size
|
|
ratio = THUMB_WIDTH / w
|
|
new_h = int(h * ratio)
|
|
img = img.convert('RGB')
|
|
img = img.resize((THUMB_WIDTH, new_h), Image.LANCZOS)
|
|
img.save(thumb_path, 'WEBP', quality=82, method=4)
|
|
except Exception as e:
|
|
print(f'Thumb error {src_path}: {e}')
|
|
|
|
def generate_thumbnails():
|
|
"""Walk ART_BASE and generate WebP thumbs for any PNG that needs one."""
|
|
os.makedirs(THUMB_DIR, exist_ok=True)
|
|
todo = []
|
|
for fname in os.listdir(ART_BASE):
|
|
if not fname.lower().endswith('.png'):
|
|
continue
|
|
src = os.path.join(ART_BASE, fname)
|
|
thumb = os.path.join(THUMB_DIR, os.path.splitext(fname)[0] + '.webp')
|
|
if not os.path.isfile(thumb):
|
|
todo.append((src, thumb))
|
|
if todo:
|
|
print(f'Generating {len(todo)} thumbnails...')
|
|
for src, thumb in todo:
|
|
_make_thumb(src, thumb)
|
|
print('Thumbnails done.')
|
|
|
|
# Run thumbnail generation in background so startup isn't blocked
|
|
threading.Thread(target=generate_thumbnails, daemon=True).start()
|
|
|
|
# ── Settings helpers ───────────────────────────────────────────────────────
|
|
DEFAULT_PASSWORD = os.environ.get('ADMIN_PASSWORD', 'n64admin')
|
|
|
|
def _hash(pw):
|
|
return hashlib.sha256(pw.encode()).hexdigest()
|
|
|
|
def load_settings():
|
|
if os.path.isfile(SETTINGS_FILE):
|
|
with open(SETTINGS_FILE) as f:
|
|
return json.load(f)
|
|
# First-run: persist hashed default
|
|
s = {'password_hash': _hash(DEFAULT_PASSWORD)}
|
|
save_settings(s)
|
|
return s
|
|
|
|
def save_settings(s):
|
|
with open(SETTINGS_FILE, 'w') as f:
|
|
json.dump(s, f, indent=2)
|
|
|
|
def check_password(pw):
|
|
return _hash(pw) == load_settings().get('password_hash', '')
|
|
|
|
def is_authed():
|
|
return session.get('authed') is True
|
|
|
|
def require_auth(f):
|
|
"""Decorator: returns 401 JSON if not authenticated."""
|
|
from functools import wraps
|
|
@wraps(f)
|
|
def wrapped(*args, **kwargs):
|
|
if not is_authed():
|
|
return jsonify({'error': 'Unauthorized'}), 401
|
|
return f(*args, **kwargs)
|
|
return wrapped
|
|
|
|
# ── Game data helpers ──────────────────────────────────────────────────────
|
|
def load_games():
|
|
with open(DATA_FILE, 'r') as f:
|
|
return json.load(f)
|
|
|
|
def save_games(games):
|
|
with open(DATA_FILE, 'w') as f:
|
|
json.dump(games, f, indent=2)
|
|
|
|
# ── Static / index ─────────────────────────────────────────────────────────
|
|
@app.route(BASE_PREFIX + '/')
|
|
@app.route(BASE_PREFIX)
|
|
def index():
|
|
return send_from_directory('static', 'index.html')
|
|
|
|
# ── Cover art ──────────────────────────────────────────────────────────────
|
|
@app.route(BASE_PREFIX + '/art/<path:art_path>')
|
|
def serve_art(art_path):
|
|
full = os.path.join(ART_BASE, art_path)
|
|
if not os.path.isfile(full):
|
|
abort(404)
|
|
|
|
# Serve pre-generated WebP thumbnail if available (much smaller)
|
|
base_name = os.path.splitext(art_path)[0]
|
|
thumb_path = os.path.join(THUMB_DIR, base_name + '.webp')
|
|
|
|
if os.path.isfile(thumb_path):
|
|
resp = send_file(thumb_path, mimetype='image/webp')
|
|
else:
|
|
# Thumb not ready yet — serve original and kick off generation
|
|
resp = send_file(full, mimetype='image/png')
|
|
threading.Thread(target=_make_thumb, args=(full, thumb_path), daemon=True).start()
|
|
|
|
# Cache aggressively — images never change
|
|
resp.headers['Cache-Control'] = 'public, max-age=31536000, immutable'
|
|
return resp
|
|
|
|
# ── Auth endpoints ─────────────────────────────────────────────────────────
|
|
@app.route(BASE_PREFIX + '/api/auth/login', methods=['POST'])
|
|
def login():
|
|
data = request.get_json(silent=True) or {}
|
|
pw = data.get('password', '')
|
|
if check_password(pw):
|
|
session['authed'] = True
|
|
return jsonify({'ok': True})
|
|
return jsonify({'ok': False, 'error': 'Wrong password'}), 401
|
|
|
|
@app.route(BASE_PREFIX + '/api/auth/logout', methods=['POST'])
|
|
def logout():
|
|
session.clear()
|
|
return jsonify({'ok': True})
|
|
|
|
@app.route(BASE_PREFIX + '/api/auth/status', methods=['GET'])
|
|
def auth_status():
|
|
return jsonify({'authed': is_authed()})
|
|
|
|
@app.route(BASE_PREFIX + '/api/auth/change-password', methods=['POST'])
|
|
@require_auth
|
|
def change_password():
|
|
data = request.get_json(silent=True) or {}
|
|
current = data.get('current', '')
|
|
new_pw = data.get('new', '')
|
|
if not check_password(current):
|
|
return jsonify({'ok': False, 'error': 'Current password incorrect'}), 400
|
|
if len(new_pw) < 4:
|
|
return jsonify({'ok': False, 'error': 'Password must be at least 4 characters'}), 400
|
|
s = load_settings()
|
|
s['password_hash'] = _hash(new_pw)
|
|
save_settings(s)
|
|
return jsonify({'ok': True})
|
|
|
|
# ── Public API ────────────────────────────────────────────────────────────
|
|
@app.route(BASE_PREFIX + '/api/games', methods=['GET'])
|
|
def get_games():
|
|
games = load_games()
|
|
query = request.args.get('q', '').lower().strip()
|
|
if query:
|
|
games = [g for g in games if query in g['title'].lower()]
|
|
return jsonify(games)
|
|
|
|
@app.route(BASE_PREFIX + '/api/games/owned', methods=['GET'])
|
|
def get_owned():
|
|
games = load_games()
|
|
return jsonify([g for g in games if g.get('owned')])
|
|
|
|
CASE_COST = 7.00
|
|
|
|
@app.route(BASE_PREFIX + '/api/stats', methods=['GET'])
|
|
def get_stats():
|
|
games = load_games()
|
|
total = len(games)
|
|
owned = sum(1 for g in games if g.get('owned'))
|
|
collection_value = sum(
|
|
g.get('loose_price') or 0
|
|
for g in games if g.get('owned') and g.get('loose_price')
|
|
)
|
|
unowned_games = [g for g in games if not g.get('owned')]
|
|
needs_cost = sum(
|
|
(g.get('loose_price') or 0) + CASE_COST
|
|
for g in unowned_games
|
|
)
|
|
return jsonify({
|
|
'total': total,
|
|
'owned': owned,
|
|
'unowned': total - owned,
|
|
'collection_value': round(collection_value, 2),
|
|
'needs_cost': round(needs_cost, 2)
|
|
})
|
|
|
|
# ── Protected write API ───────────────────────────────────────────────────
|
|
@app.route(BASE_PREFIX + '/api/games/<int:game_id>/own', methods=['POST'])
|
|
@require_auth
|
|
def toggle_own(game_id):
|
|
games = load_games()
|
|
for g in games:
|
|
if g['id'] == game_id:
|
|
g['owned'] = not g.get('owned', False)
|
|
save_games(games)
|
|
if g['owned'] and not g.get('cover_art_local'):
|
|
info = scrape_game_info(g['title'])
|
|
if info:
|
|
games = load_games()
|
|
for gg in games:
|
|
if gg['id'] == game_id:
|
|
gg.update(info)
|
|
break
|
|
save_games(games)
|
|
g.update(info)
|
|
return jsonify(g)
|
|
return jsonify({'error': 'Game not found'}), 404
|
|
|
|
@app.route(BASE_PREFIX + '/api/games/<int:game_id>/flags', methods=['POST'])
|
|
@require_auth
|
|
def update_flags(game_id):
|
|
games = load_games()
|
|
data = request.get_json(silent=True) or {}
|
|
for g in games:
|
|
if g['id'] == game_id:
|
|
for field in ('has_case', 'cleaned', 'battery_replaced'):
|
|
if field in data:
|
|
g[field] = bool(data[field])
|
|
save_games(games)
|
|
return jsonify(g)
|
|
return jsonify({'error': 'Game not found'}), 404
|
|
|
|
@app.route(BASE_PREFIX + '/api/games/<int:game_id>/info', methods=['GET'])
|
|
def get_game_info(game_id):
|
|
games = load_games()
|
|
for g in games:
|
|
if g['id'] == game_id:
|
|
if not g.get('cover_art_local') and not g.get('cover_art_remote'):
|
|
info = scrape_game_info(g['title'])
|
|
if info:
|
|
g.update(info)
|
|
save_games(games)
|
|
return jsonify(g)
|
|
return jsonify({'error': 'Game not found'}), 404
|
|
|
|
@app.route(BASE_PREFIX + '/api/rarity/refresh', methods=['POST'])
|
|
@require_auth
|
|
def refresh_rarity():
|
|
try:
|
|
rarity_data = scrape_rarityguide()
|
|
if not rarity_data:
|
|
return jsonify({'error': 'Scrape returned no data'}), 502
|
|
games = load_games()
|
|
updated = 0
|
|
for g in games:
|
|
match = _match_rarity(g['title'], rarity_data)
|
|
if match:
|
|
g['rarity_pct'] = match['rarity_pct']
|
|
g['rarity_label'] = match['rarity_label']
|
|
updated += 1
|
|
elif 'rarity_pct' not in g:
|
|
g['rarity_pct'] = None
|
|
g['rarity_label'] = None
|
|
save_games(games)
|
|
return jsonify({'updated': updated, 'total': len(games)})
|
|
except Exception as e:
|
|
print(f'Rarity refresh error: {e}')
|
|
return jsonify({'error': str(e)}), 500
|
|
|
|
@app.route(BASE_PREFIX + '/api/prices/refresh', methods=['POST'])
|
|
@require_auth
|
|
def refresh_prices():
|
|
try:
|
|
prices = scrape_pricecharting()
|
|
if not prices:
|
|
return jsonify({'error': 'Scrape returned no data'}), 502
|
|
games = load_games()
|
|
updated = 0
|
|
for g in games:
|
|
matched = _match_price(g['title'], prices)
|
|
if matched is not None:
|
|
g['loose_price'] = matched
|
|
updated += 1
|
|
elif 'loose_price' not in g:
|
|
g['loose_price'] = None
|
|
save_games(games)
|
|
collection_value = round(sum(
|
|
g.get('loose_price') or 0
|
|
for g in games if g.get('owned') and g.get('loose_price')
|
|
), 2)
|
|
return jsonify({'updated': updated, 'total': len(games), 'collection_value': collection_value})
|
|
except Exception as e:
|
|
print(f'Price refresh error: {e}')
|
|
return jsonify({'error': str(e)}), 500
|
|
|
|
# ── Scrapers ───────────────────────────────────────────────────────────────
|
|
def scrape_rarityguide():
|
|
headers = {'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36'}
|
|
base = 'https://www.rarityguide.com/n64_view.php'
|
|
results = {}
|
|
for first in [None, 101, 201]:
|
|
params = {'SortDirection': 'desc', 'SortField': '5', 'recordsPerPage': '100'}
|
|
if first:
|
|
params['FirstRecord'] = str(first)
|
|
try:
|
|
resp = requests.get(base, params=params, headers=headers, timeout=20)
|
|
if resp.status_code != 200:
|
|
continue
|
|
except Exception:
|
|
continue
|
|
soup = BeautifulSoup(resp.text, 'html.parser')
|
|
for row in soup.find_all('tr'):
|
|
tds = row.find_all('td')
|
|
if len(tds) < 5:
|
|
continue
|
|
title = tds[1].get_text(strip=True)
|
|
if not title:
|
|
continue
|
|
img = tds[4].find('img')
|
|
if not img:
|
|
continue
|
|
alt = img.get('alt', '')
|
|
m = re.match(r'(\d+)\s*percent\s*\(([^)]+)\)', alt)
|
|
if not m:
|
|
continue
|
|
results[title] = {'rarity_pct': int(m.group(1)), 'rarity_label': m.group(2).strip()}
|
|
return results if results else None
|
|
|
|
def _match_rarity(title, rarity_data):
|
|
import difflib, unicodedata
|
|
def norm(s):
|
|
s = unicodedata.normalize('NFD', s)
|
|
s = ''.join(c for c in s if unicodedata.category(c) != 'Mn')
|
|
s = s.lower()
|
|
s = re.sub(r'[^a-z0-9\s\-]', '', s)
|
|
s = re.sub(r'\s+', ' ', s).strip()
|
|
for art in ('the ', 'a ', 'an '):
|
|
if s.startswith(art): s = s[len(art):]
|
|
return s
|
|
tn = norm(title)
|
|
for k, v in rarity_data.items():
|
|
if norm(k) == tn: return v
|
|
keys = list(rarity_data.keys())
|
|
nkeys = [norm(k) for k in keys]
|
|
matches = difflib.get_close_matches(tn, nkeys, n=1, cutoff=0.85)
|
|
if matches:
|
|
return rarity_data[keys[nkeys.index(matches[0])]]
|
|
return None
|
|
|
|
def scrape_pricecharting():
|
|
headers = {'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36'}
|
|
base_url = 'https://www.pricecharting.com/console/nintendo-64'
|
|
prices = {}
|
|
session = requests.Session()
|
|
resp = session.get(base_url, headers=headers, timeout=20)
|
|
if resp.status_code != 200:
|
|
return None
|
|
|
|
def parse_page(html):
|
|
soup = BeautifulSoup(html, 'html.parser')
|
|
table = soup.find('table', id='games_table')
|
|
if not table: return soup, False
|
|
for row in table.find_all('tr'):
|
|
title_td = row.find('td', class_='title')
|
|
price_td = row.find('td', class_='used_price')
|
|
if not title_td or not price_td: continue
|
|
a = title_td.find('a')
|
|
if not a: continue
|
|
span = price_td.find('span', class_='js-price')
|
|
price_text = span.get_text(strip=True) if span else 'N/A'
|
|
try:
|
|
prices[a.get_text(strip=True)] = float(price_text.replace('$','').replace(',',''))
|
|
except (ValueError, AttributeError):
|
|
prices[a.get_text(strip=True)] = None
|
|
form = soup.find('form', {'id': re.compile(r'pagination')}) or soup.find('input', {'name': 'cursor'})
|
|
if form:
|
|
cursor_input = soup.find('input', {'name': 'cursor'})
|
|
return soup, cursor_input
|
|
return soup, None
|
|
|
|
soup, cursor_input = parse_page(resp.text)
|
|
while cursor_input:
|
|
cursor_val = cursor_input.get('value', '')
|
|
try:
|
|
resp = session.post(base_url, headers=headers, timeout=20,
|
|
data={'cursor': cursor_val, 'q': '', 'sort': 'name'})
|
|
if resp.status_code != 200: break
|
|
except Exception: break
|
|
_, cursor_input = parse_page(resp.text)
|
|
return prices
|
|
|
|
def _normalize(s):
|
|
s = s.lower()
|
|
s = re.sub(r'[éèê]', 'e', s); s = re.sub(r'[óò]', 'o', s)
|
|
s = re.sub(r'[úù]', 'u', s); s = re.sub(r'°', '', s)
|
|
s = re.sub(r'[^a-z0-9\s]', '', s)
|
|
s = re.sub(r'\s+', ' ', s).strip()
|
|
for art in ('the ', 'a ', 'an '):
|
|
if s.startswith(art): s = s[len(art):]
|
|
return s
|
|
|
|
def _match_price(title, prices):
|
|
if title in prices: return prices[title]
|
|
tl = title.lower()
|
|
for k, v in prices.items():
|
|
if k.lower() == tl: return v
|
|
tn = _normalize(title)
|
|
for k, v in prices.items():
|
|
if _normalize(k) == tn: return v
|
|
return None
|
|
|
|
def scrape_game_info(title):
|
|
try:
|
|
from urllib.parse import quote
|
|
headers = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'}
|
|
resp = requests.get(f'https://gamesdb.launchbox-app.com/games/results/{quote(title)}',
|
|
headers=headers, timeout=15)
|
|
if resp.status_code != 200: return None
|
|
soup = BeautifulSoup(resp.text, 'html.parser')
|
|
title_lower = title.lower().strip()
|
|
clean_title = re.sub(r'[^a-z0-9\s]', '', title_lower).strip()
|
|
game_link = None
|
|
for link in soup.find_all('a', href=True):
|
|
href = link['href']
|
|
if '/games/details/' not in href: continue
|
|
link_text = link.get_text(strip=True)
|
|
if 'Nintendo 64' not in link_text or 'Released' not in link_text: continue
|
|
if re.sub(r'[^a-z0-9\s]', '', link_text.lower()).startswith(clean_title):
|
|
game_link = href; break
|
|
if not game_link: return None
|
|
resp2 = requests.get(f'https://gamesdb.launchbox-app.com{game_link}', headers=headers, timeout=15)
|
|
if resp2.status_code != 200: return None
|
|
soup2 = BeautifulSoup(resp2.text, 'html.parser')
|
|
info = {}
|
|
for img in soup2.find_all('img'):
|
|
if 'launchbox-app.com' in img.get('src','') and 'm-auto' in img.get('class',[]):
|
|
info['cover_art_remote'] = img['src']; break
|
|
for dt in soup2.find_all('dt'):
|
|
dd = dt.find_next_sibling('dd')
|
|
if not dd: continue
|
|
label = dt.get_text(strip=True)
|
|
value = dd.get_text(strip=True)
|
|
if label == 'Developers': info['developer'] = value[:100]
|
|
elif label == 'Genre': info['genres'] = value[:100]
|
|
elif label == 'ESRB': info['rating'] = value[:50]
|
|
elif label == 'Max Players': info['players'] = value[:50]
|
|
if not info.get('description'):
|
|
for p in soup2.find_all('p'):
|
|
text = p.get_text(strip=True)
|
|
if len(text) > 80:
|
|
info['description'] = text[:500]; break
|
|
return info if info else None
|
|
except Exception as e:
|
|
print(f"Scraping error for '{title}': {e}")
|
|
return None
|
|
|
|
if __name__ == '__main__':
|
|
app.run(host='0.0.0.0', port=5000, debug=True)
|