223 lines
8.1 KiB
Python
223 lines
8.1 KiB
Python
import json
|
|
import sqlite3
|
|
from datetime import datetime
|
|
from typing import Any, List
|
|
|
|
from fastapi import APIRouter, HTTPException, status
|
|
|
|
from app.database import Database
|
|
from app.schemas import Product, ProductCreate, ProductSpec, ProductUpdate
|
|
from app.temple_scope import TempleId, ensure_temple_exists
|
|
|
|
|
|
router = APIRouter(prefix="/products", tags=["products"])
|
|
|
|
PRODUCT_COLUMNS = """
|
|
id, temple_id, name, alias, description, payment_button_name, donation_button_name,
|
|
sort_order, show_participant_count, cover_image, images_json, listed_at,
|
|
sale_starts_at, sale_ends_at, delisted_at, show_countdown, show_on_home,
|
|
specs_json, details_html, share_title, share_description, share_image,
|
|
merit_certificate_reserved, auto_process, is_active, sales_count
|
|
"""
|
|
|
|
|
|
def parse_json_list(value: str | None) -> list[Any]:
|
|
if not value:
|
|
return []
|
|
try:
|
|
parsed = json.loads(value)
|
|
except json.JSONDecodeError:
|
|
return []
|
|
return parsed if isinstance(parsed, list) else []
|
|
|
|
|
|
def parse_time(value: str | None) -> datetime | None:
|
|
if not value:
|
|
return None
|
|
try:
|
|
return datetime.fromisoformat(value.replace("Z", "+00:00")).replace(tzinfo=None)
|
|
except ValueError:
|
|
return None
|
|
|
|
|
|
def get_product_status(row: sqlite3.Row) -> str:
|
|
now = datetime.now()
|
|
listed_at = parse_time(row["listed_at"])
|
|
sale_starts_at = parse_time(row["sale_starts_at"])
|
|
sale_ends_at = parse_time(row["sale_ends_at"])
|
|
delisted_at = parse_time(row["delisted_at"])
|
|
|
|
if not bool(row["is_active"]):
|
|
return "inactive"
|
|
if delisted_at is not None and now >= delisted_at:
|
|
return "delisted"
|
|
if sale_ends_at is not None and now >= sale_ends_at:
|
|
return "ended"
|
|
if listed_at is not None and now < listed_at:
|
|
return "pending_list"
|
|
if sale_starts_at is not None and now < sale_starts_at:
|
|
return "pending_sale"
|
|
return "selling"
|
|
|
|
|
|
def row_to_product(row: sqlite3.Row) -> Product:
|
|
return Product(
|
|
id=row["id"],
|
|
temple_id=row["temple_id"],
|
|
name=row["name"],
|
|
alias=row["alias"],
|
|
description=row["description"],
|
|
payment_button_name=row["payment_button_name"],
|
|
donation_button_name=row["donation_button_name"],
|
|
sort_order=row["sort_order"],
|
|
show_participant_count=bool(row["show_participant_count"]),
|
|
cover_image=row["cover_image"],
|
|
images=parse_json_list(row["images_json"]),
|
|
listed_at=row["listed_at"],
|
|
sale_starts_at=row["sale_starts_at"],
|
|
sale_ends_at=row["sale_ends_at"],
|
|
delisted_at=row["delisted_at"],
|
|
show_countdown=bool(row["show_countdown"]),
|
|
show_on_home=bool(row["show_on_home"]),
|
|
specs=[ProductSpec(**item) for item in parse_json_list(row["specs_json"])],
|
|
details_html=row["details_html"],
|
|
share_title=row["share_title"],
|
|
share_description=row["share_description"],
|
|
share_image=row["share_image"],
|
|
merit_certificate_reserved=bool(row["merit_certificate_reserved"]),
|
|
auto_process=bool(row["auto_process"]),
|
|
is_active=bool(row["is_active"]),
|
|
sales_count=row["sales_count"],
|
|
status=get_product_status(row),
|
|
)
|
|
|
|
|
|
def product_payload_values(payload: ProductCreate | ProductUpdate, fallback: Product | None = None) -> tuple[Any, ...]:
|
|
data = payload.model_dump(exclude_unset=fallback is not None)
|
|
|
|
def get_value(name: str, default: Any) -> Any:
|
|
if name in data:
|
|
return data[name]
|
|
if fallback is not None:
|
|
return getattr(fallback, name)
|
|
return default
|
|
|
|
images = get_value("images", [])
|
|
specs = get_value("specs", [])
|
|
return (
|
|
get_value("name", ""),
|
|
get_value("alias", None),
|
|
get_value("description", None),
|
|
get_value("payment_button_name", "立即支付"),
|
|
get_value("donation_button_name", "随喜"),
|
|
get_value("sort_order", 0),
|
|
int(get_value("show_participant_count", False)),
|
|
get_value("cover_image", None),
|
|
json.dumps(images, ensure_ascii=False),
|
|
get_value("listed_at", None),
|
|
get_value("sale_starts_at", None),
|
|
get_value("sale_ends_at", None),
|
|
get_value("delisted_at", None),
|
|
int(get_value("show_countdown", False)),
|
|
int(get_value("show_on_home", False)),
|
|
json.dumps([spec.model_dump() if isinstance(spec, ProductSpec) else spec for spec in specs], ensure_ascii=False),
|
|
get_value("details_html", None),
|
|
get_value("share_title", None),
|
|
get_value("share_description", None),
|
|
get_value("share_image", None),
|
|
int(get_value("merit_certificate_reserved", False)),
|
|
int(get_value("auto_process", False)),
|
|
int(get_value("is_active", True)),
|
|
get_value("sales_count", 0),
|
|
)
|
|
|
|
|
|
@router.get("", response_model=List[Product])
|
|
def list_products(db: Database, temple_id: TempleId) -> List[Product]:
|
|
ensure_temple_exists(db, temple_id)
|
|
rows = db.execute(
|
|
f"""
|
|
SELECT {PRODUCT_COLUMNS}
|
|
FROM products
|
|
WHERE temple_id = ?
|
|
ORDER BY sort_order, id
|
|
""",
|
|
(temple_id,),
|
|
).fetchall()
|
|
return [row_to_product(row) for row in rows]
|
|
|
|
|
|
@router.post("", response_model=Product, status_code=status.HTTP_201_CREATED)
|
|
def create_product(payload: ProductCreate, db: Database, temple_id: TempleId) -> Product:
|
|
ensure_temple_exists(db, temple_id)
|
|
cursor = db.execute(
|
|
"""
|
|
INSERT INTO products
|
|
(
|
|
temple_id, name, alias, description, payment_button_name, donation_button_name,
|
|
sort_order, show_participant_count, cover_image, images_json, listed_at,
|
|
sale_starts_at, sale_ends_at, delisted_at, show_countdown, show_on_home,
|
|
specs_json, details_html, share_title, share_description, share_image,
|
|
merit_certificate_reserved, auto_process, is_active, sales_count
|
|
)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
""",
|
|
(temple_id, *product_payload_values(payload)),
|
|
)
|
|
db.commit()
|
|
return get_product(cursor.lastrowid, db, temple_id)
|
|
|
|
|
|
@router.get("/{product_id}", response_model=Product)
|
|
def get_product(product_id: int, db: Database, temple_id: TempleId) -> Product:
|
|
ensure_temple_exists(db, temple_id)
|
|
row = db.execute(
|
|
f"""
|
|
SELECT {PRODUCT_COLUMNS}
|
|
FROM products
|
|
WHERE id = ? AND temple_id = ?
|
|
""",
|
|
(product_id, temple_id),
|
|
).fetchone()
|
|
if row is None:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Product not found")
|
|
return row_to_product(row)
|
|
|
|
|
|
@router.patch("/{product_id}", response_model=Product)
|
|
def update_product(product_id: int, payload: ProductUpdate, db: Database, temple_id: TempleId) -> Product:
|
|
product = get_product(product_id, db, temple_id)
|
|
updates = payload.model_dump(exclude_unset=True)
|
|
|
|
if not updates:
|
|
return product
|
|
|
|
db.execute(
|
|
"""
|
|
UPDATE products
|
|
SET
|
|
name = ?, alias = ?, description = ?, payment_button_name = ?, donation_button_name = ?,
|
|
sort_order = ?, show_participant_count = ?, cover_image = ?, images_json = ?,
|
|
listed_at = ?, sale_starts_at = ?, sale_ends_at = ?, delisted_at = ?,
|
|
show_countdown = ?, show_on_home = ?, specs_json = ?, details_html = ?,
|
|
share_title = ?, share_description = ?, share_image = ?,
|
|
merit_certificate_reserved = ?, auto_process = ?, is_active = ?, sales_count = ?
|
|
WHERE id = ? AND temple_id = ?
|
|
""",
|
|
(*product_payload_values(payload, product), product_id, temple_id),
|
|
)
|
|
db.commit()
|
|
return get_product(product_id, db, temple_id)
|
|
|
|
|
|
@router.delete("/{product_id}", status_code=status.HTTP_204_NO_CONTENT)
|
|
def delete_product(product_id: int, db: Database, temple_id: TempleId) -> None:
|
|
ensure_temple_exists(db, temple_id)
|
|
cursor = db.execute(
|
|
"DELETE FROM products WHERE id = ? AND temple_id = ?",
|
|
(product_id, temple_id),
|
|
)
|
|
db.commit()
|
|
if cursor.rowcount == 0:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Product not found")
|