feat: MySQL DB 정규화(Master/History) 및 시계열 데이터 수집 시스템 통합

1. 마스터/히스토리 테이블 분리 및 마이그레이션 완료\n2. 날짜별 데이터 축적 및 대시보드 필터링 기능 추가\n3. Playwright 수집 로직(날짜필터, 좌표클릭, 정밀합산) 완전 복구
This commit is contained in:
2026-03-10 16:24:13 +09:00
parent 743cce543b
commit 4a995c11f4
9 changed files with 268 additions and 89 deletions

View File

@@ -54,24 +54,57 @@ def get_db_connection():
cursorclass=pymysql.cursors.DictCursor
)
@app.get("/project-data")
async def get_project_data():
@app.get("/available-dates")
async def get_available_dates():
"""
MySQL overseas_projects 테이블에서 프로젝트 현황 데이터를 반환
히스토리 테이블에서 유니크한 크롤링 날짜 목록을 반환
"""
try:
conn = get_db_connection()
try:
with conn.cursor() as cursor:
# 대시보드에 필요한 모든 정보를 쿼리 (short_nm 포함)
cursor.execute("SELECT project_nm, short_nm, department, master, recent_log, file_count, continent, country FROM overseas_projects ORDER BY id ASC")
cursor.execute("SELECT DISTINCT crawl_date FROM projects_history ORDER BY crawl_date DESC")
rows = cursor.fetchall()
dates = [row['crawl_date'].strftime("%Y.%m.%d") for row in rows if row['crawl_date']]
return dates
finally:
conn.close()
except Exception as e:
return {"error": str(e)}
@app.get("/project-data")
async def get_project_data(date: str = None):
"""
특정 날짜의 데이터를 JOIN하여 반환
"""
try:
conn = get_db_connection()
try:
with conn.cursor() as cursor:
if not date or date == "-":
cursor.execute("SELECT MAX(crawl_date) as last_date FROM projects_history")
target_date_row = cursor.fetchone()
target_date = target_date_row['last_date']
else:
target_date = date.replace(".", "-")
if not target_date:
return {"projects": [], "last_updated": "-"}
# 마스터 정보와 히스토리 정보를 JOIN
sql = """
SELECT m.project_nm, m.short_nm, m.department, m.master,
h.recent_log, h.file_count, m.continent, m.country
FROM projects_master m
JOIN projects_history h ON m.project_id = h.project_id
WHERE h.crawl_date = %s
ORDER BY m.project_id ASC
"""
cursor.execute(sql, (target_date,))
rows = cursor.fetchall()
# 프론트엔드 기대 형식에 맞춰 반환
# [표시될 프로젝트명(short_nm), 담당부서, 담당자, 최근활동로그, 파일수, 대륙, 국가]
projects = []
for row in rows:
# short_nm이 있으면 그것을 쓰고, 없으면 project_nm 사용
display_name = row['short_nm'] if row['short_nm'] and row['short_nm'].strip() else row['project_nm']
projects.append([
display_name,
@@ -82,11 +115,11 @@ async def get_project_data():
row['continent'],
row['country']
])
return projects
return {"projects": projects, "last_updated": target_date.strftime("%Y.%m.%d") if hasattr(target_date, 'strftime') else str(target_date).replace("-", ".")}
finally:
conn.close()
except Exception as e:
print(f"Error fetching from DB: {e}")
return {"error": str(e)}
@app.get("/")