285 lines
10 KiB
Python
285 lines
10 KiB
Python
from fastapi import APIRouter, Depends, HTTPException, Query
|
||
from sqlalchemy.orm import Session
|
||
from typing import Optional, List
|
||
from datetime import date
|
||
|
||
from app.database import get_db
|
||
from app.models.anniversary import Anniversary, AnniversaryCategory
|
||
from app.schemas.anniversary import (
|
||
AnniversaryCreate, AnniversaryUpdate, AnniversaryResponse,
|
||
AnniversaryCategoryCreate, AnniversaryCategoryUpdate, AnniversaryCategoryResponse,
|
||
)
|
||
from app.schemas.common import DeleteResponse
|
||
from app.utils.crud import get_or_404
|
||
from app.utils.datetime import utcnow
|
||
from app.utils.logger import logger
|
||
|
||
router = APIRouter(prefix="/api", tags=["纪念日"])
|
||
|
||
|
||
def compute_next_info(anniversary: Anniversary, today: date) -> tuple:
|
||
"""计算纪念日的下一次日期、距今天数、周年数"""
|
||
month, day = anniversary.date.month, anniversary.date.day
|
||
|
||
if anniversary.is_recurring:
|
||
# 计算今年和明年的日期
|
||
this_year = today.year
|
||
next_date = date(this_year, month, day)
|
||
if next_date < today:
|
||
next_date = date(this_year + 1, month, day)
|
||
|
||
days_until = (next_date - today).days
|
||
|
||
year_count = None
|
||
if anniversary.year:
|
||
year_count = next_date.year - anniversary.year
|
||
|
||
return next_date, days_until, year_count
|
||
else:
|
||
# 非重复：使用原始日期（加上年份）
|
||
if anniversary.year:
|
||
target = date(anniversary.year, month, day)
|
||
if target < today:
|
||
return None, None, None
|
||
days_until = (target - today).days
|
||
return target, days_until, 0
|
||
else:
|
||
# 无年份的日期按今年算
|
||
target = date(today.year, month, day)
|
||
if target < today:
|
||
return None, None, None
|
||
days_until = (target - today).days
|
||
return target, days_until, None
|
||
|
||
|
||
def enrich_anniversary(anniversary: Anniversary, today: date) -> dict:
|
||
"""将 SQLAlchemy 模型转换为响应字典，附加计算字段"""
|
||
data = {
|
||
"id": anniversary.id,
|
||
"title": anniversary.title,
|
||
"date": anniversary.date,
|
||
"year": anniversary.year,
|
||
"category_id": anniversary.category_id,
|
||
"description": anniversary.description,
|
||
"is_recurring": anniversary.is_recurring,
|
||
"remind_days_before": anniversary.remind_days_before,
|
||
"created_at": anniversary.created_at,
|
||
"updated_at": anniversary.updated_at,
|
||
}
|
||
|
||
next_date, days_until, year_count = compute_next_info(anniversary, today)
|
||
data["next_date"] = next_date
|
||
data["days_until"] = days_until
|
||
data["year_count"] = year_count
|
||
|
||
if anniversary.category:
|
||
data["category"] = {
|
||
"id": anniversary.category.id,
|
||
"name": anniversary.category.name,
|
||
"icon": anniversary.category.icon,
|
||
"color": anniversary.category.color,
|
||
"sort_order": anniversary.category.sort_order,
|
||
}
|
||
else:
|
||
data["category"] = None
|
||
|
||
return data
|
||
|
||
|
||
# ============ 纪念日分类 API ============
|
||
|
||
@router.get("/anniversary-categories", response_model=List[AnniversaryCategoryResponse])
|
||
def get_categories(db: Session = Depends(get_db)):
|
||
"""获取纪念日分类列表"""
|
||
try:
|
||
categories = db.query(AnniversaryCategory).order_by(
|
||
AnniversaryCategory.sort_order.asc(),
|
||
AnniversaryCategory.id.asc()
|
||
).all()
|
||
logger.info(f"获取纪念日分类列表成功，总数: {len(categories)}")
|
||
return categories
|
||
except Exception as e:
|
||
logger.error(f"获取纪念日分类列表失败: {str(e)}")
|
||
raise HTTPException(status_code=500, detail="获取纪念日分类列表失败")
|
||
|
||
|
||
@router.post("/anniversary-categories", response_model=AnniversaryCategoryResponse, status_code=201)
|
||
def create_category(data: AnniversaryCategoryCreate, db: Session = Depends(get_db)):
|
||
"""创建纪念日分类"""
|
||
try:
|
||
db_category = AnniversaryCategory(
|
||
name=data.name,
|
||
icon=data.icon,
|
||
color=data.color,
|
||
sort_order=data.sort_order,
|
||
)
|
||
db.add(db_category)
|
||
db.commit()
|
||
db.refresh(db_category)
|
||
logger.info(f"创建纪念日分类成功: id={db_category.id}, name={db_category.name}")
|
||
return db_category
|
||
except Exception as e:
|
||
db.rollback()
|
||
logger.error(f"创建纪念日分类失败: {str(e)}")
|
||
raise HTTPException(status_code=500, detail="创建纪念日分类失败")
|
||
|
||
|
||
@router.put("/anniversary-categories/{category_id}", response_model=AnniversaryCategoryResponse)
|
||
def update_category(category_id: int, data: AnniversaryCategoryUpdate, db: Session = Depends(get_db)):
|
||
"""更新纪念日分类"""
|
||
try:
|
||
category = get_or_404(db, AnniversaryCategory, category_id, "纪念日分类")
|
||
update_data = data.model_dump(exclude_unset=True)
|
||
for field, value in update_data.items():
|
||
setattr(category, field, value)
|
||
db.commit()
|
||
db.refresh(category)
|
||
logger.info(f"更新纪念日分类成功: id={category_id}")
|
||
return category
|
||
except HTTPException:
|
||
raise
|
||
except Exception as e:
|
||
db.rollback()
|
||
logger.error(f"更新纪念日分类失败: {str(e)}")
|
||
raise HTTPException(status_code=500, detail="更新纪念日分类失败")
|
||
|
||
|
||
@router.delete("/anniversary-categories/{category_id}")
|
||
def delete_category(category_id: int, db: Session = Depends(get_db)):
|
||
"""删除纪念日分类"""
|
||
try:
|
||
category = get_or_404(db, AnniversaryCategory, category_id, "纪念日分类")
|
||
# 删除分类时，将其下纪念日的 category_id 设为 NULL
|
||
anniversaries = db.query(Anniversary).filter(
|
||
Anniversary.category_id == category_id
|
||
).all()
|
||
for a in anniversaries:
|
||
a.category_id = None
|
||
db.delete(category)
|
||
db.commit()
|
||
logger.info(f"删除纪念日分类成功: id={category_id}")
|
||
return DeleteResponse(message="纪念日分类删除成功")
|
||
except HTTPException:
|
||
raise
|
||
except Exception as e:
|
||
db.rollback()
|
||
logger.error(f"删除纪念日分类失败: {str(e)}")
|
||
raise HTTPException(status_code=500, detail="删除纪念日分类失败")
|
||
|
||
|
||
# ============ 纪念日 API ============
|
||
|
||
@router.get("/anniversaries", response_model=List[AnniversaryResponse])
|
||
def get_anniversaries(
|
||
category_id: Optional[int] = Query(None, description="分类ID筛选"),
|
||
db: Session = Depends(get_db)
|
||
):
|
||
"""获取纪念日列表（包含计算字段 next_date, days_until, year_count）"""
|
||
try:
|
||
query = db.query(Anniversary)
|
||
if category_id is not None:
|
||
query = query.filter(Anniversary.category_id == category_id)
|
||
|
||
anniversaries = query.order_by(
|
||
Anniversary.date.asc(),
|
||
Anniversary.title.asc()
|
||
).all()
|
||
|
||
today = date.today()
|
||
result = [enrich_anniversary(a, today) for a in anniversaries]
|
||
|
||
# 排序：即将到来的排前面，同天数的按日期排
|
||
result.sort(key=lambda x: (
|
||
0 if (x["days_until"] is not None and x["days_until"] >= 0) else 1,
|
||
x["days_until"] if x["days_until"] is not None else 9999,
|
||
))
|
||
|
||
logger.info(f"获取纪念日列表成功，总数: {len(result)}")
|
||
return result
|
||
except Exception as e:
|
||
logger.error(f"获取纪念日列表失败: {str(e)}")
|
||
raise HTTPException(status_code=500, detail="获取纪念日列表失败")
|
||
|
||
|
||
@router.post("/anniversaries", response_model=AnniversaryResponse, status_code=201)
|
||
def create_anniversary(data: AnniversaryCreate, db: Session = Depends(get_db)):
|
||
"""创建纪念日"""
|
||
try:
|
||
db_anniversary = Anniversary(
|
||
title=data.title,
|
||
date=data.date,
|
||
year=data.year,
|
||
category_id=data.category_id,
|
||
description=data.description,
|
||
is_recurring=data.is_recurring,
|
||
remind_days_before=data.remind_days_before,
|
||
)
|
||
db.add(db_anniversary)
|
||
db.commit()
|
||
db.refresh(db_anniversary)
|
||
|
||
today = date.today()
|
||
next_date, days_until, year_count = compute_next_info(db_anniversary, today)
|
||
|
||
logger.info(f"创建纪念日成功: id={db_anniversary.id}, title={db_anniversary.title}")
|
||
return db_anniversary
|
||
except Exception as e:
|
||
db.rollback()
|
||
logger.error(f"创建纪念日失败: {str(e)}")
|
||
raise HTTPException(status_code=500, detail="创建纪念日失败")
|
||
|
||
|
||
@router.get("/anniversaries/{anniversary_id}", response_model=AnniversaryResponse)
|
||
def get_anniversary(anniversary_id: int, db: Session = Depends(get_db)):
|
||
"""获取单个纪念日"""
|
||
try:
|
||
anniversary = get_or_404(db, Anniversary, anniversary_id, "纪念日")
|
||
return anniversary
|
||
except HTTPException:
|
||
raise
|
||
except Exception as e:
|
||
logger.error(f"获取纪念日失败: {str(e)}")
|
||
raise HTTPException(status_code=500, detail="获取纪念日失败")
|
||
|
||
|
||
@router.put("/anniversaries/{anniversary_id}", response_model=AnniversaryResponse)
|
||
def update_anniversary(anniversary_id: int, data: AnniversaryUpdate, db: Session = Depends(get_db)):
|
||
"""更新纪念日"""
|
||
try:
|
||
anniversary = get_or_404(db, Anniversary, anniversary_id, "纪念日")
|
||
update_data = data.model_dump(exclude_unset=True)
|
||
|
||
for field, value in update_data.items():
|
||
if value is not None or field in data.clearable_fields:
|
||
setattr(anniversary, field, value)
|
||
|
||
anniversary.updated_at = utcnow()
|
||
db.commit()
|
||
db.refresh(anniversary)
|
||
|
||
logger.info(f"更新纪念日成功: id={anniversary_id}")
|
||
return anniversary
|
||
except HTTPException:
|
||
raise
|
||
except Exception as e:
|
||
db.rollback()
|
||
logger.error(f"更新纪念日失败: {str(e)}")
|
||
raise HTTPException(status_code=500, detail="更新纪念日失败")
|
||
|
||
|
||
@router.delete("/anniversaries/{anniversary_id}")
|
||
def delete_anniversary(anniversary_id: int, db: Session = Depends(get_db)):
|
||
"""删除纪念日"""
|
||
try:
|
||
anniversary = get_or_404(db, Anniversary, anniversary_id, "纪念日")
|
||
db.delete(anniversary)
|
||
db.commit()
|
||
logger.info(f"删除纪念日成功: id={anniversary_id}")
|
||
return DeleteResponse(message="纪念日删除成功")
|
||
except HTTPException:
|
||
raise
|
||
except Exception as e:
|
||
db.rollback()
|
||
logger.error(f"删除纪念日失败: {str(e)}")
|
||
raise HTTPException(status_code=500, detail="删除纪念日失败")
|