nocci/GameKeyManager
nocci/GameKeyManager
1
0
Fork 0
Code Issues Pull requests 1 Projects Releases 3 Packages Wiki Activity

Version 1.0

Browse source
This commit is contained in:
nocci 2025-04-26 14:32:07 +02:00
parent 4bebbb27e4
commit eaa508a8df
33 changed files with 2822 additions and 994 deletions
Download patch file Download diff file Expand all files Collapse all files

522
steam-gift-manager/app.py
View file

@ -1,26 +1,90 @@
from flask import Flask, render_template, request, redirect, url_for, flash, make_response, session, abort, send_file
import os
import logging
import warnings
from sqlalchemy.exc import LegacyAPIWarning
warnings.simplefilter("ignore", category=LegacyAPIWarning)
from flask import Flask, render_template, request, redirect, url_for, flash, make_response, session, abort, send_file, jsonify
from flask_sqlalchemy import SQLAlchemy
from flask_login import LoginManager, UserMixin, login_user, logout_user, login_required, current_user
from flask_babel import Babel, _
from werkzeug.security import generate_password_hash, check_password_hash
from datetime import datetime
import os
from datetime import datetime, timedelta
from flask_wtf import CSRFProtect
from flask import abort
import io
import warnings
import re
import io
import csv
import secrets
import requests
from dotenv import load_dotenv
load_dotenv(override=True)
from sqlalchemy.exc import IntegrityError
from apscheduler.schedulers.background import BackgroundScheduler
import atexit
from flask_migrate import Migrate
from sqlalchemy import MetaData
from reportlab.pdfgen import canvas
from reportlab.lib.pagesizes import A4, landscape, letter
from reportlab.platypus import (
SimpleDocTemplate,
Table,
TableStyle,
Paragraph,
Image,
Spacer
)
from reportlab.lib import colors
from reportlab.lib.styles import getSampleStyleSheet, ParagraphStyle
from reportlab.lib.utils import ImageReader
from reportlab.lib.units import cm, inch, mm
from io import BytesIO
import reportlab.lib
app = Flask(__name__)
app.config['SECRET_KEY'] = os.urandom(24)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:////app/data/games.db'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
app.config['BABEL_DEFAULT_LOCALE'] = 'de'
app.config['BABEL_SUPPORTED_LOCALES'] = ['de', 'en']
app.config['BABEL_TRANSLATION_DIRECTORIES'] = 'translations'
csrf = CSRFProtect(app)
db = SQLAlchemy(app)
convention = {
"ix": "ix_%(column_0_label)s",
"uq": "uq_%(table_name)s_%(column_0_name)s",
"ck": "ck_%(table_name)s_%(constraint_name)s",
"fk": "fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s",
"pk": "pk_%(table_name)s"
}
metadata = MetaData(naming_convention=convention)
load_dotenv(override=True)
# Lade Umgebungsvariablen aus .env mit override
load_dotenv(override=True)
# Konfiguration
app.config.update(
SECRET_KEY=os.getenv('SECRET_KEY'),
SQLALCHEMY_DATABASE_URI=('sqlite:////app/data/games.db'),
SQLALCHEMY_TRACK_MODIFICATIONS=False,
BABEL_DEFAULT_LOCALE=os.getenv('BABEL_DEFAULT_LOCALE'),
BABEL_SUPPORTED_LOCALES=os.getenv('BABEL_SUPPORTED_LOCALES').split(','),
BABEL_TRANSLATION_DIRECTORIES=os.getenv('BABEL_TRANSLATION_DIRECTORIES'),
SESSION_COOKIE_SECURE=os.getenv('SESSION_COOKIE_SECURE') == 'True',
WTF_CSRF_ENABLED=os.getenv('CSRF_ENABLED') == 'True',
REGISTRATION_ENABLED=os.getenv('REGISTRATION_ENABLED', 'True').lower() == 'true'
)
interval_hours = int(os.getenv('CHECK_EXPIRING_KEYS_INTERVAL_HOURS', 12))
# Initialisierung
db = SQLAlchemy(app, metadata=metadata)
migrate = Migrate(app, db)
login_manager = LoginManager(app)
login_manager.login_view = 'login'
babel = Babel(app)
# Logging
app.logger.addHandler(logging.StreamHandler())
app.logger.setLevel(logging.INFO)
@babel.localeselector
def get_locale():
if 'lang' in session and session['lang'] in app.config['BABEL_SUPPORTED_LOCALES']:
@ -34,36 +98,63 @@ def inject_template_vars():
theme='dark' if request.cookies.get('dark_mode') == 'true' else 'light'
)
class User(UserMixin, db.Model):
# Datenbankmodelle
class User(db.Model, UserMixin):
__tablename__ = 'users'
id = db.Column(db.Integer, primary_key=True)
username = db.Column(db.String(100), unique=True)
password = db.Column(db.String(100))
games = db.relationship('Game', backref='owner', lazy=True)
username = db.Column(db.String(80), unique=True, nullable=False)
password = db.Column(db.String(256), nullable=False)
games = db.relationship('Game', back_populates='owner', lazy=True)
class Game(db.Model):
id = db.Column(db.Integer, primary_key=True)
owner = db.relationship('User', back_populates='games')
name = db.Column(db.String(100), nullable=False)
steam_key = db.Column(db.String(100), nullable=False)
steam_key = db.Column(db.String(100), nullable=False, unique=True)
status = db.Column(db.String(50), nullable=False)
recipient = db.Column(db.String(100))
notes = db.Column(db.Text)
url = db.Column(db.String(200))
created_at = db.Column(db.DateTime, default=datetime.utcnow)
redeem_date = db.Column(db.DateTime)
user_id = db.Column(db.Integer, db.ForeignKey('user.id'), nullable=False)
user_id = db.Column(db.Integer, db.ForeignKey('users.id'), nullable=False)
steam_appid = db.Column(db.String(20))
class RedeemToken(db.Model):
id = db.Column(db.Integer, primary_key=True)
token = db.Column(db.String(17), unique=True, nullable=False)
game_id = db.Column(db.Integer, db.ForeignKey('game.id'), nullable=False)
expires = db.Column(db.DateTime, nullable=False)
used = db.Column(db.Boolean, default=False)
total_hours = db.Column(db.Integer, nullable=False)
with app.app_context():
db.create_all()
@login_manager.user_loader
def load_user(user_id):
return db.session.get(User, int(user_id))
def extract_steam_appid(url):
match = re.search(r'store\.steampowered\.com/app/(\d+)', url or '')
return match.group(1) if match else ''
# 404
def get_or_404(model, id):
instance = db.session.get(model, id)
if not instance:
abort(404)
return instance
@app.route('/')
@login_required
def index():
search_query = request.args.get('q', '')
query = db.session.query(Game).filter_by(user_id=current_user.id)
query = Game.query.filter_by(user_id=current_user.id)
if search_query:
query = query.filter(Game.name.ilike(f'%{search_query}%'))
games = query.order_by(Game.created_at.desc()).all()
return render_template('index.html',
games=games,
@ -88,25 +179,34 @@ def login():
username = request.form['username']
password = request.form['password']
user = User.query.filter_by(username=username).first()
if user and check_password_hash(user.password, password):
login_user(user)
return redirect(url_for('index'))
flash(_('Invalid credentials'), 'danger')
return render_template('login.html')
@app.route('/register', methods=['GET', 'POST'])
def register():
if not app.config['REGISTRATION_ENABLED']:
flash(_('Registrierungen sind deaktiviert'), 'danger')
return redirect(url_for('login'))
if request.method == 'POST':
username = request.form['username']
password = generate_password_hash(request.form['password'])
if User.query.filter_by(username=username).first():
flash(_('Username already exists'), 'danger')
return redirect(url_for('register'))
new_user = User(username=username, password=password)
db.session.add(new_user)
db.session.commit()
login_user(new_user)
return redirect(url_for('index'))
return render_template('register.html')
@app.route('/logout')
@ -115,13 +215,28 @@ def logout():
logout_user()
return redirect(url_for('login'))
import re
@app.route('/change-password', methods=['GET', 'POST'])
@login_required
def change_password():
if request.method == 'POST':
current_password = request.form['current_password']
new_password = request.form['new_password']
confirm_password = request.form['confirm_password']
def extract_steam_appid(url):
match = re.search(r'store\.steampowered\.com/app/(\d+)', url or '')
if match:
return match.group(1)
return ''
if not check_password_hash(current_user.password, current_password):
flash(_('Aktuelles Passwort ist falsch'), 'danger')
return redirect(url_for('change_password'))
if new_password != confirm_password:
flash(_('Neue Passwörter stimmen nicht überein'), 'danger')
return redirect(url_for('change_password'))
current_user.password = generate_password_hash(new_password)
db.session.commit()
flash(_('Passwort erfolgreich geändert'), 'success')
return redirect(url_for('index'))
return render_template('change_password.html')
@app.route('/add', methods=['GET', 'POST'])
@login_required
@ -130,8 +245,10 @@ def add_game():
try:
url = request.form.get('url', '')
steam_appid = request.form.get('steam_appid', '').strip()
if not steam_appid:
steam_appid = extract_steam_appid(url)
new_game = Game(
name=request.form['name'],
steam_key=request.form['steam_key'],
@ -143,13 +260,19 @@ def add_game():
redeem_date=datetime.strptime(request.form['redeem_date'], '%Y-%m-%d') if request.form['redeem_date'] else None,
user_id=current_user.id
)
db.session.add(new_game)
db.session.commit()
flash(_('Game added successfully!'), 'success')
return redirect(url_for('index'))
except IntegrityError:
db.session.rollback()
flash(_('Steam Key already exists!'), 'danger')
except Exception as e:
db.session.rollback()
flash(_('Error: ') + str(e), 'danger')
return render_template('add_game.html')
@app.route('/edit/<int:game_id>', methods=['GET', 'POST'])
@ -157,14 +280,26 @@ def add_game():
def edit_game(game_id):
game = db.session.get(Game, game_id)
if not game or game.owner != current_user:
return _("Not allowed!"), 403
abort(404)
if not game or game.owner != current_user:
abort(403)
active_redeem = RedeemToken.query.filter(
RedeemToken.game_id == game_id,
RedeemToken.expires > datetime.utcnow()
).first()
redeem_url = url_for('redeem_page', token=active_redeem.token, _external=True) if active_redeem else None
if request.method == 'POST':
try:
url = request.form.get('url', '')
steam_appid = request.form.get('steam_appid', '').strip()
if not steam_appid:
steam_appid = extract_steam_appid(url)
game.name = request.form['name']
game.steam_key = request.form['steam_key']
game.status = request.form['status']
@ -173,32 +308,40 @@ def edit_game(game_id):
game.url = url
game.steam_appid = steam_appid
game.redeem_date = datetime.strptime(request.form['redeem_date'], '%Y-%m-%d') if request.form['redeem_date'] else None
db.session.commit()
flash(_('Changes saved!'), 'success')
return redirect(url_for('index'))
except Exception as e:
db.session.rollback()
flash(_('Error: ') + str(e), 'danger')
return render_template('edit_game.html',
game=game,
redeem_date=game.redeem_date.strftime('%Y-%m-%d') if game.redeem_date else '')
game=game,
redeem_url=redeem_url,
active_redeem=active_redeem,
redeem_date=game.redeem_date.strftime('%Y-%m-%d') if game.redeem_date else '')
@app.route('/delete/<int:game_id>', methods=['POST'])
@login_required
def delete_game(game_id):
game = Game.query.get_or_404(game_id)
game = db.session.get(Game, game_id)
if not game or game.owner != current_user:
abort(404)
if game.owner != current_user:
return _("Not allowed!"), 403
abort(403)
try:
db.session.delete(game)
db.session.commit()
flash(_('Game deleted!'), 'success')
except Exception as e:
db.session.rollback()
flash(_('Error deleting: ') + str(e), 'danger')
return redirect(url_for('index'))
# --- Import/Export Funktionen ---
@app.route('/export', methods=['GET'])
@login_required
@ -206,14 +349,22 @@ def export_games():
games = Game.query.filter_by(user_id=current_user.id).all()
output = io.StringIO()
writer = csv.writer(output)
writer.writerow(['Name', 'Steam Key', 'Status', 'Recipient', 'Notes', 'URL', 'Created', 'Redeem by', 'Steam AppID'])
for game in games:
writer.writerow([
game.name, game.steam_key, game.status, game.recipient, game.notes,
game.url, game.created_at.strftime('%Y-%m-%d %H:%M:%S') if game.created_at else '',
game.name,
game.steam_key,
game.status,
game.recipient,
game.notes,
game.url,
game.created_at.strftime('%Y-%m-%d %H:%M:%S') if game.created_at else '',
game.redeem_date.strftime('%Y-%m-%d') if game.redeem_date else '',
game.steam_appid
])
output.seek(0)
return send_file(
io.BytesIO(output.getvalue().encode('utf-8')),
@ -222,35 +373,306 @@ def export_games():
download_name='games_export.csv'
)
@app.route('/export_pdf')
@login_required
def export_pdf():
excluded_statuses = ['eingelöst', 'verschenkt']
games = Game.query.filter(
Game.user_id == current_user.id,
Game.status.notin_(excluded_statuses)
).order_by(Game.created_at.desc()).all()
buffer = io.BytesIO()
doc = SimpleDocTemplate(buffer,
pagesize=landscape(A4),
leftMargin=40,
rightMargin=40,
topMargin=40,
bottomMargin=40
)
styles = getSampleStyleSheet()
elements = []
img_height = 2*cm
# Titel
elements.append(Paragraph(_("Game List (without Keys)"), styles['Title']))
elements.append(Spacer(1, 12))
# Tabellenkopf
col_widths = [
5*cm, 10*cm, 6*cm, 3*cm
]
data = [[
Paragraph('<b>Cover</b>', styles['Normal']),
Paragraph('<b>Name</b>', styles['Normal']),
Paragraph('<b>Shop-Link</b>', styles['Normal']),
Paragraph('<b>Einlösen bis</b>', styles['Normal'])
]]
for game in games:
img = None
if game.steam_appid:
try:
img_url = f"https://cdn.cloudflare.steamstatic.com/steam/apps/{game.steam_appid}/header.jpg"
img_data = io.BytesIO(requests.get(img_url, timeout=5).content)
img = Image(img_data, width=3*cm, height=img_height)
except Exception:
img = Paragraph('', styles['Normal'])
data.append([
img or '',
Paragraph(game.name, styles['Normal']),
Paragraph(game.url or '', styles['Normal']),
game.redeem_date.strftime('%d.%m.%y') if game.redeem_date else ''
])
# Tabelle formatieren
table = Table(data, colWidths=col_widths, repeatRows=1)
table.setStyle(TableStyle([
('FONTNAME', (0,0), (-1,0), 'Helvetica-Bold'),
('FONTSIZE', (0,0), (-1,0), 8),
('FONTSIZE', (0,1), (-1,-1), 8),
('VALIGN', (0,0), (-1,-1), 'MIDDLE'),
('ALIGN', (0,0), (-1,-1), 'LEFT'),
('GRID', (0,0), (-1,-1), 0.5, colors.lightgrey),
('WORDWRAP', (1,1), (1,-1), 'CJK'),
]))
elements.append(table)
doc.build(elements)
buffer.seek(0)
return send_file(
buffer,
mimetype='application/pdf',
as_attachment=True,
download_name=f'game_export_{datetime.now().strftime("%Y%m%d")}.pdf'
)
@app.route('/import', methods=['GET', 'POST'])
@login_required
def import_games():
if request.method == 'POST':
file = request.files.get('file')
if file and file.filename.endswith('.csv'):
stream = io.StringIO(file.stream.read().decode("UTF8"), newline=None)
stream = io.StringIO(file.stream.read().decode("UTF8"))
reader = csv.DictReader(stream)
for row in reader:
new_game = Game(
name=row['Name'],
steam_key=row['Steam Key'],
status=row['Status'],
recipient=row.get('Recipient', ''),
notes=row.get('Notes', ''),
url=row.get('URL', ''),
created_at=datetime.strptime(row['Created'], '%Y-%m-%d %H:%M:%S') if row.get('Created') else datetime.utcnow(),
redeem_date=datetime.strptime(row['Redeem by'], '%Y-%m-%d') if row.get('Redeem by') else None,
steam_appid=row.get('Steam AppID', ''),
user_id=current_user.id
)
db.session.add(new_game)
db.session.commit()
flash(_('Import erfolgreich!'), 'success')
new_games = 0
duplicates = 0
try:
with db.session.begin_nested():
for row in reader:
steam_key = row['Steam Key'].strip()
if Game.query.filter_by(steam_key=steam_key).first():
duplicates += 1
continue
game = Game(
name=row['Name'],
steam_key=steam_key,
status=row['Status'],
recipient=row.get('Recipient', ''),
notes=row.get('Notes', ''),
url=row.get('URL', ''),
created_at=datetime.strptime(row['Created'], '%Y-%m-%d %H:%M:%S') if row.get('Created') else datetime.utcnow(),
redeem_date=datetime.strptime(row['Redeem by'], '%Y-%m-%d') if row.get('Redeem by') else None,
steam_appid=row.get('Steam AppID', ''),
user_id=current_user.id
)
db.session.add(game)
new_games += 1
db.session.commit()
flash(_('%(new)d neue Spiele importiert, %(dup)d Duplikate übersprungen', new=new_games, dup=duplicates), 'success')
except Exception as e:
db.session.rollback()
flash(_('Importfehler: %(error)s', error=str(e)), 'danger')
return redirect(url_for('index'))
flash(_('Bitte eine gültige CSV-Datei hochladen.'), 'danger')
return render_template('import.html')
@app.route('/generate_redeem/<int:game_id>', methods=['POST'])
@login_required
def generate_redeem(game_id):
game = db.session.get(Game, game_id)
if not game or game.owner != current_user:
abort(403)
if game.owner != current_user or game.status != 'verschenkt':
abort(403)
try:
token = secrets.token_urlsafe(12)[:17]
expires = datetime.utcnow() + timedelta(hours=24)
total_hours = 24
RedeemToken.query.filter_by(game_id=game_id).delete()
new_token = RedeemToken(
token=token,
game_id=game_id,
expires=expires,
total_hours=24
)
db.session.add(new_token)
db.session.commit()
redeem_url = url_for('redeem_page', token=token, _external=True)
return jsonify({'url': redeem_url})
except Exception as e:
app.logger.error(f"Redeem error: {str(e)}")
return jsonify({'error': str(e)}), 500
@app.route('/redeem/<token>')
def redeem_page(token):
redeem_token = RedeemToken.query.filter_by(token=token).first()
if not redeem_token:
abort(404)
if redeem_token.expires < datetime.utcnow():
db.session.delete(redeem_token)
db.session.commit()
abort(404)
game = Game.query.get(redeem_token.game_id)
redeem_token.used = True
db.session.commit()
return render_template('redeem.html',
game=game,
redeem_token=redeem_token,
platform_link='https://store.steampowered.com/account/registerkey?key=' if game.steam_appid else 'https://www.gog.com/redeem')
# Benachrichtigungsfunktionen
def send_pushover_notification(user, game):
"""Sendet Pushover-Benachrichtigung für ablaufenden Key"""
if not app.config['PUSHOVER_APP_TOKEN'] or not app.config['PUSHOVER_USER_KEY']:
return False
payload = {
"token": os.getenv('PUSHOVER_APP_TOKEN'),
"user": os.getenv('PUSHOVER_USER_KEY'),
"title": "Steam-Key läuft ab!",
"message": f"Dein Key für '{game.name}' läuft in weniger als 48 Stunden ab!",
"url": url_for('edit_game', game_id=game.id, _external=True),
"url_title": "Zum Spiel",
"priority": 1
}
try:
response = requests.post(
'https://api.pushover.net/1/messages.json',
data=payload
)
return response.status_code == 200
except Exception as e:
app.logger.error(f"Pushover error: {str(e)}")
return False
def send_gotify_notification(user, game):
"""Sendet Gotify-Benachrichtigung für ablaufenden Key"""
if not GOTIFY_URL or not GOTIFY_TOKEN:
return False
payload = {
"title": "Steam-Key läuft ab!",
"message": f"Dein Key für '{game.name}' läuft in weniger als 48 Stunden ab!",
"priority": 5
}
try:
response = requests.post(
f"{GOTIFY_URL}/message?token={GOTIFY_TOKEN}",
json=payload
)
return response.status_code == 200
except Exception as e:
app.logger.error(f"Gotify error: {str(e)}")
return False
def send_matrix_notification(user, game):
"""Sendet Matrix-Benachrichtigung für ablaufenden Key"""
if not MATRIX_HOMESERVER or not MATRIX_ACCESS_TOKEN or not MATRIX_ROOM_ID:
return False
try:
from matrix_client.client import MatrixClient
client = MatrixClient(MATRIX_HOMESERVER, token=MATRIX_ACCESS_TOKEN)
room = client.join_room(MATRIX_ROOM_ID)
message = f"🎮 Dein Key für '{game.name}' läuft in weniger als 48 Stunden ab!"
room.send_text(message)
return True
except Exception as e:
app.logger.error(f"Matrix error: {str(e)}")
return False
def send_notification(user, game):
"""Sendet Benachrichtigung über den bevorzugten Dienst des Benutzers"""
if user.notification_service == 'pushover':
return send_pushover_notification(user, game)
elif user.notification_service == 'gotify':
return send_gotify_notification(user, game)
elif user.notification_service == 'matrix':
return send_matrix_notification(user, game)
return False
def check_expiring_keys():
with app.app_context():
now = datetime.utcnow()
expiry_threshold = now + timedelta(hours=48)
# Moderner Select-Aufruf
stmt = select(Game).where(
Game.status != 'eingelöst',
Game.redeem_date <= expiry_threshold,
Game.redeem_date > now
)
expiring_games = db.session.execute(stmt).scalars().all()
for game in expiring_games:
user = User.query.get(game.user_id)
if user.notification_service and user.notification_service != 'none':
send_notification(user, game)
# Optional: Cleanup-Funktion für regelmäßiges Löschen abgelaufener Tokens
def cleanup_expired_tokens():
now = datetime.utcnow()
expired = RedeemToken.query.filter(RedeemToken.expires < now).all()
for token in expired:
db.session.delete(token)
db.session.commit()
# Scheduler initialisieren und starten
scheduler = BackgroundScheduler()
scheduler.add_job(func=check_expiring_keys, trigger="interval", hours=interval_hours)
scheduler.add_job(func=cleanup_expired_tokens, trigger="interval", hours=1)
scheduler.start()
# Shutdown des Schedulers bei Beendigung der App
atexit.register(lambda: scheduler.shutdown())
if __name__ == '__main__':
with app.app_context():
db.create_all()
app.run(host='0.0.0.0', port=5000)