"""프로젝트 JSON 내보내기 / 가져오기 서비스.""" import uuid from datetime import datetime, timezone from pathlib import Path from typing import Any 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.models.catalog import ProjectRegistry, ProjectMember from app.models.project import ( Episode, EntityCanon, EntityAlias, RelationFact, RelationParticipant, SceneStill, EntityEpisodeLink, ImageAsset, WorldGuide, WebbookPackage, GenerationTrace, OperationLog, ProjectSettings, CharacterOutlook, ) def _now() -> str: return datetime.now(timezone.utc).isoformat() def _new_id() -> str: return str(uuid.uuid4()) def _row_to_dict(row) -> dict: """Convert a SQLAlchemy ORM row to a plain dict (column values only).""" return {c.name: getattr(row, c.name) for c in row.__table__.columns} # --------------------------------------------------------------------------- # Export # --------------------------------------------------------------------------- class ProjectExportService: def __init__(self, db: OrmSession, project_id: str) -> None: self._db = db self._project_id = project_id def _q(self, model, **filters): q = self._db.query(model).filter_by(project_id=self._project_id) for k, v in filters.items(): q = q.filter(getattr(model, k) == v) return q.all() def export_json(self) -> dict: """Export all project data as a JSON-serialisable dict.""" pid = self._project_id project = ( self._db.query(ProjectRegistry) .filter(ProjectRegistry.id == pid) .first() ) if not project: raise AppError( code="project.not_found", message=t("project.not_found"), status_code=404, ) members = ( self._db.query(ProjectMember) .filter(ProjectMember.project_id == pid) .all() ) episodes = self._q(Episode) entities = self._q(EntityCanon) # EntityAlias has no project_id — join via canon_id entity_ids = [e.id for e in entities] if entity_ids: entity_aliases = ( self._db.query(EntityAlias) .filter(EntityAlias.canon_id.in_(entity_ids)) .all() ) else: entity_aliases = [] relations = self._q(RelationFact) relation_ids = [r.id for r in relations] if relation_ids: relation_participants = ( self._db.query(RelationParticipant) .filter(RelationParticipant.relation_id.in_(relation_ids)) .all() ) else: relation_participants = [] scene_stills = self._q(SceneStill) entity_episode_links = self._q(EntityEpisodeLink) images = self._q(ImageAsset) world_guides = self._q(WorldGuide) webbook_packages = self._q(WebbookPackage) generation_traces = self._q(GenerationTrace) operation_logs = self._q(OperationLog) project_settings = self._q(ProjectSettings) character_outlooks = self._q(CharacterOutlook) return { "export_version": "1.0", "exported_at": _now(), "project": _row_to_dict(project), "members": [_row_to_dict(m) for m in members], "episodes": [_row_to_dict(e) for e in episodes], "entities": [_row_to_dict(e) for e in entities], "entity_aliases": [_row_to_dict(a) for a in entity_aliases], "relations": [_row_to_dict(r) for r in relations], "relation_participants": [_row_to_dict(p) for p in relation_participants], "scene_stills": [_row_to_dict(s) for s in scene_stills], "entity_episode_links": [_row_to_dict(lnk) for lnk in entity_episode_links], "images": [_row_to_dict(img) for img in images], "world_guides": [_row_to_dict(wg) for wg in world_guides], "webbook_packages": [_row_to_dict(wp) for wp in webbook_packages], "generation_traces": [_row_to_dict(gt) for gt in generation_traces], "operation_logs": [_row_to_dict(ol) for ol in operation_logs], "project_settings": [_row_to_dict(ps) for ps in project_settings], "character_outlooks": [_row_to_dict(co) for co in character_outlooks], } def export_assets_list(self) -> list[str]: """Return relative paths of every file under the project directory.""" project_dir = Path(settings.projects_dir) / self._project_id if not project_dir.exists(): return [] result: list[str] = [] for path in sorted(project_dir.rglob("*")): if path.is_file(): result.append(str(path.relative_to(project_dir))) return result # --------------------------------------------------------------------------- # Import # --------------------------------------------------------------------------- class ProjectImportService: def __init__(self, db: OrmSession, actor_id: str) -> None: self._db = db self._actor_id = actor_id def import_json(self, data: dict, new_name: str | None = None) -> str: """Import a project from an exported JSON blob. Generates fresh UUIDs for every record, remaps all internal foreign-key references, and wires the importing user as owner. Returns the new project_id. """ # ---- validate minimum structure ----------------------------------- if "export_version" not in data or "project" not in data: raise AppError( code="import.invalid_format", message=t("import.invalid_format"), status_code=400, ) # ---- new project id ----------------------------------------------- old_project_id: str = data["project"]["id"] new_project_id: str = _new_id() now = _now() # ---- build global ID remap table ---------------------------------- # Maps old_id → new_id for every entity that has a UUID PK. id_map: dict[str, str] = {old_project_id: new_project_id} def _remap(old_id: str | None) -> str | None: if old_id is None: return None return id_map.get(old_id, old_id) def _register(old_id: str) -> str: new_id = _new_id() id_map[old_id] = new_id return new_id # Pre-register IDs for all top-level records so cross-references # resolve correctly regardless of iteration order. for row in data.get("episodes", []): _register(row["id"]) for row in data.get("entities", []): _register(row["id"]) for row in data.get("entity_aliases", []): _register(row["id"]) for row in data.get("relations", []): _register(row["id"]) for row in data.get("relation_participants", []): _register(row["id"]) for row in data.get("scene_stills", []): _register(row["id"]) for row in data.get("entity_episode_links", []): _register(row["id"]) for row in data.get("images", []): _register(row["id"]) for row in data.get("world_guides", []): _register(row["id"]) for row in data.get("webbook_packages", []): _register(row["id"]) for row in data.get("generation_traces", []): _register(row["id"]) for row in data.get("operation_logs", []): _register(row["id"]) for row in data.get("project_settings", []): _register(row["id"]) for row in data.get("character_outlooks", []): _register(row["id"]) # members have their own PK too for row in data.get("members", []): _register(row["id"]) # ---- create project directory structure --------------------------- base = Path(settings.projects_dir) project_dir = base / new_project_id for sub in ["assets/screenplays", "assets/references", "assets/generated", "assets/exports"]: (project_dir / sub).mkdir(parents=True, exist_ok=True) # ---- insert ProjectRegistry --------------------------------------- proj_data: dict[str, Any] = data["project"].copy() project_name = new_name or proj_data.get("name", "Imported Project") project = ProjectRegistry( id=new_project_id, name=project_name, description=proj_data.get("description", ""), status="active", created_by=self._actor_id, created_at=now, updated_at=now, ) self._db.add(project) self._db.flush() # ---- insert owner membership for importing user ------------------ # (skip original members — they belong to a different system) owner_member = ProjectMember( id=_new_id(), project_id=new_project_id, user_id=self._actor_id, role="owner", added_by=None, created_at=now, ) self._db.add(owner_member) self._db.flush() # ---- helper to remap a row dict ---------------------------------- def _remap_row(row: dict, *fk_fields: str) -> dict: r = row.copy() for field in fk_fields: if field in r and r[field] is not None: r[field] = _remap(r[field]) return r # ---- episodes ----------------------------------------------------- for row in data.get("episodes", []): r = _remap_row(row, "id", "project_id") # Sanitize source_path: generate a safe path under the project dir original_source = r.get("source_path", "") if original_source: safe_source_name = Path(original_source).name # strip directory components safe_source_path = str(project_dir / "assets" / "screenplays" / safe_source_name) else: safe_source_path = "" self._db.add(Episode( id=r["id"], project_id=new_project_id, episode_number=r.get("episode_number", 0), title=r.get("title", ""), source_filename=r.get("source_filename", ""), source_path=safe_source_path, fulltext=r.get("fulltext"), language=r.get("language", "ko"), page_count=r.get("page_count"), status=r.get("status", "uploaded"), analysis_error=r.get("analysis_error"), created_at=r.get("created_at", now), updated_at=r.get("updated_at", now), )) # ---- entities ----------------------------------------------------- for row in data.get("entities", []): r = _remap_row(row, "id", "project_id") self._db.add(EntityCanon( id=r["id"], project_id=new_project_id, entity_type=r.get("entity_type", "character"), name=r.get("name", ""), description=r.get("description"), stable_traits=r.get("stable_traits", "{}"), status=r.get("status", "active"), created_at=r.get("created_at", now), updated_at=r.get("updated_at", now), )) # ---- entity aliases ----------------------------------------------- for row in data.get("entity_aliases", []): r = _remap_row(row, "id", "canon_id") self._db.add(EntityAlias( id=r["id"], canon_id=r["canon_id"], alias=r.get("alias", ""), )) # ---- relations ---------------------------------------------------- for row in data.get("relations", []): r = _remap_row(row, "id", "project_id") self._db.add(RelationFact( id=r["id"], project_id=new_project_id, relation_family=r.get("relation_family", ""), relation_type=r.get("relation_type", ""), directionality=r.get("directionality", ""), temporal_scope=r.get("temporal_scope", ""), continuity_priority=r.get("continuity_priority", ""), continuity_reason=r.get("continuity_reason"), created_at=r.get("created_at", now), )) # ---- relation participants ----------------------------------------- for row in data.get("relation_participants", []): r = _remap_row(row, "id", "relation_id", "canon_id") self._db.add(RelationParticipant( id=r["id"], relation_id=r["relation_id"], canon_id=r["canon_id"], participant_role=r.get("participant_role", ""), participant_order=r.get("participant_order", 1), )) # ---- scene stills ------------------------------------------------- for row in data.get("scene_stills", []): r = _remap_row(row, "id", "project_id", "episode_id") self._db.add(SceneStill( id=r["id"], project_id=new_project_id, episode_id=r["episode_id"], still_index=r.get("still_index", 0), screenplay_scene_heading=r.get("screenplay_scene_heading"), beat_title=r.get("beat_title"), still_frame_prompt=r.get("still_frame_prompt"), camera_json=r.get("camera_json", "{}"), lighting_json=r.get("lighting_json", "{}"), visible_entities_json=r.get("visible_entities_json", "[]"), status=r.get("status", "pending"), created_at=r.get("created_at", now), )) # ---- entity episode links ----------------------------------------- for row in data.get("entity_episode_links", []): r = _remap_row(row, "id", "project_id", "canon_id", "episode_id") self._db.add(EntityEpisodeLink( id=r["id"], project_id=new_project_id, canon_id=r["canon_id"], episode_id=r["episode_id"], source=r.get("source", "extracted"), )) # ---- image assets (metadata only — files not copied) ------------- for row in data.get("images", []): r = _remap_row(row, "id", "project_id", "entity_id", "still_id", "episode_id") # Sanitize file_path: generate a safe path under the project dir original_file_path = r.get("file_path", "") if original_file_path: safe_name = Path(original_file_path).name # strip directory components safe_file_path = str(project_dir / "assets" / "generated" / safe_name) else: safe_file_path = "" self._db.add(ImageAsset( id=r["id"], project_id=new_project_id, asset_type=r.get("asset_type", "scene"), entity_id=r.get("entity_id"), still_id=r.get("still_id"), episode_id=r.get("episode_id"), file_path=safe_file_path, prompt_used=r.get("prompt_used"), generation_model=r.get("generation_model"), width=r.get("width"), height=r.get("height"), status=r.get("status", "generated"), review_notes=r.get("review_notes", ""), validation_score=r.get("validation_score"), validation_result=r.get("validation_result"), created_at=r.get("created_at", now), )) # ---- world guides ------------------------------------------------ for row in data.get("world_guides", []): r = _remap_row(row, "id", "project_id", "episode_id") self._db.add(WorldGuide( id=r["id"], project_id=new_project_id, episode_id=r.get("episode_id"), guide_json=r.get("guide_json", "{}"), created_at=r.get("created_at", now), )) # ---- webbook packages -------------------------------------------- for row in data.get("webbook_packages", []): r = _remap_row(row, "id", "project_id", "episode_id") self._db.add(WebbookPackage( id=r["id"], project_id=new_project_id, episode_id=r.get("episode_id"), package_json=r.get("package_json", "{}"), prompt_version=r.get("prompt_version"), created_at=r.get("created_at", now), )) # ---- generation traces ------------------------------------------- for row in data.get("generation_traces", []): r = _remap_row(row, "id", "project_id", "image_asset_id", "still_id", "entity_id") self._db.add(GenerationTrace( id=r["id"], project_id=new_project_id, image_asset_id=r.get("image_asset_id"), still_id=r.get("still_id"), entity_id=r.get("entity_id"), attempt_number=r.get("attempt_number", 1), prompt_used=r.get("prompt_used", ""), prompt_version=r.get("prompt_version", "original"), model_name=r.get("model_name"), status=r.get("status", "success"), block_reason=r.get("block_reason"), block_categories=r.get("block_categories", "[]"), response_time_ms=r.get("response_time_ms"), sanitizer_feedback=r.get("sanitizer_feedback"), created_at=r.get("created_at", now), )) # ---- operation logs ---------------------------------------------- for row in data.get("operation_logs", []): r = _remap_row(row, "id", "project_id", "episode_id") self._db.add(OperationLog( id=r["id"], project_id=new_project_id, operation_type=r.get("operation_type", ""), episode_id=r.get("episode_id"), module_name=r.get("module_name", ""), module_version=r.get("module_version", ""), prompt_name=r.get("prompt_name"), prompt_version=r.get("prompt_version"), prompt_hash=r.get("prompt_hash"), input_summary=r.get("input_summary"), output_summary=r.get("output_summary"), status=r.get("status", "success"), error_message=r.get("error_message"), duration_ms=r.get("duration_ms"), token_usage=r.get("token_usage", "{}"), metadata_json=r.get("metadata_json", "{}"), created_at=r.get("created_at", now), )) # ---- project settings -------------------------------------------- for row in data.get("project_settings", []): r = _remap_row(row, "id", "project_id") self._db.add(ProjectSettings( id=r["id"], project_id=new_project_id, composer_system_prompt=r.get("composer_system_prompt"), composer_user_prompt=r.get("composer_user_prompt"), style_rules_json=r.get("style_rules_json"), world_summary=r.get("world_summary"), scene_split_threshold=r.get("scene_split_threshold", 600), updated_at=r.get("updated_at", now), )) # ---- character outlooks ------------------------------------------ for row in data.get("character_outlooks", []): r = _remap_row(row, "id", "character_id", "outlook_id", "project_id") self._db.add(CharacterOutlook( id=r["id"], character_id=r["character_id"], outlook_id=r["outlook_id"], project_id=new_project_id, created_at=r.get("created_at", now), )) self._db.commit() return new_project_id