239 lines
8.2 KiB
Python
239 lines
8.2 KiB
Python
import sqlite3
|
|
from typing import List
|
|
|
|
from fastapi import APIRouter, HTTPException, status
|
|
|
|
from app.database import Database
|
|
from app.schemas import FeatureIcon, FeatureIconCopyRequest, FeatureIconCreate, FeatureIconUpdate
|
|
from app.temple_scope import TempleId, ensure_temple_exists
|
|
|
|
|
|
router = APIRouter(prefix="/feature-icons", tags=["feature-icons"])
|
|
|
|
ICON_COLUMNS = """
|
|
feature_icons.id, feature_icons.temple_id, feature_icons.title, feature_icons.icon_url,
|
|
feature_icons.target_type, feature_icons.target_url, feature_icons.product_id,
|
|
products.name AS product_name, feature_icons.sort_order, feature_icons.is_active
|
|
"""
|
|
|
|
|
|
def row_to_icon(row: sqlite3.Row) -> FeatureIcon:
|
|
return FeatureIcon(
|
|
id=row["id"],
|
|
temple_id=row["temple_id"],
|
|
title=row["title"],
|
|
icon_url=row["icon_url"],
|
|
target_type=row["target_type"],
|
|
target_url=row["target_url"],
|
|
product_id=row["product_id"],
|
|
product_name=row["product_name"],
|
|
sort_order=row["sort_order"],
|
|
is_active=bool(row["is_active"]),
|
|
)
|
|
|
|
|
|
def ensure_product_in_temple(db: Database, product_id: int | None, temple_id: int) -> None:
|
|
if product_id is None:
|
|
return
|
|
row = db.execute(
|
|
"SELECT id FROM products WHERE id = ? AND temple_id = ?",
|
|
(product_id, temple_id),
|
|
).fetchone()
|
|
if row is None:
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="请选择当前寺院下的商品")
|
|
|
|
|
|
def normalize_icon_payload(
|
|
payload: FeatureIconCreate | FeatureIconUpdate,
|
|
temple_id: int,
|
|
db: Database,
|
|
fallback: FeatureIcon | None = None,
|
|
) -> tuple[str, str, str, str | None, int | None, int, int]:
|
|
data = payload.model_dump(exclude_unset=fallback is not None)
|
|
|
|
def get_value(name: str, default):
|
|
if name in data:
|
|
return data[name]
|
|
if fallback is not None:
|
|
return getattr(fallback, name)
|
|
return default
|
|
|
|
target_type = get_value("target_type", "link")
|
|
target_url = get_value("target_url", None)
|
|
product_id = get_value("product_id", None)
|
|
|
|
if target_type == "link":
|
|
product_id = None
|
|
if not target_url:
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="链接类型需要填写跳转地址")
|
|
else:
|
|
target_url = None
|
|
ensure_product_in_temple(db, product_id, temple_id)
|
|
if product_id is None:
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="应用类型需要选择商品")
|
|
|
|
return (
|
|
get_value("title", ""),
|
|
get_value("icon_url", ""),
|
|
target_type,
|
|
target_url,
|
|
product_id,
|
|
get_value("sort_order", 0),
|
|
int(get_value("is_active", True)),
|
|
)
|
|
|
|
|
|
@router.get("", response_model=List[FeatureIcon])
|
|
def list_icons(db: Database, temple_id: TempleId) -> List[FeatureIcon]:
|
|
ensure_temple_exists(db, temple_id)
|
|
rows = db.execute(
|
|
f"""
|
|
SELECT {ICON_COLUMNS}
|
|
FROM feature_icons
|
|
LEFT JOIN products ON products.id = feature_icons.product_id
|
|
WHERE feature_icons.temple_id = ?
|
|
ORDER BY feature_icons.sort_order, feature_icons.id
|
|
""",
|
|
(temple_id,),
|
|
).fetchall()
|
|
return [row_to_icon(row) for row in rows]
|
|
|
|
|
|
@router.get("/source/{source_temple_id}", response_model=List[FeatureIcon])
|
|
def list_source_icons(source_temple_id: int, db: Database) -> List[FeatureIcon]:
|
|
ensure_temple_exists(db, source_temple_id)
|
|
rows = db.execute(
|
|
f"""
|
|
SELECT {ICON_COLUMNS}
|
|
FROM feature_icons
|
|
LEFT JOIN products ON products.id = feature_icons.product_id
|
|
WHERE feature_icons.temple_id = ?
|
|
ORDER BY feature_icons.sort_order, feature_icons.id
|
|
""",
|
|
(source_temple_id,),
|
|
).fetchall()
|
|
return [row_to_icon(row) for row in rows]
|
|
|
|
|
|
@router.post("", response_model=FeatureIcon, status_code=status.HTTP_201_CREATED)
|
|
def create_icon(payload: FeatureIconCreate, db: Database, temple_id: TempleId) -> FeatureIcon:
|
|
ensure_temple_exists(db, temple_id)
|
|
cursor = db.execute(
|
|
"""
|
|
INSERT INTO feature_icons
|
|
(temple_id, title, icon_url, target_type, target_url, product_id, sort_order, is_active)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
|
|
""",
|
|
(temple_id, *normalize_icon_payload(payload, temple_id, db)),
|
|
)
|
|
db.commit()
|
|
return get_icon(cursor.lastrowid, db, temple_id)
|
|
|
|
|
|
@router.post("/copy", response_model=List[FeatureIcon], status_code=status.HTTP_201_CREATED)
|
|
def copy_icons(payload: FeatureIconCopyRequest, db: Database, temple_id: TempleId) -> List[FeatureIcon]:
|
|
ensure_temple_exists(db, temple_id)
|
|
ensure_temple_exists(db, payload.source_temple_id)
|
|
if payload.source_temple_id == temple_id:
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="请选择其他寺院的图标")
|
|
|
|
placeholders = ",".join("?" for _ in payload.icon_ids)
|
|
rows = db.execute(
|
|
f"""
|
|
SELECT id, title, icon_url, target_type, target_url, product_id, sort_order, is_active
|
|
FROM feature_icons
|
|
WHERE temple_id = ? AND id IN ({placeholders})
|
|
ORDER BY sort_order, id
|
|
""",
|
|
(payload.source_temple_id, *payload.icon_ids),
|
|
).fetchall()
|
|
if not rows:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="未找到可复制的图标")
|
|
|
|
new_ids: list[int] = []
|
|
for row in rows:
|
|
product_id = row["product_id"]
|
|
is_active = row["is_active"]
|
|
if row["target_type"] == "app":
|
|
product_id = None
|
|
is_active = 0
|
|
cursor = db.execute(
|
|
"""
|
|
INSERT INTO feature_icons
|
|
(temple_id, title, icon_url, target_type, target_url, product_id, sort_order, is_active)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
|
|
""",
|
|
(
|
|
temple_id,
|
|
row["title"],
|
|
row["icon_url"],
|
|
row["target_type"],
|
|
row["target_url"],
|
|
product_id,
|
|
row["sort_order"],
|
|
is_active,
|
|
),
|
|
)
|
|
new_ids.append(cursor.lastrowid)
|
|
|
|
db.commit()
|
|
placeholders = ",".join("?" for _ in new_ids)
|
|
copied_rows = db.execute(
|
|
f"""
|
|
SELECT {ICON_COLUMNS}
|
|
FROM feature_icons
|
|
LEFT JOIN products ON products.id = feature_icons.product_id
|
|
WHERE feature_icons.temple_id = ? AND feature_icons.id IN ({placeholders})
|
|
ORDER BY feature_icons.sort_order, feature_icons.id
|
|
""",
|
|
(temple_id, *new_ids),
|
|
).fetchall()
|
|
return [row_to_icon(row) for row in copied_rows]
|
|
|
|
|
|
@router.get("/{icon_id}", response_model=FeatureIcon)
|
|
def get_icon(icon_id: int, db: Database, temple_id: TempleId) -> FeatureIcon:
|
|
ensure_temple_exists(db, temple_id)
|
|
row = db.execute(
|
|
f"""
|
|
SELECT {ICON_COLUMNS}
|
|
FROM feature_icons
|
|
LEFT JOIN products ON products.id = feature_icons.product_id
|
|
WHERE feature_icons.id = ? AND feature_icons.temple_id = ?
|
|
""",
|
|
(icon_id, temple_id),
|
|
).fetchone()
|
|
if row is None:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Icon not found")
|
|
return row_to_icon(row)
|
|
|
|
|
|
@router.patch("/{icon_id}", response_model=FeatureIcon)
|
|
def update_icon(icon_id: int, payload: FeatureIconUpdate, db: Database, temple_id: TempleId) -> FeatureIcon:
|
|
icon = get_icon(icon_id, db, temple_id)
|
|
if not payload.model_dump(exclude_unset=True):
|
|
return icon
|
|
|
|
db.execute(
|
|
"""
|
|
UPDATE feature_icons
|
|
SET title = ?, icon_url = ?, target_type = ?, target_url = ?, product_id = ?, sort_order = ?, is_active = ?
|
|
WHERE id = ? AND temple_id = ?
|
|
""",
|
|
(*normalize_icon_payload(payload, temple_id, db, icon), icon_id, temple_id),
|
|
)
|
|
db.commit()
|
|
return get_icon(icon_id, db, temple_id)
|
|
|
|
|
|
@router.delete("/{icon_id}", status_code=status.HTTP_204_NO_CONTENT)
|
|
def delete_icon(icon_id: int, db: Database, temple_id: TempleId) -> None:
|
|
ensure_temple_exists(db, temple_id)
|
|
cursor = db.execute(
|
|
"DELETE FROM feature_icons WHERE id = ? AND temple_id = ?",
|
|
(icon_id, temple_id),
|
|
)
|
|
db.commit()
|
|
if cursor.rowcount == 0:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Icon not found")
|