← 메인 화면프로젝트 자료·코드
← 자료 목록

cc/startup_app/app.py

프로그램·코드 · 4.7 KB원본 다운로드

원본 내용을 읽기 전용으로 표시합니다.

  1. """세종 상권 살펴보기 API. 운영 환경에서는 waitress와 HTTPS 프록시 사용."""
  2. import logging
  3. import math
  4. import os
  5. from pathlib import Path
  6. from functools import lru_cache
  7. for key in ("OMP_NUM_THREADS", "OPENBLAS_NUM_THREADS", "MKL_NUM_THREADS"):
  8. os.environ.setdefault(key, "1")
  9. import numpy as np
  10. from flask import Flask, jsonify, render_template, request
  11. from werkzeug.exceptions import HTTPException
  12. try:
  13. from .engine import 엔진
  14. except ImportError:
  15. from engine import 엔진
  18. def 숫자정리(value):
  19. if isinstance(value, dict):
  20. return {key: 숫자정리(item) for key, item in value.items()}
  21. if isinstance(value, (list, tuple)):
  22. return [숫자정리(item) for item in value]
  23. if isinstance(value, np.generic):
  24. value = value.item()
  25. return None if isinstance(value, float) and not math.isfinite(value) else value
  28. @lru_cache(maxsize=1)
  29. def default_engine():
  30. return 엔진()
  33. def create_app(engine=None):
  34. app = Flask(__name__)
  35. from .materials import bp
  36. app.config['MATERIALS_DIR'] = Path(__file__).resolve().parents[1] / 'public_materials'
  37. app.register_blueprint(bp)
  38. app.json.ensure_ascii = False
  39. app.config["MAX_CONTENT_LENGTH"] = 16 * 1024
  40. if engine is not None:
  41. app.extensions["sejong_engine"] = engine
  42. def model():
  43. return app.extensions.get("sejong_engine") or default_engine()
  44. def parameters(category_required=False):
  45. data = request.get_json(silent=True)
  46. if not isinstance(data, dict) or isinstance(data.get("lat"), bool) or isinstance(data.get("lon"), bool):
  47. return None
  48. try:
  49. lat, lon = float(data["lat"]), float(data["lon"])
  50. except (KeyError, ValueError, TypeError, OverflowError):
  51. return None
  52. if not model().위치확인(lat, lon):
  53. return None
  54. category = data.get("mid")
  55. if category_required and (not isinstance(category, str) or category not in model().mid_to_cat):
  56. return None
  57. return round(lat, 6), round(lon, 6), category
  58. @app.get("/")
  59. def index():
  60. return render_template("index.html")
  61. @app.get("/healthz")
  62. def health():
  63. current = model()
  64. return jsonify({"status": "ok", "rows": len(current.now), "model_count": len(current.모델), "version": "2"})
  65. @app.get("/api/meta")
  66. def metadata():
  67. response = jsonify(숫자정리(model().메타()))
  68. response.headers["Cache-Control"] = "public, max-age=300"
  69. return response
  70. @app.get("/api/boundary")
  71. def boundary():
  72. return jsonify(model().boundary or {"type": "FeatureCollection", "features": []})
  73. @app.post("/api/analyze")
  74. def analyze():
  75. values = parameters(True)
  76. if values is None:
  77. return jsonify({"오류": "세종시 경계 안의 위치와 목록에 있는 업종을 골라 주세요."}), 400
  78. return jsonify(숫자정리(model().분석(*values)))
  79. @app.post("/api/recommend")
  80. def compare_categories():
  81. values = parameters()
  82. if values is None:
  83. return jsonify({"오류": "세종시 경계 안의 위치를 골라 주세요."}), 400
  84. return jsonify(숫자정리(model().업종추천(values[0], values[1])))
  85. @app.after_request
  86. def headers(response):
  87. response.headers.update({"X-Content-Type-Options": "nosniff", "Referrer-Policy": "strict-origin-when-cross-origin", "X-Frame-Options": "SAMEORIGIN"})
  88. response.headers.setdefault("Content-Security-Policy", "default-src 'self'; script-src 'self'; style-src 'self' 'unsafe-inline'; img-src 'self' data: https://*.tile.openstreetmap.org https://*.tile.openstreetmap.fr; connect-src 'self'; frame-src 'self'; object-src 'none'; base-uri 'self'; frame-ancestors 'self'; form-action 'self'")
  89. if request.path.startswith("/api/") and request.method == "POST":
  90. response.headers["Cache-Control"] = "no-store"
  91. return response
  92. @app.errorhandler(HTTPException)
  93. def http_error(error):
  94. messages = {404: "요청한 주소가 없습니다.", 405: "지원하지 않는 요청 방식입니다.", 413: "요청 내용이 너무 큽니다."}
  95. return jsonify({"오류": messages.get(error.code, "요청을 처리하지 못했습니다.")}), error.code
  96. @app.errorhandler(Exception)
  97. def unexpected(error):
  98. logging.getLogger(__name__).exception("요청 처리 실패")
  99. return jsonify({"오류": "일시적으로 분석하지 못했습니다. 잠시 후 다시 시도해 주세요."}), 500
  100. return app
  103. app = create_app()
  104. if __name__ == "__main__":
  105. from waitress import serve
  106. logging.basicConfig(level=logging.INFO)
  107. default_engine()
  108. serve(app, host=os.environ.get("HOST", "127.0.0.1"), port=int(os.environ.get("PORT", "5000")), threads=4)
파일 확인 정보

SHA-256: 93264d6286c5e7a0559efa06b35c6b587c47564f1ff3c966a671aab3a969f133