app.py

import subprocess

from fastapi import FastAPI, BackgroundTasks
from pydantic import BaseModel
 
from fastapi.middleware.cors import CORSMiddleware

from util.search_data import *
from util.preprocessing import *
from test_server import update_DB

from apscheduler.schedulers.asyncio import AsyncIOScheduler
from apscheduler.triggers.cron import CronTrigger

logs = []
app = FastAPI()

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

scheduler = AsyncIOScheduler()

async def run_test_server_script():
    subprocess.run(["python3", "test_server.py"])

async def startup_event():
    # 매일 05시에 `test_server.py` 실행
    scheduler.add_job(run_test_server_script, CronTrigger(hour=5, minute=0))
    scheduler.start()

def save_log(text):
    global logs
    logs.append(text)

app.add_event_handler("startup", startup_event)

@app.get("/", summary="Test page. Return 'Hellow world' message.", tags=["Hello world"])
async def Hello_world():
    save_log(f"GET hello world")
    return {"detail": "Hello World"}


@app.get("/logs", summary="Test page. Return logs", tags=["log"])
def open_log():
    global logs
    if len(logs) > 10:
        logs = logs[-5:]
    return {"detail": "logs", "data":logs}

@app.get("/word_filter/{word}", summary="검색어와 가장 유사한 시장명 및 품목명 반환", tags=["search"])
async def word_filter(word: str):
    
    data = {
        "detail": "ok",
        "data" : {
            "market" : word_to_market_name(word),
            "product" : word_to_product_name(word)
        }
    }
    save_log(f"GET /word_filter/{word}")
    return data

@app.get("/near_market/{lat}/{lon}", summary="/위도/경도 정보로 가까운 시장 10개 반환", tags=["search"])
async def near_market(lat: float, lon: float):
    nearest_markets = find_nearest_markets(lat, lon)
    data={
        "detail": "ok",
        "data" : nearest_markets
    }
    save_log(f"GET /near_market/{lat}/{lon}")
    return data

@app.get("/market/{marketname}", summary="시장 정보 반환", tags=["search"])
async def get_market(marketname: str):
    if not check_word(marketname):
        return {"detail":"fail", "message":f"{marketname}는 잘못된 시장명입니다"}

    data = dict()
    index, place_name, dist = find_nearest_place(marketname)
    lat, lon = get_market_location(marketname)
    size, area, store_count = determine_market_size(marketname)
    congestion, explain = get_congestion_data(place_name, index)
    data={
        "detail": "ok",
        "data":{
            "시장" : marketname,
            "위치" : [lat, lon],
            "규모" : size,
            "규모 정보" : {
                "면적" : area,
                "점포수" : store_count
            },
            "조회" : f"{marketname}과(와) 가장 가까운 조회 가능한 장소는 {place_name}입니다.(거리는 {int(dist*100)/100}km)", 
            "혼잡도" : congestion,
            "설명" : explain,
            "최저가격" : get_cheapest_products_this_month(marketname),
            'text': "최근 2개월 내 조사된 서울 열린데이터 광장의 서울시 생필품 농수축산물 가격 정보를 바탕으로 계산되었습니다."
        }
    }
    save_log(f"GET /market/{marketname}")
    return data

@app.get("/product/{product_name}", summary="상품 정보 반환", tags=["search"])
async def get_market(product_name: str):
    if not check_product(product_name):
        return {"detail":"fail", "message":f"{product_name}는 잘못된 상품명입니다"}
    monthly_prices = get_monthly_average_price(product_name)
    lowest_price_markets = get_lowest_price_markets(product_name)
    season_data = get_season_data_by_name(product_name)
    data={
        "detail": "ok",
        "data":{
            "상품" : product_name,
            "제철정보" : season_data,
            "최근 시세" : monthly_prices[0],
            "차트" : monthly_prices,
            "최저가 시장" : lowest_price_markets,
            'text': "최근 2개월 내 조사된 서울 열린데이터 광장의 서울시 생필품 농수축산물 가격 정보를 바탕으로 계산되었습니다."
        }
    }
    save_log(f"GET /product/{product_name}")
    return data