import json import logging from urllib.parse import urlparse from models.status import SourceType from common.utils import parse_iso_duration_seconds, format_seconds, format_clock, calc_avg_video_length, relative_date, calc_upload_frequency from common.db.run import update_run_report, update_run_plan, select_run_report_data, select_run from common.db.source import select_run_raw_data, select_mainpage_logo_url from common.db.market import select_market from integrations.llm.llm_service import LLMService from integrations.llm.prompt import report_prompt, plan_prompt, youtube_diagnosis_prompt, brand_consistency_prompt, critical_issues_prompt, transformation_prompt, roadmap_prompt, scores_prompt, other_channels_prompt from integrations.llm.schemas.report import ReportOutput, ClinicSnapshot, YouTubeAudit, BrandConsistencyOutput, CriticalIssuesOutput, DiagnosisItem, TransformationProposal, RoadmapOutput, RoadmapMonth, ScoresOutput, ChannelScore, WebsiteAudit, OtherChannelsOutput, OtherChannel from services.branding import analyze_branding from services.instagram_audit import build_instagram_audit from services.facebook_audit import build_facebook_audit from services.kpi_dashboard import build_kpi_dashboard from integrations.llm.schemas.plan import PlanOutput logger = logging.getLogger(__name__) async def generate_plan(analysis_run_id: str) -> PlanOutput: raw = await select_run_raw_data(analysis_run_id) clinic = raw.get(SourceType.MAINPAGE) or [] clinic = clinic[0]["raw_data"] # if not exist, must error branding = raw.get(SourceType.BRANDING) or [] branding = branding[0]["raw_data"] # if not exist, must error report = await select_run_report_data(analysis_run_id) market = await select_market(analysis_run_id) mainpage = raw.get(SourceType.MAINPAGE) or [] mainpage = mainpage[0]["raw_data"] # 유일 branding = raw.get(SourceType.BRANDING) or [] branding = branding[0]["raw_data"] # 유일 instagram = raw.get(SourceType.INSTAGRAM) or [] facebook = raw.get(SourceType.FACEBOOK) or [] youtube = raw.get(SourceType.YOUTUBE) or [] youtube = youtube[0]["raw_data"] if youtube else None # 유일 (기획상) gangnam_unni = raw.get(SourceType.GANGNAM_UNNI) or [] gangnam_unni = gangnam_unni[0]["raw_data"] if gangnam_unni else None# 유일 (기획상) naver_blog = raw.get(SourceType.NAVER_BLOG) or [] naver_blog = naver_blog[0]["raw_data"] if naver_blog else None# 유일 (기획상) tiktok = raw.get(SourceType.TIKTOK) or [] tiktok = tiktok[0]["raw_data"] if tiktok else None# 유일 (기획상) naver_cafe = raw.get(SourceType.NAVER_CAFE) or [] naver_cafe = naver_cafe[0]["raw_data"] if naver_cafe else None# 유일 (기획상) kakaotalk = raw.get(SourceType.KAKAOTALK) or [] kakaotalk = kakaotalk[0]["raw_data"] if kakaotalk else None# 유일 (기획상) def _json(v) -> str | None: return json.dumps(v, ensure_ascii=False) if v else None input_data = { "clinic_name": clinic.get("clinicName"), "clinic_name_en": clinic.get("clinicNameEn"), "address": clinic.get("address"), "phone": clinic.get("phone"), "slogan": clinic.get("slogan"), "services": json.dumps(clinic.get("services", []), ensure_ascii=False), "doctors": json.dumps(clinic.get("doctors", []), ensure_ascii=False), "report": _json(report), "market_competitors": _json(market.get("competitors")), "market_keywords": _json(market.get("keywords")), "market_trend": _json(market.get("trend")), "market_target_audience": _json(market.get("target_audience")), "tiktok": _json(tiktok), "instagram": _json(instagram), "facebook": _json(facebook), "naver_blog": _json(_naver_blog_summary(naver_blog)), "naver_cafe": _json(naver_cafe), "kakao_talk": _json(kakaotalk), "channel_logos": _json(branding.get("channelLogos")), "brand_assets": _json(branding.get("brandAssets")), } return await LLMService(provider="perplexity").generate(plan_prompt, input_data) def _build_clinic_snapshot(mainpage: dict, gangnam_unni: dict, brand_assets: dict, logo_url: str | None) -> dict: snapshot: dict = {} doctors = gangnam_unni.get("doctors", []) lead = max(doctors, key=lambda d: d.get("reviews", 0)) if doctors else None snapshot["name"] = mainpage["clinicName"] snapshot["name_en"] = mainpage["clinicNameEn"] snapshot["phone"] = mainpage["phone"] snapshot["location"] = mainpage["address"] snapshot["domain"] = mainpage.get("domain") or urlparse(mainpage.get("sourceUrl") or "").netloc if gangnam_unni.get("rating"): snapshot["overall_rating"] = gangnam_unni["rating"] if gangnam_unni.get("totalReviews"): snapshot["total_reviews"] = gangnam_unni["totalReviews"] if gangnam_unni.get("badges"): snapshot["certifications"] = gangnam_unni["badges"] if gangnam_unni.get("totalMajorStaffs"): snapshot["staff_count"] = gangnam_unni["totalMajorStaffs"] if lead: snapshot["lead_doctor"] = { "name": lead.get("name"), "credentials": lead.get("specialty"), "rating": lead.get("rating"), "review_count": lead.get("reviews"), } # logo URL 은 raw_info.logo_url 컬럼에서, brand_colors 는 JSON 에서 강제 주입. LLM 의 null 처리 차단. if logo_url: snapshot["logo_images"] = {"circle": None, "horizontal": logo_url, "korean": None} if brand_assets.get("brand_colors"): snapshot["brand_colors"] = brand_assets["brand_colors"] return ClinicSnapshot.model_validate(snapshot).model_dump() def _naver_blog_summary(blog: dict | None) -> dict | None: """plan 카드 한 장에 들어가는 건 전체 포스트 수와 최근 활동 시점뿐. 그 외(본문·링크·제목)는 던져봐야 토큰만 늘고 LLM 이 무관 정보로 hallucinate 함.""" if not blog: return None posts = blog.get("posts") or [] return { "totalPosts": blog.get("totalResults"), "latestPostDate": posts[0].get("postDate") if posts else None, } async def _build_youtube_audit(youtube: dict) -> dict: # 기획상 1개의 input channel, 다중 채널은 기획에 없음. videos = youtube.get("videos", []) recents = youtube.get("recents", []) yt_patch: dict = { "weekly_view_growth": {"absolute": 0, "percentage": 0.0}, "estimated_monthly_revenue": {"min": 0, "max": 0}, "linked_urls": [], "avg_video_length": calc_avg_video_length(videos), "upload_frequency": calc_upload_frequency(recents), } if youtube.get("channelName"): yt_patch["channel_name"] = youtube["channelName"] if youtube.get("handle"): yt_patch["handle"] = youtube["handle"] if youtube.get("subscribers"): yt_patch["subscribers"] = youtube["subscribers"] if youtube.get("totalVideos"): yt_patch["total_videos"] = youtube["totalVideos"] if youtube.get("totalViews"): yt_patch["total_views"] = youtube["totalViews"] if youtube.get("publishedAt"): yt_patch["channel_created_date"] = youtube["publishedAt"][:10] yt_patch["channel_description"] = youtube.get("description") or "" if youtube.get("playlists"): yt_patch["playlists"] = youtube["playlists"] if videos: yt_patch["top_videos"] = [ { "title": v["title"], "views": v["views"], "duration": format_clock(parse_iso_duration_seconds(v.get("duration", ""))), "type": "Short" if "M" not in v.get("duration", "") else "Long", "uploaded_ago": relative_date(v.get("date", "")), } for v in videos ] diagnosis_result = await LLMService(provider="perplexity").generate( youtube_diagnosis_prompt, { "channel_name": yt_patch.get("channel_name"), "subscribers": yt_patch.get("subscribers"), "total_videos": yt_patch.get("total_videos"), "total_views": yt_patch.get("total_views"), "avg_video_length": yt_patch.get("avg_video_length"), "upload_frequency": yt_patch.get("upload_frequency"), "top_videos": json.dumps(yt_patch.get("top_videos", []), ensure_ascii=False), "playlists": json.dumps(yt_patch.get("playlists", []), ensure_ascii=False), }, ) yt_patch["diagnosis"] = [item.model_dump() for item in diagnosis_result.diagnosis] return YouTubeAudit.model_validate(yt_patch).model_dump() async def _build_roadmap(analysis_run_id: str, raw: dict) -> list[dict]: result: RoadmapOutput = await LLMService(provider="perplexity").generate( roadmap_prompt, { "clinic_name": (raw.get(SourceType.MAINPAGE) or [{}])[0]["raw_data"].get("clinicName"), "data": json.dumps(raw, ensure_ascii=False), }, ) return [RoadmapMonth.model_validate(item).model_dump() for item in result.roadmap] async def _build_transformation(analysis_run_id: str, raw: dict) -> dict: result: TransformationProposal = await LLMService(provider="perplexity").generate( transformation_prompt, { "clinic_name": (raw.get(SourceType.MAINPAGE) or [{}])[0]["raw_data"].get("clinicName"), "data": json.dumps(raw, ensure_ascii=False), }, ) return result.model_dump() async def _build_critical_issues(analysis_run_id: str, raw: dict) -> list[dict]: result: CriticalIssuesOutput = await LLMService(provider="perplexity").generate( critical_issues_prompt, { "clinic_name": (raw.get(SourceType.MAINPAGE) or [{}])[0]["raw_data"].get("clinicName"), "data": json.dumps(raw, ensure_ascii=False), }, ) return [DiagnosisItem.model_validate(item).model_dump() for item in result.diagnosis] async def _build_scores(analysis_run_id: str, raw: dict) -> ScoresOutput: return await LLMService(provider="perplexity").generate( scores_prompt, { "clinic_name": (raw.get(SourceType.MAINPAGE) or [{}])[0]["raw_data"].get("clinicName"), "data": json.dumps(raw, ensure_ascii=False), }, ) def _build_website_audit(mainpage: dict) -> dict: """mainpage raw_data 에서 직접 매핑. LLM 미경유. Firecrawl 의 raw HTML 을 collect_mainpage 가 정규식 파싱해서 tracking/SNS/domain 까지 mainpage 에 다 박아둠.""" domain = mainpage.get("domain") or urlparse(mainpage.get("sourceUrl") or "").netloc sns_links = mainpage.get("snsLinks") or [] audit = { "primary_domain": domain, "additional_domains": mainpage.get("additionalDomains") or [], "sns_links_on_site": bool(sns_links), "sns_links_detail": sns_links or None, "tracking_pixels": mainpage.get("trackingPixels") or [], "main_cta": mainpage.get("mainCta") or "", } return WebsiteAudit.model_validate(audit).model_dump() async def _build_other_channels(raw: dict) -> list[dict]: result: OtherChannelsOutput = await LLMService(provider="perplexity").generate( other_channels_prompt, { "clinic_name": (raw.get(SourceType.MAINPAGE) or [{}])[0]["raw_data"].get("clinicName"), "tiktok": json.dumps(raw.get(SourceType.TIKTOK), ensure_ascii=False), "kakao_talk": json.dumps(raw.get(SourceType.KAKAOTALK), ensure_ascii=False), "naver_cafe": json.dumps(raw.get(SourceType.NAVER_CAFE), ensure_ascii=False), "naver_blog": json.dumps(raw.get(SourceType.NAVER_BLOG), ensure_ascii=False), "gangnam_unni": json.dumps(raw.get(SourceType.GANGNAM_UNNI), ensure_ascii=False), }, ) return [OtherChannel.model_validate(item).model_dump() for item in result.other_channels] async def _build_report(analysis_run_id: str) -> dict: raw = await select_run_raw_data(analysis_run_id) run = await select_run(analysis_run_id) if not raw: return {} mainpage = raw.get(SourceType.MAINPAGE) or [] mainpage = mainpage[0]["raw_data"] # 유일 branding = raw.get(SourceType.BRANDING) or [] branding = branding[0]["raw_data"] # 유일 instagram = raw.get(SourceType.INSTAGRAM) or [] facebook = raw.get(SourceType.FACEBOOK) or [] youtube = raw.get(SourceType.YOUTUBE) or [] youtube = youtube[0]["raw_data"] if youtube else None # 유일 (기획상) gangnam_unni = raw.get(SourceType.GANGNAM_UNNI) or [] gangnam_unni = gangnam_unni[0]["raw_data"] if gangnam_unni else None# 유일 (기획상) naver_blog = raw.get(SourceType.NAVER_BLOG) or [] naver_blog = naver_blog[0]["raw_data"] if naver_blog else None# 유일 (기획상) tiktok = raw.get(SourceType.TIKTOK) or [] tiktok = tiktok[0]["raw_data"] if tiktok else None# 유일 (기획상) naver_cafe = raw.get(SourceType.NAVER_CAFE) or [] naver_cafe = naver_cafe[0]["raw_data"] if naver_cafe else None# 유일 (기획상) brand_assets = branding.get("brandAssets") or {} channel_logos = branding.get("channelLogos") or {} logo_url = await select_mainpage_logo_url(analysis_run_id) brand = await generate_brand_consistency(analysis_run_id) brand_patch : list[dict] = brand.model_dump()["brand_inconsistencies"] kpi_extras = { "tiktok": tiktok, "naverCafe": naver_cafe, } scores = (await _build_scores(analysis_run_id, raw)).model_dump() report = { "id" : analysis_run_id, "created_at" : str(run["created_at"]) if run.get("created_at") else None, "target_url" : mainpage.get("domain") or urlparse(mainpage.get("sourceUrl") or "").netloc, "overall_score" : scores.get("overall_score"), "channel_scores" : scores.get("channel_scores"), "clinic_snapshot": _build_clinic_snapshot(mainpage, gangnam_unni, brand_assets, logo_url), "instagram_audit": await build_instagram_audit(instagram, channel_logos), "facebook_audit": await build_facebook_audit(facebook, brand_patch, channel_logos), "youtube_audit": await _build_youtube_audit(youtube), "other_channels": await _build_other_channels(raw), "website_audit": _build_website_audit(mainpage), "problem_diagnosis": await _build_critical_issues(analysis_run_id, raw), "transformation" : await _build_transformation(analysis_run_id, raw), "roadmap": await _build_roadmap(analysis_run_id, raw), "kpi_dashboard": build_kpi_dashboard(instagram, facebook, youtube, gangnam_unni, kpi_extras, naver_blog), } return ReportOutput(**report) def _deep_merge(base: dict, overrides: dict) -> dict: """dict 끼리 만나면 재귀로 안쪽까지 합치고, 그 외(list/scalar/None) 는 override 값으로 통째 치환.""" for k, v in overrides.items(): if isinstance(v, dict) and isinstance(base.get(k), dict): _deep_merge(base[k], v) else: base[k] = v return base async def generate_brand_consistency(analysis_run_id: str) -> BrandConsistencyOutput: raw = await select_run_raw_data(analysis_run_id) def _json(v) -> str | None: return json.dumps(v, ensure_ascii=False) if v else None mainpage = raw.get(SourceType.MAINPAGE) or [] input_data = { "clinic_name": (mainpage[0].get("clinicName") if mainpage else None), "mainpage": _json(mainpage), "instagram": _json(raw.get(SourceType.INSTAGRAM)), "facebook": _json(raw.get(SourceType.FACEBOOK)), "youtube": _json(raw.get(SourceType.YOUTUBE)), "gangnam_unni": _json(raw.get(SourceType.GANGNAM_UNNI)), } return await LLMService(provider="perplexity").generate(brand_consistency_prompt, input_data) async def run_report_task(analysis_run_id: str) -> None: logger.info("[report] start run=%s", analysis_run_id) await analyze_branding(analysis_run_id) # result = await generate_report(analysis_run_id) result = await _build_report(analysis_run_id) await update_run_report(analysis_run_id, result.model_dump()) logger.info("[report] done run=%s", analysis_run_id) def _patch_plan(result: PlanOutput, logo_desc: str) -> PlanOutput: """brand_guide.channel_branding[].profile_photo 는 LLM 안 맡기고 코드가 박는다 (모든 채널 동일값 = brand_assets.logo_description). LLM 이 fallback 문구 hallucinate 방지.""" p = result.model_dump() for ch in (p.get("brand_guide") or {}).get("channel_branding") or []: ch["profile_photo"] = logo_desc return PlanOutput(**p) async def run_plan_task(analysis_run_id: str) -> None: logger.info("[plan] start run=%s", analysis_run_id) result = await generate_plan(analysis_run_id) # profile_photo 는 brand_assets.logo_description 으로 코드가 박음 (LLM "(가이드 미보유)" 같은 hallucination 차단). raw = await select_run_raw_data(analysis_run_id) branding_list = raw.get(SourceType.BRANDING) or [] branding_data = branding_list[0]["raw_data"] if branding_list else {} logo_desc = (branding_data.get("brandAssets") or {}).get("logo_description") or "" result = _patch_plan(result, logo_desc) await update_run_plan(analysis_run_id, result.model_dump()) logger.info("[plan] done run=%s", analysis_run_id)