Init project

This commit is contained in:
2023-12-12 17:06:04 +03:00
commit b4432bc10f
39 changed files with 2794 additions and 0 deletions

2
.dockerignore Normal file
View File

@@ -0,0 +1,2 @@
*/**/__pycache__
*.pyc

2
.gitignore vendored Normal file
View File

@@ -0,0 +1,2 @@
*/**/__pycache__
*.pyc

14
Dockerfile Normal file
View File

@@ -0,0 +1,14 @@
FROM python:3.11-slim
WORKDIR /app
RUN pip install poetry
COPY ./pyproject.toml /app/pyproject.toml
COPY ./poetry.lock /app/poetry.lock
RUN poetry install --only main
COPY ./soul_diary /app/soul_diary
ENTRYPOINT ["poetry", "run", "python", "-m", "soul_diary"]
CMD ["run"]

29
README.md Normal file
View File

@@ -0,0 +1,29 @@
# Soul Diary
## User Flow
### Soul Diary Server
```mermaid
sequenceDiagram
actor user
participant client
participant server
Note over user,server: Registration
user->>server: Send username and password
server-->server: Register new user
Note over user,server: Authorization
user->>server: Send username and password
server->>client: Return access token
client-->client: Store access token
client-->client: Generate encryption key by username and password
Note over user,server: Push sense
user->>client: Enter sense data
client-->client: Encrypt sense data
client->>server: Send encrypted sense data
Note over user,server: Pull sense
user->>server: Ask sense data
server->>client: Send encrypted data
client-->client: Decrypt sense data
client->>user: Show sense data
```

1232
poetry.lock generated Normal file

File diff suppressed because it is too large Load Diff

26
pyproject.toml Normal file
View File

@@ -0,0 +1,26 @@
[tool.poetry]
name = "soul-diary"
version = "0.0.1"
description = ""
authors = ["Oleg Yurchik <oleg@yurchik.space>"]
readme = "README.md"
include = [
"soul_diary",
]
[tool.poetry.dependencies]
python = "^3.11"
flet = "^0.14.0"
flet-fastapi = "^0.14.0"
uvicorn = "^0.24.0.post1"
facet = "^0.9.1"
flet-route = "^0.3.2"
pydantic = "^2.5.2"
fastapi = "0.101.1"
typer = "^0.9.0"
pydantic-settings = "^2.1.0"
pycryptodomex = "^3.19.0"
[build-system]
requires = ["poetry-core"]
build-backend = "poetry.core.masonry.api"

0
soul_diary/__init__.py Normal file
View File

7
soul_diary/__main__.py Normal file
View File

@@ -0,0 +1,7 @@
from .cli import get_cli
if __name__ == "__main__":
cli = get_cli()
cli()

View File

20
soul_diary/cli.py Normal file
View File

@@ -0,0 +1,20 @@
import asyncio
import typer
from . import ui
def run():
ui_service = ui.get_service()
asyncio.run(ui_service.run())
def get_cli() -> typer.Typer:
cli = typer.Typer()
cli.add_typer(ui.get_cli(), name="ui")
cli.command(name="run")(run)
return cli

View File

@@ -0,0 +1,2 @@
from .cli import get_cli
from .service import UIService, get_service

View File

@@ -0,0 +1,7 @@
from .cli import get_cli
if __name__ == "__main__":
cli = get_cli()
cli()

View File

@@ -0,0 +1 @@
from .app import SoulDiaryApp

56
soul_diary/ui/app/app.py Normal file
View File

@@ -0,0 +1,56 @@
from typing import Any
import flet
from flet_route import Routing, path
from .local_storage import LocalStorage
from .middleware import middleware
from .models import BackendType
from .routes import AUTH, INDEX, SENSE_ADD, SENSE_LIST
from .views.auth import AuthView
from .views.base import BaseView
from .views.sense_add import SenseAddView
from .views.sense_list import SenseListView
class SoulDiaryApp:
def __init__(
self,
backend: BackendType | None = None,
backend_data: dict[str, Any] | None = None,
):
self._backend = backend
self._backend_data = backend_data
self._backend_client = None
def get_routes(self, page: flet.Page) -> dict[str, BaseView]:
local_storage = LocalStorage(client_storage=page.client_storage)
sense_list_view = SenseListView(local_storage=local_storage)
return {
INDEX: sense_list_view,
AUTH: AuthView(
local_storage=local_storage,
backend=self._backend,
backend_data=self._backend_data,
),
SENSE_LIST: sense_list_view,
SENSE_ADD: SenseAddView(),
}
async def run(self, page: flet.Page):
page.title = "Soul Diary"
page.app = self
routes = self.get_routes(page)
Routing(
page=page,
async_is=True,
app_routes=[
path(url=url, clear=False, view=view.entrypoint)
for url, view in routes.items()
],
middleware=middleware,
)
return await page.go_async(page.route)

View File

View File

@@ -0,0 +1,192 @@
import base64
import hashlib
import json
import uuid
from typing import Any
from Cryptodome.Cipher import AES
from soul_diary.ui.app.backend.models import SenseBackendData
from soul_diary.ui.app.local_storage import LocalStorage
from soul_diary.ui.app.models import BackendType, Emotion, Options, Sense
class BaseBackend:
BACKEND: BackendType
NONCE = b"\x00" * 16
MODE = AES.MODE_EAX
ENCODING = "utf-8"
ENCRYPTION_KEY_TEMPLATE = "backend:encryption_key:{username}:{password}"
def __init__(
self,
local_storage: LocalStorage,
username: str | None = None,
encryption_key: str | None = None,
token: str | None = None,
):
self._local_storage = local_storage
self._username = username
self._encryption_key = (
None
if encryption_key is None else
encryption_key.encode(self.ENCODING)
)
self._token = token
def generate_encryption_key(self, username: str, password: str) -> bytes:
data = (
self.ENCRYPTION_KEY_TEMPLATE
.format(username=username, password=password)
.encode(self.ENCODING)
)
return hashlib.sha256(data).hexdigest().encode(self.ENCODING)[:16]
def encode(self, data: dict[str, Any]) -> str:
if self._encryption_key is None:
raise ValueError("Need crypto key. For generating key you should authenticate.")
cipher = AES.new(self._encryption_key, self.MODE, nonce=self.NONCE)
data_string = json.dumps(data)
data_bytes = data_string.encode(self.ENCODING)
data_bytes_encoded, _ = cipher.encrypt_and_digest(data_bytes)
data_bytes_encoded_base64 = base64.b64encode(data_bytes_encoded).decode(self.ENCODING)
return data_bytes_encoded_base64
def decode(self, data: str) -> dict[str, Any]:
if self._encryption_key is None:
raise ValueError("Need crypto key. For generating key you should authenticate.")
cipher = AES.new(self._encryption_key, self.MODE, nonce=self.NONCE)
data_bytes_encoded = base64.b64decode(data)
data_bytes = cipher.decrypt(data_bytes_encoded)
data_string = data_bytes.decode(self.ENCODING)
data_decoded = json.loads(data_string)
return data_decoded
def convert_sense_data_to_sense(self, sense_data: SenseBackendData) -> Sense:
return Sense(
id=sense_data.id,
created_at=sense_data.created_at,
**self.decode(sense_data.data),
)
async def registration(self, username: str, password: str):
self._token = await self.create_user(username=username, password=password)
self._encryption_key = self.generate_encryption_key(username=username, password=password)
self._username = username
await self._local_storage.store_auth_data(
backend=self.BACKEND,
backend_data=self.get_backend_data(),
username=self._username,
encryption_key=self._encryption_key.decode(self.ENCODING),
token=self._token,
)
async def login(self, username: str, password: str):
self._token = await self.auth(username=username, password=password)
self._encryption_key = self.generate_encryption_key(username=username, password=password)
self._username = username
await self._local_storage.store_auth_data(
backend=self.BACKEND,
backend_data=self.get_backend_data(),
username=self._username,
encryption_key=self._encryption_key.decode(self.ENCODING),
token=self._token,
)
async def logout(self):
self._token = None
self._encryption_key = None
self._username = None
@property
def is_auth(self) -> bool:
return all((self._token, self._encryption_key))
async def get_sense_list(self, page: int = 1, limit: int = 10) -> list[Sense]:
sense_data_list = await self.fetch_sense_list(page=page, limit=limit)
return [
self.convert_sense_data_to_sense(sense_data)
for sense_data in sense_data_list
]
async def create_sense(
self,
emotions: list[Emotion],
feelings: str,
body: str,
desires: str,
) -> Sense:
breakpoint()
data = {
"emotions": emotions,
"feelings": feelings,
"body": body,
"desires": desires
}
encoded_data = self.encode(data)
sense_data = await self.pull_sense_data(data=encoded_data)
return self.convert_sense_data_to_sense(sense_data)
async def get_sense(self, sense_id: uuid.UUID) -> Sense:
sense_data = await self.fetch_sense(sense_id=sense_id)
return self.convert_sense_data_to_sense(sense_data)
async def edit_sense(
self,
sense_id: uuid.UUID,
emotions: list[Emotion] | None = None,
feelings: str | None = None,
body: str | None = None,
desires: str | None = None,
):
data = {
"emotions": emotions,
"feelings": feelings,
"body": body,
"desires": desires,
}
encoded_data = self.encode(data)
sense_data = await self.pull_sense_data(data=encoded_data, sense_id=sense_id)
return self.convert_sense_data_to_sense(sense_data)
def get_backend_data(self) -> dict[str, Any]:
raise NotImplementedError
async def create_user(self, username: str, password: str) -> str:
raise NotImplementedError
async def auth(self, username: str, password: str) -> str:
raise NotImplementedError
async def get_options(self) -> Options:
raise NotImplementedError
async def fetch_sense_list(
self,
page: int = 1,
limit: int = 10,
) -> list[SenseBackendData]:
raise NotImplementedError
async def fetch_sense(self, sense_id: uuid.UUID) -> SenseBackendData:
raise NotImplementedError
async def pull_sense_data(
self,
data: str,
sense_id: uuid.UUID | None = None,
) -> SenseBackendData:
raise NotImplementedError
async def delete_sense(self, sense_id: uuid.UUID):
raise NotImplementedError

View File

@@ -0,0 +1,18 @@
class BackendException(Exception):
pass
class UserAlreadyExistsException(BackendException):
pass
class IncorrectCredentialsException(BackendException):
pass
class NonAuthenticatedException(BackendException):
pass
class SenseNotFoundException(BackendException):
pass

View File

@@ -0,0 +1,135 @@
import hashlib
import uuid
from datetime import datetime
from typing import Any
from soul_diary.ui.app.models import BackendType, Options
from .base import BaseBackend
from .exceptions import (
IncorrectCredentialsException,
NonAuthenticatedException,
SenseNotFoundException,
UserAlreadyExistsException,
)
from .models import SenseBackendData
class LocalBackend(BaseBackend):
BACKEND = BackendType.LOCAL
AUTH_BLOCK_TEMPLATE = "auth_block:{username}:{password}"
AUTH_BLOCK_KEY_TEMPLATE = "soul_diary.backend.users.{username}.auth_block"
SENSE_LIST_KEY_TEMPLATE = "soul_diary.backend.users.{username}.senses"
def generate_auth_block(self, username: str, password: str) -> str:
auth_block_data = (
self.AUTH_BLOCK_TEMPLATE
.format(username=username, password=password)
.encode(self.ENCODING)
)
return hashlib.sha256(auth_block_data).hexdigest()
def get_backend_data(self) -> dict[str, Any]:
return {}
async def create_user(self, username: str, password: str) -> str | None:
auth_block_key = self.AUTH_BLOCK_KEY_TEMPLATE.format(username=username)
if await self._local_storage.raw_contains(auth_block_key):
raise UserAlreadyExistsException()
auth_block = self.generate_auth_block(username=username, password=password)
await self._local_storage.raw_write(auth_block_key, auth_block)
return auth_block
async def auth(self, username: str, password: str) -> str | None:
auth_block_key = self.AUTH_BLOCK_KEY_TEMPLATE.format(username=username)
if not await self._local_storage.raw_contains(auth_block_key):
raise IncorrectCredentialsException()
auth_block = self.generate_auth_block(username=username, password=password)
actual_auth_block = await self._local_storage.raw_read(auth_block_key)
if auth_block != actual_auth_block:
raise IncorrectCredentialsException()
return auth_block
async def get_options(self) -> Options:
return Options(registration_enabled=True)
async def _fetch_sense_list(self) -> list[SenseBackendData]:
if not self.is_auth:
raise NonAuthenticatedException()
sense_list_key = self.SENSE_LIST_KEY_TEMPLATE.format(username=self._username)
sense_list = await self._local_storage.raw_read(sense_list_key) or []
return [SenseBackendData.model_validate(sense) for sense in sense_list]
async def fetch_sense_list(
self,
page: int = 1,
limit: int = 10,
) -> list[SenseBackendData]:
sense_list = await self._fetch_sense_list()
sense_list_filtered = sense_list[(page - 1) * limit:page * limit]
return sense_list_filtered
async def fetch_sense(self, sense_id: uuid.UUID) -> SenseBackendData:
sense_list = await self._fetch_sense_list()
for sense in sense_list:
if sense.id == sense_id:
return sense
raise SenseNotFoundException()
async def pull_sense_data(self, data: str, sense_id: uuid.UUID | None = None) -> SenseBackendData:
sense_list_key = self.SENSE_LIST_KEY_TEMPLATE.format(username=self._username)
sense_list = await self._fetch_sense_list()
if sense_id is None:
sense_ids = {sense.id for sense in sense_list}
sense_id = uuid.uuid4()
while sense_id in sense_ids:
sense_id = uuid.uuid4()
sense = SenseBackendData(
id=sense_id,
data=data,
created_at=datetime.now().astimezone(),
)
sense_list.insert(0, sense)
else:
for index, sense in enumerate(sense_list):
if sense.id == sense_id:
break
else:
raise SenseNotFoundException()
sense = sense_list[index]
sense.data = data
sense_list[index] = sense
await self._local_storage.raw_write(
sense_list_key,
[sense.model_dump(mode="json") for sense in sense_list],
)
return sense
async def delete_sense(self, sense_id: uuid.UUID):
sense_list_key = self.SENSE_LIST_KEY_TEMPLATE.format(username=self._username)
sense_list = await self._fetch_sense_list()
for index, sense in enumerate(sense_list):
if sense.id == sense_id:
break
else:
raise SenseNotFoundException()
sense_list = sense_list[:index] + sense_list[index + 1:]
await self._local_storage.raw_write(
sense_list_key,
[sense.model_dump(mode="json") for sense in sense_list],
)

View File

@@ -0,0 +1,9 @@
import uuid
from pydantic import AwareDatetime, BaseModel
class SenseBackendData(BaseModel):
id: uuid.UUID
data: str
created_at: AwareDatetime

View File

@@ -0,0 +1,59 @@
from typing import Any
from pydantic import BaseModel
from soul_diary.ui.app.models import BackendType
class AuthData(BaseModel):
backend: BackendType
backend_data: dict[str, Any]
username: str
encryption_key: str
token: str
class LocalStorage:
def __init__(self, client_storage):
self._client_storage = client_storage
async def store_auth_data(
self,
backend: BackendType,
backend_data: dict[str, Any],
username: str,
encryption_key: str,
token: str,
):
auth_data = AuthData(
backend=backend,
backend_data=backend_data,
username=username,
encryption_key=encryption_key,
token=token,
)
await self.raw_write("soul_diary.client", auth_data.model_dump(mode="json"))
async def get_auth_data(self) -> AuthData | None:
if await self.raw_contains("soul_diary.client"):
data = await self.raw_read("soul_diary.client")
return AuthData.model_validate(data)
async def remove_auth_data(self):
if await self.raw_contains("soul_diary.client"):
await self.raw_remove("soul_diary.client")
async def clear(self):
await self._client_storage.clear_async()
async def raw_contains(self, key: str) -> bool:
return await self._client_storage.contains_key_async(key)
async def raw_read(self, key: str):
return await self._client_storage.get_async(key)
async def raw_write(self, key: str, value):
await self._client_storage.set_async(key, value)
async def raw_remove(self, key: str):
await self._client_storage.remove_async(key)

View File

@@ -0,0 +1,36 @@
import flet
from flet_route import Basket, Params
from soul_diary.ui.app.backend.local import LocalBackend
from soul_diary.ui.app.local_storage import LocalStorage
from soul_diary.ui.app.models import BackendType
from soul_diary.ui.app.routes import AUTH, SENSE_LIST
async def middleware(page: flet.Page, params: Params, basket: Basket):
if getattr(page.app, "backend_client", None) is not None:
if page.route == AUTH:
await page.go_async(SENSE_LIST)
return
local_storage = LocalStorage(client_storage=page.client_storage)
auth_data = await local_storage.get_auth_data()
if auth_data is None:
await page.go_async(AUTH)
return
if auth_data.backend == BackendType.LOCAL:
backend_client_class = LocalBackend
else:
await page.go_async(AUTH)
return
page.app.backend_client = backend_client_class(
local_storage=local_storage,
username=auth_data.username,
encryption_key=auth_data.encryption_key,
token=auth_data.token,
**auth_data.backend_data,
)
if page.route == AUTH:
await page.go_async(SENSE_LIST)

View File

@@ -0,0 +1,15 @@
from typing import Callable
import flet
from flet_route import Basket, Params
class BaseMiddleware:
async def __call__(
self,
page: flet.Page,
params: Params,
basket: Basket,
next_handler: Callable,
):
raise NotImplementedError

View File

@@ -0,0 +1,31 @@
import enum
import uuid
from pydantic import AwareDatetime, BaseModel, constr
class Emotion(str, enum.Enum):
JOY = "радость"
FORCE = "сила"
CALMNESS = "спокойствие"
SADNESS = "грусть"
MADNESS = "бешенство"
FEAR = "страх"
class BackendType(str, enum.Enum):
LOCAL = "local"
SOUL = "soul"
class Sense(BaseModel):
id: uuid.UUID
emotions: list[Emotion] = []
feelings: constr(min_length=1, strip_whitespace=True)
body: constr(min_length=1, strip_whitespace=True)
desires: constr(min_length=1, strip_whitespace=True)
created_at: AwareDatetime
class Options(BaseModel):
registration_enabled: bool

View File

@@ -0,0 +1,4 @@
INDEX = "/"
AUTH = "/auth"
SENSE_LIST = "/senses"
SENSE_ADD = "/senses/add"

View File

View File

@@ -0,0 +1,325 @@
import asyncio
from functools import partial
from typing import Any, Callable, Sequence
import flet
from pydantic import AnyHttpUrl
from soul_diary.ui.app.backend.exceptions import IncorrectCredentialsException, UserAlreadyExistsException
from soul_diary.ui.app.backend.local import LocalBackend
from soul_diary.ui.app.local_storage import LocalStorage
from soul_diary.ui.app.middlewares.base import BaseMiddleware
from soul_diary.ui.app.models import BackendType, Options
from soul_diary.ui.app.routes import AUTH, SENSE_LIST
from soul_diary.ui.app.views.exceptions import SoulServerIncorrectURL
from .base import BaseView, view
class AuthView(BaseView):
def __init__(
self,
local_storage: LocalStorage,
backend: BackendType | None = None,
backend_data: dict[str, Any] | None = None,
middlewares: Sequence[BaseMiddleware | Callable] = (),
):
self.top_container: flet.Container
self.center_container: flet.Container
self.bottom_container: flet.Container
self.local_storage = local_storage
self.initial_backend = self.backend = backend
self.initial_backend_data = self.backend_data = backend_data or {}
self.backend_registration_enabled: bool = True
self.username: str | None = None
self.password: str | None = None
super().__init__(middlewares=middlewares)
async def clear(self):
self.top_container.content = None
self.center_container.content = None
self.bottom_container.content = None
def clear_data(self):
self.backend = self.initial_backend
self.backend_data = self.initial_backend_data
self.username = None
self.password = None
async def setup(self):
self.top_container = flet.Container(alignment=flet.alignment.center)
self.center_container = flet.Container(alignment=flet.alignment.center)
self.bottom_container = flet.Container(alignment=flet.alignment.center)
self.container.content = flet.Column(
controls=[self.top_container, self.center_container, self.bottom_container],
width=300,
)
self.container.alignment = flet.alignment.center
self.view.route = AUTH
self.view.vertical_alignment = flet.MainAxisAlignment.CENTER
@view(initial=True)
async def entrypoint_view(self, page: flet.Page):
if self.initial_backend == BackendType.SOUL:
async def connect():
async with self.in_progress(page=page):
options = await self.connect_to_soul_server()
self.backend_registration_enabled = options.registration_enabled
await self.credentials_view(page=page)
loop = asyncio.get_running_loop()
loop.create_task(connect())
elif self.initial_backend == BackendType.LOCAL:
await self.credentials_view(page=page)
else:
await self.backend_view(page=page)
@view()
async def backend_view(self, page: flet.Page):
label = flet.Text("Выберите сервер")
self.top_container.content = label
backend_controls = flet.Column()
backend_dropdown = flet.Dropdown(
label="Бэкенд",
options=[
flet.dropdown.Option(text="Локально", key=BackendType.LOCAL.value),
flet.dropdown.Option(text="Soul Diary сервер", key=BackendType.SOUL.value),
],
value=None if self.backend is None else self.backend.value,
on_change=self.callback_change_backend,
)
container = flet.Container(
content=flet.Column(
controls=[backend_dropdown, backend_controls],
width=300,
),
)
self.center_container.content = container
connect_button = flet.ElevatedButton(
"Выбрать",
width=300,
height=50,
on_click=partial(self.callback_choose_backend, dropdown=backend_dropdown),
)
self.bottom_container.content = connect_button
@view()
async def soul_server_data_view(self, page: flet.Page):
label = flet.Text("Soul Diary сервер")
backend_button = flet.IconButton(
icon=flet.icons.ARROW_BACK,
on_click=self.callback_go_backend,
)
self.top_container.content = flet.Row(
controls=[backend_button, label],
width=300,
alignment=flet.MainAxisAlignment.START,
)
url_field = flet.TextField(
width=300,
label="URL",
value=self.backend_data.get("url"),
on_change=self.callback_change_soul_server_url,
)
self.center_container.content = url_field
connect_button = flet.ElevatedButton(
"Подключиться",
width=300,
height=50,
on_click=partial(self.callback_soul_server_connect, url_field=url_field),
)
self.bottom_container.content = connect_button
@view()
async def credentials_view(self, page: flet.Page):
controls = []
if self.initial_backend is None:
backend_data_button = flet.IconButton(
icon=flet.icons.ARROW_BACK,
on_click=self.callback_go_backend_data,
)
controls.append(backend_data_button)
self.top_container.content = flet.Row(
controls=controls,
width=300,
alignment=flet.MainAxisAlignment.START,
)
username_field = flet.TextField(
label="Логин",
on_change=self.callback_change_username,
)
password_field = flet.TextField(
label="Пароль",
password=True,
can_reveal_password=True,
on_change=self.callback_change_password,
)
self.center_container.content = flet.Column(
controls=[username_field, password_field],
width=300,
)
signin_button = flet.Container(
content=flet.ElevatedButton(
text="Войти",
width=300,
height=50,
on_click=partial(
self.callback_signin,
username_field=username_field,
password_field=password_field,
),
),
alignment=flet.alignment.center,
)
signup_button = flet.Container(
content=flet.ElevatedButton(
text="Зарегистрироваться",
width=300,
height=50,
disabled=not self.backend_registration_enabled,
on_click=partial(
self.callback_signup,
username_field=username_field,
password_field=password_field,
),
),
alignment=flet.alignment.center,
)
self.bottom_container.content = flet.Column(controls=[signin_button, signup_button])
async def callback_change_backend(self, event: flet.ControlEvent):
self.backend = BackendType(event.control.value)
event.control.error_text = None
await event.page.update_async()
async def callback_choose_backend(self, event: flet.ControlEvent, dropdown: flet.Dropdown):
if self.backend == BackendType.LOCAL:
await self.credentials_view(page=event.page)
elif self.backend == BackendType.SOUL:
await self.soul_server_data_view(page=event.page)
else:
dropdown.error_text = "Выберите тип бекенда"
await event.page.update_async()
async def callback_change_soul_server_url(self, event: flet.ControlEvent):
try:
AnyHttpUrl(event.control.value or "")
except:
event.control.error_text = "Некорректный URL"
self.backend_data["url"] = None
else:
event.control.error_text = None
self.backend_data["url"] = event.control.value
await event.page.update_async()
async def callback_soul_server_connect(
self,
event: flet.ControlEvent,
url_field: flet.TextField,
):
if self.backend == BackendType.SOUL:
async with self.in_progress(page=event.page):
try:
options = await self.connect_to_soul_server()
except SoulServerIncorrectURL:
url_field.error_text = "Некорректный URL"
except:
url_field.error_text = "Невозможно подключиться к серверу"
await event.page.update_async()
else:
self.backend_registration_enabled = options.registration_enabled
else:
await self.credentials_view(page=event.page)
async def connect_to_soul_server(self) -> Options:
try:
AnyHttpUrl(self.backend_data.get("url"))
except ValueError:
raise SoulServerIncorrectURL()
raise
async def callback_change_username(self, event: flet.ControlEvent):
self.username = event.control.value
async def callback_change_password(self, event: flet.ControlEvent):
self.password = event.control.value
async def callback_signin(
self,
event: flet.ControlEvent,
username_field: flet.TextField,
password_field: flet.TextField,
):
if not self.username:
username_field.error_text = "Заполните имя пользователя"
if not self.password:
password_field.error_text = "Заполните пароль"
if not self.username or not self.password:
await event.page.update_async()
return
if self.backend == BackendType.LOCAL:
backend_client = LocalBackend(local_storage=self.local_storage)
else:
raise
async with self.in_progress(page=event.page):
try:
await backend_client.login(username=self.username, password=self.password)
except IncorrectCredentialsException:
password_field.error_text = "Неверные имя пользователя и пароль"
await event.page.update_async()
return
event.page.app.backend_client = backend_client
await event.page.go_async(SENSE_LIST)
async def callback_signup(
self,
event: flet.ControlEvent,
username_field: flet.TextField,
password_field: flet.TextField,
):
if not self.username:
username_field.error_text = "Заполните имя пользователя"
if not self.password:
password_field.error_text = "Заполните пароль"
if not self.username or not self.password:
await event.page.update_async()
return
if self.backend == BackendType.LOCAL:
backend_client = LocalBackend(local_storage=self.local_storage)
else:
raise
async with self.in_progress(page=event.page):
try:
await backend_client.registration(username=self.username, password=self.password)
except UserAlreadyExistsException:
username_field.error_text = "Пользователь с таким именем уже существует"
await event.page.update_async()
return
event.page.app.backend_client = backend_client
await event.page.go_async(SENSE_LIST)
async def callback_go_backend(self, event: flet.ControlEvent):
await self.backend_view(page=event.page)
async def callback_go_backend_data(self, event: flet.ControlEvent):
if self.backend == BackendType.SOUL:
await self.soul_server_data_view(page=event.page)
elif self.backend == BackendType.LOCAL:
await self.backend_view(page=event.page)

View File

@@ -0,0 +1,117 @@
from contextlib import asynccontextmanager
from functools import partial, reduce
from typing import Any, Callable, Sequence
import flet
from flet_route import Basket, Params
from soul_diary.ui.app.middlewares.base import BaseMiddleware
def view(initial: bool = False, disabled: bool = False):
def decorator(function: Callable):
async def wrapper(self, page: flet.Page):
await self.clear()
await function(self, page=page)
await page.update_async()
wrapper._view_enabled = not disabled
wrapper._view_initial = initial
return wrapper
return decorator
def setup_middlewares(function: Callable):
async def wrapper(self, page: flet.Page, params: Params, basket: Basket) -> flet.View:
entrypoint = partial(function, self)
if self.middlewares:
entrypoint = partial(self.middlewares[-1], next_handler=entrypoint)
entrypoint = reduce(
lambda entrypoint, function: partial(entrypoint, next_handler=function),
self.middlewares[-2::-1],
entrypoint,
)
return await entrypoint(page=page, params=params, basket=basket)
return wrapper
class MetaView(type):
def __init__(cls, name: str, bases: tuple[type], attrs: dict[str, Any]):
super().__init__(name, bases, attrs)
is_abstract = attrs.pop("is_abstract", False)
if not is_abstract:
cls.setup_class(attrs=attrs)
def setup_class(cls, attrs: dict[str, Any]):
initial_view = None
for attr in attrs.values():
view_enabled = getattr(attr, "_view_enabled", False)
view_initial = getattr(attr, "_view_initial", False)
if view_enabled and view_initial:
if initial_view is not None:
raise ValueError(f"Initial view already defined: {initial_view.__name__}")
initial_view = attr
if initial_view is None:
raise ValueError("Initial view must be defined")
cls._initial_view = initial_view
class BaseView(metaclass=MetaView):
is_abstract = True
_initial_view: Callable | None
def __init__(self, middlewares: Sequence[BaseMiddleware | Callable] = ()):
self.middlewares = middlewares
self.container: flet.Container
self.stack: flet.Stack
self.view: flet.View
@setup_middlewares
async def entrypoint(self, page: flet.Page, params: Params, basket: Basket) -> flet.View:
self.container = flet.Container()
self.stack = flet.Stack(controls=[self.container])
self.view = flet.View(controls=[self.stack], route="/test")
await self.setup()
await self.clear()
self.clear_data()
await self.run_initial_view(page=page)
return self.view
async def setup(self):
pass
async def clear(self):
pass
def clear_data(self):
pass
@asynccontextmanager
async def in_progress(self, page: flet.Page, tooltip: str | None = None):
for control in self.stack.controls:
control.disabled = True
self.stack.controls.append(
flet.Container(
content=flet.ProgressRing(tooltip=tooltip),
alignment=flet.alignment.center,
),
)
await page.update_async()
yield
self.stack.controls.pop()
for control in self.stack.controls:
control.disabled = False
await page.update_async()
async def run_initial_view(self, page: flet.Page):
if self._initial_view is not None:
await self._initial_view(page=page)

View File

@@ -0,0 +1,2 @@
class SoulServerIncorrectURL(Exception):
pass

View File

View File

@@ -0,0 +1,235 @@
from functools import partial
from typing import Awaitable, Callable, Sequence
import flet
from soul_diary.ui.app.middlewares.base import BaseMiddleware
from soul_diary.ui.app.models import Emotion
from soul_diary.ui.app.routes import SENSE_ADD, SENSE_LIST
from .base import BaseView, view
class SenseAddView(BaseView):
def __init__(
self,
middlewares: Sequence[BaseMiddleware | Callable[[flet.Page], Awaitable]] = (),
):
self.title: flet.Text
self.content_container: flet.Container
self.buttons_row: flet.Row
self.emotions: list[Emotion] = []
self.feelings: str | None = None
self.body: str | None = None
self.desires: str | None = None
super().__init__(middlewares=middlewares)
async def clear(self):
self.title.value = ""
self.content_container.content = None
self.buttons_row.controls = []
def clear_data(self):
self.emotions = []
self.feelings = None
self.body = None
self.desires = None
async def setup(self):
# Top
self.title = flet.Text()
close_button = flet.IconButton(icon=flet.icons.CLOSE, on_click=self.callback_close)
top_container = flet.Container(
content=flet.Row(
[self.title, close_button],
alignment=flet.MainAxisAlignment.SPACE_BETWEEN,
),
margin=10,
)
# Center
self.content_container = flet.Container()
center_container = flet.Container(content=self.content_container, margin=10)
# Bottom
self.buttons_row = flet.Row()
bottom_container = flet.Container(content=self.buttons_row, margin=10)
# Build
self.container.content = flet.Column(
controls=[top_container, center_container, bottom_container],
width=600,
)
self.container.alignment = flet.alignment.center
self.view.route = SENSE_ADD
self.view.vertical_alignment = flet.MainAxisAlignment.CENTER
@view(initial=True)
async def emotions_view(self, page: flet.Page):
self.title.value = "Что ты чувствуешь?"
chips = flet.Row(
controls=[
flet.Chip(
label=flet.Text(emotion.value),
show_checkmark=False,
selected=emotion.value in self.emotions,
on_select=partial(self.callback_choose_emotion, emotion=emotion),
)
for emotion in Emotion
],
wrap=True,
)
self.content_container.content = flet.Column(
controls=[chips],
)
next_button = flet.IconButton(
flet.icons.ARROW_FORWARD,
on_click=self.callback_go_feelings_from_emotions,
)
self.buttons_row.controls = [next_button]
self.buttons_row.alignment = flet.MainAxisAlignment.END
@view()
async def feelings_view(self, page: flet.Page):
self.title.value = "Опиши свои чувства"
self.content_container.content = flet.TextField(
value=self.feelings,
multiline=True,
min_lines=10,
max_lines=10,
on_change=self.callback_change_feelings,
)
previous_button = flet.IconButton(
flet.icons.ARROW_BACK,
on_click=self.callback_go_emotions_from_feelings,
)
next_button = flet.IconButton(
flet.icons.ARROW_FORWARD,
on_click=self.callback_go_body_from_feelings,
)
self.buttons_row.controls = [previous_button, next_button]
self.buttons_row.alignment = flet.MainAxisAlignment.SPACE_BETWEEN
@view()
async def body_view(self, page: flet.Page):
self.title.value = "Опиши свои телесные ощущения"
self.content_container.content = flet.TextField(
value=self.body,
multiline=True,
min_lines=10,
max_lines=10,
on_change=self.callback_change_body,
)
previous_button = flet.IconButton(
flet.icons.ARROW_BACK,
on_click=self.callback_go_feelings_from_body,
)
next_button = flet.IconButton(
flet.icons.ARROW_FORWARD,
on_click=self.callback_go_desires_from_body,
)
self.buttons_row.controls = [previous_button, next_button]
self.buttons_row.alignment = flet.MainAxisAlignment.SPACE_BETWEEN
@view()
async def desires_view(self, page: flet.Page):
self.title.value = "Опиши свои желания на данный момент"
self.content_container.content = flet.TextField(
value=self.desires,
multiline=True,
min_lines=10,
max_lines=10,
on_change=self.callback_change_desires,
)
previous_button = flet.IconButton(
flet.icons.ARROW_BACK,
on_click=self.callback_go_body_from_desires,
)
apply_button = flet.IconButton(flet.icons.CREATE, on_click=self.callback_add_sense)
self.buttons_row.controls = [previous_button, apply_button]
self.buttons_row.alignment = flet.MainAxisAlignment.SPACE_BETWEEN
async def callback_close(self, event: flet.ControlEvent):
self.clear_data()
await event.page.go_async(SENSE_LIST)
async def callback_choose_emotion(self, event: flet.ControlEvent, emotion: Emotion):
if event.control.selected:
self.emotions.append(emotion)
emotions_column = self.content_container.content
if len(emotions_column.controls) > 1:
emotions_column.controls = emotions_column.controls[:1]
await event.page.update_async()
else:
self.emotions.remove(emotion)
async def callback_change_feelings(self, event: flet.ControlEvent):
self.feelings = event.control.value
async def callback_change_body(self, event: flet.ControlEvent):
self.body = event.control.value
async def callback_change_desires(self, event: flet.ControlEvent):
self.desires = event.control.value
async def callback_go_emotions_from_feelings(self, event: flet.ControlEvent):
await self.emotions_view(page=event.page)
async def callback_go_feelings_from_emotions(self, event: flet.ControlEvent):
if not self.emotions:
emotions_column = self.content_container.content
error_text = flet.Text("Выберите как минимум одну эмоцию", color=flet.colors.RED)
emotions_column.controls = [emotions_column.controls[0], error_text]
await event.page.update_async()
return
await self.feelings_view(page=event.page)
async def callback_go_feelings_from_body(self, event: flet.ControlEvent):
await self.feelings_view(page=event.page)
async def callback_go_body_from_feelings(self, event: flet.ControlEvent):
if self.feelings is None or not self.feelings.strip():
self.content_container.content.error_text = "Коротко опиши свои чувства"
await event.page.update_async()
return
await self.body_view(page=event.page)
async def callback_go_body_from_desires(self, event: flet.ControlEvent):
await self.body_view(page=event.page)
async def callback_go_desires_from_body(self, event: flet.ControlEvent):
if self.body is None or not self.body.strip():
self.content_container.content.error_text = "Коротко опиши свои телесные ощущения"
await event.page.update_async()
return
await self.desires_view(page=event.page)
async def callback_add_sense(self, event: flet.ControlEvent):
if self.desires is None or not self.desires.strip():
self.content_container.content.error_text = "Коротко опиши свои желания"
await event.page.update_async()
return
async with self.in_progress(page=event.page):
await event.page.app.backend_client.create_sense(
emotions=self.emotions,
feelings=self.feelings,
body=self.body,
desires=self.desires,
)
self.clear_data()
await event.page.go_async(SENSE_LIST)

View File

@@ -0,0 +1,95 @@
import asyncio
from typing import Awaitable, Callable, Sequence
import flet
from soul_diary.ui.app.local_storage import LocalStorage
from soul_diary.ui.app.middlewares.base import BaseMiddleware
from soul_diary.ui.app.models import Sense
from soul_diary.ui.app.routes import AUTH, SENSE_ADD, SENSE_LIST
from .base import BaseView, view
class SenseListView(BaseView):
def __init__(
self,
local_storage: LocalStorage,
middlewares: Sequence[BaseMiddleware | Callable[[flet.Page], Awaitable]] = (),
):
self.cards: flet.Column
self.local_storage = local_storage
super().__init__(middlewares=middlewares)
async def setup(self):
self.cards = flet.Column(alignment=flet.alignment.center, width=400)
add_button = flet.IconButton(
icon=flet.icons.ADD_CIRCLE_OUTLINE,
on_click=self.callback_add_sense,
)
settings_button = flet.IconButton(
icon=flet.icons.SETTINGS,
)
logout_button = flet.IconButton(
icon=flet.icons.LOGOUT,
on_click=self.callback_logout,
)
top_container = flet.Container(
content=flet.Row(
controls=[add_button, settings_button, logout_button],
alignment=flet.MainAxisAlignment.END,
),
)
self.container.content = flet.Column(
controls=[top_container, self.cards],
width=400,
)
self.container.alignment = flet.alignment.center
self.view.route = SENSE_LIST
self.view.vertical_alignment = flet.MainAxisAlignment.CENTER
self.view.scroll = flet.ScrollMode.ALWAYS
async def clear(self):
self.cards.controls = []
@view(initial=True)
async def sense_list_view(self, page: flet.Page):
self.cards.controls = [
flet.Container(
content=flet.ProgressRing(),
alignment=flet.alignment.center,
),
]
loop = asyncio.get_running_loop()
loop.create_task(self.render_sense_list(page=page))
async def render_sense_list(self, page: flet.Page):
senses = await page.app.backend_client.get_sense_list()
self.cards.controls = [await self.render_card_from_sense(sense) for sense in senses]
await page.update_async()
async def render_card_from_sense(self, sense: Sense) -> flet.Card:
feelings = flet.Container(content=flet.Text(sense.feelings), expand=True)
created_datetime = flet.Text(sense.created_at.strftime("%d %b %H:%M"))
return flet.Card(
content=flet.Container(
content=flet.Column(controls=[feelings, created_datetime]),
padding=10,
),
width=400,
height=100,
)
async def callback_add_sense(self, event: flet.ControlEvent):
await event.page.go_async(SENSE_ADD)
async def callback_logout(self, event: flet.ControlEvent):
await self.local_storage.remove_auth_data()
event.page.app.backend_client = None
await event.page.go_async(AUTH)

11
soul_diary/ui/cli.py Normal file
View File

@@ -0,0 +1,11 @@
import typer
from . import web
def get_cli() -> typer.Typer:
cli = typer.Typer()
cli.add_typer(web.get_cli(), name="web")
return cli

20
soul_diary/ui/service.py Normal file
View File

@@ -0,0 +1,20 @@
from facet import ServiceMixin
from .web import WebService, WebSettings, get_service as get_web_service
class UIService(ServiceMixin):
def __init__(self, web: WebService):
self._web = web
@property
def dependencies(self) -> list[ServiceMixin]:
return [
self._web,
]
def get_service() -> UIService:
settings = WebSettings()
web = get_web_service(settings=settings)
return UIService(web=web)

View File

@@ -0,0 +1,3 @@
from .cli import get_cli
from .service import WebService, get_service
from .settings import WebSettings

View File

@@ -0,0 +1,7 @@
from .cli import get_cli
if __name__ == "__main__":
cli = get_cli()
cli()

28
soul_diary/ui/web/cli.py Normal file
View File

@@ -0,0 +1,28 @@
import asyncio
import typer
from .service import get_service
from .settings import WebSettings, get_settings
def run(ctx: typer.Context):
settings: WebSettings = ctx.obj["settings"]
loop = asyncio.get_event_loop()
frontend_service = get_service(settings=settings)
loop.run_until_complete(frontend_service.run())
def settings_callback(ctx: typer.Context):
ctx.obj = ctx.obj or {}
ctx.obj["settings"] = get_settings()
def get_cli() -> typer.Typer:
cli = typer.Typer()
cli.callback()(settings_callback)
cli.command(name="run")(run)
return cli

View File

@@ -0,0 +1,39 @@
from typing import Any
import flet_fastapi
import uvicorn
from facet import ServiceMixin
from soul_diary.ui.app.models import BackendType
from soul_diary.ui.app import SoulDiaryApp
from .settings import WebSettings
class UvicornServer(uvicorn.Server):
def install_signal_handlers(self):
pass
class WebService(ServiceMixin):
def __init__(self, port: int = 8000, backend_data: dict[str, Any] | None = None):
self._port = port
self._backend_data = backend_data
@property
def port(self) -> int:
return self._port
async def start(self):
app = flet_fastapi.app(SoulDiaryApp(
# backend=BackendType.SOUL,
# backend_data=self._backend_data,
backend=BackendType.LOCAL,
).run)
config = uvicorn.Config(app=app, host="0.0.0.0", port=self._port)
server = UvicornServer(config)
self.add_task(server.serve())
def get_service(settings: WebSettings) -> WebService:
return WebService(port=settings.port, backend_data=settings.backend_data)

View File

@@ -0,0 +1,15 @@
from typing import Any
from pydantic import conint
from pydantic_settings import BaseSettings
class WebSettings(BaseSettings):
port: conint(ge=1, le=65535) = 8000
backend_data: dict[str, Any] = {
"url": "http://localhost:8001",
}
def get_settings() -> WebSettings:
return WebSettings()