주간 예상방문객 추출

lib/weekly_visitor_forecast.py

@@ -7,8 +7,8 @@ from collections import defaultdict
 import pandas as pd
 from sqlalchemy import select, func
 
-from weather_forecast import get_weekly_precip
 from conf import db, db_schema
+from lib.weather_forecast import get_weekly_precip
 from lib.holiday import is_korean_holiday
 from lib.common import load_config
 
@@ -22,12 +22,50 @@ pos = db_schema.pos
 
 engine = db.engine 
 
+def get_recent_dataframe(today=None) -> pd.DataFrame:
+    today = today or date.today()
+    weekday = today.weekday()
+    sunday = today + timedelta(days=(6 - weekday))
+    recent_dates = [sunday - timedelta(days=i) for i in reversed(range(14))]
+
+    recent_data = fetch_data_for_dates(recent_dates)
+
+    # 결측 강수량 보정
+    weekly_precip = get_weekly_precip(config['DATA_API']['serviceKey'])
+    for d in recent_dates:
+        if d >= today and (d not in recent_data or '강수량' not in recent_data[d]):
+            dt_str = d.strftime('%Y%m%d')
+            if dt_str in weekly_precip:
+                recent_data[d] = recent_data.get(d, {})
+                recent_data[d]['강수량'] = round(float(weekly_precip[dt_str]['sumRn']), 1)
+                recent_data[d]['최저기온'] = round(float(weekly_precip[dt_str]['minTa']), 1)
+                recent_data[d]['최고기온'] = round(float(weekly_precip[dt_str]['maxTa']), 1)
+                recent_data[d]['습도'] = round(float(weekly_precip[dt_str]['avgRhm']), 1)
+
+    # prophet 예측값 병합
+    prophet_forecast = load_prophet_forecast()
+    for d in recent_dates:
+        d_ts = pd.Timestamp(d)
+        if d >= today and d_ts in prophet_forecast.index:
+            recent_data[d] = recent_data.get(d, {})
+            recent_data[d]['예상 방문자'] = round(float(prophet_forecast.loc[d_ts]), 0)
+
+    return build_dataframe(recent_dates, recent_data, use_forecast_after=today)
+
+def get_last_year_dataframe(today=None) -> pd.DataFrame:
+    today = today or date.today()
+    weekday = today.weekday()
+    sunday = today + timedelta(days=(6 - weekday))
+    recent_dates = [sunday - timedelta(days=i) for i in reversed(range(14))]
+    prev_year_dates = get_last_year_same_weekdays(recent_dates)
+
+    prev_year_data = fetch_data_for_dates(prev_year_dates)
+    return build_dataframe(prev_year_dates, prev_year_data)
 
 def get_recent_dates(today=None, days=14):
     today = today or date.today()
     return [today - timedelta(days=i) for i in reversed(range(days))]
 
-
 def get_this_week_dates(today=None):
     today = today or date.today()
     weekday = today.weekday()
@@ -205,7 +243,7 @@ def main():
     recent_dates = [sunday - timedelta(days=i) for i in reversed(range(14))]
     prev_year_dates = get_last_year_same_weekdays(recent_dates)
 
-    # 이번 주 예상 대상 (오늘부터 일요일까지)
+    # 이번 주 예상 대상 (오늘부터 일요일까지    )
     this_week_dates = [today + timedelta(days=i) for i in range(7 - weekday)]
 
     # 데이터 조회
@@ -228,20 +266,8 @@ def main():
 
     # prophet 예측 결과 불러오기 및 이번 주 예상 데이터에 병합
     prophet_forecast = load_prophet_forecast()
-    for d in this_week_dates:
-        d_ts = pd.Timestamp(d)
-        has_forecast = d_ts in prophet_forecast.index
-        print(f"[DEBUG] 날짜 {d} (Timestamp {d_ts}) 예측 데이터 존재 여부: {has_forecast}")
-        if has_forecast:
-            if d not in forecast_data:
-                forecast_data[d] = {}
-            forecast_data[d]['예상 방문자'] = round(float(prophet_forecast.loc[d_ts]), 0)
-        else:
-            if d not in forecast_data:
-                forecast_data[d] = {}
-            forecast_data[d]['예상 방문자'] = None
 
-    # 최근 2주 데이터에도 오늘 이후 날짜에 대해 예상 방문자 병합
+    # 최근 2주 데이터에 오늘 이후 날짜에 대해 예상 방문자 병합
     for d in recent_dates:
         d_ts = pd.Timestamp(d)
         if d >= today and d_ts in prophet_forecast.index:
@@ -264,6 +290,19 @@ def main():
     print("\n📈 작년 동일 요일 데이터:")
     print(df_prev.to_string(index=False))
 
+    # 🔽 엑셀 파일로 저장
+    output_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), '..', 'output'))
+    os.makedirs(output_dir, exist_ok=True)
+
+    recent_excel_path = os.path.join(output_dir, 'recent_visitors.xlsx')
+    prev_excel_path = os.path.join(output_dir, 'lastyear_visitors.xlsx')
+
+    df_recent.to_excel(recent_excel_path, index=False)
+    df_prev.to_excel(prev_excel_path, index=False)
+
+    print(f"\n📁 엑셀 파일 저장 완료:")
+    print(f" - 최근 2주: {recent_excel_path}")
+    print(f" - 작년 동일 요일: {prev_excel_path}")
 
 if __name__ == "__main__":
     main()