houhuan
2bc7460f1f
- 添加FastAPI应用基础结构，包括主入口、路由和模型定义
- 实现Webhook接收端点(/webhook/{namespace})和健康检查(/health)
- 添加管理后台路由和模板，支持端点、目标、渠道和模板管理
- 包含SQLite数据库模型定义和初始化逻辑
- 添加日志记录和统计服务
- 包含Dockerfile和配置示例文件
- 添加项目文档，包括设计、流程图和验收标准
46 lines
1.5 KiB
Python
46 lines
1.5 KiB
Python
from datetime import datetime, timedelta
|
||
from app.db import SessionLocal, RequestLog
|
||
from sqlalchemy import func
|
||
|
||
# 全局变量，记录启动时间
|
||
START_TIME = datetime.utcnow()
|
||
|
||
class SystemStats:
|
||
def __init__(self):
|
||
pass
|
||
|
||
def get_uptime(self) -> str:
|
||
delta = datetime.utcnow() - START_TIME
|
||
days = delta.days
|
||
hours, remainder = divmod(delta.seconds, 3600)
|
||
minutes, seconds = divmod(remainder, 60)
|
||
if days > 0:
|
||
return f"{days}天 {hours}小时"
|
||
elif hours > 0:
|
||
return f"{hours}小时 {minutes}分"
|
||
else:
|
||
return f"{minutes}分 {seconds}秒"
|
||
|
||
def get_today_count(self) -> int:
|
||
session = SessionLocal()
|
||
try:
|
||
today_start = datetime.utcnow().replace(hour=0, minute=0, second=0, microsecond=0)
|
||
count = session.query(func.count(RequestLog.id)).filter(RequestLog.received_at >= today_start).scalar()
|
||
return count or 0
|
||
finally:
|
||
session.close()
|
||
|
||
def get_latest_log_time(self) -> str:
|
||
session = SessionLocal()
|
||
try:
|
||
log = session.query(RequestLog).order_by(RequestLog.received_at.desc()).first()
|
||
if log:
|
||
# 简单转为本地时间显示（+8）
|
||
dt = log.received_at + timedelta(hours=8)
|
||
return dt.strftime("%H:%M:%S")
|
||
return "无"
|
||
finally:
|
||
session.close()
|
||
|
||
stats_service = SystemStats()
|