85 lines
2.7 KiB
Python
85 lines
2.7 KiB
Python
"""分享链接管理路由"""
|
|
|
|
from fastapi import APIRouter, HTTPException
|
|
from pydantic import BaseModel
|
|
from typing import Optional
|
|
from database import get_connection, dict_from_row
|
|
import random
|
|
import string
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
@router.get("/{short_code}", summary="获取分享链接信息")
|
|
async def get_share(short_code: str):
|
|
conn = get_connection()
|
|
try:
|
|
row = conn.execute(
|
|
"SELECT * FROM share_links WHERE short_code = ?",
|
|
(short_code,),
|
|
).fetchone()
|
|
if not row:
|
|
raise HTTPException(status_code=404, detail="分享链接无效")
|
|
share = dict_from_row(row)
|
|
|
|
collection = conn.execute(
|
|
"SELECT * FROM stock_collections WHERE id = ?",
|
|
(share["collection_id"],),
|
|
).fetchone()
|
|
if not collection:
|
|
raise HTTPException(status_code=404, detail="集合不存在")
|
|
|
|
stocks = conn.execute(
|
|
"SELECT * FROM collection_stocks WHERE collection_id = ? ORDER BY added_at DESC",
|
|
(share["collection_id"],),
|
|
).fetchall()
|
|
|
|
coll_data = dict_from_row(collection)
|
|
coll_data["stocks"] = [dict_from_row(s) for s in stocks]
|
|
return coll_data
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
class ShareCreate(BaseModel):
|
|
collection_id: int
|
|
|
|
|
|
def generate_short_code(length: int = 8) -> str:
|
|
return "".join(random.choices(string.ascii_lowercase + string.digits, k=length))
|
|
|
|
|
|
@router.post("", summary="创建分享链接", status_code=201)
|
|
async def create_share(body: ShareCreate):
|
|
conn = get_connection()
|
|
try:
|
|
# 检查集合存在
|
|
coll = conn.execute("SELECT id FROM stock_collections WHERE id = ?", (body.collection_id,)).fetchone()
|
|
if not coll:
|
|
raise HTTPException(status_code=404, detail="集合不存在")
|
|
|
|
# 查询是否已有分享链接
|
|
existing = conn.execute(
|
|
"SELECT short_code FROM share_links WHERE collection_id = ? ORDER BY created_at DESC LIMIT 1",
|
|
(body.collection_id,),
|
|
).fetchone()
|
|
|
|
if existing:
|
|
return {"short_code": existing["short_code"]}
|
|
|
|
# 生成新链接
|
|
for _ in range(10):
|
|
short_code = generate_short_code()
|
|
try:
|
|
conn.execute(
|
|
"INSERT INTO share_links (collection_id, short_code) VALUES (?, ?)",
|
|
(body.collection_id, short_code),
|
|
)
|
|
conn.commit()
|
|
return {"short_code": short_code}
|
|
except Exception:
|
|
continue
|
|
raise HTTPException(status_code=500, detail="生成分享链接失败")
|
|
finally:
|
|
conn.close()
|