be52af70
tangwang
first commit
|
1
2
3
4
|
"""
Main FastAPI application for the search service.
Usage:
|
2a76641e
tangwang
config
|
5
|
uvicorn api.app:app --host 0.0.0.0 --port 6002 --reload
|
be52af70
tangwang
first commit
|
6
7
8
9
|
"""
import os
import sys
|
bb3c5ef8
tangwang
灌入数据流程跑通
|
10
11
|
import logging
import time
|
325eec03
tangwang
1. 日志、配置基础设施,使用优化
|
12
13
|
import argparse
import uvicorn
|
28e57bb1
tangwang
日志体系优化
|
14
|
from logging.handlers import TimedRotatingFileHandler
|
bb3c5ef8
tangwang
灌入数据流程跑通
|
15
|
from collections import defaultdict, deque
|
be52af70
tangwang
first commit
|
16
|
from typing import Optional
|
bb3c5ef8
tangwang
灌入数据流程跑通
|
17
|
from fastapi import FastAPI, Request, HTTPException
|
a7a8c6cb
tangwang
测试过滤、聚合、排序
|
18
19
|
from fastapi.responses import JSONResponse, FileResponse
from fastapi.staticfiles import StaticFiles
|
be52af70
tangwang
first commit
|
20
|
from fastapi.middleware.cors import CORSMiddleware
|
bb3c5ef8
tangwang
灌入数据流程跑通
|
21
22
23
24
25
|
from fastapi.middleware.trustedhost import TrustedHostMiddleware
from fastapi.middleware.httpsredirect import HTTPSRedirectMiddleware
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded
|
be52af70
tangwang
first commit
|
26
|
|
28e57bb1
tangwang
日志体系优化
|
27
|
# Configure backend logging
|
3c1f8031
tangwang
api/routes/indexe...
|
28
|
import pathlib
|
28e57bb1
tangwang
日志体系优化
|
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
|
def configure_backend_logging() -> None:
log_dir = pathlib.Path("logs")
log_dir.mkdir(exist_ok=True)
log_level = os.getenv("LOG_LEVEL", "INFO").upper()
numeric_level = getattr(logging, log_level, logging.INFO)
default_formatter = logging.Formatter(
"%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
root_logger = logging.getLogger()
root_logger.setLevel(numeric_level)
root_logger.handlers.clear()
console_handler = logging.StreamHandler()
console_handler.setLevel(numeric_level)
console_handler.setFormatter(default_formatter)
root_logger.addHandler(console_handler)
backend_handler = TimedRotatingFileHandler(
filename=log_dir / "api.log",
when="midnight",
interval=1,
backupCount=30,
encoding="utf-8",
)
backend_handler.setLevel(numeric_level)
backend_handler.setFormatter(default_formatter)
root_logger.addHandler(backend_handler)
verbose_logger = logging.getLogger("backend.verbose")
verbose_logger.setLevel(numeric_level)
verbose_logger.handlers.clear()
verbose_logger.propagate = False
verbose_handler = TimedRotatingFileHandler(
filename=log_dir / "backend_verbose.log",
when="midnight",
interval=1,
backupCount=30,
encoding="utf-8",
)
verbose_handler.setLevel(numeric_level)
verbose_handler.setFormatter(
logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s")
)
verbose_logger.addHandler(verbose_handler)
configure_backend_logging()
|
bb3c5ef8
tangwang
灌入数据流程跑通
|
81
82
83
84
85
|
logger = logging.getLogger(__name__)
# Initialize rate limiter
limiter = Limiter(key_func=get_remote_address)
|
be52af70
tangwang
first commit
|
86
87
88
|
# Add parent directory to path
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
0064e946
tangwang
feat: 增量索引服务、租户配置...
|
89
|
from config.env_config import ES_CONFIG, DB_CONFIG
|
9f96d6f3
tangwang
短query不用语义搜索
|
90
|
from config import ConfigLoader
|
be52af70
tangwang
first commit
|
91
|
from utils import ESClient
|
be52af70
tangwang
first commit
|
92
|
from search import Searcher
|
be52af70
tangwang
first commit
|
93
|
from query import QueryParser
|
ded6f29e
tangwang
补充suggestion模块
|
94
|
from suggestion import SuggestionService
|
bb9c626c
tangwang
搜索服务(6002)不再初始化/挂...
|
95
|
from .service_registry import set_es_client
|
be52af70
tangwang
first commit
|
96
97
|
# Global instances
|
be52af70
tangwang
first commit
|
98
99
100
|
_es_client: Optional[ESClient] = None
_searcher: Optional[Searcher] = None
_query_parser: Optional[QueryParser] = None
|
ded6f29e
tangwang
补充suggestion模块
|
101
|
_suggestion_service: Optional[SuggestionService] = None
|
9f96d6f3
tangwang
短query不用语义搜索
|
102
|
_config = None
|
be52af70
tangwang
first commit
|
103
104
|
|
4d824a77
tangwang
所有租户共用一套统一配置.tena...
|
105
|
def init_service(es_host: str = "http://localhost:9200"):
|
be52af70
tangwang
first commit
|
106
|
"""
|
4d824a77
tangwang
所有租户共用一套统一配置.tena...
|
107
|
Initialize search service with unified configuration.
|
be52af70
tangwang
first commit
|
108
109
|
Args:
|
be52af70
tangwang
first commit
|
110
111
|
es_host: Elasticsearch host URL
"""
|
ded6f29e
tangwang
补充suggestion模块
|
112
|
global _es_client, _searcher, _query_parser, _suggestion_service, _config
|
be52af70
tangwang
first commit
|
113
|
|
325eec03
tangwang
1. 日志、配置基础设施,使用优化
|
114
115
|
start_time = time.time()
logger.info("Initializing search service (multi-tenant)")
|
be52af70
tangwang
first commit
|
116
|
|
9f96d6f3
tangwang
短query不用语义搜索
|
117
118
119
120
121
122
|
# Load configuration
logger.info("Loading configuration...")
config_loader = ConfigLoader("config/config.yaml")
_config = config_loader.load_config()
logger.info("Configuration loaded")
|
325eec03
tangwang
1. 日志、配置基础设施,使用优化
|
123
124
125
|
# Get ES credentials
es_username = os.getenv('ES_USERNAME') or ES_CONFIG.get('username')
es_password = os.getenv('ES_PASSWORD') or ES_CONFIG.get('password')
|
be52af70
tangwang
first commit
|
126
|
|
325eec03
tangwang
1. 日志、配置基础设施,使用优化
|
127
128
|
# Connect to Elasticsearch
logger.info(f"Connecting to Elasticsearch at {es_host}...")
|
a406638e
tangwang
up
|
129
|
if es_username and es_password:
|
a406638e
tangwang
up
|
130
131
|
_es_client = ESClient(hosts=[es_host], username=es_username, password=es_password)
else:
|
a406638e
tangwang
up
|
132
133
|
_es_client = ESClient(hosts=[es_host])
|
be52af70
tangwang
first commit
|
134
135
|
if not _es_client.ping():
raise ConnectionError(f"Failed to connect to Elasticsearch at {es_host}")
|
325eec03
tangwang
1. 日志、配置基础设施,使用优化
|
136
|
logger.info("Elasticsearch connected")
|
bb9c626c
tangwang
搜索服务(6002)不再初始化/挂...
|
137
138
|
# expose ES client for any shared components (e.g. searcher)
set_es_client(_es_client)
|
be52af70
tangwang
first commit
|
139
|
|
325eec03
tangwang
1. 日志、配置基础设施,使用优化
|
140
141
|
# Initialize components
logger.info("Initializing query parser...")
|
9f96d6f3
tangwang
短query不用语义搜索
|
142
|
_query_parser = QueryParser(_config)
|
325eec03
tangwang
1. 日志、配置基础设施,使用优化
|
143
144
|
logger.info("Initializing searcher...")
|
9f96d6f3
tangwang
短query不用语义搜索
|
145
|
_searcher = Searcher(_es_client, _config, _query_parser)
|
ded6f29e
tangwang
补充suggestion模块
|
146
147
|
logger.info("Initializing suggestion service...")
_suggestion_service = SuggestionService(_es_client)
|
325eec03
tangwang
1. 日志、配置基础设施,使用优化
|
148
|
|
325eec03
tangwang
1. 日志、配置基础设施,使用优化
|
149
|
elapsed = time.time() - start_time
|
9f96d6f3
tangwang
短query不用语义搜索
|
150
|
logger.info(f"Search service ready! (took {elapsed:.2f}s) | Index: {_config.es_index_name}")
|
be52af70
tangwang
first commit
|
151
152
|
|
be52af70
tangwang
first commit
|
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
|
def get_es_client() -> ESClient:
"""Get Elasticsearch client."""
if _es_client is None:
raise RuntimeError("Service not initialized")
return _es_client
def get_searcher() -> Searcher:
"""Get searcher instance."""
if _searcher is None:
raise RuntimeError("Service not initialized")
return _searcher
def get_query_parser() -> QueryParser:
"""Get query parser instance."""
if _query_parser is None:
raise RuntimeError("Service not initialized")
return _query_parser
|
ded6f29e
tangwang
补充suggestion模块
|
176
177
178
179
180
181
182
|
def get_suggestion_service() -> SuggestionService:
"""Get suggestion service instance."""
if _suggestion_service is None:
raise RuntimeError("Service not initialized")
return _suggestion_service
|
9f96d6f3
tangwang
短query不用语义搜索
|
183
184
185
186
187
188
189
|
def get_config():
"""Get global config instance."""
if _config is None:
raise RuntimeError("Service not initialized")
return _config
|
bb3c5ef8
tangwang
灌入数据流程跑通
|
190
|
# Create FastAPI app with enhanced configuration
|
be52af70
tangwang
first commit
|
191
192
193
|
app = FastAPI(
title="E-Commerce Search API",
description="Configurable search engine for cross-border e-commerce",
|
bb3c5ef8
tangwang
灌入数据流程跑通
|
194
195
196
197
198
199
200
201
202
203
204
205
206
207
|
version="1.0.0",
docs_url="/docs",
redoc_url="/redoc",
openapi_url="/openapi.json"
)
# Add rate limiting middleware
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)
# Add trusted host middleware (restrict to localhost and trusted domains)
app.add_middleware(
TrustedHostMiddleware,
allowed_hosts=["*"] # Allow all hosts for development, restrict in production
|
be52af70
tangwang
first commit
|
208
209
|
)
|
bb3c5ef8
tangwang
灌入数据流程跑通
|
210
211
212
213
214
215
216
217
218
219
220
|
# Add security headers middleware
@app.middleware("http")
async def add_security_headers(request: Request, call_next):
response = await call_next(request)
response.headers["X-Content-Type-Options"] = "nosniff"
response.headers["X-Frame-Options"] = "DENY"
response.headers["X-XSS-Protection"] = "1; mode=block"
response.headers["Referrer-Policy"] = "strict-origin-when-cross-origin"
return response
# Add CORS middleware with more restrictive settings
|
be52af70
tangwang
first commit
|
221
222
|
app.add_middleware(
CORSMiddleware,
|
bb3c5ef8
tangwang
灌入数据流程跑通
|
223
|
allow_origins=["*"], # Restrict in production to specific domains
|
be52af70
tangwang
first commit
|
224
|
allow_credentials=True,
|
bb3c5ef8
tangwang
灌入数据流程跑通
|
225
|
allow_methods=["GET", "POST", "PUT", "DELETE", "OPTIONS"],
|
be52af70
tangwang
first commit
|
226
|
allow_headers=["*"],
|
bb3c5ef8
tangwang
灌入数据流程跑通
|
227
|
expose_headers=["X-Total-Count"]
|
be52af70
tangwang
first commit
|
228
229
230
231
232
233
|
)
@app.on_event("startup")
async def startup_event():
"""Initialize service on startup."""
|
f62a541c
tangwang
将 uvicorn 的默认线程池调...
|
234
235
236
237
238
239
240
241
242
|
# Configure thread pool size for uvicorn (default is 40, set to 48)
try:
import anyio.to_thread
limiter = anyio.to_thread.current_default_thread_limiter()
limiter.total_tokens = 48
logger.info(f"Thread pool size set to {limiter.total_tokens}")
except Exception as e:
logger.warning(f"Failed to set thread pool size: {e}, using default")
|
be52af70
tangwang
first commit
|
243
|
es_host = os.getenv("ES_HOST", "http://localhost:9200")
|
4d824a77
tangwang
所有租户共用一套统一配置.tena...
|
244
|
logger.info("Starting E-Commerce Search API (Multi-Tenant)")
|
bb3c5ef8
tangwang
灌入数据流程跑通
|
245
|
logger.info(f"Elasticsearch Host: {es_host}")
|
3c1f8031
tangwang
api/routes/indexe...
|
246
|
|
be52af70
tangwang
first commit
|
247
|
try:
|
4d824a77
tangwang
所有租户共用一套统一配置.tena...
|
248
|
init_service(es_host=es_host)
|
bb3c5ef8
tangwang
灌入数据流程跑通
|
249
|
logger.info("Service initialized successfully")
|
be52af70
tangwang
first commit
|
250
|
except Exception as e:
|
3c1f8031
tangwang
api/routes/indexe...
|
251
|
logger.error(f"Failed to initialize service: {e}", exc_info=True)
|
26b910bd
tangwang
refactor service ...
|
252
|
raise
|
bb3c5ef8
tangwang
灌入数据流程跑通
|
253
254
255
256
257
258
|
@app.on_event("shutdown")
async def shutdown_event():
"""Cleanup on shutdown."""
logger.info("Shutting down E-Commerce Search API")
|
be52af70
tangwang
first commit
|
259
260
261
262
|
@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
|
bb3c5ef8
tangwang
灌入数据流程跑通
|
263
264
265
266
|
"""Global exception handler with detailed logging."""
client_ip = request.client.host if request.client else "unknown"
logger.error(f"Unhandled exception from {client_ip}: {exc}", exc_info=True)
|
be52af70
tangwang
first commit
|
267
268
269
270
|
return JSONResponse(
status_code=500,
content={
"error": "Internal server error",
|
bb3c5ef8
tangwang
灌入数据流程跑通
|
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
|
"detail": "An unexpected error occurred. Please try again later.",
"timestamp": int(time.time())
}
)
@app.exception_handler(HTTPException)
async def http_exception_handler(request: Request, exc: HTTPException):
"""HTTP exception handler."""
logger.warning(f"HTTP exception from {request.client.host if request.client else 'unknown'}: {exc.status_code} - {exc.detail}")
return JSONResponse(
status_code=exc.status_code,
content={
"error": exc.detail,
"status_code": exc.status_code,
"timestamp": int(time.time())
|
be52af70
tangwang
first commit
|
288
289
290
291
|
}
)
|
a7a8c6cb
tangwang
测试过滤、聚合、排序
|
292
|
@app.get("/api")
|
bb3c5ef8
tangwang
灌入数据流程跑通
|
293
294
295
296
297
298
|
@limiter.limit("60/minute")
async def root(request: Request):
"""Root endpoint with rate limiting."""
client_ip = request.client.host if request.client else "unknown"
logger.info(f"Root endpoint accessed from {client_ip}")
|
be52af70
tangwang
first commit
|
299
300
301
|
return {
"service": "E-Commerce Search API",
"version": "1.0.0",
|
bb3c5ef8
tangwang
灌入数据流程跑通
|
302
303
|
"status": "running",
"timestamp": int(time.time())
|
be52af70
tangwang
first commit
|
304
305
306
|
}
|
bb3c5ef8
tangwang
灌入数据流程跑通
|
307
308
309
310
311
312
|
@app.get("/health")
@limiter.limit("120/minute")
async def health_check(request: Request):
"""Health check endpoint."""
try:
# Check if services are initialized
|
bb3c5ef8
tangwang
灌入数据流程跑通
|
313
|
get_es_client()
|
bf89b597
tangwang
feat(search): ada...
|
314
|
get_searcher()
|
bb3c5ef8
tangwang
灌入数据流程跑通
|
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
|
return {
"status": "healthy",
"services": {
"config": "initialized",
"elasticsearch": "connected",
"searcher": "initialized"
},
"timestamp": int(time.time())
}
except Exception as e:
logger.error(f"Health check failed: {e}")
return JSONResponse(
status_code=503,
content={
"status": "unhealthy",
"error": str(e),
"timestamp": int(time.time())
}
)
|
bb9c626c
tangwang
搜索服务(6002)不再初始化/挂...
|
337
338
|
# Include routers (search app should NOT mount indexer routes)
from .routes import search, admin
|
be52af70
tangwang
first commit
|
339
340
341
|
app.include_router(search.router)
app.include_router(admin.router)
|
be52af70
tangwang
first commit
|
342
|
|
a7a8c6cb
tangwang
测试过滤、聚合、排序
|
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
|
# Mount static files and serve frontend
frontend_path = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "frontend")
if os.path.exists(frontend_path):
# Serve frontend HTML at root
@app.get("/")
async def serve_frontend():
"""Serve the frontend HTML."""
index_path = os.path.join(frontend_path, "index.html")
if os.path.exists(index_path):
return FileResponse(index_path)
return {"service": "E-Commerce Search API", "version": "1.0.0", "status": "running"}
# Mount static files (CSS, JS, images)
app.mount("/static", StaticFiles(directory=os.path.join(frontend_path, "static")), name="static")
logger.info(f"Frontend static files mounted from: {frontend_path}")
else:
logger.warning(f"Frontend directory not found: {frontend_path}")
|
be52af70
tangwang
first commit
|
362
363
|
if __name__ == "__main__":
|
4d824a77
tangwang
所有租户共用一套统一配置.tena...
|
364
|
parser = argparse.ArgumentParser(description='Start search API service (multi-tenant)')
|
be52af70
tangwang
first commit
|
365
|
parser.add_argument('--host', default='0.0.0.0', help='Host to bind to')
|
2a76641e
tangwang
config
|
366
|
parser.add_argument('--port', type=int, default=6002, help='Port to bind to')
|
be52af70
tangwang
first commit
|
367
368
369
370
|
parser.add_argument('--es-host', default='http://localhost:9200', help='Elasticsearch host')
parser.add_argument('--reload', action='store_true', help='Enable auto-reload')
args = parser.parse_args()
|
4d824a77
tangwang
所有租户共用一套统一配置.tena...
|
371
|
# Set environment variable
|
be52af70
tangwang
first commit
|
372
373
374
375
376
377
378
379
380
|
os.environ['ES_HOST'] = args.es_host
# Run server
uvicorn.run(
"api.app:app",
host=args.host,
port=args.port,
reload=args.reload
)
|