# -*- coding: utf-8 -*- """ V2-014 PublishHistoryTracker 역할: - blog_v2_publish_history 테이블에 V2 발행/노출/비공개 이력을 기록한다. - MariaDB 10.x 호환 테이블 기준. - V2 PublishAdapter, ExposureChecker, RemovedSyncAdapter에서 공통으로 사용한다. 주의: - 이 파일은 V1 운영 테이블을 수정하지 않는다. - INSERT/UPDATE 대상은 blog_v2_publish_history 뿐이다. """ import json from datetime import datetime class PublishHistoryTracker: table_name = "blog_v2_publish_history" def __init__(self, conn): self.conn = conn def create_pending_record( self, pipeline_id, realtor_id, article_no=None, draft_id=None, publish_queue_id=None, publish_title=None, search_keyword=None, meta=None, ): """ 발행 전 또는 발행 직후 추적 레코드를 만든다. 이미 동일 source/blog_id/blog_post_no가 없을 수 있으므로 blog_url 기준 unique를 사용하지 않는다. """ meta_text = json.dumps(meta or {}, ensure_ascii=False, default=str) sql = f""" INSERT INTO {self.table_name} ( pipeline_id, pipeline_version, source, realtor_id, article_no, draft_id, publish_queue_id, publish_title, search_keyword, publish_status, search_status, meta_text, created_at, updated_at ) VALUES ( %s, 'v2', 'v2', %s, %s, %s, %s, %s, %s, 'pending', 'pending', %s, NOW(), NOW() ) """ with self.conn.cursor() as cur: cur.execute( sql, ( pipeline_id, int(realtor_id), str(article_no) if article_no else None, int(draft_id) if draft_id else None, int(publish_queue_id) if publish_queue_id else None, publish_title, search_keyword, meta_text, ), ) history_id = cur.lastrowid self.conn.commit() return history_id def mark_published( self, history_id=None, publish_queue_id=None, blog_id=None, blog_post_no=None, blog_url=None, publish_title=None, published_at=None, meta=None, ): """ 발행 성공 상태로 업데이트한다. history_id 우선, 없으면 publish_queue_id 기준으로 업데이트한다. """ published_at = published_at or datetime.now() meta_text = json.dumps(meta or {}, ensure_ascii=False, default=str) if meta is not None else None sets = [ "publish_status = 'published'", "published_at = %s", "blog_id = %s", "blog_post_no = %s", "blog_url = %s", "publish_title = COALESCE(%s, publish_title)", "publish_hour = HOUR(%s)", "publish_weekday = WEEKDAY(%s)", "updated_at = NOW()", ] params = [ published_at, blog_id, blog_post_no, blog_url, publish_title, published_at, published_at, ] if meta_text is not None: sets.append("meta_text = %s") params.append(meta_text) where, where_params = self._where_clause(history_id, publish_queue_id) params.extend(where_params) sql = f""" UPDATE {self.table_name} SET {', '.join(sets)} {where} """ with self.conn.cursor() as cur: cur.execute(sql, params) affected = cur.rowcount self.conn.commit() return affected def mark_publish_failed( self, history_id=None, publish_queue_id=None, error_message="", ): where, where_params = self._where_clause(history_id, publish_queue_id) sql = f""" UPDATE {self.table_name} SET publish_status = 'failed', publish_error_message = %s, updated_at = NOW() {where} """ with self.conn.cursor() as cur: cur.execute(sql, [str(error_message)[:5000]] + where_params) affected = cur.rowcount self.conn.commit() return affected def mark_search_result( self, history_id=None, publish_queue_id=None, is_exposed=False, exposure_rank=None, exposure_page=None, search_status=None, error_message=None, ): """ 검색 노출 체크 결과 업데이트. """ now = datetime.now() final_status = search_status or ("exposed" if is_exposed else "not_exposed") sets = [ "search_status = %s", "is_exposed = %s", "exposure_rank = %s", "exposure_page = %s", "last_checked_at = %s", "check_count = check_count + 1", "updated_at = NOW()", ] params = [ final_status, 1 if is_exposed else 0, exposure_rank, exposure_page, now, ] if is_exposed: sets.append("first_exposed_at = IF(first_exposed_at IS NULL, %s, first_exposed_at)") sets.append("last_exposed_at = %s") params.extend([now, now]) if error_message: sets.append("error_count = error_count + 1") sets.append("last_error_message = %s") params.append(str(error_message)[:5000]) where, where_params = self._where_clause(history_id, publish_queue_id) params.extend(where_params) sql = f""" UPDATE {self.table_name} SET {', '.join(sets)} {where} """ with self.conn.cursor() as cur: cur.execute(sql, params) affected = cur.rowcount self.conn.commit() return affected def mark_private_synced( self, history_id=None, publish_queue_id=None, status="success", error_message=None, ): now = datetime.now() sql_status = "success" if status == "success" else "failed" where, where_params = self._where_clause(history_id, publish_queue_id) sql = f""" UPDATE {self.table_name} SET is_private_synced = %s, private_synced_at = %s, private_sync_status = %s, private_sync_error_message = %s, updated_at = NOW() {where} """ with self.conn.cursor() as cur: cur.execute( sql, [ 1 if status == "success" else 0, now if status == "success" else None, sql_status, str(error_message)[:5000] if error_message else None, ] + where_params, ) affected = cur.rowcount self.conn.commit() return affected def _where_clause(self, history_id=None, publish_queue_id=None): if history_id: return "WHERE id = %s", [int(history_id)] if publish_queue_id: return "WHERE publish_queue_id = %s", [int(publish_queue_id)] raise ValueError("history_id 또는 publish_queue_id 중 하나는 필요합니다.")