"""분석 서비스 — 엔티티 추출 + 씬 스틸 추출 오케스트레이션.""" import json import logging import os import uuid from datetime import datetime, timezone from typing import Any, Dict, List, Optional logger = logging.getLogger(__name__) from sqlalchemy.orm import Session as OrmSession from app.core.config import settings from app.core.errors import AppError from app.i18n.loader import t from app.logging.activity_logger import ActivityLogger from app.models.project import ( Episode, EntityCanon, EntityAlias, EntityEpisodeLink, ImageAsset, RelationFact, RelationParticipant, SceneStill, CharacterOutlook, ) from app.modules.entity_extractor import extract_entities, ENTITY_PROMPT_DIR from app.modules.progress_tracker import ProgressTracker from app.modules.provenance import ProvenanceRecorder from app.modules.scene_still_extractor import extract_scene_stills, STILL_PROMPT_DIR from app.modules.llm.openai_client import OpenAIClient def _now() -> str: return datetime.now(timezone.utc).isoformat() def _new_id() -> str: return str(uuid.uuid4()) class AnalysisService: def __init__( self, db: OrmSession, project_id: str, actor_id: str, ) -> None: self._db = db self._logger = ActivityLogger(db) self._project_id = project_id self._actor_id = actor_id def _get_episode(self, episode_id: str) -> Episode: ep = ( self._db.query(Episode) .filter(Episode.id == episode_id, Episode.project_id == self._project_id) .first() ) if not ep: raise AppError( code="episode.not_found", message=t("episode.not_found"), status_code=404, ) return ep def _find_or_create_canon( self, entity: Dict[str, Any], now: str, ) -> EntityCanon: """Find existing canon by name/alias or create a new one.""" name = entity["name"] entity_type = entity["entity_type"] # Try exact name match within project existing = ( self._db.query(EntityCanon) .filter( EntityCanon.project_id == self._project_id, EntityCanon.name == name, EntityCanon.entity_type == entity_type, ) .first() ) if existing: return existing # Try alias match within project alias_match = ( self._db.query(EntityAlias) .join(EntityCanon, EntityCanon.id == EntityAlias.canon_id) .filter( EntityCanon.project_id == self._project_id, EntityAlias.alias == name, EntityCanon.entity_type == entity_type, ) .first() ) if alias_match: return self._db.query(EntityCanon).filter(EntityCanon.id == alias_match.canon_id).first() # Create new canon canon = EntityCanon( id=_new_id(), project_id=self._project_id, entity_type=entity_type, name=name, description=entity.get("description", ""), stable_traits=entity.get("stable_traits", "{}"), status="active", created_at=now, updated_at=now, ) self._db.add(canon) self._db.flush() return canon def _save_entities_v2( self, entity_result: Dict[str, Any], episode_id: str, ) -> Dict[str, str]: """v2 요소 추출 결과를 DB에 저장. 변형은 별도 EntityCanon + RelationFact. Returns: name_to_canon_id 매핑 """ # NOTE: name→id 매핑. 동일 이름의 다른 타입이 있으면 마지막 것이 남음. # 씬 visible_entities 매칭이 이름 기반이므로 현재 구조상 불가피. # 실제 시나리오에서 인물/소품 동명 확률은 극히 낮음. name_to_canon: Dict[str, str] = {} def _upsert_entity(name: str, entity_type: str, description: str, traits_json: str = "{}", t2i_prompt: str = "") -> str: existing = ( self._db.query(EntityCanon) .filter(EntityCanon.project_id == self._project_id, EntityCanon.name == name, EntityCanon.entity_type == entity_type) .first() ) if existing: existing.description = description existing.stable_traits = traits_json if t2i_prompt: existing.t2i_prompt = t2i_prompt existing.updated_at = _now() canon_id = existing.id else: canon_id = _new_id() self._db.add(EntityCanon( id=canon_id, project_id=self._project_id, entity_type=entity_type, name=name, description=description, stable_traits=traits_json, t2i_prompt=t2i_prompt, status="active", created_at=_now(), updated_at=_now(), )) # Episode link existing_link = ( self._db.query(EntityEpisodeLink) .filter(EntityEpisodeLink.canon_id == canon_id, EntityEpisodeLink.episode_id == episode_id) .first() ) if not existing_link: self._db.add(EntityEpisodeLink( id=_new_id(), project_id=self._project_id, canon_id=canon_id, episode_id=episode_id, )) name_to_canon[name] = canon_id return canon_id def _add_variant_relation(base_id: str, variant_id: str) -> None: """변형→기본 시각적 의존성 관계 추가.""" rel_id = _new_id() self._db.add(RelationFact( id=rel_id, project_id=self._project_id, relation_family="identity", relation_type="visual_variant", directionality="directed", temporal_scope="persistent", continuity_priority="critical", continuity_reason="시각적 변형 — 기본 요소에 의존", created_at=_now(), )) # base (order 1) ← variant depends on base self._db.add(RelationParticipant( id=_new_id(), relation_id=rel_id, canon_id=base_id, participant_role="base", participant_order=1, )) self._db.add(RelationParticipant( id=_new_id(), relation_id=rel_id, canon_id=variant_id, participant_role="variant", participant_order=2, )) # Characters (변형/관계 없음 — 씬 T2I 프롬프트에서 처리) for c in entity_result.get("characters", []): traits = json.dumps({ "visual_anchor_traits": c.get("visual_traits", []), }, ensure_ascii=False) _upsert_entity(c["name"], "character", c.get("description", ""), traits, c.get("t2i_prompt", "")) # Locations for loc in entity_result.get("locations", []): _upsert_entity(loc["name"], "location", loc.get("description", ""), "{}", loc.get("t2i_prompt", "")) # Props for p in entity_result.get("props", []): _upsert_entity(p["name"], "prop", p.get("description", ""), "{}", p.get("t2i_prompt", "")) self._db.flush() return name_to_canon def _save_and_merge_outlooks( self, all_outlooks: List[Dict[str, Any]], scenes: List[Dict[str, Any]], name_to_canon: Dict[str, str], episode_id: str, fulltext: str = "", ) -> None: """아웃룩을 EntityCanon에 저장하고, 중복 병합 후 T2I 치환.""" from app.modules.pipeline.outlook_merger import merge_outlooks, apply_merge_to_t2i from app.models.project import CharacterOutlook # 1. 아웃룩을 EntityCanon에 저장 outlook_name_to_id: Dict[str, str] = {} for outlook in all_outlooks: oname = outlook.get("outlook_name", "") if not oname or oname in outlook_name_to_id: continue # 이미 존재하는지 확인 existing = self._db.query(EntityCanon).filter( EntityCanon.project_id == self._project_id, EntityCanon.entity_type == "outlook", EntityCanon.name == oname, ).first() if existing: outlook_name_to_id[oname] = existing.id else: oid = _new_id() self._db.add(EntityCanon( id=oid, project_id=self._project_id, entity_type="outlook", name=oname, description=outlook.get("outlook_description", ""), t2i_prompt=outlook.get("outlook_description", ""), status="active", created_at=_now(), updated_at=_now(), )) outlook_name_to_id[oname] = oid logger.info("New outlook saved: %s", oname) # 2. CharacterOutlook 연결 for outlook in all_outlooks: oname = outlook.get("outlook_name", "") cname = outlook.get("character_name", "") oid = outlook_name_to_id.get(oname) cid = name_to_canon.get(cname) if not oid or not cid: continue # 중복 방지 exists = self._db.query(CharacterOutlook).filter( CharacterOutlook.character_id == cid, CharacterOutlook.outlook_id == oid, ).first() if not exists: self._db.add(CharacterOutlook( id=_new_id(), character_id=cid, outlook_id=oid, project_id=self._project_id, created_at=_now(), )) self._db.flush() # 3. LLM 병합 outlook_for_merge = [] for oname, oid in outlook_name_to_id.items(): # 이 아웃룩에 연결된 인물 이름 조회 char_links = self._db.query(CharacterOutlook).filter( CharacterOutlook.outlook_id == oid).all() char_names = [] for link in char_links: canon = self._db.query(EntityCanon).filter(EntityCanon.id == link.character_id).first() if canon: char_names.append(canon.name) outlook_for_merge.append({ "name": oname, "description": self._db.query(EntityCanon).filter(EntityCanon.id == oid).first().description or "", "characters": char_names, }) merge_result = merge_outlooks(outlook_for_merge, fulltext=fulltext) name_map = merge_result.get("name_map", {}) # 4. 병합 적용: DB + T2I 치환 if name_map: for old_name, new_name in name_map.items(): old_id = outlook_name_to_id.get(old_name) new_id_val = outlook_name_to_id.get(new_name) if not old_id or not new_id_val: continue # CharacterOutlook의 outlook_id를 new로 이전 self._db.query(CharacterOutlook).filter( CharacterOutlook.outlook_id == old_id ).update({"outlook_id": new_id_val}) # 중복 삭제 self._db.query(EntityCanon).filter(EntityCanon.id == old_id).delete() logger.info("Outlook merged: '%s' → '%s'", old_name, new_name) # T2I 프롬프트 치환 apply_merge_to_t2i(scenes, name_map) self._db.flush() # 5. T2I에서 사용된 모든 [[char]+[outlook]] 조합을 CharacterOutlook에 보장 import re as _re for scene in scenes: for t2i_field in [scene.get("t2i_prompt", "")] + [v.get("t2i_prompt", "") for v in scene.get("t2i_variations", [])]: for _m in _re.finditer(r'\[\[([^\]]+)\]\+\[([^\]]+)\]\]', t2i_field): char_name, outlook_name = _m.group(1), _m.group(2) if outlook_name == "미지정": continue cid = name_to_canon.get(char_name) oid = outlook_name_to_id.get(outlook_name) or name_to_canon.get(outlook_name) if not cid or not oid: continue exists = self._db.query(CharacterOutlook).filter( CharacterOutlook.character_id == cid, CharacterOutlook.outlook_id == oid, ).first() if not exists: self._db.add(CharacterOutlook( id=_new_id(), character_id=cid, outlook_id=oid, project_id=self._project_id, created_at=_now(), )) logger.info("Auto-linked CharacterOutlook: %s + %s", char_name, outlook_name) self._db.flush() def _regenerate_project_summary(self, episodes, gemini_client) -> str: """모든 에피소드 요약을 합산하여 프로젝트 요약 재생성.""" from app.modules.llm.gemini_text_client import GeminiTextClient ep_summaries = [] for ep in episodes: ep_summaries.append(f"에피소드 {ep.episode_number} ({ep.title}): {ep.summary}") combined = "\n\n".join(ep_summaries) prompt = ( f"아래는 하나의 프로젝트에 속한 여러 에피소드의 요약입니다.\n" f"모든 에피소드를 아우르는 프로젝트 전체 요약을 작성하세요.\n" f"최소 5문장 이상. 세계관, 주요 인물 관계, 핵심 갈등, 전체 스토리 흐름을 포함하세요.\n\n" f"{combined}" ) try: summary_client = GeminiTextClient() result = summary_client.send(user_message=prompt, temperature=0.3) if isinstance(result, str): return result.strip() except Exception as exc: logger.warning("Project summary regeneration failed: %s", exc) # 실패 시 fallback: 첫 에피소드 요약 사용 return episodes[0].summary or "" def _save_scenes_v2( self, scene_result: Dict[str, Any], episode_id: str, name_to_canon: Dict[str, str], ) -> None: """v2 씬 추출 결과를 DB에 저장.""" # index→id 매핑 (의존 씬 연결용) index_to_still_id: Dict[int, str] = {} for scene in scene_result.get("scenes", []): si = scene.get("scene_index", 0) still_id = _new_id() index_to_still_id[si] = still_id # visible_entities를 JSON으로 변환 vis_entities = scene.get("visible_entities", []) # entity_name → entity_id 매핑 for ve in vis_entities: ename = ve.get("entity_name", "") ve["entity_id"] = name_to_canon.get(ename, "") # T2I에서 [[인물]+[아웃룩]] 마커 파싱 → 아웃룩도 visible_entities에 추가 import re as _re t2i_text = scene.get("t2i_prompt", "") for var in scene.get("t2i_variations", []): t2i_text += " " + var.get("t2i_prompt", "") for match in _re.finditer(r'\[\[([^\]]+)\]\+\[([^\]]+)\]\]', t2i_text): char_name, outlook_name = match.group(1), match.group(2) if outlook_name != "미지정": # 아웃룩이 visible_entities에 없으면 추가 if not any(ve.get("entity_name") == outlook_name for ve in vis_entities): vis_entities.append({ "entity_name": outlook_name, "entity_type": "outlook", "entity_id": name_to_canon.get(outlook_name, ""), "variant_name": "", }) # 의존 씬 ID dep_idx = scene.get("dependent_scene_index", -1) dep_id = index_to_still_id.get(dep_idx) if dep_idx > 0 else None # N개 T2I variations t2i_variations = scene.get("t2i_variations", []) t2i_variations_json = json.dumps(t2i_variations, ensure_ascii=False) if t2i_variations else None # 세그먼트 위치 (scene_list에서 전달) seg_start = None seg_end = None for sl in scene_result.get("scene_list", []): if sl.get("scene_index") == si: seg_start = sl.get("start_char") seg_end = sl.get("end_char") break self._db.add(SceneStill( id=still_id, project_id=self._project_id, episode_id=episode_id, still_index=si, screenplay_scene_heading=scene.get("heading", ""), beat_title=scene.get("beat_title", ""), still_frame_prompt=scene.get("representative_moment", ""), visible_entities_json=json.dumps(vis_entities, ensure_ascii=False), t2i_prompt_cinematic=scene.get("t2i_prompt", ""), t2i_variations_json=t2i_variations_json, t2i_composer_version="scene_extractor_v3", segment_start_char=seg_start, segment_end_char=seg_end, dependent_scene_id=dep_id, scene_type=scene.get("scene_type", "normal"), status="pending", created_at=_now(), )) self._db.flush() def _save_entities( self, entities: List[Dict[str, Any]], episode_id: str, ) -> Dict[str, str]: """Save entities and return name->canon_id mapping.""" now = _now() name_to_canon: Dict[str, str] = {} for entity in entities: canon = self._find_or_create_canon(entity, now) name_to_canon[entity["name"]] = canon.id # Update description if better (longer) if entity.get("description") and ( not canon.description or len(entity["description"]) > len(canon.description) ): canon.description = entity["description"] canon.stable_traits = entity.get("stable_traits", canon.stable_traits) canon.updated_at = now # Add aliases for alias_text in entity.get("aliases", []): existing_alias = ( self._db.query(EntityAlias) .filter(EntityAlias.canon_id == canon.id, EntityAlias.alias == alias_text) .first() ) if not existing_alias: self._db.add(EntityAlias( id=_new_id(), canon_id=canon.id, alias=alias_text, )) # Link to episode existing_link = ( self._db.query(EntityEpisodeLink) .filter( EntityEpisodeLink.canon_id == canon.id, EntityEpisodeLink.episode_id == episode_id, ) .first() ) if not existing_link: self._db.add(EntityEpisodeLink( id=_new_id(), project_id=self._project_id, canon_id=canon.id, episode_id=episode_id, source="extracted", )) self._db.flush() return name_to_canon def _save_relations( self, relations: List[Dict[str, Any]], name_to_canon: Dict[str, str], ) -> None: """Save relation facts and participants.""" now = _now() for rel in relations: # Resolve all participants to canon IDs; skip if any unresolved participants = rel.get("participants", []) resolved = [] skip = False for p in participants: canon_id = name_to_canon.get(p["entity_name"]) if not canon_id: skip = True break resolved.append((canon_id, p["role"], p.get("order", 1))) if skip: continue fact = RelationFact( id=_new_id(), project_id=self._project_id, relation_family=rel["relation_family"], relation_type=rel["relation_type"], directionality=rel["directionality"], temporal_scope=rel["temporal_scope"], continuity_priority=rel["continuity_priority"], continuity_reason=rel.get("continuity_reason", ""), created_at=now, ) self._db.add(fact) self._db.flush() for canon_id, role, order in resolved: self._db.add(RelationParticipant( id=_new_id(), relation_id=fact.id, canon_id=canon_id, participant_role=role, participant_order=order, )) self._db.flush() def _save_stills( self, stills: List[Dict[str, Any]], episode_id: str, name_to_canon: Dict[str, str], ) -> None: """Save scene stills to DB. Map entity_name → entity_id in visible_entities.""" now = _now() # Build name/alias → canon_id lookup name_lookup: Dict[str, str] = dict(name_to_canon) for canon_id in set(name_to_canon.values()): aliases = self._db.query(EntityAlias).filter(EntityAlias.canon_id == canon_id).all() for a in aliases: name_lookup[a.alias] = canon_id for s in stills: vis = s.get("visible_entities", []) resolved_vis = [] for v in vis: if isinstance(v, dict): name = v.get("entity_name", "") canon_id = name_lookup.get(name) resolved = dict(v) if canon_id: resolved["entity_id"] = canon_id resolved_vis.append(resolved) else: resolved_vis.append(v) still = SceneStill( id=_new_id(), project_id=self._project_id, episode_id=episode_id, still_index=s["still_index"], screenplay_scene_heading=s.get("screenplay_scene_heading"), beat_title=s.get("beat_title"), still_frame_prompt=s.get("still_frame_prompt"), camera_json=json.dumps(s.get("camera", {}), ensure_ascii=False), lighting_json=json.dumps(s.get("lighting", {}), ensure_ascii=False), visible_entities_json=json.dumps(resolved_vis, ensure_ascii=False), status="pending", created_at=now, ) self._db.add(still) self._db.flush() # ── Private phase methods (split from run_analysis) ── def _run_entity_phase( self, episode: Episode, fulltext: str, language: str, checkpoint_dir: str, provenance: ProvenanceRecorder, tracker: ProgressTracker, ) -> Dict[str, Any]: """Phase 1: 요소 추출 (4턴 멀티턴) + 스타일/요약 저장. Returns: entity_result dict (characters, locations, props, style, ...) """ from app.modules.llm.gemini_text_client import GeminiTextClient from app.modules.pipeline.entity_extractor_v2 import extract_entities_multiturn from app.models.project import ProjectSettings gemini = GeminiTextClient() episode_id = episode.id # 1. 요소 추출 (4턴 멀티턴) tracker.update("요소 추출 중 (Gemini 멀티턴)", 0, 4) # 이전 에피소드의 기존 요소 (prior_entities) prior_entities_list = None prior_canons = ( self._db.query(EntityCanon) .filter(EntityCanon.project_id == self._project_id) .all() ) if prior_canons: prior_entities_list = [ {"name": e.name, "entity_type": e.entity_type, "description": (e.description or "")[:200], "t2i_prompt": (e.t2i_prompt or "")[:100]} for e in prior_canons ] with provenance.start_operation( "entity_extraction_v2", "gemini_multiturn", episode_id=episode_id, ) as op: op.set_input({"fulltext_chars": len(fulltext), "language": language}) _entity_cp_dir = os.path.join( settings.projects_dir, self._project_id, "checkpoints" ) entity_result = extract_entities_multiturn( gemini_client=gemini, fulltext=fulltext, prior_entities=prior_entities_list, checkpoint_dir=_entity_cp_dir, ) total_entities = ( len(entity_result.get("characters", [])) + len(entity_result.get("locations", [])) + len(entity_result.get("props", [])) ) op.set_output({"total_entities": total_entities}) # 1.5. 에피소드 요약 저장 + 스타일 저장 + 프로젝트 요약 재생성 style_data = entity_result.get("style", {}) if style_data: # 에피소드 요약을 episode 테이블에 저장 ep_summary = style_data.get("episode_summary", "") if ep_summary: episode.summary = ep_summary episode.updated_at = _now() # 스타일을 ProjectSettings에 저장 proj_settings = self._db.query(ProjectSettings).filter( ProjectSettings.project_id == self._project_id).first() if not proj_settings: proj_settings = ProjectSettings( id=_new_id(), project_id=self._project_id, style_rules_json=json.dumps(style_data, ensure_ascii=False), updated_at=_now(), ) self._db.add(proj_settings) else: proj_settings.style_rules_json = json.dumps(style_data, ensure_ascii=False) proj_settings.updated_at = _now() # 프로젝트 요약 재생성: 모든 에피소드 요약을 모아서 합산 요약 all_episodes = ( self._db.query(Episode) .filter(Episode.project_id == self._project_id, Episode.summary.isnot(None)) .order_by(Episode.episode_number) .all() ) if len(all_episodes) == 1: # 첫 에피소드: 에피소드 요약 = 프로젝트 요약 proj_settings.world_summary = all_episodes[0].summary elif len(all_episodes) > 1: # 여러 에피소드: Gemini로 합산 요약 재생성 proj_settings.world_summary = self._regenerate_project_summary( all_episodes, gemini ) self._db.flush() # commit은 caller(run_analysis)에서 — 에러 시 롤백 가능 return entity_result def _run_scene_phase( self, episode: Episode, fulltext: str, entity_result: Dict[str, Any], name_to_canon: Dict[str, str], checkpoint_dir: str, provenance: Optional[ProvenanceRecorder], tracker: ProgressTracker, project_llm_config: Optional[Dict] = None, ) -> Dict[str, Any]: """Phase 2: 씬 세그먼테이션 + 아웃룩 + 병렬 씬 분석. Returns: scene_result dict (scenes, scene_list, total_scenes, ...) """ from pathlib import Path from app.modules.llm.gemini_text_client import GeminiTextClient from app.modules.pipeline.scene_extractor_v2 import ( extract_scenes_multiturn, _segment_scenes, ) from app.modules.pipeline.scene_dependency_extractor import extract_scene_dependencies from app.modules.pipeline.outlook_extractor import extract_all_outlooks from app.models.project import ProjectSettings episode_id = episode.id cp_dir = Path(checkpoint_dir) cp_dir.mkdir(parents=True, exist_ok=True) _ps = self._db.query(ProjectSettings).filter( ProjectSettings.project_id == self._project_id).first() _threshold = (_ps.scene_split_threshold if _ps and _ps.scene_split_threshold else 600) # ── Step 1: 씬 세그먼테이션 (결과 저장/재활용) ── segments_file = cp_dir / "segments.json" if segments_file.exists(): segments = json.loads(segments_file.read_text(encoding="utf-8")) logger.info("Segments restored from checkpoint: %d scenes", len(segments)) else: tracker.update("씬 세그먼테이션 (Gemini Lite)", 2, 6) segments = _segment_scenes(fulltext, split_threshold=_threshold) segments_file.write_text(json.dumps(segments, ensure_ascii=False, indent=2), encoding="utf-8") logger.info("Segments saved to checkpoint: %d scenes", len(segments)) # ── Step 2: 씬 연관 추출 (결과 저장/재활용) ── deps_file = cp_dir / "scene_dependencies.json" if deps_file.exists(): scene_dependencies = {int(k): v for k, v in json.loads(deps_file.read_text(encoding="utf-8")).items()} logger.info("Scene dependencies restored from checkpoint: %d", len(scene_dependencies)) else: tracker.update("씬 연관 분석", 3, 6) scene_dependencies = extract_scene_dependencies( segments=segments, fulltext=fulltext, checkpoint_dir=checkpoint_dir, ) deps_file.write_text(json.dumps(scene_dependencies, ensure_ascii=False, indent=2), encoding="utf-8") logger.info("Scene dependencies saved to checkpoint: %d", len(scene_dependencies)) # ── Step 3: 아웃룩 추출 (결과 저장/재활용) ── outlook_file = cp_dir / "outlook_extraction.json" if outlook_file.exists(): outlook_result = json.loads(outlook_file.read_text(encoding="utf-8")) logger.info("Outlook extraction restored from checkpoint") else: tracker.update("아웃룩 추출 중", 4, 6) char_names = [c["name"] for c in entity_result.get("characters", [])] outlook_result = extract_all_outlooks( segments=segments, fulltext=fulltext, characters=char_names, checkpoint_dir=checkpoint_dir, project_llm_config=project_llm_config, ) outlook_file.write_text(json.dumps(outlook_result, ensure_ascii=False, indent=2), encoding="utf-8") logger.info("Outlook extraction saved to checkpoint") # 아웃룩을 DB에 저장 outlook_name_to_id = {} for outlook in outlook_result.get("outlooks", []): oname = outlook["name"] existing = self._db.query(EntityCanon).filter( EntityCanon.project_id == self._project_id, EntityCanon.entity_type == "outlook", EntityCanon.name == oname, ).first() if existing: outlook_name_to_id[oname] = existing.id else: oid = _new_id() self._db.add(EntityCanon( id=oid, project_id=self._project_id, entity_type="outlook", name=oname, description=outlook.get("description", ""), t2i_prompt=outlook.get("description", ""), status="active", created_at=_now(), updated_at=_now(), )) outlook_name_to_id[oname] = oid # CharacterOutlook 연결 for sa in outlook_result.get("scene_assignments", []): for ch in sa.get("characters", []): cname, oname = ch["character_name"], ch["outlook_name"] cid = name_to_canon.get(cname) oid = outlook_name_to_id.get(oname) if cid and oid: exists = self._db.query(CharacterOutlook).filter( CharacterOutlook.character_id == cid, CharacterOutlook.outlook_id == oid, ).first() if not exists: self._db.add(CharacterOutlook( id=_new_id(), character_id=cid, outlook_id=oid, project_id=self._project_id, created_at=_now(), )) # name_to_canon에 아웃룩 추가 for oname, oid in outlook_name_to_id.items(): name_to_canon[oname] = oid self._db.commit() # 씬별 아웃룩 매핑 구성 outlook_assignments = {} for sa in outlook_result.get("scene_assignments", []): outlook_assignments[sa["scene_index"]] = sa.get("characters", []) # 씬 상세 분석 (병렬) tracker.update("씬 분석 중 (병렬)", 4, 5) def _scene_progress(si, total): tracker.update(f"씬 분석 {si}/{total}", 4, 5) style = entity_result.get("style") if provenance: with provenance.start_operation( "scene_extraction_v4", "gemini_parallel", episode_id=episode_id, ) as op: scene_result = extract_scenes_multiturn( fulltext=fulltext, entities=entity_result, style=style, on_scene_progress=_scene_progress, split_threshold=_threshold, checkpoint_dir=checkpoint_dir, outlook_assignments=outlook_assignments, scene_dependencies=scene_dependencies, pre_segments=segments, scene_llm=settings.scene_detail_llm, project_llm_config=project_llm_config, ) op.set_output({"total_scenes": scene_result.get("total_scenes", 0)}) else: scene_result = extract_scenes_multiturn( fulltext=fulltext, entities=entity_result, style=style, on_scene_progress=_scene_progress, split_threshold=_threshold, checkpoint_dir=checkpoint_dir, outlook_assignments=outlook_assignments, scene_dependencies=scene_dependencies, pre_segments=segments, scene_llm=settings.scene_detail_llm, ) return scene_result # ── Public orchestration methods ── def run_analysis( self, episode_id: str, ip: Optional[str] = None, ) -> None: """Run full analysis pipeline: entity extraction + scene still extraction.""" episode = self._get_episode(episode_id) # status == "analyzing" 체크는 API 핸들러(with_for_update)에서 수행됨 # 백그라운드 스레드에서는 이미 "analyzing" 상태이므로 건너뜀 if not episode.fulltext: raise AppError( code="analysis.no_text", message=t("analysis.no_text"), status_code=400, ) if not settings.openai_api_key: raise AppError( code="analysis.openai_key_missing", message=t("analysis.openai_key_missing"), status_code=400, ) # Clean orphan ImageAssets linked to scenes being deleted still_ids = [s.id for s in self._db.query(SceneStill.id).filter( SceneStill.project_id == self._project_id, SceneStill.episode_id == episode_id, ).all()] if still_ids: self._db.query(ImageAsset).filter(ImageAsset.still_id.in_(still_ids)).delete(synchronize_session=False) # Clear previous analysis data for this episode (idempotent re-run) self._db.query(SceneStill).filter( SceneStill.project_id == self._project_id, SceneStill.episode_id == episode_id, ).delete() self._db.query(EntityEpisodeLink).filter( EntityEpisodeLink.project_id == self._project_id, EntityEpisodeLink.episode_id == episode_id, ).delete() # Don't delete EntityCanon — they may be shared across episodes # Don't delete RelationFact — they may reference entities from other episodes self._db.flush() # Mark as analyzing episode.status = "analyzing" episode.analysis_error = None episode.updated_at = _now() self._db.commit() tracker = ProgressTracker(self._db, episode_id, "analysis", self._project_id) try: language = episode.language or "ko" fulltext = episode.fulltext # deferred column — 첫 commit 전에 로드하여 캐시 provenance = ProvenanceRecorder(self._db, self._project_id) # 프로젝트 LLM 설정 로드 project_llm_config = {} try: ps = self._db.query(ProjectSettings).filter( ProjectSettings.project_id == self._project_id ).first() if ps and ps.llm_config_json: project_llm_config = json.loads(ps.llm_config_json) except Exception: pass _checkpoint_dir = os.path.join( settings.projects_dir, self._project_id, "checkpoints" ) # Phase 1: 요소 추출 entity_result = self._run_entity_phase( episode, fulltext, language, _checkpoint_dir, provenance, tracker, ) # 요소 DB 저장 (변형은 별도 EntityCanon + RelationFact) tracker.update("요소 저장 중", 1, 4) name_to_canon = self._save_entities_v2(entity_result, episode_id) self._db.commit() # Phase 2: 씬 분석 scene_result = self._run_scene_phase( episode, fulltext, entity_result, name_to_canon, _checkpoint_dir, provenance, tracker, project_llm_config=project_llm_config, ) # 씬 DB 저장 tracker.update("씬 저장 중", 6, 6) self._save_scenes_v2(scene_result, episode_id, name_to_canon) self._db.commit() # Mark as analyzed episode.status = "analyzed" episode.updated_at = _now() self._db.commit() tracker.complete() total_entities = ( len(entity_result.get("characters", [])) + len(entity_result.get("locations", [])) + len(entity_result.get("props", [])) ) self._logger.log( actor_id=self._actor_id, action="episode.analyze", resource_type="episode", resource_id=episode_id, project_id=self._project_id, detail={ "entities_count": total_entities, "scenes_count": scene_result.get("total_scenes", 0), "pipeline": "v2_gemini_multiturn", }, ip_address=ip, ) except Exception as exc: self._db.rollback() tracker.fail(str(exc)) episode = self._get_episode(episode_id) episode.status = "error" episode.analysis_error = str(exc)[:2000] episode.updated_at = _now() self._db.commit() raise def reanalyze_scenes(self, episode_id: str, ip: Optional[str] = None) -> None: """씬만 재분석 — 요소 유지, 씬 삭제 후 재추출.""" episode = self._get_episode(episode_id) if not episode.fulltext: raise AppError(code="analysis.no_text", message=t("analysis.no_text"), status_code=400) # status == "analyzing" 체크는 API 핸들러에서 수행됨 fulltext = episode.fulltext # deferred column — commit 전에 캐시 # Clean orphan ImageAssets + 씬 삭제 (status 변경 전에 수행) still_ids = [s.id for s in self._db.query(SceneStill.id).filter( SceneStill.project_id == self._project_id, SceneStill.episode_id == episode_id, ).all()] if still_ids: self._db.query(ImageAsset).filter(ImageAsset.still_id.in_(still_ids)).delete(synchronize_session=False) self._db.query(SceneStill).filter( SceneStill.project_id == self._project_id, SceneStill.episode_id == episode_id, ).delete() episode.status = "analyzing" episode.analysis_error = None episode.updated_at = _now() self._db.commit() tracker = ProgressTracker(self._db, episode_id, "analysis", self._project_id) try: from app.models.project import ProjectSettings # 기존 요소 로드 entities_orm = self._db.query(EntityCanon).filter( EntityCanon.project_id == self._project_id).all() entity_result = { "characters": [{"name": e.name, "description": e.description or "", "t2i_prompt": e.t2i_prompt or ""} for e in entities_orm if e.entity_type == "character"], "locations": [{"name": e.name, "description": e.description or "", "t2i_prompt": e.t2i_prompt or ""} for e in entities_orm if e.entity_type == "location"], "props": [{"name": e.name, "description": e.description or "", "t2i_prompt": e.t2i_prompt or ""} for e in entities_orm if e.entity_type == "prop"], } # 스타일 로드 ps = self._db.query(ProjectSettings).filter( ProjectSettings.project_id == self._project_id).first() style = json.loads(ps.style_rules_json) if ps and ps.style_rules_json else {} entity_result["style"] = style # 이름→ID 매핑 name_to_canon = {e.name: e.id for e in entities_orm} tracker.update("씬 재추출 중", 0, 2) _checkpoint_dir = os.path.join( settings.projects_dir, self._project_id, "checkpoints" ) # Delete scene phase checkpoints to force recomputation from pathlib import Path import glob as _glob for cp_name in ["segments.json", "scene_dependencies.json", "outlook_extraction.json"]: cp_file = Path(_checkpoint_dir) / cp_name if cp_file.exists(): cp_file.unlink() logger.info("Deleted checkpoint for reanalysis: %s", cp_name) # 씬 상세 체크포인트 (scene_extraction_{hash}.json)도 삭제 for stale in Path(_checkpoint_dir).glob("scene_extraction_*.json"): stale.unlink() logger.info("Deleted per-scene checkpoint for reanalysis: %s", stale.name) # Phase 2: 씬 분석 (공유 메서드 사용) scene_result = self._run_scene_phase( episode, fulltext, entity_result, name_to_canon, _checkpoint_dir, None, tracker, ) tracker.update("씬 저장 중", 1, 2) self._save_scenes_v2(scene_result, episode_id, name_to_canon) self._db.commit() episode.status = "analyzed" episode.updated_at = _now() self._db.commit() tracker.complete() self._logger.log( actor_id=self._actor_id, action="episode.reanalyze_scenes", resource_type="episode", resource_id=episode_id, project_id=self._project_id, detail={"scenes_count": scene_result.get("total_scenes", 0)}, ip_address=ip, ) except Exception as exc: self._db.rollback() tracker.fail(str(exc)) episode = self._get_episode(episode_id) episode.status = "error" episode.analysis_error = str(exc)[:2000] episode.updated_at = _now() self._db.commit() raise