116 lines
3.8 KiB
Python
116 lines
3.8 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
缓存模块
|
||
"""
|
||
|
||
# 列举导入模块
|
||
import json
|
||
from pathlib import Path
|
||
from pathlib import WindowsPath
|
||
import sys
|
||
import time
|
||
from typing import Any, Dict, Optional
|
||
|
||
sys.path.append(Path(__file__).parent.parent.as_posix())
|
||
from utils.sqlite import SQLite
|
||
|
||
|
||
class Caches(SQLite):
|
||
"""
|
||
缓存客户端，支持：
|
||
query：查询并返回单条缓存
|
||
update：新增或更新单条缓存
|
||
"""
|
||
|
||
# 自定义JSON编码器
|
||
class JSONEncoder(json.JSONEncoder):
|
||
def default(self, o):
|
||
# 若为WindowsPath对象则转为字符串路径
|
||
if isinstance(o, WindowsPath):
|
||
return o.as_posix()
|
||
return super().default(o)
|
||
|
||
def __init__(self, ttl: int = 30 * 86400):
|
||
"""
|
||
初始化
|
||
:param ttl: 缓存生存时间，单位为秒，默认为30天
|
||
"""
|
||
# 初始化
|
||
super().__init__(database=Path(__file__).parent.resolve() / "caches.db")
|
||
self.ttl = ttl
|
||
|
||
# 初始化缓存表（不清理过期缓存）
|
||
try:
|
||
with self:
|
||
self.execute(
|
||
sql="""
|
||
CREATE TABLE IF NOT EXISTS caches
|
||
(
|
||
--草稿名称
|
||
draft_name TEXT PRIMARY KEY,
|
||
--草稿生成方案
|
||
draft_generate_scheme TEXT NOT NULL,
|
||
--创建时间戳
|
||
timestamp REAL NOT NULL
|
||
)
|
||
"""
|
||
)
|
||
self.execute(
|
||
sql="""
|
||
CREATE INDEX IF NOT EXISTS idx_timestamp ON caches(timestamp)
|
||
"""
|
||
)
|
||
except Exception as exception:
|
||
raise RuntimeError(f"初始化缓存发生异常：{str(exception)}") from exception
|
||
|
||
def has_same_draft_name(self, draft_name: str) -> bool:
|
||
"""
|
||
是否存在相同草稿名称的缓存
|
||
:param draft_name: 草稿名称
|
||
:return: 是否存在相同草稿名称的缓存
|
||
"""
|
||
try:
|
||
with self:
|
||
result = self.query_one(
|
||
sql="""
|
||
SELECT draft_generate_scheme
|
||
FROM caches
|
||
WHERE draft_name = ? AND timestamp >= ?
|
||
""",
|
||
parameters=(draft_name, time.time() - self.ttl),
|
||
)
|
||
return False if result is None else True
|
||
except Exception as exception:
|
||
raise RuntimeError(
|
||
f"是否存在相同草稿方案的缓存发生异常：{str(exception)}"
|
||
) from exception
|
||
|
||
def update(
|
||
self, draft_name: str, draft_generate_scheme: list[dict]
|
||
) -> Optional[bool]:
|
||
"""
|
||
新增或更新草稿生成方案
|
||
:param draft_name: 草稿名称
|
||
:param draft_generate_scheme: 草稿生成方案
|
||
:return: 成功返回True，失败返回False
|
||
"""
|
||
try:
|
||
with self:
|
||
return self.execute(
|
||
sql="""
|
||
INSERT OR REPLACE INTO caches (draft_name, draft_generate_scheme, timestamp) VALUES (?, ?, ?)
|
||
""",
|
||
parameters=(
|
||
draft_name,
|
||
json.dumps(
|
||
obj=draft_generate_scheme,
|
||
cls=self.JSONEncoder,
|
||
sort_keys=True,
|
||
ensure_ascii=False,
|
||
),
|
||
time.time(),
|
||
),
|
||
)
|
||
except Exception as exception:
|
||
raise RuntimeError("新增或更新草稿生成方案发生异常") from exception
|