Init Commit

modules/app.py

@@ -0,0 +1,4 @@
+from flask import Flask
+
+app = Flask(__name__)
+

modules/db.py

@@ -0,0 +1,50 @@
+from flask_sqlalchemy import SQLAlchemy
+from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
+from datetime import datetime
+
+db = SQLAlchemy()
+
+class User(db.Model):
+    __tablename__ = "users"
+
+    steam_id: Mapped[int] = mapped_column(db.BigInteger, primary_key=True)
+    personaname: Mapped[str] = mapped_column(db.String(80))
+    profile_url: Mapped[str] = mapped_column(db.String(200))
+    avatar: Mapped[str] = mapped_column(db.String(200))
+    avatar_medium: Mapped[str] = mapped_column(db.String(200))
+    avatar_full: Mapped[str] = mapped_column(db.String(200))
+    community_visibility_state: Mapped[int] = mapped_column(db.Integer)
+    profile_state: Mapped[int] = mapped_column(db.Integer)
+    last_logoff: Mapped[int] = mapped_column(db.Integer)
+
+    def to_dict(self):
+        return {
+            "steam_id": self.steam_id,
+            "personaname": self.personaname,
+            "profile_url": self.profile_url,
+            "avatar": self.avatar,
+            "avatar_medium": self.avatar_medium,
+            "avatar_full": self.avatar_full,
+            "community_visibility_state": self.community_visibility_state,
+            "profile_state": self.profile_state,
+            "last_logoff": self.last_logoff,
+        }
+
+class APICall(db.Model):
+    __tablename__ = "api_calls"
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    name: Mapped[str] = mapped_column(db.String(80), unique=True, nullable=False)
+    call_count: Mapped[int] = mapped_column(db.Integer, default=0, nullable=False)
+    first_called: Mapped[datetime] = mapped_column(db.DateTime, default=None, nullable=True)
+    last_called: Mapped[datetime] = mapped_column(db.DateTime, default=datetime.utcnow, nullable=False)
+
+    def register_call(self):
+        """Increment call count and update timestamp."""
+        if self.call_count is None:
+            self.call_count = 0
+        if self.first_called is None:
+            self.first_called = datetime.now()
+        self.call_count = self.call_count + 1
+        self.last_called = datetime.now()
+        db.session.commit()

modules/steam_api.py

@@ -0,0 +1,52 @@
+import requests
+from .app import app
+from .db import db, User, APICall
+from typing import Final
+
+API_NAME: Final = "STEAM_API"
+
+def record_api_call():
+    api_call = APICall.query.filter_by(name=API_NAME).first()
+    if not api_call:
+        api_call = APICall(name=API_NAME)
+        db.session.add(api_call)
+    api_call.register_call()
+    return api_call
+
+def get_user(steam_id):
+    # 1. Check DB
+    user = User.query.filter_by(steam_id=steam_id).first()
+    if user:
+        app.logger.debug("User fetch from DB")
+        return user
+
+    # 2. Fetch from Steam API
+    url = "https://api.steampowered.com/ISteamUser/GetPlayerSummaries/v2/"
+    params = {"key": app.config['STEAM_API_KEY'], "steamids": steam_id}
+    resp = requests.get(url, params=params)
+    api = record_api_call()
+    data = resp.json()
+    players = data.get("response", {}).get("players", [])
+    if not players:
+        app.logger.debug(f"User not found (API called: {api.call_count})")
+        return None
+
+    player = players[0]
+
+    # 3. Save to DB
+    user = User(
+        steam_id=steam_id,
+        personaname=player.get("personaname"),
+        profile_url=player.get("profileurl"),
+        avatar=player.get("avatar"),
+        avatar_medium=player.get("avatarmedium"),
+        avatar_full=player.get("avatarfull"),
+        community_visibility_state=player.get("communityvisibilitystate"),
+        profile_state=player.get("profilestate"),
+        last_logoff=player.get("lastlogoff")
+    )
+    db.session.add(user)
+    db.session.commit()
+
+    app.logger.debug(f"User fetch from API (API called: {api.call_count})")
+    return user