File size: 3,131 Bytes
d0f36e8 f99684d d0f36e8 f99684d d0f36e8 f99684d d0f36e8 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 |
import os
from fastapi import FastAPI, Query, HTTPException
from fastapi.staticfiles import StaticFiles
from fastapi.responses import RedirectResponse
from typing import List, Optional
from pydantic import BaseModel
from goofish_api import GoofishAPI
from data_parser import ItemDetail, parse_search_result
# 定义搜索响应模型
class SearchResponse(BaseModel):
code: int = 0
message: str = "success"
data: List[ItemDetail]
total: int = 0
app = FastAPI(
title="闲鱼搜索API",
description="搜索闲鱼商品的API服务",
version="1.0.0"
)
# 挂载静态文件
app.mount("/static", StaticFiles(directory="static"), name="static")
# 初始化API客户端
goofish_client = GoofishAPI()
@app.get("/", include_in_schema=False)
async def root():
return RedirectResponse(url="/static/index.html")
@app.get("/api/search", response_model=SearchResponse, tags=["搜索"])
async def search(
keyword: str = Query(..., description="搜索关键词"),
page: int = Query(1, description="页码，从1开始"),
page_size: int = Query(10, description="每页结果数量，最大40"),
min_price: Optional[float] = Query(None, description="最低价格"),
max_price: Optional[float] = Query(None, description="最高价格"),
publish_days: Optional[int] = Query(None, description="最近发布天数，例如3表示最近3天发布的商品")
):
"""
搜索闲鱼商品
"""
try:
# 调用GoofishAPI的search_with_retry方法以支持自动刷新token
raw_result = goofish_client.search_with_retry(
keyword=keyword,
page_number=page,
rows_per_page=page_size,
min_price=min_price,
max_price=max_price,
publish_days=publish_days
)
# 如果搜索失败（所有重试都失败）
if raw_result is None:
raise HTTPException(status_code=502, detail="搜索服务暂时不可用，可能是令牌已过期")
# 解析结果
items = parse_search_result(raw_result)
total = len(items)
# 构建响应
return SearchResponse(
code=0,
message="success",
data=items,
total=total
)
except Exception as e:
print(f"搜索出错: {str(e)}")
raise HTTPException(status_code=500, detail=f"搜索出错：{str(e)}")
# 添加刷新token的API端点
@app.post("/api/refresh_token", tags=["系统"])
async def refresh_token():
"""
手动刷新API令牌
"""
try:
success = goofish_client.refresh_token_via_api_call()
if success:
return {"code": 0, "message": "令牌刷新成功"}
else:
return {"code": 1, "message": "令牌刷新失败，请查看日志"}
except Exception as e:
print(f"刷新令牌出错: {str(e)}")
raise HTTPException(status_code=500, detail=f"刷新令牌出错：{str(e)}")
if __name__ == "__main__":
port = int(os.environ.get("PORT", 8000))
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=port) |