""" MySQL 저장 모듈 - 업체 정보, 리뷰, 순위 데이터 저장 """ import re as _re import mysql.connector from datetime import datetime, date as _date # ───────────────────────────────────────── # DB 설정 (환경에 맞게 수정) # ───────────────────────────────────────── DB_CONFIG = { "host" : "localhost", "user" : "place_user", "password": "PlacePass1!", "database": "naver_place_monitor", "charset" : "utf8mb4", } def get_conn(): return mysql.connector.connect(**DB_CONFIG) def parse_naver_date(date_str: str): """'6.10.수' 형태의 네이버 날짜 → date 객체 (연도는 오늘 기준 추정)""" if not date_str: return None m = _re.match(r'(\d+)\.(\d+)', str(date_str)) if not m: return None month, day = int(m.group(1)), int(m.group(2)) today = _date.today() year = today.year if month > today.month or (month == today.month and day > today.day): year -= 1 try: return _date(year, month, day) except Exception: return None # ───────────────────────────────────────── # 테이블 생성 (최초 1회 실행) # ───────────────────────────────────────── CREATE_TABLES_SQL = """ -- 고객사 업체 목록 CREATE TABLE IF NOT EXISTS places ( id INT AUTO_INCREMENT PRIMARY KEY, place_id VARCHAR(20) NOT NULL UNIQUE COMMENT '네이버 플레이스 ID', client_name VARCHAR(100) COMMENT '고객사명', name VARCHAR(200) COMMENT '업체명', category VARCHAR(100), address VARCHAR(300), phone VARCHAR(30), keywords TEXT COMMENT '모니터링 키워드 (쉼표 구분)', is_active TINYINT DEFAULT 1, created_at DATETIME DEFAULT CURRENT_TIMESTAMP ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; -- 일별 통계 스냅샷 CREATE TABLE IF NOT EXISTS place_stats ( id INT AUTO_INCREMENT PRIMARY KEY, place_id VARCHAR(20) NOT NULL, rating FLOAT, visitor_review_count INT DEFAULT 0, blog_review_count INT DEFAULT 0, collected_at DATETIME DEFAULT CURRENT_TIMESTAMP, INDEX idx_place_id (place_id), INDEX idx_collected_at (collected_at) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; -- 리뷰 저장 CREATE TABLE IF NOT EXISTS place_reviews ( id INT AUTO_INCREMENT PRIMARY KEY, place_id VARCHAR(20) NOT NULL, review_id VARCHAR(50) UNIQUE, author VARCHAR(100), rating FLOAT, content TEXT, visit_count INT DEFAULT 0, review_date VARCHAR(30), review_date_parsed DATE NULL, collected_at DATETIME DEFAULT CURRENT_TIMESTAMP, INDEX idx_place_id (place_id), INDEX idx_review_date_parsed (review_date_parsed) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; -- 검색 순위 기록 CREATE TABLE IF NOT EXISTS place_ranks ( id INT AUTO_INCREMENT PRIMARY KEY, place_id VARCHAR(20) NOT NULL, keyword VARCHAR(200) NOT NULL, my_rank INT COMMENT 'NULL이면 순위권 밖', total_found INT, searched_at DATETIME DEFAULT CURRENT_TIMESTAMP, INDEX idx_place_id (place_id), INDEX idx_keyword (keyword), INDEX idx_searched_at (searched_at) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; -- 리뷰 키워드 분석 CREATE TABLE IF NOT EXISTS review_keywords ( id INT AUTO_INCREMENT PRIMARY KEY, place_id VARCHAR(20) NOT NULL, keyword VARCHAR(100) NOT NULL, count INT DEFAULT 0, analyzed_at DATETIME DEFAULT CURRENT_TIMESTAMP, INDEX idx_place_id (place_id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; """ def init_db(): """DB 및 테이블 초기화""" conn = get_conn() cursor = conn.cursor() for sql in CREATE_TABLES_SQL.strip().split(";"): sql = sql.strip() if sql: cursor.execute(sql) conn.commit() cursor.close() conn.close() print("✅ DB 초기화 완료") # ───────────────────────────────────────── # 저장 함수들 # ───────────────────────────────────────── def save_stats(summary: dict): """업체 통계 저장""" if not summary: return conn = get_conn() cursor = conn.cursor() sql = """ INSERT INTO place_stats (place_id, rating, visitor_review_count, blog_review_count, collected_at) VALUES (%s, %s, %s, %s, %s) """ cursor.execute(sql, ( summary["place_id"], summary.get("rating", 0), summary.get("visitor_review_count", 0), summary.get("blog_review_count", 0), summary.get("collected_at"), )) conn.commit() cursor.close() conn.close() def save_reviews(place_id: str, reviews: list): """리뷰 저장 (중복 무시)""" if not reviews: return conn = get_conn() cursor = conn.cursor() sql = """ INSERT IGNORE INTO place_reviews (place_id, review_id, author, rating, content, visit_count, review_date, review_date_parsed) VALUES (%s, %s, %s, %s, %s, %s, %s, %s) """ rows = [ ( place_id, r.get("review_id"), r.get("author"), r.get("rating"), r.get("content"), r.get("visit_count"), r.get("created_at"), parse_naver_date(r.get("created_at")), ) for r in reviews ] cursor.executemany(sql, rows) conn.commit() new_count = cursor.rowcount cursor.close() conn.close() return new_count def save_rank(rank_data: dict): """순위 저장""" if not rank_data: return conn = get_conn() cursor = conn.cursor() # 내 업체 순위 저장 sql = """ INSERT INTO place_ranks (place_id, keyword, my_rank, total_found, searched_at) VALUES (%s, %s, %s, %s, %s) """ # place_id는 경쟁업체 목록에서 내 업체 찾기 my_place = next( (c for c in rank_data.get("competitors", []) if c["is_my_place"]), {} ) my_place_id = my_place.get("place_id", "") cursor.execute(sql, ( my_place_id, rank_data["keyword"], rank_data.get("my_rank"), rank_data.get("total_found"), rank_data.get("searched_at"), )) conn.commit() cursor.close() conn.close() def save_keywords(place_id: str, keywords: list): """키워드 분석 결과 저장""" if not keywords: return conn = get_conn() cursor = conn.cursor() # 오늘 데이터 삭제 후 재삽입 cursor.execute( "DELETE FROM review_keywords WHERE place_id=%s AND DATE(analyzed_at)=CURDATE()", (place_id,) ) sql = "INSERT INTO review_keywords (place_id, keyword, count) VALUES (%s, %s, %s)" rows = [(place_id, kw["keyword"], kw["count"]) for kw in keywords] cursor.executemany(sql, rows) conn.commit() cursor.close() conn.close() def get_active_places() -> list: """활성 업체 목록 조회""" conn = get_conn() cursor = conn.cursor(dictionary=True) cursor.execute("SELECT * FROM places WHERE is_active=1") rows = cursor.fetchall() cursor.close() conn.close() return rows def update_place_info(place_id: str, summary: dict): """크롤링한 기본정보로 places 테이블 업데이트""" conn = get_conn() cursor = conn.cursor() cursor.execute( "UPDATE places SET name=%s, category=%s, address=%s, phone=%s WHERE place_id=%s", ( summary.get("name", ""), summary.get("category", ""), summary.get("address", ""), summary.get("phone", ""), place_id, ), ) conn.commit() cursor.close() conn.close() if __name__ == "__main__": init_db()