add icon token subdir, Docker, SPEC/AGENTS, initial commit

This commit is contained in:
proitlab committed 2026-08-16 20:13:42 +07:00
commit 73eeab7f63
17 files changed
+622

No files matched your search

+170
View File
@@ -0,0 +1,170 @@
import os
from pathlib import Path
from sqladmin import Admin, ModelView
from starlette.datastructures import UploadFile
from starlette.requests import Request
from wtforms import BooleanField, FileField, Form, IntegerField, StringField
from wtforms.validators import DataRequired
from sqlalchemy import text
from app.auth import AdminAuthBackend, get_session_secret
from app.models import ICON_DIR, SessionLocal, Token, Version
BASE_DIR = Path(__file__).resolve().parent.parent
TEMPLATES_DIR = str(BASE_DIR / "app" / "templates")
MAX_ICON_BYTES = 512 * 1024
PNG_MAGIC = b"\x89PNG\r\n\x1a\n"
PUBLIC_BASE_URL = os.environ.get("IDRS_PUBLIC_URL", "https://idrs.databisnis.id").rstrip("/")
def _is_authenticated(request: Request) -> bool:
return bool(request.session.get("auth"))
def _sanitize_png_name(name: str) -> str:
base = Path(name).name
if base in ("", ".", ".."):
return "icon.png"
if not base.lower().endswith(".png"):
base = f"{Path(base).stem or 'icon'}.png"
return base
async def _extract_icon(raw) -> bytes:
if isinstance(raw, UploadFile):
return await raw.read(MAX_ICON_BYTES + 1)
stream = raw.stream if hasattr(raw, "stream") else raw
data = stream.read(MAX_ICON_BYTES + 1)
return data if isinstance(data, bytes) else bytes(data)
def _next_token_id() -> int:
with SessionLocal() as session:
return int(session.execute(text("SELECT COALESCE(MAX(idfToken), 0) + 1 FROM tokens")).scalar())
class NullableIntegerField(IntegerField):
def process_formdata(self, valuelist):
if valuelist and valuelist[0].strip() == "":
self.data = None
return
super().process_formdata(valuelist)
class TokenForm(Form):
position = NullableIntegerField("Position", default=_next_token_id)
name = StringField("Name", validators=[DataRequired()])
symbol = StringField("Symbol", validators=[DataRequired()])
contractAddr = StringField("Contract", validators=[DataRequired()])
decimals = IntegerField("Decimals", validators=[DataRequired()])
iconUrl = StringField("Icon URL")
icon = FileField("Icon PNG")
is_live = BooleanField("Live")
class TokenAdmin(ModelView, model=Token):
name = "Tokens"
icon = "fa-solid fa-coins"
form = TokenForm
column_list = [
"position",
"name",
"symbol",
"typeBlockchain",
"contractAddr",
"decimals",
"is_live",
"iconUrl",
]
column_labels = {
"idfToken": "ID",
"position": "Position",
"typeBlockchain": "Blockchain",
"contractAddr": "Contract",
"decimals": "Decimals",
"is_live": "Live",
}
column_details_list = [
"idfToken",
"position",
"typeBlockchain",
"name",
"symbol",
"contractAddr",
"decimals",
"iconUrl",
"filename",
"is_live",
]
column_searchable_list = ["name", "symbol", "contractAddr"]
column_default_sort = [("position", True)]
page_size = 25
def is_accessible(self, request: Request) -> bool:
return _is_authenticated(request)
async def on_model_change(self, data, model, is_created, request):
data.pop("idfToken", None)
data["typeBlockchain"] = "Vexanium"
if data.get("position") in (None, ""):
data["position"] = _next_token_id()
raw = data.get("icon")
data.pop("icon", None)
if not raw:
return
content = await _extract_icon(raw)
if not content:
return
if len(content) > MAX_ICON_BYTES:
raise Exception(f"Icon must be PNG ≤ {MAX_ICON_BYTES // 1024}KB")
if not content.startswith(PNG_MAGIC):
raise Exception("Icon must be a real PNG file")
filename = _sanitize_png_name(getattr(raw, "filename", "") or "icon.png")
token_dir = ICON_DIR / "token"
token_dir.mkdir(parents=True, exist_ok=True)
(token_dir / filename).write_bytes(content)
data["filename"] = filename
data["iconUrl"] = f"{PUBLIC_BASE_URL}/img/token/{filename}"
class VersionAdmin(ModelView, model=Version):
name = "Android Version"
icon = "fa-solid fa-mobile-screen"
can_create = False
can_delete = False
column_list = ["idVersion", "appName", "acceptableVersion", "latestVersion"]
column_labels = {
"idVersion": "ID Version",
"appName": "App",
"acceptableVersion": "Acceptable",
"latestVersion": "Latest",
}
column_details_list = ["idVersion", "appName", "acceptableVersion", "latestVersion"]
form_columns = ["idVersion", "appName", "acceptableVersion", "latestVersion"]
def is_accessible(self, request: Request) -> bool:
return _is_authenticated(request)
async def on_model_change(self, data, model, is_created, request):
acceptable = int(data.get("acceptableVersion", model.acceptableVersion))
latest = int(data.get("latestVersion", model.latestVersion))
if acceptable > latest:
raise Exception("acceptableVersion must be <= latestVersion")
def create_admin(app):
auth_backend = AdminAuthBackend(get_session_secret())
admin = Admin(
app,
session_maker=SessionLocal,
title="IDRS Admin",
base_url="/admin",
templates_dir=TEMPLATES_DIR,
authentication_backend=auth_backend,
)
admin.add_view(TokenAdmin)
admin.add_view(VersionAdmin)
return admin