95 lines
3.2 KiB
Python
95 lines
3.2 KiB
Python
from app.common.errors.app_error import NotFoundAppError
|
|
from app.common.utils.id_gen import new_invite_code
|
|
from app.models.entities import RedeemCode, SystemConfig
|
|
from app.modules.system.repository import SystemRepository
|
|
|
|
|
|
class SystemService:
|
|
def __init__(self, db) -> None:
|
|
self.db = db
|
|
self.repository = SystemRepository(db)
|
|
|
|
def list_configs(self) -> list[dict]:
|
|
return [
|
|
{
|
|
"configKey": item.config_key,
|
|
"configValue": item.config_value,
|
|
"valueType": item.value_type,
|
|
"groupName": item.group_name,
|
|
"description": item.description,
|
|
"isPublic": item.is_public,
|
|
}
|
|
for item in self.repository.list_configs().all()
|
|
]
|
|
|
|
def upsert_config(self, payload) -> dict:
|
|
item = self.repository.get_config(payload.config_key)
|
|
if not item:
|
|
item = SystemConfig(**payload.model_dump())
|
|
self.db.add(item)
|
|
else:
|
|
for key, value in payload.model_dump().items():
|
|
setattr(item, key, value)
|
|
self.db.commit()
|
|
return {
|
|
"configKey": item.config_key,
|
|
"configValue": item.config_value,
|
|
"groupName": item.group_name,
|
|
}
|
|
|
|
def list_redeem_codes(self) -> list[dict]:
|
|
return [
|
|
{
|
|
"id": item.id,
|
|
"batchNo": item.batch_no,
|
|
"redeemCode": item.redeem_code,
|
|
"points": item.points,
|
|
"status": item.status,
|
|
"usedByUserId": item.used_by_user_id,
|
|
"usedAt": item.used_at.isoformat() if item.used_at else None,
|
|
}
|
|
for item in self.repository.list_redeem_codes().all()
|
|
]
|
|
|
|
def batch_create_redeem_codes(self, payload) -> list[dict]:
|
|
created = []
|
|
for _ in range(payload.quantity):
|
|
item = RedeemCode(
|
|
batch_no=payload.batch_no,
|
|
redeem_code=f"{payload.batch_no}-{new_invite_code(4)}-{new_invite_code(4)}",
|
|
points=payload.points,
|
|
status="unused",
|
|
remark=payload.remark,
|
|
)
|
|
self.db.add(item)
|
|
created.append(item)
|
|
self.db.commit()
|
|
return self.list_redeem_codes()[: payload.quantity]
|
|
|
|
def disable_redeem_code(self, redeem_id: int) -> dict:
|
|
item = self.repository.get_redeem_code(redeem_id)
|
|
if not item:
|
|
raise NotFoundAppError("redeem code not found", code=70020)
|
|
item.status = "disabled"
|
|
self.db.commit()
|
|
return {
|
|
"id": item.id,
|
|
"status": item.status,
|
|
}
|
|
|
|
def list_callback_logs(self) -> list[dict]:
|
|
return [
|
|
{
|
|
"id": item.id,
|
|
"sourceType": item.source_type,
|
|
"sourceCode": item.source_code,
|
|
"relatedNo": item.related_no,
|
|
"verifyStatus": item.verify_status,
|
|
"processStatus": item.process_status,
|
|
"errorMessage": item.error_message,
|
|
"createdAt": item.created_at.isoformat(),
|
|
}
|
|
for item in self.repository.list_callback_logs().all()
|
|
]
|
|
|