yyc12345/coconut-leaf
1
0
Fork 0
Code Activity

refactor: remove legacy backend

Browse Source
This commit is contained in:
yyc12345
2026-07-17 16:45:38 +08:00
parent bb70f3432d
commit 06001b29f8
15 changed files with 0 additions and 2070 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend-legacy/.gitignore
-19
View File
@@ -1,19 +0,0 @@
## ======== Personal ========
# Database file
*.db
# Ignore setting file
coconut-leaf.toml
## ======== Python ========
# Python-generated files
__pycache__/
*.py[oc]
build/
dist/
wheels/
*.egg-info
# Virtual environments
.venv
backend-legacy/.python-version
-1
View File
@@ -1 +0,0 @@
3.11
backend-legacy/README.md
View File
backend-legacy/coconut-leaf.py
-84
View File
@@ -1,84 +0,0 @@
import sys
from argparse import ArgumentParser
from typing import cast
from pathlib import Path
import server
import config
import utils
import database
import logger
from logger import LOGGER, LoggerLevel
def GetUsernamePassword() -> tuple[str, str]:
print("What is the first username of this calendar system?")
cache = input()
while not utils.IsValidUsername(cache):
print("Sorry, invalid data. Please try again.")
cache = input()
username = cache
print("Input this user password:")
cache = input()
while not utils.IsValidPassword(cache):
print("Sorry, invalid data. Please try again.")
cache = input()
password = cache
return (username, password)
if __name__ == "__main__":
# Set as INFO level in default first,
# and we will change it once we load the configuration file.
logger.set_level(LoggerLevel.INFO)
# Receive arguments
parser = ArgumentParser(
description="The server of light, self-host and multi-account calendar system."
)
parser.add_argument(
"-c",
"--config",
required=True,
type=Path,
action="store",
metavar="CONFIG_TOML",
dest="config",
help="The configuration file for coconut-leaf",
)
parser.add_argument(
"-i",
"--init",
action="store_true",
dest="init",
help="Set for initialize the calendar system",
)
args = parser.parse_args()
# Show splash
LOGGER.info("Coconut-leaf")
LOGGER.info("A light, self-host and multi-account calendar system")
LOGGER.info("Project: https://github.com/yyc12345/coconut-leaf")
LOGGER.info("===================")
# Load config file
try:
config.setup_config(cast(Path, args.config))
except Exception as e:
LOGGER.critical(f"Error loading config file: {e}")
sys.exit(1)
# Change logging level again according to whether enable debug mode
logging_level = LoggerLevel.DEBUG if config.get_config().others.debug else LoggerLevel.INFO
logger.set_level(logging_level)
# Initialize the calendar system if needed
if cast(bool, args.init):
gotten_data = GetUsernamePassword()
calendar = database.CalendarDatabase()
calendar.init(*gotten_data)
calendar.close()
LOGGER.info("Staring server...")
server.run()
backend-legacy/coconut-leaf.template.toml
-22
View File
@@ -1,22 +0,0 @@
[database]
driver = "sqlite"
[database.config]
path = "coconut-leaf.db"
# [database]
# driver = "mysql"
#
# [database.config]
# host = "localhost"
# port = 3306
# user = "root"
# password = "password"
# database = "coconut_leaf"
[web]
port = 8848
[others]
auto-token-clean-duration = 86400
debug = true
backend-legacy/config.py
-130
View File
@@ -1,130 +0,0 @@
import tomllib
from dataclasses import dataclass
from enum import StrEnum
from typing import Optional
from pathlib import Path
class DatabaseDriver(StrEnum):
SQLITE = "sqlite"
MYSQL = "mysql"
@staticmethod
def from_raw(raw: dict):
return DatabaseDriver(raw["driver"])
@dataclass(frozen=True)
class SqliteDatabaseConfig:
path: str
"""Database path"""
@staticmethod
def from_raw(raw: dict):
return SqliteDatabaseConfig(raw["path"])
@dataclass(frozen=True)
class MysqlDatabaseConfig:
host: str
"""Database host"""
port: int
"""Database port"""
user: str
"""Database user"""
password: str
"""Database password"""
database: str
"""Database name"""
@staticmethod
def from_raw(raw: dict):
return MysqlDatabaseConfig(
raw["host"], raw["port"], raw["user"], raw["password"], raw["database"]
)
@dataclass(frozen=True)
class DatabaseConfig:
driver: DatabaseDriver
"""Database driver"""
config: SqliteDatabaseConfig | MysqlDatabaseConfig
"""Database config"""
@staticmethod
def from_raw(raw: dict):
if raw["driver"] == DatabaseDriver.SQLITE:
return DatabaseConfig(
DatabaseDriver.SQLITE, SqliteDatabaseConfig.from_raw(raw["config"])
)
elif raw["driver"] == DatabaseDriver.MYSQL:
return DatabaseConfig(
DatabaseDriver.MYSQL, MysqlDatabaseConfig.from_raw(raw["config"])
)
else:
raise ValueError("Invalid database driver")
@dataclass(frozen=True)
class WebConfig:
port: int
"""Web server port"""
@staticmethod
def from_raw(raw: dict):
return WebConfig(raw["port"])
@dataclass(frozen=True)
class OthersConfig:
debug: bool
"""Whether enable debug mode"""
auto_token_clean_duration: int
"""Auto token clean duration"""
@staticmethod
def from_raw(raw: dict):
return OthersConfig(raw["debug"], raw["auto-token-clean-duration"])
@dataclass(frozen=True)
class Config:
database: DatabaseConfig
web: WebConfig
others: OthersConfig
@staticmethod
def from_raw(raw: dict):
return Config(
database=DatabaseConfig.from_raw(raw["database"]),
web=WebConfig.from_raw(raw["web"]),
others=OthersConfig.from_raw(raw["others"]),
)
_CONFIG: Optional[Config] = None
def setup_config(p: Path) -> None:
"""
Setup config by given path.
Raise exception if config file is invalid.
"""
with open(p, "rb") as f:
raw = tomllib.load(f)
global _CONFIG
_CONFIG = Config.from_raw(raw)
def get_config() -> Config:
"""
Get config instance.
Raises RuntimeError if config is not loaded.
"""
if _CONFIG is None:
raise RuntimeError("Config is not loaded. Call setup_config() first.")
else:
return _CONFIG
backend-legacy/database.py
-708
View File
@@ -1,708 +0,0 @@
import sqlite3
import threading
from typing import cast
from pathlib import Path
from dataclasses import dataclass
from typing import Callable, ParamSpec, TypeVar, Generic
import dt
import utils
import config
from logger import LOGGER
T = TypeVar('T')
P = ParamSpec('P')
R = TypeVar('R')
@dataclass(frozen=True)
class ResponseBody(Generic[T]):
"""The generic response body for API return."""
success: bool
"""True if this operation is successful, otherwise false."""
error: str
"""The error message provided when operation failed."""
data: T | None
"""The payload provided when operation successed."""
class DbException(Exception):
"""Error occurs when manipulating with database."""
pass
def SafeDatabaseOperation(inner: Callable[P, R]) -> Callable[P, ResponseBody[R]]:
def wrapper(*args, **kwargs) -> ResponseBody[R]:
# extract self from args
self: 'CalendarDatabase' = args[0]
# get config
cfg = config.get_config()
with self.mutex:
# try to fetching database and allocate database cursor
try:
db = self._get_db()
self._allocate_cursor()
except Exception as e:
self._free_cursor()
if cfg.others.debug:
LOGGER.exception(e)
return ResponseBody(False, str(e), None)
# do real data work
try:
currentTime = utils.GetCurrentTimestamp()
if currentTime - self.latestClean > cfg.others.auto_token_clean_duration:
self.latestClean = currentTime
LOGGER.info('Cleaning outdated token...')
self.tokenOper_clean()
result = ResponseBody(True, '', inner(*args, **kwargs))
self._free_cursor()
db.commit()
return result
except Exception as e:
self._free_cursor()
db.rollback()
if cfg.others.debug:
LOGGER.exception(e)
return ResponseBody(False, str(e), None)
return wrapper
class CalendarDatabase:
db: sqlite3.Connection | None
cursor: sqlite3.Cursor | None
mutex: threading.Lock
latestClean: int
def __init__(self):
self.db = None
self.cursor = None
self.mutex = threading.Lock()
self.latestClean = 0
def open(self):
if (self.db is not None):
raise DbException('Database is already opened')
cfg = config.get_config()
match cfg.database.driver:
case config.DatabaseDriver.SQLITE:
self.db = sqlite3.connect(cast(config.SqliteDatabaseConfig, cfg.database.config).path, check_same_thread = False)
self.db.execute('PRAGMA encoding = "UTF-8";')
self.db.execute('PRAGMA foreign_keys = ON;')
case config.DatabaseDriver.MYSQL:
raise DbException('Not implemented database')
case _:
raise DbException('Unknow database type')
def init(self, username: str, password: str):
if (self.db is not None):
raise DbException('Database is already opened')
# establish tables
cfg = config.get_config()
backend_path = Path(__file__).resolve().parent
backend_sql_path = backend_path / 'sql'
match cfg.database.driver:
case config.DatabaseDriver.SQLITE:
sql_file = backend_sql_path / 'sqlite.sql'
case config.DatabaseDriver.MYSQL:
raise DbException('Not implemented database')
case _:
raise DbException('Unknow database type')
self.open()
db = self._get_db()
self._allocate_cursor()
cursor = self._get_cursor()
# execute script for creating tables
with open(sql_file, 'r', encoding='utf-8') as fsql:
cursor.executescript(fsql.read())
# add default user in user table
cursor.execute('INSERT INTO user VALUES (?, ?, ?, ?);', (
username,
utils.ComputePasswordHash(password),
1,
utils.GenerateSalt()
))
self._free_cursor()
# commit to database
db.commit()
def close(self):
if (self.db is None):
LOGGER.warning('Try to close null database.')
else:
self._free_cursor()
self.db.close()
self.db = None
def _get_db(self) -> sqlite3.Connection:
if (self.db is None):
raise DbException('There is no opened database')
else:
return self.db
def _allocate_cursor(self) -> None:
if (self.cursor is not None):
raise DbException('There is already opened database cursor')
else:
self.cursor = self._get_db().cursor()
def _get_cursor(self) -> sqlite3.Cursor:
if (self.cursor is None):
raise DbException('There is no opened database cursor')
else:
return self.cursor
def _free_cursor(self) -> None:
if (self.cursor is None):
LOGGER.warning('Try to free null databse cursor.')
else:
self.cursor.close()
self.cursor = None
# ======================= token related internal operation
def tokenOper_clean(self):
# remove outdated token
cursor = self._get_cursor()
cursor.execute('DELETE FROM token WHERE [token_expire_on] <= ?',(utils.GetCurrentTimestamp(), ))
def tokenOper_postpone_expireOn(self, token):
cursor = self._get_cursor()
cursor.execute('UPDATE token SET [token_expire_on] = ? WHERE [token] = ?;', (
utils.GetTokenExpireOn(),
token
))
def tokenOper_check_valid(self, token):
self.tokenOper_get_username(token)
def tokenOper_is_admin(self, username):
cursor = self._get_cursor()
cursor.execute('SELECT [is_admin] FROM user WHERE [name] = ?;',(username, ))
cache = cursor.fetchone()[0]
return cache == 1
def tokenOper_get_username(self, token):
cursor = self._get_cursor()
cursor.execute('SELECT [user] FROM token WHERE [token] = ? AND [token_expire_on] > ?;',(
token,
utils.GetCurrentTimestamp()
))
result = cursor.fetchone()[0]
# need postpone expire on time
self.tokenOper_postpone_expireOn(token)
return result
# =============================== # =============================== operation function
# =============================== common
@SafeDatabaseOperation
def common_salt(self, username):
cursor = self._get_cursor()
salt = utils.GenerateSalt()
cursor.execute('UPDATE user SET [salt] = ? WHERE [name] = ?;', (
salt,
username
))
return salt
@SafeDatabaseOperation
def common_login(self, username, password, clientUa, clientIp):
cursor = self._get_cursor()
cursor.execute('SELECT [password], [salt] FROM user WHERE [name] = ?;', (username, ))
(gotten_salt, gotten_password) = cursor.fetchone()
if password == utils.ComputePasswordHashWithSalt(gotten_password, gotten_salt):
token = utils.GenerateToken(username)
cursor.execute('UPDATE user SET [salt] = ? WHERE [name] = ?;', (
utils.GenerateSalt(), # regenerate a new slat to prevent re-login try
username
))
cursor.execute('INSERT INTO token VALUES (?, ?, ?, ?, ?);', (
username,
token,
utils.GetTokenExpireOn(), # add 2 day from now
clientUa,
clientIp,
))
return token
else:
# throw a exception to indicate fail to login
raise DbException('Login authentication failed')
@SafeDatabaseOperation
def common_webLogin(self, username, password, clientUa, clientIp):
cursor = self._get_cursor()
LOGGER.debug(f'WebLogin Username: {username}')
LOGGER.debug(f'WebLogin Password: {password}')
passwordHash = utils.ComputePasswordHash(password)
LOGGER.debug(f'WebLogin Password Hash: {passwordHash}')
cursor.execute('SELECT [name] FROM user WHERE [name] = ? AND [password] = ?;', (username, passwordHash))
if len(cursor.fetchall()) != 0:
token = utils.GenerateToken(username)
cursor.execute('INSERT INTO token VALUES (?, ?, ?, ?, ?);', (
username,
token,
utils.GetTokenExpireOn(), # add 2 day from now
clientUa,
clientIp,
))
return token
else:
# throw a exception to indicate fail to login
raise DbException('Login authentication failed')
@SafeDatabaseOperation
def common_logout(self, token):
cursor = self._get_cursor()
self.tokenOper_check_valid(token)
cursor.execute('DELETE FROM token WHERE [token] = ?;', (token, ))
return True
@SafeDatabaseOperation
def common_tokenValid(self, token):
self.tokenOper_check_valid(token)
return True
# =============================== calendar
@SafeDatabaseOperation
def calendar_getFull(self, token, startDateTime, endDateTime):
cursor = self._get_cursor()
username = self.tokenOper_get_username(token)
cursor.execute('SELECT calendar.* FROM calendar INNER JOIN collection \
ON collection.uuid = calendar.belong_to \
WHERE (collection.user = ? AND calendar.loop_date_time_end >= ? AND calendar.loop_date_time_start - (calendar.event_date_time_end - calendar.event_date_time_start) <= ?);',
(username, startDateTime, endDateTime))
return cursor.fetchall()
@SafeDatabaseOperation
def calendar_getList(self, token, startDateTime, endDateTime):
cursor = self._get_cursor()
username = self.tokenOper_get_username(token)
cursor.execute('SELECT calendar.uuid FROM calendar INNER JOIN collection \
ON collection.uuid = calendar.belong_to \
WHERE (collection.user = ? AND calendar.loop_date_time_end >= ? AND calendar.loop_date_time_start - (calendar.event_date_time_end - calendar.event_date_time_start) <= ?);',
(username, startDateTime, endDateTime))
return tuple(map(lambda x: x[0], cursor.fetchall()))
@SafeDatabaseOperation
def calendar_getDetail(self, token, uuid):
cursor = self._get_cursor()
self.tokenOper_check_valid(token)
cursor.execute('SELECT * FROM calendar WHERE [uuid] = ?;', (uuid, ))
return cursor.fetchone()
@SafeDatabaseOperation
def calendar_update(self, token, uuid, lastChange, **optArgs):
cursor = self._get_cursor()
self.tokenOper_check_valid(token)
# get prev data
cursor.execute('SELECT * FROM calendar WHERE [uuid] = ? AND [last_change] = ?;', (uuid, lastChange))
analyseData = list(cursor.fetchone())
# construct update data
lastupdate = utils.GenerateUUID()
sqlList = [
'[last_change] = ?',
]
argumentsList = [
lastupdate,
]
# analyse opt arg
reAnalyseLoop = False
cache = optArgs.get('belongTo', None)
if cache is not None:
sqlList.append('[belong_to] = ?')
argumentsList.append(cache)
cache = optArgs.get('title', None)
if cache is not None:
sqlList.append('[title] = ?')
argumentsList.append(cache)
cache = optArgs.get('description', None)
if cache is not None:
sqlList.append('[description] = ?')
argumentsList.append(cache)
cache = optArgs.get('eventDateTimeStart', None)
if cache is not None:
sqlList.append('[event_date_time_start] = ?')
argumentsList.append(cache)
reAnalyseLoop = True
analyseData[5] = cache
cache = optArgs.get('eventDateTimeEnd', None)
if cache is not None:
sqlList.append('[event_date_time_end] = ?')
argumentsList.append(cache)
cache = optArgs.get('loopRules', None)
if cache is not None:
sqlList.append('[loop_rules] = ?')
argumentsList.append(cache)
reAnalyseLoop = True
analyseData[8] = cache
cache = optArgs.get('timezoneOffset', None)
if cache is not None:
sqlList.append('[timezone_offset] = ?')
argumentsList.append(cache)
reAnalyseLoop = True
analyseData[7] = cache
if reAnalyseLoop:
# re-compute loop data and upload it into list
sqlList.append('[loop_date_time_start] = ?')
argumentsList.append(analyseData[5])
sqlList.append('[loop_date_time_end] = ?')
argumentsList.append(str(dt.ResolveLoopStr(
analyseData[8],
analyseData[5],
analyseData[7]
)))
# execute
argumentsList.append(uuid)
cursor.execute('UPDATE calendar SET {} WHERE [uuid] = ?;'.format(', '.join(sqlList)),
tuple(argumentsList))
if cursor.rowcount != 1:
raise DbException('Fail to update due to no matched rows or too much rows.')
return lastupdate
@SafeDatabaseOperation
def calendar_add(self, token, belongTo, title, description, eventDateTimeStart, eventDateTimeEnd, loopRules, timezoneOffset):
cursor = self._get_cursor()
self.tokenOper_check_valid(token)
newuuid = utils.GenerateUUID()
lastupdate = utils.GenerateUUID()
# analyse loopRules and output following 2 fileds.
loopDateTimeStart = eventDateTimeStart
loopDateTimeEnd = dt.ResolveLoopStr(loopRules, eventDateTimeStart, timezoneOffset)
cursor.execute('INSERT INTO calendar VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?);',
(newuuid,
belongTo,
title,
description,
lastupdate,
eventDateTimeStart,
eventDateTimeEnd,
timezoneOffset,
loopRules,
loopDateTimeStart,
loopDateTimeEnd))
return newuuid
@SafeDatabaseOperation
def calendar_delete(self, token, uuid, lastChange):
cursor = self._get_cursor()
self.tokenOper_check_valid(token)
cursor.execute('DELETE FROM calendar WHERE [uuid] = ? AND [last_change] = ?;', (uuid, lastChange))
if cursor.rowcount != 1:
raise DbException('Fail to delete due to no matched rows or too much rows.')
return True
# =============================== collection
@SafeDatabaseOperation
def collection_getFullOwn(self, token):
cursor = self._get_cursor()
username = self.tokenOper_get_username(token)
cursor.execute('SELECT [uuid], [name], [last_change] FROM collection WHERE [user] = ?;', (username, ))
return cursor.fetchall()
@SafeDatabaseOperation
def collection_getListOwn(self, token):
cursor = self._get_cursor()
username = self.tokenOper_get_username(token)
cursor.execute('SELECT [uuid] FROM collection WHERE [user] = ?;', (username, ))
return tuple(map(lambda x: x[0], cursor.fetchall()))
@SafeDatabaseOperation
def collection_getDetailOwn(self, token, uuid):
cursor = self._get_cursor()
username = self.tokenOper_get_username(token)
cursor.execute('SELECT [uuid], [name], [last_change] FROM collection WHERE [user] = ? AND [uuid] = ?;', (username, uuid))
return cursor.fetchone()
@SafeDatabaseOperation
def collection_addOwn(self, token, newname):
cursor = self._get_cursor()
username = self.tokenOper_get_username(token)
newuuid = utils.GenerateUUID()
lastupdate = utils.GenerateUUID()
cursor.execute('INSERT INTO collection VALUES (?, ?, ?, ?);',
(newuuid, newname, username, lastupdate))
return newuuid
@SafeDatabaseOperation
def collection_updateOwn(self, token, uuid, newname, lastChange):
cursor = self._get_cursor()
self.tokenOper_check_valid(token)
lastupdate = utils.GenerateUUID()
cursor.execute('UPDATE collection SET [name] = ?, [last_change] = ? WHERE [uuid] = ? AND [last_change] = ?;', (
newname,
lastupdate,
uuid,
lastChange
))
if cursor.rowcount != 1:
raise DbException('Fail to update due to no matched rows or too much rows.')
return lastupdate
@SafeDatabaseOperation
def collection_deleteOwn(self, token, uuid, lastChange):
cursor = self._get_cursor()
self.tokenOper_check_valid(token)
cursor.execute('DELETE FROM collection WHERE [uuid] = ? AND [last_change] = ?;', (
uuid,
lastChange
))
if cursor.rowcount != 1:
raise DbException('Fail to delete due to no matched rows or too much rows.')
return True
@SafeDatabaseOperation
def collection_getSharing(self, token, uuid):
cursor = self._get_cursor()
self.tokenOper_check_valid(token)
cursor.execute('SELECT [target] FROM share WHERE [uuid] = ?;', (uuid, ))
return tuple(map(lambda x: x[0], cursor.fetchall()))
@SafeDatabaseOperation
def collection_deleteSharing(self, token, uuid, target, lastChange):
cursor = self._get_cursor()
self.tokenOper_check_valid(token)
lastupdate = utils.GenerateUUID()
cursor.execute('UPDATE collection SET [last_change] = ?, WHERE [uuid] = ? AND [last_change] = ?;', (lastupdate, uuid, lastChange))
if cursor.rowcount != 1:
raise DbException('Fail to delete due to no matched rows or too much rows.')
cursor.execute('DELETE FROM share WHERE [uuid] = ? AND [target] = ?;', (uuid, target))
if cursor.rowcount != 1:
raise DbException('Fail to delete due to no matched rows or too much rows.')
return lastupdate
@SafeDatabaseOperation
def collection_addSharing(self, token, uuid, target, lastChange):
cursor = self._get_cursor()
self.tokenOper_check_valid(token)
lastupdate = utils.GenerateUUID()
cursor.execute('UPDATE collection SET [last_change] = ? WHERE [uuid] = ? AND [last_change] = ?;', (lastupdate, uuid, lastChange))
if cursor.rowcount != 1:
raise DbException('Fail to delete due to no matched rows or too much rows.')
cursor.execute('SELECT * FROM share WHERE [uuid] = ? AND [target] = ?;', (uuid, target))
if len(cursor.fetchall()) != 0:
raise DbException('Fail to insert duplicated item.')
cursor.execute('INSERT INTO share VALUES (?, ?);', (uuid, target))
return lastupdate
@SafeDatabaseOperation
def collection_getShared(self, token):
cursor = self._get_cursor()
username = self.tokenOper_get_username(token)
cursor.execute('SELECT collection.uuid, collection.name, collection.user \
FROM share INNER JOIN collection \
ON share.uuid = collection.uuid \
WHERE share.target = ?;', (username, ))
return cursor.fetchall()
# =============================== todo
@SafeDatabaseOperation
def todo_getFull(self, token):
cursor = self._get_cursor()
username = self.tokenOper_get_username(token)
cursor.execute('SELECT * FROM todo WHERE [belong_to] = ?;', (username, ))
return cursor.fetchall()
@SafeDatabaseOperation
def todo_getList(self, token):
cursor = self._get_cursor()
username = self.tokenOper_get_username(token)
cursor.execute('SELECT [uuid] FROM todo WHERE [belong_to] = ?;', (username, ))
return tuple(map(lambda x: x[0], cursor.fetchall()))
@SafeDatabaseOperation
def todo_getDetail(self, token, uuid):
cursor = self._get_cursor()
username = self.tokenOper_get_username(token)
cursor.execute('SELECT * FROM todo WHERE [belong_to] = ? AND [uuid] = ?;', (username, uuid))
return cursor.fetchone()
@SafeDatabaseOperation
def todo_add(self, token):
cursor = self._get_cursor()
username = self.tokenOper_get_username(token)
newuuid = utils.GenerateUUID()
lastupdate = utils.GenerateUUID()
returnedData = (
newuuid,
username,
'',
lastupdate,
)
cursor.execute('INSERT INTO todo VALUES (?, ?, ?, ?);', returnedData)
return returnedData
@SafeDatabaseOperation
def todo_update(self, token, uuid, data, lastChange):
cursor = self._get_cursor()
# check valid token
self.tokenOper_check_valid(token)
# update
newLastChange = utils.GenerateUUID()
cursor.execute('UPDATE todo SET [data] = ?, [last_change] = ? WHERE [uuid] = ? AND [last_change] = ?;', (
data,
newLastChange,
uuid,
lastChange
))
if cursor.rowcount != 1:
raise DbException('Fail to update due to no matched rows or too much rows.')
return newLastChange
@SafeDatabaseOperation
def todo_delete(self, token, uuid, lastChange):
cursor = self._get_cursor()
# check valid token
self.tokenOper_check_valid(token)
# delete
cursor.execute('DELETE FROM todo WHERE [uuid] = ? AND [last_change] = ?;', (uuid, lastChange))
if cursor.rowcount != 1:
raise DbException('Fail to delete due to no matched rows or too much rows.')
return True
# =============================== admin
@SafeDatabaseOperation
def admin_get(self, token):
cursor = self._get_cursor()
username = self.tokenOper_get_username(token)
if not self.tokenOper_is_admin(username):
raise DbException('Permission denied.')
cursor.execute('SELECT [name], [is_admin] FROM user;')
return tuple(map(lambda x: (x[0], x[1] == 1), cursor.fetchall()))
@SafeDatabaseOperation
def admin_add(self, token, newname):
cursor = self._get_cursor()
username = self.tokenOper_get_username(token)
if not self.tokenOper_is_admin(username):
raise DbException('Permission denied.')
newpassword = utils.ComputePasswordHash(utils.GenerateUUID())
cursor.execute('INSERT INTO user VALUES (?, ?, ?, ?);', (
newname,
newpassword,
0,
utils.GenerateSalt()
))
return (newname, False)
@SafeDatabaseOperation
def admin_update(self, token, _username, **optArgs):
cursor = self._get_cursor()
username = self.tokenOper_get_username(token)
if not self.tokenOper_is_admin(username):
raise DbException('Permission denied.')
# construct data
sqlList = []
argumentsList = []
# analyse opt arg
cache = optArgs.get('password', None)
if cache is not None:
sqlList.append('[password] = ?')
argumentsList.append(utils.ComputePasswordHash(cache))
cache = optArgs.get('isAdmin', None)
if cache is not None:
sqlList.append('[is_admin] = ?')
argumentsList.append(1 if cache else 0)
# execute
argumentsList.append(_username)
cursor.execute('UPDATE user SET {} WHERE [name] = ?;'.format(', '.join(sqlList)),
tuple(argumentsList))
LOGGER.debug(cache)
LOGGER.debug(tuple(argumentsList))
if cursor.rowcount != 1:
raise DbException('Fail to update due to no matched rows or too much rows.')
return True
@SafeDatabaseOperation
def admin_delete(self, token, username):
cursor = self._get_cursor()
_username = self.tokenOper_get_username(token)
if not self.tokenOper_is_admin(_username):
raise DbException('Permission denied.')
# delete
cursor.execute('DELETE FROM user WHERE [name] = ?;', (username, ))
if cursor.rowcount != 1:
raise DbException('Fail to delete due to no matched rows or too much rows.')
return True
# =============================== profile
@SafeDatabaseOperation
def profile_isAdmin(self, token):
cursor = self._get_cursor()
username = self.tokenOper_get_username(token)
return self.tokenOper_is_admin(username)
@SafeDatabaseOperation
def profile_changePassword(self, token, newpassword):
cursor = self._get_cursor()
username = self.tokenOper_get_username(token)
cursor.execute('UPDATE user SET [password] = ? WHERE [name] = ?;', (
utils.ComputePasswordHash(newpassword),
username
))
return True
@SafeDatabaseOperation
def profile_getToken(self, token):
cursor = self._get_cursor()
username = self.tokenOper_get_username(token)
cursor.execute('SELECT * FROM token WHERE [user] = ?;', (
username,
))
return cursor.fetchall()
@SafeDatabaseOperation
def profile_deleteToken(self, token, deleteToken):
cursor = self._get_cursor()
_username = self.tokenOper_get_username(token)
# delete
cursor.execute('DELETE FROM token WHERE [user] = ? AND [token] = ?;', (
_username,
deleteToken
))
if cursor.rowcount != 1:
raise DbException('Fail to delete due to no matched rows or too much rows.')
return True
backend-legacy/dt.py
-289
View File
@@ -1,289 +0,0 @@
import datetime
import re
import logging
import typing
from functools import reduce
import utils
MonthDayCount = (31, 28, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31)
MIN_DATETIME = datetime.datetime(1950, 1, 1, 0, 0, 0, 0, tzinfo=datetime.timezone.utc)
MAX_DATETIME = datetime.datetime(2200, 1, 1, 0, 0, 0, 0, tzinfo=datetime.timezone.utc)
MIN_TIMESTAMP = int(MIN_DATETIME.timestamp() / 60)
MAX_TIMESTAMP = int(MAX_DATETIME.timestamp() / 60)
DAY1_SPAN = 60 * 24
DAY7_SPAN = 7 * DAY1_SPAN
LoopHandle = typing.Callable[[re.Match, int, int, int], int]
def ResolveLoopStr(strl: str, starttime: int, tzoffset: int) -> int:
# check no loop
if strl == '':
return starttime
# try compute from loopStop
(loopRules, loopStopRules) = strl.split('-')
cache = precompiledLoopStopRules['infinity'].search(loopStopRules)
if cache is not None:
return MAX_TIMESTAMP
cache = precompiledLoopStopRules['datetime'].search(loopStopRules)
if cache is not None:
return int(cache.group(1)) # group 1 is datetime
cache = precompiledLoopStopRules['times'].search(loopStopRules)
if cache is not None:
loopTimes = int(cache.group(1)) # for follwing calc
else:
raise Exception('Invalid loopStopRules') # invalid rules
for rules in precompiledLoopRules:
cache = rules[0].search(loopRules)
if cache is not None:
return rules[1](cache, starttime, loopTimes, tzoffset)
else:
raise Exception('Invalid loopRules')
def LoopHandle_Year(searchResult: re.Match, starttime: int, times: int, tzoffset: int) -> int:
clientDate = datetime.datetime.fromtimestamp(starttime * 60, UTCTimezone(tzoffset))
isStrict = searchResult.group(1) == 'S'
yearSpan = int(searchResult.group(2))
times -= 1
newYear = clientYear = clientDate.year
newMonth = clientMonth = clientDate.month
newDay = clientDay = clientDate.day
if clientMonth == 2 and clientDay == 29:
if isStrict:
realSpan = utils.LCM(yearSpan, 4)
logging.debug(realSpan)
valCache = starttime
while valCache < MAX_TIMESTAMP and times > 0:
newYear += realSpan
if not IsLeapYear(newYear):
continue
valCache = starttime + DAY1_SPAN * (DaysCount(newYear, newMonth, newDay) - DaysCount(clientYear, clientMonth, clientDay))
times -= 1
else:
newYear += times * yearSpan
if not IsLeapYear(newYear):
newDay = 28 # migrate to 28
else:
# if times == 1, no extra datetime need to be added
newYear += times * yearSpan
val = starttime + DAY1_SPAN * (DaysCount(newYear, newMonth, newDay) - DaysCount(clientYear, clientMonth, clientDay))
return val if val < MAX_TIMESTAMP else MAX_TIMESTAMP
def LoopHandle_Month(searchResult: re.Match, starttime: int, times: int, tzoffset: int) -> int:
isStrict = searchResult.group(1) == 'S'
loopType = searchResult.group(2)
monthSpan = int(searchResult.group(3))
# we should get original data in each method
times -= 1
clientDate = datetime.datetime.fromtimestamp(starttime * 60, UTCTimezone(tzoffset))
newYear = clientYear = clientDate.year
newMonth = clientMonth = clientDate.month
newDay = clientDay = clientDate.day
# data struct
# dayStatistics =
# (dayForwards || dayBackwards || weeksForward, dayOfWeek || weeksBackwards, dayOfWeek)
# ( A || B || C || D )
dayStatistics = GetDayInMonth(clientYear, clientMonth, clientDay)
if isStrict:
if loopType == 'A':
while times > 0:
newMonth += monthSpan
if newMonth > 12:
newYear += int((newMonth - 1) / 12)
newMonth = ((newMonth - 1) % 12) + 1
if newYear > MAX_DATETIME.year:
break
maxDays = MonthDayCount[newMonth - 1] + (1 if newMonth == 2 and IsLeapYear(newYear) else 0)
if dayStatistics[0] <= maxDays:
times -= 1
elif loopType == 'B':
while times > 0:
newMonth += monthSpan
if newMonth > 12:
newYear += int((newMonth - 1) / 12)
newMonth = ((newMonth - 1) % 12) + 1
if newYear > MAX_DATETIME.year:
break
maxDays = MonthDayCount[newMonth - 1] + (1 if newMonth == 2 and IsLeapYear(newYear) else 0)
if dayStatistics[1] <= maxDays:
times -= 1
elif loopType == 'C':
while times > 0:
newMonth += monthSpan
if newMonth > 12:
newYear += int((newMonth - 1) / 12)
newMonth = ((newMonth - 1) % 12) + 1
if newYear > MAX_DATETIME.year:
break
monthStatistics = GetMonthWeekStatistics(newYear, newMonth)
if dayStatistics[2] <= monthStatistics[dayStatistics[3]]:
times -= 1
elif loopType == 'D':
while times > 0:
newMonth += monthSpan
if newMonth > 12:
newYear += int((newMonth - 1) / 12)
newMonth = ((newMonth - 1) % 12) + 1
if newYear > MAX_DATETIME.year:
break
monthStatistics = GetMonthWeekStatistics(newYear, newMonth)
if dayStatistics[4] <= monthStatistics[dayStatistics[5]]:
times -= 1
else:
newMonth += times * monthSpan
newYear += int((newMonth - 1) / 12)
newMonth = ((newMonth - 1) % 12) + 1
# all method need calc newDay and it should be the last day of current selected month
# so calc it in there
newDay = MonthDayCount[newMonth - 1] + (1 if newMonth == 2 and IsLeapYear(newYear) else 0)
val = starttime + DAY1_SPAN * (DaysCount(newYear, newMonth, newDay) - DaysCount(clientYear, clientMonth, clientDay))
return val if val < MAX_TIMESTAMP else MAX_TIMESTAMP
def LoopHandle_Week(searchResult: re.Match, starttime: int, times: int, tzoffset: int) -> int:
weekOccupied = tuple(map(lambda x: x == 'T', searchResult.group(1)))
weekEventCount = reduce(lambda x, y: x + (1 if y else 0), weekOccupied, 0)
if weekEventCount == 0:
raise Exception('Invalid week format')
weekSpan = int(searchResult.group(2))
nowDayOfWeek = datetime.datetime.fromtimestamp(starttime * 60, UTCTimezone(tzoffset)).weekday()
if not weekOccupied[nowDayOfWeek]:
times-=1 # if first event is not suit for week loop rules, minus one more event to suit it.
fullWeek = int(times / weekEventCount)
remainEvent = times % weekEventCount
val = starttime + DAY7_SPAN * fullWeek * weekSpan
if val > MAX_TIMESTAMP:
return MAX_TIMESTAMP # return now, to reduce calc usage
while remainEvent != 0:
val += DAY1_SPAN
if weekOccupied[nowDayOfWeek % 7]:
remainEvent -= 1
nowDayOfWeek += 1
val -= 1
return val if val < MAX_TIMESTAMP else MAX_TIMESTAMP
def LoopHandle_Day(searchResult: re.Match, starttime: int, times: int, tzoffset: int) -> int:
val = starttime + DAY1_SPAN * times * int(searchResult.group(1))
val -= 1
return val if val < MAX_TIMESTAMP else MAX_TIMESTAMP
precompiledLoopRules: tuple[tuple[re.Pattern, LoopHandle], ...] = (
(re.compile(r'^Y([SR]{1})([1-9]\d*)$'), LoopHandle_Year),
(re.compile(r'^M([SR]{1})([ABCD]{1})([1-9]\d*)$'), LoopHandle_Month),
(re.compile(r'^W([TF]{7})([1-9]\d*)$'), LoopHandle_Week),
(re.compile(r'^D([1-9]\d*)$'), LoopHandle_Day)
)
precompiledLoopStopRules: dict[str, re.Pattern] = {
'infinity': re.compile(r'^F$'),
'datetime': re.compile(r'^D([1-9]\d*|0)$'),
'times': re.compile(r'^T([1-9]\d*)$')
}
def LeapYearCountEx(endYear: int, includeThis: bool = False, baseYear: int = 1, includeBase: bool = True):
if not includeThis:
endYear -= 1
if includeBase:
baseYear -= 1
endly = int(endYear / 4)
endly -= int(endYear / 100)
endly += int(endYear / 400)
basely = int(baseYear / 4)
basely -= int(baseYear / 100)
basely += int(baseYear / 400)
return (endly - basely)
def LeapYearCount(year: int):
return LeapYearCountEx(year, False, 1, True)
def IsLeapYear(year: int):
isLeap = False
if year % 4 == 0:
isLeap = True
if year % 100 == 0:
isLeap = False
if year % 400 == 0:
isLeap = True
return isLeap
def DaysCount(year: int, month: int, day: int):
ly = LeapYearCountEx(year, False, 1, True)
days = 365 * (year - 1)
days += ly
for index in range(1, month, 1):
days += MonthDayCount[index - 1]
if (month > 2) and IsLeapYear(year):
days += 1
days += day - 1
return days
def DayOfWeek(year: int, month: int, day: int):
# as we know, 1/1/1900 is Monday.
# via this method, we can got 1/1/1 is Monday
# compute day span
days=DaysCount(year, month, day)
# return day of week (from 0 - 6, corresponding with python)
return days % 7
def GetDayInMonth(year: int, month: int, day: int):
days = MonthDayCount[month - 1] + (1 if (month == 2 and IsLeapYear(year)) else 0)
firstDayOfWeek = DayOfWeek(year, month, 1)
dayOfWeek = (firstDayOfWeek + day - 1) % 7
dayForwards = day
dayBackwards = days - day + 1
weeksForward = int((dayForwards - 1) / 7) + 1
weeksBackwards = int((dayBackwards - 1) / 7) + 1
return (dayForwards, dayBackwards, weeksForward, dayOfWeek, weeksBackwards, dayOfWeek)
def GetMonthWeekStatistics(year: int, month: int):
days = MonthDayCount[month - 1] + (1 if (month == 2 and IsLeapYear(year)) else 0)
firstDayOfWeek = DayOfWeek(year, month, 1)
lastDayOfWeek = (firstDayOfWeek + days - 1) % 7
result = [4, 4, 4, 4, 4, 4, 4]
remain = days % 7
week = firstDayOfWeek
while remain > 0:
result[week % 7] += 1
week += 1
remain -= 1
return tuple(result)
class UTCTimezone(datetime.tzinfo):
__offset: int
def __init__(self, offset: int = 0):
self.__offset = offset
def utcoffset(self, dt):
return datetime.timedelta(minutes=self.__offset)
def tzname(self, dt):
return 'UTC {}'.format(self.__offset)
def dst(self, dt):
return datetime.timedelta(0)
backend-legacy/logger.py
-42
View File
@@ -1,42 +0,0 @@
import logging
import enum
def _build_logger() -> tuple[logging.Logger, logging.Handler]:
# Create a new logger which is independent with Flask
logger = logging.getLogger("my_console_logger")
# Avoid message was propagated to root logger or captured by Flask logger.
logger.propagate = False
# Set initial level.
logger.setLevel(logging.INFO)
# Create StreamHandler to output into stderr.
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.DEBUG)
# Set format for it.
formatter = logging.Formatter("[%(levelname)s] %(message)s")
console_handler.setFormatter(formatter)
# Add handler
logger.addHandler(console_handler)
return (logger, console_handler)
(LOGGER, CONSOLE_HANDLER) = _build_logger()
class LoggerLevel(enum.IntEnum):
DEBUG = enum.auto()
INFO = enum.auto()
def set_level(level: LoggerLevel) -> None:
logging_level: int = logging.INFO
match level:
case LoggerLevel.DEBUG:
logging_level = logging.DEBUG
case LoggerLevel.INFO:
logging_level = logging.INFO
LOGGER.setLevel(logging_level)
CONSOLE_HANDLER.setLevel(logging_level)
backend-legacy/pyproject.toml
-14
View File
@@ -1,14 +0,0 @@
[project]
name = "coleaf-backend"
version = "1.1.0"
description = "The backend of coconut-leaf."
readme = "README.md"
requires-python = ">=3.11"
dependencies = [
"flask==2.2.3",
]
[tool.uv]
constraint-dependencies = [
"Werkzeug==2.2.2",
"MarkupSafe==2.1.5"
]
backend-legacy/server.py
-513
View File
@@ -1,513 +0,0 @@
from flask import Flask
from flask import request
from dataclasses import dataclass
from typing import Any, Callable, ParamSpec, TypeVar, Generic
import config
import database
import utils
from logger import LOGGER
from database import ResponseBody
app = Flask(__name__)
calendar_db = database.CalendarDatabase()
# region: API Route
# region: Common
@app.route("/common/salt", methods=["POST"])
def api_common_saltHandle():
return SmartDbCaller(
calendar_db.common_salt, (FormField("username", str, False),), None
)
@app.route("/common/login", methods=["POST"])
def api_common_loginHandle():
clientInfo = FetchClientNetworkInfo()
return SmartDbCaller(
calendar_db.common_login,
(
FormField("username", str, False),
FormField("password", str, False),
FormField("clientUa", str, False),
FormField("clientIp", str, False),
),
{"clientUa": clientInfo.user_agent, "clientIp": clientInfo.ip_addr},
)
@app.route("/common/webLogin", methods=["POST"])
def api_common_webLoginHandle():
clientInfo = FetchClientNetworkInfo()
return SmartDbCaller(
calendar_db.common_webLogin,
(
FormField("username", str, False),
FormField("password", str, False),
FormField("clientUa", str, False),
FormField("clientIp", str, False),
),
{"clientUa": clientInfo.user_agent, "clientIp": clientInfo.ip_addr},
)
@app.route("/common/logout", methods=["POST"])
def api_common_logoutHandle():
return SmartDbCaller(
calendar_db.common_logout, (FormField("token", str, False),), None
)
@app.route("/common/tokenValid", methods=["POST"])
def api_common_tokenValidHandle():
return SmartDbCaller(
calendar_db.common_tokenValid, (FormField("token", str, False),), None
)
# endregion
# region: Calendar
@app.route("/calendar/getFull", methods=["POST"])
def api_calendar_getFullHandle():
return SmartDbCaller(
calendar_db.calendar_getFull,
(
FormField("token", str, False),
FormField("startDateTime", int, False),
FormField("endDateTime", int, False),
),
None,
)
@app.route("/calendar/getList", methods=["POST"])
def api_calendar_getListHandle():
return SmartDbCaller(
calendar_db.calendar_getList,
(
FormField("token", str, False),
FormField("startDateTime", int, False),
FormField("endDateTime", int, False),
),
None,
)
@app.route("/calendar/getDetail", methods=["POST"])
def api_calendar_getDetailHandle():
return SmartDbCaller(
calendar_db.calendar_getDetail,
(FormField("token", str, False), FormField("uuid", str, False)),
None,
)
@app.route("/calendar/update", methods=["POST"])
def api_calendar_updateHandle():
return SmartDbCaller(
calendar_db.calendar_update,
(
FormField("token", str, False),
FormField("uuid", str, False),
FormField("belongTo", str, True),
FormField("title", str, True),
FormField("description", str, True),
FormField("eventDateTimeStart", int, True),
FormField("eventDateTimeEnd", int, True),
FormField("loopRules", str, True),
FormField("timezoneOffset", int, True),
FormField("lastChange", str, False),
),
None,
)
@app.route("/calendar/add", methods=["POST"])
def api_calendar_addHandle():
return SmartDbCaller(
calendar_db.calendar_add,
(
FormField("token", str, False),
FormField("belongTo", str, False),
FormField("title", str, False),
FormField("description", str, False),
FormField("eventDateTimeStart", int, False),
FormField("eventDateTimeEnd", int, False),
FormField("loopRules", str, False),
FormField("timezoneOffset", int, False),
),
None,
)
@app.route("/calendar/delete", methods=["POST"])
def api_calendar_deleteHandle():
return SmartDbCaller(
calendar_db.calendar_delete,
(
FormField("token", str, False),
FormField("uuid", str, False),
FormField("lastChange", str, False),
),
None,
)
# endregion
# region: Collection
@app.route("/collection/getFullOwn", methods=["POST"])
def api_collection_getFullOwnHandle():
return SmartDbCaller(
calendar_db.collection_getFullOwn, (FormField("token", str, False),), None
)
@app.route("/collection/getListOwn", methods=["POST"])
def api_collection_getListOwnHandle():
return SmartDbCaller(
calendar_db.collection_getListOwn, (FormField("token", str, False),), None
)
@app.route("/collection/getDetailOwn", methods=["POST"])
def api_collection_getDetailOwnHandle():
return SmartDbCaller(
calendar_db.collection_getDetailOwn,
(FormField("token", str, False), FormField("uuid", str, False)),
None,
)
@app.route("/collection/addOwn", methods=["POST"])
def api_collection_addOwnHandle():
return SmartDbCaller(
calendar_db.collection_addOwn,
(FormField("token", str, False), FormField("name", str, False)),
None,
)
@app.route("/collection/updateOwn", methods=["POST"])
def api_collection_updateOwnHandle():
return SmartDbCaller(
calendar_db.collection_updateOwn,
(
FormField("token", str, False),
FormField("uuid", str, False),
FormField("name", str, False),
FormField("lastChange", str, False),
),
None,
)
@app.route("/collection/deleteOwn", methods=["POST"])
def api_collection_deleteOwnHandle():
return SmartDbCaller(
calendar_db.collection_deleteOwn,
(
FormField("token", str, False),
FormField("uuid", str, False),
FormField("lastChange", str, False),
),
None,
)
@app.route("/collection/getSharing", methods=["POST"])
def api_collection_getSharingHandle():
return SmartDbCaller(
calendar_db.collection_getSharing,
(FormField("token", str, False), FormField("uuid", str, False)),
None,
)
@app.route("/collection/deleteSharing", methods=["POST"])
def api_collection_deleteSharingHandle():
return SmartDbCaller(
calendar_db.collection_deleteSharing,
(
FormField("token", str, False),
FormField("uuid", str, False),
FormField("target", str, False),
FormField("lastChange", str, False),
),
None,
)
@app.route("/collection/addSharing", methods=["POST"])
def api_collection_addSharingHandle():
return SmartDbCaller(
calendar_db.collection_addSharing,
(
FormField("token", str, False),
FormField("uuid", str, False),
FormField("target", str, False),
FormField("lastChange", str, False),
),
None,
)
@app.route("/collection/getShared", methods=["POST"])
def api_collection_getSharedHandle():
return SmartDbCaller(
calendar_db.collection_getShared, (FormField("token", str, False),), None
)
# endregion
# region: Todo
@app.route("/todo/getFull", methods=["POST"])
def api_todo_getFullHandle():
return SmartDbCaller(
calendar_db.todo_getFull, (FormField("token", str, False),), None
)
@app.route("/todo/getList", methods=["POST"])
def api_todo_getListHandle():
return SmartDbCaller(
calendar_db.todo_getList, (FormField("token", str, False),), None
)
@app.route("/todo/getDetail", methods=["POST"])
def api_todo_getDetailHandle():
return SmartDbCaller(
calendar_db.todo_getDetail,
(FormField("token", str, False), FormField("uuid", str, False)),
None,
)
@app.route("/todo/add", methods=["POST"])
def api_todo_addHandle():
return SmartDbCaller(calendar_db.todo_add, (FormField("token", str, False),), None)
@app.route("/todo/update", methods=["POST"])
def api_todo_updateHandle():
return SmartDbCaller(
calendar_db.todo_update,
(
FormField("token", str, False),
FormField("uuid", str, False),
FormField("data", str, False),
FormField("lastChange", str, False),
),
None,
)
@app.route("/todo/delete", methods=["POST"])
def api_todo_deleteHandle():
return SmartDbCaller(
calendar_db.todo_delete,
(
FormField("token", str, False),
FormField("uuid", str, False),
FormField("lastChange", str, False),
),
None,
)
# endregion
# region: Admin
@app.route("/admin/get", methods=["POST"])
def api_admin_getHandle():
return SmartDbCaller(calendar_db.admin_get, (FormField("token", str, False),), None)
@app.route("/admin/add", methods=["POST"])
def api_admin_addHandle():
return SmartDbCaller(
calendar_db.admin_add,
(FormField("token", str, False), FormField("username", str, False)),
None,
)
@app.route("/admin/update", methods=["POST"])
def api_admin_updateHandle():
return SmartDbCaller(
calendar_db.admin_update,
(
FormField("token", str, False),
FormField("username", str, False),
FormField("password", str, True),
FormField("isAdmin", utils.Str2Bool, True),
),
None,
)
@app.route("/admin/delete", methods=["POST"])
def api_admin_deleteHandle():
return SmartDbCaller(
calendar_db.admin_delete,
(FormField("token", str, False), FormField("username", str, False)),
None,
)
# endregion
# region: Profile
@app.route("/profile/isAdmin", methods=["POST"])
def api_profile_isAdminHandle():
return SmartDbCaller(
calendar_db.profile_isAdmin, (FormField("token", str, False),), None
)
@app.route("/profile/changePassword", methods=["POST"])
def api_profile_changePasswordHandle():
return SmartDbCaller(
calendar_db.profile_changePassword,
(FormField("token", str, False), FormField("password", str, False)),
None,
)
@app.route("/profile/getToken", methods=["POST"])
def api_profile_getTokenHandle():
return SmartDbCaller(
calendar_db.profile_getToken, (FormField("token", str, False),), None
)
@app.route("/profile/deleteToken", methods=["POST"])
def api_profile_deleteTokenHandle():
return SmartDbCaller(
calendar_db.profile_deleteToken,
(FormField("token", str, False), FormField("deleteToken", str, False)),
None,
)
# endregion
# endregion
# region: Utilities
@dataclass(frozen=True)
class ClientNetworkInfo:
user_agent: str
"""The user agent of client."""
ip_addr: str
"""The IP address of client."""
def FetchClientNetworkInfo() -> ClientNetworkInfo:
clientUa = request.user_agent.string
forwardIpList = request.headers.getlist("X-Forwarded-For")
if forwardIpList:
clientIp = forwardIpList[0]
else:
directIp = request.remote_addr
if directIp is not None:
clientIp = directIp
else:
clientIp = "0.0.0.0"
return ClientNetworkInfo(clientUa, clientIp)
@dataclass(frozen=True)
class FormField:
name: str
"""The name of form field."""
ty: Callable[[str], Any]
"""The type of form field."""
is_optional: bool
"""True if this form field is optional, otherwise false."""
def SmartDbCaller(
db_method: Callable[..., ResponseBody[Any]],
fields: tuple[FormField, ...],
padding_form: dict[str, str] | None,
) -> dict[str, Any]:
opt_param_counter = 0
lost_required: bool = False
param_list: list[Any] = []
opt_param_dict: dict[str, Any] = {}
# fetch user passed form
user_form: dict[str, str] = request.form.to_dict()
LOGGER.debug(f"User Form: {user_form}")
# overwrite user form by our padding form
if padding_form is not None:
user_form.update(padding_form)
LOGGER.debug(f"Padded User Form: {user_form}")
# check fields one by one
for field in fields:
value = user_form.get(field.name, None)
if value is not None:
value = field.ty(value)
if field.is_optional:
# optional param
if value is not None:
opt_param_dict[field.name] = value
opt_param_counter += 1
else:
# required param
if value is None:
lost_required = True
else:
param_list.append(value)
# Only execute database function if there is no lost required fields.
# And fulfill one of following requirements:
# 1. There are all required fields (optional parameter count is zero).
# 1. Or, there is some optional parameter.
LOGGER.debug(f"Has Lost Required Parameter: {lost_required}")
LOGGER.debug(f"All Optional Parameter Count: {opt_param_counter}")
LOGGER.debug(f"Available Optional Parameter Count: {len(opt_param_dict)}")
result: ResponseBody[Any]
if lost_required == False and (opt_param_counter == 0 or len(opt_param_dict) != 0):
result = db_method(*param_list, **opt_param_dict)
else:
result = ResponseBody(False, "Invalid parameter", None)
return ConstructResponseBody(result)
def ConstructResponseBody(body: ResponseBody[Any]) -> dict[str, Any]:
return {"success": body.success, "error": body.error, "data": body.data}
# endregion
def run():
calendar_db.open()
app.run(port=config.get_config().web.port)
calendar_db.close()
backend-legacy/sql/mysql.sql
View File
backend-legacy/sql/sqlite.sql
-67
View File
@@ -1,67 +0,0 @@
CREATE TABLE user(
[name] TEXT NOT NULL,
[password] TEXT NOT NULL,
[is_admin] TINYINT NOT NULL CHECK(is_admin = 1 OR is_admin = 0),
[salt] INTEGER NOT NULL,
PRIMARY KEY (name)
);
CREATE TABLE token(
[user] TEXT NOT NULL,
[token] TEXT UNIQUE NOT NULL,
[token_expire_on] BIGINT NOT NULL,
[ua] TEXT NOT NULL,
[ip] TEXT NOT NULL,
FOREIGN KEY (user) REFERENCES user(name) ON DELETE CASCADE
);
CREATE TABLE collection(
[uuid] TEXT NOT NULL,
[name] TEXT NOT NULL,
[user] TEXT NOT NULL,
[last_change] TEXT NOT NULL,
PRIMARY KEY (uuid),
FOREIGN KEY (user) REFERENCES user(name) ON DELETE CASCADE
);
CREATE TABLE share(
[uuid] TEXT NOT NULL,
[target] TEXT NOT NULL,
FOREIGN KEY (uuid) REFERENCES collection(uuid) ON DELETE CASCADE
FOREIGN KEY (target) REFERENCES user(name) ON DELETE CASCADE
);
CREATE TABLE calendar(
[uuid] TEXT NOT NULL,
[belong_to] TEXT NOT NULL,
[title] TEXT NOT NULL,
[description] TEXT NOT NULL,
[last_change] TEXT NOT NULL,
[event_date_time_start] BIGINT NOT NULL,
[event_date_time_end] BIGINT NOT NULL,
[timezone_offset] INT NOT NULL,
[loop_rules] TEXT NOT NULL,
[loop_date_time_start] BIGINT NOT NULL,
[loop_date_time_end] BIGINT NOT NULL,
PRIMARY KEY (uuid),
FOREIGN KEY (belong_to) REFERENCES collection(uuid) ON DELETE CASCADE
);
CREATE TABLE todo(
[uuid] TEXT NOT NULL,
[belong_to] TEXT NOT NULL,
[data] TEXT NOT NULL,
[last_change] TEXT NOT NULL,
PRIMARY KEY (uuid),
FOREIGN KEY (belong_to) REFERENCES user(name) ON DELETE CASCADE
);
backend-legacy/utils.py
-64
View File
@@ -1,64 +0,0 @@
import hashlib
import random
import uuid
import time
import math
import re
USERNAME_PATTERN: re.Pattern = re.compile("^[0-9A-Za-z]+$")
PASSWORD_PATTERN: re.Pattern = re.compile("^[!-~]+$")
def IsValidUsername(strl: str) -> bool:
return USERNAME_PATTERN.match(strl) is not None
def IsValidPassword(strl: str) -> bool:
return PASSWORD_PATTERN.match(strl) is not None
def ComputePasswordHash(password: str) -> str:
s = hashlib.sha256()
s.update(password.encode("utf-8"))
return s.hexdigest()
def GenerateUUID() -> str:
return str(uuid.uuid1())
def GenerateToken(username: str) -> str:
s = hashlib.sha256()
s.update(username.encode("utf-8"))
s.update(GenerateUUID().encode("utf-8"))
return s.hexdigest()
def GenerateSalt() -> int:
return random.randint(0, 6172748)
def ComputePasswordHashWithSalt(passwordHashed: str, salt: int) -> str:
s = hashlib.sha256()
s.update((passwordHashed + str(salt)).encode("utf-8"))
return s.hexdigest()
def GetCurrentTimestamp() -> int:
return int(time.time())
def GetTokenExpireOn() -> int:
return GetCurrentTimestamp() + 60 * 60 * 24 * 2 # add 2 day from now
def Str2Bool(strl: str) -> bool:
return strl.lower() == "true"
def GCD(a: int, b: int) -> int:
return math.gcd(a, b)
def LCM(a: int, b: int) -> int:
return (a * b) // GCD(a, b)
backend-legacy/uv.lock
Generated
-117
View File
@@ -1,117 +0,0 @@
version = 1
revision = 2
requires-python = ">=3.11"
[manifest]
constraints = [
{ name = "markupsafe", specifier = "==2.1.5" },
{ name = "werkzeug", specifier = "==2.2.2" },
]
[[package]]
name = "click"
version = "8.3.3"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "colorama", marker = "sys_platform == 'win32'" },
]
sdist = { url = "https://files.pythonhosted.org/packages/bb/63/f9e1ea081ce35720d8b92acde70daaedace594dc93b693c869e0d5910718/click-8.3.3.tar.gz", hash = "sha256:398329ad4837b2ff7cbe1dd166a4c0f8900c3ca3a218de04466f38f6497f18a2", size = 328061, upload-time = "2026-04-22T15:11:27.506Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/ae/44/c1221527f6a71a01ec6fbad7fa78f1d50dfa02217385cf0fa3eec7087d59/click-8.3.3-py3-none-any.whl", hash = "sha256:a2bf429bb3033c89fa4936ffb35d5cb471e3719e1f3c8a7c3fff0b8314305613", size = 110502, upload-time = "2026-04-22T15:11:25.044Z" },
]
[[package]]
name = "coleaf-backend"
version = "1.1.0"
source = { virtual = "." }
dependencies = [
{ name = "flask" },
]
[package.metadata]
requires-dist = [{ name = "flask", specifier = "==2.2.3" }]
[[package]]
name = "colorama"
version = "0.4.6"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/d8/53/6f443c9a4a8358a93a6792e2acffb9d9d5cb0a5cfd8802644b7b1c9a02e4/colorama-0.4.6.tar.gz", hash = "sha256:08695f5cb7ed6e0531a20572697297273c47b8cae5a63ffc6d6ed5c201be6e44", size = 27697, upload-time = "2022-10-25T02:36:22.414Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/d1/d6/3965ed04c63042e047cb6a3e6ed1a63a35087b6a609aa3a15ed8ac56c221/colorama-0.4.6-py2.py3-none-any.whl", hash = "sha256:4f1d9991f5acc0ca119f9d443620b77f9d6b33703e51011c16baf57afb285fc6", size = 25335, upload-time = "2022-10-25T02:36:20.889Z" },
]
[[package]]
name = "flask"
version = "2.2.3"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "click" },
{ name = "itsdangerous" },
{ name = "jinja2" },
{ name = "werkzeug" },
]
sdist = { url = "https://files.pythonhosted.org/packages/e8/5c/ff9047989bd995b1098d14b03013f160225db2282925b517bb4a967752ee/Flask-2.2.3.tar.gz", hash = "sha256:7eb373984bf1c770023fce9db164ed0c3353cd0b53f130f4693da0ca756a2e6d", size = 697599, upload-time = "2023-02-15T22:43:57.265Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/95/9c/a3542594ce4973786236a1b7b702b8ca81dbf40ea270f0f96284f0c27348/Flask-2.2.3-py3-none-any.whl", hash = "sha256:c0bec9477df1cb867e5a67c9e1ab758de9cb4a3e52dd70681f59fa40a62b3f2d", size = 101839, upload-time = "2023-02-15T22:43:55.501Z" },
]
[[package]]
name = "itsdangerous"
version = "2.2.0"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/9c/cb/8ac0172223afbccb63986cc25049b154ecfb5e85932587206f42317be31d/itsdangerous-2.2.0.tar.gz", hash = "sha256:e0050c0b7da1eea53ffaf149c0cfbb5c6e2e2b69c4bef22c81fa6eb73e5f6173", size = 54410, upload-time = "2024-04-16T21:28:15.614Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/04/96/92447566d16df59b2a776c0fb82dbc4d9e07cd95062562af01e408583fc4/itsdangerous-2.2.0-py3-none-any.whl", hash = "sha256:c6242fc49e35958c8b15141343aa660db5fc54d4f13a1db01a3f5891b98700ef", size = 16234, upload-time = "2024-04-16T21:28:14.499Z" },
]
[[package]]
name = "jinja2"
version = "3.1.6"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "markupsafe" },
]
sdist = { url = "https://files.pythonhosted.org/packages/df/bf/f7da0350254c0ed7c72f3e33cef02e048281fec7ecec5f032d4aac52226b/jinja2-3.1.6.tar.gz", hash = "sha256:0137fb05990d35f1275a587e9aee6d56da821fc83491a0fb838183be43f66d6d", size = 245115, upload-time = "2025-03-05T20:05:02.478Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/62/a1/3d680cbfd5f4b8f15abc1d571870c5fc3e594bb582bc3b64ea099db13e56/jinja2-3.1.6-py3-none-any.whl", hash = "sha256:85ece4451f492d0c13c5dd7c13a64681a86afae63a5f347908daf103ce6d2f67", size = 134899, upload-time = "2025-03-05T20:05:00.369Z" },
]
[[package]]
name = "markupsafe"
version = "2.1.5"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/87/5b/aae44c6655f3801e81aa3eef09dbbf012431987ba564d7231722f68df02d/MarkupSafe-2.1.5.tar.gz", hash = "sha256:d283d37a890ba4c1ae73ffadf8046435c76e7bc2247bbb63c00bd1a709c6544b", size = 19384, upload-time = "2024-02-02T16:31:22.863Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/11/e7/291e55127bb2ae67c64d66cef01432b5933859dfb7d6949daa721b89d0b3/MarkupSafe-2.1.5-cp311-cp311-macosx_10_9_universal2.whl", hash = "sha256:629ddd2ca402ae6dbedfceeba9c46d5f7b2a61d9749597d4307f943ef198fc1f", size = 18219, upload-time = "2024-02-02T16:30:19.988Z" },
{ url = "https://files.pythonhosted.org/packages/6b/cb/aed7a284c00dfa7c0682d14df85ad4955a350a21d2e3b06d8240497359bf/MarkupSafe-2.1.5-cp311-cp311-macosx_10_9_x86_64.whl", hash = "sha256:5b7b716f97b52c5a14bffdf688f971b2d5ef4029127f1ad7a513973cfd818df2", size = 14098, upload-time = "2024-02-02T16:30:21.063Z" },
{ url = "https://files.pythonhosted.org/packages/1c/cf/35fe557e53709e93feb65575c93927942087e9b97213eabc3fe9d5b25a55/MarkupSafe-2.1.5-cp311-cp311-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:6ec585f69cec0aa07d945b20805be741395e28ac1627333b1c5b0105962ffced", size = 29014, upload-time = "2024-02-02T16:30:22.926Z" },
{ url = "https://files.pythonhosted.org/packages/97/18/c30da5e7a0e7f4603abfc6780574131221d9148f323752c2755d48abad30/MarkupSafe-2.1.5-cp311-cp311-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:b91c037585eba9095565a3556f611e3cbfaa42ca1e865f7b8015fe5c7336d5a5", size = 28220, upload-time = "2024-02-02T16:30:24.76Z" },
{ url = "https://files.pythonhosted.org/packages/0c/40/2e73e7d532d030b1e41180807a80d564eda53babaf04d65e15c1cf897e40/MarkupSafe-2.1.5-cp311-cp311-manylinux_2_5_i686.manylinux1_i686.manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:7502934a33b54030eaf1194c21c692a534196063db72176b0c4028e140f8f32c", size = 27756, upload-time = "2024-02-02T16:30:25.877Z" },
{ url = "https://files.pythonhosted.org/packages/18/46/5dca760547e8c59c5311b332f70605d24c99d1303dd9a6e1fc3ed0d73561/MarkupSafe-2.1.5-cp311-cp311-musllinux_1_1_aarch64.whl", hash = "sha256:0e397ac966fdf721b2c528cf028494e86172b4feba51d65f81ffd65c63798f3f", size = 33988, upload-time = "2024-02-02T16:30:26.935Z" },
{ url = "https://files.pythonhosted.org/packages/6d/c5/27febe918ac36397919cd4a67d5579cbbfa8da027fa1238af6285bb368ea/MarkupSafe-2.1.5-cp311-cp311-musllinux_1_1_i686.whl", hash = "sha256:c061bb86a71b42465156a3ee7bd58c8c2ceacdbeb95d05a99893e08b8467359a", size = 32718, upload-time = "2024-02-02T16:30:28.111Z" },
{ url = "https://files.pythonhosted.org/packages/f8/81/56e567126a2c2bc2684d6391332e357589a96a76cb9f8e5052d85cb0ead8/MarkupSafe-2.1.5-cp311-cp311-musllinux_1_1_x86_64.whl", hash = "sha256:3a57fdd7ce31c7ff06cdfbf31dafa96cc533c21e443d57f5b1ecc6cdc668ec7f", size = 33317, upload-time = "2024-02-02T16:30:29.214Z" },
{ url = "https://files.pythonhosted.org/packages/00/0b/23f4b2470accb53285c613a3ab9ec19dc944eaf53592cb6d9e2af8aa24cc/MarkupSafe-2.1.5-cp311-cp311-win32.whl", hash = "sha256:397081c1a0bfb5124355710fe79478cdbeb39626492b15d399526ae53422b906", size = 16670, upload-time = "2024-02-02T16:30:30.915Z" },
{ url = "https://files.pythonhosted.org/packages/b7/a2/c78a06a9ec6d04b3445a949615c4c7ed86a0b2eb68e44e7541b9d57067cc/MarkupSafe-2.1.5-cp311-cp311-win_amd64.whl", hash = "sha256:2b7c57a4dfc4f16f7142221afe5ba4e093e09e728ca65c51f5620c9aaeb9a617", size = 17224, upload-time = "2024-02-02T16:30:32.09Z" },
{ url = "https://files.pythonhosted.org/packages/53/bd/583bf3e4c8d6a321938c13f49d44024dbe5ed63e0a7ba127e454a66da974/MarkupSafe-2.1.5-cp312-cp312-macosx_10_9_universal2.whl", hash = "sha256:8dec4936e9c3100156f8a2dc89c4b88d5c435175ff03413b443469c7c8c5f4d1", size = 18215, upload-time = "2024-02-02T16:30:33.081Z" },
{ url = "https://files.pythonhosted.org/packages/48/d6/e7cd795fc710292c3af3a06d80868ce4b02bfbbf370b7cee11d282815a2a/MarkupSafe-2.1.5-cp312-cp312-macosx_10_9_x86_64.whl", hash = "sha256:3c6b973f22eb18a789b1460b4b91bf04ae3f0c4234a0a6aa6b0a92f6f7b951d4", size = 14069, upload-time = "2024-02-02T16:30:34.148Z" },
{ url = "https://files.pythonhosted.org/packages/51/b5/5d8ec796e2a08fc814a2c7d2584b55f889a55cf17dd1a90f2beb70744e5c/MarkupSafe-2.1.5-cp312-cp312-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:ac07bad82163452a6884fe8fa0963fb98c2346ba78d779ec06bd7a6262132aee", size = 29452, upload-time = "2024-02-02T16:30:35.149Z" },
{ url = "https://files.pythonhosted.org/packages/0a/0d/2454f072fae3b5a137c119abf15465d1771319dfe9e4acbb31722a0fff91/MarkupSafe-2.1.5-cp312-cp312-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:f5dfb42c4604dddc8e4305050aa6deb084540643ed5804d7455b5df8fe16f5e5", size = 28462, upload-time = "2024-02-02T16:30:36.166Z" },
{ url = "https://files.pythonhosted.org/packages/2d/75/fd6cb2e68780f72d47e6671840ca517bda5ef663d30ada7616b0462ad1e3/MarkupSafe-2.1.5-cp312-cp312-manylinux_2_5_i686.manylinux1_i686.manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:ea3d8a3d18833cf4304cd2fc9cbb1efe188ca9b5efef2bdac7adc20594a0e46b", size = 27869, upload-time = "2024-02-02T16:30:37.834Z" },
{ url = "https://files.pythonhosted.org/packages/b0/81/147c477391c2750e8fc7705829f7351cf1cd3be64406edcf900dc633feb2/MarkupSafe-2.1.5-cp312-cp312-musllinux_1_1_aarch64.whl", hash = "sha256:d050b3361367a06d752db6ead6e7edeb0009be66bc3bae0ee9d97fb326badc2a", size = 33906, upload-time = "2024-02-02T16:30:39.366Z" },
{ url = "https://files.pythonhosted.org/packages/8b/ff/9a52b71839d7a256b563e85d11050e307121000dcebc97df120176b3ad93/MarkupSafe-2.1.5-cp312-cp312-musllinux_1_1_i686.whl", hash = "sha256:bec0a414d016ac1a18862a519e54b2fd0fc8bbfd6890376898a6c0891dd82e9f", size = 32296, upload-time = "2024-02-02T16:30:40.413Z" },
{ url = "https://files.pythonhosted.org/packages/88/07/2dc76aa51b481eb96a4c3198894f38b480490e834479611a4053fbf08623/MarkupSafe-2.1.5-cp312-cp312-musllinux_1_1_x86_64.whl", hash = "sha256:58c98fee265677f63a4385256a6d7683ab1832f3ddd1e66fe948d5880c21a169", size = 33038, upload-time = "2024-02-02T16:30:42.243Z" },
{ url = "https://files.pythonhosted.org/packages/96/0c/620c1fb3661858c0e37eb3cbffd8c6f732a67cd97296f725789679801b31/MarkupSafe-2.1.5-cp312-cp312-win32.whl", hash = "sha256:8590b4ae07a35970728874632fed7bd57b26b0102df2d2b233b6d9d82f6c62ad", size = 16572, upload-time = "2024-02-02T16:30:43.326Z" },
{ url = "https://files.pythonhosted.org/packages/3f/14/c3554d512d5f9100a95e737502f4a2323a1959f6d0d01e0d0997b35f7b10/MarkupSafe-2.1.5-cp312-cp312-win_amd64.whl", hash = "sha256:823b65d8706e32ad2df51ed89496147a42a2a6e01c13cfb6ffb8b1e92bc910bb", size = 17127, upload-time = "2024-02-02T16:30:44.418Z" },
]
[[package]]
name = "werkzeug"
version = "2.2.2"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "markupsafe" },
]
sdist = { url = "https://files.pythonhosted.org/packages/f8/c1/1c8e539f040acd80f844c69a5ef8e2fccdf8b442dabb969e497b55d544e1/Werkzeug-2.2.2.tar.gz", hash = "sha256:7ea2d48322cc7c0f8b3a215ed73eabd7b5d75d0b50e31ab006286ccff9e00b8f", size = 844378, upload-time = "2022-08-08T21:44:15.376Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/c8/27/be6ddbcf60115305205de79c29004a0c6bc53cec814f733467b1bb89386d/Werkzeug-2.2.2-py3-none-any.whl", hash = "sha256:f979ab81f58d7318e064e99c4506445d60135ac5cd2e177a2de0089bfd4c9bd5", size = 232700, upload-time = "2022-08-08T21:44:13.251Z" },
]