70 lines
2.6 KiB
Python
70 lines
2.6 KiB
Python
from datetime import datetime, timedelta, timezone
|
|
|
|
from sqlalchemy import desc
|
|
|
|
from database.tcontentdispatch.model import TContentDispatch
|
|
|
|
|
|
def create_content(db, content: TContentDispatch):
|
|
db.add(content)
|
|
db.commit()
|
|
db.refresh(content)
|
|
return content
|
|
|
|
|
|
def create_or_update_content(db, content: TContentDispatch):
|
|
content_in_db = db.query(TContentDispatch).filter(TContentDispatch.id == content.id).first()
|
|
|
|
if content_in_db:
|
|
# Update existing content
|
|
db.commit() # Save changes to the database
|
|
db.refresh(content) # Refresh the object with the updated values
|
|
else:
|
|
# Create new content if not found in DB
|
|
db.add(content)
|
|
db.commit() # Save new content to the database
|
|
db.refresh(content) # Refresh to get the content's updated state (e.g., ID if it's auto-generated)
|
|
|
|
|
|
def get_content_by_id(db, content_id: int):
|
|
return db.query(TContentDispatch).filter(TContentDispatch.id == content_id).first()
|
|
|
|
def get_content_by_title_and_category(db, title: str, category: str):
|
|
return db.query(TContentDispatch).filter(TContentDispatch.title == title, TContentDispatch.category == category).first()
|
|
|
|
def get_contents_to_dispatch(db) -> list[TContentDispatch]:
|
|
return db.query(TContentDispatch).filter(TContentDispatch.is_sent == False).all()
|
|
|
|
def get_recent_24_hours_content_to_dispatch(db, category: str) -> TContentDispatch:
|
|
# 获取查询日期的起始和结束时间
|
|
end_time = datetime.now(timezone.utc)
|
|
start_time = end_time - timedelta(hours=24)
|
|
# 查询最近的一条未发送的内容
|
|
return db.query(TContentDispatch).filter(
|
|
TContentDispatch.category == category,
|
|
# TContentDispatch.is_sent == False,
|
|
TContentDispatch.creation_date >= start_time,
|
|
TContentDispatch.creation_date < end_time
|
|
).order_by(desc(TContentDispatch.creation_date)).first()
|
|
|
|
def finish_contents_to_dispatch(db, ids):
|
|
db.query(TContentDispatch).filter(TContentDispatch.id.in_(ids)).update({'is_sent': True})
|
|
db.commit()
|
|
|
|
def update_content(db, content_id: int, updates: dict):
|
|
content = db.query(TContentDispatch).filter(TContentDispatch.id == content_id).first()
|
|
if content:
|
|
for key, value in updates.items():
|
|
setattr(content, key, value)
|
|
db.commit()
|
|
db.refresh(content)
|
|
return content
|
|
|
|
|
|
def delete_content(db, content_id: int):
|
|
content = db.query(TContentDispatch).filter(TContentDispatch.id == content_id).first()
|
|
if content:
|
|
db.delete(content)
|
|
db.commit()
|
|
return content
|