# -*- coding: utf-8 -*- import os import sys import json import argparse from datetime import datetime BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) sys.path.append(BASE_DIR) from db import get_conn from services.fin_land_service import collect_fin_land_article_detail from services.image_downloader import download_images def fetch_pending_articles(conn, limit=10, realtor_id=None, article_no=None): sql = """ SELECT id, realtor_id, article_no, article_url, collect_status FROM blog_realtor_articles WHERE 1=1 """ params = [] if article_no: sql += " AND article_no = %s " params.append(str(article_no)) else: sql += """ AND ( collect_status IS NULL OR collect_status = '' OR collect_status = 'pending' OR collect_status = 'failed' ) """ if realtor_id: sql += " AND realtor_id = %s " params.append(int(realtor_id)) sql += """ ORDER BY id ASC LIMIT %s """ params.append(int(limit)) with conn.cursor() as cur: cur.execute(sql, params) return cur.fetchall() def save_article_detail(conn, article, detail): article_no = str(article["article_no"]) with conn.cursor() as cur: cur.execute(""" INSERT INTO blog_realestate_article_details ( realtor_article_id, article_no, complex_number, complex_name, article_title, article_desc, trade_type, real_estate_type, price_text, building_name, address, road_address, floor_info, area_info, supply_area, exclusive_area, room_count, bathroom_count, direction, parking_info, move_in_date, building_usage, realtor_office_name, realtor_name, realtor_phone, collect_status, raw_json, created_at, updated_at ) VALUES ( %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, 'collected', %s, NOW(), NOW() ) ON DUPLICATE KEY UPDATE realtor_article_id = VALUES(realtor_article_id), complex_number = VALUES(complex_number), complex_name = VALUES(complex_name), article_title = VALUES(article_title), article_desc = VALUES(article_desc), trade_type = VALUES(trade_type), real_estate_type = VALUES(real_estate_type), price_text = VALUES(price_text), building_name = VALUES(building_name), address = VALUES(address), road_address = VALUES(road_address), floor_info = VALUES(floor_info), area_info = VALUES(area_info), supply_area = VALUES(supply_area), exclusive_area = VALUES(exclusive_area), room_count = VALUES(room_count), bathroom_count = VALUES(bathroom_count), direction = VALUES(direction), parking_info = VALUES(parking_info), move_in_date = VALUES(move_in_date), building_usage = VALUES(building_usage), realtor_office_name = VALUES(realtor_office_name), realtor_name = VALUES(realtor_name), realtor_phone = VALUES(realtor_phone), collect_status = 'collected', raw_json = VALUES(raw_json), updated_at = NOW() """, ( article["id"], article_no, str(detail.get("complex_number") or ""), detail.get("complex_name", ""), detail.get("article_name", ""), detail.get("article_feature", ""), detail.get("trade_type_name") or detail.get("trade_type", ""), detail.get("real_estate_type_name") or detail.get("real_estate_type", ""), detail.get("price_text", ""), detail.get("complex_name") or detail.get("article_name", ""), detail.get("address", ""), detail.get("road_address", ""), detail.get("floor_info", ""), detail.get("area_info", ""), str(detail.get("supply_space") or ""), str(detail.get("exclusive_space") or ""), str(detail.get("room_count") or ""), str(detail.get("bathroom_count") or ""), detail.get("direction", ""), detail.get("parking_info", ""), detail.get("move_in_date", ""), detail.get("building_use", ""), detail.get("realtor_office_name", ""), detail.get("realtor_name", ""), detail.get("realtor_phone", ""), json.dumps(detail, ensure_ascii=False), )) def update_first_posted_at(conn, article_no, first_posted_at): if not first_posted_at: return with conn.cursor() as cur: cur.execute(""" UPDATE blog_realtor_articles SET first_posted_at = %s, updated_at = NOW() WHERE article_no = %s """, ( first_posted_at, str(article_no), )) def delete_existing_images(conn, article_no): with conn.cursor() as cur: cur.execute(""" DELETE FROM blog_realestate_article_images WHERE article_no = %s """, (str(article_no),)) def delete_existing_prices(conn, article_no): with conn.cursor() as cur: cur.execute(""" DELETE FROM blog_realestate_article_prices WHERE article_no = %s """, (str(article_no),)) def delete_existing_schools(conn, article_no): with conn.cursor() as cur: cur.execute(""" DELETE FROM blog_realestate_article_schools WHERE article_no = %s """, (str(article_no),)) def save_article_images(conn, article_no, downloaded_images): with conn.cursor() as cur: for idx, img in enumerate(downloaded_images, start=1): cur.execute(""" INSERT INTO blog_realestate_article_images ( article_no, image_type, image_category, image_source, image_url, local_path, file_name, file_size, width, height, sort_order, is_representative, raw_json, created_at, updated_at ) VALUES ( %s, %s, %s, 'naver_land', %s, %s, %s, %s, %s, %s, %s, %s, %s, NOW(), NOW() ) ON DUPLICATE KEY UPDATE image_type = VALUES(image_type), image_category = VALUES(image_category), image_source = VALUES(image_source), image_url = VALUES(image_url), local_path = VALUES(local_path), file_name = VALUES(file_name), file_size = VALUES(file_size), width = VALUES(width), height = VALUES(height), sort_order = VALUES(sort_order), is_representative = VALUES(is_representative), raw_json = VALUES(raw_json), updated_at = NOW() """, ( str(article_no), img.get("image_category", "building"), img.get("image_category", "building"), img.get("image_url", ""), img.get("local_public_url") or img.get("public_url") or img.get("local_path", ""), img.get("file_name", ""), int(img.get("file_size") or 0), int(img.get("width") or 0), int(img.get("height") or 0), idx, 1 if idx == 1 else 0, json.dumps(img, ensure_ascii=False), )) def save_market_prices(conn, article_no, detail): prices = detail.get("market_prices", []) or [] complex_number = str(detail.get("complex_number") or "") with conn.cursor() as cur: for item in prices: price_type = item.get("price_type", "detected_price") trade_type = detail.get("trade_type_name") or detail.get("trade_type", "") price_text = item.get("price_text", "") price_amount = int(item.get("price_amount") or 0) deal_date = item.get("deal_date", "") area_name = detail.get("supply_space_name") or detail.get("exclusive_space_name") or "" floor_info = detail.get("floor_info", "") cur.execute(""" INSERT INTO blog_realestate_article_prices ( article_no, complex_number, price_type, trade_type, price_text, price_amount, deal_date, area_name, floor_info, raw_json, created_at, updated_at ) VALUES ( %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, NOW(), NOW() ) ON DUPLICATE KEY UPDATE price_text = VALUES(price_text), price_amount = VALUES(price_amount), raw_json = VALUES(raw_json), updated_at = NOW() """, ( str(article_no), complex_number, price_type, trade_type, price_text, price_amount, deal_date, area_name, floor_info, json.dumps(item, ensure_ascii=False), )) def save_schools(conn, article_no, detail): schools = detail.get("schools", []) or [] complex_number = str(detail.get("complex_number") or "") with conn.cursor() as cur: for item in schools: school_name = item.get("school_name", "") if not school_name: continue cur.execute(""" INSERT INTO blog_realestate_article_schools ( article_no, complex_number, school_name, school_type, distance_text, address, tel, homepage, raw_json, created_at, updated_at ) VALUES ( %s, %s, %s, %s, %s, %s, %s, %s, %s, NOW(), NOW() ) ON DUPLICATE KEY UPDATE school_type = VALUES(school_type), distance_text = VALUES(distance_text), address = VALUES(address), tel = VALUES(tel), homepage = VALUES(homepage), raw_json = VALUES(raw_json), updated_at = NOW() """, ( str(article_no), complex_number, school_name, item.get("school_type", ""), item.get("distance_text", ""), item.get("address", ""), item.get("tel", ""), item.get("homepage", ""), json.dumps(item, ensure_ascii=False), )) def update_article_status(conn, article_id, status, message=""): with conn.cursor() as cur: cur.execute(""" UPDATE blog_realtor_articles SET collect_status = %s, error_message = %s, updated_at = NOW() WHERE id = %s """, ( status, message or "", article_id, )) def process_article(conn, article, force=False): article_no = str(article["article_no"]) print("-" * 80) print(f"[DETAIL START] article_no={article_no}") print(f"[FORCE] {force}") try: detail = collect_fin_land_article_detail(article_no) if force: delete_existing_images(conn, article_no) delete_existing_prices(conn, article_no) delete_existing_schools(conn, article_no) save_article_detail(conn, article, detail) update_first_posted_at( conn=conn, article_no=article_no, first_posted_at=detail.get("first_posted_at"), ) images = detail.get("images", []) or [] downloaded_images = download_images(article_no, images) if downloaded_images: save_article_images(conn, article_no, downloaded_images) price_error = "" school_error = "" try: save_market_prices(conn, article_no, detail) except Exception as e: price_error = str(e) print(f"[PRICE SAVE WARNING] {price_error}") try: save_schools(conn, article_no, detail) except Exception as e: school_error = str(e) print(f"[SCHOOL SAVE WARNING] {school_error}") message = ( f"상세수집 완료 / " f"최초게재일 {detail.get('first_posted_at') or '-'} / " f"이미지 {len(downloaded_images)}건 / " f"시세후보 {len(detail.get('market_prices', []) or [])}건 / " f"학군 {len(detail.get('schools', []) or [])}건" ) if price_error: message += f" / price_warning={price_error[:300]}" if school_error: message += f" / school_warning={school_error[:300]}" update_article_status(conn, article["id"], "collected", message) conn.commit() print(f"[DETAIL SUCCESS] article_no={article_no}") print(f"[FIRST POSTED] {detail.get('first_posted_at')}") print(f"[INFO] {message}") except Exception as e: conn.rollback() update_article_status(conn, article["id"], "failed", str(e)) conn.commit() print(f"[DETAIL FAILED] article_no={article_no}") print(f"[ERROR] {str(e)}") raise def main(): parser = argparse.ArgumentParser() parser.add_argument("--realtor-id", type=int, default=None) parser.add_argument("--article-no", type=str, default=None) parser.add_argument("--limit", type=int, default=9999) parser.add_argument("--force", action="store_true") args = parser.parse_args() conn = get_conn() try: print("=" * 80) print("[START] collect_article_details_job") print("TIME:", datetime.now().strftime("%Y-%m-%d %H:%M:%S")) print("=" * 80) articles = fetch_pending_articles( conn=conn, limit=args.limit, realtor_id=args.realtor_id, article_no=args.article_no, ) print(f"[INFO] target articles: {len(articles)}") for article in articles: process_article( conn=conn, article=article, force=args.force, ) print("=" * 80) print("[DONE] collect_article_details_job") print("=" * 80) finally: conn.close() if __name__ == "__main__": main()