생성된 백업이 없습니다.
';
+ } else {
+ data.backups.forEach(backup => {
+ const sizeInMB = formatFileSize(backup.size);
+ html += `
+ ' :
+ '
';
+
+ resultHtml += '업로드 완료!
';
+
+ data.files.forEach(file => {
+ const icon = file.status === 'success' ? '✓' : '✗';
+ resultHtml += `${icon} ${file.filename}: ${file.message}
`;
+ });
+
+ resultHtml += '
';
+ document.getElementById('upload-result').innerHTML = resultHtml;
+
+ showAlert('업로드가 완료되었습니다.', 'success');
+
+ setTimeout(() => {
+ clearFileList();
+ // 대시보드 새로고침
+ if (typeof loadDashboard === 'function') {
+ loadDashboard();
+ }
+ }, 2000);
+ } catch (error) {
+ showAlert('업로드 실패: ' + error.message, 'danger');
+ } finally {
+ document.getElementById('upload-progress').style.display = 'none';
+ document.getElementById('upload-btn').disabled = false;
+ }
+}
+
+/**
+ * 시스템 상태를 확인합니다.
+ */
+async function checkSystemStatus() {
+ try {
+ const data = await apiCall('/api/status');
+
+ const dbStatus = document.getElementById('db-status');
+ const uploadStatus = document.getElementById('upload-folder-status');
+
+ if (dbStatus) {
+ dbStatus.textContent = data.database ? '연결됨' : '연결 안됨';
+ dbStatus.className = `badge ${data.database ? 'bg-success' : 'bg-danger'}`;
+ }
+
+ if (uploadStatus) {
+ uploadStatus.textContent = data.upload_folder ? '정상' : '오류';
+ uploadStatus.className = `badge ${data.upload_folder ? 'bg-success' : 'bg-danger'}`;
+ }
+ } catch (error) {
+ console.error('시스템 상태 확인 실패:', error);
+ }
+}
diff --git a/app/templates/backup.html b/app/templates/backup.html
new file mode 100644
index 0000000..8d4e55f
--- /dev/null
+++ b/app/templates/backup.html
@@ -0,0 +1,33 @@
+{% extends "base.html" %}
+
+{% block title %}백업 관리 - First Garden POS{% endblock %}
+
+{% block extra_css %}
+
+{% endblock %}
+
+{% block content %}
+
+
+
+
+
+
+
+
+
+{% endblock %}
+
+{% block extra_js %}
+
+
+{% endblock %}
diff --git a/app/templates/base.html b/app/templates/base.html
new file mode 100644
index 0000000..a9fdcee
--- /dev/null
+++ b/app/templates/base.html
@@ -0,0 +1,72 @@
+
+
+
+
+
+
{% block title %}First Garden - POS 데이터 관리{% endblock %}
+
+
+
+
+
+
+
+
+
+
+
+ {% block extra_css %}{% endblock %}
+
+
+
+
+
+
+
+
+
+
+
+
+ {% block content %}{% endblock %}
+
+
+
+
+
+
+
+
+
+
+
+
+ {% block extra_js %}{% endblock %}
+
+
+
diff --git a/app/templates/dashboard.html b/app/templates/dashboard.html
new file mode 100644
index 0000000..cd86f57
--- /dev/null
+++ b/app/templates/dashboard.html
@@ -0,0 +1,108 @@
+{% extends "base.html" %}
+
+{% block title %}대시보드 - First Garden POS{% endblock %}
+
+{% block extra_css %}
+
+{% endblock %}
+
+{% block content %}
+
+
+
+
+
+
OKPOS 상품별
+
-
+
총 데이터
+
-
+
최종: -
+
+
+
+
+
OKPOS 영수증
+
-
+
총 데이터
+
-
+
최종: -
+
+
+
+
+
UPSolution
+
-
+
총 데이터
+
-
+
최종: -
+
+
+
+
+
날씨 데이터
+
-
+
총 데이터
+
-
+
최종: -
+
+
+
+
+
+
+
+ 이번주 예상 날씨 & 방문객
+
+
+
+
+
+ | 날짜 |
+ 최저기온 |
+ 최고기온 |
+ 강수량 |
+ 습도 |
+ 예상 방문객 |
+
+
+
+
+ | 데이터 로딩 중... |
+
+
+
+
+
+
+
+
+
+ 방문객 추이
+
+
+
+
+
+ ~
+
+
+
+
+
+
+
+
+
+
+
+{% endblock %}
+
+{% block extra_js %}
+
+
+{% endblock %}
diff --git a/app/templates/index.html b/app/templates/index.html
new file mode 100644
index 0000000..c5d77a9
--- /dev/null
+++ b/app/templates/index.html
@@ -0,0 +1,867 @@
+
+
+
+
+
+
First Garden - POS 데이터 대시보드
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+ -
+
+
+ -
+
+
+ -
+
+
+
+
+
+
+
+
+
+
+
+
+
OKPOS 상품별
+
-
+
총 데이터
+
-
+
최종: -
+
+
+
+
+
OKPOS 영수증
+
-
+
총 데이터
+
-
+
최종: -
+
+
+
+
+
UPSolution
+
-
+
총 데이터
+
-
+
최종: -
+
+
+
+
+
날씨 데이터
+
-
+
총 데이터
+
-
+
최종: -
+
+
+
+
+
+
+
+ 이번주 예상 날씨 & 방문객
+
+
+
+
+
+ | 날짜 |
+ 최저기온 |
+ 최고기온 |
+ 강수량 |
+ 습도 |
+ 예상 방문객 |
+
+
+
+
+ | 데이터 로딩 중... |
+
+
+
+
+
+
+
+
+
+ 방문객 추이
+
+
+
+
+
+ ~
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+ 시스템 상태:
+ 데이터베이스: 연결 중...
+ 업로드 폴더: 확인 중...
+
+
+
+
+
+
파일을 여기에 드래그하세요
+
또는
+
+
+ 지원 형식: OKPOS (일자별 상품별, 영수증별매출상세현황), UPSOLUTION
+ 최대 파일 크기: 100MB
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
diff --git a/app/templates/upload.html b/app/templates/upload.html
new file mode 100644
index 0000000..a023035
--- /dev/null
+++ b/app/templates/upload.html
@@ -0,0 +1,69 @@
+{% extends "base.html" %}
+
+{% block title %}파일 업로드 - First Garden POS{% endblock %}
+
+{% block extra_css %}
+
+{% endblock %}
+
+{% block content %}
+
+
+
+
+ 시스템 상태:
+ 데이터베이스: 연결 중...
+ 업로드 폴더: 확인 중...
+
+
+
+
+
+
파일을 여기에 드래그하세요
+
또는
+
+
+ 지원 형식: OKPOS (일자별 상품별, 영수증별매출상세현황), UPSOLUTION
+ 최대 파일 크기: 100MB
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+{% endblock %}
+
+{% block extra_js %}
+
+
+{% endblock %}
diff --git a/build/Dockerfile b/build/Dockerfile
index 699a0e2..d578db5 100644
--- a/build/Dockerfile
+++ b/build/Dockerfile
@@ -1,47 +1,129 @@
-FROM python:3.10-slim
+# Dockerfile - First Garden Static Analysis Service
+# Python 3.11 기반의 가벼운 이미지 사용
+
+FROM python:3.11-slim-bullseye
+
+# 메타데이터 설정
+LABEL maintainer="First Garden Team"
+LABEL description="First Garden Static Analysis - Data Collection & Visitor Forecasting Service"
# 작업 디렉토리 설정
WORKDIR /app
-# 시스템 패키지 설치
+# 타임존 설정
+ENV TZ=Asia/Seoul
+ENV PYTHONUNBUFFERED=1
+ENV PYTHONDONTWRITEBYTECODE=1
+
+# 시스템 패키지 업데이트 및 필수 도구 설치
+# mysqldump 및 mysql 클라이언트 추가 (DB 백업/복구 용)
RUN apt-get update && apt-get install -y --no-install-recommends \
- build-essential \
gcc \
- libsqlite3-dev \
+ g++ \
+ build-essential \
+ libmysqlclient-dev \
libssl-dev \
libffi-dev \
- libbz2-dev \
- libreadline-dev \
- libncurses5-dev \
- libgdbm-dev \
- liblzma-dev \
- libtk8.6 \
- tk8.6-dev \
- tcl8.6-dev \
- wget \
curl \
- unzip \
+ wget \
git \
cron \
- && rm -rf /var/lib/apt/lists/*
+ mariadb-client \
+ && rm -rf /var/lib/apt/lists/* \
+ && apt-get clean
-# requirements 설치
+# Python 의존성 설치
COPY requirements.txt ./
-RUN pip install --no-cache-dir -r requirements.txt
+RUN pip install --no-cache-dir --upgrade pip setuptools wheel && \
+ pip install --no-cache-dir -r requirements.txt
-# 앱 전체 복사
+# 앱 코드 복사
COPY . .
-# 환경 변수 설정
-ENV PYTHONUNBUFFERED=1
+# 로그 디렉토리 생성
+RUN mkdir -p /app/logs /app/data /app/output /app/uploads /app/dbbackup && \
+ chmod 755 /app/logs /app/data /app/output /app/uploads /app/dbbackup
-# 크론 작업 등록: 매일 11시에 daily_run.py 실행
-RUN echo "0 11 * * * python /app/daily_run.py >> /var/log/cron.log 2>&1" > /etc/cron.d/daily-cron \
- && chmod 0644 /etc/cron.d/daily-cron \
- && crontab /etc/cron.d/daily-cron
+# 크론 작업 설정: 매일 11시 UTC (서울 시간 20시)에 daily_run.py 실행
+RUN echo "0 11 * * * cd /app && python daily_run.py >> /app/logs/daily_run.log 2>&1" > /etc/cron.d/daily-forecast && \
+ chmod 0644 /etc/cron.d/daily-forecast && \
+ crontab /etc/cron.d/daily-forecast
-# 로그 파일 생성
-RUN touch /var/log/cron.log
+# 헬스체크 스크립트 생성
+RUN cat > /app/healthcheck.sh << 'EOF'
+#!/bin/bash
+set -e
-# 컨테이너 시작 시 cron 실행 + file_watch.py 실행 + 로그 출력 유지
-CMD cron && python lib/file_watch.py & tail -f /var/log/cron.log
+# DB 연결 확인
+python -c "
+import sys
+sys.path.insert(0, '/app')
+from conf import db
+try:
+ session = db.get_session()
+ session.execute('SELECT 1')
+ session.close()
+ print('DB Connection OK')
+except Exception as e:
+ print(f'DB Connection Failed: {e}')
+ sys.exit(1)
+" && exit 0 || exit 1
+EOF
+chmod +x /app/healthcheck.sh
+
+# 컨테이너 시작 스크립트 생성
+# Flask 웹 서버(포트 8889) + 크론 + 파일 감시 서비스 병렬 실행
+RUN cat > /app/docker-entrypoint.sh << 'EOF'
+#!/bin/bash
+set -e
+
+echo "[$(date +'%Y-%m-%d %H:%M:%S')] ========================================="
+echo "[$(date +'%Y-%m-%d %H:%M:%S')] Starting First Garden Static Service"
+echo "[$(date +'%Y-%m-%d %H:%M:%S')] ========================================="
+
+# 크론 데몬 시작 (백그라운드)
+# 일정 시간에 daily_run.py 자동 실행
+echo "[$(date +'%Y-%m-%d %H:%M:%S')] Starting cron daemon..."
+cron -f > /app/logs/cron.log 2>&1 &
+CRON_PID=$!
+echo "[$(date +'%Y-%m-%d %H:%M:%S')] Cron daemon started (PID: $CRON_PID)"
+
+# file_watch.py 실행 (백그라운드)
+# 로컬 파일 시스템 감시 및 자동 처리
+echo "[$(date +'%Y-%m-%d %H:%M:%S')] Starting file watch service..."
+cd /app
+python lib/file_watch.py > /app/logs/file_watch.log 2>&1 &
+WATCH_PID=$!
+echo "[$(date +'%Y-%m-%d %H:%M:%S')] File watch service started (PID: $WATCH_PID)"
+
+# Flask 웹 서버 시작 (포트 8889)
+# 파일 업로드, DB 백업/복구 API 제공
+echo "[$(date +'%Y-%m-%d %H:%M:%S')] Starting Flask file upload server (port 8889)..."
+cd /app
+python -c "from app.app import run_app; run_app()" > /app/logs/flask_app.log 2>&1 &
+FLASK_PID=$!
+echo "[$(date +'%Y-%m-%d %H:%M:%S')] Flask server started (PID: $FLASK_PID)"
+
+# 신호 처리: 컨테이너 종료 시 모든 하위 프로세스 정리
+trap "
+ echo '[$(date +\"%Y-%m-%d %H:%M:%S\")] Shutting down services...'
+ kill $CRON_PID $WATCH_PID $FLASK_PID 2>/dev/null || true
+ exit 0
+" SIGTERM SIGINT
+
+# 프로세스 모니터링
+# 하위 프로세스 상태 확인
+echo "[$(date +'%Y-%m-%d %H:%M:%S')] Service monitoring started"
+while true; do
+ wait
+done
+EOF
+chmod +x /app/docker-entrypoint.sh
+
+# 포트 노출 선언
+# 8889: Flask 파일 업로드 서버
+# 5000: 기타 서비스 포트
+EXPOSE 8889 5000
+
+# 엔트리포인트 설정
+ENTRYPOINT ["/app/docker-entrypoint.sh"]
diff --git a/conf/db.py b/conf/db.py
index 71cc931..0259c96 100644
--- a/conf/db.py
+++ b/conf/db.py
@@ -1,33 +1,114 @@
# db.py
import os
-from sqlalchemy import create_engine
-from sqlalchemy.orm import sessionmaker
+import logging
+from sqlalchemy import create_engine, event, exc, pool
+from sqlalchemy.orm import sessionmaker, scoped_session
import yaml
-# db.py 파일 위치 기준 상위 디렉토리 (프로젝트 루트)
+logger = logging.getLogger(__name__)
+
+# 프로젝트 루트 경로 설정
BASE_DIR = os.path.abspath(os.path.join(os.path.dirname(__file__), '..'))
CONFIG_PATH = os.path.join(BASE_DIR, 'conf', 'config.yaml')
def load_config(path=CONFIG_PATH):
- with open(path, 'r', encoding='utf-8') as f:
- return yaml.safe_load(f)
+ """설정 파일 로드"""
+ try:
+ with open(path, 'r', encoding='utf-8') as f:
+ config = yaml.safe_load(f)
+ if not config:
+ raise ValueError(f"설정 파일이 비어있음: {path}")
+ return config
+ except FileNotFoundError:
+ logger.error(f"설정 파일을 찾을 수 없음: {path}")
+ raise
+ except yaml.YAMLError as e:
+ logger.error(f"YAML 파싱 오류: {e}")
+ raise
config = load_config()
-db_cfg = config['database']
+db_cfg = config.get('database', {})
-db_url = f"mysql+pymysql://{db_cfg['user']}:{db_cfg['password']}@{db_cfg['host']}/{db_cfg['name']}?charset=utf8mb4"
-
-# MySQL 연결이 끊겼을 때 자동 재시도 옵션 포함
-engine = create_engine(
- db_url,
- pool_pre_ping=True,
- pool_recycle=3600, # 3600초 = 1시간
+# DB URL 구성
+db_url = (
+ f"mysql+pymysql://{db_cfg.get('user')}:"
+ f"{db_cfg.get('password')}@{db_cfg.get('host')}/"
+ f"{db_cfg.get('name')}?charset=utf8mb4"
)
+# MySQL 엔진 생성 (재연결 및 연결 풀 설정)
+engine = create_engine(
+ db_url,
+ poolclass=pool.QueuePool,
+ pool_pre_ping=True, # 연결 전 핸들 확인
+ pool_recycle=3600, # 3600초(1시간)마다 재연결
+ pool_size=10, # 연결 풀 크기
+ max_overflow=20, # 추가 오버플로우 연결 수
+ echo=False, # SQL 출력 (디버그용)
+ connect_args={
+ 'connect_timeout': 10,
+ 'charset': 'utf8mb4'
+ }
+)
+
+# 연결 에러 발생 시 자동 재연결
+@event.listens_for(pool.Pool, "connect")
+def receive_connect(dbapi_conn, connection_record):
+ """DB 연결 성공 로그"""
+ logger.debug("DB 연결 성공")
+
+@event.listens_for(pool.Pool, "checkout")
+def receive_checkout(dbapi_conn, connection_record, connection_proxy):
+ """연결 풀에서 체크아웃할 때"""
+ pass
+
+@event.listens_for(pool.Pool, "checkin")
+def receive_checkin(dbapi_conn, connection_record):
+ """연결 풀로 반환할 때"""
+ pass
+
+# 세션 팩토리
Session = sessionmaker(bind=engine)
+SessionLocal = scoped_session(Session)
def get_engine():
+ """엔진 반환"""
return engine
def get_session():
- return Session()
+ """새로운 세션 반환"""
+ session = Session()
+ try:
+ # 연결 테스트
+ session.execute('SELECT 1')
+ except exc.DatabaseError as e:
+ logger.error(f"DB 연결 실패: {e}")
+ session.close()
+ raise
+ return session
+
+def get_scoped_session():
+ """스코프 세션 반환 (스레드 안전)"""
+ return SessionLocal
+
+def close_session():
+ """세션 종료"""
+ SessionLocal.remove()
+
+class DBSession:
+ """컨텍스트 매니저를 사용한 세션 관리"""
+ def __init__(self):
+ self.session = None
+
+ def __enter__(self):
+ self.session = get_session()
+ return self.session
+
+ def __exit__(self, exc_type, exc_val, exc_tb):
+ if self.session:
+ if exc_type:
+ self.session.rollback()
+ logger.error(f"트랜잭션 롤백: {exc_type.__name__}: {exc_val}")
+ else:
+ self.session.commit()
+ self.session.close()
diff --git a/daily_run.py b/daily_run.py
index d2556e9..44c1ced 100644
--- a/daily_run.py
+++ b/daily_run.py
@@ -1,43 +1,113 @@
-# ./lib/weather_asos.py
-# ./lib/ga4.py
-# ./lib/air_quality.py
-# 각 파일을 모두 한번씩 실행
-# daily_run.py
+# daily_run.py
+"""
+daily_run.py
+
+매일 정기적으로 실행되는 데이터 수집 스크립트
+- weather_asos.py: 기상청 ASOS 데이터 수집
+- ga4.py: Google Analytics 4 데이터 수집
+- air_quality.py: 대기환경 API 데이터 수집
+"""
import os
import sys
+import logging
+import traceback
+from datetime import datetime
# lib 디렉토리를 path에 추가
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), 'lib')))
-from conf import db, db_schema
+
+from conf import db, db_schema
+from lib.common import setup_logging
+
+# 로거 설정
+logger = setup_logging('daily_run', 'INFO')
def run_weather():
+ """기상청 ASOS 데이터 수집"""
try:
+ logger.info("=" * 60)
+ logger.info("[RUNNING] weather_asos.py")
+ logger.info("=" * 60)
+
from weather_asos import main as weather_main
- print("\n[RUNNING] weather_asos.py")
- weather_main()
+ result = weather_main()
+ logger.info("[SUCCESS] weather_asos 완료")
+ return True
+
except Exception as e:
- print(f"[ERROR] weather_asos 실행 실패: {e}")
+ logger.error(f"[ERROR] weather_asos 실행 실패: {e}")
+ logger.error(traceback.format_exc())
+ return False
def run_ga4():
+ """Google Analytics 4 데이터 수집"""
try:
+ logger.info("=" * 60)
+ logger.info("[RUNNING] ga4.py")
+ logger.info("=" * 60)
+
from ga4 import main as ga4_main
- print("\n[RUNNING] ga4.py")
- ga4_main()
+ result = ga4_main()
+ logger.info("[SUCCESS] ga4 완료")
+ return True
+
except Exception as e:
- print(f"[ERROR] ga4 실행 실패: {e}")
+ logger.error(f"[ERROR] ga4 실행 실패: {e}")
+ logger.error(traceback.format_exc())
+ return False
def run_air_quality():
+ """대기환경 API 데이터 수집"""
try:
+ logger.info("=" * 60)
+ logger.info("[RUNNING] air_quality.py")
+ logger.info("=" * 60)
+
from air_quality import AirQualityCollector
- print("\n[RUNNING] air_quality.py")
config = db.load_config()
collector = AirQualityCollector(config, db.engine, db_schema.air)
collector.run()
+ logger.info("[SUCCESS] air_quality 완료")
+ return True
+
except Exception as e:
- print(f"[ERROR] air_quality 실행 실패: {e}")
+ logger.error(f"[ERROR] air_quality 실행 실패: {e}")
+ logger.error(traceback.format_exc())
+ return False
+
+def main():
+ """메인 실행 함수"""
+ logger.info(f"\n{'='*60}")
+ logger.info(f"[START] daily_run.py 시작: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
+ logger.info(f"{'='*60}\n")
+
+ results = {}
+
+ # 각 작업 실행
+ results['weather'] = run_weather()
+ results['ga4'] = run_ga4()
+ results['air_quality'] = run_air_quality()
+
+ # 결과 정리
+ logger.info(f"\n{'='*60}")
+ logger.info("[SUMMARY] 작업 완료 결과:")
+ logger.info(f"{'='*60}")
+ for task, success in results.items():
+ status = "✓ SUCCESS" if success else "✗ FAILED"
+ logger.info(f" {task}: {status}")
+
+ logger.info(f"[END] daily_run.py 종료: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
+ logger.info(f"{'='*60}\n")
+
+ # 모든 작업 성공 여부 반환
+ return all(results.values())
if __name__ == "__main__":
- run_weather()
- run_ga4()
- run_air_quality()
+ try:
+ success = main()
+ sys.exit(0 if success else 1)
+ except Exception as e:
+ logger.critical(f"[CRITICAL] 예상치 못한 에러: {e}")
+ logger.critical(traceback.format_exc())
+ sys.exit(1)
diff --git a/docker-compose.yml b/docker-compose.yml
index 1323511..ea58288 100644
--- a/docker-compose.yml
+++ b/docker-compose.yml
@@ -1,13 +1,103 @@
+version: '3.8'
+
services:
+ # MariaDB 데이터베이스
+ mariadb:
+ image: mariadb:11.2-jammy
+ container_name: fg-static-db
+ environment:
+ MYSQL_ROOT_PASSWORD: ${DB_ROOT_PASSWORD:-rootpassword}
+ MYSQL_DATABASE: ${DB_NAME:-firstgarden}
+ MYSQL_USER: ${DB_USER:-firstgarden}
+ MYSQL_PASSWORD: ${DB_PASSWORD:-Fg9576861!}
+ TZ: Asia/Seoul
+ volumes:
+ # 실제 볼륨 마운트 (바인드 마운트) - 데이터베이스 데이터 저장
+ - ./db_data:/var/lib/mysql
+ # 초기화 SQL
+ - ./conf/install.sql:/docker-entrypoint-initdb.d/init.sql
+ # 데이터베이스 백업 및 복구 폴더
+ - ./dbbackup:/dbbackup
+ ports:
+ - "${DB_PORT:-3306}:3306"
+ networks:
+ - static-network
+ restart: unless-stopped
+ healthcheck:
+ test: ["CMD", "mariadb-admin", "ping", "-h", "localhost"]
+ timeout: 10s
+ interval: 30s
+ retries: 3
+ start_period: 40s
+
+ # 메인 데이터 수집 및 분석 서비스 (Flask 웹 서버 포함)
fg-static:
- container_name: fg-static
+ container_name: fg-static-app
build:
context: .
dockerfile: build/Dockerfile
image: reg.firstgarden.co.kr/fg-static:latest
- volumes:
- - ./data:/app/data
- - ./conf:/app/conf
+ depends_on:
+ mariadb:
+ condition: service_healthy
environment:
- - TZ=Asia/Seoul
+ # 데이터베이스 설정
+ DB_HOST: mariadb
+ DB_PORT: 3306
+ DB_NAME: ${DB_NAME:-firstgarden}
+ DB_USER: ${DB_USER:-firstgarden}
+ DB_PASSWORD: ${DB_PASSWORD:-Fg9576861!}
+
+ # 타임존
+ TZ: Asia/Seoul
+
+ # Python 설정
+ PYTHONUNBUFFERED: 1
+ PYTHONDONTWRITEBYTECODE: 1
+
+ # 로깅 레벨
+ LOG_LEVEL: INFO
+
+ # Flask 설정
+ FLASK_ENV: production
+ FLASK_DEBUG: 0
+ volumes:
+ # 설정 파일
+ - ./conf:/app/conf:ro
+ - ./conf/config.yaml:/app/conf/config.yaml:ro
+ - ./conf/service-account-credentials.json:/app/conf/service-account-credentials.json:ro
+
+ # 실제 볼륨 마운트 (바인드 마운트)
+ - ./data:/app/data
+ - ./output:/app/output
+ - ./uploads:/app/uploads
+ - ./dbbackup:/app/dbbackup
+ - ./logs:/app/logs
+ ports:
+ # Flask 파일 업로드 서버 포트
+ - "8889:8889"
+ # 기타 포트 (필요 시)
+ - "5000:5000"
+ networks:
+ - static-network
restart: unless-stopped
+ healthcheck:
+ test: ["CMD", "curl", "-f", "http://localhost:8889/api/status", "||", "exit", "1"]
+ timeout: 10s
+ interval: 60s
+ retries: 3
+ start_period: 30s
+ logging:
+ driver: "json-file"
+ options:
+ max-size: "100m"
+ max-file: "10"
+
+networks:
+ # 컨테이너 간 통신 네트워크
+ static-network:
+ driver: bridge
+
+ # 로그
+ logs_volume:
+ driver: local
diff --git a/lib/air_quality.py b/lib/air_quality.py
index b4a80cf..b9bc3a9 100644
--- a/lib/air_quality.py
+++ b/lib/air_quality.py
@@ -1,13 +1,17 @@
-import os, sys
+import os, sys
import requests
+import json
from datetime import datetime, timedelta
from sqlalchemy import select, func, and_
from sqlalchemy.dialects.mysql import insert as mysql_insert
from sqlalchemy.exc import IntegrityError
+import traceback
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
from conf import db, db_schema
+from requests_utils import make_requests_session
+CACHE_FILE = os.path.join(os.path.dirname(__file__), '..', 'cache', 'air_num_rows.json')
class AirQualityCollector:
def __init__(self, config, engine, table):
@@ -20,6 +24,24 @@ class AirQualityCollector:
self.engine = engine
self.table = table
self.yesterday = (datetime.now() - timedelta(days=1)).date()
+ self.session = make_requests_session()
+
+ # load cache
+ self._num_rows_cache = {}
+ try:
+ if os.path.exists(CACHE_FILE):
+ with open(CACHE_FILE, 'r', encoding='utf-8') as f:
+ self._num_rows_cache = json.load(f)
+ except Exception:
+ self._num_rows_cache = {}
+
+ def _save_num_rows_cache(self):
+ try:
+ os.makedirs(os.path.dirname(CACHE_FILE), exist_ok=True)
+ with open(CACHE_FILE, 'w', encoding='utf-8') as f:
+ json.dump(self._num_rows_cache, f, ensure_ascii=False, indent=2)
+ except Exception as e:
+ print(f"[WARN] num_rows 캐시 저장 실패: {e}")
def get_latest_date(self, conn, station):
try:
@@ -30,6 +52,7 @@ class AirQualityCollector:
return result
except Exception as e:
print(f"[ERROR] 가장 최근 날짜 조회 실패: {e}")
+ traceback.print_exc()
return None
def save_data_to_db(self, items, conn, station):
@@ -37,7 +60,7 @@ class AirQualityCollector:
try:
item_date = datetime.strptime(item['msurDt'], '%Y-%m-%d').date()
except Exception as e:
- print(f"[ERROR] 날짜 파싱 오류: {item.get('msurDt')} → {e}")
+ print(f"[ERROR] 날짜 파싱 오류: {item.get('msurDt')} {e}")
continue
data = {
@@ -53,7 +76,7 @@ class AirQualityCollector:
try:
if self.debug:
- print(f"[DEBUG] {item_date} [{station}] → DB 저장 시도: {data}")
+ print(f"[DEBUG] {item_date} [{station}] DB 저장 시도: {data}")
continue
if self.force_update:
@@ -75,6 +98,7 @@ class AirQualityCollector:
print(f"[ERROR] DB 중복 오류: {e}")
except Exception as e:
print(f"[ERROR] DB 저장 실패: {e}")
+ traceback.print_exc()
def fetch_air_quality_data_range(self, start_date_str, end_date_str, station_name, num_of_rows=100, page_no=1):
url = "http://apis.data.go.kr/B552584/ArpltnStatsSvc/getMsrstnAcctoRDyrg"
@@ -88,23 +112,39 @@ class AirQualityCollector:
'msrstnName': station_name,
}
+ resp = None
try:
- resp = requests.get(url, params=params, timeout=10)
+ resp = self.session.get(url, params=params, timeout=20)
resp.raise_for_status()
data = resp.json()
return data.get('response', {}).get('body', {}).get('items', [])
- except requests.RequestException as e:
- print(f"[ERROR] 요청 실패: {e}")
- return []
- except ValueError as e:
- print(f"[ERROR] JSON 파싱 실패: {e}")
+ except Exception as e:
+ body_preview = None
+ try:
+ if resp is not None:
+ body_preview = resp.text[:1000]
+ except Exception:
+ body_preview = None
+ print(f"[ERROR] 요청 실패: {e} status={getattr(resp, 'status_code', 'n/a')} body_preview={body_preview}")
+ traceback.print_exc()
return []
def test_num_of_rows(self, station_name, date_str):
+ # 캐시 확인
+ try:
+ cache_key = f"{station_name}:{date_str}"
+ if cache_key in self._num_rows_cache:
+ val = int(self._num_rows_cache[cache_key])
+ print(f"[INFO] 캐시된 numOfRows 사용: {val} ({cache_key})")
+ return val
+ except Exception:
+ pass
+
max_rows = 1000
min_rows = 100
while max_rows >= min_rows:
+ resp = None
try:
url = "http://apis.data.go.kr/B552584/ArpltnStatsSvc/getMsrstnAcctoRDyrg"
params = {
@@ -116,13 +156,25 @@ class AirQualityCollector:
'inqEndDt': date_str,
'msrstnName': station_name,
}
- resp = requests.get(url, params=params, timeout=10)
+ resp = self.session.get(url, params=params, timeout=20)
resp.raise_for_status()
- resp.json()
+ resp.json() # 성공하면 해당 max_rows 사용 가능
print(f"[INFO] numOfRows 최대값 탐색 성공: {max_rows}")
+ # 캐시에 저장
+ try:
+ self._num_rows_cache[f"{station_name}:{date_str}"] = max_rows
+ self._save_num_rows_cache()
+ except Exception:
+ pass
return max_rows
except Exception as e:
- print(f"[WARN] numOfRows={max_rows} 실패: {e}, 100 감소 후 재시도")
+ body_preview = None
+ try:
+ if resp is not None:
+ body_preview = resp.text[:500]
+ except Exception:
+ body_preview = None
+ print(f"[WARN] numOfRows={max_rows} 실패: {e}, body_preview={body_preview} 100 감소 후 재시도")
max_rows -= 100
print("[WARN] 최소 numOfRows 값(100)도 실패했습니다. 기본값 100 사용")
@@ -160,7 +212,7 @@ class AirQualityCollector:
current_start = current_end + timedelta(days=1)
print("\n[INFO] 모든 측정소 데이터 처리 완료")
-
+
if __name__ == '__main__':
config = db.load_config()
collector = AirQualityCollector(config, db.engine, db_schema.air)
diff --git a/lib/common.py b/lib/common.py
index 41b8417..dfb5723 100644
--- a/lib/common.py
+++ b/lib/common.py
@@ -1,31 +1,136 @@
# common.py
-import os, yaml
+import os
+import yaml
import logging
import time
import glob
+from functools import wraps
+from typing import Any, Callable
-def load_config():
+# 로거 설정
+def setup_logging(name: str, level: str = 'INFO') -> logging.Logger:
"""
- conf/config.yaml 파일을 UTF-8로 읽어 파이썬 dict로 반환
+ 로거 설정 (일관된 포맷 적용)
+
+ Args:
+ name: 로거 이름
+ level: 로그 레벨 (INFO, DEBUG, WARNING, ERROR)
+
+ Returns:
+ Logger 인스턴스
"""
- path = os.path.join(os.path.dirname(__file__), '..', 'conf', 'config.yaml')
- with open(path, encoding='utf-8') as f:
- return yaml.safe_load(f)
-
-def get_logger(name):
logger = logging.getLogger(name)
+
if not logger.handlers:
handler = logging.StreamHandler()
- formatter = logging.Formatter('[%(asctime)s] %(levelname)s: %(message)s')
+ formatter = logging.Formatter(
+ '[%(asctime)s] %(name)s - %(levelname)s: %(message)s',
+ datefmt='%Y-%m-%d %H:%M:%S'
+ )
handler.setFormatter(formatter)
logger.addHandler(handler)
- logger.setLevel(logging.INFO)
+
+ logger.setLevel(getattr(logging, level.upper(), logging.INFO))
return logger
-def wait_download_complete(download_dir, ext, timeout=60):
- for _ in range(timeout):
- files = glob.glob(os.path.join(download_dir, f"*.{ext.strip('.')}"))
+def get_logger(name: str) -> logging.Logger:
+ """기존 호환성 유지"""
+ return setup_logging(name)
+
+def load_config(config_path: str = None) -> dict:
+ """
+ conf/config.yaml 파일을 UTF-8로 읽어 파이썬 dict로 반환
+
+ Args:
+ config_path: 설정 파일 경로 (없으면 기본값 사용)
+
+ Returns:
+ 설정 딕셔너리
+
+ Raises:
+ FileNotFoundError: 설정 파일을 찾을 수 없을 때
+ yaml.YAMLError: YAML 파싱 실패 시
+ """
+ if config_path is None:
+ config_path = os.path.join(os.path.dirname(__file__), '..', 'conf', 'config.yaml')
+
+ try:
+ with open(config_path, encoding='utf-8') as f:
+ config = yaml.safe_load(f)
+ if config is None:
+ raise ValueError(f"설정 파일이 비어있음: {config_path}")
+ return config
+ except FileNotFoundError:
+ raise FileNotFoundError(f"설정 파일을 찾을 수 없음: {config_path}")
+ except yaml.YAMLError as e:
+ raise yaml.YAMLError(f"YAML 파싱 오류: {e}")
+
+def retry_on_exception(max_retries: int = 3, delay: float = 1.0, backoff: float = 2.0):
+ """
+ 지정된 횟수만큼 재시도하는 데코레이터
+
+ Args:
+ max_retries: 최대 재시도 횟수
+ delay: 재시도 간격 (초)
+ backoff: 재시도마다 지연 시간 배수
+ """
+ def decorator(func: Callable) -> Callable:
+ @wraps(func)
+ def wrapper(*args, **kwargs) -> Any:
+ logger = logging.getLogger(func.__module__)
+ last_exception = None
+
+ for attempt in range(max_retries):
+ try:
+ return func(*args, **kwargs)
+ except Exception as e:
+ last_exception = e
+ if attempt < max_retries - 1:
+ wait_time = delay * (backoff ** attempt)
+ logger.warning(
+ f"{func.__name__} 재시도 {attempt + 1}/{max_retries} "
+ f"({wait_time:.1f}초 대기): {e}"
+ )
+ time.sleep(wait_time)
+ else:
+ logger.error(
+ f"{func.__name__} 모든 재시도 실패: {e}"
+ )
+
+ raise last_exception
+
+ return wrapper
+ return decorator
+
+def wait_download_complete(download_dir: str, ext: str, timeout: int = 60) -> str:
+ """
+ 파일 다운로드 완료 대기
+
+ Args:
+ download_dir: 다운로드 디렉토리
+ ext: 파일 확장자 (예: 'xlsx', 'csv')
+ timeout: 대기 시간 (초)
+
+ Returns:
+ 다운로드된 파일 경로
+
+ Raises:
+ TimeoutError: 지정 시간 내 파일이 나타나지 않을 때
+ """
+ logger = logging.getLogger(__name__)
+ ext = ext.lstrip('.')
+
+ for i in range(timeout):
+ files = glob.glob(os.path.join(download_dir, f"*.{ext}"))
if files:
+ logger.info(f"다운로드 완료: {files[0]}")
return files[0]
+
+ if i > 0 and i % 10 == 0:
+ logger.debug(f"다운로드 대기 중... ({i}초 경과)")
+
time.sleep(1)
- raise TimeoutError("다운로드 대기 시간 초과")
\ No newline at end of file
+
+ raise TimeoutError(
+ f"파일 다운로드 대기 시간 초과 ({timeout}초): {download_dir}/*.{ext}"
+ )
\ No newline at end of file
diff --git a/lib/ga4.py b/lib/ga4.py
index 1313159..bb6924f 100644
--- a/lib/ga4.py
+++ b/lib/ga4.py
@@ -1,7 +1,7 @@
-# ga4.py
+# ga4.py
'''
퍼스트가든 구글 애널리틱스 API를 활용해 관련 데이터를 DB에 저장함
-병렬 처리를 통해 처리 속도 향상
+병렬 처리를 통해 처리 속도 향상 (내부 병렬은 유지하되 에러/재시도 보강)
'''
import sys, os
@@ -9,6 +9,7 @@ sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
import yaml
import pprint
+import traceback
from datetime import datetime, timedelta
from dateutil.parser import parse
from google.analytics.data import BetaAnalyticsDataClient
@@ -38,25 +39,33 @@ def load_config():
# GA4 클라이언트 초기화
# ------------------------
def init_ga_client(service_account_file):
- os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = service_account_file
- print(f"[INFO] GA4 클라이언트 초기화 - 인증파일: {service_account_file}")
- return BetaAnalyticsDataClient()
+ try:
+ os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = service_account_file
+ print(f"[INFO] GA4 클라이언트 초기화 - 인증파일: {service_account_file}")
+ return BetaAnalyticsDataClient()
+ except Exception as e:
+ print(f"[ERROR] GA4 클라이언트 초기화 실패: {e}")
+ traceback.print_exc()
+ raise
# ------------------------
# config.yaml에 최대 rows 저장
# ------------------------
def update_config_file_with_max_rows(max_rows):
- with open(CONFIG_PATH, encoding="utf-8") as f:
- config = yaml.safe_load(f)
+ try:
+ with open(CONFIG_PATH, encoding="utf-8") as f:
+ config = yaml.safe_load(f)
- if "ga4" not in config:
- config["ga4"] = {}
- config["ga4"]["max_rows_per_request"] = max_rows
+ if "ga4" not in config:
+ config["ga4"] = {}
+ config["ga4"]["max_rows_per_request"] = int(max_rows)
- with open(CONFIG_PATH, "w", encoding="utf-8") as f:
- yaml.dump(config, f, allow_unicode=True)
+ with open(CONFIG_PATH, "w", encoding="utf-8") as f:
+ yaml.dump(config, f, allow_unicode=True)
- print(f"[INFO] config.yaml에 max_rows_per_request = {max_rows} 저장 완료")
+ print(f"[INFO] config.yaml에 max_rows_per_request = {max_rows} 저장 완료")
+ except Exception as e:
+ print(f"[WARN] config.yaml 업데이트 실패: {e}")
# ------------------------
# GA4 API로 최대 rows 감지
@@ -71,10 +80,13 @@ def detect_max_rows_supported(client, property_id):
limit=100000
)
response = client.run_report(request)
- print(f"[INFO] 최대 rows 감지: {len(response.rows)} rows 수신됨.")
- return len(response.rows)
+ nrows = len(response.rows)
+ print(f"[INFO] 최대 rows 감지: {nrows} rows 수신됨.")
+ return nrows
except Exception as e:
print(f"[WARNING] 최대 rows 감지 실패: {e}")
+ traceback.print_exc()
+ # 안전한 기본값 반환
return 10000
# ------------------------
@@ -82,21 +94,31 @@ def detect_max_rows_supported(client, property_id):
# ------------------------
def fetch_report(client, property_id, start_date, end_date, dimensions, metrics, limit=10000):
print(f"[INFO] fetch_report 호출 - 기간: {start_date} ~ {end_date}, dims={dimensions}, metrics={metrics}")
- request = RunReportRequest(
- property=f"properties/{property_id}",
- dimensions=[Dimension(name=d) for d in dimensions],
- metrics=[Metric(name=m) for m in metrics],
- date_ranges=[DateRange(start_date=start_date, end_date=end_date)],
- limit=limit,
- )
- response = client.run_report(request)
- print(f"[INFO] GA4 리포트 응답 받음: {len(response.rows)} rows")
- return response
+ try:
+ request = RunReportRequest(
+ property=f"properties/{property_id}",
+ dimensions=[Dimension(name=d) for d in dimensions],
+ metrics=[Metric(name=m) for m in metrics],
+ date_ranges=[DateRange(start_date=start_date, end_date=end_date)],
+ limit=limit,
+ )
+ response = client.run_report(request)
+ print(f"[INFO] GA4 리포트 응답 받음: {len(response.rows)} rows")
+ return response
+ except Exception as e:
+ print(f"[ERROR] GA4 fetch_report 실패: {e}")
+ traceback.print_exc()
+ # 빈 응답 형태 반환하는 대신 None 반환해서 호출부가 처리하도록 함
+ return None
# ------------------------
# 응답 데이터를 DB에 저장
# ------------------------
def save_report_to_db(engine, table, response, dimension_names, metric_names, debug=False):
+ if response is None:
+ print("[INFO] 저장할 응답 없음 (None)")
+ return
+
with engine.begin() as conn:
for row in response.rows:
dims = row.dimension_values
@@ -137,6 +159,7 @@ def save_report_to_db(engine, table, response, dimension_names, metric_names, de
print(f"[DB ERROR] 중복 오류 또는 기타: {e}")
except Exception as e:
print(f"[DB ERROR] 저장 실패: {e}")
+ traceback.print_exc()
# ------------------------
# 테이블에서 마지막 날짜 조회
@@ -175,7 +198,6 @@ def determine_date_range(table, config_start, config_end, force_update, engine):
else:
actual_start = config_start
- # 시작일이 종료일보다 뒤에 있으면 자동 교체
if actual_start > actual_end:
print(f"[WARN] 시작일({actual_start})이 종료일({actual_end})보다 뒤에 있습니다. 날짜를 교환하여 수집을 계속합니다.")
actual_start, actual_end = actual_end, actual_start
@@ -201,10 +223,10 @@ def process_dimension_metric(engine, client, property_id, config, table, dims, m
end_str = end_dt.strftime('%Y-%m-%d')
print(f"[INFO] GA4 데이터 조회: {start_str} ~ {end_str}")
response = fetch_report(client, property_id, start_str, end_str, dimensions=dims, metrics=mets, limit=max_rows)
- if len(response.rows) > 0:
+ if response and len(response.rows) > 0:
save_report_to_db(engine, table, response, dimension_names=dims, metric_names=mets, debug=debug)
else:
- print(f"[INFO] 해당 기간 {start_str} ~ {end_str} 데이터 없음")
+ print(f"[INFO] 해당 기간 {start_str} ~ {end_str} 데이터 없음 또는 요청 실패")
# ------------------------
# 메인 진입점 (병렬 처리 포함)
@@ -224,12 +246,19 @@ def main():
return
engine = db.engine
- client = init_ga_client(service_account_file)
+ try:
+ client = init_ga_client(service_account_file)
+ except Exception:
+ print("[ERROR] GA4 클라이언트 초기화 실패로 종료합니다.")
+ return
max_rows = ga4_cfg.get("max_rows_per_request")
if not max_rows:
max_rows = detect_max_rows_supported(client, property_id)
- update_config_file_with_max_rows(max_rows)
+ try:
+ update_config_file_with_max_rows(max_rows)
+ except Exception:
+ pass
print(f"[INFO] 설정된 max_rows_per_request = {max_rows}")
tasks = [
@@ -253,6 +282,7 @@ def main():
print(f"[INFO] 태스크 {i} 완료")
except Exception as e:
print(f"[ERROR] 태스크 {i} 실패: {e}")
+ traceback.print_exc()
print("[INFO] GA4 데이터 수집 및 저장 완료")
diff --git a/lib/pos_update_daily_product.py b/lib/pos_update_daily_product.py
index ea0556c..322e883 100644
--- a/lib/pos_update_daily_product.py
+++ b/lib/pos_update_daily_product.py
@@ -18,6 +18,14 @@ CONFIG = load_config()
DATA_DIR = os.path.join(os.path.dirname(__file__), '../data')
def update_pos_table(engine, table, df):
+ """
+ 데이터프레임을 테이블에 업데이트
+
+ Args:
+ engine: SQLAlchemy 엔진
+ table: DB 테이블 객체
+ df: 데이터프레임
+ """
with engine.begin() as conn:
for idx, row in df.iterrows():
data = row.to_dict()
@@ -39,6 +47,17 @@ def update_pos_table(engine, table, df):
print("[DONE] 모든 데이터 삽입 완료")
def process_file(filepath, table, engine):
+ """
+ OKPOS 파일 처리
+
+ Args:
+ filepath: 파일 경로
+ table: DB 테이블
+ engine: SQLAlchemy 엔진
+
+ Returns:
+ tuple[bool, int]: (성공 여부, 행 수)
+ """
print(f"[INFO] 처리 시작: {filepath}")
try:
ext = os.path.splitext(filepath)[-1].lower()
@@ -86,6 +105,51 @@ def process_file(filepath, table, engine):
print(f"[INFO] 처리 완료: {filepath}")
return True, len(df)
+
+def process_okpos_file(filepath):
+ """
+ OKPOS 파일을 처리하고 DB에 저장
+ 웹 업로드 인터페이스에서 사용하는 함수
+
+ Args:
+ filepath (str): 업로드된 파일 경로
+
+ Returns:
+ dict: {
+ 'success': bool,
+ 'message': str,
+ 'rows_inserted': int
+ }
+ """
+ try:
+ engine = db.engine
+ table = db_schema.pos
+
+ # 파일 처리
+ success, row_count = process_file(filepath, table, engine)
+
+ if success:
+ return {
+ 'success': True,
+ 'message': f'{row_count}행이 저장되었습니다.',
+ 'rows_inserted': row_count
+ }
+ else:
+ return {
+ 'success': False,
+ 'message': '파일 처리에 실패했습니다.',
+ 'rows_inserted': 0
+ }
+
+ except Exception as e:
+ print(f"[ERROR] OKPOS 파일 처리 오류: {e}")
+ return {
+ 'success': False,
+ 'message': f'파일 처리 중 오류: {str(e)}',
+ 'rows_inserted': 0
+ }
+
+
def batch_process_files(table, engine):
files = [f for f in os.listdir(DATA_DIR) if f.startswith("일자별 (상품별)") and f.endswith(('.xlsx', '.xls'))]
if not files:
diff --git a/lib/pos_update_upsolution.py b/lib/pos_update_upsolution.py
index 7edc385..3328249 100644
--- a/lib/pos_update_upsolution.py
+++ b/lib/pos_update_upsolution.py
@@ -71,9 +71,21 @@ def prepare_bulk_data(df):
def process_bulk_upsert(bulk_data, session, table, batch_size=1000):
+ """
+ 데이터 일괄 삽입/업데이트
+
+ Args:
+ bulk_data: 삽입할 데이터 리스트
+ session: DB 세션
+ table: 대상 테이블
+ batch_size: 배치 크기
+
+ Returns:
+ int: 삽입된 행 수
+ """
if not bulk_data:
logger.info("[INFO] 삽입할 데이터가 없습니다.")
- return
+ return 0
total = len(bulk_data)
inserted_total = 0
@@ -96,8 +108,11 @@ def process_bulk_upsert(bulk_data, session, table, batch_size=1000):
raise
logger.info(f"[DONE] 총 {total}건 처리 완료 (insert+update)")
+ return inserted_total
+
def file_reader(queue, files):
+ """파일 읽기 스레드"""
for filepath in files:
try:
logger.info(f"[READ] {os.path.basename(filepath)} 읽기 시작")
@@ -111,6 +126,7 @@ def file_reader(queue, files):
def db_writer(queue, session, table):
+ """DB 쓰기 스레드"""
while True:
item = queue.get()
if item is None:
@@ -126,6 +142,66 @@ def db_writer(queue, session, table):
logger.error(f"[FAIL] {os.path.basename(filepath)} DB 삽입 실패 - {e}")
+def process_upsolution_file(filepath):
+ """
+ UPSOLUTION 파일을 처리하고 DB에 저장
+ 웹 업로드 인터페이스에서 사용하는 함수
+
+ Args:
+ filepath (str): 업로드된 파일 경로
+
+ Returns:
+ dict: {
+ 'success': bool,
+ 'message': str,
+ 'rows_inserted': int
+ }
+ """
+ try:
+ logger.info(f"[WEB] UPSOLUTION 파일 처리 시작: {filepath}")
+
+ # 파일 읽기
+ df = load_excel_data(filepath)
+ logger.info(f"[WEB] 데이터 읽기 완료: {len(df)}행")
+
+ # 데이터 준비
+ bulk_data = prepare_bulk_data(df)
+ logger.info(f"[WEB] 데이터 준비 완료: {len(bulk_data)}행")
+
+ # DB 세션 및 테이블 가져오기
+ session = db.get_session()
+ engine = db.get_engine()
+
+ metadata = MetaData()
+ table = Table(
+ db_schema.get_full_table_name("pos_ups_billdata"),
+ metadata,
+ autoload_with=engine
+ )
+
+ # 데이터 삽입
+ inserted = process_bulk_upsert(bulk_data, session, table)
+
+ session.close()
+
+ logger.info(f"[WEB] UPSOLUTION 파일 처리 완료: {inserted}행 삽입")
+
+ return {
+ 'success': True,
+ 'message': f'{inserted}행이 저장되었습니다.',
+ 'rows_inserted': inserted
+ }
+
+ except Exception as e:
+ logger.error(f"[WEB] UPSOLUTION 파일 처리 오류: {e}", exc_info=True)
+ return {
+ 'success': False,
+ 'message': f'파일 처리 중 오류: {str(e)}',
+ 'rows_inserted': 0
+ }
+
+
+
def main():
engine = db.get_engine()
session = db.get_session()
@@ -142,7 +218,7 @@ def main():
logger.info(f"[INFO] 처리할 파일 {len(files)}개")
- queue = Queue(maxsize=2) # 2개 정도 여유 있게
+ queue = Queue(maxsize=3) # 2개 정도 여유 있게
reader_thread = threading.Thread(target=file_reader, args=(queue, files))
writer_thread = threading.Thread(target=db_writer, args=(queue, session, table))
diff --git a/lib/requests_utils.py b/lib/requests_utils.py
new file mode 100644
index 0000000..0ddf4d1
--- /dev/null
+++ b/lib/requests_utils.py
@@ -0,0 +1,21 @@
+import requests
+from requests.adapters import HTTPAdapter
+from urllib3.util import Retry
+
+def make_requests_session(retries=3, backoff_factor=0.5, status_forcelist=(429, 500, 502, 503, 504)):
+ """
+ 재시도(backoff)를 적용한 requests.Session 반환
+ """
+ session = requests.Session()
+ retry = Retry(
+ total=retries,
+ read=retries,
+ connect=retries,
+ backoff_factor=backoff_factor,
+ status_forcelist=status_forcelist,
+ allowed_methods=frozenset(["HEAD", "GET", "OPTIONS", "POST"])
+ )
+ adapter = HTTPAdapter(max_retries=retry)
+ session.mount("https://", adapter)
+ session.mount("http://", adapter)
+ return session
diff --git a/lib/to_csv.py b/lib/to_csv.py
new file mode 100644
index 0000000..405bbdc
--- /dev/null
+++ b/lib/to_csv.py
@@ -0,0 +1,44 @@
+import os, sys
+import shutil
+import pandas as pd
+
+sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
+from lib.common import get_logger
+
+logger = get_logger("TO_CSV")
+
+DATA_DIR = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "data"))
+FINISH_DIR = os.path.join(DATA_DIR, "finish")
+os.makedirs(FINISH_DIR, exist_ok=True)
+
+def convert_excel_to_csv(filepath):
+ try:
+ logger.info(f"변환 시작: {os.path.basename(filepath)}")
+ df = pd.read_excel(filepath, header=1) # 2행이 헤더
+ df.columns = [col.strip() for col in df.columns]
+
+ csv_path = filepath + '.csv'
+ df.to_csv(csv_path, index=False, encoding='utf-8-sig')
+ logger.info(f"변환 완료: {os.path.basename(csv_path)}")
+
+ # 변환 완료된 원본 엑셀 파일 이동
+ dest_path = os.path.join(FINISH_DIR, os.path.basename(filepath))
+ shutil.move(filepath, dest_path)
+ logger.info(f"원본 엑셀 파일 이동 완료: {os.path.basename(dest_path)}")
+
+ except Exception as e:
+ logger.error(f"변환 실패: {os.path.basename(filepath)} - {e}")
+
+def main():
+ files = [os.path.join(DATA_DIR, f) for f in os.listdir(DATA_DIR)
+ if (f.endswith(('.xls', '.xlsx')) and f.startswith("영수증별 상세매출"))]
+
+ logger.info(f"총 {len(files)}개 엑셀 파일 변환 시작")
+
+ for filepath in files:
+ convert_excel_to_csv(filepath)
+
+ logger.info("모든 파일 변환 완료")
+
+if __name__ == "__main__":
+ main()
\ No newline at end of file
diff --git a/lib/weather_asos.py b/lib/weather_asos.py
index 8e32ae3..2a21413 100644
--- a/lib/weather_asos.py
+++ b/lib/weather_asos.py
@@ -1,4 +1,4 @@
-import sys, os
+import sys, os
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
import yaml
@@ -6,8 +6,10 @@ import requests
from datetime import datetime, timedelta, date
from sqlalchemy.dialects.mysql import insert as mysql_insert
from sqlalchemy import select
+import traceback
from conf import db, db_schema
+from requests_utils import make_requests_session
CONFIG_PATH = os.path.join(os.path.dirname(__file__), "../conf/config.yaml")
@@ -24,7 +26,9 @@ def fetch_data_range_chunks(start_dt, end_dt, chunk_days=10):
yield current_start.strftime("%Y%m%d"), current_end.strftime("%Y%m%d")
current_start = current_end + timedelta(days=1)
-def fetch_asos_data(stn_id, start_dt, end_dt, service_key):
+def fetch_asos_data(stn_id, start_dt, end_dt, service_key, session=None):
+ if session is None:
+ session = make_requests_session()
url = "http://apis.data.go.kr/1360000/AsosDalyInfoService/getWthrDataList"
params = {
@@ -44,8 +48,9 @@ def fetch_asos_data(stn_id, start_dt, end_dt, service_key):
"Accept": "application/json"
}
+ resp = None
try:
- resp = requests.get(url, params=params, headers=headers, timeout=15)
+ resp = session.get(url, params=params, headers=headers, timeout=20)
resp.raise_for_status()
data = resp.json()
items = data.get("response", {}).get("body", {}).get("items", {}).get("item", [])
@@ -57,7 +62,14 @@ def fetch_asos_data(stn_id, start_dt, end_dt, service_key):
return items
except Exception as e:
- print(f"[ERROR] API 요청 실패: {e}")
+ body_preview = None
+ try:
+ if resp is not None:
+ body_preview = resp.text[:1000]
+ except Exception:
+ body_preview = None
+ print(f"[ERROR] ASOS API 요청 실패: {e} status={getattr(resp, 'status_code', 'n/a')} body_preview={body_preview}")
+ traceback.print_exc()
return []
def save_items_to_db(items, conn, table, force_update=False, debug=False):
@@ -98,7 +110,7 @@ def save_items_to_db(items, conn, table, force_update=False, debug=False):
data[key] = None
if debug:
- print(f"[DEBUG] {record_date} → DB 저장 시도: {data}")
+ print(f"[DEBUG] {record_date} DB 저장 시도: {data}")
continue
if force_update:
@@ -116,6 +128,7 @@ def save_items_to_db(items, conn, table, force_update=False, debug=False):
except Exception as e:
print(f"[ERROR] 저장 실패: {e}")
+ traceback.print_exc()
raise
def get_latest_date_from_db(conn, table):
@@ -149,6 +162,8 @@ def main():
chunk_days = 1000
+ session = make_requests_session()
+
with engine.begin() as conn:
print(f"[INFO] DB 저장 최종 일자 점검")
latest_date = get_latest_date_from_db(conn, table)
@@ -170,7 +185,7 @@ def main():
for stn_id in stn_ids:
for chunk_start, chunk_end in fetch_data_range_chunks(start_dt, end_dt, chunk_days):
print(f"[INFO] 지점 {stn_id} 데이터 요청 중: {chunk_start} ~ {chunk_end}")
- items = fetch_asos_data(stn_id, chunk_start, chunk_end, service_key)
+ items = fetch_asos_data(stn_id, chunk_start, chunk_end, service_key, session=session)
if items:
save_items_to_db(items, conn, table, force_update, debug)
else:
diff --git a/requirements.txt b/requirements.txt
index aed05ad..ff33c0f 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -1,16 +1,19 @@
-flask
-sqlalchemy
-pymysql
-pyyaml
-requests
-pandas
-openpyxl
-xlrd>=2.0.1
-google-analytics-data
-prophet
-statsmodels
-scikit-learn
-customtkinter
-tkcalendar
-tabulate
-watchdog
\ No newline at end of file
+flask==3.0.0
+sqlalchemy==2.0.23
+pymysql==1.1.0
+pyyaml==6.0.1
+requests==2.31.0
+pandas==2.1.3
+openpyxl==3.1.2
+xlrd==2.0.1
+google-analytics-data==0.18.5
+prophet==1.1.5
+statsmodels==0.14.0
+scikit-learn==1.3.2
+matplotlib==3.8.2
+customtkinter==5.2.0
+tkcalendar==1.6.1
+tabulate==0.9.0
+watchdog==3.0.0
+python-dotenv==1.0.0
+pytz==2023.3
\ No newline at end of file
diff --git a/static.code-workspace b/static.code-workspace
new file mode 100644
index 0000000..86488a8
--- /dev/null
+++ b/static.code-workspace
@@ -0,0 +1,15 @@
+{
+ "folders": [
+ {
+ "name": "static",
+ "path": "."
+ }
+ ],
+ "settings": {
+ "python.defaultInterpreterPath": "${workspaceFolder}\\.venv\\Scripts\\python.exe",
+ "python.terminal.activateEnvironment": true,
+ "python.analysis.extraPaths": [
+ "${workspaceFolder}\\lib"
+ ]
+ }
+}
\ No newline at end of file