"""Saved idea service for the Personal Space pre-project layer.""" from __future__ import annotations import re from typing import Any from uuid import uuid4 from sqlalchemy import and_, desc, select from sqlalchemy.ext.asyncio import AsyncSession from ..config.ai_prompts import PROMPT_SAVED_IDEA_ENRICHMENT from ..models import SavedIdea from ..schemas.pmflow_launch import PMFlowLaunchAttachment, PMFlowLaunchDraftRequest from ..schemas.space import SavedIdeaCreateRequest, SavedIdeaRead, SavedIdeaUpdateRequest from ..services.llm_service import llm_service from ..services.operation_log_service import log_operation from ..services.pmflow_launch_service import PMFlowLaunchService from ..services.ai_routing_governance_service import PLATFORM_RUNTIME_TENANT_ID from ..services.prompt_runtime_service import PromptRuntimeRenderResult, PromptRuntimeService from ..utils.timezone import now_cn_naive STATUS_LABELS = { "saved": "未整理", "enriched": "已整理", "needs_more_info": "信息不足", "draft_ready": "可生成草案", "promoted": "已转为草案", "archived": "已归档", } class SavedIdeaError(RuntimeError): def __init__(self, code: str, message: str, *, status_code: int = 400): super().__init__(message) self.code = code self.message = message self.status_code = int(status_code) class SpaceSavedIdeaService: ENRICHMENT_PROMPT_KEY = "PROMPT_SAVED_IDEA_ENRICHMENT" def __init__(self, db: AsyncSession, *, tenant_id: int): self.db = db self.tenant_id = int(tenant_id) async def list_ideas( self, *, user_id: int, limit: int = 20, include_archived: bool = False, ) -> list[SavedIdeaRead]: stmt = select(SavedIdea).where( and_(SavedIdea.tenant_id == self.tenant_id, SavedIdea.user_id == int(user_id)) ) if not include_archived: stmt = stmt.where(SavedIdea.status != "archived") stmt = stmt.order_by(desc(SavedIdea.updated_at), desc(SavedIdea.id)).limit(max(1, min(int(limit), 100))) rows = (await self.db.execute(stmt)).scalars().all() return [self._to_read(row) for row in rows] async def create_idea(self, *, body: SavedIdeaCreateRequest, user_id: int) -> SavedIdeaRead: raw_text = self._clean_text(body.raw_text) if not raw_text: raise SavedIdeaError("input_empty", "暂存想法内容不能为空", status_code=400) title = self._clean_text(body.title or "") or self._derive_title(raw_text) enrichment = await self._enrich_or_baseline(raw_text) summary = self._clean_text(body.summary or "") or str(enrichment["summary"]) row = SavedIdea( tenant_id=self.tenant_id, user_id=int(user_id), idea_id=f"idea_{uuid4().hex[:16]}", title=title[:255], raw_text=raw_text, summary=summary, source_type=str(body.source_type or "project_launch")[:32], source_ref=self._clean_text(body.source_ref or "")[:255] or None, input_mode=str(body.input_mode or "idea")[:32], attachments=[item.model_dump(mode="json") for item in body.attachments], status=str(enrichment["status"]), ai_summary=enrichment.get("ai_summary"), missing_questions=enrichment["missing_questions"], readiness_score=float(enrichment["readiness_score"]), enrichment_status=str(enrichment["enrichment_status"]), enrichment_payload=enrichment.get("enrichment_payload") or {}, enrichment_error=enrichment.get("enrichment_error"), enriched_at=enrichment.get("enriched_at"), ) self.db.add(row) await log_operation( self.db, tenant_id=self.tenant_id, user_id=int(user_id), project_id=None, entity_type="saved_idea", entity_id=row.idea_id, action="space.saved_idea.create", before_data=None, after_data={ "title": row.title, "status": row.status, "source_type": row.source_type, "readiness_score": row.readiness_score, "enrichment_status": row.enrichment_status, }, interface_path="/space/saved-ideas", caller="user", ) await self._commit() await self.db.refresh(row) return self._to_read(row) async def update_idea(self, *, idea_id: str, body: SavedIdeaUpdateRequest, user_id: int) -> SavedIdeaRead: row = await self._load_owned(idea_id=idea_id, user_id=user_id) before = self._snapshot(row) if body.raw_text is not None: row.raw_text = self._clean_text(body.raw_text) if not row.raw_text: raise SavedIdeaError("input_empty", "暂存想法内容不能为空", status_code=400) enrichment = await self._enrich_or_baseline(row.raw_text) row.summary = str(enrichment["summary"]) row.ai_summary = enrichment.get("ai_summary") row.missing_questions = enrichment["missing_questions"] row.readiness_score = float(enrichment["readiness_score"]) row.enrichment_status = str(enrichment["enrichment_status"]) row.enrichment_payload = enrichment.get("enrichment_payload") or {} row.enrichment_error = enrichment.get("enrichment_error") row.enriched_at = enrichment.get("enriched_at") if row.status in {"saved", "needs_more_info", "draft_ready", "enriched"}: row.status = str(enrichment["status"]) if body.title is not None: row.title = self._clean_text(body.title)[:255] or row.title if body.summary is not None: row.summary = self._clean_text(body.summary) or row.summary if body.status is not None: row.status = str(body.status) if row.status == "archived" and row.archived_at is None: row.archived_at = now_cn_naive() await log_operation( self.db, tenant_id=self.tenant_id, user_id=int(user_id), project_id=None, entity_type="saved_idea", entity_id=row.idea_id, action="space.saved_idea.update", before_data=before, after_data=self._snapshot(row), interface_path=f"/space/saved-ideas/{idea_id}", caller="user", ) await self._commit() await self.db.refresh(row) return self._to_read(row) async def archive_idea(self, *, idea_id: str, user_id: int) -> SavedIdeaRead: row = await self._load_owned(idea_id=idea_id, user_id=user_id) before = self._snapshot(row) row.status = "archived" row.archived_at = now_cn_naive() await log_operation( self.db, tenant_id=self.tenant_id, user_id=int(user_id), project_id=None, entity_type="saved_idea", entity_id=row.idea_id, action="space.saved_idea.archive", before_data=before, after_data=self._snapshot(row), interface_path=f"/space/saved-ideas/{idea_id}/archive", caller="user", ) await self._commit() await self.db.refresh(row) return self._to_read(row) async def promote_to_launch_draft(self, *, idea_id: str, user_id: int) -> tuple[SavedIdeaRead, dict[str, Any]]: row = await self._load_owned(idea_id=idea_id, user_id=user_id) if row.status == "archived": raise SavedIdeaError("idea_archived", "已归档想法不能生成项目草案", status_code=409) if row.launch_draft_id: row.status = "promoted" await self._commit() raise SavedIdeaError("draft_already_created", "该想法已经生成过项目草案", status_code=409) launch_service = PMFlowLaunchService(self.db, tenant_id=self.tenant_id) draft = await launch_service.create_draft( body=PMFlowLaunchDraftRequest( input=row.raw_text, input_type="template" if row.input_mode == "template" else "text", attachments=[PMFlowLaunchAttachment(**item) for item in self._safe_list(row.attachments)], source="saved_idea", ), user_id=int(user_id), ) row.status = "promoted" row.launch_draft_id = draft.draft_id row.promoted_at = now_cn_naive() await log_operation( self.db, tenant_id=self.tenant_id, user_id=int(user_id), project_id=None, entity_type="saved_idea", entity_id=row.idea_id, action="space.saved_idea.promote_to_launch_draft", before_data=None, after_data={"launch_draft_id": draft.draft_id, "title": draft.title}, interface_path=f"/space/saved-ideas/{idea_id}/promote", caller="user", ) await self._commit() await self.db.refresh(row) return self._to_read(row), draft.model_dump(mode="json") async def _load_owned(self, *, idea_id: str, user_id: int) -> SavedIdea: stmt = select(SavedIdea).where( and_( SavedIdea.tenant_id == self.tenant_id, SavedIdea.user_id == int(user_id), SavedIdea.idea_id == str(idea_id), ) ) row = (await self.db.execute(stmt)).scalar_one_or_none() if row is None: raise SavedIdeaError("idea_not_found", "暂存想法不存在或无权访问", status_code=404) return row async def _commit(self) -> None: try: await self.db.commit() except Exception: await self.db.rollback() raise def _to_read(self, row: SavedIdea) -> SavedIdeaRead: return SavedIdeaRead( id=row.idea_id, title=row.title, raw_text=row.raw_text, summary=row.summary, source_type=row.source_type, source_ref=row.source_ref, input_mode=row.input_mode, attachments=self._safe_list(row.attachments), status=row.status, status_label=STATUS_LABELS.get(row.status, "未整理"), ai_summary=row.ai_summary, missing_questions=self._safe_str_list(row.missing_questions), readiness_score=float(row.readiness_score or 0.0), enrichment_status=str(row.enrichment_status or "baseline"), enrichment_payload=self._safe_dict(row.enrichment_payload), enrichment_error=row.enrichment_error, enriched_at=row.enriched_at.isoformat() if row.enriched_at else None, launch_draft_id=row.launch_draft_id, promoted_project_id=int(row.promoted_project_id) if row.promoted_project_id is not None else None, created_at=row.created_at.isoformat() if row.created_at else "", updated_at=row.updated_at.isoformat() if row.updated_at else "", href=f"/project-launch?ideaId={row.idea_id}", ) def _snapshot(self, row: SavedIdea) -> dict[str, Any]: return { "idea_id": row.idea_id, "title": row.title, "status": row.status, "readiness_score": row.readiness_score, "enrichment_status": row.enrichment_status, "launch_draft_id": row.launch_draft_id, } @staticmethod def _clean_text(value: str) -> str: return re.sub(r"\s+", " ", str(value or "").strip()) def _derive_title(self, raw_text: str) -> str: text = self._clean_text(raw_text) separators = ["。", "！", "？", "\n", ".", "!", "?"] end = min([idx for idx in (text.find(s) for s in separators) if idx >= 0] or [len(text)]) title = text[:end].strip() or text return title[:32] if len(title) > 32 else title def _derive_summary(self, raw_text: str) -> str: text = self._clean_text(raw_text) return text[:120] + ("..." if len(text) > 120 else "") async def _enrich_or_baseline(self, raw_text: str) -> dict[str, Any]: baseline = self._baseline_enrichment(raw_text) prompt_result = await self._render_enrichment_prompt(raw_text=raw_text) try: structured = await llm_service.generate_structured_object( prompt_result.prompt, db=self.db, tenant_id=self.tenant_id, function_name="space_saved_idea_enrichment_v1", temperature=0.0, max_tokens=900, response_mode="fast", routing_meta={ "scene": "space_saved_idea_enrichment", "tier_alias": "low", "response_contract": "structured", "prompt_runtime_tenant_id": PLATFORM_RUNTIME_TENANT_ID, **prompt_result.routing_meta(), }, ) except Exception as exc: baseline["enrichment_status"] = "ai_failed" baseline["enrichment_error"] = self._clean_text(str(exc))[:500] or "AI enrichment failed" return baseline payload = structured.payload if structured is not None and isinstance(structured.payload, dict) else {} return self._normalize_ai_enrichment(raw_text, payload, baseline) async def _render_enrichment_prompt(self, *, raw_text: str) -> PromptRuntimeRenderResult: return await PromptRuntimeService(self.db).render( tenant_id=PLATFORM_RUNTIME_TENANT_ID, prompt_key=self.ENRICHMENT_PROMPT_KEY, fallback_template=PROMPT_SAVED_IDEA_ENRICHMENT, raw_text=raw_text[:4000], ) def _baseline_enrichment(self, raw_text: str) -> dict[str, Any]: summary = self._derive_summary(raw_text) missing_questions = self._missing_questions(raw_text) readiness_score = self._readiness_score(raw_text) return { "summary": summary, "status": self._initial_status(raw_text), "ai_summary": None, "missing_questions": missing_questions, "readiness_score": readiness_score, "enrichment_status": "baseline", "enrichment_payload": { "source": "system_baseline", "summary": summary, "missing_questions": missing_questions, "readiness_score": readiness_score, }, "enrichment_error": None, "enriched_at": None, } def _normalize_ai_enrichment(self, raw_text: str, payload: dict[str, Any], baseline: dict[str, Any]) -> dict[str, Any]: summary = self._clean_text(str(payload.get("summary") or "")) or str(baseline["summary"]) missing_questions = self._safe_str_list(payload.get("missing_questions"))[:3] or list(baseline["missing_questions"]) score = self._clamp_score(payload.get("readiness_score"), fallback=float(baseline["readiness_score"])) raw_status = str(payload.get("status") or "").strip() status = raw_status if raw_status in {"needs_more_info", "enriched", "draft_ready"} else self._status_from_score(score) confidence = self._clamp_score(payload.get("confidence"), fallback=0.0) enrichment_payload = { "source": "ai_structured", "summary": summary, "target_user": self._optional_text(payload.get("target_user")), "problem": self._optional_text(payload.get("problem")), "value_hypothesis": self._optional_text(payload.get("value_hypothesis")), "validation_signal": self._optional_text(payload.get("validation_signal")), "constraints": self._safe_str_list(payload.get("constraints"))[:5], "missing_questions": missing_questions, "readiness_score": score, "status": status, "confidence": confidence, "input_excerpt": self._derive_summary(raw_text), } return { "summary": summary, "status": status, "ai_summary": summary, "missing_questions": missing_questions, "readiness_score": score, "enrichment_status": "ai_enriched", "enrichment_payload": enrichment_payload, "enrichment_error": None, "enriched_at": now_cn_naive(), } def _missing_questions(self, raw_text: str) -> list[str]: text = raw_text.lower() questions: list[str] = [] if not any(keyword in text for keyword in ["面向", "用户", "客户", "人群", "who", "user"]): questions.append("目标用户是谁？") if not any(keyword in text for keyword in ["解决", "问题", "痛点", "提升", "减少", "problem"]): questions.append("要解决的核心问题是什么？") if not any(keyword in text for keyword in ["成功", "指标", "验证", "收入", "留存", "roi", "metric"]): questions.append("什么信号代表这个想法值得推进？") return questions[:3] def _readiness_score(self, raw_text: str) -> float: length_score = min(len(raw_text) / 400, 0.45) completeness = (3 - len(self._missing_questions(raw_text))) * 0.18 return round(min(0.99, 0.15 + length_score + completeness), 2) def _initial_status(self, raw_text: str) -> str: score = self._readiness_score(raw_text) return self._status_from_score(score) def _status_from_score(self, score: float) -> str: if score >= 0.78: return "draft_ready" if score >= 0.55: return "enriched" return "needs_more_info" @staticmethod def _safe_list(value: Any) -> list[dict[str, Any]]: return value if isinstance(value, list) else [] @staticmethod def _safe_str_list(value: Any) -> list[str]: if not isinstance(value, list): return [] return [str(item).strip()[:500] for item in value if str(item).strip()] @staticmethod def _safe_dict(value: Any) -> dict[str, Any]: return value if isinstance(value, dict) else {} @staticmethod def _optional_text(value: Any) -> str | None: text = str(value or "").strip() return text[:500] if text and text.lower() != "null" else None @staticmethod def _clamp_score(value: Any, *, fallback: float) -> float: try: score = float(value) except (TypeError, ValueError): score = float(fallback) return round(max(0.0, min(1.0, score)), 2) __all__ = ["SavedIdeaError", "SpaceSavedIdeaService"]