поделал сайт, сделал orm-модели, requirements.txt

functions/admin/get_log.py

@@ -0,0 +1,21 @@
+import os
+from typing import List, Dict, Any
+
+
+def get_log(log_id: int) -> List[Dict[str, Any]]:
+    log_dir = os.path.join("static/logs", str(log_id))
+    return_dir = []
+    for dir_path, _, filenames in os.walk(log_dir):
+        if dir_path != log_dir:
+            audio_file = os.path.join(dir_path, "audio.ogg")
+            if not os.path.exists(audio_file):
+                audio_file = os.path.join(dir_path, "audio.wav")
+            text_file = os.path.join(dir_path, "yandex-text.txt")
+            if not os.path.exists(text_file):
+                text_file = os.path.join(dir_path, "google-text.txt")
+            try:
+                return_dir.append({"id": log_id, "audio_file": f"/{audio_file}",
+                                   "text": open(text_file).read().split("\n")})
+            except UnicodeDecodeError:
+                pass
+    return return_dir

functions/admin/get_logs.py

@@ -0,0 +1,8 @@
+import os
+from typing import List, Tuple
+
+
+def get_logs() -> List[Tuple[int, str]]:
+    return [(int(os.path.basename(dir_path)), dir_path.split("/")[2].strip())
+            for dir_path, _, filenames in os.walk("static/logs")
+            if dir_path != "static/logs" and len(dir_path.split("/")) == 3]

functions/admin/is_logged_in.py

@@ -0,0 +1,16 @@
+import jwt
+from fastapi import Cookie, Depends
+from functions.admin.models import token, database
+
+
+def is_logged_in(access_token: str = Cookie(None), db=Depends(database.get_db)):
+    if not access_token:
+        return False
+    response = db.query(token.Token).filter(token.Token.access_token == access_token).first()
+    if not response or not response.is_active:
+        return False
+    try:
+        jwt.decode(access_token, "secret", algorithms=["HS256"])
+    except jwt.exceptions.ExpiredSignatureError:
+        return False
+    return True

functions/admin/models/database.py

@@ -0,0 +1,18 @@
+from sqlalchemy import create_engine
+from sqlalchemy.ext.declarative import declarative_base
+from sqlalchemy.orm import sessionmaker
+
+SQLALCHEMY_DATABASE_URL = "sqlite:///./db.sql"
+
+engine = create_engine(SQLALCHEMY_DATABASE_URL)
+SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
+
+Base = declarative_base()
+Base.metadata.create_all(bind=engine)
+
+def get_db():
+    db = SessionLocal()
+    try:
+        yield db
+    finally:
+        db.close()

functions/admin/models/token.py

@@ -0,0 +1,14 @@
+from sqlalchemy import Boolean, Column, ForeignKey, Integer, String
+from sqlalchemy.orm import relationship
+from functions.admin.models.database import Base
+
+class Token(Base):
+    __tablename__ = "tokens"
+
+    id = Column(Integer, primary_key=True, index=True)
+    access_token = Column(String(255), unique=True)
+    refresh_token = Column(String(255), unique=True)
+    is_active = Column(Boolean(), default=True)
+
+    user_id = Column(Integer, ForeignKey("users.id"))
+    user = relationship("User", back_populates="tokens")

functions/admin/models/user.py

@@ -0,0 +1,19 @@
+import hashlib
+
+from sqlalchemy import Boolean, Column, Integer, String
+from sqlalchemy.orm import relationship
+from functions.admin.models.database import Base
+
+class User(Base):
+    __tablename__ = "users"
+
+    id = Column(Integer, primary_key=True, index=True)
+    username = Column(String(50), unique=True)
+    email = Column(String(255), unique=True)
+    password_hash = Column(String(255))
+    is_active = Column(Boolean(), default=True)
+
+    tokens = relationship("Token", back_populates="user")
+
+    def check_password(self, password: str):
+        return self.password_hash == hashlib.sha384(password.encode()).hexdigest()

functions/admin/templates/log.py

@@ -0,0 +1,10 @@
+from fastapi import Request, Depends, responses, HTTPException
+from functions.admin import get_log, is_logged_in
+
+async def log(templates, request: Request, log_id: int, logged_in: bool = Depends(is_logged_in.is_logged_in)):
+    if logged_in:
+        if not get_log.get_log(log_id):
+            raise HTTPException(status_code=400)
+        return templates.TemplateResponse("log.html", {"request": request, "log": get_log.get_log(log_id)})
+    else:
+        return responses.RedirectResponse(url=f"/refresh?source=/logs/{log_id}", status_code=303)

functions/admin/templates/logins.py

@@ -0,0 +1,25 @@
+from fastapi import Form, Depends, responses, HTTPException
+from functions.admin.models import database, user, token
+import jwt
+from datetime import datetime, timedelta
+
+
+async def logins(username: str = Form(...), password: str = Form(...), db=Depends(database.get_db)):
+    # username = "Dmitrium12"
+    response = db.query(user.User).filter(user.User.username == username).first()
+    if not response or not response.check_password(password):
+        raise HTTPException(status_code=400, detail="Неправильное имя пользователя или пароль")
+    access_token_expires = datetime.utcnow() + timedelta(minutes=15)
+    access_token_payload = {"sub": response.username, "exp": access_token_expires}
+    access_token = jwt.encode(access_token_payload, "secret", algorithm="HS256")
+    refresh_token_expires = datetime.utcnow() + timedelta(days=7)
+    refresh_token_payload = {"sub": response.username, "exp": refresh_token_expires}
+    refresh_token = jwt.encode(refresh_token_payload, "secret", algorithm="HS256")
+    db_token = token.Token(access_token=access_token, refresh_token=refresh_token, user=response)
+    db.add(db_token)
+    db.commit()
+    response = responses.RedirectResponse(url="/logs", status_code=303)
+    response.set_cookie(key="access_token", value=access_token, expires=int(access_token_expires.timestamp()))
+    response.set_cookie(key="refresh_token", value=refresh_token,
+                        expires=int(refresh_token_expires.timestamp()))
+    return response

functions/admin/templates/logs.py

@@ -0,0 +1,8 @@
+from fastapi import Request, Depends, responses
+from functions.admin import get_logs, is_logged_in
+
+async def logs(templates, request: Request, logged_in: bool = Depends(is_logged_in.is_logged_in)):
+    if logged_in:
+        return templates.TemplateResponse("logs.html", {"request": request, "logs": get_logs.get_logs()})
+    else:
+        return responses.RedirectResponse(url="/refresh?source=/logs", status_code=303)

functions/admin/templates/refresh.py

@@ -0,0 +1,31 @@
+from fastapi import Request, Depends, Cookie, responses
+from functions.admin.models import database, user, token
+import jwt
+from datetime import datetime, timedelta
+
+async def refresh_access_token(req: Request, refresh_token: str = Cookie(None), db=Depends(database.get_db)):
+    request_args = dict(req.query_params)
+    try:
+        refresh_token_payload = jwt.decode(refresh_token, "secret", algorithms=["HS256"])
+    except jwt.exceptions.DecodeError:
+        return responses.RedirectResponse(url="/login", status_code=303)
+    response = db.query(user.User).filter(user.User.username == refresh_token_payload["sub"]).first()
+    if not response:
+        return responses.RedirectResponse(url="/login", status_code=303)
+    access_token_expires = datetime.utcnow() + timedelta(minutes=15)
+    access_token_payload = {"sub": response.username, "exp": access_token_expires}
+    access_token = jwt.encode(access_token_payload, "secret", algorithm="HS256")
+    db.query(token.Token).filter(token.Token.refresh_token == refresh_token).update({
+        token.Token.access_token: access_token,
+    })
+    db.commit()
+    if request_args:
+        response = responses.RedirectResponse(url=request_args["source"], status_code=303)
+    else:
+        response = responses.RedirectResponse(url="/", status_code=303)
+    response.set_cookie(
+        key="access_token",
+        value=access_token,
+        expires=int(access_token_expires.timestamp())
+    )
+    return response