Spaces:
Sleeping
Sleeping
| """Pydantic models for ProjectCraft and JSON parsing utilities.""" | |
| from pydantic import BaseModel, Field, field_validator | |
| from typing import Optional | |
| import json | |
| from json_repair import repair_json | |
| ERROR_MESSAGES = { | |
| "timeout": "The request took too long. Please try again.", | |
| "rate_limit": "Too many requests. Please wait 30 seconds and try again.", | |
| "bad_json": "The AI returned an unexpected format. Please try again.", | |
| "no_phases": "Plan generated but contains no phases. Add more detail.", | |
| "export_fail": "Export failed. Make sure a plan has been generated first.", | |
| "google_auth": "Google connection unavailable. Please reconnect in settings.", | |
| "share_invalid":"This share link appears to be corrupted. Request a new one.", | |
| "no_objective": "Project objective is required.", | |
| "no_teams": "At least one team is required.", | |
| "no_dates": "At least one date (start or end) must be provided.", | |
| "date_order": "End date must be after start date.", | |
| } | |
| class Phase(BaseModel): | |
| id: str | |
| name: str | |
| start_offset_days: int = 0 | |
| duration_days: int = 1 | |
| owner_team: str = "Unassigned" | |
| tasks: list[str] = Field(default_factory=list) | |
| is_critical_path: bool = False | |
| slack_days: int = 0 | |
| def must_be_positive(cls, v): return max(1, int(v or 1)) | |
| def must_be_non_negative(cls, v): return max(0, int(v or 0)) | |
| class RACIEntry(BaseModel): | |
| task: str | |
| responsible: list[str] = Field(default_factory=list) | |
| accountable: str = "Unassigned" | |
| consulted: list[str] = Field(default_factory=list) | |
| informed: list[str] = Field(default_factory=list) | |
| class Risk(BaseModel): | |
| phase_id: str = "" | |
| description: str | |
| likelihood: str = "Medium" | |
| impact: str = "Medium" | |
| mitigation: str = "" | |
| def validate_level(cls, v): | |
| return v if v in {"Low", "Medium", "High"} else "Medium" | |
| class WorkloadWeek(BaseModel): | |
| week_label: str | |
| load_score: float = 0.0 | |
| def clamp(cls, v): return max(0.0, min(10.0, float(v or 0))) | |
| class TeamWorkload(BaseModel): | |
| team_name: str | |
| weeks: list[WorkloadWeek] = Field(default_factory=list) | |
| class ProjectPlan(BaseModel): | |
| phases: list[Phase] = Field(default_factory=list) | |
| raci: list[RACIEntry] = Field(default_factory=list) | |
| risks: list[Risk] = Field(default_factory=list) | |
| workload: list[TeamWorkload] = Field(default_factory=list) | |
| estimated_start_date: Optional[str] = None | |
| estimated_end_date: Optional[str] = None | |
| timeline_note: Optional[str] = None | |
| timeline_warning: Optional[str] = None | |
| def parse_plan_json(raw: str) -> ProjectPlan: | |
| """Parse a raw JSON string into a validated ProjectPlan. Returns empty plan on error.""" | |
| try: | |
| text = raw.strip() | |
| print(f"[parse_plan_json] raw preview: {text[:300]!r}") | |
| # Strip markdown code fences (e.g. ```json ... ``` or ``` ... ```) | |
| if text.startswith("```"): | |
| # Drop the entire opening fence line (```json, ```python, or just ```) | |
| first_newline = text.find("\n") | |
| text = text[first_newline + 1:] if first_newline != -1 else text[3:] | |
| # Drop the closing fence | |
| if text.rstrip().endswith("```"): | |
| text = text.rstrip()[:-3].rstrip() | |
| print(f"[parse_plan_json] after fence strip: {text[:300]!r}") | |
| try: | |
| data = json.loads(text) | |
| except json.JSONDecodeError: | |
| # Repair common LLM JSON issues (unescaped quotes, trailing commas, etc.) | |
| print(f"[parse_plan_json] json.loads failed, attempting repair...") | |
| repaired = repair_json(text, return_objects=True) | |
| data = repaired if isinstance(repaired, dict) else json.loads(repair_json(text)) | |
| plan = ProjectPlan.model_validate(data) | |
| print(f"[parse_plan_json] phases parsed: {len(plan.phases)}") | |
| return plan | |
| except Exception as exc: | |
| print(f"[parse_plan_json] FAILED: {type(exc).__name__}: {exc}") | |
| return ProjectPlan() | |
| def merge_plan_update(existing: ProjectPlan, update: dict) -> ProjectPlan: | |
| """Merge a partial update dict into an existing ProjectPlan.""" | |
| merged = existing.model_dump() | |
| merged.update(update) | |
| return ProjectPlan.model_validate(merged) | |