Spaces:
Running
Running
| from flask import Flask, request, jsonify | |
| from flask_cors import CORS | |
| import json | |
| from datetime import datetime | |
| from typing import Optional, Dict, Any, List | |
| import re | |
| import os | |
| import io | |
| from pypdf import PdfReader | |
| from docx import Document | |
| from firestore_client import get_firestore_client | |
| from gemini_client import ask_gpt | |
| from prompt_instructions import build_system_message | |
| from role_access import get_allowed_collections | |
| from data_fetcher import fetch_data_from_firestore | |
| from data_planner import determine_data_requirements | |
| from resolver import resolve_user_context | |
| from schema_utils import has_field, resolve_field | |
| app = Flask(__name__) | |
| CORS(app) | |
| db = get_firestore_client() | |
| # -- helpers -------------------------------------------------------------- | |
| def to_jsonable(obj): | |
| """Recursively convert Firestore types (e.g., DatetimeWithNanoseconds) to JSON-safe values.""" | |
| # Firestore timestamp types come through as DatetimeWithNanoseconds (duck-type datetime) | |
| if isinstance(obj, datetime): | |
| return obj.isoformat() | |
| # Some Firestore SDKs expose Timestamp-like objects without subclassing datetime. | |
| # Fallback: detect presence of isoformat() | |
| if hasattr(obj, "isoformat") and callable(getattr(obj, "isoformat")): | |
| try: | |
| return obj.isoformat() | |
| except Exception: | |
| pass | |
| if isinstance(obj, dict): | |
| return {k: to_jsonable(v) for k, v in obj.items()} | |
| if isinstance(obj, list): | |
| return [to_jsonable(v) for v in obj] | |
| return obj | |
| # 🔧 Normalize Gemini plan into proper Firestore fetch format | |
| def normalize_plan(plan: dict, token_map: Optional[Dict[str, Any]] = None) -> dict: | |
| token_map = token_map or {} | |
| filters = plan.get("filters", {}) or {} | |
| planned_cols = plan.get("collections", []) or [] | |
| def canonical_value(key, val): | |
| # replace tokens like {{participantId}} | |
| if isinstance(val, str) and val in token_map: | |
| val = token_map[val] | |
| # special-case status normalization | |
| if key == "status" and val == "running": | |
| return "active" | |
| return val | |
| collections_out = [] | |
| for col in planned_cols: | |
| # Allow both strings ("participants") and objects ({"name":"participants","fields":[...]}). | |
| name = col["name"] if isinstance(col, dict) else col | |
| col_filters = [] | |
| for k, v in filters.items(): | |
| canon_key = resolve_field(name, k) | |
| # only add filters that exist for this collection (schema-aware) | |
| if has_field(name, canon_key): | |
| col_filters.append({ | |
| "field": canon_key, | |
| "op": "==", | |
| "value": canonical_value(k, v), | |
| }) | |
| # else: skip invalid field for this collection | |
| collections_out.append({"name": name, "filters": col_filters, "limit": 50}) | |
| return {"collections": collections_out} | |
| def _norm(v): | |
| return str(v or "").strip().lower() | |
| def _safe_number(v, default=0): | |
| try: | |
| n = float(v) | |
| return n | |
| except Exception: | |
| return default | |
| def _clamp_pct(n): | |
| try: | |
| return max(0, min(100, round(float(n)))) | |
| except Exception: | |
| return 0 | |
| def _normalize_metric(metric: Optional[str]) -> Optional[str]: | |
| m = _norm(metric) | |
| if not m: | |
| return None | |
| if "hour" in m: | |
| return "hours" | |
| if "session" in m: | |
| return "sessions" | |
| if "document" in m: | |
| return "documents" | |
| if "deliverable" in m: | |
| return "deliverables" | |
| if "report" in m: | |
| return "reports" | |
| if "submission" in m: | |
| return "submissions" | |
| return None | |
| def _proof_rules_for_metric(metric_key: Optional[str]) -> List[Dict[str, Any]]: | |
| if metric_key == "sessions": | |
| return [ | |
| { | |
| "type": "attendance", | |
| "label": "Attendance register / session proof", | |
| "reason": "Sessions usually require attendance or meeting evidence.", | |
| "required": True, | |
| }, | |
| { | |
| "type": "minutes", | |
| "label": "Session notes / minutes", | |
| "reason": "Session outcomes should be supported by notes or minutes.", | |
| "required": False, | |
| }, | |
| ] | |
| if metric_key in ["reports", "documents", "deliverables", "submissions"]: | |
| return [ | |
| { | |
| "type": "document", | |
| "label": "Output file / submitted document", | |
| "reason": "This metric usually requires the actual file or submission proof.", | |
| "required": True, | |
| } | |
| ] | |
| if metric_key == "hours": | |
| return [ | |
| { | |
| "type": "supporting_evidence", | |
| "label": "Supporting evidence", | |
| "reason": "Hours-only updates should still have some supporting evidence where possible.", | |
| "required": False, | |
| } | |
| ] | |
| return [] | |
| def _build_intervention_analysis_prompt(payload: Dict[str, Any]) -> str: | |
| return f""" | |
| You are analysing a consultant intervention progress update. | |
| Return STRICT JSON only with this shape: | |
| {{ | |
| "summary": "short summary", | |
| "polishedNotes": "professional progress notes", | |
| "activities": ["..."], | |
| "outputs": ["..."], | |
| "blockers": ["..."], | |
| "nextSteps": ["..."], | |
| "suggestedHours": 0, | |
| "suggestedUnitsCompleted": 0, | |
| "confidence": 0, | |
| "completionSignal": "none|partial|strong", | |
| "warnings": ["..."], | |
| "proofHints": [ | |
| {{ | |
| "type": "attendance|minutes|document|photo|supporting_evidence|submission_receipt|other", | |
| "label": "short label", | |
| "reason": "why this is needed" | |
| }} | |
| ] | |
| }} | |
| Rules: | |
| - Use the user's notes/transcript/message as the primary source. | |
| - Do not invent facts not grounded in the input. | |
| - suggestedHours should be conservative. | |
| - suggestedUnitsCompleted should only be given if the described work clearly maps to a measurable unit. | |
| - completionSignal means: | |
| - "none" = no sign of completion | |
| - "partial" = meaningful progress but not clearly complete | |
| - "strong" = strong signal that deliverable/work may be complete | |
| - warnings should mention ambiguity, weak evidence, or missing detail. | |
| - polishedNotes should be concise and professional. | |
| Intervention context: | |
| {json.dumps(payload, ensure_ascii=False)} | |
| """.strip() | |
| def _extract_json_block(text: str) -> Dict[str, Any]: | |
| text = (text or "").strip() | |
| try: | |
| return json.loads(text) | |
| except Exception: | |
| pass | |
| match = re.search(r"\{.*\}", text, flags=re.DOTALL) | |
| if match: | |
| try: | |
| return json.loads(match.group(0)) | |
| except Exception: | |
| pass | |
| return { | |
| "summary": "", | |
| "polishedNotes": text, | |
| "activities": [], | |
| "outputs": [], | |
| "blockers": [], | |
| "nextSteps": [], | |
| "suggestedHours": None, | |
| "suggestedUnitsCompleted": None, | |
| "confidence": 40, | |
| "completionSignal": "none", | |
| "warnings": ["AI response was not clean JSON. Parsed fallback response."], | |
| "proofHints": [], | |
| } | |
| def _merge_proof_hints(metric_rules: List[Dict[str, Any]], ai_hints: List[Dict[str, Any]]) -> List[Dict[str, Any]]: | |
| merged = [] | |
| seen = set() | |
| for item in metric_rules + (ai_hints or []): | |
| t = _norm(item.get("type")) | |
| label = str(item.get("label") or "").strip() | |
| key = f"{t}|{label.lower()}" | |
| if not t and not label: | |
| continue | |
| if key in seen: | |
| continue | |
| seen.add(key) | |
| merged.append({ | |
| "type": t or "other", | |
| "label": label or "Supporting evidence", | |
| "reason": str(item.get("reason") or "").strip(), | |
| "required": bool(item.get("required", False)), | |
| }) | |
| return merged | |
| def _calculate_progress_suggestion(intervention: Dict[str, Any], ai_result: Dict[str, Any]) -> Dict[str, Any]: | |
| target_type = _norm(intervention.get("targetType")) | |
| target_value = _safe_number(intervention.get("targetValue"), 0) | |
| current_progress = _clamp_pct(_safe_number(intervention.get("progress"), 0)) | |
| current_actual = _safe_number(intervention.get("targetActual"), 0) | |
| metric_key = _normalize_metric(intervention.get("targetMetric")) | |
| suggested_hours = ai_result.get("suggestedHours") | |
| try: | |
| suggested_hours = float(suggested_hours) if suggested_hours is not None else None | |
| except Exception: | |
| suggested_hours = None | |
| suggested_units = ai_result.get("suggestedUnitsCompleted") | |
| try: | |
| suggested_units = float(suggested_units) if suggested_units is not None else None | |
| except Exception: | |
| suggested_units = None | |
| if target_type == "percentage": | |
| # For percentage interventions, AI can only suggest a conservative bump. | |
| base_bump = 0 | |
| signal = _norm(ai_result.get("completionSignal")) | |
| if signal == "strong": | |
| base_bump = 20 | |
| elif signal == "partial": | |
| base_bump = 10 | |
| elif suggested_hours and suggested_hours > 0: | |
| base_bump = 5 | |
| suggested_progress = _clamp_pct(current_progress + base_bump) | |
| return { | |
| "targetMode": "percentage", | |
| "metricKey": metric_key, | |
| "suggestedHours": suggested_hours, | |
| "suggestedUnitsCompleted": None, | |
| "currentActual": None, | |
| "suggestedActualAfter": None, | |
| "suggestedProgressPct": suggested_progress, | |
| "overTargetBy": 0, | |
| } | |
| if target_type == "number" and target_value > 0: | |
| units = suggested_units | |
| if metric_key == "hours" and (units is None or units <= 0): | |
| units = suggested_hours | |
| if units is None or units <= 0: | |
| return { | |
| "targetMode": "number", | |
| "metricKey": metric_key, | |
| "suggestedHours": suggested_hours, | |
| "suggestedUnitsCompleted": None, | |
| "currentActual": current_actual, | |
| "suggestedActualAfter": current_actual, | |
| "suggestedProgressPct": current_progress, | |
| "overTargetBy": 0, | |
| } | |
| actual_after = current_actual + units | |
| progress_after = _clamp_pct((actual_after / target_value) * 100) | |
| over_target_by = max(0, actual_after - target_value) | |
| return { | |
| "targetMode": "number", | |
| "metricKey": metric_key, | |
| "suggestedHours": suggested_hours, | |
| "suggestedUnitsCompleted": units, | |
| "currentActual": current_actual, | |
| "suggestedActualAfter": actual_after, | |
| "suggestedProgressPct": progress_after, | |
| "overTargetBy": over_target_by, | |
| } | |
| return { | |
| "targetMode": "none", | |
| "metricKey": metric_key, | |
| "suggestedHours": suggested_hours, | |
| "suggestedUnitsCompleted": suggested_units, | |
| "currentActual": None, | |
| "suggestedActualAfter": None, | |
| "suggestedProgressPct": current_progress, | |
| "overTargetBy": 0, | |
| } | |
| ALLOWED_COURSE_SOURCE_EXTENSIONS = {"pdf", "docx"} | |
| MAX_SOURCE_TEXT_CHARS = 60000 | |
| def _allowed_course_source(filename: str) -> bool: | |
| if not filename or "." not in filename: | |
| return False | |
| return filename.rsplit(".", 1)[1].lower() in ALLOWED_COURSE_SOURCE_EXTENSIONS | |
| def _clean_extracted_text(text: str) -> str: | |
| if not text: | |
| return "" | |
| text = text.replace("\x00", " ") | |
| text = re.sub(r"[ \t]+", " ", text) | |
| text = re.sub(r"\n{3,}", "\n\n", text) | |
| return text.strip() | |
| def _extract_text_from_pdf_bytes(file_bytes: bytes) -> str: | |
| reader = PdfReader(io.BytesIO(file_bytes)) | |
| pages = [] | |
| for page in reader.pages: | |
| try: | |
| pages.append(page.extract_text() or "") | |
| except Exception: | |
| pages.append("") | |
| return _clean_extracted_text("\n\n".join(pages)) | |
| def _extract_text_from_docx_bytes(file_bytes: bytes) -> str: | |
| doc = Document(io.BytesIO(file_bytes)) | |
| lines: List[str] = [] | |
| for p in doc.paragraphs: | |
| txt = (p.text or "").strip() | |
| if txt: | |
| lines.append(txt) | |
| for table in doc.tables: | |
| for row in table.rows: | |
| row_text = " | ".join((cell.text or "").strip() for cell in row.cells if (cell.text or "").strip()) | |
| if row_text: | |
| lines.append(row_text) | |
| return _clean_extracted_text("\n".join(lines)) | |
| def _extract_course_source_text(filename: str, file_bytes: bytes) -> str: | |
| ext = filename.rsplit(".", 1)[1].lower() | |
| if ext == "pdf": | |
| return _extract_text_from_pdf_bytes(file_bytes) | |
| if ext == "docx": | |
| return _extract_text_from_docx_bytes(file_bytes) | |
| raise ValueError("Unsupported file type") | |
| def _truncate_source_text(text: str, limit: int = MAX_SOURCE_TEXT_CHARS) -> str: | |
| if len(text) <= limit: | |
| return text | |
| return text[:limit] | |
| def _build_course_outline_prompt(source_text: str, filename: str) -> str: | |
| return f""" | |
| You are designing a practical learning course outline from source material. | |
| Return STRICT JSON only with this exact shape: | |
| {{ | |
| "courseTitle": "string", | |
| "courseDescription": "string", | |
| "difficulty": "beginner|intermediate|advanced", | |
| "category": "string", | |
| "courseType": "string", | |
| "estimatedTotalDuration": "string", | |
| "learningObjectives": ["string"], | |
| "modules": [ | |
| {{ | |
| "type": "lesson|quiz|assignment|review", | |
| "title": "string", | |
| "description": "string", | |
| "duration": "e.g. 20m or 1h", | |
| "content": "only for lesson when useful", | |
| "assignmentPrompt": "only for assignment when useful", | |
| "answerKey": "only for assignment when useful", | |
| "questions": [ | |
| {{ | |
| "question": "string", | |
| "options": ["string", "string", "string", "string"], | |
| "correctAnswer": 0 | |
| }} | |
| ] | |
| }} | |
| ], | |
| "warnings": ["string"] | |
| }} | |
| Rules: | |
| - Build a course outline grounded in the uploaded document. | |
| - Prefer 4 to 12 modules unless the source strongly suggests otherwise. | |
| - Most modules should be lessons. | |
| - Include quizzes only where knowledge checks make sense. | |
| - Include assignments only when there is something practical to apply. | |
| - Include review modules only when useful for recap. | |
| - Every lesson must have a realistic duration estimate. | |
| - estimatedTotalDuration must reflect the sum of lesson durations approximately. | |
| - Keep titles practical and clean. | |
| - Do not invent niche facts that are not supported by the source. | |
| - If the document is too thin, still produce a usable outline and add a warning. | |
| - If the content looks like a scanned PDF with poor extraction, say so in warnings. | |
| Filename: {filename} | |
| Source document text: | |
| {source_text} | |
| """.strip() | |
| def _normalize_outline_json(ai_result: Dict[str, Any]) -> Dict[str, Any]: | |
| raw_modules = ai_result.get("modules") or [] | |
| out_modules = [] | |
| for idx, mod in enumerate(raw_modules): | |
| mtype = str(mod.get("type") or "lesson").strip().lower() | |
| if mtype not in ["lesson", "quiz", "assignment", "review"]: | |
| mtype = "lesson" | |
| base = { | |
| "id": f"module-{idx + 1}", | |
| "type": mtype, | |
| "title": str(mod.get("title") or f"Module {idx + 1}").strip(), | |
| "description": str(mod.get("description") or "").strip(), | |
| } | |
| if mtype == "lesson": | |
| base["duration"] = str(mod.get("duration") or "20m").strip() | |
| base["content"] = str(mod.get("content") or "").strip() | |
| base["videoUrls"] = [] | |
| base["imageUrls"] = [] | |
| elif mtype == "quiz": | |
| questions = [] | |
| for qidx, q in enumerate(mod.get("questions") or []): | |
| options = q.get("options") or [] | |
| while len(options) < 4: | |
| options.append("") | |
| questions.append({ | |
| "id": f"q-{idx + 1}-{qidx + 1}", | |
| "question": str(q.get("question") or "").strip(), | |
| "options": [str(x or "").strip() for x in options[:4]], | |
| "correctAnswer": int(q.get("correctAnswer") or 0), | |
| }) | |
| base["questions"] = questions | |
| elif mtype == "assignment": | |
| base["assignmentPrompt"] = str(mod.get("assignmentPrompt") or "").strip() | |
| base["answerKey"] = str(mod.get("answerKey") or "").strip() | |
| out_modules.append(base) | |
| return { | |
| "courseTitle": str(ai_result.get("courseTitle") or "").strip(), | |
| "courseDescription": str(ai_result.get("courseDescription") or "").strip(), | |
| "difficulty": str(ai_result.get("difficulty") or "beginner").strip().lower(), | |
| "category": str(ai_result.get("category") or "General").strip(), | |
| "courseType": str(ai_result.get("courseType") or "Foundational").strip(), | |
| "estimatedTotalDuration": str(ai_result.get("estimatedTotalDuration") or "").strip(), | |
| "learningObjectives": [ | |
| str(x).strip() for x in (ai_result.get("learningObjectives") or []) if str(x).strip() | |
| ], | |
| "modules": out_modules, | |
| "warnings": [ | |
| str(x).strip() for x in (ai_result.get("warnings") or []) if str(x).strip() | |
| ], | |
| } | |
| # --- SME Intake helpers ------------------------------------------------- | |
| ALLOWED_SME_DOC_EXTENSIONS = {"pdf", "docx", "txt", "png", "jpg", "jpeg"} | |
| MAX_SME_DOC_TEXT_CHARS = 45000 | |
| def _allowed_sme_doc(filename: str) -> bool: | |
| if not filename or "." not in filename: | |
| return False | |
| return filename.rsplit(".", 1)[1].lower() in ALLOWED_SME_DOC_EXTENSIONS | |
| def _extract_text_from_txt_bytes(file_bytes: bytes) -> str: | |
| try: | |
| return _clean_extracted_text(file_bytes.decode("utf-8", errors="ignore")) | |
| except Exception: | |
| return "" | |
| def _extract_sme_doc_text(filename: str, file_bytes: bytes) -> str: | |
| ext = filename.rsplit(".", 1)[1].lower() | |
| if ext == "pdf": | |
| return _extract_text_from_pdf_bytes(file_bytes) | |
| if ext == "docx": | |
| return _extract_text_from_docx_bytes(file_bytes) | |
| if ext == "txt": | |
| return _extract_text_from_txt_bytes(file_bytes) | |
| # For images, we are not doing OCR here. | |
| # Send metadata to AI, but text extraction stays blank. | |
| if ext in ["png", "jpg", "jpeg"]: | |
| return "" | |
| return "" | |
| def _fetch_active_programs(company_code: Optional[str] = None) -> List[Dict[str, Any]]: | |
| try: | |
| ref = db.collection("programs") | |
| if company_code: | |
| docs = ref.where("companyCode", "==", company_code).stream() | |
| else: | |
| docs = ref.stream() | |
| programs = [] | |
| for doc_snap in docs: | |
| data = doc_snap.to_dict() or {} | |
| status = _norm(data.get("status") or data.get("programStatus")) | |
| if status and status not in ["active", "open", "running"]: | |
| continue | |
| eligibility = data.get("eligibilityCriteria") or {} | |
| programs.append({ | |
| "id": doc_snap.id, | |
| "companyCode": data.get("companyCode") or "", | |
| "name": data.get("name") or "", | |
| "description": data.get("description") or "", | |
| "type": data.get("type") or "", | |
| "cohortYear": data.get("cohortYear") or "", | |
| "startDate": to_jsonable(data.get("startDate")), | |
| "endDate": to_jsonable(data.get("endDate")), | |
| "maxCapacity": data.get("maxCapacity"), | |
| "registrationLink": data.get("registrationLink") or "", | |
| "eligibilityCriteria": eligibility, | |
| "eligibilitySummary": { | |
| "minimumAge": eligibility.get("minimumAge"), | |
| "maximumAge": eligibility.get("maximumAge"), | |
| "gender": eligibility.get("gender"), | |
| "sector": eligibility.get("sector"), | |
| "province": eligibility.get("province"), | |
| "minYearsOfTrading": eligibility.get("minYearsOfTrading"), | |
| "bbbeeLevel": eligibility.get("bbbeeLevel"), | |
| "minYouthOwnershipPercent": eligibility.get("minYouthOwnershipPercent"), | |
| "minBlackOwnershipPercent": eligibility.get("minBlackOwnershipPercent"), | |
| "minFemaleOwnershipPercent": eligibility.get("minFemaleOwnershipPercent"), | |
| }, | |
| }) | |
| return programs[:50] | |
| except Exception as e: | |
| print("fetch_active_programs_failed:", e) | |
| return [] | |
| def _fetch_intervention_catalog(company_code: str) -> List[Dict[str, Any]]: | |
| try: | |
| docs = ( | |
| db.collection("interventions") | |
| .where("companyCode", "==", company_code) | |
| .stream() | |
| ) | |
| interventions = [] | |
| for doc_snap in docs: | |
| data = doc_snap.to_dict() or {} | |
| status = _norm(data.get("status")) | |
| if status and status not in ["active", "enabled", "approved"]: | |
| continue | |
| interventions.append({ | |
| "id": doc_snap.id, | |
| "title": data.get("title") or data.get("interventionTitle") or data.get("name") or "", | |
| "department": data.get("departmentName") or data.get("areaOfSupport") or data.get("department") or "", | |
| "description": data.get("description") or data.get("objective") or "", | |
| "tags": data.get("tags") or [], | |
| "stage": data.get("stage") or data.get("businessStage") or "", | |
| }) | |
| return interventions[:80] | |
| except Exception as e: | |
| print("fetch_intervention_catalog_failed:", e) | |
| return [] | |
| def _build_sme_intake_prompt(payload: Dict[str, Any]) -> str: | |
| return f""" | |
| You are analysing an unregistered SME intake for an incubation/support platform. | |
| Return STRICT JSON only with this exact shape: | |
| {{ | |
| "detectedLanguages": ["English"], | |
| "confidence": 0, | |
| "extractedProfile": {{ | |
| "businessName": "", | |
| "beneficiaryName": "", | |
| "contactPerson": "", | |
| "email": "", | |
| "phone": "", | |
| "registrationNumber": "", | |
| "taxNumber": "", | |
| "sector": "", | |
| "province": "", | |
| "city": "", | |
| "yearsOperating": "", | |
| "employeeCount": "", | |
| "monthlyRevenue": "", | |
| "mainChallenges": [] | |
| }}, | |
| "extractedApplication": {{ | |
| "motivation": "", | |
| "challenges": "", | |
| "businessDescription": "", | |
| "supportNeeded": [], | |
| "documentsMentioned": [] | |
| }}, | |
| "documentFindings": [ | |
| {{ | |
| "documentType": "id_document|cipc|tax_pin|other", | |
| "filename": "", | |
| "extractedFields": {{}}, | |
| "confidence": 0, | |
| "warnings": [] | |
| }} | |
| ], | |
| "missingFields": [ | |
| {{ | |
| "field": "", | |
| "label": "", | |
| "question": "", | |
| "reason": "" | |
| }} | |
| ], | |
| "missingDocuments": [ | |
| {{ | |
| "type": "id_document|cipc|tax_pin|other", | |
| "reason": "" | |
| }} | |
| ], | |
| "nextQuestion": "", | |
| "decision": {{ | |
| "stage": "idea|startup|early_growth|struggling|scaling|market_ready", | |
| "urgencyLevel": "low|medium|high|urgent", | |
| "urgencyScore": 0, | |
| "riskLevel": "low|medium|high", | |
| "recommendedProgramId": "", | |
| "recommendedProgramName": "", | |
| "recommendedInterventions": [ | |
| {{ | |
| "title": "", | |
| "department": "", | |
| "urgency": "low|medium|high|urgent", | |
| "reason": "" | |
| }} | |
| ], | |
| "summary": "", | |
| "classificationReasons": [], | |
| "redFlags": [], | |
| "growthSignals": [], | |
| "benefitsOfProgram": [], | |
| "benefitsOfAgents": [] | |
| }}, | |
| "warnings": [] | |
| }} | |
| Rules: | |
| - Use the SME story and uploaded document text only. | |
| - Do not invent registration numbers, tax numbers, IDs, turnover, employees, or compliance status. | |
| - If a field is missing, add it to missingFields using a friendly conversational question. | |
| - Required basic documents are: Director ID, CIPC document, SARS Tax PIN. | |
| - If a required document is not uploaded or cannot be identified, add it to missingDocuments. | |
| - Classify the SME stage: | |
| - idea: concept only, no operating history | |
| - startup: newly operating or still proving model | |
| - early_growth: some traction, some customers/revenue | |
| - struggling: stagnant revenue, compliance gaps, cashflow pressure, no growth, operational distress | |
| - scaling: growing and needing structured support | |
| - market_ready: compliant and ready for procurement/market linkage | |
| - Urgency must reflect intervention need, not emotional tone. | |
| - urgent/high urgency should be used where compliance, tax, funding, payroll, legal, safety, or survival risks are clear. | |
| - Recommend interventions only from the intervention catalog where possible. | |
| - Recommend one program from availablePrograms where possible. | |
| - Match programmes using eligibilityCriteria first, especially province, sector, gender, age, years of trading, B-BBEE level, youth ownership, black ownership, and female ownership. | |
| - If the SME does not clearly meet a programme's eligibility criteria, do not force that programme. | |
| - If required eligibility data is missing from the SME story, add friendly questions to missingFields. | |
| - If no programme fits, leave recommendedProgramId blank and explain why in summary. | |
| - benefitsOfProgram must explain why structured incubation helps. | |
| - benefitsOfAgents must explain why direct agent support helps without programme onboarding. | |
| - Keep text professional and concise. | |
| - Return JSON only. | |
| Payload: | |
| {json.dumps(payload, ensure_ascii=False)} | |
| """.strip() | |
| def _normalize_sme_intake_result(raw: Dict[str, Any]) -> Dict[str, Any]: | |
| decision = raw.get("decision") or {} | |
| allowed_stages = ["idea", "startup", "early_growth", "struggling", "scaling", "market_ready"] | |
| allowed_urgency = ["low", "medium", "high", "urgent"] | |
| allowed_risk = ["low", "medium", "high"] | |
| stage = _norm(decision.get("stage")) | |
| urgency = _norm(decision.get("urgencyLevel")) | |
| risk = _norm(decision.get("riskLevel")) | |
| if stage not in allowed_stages: | |
| stage = "startup" | |
| if urgency not in allowed_urgency: | |
| urgency = "medium" | |
| if risk not in allowed_risk: | |
| risk = "medium" | |
| interventions = [] | |
| for item in decision.get("recommendedInterventions") or []: | |
| item_urgency = _norm(item.get("urgency")) | |
| if item_urgency not in allowed_urgency: | |
| item_urgency = urgency | |
| title = str(item.get("title") or "").strip() | |
| if not title: | |
| continue | |
| interventions.append({ | |
| "title": title, | |
| "department": str(item.get("department") or "").strip(), | |
| "urgency": item_urgency, | |
| "reason": str(item.get("reason") or "").strip(), | |
| }) | |
| missing_fields = [] | |
| for item in raw.get("missingFields") or []: | |
| field = str(item.get("field") or "").strip() | |
| question = str(item.get("question") or "").strip() | |
| if not field or not question: | |
| continue | |
| missing_fields.append({ | |
| "field": field, | |
| "label": str(item.get("label") or field).strip(), | |
| "question": question, | |
| "reason": str(item.get("reason") or "").strip(), | |
| }) | |
| missing_documents = [] | |
| for item in raw.get("missingDocuments") or []: | |
| doc_type = str(item.get("type") or "").strip() | |
| if not doc_type: | |
| continue | |
| missing_documents.append({ | |
| "type": doc_type, | |
| "reason": str(item.get("reason") or "").strip(), | |
| }) | |
| document_findings = [] | |
| for item in raw.get("documentFindings") or []: | |
| document_findings.append({ | |
| "documentType": str(item.get("documentType") or "other").strip(), | |
| "filename": str(item.get("filename") or "").strip(), | |
| "extractedFields": item.get("extractedFields") or {}, | |
| "confidence": _clamp_pct(item.get("confidence")), | |
| "warnings": [ | |
| str(x).strip() | |
| for x in item.get("warnings", []) | |
| if str(x).strip() | |
| ], | |
| }) | |
| return { | |
| "detectedLanguages": [ | |
| str(x).strip() | |
| for x in raw.get("detectedLanguages", []) | |
| if str(x).strip() | |
| ], | |
| "confidence": _clamp_pct(raw.get("confidence")), | |
| "extractedProfile": raw.get("extractedProfile") or {}, | |
| "extractedApplication": raw.get("extractedApplication") or {}, | |
| "documentFindings": document_findings, | |
| "missingFields": missing_fields[:12], | |
| "missingDocuments": missing_documents[:10], | |
| "nextQuestion": str(raw.get("nextQuestion") or "").strip(), | |
| "decision": { | |
| "stage": stage, | |
| "urgencyLevel": urgency, | |
| "urgencyScore": _clamp_pct(decision.get("urgencyScore")), | |
| "riskLevel": risk, | |
| "recommendedProgramId": str(decision.get("recommendedProgramId") or "").strip(), | |
| "recommendedProgramName": str(decision.get("recommendedProgramName") or "").strip(), | |
| "recommendedInterventions": interventions[:10], | |
| "summary": str(decision.get("summary") or "").strip(), | |
| "classificationReasons": [ | |
| str(x).strip() | |
| for x in decision.get("classificationReasons", []) | |
| if str(x).strip() | |
| ][:10], | |
| "redFlags": [ | |
| str(x).strip() | |
| for x in decision.get("redFlags", []) | |
| if str(x).strip() | |
| ][:10], | |
| "growthSignals": [ | |
| str(x).strip() | |
| for x in decision.get("growthSignals", []) | |
| if str(x).strip() | |
| ][:10], | |
| "benefitsOfProgram": [ | |
| str(x).strip() | |
| for x in decision.get("benefitsOfProgram", []) | |
| if str(x).strip() | |
| ][:6], | |
| "benefitsOfAgents": [ | |
| str(x).strip() | |
| for x in decision.get("benefitsOfAgents", []) | |
| if str(x).strip() | |
| ][:6], | |
| }, | |
| "warnings": [ | |
| str(x).strip() | |
| for x in raw.get("warnings", []) | |
| if str(x).strip() | |
| ], | |
| } | |
| def build_assignments_page_context(company_code, user_id, program_id, role, user_context): | |
| """ | |
| Server-side mirror of the assignments page. | |
| Must return: | |
| - current user's department | |
| - eligible SMEs | |
| - each SME's required interventions from diagnosticPlans | |
| - confirmation status per department | |
| - existing assignedInterventions | |
| - available consultants for department | |
| - recurring / once-off assignment metadata | |
| """ | |
| # Fetch: | |
| # users / operationsStaff / consultants | |
| # departments | |
| # diagnosticPlans | |
| # applications | |
| # participants | |
| # interventions | |
| # assignedInterventions | |
| return { | |
| "page": "assignments", | |
| "companyCode": company_code, | |
| "programId": program_id or "all", | |
| "department": {}, | |
| "rules": { | |
| "smmeMustHaveAcceptedApplication": True, | |
| "diagnosticPlanMustBeConfirmedByDepartment": True, | |
| "diagnosticPlanMustBeConfirmedBySmme": True, | |
| "groupedAssignmentRequiresSharedIntervention": True, | |
| "recurringCannotSkipOpenPreviousCycle": True, | |
| "onceOffCannotRepeatAfterCompletion": True, | |
| }, | |
| "smes": [], | |
| "interventions": [], | |
| "assignments": [], | |
| "consultants": [] | |
| } | |
| # -- route --------------------------------------------------------------- | |
| def chat(): | |
| data = request.json | |
| role = data.get('role') | |
| user_input = data.get('message') | |
| company_code = data.get('companyCode') | |
| user_id = data.get('userId') | |
| if not role or not user_input or not company_code or not user_id: | |
| return jsonify({"error": "Missing role, message, companyCode, or userId"}), 400 | |
| # 🔎 Resolve current user's email + participantId (participants has no companyCode; resolve via applications) | |
| ctx = resolve_user_context(user_id, company_code) | |
| token_map = { | |
| "{{participantId}}": ctx.get("participantId"), | |
| "{{userEmail}}": ctx.get("email"), | |
| "{{userId}}": ctx.get("uid"), | |
| } | |
| # 🧠 Plan | |
| planning_result = determine_data_requirements( | |
| user_input, company_code, user_id, | |
| participant_email=ctx.get("email"), | |
| participant_id=ctx.get("participantId"), | |
| ) | |
| if "error" in planning_result: | |
| return jsonify({"reply": f"⚠️ Planning error: {planning_result['error']}"}) | |
| # 🛠️ Normalize & replace tokens (schema-aware) | |
| normalized_plan = normalize_plan(planning_result, token_map) | |
| # 📥 Fetch | |
| firestore_data = fetch_data_from_firestore(normalized_plan) | |
| # 🧩 Build messages — convert Firestore payload to JSON-safe types | |
| system_msg = build_system_message(company_code) | |
| safe_ctx = to_jsonable(ctx) | |
| safe_data = to_jsonable(firestore_data) | |
| data_msg = { | |
| "role": "system", | |
| "content": ( | |
| f"CurrentUserContext: {json.dumps(safe_ctx)}\n" | |
| f"Here is the data from Firestore:\n{json.dumps(safe_data)}" | |
| ) | |
| } | |
| user_msg = {"role": "user", "content": user_input} | |
| final_response = ask_gpt([system_msg, data_msg, user_msg]) | |
| return jsonify({"reply": final_response}) | |
| def page_assist(): | |
| data = request.json or {} | |
| page = data.get("page") | |
| message = data.get("message") | |
| role = data.get("role") | |
| company_code = data.get("companyCode") | |
| user_id = data.get("userId") | |
| program_id = data.get("programId") # optional, "all" allowed | |
| if not page or not message or not role or not company_code or not user_id: | |
| return jsonify({"error": "Missing page, message, role, companyCode, or userId"}), 400 | |
| if page != "assignments": | |
| return jsonify({"error": "Unsupported page"}), 400 | |
| ctx = resolve_user_context(user_id, company_code) | |
| assignment_context = build_assignments_page_context( | |
| company_code=company_code, | |
| user_id=user_id, | |
| program_id=None if program_id in [None, "", "all"] else program_id, | |
| role=role, | |
| user_context=ctx, | |
| ) | |
| system_msg = { | |
| "role": "system", | |
| "content": """ | |
| You are a page-aware assistant for the Interventions Assignments page. | |
| Answer using ONLY the supplied page context. | |
| Explain blockers clearly. | |
| Do not invent SMEs, interventions, departments, or assignment rules. | |
| If asked why an intervention cannot be assigned, identify the exact reason: | |
| - not part of that SME's Diagnostic Plan | |
| - Diagnostic Plan not confirmed by both department and SME | |
| - not shared across selected SMEs | |
| - previous recurring cycle still open | |
| - once-off intervention already completed | |
| - missing facilitator, due date, target mode, or sub-intervention | |
| Return concise JSON: | |
| { | |
| "reply": "", | |
| "reasonCode": "", | |
| "affectedSmes": [], | |
| "eligibleSmes": [], | |
| "blockedSmes": [], | |
| "suggestedNextAction": "" | |
| } | |
| """ | |
| } | |
| user_msg = { | |
| "role": "user", | |
| "content": json.dumps({ | |
| "question": message, | |
| "pageContext": assignment_context, | |
| }, ensure_ascii=False) | |
| } | |
| ai_raw = ask_gpt([system_msg, user_msg]) | |
| result = _extract_json_block(ai_raw) | |
| return jsonify(to_jsonable(result)) | |
| def assignments_agent(): | |
| data = request.json or {} | |
| company_code = data.get("companyCode") | |
| user_id = data.get("userId") | |
| role = data.get("role") | |
| message = data.get("message") or data.get("transcript") | |
| program_id = data.get("programId") | |
| if not company_code or not user_id or not role or not message: | |
| return jsonify({"error": "Missing companyCode, userId, role, or message"}), 400 | |
| ctx = resolve_user_context(user_id, company_code) | |
| assignment_context = build_assignments_page_context( | |
| company_code=company_code, | |
| user_id=user_id, | |
| program_id=None if program_id in [None, "", "all"] else program_id, | |
| role=role, | |
| user_context=ctx, | |
| ) | |
| system_msg = { | |
| "role": "system", | |
| "content": """ | |
| You are an assignment action planner. | |
| Extract the user's intent, but DO NOT save anything directly. | |
| Return strict JSON: | |
| { | |
| "intent": "assign_intervention|explain|unknown", | |
| "interventionName": "", | |
| "smeNames": [], | |
| "assignmentType": "singular|grouped|unknown", | |
| "assignTo": "consultant|operations|unknown", | |
| "consultantName": "", | |
| "dueDate": "", | |
| "targetMode": "hours|sessions|documents|milestones|percentage|unknown", | |
| "missingFields": [], | |
| "canExecute": false, | |
| "validationErrors": [], | |
| "confirmationMessage": "" | |
| } | |
| Required fields before execution: | |
| - interventionName | |
| - at least one SME | |
| - assignTo | |
| - dueDate | |
| - targetMode | |
| - consultantName if assignTo is consultant | |
| Validate against pageContext. | |
| """ | |
| } | |
| user_msg = { | |
| "role": "user", | |
| "content": json.dumps({ | |
| "message": message, | |
| "pageContext": assignment_context, | |
| }, ensure_ascii=False) | |
| } | |
| ai_raw = ask_gpt([system_msg, user_msg]) | |
| plan = _extract_json_block(ai_raw) | |
| if not plan.get("canExecute"): | |
| return jsonify({ | |
| "mode": "needs_more_info", | |
| "plan": plan, | |
| "reply": plan.get("confirmationMessage") or "I need more information before creating the assignment." | |
| }) | |
| validation = validate_assignment_plan(plan, assignment_context) | |
| if not validation["ok"]: | |
| return jsonify({ | |
| "mode": "blocked", | |
| "reply": validation["message"], | |
| "validationErrors": validation["errors"], | |
| "plan": plan | |
| }) | |
| # Do not write immediately unless frontend sends confirmed=true | |
| if not data.get("confirmed"): | |
| return jsonify({ | |
| "mode": "confirmation_required", | |
| "reply": validation["confirmationMessage"], | |
| "plan": plan | |
| }) | |
| saved = create_assignments_from_plan( | |
| plan=plan, | |
| company_code=company_code, | |
| user_id=user_id, | |
| user_context=ctx, | |
| assignment_context=assignment_context, | |
| ) | |
| return jsonify({ | |
| "mode": "executed", | |
| "reply": f"Created {saved['count']} assignment(s) successfully.", | |
| "createdIds": saved["ids"] | |
| }) | |
| # --- NEW: lightweight Help/Assistant intent router ------------------------- | |
| def assist(): | |
| """ | |
| Classifies a short help question and, if appropriate, returns | |
| a suggested navigation target + tour key for the frontend to action. | |
| Request JSON: | |
| { | |
| "role": "incubatee" | ..., | |
| "message": "Where do I apply?", | |
| "companyCode": "ACME", | |
| "userId": "abc123", | |
| "path": "/current/location" # optional, helps decide replace vs push if you want | |
| } | |
| Response JSON: | |
| { | |
| "reply": "bot text to show", | |
| "navigate": { | |
| "to": "/incubatee/sme", | |
| "tour": "apply", # optional | |
| "ts": 1712345678901, # force re-run effects | |
| "delayMs": 900 # UI should wait before navigating (to show the bot msg) | |
| }, | |
| "handoff": { | |
| "open": false, # whether to auto-open your drawer after nav | |
| "botText": "You're on Programs. Tap an 'Apply' button to begin.", | |
| "followups": ["track_application","profile_setup","inquiries"] | |
| } | |
| } | |
| """ | |
| data = request.json or {} | |
| role = data.get('role') | |
| user_input = (data.get('message') or '').strip().lower() | |
| company_code = data.get('companyCode') | |
| user_id = data.get('userId') | |
| if not role or not user_input or not company_code or not user_id: | |
| return jsonify({"error": "Missing role, message, companyCode, or userId"}), 400 | |
| # Simple keyword intent router (deterministic; fast) | |
| def nav_payload(to: str, tour: Optional[str], reply: str, handoff: Optional[Dict[str, Any]] = None): | |
| return jsonify({ | |
| "reply": reply, | |
| "navigate": { | |
| "to": to, | |
| "tour": tour, | |
| "ts": int(datetime.now().timestamp() * 1000), | |
| "delayMs": 900 | |
| }, | |
| "handoff": handoff or {} | |
| }) | |
| tokens = set(user_input.replace('?', '').split()) | |
| contains = lambda *words: any(w in tokens or w in user_input for w in words) | |
| # 1) Where do I apply? | |
| if contains('apply', 'application', 'register', 'program', 'submit'): | |
| return nav_payload( | |
| to="/incubatee/sme", | |
| tour="apply", | |
| reply="Taking you to Programs and starting the tour…", | |
| handoff={ | |
| "open": False, | |
| "botText": "You're on Programs. Tap an **Apply** button to begin.", | |
| "followups": ["track_application", "profile_setup", "inquiries"] | |
| } | |
| ) | |
| # 2) Track my application | |
| if contains('track', 'status', 'progress'): | |
| return nav_payload( | |
| to="/incubatee/tracker", | |
| tour="track", | |
| reply="Opening your application tracker…", | |
| handoff={ | |
| "open": False, | |
| "botText": "This is your tracker. You can view statuses and details here.", | |
| "followups": ["where_apply", "inquiries"] | |
| } | |
| ) | |
| # 3) Inquiries / helpdesk | |
| if contains('inquiry', 'inquiries', 'question', 'helpdesk', 'support', 'ticket'): | |
| return nav_payload( | |
| to="/incubatee/sme/inquiries", | |
| tour="inquiries", | |
| reply="Taking you to Inquiries…", | |
| handoff={ | |
| "open": True, | |
| "botText": "Here you can log questions or check replies.", | |
| "followups": ["where_apply", "track_application"] | |
| } | |
| ) | |
| # 4) Profile/setup questions | |
| if contains('profile', 'setup', 'account', 'redirect'): | |
| return jsonify({ | |
| "reply": ( | |
| "You need to complete your profile before applying. " | |
| "If something’s missing (e.g., name/company), we’ll redirect you to finish it first." | |
| ), | |
| "handoff": { | |
| "open": True, | |
| "botText": "Would you like to go to **My Profile** now?", | |
| "followups": ["where_apply"] | |
| } | |
| }) | |
| # 5) Contact support | |
| if contains('contact', 'email', 'help', 'assist'): | |
| return jsonify({ | |
| "reply": "You can email support at **support@smartincubation.example** or log an inquiry from the Inquiries page.", | |
| "handoff": { | |
| "open": True, | |
| "botText": "Try the **Inquiries** option if you want to log a ticket.", | |
| "followups": ["inquiries"] | |
| } | |
| }) | |
| # 6) Fallback → optionally leverage GPT to draft a helpful answer (no redirect) | |
| # Keep this light; we're not doing Firestore fetches here. | |
| try: | |
| sys = { | |
| "role": "system", | |
| "content": ( | |
| "You are Q-Bot, a concise, friendly assistant. " | |
| "Clarify the request in one or two sentences and suggest a next step. " | |
| "Do NOT mention databases or internal logic." | |
| ) | |
| } | |
| user = {"role": "user", "content": data.get('message', '')} | |
| answer = ask_gpt([sys, user]) or "I’m here to help! Could you tell me a bit more about what you need?" | |
| return jsonify({ | |
| "reply": answer, | |
| "handoff": { | |
| "open": True, | |
| "followups": ["where_apply", "track_application", "inquiries"] | |
| } | |
| }) | |
| except Exception: | |
| return jsonify({ | |
| "reply": "I’m here to help! Could you tell me a bit more about what you need?", | |
| "handoff": { | |
| "open": True, | |
| "followups": ["where_apply", "track_application", "inquiries"] | |
| } | |
| }) | |
| def analyze_intervention_update(): | |
| """ | |
| AI-assisted intervention update analysis. | |
| Request JSON: | |
| { | |
| "role": "consultant", | |
| "companyCode": "ABC", | |
| "userId": "uid123", | |
| "captureMode": "message|notes|audio", | |
| "rawInput": "what the user typed or transcript", | |
| "intervention": { | |
| "id": "...", | |
| "interventionTitle": "...", | |
| "subtitle": "...", | |
| "beneficiaryName": "...", | |
| "status": "assigned|in-progress|completed", | |
| "targetType": "percentage|number", | |
| "targetMetric": "sessions|hours|reports|...", | |
| "targetValue": 10, | |
| "targetActual": 2, | |
| "progress": 20, | |
| "dueDate": "...", | |
| "notes": "..." | |
| }, | |
| "history": [ | |
| { | |
| "notes": "...", | |
| "progressAfter": 20, | |
| "targetActualAfter": 2, | |
| "createdAt": "..." | |
| } | |
| ] | |
| } | |
| """ | |
| data = request.json or {} | |
| role = data.get('role') | |
| company_code = data.get('companyCode') | |
| user_id = data.get('userId') | |
| capture_mode = data.get('captureMode') | |
| raw_input = (data.get('rawInput') or '').strip() | |
| intervention = data.get('intervention') or {} | |
| history = data.get('history') or [] | |
| if not role or not company_code or not user_id or not capture_mode or not raw_input: | |
| return jsonify({ | |
| "error": "Missing role, companyCode, userId, captureMode, or rawInput" | |
| }), 400 | |
| if capture_mode not in ['message', 'notes', 'audio']: | |
| return jsonify({"error": "Invalid captureMode"}), 400 | |
| metric_key = _normalize_metric(intervention.get("targetMetric")) | |
| metric_rules = _proof_rules_for_metric(metric_key) | |
| prompt_payload = { | |
| "captureMode": capture_mode, | |
| "rawInput": raw_input, | |
| "role": role, | |
| "intervention": { | |
| "id": intervention.get("id"), | |
| "interventionTitle": intervention.get("interventionTitle"), | |
| "subtitle": intervention.get("subtitle"), | |
| "beneficiaryName": intervention.get("beneficiaryName"), | |
| "status": intervention.get("status"), | |
| "targetType": intervention.get("targetType"), | |
| "targetMetric": intervention.get("targetMetric"), | |
| "targetValue": intervention.get("targetValue"), | |
| "targetActual": intervention.get("targetActual"), | |
| "progress": intervention.get("progress"), | |
| "dueDate": to_jsonable(intervention.get("dueDate")), | |
| "notes": intervention.get("notes"), | |
| }, | |
| "history": history[:10], | |
| "rules": { | |
| "metricKey": metric_key, | |
| "proofRules": metric_rules, | |
| "instruction": ( | |
| "Infer what work was likely done from the raw input, " | |
| "but do not freely calculate final progress. " | |
| "Only suggest hours and measurable units if grounded in the text." | |
| ) | |
| } | |
| } | |
| try: | |
| system_msg = { | |
| "role": "system", | |
| "content": ( | |
| "You are an assistant that analyses intervention update notes for a business workflow. " | |
| "You must return strict JSON only." | |
| ) | |
| } | |
| user_msg = { | |
| "role": "user", | |
| "content": _build_intervention_analysis_prompt(prompt_payload) | |
| } | |
| ai_raw = ask_gpt([system_msg, user_msg]) | |
| ai_result = _extract_json_block(ai_raw) | |
| progress_calc = _calculate_progress_suggestion(intervention, ai_result) | |
| merged_proof = _merge_proof_hints(metric_rules, ai_result.get("proofHints", [])) | |
| completion_signal = _norm(ai_result.get("completionSignal")) | |
| readiness = "not_ready" | |
| if completion_signal == "strong": | |
| readiness = "close" | |
| elif progress_calc.get("suggestedProgressPct", 0) >= 100: | |
| readiness = "ready" | |
| response = { | |
| "summary": ai_result.get("summary", ""), | |
| "polishedNotes": ai_result.get("polishedNotes", ""), | |
| "activities": ai_result.get("activities", []), | |
| "outputs": ai_result.get("outputs", []), | |
| "blockers": ai_result.get("blockers", []), | |
| "nextSteps": ai_result.get("nextSteps", []), | |
| "warnings": ai_result.get("warnings", []), | |
| "confidence": ai_result.get("confidence", 0), | |
| "completionSignal": completion_signal or "none", | |
| "completionReadiness": readiness, | |
| "proofSuggestions": merged_proof, | |
| "calculation": progress_calc, | |
| "rawInput": raw_input, | |
| } | |
| return jsonify(to_jsonable(response)) | |
| except Exception as e: | |
| print("analyze_intervention_update_failed:", e) | |
| return jsonify({ | |
| "error": "Failed to analyse intervention update" | |
| }), 500 | |
| def generate_course_outline(): | |
| """ | |
| multipart/form-data: | |
| - file: pdf or docx | |
| - role: operations | consultant | admin | ... | |
| - companyCode: ... | |
| - userId: ... | |
| Response: | |
| { | |
| "reply": "Course outline generated successfully", | |
| "outline": { | |
| "courseTitle": "...", | |
| "courseDescription": "...", | |
| "difficulty": "...", | |
| "category": "...", | |
| "courseType": "...", | |
| "estimatedTotalDuration": "...", | |
| "learningObjectives": [], | |
| "modules": [], | |
| "warnings": [] | |
| }, | |
| "meta": { | |
| "filename": "...", | |
| "contentType": "...", | |
| "extractedChars": 12345, | |
| "truncated": false | |
| } | |
| } | |
| """ | |
| try: | |
| role = request.form.get('role') | |
| company_code = request.form.get('companyCode') | |
| user_id = request.form.get('userId') | |
| uploaded = request.files.get('file') | |
| if not role or not company_code or not user_id: | |
| return jsonify({ | |
| "error": "Missing role, companyCode, or userId" | |
| }), 400 | |
| if uploaded is None: | |
| return jsonify({ | |
| "error": "Missing file" | |
| }), 400 | |
| filename = uploaded.filename or "" | |
| if not _allowed_course_source(filename): | |
| return jsonify({ | |
| "error": "Only PDF and DOCX files are supported" | |
| }), 400 | |
| file_bytes = uploaded.read() | |
| if not file_bytes: | |
| return jsonify({ | |
| "error": "Uploaded file is empty" | |
| }), 400 | |
| extracted_text = _extract_course_source_text(filename, file_bytes) | |
| if not extracted_text: | |
| return jsonify({ | |
| "error": "Could not extract readable text from the uploaded file" | |
| }), 400 | |
| truncated_text = _truncate_source_text(extracted_text) | |
| was_truncated = len(truncated_text) < len(extracted_text) | |
| system_msg = { | |
| "role": "system", | |
| "content": ( | |
| "You generate practical LMS course outlines from uploaded documents. " | |
| "Return strict JSON only." | |
| ) | |
| } | |
| user_msg = { | |
| "role": "user", | |
| "content": _build_course_outline_prompt(truncated_text, filename) | |
| } | |
| ai_raw = ask_gpt([system_msg, user_msg]) | |
| ai_result = _extract_json_block(ai_raw) | |
| outline = _normalize_outline_json(ai_result) | |
| if was_truncated: | |
| outline["warnings"] = outline.get("warnings", []) | |
| outline["warnings"].append( | |
| "The source document was long, so only the first portion was used to generate this outline." | |
| ) | |
| return jsonify({ | |
| "reply": "Course outline generated successfully", | |
| "outline": outline, | |
| "meta": { | |
| "filename": filename, | |
| "contentType": uploaded.content_type, | |
| "extractedChars": len(extracted_text), | |
| "truncated": was_truncated, | |
| } | |
| }) | |
| except Exception as e: | |
| print("generate_course_outline_failed:", e) | |
| return jsonify({ | |
| "error": "Failed to generate course outline from file" | |
| }), 500 | |
| def analyze_sme_application_intake(): | |
| """ | |
| Multipart form-data endpoint for unregistered SME intake. | |
| Expected form fields: | |
| - mode: initial_review | final_decision | |
| - companyCode | |
| - userId optional | |
| - contactName | |
| - contactEmail | |
| - contactPhone | |
| - rawStory | |
| - missingAnswersJson optional | |
| - requiredDocumentsJson optional | |
| - files[] optional | |
| Response: | |
| { | |
| detectedLanguages, | |
| confidence, | |
| extractedProfile, | |
| extractedApplication, | |
| documentFindings, | |
| missingFields, | |
| missingDocuments, | |
| nextQuestion, | |
| decision, | |
| warnings | |
| } | |
| """ | |
| try: | |
| mode = request.form.get("mode") or "initial_review" | |
| company_code = (request.form.get("companyCode") or "").strip() | |
| user_id = request.form.get("userId") or "" | |
| contact_name = request.form.get("contactName") or "" | |
| contact_email = request.form.get("email") or request.form.get("contactEmail") or "" | |
| contact_phone = request.form.get("contactPhone") or "" | |
| raw_story = (request.form.get("rawStory") or "").strip() | |
| # companyCode is optional for public/unregistered SME intake | |
| if not raw_story: | |
| return jsonify({"error": "Missing rawStory"}), 400 | |
| try: | |
| missing_answers = json.loads(request.form.get("missingAnswersJson") or "{}") | |
| except Exception: | |
| missing_answers = {} | |
| try: | |
| required_documents = json.loads(request.form.get("requiredDocumentsJson") or "[]") | |
| except Exception: | |
| required_documents = [] | |
| uploaded_files = request.files.getlist("files") | |
| document_payloads = [] | |
| for uploaded in uploaded_files: | |
| filename = uploaded.filename or "" | |
| if not filename: | |
| continue | |
| if not _allowed_sme_doc(filename): | |
| document_payloads.append({ | |
| "filename": filename, | |
| "contentType": uploaded.content_type, | |
| "extractedText": "", | |
| "warnings": ["Unsupported document type."] | |
| }) | |
| continue | |
| file_bytes = uploaded.read() | |
| if not file_bytes: | |
| document_payloads.append({ | |
| "filename": filename, | |
| "contentType": uploaded.content_type, | |
| "extractedText": "", | |
| "warnings": ["Uploaded file was empty."] | |
| }) | |
| continue | |
| extracted_text = _extract_sme_doc_text(filename, file_bytes) | |
| truncated_text = _truncate_source_text(extracted_text, MAX_SME_DOC_TEXT_CHARS) | |
| warnings = [] | |
| ext = filename.rsplit(".", 1)[1].lower() | |
| if ext in ["png", "jpg", "jpeg"]: | |
| warnings.append("Image OCR is not enabled on this endpoint yet.") | |
| if extracted_text and len(truncated_text) < len(extracted_text): | |
| warnings.append("Document text was truncated before AI analysis.") | |
| if not extracted_text and ext not in ["png", "jpg", "jpeg"]: | |
| warnings.append("No readable text could be extracted from this document.") | |
| document_payloads.append({ | |
| "filename": filename, | |
| "contentType": uploaded.content_type, | |
| "extractedText": truncated_text, | |
| "warnings": warnings | |
| }) | |
| available_programs = _fetch_active_programs(company_code) | |
| intervention_catalog = _fetch_intervention_catalog(company_code) | |
| payload = { | |
| "mode": mode, | |
| "companyCode": company_code, | |
| "userId": user_id, | |
| "contact": { | |
| "name": contact_name, | |
| "email": contact_email, | |
| "phone": contact_phone, | |
| }, | |
| "rawStory": raw_story, | |
| "missingAnswers": missing_answers, | |
| "requiredDocuments": required_documents, | |
| "uploadedDocuments": document_payloads, | |
| "availablePrograms": available_programs, | |
| "interventionCatalog": intervention_catalog, | |
| "analysisDate": datetime.utcnow().isoformat(), | |
| } | |
| system_msg = { | |
| "role": "system", | |
| "content": ( | |
| "You analyse SME intake information for an incubation platform. " | |
| "You return strict JSON only. " | |
| "You must not invent official facts, registration numbers, tax numbers, " | |
| "document statuses, revenue, employees, or compliance claims." | |
| ) | |
| } | |
| user_msg = { | |
| "role": "user", | |
| "content": _build_sme_intake_prompt(payload) | |
| } | |
| ai_raw = ask_gpt([system_msg, user_msg]) | |
| ai_result = _extract_json_block(ai_raw) | |
| normalized = _normalize_sme_intake_result(ai_result) | |
| return jsonify(to_jsonable(normalized)) | |
| except Exception as e: | |
| print("analyze_sme_application_intake_failed:", e) | |
| return jsonify({ | |
| "error": "Failed to analyse SME application intake" | |
| }), 500 | |
| # Assessment extraction (uses this deployment's Firebase project and Gemini client) | |
| from flask_assessment_extraction import create_assessment_blueprint | |
| app.register_blueprint(create_assessment_blueprint(ask_gpt, db)) | |
| if __name__ == "__main__": | |
| app.run(host="0.0.0.0", port=7860) | |