o2o-castad-frontend/test/server.py

179 lines
5.1 KiB
Python

"""
네이버 플레이스 검색 테스트 웹 서버
Flask를 사용하여 검색 및 상세정보 조회 API 제공
"""
import asyncio
import sys
from io import StringIO
from flask import Flask, render_template, request, jsonify
from main import NaverPlaceAPI
app = Flask(__name__)
place_api = NaverPlaceAPI()
# ============================================================
# Utilities
# ============================================================
class LogCapture:
"""콘솔 출력을 캡처하는 컨텍스트 매니저"""
def __init__(self):
self.logs = []
self._stdout = None
self._capture = None
def __enter__(self):
self._stdout = sys.stdout
sys.stdout = self._capture = StringIO()
return self
def __exit__(self, *args):
self.logs = [log for log in self._capture.getvalue().split('\n') if log.strip()]
sys.stdout = self._stdout
def get_logs(self):
return self.logs
def run_async(coro):
"""비동기 함수를 동기적으로 실행"""
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
try:
return loop.run_until_complete(coro)
finally:
loop.close()
# ============================================================
# Routes - Pages
# ============================================================
@app.route('/')
def index():
"""검색 페이지"""
return render_template('search.html')
@app.route('/result')
def result():
"""결과 페이지"""
return render_template('result.html')
# ============================================================
# Routes - API
# ============================================================
@app.route('/api/autocomplete', methods=['POST'])
def api_autocomplete():
"""
빠른 자동완성 API (place_id 없음)
Request: {"query": "스테이"}
Response: {"results": [{"title": "...", "category": "...", "address": "..."}], "count": 10}
"""
try:
data = request.get_json()
query = data.get('query', '').strip()
if not query or len(query) < 2:
return jsonify({'results': []})
results = run_async(place_api.quick_search(query))
return jsonify({'results': results, 'count': len(results)})
except Exception as e:
return jsonify({'results': [], 'error': str(e)})
@app.route('/api/search', methods=['POST'])
def api_search():
"""
검색 API (place_id 포함)
Request: {"query": "스테이 머뭄"}
Response: {"results": [{"place_id": "123", "title": "...", ...}], "count": 5, "logs": [...]}
"""
try:
data = request.get_json()
query = data.get('query', '').strip()
if not query:
return jsonify({'error': '검색어를 입력해주세요.'}), 400
with LogCapture() as log_capture:
results = run_async(place_api.autocomplete_search(query))
return jsonify({
'results': results,
'count': len(results),
'logs': log_capture.get_logs(),
'query': query
})
except Exception as e:
import traceback
return jsonify({'error': str(e), 'trace': traceback.format_exc()}), 500
@app.route('/api/detail', methods=['POST'])
def api_detail():
"""
상세정보 API
Request: {"place_id": "1133638931"}
Response: {"detail": {...}, "crawling_response": {...}, "logs": [...]}
"""
try:
data = request.get_json()
place_id = data.get('place_id', '').strip()
if not place_id:
return jsonify({'error': 'place_id를 입력해주세요.'}), 400
with LogCapture() as log_capture:
detail = run_async(place_api.get_place_detail(place_id))
if not detail:
return jsonify({'error': '상세 정보를 찾을 수 없습니다.', 'logs': log_capture.get_logs()}), 404
return jsonify({
'detail': {
'place_id': detail.place_id,
'name': detail.name,
'category': detail.category,
'address': detail.address,
'road_address': detail.road_address,
'phone': detail.phone,
'description': detail.description,
'images': detail.images,
'business_hours': detail.business_hours,
'homepage': detail.homepage,
'keywords': detail.keywords,
'facilities': detail.facilities,
},
'crawling_response': place_api.convert_to_crawling_response(detail),
'logs': log_capture.get_logs()
})
except Exception as e:
import traceback
return jsonify({'error': str(e), 'trace': traceback.format_exc()}), 500
# ============================================================
# Entry Point
# ============================================================
if __name__ == '__main__':
print("=" * 60)
print("네이버 플레이스 검색 테스트 서버")
print("=" * 60)
print("브라우저에서 http://localhost:5001 으로 접속하세요")
print("=" * 60)
app.run(debug=True, port=5001, use_reloader=False)