399 lines
11 KiB
Python
399 lines
11 KiB
Python
import time
|
||
import datetime
|
||
from typing import Union, Optional
|
||
|
||
import sqlite3
|
||
|
||
from file import ObjectFileManager, LocalFileManager
|
||
|
||
|
||
class DatabaseManager:
|
||
"""数据库管理类"""
|
||
|
||
def __init__(self, db_path, online_db_uri=None):
|
||
self.db_path = db_path
|
||
self.online_db_uri = online_db_uri
|
||
self.obj = ObjectFileManager()
|
||
self.lfm = LocalFileManager("db")
|
||
|
||
def upload_db(self):
|
||
file = self.lfm.get_file("data")
|
||
self.obj.upload_file(self.online_db_uri, file)
|
||
print("DB upload success")
|
||
|
||
def download_db(self):
|
||
file = self.obj.get_file(self.online_db_uri)
|
||
self.lfm.upload_file("data", file)
|
||
|
||
def get_connection(self):
|
||
"""获取数据库连接"""
|
||
conn = sqlite3.connect(self.db_path)
|
||
conn.row_factory = sqlite3.Row
|
||
return conn
|
||
|
||
def execute_event(self, query, params=None):
|
||
conn = self.get_connection()
|
||
cursor = conn.cursor()
|
||
|
||
try:
|
||
if params:
|
||
cursor.executemany(query, params)
|
||
else:
|
||
cursor.execute(query)
|
||
conn.commit()
|
||
conn.close()
|
||
return True
|
||
except Exception as e:
|
||
print(e)
|
||
conn.close()
|
||
return False
|
||
|
||
def execute_query(self, query, params=None, fetch_all=True):
|
||
"""执行查询并返回结果"""
|
||
conn = self.get_connection()
|
||
cursor = conn.cursor()
|
||
|
||
try:
|
||
if params:
|
||
cursor.execute(query, params)
|
||
else:
|
||
cursor.execute(query)
|
||
|
||
if fetch_all:
|
||
result = cursor.fetchall()
|
||
else:
|
||
result = cursor.fetchone()
|
||
|
||
conn.close()
|
||
return result
|
||
except Exception as e:
|
||
print(e)
|
||
conn.close()
|
||
raise e
|
||
|
||
|
||
class TimeUtils:
|
||
"""
|
||
时间工具类
|
||
支持时间戳、时间字符串等多种格式的时间处理，以及相互转换和时间计算
|
||
"""
|
||
|
||
# 常用时间格式
|
||
FORMAT_DATETIME = "%Y-%m-%d %H:%M:%S"
|
||
FORMAT_DATE = "%Y-%m-%d"
|
||
FORMAT_TIME = "%H:%M:%S"
|
||
FORMAT_DATETIME_COMPACT = "%Y%m%d%H%M%S"
|
||
FORMAT_DATE_COMPACT = "%Y%m%d"
|
||
|
||
def __init__(self, time_input: Optional[Union[str, int, float, datetime.datetime]] = None):
|
||
"""
|
||
初始化时间工具
|
||
|
||
Args:
|
||
time_input: 时间输入，可以是：
|
||
- 时间戳（整数或浮点数）
|
||
- 时间字符串（支持多种格式）
|
||
- datetime对象
|
||
- None（使用当前时间）
|
||
"""
|
||
if time_input is None:
|
||
self.datetime_obj = datetime.datetime.now()
|
||
elif isinstance(time_input, (int, float)):
|
||
self.datetime_obj = datetime.datetime.fromtimestamp(time_input)
|
||
elif isinstance(time_input, str):
|
||
self.datetime_obj = self._parse_string(time_input)
|
||
elif isinstance(time_input, datetime.datetime):
|
||
self.datetime_obj = time_input
|
||
else:
|
||
raise ValueError(f"不支持的时间输入类型: {type(time_input)}")
|
||
|
||
def _parse_string(self, time_str: str) -> datetime.datetime:
|
||
"""
|
||
解析时间字符串，支持多种格式
|
||
|
||
Args:
|
||
time_str: 时间字符串
|
||
|
||
Returns:
|
||
datetime对象
|
||
"""
|
||
# 尝试多种格式进行解析
|
||
formats = [
|
||
self.FORMAT_DATETIME, # 2023-01-01 12:00:00
|
||
self.FORMAT_DATE, # 2023-01-01
|
||
self.FORMAT_DATETIME_COMPACT, # 20230101120000
|
||
self.FORMAT_DATE_COMPACT, # 20230101
|
||
"%Y/%m/%d %H:%M:%S", # 2023/01/01 12:00:00
|
||
"%Y/%m/%d", # 2023/01/01
|
||
"%Y年%m月%d日 %H时%M分%S秒", # 2023年01月01日 12时00分00秒
|
||
"%Y年%m月%d日", # 2023年01月01日
|
||
]
|
||
|
||
for fmt in formats:
|
||
try:
|
||
return datetime.datetime.strptime(time_str, fmt)
|
||
except ValueError:
|
||
continue
|
||
|
||
# 如果所有格式都失败，尝试自动解析
|
||
try:
|
||
return datetime.datetime.fromisoformat(time_str.replace('Z', '+00:00'))
|
||
except ValueError:
|
||
raise ValueError(f"无法解析时间字符串: {time_str}")
|
||
|
||
def to_timestamp(self) -> float:
|
||
"""
|
||
转换为时间戳
|
||
|
||
Returns:
|
||
时间戳（浮点数）
|
||
"""
|
||
return self.datetime_obj.timestamp()
|
||
|
||
def to_string(self, fmt: str = FORMAT_DATETIME) -> str:
|
||
"""
|
||
转换为字符串格式
|
||
|
||
Args:
|
||
fmt: 时间格式字符串
|
||
|
||
Returns:
|
||
格式化后的时间字符串
|
||
"""
|
||
return self.datetime_obj.strftime(fmt)
|
||
|
||
def to_datetime(self) -> datetime.datetime:
|
||
"""
|
||
转换为datetime对象
|
||
|
||
Returns:
|
||
datetime对象
|
||
"""
|
||
return self.datetime_obj
|
||
|
||
def add_days(self, days: int) -> 'TimeUtils':
|
||
"""
|
||
添加天数
|
||
|
||
Args:
|
||
days: 天数（可以为负数）
|
||
|
||
Returns:
|
||
新的TimeUtils对象
|
||
"""
|
||
new_datetime = self.datetime_obj + datetime.timedelta(days=days)
|
||
return TimeUtils(new_datetime)
|
||
|
||
def add_hours(self, hours: int) -> 'TimeUtils':
|
||
"""
|
||
添加小时数
|
||
|
||
Args:
|
||
hours: 小时数（可以为负数）
|
||
|
||
Returns:
|
||
新的TimeUtils对象
|
||
"""
|
||
new_datetime = self.datetime_obj + datetime.timedelta(hours=hours)
|
||
return TimeUtils(new_datetime)
|
||
|
||
def add_minutes(self, minutes: int) -> 'TimeUtils':
|
||
"""
|
||
添加分钟数
|
||
|
||
Args:
|
||
minutes: 分钟数（可以为负数）
|
||
|
||
Returns:
|
||
新的TimeUtils对象
|
||
"""
|
||
new_datetime = self.datetime_obj + datetime.timedelta(minutes=minutes)
|
||
return TimeUtils(new_datetime)
|
||
|
||
def add_seconds(self, seconds: int) -> 'TimeUtils':
|
||
"""
|
||
添加秒数
|
||
|
||
Args:
|
||
seconds: 秒数（可以为负数）
|
||
|
||
Returns:
|
||
新的TimeUtils对象
|
||
"""
|
||
new_datetime = self.datetime_obj + datetime.timedelta(seconds=seconds)
|
||
return TimeUtils(new_datetime)
|
||
|
||
def days_ago(self, days: int) -> 'TimeUtils':
|
||
"""
|
||
计算n天前的日期时间
|
||
|
||
Args:
|
||
days: 天数
|
||
|
||
Returns:
|
||
新的TimeUtils对象
|
||
"""
|
||
return self.add_days(-days)
|
||
|
||
def days_after(self, days: int) -> 'TimeUtils':
|
||
"""
|
||
计算n天后的日期时间
|
||
|
||
Args:
|
||
days: 天数
|
||
|
||
Returns:
|
||
新的TimeUtils对象
|
||
"""
|
||
return self.add_days(days)
|
||
|
||
def start_of_day(self) -> 'TimeUtils':
|
||
"""
|
||
获取当天的开始时间（00:00:00）
|
||
|
||
Returns:
|
||
新的TimeUtils对象
|
||
"""
|
||
new_datetime = self.datetime_obj.replace(hour=0, minute=0, second=0, microsecond=0)
|
||
return TimeUtils(new_datetime)
|
||
|
||
def end_of_day(self) -> 'TimeUtils':
|
||
"""
|
||
获取当天的结束时间（23:59:59）
|
||
|
||
Returns:
|
||
新的TimeUtils对象
|
||
"""
|
||
new_datetime = self.datetime_obj.replace(hour=23, minute=59, second=59, microsecond=999999)
|
||
return TimeUtils(new_datetime)
|
||
|
||
def diff_seconds(self, other: 'TimeUtils') -> float:
|
||
"""
|
||
计算两个时间的秒数差
|
||
|
||
Args:
|
||
other: 另一个TimeUtils对象
|
||
|
||
Returns:
|
||
秒数差（self - other）
|
||
"""
|
||
return (self.datetime_obj - other.datetime_obj).total_seconds()
|
||
|
||
def diff_days(self, other: 'TimeUtils') -> float:
|
||
"""
|
||
计算两个时间的天数差
|
||
|
||
Args:
|
||
other: 另一个TimeUtils对象
|
||
|
||
Returns:
|
||
天数差（self - other）
|
||
"""
|
||
return (self.datetime_obj - other.datetime_obj).days
|
||
|
||
def is_after(self, other: 'TimeUtils') -> bool:
|
||
"""
|
||
判断是否在另一个时间之后
|
||
|
||
Args:
|
||
other: 另一个TimeUtils对象
|
||
|
||
Returns:
|
||
bool值
|
||
"""
|
||
return self.datetime_obj > other.datetime_obj
|
||
|
||
def is_before(self, other: 'TimeUtils') -> bool:
|
||
"""
|
||
判断是否在另一个时间之前
|
||
|
||
Args:
|
||
other: 另一个TimeUtils对象
|
||
|
||
Returns:
|
||
bool值
|
||
"""
|
||
return self.datetime_obj < other.datetime_obj
|
||
|
||
def is_same_day(self, other: 'TimeUtils') -> bool:
|
||
"""
|
||
判断是否在同一天
|
||
|
||
Args:
|
||
other: 另一个TimeUtils对象
|
||
|
||
Returns:
|
||
bool值
|
||
"""
|
||
return (self.datetime_obj.year == other.datetime_obj.year and
|
||
self.datetime_obj.month == other.datetime_obj.month and
|
||
self.datetime_obj.day == other.datetime_obj.day)
|
||
|
||
def format_custom(self, fmt: str) -> str:
|
||
"""
|
||
自定义格式化输出
|
||
|
||
Args:
|
||
fmt: 格式化字符串
|
||
|
||
Returns:
|
||
格式化后的字符串
|
||
"""
|
||
return self.datetime_obj.strftime(fmt)
|
||
|
||
def __str__(self) -> str:
|
||
"""字符串表示"""
|
||
return self.to_string()
|
||
|
||
def __repr__(self) -> str:
|
||
"""repr表示"""
|
||
return f"TimeUtils('{self.to_string()}')"
|
||
|
||
|
||
# 工具函数
|
||
def now() -> TimeUtils:
|
||
"""获取当前时间的TimeUtils对象"""
|
||
return TimeUtils()
|
||
|
||
|
||
def from_timestamp(timestamp: Union[int, float]) -> TimeUtils:
|
||
"""从时间戳创建TimeUtils对象"""
|
||
return TimeUtils(timestamp)
|
||
|
||
|
||
def from_string(time_str: str) -> TimeUtils:
|
||
"""从时间字符串创建TimeUtils对象"""
|
||
return TimeUtils(time_str)
|
||
|
||
|
||
def from_datetime(dt: datetime.datetime) -> TimeUtils:
|
||
"""从datetime对象创建TimeUtils对象"""
|
||
return TimeUtils(dt)
|
||
|
||
|
||
def get_timestamp() -> float:
|
||
"""获取当前时间戳"""
|
||
return time.time()
|
||
|
||
|
||
def format_timestamp(timestamp: Union[int, float], fmt: str = TimeUtils.FORMAT_DATETIME) -> str:
|
||
"""格式化时间戳"""
|
||
return TimeUtils(timestamp).to_string(fmt)
|
||
|
||
|
||
def parse_string(time_str: str) -> datetime.datetime:
|
||
"""解析时间字符串为datetime对象"""
|
||
return TimeUtils(time_str).to_datetime()
|
||
|
||
|
||
def days_ago(days: int, base_time: Optional[Union[str, int, float, datetime.datetime]] = None) -> TimeUtils:
|
||
"""计算n天前的日期时间"""
|
||
base = TimeUtils(base_time) if base_time else now()
|
||
return base.days_ago(days)
|
||
|
||
|
||
def days_after(days: int, base_time: Optional[Union[str, int, float, datetime.datetime]] = None) -> TimeUtils:
|
||
"""计算n天后的日期时间"""
|
||
base = TimeUtils(base_time) if base_time else now()
|
||
return base.days_after(days)
|